fix: FIFO ordering for channel, race condition for child actors

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<GossipQueryResponse> = rt.new_inbox().unwrap();
+    let inbox_b: Inbox<GossipQueryResponse> = 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,
+    );
+}

src/channel.rs

@@ -17,6 +17,10 @@ impl<T> HybridChannel<T> {
     }
 
     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) => {

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(),
         }
     }

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);

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<DoubleResponse> = 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<EchoResponse> = rt.new_inbox().unwrap();
+    let inbox_b: Inbox<EchoResponse> = rt.new_inbox().unwrap();
+    let inbox_c: Inbox<EchoResponse> = 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<u64> = 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<u64> = 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<DoubleResponse> = 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<ChainDone> = 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<DoubleResponse> = 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<usize> = 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<EchoResponse> = rt.new_inbox().unwrap();
-    let inbox_b: Inbox<EchoResponse> = rt.new_inbox().unwrap();
-    let inbox_c: Inbox<EchoResponse> = rt.new_inbox().unwrap();
+    let buddy = rt.spawn(BuddyActor).expect("spawn buddy");
+    let parent = rt.spawn(SpawnAndBroadcastActor).expect("spawn parent");
+    let inbox: Inbox<DoubleResponse> = 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<usize> = 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 {