diff --git a/Cargo.lock b/Cargo.lock index b0ca8f0..034f52b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1327,8 +1327,10 @@ dependencies = [ name = "distribution" version = "0.1.0" dependencies = [ + "ed25519-dalek 2.2.0", "iroh", "iroh-relay", + "rand_core 0.6.4", "serde", "serde_json", "swactor", @@ -2391,6 +2393,19 @@ dependencies = [ "syn", ] +[[package]] +name = "integration-tests" +version = "0.1.0" +dependencies = [ + "dashboard", + "distribution", + "iroh", + "serde_json", + "swactor", + "swactor-datastore", + "ureq", +] + [[package]] name = "ipconfig" version = "0.3.2" @@ -5323,17 +5338,16 @@ dependencies = [ "clap", "crossbeam-queue", "ctrlc", - "dashboard", - "distribution", + "ed25519-dalek 2.2.0", "getrandom 0.2.17", "iroh", "proptest", "proptest-state-machine", + "rand_core 0.6.4", "serde", "serde_json", "stateright", "swactor", - "swactor-transport", "tempfile", "tiny_http", "tokio", diff --git a/Cargo.toml b/Cargo.toml index dfbc4ff..b940538 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -11,6 +11,7 @@ members = [ "crates/transport", "crates/node", "tests/docker", + "tests/integration", "xtask", ] exclude = ["crates/bindings/wasm-crypto"] diff --git a/crates/dashboard/examples/bench_dashboard.rs b/crates/dashboard/examples/bench_dashboard.rs deleted file mode 100644 index a2ed48f..0000000 --- a/crates/dashboard/examples/bench_dashboard.rs +++ /dev/null @@ -1,224 +0,0 @@ -use std::thread; -use std::time::{Duration, Instant}; - -use swactor::actor::{ActorAddress, ActorInterface, Ctx}; -use swactor::config::{BackoffPolicy, RuntimeConfig}; -use swactor::runtime::Runtime; - -use dashboard::collector::StatsCollector; -use dashboard::{start_dashboard, DashboardConfig}; - -// --------------------------------------------------------------------------- -// Actors -// --------------------------------------------------------------------------- - -#[derive(Clone)] -struct Work; - -struct SinkActor { - count: u64, -} - -impl SinkActor { - fn new() -> Self { - Self { count: 0 } - } -} - -impl ActorInterface for SinkActor { - type Incoming = Work; - type Response = (); - fn handle(&mut self, _ctx: &Ctx, _msg: Work) { - self.count += 1; - } -} - -#[derive(Clone)] -struct RingMsg; - -struct RingActor { - next: ActorAddress, -} - -impl ActorInterface for RingActor { - type Incoming = RingMsg; - type Response = (); - fn handle(&mut self, ctx: &Ctx, _msg: RingMsg) { - let _ = ctx.send(self.next, RingMsg); - } -} - -#[derive(Clone)] -struct SpawnCmd; - -struct SpawnerActor { - spawned: u64, -} - -impl SpawnerActor { - fn new() -> Self { - Self { spawned: 0 } - } -} - -impl ActorInterface for SpawnerActor { - type Incoming = SpawnCmd; - type Response = (); - fn handle(&mut self, ctx: &Ctx, _msg: SpawnCmd) { - for _ in 0..20 { - let _ = ctx.spawn(SinkActor::new()); - self.spawned += 1; - } - } -} - -// --------------------------------------------------------------------------- -// Helpers -// --------------------------------------------------------------------------- - -fn bench_config(threads: usize, max_actors: usize, max_messages: usize) -> RuntimeConfig { - RuntimeConfig { - num_threads: threads, - max_actors, - channel_buffer_size: max_messages, - backoff_policy: BackoffPolicy { - spin_threshold: 32, - yield_threshold: 64, - sleep_increment_us: 10, - sleep_max_us: 100, - }, - ..Default::default() - } -} - -fn run_for(duration: Duration, mut tick: impl FnMut()) { - let deadline = Instant::now() + duration; - while Instant::now() < deadline { - tick(); - } -} - -// --------------------------------------------------------------------------- -// Scenarios -// --------------------------------------------------------------------------- - -fn scenario_single_actor(dash: &dashboard::DashboardHandle) { - eprintln!(" [1/4] Single-actor bombardment (5s)"); - let collector = StatsCollector::new(4); - let mut rt = Runtime::new(bench_config(4, 64, 100_000)); - rt.set_stats_hook(collector.clone()); - let addr = rt.spawn(SinkActor::new()).unwrap(); - let handle = rt.run().unwrap(); - dash.set_runtime(handle.runtime.clone(), collector); - - run_for(Duration::from_secs(5), || { - for _ in 0..100 { - let _ = handle.runtime.send_to(addr, Work); - } - thread::sleep(Duration::from_millis(10)); - }); - - handle.shutdown(); - handle.join(); -} - -fn scenario_multi_actor(dash: &dashboard::DashboardHandle) { - eprintln!(" [2/4] Multi-actor fan-out (5s)"); - let collector = StatsCollector::new(4); - let mut rt = Runtime::new(bench_config(4, 128, 10_000)); - rt.set_stats_hook(collector.clone()); - let addrs: Vec<_> = (0..50) - .map(|_| rt.spawn(SinkActor::new()).unwrap()) - .collect(); - let handle = rt.run().unwrap(); - dash.set_runtime(handle.runtime.clone(), collector); - - run_for(Duration::from_secs(5), || { - for &addr in &addrs { - for _ in 0..10 { - let _ = handle.runtime.send_to(addr, Work); - } - } - thread::sleep(Duration::from_millis(20)); - }); - - handle.shutdown(); - handle.join(); -} - -fn scenario_ring(dash: &dashboard::DashboardHandle) { - eprintln!(" [3/4] Ring topology (5s)"); - let ring_size = 100; - let collector = StatsCollector::new(4); - let mut rt = Runtime::new(bench_config(4, ring_size + 64, 1_024)); - rt.set_stats_hook(collector.clone()); - - // Build ring backwards: last spawned actor is the entry point - let mut addrs = Vec::with_capacity(ring_size); - // First actor has no valid next yet — will be the tail of the chain - let first = rt.spawn(RingActor { next: ActorAddress::default() }).unwrap(); - addrs.push(first); - let mut prev = first; - for _ in 1..ring_size { - let addr = rt.spawn(RingActor { next: prev }).unwrap(); - addrs.push(addr); - prev = addr; - } - // The first actor's "next" should be the last actor to close the ring, - // but we can't mutate it. Instead, we inject at the last actor and - // the message flows: last -> second-to-last -> ... -> first -> (dead end). - // For dashboard visualization, a chain is fine — it creates sustained cross-worker traffic. - let entry = *addrs.last().unwrap(); - - let handle = rt.run().unwrap(); - dash.set_runtime(handle.runtime.clone(), collector); - - run_for(Duration::from_secs(5), || { - let _ = handle.runtime.send_to(entry, RingMsg); - thread::sleep(Duration::from_millis(50)); - }); - - handle.shutdown(); - handle.join(); -} - -fn scenario_spawn_storm(dash: &dashboard::DashboardHandle) { - eprintln!(" [4/4] Spawn storm (5s)"); - let collector = StatsCollector::new(4); - let mut rt = Runtime::new(bench_config(4, 50_000, 1_024)); - rt.set_stats_hook(collector.clone()); - let spawner = rt.spawn(SpawnerActor::new()).unwrap(); - let handle = rt.run().unwrap(); - dash.set_runtime(handle.runtime.clone(), collector); - - run_for(Duration::from_secs(5), || { - let _ = handle.runtime.send_to(spawner, SpawnCmd); - thread::sleep(Duration::from_millis(200)); - }); - - handle.shutdown(); - handle.join(); -} - -// --------------------------------------------------------------------------- -// Main -// --------------------------------------------------------------------------- - -fn main() { - let dash = start_dashboard(DashboardConfig { - port: 9090, - ..Default::default() - }); - dash.install_tracing(); - - eprintln!("Dashboard at http://localhost:9090"); - eprintln!("Running 4 benchmark scenarios (~20s total)...\n"); - - scenario_single_actor(&dash); - scenario_multi_actor(&dash); - scenario_ring(&dash); - scenario_spawn_storm(&dash); - - eprintln!("\nAll scenarios complete. Shutting down."); - dash.shutdown(); -} diff --git a/crates/dashboard/examples/record_and_replay_demo.rs b/crates/dashboard/examples/record_and_replay_demo.rs deleted file mode 100644 index 13d5292..0000000 --- a/crates/dashboard/examples/record_and_replay_demo.rs +++ /dev/null @@ -1,162 +0,0 @@ -use std::sync::atomic::{AtomicBool, Ordering}; -use std::sync::Arc; -use std::thread; -use std::time::Duration; - -use swactor::actor::{ActorAddress, ActorInterface, Ctx}; -use swactor::config::RuntimeConfig; -use swactor::runtime::Runtime; - -use dashboard::collector::StatsCollector; -use dashboard::{serve_replay, start_dashboard, DashboardConfig, ReplayConfig}; - -// ── Demo actors ───────────────────────────────────────────────────────── - -#[derive(Clone)] -struct Ping(ActorAddress); - -struct PingActor { - count: u32, -} - -impl PingActor { - fn new() -> Self { - Self { count: 0 } - } -} - -impl ActorInterface for PingActor { - type Incoming = Ping; - type Response = (); - - fn handle(&mut self, ctx: &Ctx, msg: Ping) { - self.count += 1; - if self.count < 100 { - let _ = ctx.send(msg.0, Ping(ctx.self_addr())); - } - } -} - -#[derive(Clone)] -struct Tick; - -struct CounterActor; - -impl ActorInterface for CounterActor { - type Incoming = Tick; - type Response = (); - - fn handle(&mut self, _ctx: &Ctx, _msg: Tick) {} -} - -// ── Main ──────────────────────────────────────────────────────────────── - -fn main() { - // ── Phase 1: Record ───────────────────────────────────────────────── - - let dash = start_dashboard(DashboardConfig { - port: 9090, - record: true, - ..Default::default() - }); - dash.install_tracing(); - - let num_threads = 4; - let collector = StatsCollector::new(num_threads); - - let mut rt = Runtime::new(RuntimeConfig { - num_threads, - max_actors: 512, - channel_buffer_size: 1000, - ..Default::default() - }); - rt.set_stats_hook(collector.clone()); - - let mut ping_addrs = Vec::new(); - for _ in 0..12 { - let addr = rt.spawn(PingActor::new()).unwrap(); - ping_addrs.push(addr); - } - - let mut counter_addrs = Vec::new(); - for _ in 0..16 { - let addr = rt.spawn(CounterActor).unwrap(); - counter_addrs.push(addr); - } - - let handle = rt.run().expect("failed to start runtime"); - dash.set_runtime(handle.runtime.clone(), collector); - - eprintln!("Recording trace for ~10 seconds..."); - eprintln!("Live dashboard at http://localhost:9090"); - - // Kick off ping-pong chains - for i in 0..ping_addrs.len() { - let target = ping_addrs[(i + 1) % ping_addrs.len()]; - let _ = handle.runtime.send_to(ping_addrs[i], Ping(target)); - } - - for round in 0..50 { - for addr in &counter_addrs { - let _ = handle.runtime.send_to(*addr, Tick); - } - - if round == 20 { - for _ in 0..8 { - let addr = handle.runtime.spawn(CounterActor).unwrap(); - counter_addrs.push(addr); - } - eprintln!(" Spawned 8 more actors"); - } - - if round == 25 { - for i in 0..ping_addrs.len() { - let target = ping_addrs[(i + 1) % ping_addrs.len()]; - let _ = handle.runtime.send_to(ping_addrs[i], Ping(target)); - } - } - - thread::sleep(Duration::from_millis(200)); - } - - handle.shutdown(); - dash.shutdown(); - handle.join(); - - // Save trace - let path = "runtime_trace.json"; - match dash.save_trace(path) { - Ok(()) => eprintln!("Trace saved to {path}"), - Err(e) => { - eprintln!("Failed to save trace: {e}"); - std::process::exit(1); - } - } - - // ── Phase 2: Replay ───────────────────────────────────────────────── - - let stop = Arc::new(AtomicBool::new(false)); - { - let stop = Arc::clone(&stop); - ctrlc::set_handler(move || { - stop.store(true, Ordering::Relaxed); - }) - .expect("failed to set Ctrl+C handler"); - } - - eprintln!("\nStarting replay at 2x speed — press Ctrl+C to stop"); - - // Spawn replay server in a background thread so we can check Ctrl+C - let replay_path = path.to_string(); - thread::spawn(move || { - if let Err(e) = serve_replay(&replay_path, ReplayConfig { port: 9091, speed: 2.0 }) { - eprintln!("Replay error: {e}"); - } - }); - - while !stop.load(Ordering::Relaxed) { - thread::sleep(Duration::from_millis(200)); - } - - eprintln!("Done."); -} diff --git a/crates/dashboard/src/actor_detail_html.rs b/crates/dashboard/src/actor_detail_html.rs index 4b4c41e..3d98f33 100644 --- a/crates/dashboard/src/actor_detail_html.rs +++ b/crates/dashboard/src/actor_detail_html.rs @@ -123,8 +123,8 @@ pub const ACTOR_DETAIL_HTML: &str = r##" diff --git a/crates/dashboard/src/actors_html.rs b/crates/dashboard/src/actors_html.rs index 9595175..8a07027 100644 --- a/crates/dashboard/src/actors_html.rs +++ b/crates/dashboard/src/actors_html.rs @@ -151,8 +151,8 @@ pub const ACTORS_HTML: &str = r##"
diff --git a/crates/dashboard/src/dashboard_html.rs b/crates/dashboard/src/dashboard_html.rs index 0aa4f7f..f777468 100644 --- a/crates/dashboard/src/dashboard_html.rs +++ b/crates/dashboard/src/dashboard_html.rs @@ -152,8 +152,8 @@ pub const DASHBOARD_HTML: &str = r##"
diff --git a/crates/dashboard/src/topology_html.rs b/crates/dashboard/src/topology_html.rs index 123937b..9f5b9da 100644 --- a/crates/dashboard/src/topology_html.rs +++ b/crates/dashboard/src/topology_html.rs @@ -46,8 +46,8 @@ pub const TOPOLOGY_HTML: &str = r##" Overview Actors Topology - Distribution - Datastore + Distribution + Datastore
diff --git a/crates/datastore/Cargo.toml b/crates/datastore/Cargo.toml index 81b2fc6..cfa67e5 100644 --- a/crates/datastore/Cargo.toml +++ b/crates/datastore/Cargo.toml @@ -8,8 +8,8 @@ unexpected_cfgs = { level = "warn", check-cfg = ['cfg(kani)'] } [dependencies] swactor = { path = "../..", features = ["serde", "transport"] } -distribution = { path = "../distribution" } -swactor-transport = { path = "../transport", default-features = false } +ed25519-dalek = { version = "2", features = ["rand_core"] } +rand_core = { version = "0.6", features = ["getrandom"] } serde = { version = "1", features = ["derive"] } serde_json = "1" blake3 = "1" @@ -28,22 +28,14 @@ serde_json = "1" proptest = "1" proptest-state-machine = "0.3" tempfile = "3" -distribution = { path = "../distribution" } swactor = { path = "../.." } ureq = { version = "2", features = ["json"] } -tiny_http = "0.12" -dashboard = { path = "../dashboard" } stateright = "0.31" [features] node = ["dep:tiny_http", "dep:clap", "dep:ctrlc", "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" diff --git a/crates/datastore/src/actors/datastore_node.rs b/crates/datastore/src/actors/datastore_node.rs index 0cf1343..01067f2 100644 --- a/crates/datastore/src/actors/datastore_node.rs +++ b/crates/datastore/src/actors/datastore_node.rs @@ -11,7 +11,7 @@ use std::sync::Arc; use swactor::actor::{ActorAddress, ActorInterface, Ctx}; use swactor::runtime::Runtime; -use swactor_transport::NodeId; +use swactor::transport::NodeId; use crate::actors::stream_downloader::StreamDownloader; use crate::actors::stream_server::StreamServer; diff --git a/crates/datastore/src/actors/gateway.rs b/crates/datastore/src/actors/gateway.rs index 73cb4be..647617a 100644 --- a/crates/datastore/src/actors/gateway.rs +++ b/crates/datastore/src/actors/gateway.rs @@ -17,7 +17,7 @@ use swactor::actor::{ActorAddress, ActorInterface, Ctx}; use crate::auth::{AccessRequestInfo, AuthzEngine, AuthzResult, DatastoreAction, DeniedReason}; use crate::messages::{DatastoreNodeMsg, DatastoreResponse, GatewayMsg}; -use swactor_transport::NodeId; +use swactor::transport::NodeId; /// The auth gateway actor wrapping an `AuthzEngine`. pub struct GatewayActor { diff --git a/crates/datastore/src/actors/metadata.rs b/crates/datastore/src/actors/metadata.rs index cc88634..df50824 100644 --- a/crates/datastore/src/actors/metadata.rs +++ b/crates/datastore/src/actors/metadata.rs @@ -11,7 +11,7 @@ use std::collections::{HashMap, HashSet}; use swactor::actor::{ActorAddress, ActorInterface, Ctx}; -use swactor_transport::NodeId; +use swactor::transport::NodeId; use crate::messages::{BlobStoreMsg, DatastoreResponse, MetadataMsg}; use crate::types::{ContentHash, DatastoreConfig, ObjectEntry, ObjectManifest}; diff --git a/crates/datastore/src/actors/transfer.rs b/crates/datastore/src/actors/transfer.rs index fd8da68..3e0f2a2 100644 --- a/crates/datastore/src/actors/transfer.rs +++ b/crates/datastore/src/actors/transfer.rs @@ -12,7 +12,7 @@ use std::collections::HashSet; use swactor::actor::{ActorAddress, ActorInterface, Ctx}; -use swactor_transport::NodeId; +use swactor::transport::NodeId; use crate::messages::{DatastoreResponse, TransferMsg}; use crate::types::{ContentHash, ObjectManifest, TransferStatus}; diff --git a/crates/datastore/src/api.rs b/crates/datastore/src/api.rs index fa973d2..27b44e5 100644 --- a/crates/datastore/src/api.rs +++ b/crates/datastore/src/api.rs @@ -12,7 +12,7 @@ use std::time::{Duration, Instant}; use swactor::actor::ActorAddress; use swactor::runtime::{Inbox, Runtime}; -use swactor_transport::NodeId; +use swactor::transport::NodeId; use crate::auth::SignedRequest; use crate::chunking::reassemble_blob; @@ -753,7 +753,7 @@ fn try_remote_get( let _ = state.runtime.send_to( peer.metadata, MetadataMsg::HandleFindObject { - from: swactor_transport::NodeId([0; 32]), // placeholder + from: swactor::transport::NodeId([0; 32]), // placeholder content_hash, reply_to: *find_inbox.addr(), }, diff --git a/crates/datastore/src/auth.rs b/crates/datastore/src/auth.rs index f57cbf3..8c31fe1 100644 --- a/crates/datastore/src/auth.rs +++ b/crates/datastore/src/auth.rs @@ -12,9 +12,9 @@ use std::path::Path; use serde::{Deserialize, Serialize}; -use swactor_transport::crypto; -use swactor_transport::NodeId; -use swactor_transport::crypto::Signature; +use crate::crypto; +use swactor::transport::NodeId; +use crate::crypto::Signature; use crate::content_hash::ContentHash; // ─── Access Request / Authorized Key Info ────────────────────────────────── diff --git a/crates/datastore/src/bin/store_cli.rs b/crates/datastore/src/bin/store_cli.rs index 29001ee..a1a2568 100644 --- a/crates/datastore/src/bin/store_cli.rs +++ b/crates/datastore/src/bin/store_cli.rs @@ -10,7 +10,7 @@ use std::time::{SystemTime, UNIX_EPOCH}; use clap::{Parser, Subcommand}; -use swactor_transport::crypto::Keypair; +use swactor_datastore::crypto::Keypair; use swactor_datastore::content_hash::ContentHash; use swactor_datastore::auth::{sign_request, DatastoreAction, SignedRequestPayload}; diff --git a/crates/datastore/src/bridge.rs b/crates/datastore/src/bridge.rs index a5d6d2e..92f8c33 100644 --- a/crates/datastore/src/bridge.rs +++ b/crates/datastore/src/bridge.rs @@ -10,7 +10,7 @@ use swactor::actor::ActorAddress; use swactor::runtime::Runtime; use swactor::std::RuntimeNaming; -use swactor_transport::NodeId; +use swactor::transport::NodeId; use crate::actors::{BlobStoreActor, DatastoreNode, GatewayActor, MetadataActor}; use crate::auth::{AccessControlList, AuthzEngine}; diff --git a/crates/datastore/src/cli.rs b/crates/datastore/src/cli.rs index df0c204..8c126c7 100644 --- a/crates/datastore/src/cli.rs +++ b/crates/datastore/src/cli.rs @@ -6,7 +6,7 @@ use std::collections::BTreeMap; use std::path::PathBuf; -use swactor_transport::NodeId; +use swactor::transport::NodeId; /// Top-level CLI commands for `swactor-store`. #[derive(Debug, Clone)] diff --git a/crates/datastore/src/crypto.rs b/crates/datastore/src/crypto.rs new file mode 100644 index 0000000..2a49f0b --- /dev/null +++ b/crates/datastore/src/crypto.rs @@ -0,0 +1,96 @@ +//! Ed25519 cryptographic primitives for the datastore. + +use std::fmt; + +use ed25519_dalek::{Signer, Verifier}; +use serde::{Deserialize, Serialize}; + +use swactor::transport::NodeId; + +// ─── Signature ────────────────────────────────────────────────────────────── + +/// An ed25519 signature (64 bytes). +#[derive(Clone, Copy, PartialEq, Eq)] +pub struct Signature(pub [u8; 64]); + +impl Serialize for Signature { + fn serialize(&self, serializer: S) -> Result { + serializer.serialize_bytes(&self.0) + } +} + +impl<'de> Deserialize<'de> for Signature { + fn deserialize>(deserializer: D) -> Result { + let bytes: Vec = Deserialize::deserialize(deserializer)?; + if bytes.len() != 64 { + return Err(serde::de::Error::custom(format!( + "expected 64 bytes for Signature, got {}", + bytes.len() + ))); + } + let mut arr = [0u8; 64]; + arr.copy_from_slice(&bytes); + Ok(Signature(arr)) + } +} + +impl fmt::Debug for Signature { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "Sig(")?; + for b in &self.0[..4] { + write!(f, "{:02x}", b)?; + } + write!(f, "\u{2026})") + } +} + +// ─── Keypair ──────────────────────────────────────────────────────────────── + +/// Node identity keypair — wraps ed25519-dalek. +pub struct Keypair { + inner: ed25519_dalek::SigningKey, +} + +impl Keypair { + /// Generate a new random keypair. + pub fn generate() -> Self { + let mut csprng = rand_core::OsRng; + Self { + inner: ed25519_dalek::SigningKey::generate(&mut csprng), + } + } + + /// Reconstruct from raw secret key bytes (32 bytes). + pub fn from_bytes(secret: &[u8; 32]) -> Self { + Self { + inner: ed25519_dalek::SigningKey::from_bytes(secret), + } + } + + /// The public key as a `NodeId`. + pub fn node_id(&self) -> NodeId { + NodeId(self.inner.verifying_key().to_bytes()) + } + + /// Raw secret key bytes. + pub fn secret_bytes(&self) -> [u8; 32] { + self.inner.to_bytes() + } + + /// Sign arbitrary bytes. + pub fn sign(&self, msg: &[u8]) -> Signature { + let sig = self.inner.sign(msg); + Signature(sig.to_bytes()) + } +} + +// ─── Verification ─────────────────────────────────────────────────────────── + +/// Verify a signature against a `NodeId` (public key) and message bytes. +pub fn verify(node_id: &NodeId, msg: &[u8], sig: &Signature) -> bool { + let Ok(vk) = ed25519_dalek::VerifyingKey::from_bytes(&node_id.0) else { + return false; + }; + let signature = ed25519_dalek::Signature::from_bytes(&sig.0); + vk.verify(msg, &signature).is_ok() +} diff --git a/crates/datastore/src/lib.rs b/crates/datastore/src/lib.rs index f3ce7a5..31f6846 100644 --- a/crates/datastore/src/lib.rs +++ b/crates/datastore/src/lib.rs @@ -1,4 +1,5 @@ pub mod content_hash; +pub mod crypto; pub mod types; pub mod messages; pub mod chunking; diff --git a/crates/datastore/src/messages.rs b/crates/datastore/src/messages.rs index eee84ef..e6a8bbd 100644 --- a/crates/datastore/src/messages.rs +++ b/crates/datastore/src/messages.rs @@ -16,7 +16,7 @@ use swactor::actor::ActorAddress; use swactor::runtime::Runtime; use swactor::transport::NetworkMessage; -use swactor_transport::NodeId; +use swactor::transport::NodeId; use crate::streams::types::StreamId; use crate::auth::{AccessRequestInfo, AuthorizedKeyInfo, DeniedReason, SignedRequest}; diff --git a/crates/datastore/src/types.rs b/crates/datastore/src/types.rs index c734a63..d41c12b 100644 --- a/crates/datastore/src/types.rs +++ b/crates/datastore/src/types.rs @@ -8,7 +8,7 @@ use std::path::PathBuf; use serde::{Deserialize, Serialize}; -use swactor_transport::NodeId; +use swactor::transport::NodeId; pub use crate::content_hash::ContentHash; diff --git a/crates/datastore/tests/acl_persistence_tests.rs b/crates/datastore/tests/acl_persistence_tests.rs index a959815..21f48f2 100644 --- a/crates/datastore/tests/acl_persistence_tests.rs +++ b/crates/datastore/tests/acl_persistence_tests.rs @@ -2,7 +2,7 @@ use std::collections::{HashMap, HashSet}; -use swactor_transport::crypto::Keypair; +use swactor_datastore::crypto::Keypair; use swactor_datastore::auth::AccessControlList; // ═══════════════════════════════════════════════════════════════════════════ diff --git a/crates/datastore/tests/auth_scenario_tests.rs b/crates/datastore/tests/auth_scenario_tests.rs index e8c8aa9..d63a16a 100644 --- a/crates/datastore/tests/auth_scenario_tests.rs +++ b/crates/datastore/tests/auth_scenario_tests.rs @@ -2,7 +2,7 @@ use std::collections::{HashMap, HashSet}; -use swactor_transport::crypto::Keypair; +use swactor_datastore::crypto::Keypair; use swactor_datastore::content_hash::ContentHash; use swactor_datastore::auth::{ sign_request, AccessControlList, AuthzEngine, AuthzResult, DatastoreAction, DeniedReason, diff --git a/crates/datastore/tests/auth_state_machine.rs b/crates/datastore/tests/auth_state_machine.rs index 596dd41..1d0eba9 100644 --- a/crates/datastore/tests/auth_state_machine.rs +++ b/crates/datastore/tests/auth_state_machine.rs @@ -18,7 +18,7 @@ use std::collections::{HashMap, HashSet}; use proptest::prelude::*; use proptest_state_machine::{prop_state_machine, ReferenceStateMachine, StateMachineTest}; -use swactor_transport::crypto::Keypair; +use swactor_datastore::crypto::Keypair; use swactor_datastore::auth::{ sign_request, AccessControlList, AuthzEngine, AuthzResult, DatastoreAction, DeniedReason, SignedRequestPayload, diff --git a/crates/datastore/tests/common/mod.rs b/crates/datastore/tests/common/mod.rs index 2bf3ebf..4a6ea4d 100644 --- a/crates/datastore/tests/common/mod.rs +++ b/crates/datastore/tests/common/mod.rs @@ -15,7 +15,7 @@ use swactor_datastore::storage::InMemoryBackend; use swactor_datastore::types::{ChunkRef, ContentHash, DatastoreConfig, ObjectEntry, ObjectManifest}; use swactor_datastore::{BlobStoreActor, DatastoreNode, MetadataActor, TransferActor}; -use swactor_transport::NodeId; +use swactor::transport::NodeId; /// Create a single-threaded runtime with StdExtension. pub fn test_runtime() -> Runtime { diff --git a/crates/datastore/tests/datastore_node_tests.rs b/crates/datastore/tests/datastore_node_tests.rs index 4b02ad9..0b7c73a 100644 --- a/crates/datastore/tests/datastore_node_tests.rs +++ b/crates/datastore/tests/datastore_node_tests.rs @@ -14,7 +14,7 @@ use swactor_datastore::types::ContentHash; use swactor_datastore::messages::{DatastoreResponse, GetChunkRequest}; use swactor_datastore::{reassemble_blob, verify_integrity}; -use swactor_transport::NodeId; +use swactor::transport::NodeId; // ═══════════════════════════════════════════════════════════════════════════ // Put & Retrieve diff --git a/crates/datastore/tests/datastore_tests.rs b/crates/datastore/tests/datastore_tests.rs index 0eb33a4..02d4160 100644 --- a/crates/datastore/tests/datastore_tests.rs +++ b/crates/datastore/tests/datastore_tests.rs @@ -9,7 +9,7 @@ use swactor_datastore::types::{ ChunkRef, ContentHash, DatastoreConfig, ObjectEntry, ObjectManifest, }; -use swactor_transport::NodeId; +use swactor::transport::NodeId; // ═══════════════════════════════════════════════════════════════════════════ // Scenario: Content-addressing round-trip diff --git a/crates/datastore/tests/gateway_tests.rs b/crates/datastore/tests/gateway_tests.rs index b9f318d..b246a6a 100644 --- a/crates/datastore/tests/gateway_tests.rs +++ b/crates/datastore/tests/gateway_tests.rs @@ -10,7 +10,7 @@ use std::time::{SystemTime, UNIX_EPOCH}; use common::{spawn_blob_store, spawn_metadata, test_runtime, tick_until_recv}; -use swactor_transport::crypto::Keypair; +use swactor_datastore::crypto::Keypair; use swactor_datastore::content_hash::ContentHash; use swactor_datastore::actors::{DatastoreNode, GatewayActor}; use swactor_datastore::auth::{ diff --git a/crates/datastore/tests/http_auth_integration.rs b/crates/datastore/tests/http_auth_integration.rs index b34e0f6..49d567f 100644 --- a/crates/datastore/tests/http_auth_integration.rs +++ b/crates/datastore/tests/http_auth_integration.rs @@ -9,7 +9,7 @@ use std::collections::{BTreeMap, HashMap, HashSet}; use std::sync::atomic::Ordering; use std::time::{Duration, SystemTime, UNIX_EPOCH}; -use swactor_transport::crypto::Keypair; +use swactor_datastore::crypto::Keypair; use swactor_datastore::content_hash::ContentHash; use swactor::config::RuntimeConfig; use swactor::runtime::Runtime; diff --git a/crates/datastore/tests/metadata_tests.rs b/crates/datastore/tests/metadata_tests.rs index 615e2c4..3281da5 100644 --- a/crates/datastore/tests/metadata_tests.rs +++ b/crates/datastore/tests/metadata_tests.rs @@ -7,7 +7,7 @@ use std::collections::{BTreeMap, HashSet}; use swactor_datastore::messages::{DatastoreResponse, MetadataMsg}; use swactor_datastore::types::{ContentHash, ObjectEntry}; -use swactor_transport::NodeId; +use swactor::transport::NodeId; use common::{ make_entry, make_manifest, spawn_metadata, test_node_id, test_runtime, tick_n, tick_until_recv, diff --git a/crates/datastore/tests/multi_node_tests.rs b/crates/datastore/tests/multi_node_tests.rs index 67be792..7d126d9 100644 --- a/crates/datastore/tests/multi_node_tests.rs +++ b/crates/datastore/tests/multi_node_tests.rs @@ -11,7 +11,7 @@ use swactor_datastore::messages::{MetadataMsg, TransferMsg}; use swactor_datastore::types::ContentHash; use swactor_datastore::TransferActor; -use swactor_transport::NodeId; +use swactor::transport::NodeId; // ═══════════════════════════════════════════════════════════════════════════ // Phase 3: Metadata dissemination tests diff --git a/crates/datastore/tests/transfer_tests.rs b/crates/datastore/tests/transfer_tests.rs index f4e1d05..6f4b470 100644 --- a/crates/datastore/tests/transfer_tests.rs +++ b/crates/datastore/tests/transfer_tests.rs @@ -6,7 +6,7 @@ use swactor_datastore::chunking::{chunk_blob, reassemble_blob, verify_integrity} use swactor_datastore::messages::{BlobStoreMsg, DatastoreResponse, TransferMsg}; use swactor_datastore::types::{ChunkRef, ContentHash, ObjectManifest}; -use swactor_transport::NodeId; +use swactor::transport::NodeId; use common::{ spawn_blob_store, spawn_transfer, test_runtime, tick_and_drain, tick_n, tick_until_recv, diff --git a/crates/distribution/Cargo.toml b/crates/distribution/Cargo.toml index 7ee30ab..c2567a3 100644 --- a/crates/distribution/Cargo.toml +++ b/crates/distribution/Cargo.toml @@ -5,13 +5,15 @@ edition = "2024" [features] default = ["tcp"] -tcp = [] +tcp = ["dep:swactor-transport", "swactor-transport/tcp"] iroh = ["dep:iroh", "dep:tokio"] relay = ["iroh", "dep:iroh-relay"] [dependencies] swactor = { path = "../..", features = ["serde", "transport"] } -swactor-transport = { path = "../transport" } +swactor-transport = { path = "../transport", optional = true } +ed25519-dalek = { version = "2", features = ["rand_core"] } +rand_core = { version = "0.6", features = ["getrandom"] } serde = { version = "1", features = ["derive"] } serde_json = "1" iroh = { version = "0.96", optional = true } diff --git a/crates/distribution/src/crypto.rs b/crates/distribution/src/crypto.rs index 5102c13..a59d25d 100644 --- a/crates/distribution/src/crypto.rs +++ b/crates/distribution/src/crypto.rs @@ -1,7 +1,101 @@ -// Re-export core crypto from transport. -pub use swactor_transport::crypto::{Keypair, Signature, verify}; +//! Ed25519 cryptographic primitives for distribution. -use crate::types::{DirectoryEntry, DirectoryEntryPayload}; +use std::fmt; + +use ed25519_dalek::{Signer, Verifier}; +use serde::{Deserialize, Serialize}; + +use crate::types::{DirectoryEntry, DirectoryEntryPayload, NodeId}; + +// ─── Signature ────────────────────────────────────────────────────────────── + +/// An ed25519 signature (64 bytes). +#[derive(Clone, Copy, PartialEq, Eq)] +pub struct Signature(pub [u8; 64]); + +impl Serialize for Signature { + fn serialize(&self, serializer: S) -> Result { + serializer.serialize_bytes(&self.0) + } +} + +impl<'de> Deserialize<'de> for Signature { + fn deserialize>(deserializer: D) -> Result { + let bytes: Vec = Deserialize::deserialize(deserializer)?; + if bytes.len() != 64 { + return Err(serde::de::Error::custom(format!( + "expected 64 bytes for Signature, got {}", + bytes.len() + ))); + } + let mut arr = [0u8; 64]; + arr.copy_from_slice(&bytes); + Ok(Signature(arr)) + } +} + +impl fmt::Debug for Signature { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "Sig(")?; + for b in &self.0[..4] { + write!(f, "{:02x}", b)?; + } + write!(f, "\u{2026})") + } +} + +// ─── Keypair ──────────────────────────────────────────────────────────────── + +/// Node identity keypair — wraps ed25519-dalek. +pub struct Keypair { + inner: ed25519_dalek::SigningKey, +} + +impl Keypair { + /// Generate a new random keypair. + pub fn generate() -> Self { + let mut csprng = rand_core::OsRng; + Self { + inner: ed25519_dalek::SigningKey::generate(&mut csprng), + } + } + + /// Reconstruct from raw secret key bytes (32 bytes). + pub fn from_bytes(secret: &[u8; 32]) -> Self { + Self { + inner: ed25519_dalek::SigningKey::from_bytes(secret), + } + } + + /// The public key as a `NodeId`. + pub fn node_id(&self) -> NodeId { + NodeId(self.inner.verifying_key().to_bytes()) + } + + /// Raw secret key bytes. + pub fn secret_bytes(&self) -> [u8; 32] { + self.inner.to_bytes() + } + + /// Sign arbitrary bytes. + pub fn sign(&self, msg: &[u8]) -> Signature { + let sig = self.inner.sign(msg); + Signature(sig.to_bytes()) + } +} + +// ─── Verification ─────────────────────────────────────────────────────────── + +/// Verify a signature against a `NodeId` (public key) and message bytes. +pub fn verify(node_id: &NodeId, msg: &[u8], sig: &Signature) -> bool { + let Ok(vk) = ed25519_dalek::VerifyingKey::from_bytes(&node_id.0) else { + return false; + }; + let signature = ed25519_dalek::Signature::from_bytes(&sig.0); + vk.verify(msg, &signature).is_ok() +} + +// ─── Directory Entry Helpers ──────────────────────────────────────────────── /// Extension methods for Keypair specific to distribution directory entries. pub trait KeypairExt { diff --git a/crates/distribution/src/driver.rs b/crates/distribution/src/driver.rs index 99412bf..26c280f 100644 --- a/crates/distribution/src/driver.rs +++ b/crates/distribution/src/driver.rs @@ -19,7 +19,7 @@ use crate::messages::*; use crate::node::{DistributedNode, DistributedNodeConfig}; use crate::snapshot::DistributionNodeSnapshot; use crate::swim::node::NodeAction; -use crate::transport::{TcpAcceptor, TcpTransport, encode_wire_envelope_with_hints}; +use swactor_transport::tcp::{TcpAcceptor, TcpTransport, encode_wire_envelope_with_hints}; use crate::types::NodeId; /// Dummy destination address used in wire envelopes for SWIM protocol messages. diff --git a/crates/distribution/src/gossip_channel.rs b/crates/distribution/src/gossip_channel.rs deleted file mode 100644 index a49dce6..0000000 --- a/crates/distribution/src/gossip_channel.rs +++ /dev/null @@ -1,268 +0,0 @@ -//! Generic gossip channel abstraction. -//! -//! Provides `GossipChannel` — a trait for any topic that wants to piggyback -//! on SWIM protocol messages — and `DisseminationBuffer` — a reusable -//! budget-limited dissemination queue that replaces the 4 independent copies -//! of the `Λ * ceil(log₂(n))` pattern. - -use serde::{Serialize, de::DeserializeOwned}; - -// ─── GossipChannel trait ──────────────────────────────────────────────────── - -/// A gossip channel that can piggyback serialized entries on SWIM messages. -/// -/// Each channel has a unique topic tag and handles its own serialization. -/// The wire transport uses type-erased `Vec` entries. -pub trait GossipChannel: Send { - /// Unique string tag identifying this channel in the piggyback payload. - fn topic_tag(&self) -> &'static str; - - /// Take up to `max_entries` pending entries, serialized as bytes. - fn take_pending_bytes(&mut self, max_entries: usize) -> Vec>; - - /// Apply incoming entries (deserialized from bytes) received from gossip. - fn apply_incoming_bytes(&mut self, entries: &[Vec], cluster_size: usize); - - /// Re-enqueue all state for dissemination (anti-entropy on membership recovery). - fn re_disseminate_all(&mut self, cluster_size: usize); - - /// Handle a node being declared dead. - fn on_node_death(&mut self, node_id: &crate::types::NodeId); - - /// Periodic garbage collection tick. - fn gc_tick(&mut self); -} - -// ─── DisseminationBuffer ──────────────────────────────────────────────── - -/// A queued entry with a remaining transmit budget. -#[derive(Debug, Clone)] -struct BufferEntry { - item: T, - remaining: usize, -} - -/// Reusable generic dissemination buffer. -/// -/// Manages budget-limited gossip dissemination for any entry type. -/// Each entry is transmitted `Λ * ceil(log₂(n))` times before eviction. -#[derive(Debug)] -pub struct DisseminationBuffer { - entries: Vec>, - lambda: usize, -} - -impl DisseminationBuffer { - /// Create a new buffer with the given dissemination multiplier (Λ). - pub fn new(lambda: usize) -> Self { - Self { - entries: Vec::new(), - lambda, - } - } - - /// Compute the transmit budget: `Λ * ceil(log₂(max(n, 2)))`. - pub 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.lambda * log_n.max(1) - } - - /// Enqueue an entry for dissemination. Does not check for duplicates. - pub fn enqueue(&mut self, item: T, cluster_size: usize) { - let budget = self.transmit_budget(cluster_size); - self.entries.push(BufferEntry { - item, - remaining: budget, - }); - } - - /// Enqueue an entry, replacing an existing one if `matcher` returns true. - /// If no match is found, pushes a new entry. - pub fn enqueue_or_replace(&mut self, item: T, cluster_size: usize, matcher: F) - where - F: Fn(&T) -> bool, - { - let budget = self.transmit_budget(cluster_size); - - if let Some(existing) = self.entries.iter_mut().find(|e| matcher(&e.item)) { - existing.item = item; - existing.remaining = budget; - return; - } - - self.entries.push(BufferEntry { - item, - remaining: budget, - }); - } - - /// Take up to `max_count` entries for piggyback. - /// Decrements remaining budget and evicts exhausted entries. - pub fn take(&mut self, max_count: usize) -> Vec { - let count = max_count.min(self.entries.len()); - let mut result = Vec::with_capacity(count); - - for entry in self.entries.iter_mut().take(count) { - result.push(entry.item.clone()); - entry.remaining = entry.remaining.saturating_sub(1); - } - - self.entries.retain(|e| e.remaining > 0); - result - } - - /// Re-enqueue all given items with fresh budgets. - pub fn re_enqueue_all(&mut self, items: impl IntoIterator, cluster_size: usize) { - for item in items { - self.enqueue(item, cluster_size); - } - } - - /// Retain only entries matching the predicate. - pub fn retain(&mut self, mut predicate: F) - where - F: FnMut(&T) -> bool, - { - self.entries.retain(|e| predicate(&e.item)); - } - - /// Number of queued entries. - pub fn len(&self) -> usize { - self.entries.len() - } - - /// Whether the buffer is empty. - pub fn is_empty(&self) -> bool { - self.entries.is_empty() - } -} - -// ─── Serialization helpers ───────────────────────────────────────────────── - -/// Serialize a slice of items to a Vec of byte vectors. -pub fn serialize_each(items: &[T]) -> Vec> { - items - .iter() - .filter_map(|item| serde_json::to_vec(item).ok()) - .collect() -} - -/// Deserialize a slice of byte vectors into items, skipping failures. -pub fn deserialize_each(entries: &[Vec]) -> Vec { - entries - .iter() - .filter_map(|bytes| serde_json::from_slice(bytes).ok()) - .collect() -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn budget_math_two_nodes() { - let buf: DisseminationBuffer = DisseminationBuffer::new(3); - // log2(2) = 1, ceil = 1, 3 * 1 = 3 - assert_eq!(buf.transmit_budget(2), 3); - } - - #[test] - fn budget_math_single_node_floors_to_two() { - let buf: DisseminationBuffer = DisseminationBuffer::new(3); - // cluster_size=1 → max(1,2)=2, log2(2)=1, 3*1=3 - assert_eq!(buf.transmit_budget(1), 3); - } - - #[test] - fn budget_math_32_nodes() { - let buf: DisseminationBuffer = DisseminationBuffer::new(3); - // log2(32) = 5, 3 * 5 = 15 - assert_eq!(buf.transmit_budget(32), 15); - } - - #[test] - fn budget_math_33_nodes() { - let buf: DisseminationBuffer = DisseminationBuffer::new(3); - // log2(33) ≈ 5.04, ceil = 6, 3 * 6 = 18 - assert_eq!(buf.transmit_budget(33), 18); - } - - #[test] - fn enqueue_take_evicts_after_budget() { - let mut buf = DisseminationBuffer::new(3); - buf.enqueue(42u32, 2); // budget = 3 - - // Take 3 times — each take decrements once - let r1 = buf.take(1); - assert_eq!(r1, vec![42]); - assert_eq!(buf.len(), 1); - - let r2 = buf.take(1); - assert_eq!(r2, vec![42]); - assert_eq!(buf.len(), 1); - - let r3 = buf.take(1); - assert_eq!(r3, vec![42]); - // After 3 takes, budget exhausted → evicted - assert_eq!(buf.len(), 0); - } - - #[test] - fn enqueue_or_replace_replaces_matching_entry() { - let mut buf = DisseminationBuffer::new(3); - buf.enqueue_or_replace(("key", 1), 2, |e| e.0 == "key"); - buf.enqueue_or_replace(("key", 2), 2, |e| e.0 == "key"); - - assert_eq!(buf.len(), 1); - let taken = buf.take(1); - assert_eq!(taken, vec![("key", 2)]); - } - - #[test] - fn enqueue_or_replace_adds_when_no_match() { - let mut buf = DisseminationBuffer::new(3); - buf.enqueue_or_replace(("a", 1), 2, |e| e.0 == "a"); - buf.enqueue_or_replace(("b", 2), 2, |e| e.0 == "b"); - - assert_eq!(buf.len(), 2); - } - - #[test] - fn re_enqueue_all_refreshes_budgets() { - let mut buf = DisseminationBuffer::new(3); - buf.enqueue(1u32, 2); - buf.enqueue(2u32, 2); - - // Drain them - for _ in 0..3 { - buf.take(2); - } - assert!(buf.is_empty()); - - // Re-enqueue - buf.re_enqueue_all(vec![1, 2, 3], 2); - assert_eq!(buf.len(), 3); - } - - #[test] - fn retain_removes_non_matching() { - let mut buf = DisseminationBuffer::new(3); - buf.enqueue(1u32, 2); - buf.enqueue(2u32, 2); - buf.enqueue(3u32, 2); - - buf.retain(|item| *item != 2); - assert_eq!(buf.len(), 2); - let taken = buf.take(3); - assert_eq!(taken, vec![1, 3]); - } - - #[test] - fn serialize_deserialize_roundtrip() { - let items = vec![1u32, 2, 3]; - let bytes = serialize_each(&items); - let recovered: Vec = deserialize_each(&bytes); - assert_eq!(recovered, items); - } -} diff --git a/crates/distribution/src/identity.rs b/crates/distribution/src/identity.rs deleted file mode 100644 index 4974508..0000000 --- a/crates/distribution/src/identity.rs +++ /dev/null @@ -1,2 +0,0 @@ -// Re-export identity utilities from transport. -pub use swactor_transport::identity::*; diff --git a/crates/distribution/src/iroh_driver.rs b/crates/distribution/src/iroh_driver.rs index 5fc5e98..14b7c7a 100644 --- a/crates/distribution/src/iroh_driver.rs +++ b/crates/distribution/src/iroh_driver.rs @@ -169,7 +169,7 @@ impl IrohDriver { if !allowed { eprintln!( "iroh driver: rejected connection from unauthorized peer {}", - crate::identity::hex_encode(&node_id.0[..4]) + swactor::transport::hex_encode(&node_id.0[..4]) ); conn.close(0u32.into(), b"unauthorized"); continue; @@ -179,13 +179,13 @@ impl IrohDriver { if negotiated_alpn == ALPN { eprintln!( "iroh driver: accepted SWIM connection from {}", - crate::identity::hex_encode(&node_id.0[..4]) + swactor::transport::hex_encode(&node_id.0[..4]) ); swim_buf.lock().unwrap().push((node_id, conn)); } else { eprintln!( "iroh driver: accepted non-SWIM connection from {} (ALPN: {})", - crate::identity::hex_encode(&node_id.0[..4]), + swactor::transport::hex_encode(&node_id.0[..4]), String::from_utf8_lossy(negotiated_alpn), ); other_buf.lock().unwrap().push((node_id, conn)); @@ -524,7 +524,7 @@ impl IrohDriver { if !self.is_peer_allowed(&node_id) { return Err(format!( "peer {} not in allow-list", - crate::identity::hex_encode(&node_id.0[..4]) + swactor::transport::hex_encode(&node_id.0[..4]) ) .into()); } diff --git a/crates/distribution/src/lib.rs b/crates/distribution/src/lib.rs index eb62498..5469829 100644 --- a/crates/distribution/src/lib.rs +++ b/crates/distribution/src/lib.rs @@ -1,11 +1,8 @@ pub mod types; pub mod crypto; -pub mod identity; pub mod peer_auth; pub mod messages; pub mod codec; -#[cfg(feature = "tcp")] -pub mod transport; pub mod swim; pub mod kademlia; pub mod cache; diff --git a/crates/distribution/src/peer_auth.rs b/crates/distribution/src/peer_auth.rs index 2e56153..d0016ae 100644 --- a/crates/distribution/src/peer_auth.rs +++ b/crates/distribution/src/peer_auth.rs @@ -9,7 +9,7 @@ use std::path::{Path, PathBuf}; use serde::{Deserialize, Serialize}; -use crate::identity::{hex_decode, hex_encode}; +use swactor::transport::{hex_decode, hex_encode}; use crate::types::NodeId; /// A single trusted peer entry. diff --git a/crates/distribution/src/swim/node.rs b/crates/distribution/src/swim/node.rs index 250b053..f8baade 100644 --- a/crates/distribution/src/swim/node.rs +++ b/crates/distribution/src/swim/node.rs @@ -3,7 +3,7 @@ //! This is the top-level SWIM state machine that a `DistributedNode` will drive. //! It produces `SwimAction`s that the caller translates into real network I/O. -use crate::identity::hex_encode; +use swactor::transport::hex_encode; use crate::messages::MembershipUpdate; use crate::types::{MemberState, NodeId, NodeRecord}; diff --git a/crates/distribution/src/transport.rs b/crates/distribution/src/transport.rs deleted file mode 100644 index 5d57799..0000000 --- a/crates/distribution/src/transport.rs +++ /dev/null @@ -1,2 +0,0 @@ -// Re-export TCP transport primitives from swactor-transport. -pub use swactor_transport::tcp::*; diff --git a/crates/distribution/src/types.rs b/crates/distribution/src/types.rs index e6e7875..e62e593 100644 --- a/crates/distribution/src/types.rs +++ b/crates/distribution/src/types.rs @@ -1,9 +1,8 @@ use serde::{Deserialize, Serialize}; use swactor::actor::ActorAddress; -// Re-export NodeId from transport and Signature from transport. -pub use swactor_transport::NodeId; -pub use swactor_transport::crypto::Signature; +pub use swactor::transport::NodeId; +pub use crate::crypto::Signature; // ─── NodeId extensions (Kademlia-specific) ───────────────────────────────── diff --git a/crates/distribution/tests/transport_and_codec.rs b/crates/distribution/tests/transport_and_codec.rs index a5bb710..48f9a39 100644 --- a/crates/distribution/tests/transport_and_codec.rs +++ b/crates/distribution/tests/transport_and_codec.rs @@ -83,7 +83,7 @@ mod tcp_transport { use distribution::codec::distribution_codec_registry; use distribution::messages::*; - use distribution::transport::{TcpAcceptor, TcpTransport}; + use swactor_transport::tcp::{TcpAcceptor, TcpTransport}; use distribution::types::NodeId; #[test] diff --git a/crates/node/Cargo.toml b/crates/node/Cargo.toml index d8b3934..1fc6a68 100644 --- a/crates/node/Cargo.toml +++ b/crates/node/Cargo.toml @@ -17,6 +17,10 @@ serde = { version = "1", features = ["derive"] } serde_json = "1" libc = "0.2" +[lib] +name = "swactor_node" +path = "src/lib.rs" + [features] default = ["iroh", "relay"] tcp = ["distribution/tcp"] diff --git a/crates/node/src/lib.rs b/crates/node/src/lib.rs new file mode 100644 index 0000000..66695e9 --- /dev/null +++ b/crates/node/src/lib.rs @@ -0,0 +1,6 @@ +pub mod config; +pub mod names; +pub mod plugins; +#[cfg(feature = "relay")] +pub mod relay; +pub mod install; diff --git a/crates/node/src/main.rs b/crates/node/src/main.rs index c679b92..e816508 100644 --- a/crates/node/src/main.rs +++ b/crates/node/src/main.rs @@ -14,7 +14,7 @@ use clap::{Parser, Subcommand}; use swactor::actor::{ActorInterface, Ctx}; use swactor::config::RuntimeConfig; use swactor::runtime::Runtime; -use swactor_transport::NodeId; +use swactor::transport::NodeId; use swactor_transport::crypto::Keypair; use swactor_transport::identity::{base58_encode, base58_decode, hex_encode, load_or_generate_keypair}; use distribution::node::DistributedNodeConfig; @@ -27,12 +27,9 @@ use dashboard::{start_dashboard, DashboardConfig}; use swactor_datastore::{DatastoreAuthConfig, DatastoreGroup, DatastoreGroupConfig}; -mod config; -mod install; -mod names; -mod plugins; +use swactor_node::{config, install, names, plugins}; #[cfg(feature = "relay")] -mod relay; +use swactor_node::relay; // ── CLI ────────────────────────────────────────────────────────────────── @@ -339,10 +336,15 @@ fn main() { cfg.listen.as_ref().and_then(|s| s.parse().ok()) }); - // Signal handler + // Signal handler — second Ctrl+C forces immediate exit { let stop = Arc::clone(&stop); ctrlc::set_handler(move || { + if stop.load(Ordering::Relaxed) { + eprintln!("\nForce exit."); + std::process::exit(1); + } + eprintln!("\nShutting down (press Ctrl+C again to force)..."); stop.store(true, Ordering::Relaxed); }) .expect("failed to set signal handler"); @@ -527,6 +529,7 @@ fn main() { }; // Distribution config + eprintln!("Distribution: SWIM (transport: {transport})"); let swim_config = SwimConfig { probe_interval: 5, probe_timeout: 6, // 600ms — allows relay round-trip @@ -600,7 +603,6 @@ fn main() { } } - eprintln!("\nShutting down..."); handle.shutdown(); dash.shutdown(); handle.join(); @@ -628,7 +630,8 @@ fn run_tcp( use distribution::driver::NodeDriver; let listen_addr = listen.expect("--listen is required for TCP mode"); - let mut driver = NodeDriver::with_keypair(listen_addr, keypair, node_config) + let dist_keypair = distribution::crypto::Keypair::from_bytes(&keypair.secret_bytes()); + let mut driver = NodeDriver::with_keypair(listen_addr, dist_keypair, node_config) .expect("failed to create node driver"); eprintln!( diff --git a/crates/node/src/plugins/datastore.rs b/crates/node/src/plugins/datastore.rs index 4fb90fa..2e232c9 100644 --- a/crates/node/src/plugins/datastore.rs +++ b/crates/node/src/plugins/datastore.rs @@ -231,7 +231,7 @@ fn start_datastore_group( chunk_size: u32, storage_path: Option, ) -> Result { - use swactor_transport::NodeId; + use swactor::transport::NodeId; use std::time::{SystemTime, UNIX_EPOCH}; // Generate a unique node ID diff --git a/crates/node/src/plugins/datastore_page.html b/crates/node/src/plugins/datastore_page.html index 0df8d49..4671a62 100644 --- a/crates/node/src/plugins/datastore_page.html +++ b/crates/node/src/plugins/datastore_page.html @@ -193,7 +193,7 @@ diff --git a/crates/node/src/plugins/distribution_page.html b/crates/node/src/plugins/distribution_page.html index 3b73368..81032f5 100644 --- a/crates/node/src/plugins/distribution_page.html +++ b/crates/node/src/plugins/distribution_page.html @@ -167,7 +167,7 @@ Overview Actors Distribution - Datastore + Datastore
diff --git a/crates/node/src/plugins/peers.rs b/crates/node/src/plugins/peers.rs index 2a1d537..b363529 100644 --- a/crates/node/src/plugins/peers.rs +++ b/crates/node/src/plugins/peers.rs @@ -8,7 +8,7 @@ use std::sync::{Arc, Mutex}; use dashboard::plugin::{DashboardPlugin, PluginResponse}; use dashboard::JoinPeerInfo; -use swactor_transport::NodeId; +use swactor::transport::NodeId; use swactor_transport::identity::{base58_decode, hex_decode}; use distribution::peer_auth::PeerAllowList; diff --git a/crates/simulation/Cargo.toml b/crates/simulation/Cargo.toml index 93c8b35..a6ce71b 100644 --- a/crates/simulation/Cargo.toml +++ b/crates/simulation/Cargo.toml @@ -5,11 +5,12 @@ edition = "2024" [features] default = [] +distribution = ["dep:distribution"] gossip = ["dep:log"] dashboard = ["gossip", "dep:tiny_http", "dep:toml"] [dependencies] -distribution = { path = "../distribution" } +distribution = { path = "../distribution", optional = true } swactor = { path = "../..", features = ["serde"] } serde = { version = "1", features = ["derive"] } serde_json = "1" @@ -19,7 +20,7 @@ tiny_http = { version = "0.12", optional = true } toml = { version = "0.8", optional = true } [dev-dependencies] -simulation = { path = ".", features = ["gossip"] } +simulation = { path = ".", features = ["gossip", "distribution"] } [[example]] name = "gossip_sim" diff --git a/crates/simulation/src/distribution/sim.rs b/crates/simulation/src/distribution/sim.rs index 8a3e267..7184aab 100644 --- a/crates/simulation/src/distribution/sim.rs +++ b/crates/simulation/src/distribution/sim.rs @@ -1,45 +1,15 @@ -use std::collections::{HashMap, HashSet}; - use distribution::node::{DistributedNode, DistributedNodeConfig, ResolveResult}; use distribution::swim::node::NodeAction; use distribution::swim::probe::SwimConfig; use distribution::types::NodeId; use swactor::actor::ActorAddress; +use crate::runner::NetworkState; +pub use crate::runner::{NetworkFault, NetworkTopology, NodeLocation, Partition}; use crate::trace::{Event, SimulationTrace}; use super::trace::{DistributionEventKind, DistributionSnapshot}; -/// Network location of a simulated node. -#[derive(Debug, Clone, PartialEq, Eq)] -pub enum NodeLocation { - /// Publicly reachable (e.g. cloud VPS). Can receive inbound from anyone. - Public, - /// Behind NAT. Can only receive inbound from same LAN group or via relay. - Nat { group: String }, - /// Completely firewalled — no inbound or outbound. - Firewalled, -} - -/// Network topology describing NAT/firewall/relay placement. -#[derive(Debug, Clone)] -pub struct NetworkTopology { - /// Per-node location (indexed by node_idx). Length must equal num_nodes. - pub locations: Vec, - /// Node indices that act as relay forwarders for cross-NAT traffic. - pub relay_nodes: Vec, -} - -/// A network partition between two sets of nodes. -/// Nodes in `side_a` cannot communicate with nodes in `side_b`. -#[derive(Debug, Clone)] -pub struct Partition { - pub side_a: Vec, - pub side_b: Vec, - /// If true, A→B is blocked but B→A works (asymmetric). - pub asymmetric: bool, -} - /// An action to execute at a specific round during the simulation. #[derive(Debug, Clone)] pub enum SimAction { @@ -57,21 +27,6 @@ pub enum SimAction { Introduce { node_a: usize, node_b: usize }, } -/// Schedule entry for network faults. -#[derive(Debug, Clone)] -pub enum NetworkFault { - /// Introduce a partition at the given round. - Partition { round: usize, partition: Partition }, - /// Heal a partition at the given round (restores full connectivity). - Heal { round: usize }, - /// Set message drop rate (0.0 = no drops, 1.0 = drop all). - SetDropRate { round: usize, rate: f64 }, - /// Per-link drop rate. rate=0.0 clears the fault. - LinkFault { round: usize, from: usize, to: usize, rate: f64, bidirectional: bool }, - /// Relay penalty — extra drop probability for relay-routed messages. - SetRelayPenalty { round: usize, rate: f64 }, -} - /// Configuration for a distribution simulation run. #[derive(Debug, Clone)] pub struct DistributionSimConfig { @@ -130,185 +85,6 @@ impl Default for DistributionSimConfig { } } -/// Tracks active network state during simulation. -struct NetworkState { - /// Set of (from_idx, to_idx) pairs where messages are blocked. - blocked: HashSet<(usize, usize)>, - /// Probability of dropping a message [0.0, 1.0]. - drop_rate: f64, - /// Simple counter-based deterministic "random" for drop decisions. - drop_counter: u64, - /// Optional NAT/firewall topology. - topology: Option, - /// Per-node alive status (indexed by node_idx). - alive: Vec, - /// Per-link drop rates (from, to) -> rate. - link_drop_rates: HashMap<(usize, usize), f64>, - /// Extra drop probability for relay-routed messages. - relay_penalty: f64, -} - -impl NetworkState { - fn new() -> Self { - Self { - blocked: HashSet::new(), - drop_rate: 0.0, - drop_counter: 0x853c49e6748fea9b, - topology: None, - alive: Vec::new(), - link_drop_rates: HashMap::new(), - relay_penalty: 0.0, - } - } - - fn new_with_topology(topology: Option, num_nodes: usize) -> Self { - Self { - blocked: HashSet::new(), - drop_rate: 0.0, - drop_counter: 0x853c49e6748fea9b, - topology, - alive: vec![true; num_nodes], - link_drop_rates: HashMap::new(), - relay_penalty: 0.0, - } - } - - fn set_alive(&mut self, idx: usize, alive: bool) { - if idx < self.alive.len() { - self.alive[idx] = alive; - } - } - - fn apply_fault(&mut self, fault: &NetworkFault, num_nodes: usize) { - match fault { - NetworkFault::Partition { partition, .. } => { - for &a in &partition.side_a { - for &b in &partition.side_b { - if a < num_nodes && b < num_nodes { - self.blocked.insert((a, b)); - if !partition.asymmetric { - self.blocked.insert((b, a)); - } - } - } - } - } - NetworkFault::Heal { .. } => { - self.blocked.clear(); - } - NetworkFault::SetDropRate { rate, .. } => { - self.drop_rate = rate.clamp(0.0, 1.0); - } - NetworkFault::LinkFault { from, to, rate, bidirectional, .. } => { - let rate = rate.clamp(0.0, 1.0); - if rate == 0.0 { - self.link_drop_rates.remove(&(*from, *to)); - if *bidirectional { - self.link_drop_rates.remove(&(*to, *from)); - } - } else { - self.link_drop_rates.insert((*from, *to), rate); - if *bidirectional { - self.link_drop_rates.insert((*to, *from), rate); - } - } - } - NetworkFault::SetRelayPenalty { rate, .. } => { - self.relay_penalty = rate.clamp(0.0, 1.0); - } - } - } - - /// Check if `from` can directly initiate a connection to `to`. - fn directly_reachable(&self, from: usize, to: usize) -> bool { - let topo = match &self.topology { - Some(t) => t, - None => return true, // No topology = full connectivity - }; - if from >= topo.locations.len() || to >= topo.locations.len() { - return true; - } - match (&topo.locations[from], &topo.locations[to]) { - (_, NodeLocation::Firewalled) => false, - (NodeLocation::Firewalled, _) => false, - (_, NodeLocation::Public) => true, // Anyone can reach public - (NodeLocation::Public, NodeLocation::Nat { .. }) => false, // Can't initiate inbound to NAT - (NodeLocation::Nat { group: g1 }, NodeLocation::Nat { group: g2 }) => g1 == g2, // Same LAN - } - } - - /// Check if two nodes can communicate (bidirectional once established). - /// Either direct reachability in either direction, or via a relay. - fn can_reach(&self, from: usize, to: usize) -> bool { - let topo = match &self.topology { - Some(t) => t, - None => return true, - }; - // Direct: if either side can initiate, the connection is bidirectional - if self.directly_reachable(from, to) || self.directly_reachable(to, from) { - return true; - } - // Relay path: any alive relay R where both endpoints can bidirectionally reach R - for &r in &topo.relay_nodes { - if r == from || r == to { - continue; - } - if !self.alive.get(r).copied().unwrap_or(false) { - continue; - } - let from_reaches_r = self.directly_reachable(from, r) || self.directly_reachable(r, from); - let to_reaches_r = self.directly_reachable(to, r) || self.directly_reachable(r, to); - if from_reaches_r && to_reaches_r { - return true; - } - } - false - } - - /// Returns true when neither direction is directly reachable but a relay path exists. - fn requires_relay(&self, from: usize, to: usize) -> bool { - if self.topology.is_none() { - return false; - } - if self.directly_reachable(from, to) || self.directly_reachable(to, from) { - return false; - } - self.can_reach(from, to) - } - - /// Returns true if this message should be delivered. - fn should_deliver(&mut self, from_idx: usize, to_idx: usize) -> bool { - // 1. Check partition blocks - if self.blocked.contains(&(from_idx, to_idx)) { - return false; - } - // 2. Check NAT reachability (only if topology is set) - if self.topology.is_some() && !self.can_reach(from_idx, to_idx) { - return false; - } - // 3. Determine effective drop rate: per-link if set, else global - let base_rate = self.link_drop_rates - .get(&(from_idx, to_idx)) - .copied() - .unwrap_or(self.drop_rate); - // 4. Compose relay penalty if applicable - let effective_rate = if self.relay_penalty > 0.0 && self.requires_relay(from_idx, to_idx) { - 1.0 - (1.0 - base_rate) * (1.0 - self.relay_penalty) - } else { - base_rate - }; - // 5. Apply effective rate via LCG PRNG - if effective_rate > 0.0 { - self.drop_counter = self.drop_counter.wrapping_mul(6364136223846793005).wrapping_add(1); - let r = (self.drop_counter >> 33) as f64 / (u32::MAX as f64); - if r < effective_rate { - return false; - } - } - true - } -} - pub type DistTrace = SimulationTrace; /// Run a distribution simulation, returning both the trace and the final node states. @@ -446,14 +222,7 @@ fn run_simulation_inner(config: DistributionSimConfig) -> (DistTrace, Vec