From ebd2034d67d8f9184fc7238b40e57d810b57ed19 Mon Sep 17 00:00:00 2001 From: Vladimir Logachev Date: Thu, 16 Jul 2026 01:40:11 +0400 Subject: [PATCH] fix(sync): retry missing content downloads on every successful sync --- src/engine/live.rs | 76 +++++++++++++++++++---- tests/sync.rs | 150 +++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 215 insertions(+), 11 deletions(-) diff --git a/src/engine/live.rs b/src/engine/live.rs index 7fe587fc..1238932d 100644 --- a/src/engine/live.rs +++ b/src/engine/live.rs @@ -166,8 +166,15 @@ pub struct LiveActor { running_sync_accept: JoinSet, /// Running download futures. download_tasks: JoinSet, - /// Content hashes which are wanted but not yet queued because no provider was found. - missing_hashes: HashSet, + /// 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 = self + .queued_hashes + .by_namespace + .get(&namespace) + .map(|hashes| hashes.iter().copied().collect()) + .unwrap_or_default(); + let parked: Vec = 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,6 +691,7 @@ 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; @@ -658,7 +699,14 @@ impl LiveActor { 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,11 +796,6 @@ 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() @@ -760,9 +803,20 @@ impl LiveActor { .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, diff --git a/tests/sync.rs b/tests/sync.rs index 6a096c78..604feeb2 100644 --- a/tests/sync.rs +++ b/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> + 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(()) +}