swactor/tests/runtime_stress.rs

//! Runtime Stress Tests — multi-threaded execution, parking, placement, and scale.
//!
//! Covers: single vs multi-threaded processing, high-volume MT delivery,
//! panic isolation under load, worker parking/shutdown, sustained throughput,
//! and load-aware actor placement.

mod common;
use common::*;

use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};

// ── Helpers ──────────────────────────────────────────────────────────────────

/// Poll `inbox` until a message arrives or `timeout` elapses.
fn poll_inbox<M: swactor::actor::Message>(
    inbox: &Inbox<M>,
    timeout: Duration,
) -> Option<M> {
    let deadline = Instant::now() + timeout;
    loop {
        if let Some(msg) = inbox.try_recv() {
            return Some(msg);
        }
        if Instant::now() > deadline {
            return None;
        }
        std::thread::sleep(Duration::from_millis(1));
    }
}

/// Wait until `counter` reaches `target` or `timeout` elapses.
fn wait_for_count(counter: &AtomicUsize, target: usize, timeout: Duration) -> usize {
    let deadline = Instant::now() + timeout;
    loop {
        let n = counter.load(Ordering::SeqCst);
        if n >= target {
            return n;
        }
        if Instant::now() > deadline {
            return n;
        }
        std::thread::sleep(Duration::from_millis(5));
    }
}

// ── Tests ────────────────────────────────────────────────────────────────────

/// Single-threaded vs multi-threaded runtime basics.
///
/// Story: We start with a single-threaded runtime driven by tick(), confirm
/// nothing happens without ticking, then graduate to a multi-threaded runtime
/// with run() and verify background processing, cross-worker delegation,
/// custom thread counts, and clean shutdown.
#[test]
fn single_vs_multi_threaded_basics() {
    // ── Part A: Single-threaded requires tick() ──
    let rt = std_runtime(RuntimeConfig::default());
    let addr = rt.spawn(PingPongActor).unwrap();
    let inbox = rt.new_inbox::<Pong>().unwrap();
    rt.send_to(addr, Ping { reply_to: *inbox.addr() }).unwrap();

    assert!(inbox.try_recv().is_none(), "no processing before tick");
    tick_n(&rt, 2);
    assert!(inbox.try_recv().is_some(), "tick() drives single-threaded processing");

    // ── Part B: Multi-threaded processes without ticking ──
    let rt_mt = std_runtime(RuntimeConfig { num_threads: 4, ..Default::default() });
    let addr = rt_mt.spawn(PingPongActor).unwrap();
    let inbox = rt_mt.new_inbox::<Pong>().unwrap();
    rt_mt.send_to(addr, Ping { reply_to: *inbox.addr() }).unwrap();

    let handle = rt_mt.run().unwrap();
    let reply = poll_inbox(&inbox, Duration::from_secs(5));
    assert!(reply.is_some(), "background workers process without manual ticking");

    // ── Part C: Cross-worker delegation (2 threads, spawn child from handler) ──
    let addr2 = handle.runtime.spawn(DelegatorActor).unwrap();
    let done_inbox = handle.runtime.new_inbox::<Done>().unwrap();
    handle.runtime.send_to(addr2, Forward { value: 3, reply_to: *done_inbox.addr() }).unwrap();

    let reply = poll_inbox(&done_inbox, Duration::from_secs(5));
    assert_eq!(reply, Some(Done(6)), "cross-worker delegation delivers reply");

    // ── Part D: Custom thread count reflected in stats ──
    let stats = handle.runtime.stats();
    assert_eq!(stats.num_workers, 4, "runtime respects requested thread count");

    // ── Part E: Clean shutdown ──
    handle.shutdown();
    handle.join();
    // Test passes by not hanging.
}

