diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/swarm/peer/inbound.rs | 24 | ||||
| -rw-r--r-- | src/swarm/peer/keepalive.rs | 4 | ||||
| -rw-r--r-- | src/swarm/reach.rs | 8 | ||||
| -rw-r--r-- | src/swarm/relay.rs | 8 | ||||
| -rw-r--r-- | src/transport/mod.rs | 4 |
5 files changed, 36 insertions, 12 deletions
diff --git a/src/swarm/peer/inbound.rs b/src/swarm/peer/inbound.rs index 1454c3e..28066cb 100644 --- a/src/swarm/peer/inbound.rs +++ b/src/swarm/peer/inbound.rs @@ -143,13 +143,31 @@ pub(crate) async fn handle_message( crate::swarm::blob::handle(state, &peer, blob).await; } transport::ParsedMessage::Status(remote) => { - // An announcement is no keepalive; it is answered only if our heads differ, - // so the announcer learns what we have that it lacks. + // An announcement is no keepalive; it is answered only if we lack some of the + // heads it announced (`relay::lacks`), so the announcer learns what we have + // that it lacks while we pull. Having all of them, we are as far or further, + // and what we have further we announced ourselves; differing heads alone are + // no reason, as changes on their way to us make them differ while writes go on. if !remote.announcing { status_received(state, &peer).await; } let local = on_status(state, &peer, &remote, reactivated).await; - if remote.announcing && local.heads == remote.heads { + if remote.announcing { + if local.heads != remote.heads { + // Off the consumer: telling which heads we lack takes the document's + // lock. + let (state, peer) = (state.clone(), peer.clone()); + tokio::spawn(async move { + if crate::swarm::relay::lacks(&state, &remote.heads).await + && let Err(error) = state + .transport + .send_message(&peer, &ParsedMessage::StatusReply(local)) + .await + { + debug!(peer = %peer, error = %format!("{error:#}"), "STATUS_REPLY failed"); + } + }); + } return Ok(()); } if let Err(error) = state diff --git a/src/swarm/peer/keepalive.rs b/src/swarm/peer/keepalive.rs index 8362d73..f4e961f 100644 --- a/src/swarm/peer/keepalive.rs +++ b/src/swarm/peer/keepalive.rs @@ -33,8 +33,8 @@ pub(crate) async fn send_status(state: &AppState, peer: &str) -> Result<()> { } /// 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`). Whom we reach only `with_reaches` (`swarm::reach`): +/// leaves the keepalives' schedule as it is, and it is answered only if the peer lacks +/// some of our heads (`Status::announcing`). Whom we reach only `with_reaches` (`swarm::reach`): /// announcements of heads go often. pub(crate) async fn announce_heads(state: &AppState, peer: &str, with_reaches: bool) -> Result<()> { let local = local_status(state, peer).await; diff --git a/src/swarm/reach.rs b/src/swarm/reach.rs index 7bd0f73..5c89d04 100644 --- a/src/swarm/reach.rs +++ b/src/swarm/reach.rs @@ -72,8 +72,8 @@ pub(crate) fn published(state: &AppState) -> Vec<u64> { /// With the peer worker's periodic pass (every 5 s), and at once when a member we said /// we reach stops answering (`reached_stops_answering`): whom we reach now; if it changed, -/// our neighbours are told at once (a STATUS each, unanswered if the heads are the -/// same). A change that is undone within a pass isn't told. +/// our neighbours are told at once (a STATUS each, unanswered unless they lack some of +/// our heads). A change that is undone within a pass isn't told. pub(crate) async fn check(state: &AppState) { let now = measure(state).await; { @@ -109,8 +109,8 @@ pub(crate) fn reached_stops_answering(state: &AppState, peer: &str) -> bool { .is_some_and(|keys| keys.contains(&topology::pair_key(peer))) } -/// A STATUS to `peer` with whom we reach: as an announcement (unanswered if the heads -/// are the same), but with `reaches`. +/// A STATUS to `peer` with whom we reach: as an announcement (unanswered unless it lacks +/// some of our heads), but with `reaches`. async fn tell(state: &AppState, peer: &str) -> anyhow::Result<()> { announce_heads(state, peer, true).await } diff --git a/src/swarm/relay.rs b/src/swarm/relay.rs index 53c1065..99c1381 100644 --- a/src/swarm/relay.rs +++ b/src/swarm/relay.rs @@ -5,7 +5,7 @@ //! further by itself but along a writer's tree. So after applying changes from one //! neighbour, a node *announces* //! 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 +//! changes; no keepalive, and answered only by a neighbour that lacks some of them). 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. Regular @@ -218,6 +218,12 @@ pub(crate) struct Pulling { waiting: HashMap<String, transport::Status>, } +/// Whether we lack any of `heads` (with a digest, any of the sample it lists). +pub(crate) async fn lacks(state: &AppState, heads: &HeadsSummary) -> bool { + let replica = state.doc.read().await; + !crate::document::changes::missing(&replica.doc, heads.listed()).is_empty() +} + /// A keepalive from `peer`: if it has heads we lack, pull them (one peer at a time; a /// busy slot queues the peer). Returns whether a pull started. pub(crate) async fn pull_if_behind( diff --git a/src/transport/mod.rs b/src/transport/mod.rs index e27ce2d..17d6d4c 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -203,8 +203,8 @@ pub(crate) struct Status { /// 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. + /// STATUS answered only if the recipient lacks some of the heads. Most announcements + /// reach a neighbour that has the changes already, which has nothing to tell. pub(crate) announcing: bool, /// Whom the sender reaches (`swarm::reach`): the pair keys (`topology::pair_key`) /// of the members it has an active session with that answer, sorted. In keepalive STATUSes and |