From b74d3f7b5fb13994123a57fff81eef5860f3d704 Mon Sep 17 00:00:00 2001 From: Zachery Aaron Shores-Chmielewski Date: Mon, 16 Feb 2026 00:01:30 +0700 Subject: [PATCH] feat: datastore complete with manual testing and docs --- .cargo/config.toml | 2 + Cargo.lock | 5 + Cargo.toml | 2 +- crates/datastore/Cargo.toml | 6 +- crates/datastore/README.md | 141 ++++++++ crates/datastore/src/api.rs | 36 +- crates/datastore/src/bin/store_node.rs | 94 +++-- crates/datastore/src/lib.rs | 3 + crates/datastore/src/metrics.rs | 197 +++++++++++ crates/datastore/src/ui_html.rs | 326 ++++++++++++++++++ .../datastore/tests/api_integration_test.rs | 190 ++++++++++ .../tests/dashboard_integration_test.rs | 205 +++++++++++ .../src/kademlia/routing_table.rs | 3 +- crates/distribution/src/transport.rs | 1 + crates/distribution/tests/common/mod.rs | 1 + .../distribution/tests/kademlia_directory.rs | 2 +- .../src/actor_detail_html.rs | 2 + crates/runtime-dashboard/src/actors_html.rs | 1 + .../runtime-dashboard/src/dashboard_html.rs | 1 + .../src/datastore_collector.rs | 16 + .../runtime-dashboard/src/datastore_html.rs | 286 +++++++++++++++ .../src/distribution_html.rs | 1 + crates/runtime-dashboard/src/lib.rs | 14 + crates/runtime-dashboard/src/server.rs | 46 +++ crates/runtime-dashboard/src/topology_html.rs | 1 + crates/runtime-dashboard/tests/command_api.rs | 2 +- docs/datastore/actors.md | 278 +++++++++++++++ docs/datastore/streaming.md | 103 ++++++ docs/diagrams/datastore_chunk_lifecycle.svg | 170 +++++++++ docs/diagrams/datastore_transfer.svg | 153 ++++++++ src/runtime.rs | 5 - xtask/Cargo.toml | 4 + xtask/src/main.rs | 197 +++++++++++ 33 files changed, 2461 insertions(+), 33 deletions(-) create mode 100644 .cargo/config.toml create mode 100644 crates/datastore/README.md create mode 100644 crates/datastore/src/metrics.rs create mode 100644 crates/datastore/src/ui_html.rs create mode 100644 crates/datastore/tests/api_integration_test.rs create mode 100644 crates/datastore/tests/dashboard_integration_test.rs create mode 100644 crates/runtime-dashboard/src/datastore_collector.rs create mode 100644 crates/runtime-dashboard/src/datastore_html.rs create mode 100644 docs/datastore/actors.md create mode 100644 docs/datastore/streaming.md create mode 100644 docs/diagrams/datastore_chunk_lifecycle.svg create mode 100644 docs/diagrams/datastore_transfer.svg create mode 100644 xtask/Cargo.toml create mode 100644 xtask/src/main.rs diff --git a/.cargo/config.toml b/.cargo/config.toml new file mode 100644 index 0000000..35049cb --- /dev/null +++ b/.cargo/config.toml @@ -0,0 +1,2 @@ +[alias] +xtask = "run --package xtask --" diff --git a/Cargo.lock b/Cargo.lock index e112d14..d295a32 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4391,6 +4391,7 @@ dependencies = [ "swactor-std", "tempfile", "tiny_http", + "toml", "ureq", ] @@ -6231,6 +6232,10 @@ dependencies = [ "xml-rs", ] +[[package]] +name = "xtask" +version = "0.1.0" + [[package]] name = "yoke" version = "0.8.1" diff --git a/Cargo.toml b/Cargo.toml index 755ce1f..60a0454 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,5 +1,5 @@ [workspace] -members = [".", "crates/python", "crates/wasm", "crates/bin-runner", "crates/simulation", "crates/runtime-dashboard", "crates/distribution", "crates/std", "crates/datastore", "tests/docker"] +members = [".", "crates/python", "crates/wasm", "crates/bin-runner", "crates/simulation", "crates/runtime-dashboard", "crates/distribution", "crates/std", "crates/datastore", "tests/docker", "xtask"] exclude = ["tools/depgraph"] [package] diff --git a/crates/datastore/Cargo.toml b/crates/datastore/Cargo.toml index cab55a4..4aa7b5c 100644 --- a/crates/datastore/Cargo.toml +++ b/crates/datastore/Cargo.toml @@ -14,6 +14,7 @@ clap = { version = "4", features = ["derive"], optional = true } ureq = { version = "2", features = ["json"], optional = true } ctrlc = { version = "3", optional = true } runtime-dashboard = { path = "../runtime-dashboard", optional = true } +toml = { version = "0.8", optional = true } [dev-dependencies] serde_json = "1" @@ -21,9 +22,12 @@ proptest = "1" tempfile = "3" swactor = { path = "../.." } swactor-std = { path = "../std" } +ureq = { version = "2", features = ["json"] } +tiny_http = "0.12" +runtime-dashboard = { path = "../runtime-dashboard" } [features] -node = ["dep:tiny_http", "dep:clap", "dep:ctrlc", "dep:runtime-dashboard"] +node = ["dep:tiny_http", "dep:clap", "dep:ctrlc", "dep:runtime-dashboard", "dep:toml"] cli = ["dep:clap", "dep:ureq"] [[bin]] diff --git a/crates/datastore/README.md b/crates/datastore/README.md new file mode 100644 index 0000000..129a83e --- /dev/null +++ b/crates/datastore/README.md @@ -0,0 +1,141 @@ +# swactor-datastore + +Distributed content-addressed datastore built on [swactor](../../README.md). Objects are split into fixed-size chunks, identified by their blake3 hash, and replicated across a peer-to-peer network via epidemic gossip. + +## Building + +Node binary (HTTP server + actor runtime): + +```sh +cargo build -p swactor-datastore --features node +``` + +CLI client: + +```sh +cargo build -p swactor-datastore --features cli +``` + +Both at once: + +```sh +cargo build -p swactor-datastore --features node,cli +``` + +## Node + +Start a datastore node: + +```sh +swactor-store-node +``` + +### Flags + +| Flag | Default | Description | +|------|---------|-------------| +| `--port` | `9091` | HTTP API port | +| `--storage-path` | *(in-memory)* | Directory for persistent storage | +| `--dashboard-port` | *(disabled)* | Runtime dashboard port | +| `--chunk-size` | `1048576` | Chunk size in bytes (1 MB) | +| `--gc-interval` | `1000` | GC interval in ticks (~100s) | +| `--disseminate-interval` | `50` | Gossip interval in ticks (~5s) | +| `--config` | *(none)* | Path to a TOML config file | + +Example with persistent storage and dashboard: + +```sh +swactor-store-node --storage-path ./data --dashboard-port 9090 +``` + +### Config file + +Create a `store.toml` and pass it with `--config`: + +```toml +port = 9091 +storage_path = "./my-data" +dashboard_port = 9090 +chunk_size = 1048576 +gc_interval = 1000 +disseminate_interval = 50 +``` + +CLI flags override config file values. Omitted fields use built-in defaults. + +```sh +swactor-store-node --config store.toml --port 8080 +``` + +## CLI + +The `swactor-store` command talks to a running node over HTTP. + +### Status + +```sh +swactor-store status +``` + +### Put + +```sh +swactor-store put photo.jpg --name "vacation" +``` + +### Get (metadata) + +```sh +swactor-store get +``` + +### Get (download) + +```sh +swactor-store get --output photo.jpg +``` + +### Delete + +```sh +swactor-store delete +``` + +### List (local) + +```sh +swactor-store list +``` + +### List (swarm-wide) + +```sh +swactor-store list --all +``` + +### Filter by name + +```sh +swactor-store list --name vacation +``` + +Use `--url` to point at a different node: + +```sh +swactor-store --url http://192.168.1.50:9091 list +``` + +## Web UI + +Visit `http://:/` in a browser. The UI supports uploading, listing, downloading, inspecting, and deleting objects — works on desktop and mobile. + +## HTTP API + +| Method | Path | Description | +|--------|------|-------------| +| `GET` | `/api/status` | Node identity | +| `POST` | `/api/put?name=...` | Upload (body = raw bytes) | +| `GET` | `/api/list` | List objects (`?all=true` for swarm) | +| `GET` | `/api/get?hash=...` | Object metadata + manifest | +| `GET` | `/api/data?hash=...` | Download reassembled binary | +| `POST` | `/api/delete?hash=...` | Delete object | diff --git a/crates/datastore/src/api.rs b/crates/datastore/src/api.rs index fc0033a..28046fc 100644 --- a/crates/datastore/src/api.rs +++ b/crates/datastore/src/api.rs @@ -14,6 +14,7 @@ use swactor::runtime::{Inbox, Runtime}; use crate::chunking::reassemble_blob; use crate::messages::{BlobStoreMsg, DatastoreNodeMsg, DatastoreResponse, MetadataMsg}; +use crate::metrics::DatastoreMetrics; use crate::types::ContentHash; /// Per-peer actor addresses needed for remote operations. @@ -30,6 +31,7 @@ struct ApiState { metadata_addr: ActorAddress, blob_store_addr: ActorAddress, peers: Arc>>, + metrics: Arc, } const POLL_TIMEOUT: Duration = Duration::from_secs(5); @@ -67,6 +69,16 @@ fn respond_bytes(request: tiny_http::Request, data: &[u8]) { let _ = request.respond(response); } +fn respond_html(request: tiny_http::Request) { + let response = + tiny_http::Response::from_string(crate::ui_html::DATASTORE_UI_HTML).with_header( + "Content-Type: text/html; charset=utf-8" + .parse::() + .unwrap(), + ); + let _ = request.respond(response); +} + fn respond_error(request: tiny_http::Request, status: u16, msg: &str) { let json = serde_json::json!({ "error": msg }).to_string(); let response = tiny_http::Response::from_string(json) @@ -186,6 +198,7 @@ fn handle_put(mut request: tiny_http::Request, url: &str, state: &ApiState) { // Can't respond — request consumed return; } + let body_len = body.len(); let inbox = match state.runtime.new_inbox::() { Ok(i) => i, @@ -207,7 +220,13 @@ fn handle_put(mut request: tiny_http::Request, url: &str, state: &ApiState) { match poll_response(&inbox, POLL_TIMEOUT) { Some(DatastoreResponse::PutOk { content_hash }) => { - let json = serde_json::json!({ "content_hash": content_hash.to_hex() }).to_string(); + let hex = content_hash.to_hex(); + state.metrics.record_put( + &hex, + params.get("name").map(|s| s.as_str()), + body_len as u64, + ); + let json = serde_json::json!({ "content_hash": hex }).to_string(); respond_json(request, &json); } Some(DatastoreResponse::Error { reason }) => { @@ -257,6 +276,7 @@ fn handle_get(request: tiny_http::Request, url: &str, state: &ApiState) { match poll_response(&inbox, POLL_TIMEOUT) { Some(DatastoreResponse::GetOk { entry, manifest }) => { + state.metrics.record_get(&content_hash.to_hex()); let json = serde_json::json!({ "entry": entry_to_json(&entry), "manifest": manifest_to_json(&manifest), @@ -313,6 +333,8 @@ fn handle_data(request: tiny_http::Request, url: &str, state: &ApiState) { }, ); + state.metrics.record_get(&content_hash.to_hex()); + let (entry, manifest) = match poll_response(&inbox, POLL_TIMEOUT) { Some(DatastoreResponse::GetOk { entry, manifest }) => (entry, manifest), Some(DatastoreResponse::NotFound) => { @@ -411,7 +433,9 @@ fn handle_delete(request: tiny_http::Request, url: &str, state: &ApiState) { match poll_response(&inbox, POLL_TIMEOUT) { Some(DatastoreResponse::DeleteOk { content_hash }) => { - let json = serde_json::json!({ "content_hash": content_hash.to_hex() }).to_string(); + let hex = content_hash.to_hex(); + state.metrics.record_delete(&hex, 0); + let json = serde_json::json!({ "content_hash": hex }).to_string(); respond_json(request, &json); } Some(DatastoreResponse::NotFound) => { @@ -609,6 +633,8 @@ fn try_remote_get( }; // Fetch each chunk from peer and store locally + let hash_hex = content_hash.to_hex(); + state.metrics.begin_transfer(&hash_hex, manifest.chunks.len()); let mut all_ok = true; for chunk_ref in &manifest.chunks { let chunk_inbox = match state.runtime.new_inbox::() { @@ -648,6 +674,7 @@ fn try_remote_get( ); // Wait for confirmation let _ = poll_response(&store_inbox, Duration::from_secs(2)); + state.metrics.advance_transfer(&hash_hex); } _ => { all_ok = false; @@ -656,6 +683,8 @@ fn try_remote_get( } } + state.metrics.end_transfer(&hash_hex); + if !all_ok { continue; } @@ -707,6 +736,7 @@ pub fn start_api_server( metadata_addr: ActorAddress, blob_store_addr: ActorAddress, port: u16, + metrics: Arc, ) -> (Arc, Arc>>) { let shutdown = Arc::new(AtomicBool::new(false)); let peers: Arc>> = Arc::new(Mutex::new(Vec::new())); @@ -717,6 +747,7 @@ pub fn start_api_server( metadata_addr, blob_store_addr, peers: Arc::clone(&peers), + metrics, }); let addr = format!("0.0.0.0:{port}"); @@ -749,6 +780,7 @@ pub fn start_api_server( ("POST", "/api/delete") => handle_delete(request, &url, &state), ("GET", "/api/list") => handle_list(request, &url, &state), ("GET", "/api/status") => handle_status(request, &state), + ("GET", "/") => respond_html(request), _ => { respond_error(request, 404, "not found"); } diff --git a/crates/datastore/src/bin/store_node.rs b/crates/datastore/src/bin/store_node.rs index 0371585..f8e6c6e 100644 --- a/crates/datastore/src/bin/store_node.rs +++ b/crates/datastore/src/bin/store_node.rs @@ -9,6 +9,7 @@ use std::thread; use std::time::Duration; use clap::Parser; +use serde::Deserialize; use swactor::config::RuntimeConfig; use swactor::runtime::Runtime; @@ -16,6 +17,7 @@ use swactor::runtime::Runtime; use swactor_datastore::actors::{BlobStoreActor, DatastoreNode, MetadataActor}; use swactor_datastore::api::start_api_server; use swactor_datastore::messages::MetadataMsg; +use swactor_datastore::metrics::DatastoreMetrics; use swactor_datastore::storage::{FilesystemBackend, InMemoryBackend}; use swactor_datastore::DatastoreConfig; @@ -24,9 +26,13 @@ use distribution::types::NodeId; #[derive(Parser)] #[command(name = "swactor-store-node", about = "Swactor distributed datastore node")] struct Args { + /// Path to a TOML config file + #[arg(long)] + config: Option, + /// HTTP API port - #[arg(long, default_value = "9091")] - port: u16, + #[arg(long)] + port: Option, /// Storage directory (omit for in-memory) #[arg(long)] @@ -37,20 +43,62 @@ struct Args { dashboard_port: Option, /// Chunk size in bytes - #[arg(long, default_value = "1048576")] - chunk_size: u32, + #[arg(long)] + chunk_size: Option, /// GC interval in ticks (each tick is ~100ms) - #[arg(long, default_value = "1000")] - gc_interval: u64, + #[arg(long)] + gc_interval: Option, /// Dissemination interval in ticks - #[arg(long, default_value = "50")] + #[arg(long)] + disseminate_interval: Option, +} + +#[derive(Deserialize, Default)] +struct NodeConfig { + port: Option, + storage_path: Option, + dashboard_port: Option, + chunk_size: Option, + gc_interval: Option, + disseminate_interval: Option, +} + +/// Resolved configuration with CLI > config file > defaults applied. +struct ResolvedConfig { + port: u16, + storage_path: Option, + dashboard_port: Option, + chunk_size: u32, + gc_interval: u64, disseminate_interval: u64, } +fn resolve_config(args: &Args) -> ResolvedConfig { + let file_cfg = match &args.config { + Some(path) => { + let contents = std::fs::read_to_string(path) + .unwrap_or_else(|e| panic!("failed to read config file {}: {e}", path.display())); + toml::from_str::(&contents) + .unwrap_or_else(|e| panic!("failed to parse config file {}: {e}", path.display())) + } + None => NodeConfig::default(), + }; + + ResolvedConfig { + port: args.port.or(file_cfg.port).unwrap_or(9091), + storage_path: args.storage_path.clone().or(file_cfg.storage_path), + dashboard_port: args.dashboard_port.or(file_cfg.dashboard_port), + chunk_size: args.chunk_size.or(file_cfg.chunk_size).unwrap_or(1_048_576), + gc_interval: args.gc_interval.or(file_cfg.gc_interval).unwrap_or(1000), + disseminate_interval: args.disseminate_interval.or(file_cfg.disseminate_interval).unwrap_or(50), + } +} + fn main() { let args = Args::parse(); + let cfg = resolve_config(&args); let stop = Arc::new(AtomicBool::new(false)); // Signal handler @@ -63,7 +111,7 @@ fn main() { } // Optionally start dashboard - let dash = args.dashboard_port.map(|port| { + let dash = cfg.dashboard_port.map(|port| { let d = runtime_dashboard::start_dashboard(runtime_dashboard::DashboardConfig { port, ..Default::default() @@ -106,18 +154,18 @@ fn main() { // Datastore config let config = DatastoreConfig { - chunk_size: args.chunk_size, - storage_path: args + chunk_size: cfg.chunk_size, + storage_path: cfg .storage_path .as_ref() .map(|s| s.into()) .unwrap_or_else(|| "datastore".into()), - gc_interval: args.gc_interval, + gc_interval: cfg.gc_interval, ..Default::default() }; // Create storage backend - let backend: Box = match &args.storage_path { + let backend: Box = match &cfg.storage_path { Some(path) => { let p = std::path::PathBuf::from(path); std::fs::create_dir_all(&p).expect("failed to create storage directory"); @@ -145,8 +193,14 @@ fn main() { // Start runtime let handle = rt.run().expect("failed to start runtime"); + // Create datastore metrics + let node_hex: String = node_id.0.iter().map(|b| format!("{b:02x}")).collect(); + let metrics = Arc::new(DatastoreMetrics::new()); + metrics.set_node_id(node_hex.clone()); + if let Some(ref d) = dash { d.set_runtime(handle.runtime.clone(), collector); + d.set_datastore(Arc::clone(&metrics) as Arc); } // Start HTTP API @@ -155,17 +209,17 @@ fn main() { datastore_addr, metadata_addr, blob_store_addr, - args.port, + cfg.port, + Arc::clone(&metrics), ); - let node_hex: String = node_id.0.iter().map(|b| format!("{b:02x}")).collect(); eprintln!("Node {} started", &node_hex[..8]); - eprintln!("API at http://0.0.0.0:{}", args.port); - if let Some(port) = args.dashboard_port { + eprintln!("API at http://0.0.0.0:{}", cfg.port); + if let Some(port) = cfg.dashboard_port { eprintln!("Dashboard at http://0.0.0.0:{port}"); } - if args.storage_path.is_some() { - eprintln!("Storage: {}", args.storage_path.as_ref().unwrap()); + if cfg.storage_path.is_some() { + eprintln!("Storage: {}", cfg.storage_path.as_ref().unwrap()); } else { eprintln!("Storage: in-memory"); } @@ -175,13 +229,13 @@ fn main() { while !stop.load(Ordering::Relaxed) { round += 1; - if round % args.gc_interval == 0 { + if round % cfg.gc_interval == 0 { let _ = handle .runtime .send_to(metadata_addr, MetadataMsg::GcTick); } - if round % args.disseminate_interval == 0 { + if round % cfg.disseminate_interval == 0 { let _ = handle .runtime .send_to(metadata_addr, MetadataMsg::DisseminateTick); diff --git a/crates/datastore/src/lib.rs b/crates/datastore/src/lib.rs index 0c90686..f7ed2d1 100644 --- a/crates/datastore/src/lib.rs +++ b/crates/datastore/src/lib.rs @@ -4,8 +4,11 @@ pub mod chunking; pub mod storage; pub mod actors; pub mod cli; +pub mod metrics; #[cfg(feature = "node")] pub mod api; +#[cfg(feature = "node")] +pub mod ui_html; pub use types::{ChunkRef, ContentHash, DatastoreConfig, ObjectEntry, ObjectManifest}; pub use messages::{BlobStoreMsg, DatastoreNodeMsg, DatastoreResponse, MetadataMsg, TransferMsg}; diff --git a/crates/datastore/src/metrics.rs b/crates/datastore/src/metrics.rs new file mode 100644 index 0000000..5e6f2b0 --- /dev/null +++ b/crates/datastore/src/metrics.rs @@ -0,0 +1,197 @@ +//! Thread-safe metrics for the datastore, consumed by the runtime dashboard. +//! +//! `DatastoreMetrics` accumulates counters and event history from any thread +//! (API handlers run on `tiny_http` worker threads). The dashboard polls +//! `snapshot()` every ~200ms via the `DatastoreStatsProvider` trait. + +use std::collections::VecDeque; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::Mutex; +use std::time::{SystemTime, UNIX_EPOCH}; + +use serde::{Deserialize, Serialize}; + +// ── Snapshot types (serializable, sent to the dashboard) ──────────────────── + +/// Point-in-time snapshot of the datastore's state and metrics. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct DatastoreSnapshot { + pub node_id: String, + pub object_count: u64, + pub total_bytes: u64, + pub put_ops: u64, + pub get_ops: u64, + pub delete_ops: u64, + pub objects: Vec, + pub recent_events: Vec, + pub active_transfers: Vec, +} + +/// Summary of a single stored object (for the dashboard table). +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ObjectSummary { + pub hash: String, + pub name: Option, + pub size_bytes: u64, +} + +/// A recorded datastore operation. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct DatastoreEvent { + pub timestamp_ms: u64, + pub kind: String, + pub hash: String, + pub name: Option, + pub size_bytes: u64, +} + +/// Progress of an in-flight chunk transfer. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct TransferProgress { + pub hash: String, + pub chunks_received: usize, + pub chunks_total: usize, +} + +// ── Live metrics (thread-safe, mutated from API handlers) ─────────────────── + +const MAX_EVENTS: usize = 200; + +/// Thread-safe metrics accumulator for the datastore. +/// +/// Atomic counters for the hot path (put/get/delete counts). A `Mutex`-guarded +/// ring buffer for the event timeline and a small vec for active transfers. +pub struct DatastoreMetrics { + node_id: Mutex, + put_ops: AtomicU64, + get_ops: AtomicU64, + delete_ops: AtomicU64, + objects: Mutex>, + events: Mutex>, + transfers: Mutex>, +} + +impl DatastoreMetrics { + pub fn new() -> Self { + Self { + node_id: Mutex::new(String::new()), + put_ops: AtomicU64::new(0), + get_ops: AtomicU64::new(0), + delete_ops: AtomicU64::new(0), + objects: Mutex::new(Vec::new()), + events: Mutex::new(VecDeque::with_capacity(MAX_EVENTS + 1)), + transfers: Mutex::new(Vec::new()), + } + } + + pub fn set_node_id(&self, hex: String) { + *self.node_id.lock().unwrap() = hex; + } + + /// Record a successful PUT operation. + pub fn record_put(&self, hash: &str, name: Option<&str>, size_bytes: u64) { + self.put_ops.fetch_add(1, Ordering::Relaxed); + self.push_event("put", hash, name, size_bytes); + let mut objs = self.objects.lock().unwrap(); + objs.push(ObjectSummary { + hash: hash.to_string(), + name: name.map(|s| s.to_string()), + size_bytes, + }); + } + + /// Record a successful GET operation. + pub fn record_get(&self, hash: &str) { + self.get_ops.fetch_add(1, Ordering::Relaxed); + self.push_event("get", hash, None, 0); + } + + /// Record a successful DELETE operation. + pub fn record_delete(&self, hash: &str, size_bytes: u64) { + self.delete_ops.fetch_add(1, Ordering::Relaxed); + self.push_event("delete", hash, None, size_bytes); + let mut objs = self.objects.lock().unwrap(); + objs.retain(|o| o.hash != hash); + } + + /// Begin tracking a chunk transfer. + pub fn begin_transfer(&self, hash: &str, chunks_total: usize) { + let mut transfers = self.transfers.lock().unwrap(); + transfers.push(TransferProgress { + hash: hash.to_string(), + chunks_received: 0, + chunks_total, + }); + } + + /// Advance a tracked transfer by one chunk. + pub fn advance_transfer(&self, hash: &str) { + let mut transfers = self.transfers.lock().unwrap(); + if let Some(t) = transfers.iter_mut().find(|t| t.hash == hash) { + t.chunks_received += 1; + } + } + + /// Remove a completed/failed transfer from tracking. + pub fn end_transfer(&self, hash: &str) { + let mut transfers = self.transfers.lock().unwrap(); + transfers.retain(|t| t.hash != hash); + } + + /// Seed the object list (e.g. from an initial LIST query at startup). + pub fn seed_objects(&self, objects: Vec) { + *self.objects.lock().unwrap() = objects; + } + + /// Capture a serializable snapshot of the current metrics. + pub fn snapshot(&self) -> DatastoreSnapshot { + let objs = self.objects.lock().unwrap(); + let total_bytes: u64 = objs.iter().map(|o| o.size_bytes).sum(); + let events = self.events.lock().unwrap(); + let transfers = self.transfers.lock().unwrap(); + + DatastoreSnapshot { + node_id: self.node_id.lock().unwrap().clone(), + object_count: objs.len() as u64, + total_bytes, + put_ops: self.put_ops.load(Ordering::Relaxed), + get_ops: self.get_ops.load(Ordering::Relaxed), + delete_ops: self.delete_ops.load(Ordering::Relaxed), + objects: objs.clone(), + recent_events: events.iter().cloned().collect(), + active_transfers: transfers.clone(), + } + } + + fn push_event(&self, kind: &str, hash: &str, name: Option<&str>, size_bytes: u64) { + let event = DatastoreEvent { + timestamp_ms: now_ms(), + kind: kind.to_string(), + hash: hash.to_string(), + name: name.map(|s| s.to_string()), + size_bytes, + }; + let mut events = self.events.lock().unwrap(); + if events.len() >= MAX_EVENTS { + events.pop_front(); + } + events.push_back(event); + } +} + +fn now_ms() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_millis() as u64 +} + +// ── Dashboard integration (only when runtime-dashboard is available) ──────── + +#[cfg(feature = "node")] +impl runtime_dashboard::datastore_collector::DatastoreStatsProvider for DatastoreMetrics { + fn snapshot_json(&self) -> Option { + let snap = self.snapshot(); + serde_json::to_string(&snap).ok() + } +} diff --git a/crates/datastore/src/ui_html.rs b/crates/datastore/src/ui_html.rs new file mode 100644 index 0000000..592a878 --- /dev/null +++ b/crates/datastore/src/ui_html.rs @@ -0,0 +1,326 @@ +pub const DATASTORE_UI_HTML: &str = r##" + + + + +swactor-store + + + + +
+
+

