diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/simtest/identity.rs | 1 | ||||
| -rw-r--r-- | src/simtest/timing.rs | 1 | ||||
| -rw-r--r-- | src/swarm/peer/inbound.rs | 9 | ||||
| -rw-r--r-- | src/swarm/peer/keepalive.rs | 15 | ||||
| -rw-r--r-- | src/swarm/peer/mod.rs | 2 | ||||
| -rw-r--r-- | src/swarm/relay.rs | 11 | ||||
| -rw-r--r-- | src/transport/libp2p/tests.rs | 1 | ||||
| -rw-r--r-- | src/transport/mod.rs | 4 | ||||
| -rw-r--r-- | src/transport/mycelium/mod.rs | 1 | ||||
| -rw-r--r-- | src/transport/paths.rs | 1 | ||||
| -rw-r--r-- | src/transport/sim.rs | 1 | ||||
| -rw-r--r-- | src/transport/wire.rs | 8 |
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 { |
