liminal-rs 0.14.1

//! LIM-002 behavioural tests for the process-backed channel actor.
//!
//! Every test runs against a dedicated [`ChannelSupervisor`] (its own beamr
//! scheduler) so they are isolated from the process-global default supervisor
//! and from each other. Delivery is performed synchronously inside the actor's
//! `Publish` handler before its reply is sent, so a message is observable on a
//! subscriber's inbox as soon as `publish` returns; lifecycle effects driven by
//! a process EXIT (subscriber death) are asynchronous and are awaited on the
//! actor's own departure report. Only a crashed actor's removal from the
//! process table is still awaited with a bounded poll: beamr 0.20.0 signals an
//! exit before the pid leaves the table (see `crash_actor`).

use beamr::process::ExitReason;

use crate::channel::{
    ChannelConfig, ChannelHandle, ChannelMode, PublishRecord, Schema, SchemaValidationError,
};
use crate::channel::registry::ChannelRegistry;
use crate::channel::subscription::SubscriptionHandle;
use crate::channel::supervisor::{ChannelRestartPolicy, ChannelSupervisor};
use crate::envelope::{Envelope, PublisherId};
use crate::error::LiminalError;

use serde_json::{Value, json};

use std::error::Error;
use std::sync::Arc;
use std::sync::mpsc::Receiver;
use std::time::{Duration, Instant};

fn order_schema() -> Result<Schema, SchemaValidationError> {
    Schema::new(json!({
        "type": "object",
        "properties": {
            "order_id": {"type": "string"},
            "quantity": {"type": "integer", "minimum": 1}
        },
        "required": ["order_id", "quantity"],
        "additionalProperties": false
    }))
}

fn order_channel(supervisor: &ChannelSupervisor) -> Result<ChannelHandle, Box<dyn Error>> {
    let config = ChannelConfig::new("orders".to_owned(), order_schema()?, ChannelMode::Ephemeral);
    Ok(ChannelHandle::with_supervisor(config, supervisor.clone()))
}

/// The next envelope delivered to `subscription`. The channel actor pushes
/// each envelope onto the subscriber's inbox inside its `Publish` handler,
/// before it replies, so once `publish` has returned the envelope is already
/// there: nothing to wait for. `try_next` pops that inbox without blocking
/// (`subscription.rs:1231-1246`).
fn await_envelope(subscription: &SubscriptionHandle) -> Result<Envelope, Box<dyn Error>> {
    Ok(subscription
        .try_next()?
        .ok_or("no envelope was delivered by the time publish returned")?)
}

/// Each departure the channel actor reports: the channel's name and the
/// subscriber's pid.
type Departures = Receiver<(String, u64)>;

/// A supervisor whose channels report every departure to the returned
/// receiver. The actor reports a subscriber's exit only after it has removed
/// the subscriber from its fan-out list (`ChannelActorCore::handle_subscriber_exit`),
/// so a test that has heard the report reads the count directly.
fn observed_supervisor() -> Result<(ChannelSupervisor, Departures), Box<dyn Error>> {
    let supervisor = ChannelSupervisor::new()?;
    let (reports, departures) = std::sync::mpsc::channel();
    supervisor.install_observer(Arc::new(DepartureReports { reports }));
    Ok((supervisor, departures))
}

/// Drops `subscription`, which ends its process, and returns once the
/// channel actor has handled that exit.
///
/// Dropping terminates the subscriber, and beamr sends its link EXIT to the
/// actor before `terminate_process` returns (cleanup runs on the caller's
/// thread, `execution/core.rs:1645` at beamr 0.20.0). The actor takes an EXIT
/// before any later command's wake, in a parked mailbox and at a running
/// slice's store-back alike (`core.rs:389-415` merges exits first). So once a
/// later round trip through the actor (`subscriber_count`) is answered, the
/// departure report is already sent, and a missing one is a failure here and
/// not a hang.
fn drop_and_await_departure(
    subscription: SubscriptionHandle,
    handle: &ChannelHandle,
    departures: &Departures,
) -> Result<(), Box<dyn Error>> {
    let pid = subscription.pid();
    drop(subscription);
    handle.subscriber_count()?;
    match departures.try_recv() {
        Ok((_, departed)) if departed == pid => Ok(()),
        Ok((_, departed)) => {
            Err(format!("subscriber {departed} departed where {pid} was dropped").into())
        }
        Err(_) => Err(format!(
            "no departure was reported for subscriber {pid} by the time the actor answered a \
             later command: the actor never saw its exit"
        )
        .into()),
    }
}

