liminal-rs 0.16.2

//! LIM-009: a channel actor's death is told, never polled for.
//!
//! A command's reply sender sits in the channel core's queue, which outlives
//! the actor process. These tests pin that a waiter on a dead actor is answered
//! by the death itself (the exit watcher retires the queue and so drops the
//! sender), that retirement is one idempotent path whichever of the exit and a
//! restart reaches it first, and that the watcher costs nothing while no actor
//! exits. Every wait is on a signal; each bound is a guard against a hang,
//! never the witness.

use beamr::process::ExitReason;
use serde_json::json;

use std::error::Error;
use std::time::Duration;

use crate::channel::supervisor::{ChannelRestartPolicy, ChannelSupervisor};
use crate::channel::{ChannelConfig, ChannelHandle, ChannelMode, Schema};
use crate::envelope::PublisherId;

/// A guard on a signal that comes within one scheduler slice; never the witness.
const SIGNAL_GUARD: Duration = Duration::from_secs(5);

fn order_channel(supervisor: &ChannelSupervisor) -> Result<ChannelHandle, Box<dyn Error>> {
    let schema = Schema::new(json!({
        "type": "object",
        "properties": {
            "order_id": {"type": "string"},
            "quantity": {"type": "integer", "minimum": 1}
        },
        "required": ["order_id", "quantity"],
        "additionalProperties": false
    }))?;
    let config = ChannelConfig::new("orders".to_owned(), schema, ChannelMode::Ephemeral);
    Ok(ChannelHandle::with_supervisor(config, supervisor.clone()))
}

fn order(id: &str) -> Vec<u8> {
    format!(r#"{{"order_id":"{id}","quantity":1}}"#).into_bytes()
}

/// R1. P0's wake parks at the NIF entry; P1 queues behind it; the actor is then
/// terminated, so P1 is left on the queue with no process to serve it. No
/// command touches the core after that. The test hears the retirement from the
/// watcher, and only then waits on P1: its answer is the died error.
#[test]
fn a_waiter_on_a_dead_actor_is_told_by_the_exit() -> Result<(), Box<dyn Error>> {
    let supervisor = ChannelSupervisor::with_policy(ChannelRestartPolicy::never())?;
    let handle = order_channel(&supervisor)?;
    let core = handle.core()?;
    let pid = handle.actor_pid()?;
    let retired = supervisor.retirements()?;
    let hold = supervisor.entry_hold();

    hold.arm()?;
    let first = core.enqueue_publish(order("P0"), PublisherId::default(), None, None, None)?;
    hold.await_parked()?;
    let second = core.enqueue_publish(order("P1"), PublisherId::default(), None, None, None)?;
    handle
        .scheduler()?
        .terminate_process(pid, ExitReason::Error);
    hold.release()?;

    assert_eq!(
        retired.recv_timeout(SIGNAL_GUARD)?,
        pid,
        "the actor's exit itself retires its queue, with no later command to notice it"
    );
    assert!(
        core.await_publish(first).is_ok(),
        "P0 was served by the held slice"
    );
    let refused = core
        .await_publish(second)
        .err()
        .ok_or("P1's actor died before serving it, so its wait must fail")?;
    assert!(
        refused
            .to_string()
            .contains(&format!("channel actor pid {pid} died before replying")),
        "the wait names the death: {refused}"
    );
    supervisor.shutdown();
    Ok(())
}

/// R1: the reply wait holds no liveness poll and no loop around its receive.
#[test]
fn the_reply_wait_holds_no_liveness_poll() {
    let source = include_str!("actor/wait.rs");
    for polled in ["LIVENESS_POLL", "process_table", "loop {"] {
        assert!(
            !source.contains(polled),
            "channel/actor/wait.rs still holds `{polled}`: a waiter is told by the death"
        );
    }
}

/// R2: the watcher retires the dead actor; a second retire of the same pid has
/// no effect; the next command restarts the actor and is answered.
#[test]
fn a_retired_actor_restarts_on_the_next_command_and_a_second_retire_is_nothing()
-> Result<(), Box<dyn Error>> {
    let supervisor = ChannelSupervisor::new()?;
    let handle = order_channel(&supervisor)?;
    let subscription = handle.subscribe()?;
    let core = handle.core()?;
    let pid = handle.actor_pid()?;
    let retired = supervisor.retirements()?;

    handle
        .scheduler()?
        .terminate_process(pid, ExitReason::Error);
    assert_eq!(retired.recv_timeout(SIGNAL_GUARD)?, pid);
    supervisor.retire_again(pid, &core)?;
    assert!(
        retired.try_recv().is_err(),
        "a second retire of a retired pid takes no effect"
    );

    handle.publish(order("after"))?;
    assert_ne!(handle.actor_pid()?, pid, "the command restarted the actor");
    let delivered = subscription
        .try_next()?
        .ok_or("the restarted actor delivered the publish before replying")?;
    assert_eq!(delivered.payload, order("after"));
    supervisor.shutdown();
    Ok(())
}

/// R2: a restart racing the exit notification. Every command is answered or
/// refused by name; none waits to its deadline.
#[test]
fn a_restart_racing_the_exit_never_leaves_a_command_to_its_deadline() -> Result<(), Box<dyn Error>>
{
    const ROUNDS: u32 = 64;
    let supervisor = ChannelSupervisor::with_policy(ChannelRestartPolicy::one_for_one(ROUNDS + 1))?;
    let handle = order_channel(&supervisor)?;
    let scheduler = handle.scheduler()?;
    for round in 0..ROUNDS {
        let pid = handle.actor_pid()?;
        scheduler.terminate_process(pid, ExitReason::Error);
        if let Err(refused) = handle.publish(order(&format!("R{round}"))) {
            assert!(
                !refused.to_string().contains("timed out"),
                "round {round}: a command waited to its deadline: {refused}"
            );
        }
    }
    supervisor.shutdown();
    Ok(())
}

/// R3: while no channel actor exits, the watcher runs no slice. Commands and
/// deliveries (the unrelated counters) grow; the watcher's wake count stays at
/// what the actor's spawn left it.
#[test]
fn the_exit_watcher_costs_nothing_while_no_actor_exits() -> Result<(), Box<dyn Error>> {
    let supervisor = ChannelSupervisor::new()?;
    let handle = order_channel(&supervisor)?;
    let subscription = handle.subscribe()?;
    let before = supervisor.watcher_wakes();
    let mut delivered = 0_u32;
    for n in 0..256 {
        handle.publish(order(&format!("idle-{n}")))?;
        if subscription.try_next()?.is_some() {
            delivered += 1;
        }
    }
    assert_eq!(
        delivered, 256,
        "the unrelated work ran: every publish was delivered"
    );
    assert_eq!(
        supervisor.watcher_wakes(),
        before,
        "the exit watcher ran a slice with no actor exiting"
    );
    supervisor.shutdown();
    Ok(())
}