Verify pushed content by polling observes, not one live watch

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>

varde-daemon/src/transfer.rs

@@ -245,34 +245,36 @@ impl Transfer {
         Ok(bytes)
     }
 
-    /// Watch `hash` on the remote end of `conn` until its bitfield
+    /// Check `hash` on the remote end of `conn` until its bitfield
-    /// reports the blob complete. Bounded: the peer has already received
+    /// reports the blob complete. Polls with fresh observe requests
-    /// the bytes, so verification is bookkeeping, not transfer.
+    /// 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 deadline = tokio::time::Instant::now() + Duration::from_secs(60);
+        loop {
             let observe = self.store.remote().observe(
                 conn.clone(),
                 iroh_blobs::protocol::ObserveRequest::new(hash),
             );
             let mut observe = std::pin::pin!(observe);
-        loop {
+            match tokio::time::timeout(Duration::from_secs(10), observe.next()).await {
-            let next = tokio::time::timeout(Duration::from_secs(60), observe.next())
+                Ok(Some(Ok(bitfield))) if bitfield.is_complete() => return Ok(()),
-                .await
+                // Present but not complete yet, or no snapshot in time:
-                .map_err(|_| anyhow::anyhow!("verifying pushed content on the peer timed out"))?;
+                // poll again from scratch.
-            match next {
+                Ok(Some(Ok(_))) | Ok(None) | Err(_) => {}
-                Some(bitfield) => {
+                Ok(Some(Err(e))) => return Err(e).context("observing pushed content on the peer"),
-                    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")
             }
+            if tokio::time::Instant::now() >= deadline {
+                anyhow::bail!("verifying pushed content on the peer timed out");
             }
+            tokio::time::sleep(Duration::from_millis(500)).await;
         }
     }