/// Crash the channel actor process `handle` is bound to with an abnormal exit
/// (the same `terminate_process(_, ExitReason::Error)` path the conversation
/// crash tests use) and block until that pid has actually left the scheduler's
/// process table, so the subsequent operation deterministically observes a dead
/// actor and drives a restart. Returns the crashed pid.
///
/// This is the one bounded poll left. beamr 0.20.0 fires an exit watch before
/// the pid leaves the table (`execution/core.rs:1638-1682`), and the
/// supervisor's `live_pid` checks the table, so no signal yet says the next
/// operation will restart. The work list's beamr row asks for one.
fn crash_actor(handle: &ChannelHandle) -> Result<u64, Box<dyn Error>> {
    let pid = handle.actor_pid()?;
    let scheduler = handle.scheduler()?;
    scheduler.terminate_process(pid, ExitReason::Error);
    let deadline = Instant::now() + Duration::from_secs(2);
    while Instant::now() < deadline {
        if scheduler.process_table().get(pid).is_none() {
            return Ok(pid);
        }
        std::thread::sleep(Duration::from_millis(5));
    }
    Err(format!("channel actor pid {pid} never left the process table after crash").into())
}

#[test]
fn channel_restarts_after_crash_and_publish_succeeds() -> Result<(), Box<dyn Error>> {
    let supervisor = ChannelSupervisor::new()?;
    let handle = order_channel(&supervisor)?;
    let subscription = handle.subscribe()?;
    let original_pid = handle.actor_pid()?;

    // Crash the live actor, then publish: the supervisor must restart it.
    crash_actor(&handle)?;
    handle.publish(br#"{"order_id":"A1","quantity":3}"#)?;

    // (a) The channel came back on a NEW pid and the publish was delivered.
    let restarted_pid = handle.actor_pid()?;
    assert_ne!(
        original_pid, restarted_pid,
        "the restarted actor must run on a fresh pid"
    );
    let envelope = await_envelope(&subscription)?;
    assert_eq!(
        envelope.payload,
        br#"{"order_id":"A1","quantity":3}"#.to_vec()
    );
    supervisor.shutdown();
    Ok(())
}

/// The pin depends on beamr 0.20.0's ordering (git 05adf473): a dropped
/// subscriber's link EXIT is sent before `terminate_process` returns
/// (`execution/core.rs:1645`), and the actor takes it before any later
/// command's wake, at a running slice's store-back too (`core.rs:389-415`
/// merges exits ahead of mail). So `drop_and_await_departure` reads the
/// departure report after one round trip, without waiting, and a missing
/// relink fails on its named error in milliseconds rather than hanging.
#[test]
fn subscriber_death_detection_survives_restart() -> Result<(), Box<dyn Error>> {
    // MAJOR-1 regression guard. A subscriber created BEFORE the crash must still
    // have its death detected AFTER the actor restarts. This only works if the
    // restarted actor re-links to the surviving subscriber on boot; without the
    // re-link the EXIT never reaches the new actor and no departure is ever
    // reported, so this test fails in `drop_and_await_departure` ("the actor
    // never saw its exit") before the fix and passes after it.
    let (supervisor, departures) = observed_supervisor()?;
    let handle = order_channel(&supervisor)?;
    let keep = handle.subscribe()?;
    let transient = handle.subscribe()?;
    assert_eq!(handle.subscriber_count()?, 2);

    // Crash + restart (the publish drives the restart and the boot re-link).
    crash_actor(&handle)?;
    handle.publish(br#"{"order_id":"A1","quantity":3}"#)?;
    assert_eq!(
        handle.subscriber_count()?,
        2,
        "both pre-crash subscribers must still be registered after restart"
    );

    // Drop one subscriber AFTER the restart: its EXIT must prune it from the
    // restarted actor's fan-out — proving the re-link was re-established.
    drop_and_await_departure(transient, &handle, &departures)?;
    assert_eq!(handle.subscriber_count()?, 1);

    // The survivor still receives published messages on the restarted actor.
    handle.publish(br#"{"order_id":"A2","quantity":4}"#)?;
    let _ = await_envelope(&keep)?;
    supervisor.shutdown();
    Ok(())
}

#[test]
fn max_restarts_budget_bounds_restarts() -> Result<(), Box<dyn Error>> {
    // A budget of exactly one restart: the first crash restarts, the second
    // crash exhausts the budget and the next operation must surface an error
    // rather than silently respawning again.
    let supervisor = ChannelSupervisor::with_policy(ChannelRestartPolicy::one_for_one(1))?;
    let handle = order_channel(&supervisor)?;
    let _subscription = handle.subscribe()?;

    // First crash + restart (consumes the single restart from the budget).
    crash_actor(&handle)?;
    handle.publish(br#"{"order_id":"A1","quantity":3}"#)?;

    // Second crash: the budget is now exhausted, so the next operation must fail
    // and the actor must NOT be respawned.
    crash_actor(&handle)?;
    let result = handle.publish(br#"{"order_id":"A2","quantity":4}"#);
    assert!(
        matches!(result, Err(LiminalError::DeliveryFailed { .. })),
        "exhausting the restart budget must surface a DeliveryFailed error, got {result:?}"
    );
    supervisor.shutdown();
    Ok(())
}

#[test]
fn command_on_dead_actor_returns_promptly_not_after_timeout() -> Result<(), Box<dyn Error>> {
    // MINOR-2: if the actor dies while a command's reply sender sits in the
    // command queue, the host must NOT block the full 5s COMMAND_TIMEOUT. With a
    // `never` restart policy a crashed actor stays dead, so a subsequent
    // operation must surface an error well within the timeout. We assert a tight
    // wall-clock bound: a hang would blow far past it.
    let supervisor = ChannelSupervisor::with_policy(ChannelRestartPolicy::never())?;
    let handle = order_channel(&supervisor)?;
    let _subscription = handle.subscribe()?;
    // Confirm the actor is live, then crash it so it can never reply.
    let _ = handle.actor_pid()?;
    crash_actor(&handle)?;

    let started = Instant::now();
    let result = handle.publish(br#"{"order_id":"A1","quantity":3}"#);
    let elapsed = started.elapsed();
    assert!(
        result.is_err(),
        "publish to a dead non-restarting actor must fail"
    );
    assert!(
        elapsed < Duration::from_secs(2),
        "command on a dead actor must return promptly (took {elapsed:?}), not block the 5s timeout"
    );
    supervisor.shutdown();
    Ok(())
}

#[test]
fn restarting_one_channel_does_not_disturb_another() -> Result<(), Box<dyn Error>> {
    let supervisor = ChannelSupervisor::new()?;
    let orders = ChannelHandle::with_supervisor(
        ChannelConfig::new("orders".to_owned(), order_schema()?, ChannelMode::Ephemeral),
        supervisor.clone(),
    );
    let events = ChannelHandle::with_supervisor(
        ChannelConfig::new("events".to_owned(), order_schema()?, ChannelMode::Ephemeral),
        supervisor.clone(),
    );
    let orders_sub = orders.subscribe()?;
    let events_sub = events.subscribe()?;
    let events_pid = events.actor_pid()?;

    // Crash + restart ONLY the orders channel.
    crash_actor(&orders)?;
    orders.publish(br#"{"order_id":"only-orders","quantity":2}"#)?;
    let delivered = await_envelope(&orders_sub)?;
    assert_eq!(
        delivered.payload,
        br#"{"order_id":"only-orders","quantity":2}"#.to_vec()
    );

    // The events channel was untouched: same pid, still delivering.
    assert_eq!(
        events.actor_pid()?,
        events_pid,
        "the unrelated channel actor must keep its original pid"
    );
    events.publish(br#"{"order_id":"events-live","quantity":1}"#)?;
    let event = await_envelope(&events_sub)?;
    assert_eq!(
        event.payload,
        br#"{"order_id":"events-live","quantity":1}"#.to_vec()
    );
    supervisor.shutdown();
    Ok(())
}

#[test]
fn publish_valid_message_delivers_envelope() -> Result<(), Box<dyn Error>> {
    let supervisor = ChannelSupervisor::new()?;
    let handle = order_channel(&supervisor)?;
    let subscription = handle.subscribe()?;

    handle.publish_from("publisher-1", br#"{"order_id":"A1","quantity":3}"#)?;
    let envelope = await_envelope(&subscription)?;

    assert_eq!(
        envelope.payload,
        br#"{"order_id":"A1","quantity":3}"#.to_vec()
    );
    assert_eq!(envelope.publisher_id.as_str(), "publisher-1");
    assert!(envelope.causal_context.is_none());
    supervisor.shutdown();
    Ok(())
}

#[test]
fn publish_invalid_message_returns_schema_mismatch_without_delivery() -> Result<(), Box<dyn Error>>
{
    let supervisor = ChannelSupervisor::new()?;
    let handle = order_channel(&supervisor)?;
    let subscription = handle.subscribe()?;

    let result = handle.publish(br#"{"order_id":"A1","quantity":0}"#);

    assert!(matches!(result, Err(LiminalError::SchemaMismatch { .. })));
    assert!(subscription.try_next()?.is_none());
    supervisor.shutdown();
    Ok(())
}

#[test]
fn evolved_schema_keeps_existing_subscriber_and_applies_default() -> Result<(), Box<dyn Error>> {
    let supervisor = ChannelSupervisor::new()?;
    let handle = order_channel(&supervisor)?;
    let subscription = handle.subscribe()?;
    let schema_id =
        handle.evolve_schema_add_field("priority", json!({"type":"string"}), json!("normal"))?;

    handle.publish(br#"{"order_id":"A1","quantity":3}"#)?;
    let envelope = await_envelope(&subscription)?;
    let payload: Value = serde_json::from_slice(&envelope.payload)?;

    assert_eq!(envelope.schema_id, schema_id);
    assert_eq!(payload.get("priority"), Some(&json!("normal")));
    supervisor.shutdown();
    Ok(())
}

#[test]
fn predicate_subscription_only_receives_matching_messages() -> Result<(), Box<dyn Error>> {
    let supervisor = ChannelSupervisor::new()?;
    let handle = order_channel(&supervisor)?;
    // Only deliver orders with quantity >= 5.
    let filtered = handle.subscribe_filtered(|envelope: &Envelope| {
        serde_json::from_slice::<Value>(&envelope.payload)
            .ok()
            .and_then(|value| value.get("quantity").and_then(Value::as_u64))
            .is_some_and(|quantity| quantity >= 5)
    })?;
    let unfiltered = handle.subscribe()?;

    handle.publish(br#"{"order_id":"small","quantity":1}"#)?;
    handle.publish(br#"{"order_id":"big","quantity":9}"#)?;

    // The unfiltered subscriber receives both, in order.
    let first = await_envelope(&unfiltered)?;
    let second = await_envelope(&unfiltered)?;
    assert_eq!(
        first.payload,
        br#"{"order_id":"small","quantity":1}"#.to_vec()
    );
    assert_eq!(
        second.payload,
        br#"{"order_id":"big","quantity":9}"#.to_vec()
    );

    // The filtered subscriber receives ONLY the matching (big) order.
    let matched = await_envelope(&filtered)?;
    assert_eq!(
        matched.payload,
        br#"{"order_id":"big","quantity":9}"#.to_vec()
    );
    assert!(filtered.try_next()?.is_none());
    supervisor.shutdown();
    Ok(())
}

#[test]
fn dropping_subscription_removes_subscriber_via_exit() -> Result<(), Box<dyn Error>> {
    let (supervisor, departures) = observed_supervisor()?;
    let handle = order_channel(&supervisor)?;
    let keep = handle.subscribe()?;
    let transient = handle.subscribe()?;
    assert_eq!(handle.subscriber_count()?, 2);

    // Dropping the handle terminates the subscriber process; the channel actor
    // traps the resulting EXIT and prunes it from the fan-out list — no polling
    // of a weak pointer is involved.
    drop_and_await_departure(transient, &handle, &departures)?;
    assert_eq!(handle.subscriber_count()?, 1);

    // The surviving subscriber still receives published messages.
    handle.publish(br#"{"order_id":"A1","quantity":3}"#)?;
    let envelope = await_envelope(&keep)?;
    assert_eq!(
        envelope.payload,
        br#"{"order_id":"A1","quantity":3}"#.to_vec()
    );
    supervisor.shutdown();
    Ok(())
}

#[test]
fn explicit_unsubscribe_removes_subscriber() -> Result<(), Box<dyn Error>> {
    let supervisor = ChannelSupervisor::new()?;
    let handle = order_channel(&supervisor)?;
    let subscription = handle.subscribe()?;
    assert_eq!(handle.subscriber_count()?, 1);

    handle.unsubscribe(&subscription)?;
    assert_eq!(handle.subscriber_count()?, 0);
    supervisor.shutdown();
    Ok(())
}

#[test]
fn publish_after_close_is_rejected() -> Result<(), Box<dyn Error>> {
    let supervisor = ChannelSupervisor::new()?;
    let handle = order_channel(&supervisor)?;
    handle.close()?;

    // The actor process stopped on close; a publish therefore cannot be
    // delivered. It must fail rather than silently succeed.
    let result = handle.publish(br#"{"order_id":"A1","quantity":3}"#);
    assert!(result.is_err());
    supervisor.shutdown();
    Ok(())
}

#[test]
fn registry_creates_looks_up_lists_and_rejects_duplicates() -> Result<(), Box<dyn Error>> {
    let registry = ChannelRegistry::new()?;
    let config = ChannelConfig::new("orders".to_owned(), order_schema()?, ChannelMode::Ephemeral);

    let handle = registry.create(config.clone())?;
    let _subscription = handle.subscribe()?;

    // Lookup returns the same channel.
    assert!(registry.lookup("orders")?.is_some());
    assert!(registry.lookup("missing")?.is_none());

    // Duplicate names are rejected.
    assert!(matches!(
        registry.create(config),
        Err(LiminalError::PublishFailed { .. })
    ));

    // List reports the channel with its live subscriber count.
    let summaries = registry.list()?;
    assert_eq!(summaries.len(), 1);
    assert_eq!(summaries[0].name, "orders");
    assert_eq!(summaries[0].subscriber_count, 1);

    // Close removes it.
    assert!(registry.close("orders")?);
    assert!(registry.lookup("orders")?.is_none());
    assert!(!registry.close("orders")?);
    registry.shutdown();
    Ok(())
}

#[test]
fn separate_channels_are_independent_processes() -> Result<(), Box<dyn Error>> {
    let supervisor = ChannelSupervisor::new()?;
    let orders = ChannelHandle::with_supervisor(
        ChannelConfig::new("orders".to_owned(), order_schema()?, ChannelMode::Ephemeral),
        supervisor.clone(),
    );
    let events = ChannelHandle::with_supervisor(
        ChannelConfig::new("events".to_owned(), order_schema()?, ChannelMode::Ephemeral),
        supervisor.clone(),
    );
    let orders_sub = orders.subscribe()?;
    let events_sub = events.subscribe()?;

    orders.publish(br#"{"order_id":"only-orders","quantity":2}"#)?;
    let delivered = await_envelope(&orders_sub)?;
    assert_eq!(
        delivered.payload,
        br#"{"order_id":"only-orders","quantity":2}"#.to_vec()
    );
    // The events channel's subscriber receives nothing — separate process,
    // separate subscriber list.
    assert!(events_sub.try_next()?.is_none());
    supervisor.shutdown();
    Ok(())
}

#[test]
fn is_actor_spawned_reports_an_idle_channel_without_spawning_it() -> Result<(), Box<dyn Error>> {
    let supervisor = ChannelSupervisor::new()?;
    let handle = order_channel(&supervisor)?;
    let scheduler = supervisor.scheduler();
    let idle_processes = scheduler.process_table().len();

    // Asking repeatedly must keep answering `false`: the probe reports the
    // lazy-spawn state without consuming it, unlike every accessor that routes
    // through `core()`.
    for probe in 0..3 {
        assert!(
            !handle.is_actor_spawned(),
            "probe {probe} must find a configured-but-untouched channel unspawned"
        );
    }

    // Independent witness for the same claim: the supervisor's own scheduler
    // gained no process across the probes, so `false` is not the answer of a
    // method that quietly spawned one behind it.
    assert_eq!(
        scheduler.process_table().len(),
        idle_processes,
        "probing spawn-state must add no process to the scheduler"
    );
    supervisor.shutdown();
    Ok(())
}

#[test]
fn is_actor_spawned_is_true_once_an_operation_forces_the_actor() -> Result<(), Box<dyn Error>> {
    // The other direction: without this arm, a method that always answered
    // `false` would pass the idle test above.
    let supervisor = ChannelSupervisor::new()?;
    let handle = order_channel(&supervisor)?;
    assert!(!handle.is_actor_spawned());

    let subscription = handle.subscribe()?;

    assert!(
        handle.is_actor_spawned(),
        "an operation that drove the actor must leave the spawn observable"
    );
    drop(subscription);
    supervisor.shutdown();
    Ok(())
}

#[test]
fn restart_policy_never_is_configurable() -> Result<(), Box<dyn Error>> {
    // Supervision is configurable (R4): a `never` policy is accepted and exposed.
    let supervisor = ChannelSupervisor::with_policy(ChannelRestartPolicy::never())?;
    assert_eq!(supervisor.policy(), &ChannelRestartPolicy::never());
    let handle = order_channel(&supervisor)?;
    let subscription = handle.subscribe()?;
    handle.publish(br#"{"order_id":"A1","quantity":3}"#)?;
    let _ = await_envelope(&subscription)?;
    supervisor.shutdown();
    Ok(())
}

#[test]
fn invalid_publish_to_a_durable_channel_burns_no_store_sequence() -> Result<(), Box<dyn Error>> {
    // G4 (docs/design/wire-durability-gaps-20260831.md): before the
    // pre-publish gate, a durable channel persisted the payload FIRST and
    // validated inside the actor afterwards, so a payload the channel refused
    // still consumed a haematite sequence and left durable bytes the channel
    // never accepted. The refusal must leave the log untouched.
    let supervisor = ChannelSupervisor::new()?;
    let store: std::sync::Arc<dyn crate::durability::store::DurableStore> =
        std::sync::Arc::new(crate::durability::store::open_ephemeral(1)?);
    let config = ChannelConfig::new("orders".to_owned(), order_schema()?, ChannelMode::Durable);
    let handle = ChannelHandle::new_durable_with_supervisor(
        config,
        std::sync::Arc::clone(&store),
        supervisor.clone(),
    )?;

    // quantity violates the schema's minimum.
    let refused = handle.publish(br#"{"order_id":"A1","quantity":0}"#);
    assert!(matches!(refused, Err(LiminalError::SchemaMismatch { .. })));

    let after_refusal = crate::durability::bridge::block_on(store.read_from("orders:0", 0, 16))??;
    assert!(
        after_refusal.is_empty(),
        "a refused publish burned {} durable record(s)",
        after_refusal.len()
    );

    // Positive control through the same predicate: a valid publish appends
    // exactly one record to the same stream key the refusal was checked on.
    handle.publish(br#"{"order_id":"A1","quantity":3}"#)?;
    let after_accept = crate::durability::bridge::block_on(store.read_from("orders:0", 0, 16))??;
    assert_eq!(
        after_accept.len(),
        1,
        "the control publish must land on the stream key the refusal was measured against"
    );
    supervisor.shutdown();
    Ok(())
}

#[test]
fn a_durable_publish_reports_the_store_assigned_position() -> Result<(), Box<dyn Error>> {
    // G1 (docs/design/wire-durability-gaps-20260831.md): haematite's append
    // assigns the durable position and `persist_durable` used to discard it,
    // so no caller of a durable channel could ever learn where its publish
    // landed. The delivery now carries it, and the claim is pinned against
    // the store's own rows — consecutive publishes report consecutive
    // positions, and the row count agrees.
    let supervisor = ChannelSupervisor::new()?;
    let store: std::sync::Arc<dyn crate::durability::store::DurableStore> =
        std::sync::Arc::new(crate::durability::store::open_ephemeral(1)?);
    let config = ChannelConfig::new("orders".to_owned(), order_schema()?, ChannelMode::Durable);
    let handle = ChannelHandle::new_durable_with_supervisor(
        config,
        std::sync::Arc::clone(&store),
        supervisor.clone(),
    )?;

    let first = handle.publish_with_delivery(
        br#"{"order_id":"A1","quantity":3}"#,
        PublisherId::default(),
        None,
    )?;
    let second = handle.publish_with_delivery(
        br#"{"order_id":"A2","quantity":4}"#,
        PublisherId::default(),
        None,
    )?;
    let first_position = first
        .durable_position()
        .ok_or("a durable publish reported no durable position")?;
    let second_position = second
        .durable_position()
        .ok_or("the second durable publish reported no durable position")?;
    assert_eq!(
        second_position,
        first_position
            .checked_add(1)
            .ok_or("durable position overflowed")?,
        "consecutive durable publishes must report consecutive store positions"
    );
    let rows = crate::durability::bridge::block_on(store.read_from("orders:0", 0, 16))??;
    assert_eq!(
        rows.len(),
        2,
        "the store must hold exactly the two publishes whose positions were reported"
    );

    // The other direction: a memory channel has no store, so it must say so
    // rather than mint a fictional position.
    let memory = order_channel(&supervisor)?;
    let memory_delivery = memory.publish_with_delivery(
        br#"{"order_id":"A3","quantity":5}"#,
        PublisherId::default(),
        None,
    )?;
    assert_eq!(
        memory_delivery.durable_position(),
        None,
        "a memory channel must not report a durable position"
    );
    supervisor.shutdown();
    Ok(())
}

/// SRV-012 R6: the matched count tells "nobody subscribed" from "subscribed
/// and taken", and a subscriber whose predicate does not match is neither.
#[test]
fn matched_count_tells_no_subscriber_from_a_taker() -> Result<(), Box<dyn Error>> {
    let supervisor = ChannelSupervisor::new()?;
    let channel = order_channel(&supervisor)?;
    let payload = br#"{"order_id":"A1","quantity":3}"#;

    let alone = channel.publish_with_delivery(payload, PublisherId::default(), None)?;
    assert_eq!((alone.matched_count(), alone.delivered_count()), (0, 0));

    let filtered = channel.subscribe_filtered(|_| false)?;
    let skipped = channel.publish_with_delivery(payload, PublisherId::default(), None)?;
    assert_eq!((skipped.matched_count(), skipped.delivered_count()), (0, 0));

    let taker = channel.subscribe()?;
    let delivery = channel.publish_with_delivery(payload, PublisherId::default(), None)?;
    assert_eq!(
        (delivery.matched_count(), delivery.delivered_count()),
        (1, 1)
    );
    drop(filtered);
    drop(taker);
    Ok(())
}

/// SRV-012 `to_stations`: a recording publish names exactly the subscribers
/// whose inbox queued the message. One whose inbox refused it (its depth cap
/// is full) is matched and counted, never recorded, and the buffer keeps its
/// allocation from one publish to the next.
#[test]
fn a_recording_publish_names_only_the_subscribers_that_queued_it() -> Result<(), Box<dyn Error>> {
    use crate::channel::{ConnectionInboxBudget, InboxInstall};

    let supervisor = ChannelSupervisor::new()?;
    let channel = order_channel(&supervisor)?;
    let payload = br#"{"order_id":"A1","quantity":3}"#;
    let taker = channel.subscribe()?;
    let refuser = channel.subscribe_with_install(InboxInstall {
        budget: ConnectionInboxBudget::new(1 << 20),
        depth_cap: 1,
        notifier: None,
        capacity: None,
    })?;
    let mut record = PublishRecord::default();
    let pids = |record: &PublishRecord| -> Vec<u64> {
        let mut pids: Vec<u64> = record.recipients().iter().map(|(pid, _)| *pid).collect();
        pids.sort_unstable();
        pids
    };

    let first =
        channel.publish_recording(payload, PublisherId::default(), None, None, &mut record)?;
    let mut both = vec![taker.subscriber_pid(), refuser.subscriber_pid()];
    both.sort_unstable();
    assert_eq!(pids(&record), both);
    assert_eq!((first.matched_count(), first.delivered_count()), (2, 2));

    let reused = record.recipients().as_ptr();
    let second =
        channel.publish_recording(payload, PublisherId::default(), None, None, &mut record)?;
    assert_eq!(
        pids(&record),
        [taker.subscriber_pid()],
        "the full inbox refused it"
    );
    assert_eq!((second.matched_count(), second.delivered_count()), (2, 1));
    assert_eq!(
        record.recipients().as_ptr(),
        reused,
        "the buffer's allocation is reused"
    );
    drop((taker, refuser));
    Ok(())
}

/// Lane brief ruling 7 (Hermes 7 Oct 03:54): a publish that lends no record
/// reads no clock, and a recording one reads it once per subscriber its
/// fan-out queued the envelope for.
/// Partner of `each_recipient_carries_the_instant_its_inbox_queued_it`: this
/// pin catches one instant shared by every recipient (one read, not two), and
/// that one catches an instant from outside the publish. Neither alone does.
#[test]
fn a_publish_with_no_record_reads_no_clock() -> Result<(), Box<dyn Error>> {
    use std::sync::atomic::Ordering;

    use crate::channel::record::CLOCK_READS;

    let supervisor = ChannelSupervisor::new()?;
    let channel = order_channel(&supervisor)?;
    let payload = br#"{"order_id":"A1","quantity":3}"#;
    let subscribers = [channel.subscribe()?, channel.subscribe()?];
    let before = CLOCK_READS.load(Ordering::Relaxed);
    for _ in 0..3 {
        channel.publish_with_delivery(payload, PublisherId::default(), None)?;
    }
    assert_eq!(
        CLOCK_READS.load(Ordering::Relaxed),
        before,
        "no record lent, no clock read"
    );
    let mut record = PublishRecord::default();
    channel.publish_recording(payload, PublisherId::default(), None, None, &mut record)?;
    assert_eq!(
        CLOCK_READS.load(Ordering::Relaxed) - before,
        2,
        "one read per queued subscriber"
    );
    drop(subscribers);
    Ok(())
}

/// Lane brief ruling 7: each recipient carries the instant its inbox queued
/// the envelope, read during the publish; a channel that is not durable
/// records no append.
/// Partner of `a_publish_with_no_record_reads_no_clock`: this pin catches an
/// instant from outside the publish, and that one catches one instant shared
/// by every recipient. Neither alone does.
#[test]
fn each_recipient_carries_the_instant_its_inbox_queued_it() -> Result<(), Box<dyn Error>> {
    let supervisor = ChannelSupervisor::new()?;
    let channel = order_channel(&supervisor)?;
    let payload = br#"{"order_id":"A1","quantity":3}"#;
    let subscribers = [channel.subscribe()?, channel.subscribe()?];
    let mut record = PublishRecord::default();
    let before = std::time::Instant::now();
    channel.publish_recording(payload, PublisherId::default(), None, None, &mut record)?;
    let after = std::time::Instant::now();
    assert_eq!(record.recipients().len(), 2);
    for (_, queued) in record.recipients() {
        assert!(before <= *queued && *queued <= after, "{queued:?}");
    }
    assert_eq!(record.appended_at(), None);
    drop(subscribers);
    Ok(())
}

/// An observer that reports each departure on a channel, so a test waits on
/// the report and never on the clock.
#[derive(Debug)]
struct DepartureReports {
    reports: std::sync::mpsc::Sender<(String, u64)>,
}

impl crate::channel::ClusterObserver for DepartureReports {
    fn on_subscribe(&self, _channel: &str, _subscriber_pid: u64) {}

    fn on_unsubscribe(&self, channel: &str, subscriber_pid: u64) {
        // A closed receiver means the test already returned; there is no one
        // left to tell.
        self.reports
            .send((channel.to_owned(), subscriber_pid))
            .unwrap_or_default();
    }

    fn on_publish(&self, _channel: &str, _envelope: &Envelope) {}
}

/// The longest a test waits for a departure report: a guard against a hang,
/// never a deadline the behaviour is measured against.
const DEPARTURE_GUARD: Duration = Duration::from_secs(60);

/// SRV-012 R6: a subscription dropped without an explicit unsubscribe (the
/// server's close and shed paths) is reported to the observer from the real
/// drop site, once per subscriber; an explicit unsubscribe is reported once
/// and its later drop adds nothing.
#[test]
fn every_dropped_subscription_reaches_the_observer_once() -> Result<(), Box<dyn Error>> {
    let supervisor = ChannelSupervisor::new()?;
    let (reports, departures) = std::sync::mpsc::channel();
    supervisor.install_observer(Arc::new(DepartureReports { reports }));
    let channel = order_channel(&supervisor)?;

    let dropped: Vec<SubscriptionHandle> = (0..3)
        .map(|_| channel.subscribe())
        .collect::<Result<_, _>>()?;
    let mut expected: Vec<u64> = dropped.iter().map(SubscriptionHandle::pid).collect();
    drop(dropped);
    let mut heard: Vec<u64> = Vec::new();
    for _ in 0..3 {
        let (name, pid) = departures.recv_timeout(DEPARTURE_GUARD)?;
        assert_eq!(name, "orders");
        heard.push(pid);
    }
    expected.sort_unstable();
    heard.sort_unstable();
    assert_eq!(heard, expected, "each dropped subscriber is reported once");

    // Explicit unsubscribe: reported by the unsubscribe itself...
    let explicit = channel.subscribe()?;
    let explicit_pid = explicit.pid();
    channel.unsubscribe(&explicit)?;
    assert_eq!(
        departures.recv_timeout(DEPARTURE_GUARD)?,
        ("orders".to_owned(), explicit_pid)
    );
    // ...and its drop adds nothing. The actor handles EXITs in order, so the
    // next report must be the sentinel's: had the explicit subscriber's exit
    // been reported again, it would have come first.
    drop(explicit);
    let sentinel = channel.subscribe()?;
    let sentinel_pid = sentinel.pid();
    drop(sentinel);
    assert_eq!(
        departures.recv_timeout(DEPARTURE_GUARD)?,
        ("orders".to_owned(), sentinel_pid)
    );
    Ok(())
}

/// A command whose wake reached an actor that then died stays on the core's
/// queue, because the queue outlives the process. The restarted actor must
/// still boot and serve the next publish: one stale command must not leave
/// every later command (Boot first) one wake behind.
///
/// The actor runs on one scheduler thread, so the held slice finishes before
/// the restarted actor's first slice. The hold is released before the
/// restart, so it never blocks the new actor.
#[test]
fn a_command_left_by_a_dead_actor_does_not_hold_the_restart_one_wake_behind()
-> 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 hold = supervisor.entry_hold();

    // P0's wake starts a slice that parks at the NIF's entry, before it pops.
    hold.arm()?;
    let first = core.enqueue_publish(
        br#"{"order_id":"P0","quantity":1}"#.to_vec(),
        PublisherId::default(),
        None,
        None,
        None,
    )?;
    hold.await_parked()?;
    // P1's wake lands in the executing slot, behind the held slice.
    let second = core.enqueue_publish(
        br#"{"order_id":"P1","quantity":1}"#.to_vec(),
        PublisherId::default(),
        None,
        None,
        None,
    )?;
    handle
        .scheduler()?
        .terminate_process(pid, ExitReason::Error);
    // The held slice pops and serves P0. At store-back the slot is dead, so
    // P1's wake goes with it and P1 is left on the queue.
    hold.release()?;

    // This publish restarts the actor; its Boot queues behind P1.
    handle.publish(br#"{"order_id":"P2","quantity":1}"#)?;

    assert!(
        core.await_publish(first).is_ok(),
        "P0 was served by the held slice before it ended"
    );
    assert!(
        core.await_publish(second).is_err(),
        "P1's actor died before serving it, so its publish failed"
    );
    let delivered: Vec<Vec<u8>> = std::iter::from_fn(|| subscription.try_next().transpose())
        .map(|envelope| envelope.map(|envelope| envelope.payload))
        .collect::<Result<_, _>>()?;
    assert_eq!(
        delivered,
        vec![
            br#"{"order_id":"P0","quantity":1}"#.to_vec(),
            br#"{"order_id":"P2","quantity":1}"#.to_vec(),
        ],
        "only the publishes that returned Ok reach the subscriber"
    );
    supervisor.shutdown();
    Ok(())
}