From d3c23c71482fd6a946e5ddb0af8b84f45a6862b7 Mon Sep 17 00:00:00 2001 From: Zachery Aaron Shores-Chmielewski Date: Fri, 6 Feb 2026 21:42:47 +0700 Subject: [PATCH] refactor: better worker thread tests --- src/worker/mod.rs | 2 +- src/worker/tests.rs | 882 +++++++++++++++++--------------------------- 2 files changed, 346 insertions(+), 538 deletions(-) diff --git a/src/worker/mod.rs b/src/worker/mod.rs index f113ba6..493b9c4 100644 --- a/src/worker/mod.rs +++ b/src/worker/mod.rs @@ -274,7 +274,7 @@ impl ActorPool { -pub struct Mailbox { +pub(crate) struct Mailbox { queue: VecDeque, waterlevel: usize, } diff --git a/src/worker/tests.rs b/src/worker/tests.rs index 0048b61..720d968 100644 --- a/src/worker/tests.rs +++ b/src/worker/tests.rs @@ -1,4 +1,5 @@ use std::any::Any; +use std::collections::HashMap; use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use std::sync::Arc; use std::thread; @@ -7,587 +8,398 @@ use crate::actor::{ActorAddress, AnyActor}; use crate::address_map::{AddressMap, Placement, WorkerId}; use crate::channel::Receiver; use crate::config::{BackoffPolicy, RuntimeConfig}; -use crate::runtime::{ContextInner, Ctx, Envelope, InboxRegistry}; -use super::{ActorPool, Mailbox, TickContext, Worker, WorkerStats}; -use crate::Error; +use crate::runtime::{Ctx, Envelope, InboxRegistry}; -// ── Test helpers ──────────────────────────────────────────────────── +use super::{TickContext, Worker, WorkerStats}; -#[derive(Clone, Debug, PartialEq)] -struct TestMsg(u64); +// ── Actors ───────────────────────────────────────────────────────── -/// A minimal actor that counts how many messages it handled. -struct CounterActor { - counter: Arc, -} +/// Counts how many u64 messages it successfully handled. +struct CounterActor(Arc); impl AnyActor for CounterActor { fn handle_any(&mut self, _ctx: &Ctx, msg: Box) { if msg.downcast::().is_ok() { - self.counter.fetch_add(1, Ordering::Relaxed); + self.0.fetch_add(1, Ordering::Relaxed); } } } -fn make_test_actor( - _addr: ActorAddress, -) -> (Box, Arc) { - let counter = Arc::new(AtomicUsize::new(0)); - let actor = CounterActor { - counter: counter.clone(), - }; - (Box::new(actor), counter) -} +// ── Harness ──────────────────────────────────────────────────────── -/// No-op ContextInner for ActorPool tests. -struct StubContextInner { - waterlevel: usize, -} - -impl ContextInner for StubContextInner { - fn send_any(&self, _addr: ActorAddress, _msg: Box) -> Result<(), Error> { - Ok(()) - } - - fn spawn_any(&self, _addr: ActorAddress, _actor: Box) -> Result<(), Error> { - Ok(()) - } - - fn mailbox_waterlevel(&self) -> usize { - self.waterlevel - } -} - -fn make_addr(id: u8) -> ActorAddress { +fn addr(id: u8) -> ActorAddress { let mut bytes = [0u8; 32]; bytes[0] = id; ActorAddress(bytes) } -// ── Mailbox tests ─────────────────────────────────────────────────── - -#[test] -fn mailbox_new_is_empty() { - let mb: Mailbox = Mailbox::new(10); - assert_eq!(mb.len(), 0); - assert!(mb.is_empty()); +/// Self-contained single-worker test environment. +/// +/// Holds the worker, its channels, shared state for TickContext, and +/// per-actor handle counters — everything needed to drive scenarios. +struct Env { + worker: Worker, + stats: Arc, + feed_transfer: crate::channel::Sender, + feed_spawn: crate::channel::Sender<(ActorAddress, Box)>, + address_map: AddressMap, + placement: Placement, + inbox_registry: InboxRegistry, + config: RuntimeConfig, + tc_transfer_txs: Vec>, + tc_spawn_txs: Vec)>>, + counters: HashMap>, } -#[test] -fn mailbox_push_increments_len() { - let mut mb: Mailbox = Mailbox::new(10); - mb.push(1); - assert_eq!(mb.len(), 1); - mb.push(2); - assert_eq!(mb.len(), 2); -} - -#[test] -fn mailbox_pop_returns_fifo_order() { - let mut mb: Mailbox = Mailbox::new(10); - mb.push(10); - mb.push(20); - mb.push(30); - assert_eq!(mb.pop(), Some(10)); - assert_eq!(mb.pop(), Some(20)); - assert_eq!(mb.pop(), Some(30)); -} - -#[test] -fn mailbox_pop_empty_returns_none() { - let mut mb: Mailbox = Mailbox::new(10); - assert_eq!(mb.pop(), None); -} - -#[test] -fn mailbox_pop_drains_to_empty() { - let mut mb: Mailbox = Mailbox::new(10); - mb.push(1); - mb.push(2); - mb.pop(); - mb.pop(); - assert!(mb.is_empty()); - assert_eq!(mb.len(), 0); -} - -#[test] -fn mailbox_push_pop_interleaved() { - let mut mb: Mailbox = Mailbox::new(10); - mb.push(1); - mb.push(2); - assert_eq!(mb.pop(), Some(1)); - mb.push(3); - assert_eq!(mb.pop(), Some(2)); - assert_eq!(mb.pop(), Some(3)); - assert_eq!(mb.pop(), None); -} - -#[test] -fn drain_count_empty() { - let mb: Mailbox = Mailbox::new(10); - assert_eq!(mb.drain_count(), 0); -} - -#[test] -fn drain_count_below_waterlevel() { - let mut mb: Mailbox = Mailbox::new(10); - for i in 0..5 { - mb.push(i); +impl Env { + fn new() -> Self { + Self::with_config(RuntimeConfig::default()) } - assert_eq!(mb.drain_count(), 5); -} -#[test] -fn drain_count_at_waterlevel_minus_one() { - let mut mb: Mailbox = Mailbox::new(10); - for i in 0..9 { - mb.push(i); + fn with_config(config: RuntimeConfig) -> Self { + let transfer_rx = Receiver::::new(256); + let spawn_rx = Receiver::<(ActorAddress, Box)>::new(256); + + let feed_transfer = transfer_rx.new_sender(); + let feed_spawn = spawn_rx.new_sender(); + let tc_transfer = transfer_rx.new_sender(); + let tc_spawn = spawn_rx.new_sender(); + + let stats = Arc::new(WorkerStats::new()); + let worker = Worker::new(WorkerId(0), transfer_rx, spawn_rx, stats.clone()); + + Self { + worker, + stats, + feed_transfer, + feed_spawn, + address_map: AddressMap::new(), + placement: Placement::new(1), + inbox_registry: InboxRegistry::new(), + config, + tc_transfer_txs: vec![tc_transfer], + tc_spawn_txs: vec![tc_spawn], + counters: HashMap::new(), + } } - // len=9 < waterlevel=10 → returns len - assert_eq!(mb.drain_count(), 9); -} -#[test] -fn drain_count_at_waterlevel() { - let mut mb: Mailbox = Mailbox::new(10); - for i in 0..10 { - mb.push(i); + /// Enqueue an actor spawn (drained on next tick, phase 1). + fn spawn(&mut self, id: u8) { + let counter = Arc::new(AtomicUsize::new(0)); + let actor: Box = Box::new(CounterActor(counter.clone())); + self.feed_spawn.try_send((addr(id), actor)).ok().unwrap(); + self.counters.insert(id, counter); } - // len=10 >= waterlevel=10 → returns len >> 1 = 5 - assert_eq!(mb.drain_count(), 5); -} -#[test] -fn drain_count_at_waterlevel_plus_one() { - let mut mb: Mailbox = Mailbox::new(10); - for i in 0..11 { - mb.push(i); - } - // len=11 >= waterlevel=10 → returns 11 >> 1 = 5 - assert_eq!(mb.drain_count(), 5); -} - -#[test] -fn drain_count_well_above_waterlevel() { - let mut mb: Mailbox = Mailbox::new(10); - for i in 0..100 { - mb.push(i); - } - // len=100 >= waterlevel=10 → returns 100 >> 1 = 50 - assert_eq!(mb.drain_count(), 50); -} - -#[test] -fn drain_count_odd_len_truncates() { - let mut mb: Mailbox = Mailbox::new(1); - for i in 0..7 { - mb.push(i); - } - // len=7 >= waterlevel=1 → returns 7 >> 1 = 3 - assert_eq!(mb.drain_count(), 3); -} - -#[test] -fn drain_count_waterlevel_one() { - let mut mb: Mailbox = Mailbox::new(1); - mb.push(42); - // len=1 >= waterlevel=1 → returns 1 >> 1 = 0 - assert_eq!(mb.drain_count(), 0); -} - -#[test] -fn drain_count_waterlevel_zero() { - // waterlevel=0 means len >= 0 is always true → always half-drain - let mb: Mailbox = Mailbox::new(0); - assert_eq!(mb.drain_count(), 0); // empty: 0 >> 1 = 0 - - let mut mb2: Mailbox = Mailbox::new(0); - mb2.push(1); - // len=1 >= waterlevel=0 → returns 1 >> 1 = 0 - assert_eq!(mb2.drain_count(), 0); - - let mut mb3: Mailbox = Mailbox::new(0); - mb3.push(1); - mb3.push(2); - // len=2 >= waterlevel=0 → returns 2 >> 1 = 1 - assert_eq!(mb3.drain_count(), 1); -} - -#[test] -fn drain_count_does_not_mutate() { - let mut mb: Mailbox = Mailbox::new(10); - for i in 0..5 { - mb.push(i); - } - let dc1 = mb.drain_count(); - let dc2 = mb.drain_count(); - assert_eq!(dc1, dc2); - assert_eq!(mb.len(), 5); -} - -#[test] -fn drain_count_large_waterlevel() { - let mut mb: Mailbox = Mailbox::new(usize::MAX); - for i in 0..100 { - mb.push(i); - } - // len=100 < waterlevel=usize::MAX → always full drain - assert_eq!(mb.drain_count(), 100); -} - -#[test] -fn mailbox_with_struct_messages() { - let mut mb: Mailbox = Mailbox::new(10); - mb.push(TestMsg(1)); - mb.push(TestMsg(2)); - assert_eq!(mb.pop(), Some(TestMsg(1))); - assert_eq!(mb.pop(), Some(TestMsg(2))); - assert!(mb.is_empty()); -} - -// ── ActorPool tests ───────────────────────────────────────────────── - -#[test] -fn pool_new_is_empty() { - let pool = ActorPool::new(); - assert_eq!(pool.len(), 0); -} - -#[test] -fn pool_insert_increments_len() { - let mut pool = ActorPool::new(); - let addr = make_addr(1); - let (actor, _) = make_test_actor(addr); - pool.insert(addr, actor); - assert_eq!(pool.len(), 1); - - let addr2 = make_addr(2); - let (actor2, _) = make_test_actor(addr2); - pool.insert(addr2, actor2); - assert_eq!(pool.len(), 2); -} - -#[test] -fn pool_remove_returns_actor() { - let mut pool = ActorPool::new(); - let addr = make_addr(1); - let (actor, _) = make_test_actor(addr); - pool.insert(addr, actor); - assert!(pool.remove(&addr).is_some()); - assert_eq!(pool.len(), 0); -} - -#[test] -fn pool_remove_unknown_returns_none() { - let mut pool = ActorPool::new(); - let addr = make_addr(99); - assert!(pool.remove(&addr).is_none()); -} - -#[test] -fn pool_deliver_correct_type() { - let mut pool = ActorPool::new(); - let addr = make_addr(1); - let (actor, _) = make_test_actor(addr); - pool.insert(addr, actor); - - let msg: Box = Box::new(42u64); - assert!(pool.deliver(&addr, msg)); -} - -#[test] -fn pool_deliver_unknown_addr() { - let mut pool = ActorPool::new(); - let addr = make_addr(99); - let msg: Box = Box::new(42u64); - assert!(!pool.deliver(&addr, msg)); -} - -#[test] -fn pool_deliver_wrong_type() { - let mut pool = ActorPool::new(); - let addr = make_addr(1); - let (actor, _) = make_test_actor(addr); - pool.insert(addr, actor); - - // Actor expects u64, we send String — queued (type check deferred to tick) - let msg: Box = Box::new("wrong type".to_string()); - assert!(pool.deliver(&addr, msg)); -} - -#[test] -fn pool_tick_all_processes_messages() { - let mut pool = ActorPool::new(); - let addr = make_addr(1); - let (actor, counter) = make_test_actor(addr); - pool.insert(addr, actor); - - // Deliver 3 messages - pool.deliver(&addr, Box::new(1u64)); - pool.deliver(&addr, Box::new(2u64)); - pool.deliver(&addr, Box::new(3u64)); - - let stub = StubContextInner { waterlevel: 100 }; - let processed = pool.tick_all(&stub); - assert_eq!(processed, 3); - assert_eq!(counter.load(Ordering::Relaxed), 3); -} - -#[test] -fn pool_tick_all_empty_returns_zero() { - let mut pool = ActorPool::new(); - let addr = make_addr(1); - let (actor, _) = make_test_actor(addr); - pool.insert(addr, actor); - - // No messages delivered - let stub = StubContextInner { waterlevel: 100 }; - let processed = pool.tick_all(&stub); - assert_eq!(processed, 0); -} - -// ── Worker tests ──────────────────────────────────────────────────── - -#[test] -fn worker_tick_once_no_work() { - let transfer_rx = Receiver::::new(64); - let spawn_rx = Receiver::<(ActorAddress, Box)>::new(64); - - let transfer_tx = transfer_rx.new_sender(); - let spawn_tx = spawn_rx.new_sender(); - - let mut worker = Worker::new(WorkerId(0), transfer_rx, spawn_rx, Arc::new(WorkerStats::new())); - - let address_map = AddressMap::new(); - let placement = Placement::new(1); - let inbox_registry = InboxRegistry::new(); - let config = RuntimeConfig::default(); - - let tc = TickContext { - address_map: &address_map, - transfer_txs: &[transfer_tx], - spawn_txs: &[spawn_tx], - placement: &placement, - inbox_registry: &inbox_registry, - config: &config, - }; - - assert!(!worker.tick_once(&tc)); -} - -#[test] -fn worker_tick_once_drains_spawns() { - let transfer_rx = Receiver::::new(64); - let spawn_rx = Receiver::<(ActorAddress, Box)>::new(64); - - let transfer_tx = transfer_rx.new_sender(); - let spawn_tx = spawn_rx.new_sender(); - - let addr = make_addr(1); - let (actor, _) = make_test_actor(addr); - spawn_tx.try_send((addr, actor)).ok().unwrap(); - - let mut worker = Worker::new(WorkerId(0), transfer_rx, spawn_rx, Arc::new(WorkerStats::new())); - - let address_map = AddressMap::new(); - let placement = Placement::new(1); - let inbox_registry = InboxRegistry::new(); - let config = RuntimeConfig::default(); - - let tc = TickContext { - address_map: &address_map, - transfer_txs: &[transfer_tx], - spawn_txs: &[spawn_tx], - placement: &placement, - inbox_registry: &inbox_registry, - config: &config, - }; - - assert!(worker.tick_once(&tc)); - assert_eq!(worker.pool.len(), 1); -} - -#[test] -fn worker_tick_once_drains_transfers() { - let transfer_rx = Receiver::::new(64); - let spawn_rx = Receiver::<(ActorAddress, Box)>::new(64); - - let transfer_tx = transfer_rx.new_sender(); - let transfer_tx2 = transfer_rx.new_sender(); - let spawn_tx = spawn_rx.new_sender(); - let spawn_tx2 = spawn_rx.new_sender(); - - let addr = make_addr(1); - let (actor, counter) = make_test_actor(addr); - - let mut worker = Worker::new(WorkerId(0), transfer_rx, spawn_rx, Arc::new(WorkerStats::new())); - - // Spawn the actor first - spawn_tx.try_send((addr, actor)).ok().unwrap(); - - let address_map = AddressMap::new(); - let placement = Placement::new(1); - let inbox_registry = InboxRegistry::new(); - let config = RuntimeConfig::default(); - - let tc = TickContext { - address_map: &address_map, - transfer_txs: &[transfer_tx2], - spawn_txs: &[spawn_tx2], - placement: &placement, - inbox_registry: &inbox_registry, - config: &config, - }; - - worker.tick_once(&tc); // drain spawns - - // Now send a transfer envelope - let envelope = Envelope::new(addr, Box::new(42u64)); - transfer_tx.try_send(envelope).ok().unwrap(); - - // Tick again to drain transfers + tick actors - let did_work = worker.tick_once(&tc); - assert!(did_work); - assert_eq!(counter.load(Ordering::Relaxed), 1); -} - -#[test] -fn worker_tick_once_processes_messages() { - let transfer_rx = Receiver::::new(64); - let spawn_rx = Receiver::<(ActorAddress, Box)>::new(64); - - let transfer_tx = transfer_rx.new_sender(); - let transfer_tx2 = transfer_rx.new_sender(); - let spawn_tx = spawn_rx.new_sender(); - let spawn_tx2 = spawn_rx.new_sender(); - - let addr = make_addr(1); - let (actor, counter) = make_test_actor(addr); - - let mut worker = Worker::new(WorkerId(0), transfer_rx, spawn_rx, Arc::new(WorkerStats::new())); - - spawn_tx.try_send((addr, actor)).ok().unwrap(); - - let address_map = AddressMap::new(); - let placement = Placement::new(1); - let inbox_registry = InboxRegistry::new(); - let config = RuntimeConfig::default(); - - let tc = TickContext { - address_map: &address_map, - transfer_txs: &[transfer_tx2], - spawn_txs: &[spawn_tx2], - placement: &placement, - inbox_registry: &inbox_registry, - config: &config, - }; - - worker.tick_once(&tc); // spawn - - // Deliver multiple messages - for i in 0..5u64 { - transfer_tx - .try_send(Envelope::new(addr, Box::new(i))) + /// Enqueue a u64 message (drained on next tick, phase 2). + fn send(&self, id: u8, val: u64) { + self.feed_transfer + .try_send(Envelope::new(addr(id), Box::new(val))) .ok() .unwrap(); } - worker.tick_once(&tc); // transfer + tick - assert_eq!(counter.load(Ordering::Relaxed), 5); -} - -#[test] -fn worker_tick_once_multiple_spawns() { - let transfer_rx = Receiver::::new(64); - let spawn_rx = Receiver::<(ActorAddress, Box)>::new(64); - - let transfer_tx = transfer_rx.new_sender(); - let spawn_tx = spawn_rx.new_sender(); - - for i in 0..5u8 { - let addr = make_addr(i); - let (actor, _) = make_test_actor(addr); - spawn_tx.try_send((addr, actor)).ok().unwrap(); + /// Enqueue a wrong-typed message (String instead of u64). + fn send_bad(&self, id: u8) { + self.feed_transfer + .try_send(Envelope::new(addr(id), Box::new("bad".to_string()))) + .ok() + .unwrap(); } - let mut worker = Worker::new(WorkerId(0), transfer_rx, spawn_rx, Arc::new(WorkerStats::new())); + /// Remove an actor from the pool (immediate, no tick needed). + fn remove(&mut self, id: u8) { + self.worker.pool.remove(&addr(id)); + } - let address_map = AddressMap::new(); - let placement = Placement::new(1); - let inbox_registry = InboxRegistry::new(); - let config = RuntimeConfig::default(); + /// Run one tick of the worker loop. + fn tick(&mut self) { + let tc = TickContext { + address_map: &self.address_map, + transfer_txs: &self.tc_transfer_txs, + spawn_txs: &self.tc_spawn_txs, + placement: &self.placement, + inbox_registry: &self.inbox_registry, + config: &self.config, + }; + self.worker.tick_once(&tc); + } - let tc = TickContext { - address_map: &address_map, - transfer_txs: &[transfer_tx], - spawn_txs: &[spawn_tx], - placement: &placement, - inbox_registry: &inbox_registry, - config: &config, - }; + // ── Readouts ─────────────────────────────────────────────────── - assert!(worker.tick_once(&tc)); - assert_eq!(worker.pool.len(), 5); + fn handled(&self, id: u8) -> usize { + self.counters[&id].load(Ordering::Relaxed) + } + + fn pool_len(&self) -> usize { + self.worker.pool.len() + } + + fn depth(&self) -> usize { + self.stats.total_mailbox_depth.load(Ordering::Relaxed) + } + + fn processed(&self) -> u64 { + self.stats.messages_processed.load(Ordering::Relaxed) + } + + fn num_actors_stat(&self) -> usize { + self.stats.num_actors.load(Ordering::Relaxed) + } +} + +// ── Step-driven runner ───────────────────────────────────────────── + +enum Step { + Spawn(u8), + Send(u8, u64), + SendBad(u8), + Remove(u8), + Tick, + Expect { pool_len: usize, depth: usize, processed: u64 }, + ExpectHandled(u8, usize), +} + +fn run(steps: &[Step]) { + run_with(RuntimeConfig::default(), steps); +} + +fn run_with(config: RuntimeConfig, steps: &[Step]) { + let mut env = Env::with_config(config); + for (i, step) in steps.iter().enumerate() { + match step { + Step::Spawn(id) => env.spawn(*id), + Step::Send(id, val) => env.send(*id, *val), + Step::SendBad(id) => env.send_bad(*id), + Step::Remove(id) => env.remove(*id), + Step::Tick => env.tick(), + Step::Expect { pool_len, depth, processed } => { + assert_eq!(env.pool_len(), *pool_len, "step {i}: pool_len"); + assert_eq!(env.depth(), *depth, "step {i}: depth"); + assert_eq!(env.processed(), *processed, "step {i}: processed"); + } + Step::ExpectHandled(id, n) => { + assert_eq!(env.handled(*id), *n, "step {i}: handled({})", id); + } + } + } +} + +// ── Tests ────────────────────────────────────────────────────────── + +#[test] +fn spawn_send_process() { + run(&[ + Step::Spawn(1), + Step::Spawn(2), + Step::Tick, + Step::Expect { pool_len: 2, depth: 0, processed: 0 }, + + Step::Send(1, 10), + Step::Send(1, 20), + Step::Send(2, 30), + Step::Tick, + Step::Expect { pool_len: 2, depth: 0, processed: 3 }, + Step::ExpectHandled(1, 2), + Step::ExpectHandled(2, 1), + ]); } #[test] -fn worker_tick_once_wrong_type_no_panic() { - let transfer_rx = Receiver::::new(64); - let spawn_rx = Receiver::<(ActorAddress, Box)>::new(64); +fn remove_drops_future_messages() { + run(&[ + Step::Spawn(1), + Step::Spawn(2), + Step::Tick, - let transfer_tx = transfer_rx.new_sender(); - let transfer_tx2 = transfer_rx.new_sender(); - let spawn_tx = spawn_rx.new_sender(); - let spawn_tx2 = spawn_rx.new_sender(); + // Remove actor 1 directly from pool + Step::Remove(1), + Step::Expect { pool_len: 1, depth: 0, processed: 0 }, - let addr = make_addr(1); - let (actor, counter) = make_test_actor(addr); + // Messages to actor 1 are drained from the transfer queue + // but pool.deliver finds no slot — silently dropped + Step::Send(1, 42), + Step::Send(2, 99), + Step::Tick, - let mut worker = Worker::new(WorkerId(0), transfer_rx, spawn_rx, Arc::new(WorkerStats::new())); - - spawn_tx.try_send((addr, actor)).ok().unwrap(); - - let address_map = AddressMap::new(); - let placement = Placement::new(1); - let inbox_registry = InboxRegistry::new(); - let config = RuntimeConfig::default(); - - let tc = TickContext { - address_map: &address_map, - transfer_txs: &[transfer_tx2], - spawn_txs: &[spawn_tx2], - placement: &placement, - inbox_registry: &inbox_registry, - config: &config, - }; - - worker.tick_once(&tc); // spawn - - // Send wrong type (String instead of u64) — should not panic - let bad_envelope = Envelope::new(addr, Box::new("wrong".to_string())); - transfer_tx.try_send(bad_envelope).ok().unwrap(); - - // Send correct type after - let good_envelope = Envelope::new(addr, Box::new(99u64)); - transfer_tx.try_send(good_envelope).ok().unwrap(); - - worker.tick_once(&tc); // transfer + tick - // The correct message should still be processed - assert_eq!(counter.load(Ordering::Relaxed), 1); + Step::Expect { pool_len: 1, depth: 0, processed: 1 }, + Step::ExpectHandled(1, 0), + Step::ExpectHandled(2, 1), + ]); } #[test] -fn worker_run_stops_on_signal() { +fn wrong_type_silently_dropped() { + run(&[ + Step::Spawn(1), + Step::Tick, + + // Mix correct (u64) and incorrect (String) types + Step::Send(1, 1), + Step::SendBad(1), + Step::Send(1, 2), + Step::SendBad(1), + Step::SendBad(1), + Step::Send(1, 3), + Step::Tick, + + // All 6 popped from mailbox ("processed" by the pool), + // but only the 3 u64 messages were handled by the actor + Step::Expect { pool_len: 1, depth: 0, processed: 6 }, + Step::ExpectHandled(1, 3), + ]); +} + +#[test] +fn backpressure_drains_half() { + let config = RuntimeConfig { + mailbox_waterlevel: 4, + ..Default::default() + }; + + run_with(config, &[ + Step::Spawn(1), + Step::Tick, + + // Send 10 messages + Step::Send(1, 0), Step::Send(1, 1), Step::Send(1, 2), Step::Send(1, 3), + Step::Send(1, 4), Step::Send(1, 5), Step::Send(1, 6), Step::Send(1, 7), + Step::Send(1, 8), Step::Send(1, 9), + + // Tick 1: 10 in mailbox, drain_count(10, 4) = 5 + Step::Tick, + Step::Expect { pool_len: 1, depth: 5, processed: 5 }, + + // Tick 2: 5 remaining, drain_count(5, 4) = 2 + Step::Tick, + Step::Expect { pool_len: 1, depth: 3, processed: 7 }, + + // Tick 3: 3 remaining, drain_count(3, 4) = 3 (below waterlevel → all) + Step::Tick, + Step::Expect { pool_len: 1, depth: 0, processed: 10 }, + Step::ExpectHandled(1, 10), + ]); +} + +#[test] +fn spawn_and_send_same_tick() { + // Spawn is phase 1, transfer is phase 2, processing is phase 3. + // All three happen within a single tick_once call. + run(&[ + Step::Spawn(1), + Step::Send(1, 42), + Step::Tick, + Step::Expect { pool_len: 1, depth: 0, processed: 1 }, + Step::ExpectHandled(1, 1), + ]); +} + +#[test] +fn stats_track_pool_mutations() { + let mut env = Env::new(); + + // Before any tick, stats are zeroed + assert_eq!(env.num_actors_stat(), 0); + assert_eq!(env.depth(), 0); + assert_eq!(env.processed(), 0); + + // Spawn 3 + tick → stats reflect 3 actors + env.spawn(1); + env.spawn(2); + env.spawn(3); + env.tick(); + assert_eq!(env.num_actors_stat(), 3); + + // Send 5 to actor 1 + tick → processed increases + for i in 0..5 { + env.send(1, i); + } + env.tick(); + assert_eq!(env.processed(), 5); + assert_eq!(env.depth(), 0); + assert_eq!(env.num_actors_stat(), 3); + + // Remove actor 2 + tick → stats update + env.remove(2); + env.tick(); + assert_eq!(env.num_actors_stat(), 2); + assert_eq!(env.pool_len(), 2); +} + +/// A long mixed-action sequence: spawns, sends, removes, wrong types, +/// and backpressure — all in one run. +#[test] +fn interleaved_lifecycle() { + run(&[ + // ── Phase 1: build the pool ──────────────────────────────── + Step::Spawn(1), + Step::Spawn(2), + Step::Spawn(3), + Step::Tick, + Step::Expect { pool_len: 3, depth: 0, processed: 0 }, + + // ── Phase 2: normal message flow ─────────────────────────── + Step::Send(1, 100), + Step::Send(2, 200), + Step::Send(3, 300), + Step::Tick, + Step::ExpectHandled(1, 1), + Step::ExpectHandled(2, 1), + Step::ExpectHandled(3, 1), + Step::Expect { pool_len: 3, depth: 0, processed: 3 }, + + // ── Phase 3: remove actor 2, send to all 3 ──────────────── + Step::Remove(2), + Step::Send(1, 101), + Step::Send(2, 201), // actor 2 gone — dropped at deliver + Step::Send(3, 301), + Step::Tick, + Step::Expect { pool_len: 2, depth: 0, processed: 5 }, + Step::ExpectHandled(1, 2), + Step::ExpectHandled(2, 1), // unchanged since removal + Step::ExpectHandled(3, 2), + + // ── Phase 4: late spawn + immediate send ─────────────────── + Step::Spawn(4), + Step::Send(4, 400), + Step::Tick, + Step::Expect { pool_len: 3, depth: 0, processed: 6 }, + Step::ExpectHandled(4, 1), + + // ── Phase 5: bad types mixed with good ───────────────────── + Step::SendBad(1), + Step::SendBad(1), + Step::SendBad(1), + Step::Send(1, 999), + Step::Tick, + // 4 popped (3 bad + 1 good), only 1 handled by actor + Step::Expect { pool_len: 3, depth: 0, processed: 10 }, + Step::ExpectHandled(1, 3), // 2 from prior phases + 1 good + + // ── Phase 6: remove all, send to ghosts ──────────────────── + Step::Remove(1), + Step::Remove(3), + Step::Remove(4), + Step::Expect { pool_len: 0, depth: 0, processed: 10 }, + Step::Send(1, 0), + Step::Send(3, 0), + Step::Tick, + Step::Expect { pool_len: 0, depth: 0, processed: 10 }, + ]); +} + +#[test] +fn run_loop_stops_on_shutdown() { let transfer_rx = Receiver::::new(64); let spawn_rx = Receiver::<(ActorAddress, Box)>::new(64); - let transfer_tx = transfer_rx.new_sender(); let spawn_tx = spawn_rx.new_sender(); - let mut worker = Worker::new(WorkerId(0), transfer_rx, spawn_rx, Arc::new(WorkerStats::new())); - let is_running = AtomicBool::new(false); // start as false → should exit immediately + let stats = Arc::new(WorkerStats::new()); + let mut worker = Worker::new(WorkerId(0), transfer_rx, spawn_rx, stats); + + let is_running = AtomicBool::new(false); let backoff = BackoffPolicy::default(); - let address_map = AddressMap::new(); let placement = Placement::new(1); let inbox_registry = InboxRegistry::new(); @@ -602,11 +414,7 @@ fn worker_run_stops_on_signal() { config: &config, }; - // Run in a scoped thread to verify it actually terminates thread::scope(|s| { - s.spawn(|| { - worker.run(&tc, &is_running, &backoff); - }); + s.spawn(|| worker.run(&tc, &is_running, &backoff)); }); - // If we get here, the thread exited — test passes }