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 c196d3c..d295a32 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" @@ -4343,6 +4375,26 @@ dependencies = [ "wat", ] +[[package]] +name = "swactor-datastore" +version = "0.1.0" +dependencies = [ + "blake3", + "clap", + "ctrlc", + "distribution", + "proptest", + "runtime-dashboard", + "serde", + "serde_json", + "swactor", + "swactor-std", + "tempfile", + "tiny_http", + "toml", + "ureq", +] + [[package]] name = "swactor-std" version = "0.1.0" @@ -4928,6 +4980,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" @@ -5557,6 +5627,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" @@ -6153,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 876270b..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", "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 new file mode 100644 index 0000000..4aa7b5c --- /dev/null +++ b/crates/datastore/Cargo.toml @@ -0,0 +1,41 @@ +[package] +name = "swactor-datastore" +version = "0.1.0" +edition = "2024" + +[dependencies] +swactor = { path = "../..", features = ["serde", "transport"] } +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 } +toml = { version = "0.8", optional = true } + +[dev-dependencies] +serde_json = "1" +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", "dep:toml"] +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/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/actors/blob_store.rs b/crates/datastore/src/actors/blob_store.rs new file mode 100644 index 0000000..5e860b9 --- /dev/null +++ b/crates/datastore/src/actors/blob_store.rs @@ -0,0 +1,178 @@ +//! BlobStoreActor — content-addressed chunk and manifest storage. +//! +//! Delegates all I/O through a `StorageBackend` trait, allowing pluggable +//! backends (filesystem for MVP, IndexedDB for browser, etc.). + +use std::collections::HashSet; + +use swactor::actor::{ActorInterface, Ctx}; + +use crate::messages::{BlobStoreMsg, DatastoreResponse}; +use crate::storage::StorageBackend; +use crate::types::{ContentHash, ObjectManifest}; + +/// Manages chunk and manifest storage via a pluggable backend. +pub struct BlobStoreActor { + backend: Box, +} + +impl BlobStoreActor { + pub fn new(backend: Box) -> Self { + Self { backend } + } + + fn handle_write_chunk( + &mut self, + ctx: &Ctx, + hash: ContentHash, + data: Vec, + reply_to: swactor::actor::ActorAddress, + ) { + match self.backend.write_chunk(&hash, &data) { + Ok(()) => { + let _ = ctx.send(reply_to, DatastoreResponse::ChunkStored { hash }); + } + Err(e) => { + let _ = ctx.send( + reply_to, + DatastoreResponse::Error { + reason: format!("write chunk failed: {e}"), + }, + ); + } + } + } + + fn handle_read_chunk( + &self, + ctx: &Ctx, + hash: ContentHash, + reply_to: swactor::actor::ActorAddress, + ) { + match self.backend.read_chunk(&hash) { + Ok(Some(data)) => { + let _ = ctx.send(reply_to, DatastoreResponse::ChunkOk { hash, data }); + } + Ok(None) => { + let _ = ctx.send(reply_to, DatastoreResponse::NotFound); + } + Err(e) => { + let _ = ctx.send( + reply_to, + DatastoreResponse::Error { + reason: format!("read chunk failed: {e}"), + }, + ); + } + } + } + + fn handle_delete_chunk(&mut self, hash: ContentHash) { + let _ = self.backend.delete_chunk(&hash); + } + + fn handle_has_chunk( + &self, + ctx: &Ctx, + hash: ContentHash, + reply_to: swactor::actor::ActorAddress, + ) { + let exists = self.backend.has_chunk(&hash); + let _ = ctx.send(reply_to, DatastoreResponse::Bool(exists)); + } + + fn handle_list_chunks(&self, ctx: &Ctx, reply_to: swactor::actor::ActorAddress) { + let hashes = self.backend.list_chunks(); + let _ = ctx.send(reply_to, DatastoreResponse::ChunkList { hashes }); + } + + fn handle_gc_unreferenced(&mut self, referenced: HashSet) { + let all_chunks = self.backend.list_chunks(); + for hash in all_chunks { + if !referenced.contains(&hash) { + let _ = self.backend.delete_chunk(&hash); + } + } + } + + fn handle_write_manifest( + &mut self, + ctx: &Ctx, + manifest: ObjectManifest, + reply_to: swactor::actor::ActorAddress, + ) { + let hash = manifest.content_hash; + match self.backend.write_manifest(&manifest) { + Ok(()) => { + let _ = ctx.send(reply_to, DatastoreResponse::ManifestStored { hash }); + } + Err(e) => { + let _ = ctx.send( + reply_to, + DatastoreResponse::Error { + reason: format!("write manifest failed: {e}"), + }, + ); + } + } + } + + fn handle_read_manifest( + &self, + ctx: &Ctx, + hash: ContentHash, + reply_to: swactor::actor::ActorAddress, + ) { + match self.backend.read_manifest(&hash) { + Ok(Some(manifest)) => { + let _ = ctx.send(reply_to, DatastoreResponse::ManifestOk { manifest }); + } + Ok(None) => { + let _ = ctx.send(reply_to, DatastoreResponse::NotFound); + } + Err(e) => { + let _ = ctx.send( + reply_to, + DatastoreResponse::Error { + reason: format!("read manifest failed: {e}"), + }, + ); + } + } + } +} + +impl ActorInterface for BlobStoreActor { + type Incoming = BlobStoreMsg; + type Response = DatastoreResponse; + + fn handle(&mut self, ctx: &Ctx, msg: BlobStoreMsg) { + match msg { + BlobStoreMsg::WriteChunk { + hash, + data, + reply_to, + } => self.handle_write_chunk(ctx, hash, data, reply_to), + BlobStoreMsg::ReadChunk { hash, reply_to } => { + self.handle_read_chunk(ctx, hash, reply_to) + } + BlobStoreMsg::DeleteChunk { hash } => self.handle_delete_chunk(hash), + BlobStoreMsg::HasChunk { hash, reply_to } => { + self.handle_has_chunk(ctx, hash, reply_to) + } + BlobStoreMsg::ListChunks { reply_to } => { + self.handle_list_chunks(ctx, reply_to) + } + BlobStoreMsg::GcUnreferenced { referenced } => { + self.handle_gc_unreferenced(referenced) + } + BlobStoreMsg::WriteManifest { + manifest, + reply_to, + } => self.handle_write_manifest(ctx, manifest, reply_to), + BlobStoreMsg::ReadManifest { hash, reply_to } => { + self.handle_read_manifest(ctx, hash, reply_to) + } + } + } +} diff --git a/crates/datastore/src/actors/datastore_node.rs b/crates/datastore/src/actors/datastore_node.rs new file mode 100644 index 0000000..3970e0e --- /dev/null +++ b/crates/datastore/src/actors/datastore_node.rs @@ -0,0 +1,282 @@ +//! DatastoreNode — coordinator/facade actor for the datastore stack. +//! +//! Encapsulates the internal actor topology (BlobStoreActor, MetadataActor) +//! behind a single address. Callers send high-level commands (Put, Get, +//! Delete, List, Status) and receive responses. Also routes incoming network +//! protocol messages to the appropriate internal actors. + +use std::collections::BTreeMap; + +use swactor::actor::{ActorAddress, ActorInterface, Ctx}; + +use distribution::types::NodeId; + +use crate::chunking::chunk_blob; +use crate::messages::{ + BlobStoreMsg, DatastoreNodeMsg, DatastoreResponse, MetadataMsg, +}; +use crate::types::{ContentHash, DatastoreConfig, ObjectEntry}; + +/// Top-level coordinator actor for the datastore. +/// +/// Pure router/facade — delegates all work to BlobStoreActor and MetadataActor. +/// Callers interact with a single address instead of knowing about internal actors. +pub struct DatastoreNode { + node_id: NodeId, + blob_store: ActorAddress, + metadata: ActorAddress, + config: DatastoreConfig, +} + +impl DatastoreNode { + pub fn new( + node_id: NodeId, + blob_store: ActorAddress, + metadata: ActorAddress, + config: DatastoreConfig, + ) -> Self { + Self { + node_id, + blob_store, + metadata, + config, + } + } + + fn handle_put( + &self, + ctx: &Ctx, + data: Vec, + name: Option, + tags: BTreeMap, + reply_to: ActorAddress, + ) { + let (content_hash, manifest, chunks) = chunk_blob(&data, self.config.chunk_size); + + // Fire-and-forget chunk writes to BlobStoreActor. + // reply_to: self — ChunkStored responses are silently dropped (type mismatch). + for (hash, chunk_data) in chunks { + let _ = ctx.send( + self.blob_store, + BlobStoreMsg::WriteChunk { + hash, + data: chunk_data, + reply_to: ctx.self_addr(), + }, + ); + } + + // Fire-and-forget manifest write to BlobStoreActor. + let _ = ctx.send( + self.blob_store, + BlobStoreMsg::WriteManifest { + manifest: manifest.clone(), + reply_to: ctx.self_addr(), + }, + ); + + // Build ObjectEntry and send to MetadataActor with caller's reply_to. + let entry = ObjectEntry { + content_hash, + name, + node_id: self.node_id, + tags, + size_bytes: data.len() as u64, + created_at: 0, + }; + + let _ = ctx.send( + self.metadata, + MetadataMsg::PutObject { + entry, + manifest, + reply_to, + }, + ); + } + + fn handle_get(&self, ctx: &Ctx, content_hash: ContentHash, reply_to: ActorAddress) { + let _ = ctx.send( + self.metadata, + MetadataMsg::GetObject { + content_hash, + reply_to, + }, + ); + } + + fn handle_read_chunk(&self, ctx: &Ctx, hash: ContentHash, reply_to: ActorAddress) { + let _ = ctx.send( + self.blob_store, + BlobStoreMsg::ReadChunk { hash, reply_to }, + ); + } + + fn handle_delete(&self, ctx: &Ctx, content_hash: ContentHash, reply_to: ActorAddress) { + let _ = ctx.send( + self.metadata, + MetadataMsg::DeleteObject { + content_hash, + reply_to, + }, + ); + } + + fn handle_list( + &self, + ctx: &Ctx, + name_filter: Option, + all: bool, + reply_to: ActorAddress, + ) { + if all { + let _ = ctx.send( + self.metadata, + MetadataMsg::ListSwarm { + name_filter, + reply_to, + }, + ); + } else { + let _ = ctx.send( + self.metadata, + MetadataMsg::ListLocal { + name_filter, + reply_to, + }, + ); + } + } + + fn handle_status(&self, ctx: &Ctx, reply_to: ActorAddress) { + let _ = ctx.send( + reply_to, + DatastoreResponse::NodeStatus { + node_id: self.node_id, + }, + ); + } + + fn handle_incoming_get_chunk( + &self, + ctx: &Ctx, + request: crate::messages::GetChunkRequest, + reply_to: ActorAddress, + ) { + let _ = ctx.send( + self.blob_store, + BlobStoreMsg::ReadChunk { + hash: request.hash, + reply_to, + }, + ); + } + + fn handle_incoming_get_manifest( + &self, + ctx: &Ctx, + request: crate::messages::GetManifestRequest, + reply_to: ActorAddress, + ) { + let _ = ctx.send( + self.blob_store, + BlobStoreMsg::ReadManifest { + hash: request.hash, + reply_to, + }, + ); + } + + fn handle_incoming_store_object( + &self, + ctx: &Ctx, + request: crate::messages::StoreObjectRequest, + ) { + let _ = ctx.send( + self.metadata, + MetadataMsg::HandleStoreObject { + entry: request.entry, + manifest: None, + }, + ); + } + + fn handle_incoming_find_object( + &self, + ctx: &Ctx, + request: crate::messages::FindObjectRequest, + reply_to: ActorAddress, + ) { + let _ = ctx.send( + self.metadata, + MetadataMsg::HandleFindObject { + from: request.from, + content_hash: request.content_hash, + reply_to, + }, + ); + } + + fn handle_incoming_list_objects( + &self, + ctx: &Ctx, + request: crate::messages::ListObjectsRequest, + reply_to: ActorAddress, + ) { + let _ = ctx.send( + self.metadata, + MetadataMsg::ListLocal { + name_filter: request.name_filter, + reply_to, + }, + ); + } +} + +impl ActorInterface for DatastoreNode { + type Incoming = DatastoreNodeMsg; + type Response = DatastoreResponse; + + fn handle(&mut self, ctx: &Ctx, msg: DatastoreNodeMsg) { + match msg { + DatastoreNodeMsg::Put { + data, + name, + tags, + reply_to, + } => self.handle_put(ctx, data, name, tags, reply_to), + DatastoreNodeMsg::Get { + content_hash, + reply_to, + } => self.handle_get(ctx, content_hash, reply_to), + DatastoreNodeMsg::Delete { + content_hash, + reply_to, + } => self.handle_delete(ctx, content_hash, reply_to), + DatastoreNodeMsg::List { + name_filter, + all, + reply_to, + } => self.handle_list(ctx, name_filter, all, reply_to), + DatastoreNodeMsg::Status { reply_to } => self.handle_status(ctx, reply_to), + DatastoreNodeMsg::ReadChunk { hash, reply_to } => { + self.handle_read_chunk(ctx, hash, reply_to) + } + DatastoreNodeMsg::IncomingGetChunk { request, reply_to } => { + self.handle_incoming_get_chunk(ctx, request, reply_to) + } + DatastoreNodeMsg::IncomingGetManifest { request, reply_to } => { + self.handle_incoming_get_manifest(ctx, request, reply_to) + } + DatastoreNodeMsg::IncomingStoreObject { request } => { + self.handle_incoming_store_object(ctx, request) + } + DatastoreNodeMsg::IncomingFindObject { request, reply_to } => { + self.handle_incoming_find_object(ctx, request, reply_to) + } + DatastoreNodeMsg::IncomingListObjects { request, reply_to } => { + self.handle_incoming_list_objects(ctx, request, reply_to) + } + } + } +} diff --git a/crates/datastore/src/actors/metadata.rs b/crates/datastore/src/actors/metadata.rs new file mode 100644 index 0000000..f93403b --- /dev/null +++ b/crates/datastore/src/actors/metadata.rs @@ -0,0 +1,339 @@ +//! MetadataActor — object metadata index with DHT overlay. +//! +//! Owns: +//! - Local object index: `HashMap` keyed by content hash +//! - Dissemination queue for DHT replication (reuses `ClusterRegistry` pattern) +//! +//! The metadata DHT is a separate Kademlia overlay from the actor directory. +//! Objects are keyed by `blake3(blob_bytes)` — the content hash of the entire blob. + +use std::collections::{HashMap, HashSet}; + +use swactor::actor::{ActorAddress, ActorInterface, Ctx}; + +use distribution::types::NodeId; + +use crate::messages::{BlobStoreMsg, DatastoreResponse, MetadataMsg}; +use crate::types::{ContentHash, DatastoreConfig, ObjectEntry, ObjectManifest}; + +// ─── Dissemination entry (reuses registry.rs pattern) ─────────────────────── + +#[derive(Debug, Clone)] +struct DisseminationEntry { + entry: ObjectEntry, + manifest: Option, + remaining: usize, +} + +// ─── MetadataActor ────────────────────────────────────────────────────────── + +/// Manages the object metadata index for a single node. +pub struct MetadataActor { + /// This node's identity. + node_id: NodeId, + /// Local object index: content_hash → ObjectEntry. + entries: HashMap, + /// Local manifest cache: content_hash → ObjectManifest. + manifests: HashMap, + /// Pending entries to disseminate to DHT peers. + dissemination: Vec, + /// Tick counter for periodic GC. + tick_count: u64, + /// Configuration. + gc_interval: u64, + /// Dissemination multiplier (Λ) — same role as in SWIM. + dissemination_lambda: usize, + /// Address of the local BlobStoreActor (for forwarding manifest writes). + blob_store_addr: Option, + /// Addresses of peer MetadataActors for epidemic dissemination. + peers: Vec, +} + +impl MetadataActor { + pub fn new(node_id: NodeId, config: &DatastoreConfig) -> Self { + Self { + node_id, + entries: HashMap::new(), + manifests: HashMap::new(), + dissemination: Vec::new(), + tick_count: 0, + gc_interval: config.gc_interval, + dissemination_lambda: 3, + blob_store_addr: None, + peers: Vec::new(), + } + } + + /// Set the address of the co-located BlobStoreActor. + pub fn set_blob_store(&mut self, addr: ActorAddress) { + self.blob_store_addr = Some(addr); + } + + fn transmit_budget(&self, cluster_size: usize) -> usize { + let n = cluster_size.max(2) as f64; + let log_n = n.log2().ceil() as usize; + self.dissemination_lambda * log_n.max(1) + } + + fn enqueue(&mut self, entry: ObjectEntry, manifest: Option, cluster_size: usize) { + let budget = self.transmit_budget(cluster_size); + + // Replace existing entry for same content hash if present. + if let Some(existing) = self + .dissemination + .iter_mut() + .find(|e| e.entry.content_hash == entry.content_hash) + { + existing.entry = entry; + if manifest.is_some() { + existing.manifest = manifest; + } + existing.remaining = budget; + return; + } + + self.dissemination.push(DisseminationEntry { + entry, + manifest, + remaining: budget, + }); + } + + /// Take pending entries for dissemination, up to `max_count`. + /// Returns `(ObjectEntry, Option)` pairs. + pub fn take_pending(&mut self, max_count: usize) -> Vec<(ObjectEntry, Option)> { + let count = max_count.min(self.dissemination.len()); + let mut result = Vec::with_capacity(count); + + for entry in self.dissemination.iter_mut().take(count) { + result.push((entry.entry.clone(), entry.manifest.clone())); + entry.remaining = entry.remaining.saturating_sub(1); + } + + // Evict exhausted entries. + self.dissemination.retain(|e| e.remaining > 0); + + result + } + + /// Periodic GC: build referenced chunk set from all manifests and send + /// `GcUnreferenced` to BlobStoreActor to delete orphaned chunks. + fn gc_tick(&mut self, ctx: &Ctx) { + self.tick_count += 1; + if self.tick_count % self.gc_interval != 0 { + return; + } + + let blob_store_addr = match self.blob_store_addr { + Some(addr) => addr, + None => return, + }; + + let mut referenced = HashSet::new(); + for manifest in self.manifests.values() { + for chunk_ref in &manifest.chunks { + referenced.insert(chunk_ref.hash); + } + } + + let _ = ctx.send(blob_store_addr, BlobStoreMsg::GcUnreferenced { referenced }); + } + + // ─── Message handlers ─────────────────────────────────────────────── + + fn handle_put_object( + &mut self, + ctx: &Ctx, + entry: ObjectEntry, + manifest: ObjectManifest, + reply_to: ActorAddress, + ) { + let content_hash = entry.content_hash; + + // Store manifest locally. + self.manifests.insert(content_hash, manifest.clone()); + + // Insert entry keyed by content hash. + let mut entry = entry; + entry.node_id = self.node_id; + self.entries.insert(content_hash, entry.clone()); + + // Enqueue for DHT dissemination (include manifest for peer replication). + self.enqueue(entry, Some(manifest), 3); + + let _ = ctx.send( + reply_to, + DatastoreResponse::PutOk { content_hash }, + ); + } + + fn handle_get_object(&self, ctx: &Ctx, content_hash: ContentHash, reply_to: ActorAddress) { + match self.entries.get(&content_hash) { + Some(entry) => { + if let Some(manifest) = self.manifests.get(&content_hash) { + let _ = ctx.send( + reply_to, + DatastoreResponse::GetOk { + entry: entry.clone(), + manifest: manifest.clone(), + }, + ); + } else { + let _ = ctx.send( + reply_to, + DatastoreResponse::Error { + reason: format!("manifest not found for content hash: {content_hash}"), + }, + ); + } + } + None => { + let _ = ctx.send(reply_to, DatastoreResponse::NotFound); + } + } + } + + fn handle_delete_object(&mut self, ctx: &Ctx, content_hash: ContentHash, reply_to: ActorAddress) { + if self.entries.remove(&content_hash).is_some() { + self.manifests.remove(&content_hash); + let _ = ctx.send(reply_to, DatastoreResponse::DeleteOk { content_hash }); + } else { + let _ = ctx.send(reply_to, DatastoreResponse::NotFound); + } + } + + fn handle_list_local( + &self, + ctx: &Ctx, + name_filter: Option, + reply_to: ActorAddress, + ) { + let entries: Vec = self + .entries + .values() + .filter(|e| { + match (&name_filter, &e.name) { + (Some(filter), Some(name)) => name.contains(filter.as_str()), + (Some(_), None) => false, + (None, _) => true, + } + }) + .cloned() + .collect(); + + let _ = ctx.send(reply_to, DatastoreResponse::ListOk { entries }); + } + + fn handle_list_swarm( + &self, + ctx: &Ctx, + name_filter: Option, + reply_to: ActorAddress, + ) { + // Delegates to local index. Swarm-wide fan-out to peer MetadataActors + // will be wired when networking integration is added. + self.handle_list_local(ctx, name_filter, reply_to); + } + + fn handle_find_object( + &self, + ctx: &Ctx, + _from: NodeId, + content_hash: ContentHash, + reply_to: ActorAddress, + ) { + match self.entries.get(&content_hash) { + Some(entry) => { + let _ = ctx.send( + reply_to, + DatastoreResponse::GetOk { + entry: entry.clone(), + manifest: self + .manifests + .get(&content_hash) + .cloned() + .unwrap_or_else(|| ObjectManifest { + content_hash, + chunks: vec![], + total_size: entry.size_bytes, + chunk_size: 0, + content_type: None, + }), + }, + ); + } + None => { + let _ = ctx.send(reply_to, DatastoreResponse::NotFound); + } + } + } + + fn handle_set_peers(&mut self, peers: Vec) { + self.peers = peers; + } + + fn handle_disseminate_tick(&mut self, ctx: &Ctx) { + if self.peers.is_empty() { + return; + } + let pending = self.take_pending(10); + for (entry, manifest) in pending { + for &peer in &self.peers { + let _ = ctx.send(peer, MetadataMsg::HandleStoreObject { + entry: entry.clone(), + manifest: manifest.clone(), + }); + } + } + } + + fn handle_store_object(&mut self, entry: ObjectEntry, manifest: Option) { + // Insert if absent — content-addressed entries don't conflict. + let content_hash = entry.content_hash; + if !self.entries.contains_key(&content_hash) { + if let Some(ref m) = manifest { + self.manifests.insert(content_hash, m.clone()); + } + self.entries.insert(content_hash, entry.clone()); + self.enqueue(entry, manifest, 3); + } + } +} + +impl ActorInterface for MetadataActor { + type Incoming = MetadataMsg; + type Response = DatastoreResponse; + + fn handle(&mut self, ctx: &Ctx, msg: MetadataMsg) { + match msg { + MetadataMsg::PutObject { + entry, + manifest, + reply_to, + } => self.handle_put_object(ctx, entry, manifest, reply_to), + MetadataMsg::GetObject { content_hash, reply_to } => { + self.handle_get_object(ctx, content_hash, reply_to) + } + MetadataMsg::DeleteObject { content_hash, reply_to } => { + self.handle_delete_object(ctx, content_hash, reply_to) + } + MetadataMsg::ListLocal { name_filter, reply_to } => { + self.handle_list_local(ctx, name_filter, reply_to) + } + MetadataMsg::ListSwarm { name_filter, reply_to } => { + self.handle_list_swarm(ctx, name_filter, reply_to) + } + MetadataMsg::HandleFindObject { + from, + content_hash, + reply_to, + } => self.handle_find_object(ctx, from, content_hash, reply_to), + MetadataMsg::HandleStoreObject { entry, manifest } => { + self.handle_store_object(entry, manifest) + } + MetadataMsg::SetPeers { peers } => self.handle_set_peers(peers), + MetadataMsg::DisseminateTick => self.handle_disseminate_tick(ctx), + MetadataMsg::GcTick => self.gc_tick(ctx), + } + } +} diff --git a/crates/datastore/src/actors/mod.rs b/crates/datastore/src/actors/mod.rs new file mode 100644 index 0000000..554c243 --- /dev/null +++ b/crates/datastore/src/actors/mod.rs @@ -0,0 +1,9 @@ +pub mod blob_store; +pub mod datastore_node; +pub mod metadata; +pub mod transfer; + +pub use blob_store::BlobStoreActor; +pub use datastore_node::DatastoreNode; +pub use metadata::MetadataActor; +pub use transfer::TransferActor; diff --git a/crates/datastore/src/actors/transfer.rs b/crates/datastore/src/actors/transfer.rs new file mode 100644 index 0000000..774f1b0 --- /dev/null +++ b/crates/datastore/src/actors/transfer.rs @@ -0,0 +1,180 @@ +//! TransferActor — ephemeral actor for downloading an object from a remote node. +//! +//! One TransferActor is spawned per download. It walks the manifest's chunk list, +//! requests each chunk from the source node's BlobStoreActor, forwards received +//! chunks to the local BlobStoreActor for persistence, and replies to the +//! original requester when all chunks are received (or on failure). +//! +//! Sequential chunk fetching for MVP (parallel fetching planned for later). +//! Self-terminates via `ctx.stop_self()` on completion, failure, or cancel. + +use std::collections::HashSet; + +use swactor::actor::{ActorAddress, ActorInterface, Ctx}; + +use distribution::types::NodeId; + +use crate::messages::{DatastoreResponse, TransferMsg}; +use crate::types::{ContentHash, ObjectManifest, TransferStatus}; + +/// Ephemeral actor that manages a single object download. +pub struct TransferActor { + /// The manifest describing which chunks to download. + manifest: Option, + /// The remote node to fetch chunks from. + source_node: Option, + /// Address to send the final result to. + reply_to: Option, + /// Address of the local BlobStoreActor for storing received chunks. + blob_store_addr: ActorAddress, + /// Chunk hashes still pending download. + pending: HashSet, + /// Chunk hashes successfully received and stored. + received: HashSet, + /// Current status of the transfer. + status: TransferStatus, + /// Number of retry attempts per chunk. + max_retries: usize, + /// Tracks which chunks have been retried and how many times. + retry_counts: std::collections::HashMap, +} + +impl TransferActor { + /// Create a new transfer actor. + /// + /// `blob_store_addr` is the address of the local BlobStoreActor where + /// downloaded chunks will be persisted. + pub fn new(blob_store_addr: ActorAddress) -> Self { + Self { + manifest: None, + source_node: None, + reply_to: None, + blob_store_addr, + pending: HashSet::new(), + received: HashSet::new(), + status: TransferStatus::Downloading { + chunks_received: 0, + chunks_total: 0, + }, + max_retries: 1, + retry_counts: std::collections::HashMap::new(), + } + } + + fn handle_start_download( + &mut self, + _ctx: &Ctx, + manifest: ObjectManifest, + source_node: NodeId, + reply_to: ActorAddress, + ) { + let total = manifest.chunks.len(); + self.pending = manifest.chunks.iter().map(|c| c.hash).collect(); + self.manifest = Some(manifest); + self.source_node = Some(source_node); + self.reply_to = Some(reply_to); + self.status = TransferStatus::Downloading { + chunks_received: 0, + chunks_total: total, + }; + + // Chunks are fed externally via ChunkReceived/ChunkFailed messages. + // In production, a network adapter (or DatastoreNode) reads chunks from + // the remote BlobStoreActor and forwards them here. In simulation, the + // test harness plays this role. + } + + fn handle_chunk_received( + &mut self, + ctx: &Ctx, + hash: ContentHash, + data: Vec, + ) { + if !self.pending.remove(&hash) { + return; // Duplicate or unexpected chunk. + } + + // Forward chunk to local BlobStoreActor for persistence. + let _ = ctx.send( + self.blob_store_addr, + crate::messages::BlobStoreMsg::WriteChunk { + hash, + data, + reply_to: ctx.self_addr(), + }, + ); + + self.received.insert(hash); + + let total = self.received.len() + self.pending.len(); + self.status = TransferStatus::Downloading { + chunks_received: self.received.len(), + chunks_total: total, + }; + + // Check if all chunks are received. + if self.pending.is_empty() { + self.status = TransferStatus::Complete; + if let (Some(manifest), Some(reply_to)) = (&self.manifest, self.reply_to) { + let _ = ctx.send( + reply_to, + DatastoreResponse::TransferComplete { + content_hash: manifest.content_hash, + }, + ); + } + ctx.stop_self(); + } + } + + fn handle_chunk_failed( + &mut self, + ctx: &Ctx, + hash: ContentHash, + reason: String, + ) { + let retries = self.retry_counts.entry(hash).or_insert(0); + if *retries < self.max_retries { + *retries += 1; + // Retry is tracked; the external driver (network adapter or test + // harness) is expected to re-send the chunk on retry. + return; + } + + // Exhausted retries — fail the whole transfer. + self.status = TransferStatus::Failed { + reason: reason.clone(), + }; + if let Some(reply_to) = self.reply_to { + let _ = ctx.send(reply_to, DatastoreResponse::TransferFailed { reason }); + } + ctx.stop_self(); + } + + fn handle_cancel(&mut self, ctx: &Ctx) { + self.status = TransferStatus::Cancelled; + ctx.stop_self(); + } +} + +impl ActorInterface for TransferActor { + type Incoming = TransferMsg; + type Response = DatastoreResponse; + + fn handle(&mut self, ctx: &Ctx, msg: TransferMsg) { + match msg { + TransferMsg::StartDownload { + manifest, + source_node, + reply_to, + } => self.handle_start_download(ctx, manifest, source_node, reply_to), + TransferMsg::ChunkReceived { hash, data } => { + self.handle_chunk_received(ctx, hash, data) + } + TransferMsg::ChunkFailed { hash, reason } => { + self.handle_chunk_failed(ctx, hash, reason) + } + TransferMsg::Cancel => self.handle_cancel(ctx), + } + } +} diff --git a/crates/datastore/src/api.rs b/crates/datastore/src/api.rs new file mode 100644 index 0000000..28046fc --- /dev/null +++ b/crates/datastore/src/api.rs @@ -0,0 +1,793 @@ +//! 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::metrics::DatastoreMetrics; +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>>, + metrics: 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_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) + .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 body_len = body.len(); + + 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 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 }) => { + 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 }) => { + state.metrics.record_get(&content_hash.to_hex()); + 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(), + }, + ); + + 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) => { + // 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 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) => { + 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 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::() { + 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)); + state.metrics.advance_transfer(&hash_hex); + } + _ => { + all_ok = false; + break; + } + } + } + + state.metrics.end_transfer(&hash_hex); + + 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, + metrics: Arc, +) -> (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), + metrics, + }); + + 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), + ("GET", "/") => respond_html(request), + _ => { + 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..f8e6c6e --- /dev/null +++ b/crates/datastore/src/bin/store_node.rs @@ -0,0 +1,254 @@ +//! 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 serde::Deserialize; + +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::metrics::DatastoreMetrics; +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 { + /// Path to a TOML config file + #[arg(long)] + config: Option, + + /// HTTP API port + #[arg(long)] + port: Option, + + /// 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)] + chunk_size: Option, + + /// GC interval in ticks (each tick is ~100ms) + #[arg(long)] + gc_interval: Option, + + /// Dissemination interval in ticks + #[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 + { + 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 = cfg.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: cfg.chunk_size, + storage_path: cfg + .storage_path + .as_ref() + .map(|s| s.into()) + .unwrap_or_else(|| "datastore".into()), + gc_interval: cfg.gc_interval, + ..Default::default() + }; + + // Create storage backend + 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"); + 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"); + + // 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 + let (api_shutdown, _peers) = start_api_server( + handle.runtime.clone(), + datastore_addr, + metadata_addr, + blob_store_addr, + cfg.port, + Arc::clone(&metrics), + ); + + eprintln!("Node {} started", &node_hex[..8]); + 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 cfg.storage_path.is_some() { + eprintln!("Storage: {}", cfg.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 % cfg.gc_interval == 0 { + let _ = handle + .runtime + .send_to(metadata_addr, MetadataMsg::GcTick); + } + + if round % cfg.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/chunking.rs b/crates/datastore/src/chunking.rs new file mode 100644 index 0000000..33035b8 --- /dev/null +++ b/crates/datastore/src/chunking.rs @@ -0,0 +1,138 @@ +//! Chunking engine — pure functions for splitting blobs into content-addressed +//! chunks and reassembling them. +//! +//! No I/O. All functions are deterministic and side-effect-free. + +use crate::types::{ChunkRef, ContentHash, ObjectManifest}; + +/// Errors that can occur during chunk reassembly. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum ChunkingError { + /// A chunk referenced by the manifest was not provided. + MissingChunk { hash: ContentHash }, + /// The reassembled data size does not match the manifest's `total_size`. + SizeMismatch { expected: u64, actual: u64 }, + /// The reassembled data's content hash does not match the manifest's `content_hash`. + HashMismatch { + expected: ContentHash, + actual: ContentHash, + }, +} + +impl std::fmt::Display for ChunkingError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + ChunkingError::MissingChunk { hash } => write!(f, "missing chunk: {hash}"), + ChunkingError::SizeMismatch { expected, actual } => { + write!(f, "size mismatch: expected {expected}, got {actual}") + } + ChunkingError::HashMismatch { expected, actual } => { + write!(f, "hash mismatch: expected {expected}, got {actual}") + } + } + } +} + +impl std::error::Error for ChunkingError {} + +/// Split a blob into fixed-size chunks and produce a manifest. +/// +/// Returns `(content_hash, manifest, chunks)` where: +/// - `content_hash` is `blake3(data)` — the whole-blob hash +/// - `manifest` describes the chunk layout +/// - `chunks` is a vec of `(chunk_hash, chunk_bytes)` pairs +/// +/// `chunk_size` must be > 0. +pub fn chunk_blob( + data: &[u8], + chunk_size: u32, +) -> (ContentHash, ObjectManifest, Vec<(ContentHash, Vec)>) { + assert!(chunk_size > 0, "chunk_size must be > 0"); + + let content_hash = ContentHash::of(data); + let mut chunks = Vec::new(); + let mut chunk_refs = Vec::new(); + let mut offset: u64 = 0; + + if data.is_empty() { + let manifest = ObjectManifest { + content_hash, + chunks: chunk_refs, + total_size: 0, + chunk_size, + content_type: None, + }; + return (content_hash, manifest, chunks); + } + + for chunk_data in data.chunks(chunk_size as usize) { + let hash = ContentHash::of(chunk_data); + chunk_refs.push(ChunkRef { + hash, + offset, + size: chunk_data.len() as u32, + }); + chunks.push((hash, chunk_data.to_vec())); + offset += chunk_data.len() as u64; + } + + let manifest = ObjectManifest { + content_hash, + chunks: chunk_refs, + total_size: data.len() as u64, + chunk_size, + content_type: None, + }; + + (content_hash, manifest, chunks) +} + +/// Reassemble a blob from its manifest and chunk data. +/// +/// Chunks are looked up by hash from the provided slice. The manifest's +/// `chunks` field determines the ordering. Verifies total size and +/// content hash after reassembly. +pub fn reassemble_blob( + manifest: &ObjectManifest, + chunks: &[(ContentHash, Vec)], +) -> Result, ChunkingError> { + let mut result = Vec::with_capacity(manifest.total_size as usize); + + for chunk_ref in &manifest.chunks { + let chunk_data = chunks + .iter() + .find(|(h, _)| *h == chunk_ref.hash) + .map(|(_, d)| d); + + match chunk_data { + Some(data) => result.extend_from_slice(data), + None => { + return Err(ChunkingError::MissingChunk { + hash: chunk_ref.hash, + }) + } + } + } + + if result.len() as u64 != manifest.total_size { + return Err(ChunkingError::SizeMismatch { + expected: manifest.total_size, + actual: result.len() as u64, + }); + } + + let actual_hash = ContentHash::of(&result); + if actual_hash != manifest.content_hash { + return Err(ChunkingError::HashMismatch { + expected: manifest.content_hash, + actual: actual_hash, + }); + } + + Ok(result) +} + +/// Verify that `data` hashes to `expected`. +pub fn verify_integrity(data: &[u8], expected: &ContentHash) -> bool { + ContentHash::of(data) == *expected +} diff --git a/crates/datastore/src/cli.rs b/crates/datastore/src/cli.rs new file mode 100644 index 0000000..1dbc058 --- /dev/null +++ b/crates/datastore/src/cli.rs @@ -0,0 +1,72 @@ +//! CLI command type definitions for `swactor-store`. +//! +//! Types only — no implementation. These define the CLI interface that will +//! be wired to the actor system in a future milestone. + +use std::collections::BTreeMap; +use std::path::PathBuf; + +use distribution::types::NodeId; + +/// Top-level CLI commands for `swactor-store`. +#[derive(Debug, Clone)] +pub enum CliCommand { + /// Store a local file as a distributed object. + /// + /// ```text + /// swactor-store put [--name