claw-store/src/cluster/rpc.rs

@@ -193,6 +193,13 @@ pub enum Method {
     /// `payload`: empty.
     /// Reply: JSON `DashboardStatusReply`.
     DashboardStatus = 0x1c,
+    /// Dashboard-v2: storage-inventory payload (tags, snapshots,
+    /// ref-tracking, plus sampled blob + ref lists). Fed to the
+    /// aggregator's storage-browser tabs.
+    ///
+    /// `payload`: empty.
+    /// Reply: JSON `DashboardStorageReply`.
+    DashboardStorage = 0x1d,
 }
 
 impl Method {
@@ -228,6 +235,7 @@ impl Method {
             0x1a => Some(Method::SetTagExpiry),
             0x1b => Some(Method::GetTagExpiry),
             0x1c => Some(Method::DashboardStatus),
+            0x1d => Some(Method::DashboardStorage),
             _ => None,
         }
     }
@@ -346,6 +354,61 @@ pub struct PutManifestReply {
 ///
 /// Not `Clone` on its own — wrap in `Arc<RpcRouter>` so a single
 /// instance backs the accept loop plus any explicit dispatch calls.
+/// Dashboard-v2 storage-inventory payload. Small-and-bounded lists
+/// (tags, snapshots, ref-tracking) come back in full. Large lists
+/// (blobs, refs) come back as bounded samples — pagination lives
+/// in a follow-on RPC when it's actually needed.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct DashboardStorageReply {
+    pub node_name: String,
+    pub tags: Vec<DashboardTag>,
+    pub snapshots: Vec<DashboardSnapshot>,
+    pub ref_tracking: Vec<DashboardRefTracking>,
+    /// Up to 200 blob summaries (id + size + chunk count). Ordered
+    /// by blob-id hex ascending.
+    pub blobs_sample: Vec<DashboardBlob>,
+    pub blobs_sample_capped_at: usize,
+    /// Up to 200 (fingerprint, blob-id) pairs. Ordered by fp hex.
+    pub refs_sample: Vec<DashboardRef>,
+    pub refs_sample_capped_at: usize,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct DashboardTag {
+    pub key: String,
+    pub value_hex: String,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct DashboardSnapshot {
+    pub name: String,
+    pub created_at_unix: u64,
+    pub blob_count: usize,
+    pub file_bytes: u64,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct DashboardRefTracking {
+    pub fingerprint_hex: String,
+    pub repo: String,
+    pub refs: Vec<String>,
+    pub first_seen_unix: u64,
+    pub last_seen_unix: u64,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct DashboardBlob {
+    pub blob_id_hex: String,
+    pub size_bytes: u64,
+    pub chunk_count: usize,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct DashboardRef {
+    pub fingerprint_hex: String,
+    pub blob_id_hex: String,
+}
+
 /// Recursive byte count under a directory. Used by the dashboard
 /// handler for reporting only; silent on read errors.
 fn dir_size_bytes(root: &std::path::Path) -> u64 {
@@ -574,6 +637,98 @@ impl RpcRouter {
                     .context("encoding DashboardStatusReply as JSON")?;
                 Ok(HandlerOutcome::Reply(json))
             }
+            Method::DashboardStorage => {
+                const SAMPLE_CAP: usize = 200;
+                let blob_root = self.blob_store.as_ref().map(|s| s.root().to_path_buf());
+                let mut tags = Vec::new();
+                if let Some(ts) = &self.tag_store {
+                    if let Ok(list) = ts.list().await {
+                        for e in list {
+                            tags.push(DashboardTag {
+                                key: e.key,
+                                value_hex: e.value_hex,
+                            });
+                        }
+                    }
+                }
+                let mut snapshots = Vec::new();
+                if let Some(p) = &blob_root {
+                    if let Ok(store) = crate::cluster::snapshot::SnapshotStore::open(p.clone()) {
+                        if let Ok(list) = store.list().await {
+                            for s in list {
+                                snapshots.push(DashboardSnapshot {
+                                    name: s.name,
+                                    created_at_unix: s.created_at_unix,
+                                    blob_count: s.blob_count,
+                                    file_bytes: s.file_bytes,
+                                });
+                            }
+                        }
+                    }
+                }
+                let mut ref_tracking = Vec::new();
+                if let Some(p) = &blob_root {
+                    if let Ok(store) = crate::cluster::ref_tracking::RefTracking::open(p.clone()) {
+                        if let Ok(list) = store.list_all().await {
+                            for e in list {
+                                let mut hex = String::with_capacity(64);
+                                for b in &e.fingerprint {
+                                    hex.push_str(&format!("{b:02x}"));
+                                }
+                                ref_tracking.push(DashboardRefTracking {
+                                    fingerprint_hex: hex,
+                                    repo: e.repo,
+                                    refs: e.refs,
+                                    first_seen_unix: e.first_seen_unix,
+                                    last_seen_unix: e.last_seen_unix,
+                                });
+                            }
+                        }
+                    }
+                }
+                let mut blobs_sample = Vec::new();
+                if let Some(store) = &self.blob_store {
+                    if let Ok(mut ids) = store.list_blob_ids().await {
+                        ids.sort();
+                        for id in ids.into_iter().take(SAMPLE_CAP) {
+                            let m = store.load_manifest(&id).await.ok().flatten();
+                            blobs_sample.push(DashboardBlob {
+                                blob_id_hex: id.to_hex(),
+                                size_bytes: m.as_ref().map(|m| m.total_size).unwrap_or(0),
+                                chunk_count: m.map(|m| m.chunks.len()).unwrap_or(0),
+                            });
+                        }
+                    }
+                }
+                let mut refs_sample = Vec::new();
+                if let Some(store) = &self.ref_store {
+                    if let Ok(list) = store.list().await {
+                        for (k, v) in list.into_iter().take(SAMPLE_CAP) {
+                            let mut kh = String::with_capacity(64);
+                            for b in &k {
+                                kh.push_str(&format!("{b:02x}"));
+                            }
+                            refs_sample.push(DashboardRef {
+                                fingerprint_hex: kh,
+                                blob_id_hex: crate::cluster::blob::BlobId::from_bytes(v).to_hex(),
+                            });
+                        }
+                    }
+                }
+                let reply = DashboardStorageReply {
+                    node_name: self.local_name.clone(),
+                    tags,
+                    snapshots,
+                    ref_tracking,
+                    blobs_sample_capped_at: SAMPLE_CAP,
+                    blobs_sample,
+                    refs_sample_capped_at: SAMPLE_CAP,
+                    refs_sample,
+                };
+                let json = serde_json::to_vec(&reply)
+                    .context("encoding DashboardStorageReply as JSON")?;
+                Ok(HandlerOutcome::Reply(json))
+            }
             Method::BlobStat => {
                 let store = match &self.blob_store {
                     Some(s) => s,

claw-store/src/cluster/rpc/client.rs

@@ -182,6 +182,16 @@ pub async fn call_dashboard_status(conn: &Connection) -> Result<DashboardStatusR
     serde_json::from_slice(&reply).context("decoding DashboardStatusReply JSON")
 }
 
+pub async fn call_dashboard_storage(conn: &Connection) -> Result<DashboardStorageReply> {
+    let reply = rpc_call(conn, Method::DashboardStorage, &[]).await?;
+    if reply.len() == 1 {
+        if let Some(code) = decode_error(reply[0]) {
+            bail!("peer replied with error: {}", code.describe());
+        }
+    }
+    serde_json::from_slice(&reply).context("decoding DashboardStorageReply JSON")
+}
+
 /// Recognise a single-byte reply as one of our error codes. Returns
 /// `None` for any other single-byte value (which is a valid reply,
 /// just an unusually short one).

claw-store/src/serve_v2.rs

@@ -22,7 +22,10 @@ use std::net::SocketAddr;
 use std::sync::Arc;
 use std::time::Duration;
 
-use crate::cluster::rpc::{call_dashboard_status, DashboardStatusReply};
+use crate::cluster::rpc::{
+    call_dashboard_status, call_dashboard_storage, DashboardStatusReply,
+    DashboardStorageReply,
+};
 use crate::cluster::transport::{NodeIdentity, QuicClient};
 use crate::config::{Config, PeerEntry};
 
@@ -34,7 +37,10 @@ use crate::config::{Config, PeerEntry};
 pub struct V2State {
     pub aggregator_name: String,
     pub peers: Vec<PeerEntry>,
-    pub client: QuicClient,
+    /// QuicClient wraps a single UDP endpoint + a connection pool.
+    /// Arc so the storage handler's per-peer JoinSet can hand the
+    /// client to spawned tasks without re-creating the endpoint.
+    pub client: std::sync::Arc<QuicClient>,
     /// Fallback RPC port when a peer entry has `lan_addr = "10.x:7701"`
     /// (gossip port, not RPC). We assume RPC = gossip + 1 in that
     /// case unless the operator sets an explicit `rpc_addr`.
@@ -58,7 +64,7 @@ impl V2State {
         Ok(Self {
             aggregator_name: cfg.node.name.clone(),
             peers: cluster.peers.clone(),
-            client,
+            client: std::sync::Arc::new(client),
             default_rpc_port_offset: 1,
         })
     }
@@ -86,6 +92,21 @@ impl V2State {
 
     /// Fetch a single node's dashboard payload via RPC.
     async fn fetch_node(&self, peer: &PeerEntry) -> anyhow::Result<DashboardStatusReply> {
+        let conn = self.dial(peer).await?;
+        let reply = call_dashboard_status(&conn).await?;
+        conn.close(quinn::VarInt::from_u32(0), b"done");
+        Ok(reply)
+    }
+
+    /// Fetch a single node's storage inventory via RPC.
+    async fn fetch_storage(&self, peer: &PeerEntry) -> anyhow::Result<DashboardStorageReply> {
+        let conn = self.dial(peer).await?;
+        let reply = call_dashboard_storage(&conn).await?;
+        conn.close(quinn::VarInt::from_u32(0), b"done");
+        Ok(reply)
+    }
+
+    async fn dial(&self, peer: &PeerEntry) -> anyhow::Result<quinn::Connection> {
         let addr = self.peer_rpc_addr(peer).ok_or_else(|| {
             anyhow::anyhow!("peer {} has no reachable RPC address", peer.name)
         })?;
@@ -95,9 +116,7 @@ impl V2State {
         )
         .await
         .map_err(|_| anyhow::anyhow!("connect to {} timed out", peer.name))??;
-        let reply = call_dashboard_status(&conn).await?;
-        conn.close(quinn::VarInt::from_u32(0), b"done");
-        Ok(reply)
+        Ok(conn)
     }
 }
 
@@ -169,6 +188,49 @@ pub struct FleetSnapshot {
     pub nodes: Vec<NodeStatusV2>,
 }
 
+/// One tag with the node it came from. Frontend renders `node` as a
+/// column so operators see where each artifact lives.
+#[derive(Serialize)]
+pub struct TagRow {
+    pub node: String,
+    pub key: String,
+    pub value_hex: String,
+}
+
+#[derive(Serialize)]
+pub struct SnapshotRow {
+    pub node: String,
+    pub name: String,
+    pub created_at_unix: u64,
+    pub blob_count: usize,
+    pub file_bytes: u64,
+}
+
+#[derive(Serialize)]
+pub struct RefTrackingRow {
+    pub node: String,
+    pub fingerprint_hex: String,
+    pub repo: String,
+    pub refs: Vec<String>,
+    pub first_seen_unix: u64,
+    pub last_seen_unix: u64,
+}
+
+#[derive(Serialize)]
+pub struct BlobRow {
+    pub node: String,
+    pub blob_id_hex: String,
+    pub size_bytes: u64,
+    pub chunk_count: usize,
+}
+
+#[derive(Serialize)]
+pub struct RefRow {
+    pub node: String,
+    pub fingerprint_hex: String,
+    pub blob_id_hex: String,
+}
+
 // ── handlers ─────────────────────────────────────────────────────
 
 async fn handle_fleet(State(s): State<Arc<V2State>>) -> Json<FleetSnapshot> {
@@ -227,10 +289,141 @@ async fn handle_node_status(
     Json(node)
 }
 
+// ── storage handlers (fleet-aggregated) ──────────────────────────
+
+/// Fan out DashboardStorage to every peer, return `(node, reply)`
+/// pairs where the RPC succeeded. Failures logged but skipped
+/// silently — a browser view of "N nodes, one is down" is more
+/// useful than a hard 500.
+async fn gather_storage(s: &V2State) -> Vec<(String, DashboardStorageReply)> {
+    let mut set = tokio::task::JoinSet::new();
+    for p in &s.peers {
+        let name = p.name.clone();
+        let peer = p.clone();
+        let state = Arc::new(s.clone_shallow());
+        set.spawn(async move {
+            match state.fetch_storage(&peer).await {
+                Ok(r) => Some((name, r)),
+                Err(e) => {
+                    tracing::debug!(peer = %name, error = %e, "DashboardStorage RPC failed");
+                    None
+                }
+            }
+        });
+    }
+    let mut out = Vec::new();
+    while let Some(res) = set.join_next().await {
+        if let Ok(Some(pair)) = res {
+            out.push(pair);
+        }
+    }
+    out
+}
+
+async fn handle_tags(State(s): State<Arc<V2State>>) -> Json<Vec<TagRow>> {
+    let mut rows = Vec::new();
+    for (node, r) in gather_storage(&s).await {
+        for t in r.tags {
+            rows.push(TagRow {
+                node: node.clone(),
+                key: t.key,
+                value_hex: t.value_hex,
+            });
+        }
+    }
+    rows.sort_by(|a, b| a.key.cmp(&b.key).then_with(|| a.node.cmp(&b.node)));
+    Json(rows)
+}
+
+async fn handle_snapshots(State(s): State<Arc<V2State>>) -> Json<Vec<SnapshotRow>> {
+    let mut rows = Vec::new();
+    for (node, r) in gather_storage(&s).await {
+        for snap in r.snapshots {
+            rows.push(SnapshotRow {
+                node: node.clone(),
+                name: snap.name,
+                created_at_unix: snap.created_at_unix,
+                blob_count: snap.blob_count,
+                file_bytes: snap.file_bytes,
+            });
+        }
+    }
+    // Oldest first — matches CLI convention.
+    rows.sort_by_key(|s| s.created_at_unix);
+    Json(rows)
+}
+
+async fn handle_ref_tracking(State(s): State<Arc<V2State>>) -> Json<Vec<RefTrackingRow>> {
+    let mut rows = Vec::new();
+    for (node, r) in gather_storage(&s).await {
+        for e in r.ref_tracking {
+            rows.push(RefTrackingRow {
+                node: node.clone(),
+                fingerprint_hex: e.fingerprint_hex,
+                repo: e.repo,
+                refs: e.refs,
+                first_seen_unix: e.first_seen_unix,
+                last_seen_unix: e.last_seen_unix,
+            });
+        }
+    }
+    rows.sort_by(|a, b| a.repo.cmp(&b.repo).then_with(|| a.node.cmp(&b.node)));
+    Json(rows)
+}
+
+async fn handle_blobs(State(s): State<Arc<V2State>>) -> Json<Vec<BlobRow>> {
+    let mut rows = Vec::new();
+    for (node, r) in gather_storage(&s).await {
+        for b in r.blobs_sample {
+            rows.push(BlobRow {
+                node: node.clone(),
+                blob_id_hex: b.blob_id_hex,
+                size_bytes: b.size_bytes,
+                chunk_count: b.chunk_count,
+            });
+        }
+    }
+    rows.sort_by(|a, b| a.blob_id_hex.cmp(&b.blob_id_hex));
+    Json(rows)
+}
+
+async fn handle_refs(State(s): State<Arc<V2State>>) -> Json<Vec<RefRow>> {
+    let mut rows = Vec::new();
+    for (node, r) in gather_storage(&s).await {
+        for x in r.refs_sample {
+            rows.push(RefRow {
+                node: node.clone(),
+                fingerprint_hex: x.fingerprint_hex,
+                blob_id_hex: x.blob_id_hex,
+            });
+        }
+    }
+    rows.sort_by(|a, b| a.fingerprint_hex.cmp(&b.fingerprint_hex));
+    Json(rows)
+}
+
+impl V2State {
+    /// Cheap clone for JoinSet handoff — reuses the underlying
+    /// QuicClient (Arc'd inside) + clones the small peer list.
+    fn clone_shallow(&self) -> Self {
+        Self {
+            aggregator_name: self.aggregator_name.clone(),
+            peers: self.peers.clone(),
+            client: self.client.clone(),
+            default_rpc_port_offset: self.default_rpc_port_offset,
+        }
+    }
+}
+
 // ── route registration ──────────────────────────────────────────
 
 pub fn routes() -> Router<Arc<V2State>> {
     Router::new()
         .route("/api/v2/fleet", get(handle_fleet))
         .route("/api/v2/node/:name/status", get(handle_node_status))
+        .route("/api/v2/storage/blobs", get(handle_blobs))
+        .route("/api/v2/storage/tags", get(handle_tags))
+        .route("/api/v2/storage/refs", get(handle_refs))
+        .route("/api/v2/storage/snapshots", get(handle_snapshots))
+        .route("/api/v2/storage/ref-tracking", get(handle_ref_tracking))
 }