smarm/tests/gen_server.rs

//! gen_server tests: call round-trip, cast, lifecycle callbacks, and the two
//! server-down detection paths (reply-channel close vs. inbox-send failure).

use smarm::gen_server::{start, CallError, GenServer};
use smarm::run;
use std::sync::{Arc, Mutex};

// ---------------------------------------------------------------------------
// A trivial counter server: casts mutate, calls read (or blow up).
// ---------------------------------------------------------------------------

struct Counter {
    n: i64,
}

enum Req {
    Get,
    Boom,
}

enum Op {
    Add(i64),
}

impl GenServer for Counter {
    type Call = Req;
    type Reply = i64;
    type Cast = Op;

    fn handle_call(&mut self, req: Req) -> i64 {
        match req {
            Req::Get => self.n,
            Req::Boom => panic!("boom"),
        }
    }

    fn handle_cast(&mut self, op: Op) {
        match op {
            Op::Add(x) => self.n += x,
        }
    }
}

// Casts are applied in order and a later call observes the accumulated state.
#[test]
fn cast_then_call_roundtrip() {
    let got = Arc::new(Mutex::new(0i64));
    let got2 = got.clone();
    run(move || {
        let server = start(Counter { n: 0 });
        server.cast(Op::Add(5)).unwrap();
        server.cast(Op::Add(3)).unwrap();
        let n = server.call(Req::Get).unwrap();
        *got2.lock().unwrap() = n;
    });
    assert_eq!(*got.lock().unwrap(), 8);
}

// ---------------------------------------------------------------------------
// Lifecycle: init runs before the first message, terminate on graceful exit.
// ---------------------------------------------------------------------------

struct Lifecycle {
    log: Arc<Mutex<Vec<&'static str>>>,
}

impl GenServer for Lifecycle {
    type Call = ();
    type Reply = ();
    type Cast = ();

    fn init(&mut self) {
        self.log.lock().unwrap().push("init");
    }

    fn handle_call(&mut self, _req: ()) {
        self.log.lock().unwrap().push("call");
    }

    fn handle_cast(&mut self, _req: ()) {}

    fn terminate(&mut self) {
        self.log.lock().unwrap().push("terminate");
    }
}

// init -> handle_call -> (drop last ref closes inbox) -> terminate.
#[test]
fn init_and_terminate_run() {
    let log = Arc::new(Mutex::new(Vec::new()));
    let log2 = log.clone();
    run(move || {
        let server = start(Lifecycle { log: log2 });
        server.call(()).unwrap();
        // Dropping the only ref closes the inbox; the server breaks out of its
        // recv loop and runs terminate. run() will not return until it has.
        drop(server);
    });
    assert_eq!(*log.lock().unwrap(), vec!["init", "call", "terminate"]);
}

// ---------------------------------------------------------------------------
// Server-down detection.
// ---------------------------------------------------------------------------

// The server dies *while* a call is in flight (handler panics): the reply
// sender drops on unwind, closing the reply channel, so the parked caller's
// recv returns Err -> ServerDown.
#[test]
fn call_to_panicking_handler_is_server_down() {
    let got = Arc::new(Mutex::new(None));
    let got2 = got.clone();
    run(move || {
        let server = start(Counter { n: 0 });
        let r = server.call(Req::Boom);
        *got2.lock().unwrap() = Some(r);
    });
    assert_eq!(*got.lock().unwrap(), Some(Err(CallError::ServerDown)));
}

// A call issued *after* the server is already gone: the inbox is closed, so the
// send itself fails -> ServerDown (the other detection path).
#[test]
fn call_after_server_gone_is_server_down() {
    let got = Arc::new(Mutex::new(None));
    let got2 = got.clone();
    run(move || {
        let server = start(Counter { n: 0 });
        let server2 = server.clone();
        // This kills the server (and is itself ServerDown).
        assert_eq!(server.call(Req::Boom), Err(CallError::ServerDown));
        // Inbox now closed; a fresh call can't even be enqueued.
        let r = server2.call(Req::Get);
        *got2.lock().unwrap() = Some(r);
    });
    assert_eq!(*got.lock().unwrap(), Some(Err(CallError::ServerDown)));
}

// ---------------------------------------------------------------------------
// call_timeout
// ---------------------------------------------------------------------------

use smarm::gen_server::CallTimeoutError;
use std::time::{Duration, Instant};

/// Replies after sleeping `delay_ms` (parking the server actor, not the OS
/// thread), so callers can race a deadline against the reply.
struct Slow;

impl GenServer for Slow {
    type Call = u64; // delay in ms
    type Reply = u64;
    type Cast = ();

    fn handle_call(&mut self, delay_ms: u64) -> u64 {
        if delay_ms > 0 {
            smarm::sleep(Duration::from_millis(delay_ms));
        }
        delay_ms
    }

    fn handle_cast(&mut self, _: ()) {}
}

#[test]
fn call_timeout_returns_reply_within_deadline() {
    run(|| {
        let srv = start(Slow);
        assert_eq!(srv.call_timeout(0, Duration::from_secs(10)), Ok(0));
    });
}

#[test]
fn call_timeout_times_out_on_slow_handler() {
    run(|| {
        let srv = start(Slow);
        let start_t = Instant::now();
        let r = srv.call_timeout(500, Duration::from_millis(50));
        assert_eq!(r, Err(CallTimeoutError::Timeout));
        let elapsed = start_t.elapsed();
        // Gave up at the deadline, not at the reply.
        assert!(elapsed >= Duration::from_millis(50));
        assert!(elapsed < Duration::from_millis(500));
    });
}

#[test]
fn server_survives_an_abandoned_call_and_late_reply_is_discarded() {
    run(|| {
        let srv = start(Slow);
        assert_eq!(
            srv.call_timeout(100, Duration::from_millis(20)),
            Err(CallTimeoutError::Timeout)
        );
        // The timed-out request is still handled; its reply send fails
        // harmlessly (receiver dropped). The server must keep serving, and
        // the late reply must not leak into THIS call's reply channel.
        assert_eq!(srv.call_timeout(0, Duration::from_secs(10)), Ok(0));
        // Plain unbounded call still fine too.
        assert_eq!(srv.call(0), Ok(0));
    });
}

#[test]
fn call_timeout_to_dead_server_is_server_down_not_timeout() {
    struct Bomb;
    impl GenServer for Bomb {
        type Call = ();
        type Reply = ();
        type Cast = ();
        fn handle_call(&mut self, _: ()) {
            panic!("kaboom");
        }
        fn handle_cast(&mut self, _: ()) {}
    }

    run(|| {
        let srv = start(Bomb);
        // Dies mid-call: reply channel closes -> ServerDown (even though the
        // generous deadline never fires).
        assert_eq!(
            srv.call_timeout((), Duration::from_secs(10)),
            Err(CallTimeoutError::ServerDown)
        );
        // Already gone: inbox send fails -> ServerDown.
        assert_eq!(
            srv.call_timeout((), Duration::from_secs(10)),
            Err(CallTimeoutError::ServerDown)
        );
    });
}