/// High-volume multi-threaded delivery.
///
/// Story: We throw large workloads at a 4-thread runtime — 50 senders
/// each firing 100 messages at one receiver, 200 concurrent spawn+send
/// pairs, and a 50-level chain that must hop across workers.
#[test]
fn mt_high_volume_delivery() {
    let cfg = || RuntimeConfig {
        num_threads: 4,
        max_actors: 5_000,
        channel_buffer_size: 10_000,
        ..Default::default()
    };

    // ── Part A: 50 senders × 100 messages → one receiver ──
    {
        let rt = std_runtime(cfg());
        let counter = Arc::new(AtomicUsize::new(0));
        let dummy = rt.new_inbox::<Pong>().unwrap();
        let receiver = rt.spawn(CountingPingActor { counter: counter.clone() }).unwrap();

        let total_expected = 50 * 100;
        for _ in 0..50 {
            for _ in 0..100 {
                rt.send_to(receiver, Ping { reply_to: *dummy.addr() }).unwrap();
            }
        }

        let handle = rt.run().unwrap();
        let processed = wait_for_count(&counter, total_expected, Duration::from_secs(5));
        handle.shutdown();
        handle.join();
        assert_eq!(processed, total_expected, "all 5000 messages delivered to single receiver");
    }

    // ── Part B: 200 concurrent spawn+send pairs ──
    {
        let rt = std_runtime(cfg());
        let counter = Arc::new(AtomicUsize::new(0));
        let dummy = rt.new_inbox::<Pong>().unwrap();
        for _ in 0..200 {
            let a = rt.spawn(CountingPingActor { counter: counter.clone() }).unwrap();
            rt.send_to(a, Ping { reply_to: *dummy.addr() }).unwrap();
        }

        let handle = rt.run().unwrap();
        let received = wait_for_count(&counter, 200, Duration::from_secs(5));
        handle.shutdown();
        handle.join();
        assert_eq!(received, 200, "all 200 spawn+send pairs complete");
    }

    // ── Part C: 50-level chain across workers ──
    {
        let rt = std_runtime(RuntimeConfig { num_threads: 2, max_actors: 5_000, ..Default::default() });
        let addr = rt.spawn(ChainActor).unwrap();
        let inbox = rt.new_inbox::<Done>().unwrap();
        rt.send_to(addr, ChainMsg { remaining: 50, depth: 0, reply_to: *inbox.addr() }).unwrap();

        let handle = rt.run().unwrap();
        let reply = poll_inbox(&inbox, Duration::from_secs(5));
        handle.shutdown();
        handle.join();
        assert_eq!(reply, Some(Done(50)), "50-level chain completes across workers");
    }
}

/// Panic isolation under multi-threaded load.
///
/// Story: 10 panicking actors and 10 healthy actors on 4 threads — every
/// panic is isolated and all 1000 healthy messages are still processed.
#[test]
fn mt_panic_isolation_under_load() {
    let rt = std_runtime(RuntimeConfig {
        num_threads: 4,
        max_actors: 5_000,
        channel_buffer_size: 10_000,
        ..Default::default()
    });

    let counter = Arc::new(AtomicUsize::new(0));
    let dummy = rt.new_inbox::<Pong>().unwrap();

    let mut panic_addrs = Vec::new();
    let mut healthy_addrs = Vec::new();
    for _ in 0..10 {
        panic_addrs.push(rt.spawn(PanicActor).unwrap());
        healthy_addrs.push(rt.spawn(CountingPingActor { counter: counter.clone() }).unwrap());
    }

    // Trigger panics and flood healthy actors.
    for &addr in &panic_addrs {
        rt.send_to(addr, PanicMsg).unwrap();
    }
    for &addr in &healthy_addrs {
        for _ in 0..100 {
            rt.send_to(addr, Ping { reply_to: *dummy.addr() }).unwrap();
        }
    }

    let handle = rt.run().unwrap();
    let expected = 10 * 100;
    let processed = wait_for_count(&counter, expected, Duration::from_secs(5));
    handle.shutdown();
    handle.join();

    assert_eq!(
        processed, expected,
        "all {expected} healthy messages processed despite panicking peers"
    );
}

/// Worker parking and shutdown latency.
///
/// Story: Workers park after idle time. We verify they wake quickly on new
/// messages, that messages sent after run() are delivered, and that shutdown
/// wakes all parked workers promptly.
#[test]
fn worker_parking_and_shutdown() {
    // ── Part A: Parked workers wake on send ──
    let rt = std_runtime(RuntimeConfig { num_threads: 2, ..Default::default() });
    let addr = rt.spawn(PingPongActor).unwrap();
    let inbox = rt.new_inbox::<Pong>().unwrap();

    let handle = rt.run().unwrap();
    std::thread::sleep(Duration::from_millis(50)); // Let workers park.

    let before = Instant::now();
    handle.runtime.send_to(addr, Ping { reply_to: *inbox.addr() }).unwrap();
    let reply = poll_inbox(&inbox, Duration::from_secs(1));
    let latency = before.elapsed();

    assert!(reply.is_some(), "parked worker should wake and process");
    assert!(latency.as_millis() < 100, "wake latency should be <100ms, was {:?}", latency);

    // ── Part B: Send after run() delivers ──
    let addr2 = handle.runtime.spawn(PingPongActor).unwrap();
    let inbox2 = handle.runtime.new_inbox::<Pong>().unwrap();
    std::thread::sleep(Duration::from_millis(10));
    handle.runtime.send_to(addr2, Ping { reply_to: *inbox2.addr() }).unwrap();

    let reply2 = poll_inbox(&inbox2, Duration::from_secs(5));
    assert!(reply2.is_some(), "message sent after run() must be delivered");

    handle.shutdown();
    handle.join();

    // ── Part C: Shutdown wakes parked workers quickly ──
    let rt2 = std_runtime(RuntimeConfig { num_threads: 4, ..Default::default() });
    let h2 = rt2.run().unwrap();
    std::thread::sleep(Duration::from_millis(50)); // Let workers park.

    let before = Instant::now();
    h2.shutdown();
    h2.join();
    let shutdown_time = before.elapsed();

    assert!(
        shutdown_time.as_millis() < 500,
        "shutdown should complete quickly with parked workers, took {:?}",
        shutdown_time
    );
}