swactor-store

+ connecting... +
+
+ +
+ +
+

Upload

+
+ + + +
+
+ + +
+

Objects

+
+
+
+ + + + +
+ + + +"##; diff --git a/crates/datastore/tests/api_integration_test.rs b/crates/datastore/tests/api_integration_test.rs new file mode 100644 index 0000000..6be565a --- /dev/null +++ b/crates/datastore/tests/api_integration_test.rs @@ -0,0 +1,190 @@ +#![cfg(feature = "node")] + +//! Integration test: spins up a real datastore node with HTTP API and exercises +//! the full CRUD lifecycle over HTTP. + +use std::sync::Arc; +use std::thread; +use std::time::Duration; + +use swactor::config::RuntimeConfig; +use swactor::runtime::Runtime; + +use swactor_datastore::actors::{BlobStoreActor, DatastoreNode, MetadataActor}; +use swactor_datastore::api::start_api_server; +use swactor_datastore::metrics::DatastoreMetrics; +use swactor_datastore::storage::InMemoryBackend; +use swactor_datastore::DatastoreConfig; + +use distribution::types::NodeId; + +fn find_free_port() -> u16 { + std::net::TcpListener::bind("127.0.0.1:0") + .unwrap() + .local_addr() + .unwrap() + .port() +} + +/// Full CRUD lifecycle over HTTP: +/// status → put → list → get metadata → get data → delete → list (empty) → get (404) +#[test] +fn http_crud_lifecycle() { + let port = find_free_port(); + let base = format!("http://127.0.0.1:{port}"); + + // Set up runtime with worker threads (needed for HTTP server) + let collector = runtime_dashboard::collector::StatsCollector::new(2); + let mut rt = Runtime::new(RuntimeConfig { + num_threads: 2, + max_actors: 1024, + channel_buffer_size: 2000, + ..Default::default() + }); + rt.set_stats_hook(collector); + + let node_id = NodeId([0xAA; 32]); + + let config = DatastoreConfig { + chunk_size: 1_048_576, + gc_interval: 1000, + ..Default::default() + }; + + let backend: Box = Box::new(InMemoryBackend::new()); + let blob_store_addr = rt.spawn(BlobStoreActor::new(backend)).unwrap(); + + let mut metadata = MetadataActor::new(node_id, &config); + metadata.set_blob_store(blob_store_addr); + let metadata_addr = rt.spawn(metadata).unwrap(); + + let datastore_node = DatastoreNode::new(node_id, blob_store_addr, metadata_addr, config); + let datastore_addr = rt.spawn(datastore_node).unwrap(); + + let handle = rt.run().expect("failed to start runtime"); + + let metrics = Arc::new(DatastoreMetrics::new()); + let (api_shutdown, _peers) = start_api_server( + handle.runtime.clone(), + datastore_addr, + metadata_addr, + blob_store_addr, + port, + Arc::clone(&metrics), + ); + + // Give the HTTP server time to bind + thread::sleep(Duration::from_millis(200)); + + let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + run_crud_scenario(&base, &metrics); + })); + + // Cleanup + api_shutdown.store(true, std::sync::atomic::Ordering::Relaxed); + handle.shutdown(); + handle.join(); + + if let Err(e) = result { + std::panic::resume_unwind(e); + } +} + +fn run_crud_scenario(base: &str, metrics: &Arc) { + // 1. Status — should return a node_id + let status: serde_json::Value = ureq::get(&format!("{base}/api/status")) + .call() + .expect("status request failed") + .into_json() + .unwrap(); + let node_id = status["node_id"].as_str().expect("node_id should be a string"); + assert_eq!(node_id.len(), 64, "node_id should be 64 hex chars"); + + // 2. Put — upload some content + let content = b"hello from the integration test!"; + let put_resp: serde_json::Value = ureq::post(&format!("{base}/api/put?name=greeting")) + .send_bytes(content) + .expect("put request failed") + .into_json() + .unwrap(); + let hash = put_resp["content_hash"] + .as_str() + .expect("put should return content_hash"); + assert_eq!(hash.len(), 64, "content_hash should be 64 hex chars"); + + // 3. List — should contain exactly one entry matching our upload + let list_resp: serde_json::Value = ureq::get(&format!("{base}/api/list")) + .call() + .expect("list request failed") + .into_json() + .unwrap(); + let entries = list_resp["entries"].as_array().expect("entries should be an array"); + assert_eq!(entries.len(), 1, "should have exactly 1 entry after put"); + assert_eq!(entries[0]["content_hash"].as_str().unwrap(), hash); + assert_eq!(entries[0]["name"].as_str().unwrap(), "greeting"); + + // 4. Get metadata — entry + manifest for the uploaded object + let get_resp: serde_json::Value = ureq::get(&format!("{base}/api/get?hash={hash}")) + .call() + .expect("get request failed") + .into_json() + .unwrap(); + let entry = &get_resp["entry"]; + assert_eq!(entry["content_hash"].as_str().unwrap(), hash); + assert_eq!(entry["name"].as_str().unwrap(), "greeting"); + assert_eq!(entry["size_bytes"].as_u64().unwrap(), content.len() as u64); + let manifest = &get_resp["manifest"]; + let chunks = manifest["chunks"].as_array().expect("manifest should have chunks"); + assert!(!chunks.is_empty(), "manifest should have at least one chunk"); + + // 5. Get data — download the raw bytes and verify content matches + let data_resp = ureq::get(&format!("{base}/api/data?hash={hash}")) + .call() + .expect("data request failed"); + let mut downloaded = Vec::new(); + data_resp + .into_reader() + .read_to_end(&mut downloaded) + .unwrap(); + assert_eq!(downloaded, content, "downloaded bytes should match uploaded content"); + + // 6. Delete — remove the object + let del_resp: serde_json::Value = ureq::post(&format!("{base}/api/delete?hash={hash}")) + .call() + .expect("delete request failed") + .into_json() + .unwrap(); + assert_eq!(del_resp["content_hash"].as_str().unwrap(), hash); + + // 7. List after delete — should be empty + let list_resp2: serde_json::Value = ureq::get(&format!("{base}/api/list")) + .call() + .expect("list request failed") + .into_json() + .unwrap(); + let entries2 = list_resp2["entries"] + .as_array() + .expect("entries should be an array"); + assert!(entries2.is_empty(), "list should be empty after delete"); + + // 8. Get after delete — should 404 + let get_err = ureq::get(&format!("{base}/api/get?hash={hash}")).call(); + match get_err { + Err(ureq::Error::Status(404, _)) => {} // expected + Err(e) => panic!("expected 404, got error: {e}"), + Ok(_) => panic!("expected 404, got 200"), + } + + // 9. Verify metrics snapshot reflects the full lifecycle + let snap = metrics.snapshot(); + assert_eq!(snap.put_ops, 1, "one put recorded"); + // handle_get + handle_data = 2 get operations + assert_eq!(snap.get_ops, 2, "metadata-get + data-get recorded"); + assert_eq!(snap.delete_ops, 1, "one delete recorded"); + assert!(snap.objects.is_empty(), "no objects after delete"); + assert!( + snap.recent_events.len() >= 4, + "at least 4 events (put + get + get + delete), got {}", + snap.recent_events.len() + ); +} diff --git a/crates/datastore/tests/dashboard_integration_test.rs b/crates/datastore/tests/dashboard_integration_test.rs new file mode 100644 index 0000000..2608f21 --- /dev/null +++ b/crates/datastore/tests/dashboard_integration_test.rs @@ -0,0 +1,205 @@ +#![cfg(feature = "node")] + +//! End-to-end integration test: spins up a datastore node with an HTTP API +//! **and** a runtime dashboard, performs CRUD over HTTP, then verifies: +//! +//! - `DatastoreMetrics::snapshot()` reflects the operations +//! - Dashboard `/datastore` serves HTML +//! - Dashboard `/api/datastore` returns a JSON snapshot matching the metrics + +use std::sync::Arc; +use std::thread; +use std::time::Duration; + +use swactor::config::RuntimeConfig; +use swactor::runtime::Runtime; + +use swactor_datastore::actors::{BlobStoreActor, DatastoreNode, MetadataActor}; +use swactor_datastore::api::start_api_server; +use swactor_datastore::metrics::DatastoreMetrics; +use swactor_datastore::storage::InMemoryBackend; +use swactor_datastore::DatastoreConfig; + +use distribution::types::NodeId; + +fn find_free_port() -> u16 { + std::net::TcpListener::bind("127.0.0.1:0") + .unwrap() + .local_addr() + .unwrap() + .port() +} + +/// Scenario: datastore CRUD → dashboard reflects live operation metrics. +/// +/// Story: +/// Two files are uploaded. One is fetched (metadata + data). The other +/// is deleted. Afterwards we check that the metrics snapshot, the +/// dashboard HTML page, and the dashboard JSON API all agree on what +/// happened. +#[test] +fn dashboard_reflects_datastore_operations() { + let api_port = find_free_port(); + let dash_port = find_free_port(); + let api_base = format!("http://127.0.0.1:{api_port}"); + let dash_base = format!("http://127.0.0.1:{dash_port}"); + + // ── Infrastructure: runtime + actors + dashboard + API ────────────── + + let dash = runtime_dashboard::start_dashboard(runtime_dashboard::DashboardConfig { + port: dash_port, + ..Default::default() + }); + + let collector = runtime_dashboard::collector::StatsCollector::new(2); + let mut rt = Runtime::new(RuntimeConfig { + num_threads: 2, + max_actors: 1024, + channel_buffer_size: 2000, + ..Default::default() + }); + rt.set_stats_hook(collector.clone()); + + let node_id = NodeId([0xBB; 32]); + let config = DatastoreConfig { + chunk_size: 1_048_576, + gc_interval: 1000, + ..Default::default() + }; + + let backend: Box = Box::new(InMemoryBackend::new()); + let blob_store_addr = rt.spawn(BlobStoreActor::new(backend)).unwrap(); + + let mut metadata = MetadataActor::new(node_id, &config); + metadata.set_blob_store(blob_store_addr); + let metadata_addr = rt.spawn(metadata).unwrap(); + + let datastore_node = DatastoreNode::new(node_id, blob_store_addr, metadata_addr, config); + let datastore_addr = rt.spawn(datastore_node).unwrap(); + + let handle = rt.run().expect("failed to start runtime"); + + let metrics = Arc::new(DatastoreMetrics::new()); + let node_hex: String = node_id.0.iter().map(|b| format!("{b:02x}")).collect(); + metrics.set_node_id(node_hex); + + dash.set_runtime(handle.runtime.clone(), collector); + dash.set_datastore( + Arc::clone(&metrics) + as Arc, + ); + + let (api_shutdown, _peers) = start_api_server( + handle.runtime.clone(), + datastore_addr, + metadata_addr, + blob_store_addr, + api_port, + Arc::clone(&metrics), + ); + + thread::sleep(Duration::from_millis(300)); + + let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + run_dashboard_scenario(&api_base, &dash_base, &metrics); + })); + + // Cleanup + api_shutdown.store(true, std::sync::atomic::Ordering::Relaxed); + dash.shutdown(); + handle.shutdown(); + handle.join(); + + if let Err(e) = result { + std::panic::resume_unwind(e); + } +} + +fn run_dashboard_scenario(api: &str, dash: &str, metrics: &Arc) { + // ── 1. Upload two objects ─────────────────────────────────────────── + + let put_a: serde_json::Value = ureq::post(&format!("{api}/api/put?name=alpha")) + .send_bytes(b"payload-alpha") + .expect("put A failed") + .into_json() + .unwrap(); + let hash_a = put_a["content_hash"].as_str().unwrap().to_string(); + + let _put_b: serde_json::Value = ureq::post(&format!("{api}/api/put?name=bravo")) + .send_bytes(b"payload-bravo") + .expect("put B failed") + .into_json() + .unwrap(); + let hash_b = _put_b["content_hash"].as_str().unwrap().to_string(); + + // ── 2. GET object A (metadata + raw data → 2 get ops) ────────────── + + let _: serde_json::Value = ureq::get(&format!("{api}/api/get?hash={hash_a}")) + .call() + .unwrap() + .into_json() + .unwrap(); + + let data_resp = ureq::get(&format!("{api}/api/data?hash={hash_a}")) + .call() + .unwrap(); + let mut body = Vec::new(); + data_resp.into_reader().read_to_end(&mut body).unwrap(); + assert_eq!(body, b"payload-alpha", "downloaded data should match"); + + // ── 3. DELETE object B ────────────────────────────────────────────── + + let _: serde_json::Value = ureq::post(&format!("{api}/api/delete?hash={hash_b}")) + .call() + .unwrap() + .into_json() + .unwrap(); + + // ── 4. Assert: in-process metrics snapshot ────────────────────────── + + let snap = metrics.snapshot(); + + assert_eq!(snap.put_ops, 2, "two puts recorded"); + assert_eq!(snap.get_ops, 2, "metadata-get + data-get recorded"); + assert_eq!(snap.delete_ops, 1, "one delete recorded"); + assert_eq!(snap.objects.len(), 1, "only alpha remains after deleting bravo"); + assert_eq!(snap.objects[0].hash, hash_a); + assert!( + snap.recent_events.len() >= 5, + "at least 5 events (2 put + 2 get + 1 delete), got {}", + snap.recent_events.len() + ); + + // ── 5. Assert: dashboard /datastore serves HTML ───────────────────── + + let page = ureq::get(&format!("{dash}/datastore")).call().unwrap(); + assert_eq!(page.status(), 200); + assert!( + page.header("Content-Type") + .unwrap_or("") + .contains("text/html"), + ); + let html = page.into_string().unwrap(); + assert!(html.contains("Datastore"), "page should mention Datastore"); + + // ── 6. Assert: /api/datastore JSON matches metrics ────────────────── + + let ds: serde_json::Value = ureq::get(&format!("{dash}/api/datastore")) + .call() + .unwrap() + .into_json() + .unwrap(); + + assert_eq!(ds["put_ops"].as_u64().unwrap(), 2); + assert_eq!(ds["get_ops"].as_u64().unwrap(), 2); + assert_eq!(ds["delete_ops"].as_u64().unwrap(), 1); + + let objects = ds["objects"].as_array().expect("objects should be array"); + assert_eq!(objects.len(), 1); + assert_eq!(objects[0]["hash"].as_str().unwrap(), hash_a); + + let events = ds["recent_events"] + .as_array() + .expect("recent_events should be array"); + assert!(events.len() >= 5); +} diff --git a/crates/distribution/src/kademlia/routing_table.rs b/crates/distribution/src/kademlia/routing_table.rs index fb11ee3..101f0c8 100644 --- a/crates/distribution/src/kademlia/routing_table.rs +++ b/crates/distribution/src/kademlia/routing_table.rs @@ -96,7 +96,6 @@ impl KBucket { pub struct RoutingTable { self_id: NodeId, buckets: Vec, - k: usize, } impl RoutingTable { @@ -109,7 +108,7 @@ impl RoutingTable { for _ in 0..NUM_BUCKETS { buckets.push(KBucket::new(k)); } - Self { self_id, buckets, k } + Self { self_id, buckets } } pub fn self_id(&self) -> NodeId { diff --git a/crates/distribution/src/transport.rs b/crates/distribution/src/transport.rs index 6cdfdaa..677cd4e 100644 --- a/crates/distribution/src/transport.rs +++ b/crates/distribution/src/transport.rs @@ -195,6 +195,7 @@ pub fn encode_wire_envelope(envelope: &WireEnvelope) -> Vec { enum ReadError { WouldBlock, Disconnected, + #[allow(dead_code)] Other(std::io::Error), } diff --git a/crates/distribution/tests/common/mod.rs b/crates/distribution/tests/common/mod.rs index 89d2c63..51c58ae 100644 --- a/crates/distribution/tests/common/mod.rs +++ b/crates/distribution/tests/common/mod.rs @@ -3,6 +3,7 @@ //! `TestCluster` makes sender-misattribution structurally impossible by //! tagging every response with the responder's index, mirroring the //! simulation crate's `deliver_actions_tagged_with_net`. +#![allow(dead_code)] use distribution::node::{DistributedNode, DistributedNodeConfig}; use distribution::registry::RegistryConfig; diff --git a/crates/distribution/tests/kademlia_directory.rs b/crates/distribution/tests/kademlia_directory.rs index 5c7cb91..3c00a0f 100644 --- a/crates/distribution/tests/kademlia_directory.rs +++ b/crates/distribution/tests/kademlia_directory.rs @@ -3,7 +3,7 @@ use distribution::crypto::Keypair; use distribution::kademlia::directory::{ actor_addr_as_node_id, resolve_quorum, DirectoryShard, QuorumResult, }; -use distribution::types::{DirectoryEntry, NodeId, Signature}; +use distribution::types::{NodeId, Signature}; // ─── DirectoryShard ───────────────────────────────────────────────────────── diff --git a/crates/runtime-dashboard/src/actor_detail_html.rs b/crates/runtime-dashboard/src/actor_detail_html.rs index 788a54e..c5c2c27 100644 --- a/crates/runtime-dashboard/src/actor_detail_html.rs +++ b/crates/runtime-dashboard/src/actor_detail_html.rs @@ -123,6 +123,8 @@ pub const ACTOR_DETAIL_HTML: &str = r##" diff --git a/crates/runtime-dashboard/src/actors_html.rs b/crates/runtime-dashboard/src/actors_html.rs index 7b0f011..2c139fb 100644 --- a/crates/runtime-dashboard/src/actors_html.rs +++ b/crates/runtime-dashboard/src/actors_html.rs @@ -152,6 +152,7 @@ pub const ACTORS_HTML: &str = r##" Overview Actors Distribution + Datastore
diff --git a/crates/runtime-dashboard/src/dashboard_html.rs b/crates/runtime-dashboard/src/dashboard_html.rs index 1917297..0054723 100644 --- a/crates/runtime-dashboard/src/dashboard_html.rs +++ b/crates/runtime-dashboard/src/dashboard_html.rs @@ -153,6 +153,7 @@ pub const DASHBOARD_HTML: &str = r##" Overview Actors Distribution + Datastore
diff --git a/crates/runtime-dashboard/src/datastore_collector.rs b/crates/runtime-dashboard/src/datastore_collector.rs new file mode 100644 index 0000000..47ec74d --- /dev/null +++ b/crates/runtime-dashboard/src/datastore_collector.rs @@ -0,0 +1,16 @@ +//! Datastore stats provider for the runtime dashboard. +//! +//! The trait returns a pre-serialized JSON string so that `runtime-dashboard` +//! has no compile-time dependency on `swactor-datastore` (which would create a +//! circular dependency since `swactor-datastore[node]` depends on us). +//! +//! The `swactor-datastore` crate implements this trait in its `node` feature. + +/// Trait for providing datastore stats to the dashboard. +/// +/// Implementations capture a point-in-time snapshot as serialized JSON. +/// The dashboard polls this every ~200ms via SSE. +pub trait DatastoreStatsProvider: Send + Sync { + /// Return a JSON-serialized datastore snapshot, or `None` if unavailable. + fn snapshot_json(&self) -> Option; +} diff --git a/crates/runtime-dashboard/src/datastore_html.rs b/crates/runtime-dashboard/src/datastore_html.rs new file mode 100644 index 0000000..3afd976 --- /dev/null +++ b/crates/runtime-dashboard/src/datastore_html.rs @@ -0,0 +1,286 @@ +pub const DATASTORE_HTML: &str = r##" + + + + +Swactor Runtime – Datastore + + + +
+
+

