swactor/tests/runtime_hooks.rs

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<AtomicUsize>,
    stopped: Arc<AtomicUsize>,
    handled: Arc<AtomicUsize>,
}

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<AtomicUsize>,
}

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<AtomicUsize>,
}

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::<Pong>().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::<Pong>().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::<Done>().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::<Pong>().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::<Done>().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::<Done>().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::<Pong>().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::<Pong>().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::<Pong>().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");
}