fix(sync): retry missing content downloads on every successful sync

src/engine/live.rs

@@ -166,8 +166,15 @@ pub struct LiveActor {
     running_sync_accept: JoinSet<SyncAcceptRes>,
     /// Running download futures.
     download_tasks: JoinSet<DownloadRes>,
-    /// Content hashes which are wanted but not yet queued because no provider was found.
-    missing_hashes: HashSet<Hash>,
+    /// Content hashes which are wanted but not yet queued because no provider was found,
+    /// keyed by the namespace whose entry wants them (the namespace drives retries on
+    /// sync-finished and keeps `PendingContentReady` attribution correct).
+    missing_hashes: HashSet<(NamespaceId, Hash)>,
+    /// Queued content whose running download should be retried once if it fails: a fresh
+    /// provider was registered (through a finished sync) after the running download had
+    /// already snapshotted its provider set, so the provider is only reachable by a new
+    /// download attempt.
+    retry_after_failure: HashSet<(NamespaceId, Hash)>,
     /// Content hashes queued in downloader.
     queued_hashes: QueuedHashes,
     /// Nodes known to have a hash
@@ -214,6 +221,7 @@ impl LiveActor {
             download_tasks: Default::default(),
             state: Default::default(),
             missing_hashes: Default::default(),
+            retry_after_failure: Default::default(),
             queued_hashes: Default::default(),
             hash_providers: Default::default(),
             metrics,
@@ -566,6 +574,38 @@ impl LiveActor {
                     debug!(%e, "failed to register peer for document")
                 }
 
+                // Retry content that is still missing for this namespace: the peer we
+                // just synced with is a fresh provider candidate. Entries whose records
+                // arrive ahead of their content are parked in `missing_hashes` and are
+                // otherwise unparked only by a best-effort gossip `ContentReady`
+                // broadcast — if that one message is lost, the content would starve
+                // until an unrelated insert. `start_download` skips content that
+                // arrived in the meantime and dedupes in-flight downloads. Content
+                // already being downloaded gets the peer registered as a provider,
+                // and the download is retried once if it fails: the running attempt
+                // snapshotted its provider set before this peer joined it.
+                let queued: Vec<Hash> = self
+                    .queued_hashes
+                    .by_namespace
+                    .get(&namespace)
+                    .map(|hashes| hashes.iter().copied().collect())
+                    .unwrap_or_default();
+                let parked: Vec<Hash> = self
+                    .missing_hashes
+                    .iter()
+                    .filter(|(ns, _)| *ns == namespace)
+                    .map(|(_, hash)| *hash)
+                    .collect();
+                for hash in parked {
+                    debug!(peer=%peer.fmt_short(), %hash, "retrying parked content");
+                    self.start_download(namespace, hash, peer, true).await;
+                }
+                for hash in queued {
+                    debug!(peer=%peer.fmt_short(), %hash, "registering sync peer for queued content");
+                    self.retry_after_failure.insert((namespace, hash));
+                    self.start_download(namespace, hash, peer, true).await;
+                }
+
                 // broadcast a sync report to our neighbors, but only if we received new entries.
                 if details.outcome.num_recv > 0 {
                     info!("broadcast sync report to neighbors");
@@ -651,14 +691,22 @@ impl LiveActor {
         let completed_namespaces = self.queued_hashes.remove_hash(&hash);
         debug!(namespace=%namespace.fmt_short(), success=res.is_ok(), completed_namespaces=completed_namespaces.len(), "download ready");
         if res.is_ok() {
+            self.retry_after_failure.retain(|(_, h)| *h != hash);
             self.subscribers
                 .send(&namespace, Event::ContentReady { hash })
                 .await;
             // Inform our neighbors that we have new content ready.
             self.broadcast_neighbors(namespace, &Op::ContentReady(hash))
                 .await;
         } else {
-            self.missing_hashes.insert(hash);
+            self.missing_hashes.insert((namespace, hash));
+            if self.retry_after_failure.remove(&(namespace, hash)) {
+                // A provider was registered while the failed download was already
+                // running with an older provider snapshot: retry once with the
+                // enriched set.
+                debug!(%hash, "retrying failed download with providers registered meanwhile");
+                self.queue_download(namespace, hash, true).await;
+            }
         }
         for namespace in completed_namespaces.iter() {
             if let Some(true) = self.state.may_emit_ready(namespace) {
@@ -732,7 +780,7 @@ impl LiveActor {
                         let node_id = PublicKey::from_bytes(&from)?;
                         self.start_download(namespace, hash, node_id, false).await;
                     } else {
-                        self.missing_hashes.insert(hash);
+                        self.missing_hashes.insert((namespace, hash));
                     }
                 }
             }
@@ -748,21 +796,27 @@ impl LiveActor {
         node: PublicKey,
         only_if_missing: bool,
     ) {
-        let entry_status = self.bao_store.blobs().status(hash).await;
-        if matches!(entry_status, Ok(BlobStatus::Complete { .. })) {
-            self.missing_hashes.remove(&hash);
-            return;
-        }
         self.hash_providers
             .0
             .lock()
             .expect("poisoned")
             .entry(hash)
             .or_default()
             .insert(node);
+        self.queue_download(namespace, hash, only_if_missing).await;
+    }
+
+    /// Queue a download for `hash` from the providers registered so far, unless the
+    /// content is already complete or a download is already running.
+    async fn queue_download(&mut self, namespace: NamespaceId, hash: Hash, only_if_missing: bool) {
+        let entry_status = self.bao_store.blobs().status(hash).await;
+        if matches!(entry_status, Ok(BlobStatus::Complete { .. })) {
+            self.missing_hashes.remove(&(namespace, hash));
+            return;
+        }
         if self.queued_hashes.contains_hash(&hash) {
             self.queued_hashes.insert(hash, namespace);
-        } else if !only_if_missing || self.missing_hashes.contains(&hash) {
+        } else if !only_if_missing || self.missing_hashes.contains(&(namespace, hash)) {
             let req = DownloadRequest::new(
                 HashAndFormat::raw(hash),
                 self.hash_providers.clone(),
@@ -771,7 +825,7 @@ impl LiveActor {
             let handle = self.downloader.download_with_opts(req);
 
             self.queued_hashes.insert(hash, namespace);
-            self.missing_hashes.remove(&hash);
+            self.missing_hashes.remove(&(namespace, hash));
             self.download_tasks.spawn(async move {
                 (
                     namespace,

tests/sync.rs

@@ -1384,3 +1384,153 @@ fn match_sync_finished(event: &LiveEvent, peer: PublicKey) -> bool {
     };
     e.peer == peer && e.result.is_ok()
 }
+
+/// Receive events from the stream until one matches, discarding the rest.
+/// Panics when the timeout elapses first. For scenarios where gossip emits
+/// a nondeterministic set of surrounding events (neighbor changes, repeated
+/// readiness notifications) and only one specific event is the checkpoint.
+async fn next_event_matching(
+    stream: &mut (impl Stream<Item = Result<LiveEvent>> + Unpin + Send),
+    timeout: Duration,
+    matcher: impl Fn(&LiveEvent) -> bool,
+) -> LiveEvent {
+    let fut = async {
+        loop {
+            let event = stream
+                .try_next()
+                .await
+                .expect("event stream errored")
+                .expect("event stream ended");
+            if matcher(&event) {
+                break event;
+            }
+        }
+    };
+    n0_future::time::timeout(timeout, fut)
+        .await
+        .expect("timeout waiting for matching event")
+}
+
+/// A record can arrive from a peer that does not have the record's content:
+/// the sender is a relay that received the record but never fetched the
+/// bytes. The receiver then has no provider to download from, and the
+/// content hash is parked. Unparking used to depend solely on a
+/// best-effort gossip `ContentReady` broadcast from some neighbor that
+/// downloaded the content; when no such broadcast ever comes — nobody else
+/// downloads, or the message is lost — the content starved forever, even
+/// though the receiver keeps completing sync exchanges with peers that do
+/// have the bytes. Every successful sync now retries the namespace's
+/// parked hashes against the just-synced peer, so the first sync with a
+/// peer that has the content delivers it.
+///
+/// The writer leaves the document while the receiver joins, so the record
+/// can only reach the receiver through the relay — without that, gossip
+/// may connect the receiver to the writer directly and deliver the content
+/// through an ordinary insert, masking the starvation.
+#[tokio::test]
+#[traced_test]
+async fn sync_fetches_parked_content_from_later_sync_peer() -> Result<()> {
+    let mut rng = test_rng(b"sync_fetches_parked_content_from_later_sync_peer");
+
+    // Three nodes on loopback: the test drives every exchange explicitly
+    // and must not depend on the host's external interfaces.
+    let mut nodes = Vec::new();
+    for _ in 0..3 {
+        let secret_key = SecretKey::from_bytes(&rng.random());
+        let ep = Endpoint::builder(presets::Minimal)
+            .secret_key(secret_key)
+            .bind_addr("127.0.0.1:0")?
+            .bind()
+            .await?;
+        nodes.push(Node::memory(ep).spawn().await?);
+    }
+    let (writer, relay, receiver) = (&nodes[0], &nodes[1], &nodes[2]);
+    let writer_id = writer.id();
+    let relay_id = relay.id();
+
+    // The writer holds the only copy of the content.
+    let author = writer.docs().author_create().await?;
+    let doc_writer = writer.docs().create().await?;
+    let hash = doc_writer
+        .set_bytes(author, b"k".to_vec(), b"v".to_vec())
+        .await?;
+    let ticket_writer = doc_writer
+        .share(ShareMode::Read, AddrInfoOptions::RelayAndAddresses)
+        .await?;
+
+    // The relay receives the record but never the content: its download
+    // policy forbids fetching anything, and is set before its first sync so
+    // no download can race it.
+    let doc_relay = relay
+        .docs()
+        .import_namespace(ticket_writer.capability.clone())
+        .await?;
+    doc_relay
+        .set_download_policy(DownloadPolicy::NothingExcept(vec![]))
+        .await?;
+    let mut events_relay = doc_relay.subscribe().await?;
+    doc_relay.start_sync(ticket_writer.nodes.clone()).await?;
+    next_event_matching(
+        &mut events_relay,
+        TIMEOUT,
+        |e| matches!(e, LiveEvent::InsertRemote { from, .. } if *from == writer_id),
+    )
+    .await;
+    assert!(
+        !relay.blobs().has(hash).await?,
+        "the relay's download policy must keep it content-less"
+    );
+
+    // The writer steps away: from here on the record can reach the receiver
+    // only through the relay.
+    doc_writer.leave().await?;
+
+    // The receiver learns the record from the relay — the only live peer —
+    // so the record arrives with the content missing at its sender, and the
+    // content hash parks with no usable provider.
+    let ticket_relay = doc_relay
+        .share(ShareMode::Read, AddrInfoOptions::RelayAndAddresses)
+        .await?;
+    let (doc_receiver, mut events_receiver) = receiver
+        .docs()
+        .import_and_subscribe(ticket_relay.clone())
+        .await?;
+    next_event_matching(&mut events_receiver, TIMEOUT, |e| {
+        matches!(
+            e,
+            LiveEvent::InsertRemote { from, content_status: ContentStatus::Missing, .. }
+                if *from == relay_id
+        )
+    })
+    .await;
+    assert!(
+        !receiver.blobs().has(hash).await?,
+        "the receiver cannot have content its only peer does not hold"
+    );
+
+    // The writer returns, and the receiver completes a sync with it. The
+    // sync exchanges no records — the receiver already has them all — so
+    // before the fix nothing requested the parked content and it never
+    // arrived; now the finished sync retries the parked hash against the
+    // writer, which has the bytes.
+    doc_writer.start_sync(vec![]).await?;
+    doc_receiver.start_sync(ticket_writer.nodes.clone()).await?;
+    next_event_matching(&mut events_receiver, TIMEOUT, |e| {
+        match_sync_finished(e, writer_id)
+    })
+    .await;
+    let deadline = Instant::now() + TIMEOUT;
+    while !receiver.blobs().has(hash).await? {
+        assert!(
+            Instant::now() < deadline,
+            "parked content did not arrive from a later sync peer"
+        );
+        n0_future::time::sleep(Duration::from_millis(100)).await;
+    }
+    assert_latest(receiver.blobs(), &doc_receiver, b"k", b"v").await;
+
+    for node in nodes {
+        node.shutdown().await?;
+    }
+    Ok(())
+}