+ Swactor Runtime Dashboard + +

+ +
+
+ Waiting for data... +
+
+ +
+ +
+

Datastore Stats

+
+
0
Objects
+
0
Total Size
+
0
Puts
+
0
Gets
+
0
Deletes
+
+
+ + +
+

Event Timeline

+
+
+ + +
+

Objects

+
+ + + +
HashNameSize
+
+
+ + +
+

Active Transfers

+
+
+
+ + + + +"##; diff --git a/crates/runtime-dashboard/src/distribution_html.rs b/crates/runtime-dashboard/src/distribution_html.rs index d9931e3..ae9af8d 100644 --- a/crates/runtime-dashboard/src/distribution_html.rs +++ b/crates/runtime-dashboard/src/distribution_html.rs @@ -121,6 +121,7 @@ pub const DISTRIBUTION_HTML: &str = r##" Overview Actors Distribution + Datastore
diff --git a/crates/runtime-dashboard/src/lib.rs b/crates/runtime-dashboard/src/lib.rs index 08cbd73..1db0871 100644 --- a/crates/runtime-dashboard/src/lib.rs +++ b/crates/runtime-dashboard/src/lib.rs @@ -20,6 +20,9 @@ mod distribution_html; #[cfg(feature = "distribution")] pub mod distribution_collector; +mod datastore_html; +pub mod datastore_collector; + use std::io; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Mutex}; @@ -91,6 +94,7 @@ pub struct DashboardHandle { recording: bool, #[cfg(feature = "distribution")] distribution: Arc>>>, + datastore: Arc>>>, } impl DashboardHandle { @@ -123,6 +127,11 @@ impl DashboardHandle { *self.distribution.lock().unwrap() = Some(provider); } + /// Attach a datastore stats provider, enabling the `/datastore` page. + pub fn set_datastore(&self, provider: Arc) { + *self.datastore.lock().unwrap() = Some(provider); + } + /// Access the time-series history store (for TUI sparklines, etc.). pub fn history(&self) -> &Arc { &self.history @@ -178,6 +187,9 @@ pub fn start_dashboard(config: DashboardConfig) -> DashboardHandle { let distribution: Arc>>> = Arc::new(Mutex::new(None)); + let datastore: Arc>>> = + Arc::new(Mutex::new(None)); + server::spawn_http_server( Arc::clone(&store), Arc::clone(&runtime), @@ -187,6 +199,7 @@ pub fn start_dashboard(config: DashboardConfig) -> DashboardHandle { config.port, #[cfg(feature = "distribution")] Arc::clone(&distribution), + Arc::clone(&datastore), ); // Start stats recorder thread when recording is enabled @@ -229,6 +242,7 @@ pub fn start_dashboard(config: DashboardConfig) -> DashboardHandle { recording: config.record, #[cfg(feature = "distribution")] distribution, + datastore, } } diff --git a/crates/runtime-dashboard/src/server.rs b/crates/runtime-dashboard/src/server.rs index ee9bba5..ee95b19 100644 --- a/crates/runtime-dashboard/src/server.rs +++ b/crates/runtime-dashboard/src/server.rs @@ -24,6 +24,9 @@ use crate::distribution_collector::DistributionStatsProvider; #[cfg(feature = "distribution")] use crate::distribution_html::DISTRIBUTION_HTML; +use crate::datastore_collector::DatastoreStatsProvider; +use crate::datastore_html::DATASTORE_HTML; + /// Format a server-sent event. fn format_sse(event: &str, data: &str) -> Vec { format!("event: {event}\ndata: {data}\n\n").into_bytes() @@ -128,6 +131,7 @@ pub(crate) fn spawn_http_server( port: u16, #[cfg(feature = "distribution")] distribution: Arc>>>, + datastore: Arc>>>, ) { let addr = format!("0.0.0.0:{port}"); let server = tiny_http::Server::http(&addr).expect("failed to bind HTTP server"); @@ -144,6 +148,7 @@ pub(crate) fn spawn_http_server( let cmd_router = Arc::clone(&cmd_router); #[cfg(feature = "distribution")] let distribution = Arc::clone(&distribution); + let datastore = Arc::clone(&datastore); thread::spawn(move || { loop { let request = match server.recv() { @@ -159,6 +164,7 @@ pub(crate) fn spawn_http_server( "/topology" => respond_html(request, TOPOLOGY_HTML, "live"), #[cfg(feature = "distribution")] "/distribution" => respond_html(request, DISTRIBUTION_HTML, "live"), + "/datastore" => respond_html(request, DATASTORE_HTML, "live"), "/events" => { handle_live_sse( request, @@ -169,6 +175,7 @@ pub(crate) fn spawn_http_server( Arc::clone(&history), #[cfg(feature = "distribution")] Arc::clone(&distribution), + Arc::clone(&datastore), ); } "/api/stats" => { @@ -204,6 +211,12 @@ pub(crate) fn spawn_http_server( Arc::clone(&distribution), ); } + "/api/datastore" => { + handle_datastore_api( + request, + Arc::clone(&datastore), + ); + } "/api/logs" => { handle_logs_api(request, &url, Arc::clone(&store)); } @@ -239,6 +252,7 @@ fn handle_live_sse( history: Arc, #[cfg(feature = "distribution")] distribution: Arc>>>, + datastore: Arc>>>, ) { let (tx, rx) = mpsc::channel::>(); let response = make_sse_response(rx); @@ -309,6 +323,18 @@ fn handle_live_sse( } } + // Send datastore snapshot if provider is attached + { + let maybe_ds = datastore.lock().unwrap().clone(); + if let Some(provider) = maybe_ds { + if let Some(json) = provider.snapshot_json() { + if tx.send(format_sse("datastore", &json)).is_err() { + return; + } + } + } + } + // Send new activity events let (batch, new_cursor) = store.read_from(cursor); if !batch.is_empty() { @@ -423,6 +449,26 @@ fn handle_distribution_api( let _ = request.respond(response); } +fn handle_datastore_api( + request: tiny_http::Request, + datastore: Arc>>>, +) { + let json = match datastore.lock().unwrap().as_ref() { + Some(provider) => provider.snapshot_json().unwrap_or_else(|| "{}".into()), + None => serde_json::json!({ + "error": "datastore provider not attached" + }) + .to_string(), + }; + + let response = tiny_http::Response::from_string(json).with_header( + "Content-Type: application/json" + .parse::() + .unwrap(), + ); + let _ = request.respond(response); +} + fn handle_topology_api( request: tiny_http::Request, runtime: Arc>>>, diff --git a/crates/runtime-dashboard/src/topology_html.rs b/crates/runtime-dashboard/src/topology_html.rs index a52ea56..123937b 100644 --- a/crates/runtime-dashboard/src/topology_html.rs +++ b/crates/runtime-dashboard/src/topology_html.rs @@ -47,6 +47,7 @@ pub const TOPOLOGY_HTML: &str = r##" Actors Topology Distribution + Datastore
diff --git a/crates/runtime-dashboard/tests/command_api.rs b/crates/runtime-dashboard/tests/command_api.rs index 620c266..3f68142 100644 --- a/crates/runtime-dashboard/tests/command_api.rs +++ b/crates/runtime-dashboard/tests/command_api.rs @@ -8,7 +8,7 @@ use std::sync::Arc; use swactor::actor::ActorInterface; use swactor::runtime::{Ctx, Runtime, RuntimeConfig}; use runtime_dashboard::command::{ - from_query_params, parse_line, CommandContext, CommandRequest, CommandResponse, CommandRouter, + from_query_params, parse_line, CommandContext, CommandResponse, CommandRouter, }; // ── Test Helpers ───────────────────────────────────────────────────────────── diff --git a/docs/datastore/actors.md b/docs/datastore/actors.md new file mode 100644 index 0000000..f9b9ebe --- /dev/null +++ b/docs/datastore/actors.md @@ -0,0 +1,278 @@ +# Datastore Actor Reference + +## Overview + +The datastore is built from four actors within the swactor runtime. `DatastoreNode` is the public facade — all external requests (HTTP API, network protocol) enter through it and are routed to two long-lived worker actors: `BlobStoreActor` (content-addressed chunk/manifest I/O) and `MetadataActor` (object index, DHT replication, GC). A fourth actor, `TransferActor`, is spawned ephemerally for each remote download and self-terminates on completion or failure. + +``` + ┌─────────────────────────┐ + │ store_node (main) │ + │ spawns all 3 long-lived │ + │ actors, drives ticks │ + └────┬──────┬──────┬───────┘ + │ │ │ + spawn │ │ │ spawn + ┌──────────────┘ │ └──────────────┐ + ▼ │ spawn ▼ + ┌───────────────────┐ │ ┌───────────────────┐ + │ BlobStoreActor │ │ │ MetadataActor │ + │ (chunks, manifests)│ │ │ (index, DHT, GC) │ + └─────────▲─────────┘ │ └──▲────────┬───────┘ + │ ▼ │ │ + │ ┌───────────────────┐ │ │ + │ │ DatastoreNode │────┘ │ + │ │ (facade/router) │─────────────┘ + └────────────│ │ + └────────┬──────────┘ + │ spawns (per download) + ▼ + ┌───────────────────┐ + │ TransferActor │ + │ (ephemeral) │ + └───────────────────┘ + + Arrows: ──▶ sends messages to +``` + +## Actors + +### DatastoreNode + +| | | +|---|---| +| **Role** | Top-level coordinator/facade. Accepts user-facing commands and incoming network protocol messages, delegates all work to `BlobStoreActor` and `MetadataActor`. | +| **Source** | `crates/datastore/src/actors/datastore_node.rs` | +| **Spawned by** | `store_node` binary (`crates/datastore/src/bin/store_node.rs:188`) | +| **Lifecycle** | Long-lived — runs for the lifetime of the process | + +**Inbound messages** (`DatastoreNodeMsg` — 11 variants): + +User-facing commands: +- `Put { data, name, tags, reply_to }` — chunk a blob, write chunks + manifest to `BlobStoreActor`, register in `MetadataActor` +- `Get { content_hash, reply_to }` — retrieve object metadata + manifest via `MetadataActor` +- `Delete { content_hash, reply_to }` — remove object via `MetadataActor` +- `List { name_filter, all, reply_to }` — list objects (local or swarm-wide) via `MetadataActor` +- `Status { reply_to }` — return this node's `NodeId` +- `ReadChunk { hash, reply_to }` — read a single chunk via `BlobStoreActor` + +Protocol routing (incoming network messages): +- `IncomingGetChunk` — forwards to `BlobStoreActor::ReadChunk` +- `IncomingGetManifest` — forwards to `BlobStoreActor::ReadManifest` +- `IncomingStoreObject` — forwards to `MetadataActor::HandleStoreObject` +- `IncomingFindObject` — forwards to `MetadataActor::HandleFindObject` +- `IncomingListObjects` — forwards to `MetadataActor::ListLocal` + +**Key outbound messages:** +- `BlobStoreMsg::WriteChunk`, `WriteManifest`, `ReadChunk`, `ReadManifest` — to `BlobStoreActor` +- `MetadataMsg::PutObject`, `GetObject`, `DeleteObject`, `ListLocal`, `ListSwarm`, `HandleStoreObject`, `HandleFindObject` — to `MetadataActor` +- `DatastoreResponse::NodeStatus` — directly to caller for `Status` + +--- + +### BlobStoreActor + +| | | +|---|---| +| **Role** | Content-addressed storage for chunks and manifests. All I/O goes through a pluggable `StorageBackend` (filesystem or in-memory). | +| **Source** | `crates/datastore/src/actors/blob_store.rs` | +| **Spawned by** | `store_node` binary (`store_node.rs:178`) | +| **Lifecycle** | Long-lived — runs for the lifetime of the process | + +**Inbound messages** (`BlobStoreMsg` — 8 variants): + +Chunk operations: +- `WriteChunk { hash, data, reply_to }` — persist a chunk, reply `ChunkStored` +- `ReadChunk { hash, reply_to }` — read a chunk, reply `ChunkOk` or `NotFound` +- `DeleteChunk { hash }` — remove a chunk (fire-and-forget) +- `HasChunk { hash, reply_to }` — existence check, reply `Bool` +- `ListChunks { reply_to }` — list all chunk hashes, reply `ChunkList` +- `GcUnreferenced { referenced }` — delete chunks not in the referenced set (fire-and-forget) + +Manifest operations: +- `WriteManifest { manifest, reply_to }` — persist a manifest, reply `ManifestStored` +- `ReadManifest { hash, reply_to }` — read a manifest, reply `ManifestOk` or `NotFound` + +**Key outbound messages:** +- `DatastoreResponse` variants (`ChunkStored`, `ChunkOk`, `ManifestStored`, `ManifestOk`, `NotFound`, `Error`, `Bool`, `ChunkList`) — always back to the `reply_to` address + +--- + +### MetadataActor + +| | | +|---|---| +| **Role** | Object metadata index. Maintains a `HashMap` and a manifest cache. Handles DHT-style find/store operations, epidemic dissemination of entries to peers, and periodic garbage collection. | +| **Source** | `crates/datastore/src/actors/metadata.rs` | +| **Spawned by** | `store_node` binary (`store_node.rs:185`) | +| **Lifecycle** | Long-lived — runs for the lifetime of the process | + +**Inbound messages** (`MetadataMsg` — 11 variants): + +Object operations: +- `PutObject { entry, manifest, reply_to }` — store metadata + manifest locally, enqueue for dissemination, reply `PutOk` +- `GetObject { content_hash, reply_to }` — local lookup, reply `GetOk` or `NotFound` +- `DeleteObject { content_hash, reply_to }` — remove from local index, reply `DeleteOk` or `NotFound` +- `ListLocal { name_filter, reply_to }` — list local entries with optional name filter, reply `ListOk` +- `ListSwarm { name_filter, reply_to }` — swarm-wide list (currently delegates to `ListLocal`) + +DHT protocol: +- `HandleFindObject { from, content_hash, reply_to }` — answer an incoming FIND_VALUE from a peer +- `HandleStoreObject { entry, manifest }` — accept an incoming STORE from a peer (fire-and-forget) + +Peer management: +- `SetPeers { peers }` — update the list of peer `MetadataActor` addresses for dissemination + +Periodic ticks (driven by the `store_node` main loop): +- `DisseminateTick` — send pending entries to all known peers +- `GcTick` — collect referenced chunks from all manifests, send `BlobStoreMsg::GcUnreferenced` to `BlobStoreActor` + +**Key outbound messages:** +- `DatastoreResponse` variants (`PutOk`, `GetOk`, `DeleteOk`, `ListOk`, `NotFound`, `Error`) — to caller +- `MetadataMsg::HandleStoreObject` — to peer `MetadataActor` addresses during dissemination +- `BlobStoreMsg::GcUnreferenced` — to local `BlobStoreActor` during GC + +--- + +### TransferActor + +| | | +|---|---| +| **Role** | Manages a single object download from a remote node. Tracks pending/received chunks, forwards received data to the local `BlobStoreActor`, and reports completion or failure to the original requester. | +| **Source** | `crates/datastore/src/actors/transfer.rs` | +| **Spawned by** | `DatastoreNode` (one per remote download) | +| **Lifecycle** | Ephemeral — self-terminates via `ctx.stop_self()` on completion, failure, or cancel | + +**Inbound messages** (`TransferMsg` — 4 variants): + +- `StartDownload { manifest, source_node, reply_to }` — initialize the download with a manifest and source +- `ChunkReceived { hash, data }` — a chunk arrived from the remote node +- `ChunkFailed { hash, reason }` — a chunk fetch failed (retries up to `max_retries`, then fails the whole transfer) +- `Cancel` — abort the transfer immediately + +**Key outbound messages:** +- `BlobStoreMsg::WriteChunk` — to local `BlobStoreActor` for each received chunk +- `DatastoreResponse::TransferComplete` — to `reply_to` when all chunks received +- `DatastoreResponse::TransferFailed` — to `reply_to` when retries are exhausted + +--- + +## Message Reference + +All message types are defined in `crates/datastore/src/messages.rs`. + +### Intra-node actor messages + +| Enum | Variants | Handled by | +|------|----------|------------| +| `DatastoreNodeMsg` | 11 (6 user-facing + 5 protocol routing) | `DatastoreNode` | +| `BlobStoreMsg` | 8 (5 chunk ops + 1 GC + 2 manifest ops) | `BlobStoreActor` | +| `MetadataMsg` | 11 (5 object ops + 2 DHT + 1 peer mgmt + 2 ticks) | `MetadataActor` | +| `TransferMsg` | 4 (start + chunk received + chunk failed + cancel) | `TransferActor` | + +### Shared response enum + +`DatastoreResponse` — 15 variants used as the return type for all four actors: + +| Variant | Meaning | +|---------|---------| +| `PutOk { content_hash }` | Object stored successfully | +| `GetOk { entry, manifest }` | Object found | +| `DeleteOk { content_hash }` | Object deleted | +| `ListOk { entries }` | List result | +| `ChunkOk { hash, data }` | Chunk data retrieved | +| `ChunkStored { hash }` | Chunk written to storage | +| `ManifestStored { hash }` | Manifest written to storage | +| `ManifestOk { manifest }` | Manifest retrieved | +| `TransferComplete { content_hash }` | All chunks downloaded | +| `TransferFailed { reason }` | Transfer failed | +| `NodeStatus { node_id }` | Node identity | +| `NotFound` | Resource not found | +| `Error { reason }` | Generic error | +| `Bool(bool)` | Boolean result (e.g. `HasChunk`) | +| `ChunkList { hashes }` | List of chunk hashes | + +### Inter-node wire messages (NetworkMessage) + +| Struct | Direction | Purpose | +|--------|-----------|---------| +| `GetChunkRequest` | requester → holder | Fetch a chunk by hash | +| `GetChunkResponse` | holder → requester | Return chunk data (or `None`) | +| `StoreObjectRequest` | origin → DHT peer | Kademlia STORE for object metadata | +| `FindObjectRequest` | requester → DHT peer | Kademlia FIND_VALUE for object metadata | +| `FindObjectResponse` | DHT peer → requester | Return `Found(entry)` or `Closer(nodes)` | +| `GetManifestRequest` | requester → holder | Fetch a manifest by content hash | +| `GetManifestResponse` | holder → requester | Return manifest (or `None`) | +| `ListObjectsRequest` | requester → peer | List objects with optional name filter | +| `ListObjectsResponse` | peer → requester | Return matching entries | + +Wire messages are distinguished from intra-node messages by implementing the `NetworkMessage` trait with a stable `type_tag()` string. They are serialized with serde for transport over iroh/QUIC. + +--- + +## Key Flows + +### Put (store a blob) + +``` +Client → DatastoreNode::Put + → chunk_blob() splits data into chunks + → BlobStoreActor::WriteChunk (for each chunk, fire-and-forget) + → BlobStoreActor::WriteManifest + → MetadataActor::PutObject + → stores entry + manifest locally + → enqueues for dissemination + → replies DatastoreResponse::PutOk +``` + +### Get (retrieve metadata) + +``` +Client → DatastoreNode::Get + → MetadataActor::GetObject + → local index lookup + → replies DatastoreResponse::GetOk (or NotFound) +``` + +### Data (reassemble from chunks) + +See [streaming.md](streaming.md) for the full transfer protocol. In summary: + +``` +API server → DatastoreNode::Get → MetadataActor (local miss) + → iterate peers: + → FindObjectRequest (wire) → peer MetadataActor + → GetManifestRequest (wire) → peer BlobStoreActor + → GetChunkRequest (wire) → peer BlobStoreActor (per chunk) + → BlobStoreActor::WriteChunk (store locally) + → BlobStoreActor::WriteManifest + → MetadataActor::PutObject + → reassemble_blob() → verify blake3 → respond +``` + +### Dissemination (epidemic replication) + +``` +store_node main loop (every disseminate_interval ticks) + → MetadataActor::DisseminateTick + → take_pending() selects entries with remaining budget + → for each peer: MetadataActor::HandleStoreObject + → peer inserts if absent, re-enqueues for further dissemination +``` + +Budget per entry = `Λ * ceil(log2(cluster_size))` (SWIM-style, Λ=3). + +### GC (garbage collection) + +``` +store_node main loop (every gc_interval ticks) + → MetadataActor::GcTick + → scans all manifests → builds referenced chunk set + → BlobStoreActor::GcUnreferenced { referenced } + → deletes any chunk not in the referenced set +``` + +--- + +## Related + +- [streaming.md](streaming.md) — chunking, transfer protocol, reassembly, and progress tracking diff --git a/docs/datastore/streaming.md b/docs/datastore/streaming.md new file mode 100644 index 0000000..153876a --- /dev/null +++ b/docs/datastore/streaming.md @@ -0,0 +1,103 @@ +# Datastore Streaming Architecture + +## Overview + +"Streaming" in the swactor datastore refers to **progressive chunk-based transfer**, not byte-level streaming. When an object is stored, it is split into fixed-size chunks, each content-addressed with blake3. When retrieved from a remote peer, chunks are fetched individually and reassembled — enabling progress tracking and partial recovery. + +This design trades a small amount of per-chunk overhead for: +- **Progress visibility**: the dashboard shows `chunks_received / chunks_total` in real time +- **Resumability**: a failed transfer can (in principle) restart from the last chunk +- **Deduplication**: identical chunks across objects are stored once + +## Content-Addressed Chunking + +The `chunk_blob()` function (`chunking.rs`) splits raw bytes into fixed-size pieces: + +1. Compute `ContentHash = blake3(entire_blob)` — this is the object's identity +2. Split the blob into `ceil(total_size / chunk_size)` pieces (default chunk size: 1 MB) +3. For each piece, compute `chunk_hash = blake3(piece_bytes)` +4. Build a `ChunkRef { hash, offset, size }` for each piece +5. Return an `ObjectManifest` containing the full list of `ChunkRef`s + +``` +Blob (5.2 MB, chunk_size=1MB) +├── Chunk 0: hash=abc1…, offset=0, size=1048576 +├── Chunk 1: hash=def2…, offset=1048576, size=1048576 +├── Chunk 2: hash=789a…, offset=2097152, size=1048576 +├── Chunk 3: hash=bcd3…, offset=3145728, size=1048576 +└── Chunk 4: hash=ef45…, offset=4194304, size=1048576 (last: 209920 bytes) +``` + +The object's identity (`ContentHash`) is the hash of the *entire* blob, not of the manifest. This means the same data always produces the same hash regardless of chunk size. + +## Transfer Protocol + +When a client requests an object via `GET /api/data?hash=...`, the API server: + +1. Sends a `DatastoreNodeMsg::Get` to the local `DatastoreNode` actor +2. If found locally, reads all chunks from the local `BlobStore` and reassembles +3. If **not found locally**, enters `try_remote_get()`: + +![Transfer Flow](../diagrams/datastore_transfer.svg) + +### Remote GET step-by-step + +1. **Iterate peers**: for each known peer node: +2. **FindObject**: send `MetadataMsg::HandleFindObject` to the peer's `MetadataActor` +3. **Read manifest**: send `BlobStoreMsg::ReadManifest` to the peer's `BlobStore` +4. **Fetch chunks**: for each `ChunkRef` in the manifest: + - Send `BlobStoreMsg::ReadChunk` to the peer + - Receive `DatastoreResponse::ChunkOk { hash, data }` + - Store locally via `BlobStoreMsg::WriteChunk` + - Update `DatastoreMetrics::advance_transfer()` for dashboard progress +5. **Store manifest locally**: `BlobStoreMsg::WriteManifest` +6. **Store metadata locally**: `MetadataMsg::PutObject` +7. **Reassemble and respond**: `reassemble_blob()` concatenates chunks and verifies integrity + +If a peer doesn't have the object (or any step fails), the loop continues to the next peer. + +## Reassembly + +`reassemble_blob()` (`chunking.rs`) takes a manifest and a set of `(hash, data)` pairs: + +1. For each `ChunkRef` in manifest order, find the matching `(hash, data)` pair +2. Concatenate all chunk data into a single buffer +3. Compute `blake3(result)` and verify it matches `manifest.content_hash` +4. Return the reassembled blob (or a `ChunkingError` on mismatch) + +This integrity check ensures that even if individual chunks are corrupted or swapped, the final result is always verified against the original content hash. + +## Progress Tracking + +The `DatastoreMetrics` struct provides thread-safe transfer tracking: + +``` +begin_transfer(hash, chunks_total) // called when remote GET starts +advance_transfer(hash) // called after each chunk is stored locally +end_transfer(hash) // called on completion or failure +``` + +The dashboard SSE stream includes a `datastore` event every ~200ms with a `DatastoreSnapshot` containing `active_transfers: Vec`. The web UI renders these as animated progress bars. + +``` +TransferProgress { + hash: "abc123...", + chunks_received: 3, + chunks_total: 5, +} +``` + +## GC Integration + +When an object is deleted, its `ObjectEntry` and `ObjectManifest` are removed from the `MetadataActor`. However, the underlying chunks are **not immediately deleted** — they may be referenced by other manifests (deduplication). + +Instead, garbage collection runs periodically: + +1. `MetadataMsg::GcTick` triggers a scan +2. The `MetadataActor` collects all chunk hashes referenced by any live manifest +3. Sends `BlobStoreMsg::GcUnreferenced` with the referenced set +4. The `BlobStore` deletes any chunks **not** in the referenced set + +This two-phase approach prevents data loss when chunks are shared between objects. + +![Chunk Lifecycle](../diagrams/datastore_chunk_lifecycle.svg) diff --git a/docs/diagrams/datastore_chunk_lifecycle.svg b/docs/diagrams/datastore_chunk_lifecycle.svg new file mode 100644 index 0000000..f0dffdc --- /dev/null +++ b/docs/diagrams/datastore_chunk_lifecycle.svg @@ -0,0 +1,170 @@ + + + + + + + + + + + + + + + + + + + + + Chunk Lifecycle + chunking.rs · storage.rs · actors/blob_store.rs + + + + + + + + Blob Data + raw bytes + + + + + + + blake3 hash + ContentHash + + + + + + + chunk_blob() + split by chunk_size + + + + + + + + + ChunkRef[0] + + + ChunkRef[1] + + + ChunkRef[N] + + + + + + + + + ObjectManifest + chunks + total_size + + + + + + Storage + + + + BlobStoreMsg + ::WriteChunk { hash, data } + + + + + + StorageBackend + ::write_chunk(hash, data) + + + + + + Filesystem / + InMemoryBackend + + + + + + Reference Tracking + + + ObjectManifest + references chunk hashes + + + owns + + + MetadataActor + + tracks all manifests + + + + + + Garbage Collection + + + GcTick + periodic trigger + + + + + Collect referenced + hashes from manifests + + + + + GcUnreferenced + delete orphan chunks + + + + + + Deleted + + + + + Legend + + + Transition + + + Data flow + + + Deletion + + + Actor / Component + + + Data type + + + Success state + diff --git a/docs/diagrams/datastore_transfer.svg b/docs/diagrams/datastore_transfer.svg new file mode 100644 index 0000000..dd09007 --- /dev/null +++ b/docs/diagrams/datastore_transfer.svg @@ -0,0 +1,153 @@ + + + + + + + + + + + + + + + + + + + + + + + + Datastore Remote GET Transfer Flow + api.rs:583-706 · types.rs + + + + Client + + + + GET /api/data + + + + API Server + + + + Get msg + + + + DatastoreNode + + + + lookup + + + + MetadataActor + (local) + + + + Not Found + + + + try_remote_get() + iterate peers + + + + FindObject + + + + Peer MetadataActor + (remote node) + + + + Found! + + + + Peer BlobStore + ReadChunk × N + + + + + + + + + Chunk 1 + + + Chunk 2 + + + Chunk N + + + + + + + + + Local BlobStore + WriteChunk × N + + + + + + + reassemble_blob + verify integrity + + + + 200 OK + blob data + + + + DatastoreMetrics + begin/advance/end + + + transfer progress + + + + + Legend + + + Transition + + + Success path + + + Not found / fallback + + + Chunk transfer + + + Metrics tracking + diff --git a/src/runtime.rs b/src/runtime.rs index ce04231..36f6bac 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -265,11 +265,6 @@ impl Runtime { /// Creates a temporary inbox, calls `msg_builder` with the inbox's address /// (so you can embed it as `reply_to`), sends the message, and returns an /// [`Ask`] handle for receiving the response. - /// - /// ```ignore - /// let ask = rt.ask(actor, |reply_to| GetValue { reply_to })?; - /// let value = ask.recv_ticking(&rt, 10)?; - /// ``` pub fn ask( &self, addr: ActorAddress, diff --git a/xtask/Cargo.toml b/xtask/Cargo.toml new file mode 100644 index 0000000..37518d7 --- /dev/null +++ b/xtask/Cargo.toml @@ -0,0 +1,4 @@ +[package] +name = "xtask" +version = "0.1.0" +edition = "2024" diff --git a/xtask/src/main.rs b/xtask/src/main.rs new file mode 100644 index 0000000..fb07d80 --- /dev/null +++ b/xtask/src/main.rs @@ -0,0 +1,197 @@ +use std::process::Command; +use std::time::Instant; + +struct TestStep { + label: &'static str, + args: &'static [&'static str], +} + +struct Group { + name: &'static str, + description: &'static str, + steps: &'static [TestStep], +} + +const CORE: Group = Group { + name: "core", + description: "Actor runtime, message delivery, property tests", + steps: &[TestStep { + label: "actor runtime", + args: &["test", "-p", "swactor", "--features", "transport"], + }], +}; + +const DISTRIBUTION: Group = Group { + name: "distribution", + description: "Distribution protocol + datastore", + steps: &[ + TestStep { + label: "distribution protocol", + args: &["test", "-p", "distribution"], + }, + TestStep { + label: "datastore", + args: &["test", "-p", "swactor-datastore"], + }, + ], +}; + +const CLUSTER_SIMS: Group = Group { + name: "cluster-sims", + description: "Deterministic cluster simulations", + steps: &[TestStep { + label: "cluster simulations", + args: &["test", "-p", "simulation"], + }], +}; + +const INTEGRATED: Group = Group { + name: "integrated", + description: "HTTP API + dashboard end-to-end tests", + steps: &[ + TestStep { + label: "datastore integration (node features)", + args: &[ + "test", + "-p", + "swactor-datastore", + "--features", + "node", + "--test", + "api_integration_test", + "--test", + "dashboard_integration_test", + ], + }, + TestStep { + label: "runtime dashboard", + args: &["test", "-p", "runtime-dashboard"], + }, + ], +}; + +fn groups_for(name: &str) -> Option> { + match name { + "core" => Some(vec![&CORE]), + "distribution" => Some(vec![&DISTRIBUTION]), + "cluster-sims" => Some(vec![&CLUSTER_SIMS]), + "integrated" => Some(vec![&INTEGRATED]), + "essential" => Some(vec![&CORE, &DISTRIBUTION, &INTEGRATED]), + "all" => Some(vec![&CORE, &DISTRIBUTION, &CLUSTER_SIMS, &INTEGRATED]), + _ => None, + } +} + +fn run_step(group_name: &str, step: &TestStep) -> bool { + println!("\n=== {group_name}: {} ===", step.label); + println!(" cargo {}", step.args.join(" ")); + println!(); + + let status = Command::new("cargo") + .args(step.args) + .status(); + + match status { + Ok(s) => s.success(), + Err(e) => { + eprintln!("Failed to execute cargo: {e}"); + false + } + } +} + +fn print_usage() { + println!( + "\ +USAGE: cargo xtask test + +GROUPS: + core Actor runtime, message delivery, property tests + distribution Distribution protocol + datastore + cluster-sims Deterministic cluster simulations + integrated HTTP API + dashboard end-to-end tests + essential core + distribution + integrated (merge gate) + all Every test group + +FLAGS: + --list Show all groups and the cargo commands they run" + ); +} + +fn print_list() { + let all_groups: &[(&[&str], &Group)] = &[ + (&[], &CORE), + (&[], &DISTRIBUTION), + (&[], &CLUSTER_SIMS), + (&[], &INTEGRATED), + ]; + + println!("Available test groups:\n"); + + for &(_, group) in all_groups { + println!(" {:<14}{}", group.name, group.description); + for step in group.steps { + println!(" → cargo {}", step.args.join(" ")); + } + println!(); + } + + println!(" {:<14}core + distribution + integrated (merge gate)", "essential"); + println!(" {:<14}Every test group", "all"); +} + +fn main() { + let args: Vec = std::env::args().collect(); + + if args.len() < 2 || args[1] != "test" { + print_usage(); + std::process::exit(if args.len() < 2 { 1 } else { 1 }); + } + + if args.len() < 3 { + print_usage(); + std::process::exit(1); + } + + let target = &args[2]; + + if target == "--list" { + print_list(); + return; + } + + let groups = match groups_for(target) { + Some(g) => g, + None => { + eprintln!("Unknown test group: {target}\n"); + print_usage(); + std::process::exit(1); + } + }; + + let start = Instant::now(); + let mut passed = 0usize; + let mut failed = 0usize; + + for group in &groups { + for step in group.steps { + if run_step(group.name, step) { + passed += 1; + } else { + failed += 1; + let elapsed = start.elapsed(); + println!( + "\n--- FAILED after {:.1}s ({passed} passed, {failed} failed) ---", + elapsed.as_secs_f64() + ); + std::process::exit(1); + } + } + } + + let elapsed = start.elapsed(); + println!( + "\n--- All {passed} step(s) passed in {:.1}s ---", + elapsed.as_secs_f64() + ); +}