mod common; use common::*; use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; // ── Lifecycle Hook Helpers ──────────────────────────────────────────────── /// An actor that records lifecycle events to shared counters. struct LifecycleActor { started: Arc, stopped: Arc, handled: Arc, } impl ActorInterface for LifecycleActor { type Incoming = Ping; type Response = Pong; fn on_start(&mut self, _ctx: &Ctx) { self.started.fetch_add(1, Ordering::Relaxed); } fn on_stop(&mut self, _ctx: &Ctx) { self.stopped.fetch_add(1, Ordering::Relaxed); } fn handle(&mut self, ctx: &Ctx, msg: Ping) { self.handled.fetch_add(1, Ordering::Relaxed); let _ = ctx.send(msg.reply_to, Pong); } } /// An actor that stops itself after processing N messages. struct SelfStopActor { count: usize, stop_after: usize, stopped: Arc, } impl ActorInterface for SelfStopActor { type Incoming = Forward; type Response = Done; fn on_stop(&mut self, _ctx: &Ctx) { self.stopped.fetch_add(1, Ordering::Relaxed); } fn handle(&mut self, ctx: &Ctx, msg: Forward) { self.count += 1; let _ = ctx.send(msg.reply_to, Done(msg.value)); if self.count >= self.stop_after { ctx.stop_self(); } } } /// An actor that sends a farewell message in on_stop. struct FarewellActor { farewell_to: ActorAddress, } impl ActorInterface for FarewellActor { type Incoming = Ping; type Response = Pong; fn on_stop(&mut self, ctx: &Ctx) { let _ = ctx.send(self.farewell_to, Pong); } fn handle(&mut self, ctx: &Ctx, msg: Ping) { let _ = ctx.send(msg.reply_to, Pong); } } /// An actor whose on_start panics. struct PanicOnStartActor { handled: Arc, } impl ActorInterface for PanicOnStartActor { type Incoming = Ping; type Response = Pong; fn on_start(&mut self, _ctx: &Ctx) { panic!("on_start panic"); } fn handle(&mut self, _ctx: &Ctx, _msg: Ping) { self.handled.fetch_add(1, Ordering::Relaxed); } } /// Handles Forward messages, replies Done(value * 2), panics on the panic_at-th message. /// (Needed for on_start_called_again_after_restart and stop_vs_panic tests.) struct RestartTestActor { count: usize, panic_at: usize, } impl ActorInterface for RestartTestActor { type Incoming = Forward; type Response = Done; fn handle(&mut self, ctx: &Ctx, msg: Forward) { self.count += 1; if self.count >= self.panic_at { panic!("intentional panic at message {}", self.count); } let _ = ctx.send(msg.reply_to, Done(msg.value * 2)); } } // ── Lifecycle Hook Tests ────────────────────────────────────────────────── /// Given an actor with on_start implemented, /// when it is spawned and the runtime ticks, /// then on_start is called exactly once before the first message. #[test] fn on_start_called_before_first_message() { let started = Arc::new(AtomicUsize::new(0)); let stopped = Arc::new(AtomicUsize::new(0)); let handled = Arc::new(AtomicUsize::new(0)); let rt = std_runtime(RuntimeConfig::default()); let inbox = rt.new_inbox::().unwrap(); let addr = rt.spawn(LifecycleActor { started: started.clone(), stopped: stopped.clone(), handled: handled.clone(), }).unwrap(); // First tick — should call on_start rt.tick(); assert_eq!(started.load(Ordering::Relaxed), 1, "on_start called on first tick"); assert_eq!(handled.load(Ordering::Relaxed), 0, "no messages processed yet"); // Send messages and tick more let _ = rt.send_to(addr, Ping { reply_to: *inbox.addr() }); rt.tick(); assert_eq!(started.load(Ordering::Relaxed), 1, "on_start not called again"); assert_eq!(handled.load(Ordering::Relaxed), 1, "message processed after on_start"); } /// Given an actor with on_start, /// when multiple actors are spawned, /// then each gets its own on_start call exactly once. #[test] fn on_start_called_per_actor() { let started = Arc::new(AtomicUsize::new(0)); let stopped = Arc::new(AtomicUsize::new(0)); let handled = Arc::new(AtomicUsize::new(0)); let rt = std_runtime(RuntimeConfig::default()); for _ in 0..5 { let _ = rt.spawn(LifecycleActor { started: started.clone(), stopped: stopped.clone(), handled: handled.clone(), }).unwrap(); } rt.tick(); assert_eq!(started.load(Ordering::Relaxed), 5, "on_start called for each of 5 actors"); // Subsequent ticks don't repeat on_start rt.tick(); rt.tick(); assert_eq!(started.load(Ordering::Relaxed), 5, "on_start still 5 after more ticks"); } /// Given an actor whose on_start panics, /// when it is spawned and the runtime ticks, /// then it is immediately poisoned and never processes messages. #[test] fn on_start_panic_poisons_actor() { let handled = Arc::new(AtomicUsize::new(0)); let rt = std_runtime(RuntimeConfig::default()); let inbox = rt.new_inbox::().unwrap(); let addr = rt.spawn(PanicOnStartActor { handled: handled.clone() }).unwrap(); let _ = rt.send_to(addr, Ping { reply_to: *inbox.addr() }); for _ in 0..5 { rt.tick(); } assert_eq!(handled.load(Ordering::Relaxed), 0, "actor never processed messages"); let stats = rt.stats(); let total_panics: u64 = stats.workers.iter().map(|w| w.panics).sum(); assert_eq!(total_panics, 1, "on_start panic counted"); } // ── Graceful Stop Tests ─────────────────────────────────────────────────── /// Given an actor that calls ctx.stop_self() after 3 messages, /// when 5 messages are sent, /// then only 3 are processed, the actor is removed, and on_stop is called. #[test] fn actor_can_stop_self() { let stopped = Arc::new(AtomicUsize::new(0)); let rt = std_runtime(RuntimeConfig::default()); let inbox = rt.new_inbox::().unwrap(); let addr = rt.spawn(SelfStopActor { count: 0, stop_after: 3, stopped: stopped.clone(), }).unwrap(); for i in 0..5 { let _ = rt.send_to(addr, Forward { value: i, reply_to: *inbox.addr() }); } for _ in 0..10 { rt.tick(); } // Only 3 messages should be processed (stop_self after 3rd) let mut replies = Vec::new(); while let Some(Done(v)) = inbox.try_recv() { replies.push(v); } assert_eq!(replies.len(), 3, "only 3 messages processed before stop"); assert!(replies.contains(&0)); assert!(replies.contains(&1)); assert!(replies.contains(&2)); assert_eq!(stopped.load(Ordering::Relaxed), 1, "on_stop called exactly once"); // Actor should be removed from address map let stats = rt.stats(); assert_eq!(stats.actors.len(), 0, "stopped actor removed from address map"); } /// Given a running actor, /// when runtime.stop_actor(addr) is called, /// then the actor stops, on_stop is called, and it's removed from the pool. #[test] fn runtime_can_stop_actor() { let started = Arc::new(AtomicUsize::new(0)); let stopped = Arc::new(AtomicUsize::new(0)); let handled = Arc::new(AtomicUsize::new(0)); let rt = std_runtime(RuntimeConfig::default()); let inbox = rt.new_inbox::().unwrap(); let addr = rt.spawn(LifecycleActor { started: started.clone(), stopped: stopped.clone(), handled: handled.clone(), }).unwrap(); // Let it start and process a message let _ = rt.send_to(addr, Ping { reply_to: *inbox.addr() }); for _ in 0..3 { rt.tick(); } assert_eq!(handled.load(Ordering::Relaxed), 1); // Stop it externally rt.stop_actor(addr).unwrap(); for _ in 0..3 { rt.tick(); } assert_eq!(stopped.load(Ordering::Relaxed), 1, "on_stop called"); // Actor should be gone let stats = rt.stats(); assert_eq!(stats.actors.len(), 0, "stopped actor removed"); assert_eq!(stats.workers[0].num_actors, 0); } /// Given a stopped actor, /// when new messages are sent to it, /// then sends return Err (address not found). #[test] fn send_to_stopped_actor_returns_error() { let stopped = Arc::new(AtomicUsize::new(0)); let rt = std_runtime(RuntimeConfig::default()); let inbox = rt.new_inbox::().unwrap(); let addr = rt.spawn(SelfStopActor { count: 0, stop_after: 1, stopped: stopped.clone(), }).unwrap(); // One message triggers stop let _ = rt.send_to(addr, Forward { value: 1, reply_to: *inbox.addr() }); for _ in 0..10 { rt.tick(); } // Actor is now removed — send should fail let result = rt.send_to(addr, Forward { value: 2, reply_to: *inbox.addr() }); assert!(result.is_err(), "send to stopped actor should return Err"); } /// Given a gracefully stopped actor and a panicked actor, /// then stats.stops and stats.panics track them separately. #[test] fn stop_vs_panic_tracked_separately_in_stats() { let stopped = Arc::new(AtomicUsize::new(0)); let rt = std_runtime(RuntimeConfig::default()); let inbox = rt.new_inbox::().unwrap(); // Actor that stops itself after 1 message let _stop_addr = rt.spawn(SelfStopActor { count: 0, stop_after: 1, stopped: stopped.clone(), }).unwrap(); // Actor that panics on first message let panic_addr = rt.spawn(RestartTestActor { count: 0, panic_at: 1 }).unwrap(); let _ = rt.send_to(_stop_addr, Forward { value: 1, reply_to: *inbox.addr() }); let _ = rt.send_to(panic_addr, Forward { value: 1, reply_to: *inbox.addr() }); for _ in 0..10 { rt.tick(); } let stats = rt.stats(); let total_stops: u64 = stats.workers.iter().map(|w| w.stops).sum(); let total_panics: u64 = stats.workers.iter().map(|w| w.panics).sum(); assert_eq!(total_stops, 1, "one graceful stop"); assert_eq!(total_panics, 1, "one panic"); } /// Given an actor with on_stop that sends a farewell message, /// when the actor is stopped, /// then the farewell message is delivered. #[test] fn on_stop_can_send_messages() { let rt = std_runtime(RuntimeConfig::default()); let inbox = rt.new_inbox::().unwrap(); let addr = rt.spawn(FarewellActor { farewell_to: *inbox.addr(), }).unwrap(); // Let it start rt.tick(); // Stop it rt.stop_actor(addr).unwrap(); for _ in 0..5 { rt.tick(); } // Should receive farewell Pong from on_stop let farewell = inbox.try_recv(); assert_eq!(farewell, Some(Pong), "farewell message delivered from on_stop"); } /// Given a supervisor with a child that panics and is restarted, /// when the child is respawned by the supervisor, /// then on_start is called again on the fresh instance. #[test] fn on_start_called_again_after_restart() { let started = Arc::new(AtomicUsize::new(0)); let rt = std_runtime(RuntimeConfig::default()); let started_c = started.clone(); let _sup_addr = rt.spawn(Supervisor::new( SupervisorStrategy::OneForOne, 5, vec![ChildSpec::new("child", RestartPolicy::Permanent, move |ctx| { ctx.spawn(LifecycleActor { started: started_c.clone(), stopped: Arc::new(AtomicUsize::new(0)), handled: Arc::new(AtomicUsize::new(0)), }) })], )).unwrap(); // First tick: supervisor starts, spawns child, on_start called for _ in 0..3 { rt.tick(); } assert_eq!(started.load(Ordering::Relaxed), 1, "on_start called once"); } /// Given an actor stopped via stop_actor() with messages already queued, /// when the stop signal arrives after the queued messages (PoisonPill semantics), /// then messages ahead of the signal are processed, then the actor stops. #[test] fn external_stop_is_queued_after_pending_messages() { let stopped = Arc::new(AtomicUsize::new(0)); let handled = Arc::new(AtomicUsize::new(0)); let rt = std_runtime(RuntimeConfig::default()); let started = Arc::new(AtomicUsize::new(0)); let addr = rt.spawn(LifecycleActor { started: started.clone(), stopped: stopped.clone(), handled: handled.clone(), }).unwrap(); let inbox = rt.new_inbox::().unwrap(); // Queue 10 messages, then stop — StopSignal is queued AFTER the 10 for _ in 0..10 { let _ = rt.send_to(addr, Ping { reply_to: *inbox.addr() }); } rt.stop_actor(addr).unwrap(); for _ in 0..10 { rt.tick(); } // All 10 messages processed (they were ahead of StopSignal in the queue) let total_handled = handled.load(Ordering::Relaxed); assert_eq!(total_handled, 10, "all messages processed before stop signal"); assert_eq!(stopped.load(Ordering::Relaxed), 1, "on_stop called"); // Actor is removed let stats = rt.stats(); assert_eq!(stats.actors.len(), 0, "stopped actor removed"); } /// Given a running actor with no pending messages, /// when stop_actor() is called and then new messages are sent, /// then the stop takes priority and new messages are not processed. #[test] fn external_stop_before_new_messages_prevents_processing() { let stopped = Arc::new(AtomicUsize::new(0)); let handled = Arc::new(AtomicUsize::new(0)); let started = Arc::new(AtomicUsize::new(0)); let rt = std_runtime(RuntimeConfig::default()); let inbox = rt.new_inbox::().unwrap(); let addr = rt.spawn(LifecycleActor { started: started.clone(), stopped: stopped.clone(), handled: handled.clone(), }).unwrap(); // Let actor start rt.tick(); // Stop first, then send messages rt.stop_actor(addr).unwrap(); for _ in 0..5 { let _ = rt.send_to(addr, Ping { reply_to: *inbox.addr() }); } for _ in 0..10 { rt.tick(); } // Stop signal was first in queue, so no messages processed assert_eq!(handled.load(Ordering::Relaxed), 0, "no messages processed after stop"); assert_eq!(stopped.load(Ordering::Relaxed), 1, "on_stop called"); } /// Given stop_actor is called on a nonexistent address, /// then it returns Err. #[test] fn stop_nonexistent_actor_returns_error() { let rt = std_runtime(RuntimeConfig::default()); let fake_addr = swactor::actor::ActorAddress::default(); let result = rt.stop_actor(fake_addr); assert!(result.is_err(), "stop_actor on nonexistent address should return Err"); }