diff --git a/.gitignore b/.gitignore index 81a3525..59fb7e8 100644 --- a/.gitignore +++ b/.gitignore @@ -24,4 +24,6 @@ deploy.toml !.deploy/deploy.example.toml # Local dev node state -.dev-node/ \ No newline at end of file +.dev-node/ +.dev-cluster/ +.sim-cluster/ \ No newline at end of file diff --git a/Cargo.lock b/Cargo.lock index e00b73d..2f5c85a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6949,9 +6949,12 @@ name = "xtask" version = "0.1.0" dependencies = [ "clap", + "iroh-relay", "libc", + "reqwest 0.12.28", "serde", "serde_json", + "tokio", "toml 0.8.23", ] diff --git a/crates/dashboard/src/pool_html.rs b/crates/dashboard/src/pool_html.rs new file mode 100644 index 0000000..71c7a53 --- /dev/null +++ b/crates/dashboard/src/pool_html.rs @@ -0,0 +1,392 @@ +pub const POOL_HTML: &str = r##" + + + + +Swactor Runtime – Pool + + + +
+
+

+ Swactor Runtime Dashboard + +

+ +
+
+ + +
+
+ + + + + +
+ + + + +"##; diff --git a/crates/datastore/src/pool/coordinator.rs b/crates/datastore/src/pool/coordinator.rs new file mode 100644 index 0000000..e9f7dd5 --- /dev/null +++ b/crates/datastore/src/pool/coordinator.rs @@ -0,0 +1,274 @@ +//! Pool coordinator actor — placement-aware CRUD facade. +//! +//! Owns an `Arc>` for query access and delegates +//! storage operations to the co-located `DatastoreNode` actor. + +use std::collections::BTreeMap; +use std::sync::{Arc, Mutex}; + +use swactor::actor::{ActorAddress, ActorInterface, Ctx}; + +use distribution::types::NodeId; +use shared_types::ContentHash; +use shared_types::pool::PoolConfig; + +use crate::messages::{DatastoreNodeMsg, DatastoreResponse}; +use super::disseminator::PoolDisseminator; +use super::messages::PoolCoordinatorMsg; + +/// The pool coordinator actor. +pub struct PoolCoordinator { + node_id: NodeId, + pool_config: PoolConfig, + disseminator: Arc>, + + // Co-located actor addresses + datastore_addr: ActorAddress, + + tick_count: u64, +} + +impl PoolCoordinator { + pub fn new( + node_id: NodeId, + pool_config: PoolConfig, + disseminator: Arc>, + datastore_addr: ActorAddress, + ) -> Self { + Self { + node_id, + pool_config, + disseminator, + datastore_addr, + tick_count: 0, + } + } + + /// Access the shared disseminator. + pub fn disseminator(&self) -> &Arc> { + &self.disseminator + } + + fn cluster_size(&self) -> usize { + let d = self.disseminator.lock().unwrap(); + d.member_count().max(1) + } + + // ─── Message handlers ────────────────────────────────────────────── + + fn handle_pool_put( + &self, + ctx: &Ctx, + data: Vec, + name: Option, + tags: BTreeMap, + reply_to: ActorAddress, + ) { + // Delegate to local DatastoreNode for now. + // Future: check capacity and redirect to best node. + let _ = ctx.send( + self.datastore_addr, + DatastoreNodeMsg::Put { + data, + name, + tags, + reply_to, + }, + ); + + // Note: content announcement happens after PutOk is received. + // For now, caller is responsible for announcing content via PoolTick + // or a future PutOk callback. + } + + fn handle_pool_get( + &self, + ctx: &Ctx, + content_hash: ContentHash, + reply_to: ActorAddress, + ) { + // Try local first via DatastoreNode + let _ = ctx.send( + self.datastore_addr, + DatastoreNodeMsg::Get { + content_hash, + reply_to, + }, + ); + + // Future: if local not found, use disseminator.locate_content() + // to fetch from a specific peer instead of fan-out. + } + + fn handle_pool_delete( + &self, + ctx: &Ctx, + content_hash: ContentHash, + reply_to: ActorAddress, + ) { + // Delete locally + let _ = ctx.send( + self.datastore_addr, + DatastoreNodeMsg::Delete { + content_hash, + reply_to, + }, + ); + + // Announce tombstone via gossip + let cluster_size = self.cluster_size(); + self.disseminator + .lock() + .unwrap() + .remove_content(content_hash, cluster_size); + } + + fn handle_pool_list( + &self, + ctx: &Ctx, + name_filter: Option, + reply_to: ActorAddress, + ) { + let _ = ctx.send( + self.datastore_addr, + DatastoreNodeMsg::List { + name_filter, + all: false, + reply_to, + }, + ); + } + + fn handle_pool_status(&self, ctx: &Ctx, reply_to: ActorAddress) { + let d = self.disseminator.lock().unwrap(); + let (total_bytes, used_bytes) = d.pool_capacity_summary(); + let members: Vec = d + .active_members() + .iter() + .map(|id| id.0.iter().map(|b| format!("{b:02x}")).collect()) + .collect(); + + let member_count = d.member_count(); + let content_count = d.content_count(); + drop(d); + + let json = serde_json::json!({ + "pool_name": self.pool_config.pool_name, + "pool_id": self.pool_config.pool_id.to_hex(), + "member_count": member_count, + "content_count": content_count, + "total_bytes": total_bytes, + "used_bytes": used_bytes, + "members": members, + }); + + let _ = ctx.send( + reply_to, + DatastoreResponse::PoolStatus { + json: json.to_string(), + }, + ); + } + + fn handle_join_pool(&mut self, ctx: &Ctx, reply_to: ActorAddress) { + let cluster_size = self.cluster_size(); + let mut d = self.disseminator.lock().unwrap(); + + if !d.is_node_authorized(&self.node_id) { + let _ = ctx.send( + reply_to, + DatastoreResponse::Error { + reason: "not authorized to join pool".into(), + }, + ); + return; + } + + d.join(cluster_size); + d.announce_capacity(self.pool_config.capacity_bytes, 0, cluster_size); + drop(d); + + let _ = ctx.send(reply_to, DatastoreResponse::Bool(true)); + } + + fn handle_leave_pool(&mut self, ctx: &Ctx, reply_to: ActorAddress) { + let cluster_size = self.cluster_size(); + self.disseminator.lock().unwrap().leave(cluster_size); + let _ = ctx.send(reply_to, DatastoreResponse::Bool(true)); + } + + fn handle_grant_access(&mut self, ctx: &Ctx, target: NodeId, reply_to: ActorAddress) { + let cluster_size = self.cluster_size(); + self.disseminator + .lock() + .unwrap() + .grant_access(target, cluster_size); + let _ = ctx.send(reply_to, DatastoreResponse::Bool(true)); + } + + fn handle_revoke_access(&mut self, ctx: &Ctx, target: NodeId, reply_to: ActorAddress) { + let cluster_size = self.cluster_size(); + self.disseminator + .lock() + .unwrap() + .revoke_access(target, cluster_size); + let _ = ctx.send(reply_to, DatastoreResponse::Bool(true)); + } + + fn handle_pool_tick(&mut self) { + self.tick_count += 1; + // Periodic capacity re-announcement (every 100 ticks) + if self.tick_count % 100 == 0 { + let cluster_size = self.cluster_size(); + self.disseminator.lock().unwrap().announce_capacity( + self.pool_config.capacity_bytes, + 0, // TODO: query actual usage from BlobStore + cluster_size, + ); + } + } +} + +impl ActorInterface for PoolCoordinator { + type Incoming = PoolCoordinatorMsg; + type Response = DatastoreResponse; + + fn handle(&mut self, ctx: &Ctx, msg: PoolCoordinatorMsg) { + match msg { + PoolCoordinatorMsg::PoolPut { + data, + name, + tags, + reply_to, + } => self.handle_pool_put(ctx, data, name, tags, reply_to), + PoolCoordinatorMsg::PoolGet { + content_hash, + reply_to, + } => self.handle_pool_get(ctx, content_hash, reply_to), + PoolCoordinatorMsg::PoolDelete { + content_hash, + reply_to, + } => self.handle_pool_delete(ctx, content_hash, reply_to), + PoolCoordinatorMsg::PoolList { + name_filter, + reply_to, + } => self.handle_pool_list(ctx, name_filter, reply_to), + PoolCoordinatorMsg::PoolStatus { reply_to } => { + self.handle_pool_status(ctx, reply_to) + } + PoolCoordinatorMsg::JoinPool { reply_to } => { + self.handle_join_pool(ctx, reply_to) + } + PoolCoordinatorMsg::LeavePool { reply_to } => { + self.handle_leave_pool(ctx, reply_to) + } + PoolCoordinatorMsg::GrantPoolAccess { target, reply_to } => { + self.handle_grant_access(ctx, target, reply_to) + } + PoolCoordinatorMsg::RevokePoolAccess { target, reply_to } => { + self.handle_revoke_access(ctx, target, reply_to) + } + PoolCoordinatorMsg::PoolTick => self.handle_pool_tick(), + } + } +} diff --git a/crates/datastore/src/pool/disseminator.rs b/crates/datastore/src/pool/disseminator.rs new file mode 100644 index 0000000..8af5040 --- /dev/null +++ b/crates/datastore/src/pool/disseminator.rs @@ -0,0 +1,716 @@ +//! Pool disseminator — gossip-converged state for pool membership, +//! capacity, content locations, and ACL. +//! +//! Implements `GossipChannel` to plug into the generic gossip system +//! via `DistributedNode::register_channel()`. + +use std::collections::HashMap; +use std::sync::{Arc, Mutex}; + +use distribution::gossip_channel::{DisseminationBuffer, GossipChannel, deserialize_each, serialize_each}; +use distribution::types::NodeId; +use shared_types::ContentHash; +use shared_types::pool::*; + +// ─── PoolDisseminator ────────────────────────────────────────────────────── + +/// Configuration for the pool disseminator. +#[derive(Debug, Clone)] +pub struct PoolDisseminatorConfig { + pub tombstone_ttl: u64, + pub gc_interval: u64, +} + +impl Default for PoolDisseminatorConfig { + fn default() -> Self { + Self { + tombstone_ttl: 3600, + gc_interval: 1000, + } + } +} + +/// Manages converged pool state via gossip dissemination. +#[derive(Debug)] +pub struct PoolDisseminator { + pool_id: PoolId, + pool_name: String, + local_node_id: NodeId, + + // Converged state maps + members: HashMap<[u8; 32], PoolMemberEntry>, + capacity: HashMap<[u8; 32], PoolCapacityEntry>, + content_locations: HashMap<(ContentHash, [u8; 32]), ContentLocationEntry>, + acl: HashMap<[u8; 32], PoolACLEntry>, + + // Dissemination buffer + buffer: DisseminationBuffer, + + // Local generation counters + local_member_gen: u64, + local_capacity_gen: u64, + + config: PoolDisseminatorConfig, + tick_count: u64, +} + +impl PoolDisseminator { + pub fn new(pool_id: PoolId, pool_name: String, local_node_id: NodeId, lambda: usize) -> Self { + Self { + pool_id, + pool_name, + local_node_id, + members: HashMap::new(), + capacity: HashMap::new(), + content_locations: HashMap::new(), + acl: HashMap::new(), + buffer: DisseminationBuffer::new(lambda), + local_member_gen: 0, + local_capacity_gen: 0, + config: PoolDisseminatorConfig::default(), + tick_count: 0, + } + } + + pub fn with_config(mut self, config: PoolDisseminatorConfig) -> Self { + self.config = config; + self + } + + pub fn pool_id(&self) -> PoolId { + self.pool_id + } + + // ─── Lifecycle methods ────────────────────────────────────────────── + + /// Join the pool. Announces Active membership. + pub fn join(&mut self, cluster_size: usize) { + self.local_member_gen += 1; + let entry = PoolMemberEntry { + pool_id: self.pool_id, + node_id: self.local_node_id.0, + state: PoolMemberState::Active, + generation: self.local_member_gen, + }; + self.merge_membership(entry.clone()); + self.buffer.enqueue(PoolEntry::Membership(entry), cluster_size); + } + + /// Leave the pool. Announces Left membership. + pub fn leave(&mut self, cluster_size: usize) { + self.local_member_gen += 1; + let entry = PoolMemberEntry { + pool_id: self.pool_id, + node_id: self.local_node_id.0, + state: PoolMemberState::Left, + generation: self.local_member_gen, + }; + self.merge_membership(entry.clone()); + self.buffer.enqueue(PoolEntry::Membership(entry), cluster_size); + } + + /// Announce storage capacity. + pub fn announce_capacity(&mut self, total: u64, used: u64, cluster_size: usize) { + self.local_capacity_gen += 1; + let entry = PoolCapacityEntry { + pool_id: self.pool_id, + node_id: self.local_node_id.0, + total_bytes: total, + used_bytes: used, + generation: self.local_capacity_gen, + }; + self.merge_capacity(entry.clone()); + self.buffer.enqueue(PoolEntry::Capacity(entry), cluster_size); + } + + /// Announce that this node has a piece of content. + pub fn announce_content(&mut self, hash: ContentHash, cluster_size: usize) { + let key = (hash, self.local_node_id.0); + let next_gen = self.content_locations.get(&key).map_or(1, |e| e.generation + 1); + let entry = ContentLocationEntry { + pool_id: self.pool_id, + content_hash: hash, + node_id: self.local_node_id.0, + generation: next_gen, + tombstone: false, + }; + self.merge_content_location(entry.clone()); + self.buffer.enqueue(PoolEntry::ContentLocation(entry), cluster_size); + } + + /// Remove content announcement (tombstone). + pub fn remove_content(&mut self, hash: ContentHash, cluster_size: usize) { + let key = (hash, self.local_node_id.0); + let next_gen = self.content_locations.get(&key).map_or(1, |e| e.generation + 1); + let entry = ContentLocationEntry { + pool_id: self.pool_id, + content_hash: hash, + node_id: self.local_node_id.0, + generation: next_gen, + tombstone: true, + }; + self.merge_content_location(entry.clone()); + self.buffer.enqueue(PoolEntry::ContentLocation(entry), cluster_size); + } + + /// Grant access to a node. + pub fn grant_access(&mut self, target: NodeId, cluster_size: usize) { + let next_gen = self.acl.get(&target.0).map_or(1, |e| e.generation + 1); + let entry = PoolACLEntry { + pool_id: self.pool_id, + node_id: target.0, + granted_by: self.local_node_id.0, + generation: next_gen, + revoked: false, + }; + self.merge_acl(entry.clone()); + self.buffer.enqueue(PoolEntry::ACL(entry), cluster_size); + } + + /// Revoke access from a node. + pub fn revoke_access(&mut self, target: NodeId, cluster_size: usize) { + let next_gen = self.acl.get(&target.0).map_or(1, |e| e.generation + 1); + let entry = PoolACLEntry { + pool_id: self.pool_id, + node_id: target.0, + granted_by: self.local_node_id.0, + generation: next_gen, + revoked: true, + }; + self.merge_acl(entry.clone()); + self.buffer.enqueue(PoolEntry::ACL(entry), cluster_size); + } + + // ─── Query API ────────────────────────────────────────────────────── + + /// All active pool members. + pub fn active_members(&self) -> Vec { + self.members + .values() + .filter(|m| m.state == PoolMemberState::Active) + .map(|m| NodeId(m.node_id)) + .collect() + } + + /// Total and used capacity across the pool. + pub fn pool_capacity_summary(&self) -> (u64, u64) { + let mut total = 0u64; + let mut used = 0u64; + for cap in self.capacity.values() { + // Only count active members + if let Some(m) = self.members.get(&cap.node_id) { + if m.state == PoolMemberState::Active { + total = total.saturating_add(cap.total_bytes); + used = used.saturating_add(cap.used_bytes); + } + } + } + (total, used) + } + + /// Find which nodes have a given content hash. + pub fn locate_content(&self, hash: &ContentHash) -> Vec { + self.content_locations + .iter() + .filter(|((h, _), entry)| h == hash && !entry.tombstone) + .map(|((_, node_id), _)| NodeId(*node_id)) + .collect() + } + + /// Find the node with the most free space. + pub fn node_with_most_free_space(&self) -> Option { + self.capacity + .values() + .filter(|cap| { + self.members + .get(&cap.node_id) + .is_some_and(|m| m.state == PoolMemberState::Active) + }) + .max_by_key(|cap| cap.total_bytes.saturating_sub(cap.used_bytes)) + .map(|cap| NodeId(cap.node_id)) + } + + /// Check if a node is authorized to join this pool. + pub fn is_node_authorized(&self, node_id: &NodeId) -> bool { + // If no ACL entries exist, the pool is open + if self.acl.is_empty() { + return true; + } + self.acl + .get(&node_id.0) + .is_some_and(|entry| !entry.revoked) + } + + /// Number of active members. + pub fn member_count(&self) -> usize { + self.members + .values() + .filter(|m| m.state == PoolMemberState::Active) + .count() + } + + /// Number of live content location entries (non-tombstone). + pub fn content_count(&self) -> usize { + self.content_locations + .values() + .filter(|e| !e.tombstone) + .count() + } + + /// Serialize the current pool state to a JSON string for the dashboard. + pub fn snapshot_json(&self) -> String { + fn hex(bytes: &[u8; 32]) -> String { + bytes.iter().map(|b| format!("{b:02x}")).collect() + } + + let (total_bytes, used_bytes) = self.pool_capacity_summary(); + + let members: Vec = self + .members + .values() + .filter(|m| m.state == PoolMemberState::Active) + .map(|m| { + let cap = self.capacity.get(&m.node_id); + serde_json::json!({ + "node_id": hex(&m.node_id), + "state": format!("{:?}", m.state), + "generation": m.generation, + "total_bytes": cap.map_or(0, |c| c.total_bytes), + "used_bytes": cap.map_or(0, |c| c.used_bytes), + }) + }) + .collect(); + + // Group content locations by hash + let mut by_hash: HashMap> = HashMap::new(); + for ((hash, _), entry) in &self.content_locations { + if !entry.tombstone { + by_hash.entry(*hash).or_default().push(entry.node_id); + } + } + let content_locations: Vec = by_hash + .iter() + .map(|(hash, nodes)| { + serde_json::json!({ + "content_hash": hash.to_hex(), + "nodes": nodes.iter().map(hex).collect::>(), + "replica_count": nodes.len(), + }) + }) + .collect(); + + let acl: Vec = self + .acl + .values() + .map(|a| { + serde_json::json!({ + "node_id": hex(&a.node_id), + "granted_by": hex(&a.granted_by), + "revoked": a.revoked, + }) + }) + .collect(); + + let acl_mode = if self.acl.is_empty() { "open" } else { "allow-list" }; + + serde_json::json!({ + "pool_name": self.pool_name, + "pool_id": self.pool_id.to_hex(), + "member_count": self.member_count(), + "content_count": self.content_count(), + "total_bytes": total_bytes, + "used_bytes": used_bytes, + "members": members, + "content_locations": content_locations, + "acl": acl, + "acl_mode": acl_mode, + }) + .to_string() + } + + // ─── Internal merge logic ─────────────────────────────────────────── + + fn merge_membership(&mut self, entry: PoolMemberEntry) -> bool { + let key = entry.node_id; + if let Some(existing) = self.members.get(&key) { + if entry.generation <= existing.generation { + return false; + } + } + self.members.insert(key, entry); + true + } + + fn merge_capacity(&mut self, entry: PoolCapacityEntry) -> bool { + let key = entry.node_id; + if let Some(existing) = self.capacity.get(&key) { + if entry.generation <= existing.generation { + return false; + } + } + self.capacity.insert(key, entry); + true + } + + fn merge_content_location(&mut self, entry: ContentLocationEntry) -> bool { + let key = (entry.content_hash, entry.node_id); + if let Some(existing) = self.content_locations.get(&key) { + if entry.generation <= existing.generation { + return false; + } + } + self.content_locations.insert(key, entry); + true + } + + fn merge_acl(&mut self, entry: PoolACLEntry) -> bool { + let key = entry.node_id; + if let Some(existing) = self.acl.get(&key) { + if entry.generation <= existing.generation { + return false; + } + } + self.acl.insert(key, entry); + true + } + + /// Merge a single pool entry and return whether state changed. + fn merge_entry(&mut self, entry: PoolEntry) -> bool { + match entry { + PoolEntry::Membership(m) => self.merge_membership(m), + PoolEntry::Capacity(c) => self.merge_capacity(c), + PoolEntry::ContentLocation(cl) => self.merge_content_location(cl), + PoolEntry::ACL(a) => self.merge_acl(a), + } + } + + /// Take pending entries (internal, typed). + fn take_pending_inner(&mut self, max_count: usize) -> Vec { + self.buffer.take(max_count) + } + + /// Apply incoming entries (internal, typed). + fn apply_incoming_inner(&mut self, entries: Vec, cluster_size: usize) { + for entry in entries { + if self.merge_entry(entry.clone()) { + self.buffer.enqueue(entry, cluster_size); + } + } + } + + /// Re-enqueue all state (internal). + fn re_disseminate_all_inner(&mut self, cluster_size: usize) { + let mut all_entries: Vec = Vec::new(); + + for m in self.members.values().cloned() { + all_entries.push(PoolEntry::Membership(m)); + } + for c in self.capacity.values().cloned() { + all_entries.push(PoolEntry::Capacity(c)); + } + for cl in self.content_locations.values().cloned() { + all_entries.push(PoolEntry::ContentLocation(cl)); + } + for a in self.acl.values().cloned() { + all_entries.push(PoolEntry::ACL(a)); + } + + self.buffer.re_enqueue_all(all_entries, cluster_size); + } + + /// GC: evict tombstones past TTL. + fn gc_tick_inner(&mut self) { + self.tick_count += 1; + if self.tick_count % self.config.gc_interval != 0 { + return; + } + + let ttl = self.config.tombstone_ttl; + let tick = self.tick_count; + + // GC left members + self.members.retain(|_, m| { + if m.state == PoolMemberState::Left { + m.generation + ttl > tick + } else { + true + } + }); + + // GC tombstoned content locations + self.content_locations.retain(|_, cl| { + if cl.tombstone { + cl.generation + ttl > tick + } else { + true + } + }); + + // GC revoked ACL entries + self.acl.retain(|_, a| { + if a.revoked { + a.generation + ttl > tick + } else { + true + } + }); + } +} + +// ─── SharedPoolChannel ───────────────────────────────────────────────────── + +/// Wrapper around `Arc>` that implements `GossipChannel`. +/// +/// This enables shared ownership between the `PoolCoordinator` actor +/// (which needs query/lifecycle access) and `DistributedNode` (which +/// drives gossip piggyback). +pub struct SharedPoolChannel { + inner: Arc>, +} + +impl SharedPoolChannel { + pub fn new(disseminator: Arc>) -> Self { + Self { inner: disseminator } + } + + /// Consume the channel and return the underlying `Arc>`. + pub fn into_inner(self) -> Arc> { + self.inner + } +} + +impl GossipChannel for SharedPoolChannel { + fn topic_tag(&self) -> &'static str { + "pool" + } + + fn take_pending_bytes(&mut self, max_entries: usize) -> Vec> { + let entries = self.inner.lock().unwrap().take_pending_inner(max_entries); + serialize_each(&entries) + } + + fn apply_incoming_bytes(&mut self, entries: &[Vec], cluster_size: usize) { + let parsed: Vec = deserialize_each(entries); + self.inner.lock().unwrap().apply_incoming_inner(parsed, cluster_size); + } + + fn re_disseminate_all(&mut self, cluster_size: usize) { + self.inner.lock().unwrap().re_disseminate_all_inner(cluster_size); + } + + fn on_node_death(&mut self, _node_id: &NodeId) { + // Pool membership is explicit (join/leave), not auto-removed on node death. + // Capacity becomes unreliable but we don't remove it. + } + + fn gc_tick(&mut self) { + self.inner.lock().unwrap().gc_tick_inner(); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn node_id(b: u8) -> NodeId { + NodeId([b; 32]) + } + + fn make_disseminator(b: u8) -> PoolDisseminator { + PoolDisseminator::new( + PoolId::from_name("test-pool"), + "test-pool".into(), + node_id(b), + 3, + ) + } + + #[test] + fn join_and_query_members() { + let mut d = make_disseminator(1); + d.join(2); + + assert_eq!(d.member_count(), 1); + assert_eq!(d.active_members(), vec![node_id(1)]); + } + + #[test] + fn leave_removes_from_active() { + let mut d = make_disseminator(1); + d.join(2); + d.leave(2); + + assert_eq!(d.member_count(), 0); + assert!(d.active_members().is_empty()); + } + + #[test] + fn announce_and_locate_content() { + let mut d = make_disseminator(1); + d.join(2); + let hash = ContentHash::of(b"test-data"); + d.announce_content(hash, 2); + + let locations = d.locate_content(&hash); + assert_eq!(locations, vec![node_id(1)]); + } + + #[test] + fn remove_content_tombstones() { + let mut d = make_disseminator(1); + d.join(2); + let hash = ContentHash::of(b"test-data"); + d.announce_content(hash, 2); + d.remove_content(hash, 2); + + assert!(d.locate_content(&hash).is_empty()); + } + + #[test] + fn capacity_summary() { + let mut d = make_disseminator(1); + d.join(2); + d.announce_capacity(1000, 300, 2); + + let (total, used) = d.pool_capacity_summary(); + assert_eq!(total, 1000); + assert_eq!(used, 300); + } + + #[test] + fn acl_grant_and_check() { + let mut d = make_disseminator(1); + d.grant_access(node_id(2), 2); + + assert!(d.is_node_authorized(&node_id(2))); + assert!(!d.is_node_authorized(&node_id(3))); + } + + #[test] + fn acl_revoke() { + let mut d = make_disseminator(1); + d.grant_access(node_id(2), 2); + d.revoke_access(node_id(2), 2); + + assert!(!d.is_node_authorized(&node_id(2))); + } + + #[test] + fn empty_acl_means_open() { + let d = make_disseminator(1); + assert!(d.is_node_authorized(&node_id(99))); + } + + #[test] + fn higher_generation_wins_merge() { + let mut d1 = make_disseminator(1); + let mut d2 = make_disseminator(2); + + // d1 joins + d1.join(2); + // d1 leaves + d1.leave(2); + + // Gossip d1's entries to d2 out of order: + // First send the Active (gen 1), then the Left (gen 2) + let active_entry = PoolEntry::Membership(PoolMemberEntry { + pool_id: PoolId::from_name("test-pool"), + node_id: [1u8; 32], + state: PoolMemberState::Active, + generation: 1, + }); + let left_entry = PoolEntry::Membership(PoolMemberEntry { + pool_id: PoolId::from_name("test-pool"), + node_id: [1u8; 32], + state: PoolMemberState::Left, + generation: 2, + }); + + // Apply Left first (gen 2), then Active (gen 1) — Active should be rejected + d2.merge_entry(left_entry); + let changed = d2.merge_entry(active_entry); + assert!(!changed, "lower generation should not win"); + + // d2 should see node_id(1) as Left + assert_eq!(d2.member_count(), 0); // Active count is 0 + } + + #[test] + fn two_disseminators_converge_via_gossip_exchange() { + let mut d1 = make_disseminator(1); + let mut d2 = make_disseminator(2); + + // d1 joins and announces content + d1.join(2); + let hash = ContentHash::of(b"shared-file"); + d1.announce_content(hash, 2); + + // d2 joins + d2.join(2); + + // Simulate gossip: d1 → d2 + let pending = d1.take_pending_inner(100); + let bytes = serialize_each(&pending); + let parsed: Vec = deserialize_each(&bytes); + d2.apply_incoming_inner(parsed, 2); + + // d2 should now see d1 as a member and know about the content + assert_eq!(d2.member_count(), 2); + assert_eq!(d2.locate_content(&hash), vec![node_id(1)]); + + // Simulate gossip: d2 → d1 + let pending = d2.take_pending_inner(100); + let bytes = serialize_each(&pending); + let parsed: Vec = deserialize_each(&bytes); + d1.apply_incoming_inner(parsed, 2); + + // d1 should now see d2 as a member + assert_eq!(d1.member_count(), 2); + } + + #[test] + fn three_node_convergence_loop() { + let pool = PoolId::from_name("test-pool"); + let mut nodes: Vec = (0..3) + .map(|i| PoolDisseminator::new(pool, "test-pool".into(), node_id(i as u8), 3)) + .collect(); + + // Each node joins + for n in &mut nodes { + n.join(3); + } + + // Node 0 announces content + let hash = ContentHash::of(b"convergence-test"); + nodes[0].announce_content(hash, 3); + + // Run 5 gossip rounds where each node exchanges with all others + for _ in 0..5 { + // Collect pending from each node + let pending_bytes: Vec>> = nodes + .iter_mut() + .map(|n| serialize_each(&n.take_pending_inner(100))) + .collect(); + + // Apply each node's pending to all other nodes + for (sender_idx, bytes) in pending_bytes.iter().enumerate() { + for (receiver_idx, node) in nodes.iter_mut().enumerate() { + if sender_idx != receiver_idx { + let parsed: Vec = deserialize_each(bytes); + node.apply_incoming_inner(parsed, 3); + } + } + } + } + + // All nodes should agree on membership and content locations + for (i, node) in nodes.iter().enumerate() { + assert_eq!(node.member_count(), 3, "node {i} should see 3 members"); + assert_eq!( + node.locate_content(&hash), + vec![node_id(0)], + "node {i} should know content is on node 0" + ); + } + } +} diff --git a/crates/datastore/src/pool/messages.rs b/crates/datastore/src/pool/messages.rs new file mode 100644 index 0000000..de3fd7e --- /dev/null +++ b/crates/datastore/src/pool/messages.rs @@ -0,0 +1,79 @@ +//! Messages for the `PoolCoordinator` actor. + +use std::collections::BTreeMap; + +use distribution::types::NodeId; +use shared_types::ContentHash; + +/// Messages handled by the `PoolCoordinator` actor. +#[derive(Debug, Clone)] +pub enum PoolCoordinatorMsg { + // ── User-facing operations ────────────────────────────────────────── + /// Store data in the pool (placement-aware). + PoolPut { + data: Vec, + name: Option, + tags: BTreeMap, + reply_to: swactor::actor::ActorAddress, + }, + /// Retrieve data from the pool (location-aware). + PoolGet { + content_hash: ContentHash, + reply_to: swactor::actor::ActorAddress, + }, + /// Delete data from the pool. + PoolDelete { + content_hash: ContentHash, + reply_to: swactor::actor::ActorAddress, + }, + /// List objects in the pool. + PoolList { + name_filter: Option, + reply_to: swactor::actor::ActorAddress, + }, + /// Pool status (members, capacity, content count). + PoolStatus { + reply_to: swactor::actor::ActorAddress, + }, + + // ── Pool lifecycle ────────────────────────────────────────────────── + /// Join the pool. + JoinPool { + reply_to: swactor::actor::ActorAddress, + }, + /// Leave the pool. + LeavePool { + reply_to: swactor::actor::ActorAddress, + }, + + // ── Auth management ───────────────────────────────────────────────── + /// Grant a node access to the pool. + GrantPoolAccess { + target: NodeId, + reply_to: swactor::actor::ActorAddress, + }, + /// Revoke a node's access to the pool. + RevokePoolAccess { + target: NodeId, + reply_to: swactor::actor::ActorAddress, + }, + + // ── Periodic ──────────────────────────────────────────────────────── + /// Periodic tick: announce capacity, drive dissemination. + PoolTick, +} + +/// Pool-specific response variants. +#[derive(Debug, Clone)] +pub enum PoolResponse { + /// Pool status snapshot. + PoolStatus { + pool_name: String, + pool_id_hex: String, + member_count: usize, + content_count: usize, + total_bytes: u64, + used_bytes: u64, + members: Vec, + }, +} diff --git a/crates/datastore/src/pool/mod.rs b/crates/datastore/src/pool/mod.rs new file mode 100644 index 0000000..e9d8ae3 --- /dev/null +++ b/crates/datastore/src/pool/mod.rs @@ -0,0 +1,8 @@ +//! Pooled datastore protocol. +//! +//! A shared storage pool where multiple nodes contribute storage capacity +//! and converge on a shared view of what content lives where. + +pub mod disseminator; +pub mod messages; +pub mod coordinator; diff --git a/crates/datastore/tests/dashboard_integration_test.rs b/crates/datastore/tests/dashboard_integration_test.rs index 74c7adf..8d89d8e 100644 --- a/crates/datastore/tests/dashboard_integration_test.rs +++ b/crates/datastore/tests/dashboard_integration_test.rs @@ -48,6 +48,7 @@ fn dashboard_reflects_datastore_operations() { port: dash_port, ..Default::default() }); + dash.start_http_standalone(); let collector = dashboard::collector::StatsCollector::new(2); let mut rt = Runtime::new(RuntimeConfig { diff --git a/crates/datastore/tests/pool_tests.rs b/crates/datastore/tests/pool_tests.rs new file mode 100644 index 0000000..9607c65 --- /dev/null +++ b/crates/datastore/tests/pool_tests.rs @@ -0,0 +1,341 @@ +//! Pool protocol integration tests. +//! +//! Tests the pool disseminator's convergence behavior, the gossip channel +//! integration, and the coordinator actor's lifecycle. + +use std::sync::{Arc, Mutex}; + +use distribution::gossip_channel::{GossipChannel, serialize_each}; +use distribution::types::NodeId; +use shared_types::ContentHash; +use shared_types::pool::*; +use swactor_datastore::pool::disseminator::{PoolDisseminator, SharedPoolChannel}; + +fn node_id(b: u8) -> NodeId { + NodeId([b; 32]) +} + +fn make_pool_disseminator(node_byte: u8) -> PoolDisseminator { + PoolDisseminator::new( + PoolId::from_name("test-pool"), + "test-pool".into(), + node_id(node_byte), + 3, + ) +} + +// ─── Disseminator convergence scenarios ──────────────────────────────────── + +/// Two nodes join a pool and exchange gossip until they converge +/// on identical state. +#[test] +fn two_nodes_converge_on_membership() { + let mut d1 = make_pool_disseminator(1); + let mut d2 = make_pool_disseminator(2); + + d1.join(2); + d2.join(2); + + // Simulate 3 gossip rounds + for _ in 0..3 { + let mut ch1 = SharedPoolChannel::new(Arc::new(Mutex::new(d1))); + let mut ch2 = SharedPoolChannel::new(Arc::new(Mutex::new(d2))); + + let bytes1 = ch1.take_pending_bytes(100); + let bytes2 = ch2.take_pending_bytes(100); + + ch1.apply_incoming_bytes(&bytes2, 2); + ch2.apply_incoming_bytes(&bytes1, 2); + + d1 = Arc::try_unwrap(ch1.into_inner()).unwrap().into_inner().unwrap(); + d2 = Arc::try_unwrap(ch2.into_inner()).unwrap().into_inner().unwrap(); + } + + assert_eq!(d1.member_count(), 2); + assert_eq!(d2.member_count(), 2); +} + +/// Content announced on one node is visible on another after gossip. +#[test] +fn content_location_propagates_via_gossip() { + let mut d1 = make_pool_disseminator(1); + let mut d2 = make_pool_disseminator(2); + let hash = ContentHash::of(b"test-content"); + + d1.join(2); + d2.join(2); + d1.announce_content(hash, 2); + + // Gossip d1 → d2 + let mut ch1 = SharedPoolChannel::new(Arc::new(Mutex::new(d1))); + let bytes = ch1.take_pending_bytes(100); + let _d1 = Arc::try_unwrap(ch1.into_inner()).unwrap().into_inner().unwrap(); + + let mut ch2 = SharedPoolChannel::new(Arc::new(Mutex::new(d2))); + ch2.apply_incoming_bytes(&bytes, 2); + let d2 = Arc::try_unwrap(ch2.into_inner()).unwrap().into_inner().unwrap(); + + assert_eq!(d2.locate_content(&hash), vec![node_id(1)]); +} + +/// A node's leave is propagated and removes it from the active member list +/// on the receiving node. +#[test] +fn leave_propagates_via_gossip() { + let mut d1 = make_pool_disseminator(1); + let mut d2 = make_pool_disseminator(2); + + d1.join(2); + d2.join(2); + + // Exchange so both see each other + let mut ch1 = SharedPoolChannel::new(Arc::new(Mutex::new(d1))); + let mut ch2 = SharedPoolChannel::new(Arc::new(Mutex::new(d2))); + let bytes1 = ch1.take_pending_bytes(100); + let bytes2 = ch2.take_pending_bytes(100); + ch1.apply_incoming_bytes(&bytes2, 2); + ch2.apply_incoming_bytes(&bytes1, 2); + d1 = Arc::try_unwrap(ch1.into_inner()).unwrap().into_inner().unwrap(); + d2 = Arc::try_unwrap(ch2.into_inner()).unwrap().into_inner().unwrap(); + + assert_eq!(d1.member_count(), 2); + assert_eq!(d2.member_count(), 2); + + // d1 leaves + d1.leave(2); + + // Gossip leave to d2 + let mut ch1 = SharedPoolChannel::new(Arc::new(Mutex::new(d1))); + let bytes = ch1.take_pending_bytes(100); + let _d1 = Arc::try_unwrap(ch1.into_inner()).unwrap().into_inner().unwrap(); + + let mut ch2 = SharedPoolChannel::new(Arc::new(Mutex::new(d2))); + ch2.apply_incoming_bytes(&bytes, 2); + let d2 = Arc::try_unwrap(ch2.into_inner()).unwrap().into_inner().unwrap(); + + assert_eq!(d2.member_count(), 1); // only d2 remains active +} + +/// Content tombstone propagates and removes the location. +#[test] +fn content_deletion_propagates() { + let mut d1 = make_pool_disseminator(1); + let mut d2 = make_pool_disseminator(2); + let hash = ContentHash::of(b"ephemeral"); + + d1.join(2); + d2.join(2); + d1.announce_content(hash, 2); + + // Propagate content announcement + let mut ch1 = SharedPoolChannel::new(Arc::new(Mutex::new(d1))); + let bytes = ch1.take_pending_bytes(100); + d1 = Arc::try_unwrap(ch1.into_inner()).unwrap().into_inner().unwrap(); + let mut ch2 = SharedPoolChannel::new(Arc::new(Mutex::new(d2))); + ch2.apply_incoming_bytes(&bytes, 2); + d2 = Arc::try_unwrap(ch2.into_inner()).unwrap().into_inner().unwrap(); + + assert_eq!(d2.locate_content(&hash).len(), 1); + + // d1 removes content + d1.remove_content(hash, 2); + + // Propagate tombstone + let mut ch1 = SharedPoolChannel::new(Arc::new(Mutex::new(d1))); + let bytes = ch1.take_pending_bytes(100); + let _ = Arc::try_unwrap(ch1.into_inner()).unwrap().into_inner().unwrap(); + let mut ch2 = SharedPoolChannel::new(Arc::new(Mutex::new(d2))); + ch2.apply_incoming_bytes(&bytes, 2); + d2 = Arc::try_unwrap(ch2.into_inner()).unwrap().into_inner().unwrap(); + + assert!(d2.locate_content(&hash).is_empty()); +} + +/// ACL grant propagates to other nodes. +#[test] +fn acl_grant_propagates() { + let mut d1 = make_pool_disseminator(1); + let mut d2 = make_pool_disseminator(2); + + d1.grant_access(node_id(3), 2); + + let mut ch1 = SharedPoolChannel::new(Arc::new(Mutex::new(d1))); + let bytes = ch1.take_pending_bytes(100); + let _ = Arc::try_unwrap(ch1.into_inner()).unwrap().into_inner().unwrap(); + + let mut ch2 = SharedPoolChannel::new(Arc::new(Mutex::new(d2))); + ch2.apply_incoming_bytes(&bytes, 2); + d2 = Arc::try_unwrap(ch2.into_inner()).unwrap().into_inner().unwrap(); + + assert!(d2.is_node_authorized(&node_id(3))); + assert!(!d2.is_node_authorized(&node_id(4))); +} + +/// Capacity announcement propagates and is reflected in summary. +#[test] +fn capacity_propagates_and_summarizes() { + let mut d1 = make_pool_disseminator(1); + let mut d2 = make_pool_disseminator(2); + + d1.join(2); + d2.join(2); + d1.announce_capacity(1_000_000, 100_000, 2); + d2.announce_capacity(2_000_000, 200_000, 2); + + // Exchange + let mut ch1 = SharedPoolChannel::new(Arc::new(Mutex::new(d1))); + let mut ch2 = SharedPoolChannel::new(Arc::new(Mutex::new(d2))); + let bytes1 = ch1.take_pending_bytes(100); + let bytes2 = ch2.take_pending_bytes(100); + ch1.apply_incoming_bytes(&bytes2, 2); + ch2.apply_incoming_bytes(&bytes1, 2); + d1 = Arc::try_unwrap(ch1.into_inner()).unwrap().into_inner().unwrap(); + d2 = Arc::try_unwrap(ch2.into_inner()).unwrap().into_inner().unwrap(); + + let (total1, used1) = d1.pool_capacity_summary(); + let (total2, used2) = d2.pool_capacity_summary(); + + assert_eq!(total1, 3_000_000); + assert_eq!(used1, 300_000); + assert_eq!(total2, 3_000_000); + assert_eq!(used2, 300_000); +} + +/// The node_with_most_free_space query returns the correct node. +#[test] +fn placement_query_picks_node_with_most_space() { + let mut d = make_pool_disseminator(1); + d.join(3); + + // Simulate node 2 joining and having lots of space + let member2 = PoolEntry::Membership(PoolMemberEntry { + pool_id: PoolId::from_name("test-pool"), + node_id: [2u8; 32], + state: PoolMemberState::Active, + generation: 1, + }); + let cap2 = PoolEntry::Capacity(PoolCapacityEntry { + pool_id: PoolId::from_name("test-pool"), + node_id: [2u8; 32], + total_bytes: 10_000_000, + used_bytes: 1_000_000, + generation: 1, + }); + + // Simulate node 3 with less free space + let member3 = PoolEntry::Membership(PoolMemberEntry { + pool_id: PoolId::from_name("test-pool"), + node_id: [3u8; 32], + state: PoolMemberState::Active, + generation: 1, + }); + let cap3 = PoolEntry::Capacity(PoolCapacityEntry { + pool_id: PoolId::from_name("test-pool"), + node_id: [3u8; 32], + total_bytes: 5_000_000, + used_bytes: 4_000_000, + generation: 1, + }); + + let entries = vec![member2, cap2, member3, cap3]; + let bytes = serialize_each(&entries); + let mut ch = SharedPoolChannel::new(Arc::new(Mutex::new(d))); + ch.apply_incoming_bytes(&bytes, 3); + d = Arc::try_unwrap(ch.into_inner()).unwrap().into_inner().unwrap(); + + // d1 has no capacity announced, node 2 has 9M free, node 3 has 1M free + let best = d.node_with_most_free_space().unwrap(); + assert_eq!(best, node_id(2)); +} + +/// Five-node convergence: all nodes join, one announces content, everyone converges. +#[test] +fn five_node_pool_converges() { + let pool = PoolId::from_name("five-pool"); + let mut nodes: Vec = (0..5) + .map(|i| PoolDisseminator::new(pool, "five-pool".into(), node_id(i as u8), 3)) + .collect(); + + // All join + for n in &mut nodes { + n.join(5); + } + + // Node 0 and 2 announce content + let hash_a = ContentHash::of(b"file-a"); + let hash_b = ContentHash::of(b"file-b"); + nodes[0].announce_content(hash_a, 5); + nodes[2].announce_content(hash_b, 5); + + // Run 10 gossip rounds + for _ in 0..10 { + let mut channels: Vec = nodes + .into_iter() + .map(|d| SharedPoolChannel::new(Arc::new(Mutex::new(d)))) + .collect(); + + let pending: Vec>> = channels + .iter_mut() + .map(|ch| ch.take_pending_bytes(100)) + .collect(); + + for (recv_idx, ch) in channels.iter_mut().enumerate() { + for (send_idx, bytes) in pending.iter().enumerate() { + if recv_idx != send_idx { + ch.apply_incoming_bytes(bytes, 5); + } + } + } + + nodes = channels + .into_iter() + .map(|ch| Arc::try_unwrap(ch.into_inner()).unwrap().into_inner().unwrap()) + .collect(); + } + + for (i, node) in nodes.iter().enumerate() { + assert_eq!(node.member_count(), 5, "node {i} should see 5 members"); + assert_eq!( + node.locate_content(&hash_a), + vec![node_id(0)], + "node {i} should locate file-a on node 0" + ); + assert_eq!( + node.locate_content(&hash_b), + vec![node_id(2)], + "node {i} should locate file-b on node 2" + ); + } +} + +// ─── GossipChannel wire format tests ─────────────────────────────────────── + +/// Verify the GossipChannel topic tag is correct. +#[test] +fn shared_pool_channel_topic_tag() { + let d = make_pool_disseminator(1); + let ch = SharedPoolChannel::new(Arc::new(Mutex::new(d))); + assert_eq!(GossipChannel::topic_tag(&ch), "pool"); +} + +/// Verify serialization round-trip through GossipChannel bytes interface. +#[test] +fn gossip_channel_bytes_roundtrip() { + let mut d = make_pool_disseminator(1); + d.join(2); + let hash = ContentHash::of(b"roundtrip"); + d.announce_content(hash, 2); + + let mut ch = SharedPoolChannel::new(Arc::new(Mutex::new(d))); + let bytes = ch.take_pending_bytes(100); + assert!(!bytes.is_empty()); + + // Apply to a fresh disseminator + let d2 = make_pool_disseminator(2); + let mut ch2 = SharedPoolChannel::new(Arc::new(Mutex::new(d2))); + ch2.apply_incoming_bytes(&bytes, 2); + let d2 = Arc::try_unwrap(ch2.into_inner()).unwrap().into_inner().unwrap(); + + assert_eq!(d2.member_count(), 1); + assert_eq!(d2.locate_content(&hash), vec![node_id(1)]); +} diff --git a/crates/distribution/src/gossip_channel.rs b/crates/distribution/src/gossip_channel.rs new file mode 100644 index 0000000..a49dce6 --- /dev/null +++ b/crates/distribution/src/gossip_channel.rs @@ -0,0 +1,268 @@ +//! 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/iroh_driver.rs b/crates/distribution/src/iroh_driver.rs index cab5ee7..66fc1e1 100644 --- a/crates/distribution/src/iroh_driver.rs +++ b/crates/distribution/src/iroh_driver.rs @@ -507,15 +507,31 @@ impl IrohDriver { self.connections.remove(&node_id); } + // Resolve relay URL: explicit cache → SWIM metadata gossip → own home relay + let relay = self.peer_relay_urls.get(&node_id).cloned() + .or_else(|| { + self.node + .relay_url(&node_id) + .and_then(|s| s.parse::().ok()) + }) + .or_else(|| self.endpoint.addr().relay_urls().next().cloned()); + let endpoint = self.endpoint.clone(); - let conn = if let Some(relay) = self.peer_relay_urls.get(&node_id) { - let addr = EndpointAddr::new(key).with_relay_url(relay.clone()); + let connect_timeout = Duration::from_secs(2); + let conn = if let Some(relay) = relay { + let addr = EndpointAddr::new(key).with_relay_url(relay); self.rt.block_on(async { - endpoint.connect(addr, ALPN).await + match tokio::time::timeout(connect_timeout, endpoint.connect(addr, ALPN)).await { + Ok(result) => result.map_err(|e| -> Box { Box::new(e) }), + Err(_) => Err("connect timeout".into()), + } })? } else { self.rt.block_on(async { - endpoint.connect(key, ALPN).await + match tokio::time::timeout(connect_timeout, endpoint.connect(key, ALPN)).await { + Ok(result) => result.map_err(|e| -> Box { Box::new(e) }), + Err(_) => Err("connect timeout".into()), + } })? }; diff --git a/crates/shared-types/src/pool.rs b/crates/shared-types/src/pool.rs new file mode 100644 index 0000000..d19cc4f --- /dev/null +++ b/crates/shared-types/src/pool.rs @@ -0,0 +1,252 @@ +//! Shared types for the pooled datastore protocol. +//! +//! Types live in `shared-types` because both `distribution` and `datastore` +//! depend on them. + +use std::fmt; + +use serde::{Deserialize, Serialize}; + +use crate::ContentHash; + +// Re-export NodeId-shaped bytes — pool uses the same 32-byte node identifier. +// The actual NodeId type lives in `distribution::types`, but we use raw [u8; 32] +// here to avoid a circular dependency. Callers convert as needed. + +// ─── PoolId ──────────────────────────────────────────────────────────────── + +/// Unique pool identifier: blake3(name_bytes). +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] +pub struct PoolId(pub [u8; 32]); + +impl PoolId { + /// Create a pool ID from a human-readable name. + pub fn from_name(name: &str) -> Self { + PoolId(*blake3::hash(name.as_bytes()).as_bytes()) + } + + /// Encode as lowercase hex string. + pub fn to_hex(&self) -> String { + let mut s = String::with_capacity(64); + for b in &self.0 { + use fmt::Write; + write!(s, "{:02x}", b).unwrap(); + } + s + } + + /// Parse a 64-character hex string into a PoolId. + pub fn from_hex(hex: &str) -> Option { + if hex.len() != 64 { + return None; + } + let mut bytes = [0u8; 32]; + for (i, chunk) in hex.as_bytes().chunks(2).enumerate() { + let hi = hex_digit(chunk[0])?; + let lo = hex_digit(chunk[1])?; + bytes[i] = (hi << 4) | lo; + } + Some(PoolId(bytes)) + } +} + +impl fmt::Display for PoolId { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + for b in &self.0[..8] { + write!(f, "{:02x}", b)?; + } + write!(f, "\u{2026}") + } +} + +// ─── Pool Membership ─────────────────────────────────────────────────────── + +/// Whether a node is actively participating in a pool. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +pub enum PoolMemberState { + Active, + Left, +} + +/// A node's membership in a pool. Higher generation always wins. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct PoolMemberEntry { + pub pool_id: PoolId, + pub node_id: [u8; 32], + pub state: PoolMemberState, + pub generation: u64, +} + +// ─── Pool Capacity ───────────────────────────────────────────────────────── + +/// A node's storage capacity announcement. Higher generation always wins. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct PoolCapacityEntry { + pub pool_id: PoolId, + pub node_id: [u8; 32], + pub total_bytes: u64, + pub used_bytes: u64, + pub generation: u64, +} + +// ─── Content Location ────────────────────────────────────────────────────── + +/// Where a content hash is stored. Key: (pool_id, content_hash, node_id). +/// Higher generation wins. Tombstones indicate deletion. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct ContentLocationEntry { + pub pool_id: PoolId, + pub content_hash: ContentHash, + pub node_id: [u8; 32], + pub generation: u64, + pub tombstone: bool, +} + +// ─── Pool ACL ────────────────────────────────────────────────────────────── + +/// Authorization for a node to join a pool. Higher generation wins. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct PoolACLEntry { + pub pool_id: PoolId, + pub node_id: [u8; 32], + pub granted_by: [u8; 32], + pub generation: u64, + pub revoked: bool, +} + +// ─── Tagged Union ────────────────────────────────────────────────────────── + +/// All pool entry variants, used for gossip serialization. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub enum PoolEntry { + Membership(PoolMemberEntry), + Capacity(PoolCapacityEntry), + ContentLocation(ContentLocationEntry), + ACL(PoolACLEntry), +} + +// ─── Pool Config ─────────────────────────────────────────────────────────── + +/// Configuration for a pool. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct PoolConfig { + pub pool_name: String, + pub pool_id: PoolId, + /// Total bytes this node pledges to the pool. + pub capacity_bytes: u64, + /// Tombstone TTL in ticks before GC. + pub tombstone_ttl: u64, + /// GC interval in ticks. + pub gc_interval: u64, + /// Dissemination multiplier (Λ). + pub dissemination_lambda: usize, +} + +impl PoolConfig { + pub fn new(pool_name: &str, capacity_bytes: u64) -> Self { + Self { + pool_name: pool_name.to_string(), + pool_id: PoolId::from_name(pool_name), + capacity_bytes, + tombstone_ttl: 3600, + gc_interval: 1000, + dissemination_lambda: 3, + } + } +} + +// ─── Helpers ─────────────────────────────────────────────────────────────── + +fn hex_digit(b: u8) -> Option { + match b { + b'0'..=b'9' => Some(b - b'0'), + b'a'..=b'f' => Some(b - b'a' + 10), + b'A'..=b'F' => Some(b - b'A' + 10), + _ => None, + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn pool_id_from_name_is_deterministic() { + let a = PoolId::from_name("test-pool"); + let b = PoolId::from_name("test-pool"); + assert_eq!(a, b); + } + + #[test] + fn pool_id_different_names_differ() { + let a = PoolId::from_name("pool-a"); + let b = PoolId::from_name("pool-b"); + assert_ne!(a, b); + } + + #[test] + fn pool_id_hex_roundtrip() { + let id = PoolId::from_name("roundtrip-test"); + let hex = id.to_hex(); + let recovered = PoolId::from_hex(&hex).unwrap(); + assert_eq!(id, recovered); + } + + #[test] + fn pool_entry_serde_roundtrip() { + let entry = PoolEntry::Membership(PoolMemberEntry { + pool_id: PoolId::from_name("test"), + node_id: [1u8; 32], + state: PoolMemberState::Active, + generation: 1, + }); + let bytes = serde_json::to_vec(&entry).unwrap(); + let recovered: PoolEntry = serde_json::from_slice(&bytes).unwrap(); + match recovered { + PoolEntry::Membership(m) => { + assert_eq!(m.node_id, [1u8; 32]); + assert_eq!(m.state, PoolMemberState::Active); + assert_eq!(m.generation, 1); + } + _ => panic!("wrong variant"), + } + } + + #[test] + fn higher_generation_wins_for_membership() { + let old = PoolMemberEntry { + pool_id: PoolId::from_name("test"), + node_id: [1u8; 32], + state: PoolMemberState::Active, + generation: 1, + }; + let new = PoolMemberEntry { + pool_id: PoolId::from_name("test"), + node_id: [1u8; 32], + state: PoolMemberState::Left, + generation: 2, + }; + assert!(new.generation > old.generation); + } + + #[test] + fn content_location_tombstone_semantics() { + let live = ContentLocationEntry { + pool_id: PoolId::from_name("test"), + content_hash: ContentHash::of(b"hello"), + node_id: [1u8; 32], + generation: 1, + tombstone: false, + }; + let dead = ContentLocationEntry { + pool_id: PoolId::from_name("test"), + content_hash: ContentHash::of(b"hello"), + node_id: [1u8; 32], + generation: 2, + tombstone: true, + }; + assert!(!live.tombstone); + assert!(dead.tombstone); + assert!(dead.generation > live.generation); + } +} diff --git a/crates/swactor-node/src/main.rs b/crates/swactor-node/src/main.rs index d8ec184..af46d82 100644 --- a/crates/swactor-node/src/main.rs +++ b/crates/swactor-node/src/main.rs @@ -789,8 +789,16 @@ fn run_iroh( let seed_bytes = parse_node_id_str(&seed_str).expect("invalid seed node ID (expected hex or base58)"); let seed_key = iroh::PublicKey::from_bytes(&seed_bytes).expect("invalid seed public key"); + let mut seed_addr = iroh::EndpointAddr::from(seed_key); + // Include relay URLs so iroh can locate the seed through the relay + for host in &relay_hosts { + let url_str = format!("http://{host}:{relay_port}/"); + if let Ok(url) = url_str.parse::() { + seed_addr = seed_addr.with_relay_url(url); + } + } eprintln!("Joining cluster via seed {}", base58_encode(&seed_bytes)); - driver.join(&[iroh::EndpointAddr::from(seed_key)]); + driver.join(&[seed_addr]); } // Spawn and register actors diff --git a/docs/development_history/pooled-datastore/POOLED_DATASTORE.md b/docs/development_history/pooled-datastore/POOLED_DATASTORE.md new file mode 100644 index 0000000..4060b98 --- /dev/null +++ b/docs/development_history/pooled-datastore/POOLED_DATASTORE.md @@ -0,0 +1,423 @@ +# Pooled Datastore & Sim-Cluster — Development History + +> Adds a gossip-converged pooled storage protocol, a generic gossip channel +> abstraction, a pool dashboard page, shared pool types, iroh connection +> hardening, and a Docker-free multi-process sim-cluster test harness. +> +> ~18 new/modified files · ~2,400 insertions +> +> *Branch: `pooled-datastore`* + +--- + +## Table of Contents + +1. [Overview & Motivation](#1-overview--motivation) +2. [What Was Built](#2-what-was-built) +3. [Pooled Storage Protocol](#3-pooled-storage-protocol) +4. [Generic Gossip Channel Abstraction](#4-generic-gossip-channel-abstraction) +5. [Pool Disseminator](#5-pool-disseminator) +6. [Pool Coordinator Actor](#6-pool-coordinator-actor) +7. [Shared Pool Types](#7-shared-pool-types) +8. [Dashboard Pool Page](#8-dashboard-pool-page) +9. [Iroh Connection Hardening](#9-iroh-connection-hardening) +10. [Sim-Cluster Test Harness](#10-sim-cluster-test-harness) +11. [Design Decisions & Tradeoffs](#11-design-decisions--tradeoffs) +12. [Test Coverage](#12-test-coverage) +13. [Known Gaps & Future Work](#13-known-gaps--future-work) + +--- + +## 1. Overview & Motivation + +The existing datastore provides content-addressed storage on individual nodes, +but there is no mechanism for multiple nodes to form a shared storage pool — +knowing who has what content, how much capacity each node offers, or where to +place new data. + +This branch introduces a **pooled datastore protocol** that layers on top of +the existing content-addressed datastore. Multiple nodes join a named pool, +gossip their membership/capacity/content-locations via SWIM piggyback, and +converge on a shared view of the pool's state. This enables: + +- **Content location**: find which node(s) hold a given content hash without + fan-out queries. +- **Capacity-aware placement**: route new writes to the node with the most free + space. +- **Pool ACL**: optional allow-list to restrict which nodes can join a pool. +- **Live observability**: a new dashboard page shows pool membership, capacity + bars, content location map, and ACL state in real time via SSE. + +Separately, the branch also introduces: + +- A **generic gossip channel abstraction** (`GossipChannel` trait + + `DisseminationBuffer`) that replaces the 4 duplicated dissemination + patterns in the distribution crate. +- A **sim-cluster test harness** (`cargo xtask test sim-cluster` / `cargo + xtask sim-cluster`) that spawns real multi-process clusters with a local iroh + relay — no Docker required. +- **Iroh connection hardening**: relay URL resolution cascade and connect + timeouts to prevent indefinite hangs during peer connection. + +--- + +## 2. What Was Built + +| Component | Crate / Location | Lines | +|-----------|-----------------|-------| +| Pool types (PoolId, entries, config) | `crates/shared-types/src/pool.rs` | ~170 | +| Gossip channel trait + DisseminationBuffer | `crates/distribution/src/gossip_channel.rs` | ~270 | +| Pool disseminator (CRDT state + gossip) | `crates/datastore/src/pool/disseminator.rs` | ~720 | +| Pool coordinator actor | `crates/datastore/src/pool/coordinator.rs` | ~275 | +| Pool messages | `crates/datastore/src/pool/messages.rs` | ~80 | +| Pool dashboard HTML/JS | `crates/dashboard/src/pool_html.rs` | ~390 | +| Sim-cluster harness | `xtask/src/sim_cluster.rs` | ~700 | +| Pool integration tests | `crates/datastore/tests/pool_tests.rs` | ~340 | +| Docker compose (dev cluster) | `tests/docker/docker-compose.dev-cluster.yml` | ~75 | + +Modified files: + +| File | Change | +|------|--------| +| `crates/distribution/src/iroh_driver.rs` | Relay URL cascade + connect timeout | +| `crates/swactor-node/src/main.rs` | Seed node relay URL hints for iroh | +| `crates/datastore/tests/dashboard_integration_test.rs` | Start HTTP standalone | +| `xtask/src/main.rs` | `sim-cluster` subcommand + test group | +| `xtask/Cargo.toml` | reqwest, tokio, iroh-relay deps | + +--- + +## 3. Pooled Storage Protocol + +The pool protocol is a set of four CRDT entry types that converge via gossip: + +``` +┌────────────────────────────────────────────────────┐ +│ Pool State (per node) │ +├──────────────┬─────────────┬───────────┬───────────┤ +│ Membership │ Capacity │ Content │ ACL │ +│ │ │ Location │ │ +│ node→state │ node→bytes │ (hash, │ node→ │ +│ (Active/Left)│ (total/used)│ node)→ │ grant/ │ +│ │ │ tombstone│ revoke │ +├──────────────┴─────────────┴───────────┴───────────┤ +│ Higher generation always wins │ +│ (last-writer-wins register per key) │ +└────────────────────────────────────────────────────┘ +``` + +**Convergence rule**: For each entry type, the key is derived from the entry +(e.g. `node_id` for membership, `(content_hash, node_id)` for content +locations). When two entries share a key, the one with the higher `generation` +wins. This makes all merges commutative, associative, and idempotent — a CRDT. + +**Deletion**: Content locations and ACL entries use tombstones (`tombstone: +true` / `revoked: true`) with a generation bump. Tombstones are garbage +collected after a configurable TTL. + +**Dissemination**: All entries go through a shared `DisseminationBuffer` +which transmits each entry `Λ * ceil(log₂(n))` times before eviction, matching +the standard SWIM protocol budget. + +--- + +## 4. Generic Gossip Channel Abstraction + +**File**: `crates/distribution/src/gossip_channel.rs` + +Before this branch, SWIM piggyback dissemination was hardcoded for membership +updates, directory entries, and dead-letter notifications — each with its own +copy of the `Λ * ceil(log₂(n))` budget logic. + +The new abstraction provides: + +- **`GossipChannel` trait**: A topic-tagged channel that produces/consumes + `Vec` entries for piggyback. Methods: `topic_tag()`, + `take_pending_bytes()`, `apply_incoming_bytes()`, `re_disseminate_all()`, + `on_node_death()`, `gc_tick()`. + +- **`DisseminationBuffer`**: A generic budget-limited queue. Entries are + enqueued with a transmit budget of `Λ * ceil(log₂(n))` and evicted after + exhaustion. Supports `enqueue`, `enqueue_or_replace` (idempotent upsert), + `take`, `re_enqueue_all`, and `retain`. + +- **Serialization helpers**: `serialize_each()` and `deserialize_each()` for + converting between typed entries and `Vec`. + +The pool disseminator is the first consumer, plugging into the distribution +layer via `SharedPoolChannel` which wraps `Arc>`. + +--- + +## 5. Pool Disseminator + +**File**: `crates/datastore/src/pool/disseminator.rs` + +The core state machine. Manages four `HashMap` tables (membership, capacity, +content locations, ACL) and a single `DisseminationBuffer`. + +Key methods: + +- **Lifecycle**: `join()`, `leave()` — announce membership state changes. +- **Storage**: `announce_content()`, `remove_content()`, `announce_capacity()`. +- **ACL**: `grant_access()`, `revoke_access()`, `is_node_authorized()`. +- **Queries**: `active_members()`, `member_count()`, `content_count()`, + `locate_content()`, `node_with_most_free_space()`, `pool_capacity_summary()`. +- **Dashboard**: `snapshot_json()` — full JSON snapshot for SSE. +- **Internal**: `merge_entry()` applies the higher-generation-wins rule. + `take_pending_inner()` / `apply_incoming_inner()` drive gossip exchange. + +`SharedPoolChannel` wraps this in an `Arc>` and implements +`GossipChannel`, bridging ownership between the `PoolCoordinator` actor +(lifecycle/queries) and `DistributedNode` (gossip transport). + +--- + +## 6. Pool Coordinator Actor + +**File**: `crates/datastore/src/pool/coordinator.rs` + +An actor implementing `ActorInterface` for `PoolCoordinatorMsg`. It acts as a +placement-aware CRUD facade: + +- **PoolPut/Get/Delete/List**: Delegates to the co-located `DatastoreNode` + actor. Future: redirect to best node based on capacity. +- **PoolStatus**: Queries the disseminator and returns a JSON status snapshot. +- **JoinPool/LeavePool**: Checks ACL authorization, then calls the + disseminator. +- **GrantPoolAccess/RevokePoolAccess**: Manages the allow-list. +- **PoolTick**: Periodic capacity re-announcement (every 100 ticks). + +--- + +## 7. Shared Pool Types + +**File**: `crates/shared-types/src/pool.rs` + +Types live in `shared-types` to avoid circular dependencies between +`distribution` and `datastore`: + +- **`PoolId`**: `blake3(name_bytes)` — 32-byte deterministic pool identifier. + Supports hex encoding/decoding and truncated display. +- **`PoolMemberEntry`**: Node membership with `Active`/`Left` state. +- **`PoolCapacityEntry`**: Storage capacity announcement (total/used bytes). +- **`ContentLocationEntry`**: Where a content hash is stored, with tombstone + support. +- **`PoolACLEntry`**: Authorization grant/revoke with `granted_by` provenance. +- **`PoolEntry`**: Tagged enum wrapping all four entry types for gossip + serialization. +- **`PoolConfig`**: Pool configuration (name, capacity, TTL, GC interval, Λ). + +--- + +## 8. Dashboard Pool Page + +**File**: `crates/dashboard/src/pool_html.rs` + +A new `/pool` page in the dashboard with: + +- **Summary cards**: pool name, member count, content count, total/used + capacity. +- **Capacity bars**: per-node usage with color thresholds (green < 70%, orange + < 90%, red >= 90%). +- **Members table**: node ID (truncated with tooltip), state, total/used/free. +- **Content location map**: content hash → replica count → node list. +- **ACL panel**: open mode indicator or allow-list table. +- **Join/Leave buttons**: POST to `/api/pool/join` and `/api/pool/leave`. +- **Live updates**: SSE `pool` events drive real-time state refresh. + +--- + +## 9. Iroh Connection Hardening + +**File**: `crates/distribution/src/iroh_driver.rs` + +Two problems fixed: + +1. **Relay URL resolution cascade**: When connecting to a peer, the driver now + tries three sources in order: (a) explicit relay URL cache from prior + connections, (b) SWIM metadata gossip (via `node.relay_url()`), (c) the + local node's own home relay. Previously only the explicit cache was checked, + causing connections to fail when the cache was empty. + +2. **Connect timeout**: All `endpoint.connect()` calls now have a 2-second + `tokio::time::timeout` wrapper. Previously, connections could hang + indefinitely if a peer was unreachable. + +**File**: `crates/swactor-node/src/main.rs` + +Seed node addresses now include relay URLs so iroh can locate the seed through +the relay server, rather than relying solely on direct addressing. + +--- + +## 10. Sim-Cluster Test Harness + +**File**: `xtask/src/sim_cluster.rs` + +A new test stage and development tool that spawns real multi-process swactor +clusters without Docker: + +### Test mode: `cargo xtask test sim-cluster` + +Runs 4 scenarios sequentially, each with a fresh 5-node cluster: + +| # | Scenario | Validates | +|---|----------|-----------| +| 1 | Cluster convergence | All 5 nodes see >= 4 alive peers, routing table >= 4 | +| 2 | Node death detection | Kill node 2, survivors detect alive drop, dead count >= 1 | +| 3 | Killed node rejoins | Kill node 2, restart it, rejoined node sees alive >= 1 | +| 4 | Actors resolvable | Each node has >= 2 directory entries, total >= 10 | + +### Interactive mode: `cargo xtask sim-cluster --nodes N` + +Spawns a persistent cluster for development. Prints dashboard URLs and blocks +until Ctrl-C. + +### Infrastructure + +- **Local relay server**: Embedded `iroh-relay` server on an ephemeral port. + Nodes connect through the relay rather than requiring direct connectivity. +- **RAII lifecycle**: `SimCluster` owns child processes and SIGTERM's them on + drop. `RelayServer` owns its tokio runtime. +- **Config generation**: Each node gets a `node.toml` with dashboard port, + actor count, relay host/port, and optional seed node ID. +- **Seed key discovery**: Polls the seed node's key file on disk to extract the + public key before spawning joiner nodes. +- **HTTP observation**: Polls `/api/distribution` on each node's dashboard. + Uses `serde_json::Value` to avoid compile-time coupling to protocol types. +- **Node lifecycle**: `kill_node()` sends SIGTERM, `restart_node()` re-spawns + with the same config (non-seed nodes get the seed's public key). + +### Dev cluster compose + +**File**: `tests/docker/docker-compose.dev-cluster.yml` + +A 3-node Docker Compose file for development with pool configuration +(`--pool-name dev-pool --pool-capacity 104857600`). Uses a bridge network with +static IPs. + +--- + +## 11. Design Decisions & Tradeoffs + +**Higher-generation-wins CRDT over vector clocks**: Pool entries use a simple +monotonic generation counter per entry key. This is sufficient because each +entry has a single writer (the node that owns it). Vector clocks would add +complexity without benefit since there are no concurrent writers for the same +key. + +**Tombstones with TTL over immediate deletion**: Content locations and ACL +revocations use tombstones that propagate via gossip before being GC'd. Without +tombstones, a deleted entry could be re-introduced by a node that hasn't yet +received the deletion. + +**Shared `Arc>` over message-passing for disseminator**: The pool +disseminator needs to be accessed by both the coordinator actor (for +lifecycle/queries) and the distribution layer (for gossip). Rather than adding +an actor-to-actor message protocol, the disseminator is wrapped in +`Arc>`. The lock is held only briefly for individual +operations. + +**Sim-cluster over Docker for testing**: Docker adds build time, image +management, and network configuration complexity. The sim-cluster spawns bare +processes on localhost, uses an embedded iroh relay, and tears down in +milliseconds. Scenarios that previously required Docker Compose now run with +`cargo xtask test sim-cluster`. + +**HTTP polling over direct protocol observation**: The sim-cluster observes +node state via HTTP (`/api/distribution`) rather than linking against protocol +types. This makes the test harness resilient to protocol changes and mirrors +how an operator would observe a real cluster. + +**Pool types in `shared-types`**: Pool entry types live in `shared-types` +rather than `datastore` to avoid a circular dependency — `distribution` needs +to know about pool entries for gossip serialization, and `datastore` depends on +`distribution`. + +--- + +## 12. Test Coverage + +### Unit tests (disseminator internals) + +In `crates/datastore/src/pool/disseminator.rs`: + +- `join_and_query_members` — join lifecycle +- `leave_removes_from_active` — leave lifecycle +- `announce_and_locate_content` — content announcement + query +- `remove_content_tombstones` — tombstone semantics +- `capacity_summary` — capacity aggregation +- `acl_grant_and_check` / `acl_revoke` / `empty_acl_means_open` — ACL logic +- `higher_generation_wins_merge` — CRDT merge rule +- `two_disseminators_converge_via_gossip_exchange` — two-node gossip +- `three_node_convergence_loop` — multi-round gossip convergence + +### Unit tests (gossip channel) + +In `crates/distribution/src/gossip_channel.rs`: + +- `budget_math_*` (4 tests) — transmit budget calculation +- `enqueue_take_evicts_after_budget` — budget exhaustion +- `enqueue_or_replace_*` (2 tests) — idempotent upsert +- `re_enqueue_all_refreshes_budgets` — anti-entropy +- `retain_removes_non_matching` — predicate-based eviction +- `serialize_deserialize_roundtrip` — wire format + +### Unit tests (shared types) + +In `crates/shared-types/src/pool.rs`: + +- `pool_id_from_name_is_deterministic` / `pool_id_different_names_differ` +- `pool_id_hex_roundtrip` +- `pool_entry_serde_roundtrip` +- `higher_generation_wins_for_membership` +- `content_location_tombstone_semantics` + +### Integration tests (pool protocol) + +In `crates/datastore/tests/pool_tests.rs`: + +- `two_nodes_converge_on_membership` — two-node gossip convergence +- `content_location_propagates_via_gossip` — cross-node content discovery +- `leave_propagates_via_gossip` — membership leave propagation +- `content_deletion_propagates` — tombstone propagation +- `acl_grant_propagates` — ACL gossip +- `capacity_propagates_and_summarizes` — capacity gossip + aggregation +- `placement_query_picks_node_with_most_space` — capacity-aware placement +- `five_node_pool_converges` — 5-node full convergence +- `shared_pool_channel_topic_tag` — GossipChannel interface +- `gossip_channel_bytes_roundtrip` — wire format through GossipChannel + +### Sim-cluster scenarios (multi-process) + +In `xtask/src/sim_cluster.rs`: + +- Cluster convergence (5 nodes) +- Node death detection (kill + observe) +- Killed node rejoins (kill + restart + observe) +- Actors resolvable (directory entry propagation) + +--- + +## 13. Known Gaps & Future Work + +- **Remote content fetch**: `PoolGet` currently only checks the local + datastore. It should use `locate_content()` to fetch from the node that + actually has the content. +- **Capacity-aware placement**: `PoolPut` delegates to the local datastore. + It should use `node_with_most_free_space()` to route writes to the best node. +- **Actual usage tracking**: `PoolTick` re-announces capacity with `used: 0`. + It should query the `BlobStore` for actual disk usage. +- **GossipChannel integration**: The `GossipChannel` trait and + `SharedPoolChannel` are built but not yet wired into `DistributedNode`'s + piggyback system. The existing hardcoded dissemination channels need to be + migrated to the new trait. +- **Dashboard SSE integration**: The pool dashboard HTML is built, but the + server-side SSE event source for `pool` events needs to be wired to the + `PoolDisseminator::snapshot_json()` method. +- **Sim-cluster namespace isolation**: The harness uses high ephemeral ports + for isolation. Full Linux network namespace isolation (as designed in + `TEST_ISOLATION.md`) is a future enhancement. +- **Sim-cluster in CI**: The sim-cluster test group is opt-in and excluded + from `essential`/`all`. Once proven stable, it should be added to CI. diff --git a/tests/docker/docker-compose.dev-cluster.yml b/tests/docker/docker-compose.dev-cluster.yml new file mode 100644 index 0000000..f46d0e9 --- /dev/null +++ b/tests/docker/docker-compose.dev-cluster.yml @@ -0,0 +1,77 @@ +services: + seed: + image: swactor-dev-cluster + command: + - "--identity-dir" + - "/identity" + - "--listen" + - "0.0.0.0:11204" + - "--no-relay" + - "--pool-name" + - "dev-pool" + - "--pool-capacity" + - "104857600" + - "--actors" + - "2" + volumes: + - /home/aaron/swactor-pooled-datastore/.dev-cluster/seed-identity:/identity:ro + networks: + dev-cluster: + ipv4_address: 10.0.2.10 + ports: + - "9100:9090" + + node-2: + image: swactor-dev-cluster + command: + - "--seed-node-id" + - "d5469553d4408073022fee3e41c99cf9778bf3621e430f75f8c8ae7d6de0153f" + - "--seed-addrs" + - "10.0.2.10:11204" + - "--listen" + - "0.0.0.0:11204" + - "--no-relay" + - "--pool-name" + - "dev-pool" + - "--pool-capacity" + - "104857600" + - "--actors" + - "2" + networks: + dev-cluster: + ipv4_address: 10.0.2.11 + ports: + - "9101:9090" + depends_on: + - seed + + node-3: + image: swactor-dev-cluster + command: + - "--seed-node-id" + - "d5469553d4408073022fee3e41c99cf9778bf3621e430f75f8c8ae7d6de0153f" + - "--seed-addrs" + - "10.0.2.10:11204" + - "--listen" + - "0.0.0.0:11204" + - "--no-relay" + - "--pool-name" + - "dev-pool" + - "--pool-capacity" + - "104857600" + - "--actors" + - "2" + networks: + dev-cluster: + ipv4_address: 10.0.2.12 + ports: + - "9102:9090" + depends_on: + - seed + +networks: + dev-cluster: + driver: bridge + ipam: + config: + - subnet: 10.0.2.0/24 diff --git a/xtask/Cargo.toml b/xtask/Cargo.toml index a63becf..29eb9e2 100644 --- a/xtask/Cargo.toml +++ b/xtask/Cargo.toml @@ -9,3 +9,6 @@ toml = "0.8" serde = { version = "1", features = ["derive"] } serde_json = "1" libc = "0.2" +reqwest = { version = "0.12", features = ["blocking", "json"] } +tokio = { version = "1", features = ["rt-multi-thread"] } +iroh-relay = { version = "0.96", features = ["server"] } diff --git a/xtask/src/main.rs b/xtask/src/main.rs index 7462fd8..7773971 100644 --- a/xtask/src/main.rs +++ b/xtask/src/main.rs @@ -1,4 +1,5 @@ mod deploy; +mod sim_cluster; use std::path::{Path, PathBuf}; use std::process::Command; @@ -107,6 +108,13 @@ enum Cmd { dirs: Vec, }, + /// Launch a local sim-cluster (relay + N nodes) for development + SimCluster { + /// Number of nodes (default: 5) + #[arg(long, default_value = "5")] + nodes: usize, + }, + /// Deploy swactor to remote machines Deploy { /// Deploy via Docker over SSH (build image, push, run containers) @@ -285,6 +293,7 @@ TEST GROUPS: distribution Distribution protocol + datastore cluster-sims Deterministic cluster simulations integrated HTTP API + dashboard end-to-end tests + sim-cluster Multi-process cluster with local iroh relay essential core + distribution + integrated (merge gate) all Every test group @@ -317,6 +326,7 @@ fn print_list() { println!(); } + println!(" {:<14}Multi-process cluster with local iroh relay", "sim-cluster"); println!(" {:<14}core + distribution + integrated (merge gate)", "essential"); println!(" {:<14}Every test group", "all"); } @@ -337,6 +347,11 @@ fn run_test(group: Option, list: bool) { } }; + if group_name == "sim-cluster" { + sim_cluster::run(); + return; + } + let groups = match groups_for(&group_name) { Some(g) => g, None => { @@ -930,6 +945,7 @@ fn main() { } => run_cli(url, key, extra, &config.cli), Cmd::InitNode { role, dir } => run_init_node(&role, dir.as_deref()), Cmd::GenPeers { dirs } => run_gen_peers(&dirs), + Cmd::SimCluster { nodes } => sim_cluster::run_interactive(nodes), Cmd::Deploy { docker, config, skip_build, skip_verify, skip_peers } => { let config = config.unwrap_or_else(|| { if docker { ".deploy/docker.toml" } else { ".deploy/deploy.toml" }.into() diff --git a/xtask/src/sim_cluster.rs b/xtask/src/sim_cluster.rs new file mode 100644 index 0000000..11cad7e --- /dev/null +++ b/xtask/src/sim_cluster.rs @@ -0,0 +1,701 @@ +//! sim-cluster: multi-process cluster tests using iroh transport. +//! +//! Spawns 5 swactor nodes with iroh transport connected through a local +//! relay server, then runs the same scenarios as `tests/docker/tests/cluster.rs` +//! without Docker. + +use std::fs; +use std::net::Ipv4Addr; +use std::path::{Path, PathBuf}; +use std::process::{Child, Command, Stdio}; +use std::thread; +use std::time::{Duration, Instant}; + +use crate::workspace_root; + +// ── Constants ─────────────────────────────────────────────────────────── + +const NODE_COUNT: usize = 5; +const ACTORS_PER_NODE: usize = 2; +const TIMEOUT: Duration = Duration::from_secs(30); + +/// Dashboard HTTP ports — high ports to avoid conflicts. +fn dashboard_port(index: usize) -> u16 { + 19091 + index as u16 +} + +fn all_dashboard_ports(node_count: usize) -> Vec { + (0..node_count).map(dashboard_port).collect() +} + +// ── Local relay server ────────────────────────────────────────────────── + +/// Owns a tokio runtime + iroh-relay server. RAII cleanup on drop. +struct RelayServer { + _server: iroh_relay::server::Server, + _rt: tokio::runtime::Runtime, + port: u16, +} + +impl RelayServer { + fn start() -> Self { + let rt = tokio::runtime::Builder::new_multi_thread() + .worker_threads(1) + .enable_all() + .build() + .expect("failed to create tokio runtime for relay"); + + let server = rt.block_on(async { + iroh_relay::server::Server::spawn(iroh_relay::server::ServerConfig::<(), ()> { + relay: Some(iroh_relay::server::RelayConfig { + http_bind_addr: (Ipv4Addr::LOCALHOST, 0).into(), + tls: None, + limits: Default::default(), + key_cache_capacity: Some(256), + access: iroh_relay::server::AccessConfig::Everyone, + }), + quic: None, + metrics_addr: None, + }) + .await + }) + .expect("failed to spawn relay server"); + + let addr = server.http_addr().expect("relay has no HTTP address"); + let port = addr.port(); + + RelayServer { + _server: server, + _rt: rt, + port, + } + } +} + +// ── Build ─────────────────────────────────────────────────────────────── + +fn build_swactor(root: &Path) -> PathBuf { + let status = Command::new("cargo") + .args(["build", "-p", "swactor-node"]) + .current_dir(root) + .status() + .expect("failed to run cargo build"); + if !status.success() { + eprintln!("cargo build failed"); + std::process::exit(1); + } + + let binary = root.join("target/debug/swactor"); + if !binary.exists() { + eprintln!("binary not found at {}", binary.display()); + std::process::exit(1); + } + binary +} + +// ── Config generation ─────────────────────────────────────────────────── + +fn write_node_config( + dir: &Path, + index: usize, + relay_port: u16, + seed_public_key: Option<&str>, +) -> PathBuf { + let identity_dir = dir.join("identity"); + fs::create_dir_all(&identity_dir).expect("failed to create identity dir"); + + let config_path = dir.join("node.toml"); + let dashboard = dashboard_port(index); + + let mut config = format!( + r#"dashboard_port = {dashboard} +actors = {ACTORS_PER_NODE} +no_datastore = true +identity_dir = "{identity}" +relay = false +relay_port = {relay_port} +relay_hosts = ["127.0.0.1"] +"#, + identity = identity_dir.display(), + ); + + if let Some(seed_id) = seed_public_key { + config.push_str(&format!("seed_node_id = \"{seed_id}\"\n")); + } + + fs::write(&config_path, &config).expect("failed to write node config"); + config_path +} + +// ── Seed key discovery ────────────────────────────────────────────────── + +fn read_seed_public_key(run_dir: &Path) -> String { + let key_path = run_dir.join("node-0/identity/node.key.json"); + let deadline = Instant::now() + Duration::from_secs(10); + + loop { + if Instant::now() > deadline { + panic!( + "timed out waiting for seed key file: {}", + key_path.display() + ); + } + + if key_path.exists() { + if let Ok(data) = fs::read_to_string(&key_path) { + if let Ok(json) = serde_json::from_str::(&data) { + if let Some(pk) = json.get("public_key").and_then(|v| v.as_str()) { + return pk.to_string(); + } + } + } + } + + thread::sleep(Duration::from_millis(100)); + } +} + +// ── Dashboard readiness ───────────────────────────────────────────────── + +fn wait_for_dashboard(port: u16, timeout: Duration) { + let start = Instant::now(); + loop { + if start.elapsed() > timeout { + panic!("timed out waiting for dashboard on port {port}"); + } + if poll_distribution(port).is_some() { + return; + } + thread::sleep(Duration::from_millis(200)); + } +} + +// ── Cluster handle (RAII) ─────────────────────────────────────────────── + +struct SimCluster { + children: Vec<(usize, Child)>, + run_dir: PathBuf, + binary: PathBuf, + relay: RelayServer, + seed_public_key: String, + _node_count: usize, +} + +impl SimCluster { + fn spawn(binary: &Path, run_dir: &Path, node_count: usize) -> Self { + fs::create_dir_all(run_dir).expect("failed to create run dir"); + + // Phase 0: start relay + let relay = RelayServer::start(); + println!( + " relay at http://127.0.0.1:{}/ (port {})", + relay.port, relay.port + ); + + let mut children = Vec::new(); + + // Phase 1: spawn seed node (index 0) — no seed_node_id + let child = spawn_node(binary, run_dir, 0, relay.port, None); + children.push((0, child)); + + // Phase 2: wait for seed's key file + let seed_public_key = read_seed_public_key(run_dir); + println!(" seed key: {seed_public_key}"); + + // Wait for seed's dashboard to be ready before spawning joiners + wait_for_dashboard(dashboard_port(0), Duration::from_secs(15)); + + // Phase 3: spawn remaining nodes with seed_node_id + for i in 1..node_count { + let child = spawn_node(binary, run_dir, i, relay.port, Some(&seed_public_key)); + children.push((i, child)); + } + + SimCluster { + children, + run_dir: run_dir.to_path_buf(), + binary: binary.to_path_buf(), + relay, + seed_public_key, + _node_count: node_count, + } + } + + fn kill_node(&mut self, index: usize) { + if let Some(pos) = self.children.iter().position(|(i, _)| *i == index) { + let (_, mut child) = self.children.remove(pos); + let _ = signal_term(child.id()); + let _ = child.wait(); + } + } + + fn restart_node(&mut self, index: usize) { + // Non-seed nodes need the seed's public key; the seed itself doesn't + let seed_id = if index != 0 { + Some(self.seed_public_key.as_str()) + } else { + None + }; + let child = spawn_node( + &self.binary, + &self.run_dir, + index, + self.relay.port, + seed_id, + ); + self.children.push((index, child)); + } +} + +impl Drop for SimCluster { + fn drop(&mut self) { + for (_, child) in &self.children { + let _ = signal_term(child.id()); + } + for (_, child) in &mut self.children { + let _ = child.wait(); + } + // relay drops automatically via RelayServer Drop + // NOTE: leaving run_dir for debugging; uncomment below for production + // let _ = fs::remove_dir_all(&self.run_dir); + } +} + +fn spawn_node( + binary: &Path, + run_dir: &Path, + index: usize, + relay_port: u16, + seed_public_key: Option<&str>, +) -> Child { + let node_dir = run_dir.join(format!("node-{index}")); + let config_path = write_node_config(&node_dir, index, relay_port, seed_public_key); + + let log_path = node_dir.join("stderr.log"); + let log_file = fs::File::create(&log_path) + .unwrap_or_else(|e| panic!("failed to create log for node {index}: {e}")); + + Command::new(binary) + .args(["--config", &config_path.to_string_lossy()]) + .stdout(Stdio::null()) + .stderr(Stdio::from(log_file)) + .spawn() + .unwrap_or_else(|e| panic!("failed to spawn node {index}: {e}")) +} + +fn signal_term(pid: u32) -> std::io::Result<()> { + unsafe { libc::kill(pid as i32, libc::SIGTERM) }; + Ok(()) +} + +// ── HTTP polling (decoupled from distribution crate) ──────────────────── + +fn poll_distribution(port: u16) -> Option { + let url = format!("http://127.0.0.1:{port}/api/distribution"); + let client = reqwest::blocking::Client::builder() + .timeout(Duration::from_secs(2)) + .build() + .ok()?; + let resp = client.get(&url).send().ok()?; + if !resp.status().is_success() { + return None; + } + let text = resp.text().ok()?; + if text == "{}" { + return None; + } + serde_json::from_str(&text).ok() +} + +fn get_usize(val: &serde_json::Value, key: &str) -> Option { + val.get(key).and_then(|v| v.as_u64()).map(|n| n as usize) +} + +fn wait_for_convergence( + ports: &[u16], + expected_alive: usize, + timeout: Duration, +) -> Result { + let start = Instant::now(); + loop { + if start.elapsed() > timeout { + let mut diag = String::from("Convergence timeout. Last seen: "); + for &port in ports { + match poll_distribution(port) { + Some(snap) => { + let alive = get_usize(&snap, "alive_count").unwrap_or(0); + diag.push_str(&format!("port {port}={alive}, ")); + } + None => diag.push_str(&format!("port {port}=unreachable, ")), + } + } + return Err(diag); + } + + let all_converged = ports.iter().all(|&port| { + poll_distribution(port) + .and_then(|snap| get_usize(&snap, "alive_count")) + .map(|alive| alive >= expected_alive) + .unwrap_or(false) + }); + + if all_converged { + return Ok(start.elapsed()); + } + + thread::sleep(Duration::from_secs(1)); + } +} + +fn wait_for_death_detection( + ports: &[u16], + max_alive: usize, + timeout: Duration, +) -> Result { + let start = Instant::now(); + loop { + if start.elapsed() > timeout { + let mut diag = String::from("Death detection timeout. Last seen: "); + for &port in ports { + match poll_distribution(port) { + Some(snap) => { + let alive = get_usize(&snap, "alive_count").unwrap_or(0); + diag.push_str(&format!("port {port}={alive} alive, ")); + } + None => diag.push_str(&format!("port {port}=unreachable, ")), + } + } + return Err(diag); + } + + let all_detected = ports.iter().all(|&port| { + poll_distribution(port) + .and_then(|snap| get_usize(&snap, "alive_count")) + .map(|alive| alive <= max_alive) + .unwrap_or(false) + }); + + if all_detected { + return Ok(start.elapsed()); + } + + thread::sleep(Duration::from_secs(1)); + } +} + +fn wait_for_dead_count(ports: &[u16], min_dead: usize, timeout: Duration) -> bool { + let start = Instant::now(); + loop { + if start.elapsed() > timeout { + return false; + } + + let any_sees_dead = ports.iter().any(|&port| { + poll_distribution(port) + .and_then(|snap| get_usize(&snap, "dead_count")) + .map(|dead| dead >= min_dead) + .unwrap_or(false) + }); + + if any_sees_dead { + return true; + } + + thread::sleep(Duration::from_secs(1)); + } +} + +// ── Scenarios ─────────────────────────────────────────────────────────── + +fn scenario_cluster_convergence(binary: &Path, base_dir: &Path) -> Result<(), String> { + let run_dir = base_dir.join("scenario-1"); + let cluster = SimCluster::spawn(binary, &run_dir, NODE_COUNT); + let ports = all_dashboard_ports(NODE_COUNT); + + // alive_count excludes self, so each node sees NODE_COUNT - 1 peers + let expected_alive = NODE_COUNT - 1; + let elapsed = wait_for_convergence(&ports, expected_alive, TIMEOUT)?; + println!(" converged in {:.1}s", elapsed.as_secs_f64()); + + // Verify each node's snapshot + for (i, &port) in ports.iter().enumerate() { + let snap = poll_distribution(port) + .ok_or_else(|| format!("node {i} (port {port}) unreachable after convergence"))?; + let alive = get_usize(&snap, "alive_count").unwrap_or(0); + let routing = get_usize(&snap, "routing_table_size").unwrap_or(0); + if alive < expected_alive { + return Err(format!("node {i} sees {alive} alive, expected >= {expected_alive}")); + } + if routing < NODE_COUNT - 1 { + return Err(format!( + "node {i} has routing_table_size {routing}, expected >= {}", + NODE_COUNT - 1 + )); + } + } + + drop(cluster); + Ok(()) +} + +fn scenario_node_death_detection(binary: &Path, base_dir: &Path) -> Result<(), String> { + let run_dir = base_dir.join("scenario-2"); + let mut cluster = SimCluster::spawn(binary, &run_dir, NODE_COUNT); + let ports = all_dashboard_ports(NODE_COUNT); + + let expected_alive = NODE_COUNT - 1; + wait_for_convergence(&ports, expected_alive, TIMEOUT) + .map_err(|e| format!("pre-kill convergence failed: {e}"))?; + + // Kill node 2 + cluster.kill_node(2); + + // Survivors: all except index 2 + let survivor_ports: Vec = (0..NODE_COUNT) + .filter(|&i| i != 2) + .map(dashboard_port) + .collect(); + + // Wait for alive count to drop (alive_count excludes self, so 5-node cluster + // sees 4 alive; after killing 1, survivors should see <= 3) + let elapsed = wait_for_death_detection(&survivor_ports, NODE_COUNT - 2, TIMEOUT)?; + println!(" detected in {:.1}s", elapsed.as_secs_f64()); + + // Poll for dead_count — SWIM transitions suspect→dead with a delay + let dead_detected = wait_for_dead_count(&survivor_ports, 1, TIMEOUT); + if !dead_detected { + return Err("no survivor detected a dead member".into()); + } + + drop(cluster); + Ok(()) +} + +fn scenario_killed_node_rejoins(binary: &Path, base_dir: &Path) -> Result<(), String> { + let run_dir = base_dir.join("scenario-3"); + let mut cluster = SimCluster::spawn(binary, &run_dir, NODE_COUNT); + let ports = all_dashboard_ports(NODE_COUNT); + + let expected_alive = NODE_COUNT - 1; + wait_for_convergence(&ports, expected_alive, TIMEOUT) + .map_err(|e| format!("pre-kill convergence failed: {e}"))?; + + // Kill node 2 + cluster.kill_node(2); + + let survivor_ports: Vec = (0..NODE_COUNT) + .filter(|&i| i != 2) + .map(dashboard_port) + .collect(); + wait_for_death_detection(&survivor_ports, NODE_COUNT - 2, TIMEOUT) + .map_err(|e| format!("death detection failed: {e}"))?; + + // Restart node 2 — identity persists, so cluster recognizes it + cluster.restart_node(2); + + let rejoined_port = dashboard_port(2); + let elapsed = wait_for_convergence(&[rejoined_port], 1, TIMEOUT)?; + println!(" rejoined in {:.1}s", elapsed.as_secs_f64()); + + let snap = poll_distribution(rejoined_port) + .ok_or("rejoined node unreachable")?; + let alive = get_usize(&snap, "alive_count").unwrap_or(0); + if alive < 1 { + return Err(format!("rejoined node sees {alive} alive, expected >= 1")); + } + + drop(cluster); + Ok(()) +} + +fn scenario_actors_resolvable(binary: &Path, base_dir: &Path) -> Result<(), String> { + let run_dir = base_dir.join("scenario-4"); + let cluster = SimCluster::spawn(binary, &run_dir, NODE_COUNT); + let ports = all_dashboard_ports(NODE_COUNT); + + let expected_alive = NODE_COUNT - 1; + wait_for_convergence(&ports, expected_alive, TIMEOUT) + .map_err(|e| format!("convergence failed: {e}"))?; + + let mut total_directory_entries = 0usize; + + for (i, &port) in ports.iter().enumerate() { + let snap = poll_distribution(port) + .ok_or_else(|| format!("node {i} unreachable"))?; + let dir_count = get_usize(&snap, "directory_entry_count").unwrap_or(0); + if dir_count < ACTORS_PER_NODE { + return Err(format!( + "node {i} has {dir_count} directory entries, expected >= {ACTORS_PER_NODE}" + )); + } + total_directory_entries += dir_count; + } + + let expected_total = NODE_COUNT * ACTORS_PER_NODE; + if total_directory_entries < expected_total { + return Err(format!( + "total directory entries {total_directory_entries}, expected >= {expected_total}" + )); + } + println!(" total: {total_directory_entries} directory entries"); + + drop(cluster); + Ok(()) +} + +// ── Interactive mode ───────────────────────────────────────────────────── + +pub fn run_interactive(node_count: usize) { + use std::sync::atomic::{AtomicBool, Ordering}; + use std::sync::{Arc, Condvar, Mutex}; + + let root = workspace_root(); + + println!("=== sim-cluster: building swactor (iroh) ==="); + let binary = build_swactor(&root); + + let base_dir = root.join(".sim-cluster"); + let _ = fs::remove_dir_all(&base_dir); + + let run_dir = base_dir.join("interactive"); + println!("=== sim-cluster: spawning {node_count} nodes ==="); + let cluster = SimCluster::spawn(&binary, &run_dir, node_count); + + let ports = all_dashboard_ports(node_count); + + // Wait for all dashboards + println!("=== sim-cluster: waiting for dashboards ==="); + for &port in &ports { + wait_for_dashboard(port, Duration::from_secs(30)); + } + + // Wait for convergence + println!("=== sim-cluster: waiting for convergence ==="); + let expected_alive = node_count - 1; + let convergence = wait_for_convergence(&ports, expected_alive, TIMEOUT); + + let converged_msg = match &convergence { + Ok(elapsed) => format!("{node_count} nodes, all converged in {:.1}s", elapsed.as_secs_f64()), + Err(e) => format!("{node_count} nodes, convergence issue: {e}"), + }; + + // Print summary + println!(); + println!("=== sim-cluster ready ==="); + println!(" relay: http://127.0.0.1:{}/", cluster.relay.port); + for i in 0..node_count { + let port = dashboard_port(i); + let label = if i == 0 { " (seed)" } else { "" }; + println!(" node-{i}: http://127.0.0.1:{port}/{label}"); + } + println!(" cluster: {converged_msg}"); + println!(); + println!(" Logs: .sim-cluster/interactive/node-N/stderr.log"); + println!(" Press Ctrl-C to shut down."); + + // Block on Ctrl-C + let shutdown = Arc::new((Mutex::new(false), Condvar::new())); + let shutdown2 = Arc::clone(&shutdown); + let flag = Arc::new(AtomicBool::new(false)); + let flag2 = Arc::clone(&flag); + + unsafe { + let shutdown_ptr = Arc::into_raw(shutdown2) as usize; + let flag_ptr = Arc::into_raw(flag2) as usize; + libc::signal(libc::SIGINT, handler as *const () as libc::sighandler_t); + SHUTDOWN_PTR.store(shutdown_ptr, Ordering::SeqCst); + FLAG_PTR.store(flag_ptr, Ordering::SeqCst); + } + + let (lock, cvar) = &*shutdown; + let mut stopped = lock.lock().unwrap(); + while !*stopped { + stopped = cvar.wait(stopped).unwrap(); + } + + println!("\n=== sim-cluster: shutting down ==="); + drop(cluster); + println!("=== sim-cluster: stopped ==="); +} + +// Signal handler support for run_interactive +use std::sync::atomic::{AtomicUsize, Ordering}; + +static SHUTDOWN_PTR: AtomicUsize = AtomicUsize::new(0); +static FLAG_PTR: AtomicUsize = AtomicUsize::new(0); + +extern "C" fn handler(_sig: libc::c_int) { + use std::sync::atomic::AtomicBool; + use std::sync::{Condvar, Mutex}; + + let flag_ptr = FLAG_PTR.load(Ordering::SeqCst); + if flag_ptr != 0 { + let flag = unsafe { &*(flag_ptr as *const AtomicBool) }; + if flag.swap(true, Ordering::SeqCst) { + // Second Ctrl-C — force exit + std::process::exit(1); + } + } + + let ptr = SHUTDOWN_PTR.load(Ordering::SeqCst); + if ptr != 0 { + let pair = unsafe { &*(ptr as *const (Mutex, Condvar)) }; + if let Ok(mut stopped) = pair.0.lock() { + *stopped = true; + pair.1.notify_one(); + } + } +} + +// ── Entry point (test mode) ───────────────────────────────────────────── + +pub fn run() { + let root = workspace_root(); + let overall_start = Instant::now(); + + println!("=== sim-cluster: building swactor (iroh) ==="); + let binary = build_swactor(&root); + + let base_dir = root.join(".sim-cluster"); + // Clean any stale runs + let _ = fs::remove_dir_all(&base_dir); + + let scenarios: &[(&str, fn(&Path, &Path) -> Result<(), String>)] = &[ + ("cluster convergence", scenario_cluster_convergence), + ("node death detection", scenario_node_death_detection), + ("killed node rejoins", scenario_killed_node_rejoins), + ("actors resolvable", scenario_actors_resolvable), + ]; + + let total = scenarios.len(); + let mut passed = 0usize; + + for (i, (name, func)) in scenarios.iter().enumerate() { + println!( + "=== sim-cluster: scenario {}/{total} \u{2014} {name} ===", + i + 1 + ); + match func(&binary, &base_dir) { + Ok(()) => passed += 1, + Err(e) => { + let elapsed = overall_start.elapsed(); + eprintln!(" FAILED: {e}"); + eprintln!( + "\n--- FAILED after {:.1}s ({passed}/{total} passed) ---", + elapsed.as_secs_f64() + ); + // Leave .sim-cluster for debugging + std::process::exit(1); + } + } + } + + // Clean up base dir + let _ = fs::remove_dir_all(&base_dir); + + let elapsed = overall_start.elapsed(); + println!( + "\n--- All {total} scenario(s) passed in {:.1}s ---", + elapsed.as_secs_f64() + ); +}