zacheryasc/swactor
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

datastore #41

Merged
zacheryasc merged 3 commits from datastore into master 2026-02-15 17:03:32 +00:00
Conversation 0 Commits 3 Files changed 57 +1462 -27
10 changed files with 1462 additions and 27 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files
Showing only changes of commit ebf778fc3f - Show all commits

72
Cargo.lock generated
View file

@ -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"

19
crates/datastore/Cargo.toml
View file

@ -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"]

761
crates/datastore/src/api.rs Normal file
View file

@ -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<Runtime>,
datastore_addr: ActorAddress,
metadata_addr: ActorAddress,
blob_store_addr: ActorAddress,
peers: Arc<Mutex<Vec<PeerInfo>>>,
}
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<DatastoreResponse>, timeout: Duration) -> Option<DatastoreResponse> {
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::<tiny_http::Header>()
.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::<tiny_http::Header>()
.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::<tiny_http::Header>()
.unwrap(),
);
let _ = request.respond(response);
}
fn parse_query_string(url: &str) -> BTreeMap<String, String> {
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<u8> {
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<serde_json::Value> = 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<serde_json::Value> {
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 &params {
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::<DatastoreResponse>() {
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::<DatastoreResponse>() {
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::<DatastoreResponse>() {
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::<DatastoreResponse>() {
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::<DatastoreResponse>() {
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<String>,
state: &ApiState,
) {
let inbox = match state.runtime.new_inbox::<DatastoreResponse>() {
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<String>,
state: &ApiState,
) {
let mut all_entries = Vec::new();
// Query local
let inbox = match state.runtime.new_inbox::<DatastoreResponse>() {
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::<DatastoreResponse>() {
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::<DatastoreResponse>() {
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::<DatastoreResponse>().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::<DatastoreResponse>().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::<DatastoreResponse>() {
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::<DatastoreResponse>() {
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::<DatastoreResponse>() {
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::<DatastoreResponse>() {
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<Runtime>,
datastore_addr: ActorAddress,
metadata_addr: ActorAddress,
blob_store_addr: ActorAddress,
port: u16,
) -> (Arc<AtomicBool>, Arc<Mutex<Vec<PeerInfo>>>) {
let shutdown = Arc::new(AtomicBool::new(false));
let peers: Arc<Mutex<Vec<PeerInfo>>> = 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)
}

329
crates/datastore/src/bin/store_cli.rs Normal file
View file

@ -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<String>,
},
/// Retrieve object metadata (or download with --output)
Get {
/// Content hash (hex)
hash: String,
/// Download file to this path
#[arg(long)]
output: Option<PathBuf>,
},
/// Delete an object
Delete {
/// Content hash (hex)
hash: String,
},
/// List stored objects
List {
/// Filter by name substring
#[arg(long)]
name: Option<String>,
/// 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
}

200
crates/datastore/src/bin/store_node.rs Normal file
View file

@ -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<String>,
/// Dashboard HTTP port (omit to disable dashboard)
#[arg(long)]
dashboard_port: Option<u16>,
/// 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<dyn swactor_datastore::StorageBackend> = 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();
}

2
crates/datastore/src/lib.rs
View file

@ -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};

24
crates/datastore/src/storage/mod.rs
View file

@ -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<ContentHash> {
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<u8> {
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,
}
}

24
crates/datastore/src/types.rs
View file

@ -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<Self> {
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<u8> {
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)`.

58
crates/datastore/tests/datastore_tests.rs
View file

@ -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::<u8>(), 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]

0
crates/datastore/PROTOCOL.md → docs/development_history/datastore/PROTOCOL.md
View file