aboutsummaryrefslogtreecommitdiffstats
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/simtest/identity.rs1
-rw-r--r--src/simtest/timing.rs1
-rw-r--r--src/swarm/peer/inbound.rs9
-rw-r--r--src/swarm/peer/keepalive.rs15
-rw-r--r--src/swarm/peer/mod.rs2
-rw-r--r--src/swarm/relay.rs11
-rw-r--r--src/transport/libp2p/tests.rs1
-rw-r--r--src/transport/mod.rs4
-rw-r--r--src/transport/mycelium/mod.rs1
-rw-r--r--src/transport/paths.rs1
-rw-r--r--src/transport/sim.rs1
-rw-r--r--src/transport/wire.rs8
12 files changed, 46 insertions, 9 deletions
diff --git a/src/simtest/identity.rs b/src/simtest/identity.rs
index 7dbb722..04dbed3 100644
--- a/src/simtest/identity.rs
+++ b/src/simtest/identity.rs
@@ -110,6 +110,7 @@ fn status(node: &AppState) -> ParsedMessage {
ready: false,
syncs_with_you: false,
leaving: false,
+ announcing: false,
})
}
diff --git a/src/simtest/timing.rs b/src/simtest/timing.rs
index daa588f..7879ac7 100644
--- a/src/simtest/timing.rs
+++ b/src/simtest/timing.rs
@@ -335,6 +335,7 @@ async fn late_epoch_does_not_flip_a_peers_epoch() {
ready: true,
syncs_with_you: false,
leaving: false,
+ announcing: false,
})
};
let a = &a;
diff --git a/src/swarm/peer/inbound.rs b/src/swarm/peer/inbound.rs
index 18d29d5..1454c3e 100644
--- a/src/swarm/peer/inbound.rs
+++ b/src/swarm/peer/inbound.rs
@@ -143,8 +143,15 @@ pub(crate) async fn handle_message(
crate::swarm::blob::handle(state, &peer, blob).await;
}
transport::ParsedMessage::Status(remote) => {
- status_received(state, &peer).await;
+ // An announcement is no keepalive; it is answered only if our heads differ,
+ // so the announcer learns what we have that it lacks.
+ if !remote.announcing {
+ status_received(state, &peer).await;
+ }
let local = on_status(state, &peer, &remote, reactivated).await;
+ if remote.announcing && local.heads == remote.heads {
+ return Ok(());
+ }
if let Err(error) = state
.transport
.send_message(&peer, &ParsedMessage::StatusReply(local))
diff --git a/src/swarm/peer/keepalive.rs b/src/swarm/peer/keepalive.rs
index 8597dcd..196b8b6 100644
--- a/src/swarm/peer/keepalive.rs
+++ b/src/swarm/peer/keepalive.rs
@@ -32,6 +32,20 @@ pub(crate) async fn send_status(state: &AppState, peer: &str) -> Result<()> {
.await
}
+/// Sends `peer` a STATUS announcing our new heads (`swarm::relay`): no keepalive, so it
+/// leaves the keepalives' schedule as it is, and it is answered only if the peer's heads
+/// differ (`Status::announcing`).
+pub(crate) async fn announce_heads(state: &AppState, peer: &str) -> Result<()> {
+ let status = transport::Status {
+ announcing: true,
+ ..local_status(state, peer).await
+ };
+ state
+ .transport
+ .send_message(peer, &ParsedMessage::Status(status))
+ .await
+}
+
/// Sends `peer` a PING (the paths layer measures the path with it).
pub(crate) async fn send_ping(state: &AppState, peer: &str) -> Result<()> {
let nonce = crate::transport::paths::random_nonce();
@@ -173,6 +187,7 @@ pub(crate) async fn local_status(state: &AppState, peer: &str) -> transport::Sta
ready: state.is_ready().await && !state.is_closing(),
syncs_with_you,
leaving: state.is_closing(),
+ announcing: false,
}
}
diff --git a/src/swarm/peer/mod.rs b/src/swarm/peer/mod.rs
index cf427a2..1826ec6 100644
--- a/src/swarm/peer/mod.rs
+++ b/src/swarm/peer/mod.rs
@@ -26,6 +26,6 @@ pub(crate) use handshake::{
send_hello_once,
};
pub(crate) use inbound::handle_message;
-pub(crate) use keepalive::{announce, farewell, send_status};
+pub(crate) use keepalive::{announce, announce_heads, farewell, send_status};
pub(crate) use lanes::SyncLanes;
pub(crate) use worker::{Due, run_peer_worker};
diff --git a/src/swarm/relay.rs b/src/swarm/relay.rs
index 1393d50..a4a4034 100644
--- a/src/swarm/relay.rs
+++ b/src/swarm/relay.rs
@@ -2,8 +2,9 @@
//!
//! A node pushes its own changes to its neighbours, but nothing it receives goes any
//! further by itself. So after applying changes from one neighbour, a node *announces*
-//! its new heads to its other neighbours with a STATUS (a keepalive: heads, about 100
-//! bytes, not the changes). A neighbour that sees heads it lacks *pulls*: it enters the
+//! its new heads to its other neighbours with a STATUS (heads, about 100 bytes, not the
+//! changes; no keepalive, and answered only by a neighbour whose heads differ). A
+//! neighbour that sees heads it lacks *pulls*: it enters the
//! announced heads as the peer's heads in its sync state, Automerge asks for what's
//! missing (`need`), and the announcer answers with exactly those changes. Each node
//! receives a change about once; the rest is announcements. Regular keepalives work the
@@ -11,7 +12,7 @@
//!
//! Announcements are coalesced: the first applied change schedules one announcement
//! round `ANNOUNCE_DELAY` later, and every change applied until then rides along. A
-//! round costs one STATUS and one STATUS_REPLY per neighbour (plus the pulls), whatever the
+//! round costs one STATUS per neighbour (plus the pulls and their replies), whatever the
//! write rate; the delay is the price per hop. Without coalescing, every applied sync
//! message would announce to every neighbour, and announcements would outnumber the
//! data they announce (about 15 to 1 at 250 nodes).
@@ -31,7 +32,7 @@ use tracing::debug;
use crate::locks::MutexExt;
use crate::state::{AppState, Standing};
-use crate::swarm::peer::send_status;
+use crate::swarm::peer::announce_heads;
use crate::swarm::topology;
use crate::transport::{self, HeadsSummary};
@@ -86,7 +87,7 @@ async fn announce_now(state: &AppState, senders: &HashSet<String>) {
}
state.stats.announce_round();
for peer in neighbours {
- if let Err(error) = send_status(state, &peer).await {
+ if let Err(error) = announce_heads(state, &peer).await {
debug!(peer = %peer, error = %format!("{error:#}"), "Announcing our heads failed");
}
}
diff --git a/src/transport/libp2p/tests.rs b/src/transport/libp2p/tests.rs
index 0f52e23..1d7c6ac 100644
--- a/src/transport/libp2p/tests.rs
+++ b/src/transport/libp2p/tests.rs
@@ -59,6 +59,7 @@ fn status() -> ParsedMessage {
ready: true,
syncs_with_you: false,
leaving: false,
+ announcing: false,
})
}
diff --git a/src/transport/mod.rs b/src/transport/mod.rs
index b8b57af..25f9b7c 100644
--- a/src/transport/mod.rs
+++ b/src/transport/mod.rs
@@ -202,6 +202,10 @@ pub(crate) struct Status {
/// The sender is shutting down: its farewell (`keepalive::farewell`). The
/// recipient holds it dormant at once, rather than after `dormant_after` of silence.
pub(crate) leaving: bool,
+ /// The sender announces new heads (`swarm::relay`), rather than keeping alive: a
+ /// STATUS answered only if the recipient's heads differ. Most announcements reach a
+ /// neighbour that has the changes already, which has nothing to tell.
+ pub(crate) announcing: bool,
}
/// Cheap to clone: large parts are shared (`bytes::Bytes`).
diff --git a/src/transport/mycelium/mod.rs b/src/transport/mycelium/mod.rs
index 9078134..9850ef0 100644
--- a/src/transport/mycelium/mod.rs
+++ b/src/transport/mycelium/mod.rs
@@ -717,6 +717,7 @@ mod tests {
ready: true,
syncs_with_you: true,
leaving: false,
+ announcing: false,
};
let payload = wire::encode(&ParsedMessage::StatusReply(status.clone())).unwrap();
// Mycelium's message data: a length byte, the topic, then our payload.
diff --git a/src/transport/paths.rs b/src/transport/paths.rs
index 164574d..29b3c23 100644
--- a/src/transport/paths.rs
+++ b/src/transport/paths.rs
@@ -1513,6 +1513,7 @@ mod tests {
ready: false,
syncs_with_you: false,
leaving: false,
+ announcing: false,
})
}
diff --git a/src/transport/sim.rs b/src/transport/sim.rs
index 2f167dc..7a679fc 100644
--- a/src/transport/sim.rs
+++ b/src/transport/sim.rs
@@ -387,6 +387,7 @@ mod tests {
ready: true,
syncs_with_you: false,
leaving: false,
+ announcing: false,
})
}
diff --git a/src/transport/wire.rs b/src/transport/wire.rs
index d72fba4..d3bf39d 100644
--- a/src/transport/wire.rs
+++ b/src/transport/wire.rs
@@ -14,7 +14,7 @@ use super::{EncodedMessage, MessageKind, ParsedMessage};
/// The wire format's version, the first byte of every payload. Bump it with every change
/// to a message type or a codec (`golden_bytes` fails then).
-pub(crate) const VERSION: u8 = 15;
+pub(crate) const VERSION: u8 = 16;
/// What the second byte of a payload says.
fn tag(kind: MessageKind) -> u8 {
@@ -369,6 +369,7 @@ pub(crate) fn samples() -> Vec<ParsedMessage> {
ready: true,
syncs_with_you: true,
leaving: false,
+ announcing: false,
};
let proof = IdentityProof {
key: [8; 32],
@@ -549,6 +550,7 @@ mod tests {
ready: true,
syncs_with_you: false,
leaving: false,
+ announcing: false,
};
let decoded =
decode(&encode(&ParsedMessage::StatusReply(status.clone())).unwrap()).unwrap();
@@ -572,6 +574,7 @@ mod tests {
ready: true,
syncs_with_you: true,
leaving: false,
+ announcing: false,
};
assert!(decode(&encode(&ParsedMessage::Status(status)).unwrap()).is_err());
}
@@ -669,10 +672,11 @@ mod tests {
ready: true,
syncs_with_you: false,
leaving: false,
+ announcing: true,
});
assert_eq!(
hex(&encode(&status).unwrap()),
- format!("{v}03969290c400{epoch_hex}01c3c2c2")
+ format!("{v}03979290c400{epoch_hex}01c3c2c2c3")
);
let who = ParsedMessage::Who(IdentityProof {