feat(T3.5): speculative blob prefetch from peers during idle periods

- services.rs: add prefetch_task that wakes every 10 min, checks if
  hot-tier fill is below 60%, then pulls up to 5 blobs per live peer
  that this node doesn't already have locally; sets the ref locally
  after each successful pull so future GetRef calls are pure local hits
- rpc.rs: make pull_blob_locally pub(crate) so services.rs can call it;
  add outbound_client() accessor on RpcRouter for the prefetch task

The prefetch task is a no-op when hot-tier fill >= 60% or when the node
has no outbound QUIC client configured (i.e. in test environments).

Co-Authored-By: Claude Sonnet 4.6 <[email protected]>

claw-store/src/cluster/rpc.rs

@@ -869,6 +869,14 @@ impl RpcRouter {
         &self.metrics
     }
 
+    /// Read-only access to the outbound QUIC client. Used by T3.5
+    /// prefetch task in `ClusterServices` to open peer connections.
+    pub fn outbound_client(
+        &self,
+    ) -> Option<Arc<crate::cluster::transport::QuicClient>> {
+        self.outbound_client.clone()
+    }
+
     /// Attach a local blob store. Enables the `Blob*` methods; nodes
     /// without a store return [`ErrorCode::NotConfigured`] for those.
     pub fn with_blob_store(mut self, store: Arc<BlobStore>) -> Self {
@@ -1839,7 +1847,7 @@ impl RpcRouter {
 /// missing chunks from `conn` into the local `BlobStore`. Same shape
 /// as `prewarm_missing_chunks_between_parallel` but the downstream is
 /// in-process rather than another peer.
-async fn pull_blob_locally(
+pub(crate) async fn pull_blob_locally(
     conn: &Connection,
     local: &BlobStore,
     id: &crate::cluster::blob::BlobId,

claw-store/src/cluster/services.rs

@@ -38,6 +38,20 @@ const HOT_METRIC_INTERVAL: Duration = Duration::from_secs(30);
 /// cadence still sees at-worst-60s-stale counters.
 const CACHE_METRIC_INTERVAL: Duration = Duration::from_secs(60);
 
+/// T3.5: how often the idle prefetch task wakes and checks peers.
+/// 10-minute cadence keeps cross-node traffic minimal; each tick pulls
+/// at most `PREFETCH_MAX_PER_TICK` blobs per peer.
+const PREFETCH_INTERVAL: Duration = Duration::from_secs(600);
+
+/// T3.5: maximum blobs pulled from any single peer per prefetch tick.
+/// Caps burst bandwidth to ≈ N × blob_size during each idle window.
+const PREFETCH_MAX_PER_TICK: usize = 5;
+
+/// T3.5: hot-tier fill fraction below which the node is considered idle
+/// enough to speculate-pull blobs from peers. Above this threshold we
+/// don't want to grow disk usage further.
+const PREFETCH_IDLE_THRESHOLD: f64 = 0.60;
+
 /// Live cluster services attached to a running daemon.
 ///
 /// Drop shuts down all background tasks. Ownership is single: the
@@ -84,6 +98,9 @@ pub struct ClusterServices {
     /// `cluster.gc_interval_hours` is unset or the daemon has no blob
     /// store (nothing to sweep).
     gc_task: Option<JoinHandle<()>>,
+    /// T3.5: speculative blob prefetch from peers during idle periods.
+    /// Active only when the daemon has both a blob store and a ref store.
+    prefetch_task: Option<JoinHandle<()>>,
 }
 
 impl std::fmt::Debug for ClusterServices {
@@ -471,6 +488,132 @@ impl ClusterServices {
             _ => None,
         };
 
+        // T3.5: speculative blob prefetch during idle periods. Active only
+        // when we have an outbound RPC client (blob store + ref store +
+        // outbound_client all wired). The task wakes every PREFETCH_INTERVAL,
+        // checks if the node is idle (hot-tier < PREFETCH_IDLE_THRESHOLD),
+        // and pulls up to PREFETCH_MAX_PER_TICK blobs from each live peer.
+        let prefetch_task = match (&blob_store, &ref_store, &router) {
+            (Some(blob_store), Some(ref_store), Some(router))
+                if router.outbound_client().is_some() =>
+            {
+                let blob_store = blob_store.clone();
+                let ref_store = ref_store.clone();
+                let gossip = gossip.clone();
+                let router = router.clone();
+                Some(tokio::spawn(async move {
+                    use crate::cluster::gossip::keys;
+                    use crate::cluster::rpc::{call_dashboard_storage, pull_blob_locally};
+                    use crate::cluster::blob::BlobId;
+                    let mut ticker = tokio::time::interval(PREFETCH_INTERVAL);
+                    ticker.tick().await; // skip immediate first tick
+                    loop {
+                        ticker.tick().await;
+                        // Idle check: hot-tier fill below threshold.
+                        let hot_used = gossip
+                            .self_kv(keys::HOT_USED_BYTES)
+                            .await
+                            .and_then(|s| s.parse::<u64>().ok())
+                            .unwrap_or(0);
+                        let hot_max = gossip
+                            .self_kv(keys::HOT_MAX_BYTES)
+                            .await
+                            .and_then(|s| s.parse::<u64>().ok())
+                            .unwrap_or(1);
+                        let fill = if hot_max > 0 {
+                            hot_used as f64 / hot_max as f64
+                        } else {
+                            1.0
+                        };
+                        if fill >= PREFETCH_IDLE_THRESHOLD {
+                            continue;
+                        }
+                        // Prefetch from each live peer.
+                        let peers = gossip
+                            .peers()
+                            .await
+                            .into_iter()
+                            .filter(|p| p.alive && p.rpc_lan.or(p.rpc_tailscale).is_some())
+                            .collect::<Vec<_>>();
+                        for peer in peers {
+                            let Some(client) = router.outbound_client() else { break };
+                            let addr = match peer.rpc_lan.or(peer.rpc_tailscale) {
+                                Some(a) => a,
+                                None => continue,
+                            };
+                            let conn = match tokio::time::timeout(
+                                Duration::from_secs(5),
+                                client.connect(addr, &peer.name),
+                            )
+                            .await
+                            {
+                                Ok(Ok(c)) => c,
+                                _ => continue,
+                            };
+                            let storage = match call_dashboard_storage(&conn).await {
+                                Ok(s) => s,
+                                Err(e) => {
+                                    tracing::debug!(
+                                        peer = %peer.name,
+                                        error = %e,
+                                        "prefetch: DashboardStorage failed"
+                                    );
+                                    continue;
+                                }
+                            };
+                            let mut pulled = 0usize;
+                            for r in &storage.refs_sample {
+                                if pulled >= PREFETCH_MAX_PER_TICK {
+                                    break;
+                                }
+                                let fp_id = match BlobId::from_hex(&r.fingerprint_hex) {
+                                    Ok(id) => id,
+                                    Err(_) => continue,
+                                };
+                                let fp: [u8; 32] = *fp_id.as_bytes();
+                                // Skip refs we already have locally.
+                                if ref_store.contains(&fp).await.unwrap_or(true) {
+                                    continue;
+                                }
+                                let blob_id = match BlobId::from_hex(&r.blob_id_hex) {
+                                    Ok(id) => id,
+                                    Err(_) => continue,
+                                };
+                                // Pull chunks + manifest from peer.
+                                match pull_blob_locally(&conn, &blob_store, &blob_id).await {
+                                    Ok(()) => {
+                                        let _ = ref_store.put(&fp, blob_id.as_bytes()).await;
+                                        pulled += 1;
+                                        tracing::debug!(
+                                            peer = %peer.name,
+                                            blob = %r.blob_id_hex,
+                                            "prefetch: pulled blob"
+                                        );
+                                    }
+                                    Err(e) => {
+                                        tracing::debug!(
+                                            peer = %peer.name,
+                                            blob = %r.blob_id_hex,
+                                            error = %e,
+                                            "prefetch: pull failed"
+                                        );
+                                    }
+                                }
+                            }
+                            if pulled > 0 {
+                                tracing::info!(
+                                    peer = %peer.name,
+                                    count = pulled,
+                                    "prefetch: idle-pulled blobs from peer"
+                                );
+                            }
+                        }
+                    }
+                }))
+            }
+            _ => None,
+        };
+
         Ok(Self {
             gossip,
             blob_store,
@@ -482,6 +625,7 @@ impl ClusterServices {
             cache_metric_task,
             prom_server,
             gc_task,
+            prefetch_task,
         })
     }
 
@@ -524,6 +668,9 @@ impl ClusterServices {
         if let Some(task) = self.gc_task {
             task.abort();
         }
+        if let Some(task) = self.prefetch_task {
+            task.abort();
+        }
     }
 }