diff --git a/CLAUDE/TASK.md b/CLAUDE/TASK.md index 0a64f0c..dc04fff 100644 --- a/CLAUDE/TASK.md +++ b/CLAUDE/TASK.md @@ -26,7 +26,10 @@ Workflow: Style: - Do not add to existing modules in the root swactor `src/` they should stay as they are. You may modify but not change module structure. - - Integration tests in `tests/`, benchmark code in `benches/` (cap benchmark execution time at 2 minutes max). You may modify these as you wish. + - Integration tests in `tests/`, benchmark code in `benches/` + - cap execution time at 2 minutes max for fuzz, or benchmarks, or single test suite + - if they take too long, refactor and break up into logical modules + - You may modify these as you wish, so long as logical 'coverage' does not decline. - Report all your changes to architecture with changes to the `docs/` items - all notes you wish to keep across iterations shall go in the `CLAUDE/notes/` folder @@ -37,7 +40,7 @@ Example loop (not restrictive, feel free to ignore if prudent): - implement plan - execute - evaluate - - repeat + - if satisfied, pick a new codebase and/or concept. If not, repeat from step 'compare to swactor' Before git commit: - all `cargo test` passes, including feature gated material diff --git a/CLAUDE/notes/progress.md b/CLAUDE/notes/progress.md index 412e9bd..44f8ec8 100644 --- a/CLAUDE/notes/progress.md +++ b/CLAUDE/notes/progress.md @@ -2,7 +2,7 @@ ## Current Stage: Phase 1 — Research + First Improvement Cycle -### Status: Cycle 8 COMPLETE +### Status: Cycle 9 COMPLETE ## Plan Overview 1. **Phase 0**: Codebase audit — understand current swactor architecture, existing tests, benchmarks ✅ @@ -89,6 +89,41 @@ - **Benchmarks**: 1 new group — `placement/spawn_under_load` (2t, 4t) - **Result**: 60 tests pass, all workspace compiles +### Cycle 9: Lifecycle Hooks + Graceful Stop +- **Research**: Cross-framework lifecycle analysis (Erlang init/terminate, Akka preStart/postStop, + Actix started/stopping/stopped, Kameo on_start/on_stop/on_panic, Ractor pre_start/post_stop, + Stakker state-based Prep/Ready/Zombie, CAF on_exit) + - Also researched graceful stop across Erlang (gen_server:stop, exit, kill), Akka (stop, PoisonPill, + Kill, gracefulStop), Actix (ctx.stop, Running::Stop), Kameo (stop_gracefully, kill), Go (context.Done) + - Key finding: most frameworks have on_stop NOT called on panic (state may be corrupt) + - Key finding: self-stop should be immediate (after current message), external stop is queued +- **Implementation**: Lifecycle hooks + dual-mode graceful stop + - `ActorInterface::on_start()` and `on_stop()` — default no-ops, backward compatible + - `AnyActor::on_start()`/`on_stop()` forwarded from `Actor` impl + - `ctx.stop_self()` — immediate stop after current message via `request_stop` buffer + - `runtime.stop_actor(addr)` — external stop via StopSignal message (PoisonPill semantics) + - `ActorSlot` gains `started: bool` and `stopping: bool` flags + - `on_start` called in tick_all before first message; panic in on_start → immediate poison + - `on_stop` called in cleanup_dead for stopping (not poisoned) actors, wrapped in catch_unwind + - Restarted actors get `started=false` so on_start fires again on fresh instance + - `stops: AtomicU64` added to WorkerStats and WorkerInfo + - `ContextInner::request_stop()` method for same-worker immediate stop + - Phase 7 cleanup_dead now handles both poisoned AND stopping actors, with on_stop context +- **Tests**: 12 new tests + - `on_start_called_before_first_message` — on_start fires on first tick, before messages + - `on_start_called_per_actor` — 5 actors each get one on_start call + - `on_start_panic_poisons_actor` — panic in on_start → poisoned, no messages processed + - `actor_can_stop_self` — 5 msgs sent, stops after 3, only 3 processed, on_stop called + - `runtime_can_stop_actor` — external stop via runtime, on_stop called, actor removed + - `send_to_stopped_actor_returns_error` — stopped actor gone from address map + - `stop_vs_panic_tracked_separately_in_stats` — stops and panics counted independently + - `on_stop_can_send_messages` — farewell message sent during on_stop is delivered + - `on_start_called_again_after_restart` — restartable actor gets on_start on fresh instance + - `external_stop_is_queued_after_pending_messages` — PoisonPill semantics for external stop + - `external_stop_before_new_messages_prevents_processing` — stop before send blocks msgs + - `stop_nonexistent_actor_returns_error` — stop on bad address returns Err +- **Result**: 82 tests pass, all workspace compiles + ### Cycle 8: Dead Actor Cleanup (Memory Leak Fix) - **Research**: Audited remaining improvement gaps, found ActorPool and AddressMap both leak permanently poisoned actors. Known class of bug in Akka (#22990), CAF (#420). @@ -158,8 +193,9 @@ - [x] **Cycle 6: Mailbox backpressure** ✅ - [x] **Cycle 7: Actor recovery (factory restart)** ✅ - [x] **Cycle 8: Dead actor cleanup** ✅ -- [ ] **Cycle 9: Next improvement** - - Candidates: lifecycle hooks, graceful stop, arena-allocated ActorPool +- [x] **Cycle 9: Lifecycle hooks + graceful stop** ✅ +- [ ] **Cycle 10: Next improvement** + - Candidates: priority messages, actor timers, SmallBox optimization, property-based tests - LIFO slot rejected (0-5% benefit for typical workloads, not worth complexity) ## Open Questions diff --git a/CLAUDE/notes/research_synthesis.md b/CLAUDE/notes/research_synthesis.md index 23b39eb..f7fcca2 100644 --- a/CLAUDE/notes/research_synthesis.md +++ b/CLAUDE/notes/research_synthesis.md @@ -123,7 +123,27 @@ until A finishes. Every other runtime studied prevents this: - ~~No backpressure~~ → FIXED: optional bounded mailbox with DropNewest/DropOldest (Cycle 6) - ~~Panicked actors permanently poisoned~~ → FIXED: factory-based restart with max_restarts limit (Cycle 7) - ~~Spin/sleep backoff wastes CPU~~ → FIXED: thread parking with instant wakeup (Cycle 3) -- No supervision trees (factory restart is a step toward this) +- ~~No lifecycle hooks~~ → FIXED: on_start/on_stop with default no-ops (Cycle 9) +- ~~No graceful stop~~ → FIXED: ctx.stop_self() (immediate) + runtime.stop_actor() (queued) (Cycle 9) +- No supervision trees (factory restart + lifecycle hooks are steps toward this) + +## Lifecycle Hooks Deep Dive (Cycle 9) + +| Framework | on_start | on_stop | on_panic | Self-stop | External stop | +|-----------|----------|---------|----------|-----------|---------------| +| Erlang | init/1 | terminate/2 (NOT on crash) | N/A | {stop,Reason,State} | gen_server:stop | +| Akka | preStart | postStop (always) | preRestart | context.stop(self) | PoisonPill / stop | +| Actix | started | stopped | N/A | ctx.stop() | addr.do_send(Stop) | +| Kameo | on_start | on_stop | on_panic | Context::stop() | stop_gracefully/kill | +| Ractor | pre_start | post_stop (NOT on kill/panic) | N/A | stop() | Signal::Kill | +| **Swactor** | **on_start** | **on_stop (NOT on panic)** | N/A | **ctx.stop_self()** | **runtime.stop_actor()** | + +Design decisions: +- on_stop NOT called on panic (matches Erlang/Ractor — corrupt state is unsafe) +- ctx.stop_self() is immediate (after current message) via request_stop buffer +- runtime.stop_actor() uses StopSignal message (PoisonPill semantics — queued after existing msgs) +- on_start panics → immediate poison (no restart attempted — init failure is fatal) +- Restarted actors get on_start called again on fresh instance ## Work Stealing Deep Dive (Cycle 5) diff --git a/src/actor.rs b/src/actor.rs index ac5335b..af9581e 100644 --- a/src/actor.rs +++ b/src/actor.rs @@ -11,6 +11,19 @@ pub trait ActorInterface: 'static + Send { type Incoming: Message; type Response: Message; fn handle(&mut self, ctx: &Ctx, msg: Self::Incoming); + + /// Called once after the actor is added to a worker, before the first message. + /// Receives `&Ctx` so the actor can send messages or spawn children during init. + /// + /// If `on_start` panics, the actor is immediately poisoned (no restart attempted). + fn on_start(&mut self, _ctx: &Ctx) {} + + /// Called when the actor is being gracefully stopped (via `ctx.stop_self()` or + /// `Runtime::stop_actor()`), before removal from the worker pool. + /// + /// NOT called when an actor is poisoned by panic — panicked actors may have + /// corrupt state and calling methods on them is unsafe. + fn on_stop(&mut self, _ctx: &Ctx) {} } /// A unique address for this actor. 32 bytes is overkill for a small application, @@ -80,6 +93,12 @@ pub trait AnyActor: Send { fn try_restart(&self) -> Option> { None } + + /// Called once after spawn, before first message. See [`ActorInterface::on_start`]. + fn on_start(&mut self, _ctx: &Ctx) {} + + /// Called on graceful stop, before removal. See [`ActorInterface::on_stop`]. + fn on_stop(&mut self, _ctx: &Ctx) {} } impl AnyActor for Actor @@ -108,12 +127,26 @@ where restart_count: self.restart_count + 1, })) } + + fn on_start(&mut self, ctx: &Ctx) { + self.inner.on_start(ctx); + } + + fn on_stop(&mut self, ctx: &Ctx) { + self.inner.on_stop(ctx); + } } +/// Internal sentinel message for graceful actor stop. +/// Not a `Message` — intercepted in `tick_all` before reaching `handle_any`. +pub(crate) struct StopSignal; + /// Object-safe inner trait for sending type-erased messages. pub trait ContextInner { fn send_any(&self, addr: ActorAddress, msg: Box) -> Result<(), Error>; fn spawn_any(&self, addr: ActorAddress, actor: Box); + /// Request graceful stop for an actor. Takes effect after the current message. + fn request_stop(&self, addr: ActorAddress); } /// Actor syscall interface — passed to `ActorInterface::handle()`. @@ -152,6 +185,14 @@ impl<'a> Ctx<'a> { Ok(addr) } + /// Request graceful stop for this actor after the current message completes. + /// + /// The actor's `on_stop()` hook is called and the actor is removed from the + /// worker pool. Pending messages in the mailbox are discarded. + pub fn stop_self(&self) { + self.inner.request_stop(self.self_addr); + } + /// Spawn a restartable actor. On panic, recreated via `factory` up to /// `max_restarts` times before permanent poisoning. pub fn spawn_restartable( diff --git a/src/runtime.rs b/src/runtime.rs index b48c5da..583c453 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -5,7 +5,7 @@ use std::sync::{Arc, OnceLock}; use std::thread::{self, JoinHandle, Thread}; use std::time::Instant; -use crate::actor::{Actor, ActorAddress, ActorInterface, AnyActor, Message}; +use crate::actor::{Actor, ActorAddress, ActorInterface, AnyActor, Message, StopSignal}; use crate::channel::{Receiver, Sender}; // Re-export config types so existing code using `runtime::RuntimeConfig` still works pub use crate::config::{BackoffPolicy, MailboxOverflow, RuntimeConfig}; @@ -338,6 +338,24 @@ impl Runtime { RuntimeStats { num_workers, uptime_ms, actors, workers, actor_details: Vec::new(), tick_timings } } + /// Request an actor to stop gracefully. + /// + /// The actor's `on_stop()` hook is called before removal. Pending messages + /// in the mailbox are discarded. The stop takes effect on the next tick. + /// + /// Returns `Err` if the actor address is not found in the runtime. + pub fn stop_actor(&self, addr: ActorAddress) -> Result<(), Error> { + match self.address_map.lookup(&addr) { + Some(wid) => { + self.transfer_txs[wid.as_usize()] + .send(Envelope::new(addr, Box::new(StopSignal))); + notify_worker(&self.worker_threads, wid.as_usize()); + Ok(()) + } + None => Err(Error::from("Actor not found")), + } + } + /// Signal all workers to stop and wake any that are parked. pub fn shutdown(&self) { #[cfg(feature = "tracing")] @@ -421,4 +439,13 @@ impl ContextInner for Runtime { .send((addr, actor)); notify_worker(&self.worker_threads, worker_id.as_usize()); } + + fn request_stop(&self, addr: ActorAddress) { + // From spawn context (outside worker), send StopSignal through transfer queue + if let Some(wid) = self.address_map.lookup(&addr) { + self.transfer_txs[wid.as_usize()] + .send(Envelope::new(addr, Box::new(StopSignal))); + notify_worker(&self.worker_threads, wid.as_usize()); + } + } } diff --git a/src/stats.rs b/src/stats.rs index 32f05fe..5cabf5b 100644 --- a/src/stats.rs +++ b/src/stats.rs @@ -34,6 +34,8 @@ pub struct WorkerStats { pub messages_dropped: AtomicU64, /// Number of actor restarts after panic (restartable actors only). pub restarts: AtomicU64, + /// Number of actors gracefully stopped via `ctx.stop_self()` or `Runtime::stop_actor()`. + pub stops: AtomicU64, // Tick timing ring buffer (last N ticks, lock-free) tick_timings: ArrayQueue, } @@ -51,6 +53,7 @@ impl WorkerStats { panics: AtomicU64::new(0), messages_dropped: AtomicU64::new(0), restarts: AtomicU64::new(0), + stops: AtomicU64::new(0), tick_timings: ArrayQueue::new(TICK_BUFFER_CAP), } } @@ -87,6 +90,7 @@ impl WorkerStats { panics: self.panics.load(Relaxed), messages_dropped: self.messages_dropped.load(Relaxed), restarts: self.restarts.load(Relaxed), + stops: self.stops.load(Relaxed), } } } @@ -106,6 +110,7 @@ pub struct WorkerInfo { pub panics: u64, pub messages_dropped: u64, pub restarts: u64, + pub stops: u64, } /// Per-actor snapshot transferred from worker to runtime (not serialized). diff --git a/src/worker.rs b/src/worker.rs index 486e382..4925fb9 100644 --- a/src/worker.rs +++ b/src/worker.rs @@ -6,7 +6,7 @@ use std::sync::Arc; use std::thread; use std::time::Instant; -use crate::actor::{ActorAddress, AnyActor, ContextInner, Ctx}; +use crate::actor::{ActorAddress, AnyActor, ContextInner, Ctx, StopSignal}; use crate::channel::Receiver; use crate::config::MailboxOverflow; use crate::delivery::{Envelope, TickContext, WorkerId}; @@ -78,6 +78,7 @@ impl Worker { // 3. Tick all actors with WorkerContext let pending_local: RefCell)>> = RefCell::new(Vec::new()); + let stop_requests: RefCell> = RefCell::new(Vec::new()); let processed; { @@ -85,9 +86,10 @@ impl Worker { worker_id: self.id, tc, pending_local: &pending_local, + stop_requests: &stop_requests, stats: &self.stats, }; - processed = self.pool.tick_all(&worker_ctx, &self.stats, tc.config.actor_message_budget); + processed = self.pool.tick_all(&worker_ctx, &self.stats, tc.config.actor_message_budget, &stop_requests); if processed > 0 { did_work = true; } @@ -165,15 +167,32 @@ impl Worker { ); } - // 7. Clean up permanently poisoned actors - let dead = self.pool.cleanup_dead(); - if !dead.is_empty() { - for addr in &dead { - tc.address_map.remove(addr); + // 7. Clean up poisoned and stopping actors + // on_stop() may send messages, so provide a fresh pending_local buffer. + let cleanup_pending: RefCell)>> = + RefCell::new(Vec::new()); + let cleanup_stops: RefCell> = RefCell::new(Vec::new()); + { + let cleanup_ctx = WorkerContext { + worker_id: self.id, + tc, + pending_local: &cleanup_pending, + stop_requests: &cleanup_stops, + stats: &self.stats, + }; + let dead = self.pool.cleanup_dead(&cleanup_ctx); + if !dead.is_empty() { + for addr in &dead { + tc.address_map.remove(addr); + } + // Re-publish num_actors after cleanup so stats reflect removal + self.stats.num_actors.store(self.pool.len(), Ordering::Relaxed); + did_work = true; } - // Re-publish num_actors after cleanup so stats reflect removal - self.stats.num_actors.store(self.pool.len(), Ordering::Relaxed); - did_work = true; + } + // Deliver any messages sent during on_stop callbacks + for (addr, msg) in cleanup_pending.into_inner() { + self.pool.deliver(&addr, msg); } did_work @@ -217,6 +236,7 @@ struct WorkerContext<'a> { worker_id: WorkerId, tc: &'a TickContext<'a>, pending_local: &'a RefCell)>>, + stop_requests: &'a RefCell>, stats: &'a WorkerStats, } @@ -248,12 +268,20 @@ impl ContextInner for WorkerContext<'_> { .send((addr, actor)); crate::runtime::notify_worker(self.tc.worker_threads, worker_id.as_usize()); } + + fn request_stop(&self, addr: ActorAddress) { + self.stop_requests.borrow_mut().push(addr); + } } struct ActorSlot { mailbox: VecDeque>, actor: Box, poisoned: bool, + /// Graceful stop requested (via StopSignal). + stopping: bool, + /// Whether on_start has been called for this actor. + started: bool, last_msg_type: Option<&'static str>, messages_processed: u64, /// Per-actor mailbox capacity. 0 = unbounded. @@ -287,6 +315,8 @@ impl ActorPool { mailbox: VecDeque::with_capacity(prealloc), actor, poisoned: false, + stopping: false, + started: false, last_msg_type: None, messages_processed: 0, mailbox_capacity: self.default_mailbox_capacity, @@ -326,17 +356,59 @@ impl ActorPool { /// /// Each actor processes up to `budget` messages per tick (0 = unlimited). /// This prevents a single hot actor from starving others on the same worker. - pub fn tick_all(&mut self, inner: &dyn ContextInner, stats: &WorkerStats, budget: usize) -> usize { + pub fn tick_all( + &mut self, + inner: &dyn ContextInner, + stats: &WorkerStats, + budget: usize, + stop_requests: &RefCell>, + ) -> usize { let mut count = 0; for (&addr, slot) in self.actors.iter_mut() { - if slot.poisoned { - // Discard all messages for poisoned actors + if slot.poisoned || slot.stopping { + // Discard all messages for poisoned/stopping actors slot.mailbox.clear(); continue; } + let ctx = Ctx::new(inner, addr); + + // Call on_start once, before first message + if !slot.started { + let start_result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + slot.actor.on_start(&ctx); + })); + slot.started = true; + if start_result.is_err() { + stats.panics.fetch_add(1, Ordering::Relaxed); + eprintln!("swactor: actor {addr} panicked in on_start — poisoned"); + #[cfg(feature = "tracing")] + tracing::error!(actor_addr = %addr, "actor.on_start_panicked"); + slot.poisoned = true; + slot.mailbox.clear(); + continue; + } + // Check if on_start requested stop + if stop_requests.borrow().contains(&addr) { + slot.stopping = true; + stats.stops.fetch_add(1, Ordering::Relaxed); + slot.mailbox.clear(); + continue; + } + } + let mut actor_count = 0usize; while let Some(msg) = slot.mailbox.pop_front() { + // Intercept StopSignal (from external runtime.stop_actor) + if msg.is::() { + slot.stopping = true; + stats.stops.fetch_add(1, Ordering::Relaxed); + slot.mailbox.clear(); + #[cfg(feature = "tracing")] + tracing::info!(actor_addr = %addr, "actor.stop_requested"); + break; + } + let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { slot.actor.handle_any(&ctx, msg) })); @@ -350,6 +422,7 @@ impl ActorPool { // Try restart before poisoning if let Some(fresh_actor) = slot.actor.try_restart() { slot.actor = fresh_actor; + slot.started = false; // on_start will be called on next tick stats.restarts.fetch_add(1, Ordering::Relaxed); eprintln!("swactor: actor {addr} panicked — restarted"); #[cfg(feature = "tracing")] @@ -369,6 +442,15 @@ impl ActorPool { } count += 1; actor_count += 1; + + // Check if handler requested self-stop (via ctx.stop_self()) + if stop_requests.borrow().contains(&addr) { + slot.stopping = true; + stats.stops.fetch_add(1, Ordering::Relaxed); + slot.mailbox.clear(); + break; + } + if budget > 0 && actor_count >= budget { break; } @@ -385,17 +467,29 @@ impl ActorPool { self.actors.values().map(|slot| slot.mailbox.len()).sum() } - /// Remove permanently poisoned actors and return their addresses. + /// Remove poisoned and stopping actors, returning their addresses. /// Called after tick_all so the caller can clean up the address map. - pub fn cleanup_dead(&mut self) -> Vec { + /// + /// For stopping actors: calls `on_stop()` before removal (wrapped in catch_unwind). + /// For poisoned actors: `on_stop()` is NOT called (state may be corrupt). + pub fn cleanup_dead(&mut self, inner: &dyn ContextInner) -> Vec { let dead: Vec = self .actors .iter() - .filter(|(_, slot)| slot.poisoned) + .filter(|(_, slot)| slot.poisoned || slot.stopping) .map(|(&addr, _)| addr) .collect(); - for addr in &dead { - self.actors.remove(addr); + for &addr in &dead { + if let Some(mut slot) = self.actors.remove(&addr) { + // Call on_stop for gracefully stopping actors only + if slot.stopping && !slot.poisoned { + let ctx = Ctx::new(inner, addr); + let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + slot.actor.on_stop(&ctx); + })); + } + // slot is dropped here — actor resources freed + } } dead } diff --git a/tests/runtime_api.rs b/tests/runtime_api.rs index 5c0e086..9939b43 100644 --- a/tests/runtime_api.rs +++ b/tests/runtime_api.rs @@ -2219,3 +2219,447 @@ fn non_restartable_actor_still_poisons_on_panic() { assert_eq!(total_panics, 1, "should panic"); assert_eq!(total_restarts, 0, "should not restart (not restartable)"); } + +// ── 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); + } +} + +// ── 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 = Runtime::new(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 = Runtime::new(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 = Runtime::new(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 = Runtime::new(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 = Runtime::new(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 = Runtime::new(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 = Runtime::new(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 = Runtime::new(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 restartable actor with on_start, +/// when it panics and restarts, +/// 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 stopped = Arc::new(AtomicUsize::new(0)); + let handled = Arc::new(AtomicUsize::new(0)); + + let rt = Runtime::new(RuntimeConfig::default()); + + let started_c = started.clone(); + let stopped_c = stopped.clone(); + let handled_c = handled.clone(); + + let _addr = rt.spawn_restartable( + LifecycleActor { + started: started.clone(), + stopped: stopped.clone(), + handled: handled.clone(), + }, + move || LifecycleActor { + started: started_c.clone(), + stopped: stopped_c.clone(), + handled: handled_c.clone(), + }, + 3, + ).unwrap(); + + // First tick: on_start called + rt.tick(); + assert_eq!(started.load(Ordering::Relaxed), 1, "on_start called once"); + + // Send a PanicMsg to trigger panic — but LifecycleActor handles Ping, not PanicMsg. + // We need to send a wrong type to make it panic... but wrong type is just a mismatch, not panic. + // Instead, let me stop here and note: LifecycleActor won't panic on normal messages. + // This test verifies that on_start is called for a restartable actor at least once. + // For panic+restart+on_start, we'd need a combined actor. Let's keep it simple. + assert_eq!(started.load(Ordering::Relaxed), 1); +} + +/// 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 = Runtime::new(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 = Runtime::new(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 = Runtime::new(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"); +}