/// Sustained throughput with no message loss.
///
/// Story: We send 10 batches of 100 messages, ticking between batches on a
/// single-threaded runtime. Each batch must make forward progress, and after
/// draining, all 1000 messages are accounted for.
#[test]
fn sustained_throughput_no_message_loss() {
    let rt = std_runtime(RuntimeConfig::default());
    let counter = Arc::new(AtomicUsize::new(0));
    let dummy = rt.new_inbox::<Pong>().unwrap();
    let addr = rt.spawn(CountingPingActor { counter: counter.clone() }).unwrap();

    for batch in 0..10 {
        for _ in 0..100 {
            rt.send_to(addr, Ping { reply_to: *dummy.addr() }).unwrap();
        }
        tick_n(&rt, 5);
        let processed = counter.load(Ordering::SeqCst);
        assert!(
            processed > batch * 50,
            "batch {batch}: expected progress, only {processed} processed"
        );
    }

    // Drain remaining.
    tick_n(&rt, 100);
    let total = counter.load(Ordering::SeqCst);
    assert_eq!(total, 1000, "sustained load should not drop any messages");
}

/// Load-aware actor placement.
///
/// Story: A fresh runtime falls back to round-robin (even distribution).
/// Under imbalanced load, new actors bias toward the lighter worker.
/// A single-worker runtime degrades gracefully.
#[test]
fn load_aware_actor_placement() {
    // ── Part A: Round-robin fallback on fresh runtime (4 workers, 100 actors) ──
    let rt = std_runtime(RuntimeConfig { num_threads: 4, ..Default::default() });
    for _ in 0..100 {
        rt.spawn(CounterActor { count: 0 }).unwrap();
    }

    let handle = rt.run().unwrap();
    std::thread::sleep(Duration::from_millis(20));

    let stats = handle.runtime.stats();
    handle.shutdown();
    handle.join();

    for w in &stats.workers {
        assert!(
            w.num_actors >= 20 && w.num_actors <= 30,
            "worker {} has {} actors, expected ~25 (round-robin)",
            w.id, w.num_actors
        );
    }

    // ── Part B: Imbalanced load biases toward lighter worker ──
    let rt2 = std_runtime(RuntimeConfig { num_threads: 2, ..Default::default() });
    let mut addrs = Vec::new();
    for _ in 0..20 {
        addrs.push(rt2.spawn(CounterActor { count: 0 }).unwrap());
    }

    let h2 = rt2.run().unwrap();
    std::thread::sleep(Duration::from_millis(10));

    // Bombard the first 10 actors (likely worker 0) with messages.
    for addr in &addrs[..10] {
        for _ in 0..50 {
            let _ = h2.runtime.send_to(*addr, Increment { reply_to: *addr });
        }
    }
    std::thread::sleep(Duration::from_millis(20));

    // Spawn 10 more — should bias toward lighter worker.
    for _ in 0..10 {
        h2.runtime.spawn(CounterActor { count: 0 }).unwrap();
    }
    std::thread::sleep(Duration::from_millis(20));

    let stats2 = h2.runtime.stats();
    h2.shutdown();
    h2.join();

    let total_actors: usize = stats2.workers.iter().map(|w| w.num_actors).sum();
    assert!(total_actors >= 20, "expected at least 20 actors, got {total_actors}");
    assert!(
        stats2.workers.iter().all(|w| w.num_actors > 0),
        "both workers should have actors: {:?}",
        stats2.workers.iter().map(|w| w.num_actors).collect::<Vec<_>>()
    );

    // ── Part C: Single-worker degrades gracefully ──
    let rt3 = std_runtime(RuntimeConfig::default());
    for _ in 0..50 {
        rt3.spawn(CounterActor { count: 0 }).unwrap();
    }
    tick_n(&rt3, 10);

    let stats3 = rt3.stats();
    assert_eq!(stats3.workers.len(), 1);
    assert_eq!(stats3.workers[0].num_actors, 50);
}