From ebf778fc3fad910e8a3eef529e6b2a9c4050b505 Mon Sep 17 00:00:00 2001 From: Zachery Aaron Shores-Chmielewski Date: Sun, 15 Feb 2026 20:43:10 +0700 Subject: [PATCH] fix: cli for datastore works --- Cargo.lock | 72 +- crates/datastore/Cargo.toml | 19 + crates/datastore/src/api.rs | 761 ++++++++++++++++++ crates/datastore/src/bin/store_cli.rs | 329 ++++++++ crates/datastore/src/bin/store_node.rs | 200 +++++ crates/datastore/src/lib.rs | 2 + crates/datastore/src/storage/mod.rs | 24 +- crates/datastore/src/types.rs | 24 + crates/datastore/tests/datastore_tests.rs | 58 ++ .../datastore/PROTOCOL.md | 0 10 files changed, 1462 insertions(+), 27 deletions(-) create mode 100644 crates/datastore/src/api.rs create mode 100644 crates/datastore/src/bin/store_cli.rs create mode 100644 crates/datastore/src/bin/store_node.rs rename {crates => docs/development_history}/datastore/PROTOCOL.md (100%) diff --git a/Cargo.lock b/Cargo.lock index b8bb399..e112d14 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -11,6 +11,12 @@ dependencies = [ "gimli", ] +[[package]] +name = "adler2" +version = "2.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "320119579fcad9c21884f5c4861d16174d0e06250625266f50fe6898340abefa" + [[package]] name = "ahash" version = "0.8.12" @@ -1323,6 +1329,16 @@ version = "0.1.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5baebc0774151f905a1a2cc41989300b1e6fbb29aff0ceffa1064fdd3088d582" +[[package]] +name = "flate2" +version = "1.1.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "843fba2746e448b37e26a819579957415c8cef339bf08564fe8b7ddbd959573c" +dependencies = [ + "crc32fast", + "miniz_oxide", +] + [[package]] name = "fnv" version = "1.0.7" @@ -1831,7 +1847,7 @@ dependencies = [ "tokio", "tokio-rustls", "tower-service", - "webpki-roots", + "webpki-roots 1.0.6", ] [[package]] @@ -2152,7 +2168,7 @@ dependencies = [ "tracing", "url", "wasm-bindgen-futures", - "webpki-roots", + "webpki-roots 1.0.6", ] [[package]] @@ -2304,7 +2320,7 @@ dependencies = [ "tracing", "url", "vergen-gitcl", - "webpki-roots", + "webpki-roots 1.0.6", "ws_stream_wasm", "z32", ] @@ -2569,6 +2585,16 @@ version = "0.3.17" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6877bb514081ee2a7ff5ef9de3281f14a4dd4bceac4c09388074a6b5df8a139a" +[[package]] +name = "miniz_oxide" +version = "0.8.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1fa76a2c86f704bdb222d66965fb3d63269ce38518b83cb0575fca855ebb6316" +dependencies = [ + "adler2", + "simd-adler32", +] + [[package]] name = "mio" version = "1.1.1" @@ -3723,7 +3749,7 @@ dependencies = [ "wasm-bindgen-futures", "wasm-streams", "web-sys", - "webpki-roots", + "webpki-roots 1.0.6", ] [[package]] @@ -4120,6 +4146,12 @@ version = "3.0.0-rc.10" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7f1880df446116126965eeec169136b2e0251dba37c6223bcc819569550edea3" +[[package]] +name = "simd-adler32" +version = "0.3.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e320a6c5ad31d271ad523dcf3ad13e2767ad8b1cb8f047f75a8aeaf8da139da2" + [[package]] name = "simdutf8" version = "0.1.5" @@ -4348,13 +4380,18 @@ name = "swactor-datastore" version = "0.1.0" dependencies = [ "blake3", + "clap", + "ctrlc", "distribution", "proptest", + "runtime-dashboard", "serde", "serde_json", "swactor", "swactor-std", "tempfile", + "tiny_http", + "ureq", ] [[package]] @@ -4942,6 +4979,24 @@ version = "0.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8ecb6da28b8a351d773b68d5825ac39017e680750f980f3a1a85cd8dd28a47c1" +[[package]] +name = "ureq" +version = "2.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "02d1a66277ed75f640d608235660df48c8e3c19f3b4edb6a263315626cc3c01d" +dependencies = [ + "base64 0.22.1", + "flate2", + "log", + "once_cell", + "rustls", + "rustls-pki-types", + "serde", + "serde_json", + "url", + "webpki-roots 0.26.11", +] + [[package]] name = "url" version = "2.5.8" @@ -5571,6 +5626,15 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "webpki-roots" +version = "0.26.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "521bc38abb08001b01866da9f51eb7c5d647a19260e00054a8c7fd5f9e57f7a9" +dependencies = [ + "webpki-roots 1.0.6", +] + [[package]] name = "webpki-roots" version = "1.0.6" diff --git a/crates/datastore/Cargo.toml b/crates/datastore/Cargo.toml index cfda4f7..cab55a4 100644 --- a/crates/datastore/Cargo.toml +++ b/crates/datastore/Cargo.toml @@ -9,6 +9,11 @@ distribution = { path = "../distribution" } serde = { version = "1", features = ["derive"] } serde_json = "1" blake3 = "1" +tiny_http = { version = "0.12", optional = true } +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 } [dev-dependencies] serde_json = "1" @@ -16,3 +21,17 @@ proptest = "1" tempfile = "3" swactor = { path = "../.." } swactor-std = { path = "../std" } + +[features] +node = ["dep:tiny_http", "dep:clap", "dep:ctrlc", "dep:runtime-dashboard"] +cli = ["dep:clap", "dep:ureq"] + +[[bin]] +name = "swactor-store-node" +path = "src/bin/store_node.rs" +required-features = ["node"] + +[[bin]] +name = "swactor-store" +path = "src/bin/store_cli.rs" +required-features = ["cli"] diff --git a/crates/datastore/src/api.rs b/crates/datastore/src/api.rs new file mode 100644 index 0000000..fc0033a --- /dev/null +++ b/crates/datastore/src/api.rs @@ -0,0 +1,761 @@ +//! HTTP API for the datastore node. +//! +//! Bridges HTTP requests to actor messages using `Runtime::new_inbox()` + +//! `try_recv()` polling for synchronous request/response with actors. + +use std::collections::BTreeMap; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, Mutex}; +use std::thread; +use std::time::{Duration, Instant}; + +use swactor::actor::ActorAddress; +use swactor::runtime::{Inbox, Runtime}; + +use crate::chunking::reassemble_blob; +use crate::messages::{BlobStoreMsg, DatastoreNodeMsg, DatastoreResponse, MetadataMsg}; +use crate::types::ContentHash; + +/// Per-peer actor addresses needed for remote operations. +#[derive(Clone)] +pub struct PeerInfo { + pub metadata: ActorAddress, + pub blob_store: ActorAddress, +} + +/// Shared state passed to HTTP handler threads. +struct ApiState { + runtime: Arc, + datastore_addr: ActorAddress, + metadata_addr: ActorAddress, + blob_store_addr: ActorAddress, + peers: Arc>>, +} + +const POLL_TIMEOUT: Duration = Duration::from_secs(5); +const POLL_INTERVAL: Duration = Duration::from_millis(1); + +/// Poll an inbox for a response with timeout. +fn poll_response(inbox: &Inbox, timeout: Duration) -> Option { + let start = Instant::now(); + loop { + if let Some(resp) = inbox.try_recv() { + return Some(resp); + } + if start.elapsed() > timeout { + return None; + } + thread::sleep(POLL_INTERVAL); + } +} + +fn respond_json(request: tiny_http::Request, json: &str) { + let response = tiny_http::Response::from_string(json).with_header( + "Content-Type: application/json" + .parse::() + .unwrap(), + ); + let _ = request.respond(response); +} + +fn respond_bytes(request: tiny_http::Request, data: &[u8]) { + let response = tiny_http::Response::from_data(data.to_vec()).with_header( + "Content-Type: application/octet-stream" + .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) + .with_status_code(status) + .with_header( + "Content-Type: application/json" + .parse::() + .unwrap(), + ); + let _ = request.respond(response); +} + +fn parse_query_string(url: &str) -> BTreeMap { + let mut params = BTreeMap::new(); + if let Some(qs) = url.split('?').nth(1) { + for pair in qs.split('&') { + let mut kv = pair.splitn(2, '='); + if let (Some(k), Some(v)) = (kv.next(), kv.next()) { + params.insert( + url_decode(k), + url_decode(v), + ); + } + } + } + params +} + +fn url_decode(s: &str) -> String { + let mut result = String::with_capacity(s.len()); + let mut chars = s.bytes(); + while let Some(b) = chars.next() { + match b { + b'%' => { + let hi = chars.next().and_then(hex_val); + let lo = chars.next().and_then(hex_val); + if let (Some(h), Some(l)) = (hi, lo) { + result.push((h << 4 | l) as char); + } + } + b'+' => result.push(' '), + _ => result.push(b as char), + } + } + result +} + +fn hex_val(b: u8) -> Option { + match b { + b'0'..=b'9' => Some(b - b'0'), + b'a'..=b'f' => Some(b - b'a' + 10), + b'A'..=b'F' => Some(b - b'A' + 10), + _ => None, + } +} + +// ── JSON serialization helpers ─────────────────────────────────────────── +// +// ContentHash/NodeId derive Serialize as byte arrays ([u8; 32]). +// The API should expose them as hex strings. These helpers convert +// domain types into JSON with human-readable hex fields. + +fn entry_to_json(entry: &crate::types::ObjectEntry) -> serde_json::Value { + let node_hex: String = entry.node_id.0.iter().map(|b| format!("{b:02x}")).collect(); + serde_json::json!({ + "content_hash": entry.content_hash.to_hex(), + "name": entry.name, + "node_id": node_hex, + "tags": entry.tags, + "size_bytes": entry.size_bytes, + "created_at": entry.created_at, + }) +} + +fn manifest_to_json(manifest: &crate::types::ObjectManifest) -> serde_json::Value { + let chunks: Vec = manifest + .chunks + .iter() + .map(|c| { + serde_json::json!({ + "hash": c.hash.to_hex(), + "offset": c.offset, + "size": c.size, + }) + }) + .collect(); + serde_json::json!({ + "content_hash": manifest.content_hash.to_hex(), + "chunks": chunks, + "total_size": manifest.total_size, + "chunk_size": manifest.chunk_size, + "content_type": manifest.content_type, + }) +} + +fn entries_to_json(entries: &[crate::types::ObjectEntry]) -> Vec { + entries.iter().map(entry_to_json).collect() +} + +// ── PUT handler ───────────────────────────────────────────────────────── + +fn handle_put(mut request: tiny_http::Request, url: &str, state: &ApiState) { + let params = parse_query_string(url); + let name = params.get("name").cloned(); + + // Collect tags from query params (skip "name") + let mut tags = BTreeMap::new(); + for (k, v) in ¶ms { + if k != "name" { + tags.insert(k.clone(), v.clone()); + } + } + + // Read body + let mut body = Vec::new(); + if request.as_reader().read_to_end(&mut body).is_err() { + // Can't respond — request consumed + return; + } + + let inbox = match state.runtime.new_inbox::() { + Ok(i) => i, + Err(_) => { + respond_error(request, 500, "failed to create inbox"); + return; + } + }; + + let _ = state.runtime.send_to( + state.datastore_addr, + DatastoreNodeMsg::Put { + data: body, + name, + tags, + reply_to: *inbox.addr(), + }, + ); + + match poll_response(&inbox, POLL_TIMEOUT) { + Some(DatastoreResponse::PutOk { content_hash }) => { + let json = serde_json::json!({ "content_hash": content_hash.to_hex() }).to_string(); + respond_json(request, &json); + } + Some(DatastoreResponse::Error { reason }) => { + respond_error(request, 500, &reason); + } + _ => { + respond_error(request, 504, "timeout waiting for put response"); + } + } +} + +// ── GET handler (metadata) ────────────────────────────────────────────── + +fn handle_get(request: tiny_http::Request, url: &str, state: &ApiState) { + let params = parse_query_string(url); + let hash_hex = match params.get("hash") { + Some(h) => h, + None => { + respond_error(request, 400, "missing ?hash= parameter"); + return; + } + }; + + let content_hash = match ContentHash::from_hex(hash_hex) { + Some(h) => h, + None => { + respond_error(request, 400, "invalid content hash hex"); + return; + } + }; + + let inbox = match state.runtime.new_inbox::() { + Ok(i) => i, + Err(_) => { + respond_error(request, 500, "failed to create inbox"); + return; + } + }; + + let _ = state.runtime.send_to( + state.datastore_addr, + DatastoreNodeMsg::Get { + content_hash, + reply_to: *inbox.addr(), + }, + ); + + match poll_response(&inbox, POLL_TIMEOUT) { + Some(DatastoreResponse::GetOk { entry, manifest }) => { + let json = serde_json::json!({ + "entry": entry_to_json(&entry), + "manifest": manifest_to_json(&manifest), + }) + .to_string(); + respond_json(request, &json); + } + Some(DatastoreResponse::NotFound) => { + respond_error(request, 404, "not found"); + } + Some(DatastoreResponse::Error { reason }) => { + respond_error(request, 500, &reason); + } + _ => { + respond_error(request, 504, "timeout"); + } + } +} + +// ── DATA handler (reassembled binary) ─────────────────────────────────── + +fn handle_data(request: tiny_http::Request, url: &str, state: &ApiState) { + let params = parse_query_string(url); + let hash_hex = match params.get("hash") { + Some(h) => h, + None => { + respond_error(request, 400, "missing ?hash= parameter"); + return; + } + }; + + let content_hash = match ContentHash::from_hex(hash_hex) { + Some(h) => h, + None => { + respond_error(request, 400, "invalid content hash hex"); + return; + } + }; + + // Step 1: Get entry + manifest (try local first) + let inbox = match state.runtime.new_inbox::() { + Ok(i) => i, + Err(_) => { + respond_error(request, 500, "failed to create inbox"); + return; + } + }; + + let _ = state.runtime.send_to( + state.datastore_addr, + DatastoreNodeMsg::Get { + content_hash, + reply_to: *inbox.addr(), + }, + ); + + let (entry, manifest) = match poll_response(&inbox, POLL_TIMEOUT) { + Some(DatastoreResponse::GetOk { entry, manifest }) => (entry, manifest), + Some(DatastoreResponse::NotFound) => { + // Try remote GET + match try_remote_get(content_hash, state) { + Some((e, m)) => (e, m), + None => { + respond_error(request, 404, "not found"); + return; + } + } + } + Some(DatastoreResponse::Error { reason }) => { + respond_error(request, 500, &reason); + return; + } + _ => { + respond_error(request, 504, "timeout"); + return; + } + }; + + // Step 2: Read all chunks + let _ = entry; // entry used for metadata context, manifest for chunks + let mut chunk_data = Vec::new(); + for chunk_ref in &manifest.chunks { + let chunk_inbox = match state.runtime.new_inbox::() { + Ok(i) => i, + Err(_) => { + respond_error(request, 500, "failed to create inbox"); + return; + } + }; + + let _ = state.runtime.send_to( + state.datastore_addr, + DatastoreNodeMsg::ReadChunk { + hash: chunk_ref.hash, + reply_to: *chunk_inbox.addr(), + }, + ); + + match poll_response(&chunk_inbox, POLL_TIMEOUT) { + Some(DatastoreResponse::ChunkOk { hash, data }) => { + chunk_data.push((hash, data)); + } + _ => { + respond_error(request, 500, "failed to read chunk"); + return; + } + } + } + + // Step 3: Reassemble + match reassemble_blob(&manifest, &chunk_data) { + Ok(data) => respond_bytes(request, &data), + Err(e) => respond_error(request, 500, &format!("reassembly failed: {e:?}")), + } +} + +// ── DELETE handler ────────────────────────────────────────────────────── + +fn handle_delete(request: tiny_http::Request, url: &str, state: &ApiState) { + let params = parse_query_string(url); + let hash_hex = match params.get("hash") { + Some(h) => h, + None => { + respond_error(request, 400, "missing ?hash= parameter"); + return; + } + }; + + let content_hash = match ContentHash::from_hex(hash_hex) { + Some(h) => h, + None => { + respond_error(request, 400, "invalid content hash hex"); + return; + } + }; + + let inbox = match state.runtime.new_inbox::() { + Ok(i) => i, + Err(_) => { + respond_error(request, 500, "failed to create inbox"); + return; + } + }; + + let _ = state.runtime.send_to( + state.datastore_addr, + DatastoreNodeMsg::Delete { + content_hash, + reply_to: *inbox.addr(), + }, + ); + + match poll_response(&inbox, POLL_TIMEOUT) { + Some(DatastoreResponse::DeleteOk { content_hash }) => { + let json = serde_json::json!({ "content_hash": content_hash.to_hex() }).to_string(); + respond_json(request, &json); + } + Some(DatastoreResponse::NotFound) => { + respond_error(request, 404, "not found"); + } + Some(DatastoreResponse::Error { reason }) => { + respond_error(request, 500, &reason); + } + _ => { + respond_error(request, 504, "timeout"); + } + } +} + +// ── LIST handler ──────────────────────────────────────────────────────── + +fn handle_list(request: tiny_http::Request, url: &str, state: &ApiState) { + let params = parse_query_string(url); + let name_filter = params.get("name").cloned(); + let all = params.get("all").map_or(false, |v| v == "true" || v == "1"); + + if all { + handle_list_swarm(request, name_filter, state); + } else { + handle_list_local(request, name_filter, state); + } +} + +fn handle_list_local( + request: tiny_http::Request, + name_filter: Option, + state: &ApiState, +) { + let inbox = match state.runtime.new_inbox::() { + Ok(i) => i, + Err(_) => { + respond_error(request, 500, "failed to create inbox"); + return; + } + }; + + let _ = state.runtime.send_to( + state.datastore_addr, + DatastoreNodeMsg::List { + name_filter, + all: false, + reply_to: *inbox.addr(), + }, + ); + + match poll_response(&inbox, POLL_TIMEOUT) { + Some(DatastoreResponse::ListOk { entries }) => { + let json = serde_json::json!({ "entries": entries_to_json(&entries) }).to_string(); + respond_json(request, &json); + } + Some(DatastoreResponse::Error { reason }) => { + respond_error(request, 500, &reason); + } + _ => { + respond_error(request, 504, "timeout"); + } + } +} + +/// ListSwarm fan-out: query local + all peers, merge and deduplicate. +fn handle_list_swarm( + request: tiny_http::Request, + name_filter: Option, + state: &ApiState, +) { + let mut all_entries = Vec::new(); + + // Query local + let inbox = match state.runtime.new_inbox::() { + Ok(i) => i, + Err(_) => { + respond_error(request, 500, "failed to create inbox"); + return; + } + }; + + let _ = state.runtime.send_to( + state.metadata_addr, + MetadataMsg::ListLocal { + name_filter: name_filter.clone(), + reply_to: *inbox.addr(), + }, + ); + + if let Some(DatastoreResponse::ListOk { entries }) = poll_response(&inbox, POLL_TIMEOUT) { + all_entries.extend(entries); + } + + // Query each peer + let peers = state.peers.lock().unwrap().clone(); + for peer in &peers { + let peer_inbox = match state.runtime.new_inbox::() { + Ok(i) => i, + Err(_) => continue, + }; + + let _ = state.runtime.send_to( + peer.metadata, + MetadataMsg::ListLocal { + name_filter: name_filter.clone(), + reply_to: *peer_inbox.addr(), + }, + ); + + if let Some(DatastoreResponse::ListOk { entries }) = + poll_response(&peer_inbox, Duration::from_secs(2)) + { + all_entries.extend(entries); + } + } + + // Deduplicate by content_hash + let mut seen = std::collections::HashSet::new(); + all_entries.retain(|e| seen.insert(e.content_hash)); + + let json = serde_json::json!({ "entries": entries_to_json(&all_entries) }).to_string(); + respond_json(request, &json); +} + +// ── STATUS handler ────────────────────────────────────────────────────── + +fn handle_status(request: tiny_http::Request, state: &ApiState) { + let inbox = match state.runtime.new_inbox::() { + Ok(i) => i, + Err(_) => { + respond_error(request, 500, "failed to create inbox"); + return; + } + }; + + let _ = state.runtime.send_to( + state.datastore_addr, + DatastoreNodeMsg::Status { + reply_to: *inbox.addr(), + }, + ); + + match poll_response(&inbox, POLL_TIMEOUT) { + Some(DatastoreResponse::NodeStatus { node_id }) => { + let hex: String = node_id.0.iter().map(|b| format!("{b:02x}")).collect(); + let json = serde_json::json!({ "node_id": hex }).to_string(); + respond_json(request, &json); + } + _ => { + respond_error(request, 504, "timeout"); + } + } +} + +// ── Remote GET orchestration ──────────────────────────────────────────── + +/// Try to fetch an object from peers when not found locally. +/// Returns (entry, manifest) on success, stores chunks locally as a side effect. +fn try_remote_get( + content_hash: ContentHash, + state: &ApiState, +) -> Option<(crate::types::ObjectEntry, crate::types::ObjectManifest)> { + let peers = state.peers.lock().unwrap().clone(); + + for peer in &peers { + // Ask peer's metadata actor for the object + let find_inbox = state.runtime.new_inbox::().ok()?; + let _ = state.runtime.send_to( + peer.metadata, + MetadataMsg::HandleFindObject { + from: distribution::types::NodeId([0; 32]), // placeholder + content_hash, + reply_to: *find_inbox.addr(), + }, + ); + + let (entry, _) = match poll_response(&find_inbox, Duration::from_secs(2)) { + Some(DatastoreResponse::GetOk { entry, manifest }) => (entry, manifest), + _ => continue, + }; + + // Get manifest from peer's blob store + let manifest_inbox = state.runtime.new_inbox::().ok()?; + let _ = state.runtime.send_to( + peer.blob_store, + BlobStoreMsg::ReadManifest { + hash: content_hash, + reply_to: *manifest_inbox.addr(), + }, + ); + + let manifest = match poll_response(&manifest_inbox, Duration::from_secs(2)) { + Some(DatastoreResponse::ManifestOk { manifest }) => manifest, + _ => continue, + }; + + // Fetch each chunk from peer and store locally + let mut all_ok = true; + for chunk_ref in &manifest.chunks { + let chunk_inbox = match state.runtime.new_inbox::() { + Ok(i) => i, + Err(_) => { + all_ok = false; + break; + } + }; + + let _ = state.runtime.send_to( + peer.blob_store, + BlobStoreMsg::ReadChunk { + hash: chunk_ref.hash, + reply_to: *chunk_inbox.addr(), + }, + ); + + match poll_response(&chunk_inbox, Duration::from_secs(2)) { + Some(DatastoreResponse::ChunkOk { hash, data }) => { + // Store locally + let store_inbox = + match state.runtime.new_inbox::() { + Ok(i) => i, + Err(_) => { + all_ok = false; + break; + } + }; + let _ = state.runtime.send_to( + state.blob_store_addr, + BlobStoreMsg::WriteChunk { + hash, + data, + reply_to: *store_inbox.addr(), + }, + ); + // Wait for confirmation + let _ = poll_response(&store_inbox, Duration::from_secs(2)); + } + _ => { + all_ok = false; + break; + } + } + } + + if !all_ok { + continue; + } + + // Store manifest locally + let m_inbox = match state.runtime.new_inbox::() { + Ok(i) => i, + Err(_) => continue, + }; + let _ = state.runtime.send_to( + state.blob_store_addr, + BlobStoreMsg::WriteManifest { + manifest: manifest.clone(), + reply_to: *m_inbox.addr(), + }, + ); + let _ = poll_response(&m_inbox, Duration::from_secs(2)); + + // Store entry+manifest in local metadata + let put_inbox = match state.runtime.new_inbox::() { + Ok(i) => i, + Err(_) => continue, + }; + let _ = state.runtime.send_to( + state.metadata_addr, + MetadataMsg::PutObject { + entry: entry.clone(), + manifest: manifest.clone(), + reply_to: *put_inbox.addr(), + }, + ); + let _ = poll_response(&put_inbox, Duration::from_secs(2)); + + return Some((entry, manifest)); + } + + None +} + +// ── Server startup ────────────────────────────────────────────────────── + +/// Start the HTTP API server for the datastore. +/// +/// Returns a shared shutdown flag (set to `true` to stop the server) +/// and a peer list that can be updated to enable remote operations. +pub fn start_api_server( + runtime: Arc, + datastore_addr: ActorAddress, + metadata_addr: ActorAddress, + blob_store_addr: ActorAddress, + port: u16, +) -> (Arc, Arc>>) { + let shutdown = Arc::new(AtomicBool::new(false)); + let peers: Arc>> = Arc::new(Mutex::new(Vec::new())); + + let state = Arc::new(ApiState { + runtime, + datastore_addr, + metadata_addr, + blob_store_addr, + peers: Arc::clone(&peers), + }); + + let addr = format!("0.0.0.0:{port}"); + let server = tiny_http::Server::http(&addr).expect("failed to bind datastore API server"); + let server = Arc::new(server); + + for _ in 0..4 { + let server = Arc::clone(&server); + let state = Arc::clone(&state); + let shutdown = Arc::clone(&shutdown); + thread::spawn(move || { + loop { + if shutdown.load(Ordering::Relaxed) { + break; + } + let request = match server.recv_timeout(Duration::from_millis(500)) { + Ok(Some(r)) => r, + Ok(None) => continue, + Err(_) => break, + }; + + let url = request.url().to_string(); + let path = url.split('?').next().unwrap_or(&url); + let method = request.method().as_str(); + + match (method, path) { + ("POST", "/api/put") => handle_put(request, &url, &state), + ("GET", "/api/get") => handle_get(request, &url, &state), + ("GET", "/api/data") => handle_data(request, &url, &state), + ("POST", "/api/delete") => handle_delete(request, &url, &state), + ("GET", "/api/list") => handle_list(request, &url, &state), + ("GET", "/api/status") => handle_status(request, &state), + _ => { + respond_error(request, 404, "not found"); + } + } + } + }); + } + + (shutdown, peers) +} diff --git a/crates/datastore/src/bin/store_cli.rs b/crates/datastore/src/bin/store_cli.rs new file mode 100644 index 0000000..6b09b27 --- /dev/null +++ b/crates/datastore/src/bin/store_cli.rs @@ -0,0 +1,329 @@ +//! swactor-store — CLI client for the datastore node. +//! +//! Talks to a running `swactor-store-node` over its HTTP API. + +use std::fs; +use std::io::Read; +use std::path::PathBuf; + +use clap::{Parser, Subcommand}; + +#[derive(Parser)] +#[command(name = "swactor-store", about = "Swactor datastore CLI")] +struct Args { + /// Base URL of the datastore node + #[arg(long, default_value = "http://localhost:9091")] + url: String, + + #[command(subcommand)] + command: Command, +} + +#[derive(Subcommand)] +enum Command { + /// Store a file in the datastore + Put { + /// Path to the local file to store + path: PathBuf, + /// Optional name label + #[arg(long)] + name: Option, + }, + /// Retrieve object metadata (or download with --output) + Get { + /// Content hash (hex) + hash: String, + /// Download file to this path + #[arg(long)] + output: Option, + }, + /// Delete an object + Delete { + /// Content hash (hex) + hash: String, + }, + /// List stored objects + List { + /// Filter by name substring + #[arg(long)] + name: Option, + /// List from all nodes (swarm-wide) + #[arg(long)] + all: bool, + }, + /// Query node status + Status, +} + +fn main() { + let args = Args::parse(); + let base = args.url.trim_end_matches('/'); + + match args.command { + Command::Put { path, name } => cmd_put(base, &path, name.as_deref()), + Command::Get { hash, output } => cmd_get(base, &hash, output.as_deref()), + Command::Delete { hash } => cmd_delete(base, &hash), + Command::List { name, all } => cmd_list(base, name.as_deref(), all), + Command::Status => cmd_status(base), + } +} + +fn cmd_put(base: &str, path: &PathBuf, name: Option<&str>) { + let data = match fs::read(path) { + Ok(d) => d, + Err(e) => { + eprintln!("Error reading {}: {e}", path.display()); + std::process::exit(1); + } + }; + + let label = name + .map(|n| n.to_string()) + .or_else(|| { + path.file_name() + .and_then(|f| f.to_str()) + .map(|s| s.to_string()) + }); + + let mut url = format!("{base}/api/put"); + if let Some(ref n) = label { + url.push_str(&format!("?name={}", url_encode(n))); + } + + let resp = match ureq::post(&url).send_bytes(&data) { + Ok(r) => r, + Err(e) => { + eprintln!("Error: {e}"); + std::process::exit(1); + } + }; + + let body: serde_json::Value = match resp.into_json() { + Ok(v) => v, + Err(e) => { + eprintln!("Error parsing response: {e}"); + std::process::exit(1); + } + }; + + if let Some(hash) = body.get("content_hash").and_then(|v| v.as_str()) { + println!("{hash}"); + } else if let Some(err) = body.get("error").and_then(|v| v.as_str()) { + eprintln!("Error: {err}"); + std::process::exit(1); + } +} + +fn cmd_get(base: &str, hash: &str, output: Option<&std::path::Path>) { + if let Some(out_path) = output { + // Download raw data + let url = format!("{base}/api/data?hash={hash}"); + let resp = match ureq::get(&url).call() { + Ok(r) => r, + Err(e) => { + eprintln!("Error: {e}"); + std::process::exit(1); + } + }; + + if resp.status() != 200 { + let body = resp.into_string().unwrap_or_default(); + eprintln!("Error: {body}"); + std::process::exit(1); + } + + let mut data = Vec::new(); + if let Err(e) = resp.into_reader().read_to_end(&mut data) { + eprintln!("Error reading response: {e}"); + std::process::exit(1); + } + + if let Err(e) = fs::write(out_path, &data) { + eprintln!("Error writing {}: {e}", out_path.display()); + std::process::exit(1); + } + println!("Written {} bytes to {}", data.len(), out_path.display()); + } else { + // Metadata only + let url = format!("{base}/api/get?hash={hash}"); + let resp = match ureq::get(&url).call() { + Ok(r) => r, + Err(e) => { + eprintln!("Error: {e}"); + std::process::exit(1); + } + }; + + let body: serde_json::Value = match resp.into_json() { + Ok(v) => v, + Err(e) => { + eprintln!("Error parsing response: {e}"); + std::process::exit(1); + } + }; + + if let Some(err) = body.get("error").and_then(|v| v.as_str()) { + eprintln!("Error: {err}"); + std::process::exit(1); + } + + if let Some(entry) = body.get("entry") { + println!("Hash: {}", entry.get("content_hash").and_then(|v| v.as_str()).unwrap_or("?")); + println!( + "Name: {}", + entry.get("name").and_then(|v| v.as_str()).unwrap_or("(none)") + ); + println!( + "Size: {} bytes", + entry.get("size_bytes").and_then(|v| v.as_u64()).unwrap_or(0) + ); + println!( + "Node: {}", + entry.get("node_id").and_then(|v| v.as_str()).unwrap_or("?") + ); + if let Some(tags) = entry.get("tags").and_then(|v| v.as_object()) { + if !tags.is_empty() { + println!("Tags:"); + for (k, v) in tags { + println!(" {k}: {v}"); + } + } + } + } + if let Some(manifest) = body.get("manifest") { + println!( + "Chunks: {}", + manifest + .get("chunks") + .and_then(|v| v.as_array()) + .map(|a| a.len()) + .unwrap_or(0) + ); + } + } +} + +fn cmd_delete(base: &str, hash: &str) { + let url = format!("{base}/api/delete?hash={hash}"); + let resp = match ureq::post(&url).send_bytes(&[]) { + Ok(r) => r, + Err(e) => { + eprintln!("Error: {e}"); + std::process::exit(1); + } + }; + + let body: serde_json::Value = match resp.into_json() { + Ok(v) => v, + Err(e) => { + eprintln!("Error parsing response: {e}"); + std::process::exit(1); + } + }; + + if let Some(h) = body.get("content_hash").and_then(|v| v.as_str()) { + println!("Deleted {h}"); + } else if let Some(err) = body.get("error").and_then(|v| v.as_str()) { + eprintln!("Error: {err}"); + std::process::exit(1); + } +} + +fn cmd_list(base: &str, name: Option<&str>, all: bool) { + let mut url = format!("{base}/api/list"); + let mut sep = '?'; + if let Some(n) = name { + url.push_str(&format!("{sep}name={}", url_encode(n))); + sep = '&'; + } + if all { + url.push_str(&format!("{sep}all=true")); + } + + let resp = match ureq::get(&url).call() { + Ok(r) => r, + Err(e) => { + eprintln!("Error: {e}"); + std::process::exit(1); + } + }; + + let body: serde_json::Value = match resp.into_json() { + Ok(v) => v, + Err(e) => { + eprintln!("Error parsing response: {e}"); + std::process::exit(1); + } + }; + + if let Some(err) = body.get("error").and_then(|v| v.as_str()) { + eprintln!("Error: {err}"); + std::process::exit(1); + } + + if let Some(entries) = body.get("entries").and_then(|v| v.as_array()) { + if entries.is_empty() { + println!("(no entries)"); + return; + } + // Print header + println!("{:<64} {:>10} {}", "HASH", "SIZE", "NAME"); + println!("{}", "-".repeat(90)); + for entry in entries { + let hash = entry + .get("content_hash") + .and_then(|v| v.as_str()) + .unwrap_or("?"); + let size = entry + .get("size_bytes") + .and_then(|v| v.as_u64()) + .unwrap_or(0); + let name = entry + .get("name") + .and_then(|v| v.as_str()) + .unwrap_or("(none)"); + println!("{hash:<64} {size:>10} {name}"); + } + } +} + +fn cmd_status(base: &str) { + let url = format!("{base}/api/status"); + let resp = match ureq::get(&url).call() { + Ok(r) => r, + Err(e) => { + eprintln!("Error: {e}"); + std::process::exit(1); + } + }; + + let body: serde_json::Value = match resp.into_json() { + Ok(v) => v, + Err(e) => { + eprintln!("Error parsing response: {e}"); + std::process::exit(1); + } + }; + + if let Some(node_id) = body.get("node_id").and_then(|v| v.as_str()) { + println!("Node ID: {node_id}"); + } else if let Some(err) = body.get("error").and_then(|v| v.as_str()) { + eprintln!("Error: {err}"); + std::process::exit(1); + } +} + +fn url_encode(s: &str) -> String { + let mut result = String::with_capacity(s.len()); + for b in s.bytes() { + match b { + b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => { + result.push(b as char); + } + _ => { + result.push_str(&format!("%{b:02X}")); + } + } + } + result +} diff --git a/crates/datastore/src/bin/store_node.rs b/crates/datastore/src/bin/store_node.rs new file mode 100644 index 0000000..0371585 --- /dev/null +++ b/crates/datastore/src/bin/store_node.rs @@ -0,0 +1,200 @@ +//! swactor-store-node — standalone datastore node with HTTP API. +//! +//! Starts the actor runtime, spawns datastore actors, and serves +//! a REST API for external tools (the `swactor-store` CLI). + +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::Arc; +use std::thread; +use std::time::Duration; + +use clap::Parser; + +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::messages::MetadataMsg; +use swactor_datastore::storage::{FilesystemBackend, InMemoryBackend}; +use swactor_datastore::DatastoreConfig; + +use distribution::types::NodeId; + +#[derive(Parser)] +#[command(name = "swactor-store-node", about = "Swactor distributed datastore node")] +struct Args { + /// HTTP API port + #[arg(long, default_value = "9091")] + port: u16, + + /// Storage directory (omit for in-memory) + #[arg(long)] + storage_path: Option, + + /// Dashboard HTTP port (omit to disable dashboard) + #[arg(long)] + dashboard_port: Option, + + /// Chunk size in bytes + #[arg(long, default_value = "1048576")] + chunk_size: u32, + + /// GC interval in ticks (each tick is ~100ms) + #[arg(long, default_value = "1000")] + gc_interval: u64, + + /// Dissemination interval in ticks + #[arg(long, default_value = "50")] + disseminate_interval: u64, +} + +fn main() { + let args = Args::parse(); + let stop = Arc::new(AtomicBool::new(false)); + + // Signal handler + { + let stop = Arc::clone(&stop); + ctrlc::set_handler(move || { + stop.store(true, Ordering::Relaxed); + }) + .expect("failed to set signal handler"); + } + + // Optionally start dashboard + let dash = args.dashboard_port.map(|port| { + let d = runtime_dashboard::start_dashboard(runtime_dashboard::DashboardConfig { + port, + ..Default::default() + }); + d.install_tracing(); + d + }); + + // Create runtime + let num_threads = 2; + let collector = runtime_dashboard::collector::StatsCollector::new(num_threads); + let mut rt = Runtime::new(RuntimeConfig { + num_threads, + max_actors: 1024, + channel_buffer_size: 2000, + ..Default::default() + }); + rt.set_stats_hook(collector.clone()); + + // Generate node ID from random bytes + let node_id = { + let mut bytes = [0u8; 32]; + for (i, b) in std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_nanos() + .to_le_bytes() + .iter() + .enumerate() + { + bytes[i % 32] ^= *b; + } + // Mix in process id for uniqueness + let pid = std::process::id(); + for (i, b) in pid.to_le_bytes().iter().enumerate() { + bytes[i + 16] ^= *b; + } + NodeId(bytes) + }; + + // Datastore config + let config = DatastoreConfig { + chunk_size: args.chunk_size, + storage_path: args + .storage_path + .as_ref() + .map(|s| s.into()) + .unwrap_or_else(|| "datastore".into()), + gc_interval: args.gc_interval, + ..Default::default() + }; + + // Create storage backend + let backend: Box = match &args.storage_path { + Some(path) => { + let p = std::path::PathBuf::from(path); + std::fs::create_dir_all(&p).expect("failed to create storage directory"); + Box::new(FilesystemBackend::new(p)) + } + None => Box::new(InMemoryBackend::new()), + }; + + // Spawn actors before starting runtime threads + let blob_store_addr = rt + .spawn(BlobStoreActor::new(backend)) + .expect("failed to spawn BlobStoreActor"); + + let mut metadata = MetadataActor::new(node_id, &config); + metadata.set_blob_store(blob_store_addr); + let metadata_addr = rt + .spawn(metadata) + .expect("failed to spawn MetadataActor"); + + let datastore_node = DatastoreNode::new(node_id, blob_store_addr, metadata_addr, config); + let datastore_addr = rt + .spawn(datastore_node) + .expect("failed to spawn DatastoreNode"); + + // Start runtime + let handle = rt.run().expect("failed to start runtime"); + + if let Some(ref d) = dash { + d.set_runtime(handle.runtime.clone(), collector); + } + + // Start HTTP API + let (api_shutdown, _peers) = start_api_server( + handle.runtime.clone(), + datastore_addr, + metadata_addr, + blob_store_addr, + args.port, + ); + + 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!("Dashboard at http://0.0.0.0:{port}"); + } + if args.storage_path.is_some() { + eprintln!("Storage: {}", args.storage_path.as_ref().unwrap()); + } else { + eprintln!("Storage: in-memory"); + } + + // Main loop + let mut round: u64 = 0; + while !stop.load(Ordering::Relaxed) { + round += 1; + + if round % args.gc_interval == 0 { + let _ = handle + .runtime + .send_to(metadata_addr, MetadataMsg::GcTick); + } + + if round % args.disseminate_interval == 0 { + let _ = handle + .runtime + .send_to(metadata_addr, MetadataMsg::DisseminateTick); + } + + thread::sleep(Duration::from_millis(100)); + } + + eprintln!("\nShutting down..."); + api_shutdown.store(true, Ordering::Relaxed); + handle.shutdown(); + if let Some(d) = dash { + d.shutdown(); + } + handle.join(); +} diff --git a/crates/datastore/src/lib.rs b/crates/datastore/src/lib.rs index 28a3cef..0c90686 100644 --- a/crates/datastore/src/lib.rs +++ b/crates/datastore/src/lib.rs @@ -4,6 +4,8 @@ pub mod chunking; pub mod storage; pub mod actors; pub mod cli; +#[cfg(feature = "node")] +pub mod api; pub use types::{ChunkRef, ContentHash, DatastoreConfig, ObjectEntry, ObjectManifest}; pub use messages::{BlobStoreMsg, DatastoreNodeMsg, DatastoreResponse, MetadataMsg, TransferMsg}; diff --git a/crates/datastore/src/storage/mod.rs b/crates/datastore/src/storage/mod.rs index e1711d7..6623934 100644 --- a/crates/datastore/src/storage/mod.rs +++ b/crates/datastore/src/storage/mod.rs @@ -85,7 +85,7 @@ impl FilesystemBackend { }; for file in files.flatten() { if let Some(name) = file.file_name().to_str() { - if let Some(hash) = hex_to_content_hash(name) { + if let Some(hash) = ContentHash::from_hex(name) { self.chunk_index.insert(hash); } } @@ -173,25 +173,3 @@ impl StorageBackend for FilesystemBackend { } } -/// Parse a 64-character hex string into a ContentHash. -fn hex_to_content_hash(hex: &str) -> Option { - if hex.len() != 64 { - return None; - } - let mut bytes = [0u8; 32]; - for (i, chunk) in hex.as_bytes().chunks(2).enumerate() { - let hi = hex_digit(chunk[0])?; - let lo = hex_digit(chunk[1])?; - bytes[i] = (hi << 4) | lo; - } - Some(ContentHash(bytes)) -} - -fn hex_digit(b: u8) -> Option { - match b { - b'0'..=b'9' => Some(b - b'0'), - b'a'..=b'f' => Some(b - b'a' + 10), - b'A'..=b'F' => Some(b - b'A' + 10), - _ => None, - } -} diff --git a/crates/datastore/src/types.rs b/crates/datastore/src/types.rs index 73acd8a..5ad9be9 100644 --- a/crates/datastore/src/types.rs +++ b/crates/datastore/src/types.rs @@ -53,6 +53,21 @@ impl ContentHash { zeros } + /// Parse a 64-character hex string into a ContentHash. + /// Returns `None` if the string is not exactly 64 hex characters. + pub fn from_hex(hex: &str) -> Option { + if hex.len() != 64 { + return None; + } + let mut bytes = [0u8; 32]; + for (i, chunk) in hex.as_bytes().chunks(2).enumerate() { + let hi = hex_digit(chunk[0])?; + let lo = hex_digit(chunk[1])?; + bytes[i] = (hi << 4) | lo; + } + Some(ContentHash(bytes)) + } + /// Encode as lowercase hex string. pub fn to_hex(&self) -> String { let mut s = String::with_capacity(64); @@ -86,6 +101,15 @@ impl fmt::Display for ContentHash { } } +fn hex_digit(b: u8) -> Option { + match b { + b'0'..=b'9' => Some(b - b'0'), + b'a'..=b'f' => Some(b - b'a' + 10), + b'A'..=b'F' => Some(b - b'A' + 10), + _ => None, + } +} + // ─── ObjectEntry ──────────────────────────────────────────────────────────── /// Metadata record for a stored object — content-addressed by `blake3(blob_bytes)`. diff --git a/crates/datastore/tests/datastore_tests.rs b/crates/datastore/tests/datastore_tests.rs index 789eb84..cb87f66 100644 --- a/crates/datastore/tests/datastore_tests.rs +++ b/crates/datastore/tests/datastore_tests.rs @@ -111,6 +111,53 @@ fn manifest_survives_json_round_trip() { assert_eq!(manifest, deserialized); } +// ═══════════════════════════════════════════════════════════════════════════ +// Scenario: ContentHash hex round-trip +// ═══════════════════════════════════════════════════════════════════════════ + +/// to_hex → from_hex round-trips for any content hash. +#[test] +fn content_hash_hex_round_trip() { + let data = b"round-trip through hex encoding"; + let hash = ContentHash::of(data); + let hex = hash.to_hex(); + let recovered = ContentHash::from_hex(&hex).expect("valid hex should parse"); + assert_eq!(hash, recovered); +} + +/// from_hex rejects strings that are not exactly 64 hex characters. +#[test] +fn from_hex_rejects_wrong_length() { + assert!(ContentHash::from_hex("abcd").is_none()); + assert!(ContentHash::from_hex("").is_none()); + // 63 chars + assert!(ContentHash::from_hex( + &"a".repeat(63) + ).is_none()); + // 65 chars + assert!(ContentHash::from_hex( + &"a".repeat(65) + ).is_none()); +} + +/// from_hex rejects non-hex characters. +#[test] +fn from_hex_rejects_non_hex_chars() { + // 'g' is not valid hex + let bad = format!("{}g{}", "a".repeat(31), "a".repeat(32)); + assert_eq!(bad.len(), 64); + assert!(ContentHash::from_hex(&bad).is_none()); +} + +/// from_hex accepts uppercase hex. +#[test] +fn from_hex_accepts_uppercase() { + let hash = ContentHash::of(b"uppercase test"); + let hex_upper = hash.to_hex().to_uppercase(); + let recovered = ContentHash::from_hex(&hex_upper).expect("uppercase hex should parse"); + assert_eq!(hash, recovered); +} + // ═══════════════════════════════════════════════════════════════════════════ // Scenario: ContentHash DHT properties // ═══════════════════════════════════════════════════════════════════════════ @@ -216,6 +263,17 @@ mod proptests { } } + // ContentHash::from_hex is the inverse of to_hex. + proptest! { + #[test] + fn hex_round_trip(data in proptest::collection::vec(any::(), 1..1024)) { + let hash = ContentHash::of(&data); + let hex = hash.to_hex(); + let recovered = ContentHash::from_hex(&hex).unwrap(); + prop_assert_eq!(hash, recovered); + } + } + // Chunking any data and reassembling preserves the original. proptest! { #[test] diff --git a/crates/datastore/PROTOCOL.md b/docs/development_history/datastore/PROTOCOL.md similarity index 100% rename from crates/datastore/PROTOCOL.md rename to docs/development_history/datastore/PROTOCOL.md