uhhm/varde
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

Verify pushed content by polling observes, not one live watch

Browse Source 
CI showed the delivery verification timing out while the push itself
succeeded: an observe watch opened while the peer's import is still
creating the blob can miss the entry and never report again (locally
the import always won the race, so a single complete bitfield arrived
and the watch looked fine). Poll with fresh short-lived observe
requests instead — every iteration reads the peer's current state, so
the race disappears.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Bendik Lynghaug
2026-08-16 15:50:24 +02:00
co-authored by Claude Fable 5
parent b53b5cc000
commit e762e0ca9f
1 changed files with 23 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
varde-daemon/src/transfer.rs
+23 -21
View File
@@ -245,34 +245,36 @@ impl Transfer {
Ok(bytes)
}
/// Watch `hash` on the remote end of `conn` until its bitfield
/// reports the blob complete. Bounded: the peer has already received
/// the bytes, so verification is bookkeeping, not transfer.
/// Check `hash` on the remote end of `conn` until its bitfield
/// reports the blob complete. Polls with fresh observe requests
/// rather than holding one live watch: a watch opened while the
/// peer's import is still creating the blob can miss the entry and
/// stay silent forever, while a fresh request always reads current
/// state. Bounded: the peer has already received the bytes, so
/// verification is bookkeeping, not transfer.
async fn wait_remote_complete(
&self,
conn: &iroh::endpoint::Connection,
hash: Hash,
) -> Result<()> {
let observe = self.store.remote().observe(
conn.clone(),
iroh_blobs::protocol::ObserveRequest::new(hash),
);
let mut observe = std::pin::pin!(observe);
let deadline = tokio::time::Instant::now() + Duration::from_secs(60);
loop {
let next = tokio::time::timeout(Duration::from_secs(60), observe.next())
.await
.map_err(|_| anyhow::anyhow!("verifying pushed content on the peer timed out"))?;
match next {
Some(bitfield) => {
let bitfield = bitfield.context("observing pushed content on the peer")?;
if bitfield.is_complete() {
return Ok(());
}
}
None => {
anyhow::bail!("peer stopped reporting before the pushed content completed")
}
let observe = self.store.remote().observe(
conn.clone(),
iroh_blobs::protocol::ObserveRequest::new(hash),
);
let mut observe = std::pin::pin!(observe);
match tokio::time::timeout(Duration::from_secs(10), observe.next()).await {
Ok(Some(Ok(bitfield))) if bitfield.is_complete() => return Ok(()),
// Present but not complete yet, or no snapshot in time:
// poll again from scratch.
Ok(Some(Ok(_))) | Ok(None) | Err(_) => {}
Ok(Some(Err(e))) => return Err(e).context("observing pushed content on the peer"),
}
if tokio::time::Instant::now() >= deadline {
anyhow::bail!("verifying pushed content on the peer timed out");
}
tokio::time::sleep(Duration::from_millis(500)).await;
}
}