dashboard-v2 PR 3: fleet aggregator via DashboardStatus RPC #94
@@ -183,6 +183,16 @@ pub enum Method {
|
|||||||
/// Reply: 8 bytes (`u64` LE) on hit, single-byte
|
/// Reply: 8 bytes (`u64` LE) on hit, single-byte
|
||||||
/// [`ErrorCode::NotFound`] when no sidecar is present.
|
/// [`ErrorCode::NotFound`] when no sidecar is present.
|
||||||
GetTagExpiry = 0x1b,
|
GetTagExpiry = 0x1b,
|
||||||
|
/// Dashboard-v2 (2026-07-14): return a JSON snapshot of this
|
||||||
|
/// node's counts (blobs / tags / refs / snapshots / ref-tracking
|
||||||
|
/// + on-disk bytes). Feeds the fleet aggregator dashboard so an
|
||||||
|
/// operator gets a single-pane view of the fleet from any client
|
||||||
|
/// that can reach cluster RPC — no HTTP server on the daemon
|
||||||
|
/// nodes required.
|
||||||
|
///
|
||||||
|
/// `payload`: empty.
|
||||||
|
/// Reply: JSON `DashboardStatusReply`.
|
||||||
|
DashboardStatus = 0x1c,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl Method {
|
impl Method {
|
||||||
@@ -217,6 +227,7 @@ impl Method {
|
|||||||
0x19 => Some(Method::GetTagVersioned),
|
0x19 => Some(Method::GetTagVersioned),
|
||||||
0x1a => Some(Method::SetTagExpiry),
|
0x1a => Some(Method::SetTagExpiry),
|
||||||
0x1b => Some(Method::GetTagExpiry),
|
0x1b => Some(Method::GetTagExpiry),
|
||||||
|
0x1c => Some(Method::DashboardStatus),
|
||||||
_ => None,
|
_ => None,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -297,6 +308,28 @@ pub struct PeerStatusReply {
|
|||||||
pub peers: Vec<PeerView>,
|
pub peers: Vec<PeerView>,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Dashboard-v2 aggregator payload. One per node. Wire: JSON.
|
||||||
|
///
|
||||||
|
/// Read by `serve_v2::AggregatorState` from any client that can
|
||||||
|
/// reach cluster RPC (typically the operator's laptop running
|
||||||
|
/// `claw-store serve --aggregator`).
|
||||||
|
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
|
||||||
|
pub struct DashboardStatusReply {
|
||||||
|
pub node_name: String,
|
||||||
|
pub zone: String,
|
||||||
|
pub blob_store_root: Option<String>,
|
||||||
|
pub blob_count: usize,
|
||||||
|
pub tag_count: usize,
|
||||||
|
pub ref_count: usize,
|
||||||
|
pub snapshot_count: usize,
|
||||||
|
pub ref_tracking_count: usize,
|
||||||
|
pub blob_store_bytes: u64,
|
||||||
|
/// `local_rustc_release` from PeerStatus, echoed here so the
|
||||||
|
/// aggregator doesn't need a second round-trip.
|
||||||
|
#[serde(default)]
|
||||||
|
pub rustc_release: Option<String>,
|
||||||
|
}
|
||||||
|
|
||||||
/// Reply payload for [`Method::PutManifest`]. When `missing` is empty
|
/// Reply payload for [`Method::PutManifest`]. When `missing` is empty
|
||||||
/// the manifest was persisted successfully; otherwise the client must
|
/// the manifest was persisted successfully; otherwise the client must
|
||||||
/// upload the listed chunks (typically via [`Method::PutChunk`]) and
|
/// upload the listed chunks (typically via [`Method::PutChunk`]) and
|
||||||
@@ -313,6 +346,33 @@ pub struct PutManifestReply {
|
|||||||
///
|
///
|
||||||
/// Not `Clone` on its own — wrap in `Arc<RpcRouter>` so a single
|
/// Not `Clone` on its own — wrap in `Arc<RpcRouter>` so a single
|
||||||
/// instance backs the accept loop plus any explicit dispatch calls.
|
/// instance backs the accept loop plus any explicit dispatch calls.
|
||||||
|
/// 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 {
|
||||||
|
let mut total: u64 = 0;
|
||||||
|
let mut stack = vec![root.to_path_buf()];
|
||||||
|
while let Some(dir) = stack.pop() {
|
||||||
|
let entries = match std::fs::read_dir(&dir) {
|
||||||
|
Ok(e) => e,
|
||||||
|
Err(_) => continue,
|
||||||
|
};
|
||||||
|
for entry in entries.flatten() {
|
||||||
|
let ft = match entry.file_type() {
|
||||||
|
Ok(t) => t,
|
||||||
|
Err(_) => continue,
|
||||||
|
};
|
||||||
|
if ft.is_dir() {
|
||||||
|
stack.push(entry.path());
|
||||||
|
} else if ft.is_file() {
|
||||||
|
if let Ok(m) = entry.metadata() {
|
||||||
|
total = total.saturating_add(m.len());
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
total
|
||||||
|
}
|
||||||
|
|
||||||
pub struct RpcRouter {
|
pub struct RpcRouter {
|
||||||
gossip: Arc<ClusterGossip>,
|
gossip: Arc<ClusterGossip>,
|
||||||
blob_store: Option<Arc<BlobStore>>,
|
blob_store: Option<Arc<BlobStore>>,
|
||||||
@@ -450,6 +510,70 @@ impl RpcRouter {
|
|||||||
}
|
}
|
||||||
Ok(HandlerOutcome::Reply(json))
|
Ok(HandlerOutcome::Reply(json))
|
||||||
}
|
}
|
||||||
|
Method::DashboardStatus => {
|
||||||
|
// Aggregator payload for dashboard-v2. Reuses gossip
|
||||||
|
// for name+zone+rustc; walks the on-disk stores for
|
||||||
|
// count + byte totals. Cheap: 5 filesystem walks.
|
||||||
|
let (blob_root, blob_count) = match &self.blob_store {
|
||||||
|
Some(s) => {
|
||||||
|
let ids = s.list_blob_ids().await.unwrap_or_default();
|
||||||
|
(
|
||||||
|
Some(s.root().display().to_string()),
|
||||||
|
ids.len(),
|
||||||
|
)
|
||||||
|
}
|
||||||
|
None => (None, 0),
|
||||||
|
};
|
||||||
|
let root_path = blob_root.as_ref().map(std::path::PathBuf::from);
|
||||||
|
let tag_count = match &self.tag_store {
|
||||||
|
Some(t) => t.list().await.map(|v| v.len()).unwrap_or(0),
|
||||||
|
None => 0,
|
||||||
|
};
|
||||||
|
let ref_count = match &self.ref_store {
|
||||||
|
Some(r) => r.list().await.map(|v| v.len()).unwrap_or(0),
|
||||||
|
None => 0,
|
||||||
|
};
|
||||||
|
// Snapshot + ref-tracking aren't held on the router;
|
||||||
|
// open by path derived from blob_store_root when
|
||||||
|
// possible.
|
||||||
|
let snapshot_count = match &root_path {
|
||||||
|
Some(p) => match crate::cluster::snapshot::SnapshotStore::open(p.clone()) {
|
||||||
|
Ok(s) => s.list().await.map(|v| v.len()).unwrap_or(0),
|
||||||
|
Err(_) => 0,
|
||||||
|
},
|
||||||
|
None => 0,
|
||||||
|
};
|
||||||
|
let ref_tracking_count = match &root_path {
|
||||||
|
Some(p) => match crate::cluster::ref_tracking::RefTracking::open(p.clone()) {
|
||||||
|
Ok(r) => r.list_all().await.map(|v| v.len()).unwrap_or(0),
|
||||||
|
Err(_) => 0,
|
||||||
|
},
|
||||||
|
None => 0,
|
||||||
|
};
|
||||||
|
let blob_store_bytes = match &root_path {
|
||||||
|
Some(p) => dir_size_bytes(p),
|
||||||
|
None => 0,
|
||||||
|
};
|
||||||
|
let rustc_release = self
|
||||||
|
.gossip
|
||||||
|
.self_kv(crate::cluster::gossip::keys::RUSTC_RELEASE)
|
||||||
|
.await;
|
||||||
|
let reply = DashboardStatusReply {
|
||||||
|
node_name: self.local_name.clone(),
|
||||||
|
zone: self.local_zone.clone(),
|
||||||
|
blob_store_root: blob_root,
|
||||||
|
blob_count,
|
||||||
|
tag_count,
|
||||||
|
ref_count,
|
||||||
|
snapshot_count,
|
||||||
|
ref_tracking_count,
|
||||||
|
blob_store_bytes,
|
||||||
|
rustc_release,
|
||||||
|
};
|
||||||
|
let json = serde_json::to_vec(&reply)
|
||||||
|
.context("encoding DashboardStatusReply as JSON")?;
|
||||||
|
Ok(HandlerOutcome::Reply(json))
|
||||||
|
}
|
||||||
Method::BlobStat => {
|
Method::BlobStat => {
|
||||||
let store = match &self.blob_store {
|
let store = match &self.blob_store {
|
||||||
Some(s) => s,
|
Some(s) => s,
|
||||||
|
|||||||
@@ -169,6 +169,19 @@ pub async fn call_peer_status(conn: &Connection) -> Result<PeerStatusReply> {
|
|||||||
serde_json::from_slice(&reply).context("decoding PeerStatusReply JSON")
|
serde_json::from_slice(&reply).context("decoding PeerStatusReply JSON")
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Convenience wrapper for [`Method::DashboardStatus`]. Feeds the
|
||||||
|
/// dashboard-v2 aggregator. Empty payload → JSON reply with counts
|
||||||
|
/// + on-disk bytes.
|
||||||
|
pub async fn call_dashboard_status(conn: &Connection) -> Result<DashboardStatusReply> {
|
||||||
|
let reply = rpc_call(conn, Method::DashboardStatus, &[]).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 DashboardStatusReply JSON")
|
||||||
|
}
|
||||||
|
|
||||||
/// Recognise a single-byte reply as one of our error codes. Returns
|
/// Recognise a single-byte reply as one of our error codes. Returns
|
||||||
/// `None` for any other single-byte value (which is a valid reply,
|
/// `None` for any other single-byte value (which is a valid reply,
|
||||||
/// just an unusually short one).
|
/// just an unusually short one).
|
||||||
|
|||||||
+14
-10
@@ -706,26 +706,30 @@ pub fn build_app_with_v2(
|
|||||||
api = api.fallback_service(tower_http::services::ServeDir::new(dir));
|
api = api.fallback_service(tower_http::services::ServeDir::new(dir));
|
||||||
}
|
}
|
||||||
|
|
||||||
// dashboard-v2 backend (docs/dashboard-v2.md). Additive alongside
|
// dashboard-v2 aggregator backend (docs/dashboard-v2.md).
|
||||||
// the legacy /api/* routes above so the cutover doesn't break the
|
// Builds only when [cluster] + [cluster.tls] + peers are
|
||||||
// old dashboard while the new one is being iterated on.
|
// present; otherwise v2 API routes are skipped, and the /v2/
|
||||||
let v2_state = std::sync::Arc::new(crate::serve_v2::V2State::from_config(&state.cfg));
|
// static mount (if any) still works so operators can see the
|
||||||
let mut v2_routes = crate::serve_v2::routes();
|
// config-missing error message the SPA renders.
|
||||||
|
let v2_state = crate::serve_v2::V2State::from_config(&state.cfg)
|
||||||
|
.map(std::sync::Arc::new)
|
||||||
|
.ok();
|
||||||
|
let mut v2_router: Router<()> = Router::new();
|
||||||
|
if let Some(v2s) = v2_state {
|
||||||
|
v2_router = v2_router.merge(crate::serve_v2::routes().with_state(v2s));
|
||||||
|
}
|
||||||
if let Some(dir) = v2_static_dir {
|
if let Some(dir) = v2_static_dir {
|
||||||
// Serve the compiled SPA at /v2/*. Any unknown /v2/* path
|
v2_router = v2_router.nest_service(
|
||||||
// falls back to index.html so client-side wouter routing works.
|
|
||||||
v2_routes = v2_routes.nest_service(
|
|
||||||
"/v2",
|
"/v2",
|
||||||
tower_http::services::ServeDir::new(&dir).fallback(
|
tower_http::services::ServeDir::new(&dir).fallback(
|
||||||
tower_http::services::ServeFile::new(dir.join("index.html")),
|
tower_http::services::ServeFile::new(dir.join("index.html")),
|
||||||
),
|
),
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
let v2_routes = v2_routes.with_state(v2_state);
|
|
||||||
|
|
||||||
Router::new()
|
Router::new()
|
||||||
.merge(api.layer(cors.clone()).with_state(state))
|
.merge(api.layer(cors.clone()).with_state(state))
|
||||||
.merge(v2_routes.layer(cors))
|
.merge(v2_router.layer(cors))
|
||||||
}
|
}
|
||||||
|
|
||||||
pub async fn run_server(
|
pub async fn run_server(
|
||||||
|
|||||||
+172
-497
@@ -1,561 +1,236 @@
|
|||||||
//! dashboard-v2 backend — additive `/api/v2/*` handlers.
|
//! dashboard-v2 aggregator backend — additive `/api/v2/*` handlers.
|
||||||
//!
|
//!
|
||||||
//! Reads the distributed-architecture stores (BlobStore, TagStore,
|
//! Fleet-single-pane-of-glass model: `claw-store serve` runs on the
|
||||||
//! RefStore, SnapshotStore, RefTracking). Kept in a separate file
|
//! operator's laptop (typically). It holds a fleet-CA leaf cert +
|
||||||
//! from `serve.rs` so the legacy ZFS-era handlers there can be
|
//! the peer list, connects to each daemon's cluster RPC port over
|
||||||
//! removed cleanly at cutover without a big rebase.
|
//! mTLS, and issues `DashboardStatus` RPCs. Peers don't need to
|
||||||
|
//! run any HTTP server of their own — RPC over the existing QUIC
|
||||||
|
//! socket is enough.
|
||||||
//!
|
//!
|
||||||
//! Design + endpoint spec: `docs/dashboard-v2.md`.
|
//! Config: peers come from `[[cluster.peers]]` in the aggregator's
|
||||||
|
//! `config.toml`. TLS material via `[cluster.tls]`.
|
||||||
//!
|
//!
|
||||||
//! Single-node reads in this cut. Cross-node fleet aggregation
|
//! Design doc: `docs/dashboard-v2.md`.
|
||||||
//! (fan-out via QUIC RPC) lands in a follow-on PR.
|
|
||||||
|
|
||||||
use axum::{
|
use axum::{
|
||||||
extract::{Path, Query, State},
|
extract::{Path, State},
|
||||||
routing::get,
|
routing::get,
|
||||||
Json, Router,
|
Json, Router,
|
||||||
};
|
};
|
||||||
use serde::{Deserialize, Serialize};
|
use serde::{Deserialize, Serialize};
|
||||||
use std::path::PathBuf;
|
use std::net::SocketAddr;
|
||||||
use std::sync::Arc;
|
use std::sync::Arc;
|
||||||
|
use std::time::Duration;
|
||||||
|
|
||||||
use crate::cluster::blob::{BlobId, BlobStore};
|
use crate::cluster::rpc::{call_dashboard_status, DashboardStatusReply};
|
||||||
use crate::cluster::ref_tracking::{RefEntry, RefTracking};
|
use crate::cluster::transport::{NodeIdentity, QuicClient};
|
||||||
use crate::cluster::refs::RefStore;
|
use crate::config::{Config, PeerEntry};
|
||||||
use crate::cluster::snapshot::SnapshotStore;
|
|
||||||
use crate::cluster::tags::TagStore;
|
|
||||||
use crate::config::Config;
|
|
||||||
|
|
||||||
/// Runtime shared by all v2 handlers. Cheap clones — inner types
|
/// Aggregator runtime: one QuicClient, one peer list, one identity.
|
||||||
/// are `Arc<_>` or filesystem-backed with no in-memory state.
|
///
|
||||||
#[derive(Clone)]
|
/// The client is reused across every RPC (quinn holds one UDP
|
||||||
|
/// endpoint + a pool of connections). Peer addresses are resolved
|
||||||
|
/// per-request so a hot-swapped config doesn't require restart.
|
||||||
pub struct V2State {
|
pub struct V2State {
|
||||||
pub node_name: String,
|
pub aggregator_name: String,
|
||||||
pub blob_store_root: Option<PathBuf>,
|
pub peers: Vec<PeerEntry>,
|
||||||
|
pub client: 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`.
|
||||||
|
pub default_rpc_port_offset: u16,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl V2State {
|
impl V2State {
|
||||||
pub fn from_config(cfg: &Config) -> Self {
|
/// Build an aggregator state from the aggregator's own config.
|
||||||
let blob_store_root = cfg
|
/// Requires `[cluster]` + `[cluster.tls]` + at least one
|
||||||
|
/// `[[cluster.peers]]`. The aggregator doesn't need a
|
||||||
|
/// blob_store_root itself.
|
||||||
|
pub fn from_config(cfg: &Config) -> anyhow::Result<Self> {
|
||||||
|
let cluster = cfg
|
||||||
.cluster
|
.cluster
|
||||||
.as_ref()
|
.as_ref()
|
||||||
.and_then(|c| c.blob_store_root.clone());
|
.ok_or_else(|| anyhow::anyhow!("dashboard-v2 aggregator needs [cluster] config"))?;
|
||||||
Self {
|
let identity = NodeIdentity::from_cluster_config(cluster)
|
||||||
node_name: cfg.node.name.clone(),
|
.map_err(|e| anyhow::anyhow!("loading fleet-CA identity: {e}"))?;
|
||||||
blob_store_root,
|
let client = QuicClient::new("0.0.0.0:0".parse()?, identity)
|
||||||
|
.map_err(|e| anyhow::anyhow!("building QUIC client: {e}"))?;
|
||||||
|
Ok(Self {
|
||||||
|
aggregator_name: cfg.node.name.clone(),
|
||||||
|
peers: cluster.peers.clone(),
|
||||||
|
client,
|
||||||
|
default_rpc_port_offset: 1,
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Resolve a peer name → RPC socket. Prefers `rpc_addr` when the
|
||||||
|
/// peer entry sets one; otherwise takes `lan_addr` and adds the
|
||||||
|
/// port offset (gossip → RPC = +1 by fleet convention).
|
||||||
|
fn peer_rpc_addr(&self, peer: &PeerEntry) -> Option<SocketAddr> {
|
||||||
|
// Prefer LAN (fast), fall back to tailnet. RPC port =
|
||||||
|
// gossip port + 1 by fleet convention (see fleet configs).
|
||||||
|
if let Some(lan) = peer.lan_addr {
|
||||||
|
return Some(SocketAddr::new(
|
||||||
|
lan.ip(),
|
||||||
|
lan.port() + self.default_rpc_port_offset,
|
||||||
|
));
|
||||||
}
|
}
|
||||||
|
if let Some(ts) = peer.tailscale_addr {
|
||||||
|
return Some(SocketAddr::new(
|
||||||
|
ts.ip(),
|
||||||
|
ts.port() + self.default_rpc_port_offset,
|
||||||
|
));
|
||||||
|
}
|
||||||
|
None
|
||||||
}
|
}
|
||||||
|
|
||||||
fn open_blob(&self) -> Option<BlobStore> {
|
/// Fetch a single node's dashboard payload via RPC.
|
||||||
let root = self.blob_store_root.as_ref()?;
|
async fn fetch_node(&self, peer: &PeerEntry) -> anyhow::Result<DashboardStatusReply> {
|
||||||
BlobStore::open(root.clone()).ok()
|
let addr = self.peer_rpc_addr(peer).ok_or_else(|| {
|
||||||
}
|
anyhow::anyhow!("peer {} has no reachable RPC address", peer.name)
|
||||||
|
})?;
|
||||||
fn open_tags(&self) -> Option<TagStore> {
|
let conn = tokio::time::timeout(
|
||||||
let root = self.blob_store_root.as_ref()?;
|
Duration::from_secs(5),
|
||||||
TagStore::open(root.join("tags-db")).ok()
|
self.client.connect(addr, &peer.name),
|
||||||
}
|
)
|
||||||
|
.await
|
||||||
fn open_refs(&self) -> Option<RefStore> {
|
.map_err(|_| anyhow::anyhow!("connect to {} timed out", peer.name))??;
|
||||||
let root = self.blob_store_root.as_ref()?;
|
let reply = call_dashboard_status(&conn).await?;
|
||||||
RefStore::open(root.join("refs-db")).ok()
|
conn.close(quinn::VarInt::from_u32(0), b"done");
|
||||||
}
|
Ok(reply)
|
||||||
|
|
||||||
fn open_snapshots(&self) -> Option<SnapshotStore> {
|
|
||||||
let root = self.blob_store_root.as_ref()?;
|
|
||||||
SnapshotStore::open(root.clone()).ok()
|
|
||||||
}
|
|
||||||
|
|
||||||
fn open_ref_tracking(&self) -> Option<RefTracking> {
|
|
||||||
let root = self.blob_store_root.as_ref()?;
|
|
||||||
RefTracking::open(root.clone()).ok()
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// ── request / response shapes ────────────────────────────────────
|
// ── response shapes ──────────────────────────────────────────────
|
||||||
|
|
||||||
#[derive(Serialize, Deserialize)]
|
#[derive(Serialize, Deserialize)]
|
||||||
pub struct NodeStatusV2 {
|
pub struct NodeStatusV2 {
|
||||||
pub node_name: String,
|
pub node_name: String,
|
||||||
pub blob_store_root: Option<PathBuf>,
|
pub zone: String,
|
||||||
|
pub blob_store_root: Option<String>,
|
||||||
pub blob_count: usize,
|
pub blob_count: usize,
|
||||||
pub tag_count: usize,
|
pub tag_count: usize,
|
||||||
pub ref_count: usize,
|
pub ref_count: usize,
|
||||||
pub snapshot_count: usize,
|
pub snapshot_count: usize,
|
||||||
pub ref_tracking_count: usize,
|
pub ref_tracking_count: usize,
|
||||||
pub blob_store_bytes: u64,
|
pub blob_store_bytes: u64,
|
||||||
|
pub rustc_release: Option<String>,
|
||||||
|
/// `true` when the aggregator successfully talked to the peer;
|
||||||
|
/// `false` when the RPC failed. Frontend uses this to badge the
|
||||||
|
/// card as offline.
|
||||||
|
pub online: bool,
|
||||||
|
/// Human-readable error when `online = false`.
|
||||||
|
pub error: Option<String>,
|
||||||
}
|
}
|
||||||
|
|
||||||
#[derive(Deserialize)]
|
impl NodeStatusV2 {
|
||||||
pub struct Pagination {
|
fn ok(name: &str, r: DashboardStatusReply) -> Self {
|
||||||
#[serde(default = "default_limit")]
|
let mut n = Self {
|
||||||
pub limit: usize,
|
node_name: r.node_name,
|
||||||
#[serde(default)]
|
zone: r.zone,
|
||||||
pub offset: usize,
|
blob_store_root: r.blob_store_root,
|
||||||
}
|
blob_count: r.blob_count,
|
||||||
|
tag_count: r.tag_count,
|
||||||
fn default_limit() -> usize {
|
ref_count: r.ref_count,
|
||||||
200
|
snapshot_count: r.snapshot_count,
|
||||||
}
|
ref_tracking_count: r.ref_tracking_count,
|
||||||
|
blob_store_bytes: r.blob_store_bytes,
|
||||||
#[derive(Serialize, Deserialize)]
|
rustc_release: r.rustc_release,
|
||||||
pub struct BlobSummary {
|
online: true,
|
||||||
pub blob_id_hex: String,
|
error: None,
|
||||||
pub size_bytes: u64,
|
};
|
||||||
pub chunk_count: usize,
|
if n.node_name.is_empty() {
|
||||||
}
|
n.node_name = name.to_string();
|
||||||
|
|
||||||
#[derive(Serialize, Deserialize)]
|
|
||||||
pub struct TagSummaryV2 {
|
|
||||||
pub key: String,
|
|
||||||
pub value_hex: String,
|
|
||||||
}
|
|
||||||
|
|
||||||
#[derive(Deserialize)]
|
|
||||||
pub struct TagFilter {
|
|
||||||
#[serde(default)]
|
|
||||||
pub prefix: String,
|
|
||||||
}
|
|
||||||
|
|
||||||
#[derive(Serialize, Deserialize)]
|
|
||||||
pub struct RefSummary {
|
|
||||||
pub fingerprint_hex: String,
|
|
||||||
pub blob_id_hex: String,
|
|
||||||
}
|
|
||||||
|
|
||||||
#[derive(Serialize, Deserialize)]
|
|
||||||
pub struct SnapshotSummaryV2 {
|
|
||||||
pub name: String,
|
|
||||||
pub created_at_unix: u64,
|
|
||||||
pub blob_count: usize,
|
|
||||||
pub file_bytes: u64,
|
|
||||||
}
|
|
||||||
|
|
||||||
#[derive(Serialize, Deserialize)]
|
|
||||||
pub struct RefTrackingItem {
|
|
||||||
pub fingerprint_hex: String,
|
|
||||||
pub repo: String,
|
|
||||||
pub refs: Vec<String>,
|
|
||||||
pub first_seen_unix: u64,
|
|
||||||
pub last_seen_unix: u64,
|
|
||||||
}
|
|
||||||
|
|
||||||
impl From<RefEntry> for RefTrackingItem {
|
|
||||||
fn from(e: RefEntry) -> Self {
|
|
||||||
let mut hex = String::with_capacity(64);
|
|
||||||
for b in &e.fingerprint {
|
|
||||||
hex.push_str(&format!("{b:02x}"));
|
|
||||||
}
|
}
|
||||||
|
n
|
||||||
|
}
|
||||||
|
fn err(name: &str, e: String) -> Self {
|
||||||
Self {
|
Self {
|
||||||
fingerprint_hex: hex,
|
node_name: name.to_string(),
|
||||||
repo: e.repo,
|
zone: String::new(),
|
||||||
refs: e.refs,
|
blob_store_root: None,
|
||||||
first_seen_unix: e.first_seen_unix,
|
blob_count: 0,
|
||||||
last_seen_unix: e.last_seen_unix,
|
tag_count: 0,
|
||||||
|
ref_count: 0,
|
||||||
|
snapshot_count: 0,
|
||||||
|
ref_tracking_count: 0,
|
||||||
|
blob_store_bytes: 0,
|
||||||
|
rustc_release: None,
|
||||||
|
online: false,
|
||||||
|
error: Some(e),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
#[derive(Deserialize)]
|
|
||||||
pub struct RefTrackingFilter {
|
|
||||||
#[serde(default)]
|
|
||||||
pub repo: String,
|
|
||||||
}
|
|
||||||
|
|
||||||
#[derive(Serialize)]
|
#[derive(Serialize)]
|
||||||
pub struct EmptyReason {
|
pub struct FleetSnapshot {
|
||||||
pub reason: String,
|
pub aggregator_name: String,
|
||||||
|
pub fetched_at_unix: u64,
|
||||||
|
pub nodes: Vec<NodeStatusV2>,
|
||||||
}
|
}
|
||||||
|
|
||||||
// ── handlers ─────────────────────────────────────────────────────
|
// ── handlers ─────────────────────────────────────────────────────
|
||||||
|
|
||||||
async fn handle_status(State(s): State<Arc<V2State>>) -> Json<NodeStatusV2> {
|
async fn handle_fleet(State(s): State<Arc<V2State>>) -> Json<FleetSnapshot> {
|
||||||
let blob = s.open_blob();
|
// Fan out concurrently.
|
||||||
let tags = s.open_tags();
|
// Fan out with a JoinSet — bounded concurrency, no futures crate.
|
||||||
let refs = s.open_refs();
|
let mut set = tokio::task::JoinSet::new();
|
||||||
let snaps = s.open_snapshots();
|
for p in &s.peers {
|
||||||
let rt = s.open_ref_tracking();
|
let name = p.name.clone();
|
||||||
let root = s.blob_store_root.clone();
|
let state = s.clone();
|
||||||
|
let peer = p.clone();
|
||||||
let blob_count = match &blob {
|
set.spawn(async move {
|
||||||
Some(b) => b.list_blob_ids().await.map(|v| v.len()).unwrap_or(0),
|
match state.fetch_node(&peer).await {
|
||||||
None => 0,
|
Ok(r) => NodeStatusV2::ok(&name, r),
|
||||||
};
|
Err(e) => NodeStatusV2::err(&name, format!("{e:#}")),
|
||||||
let tag_count = match &tags {
|
}
|
||||||
Some(t) => t.list().await.map(|v| v.len()).unwrap_or(0),
|
});
|
||||||
None => 0,
|
}
|
||||||
};
|
let mut nodes: Vec<NodeStatusV2> = Vec::new();
|
||||||
let ref_count = match &refs {
|
while let Some(res) = set.join_next().await {
|
||||||
Some(r) => r.list().await.map(|v| v.len()).unwrap_or(0),
|
if let Ok(n) = res {
|
||||||
None => 0,
|
nodes.push(n);
|
||||||
};
|
}
|
||||||
let snapshot_count = match &snaps {
|
}
|
||||||
Some(s) => s.list().await.map(|v| v.len()).unwrap_or(0),
|
// Deterministic ordering — by name — so the UI doesn't shuffle
|
||||||
None => 0,
|
// between polls.
|
||||||
};
|
nodes.sort_by(|a, b| a.node_name.cmp(&b.node_name));
|
||||||
let ref_tracking_count = match &rt {
|
let now = std::time::SystemTime::now()
|
||||||
Some(r) => r.list_all().await.map(|v| v.len()).unwrap_or(0),
|
.duration_since(std::time::UNIX_EPOCH)
|
||||||
None => 0,
|
.map(|d| d.as_secs())
|
||||||
};
|
.unwrap_or(0);
|
||||||
let blob_store_bytes = match &root {
|
Json(FleetSnapshot {
|
||||||
Some(p) => dir_size_bytes(p),
|
aggregator_name: s.aggregator_name.clone(),
|
||||||
None => 0,
|
fetched_at_unix: now,
|
||||||
};
|
nodes,
|
||||||
Json(NodeStatusV2 {
|
|
||||||
node_name: s.node_name.clone(),
|
|
||||||
blob_store_root: root,
|
|
||||||
blob_count,
|
|
||||||
tag_count,
|
|
||||||
ref_count,
|
|
||||||
snapshot_count,
|
|
||||||
ref_tracking_count,
|
|
||||||
blob_store_bytes,
|
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn handle_node_status(
|
async fn handle_node_status(
|
||||||
State(s): State<Arc<V2State>>,
|
State(s): State<Arc<V2State>>,
|
||||||
Path(name): Path<String>,
|
Path(name): Path<String>,
|
||||||
) -> Result<Json<NodeStatusV2>, (axum::http::StatusCode, Json<EmptyReason>)> {
|
) -> Json<NodeStatusV2> {
|
||||||
// Single-node cut: only "local" or "<own name>" resolves.
|
let peer = s.peers.iter().find(|p| p.name == name).cloned();
|
||||||
// Cross-node lookup lands with the fleet fan-out follow-on.
|
let peer = match peer {
|
||||||
if name != "local" && name != s.node_name {
|
Some(p) => p,
|
||||||
return Err((
|
None => {
|
||||||
axum::http::StatusCode::NOT_IMPLEMENTED,
|
return Json(NodeStatusV2::err(
|
||||||
Json(EmptyReason {
|
&name,
|
||||||
reason: format!(
|
format!("no peer named {name} in aggregator config"),
|
||||||
"cross-node lookup not yet implemented; requested {name}, this is {}",
|
))
|
||||||
s.node_name
|
|
||||||
),
|
|
||||||
}),
|
|
||||||
));
|
|
||||||
}
|
|
||||||
Ok(handle_status(State(s)).await)
|
|
||||||
}
|
|
||||||
|
|
||||||
async fn handle_blobs(
|
|
||||||
State(s): State<Arc<V2State>>,
|
|
||||||
Query(p): Query<Pagination>,
|
|
||||||
) -> Json<Vec<BlobSummary>> {
|
|
||||||
let store = match s.open_blob() {
|
|
||||||
Some(b) => b,
|
|
||||||
None => return Json(Vec::new()),
|
|
||||||
};
|
|
||||||
let mut ids = store.list_blob_ids().await.unwrap_or_default();
|
|
||||||
ids.sort();
|
|
||||||
let store = std::sync::Arc::new(store);
|
|
||||||
let out = ids.into_iter().skip(p.offset).take(p.limit).map(|id| {
|
|
||||||
let store = store.clone();
|
|
||||||
async move {
|
|
||||||
let manifest = store.load_manifest(&id).await.ok().flatten();
|
|
||||||
BlobSummary {
|
|
||||||
blob_id_hex: id.to_hex(),
|
|
||||||
size_bytes: manifest.as_ref().map(|m| m.total_size).unwrap_or(0),
|
|
||||||
chunk_count: manifest.map(|m| m.chunks.len()).unwrap_or(0),
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
});
|
|
||||||
let collected: Vec<BlobSummary> = futures_join(out).await;
|
|
||||||
Json(collected)
|
|
||||||
}
|
|
||||||
|
|
||||||
async fn handle_tags(
|
|
||||||
State(s): State<Arc<V2State>>,
|
|
||||||
Query(f): Query<TagFilter>,
|
|
||||||
) -> Json<Vec<TagSummaryV2>> {
|
|
||||||
let store = match s.open_tags() {
|
|
||||||
Some(t) => t,
|
|
||||||
None => return Json(Vec::new()),
|
|
||||||
};
|
};
|
||||||
let entries = store.list().await.unwrap_or_default();
|
let node = match s.fetch_node(&peer).await {
|
||||||
let out: Vec<_> = entries
|
Ok(r) => NodeStatusV2::ok(&name, r),
|
||||||
.into_iter()
|
Err(e) => NodeStatusV2::err(&name, format!("{e:#}")),
|
||||||
.filter(|e| f.prefix.is_empty() || e.key.starts_with(&f.prefix))
|
|
||||||
.map(|e| TagSummaryV2 {
|
|
||||||
key: e.key,
|
|
||||||
value_hex: e.value_hex,
|
|
||||||
})
|
|
||||||
.collect();
|
|
||||||
Json(out)
|
|
||||||
}
|
|
||||||
|
|
||||||
async fn handle_refs(
|
|
||||||
State(s): State<Arc<V2State>>,
|
|
||||||
Query(p): Query<Pagination>,
|
|
||||||
) -> Json<Vec<RefSummary>> {
|
|
||||||
let store = match s.open_refs() {
|
|
||||||
Some(r) => r,
|
|
||||||
None => return Json(Vec::new()),
|
|
||||||
};
|
};
|
||||||
let all = store.list().await.unwrap_or_default();
|
Json(node)
|
||||||
let out: Vec<_> = all
|
|
||||||
.into_iter()
|
|
||||||
.skip(p.offset)
|
|
||||||
.take(p.limit)
|
|
||||||
.map(|(k, v)| {
|
|
||||||
let mut kh = String::with_capacity(64);
|
|
||||||
for b in &k {
|
|
||||||
kh.push_str(&format!("{b:02x}"));
|
|
||||||
}
|
|
||||||
RefSummary {
|
|
||||||
fingerprint_hex: kh,
|
|
||||||
blob_id_hex: BlobId::from_bytes(v).to_hex(),
|
|
||||||
}
|
|
||||||
})
|
|
||||||
.collect();
|
|
||||||
Json(out)
|
|
||||||
}
|
|
||||||
|
|
||||||
async fn handle_snapshots(State(s): State<Arc<V2State>>) -> Json<Vec<SnapshotSummaryV2>> {
|
|
||||||
let store = match s.open_snapshots() {
|
|
||||||
Some(s) => s,
|
|
||||||
None => return Json(Vec::new()),
|
|
||||||
};
|
|
||||||
let entries = store.list().await.unwrap_or_default();
|
|
||||||
let out: Vec<_> = entries
|
|
||||||
.into_iter()
|
|
||||||
.map(|s| SnapshotSummaryV2 {
|
|
||||||
name: s.name,
|
|
||||||
created_at_unix: s.created_at_unix,
|
|
||||||
blob_count: s.blob_count,
|
|
||||||
file_bytes: s.file_bytes,
|
|
||||||
})
|
|
||||||
.collect();
|
|
||||||
Json(out)
|
|
||||||
}
|
|
||||||
|
|
||||||
async fn handle_ref_tracking(
|
|
||||||
State(s): State<Arc<V2State>>,
|
|
||||||
Query(f): Query<RefTrackingFilter>,
|
|
||||||
) -> Json<Vec<RefTrackingItem>> {
|
|
||||||
let store = match s.open_ref_tracking() {
|
|
||||||
Some(r) => r,
|
|
||||||
None => return Json(Vec::new()),
|
|
||||||
};
|
|
||||||
let all = store.list_all().await.unwrap_or_default();
|
|
||||||
let out: Vec<_> = all
|
|
||||||
.into_iter()
|
|
||||||
.filter(|e| f.repo.is_empty() || e.repo == f.repo)
|
|
||||||
.map(RefTrackingItem::from)
|
|
||||||
.collect();
|
|
||||||
Json(out)
|
|
||||||
}
|
|
||||||
|
|
||||||
// ── helpers ──────────────────────────────────────────────────────
|
|
||||||
|
|
||||||
/// Recursive byte count. Silent on read errors — used only for
|
|
||||||
/// reporting, not correctness.
|
|
||||||
fn dir_size_bytes(root: &std::path::Path) -> u64 {
|
|
||||||
let mut total: u64 = 0;
|
|
||||||
let mut stack = vec![root.to_path_buf()];
|
|
||||||
while let Some(dir) = stack.pop() {
|
|
||||||
let entries = match std::fs::read_dir(&dir) {
|
|
||||||
Ok(e) => e,
|
|
||||||
Err(_) => continue,
|
|
||||||
};
|
|
||||||
for entry in entries.flatten() {
|
|
||||||
let ft = match entry.file_type() {
|
|
||||||
Ok(t) => t,
|
|
||||||
Err(_) => continue,
|
|
||||||
};
|
|
||||||
if ft.is_dir() {
|
|
||||||
stack.push(entry.path());
|
|
||||||
} else if ft.is_file() {
|
|
||||||
if let Ok(m) = entry.metadata() {
|
|
||||||
total = total.saturating_add(m.len());
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
total
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Sequentially resolve N per-item async closures. Kept simple —
|
|
||||||
/// blob-metadata reads are I/O light and page-sized (200 default);
|
|
||||||
/// concurrent-fan-out isn't worth the complexity here.
|
|
||||||
async fn futures_join<F, T>(iter: impl Iterator<Item = F>) -> Vec<T>
|
|
||||||
where
|
|
||||||
F: std::future::Future<Output = T>,
|
|
||||||
{
|
|
||||||
let mut out = Vec::new();
|
|
||||||
for f in iter {
|
|
||||||
out.push(f.await);
|
|
||||||
}
|
|
||||||
out
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// ── route registration ──────────────────────────────────────────
|
// ── route registration ──────────────────────────────────────────
|
||||||
|
|
||||||
/// Attach the v2 route tree onto an existing axum Router.
|
|
||||||
///
|
|
||||||
/// Caller supplies the v2 state via `.with_state(state)` when
|
|
||||||
/// building the final router.
|
|
||||||
pub fn routes() -> Router<Arc<V2State>> {
|
pub fn routes() -> Router<Arc<V2State>> {
|
||||||
Router::new()
|
Router::new()
|
||||||
.route("/api/v2/node/local/status", get(handle_status))
|
.route("/api/v2/fleet", get(handle_fleet))
|
||||||
.route("/api/v2/node/:name/status", get(handle_node_status))
|
.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))
|
|
||||||
}
|
|
||||||
|
|
||||||
#[cfg(test)]
|
|
||||||
mod tests {
|
|
||||||
use super::*;
|
|
||||||
use axum::body::to_bytes;
|
|
||||||
use axum::{body::Body, http::Request};
|
|
||||||
use tempfile::TempDir;
|
|
||||||
use tower::ServiceExt;
|
|
||||||
|
|
||||||
async fn seed_state() -> (TempDir, Arc<V2State>) {
|
|
||||||
let tmp = TempDir::new().unwrap();
|
|
||||||
let root = tmp.path().to_path_buf();
|
|
||||||
// Seed one blob so /status has non-zero numbers.
|
|
||||||
let store = BlobStore::open(root.clone()).unwrap();
|
|
||||||
store.put_bytes(b"hello v2").await.unwrap();
|
|
||||||
let snap = SnapshotStore::open(root.clone()).unwrap();
|
|
||||||
snap.create("seed", &store, 42).await.unwrap();
|
|
||||||
let state = Arc::new(V2State {
|
|
||||||
node_name: "test-node".into(),
|
|
||||||
blob_store_root: Some(root),
|
|
||||||
});
|
|
||||||
(tmp, state)
|
|
||||||
}
|
|
||||||
|
|
||||||
#[tokio::test]
|
|
||||||
async fn status_reports_seeded_blob_and_snapshot() {
|
|
||||||
let (_tmp, state) = seed_state().await;
|
|
||||||
let app: Router = routes().with_state(state);
|
|
||||||
let resp = app
|
|
||||||
.oneshot(
|
|
||||||
Request::builder()
|
|
||||||
.uri("/api/v2/node/local/status")
|
|
||||||
.body(Body::empty())
|
|
||||||
.unwrap(),
|
|
||||||
)
|
|
||||||
.await
|
|
||||||
.unwrap();
|
|
||||||
assert_eq!(resp.status(), axum::http::StatusCode::OK);
|
|
||||||
let bytes = to_bytes(resp.into_body(), 65_536).await.unwrap();
|
|
||||||
let body: NodeStatusV2 = serde_json::from_slice(&bytes).unwrap();
|
|
||||||
assert_eq!(body.node_name, "test-node");
|
|
||||||
assert_eq!(body.blob_count, 1);
|
|
||||||
assert_eq!(body.snapshot_count, 1);
|
|
||||||
assert!(body.blob_store_bytes > 0);
|
|
||||||
}
|
|
||||||
|
|
||||||
#[tokio::test]
|
|
||||||
async fn cross_node_status_returns_not_implemented() {
|
|
||||||
let (_tmp, state) = seed_state().await;
|
|
||||||
let app: Router = routes().with_state(state);
|
|
||||||
let resp = app
|
|
||||||
.oneshot(
|
|
||||||
Request::builder()
|
|
||||||
.uri("/api/v2/node/architect/status")
|
|
||||||
.body(Body::empty())
|
|
||||||
.unwrap(),
|
|
||||||
)
|
|
||||||
.await
|
|
||||||
.unwrap();
|
|
||||||
assert_eq!(resp.status(), axum::http::StatusCode::NOT_IMPLEMENTED);
|
|
||||||
}
|
|
||||||
|
|
||||||
#[tokio::test]
|
|
||||||
async fn blobs_endpoint_paginates() {
|
|
||||||
let (_tmp, state) = seed_state().await;
|
|
||||||
// Seed more blobs.
|
|
||||||
if let Some(root) = &state.blob_store_root {
|
|
||||||
let store = BlobStore::open(root.clone()).unwrap();
|
|
||||||
for i in 0..5 {
|
|
||||||
store
|
|
||||||
.put_bytes(format!("payload-{i}").as_bytes())
|
|
||||||
.await
|
|
||||||
.unwrap();
|
|
||||||
}
|
|
||||||
}
|
|
||||||
let app: Router = routes().with_state(state);
|
|
||||||
let resp = app
|
|
||||||
.oneshot(
|
|
||||||
Request::builder()
|
|
||||||
.uri("/api/v2/storage/blobs?limit=2&offset=1")
|
|
||||||
.body(Body::empty())
|
|
||||||
.unwrap(),
|
|
||||||
)
|
|
||||||
.await
|
|
||||||
.unwrap();
|
|
||||||
assert_eq!(resp.status(), axum::http::StatusCode::OK);
|
|
||||||
let bytes = to_bytes(resp.into_body(), 65_536).await.unwrap();
|
|
||||||
let body: Vec<BlobSummary> = serde_json::from_slice(&bytes).unwrap();
|
|
||||||
assert_eq!(body.len(), 2);
|
|
||||||
}
|
|
||||||
|
|
||||||
#[tokio::test]
|
|
||||||
async fn tags_endpoint_filters_by_prefix() {
|
|
||||||
let (_tmp, state) = seed_state().await;
|
|
||||||
if let Some(root) = &state.blob_store_root {
|
|
||||||
let ts = TagStore::open(root.join("tags-db")).unwrap();
|
|
||||||
ts.put("clawverse:main:latest", &[0xAA; 32]).await.unwrap();
|
|
||||||
ts.put("other:tag", &[0xBB; 32]).await.unwrap();
|
|
||||||
}
|
|
||||||
let app: Router = routes().with_state(state);
|
|
||||||
let resp = app
|
|
||||||
.oneshot(
|
|
||||||
Request::builder()
|
|
||||||
.uri("/api/v2/storage/tags?prefix=clawverse:")
|
|
||||||
.body(Body::empty())
|
|
||||||
.unwrap(),
|
|
||||||
)
|
|
||||||
.await
|
|
||||||
.unwrap();
|
|
||||||
let bytes = to_bytes(resp.into_body(), 65_536).await.unwrap();
|
|
||||||
let body: Vec<TagSummaryV2> = serde_json::from_slice(&bytes).unwrap();
|
|
||||||
assert_eq!(body.len(), 1);
|
|
||||||
assert_eq!(body[0].key, "clawverse:main:latest");
|
|
||||||
}
|
|
||||||
|
|
||||||
#[tokio::test]
|
|
||||||
async fn snapshots_endpoint_returns_seeded() {
|
|
||||||
let (_tmp, state) = seed_state().await;
|
|
||||||
let app: Router = routes().with_state(state);
|
|
||||||
let resp = app
|
|
||||||
.oneshot(
|
|
||||||
Request::builder()
|
|
||||||
.uri("/api/v2/storage/snapshots")
|
|
||||||
.body(Body::empty())
|
|
||||||
.unwrap(),
|
|
||||||
)
|
|
||||||
.await
|
|
||||||
.unwrap();
|
|
||||||
let bytes = to_bytes(resp.into_body(), 65_536).await.unwrap();
|
|
||||||
let body: Vec<SnapshotSummaryV2> = serde_json::from_slice(&bytes).unwrap();
|
|
||||||
assert_eq!(body.len(), 1);
|
|
||||||
assert_eq!(body[0].name, "seed");
|
|
||||||
assert_eq!(body[0].created_at_unix, 42);
|
|
||||||
}
|
|
||||||
|
|
||||||
#[tokio::test]
|
|
||||||
async fn ref_tracking_filters_by_repo() {
|
|
||||||
let (_tmp, state) = seed_state().await;
|
|
||||||
if let Some(root) = &state.blob_store_root {
|
|
||||||
let rt = RefTracking::open(root.clone()).unwrap();
|
|
||||||
rt.record([0x11; 32], "r/x", "main", 100).await.unwrap();
|
|
||||||
rt.record([0x22; 32], "r/y", "main", 100).await.unwrap();
|
|
||||||
}
|
|
||||||
let app: Router = routes().with_state(state);
|
|
||||||
let resp = app
|
|
||||||
.oneshot(
|
|
||||||
Request::builder()
|
|
||||||
.uri("/api/v2/storage/ref-tracking?repo=r/x")
|
|
||||||
.body(Body::empty())
|
|
||||||
.unwrap(),
|
|
||||||
)
|
|
||||||
.await
|
|
||||||
.unwrap();
|
|
||||||
let bytes = to_bytes(resp.into_body(), 65_536).await.unwrap();
|
|
||||||
let body: Vec<RefTrackingItem> = serde_json::from_slice(&bytes).unwrap();
|
|
||||||
assert_eq!(body.len(), 1);
|
|
||||||
assert_eq!(body[0].repo, "r/x");
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user