smarm/tests/pg.rs

//! Process-group tests that run under the scheduler: `join` installs a real
//! monitor on a live actor, and a real death drives eviction on next contact.
//! (Pure structural invariants live in the `pg` unit tests.)

use smarm::{channel, members, pick, run, spawn};
use smarm::{join, leave};

#[test]
fn join_then_members_lists_a_live_member() {
    run(|| {
        let (tx, rx) = channel::<()>();
        let h = spawn(move || {
            rx.recv().unwrap();
        });
        let pid = h.pid();
        assert!(join("workers", pid), "first join is new");
        assert!(!join("workers", pid), "second join is idempotent");
        assert_eq!(members("workers"), vec![pid]);
        assert_eq!(pick("workers"), Some(pid));
        tx.send(()).unwrap();
        h.join().unwrap();
    });
}

#[test]
fn a_dead_actor_vanishes_from_every_group_it_joined() {
    run(|| {
        let (tx, rx) = channel::<()>();
        let h = spawn(move || {
            rx.recv().unwrap();
        });
        let pid = h.pid();
        join("g1", pid);
        join("g2", pid);
        assert_eq!(members("g1"), vec![pid]);
        assert_eq!(members("g2"), vec![pid]);

        // Release and reap the actor. finalize_actor queues the Down to our
        // monitors before unparking joiners, so by the time join() returns the
        // Down is already waiting in the membership channel.
        tx.send(()).unwrap();
        h.join().unwrap();

        // Drain-on-contact: touching g1 detects the death and sweeps the pid
        // out of every group (g2 included), not just g1.
        assert!(members("g1").is_empty(), "evicted from the touched group");
        assert!(members("g2").is_empty(), "and swept from the untouched group");
        assert_eq!(pick("g1"), None);
    });
}

#[test]
fn pick_returns_none_when_the_only_member_is_dead() {
    run(|| {
        let (tx, rx) = channel::<()>();
        let h = spawn(move || {
            rx.recv().unwrap();
        });
        let pid = h.pid();
        join("pool", pid);
        tx.send(()).unwrap();
        h.join().unwrap();
        assert_eq!(pick("pool"), None);
        assert!(members("pool").is_empty());
    });
}

#[test]
fn live_members_survive_a_peers_death() {
    run(|| {
        let (tx_a, rx_a) = channel::<()>();
        let (tx_b, rx_b) = channel::<()>();
        let a = spawn(move || {
            rx_a.recv().unwrap();
        });
        let b = spawn(move || {
            rx_b.recv().unwrap();
        });
        join("svc", a.pid());
        join("svc", b.pid());

        // Kill a; b is still parked on its channel.
        tx_a.send(()).unwrap();
        a.join().unwrap();

        assert_eq!(members("svc"), vec![b.pid()], "only the dead peer is reaped");
        assert_eq!(pick("svc"), Some(b.pid()));

        tx_b.send(()).unwrap();
        b.join().unwrap();
    });
}

#[test]
fn leave_drops_a_membership_without_affecting_others() {
    run(|| {
        let (tx_a, rx_a) = channel::<()>();
        let (tx_b, rx_b) = channel::<()>();
        let a = spawn(move || {
            rx_a.recv().unwrap();
        });
        let b = spawn(move || {
            rx_b.recv().unwrap();
        });
        join("g", a.pid());
        join("g", b.pid());
        assert!(leave("g", a.pid()), "a was a member");
        assert!(!leave("g", a.pid()), "leaving twice finds nothing");
        assert_eq!(members("g"), vec![b.pid()]);

        tx_a.send(()).unwrap();
        tx_b.send(()).unwrap();
        a.join().unwrap();
        b.join().unwrap();
    });
}

#[test]
fn joining_an_already_dead_pid_is_evicted_on_next_contact() {
    run(|| {
        let h = spawn(|| {});
        let pid = h.pid();
        h.join().unwrap(); // actor is finalized before we join it to anything

        // monitor() on a gone pid queues a NoProc Down immediately, so the
        // membership is reaped the next time the group is touched.
        join("late", pid);
        assert!(members("late").is_empty(), "dead-at-join member is reaped on read");
        assert_eq!(pick("late"), None);
    });
}