From 523b0d5545215969e6653e239bad70db65464663 Mon Sep 17 00:00:00 2001 From: Zachery Aaron Shores-Chmielewski Date: Mon, 16 Feb 2026 22:25:17 +0700 Subject: [PATCH] feat: dev-node in xtask --- Cargo.lock | 13 + Cargo.toml | 1 + crates/datastore/src/bridge.rs | 407 +++++++++++++++ crates/datastore/src/lib.rs | 2 + .../src/datastore_collector.rs | 61 ++- .../runtime-dashboard/src/datastore_html.rs | 335 +++++++++++- crates/runtime-dashboard/src/lib.rs | 16 + crates/runtime-dashboard/src/server.rs | 284 +++++++++- crates/swactor-node/Cargo.toml | 22 + crates/swactor-node/src/main.rs | 491 ++++++++++++++++++ xtask/src/main.rs | 152 +++++- 11 files changed, 1736 insertions(+), 48 deletions(-) create mode 100644 crates/datastore/src/bridge.rs create mode 100644 crates/swactor-node/Cargo.toml create mode 100644 crates/swactor-node/src/main.rs diff --git a/Cargo.lock b/Cargo.lock index 7da30f8..4c8d0cf 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4647,6 +4647,19 @@ dependencies = [ "ureq", ] +[[package]] +name = "swactor-node" +version = "0.1.0" +dependencies = [ + "clap", + "ctrlc", + "distribution", + "iroh", + "runtime-dashboard", + "swactor", + "swactor-datastore", +] + [[package]] name = "swactor-std" version = "0.1.0" diff --git a/Cargo.toml b/Cargo.toml index 3fb1f43..7787d60 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -11,6 +11,7 @@ members = [ "crates/datastore", "crates/shared-types", "crates/crypto-wasm", + "crates/swactor-node", "tests/docker", "crates/ci", "crates/local-runner", diff --git a/crates/datastore/src/bridge.rs b/crates/datastore/src/bridge.rs new file mode 100644 index 0000000..a79e757 --- /dev/null +++ b/crates/datastore/src/bridge.rs @@ -0,0 +1,407 @@ +//! Bridge between the runtime dashboard's `DatastoreStatsProvider` trait and +//! the datastore actor system. Allows the dashboard to perform CRUD operations +//! and lifecycle management without depending on `swactor-datastore` types. + +use std::collections::BTreeMap; +use std::sync::Arc; +use std::thread; +use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; + +use swactor::actor::ActorAddress; +use swactor::runtime::{Inbox, Runtime}; + +use distribution::types::NodeId; +use runtime_dashboard::datastore_collector::{ + DatastoreFactory, DatastoreStatsProvider, ListScope, +}; + +use crate::actors::{BlobStoreActor, DatastoreNode, MetadataActor}; +use crate::chunking::reassemble_blob; +use crate::messages::{DatastoreNodeMsg, DatastoreResponse, MetadataMsg}; +use crate::metrics::DatastoreMetrics; +use crate::storage::{FilesystemBackend, InMemoryBackend}; +use crate::types::{ContentHash, DatastoreConfig}; + +const POLL_TIMEOUT: Duration = Duration::from_secs(5); +const POLL_INTERVAL: Duration = Duration::from_millis(1); + +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 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() +} + +/// Bridges the dashboard trait to the datastore actor system. +pub struct DatastoreBridge { + metrics: Arc, + runtime: Arc, + datastore_addr: ActorAddress, + metadata_addr: ActorAddress, + #[allow(dead_code)] + blob_store_addr: ActorAddress, +} + +impl DatastoreBridge { + pub fn new( + metrics: Arc, + runtime: Arc, + datastore_addr: ActorAddress, + metadata_addr: ActorAddress, + blob_store_addr: ActorAddress, + ) -> Self { + Self { + metrics, + runtime, + datastore_addr, + metadata_addr, + blob_store_addr, + } + } +} + +impl DatastoreStatsProvider for DatastoreBridge { + fn snapshot_json(&self) -> Option { + let snap = self.metrics.snapshot(); + serde_json::to_string(&snap).ok() + } + + fn is_running(&self) -> bool { + true + } + + fn list_objects(&self, name_filter: Option<&str>, scope: ListScope) -> Result { + let inbox = self.runtime.new_inbox::() + .map_err(|e| format!("failed to create inbox: {e}"))?; + + match scope { + ListScope::Local => { + let _ = self.runtime.send_to( + self.datastore_addr, + DatastoreNodeMsg::List { + name_filter: name_filter.map(|s| s.to_string()), + all: false, + reply_to: *inbox.addr(), + }, + ); + } + ListScope::Swarm => { + let _ = self.runtime.send_to( + self.metadata_addr, + MetadataMsg::ListLocal { + name_filter: name_filter.map(|s| s.to_string()), + 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(); + Ok(json) + } + Some(DatastoreResponse::Error { reason }) => Err(reason), + _ => Err("timeout".into()), + } + } + + fn get_object(&self, hash: &str) -> Result { + let content_hash = ContentHash::from_hex(hash) + .ok_or_else(|| "invalid content hash hex".to_string())?; + + let inbox = self.runtime.new_inbox::() + .map_err(|e| format!("failed to create inbox: {e}"))?; + + let _ = self.runtime.send_to( + self.datastore_addr, + DatastoreNodeMsg::Get { + content_hash, + reply_to: *inbox.addr(), + }, + ); + + match poll_response(&inbox, POLL_TIMEOUT) { + Some(DatastoreResponse::GetOk { entry, manifest }) => { + self.metrics.record_get(&content_hash.to_hex()); + let json = serde_json::json!({ + "entry": entry_to_json(&entry), + "manifest": manifest_to_json(&manifest), + }) + .to_string(); + Ok(json) + } + Some(DatastoreResponse::NotFound) => Err("not found".into()), + Some(DatastoreResponse::Error { reason }) => Err(reason), + _ => Err("timeout".into()), + } + } + + fn get_data(&self, hash: &str) -> Result, String> { + let content_hash = ContentHash::from_hex(hash) + .ok_or_else(|| "invalid content hash hex".to_string())?; + + // Get manifest + let inbox = self.runtime.new_inbox::() + .map_err(|e| format!("failed to create inbox: {e}"))?; + + let _ = self.runtime.send_to( + self.datastore_addr, + DatastoreNodeMsg::Get { + content_hash, + reply_to: *inbox.addr(), + }, + ); + + self.metrics.record_get(&content_hash.to_hex()); + + let manifest = match poll_response(&inbox, POLL_TIMEOUT) { + Some(DatastoreResponse::GetOk { manifest, .. }) => manifest, + Some(DatastoreResponse::NotFound) => return Err("not found".into()), + Some(DatastoreResponse::Error { reason }) => return Err(reason), + _ => return Err("timeout".into()), + }; + + // Read chunks + let mut chunk_data = Vec::new(); + for chunk_ref in &manifest.chunks { + let chunk_inbox = self.runtime.new_inbox::() + .map_err(|e| format!("failed to create inbox: {e}"))?; + + let _ = self.runtime.send_to( + self.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)); + } + _ => return Err("failed to read chunk".into()), + } + } + + reassemble_blob(&manifest, &chunk_data) + .map_err(|e| format!("reassembly failed: {e:?}")) + } + + fn put_data(&self, data: Vec, name: Option) -> Result { + let body_len = data.len(); + + let inbox = self.runtime.new_inbox::() + .map_err(|e| format!("failed to create inbox: {e}"))?; + + let _ = self.runtime.send_to( + self.datastore_addr, + DatastoreNodeMsg::Put { + data, + name: name.clone(), + tags: BTreeMap::new(), + reply_to: *inbox.addr(), + }, + ); + + match poll_response(&inbox, POLL_TIMEOUT) { + Some(DatastoreResponse::PutOk { content_hash }) => { + let hex = content_hash.to_hex(); + self.metrics.record_put(&hex, name.as_deref(), body_len as u64); + let json = serde_json::json!({ "content_hash": hex }).to_string(); + Ok(json) + } + Some(DatastoreResponse::Error { reason }) => Err(reason), + _ => Err("timeout waiting for put response".into()), + } + } + + fn delete_object(&self, hash: &str) -> Result { + let content_hash = ContentHash::from_hex(hash) + .ok_or_else(|| "invalid content hash hex".to_string())?; + + let inbox = self.runtime.new_inbox::() + .map_err(|e| format!("failed to create inbox: {e}"))?; + + let _ = self.runtime.send_to( + self.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(); + self.metrics.record_delete(&hex, 0); + let json = serde_json::json!({ "content_hash": hex }).to_string(); + Ok(json) + } + Some(DatastoreResponse::NotFound) => Err("not found".into()), + Some(DatastoreResponse::Error { reason }) => Err(reason), + _ => Err("timeout".into()), + } + } + + fn node_status(&self) -> Result { + let inbox = self.runtime.new_inbox::() + .map_err(|e| format!("failed to create inbox: {e}"))?; + + let _ = self.runtime.send_to( + self.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(); + Ok(json) + } + _ => Err("timeout".into()), + } + } + + fn shutdown_datastore(&self) -> Result<(), String> { + // We can't actually stop the actors from here without a runtime handle, + // but we can signal shutdown. The caller (server handler) clears the + // provider reference which effectively disables the datastore. + Ok(()) + } +} + +/// Factory that can spawn a new set of datastore actors on a shared runtime. +pub struct DatastoreNodeFactory { + runtime: Arc, + default_chunk_size: u32, +} + +impl DatastoreNodeFactory { + pub fn new(runtime: Arc, default_chunk_size: u32) -> Self { + Self { + runtime, + default_chunk_size, + } + } +} + +impl DatastoreFactory for DatastoreNodeFactory { + fn start_datastore( + &self, + storage_path: Option, + ) -> Result, String> { + // Generate a unique node ID + let node_id = { + let mut bytes = [0u8; 32]; + let nanos = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_nanos(); + for (i, b) in nanos.to_le_bytes().iter().enumerate() { + bytes[i % 32] ^= *b; + } + let pid = std::process::id(); + for (i, b) in pid.to_le_bytes().iter().enumerate() { + bytes[i + 16] ^= *b; + } + NodeId(bytes) + }; + + let config = DatastoreConfig { + chunk_size: self.default_chunk_size, + storage_path: storage_path + .as_ref() + .map(|s| s.into()) + .unwrap_or_else(|| "datastore".into()), + ..Default::default() + }; + + let backend: Box = match &storage_path { + Some(path) => { + let p = std::path::PathBuf::from(path); + std::fs::create_dir_all(&p) + .map_err(|e| format!("failed to create storage directory: {e}"))?; + Box::new(FilesystemBackend::new(p)) + } + None => Box::new(InMemoryBackend::new()), + }; + + let blob_store_addr = self + .runtime + .spawn(BlobStoreActor::new(backend)) + .map_err(|e| format!("failed to spawn BlobStoreActor: {e}"))?; + + let mut metadata = MetadataActor::new(node_id, &config); + metadata.set_blob_store(blob_store_addr); + let metadata_addr = self + .runtime + .spawn(metadata) + .map_err(|e| format!("failed to spawn MetadataActor: {e}"))?; + + let datastore_node = DatastoreNode::new(node_id, blob_store_addr, metadata_addr, config); + let datastore_addr = self + .runtime + .spawn(datastore_node) + .map_err(|e| format!("failed to spawn DatastoreNode: {e}"))?; + + 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); + + let bridge = DatastoreBridge::new( + metrics, + Arc::clone(&self.runtime), + datastore_addr, + metadata_addr, + blob_store_addr, + ); + + Ok(Arc::new(bridge)) + } +} diff --git a/crates/datastore/src/lib.rs b/crates/datastore/src/lib.rs index 1768caa..b3f928b 100644 --- a/crates/datastore/src/lib.rs +++ b/crates/datastore/src/lib.rs @@ -10,6 +10,8 @@ pub mod metrics; pub mod api; #[cfg(feature = "node")] pub mod ui_html; +#[cfg(feature = "node")] +pub mod bridge; pub use types::{ChunkRef, ContentHash, DatastoreConfig, ObjectEntry, ObjectManifest}; pub use messages::{BlobStoreMsg, DatastoreNodeMsg, DatastoreResponse, MetadataMsg, TransferMsg}; diff --git a/crates/runtime-dashboard/src/datastore_collector.rs b/crates/runtime-dashboard/src/datastore_collector.rs index 47ec74d..b2e38b3 100644 --- a/crates/runtime-dashboard/src/datastore_collector.rs +++ b/crates/runtime-dashboard/src/datastore_collector.rs @@ -6,11 +6,70 @@ //! //! The `swactor-datastore` crate implements this trait in its `node` feature. -/// Trait for providing datastore stats to the dashboard. +use std::sync::Arc; + +/// Scope filter for listing objects. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ListScope { + Local, + Swarm, +} + +/// Trait for providing datastore stats and CRUD operations to the dashboard. /// /// Implementations capture a point-in-time snapshot as serialized JSON. /// The dashboard polls this every ~200ms via SSE. +/// +/// All command methods have default implementations returning `Err` so that +/// existing `DatastoreMetrics` impls continue to compile without changes. pub trait DatastoreStatsProvider: Send + Sync { /// Return a JSON-serialized datastore snapshot, or `None` if unavailable. fn snapshot_json(&self) -> Option; + + /// List objects as a JSON string. `scope` selects local-only or swarm-wide. + fn list_objects(&self, _name_filter: Option<&str>, _scope: ListScope) -> Result { + Err("not supported".into()) + } + + /// Get a single object's metadata + manifest as JSON. + fn get_object(&self, _hash: &str) -> Result { + Err("not supported".into()) + } + + /// Get the raw binary data for an object. + fn get_data(&self, _hash: &str) -> Result, String> { + Err("not supported".into()) + } + + /// Store data, optionally with a name. Returns JSON with `content_hash`. + fn put_data(&self, _data: Vec, _name: Option) -> Result { + Err("not supported".into()) + } + + /// Delete an object by hash. Returns JSON confirmation. + fn delete_object(&self, _hash: &str) -> Result { + Err("not supported".into()) + } + + /// Get node status as JSON. + fn node_status(&self) -> Result { + Err("not supported".into()) + } + + /// Whether the datastore is currently running. + fn is_running(&self) -> bool { + false + } + + /// Shut down the datastore actors. + fn shutdown_datastore(&self) -> Result<(), String> { + Err("not supported".into()) + } +} + +/// Factory for creating a new datastore instance from the dashboard. +pub trait DatastoreFactory: Send + Sync { + /// Start a datastore with optional persistent storage path. + /// Returns a provider that can be installed into the dashboard. + fn start_datastore(&self, storage_path: Option) -> Result, String>; } diff --git a/crates/runtime-dashboard/src/datastore_html.rs b/crates/runtime-dashboard/src/datastore_html.rs index 3afd976..4522338 100644 --- a/crates/runtime-dashboard/src/datastore_html.rs +++ b/crates/runtime-dashboard/src/datastore_html.rs @@ -80,6 +80,8 @@ pub const DATASTORE_HTML: &str = r##" white-space: nowrap; } .objects-table th { color: #888; font-weight: 500; position: sticky; top: 0; background: #161822; } + .objects-table tr { cursor: pointer; } + .objects-table tr:hover td { background: #1c1f2e; } .transfers-section { display: none; } .transfers-section.visible { display: block; } @@ -96,6 +98,87 @@ pub const DATASTORE_HTML: &str = r##" } .transfer-label { color: #888; font-size: 11px; min-width: 80px; text-align: right; } + /* Upload panel */ + .upload-row { + display: flex; gap: 10px; align-items: center; flex-wrap: wrap; + } + input[type="file"] { + background: #1c1f2e; color: #e0e0e0; border: 1px solid #2a2d3e; + border-radius: 4px; padding: 8px; font-family: inherit; font-size: 13px; + min-height: 38px; cursor: pointer; + } + input[type="file"]::file-selector-button { + background: #1e2030; color: #e0e0e0; border: 1px solid #2a2d3e; + border-radius: 4px; padding: 4px 10px; font-family: inherit; + font-size: 12px; cursor: pointer; margin-right: 8px; + } + input[type="file"]::file-selector-button:hover { border-color: #6366f1; } + input[type="text"] { + background: #1c1f2e; color: #e0e0e0; border: 1px solid #2a2d3e; + border-radius: 4px; padding: 6px 10px; font-family: inherit; + font-size: 13px; min-height: 38px; width: 180px; + } + input[type="text"]:focus { outline: none; border-color: #6366f1; } + + /* Buttons */ + button { + background: #1e2030; color: #e0e0e0; border: 1px solid #2a2d3e; + border-radius: 4px; padding: 6px 14px; font-family: inherit; + font-size: 13px; cursor: pointer; min-height: 38px; + transition: border-color 0.15s; + } + button:hover { border-color: #6366f1; color: #fff; } + button:disabled { opacity: 0.4; cursor: default; } + button.danger:hover { border-color: #f44336; } + button.primary { background: #6366f1; border-color: #6366f1; color: #fff; font-weight: 600; } + button.primary:hover { background: #5558e6; } + + /* Actions in table */ + .actions-cell { white-space: nowrap; text-align: right; } + .actions-cell button { min-height: 28px; padding: 2px 8px; font-size: 11px; } + + /* Origin badge */ + .origin-badge { + display: inline-block; font-size: 10px; padding: 1px 6px; + border-radius: 3px; font-weight: 600; + } + .origin-badge.local { background: #1b3a2a; color: #4caf50; } + .origin-badge.remote { background: #1a2a3e; color: #2196f3; } + + /* Toast */ + .toast { + position: fixed; bottom: 20px; right: 20px; padding: 10px 16px; + border-radius: 4px; font-size: 12px; z-index: 100; opacity: 0; + transition: opacity 0.3s; pointer-events: none; + } + .toast.show { opacity: 1; } + .toast.success { background: #4caf50; color: #fff; } + .toast.error { background: #f44336; color: #fff; } + + /* Modal */ + .modal-overlay { + display: none; position: fixed; inset: 0; + background: rgba(0,0,0,0.6); z-index: 50; + align-items: center; justify-content: center; + } + .modal-overlay.open { display: flex; } + .modal { + background: #161822; border: 1px solid #2a2d3e; border-radius: 6px; + padding: 20px; width: 90%; max-width: 560px; max-height: 80vh; + overflow-y: auto; + } + .modal h2 { font-size: 14px; color: #fff; margin-bottom: 16px; text-transform: none; letter-spacing: 0; } + .modal-close { + float: right; background: none; border: none; color: #888; + font-size: 18px; cursor: pointer; min-height: auto; padding: 0; + } + .modal-close:hover { color: #fff; border: none; } + .detail-row { display: flex; margin-bottom: 8px; } + .detail-label { color: #888; width: 110px; flex-shrink: 0; font-size: 11px; text-transform: uppercase; padding-top: 2px; } + .detail-value { color: #e0e0e0; word-break: break-all; font-size: 13px; } + .chunk-list { margin-top: 8px; } + .chunk-item { color: #888; font-size: 11px; padding: 2px 0; } + ::-webkit-scrollbar { width: 6px; } ::-webkit-scrollbar-track { background: #0f1117; } ::-webkit-scrollbar-thumb { background: #2a2d3e; border-radius: 3px; } @@ -117,10 +200,22 @@ pub const DATASTORE_HTML: &str = r##"
Waiting for data... + +
+ + +

Datastore Stats

@@ -142,9 +237,9 @@ pub const DATASTORE_HTML: &str = r##"

Objects

-
+
- +
HashNameSize
HashNameOriginSizeActions
@@ -157,10 +252,25 @@ pub const DATASTORE_HTML: &str = r##"
+ + + +
+