diff --git a/crates/swactor-gossip/tests/bug_hunt_gossip.rs b/crates/swactor-gossip/tests/bug_hunt_gossip.rs new file mode 100644 index 0000000..e438ce0 --- /dev/null +++ b/crates/swactor-gossip/tests/bug_hunt_gossip.rs @@ -0,0 +1,139 @@ +//! Bug-hunt tests for swactor-gossip. +//! +//! Each test asserts correct protocol behaviour and **fails** when +//! the bug is present. + +use swactor::runtime::{Inbox, Runtime, RuntimeConfig}; +use swactor_gossip::{GossipActor, GossipMessage, GossipQueryResponse, GossipState}; + +// ═════════════════════════════════════════════════════════════════════════ +// Bug 1 — GossipState::merge drops remote value on version ties +// +// `merge` uses `remote.version > local.version`. When two nodes +// independently `set` the same key, both create version 1. Neither +// value dominates the other, so the nodes never converge — permanent +// split-brain. +// +// A correct LWW register needs a tiebreaker (e.g. node-id comparison +// or a Lamport timestamp) so that one value always wins. +// ═════════════════════════════════════════════════════════════════════════ + +/// Direct GossipState unit test — no runtime needed. +#[test] +fn merge_equal_version_causes_permanent_divergence() { + let mut state_a = GossipState::new(); + let mut state_b = GossipState::new(); + + // Both independently set "x" → both get version 1. + state_a.set("x".into(), b"alpha".to_vec()); + state_b.set("x".into(), b"beta".to_vec()); + + assert_eq!(state_a.get("x").unwrap().version, 1); + assert_eq!(state_b.get("x").unwrap().version, 1); + + // Simulate a full gossip exchange (both directions). + state_a.merge(&state_b); + state_b.merge(&state_a); + + // Extra rounds — shouldn't matter, but proves the deadlock is permanent. + for _ in 0..10 { + state_a.merge(&state_b); + state_b.merge(&state_a); + } + + let val_a = &state_a.get("x").unwrap().value; + let val_b = &state_b.get("x").unwrap().value; + + assert_eq!( + val_a, val_b, + "BUG: permanent divergence — A holds {:?}, B holds {:?}, \ + both at version 1. merge() rejects the remote value when \ + versions are equal, so neither node ever adopts the other's value.", + String::from_utf8_lossy(val_a), + String::from_utf8_lossy(val_b), + ); +} + +/// Full actor-level test: two gossip nodes with concurrent writes. +#[test] +fn concurrent_writes_diverge_through_gossip_actors() { + let rt = Runtime::new(RuntimeConfig::default()); + + let node_a = rt.spawn(GossipActor::new()).unwrap(); + let node_b = rt.spawn(GossipActor::new()).unwrap(); + + // Wire bidirectional peering. + rt.send_to(node_a, GossipMessage::AddPeer(node_b)).unwrap(); + rt.send_to(node_b, GossipMessage::AddPeer(node_a)).unwrap(); + for _ in 0..5 { + rt.tick(); + } + + // Both set the same key to different values → version 1 on each. + rt.send_to( + node_a, + GossipMessage::Set { + key: "x".into(), + value: b"alpha".to_vec(), + }, + ) + .unwrap(); + rt.send_to( + node_b, + GossipMessage::Set { + key: "x".into(), + value: b"beta".to_vec(), + }, + ) + .unwrap(); + for _ in 0..5 { + rt.tick(); + } + + // Run many gossip rounds. + for _ in 0..50 { + rt.send_to(node_a, GossipMessage::DoGossipRound).unwrap(); + rt.send_to(node_b, GossipMessage::DoGossipRound).unwrap(); + for _ in 0..10 { + rt.tick(); + } + } + + // Query both nodes. + let inbox_a: Inbox = rt.new_inbox().unwrap(); + let inbox_b: Inbox = rt.new_inbox().unwrap(); + + rt.send_to( + node_a, + GossipMessage::Query { + key: "x".into(), + reply_to: *inbox_a.addr(), + }, + ) + .unwrap(); + rt.send_to( + node_b, + GossipMessage::Query { + key: "x".into(), + reply_to: *inbox_b.addr(), + }, + ) + .unwrap(); + for _ in 0..10 { + rt.tick(); + } + + let resp_a = inbox_a.try_recv().expect("response from A"); + let resp_b = inbox_b.try_recv().expect("response from B"); + + assert_eq!( + resp_a.value, resp_b.value, + "BUG: permanent split-brain — node A has {:?}, node B has {:?} \ + (versions {:?} vs {:?}). merge() rejects the remote value on \ + equal versions, so the two nodes can never agree.", + resp_a.value.as_ref().map(|v| String::from_utf8_lossy(v).to_string()), + resp_b.value.as_ref().map(|v| String::from_utf8_lossy(v).to_string()), + resp_a.version, + resp_b.version, + ); +} diff --git a/src/channel.rs b/src/channel.rs index c487ec4..7f0c92a 100644 --- a/src/channel.rs +++ b/src/channel.rs @@ -17,6 +17,10 @@ impl HybridChannel { } pub fn push(&self, value: T) -> Result<(), T> { + if !self.overflow.is_empty() { + self.overflow.push(value); + return Ok(()); + } match self.ring.push(value) { Ok(()) => Ok(()), Err(v) => { diff --git a/src/config.rs b/src/config.rs index 95eab7b..9dec626 100644 --- a/src/config.rs +++ b/src/config.rs @@ -48,7 +48,8 @@ impl Default for RuntimeConfig { max_actors: DEFAULT_MAX_ACTORS, actor_max_messages: DEFAULT_ACTOR_MAX_MESSAGES, num_threads: 1, - mailbox_waterlevel: DEFAULT_MAILBOX_WATERLEVEL, + // Clamp to minimum 2: drain_count(1,1)==0 causes livelock + mailbox_waterlevel: DEFAULT_MAILBOX_WATERLEVEL.max(2), backoff_policy: BackoffPolicy::default(), } } diff --git a/src/worker.rs b/src/worker.rs index 3cd5c34..24c981c 100644 --- a/src/worker.rs +++ b/src/worker.rs @@ -71,7 +71,14 @@ impl Worker { } } - // 4. Drain pending_local buffer → deliver to local actors + // 4. Drain spawn queue again — actors spawned during step 3 + // must be in the pool before pending_local delivery. + while let Some((addr, actor)) = self.spawn_rx.try_recv() { + self.pool.insert(addr, actor); + did_work = true; + } + + // 5. Drain pending_local buffer → deliver to local actors let pending = pending_local.into_inner(); if !pending.is_empty() { did_work = true; @@ -80,7 +87,7 @@ impl Worker { self.pool.deliver(&addr, msg); } - // 5. Publish stats + // 6. Publish stats self.stats.num_actors.store(self.pool.len(), Ordering::Relaxed); self.stats.total_mailbox_depth.store(self.pool.total_mailbox_depth(), Ordering::Relaxed); self.stats.messages_processed.fetch_add(processed as u64, Ordering::Relaxed); diff --git a/tests/runtime_api.rs b/tests/runtime_api.rs index 824f2f2..51c6389 100644 --- a/tests/runtime_api.rs +++ b/tests/runtime_api.rs @@ -156,9 +156,314 @@ fn send_to_unknown_address_fails() { #[test] fn actor_spawns_child_and_delegates() { - // Uses 2 workers so parent and child land on different workers, - // avoiding the single-worker timing issue where pending_local - // delivery precedes spawn-queue draining. + let rt = Runtime::new(RuntimeConfig { + num_threads: 1, + ..Default::default() + }); + + let parent = rt.spawn(DelegateActor).expect("spawn parent"); + let inbox: Inbox = rt.new_inbox().unwrap(); + + rt.send_to( + parent, + DelegateRequest { + value: 7, + reply_to: *inbox.addr(), + }, + ) + .unwrap(); + + for _ in 0..10 { + rt.tick(); + if let Some(resp) = inbox.try_recv() { + assert_eq!(resp, DoubleResponse(14)); + return; + } + } + + panic!("Did not receive DoubleResponse"); +} + +#[test] +fn multiple_actors_independent_mailboxes() { + let rt = Runtime::new(RuntimeConfig::default()); + + let inbox_a: Inbox = rt.new_inbox().unwrap(); + let inbox_b: Inbox = rt.new_inbox().unwrap(); + let inbox_c: Inbox = rt.new_inbox().unwrap(); + + let actor_a = rt.spawn(EchoActor).unwrap(); + let actor_b = rt.spawn(EchoActor).unwrap(); + let actor_c = rt.spawn(EchoActor).unwrap(); + + rt.send_to(actor_a, EchoMessage { payload: 10, reply_to: *inbox_a.addr() }).unwrap(); + rt.send_to(actor_b, EchoMessage { payload: 20, reply_to: *inbox_b.addr() }).unwrap(); + rt.send_to(actor_c, EchoMessage { payload: 30, reply_to: *inbox_c.addr() }).unwrap(); + + for _ in 0..10 { + rt.tick(); + } + + assert_eq!(inbox_a.try_recv(), Some(EchoResponse(10))); + assert_eq!(inbox_b.try_recv(), Some(EchoResponse(20))); + assert_eq!(inbox_c.try_recv(), Some(EchoResponse(30))); + // No cross-contamination + assert_eq!(inbox_a.try_recv(), None); + assert_eq!(inbox_b.try_recv(), None); + assert_eq!(inbox_c.try_recv(), None); +} + +// --------------------------------------------------------------------------- +// Additional fixtures for spawn/delegate tests +// --------------------------------------------------------------------------- + +/// Three-deep chain: Parent → Child → Grandchild → reply_to +struct ChainActor { + depth: usize, +} + +#[derive(Clone)] +struct ChainRequest { + remaining: usize, + reply_to: ActorAddress, +} + +#[derive(Clone, Debug, PartialEq)] +struct ChainDone(usize); + +impl ActorInterface for ChainActor { + type Incoming = ChainRequest; + type Response = ChainDone; + + fn handle(&mut self, ctx: &Ctx, msg: ChainRequest) { + if msg.remaining == 0 { + let _ = ctx.send(msg.reply_to, ChainDone(self.depth)); + } else { + let child = ctx + .spawn(ChainActor { + depth: self.depth + 1, + }) + .expect("spawn chain child"); + let _ = ctx.send( + child, + ChainRequest { + remaining: msg.remaining - 1, + reply_to: msg.reply_to, + }, + ); + } + } +} + +/// Spawns N children and sends work to each. Each child replies to reply_to. +struct FanOutActor; + +#[derive(Clone)] +struct FanOutRequest { + count: usize, + reply_to: ActorAddress, +} + +impl ActorInterface for FanOutActor { + type Incoming = FanOutRequest; + type Response = (); + + fn handle(&mut self, ctx: &Ctx, msg: FanOutRequest) { + for i in 0..msg.count { + let child = ctx.spawn(DoubleActor).expect("spawn fan-out child"); + let _ = ctx.send( + child, + DoubleRequest { + value: i + 1, + reply_to: msg.reply_to, + }, + ); + } + } +} + +/// Actor A: spawns a child, sends work to it, and also forwards the child's +/// address to a "buddy" so the buddy can send to the child too. +struct SpawnAndBroadcastActor; + +#[derive(Clone)] +struct SpawnAndBroadcastRequest { + buddy: ActorAddress, + reply_to: ActorAddress, +} + +/// Sent from A to buddy B, carrying the child's address. +#[derive(Clone)] +struct ForwardToChild { + child: ActorAddress, + reply_to: ActorAddress, +} + +impl ActorInterface for SpawnAndBroadcastActor { + type Incoming = SpawnAndBroadcastRequest; + type Response = (); + + fn handle(&mut self, ctx: &Ctx, msg: SpawnAndBroadcastRequest) { + let child = ctx.spawn(DoubleActor).expect("spawn child"); + // Send work from self + let _ = ctx.send( + child, + DoubleRequest { + value: 10, + reply_to: msg.reply_to, + }, + ); + // Tell buddy about the child + let _ = ctx.send( + msg.buddy, + ForwardToChild { + child, + reply_to: msg.reply_to, + }, + ); + } +} + +/// Buddy actor B: receives a child address and sends work to it. +struct BuddyActor; + +impl ActorInterface for BuddyActor { + type Incoming = ForwardToChild; + type Response = (); + + fn handle(&mut self, ctx: &Ctx, msg: ForwardToChild) { + let _ = ctx.send( + msg.child, + DoubleRequest { + value: 20, + reply_to: msg.reply_to, + }, + ); + } +} + +// --------------------------------------------------------------------------- +// Bug-fix regression tests +// --------------------------------------------------------------------------- + +#[test] +fn inbox_preserves_fifo_across_overflow() { + let rt = Runtime::new(RuntimeConfig { + actor_max_messages: 2, + ..Default::default() + }); + + let inbox: Inbox = rt.new_inbox().unwrap(); + let addr = *inbox.addr(); + + // Push 4 items: ring gets [1,2], overflow gets [3,4]. + for v in 1..=4u64 { + rt.send_to(addr, v).unwrap(); + } + + // Drain the ring. + assert_eq!(inbox.try_recv(), Some(1)); + assert_eq!(inbox.try_recv(), Some(2)); + + // Push a new item — must land behind 3 and 4 in overflow. + rt.send_to(addr, 5u64).unwrap(); + + let rest: Vec = std::iter::from_fn(|| inbox.try_recv()).collect(); + assert_eq!(rest, vec![3, 4, 5]); +} + +#[test] +fn delegate_works_on_single_worker() { + let rt = Runtime::new(RuntimeConfig { + num_threads: 1, + ..Default::default() + }); + + let parent = rt.spawn(DelegateActor).expect("spawn parent"); + let inbox: Inbox = rt.new_inbox().unwrap(); + + rt.send_to( + parent, + DelegateRequest { + value: 5, + reply_to: *inbox.addr(), + }, + ) + .unwrap(); + + for _ in 0..10 { + rt.tick(); + if let Some(resp) = inbox.try_recv() { + assert_eq!(resp, DoubleResponse(10)); + return; + } + } + + panic!("Did not receive DoubleResponse on single worker"); +} + +#[test] +fn spawn_chain_three_deep() { + let rt = Runtime::new(RuntimeConfig { + num_threads: 1, + ..Default::default() + }); + + let root = rt.spawn(ChainActor { depth: 0 }).expect("spawn root"); + let inbox: Inbox = rt.new_inbox().unwrap(); + + rt.send_to( + root, + ChainRequest { + remaining: 2, + reply_to: *inbox.addr(), + }, + ) + .unwrap(); + + for _ in 0..50 { + rt.tick(); + if let Some(resp) = inbox.try_recv() { + assert_eq!(resp, ChainDone(2)); + return; + } + } + + panic!("Did not receive ChainDone from grandchild"); +} + +#[test] +fn spawn_fan_out() { + let rt = Runtime::new(RuntimeConfig { + num_threads: 1, + ..Default::default() + }); + + let fan = rt.spawn(FanOutActor).expect("spawn fan-out"); + let inbox: Inbox = rt.new_inbox().unwrap(); + + rt.send_to( + fan, + FanOutRequest { + count: 5, + reply_to: *inbox.addr(), + }, + ) + .unwrap(); + + for _ in 0..50 { + rt.tick(); + } + + let mut results: Vec = std::iter::from_fn(|| inbox.try_recv()) + .map(|r| r.0) + .collect(); + results.sort(); + assert_eq!(results, vec![2, 4, 6, 8, 10]); +} + +#[test] +fn delegate_works_cross_worker() { let config = RuntimeConfig { num_threads: 2, ..Default::default() @@ -196,34 +501,40 @@ fn actor_spawns_child_and_delegates() { } #[test] -fn multiple_actors_independent_mailboxes() { - let rt = Runtime::new(RuntimeConfig::default()); +fn multiple_senders_to_new_child() { + let rt = Runtime::new(RuntimeConfig { + num_threads: 1, + ..Default::default() + }); - let inbox_a: Inbox = rt.new_inbox().unwrap(); - let inbox_b: Inbox = rt.new_inbox().unwrap(); - let inbox_c: Inbox = rt.new_inbox().unwrap(); + let buddy = rt.spawn(BuddyActor).expect("spawn buddy"); + let parent = rt.spawn(SpawnAndBroadcastActor).expect("spawn parent"); + let inbox: Inbox = rt.new_inbox().unwrap(); - let actor_a = rt.spawn(EchoActor).unwrap(); - let actor_b = rt.spawn(EchoActor).unwrap(); - let actor_c = rt.spawn(EchoActor).unwrap(); + rt.send_to( + parent, + SpawnAndBroadcastRequest { + buddy, + reply_to: *inbox.addr(), + }, + ) + .unwrap(); - rt.send_to(actor_a, EchoMessage { payload: 10, reply_to: *inbox_a.addr() }).unwrap(); - rt.send_to(actor_b, EchoMessage { payload: 20, reply_to: *inbox_b.addr() }).unwrap(); - rt.send_to(actor_c, EchoMessage { payload: 30, reply_to: *inbox_c.addr() }).unwrap(); - - for _ in 0..10 { + for _ in 0..50 { rt.tick(); } - assert_eq!(inbox_a.try_recv(), Some(EchoResponse(10))); - assert_eq!(inbox_b.try_recv(), Some(EchoResponse(20))); - assert_eq!(inbox_c.try_recv(), Some(EchoResponse(30))); - // No cross-contamination - assert_eq!(inbox_a.try_recv(), None); - assert_eq!(inbox_b.try_recv(), None); - assert_eq!(inbox_c.try_recv(), None); + let mut results: Vec = std::iter::from_fn(|| inbox.try_recv()) + .map(|r| r.0) + .collect(); + results.sort(); + assert_eq!(results, vec![20, 40]); } +// --------------------------------------------------------------------------- +// Distribution tests +// --------------------------------------------------------------------------- + #[test] fn round_robin_distributes_across_workers() { let config = RuntimeConfig {