diff --git a/varde-daemon/src/transfer.rs b/varde-daemon/src/transfer.rs index e977d27..9160fb9 100644 --- a/varde-daemon/src/transfer.rs +++ b/varde-daemon/src/transfer.rs @@ -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; } }