liminal-rs 0.15.0

use beamr::process::ExitReason;
use beamr::scheduler::{ExitWatch, ExitWatchState, Scheduler};

use crate::durability::bridge::block_on;
use crate::durability::{
    CountingStore, DurabilityError, DurableStore, MessageEnvelope, ScanCursor, ScanPage,
    StoreCounts, StoredEntry, open_ephemeral,
};
use crate::envelope::PublisherId;
use crate::error::LiminalError;

use std::error::Error;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};

use super::{ChannelConfig, ChannelHandle, ChannelMode, ChannelSupervisor, Schema};

struct ActorFault {
    scheduler: Arc<Scheduler>,
    pid: u64,
    watch: ExitWatch,
}

struct BatchFaultStore {
    inner: Arc<dyn DurableStore>,
    fault: Arc<Mutex<Option<ActorFault>>>,
    exits: Arc<AtomicU64>,
}

impl BatchFaultStore {
    fn finish_fault(&self) -> Result<(), DurabilityError> {
        let fault = self.fault.lock().map_err(fault_error)?.take();
        if let Some(fault) = fault {
            fault
                .scheduler
                .terminate_process(fault.pid, ExitReason::Error);
            let outcome = fault.watch.recv().map_err(fault_error)?;
            if outcome != (fault.pid, ExitReason::Error) {
                return Err(fault_error(format!("unexpected actor exit {outcome:?}")));
            }
            // The watch fires once the exit tombstone and outcome are installed,
            // before beamr removes the pid from its table; the tombstone is what
            // makes every later send to the actor refuse, so that is the state
            // read here (R3b review row 14; design R3B-ROWS-13-14-2026-10-07.md).
            if fault.scheduler.peek_exit_reason(fault.pid) != Some(ExitReason::Error) {
                return Err(fault_error(
                    "the exit watch fired before the outcome was installed",
                ));
            }
            self.exits.fetch_add(1, Ordering::Release);
        }
        Ok(())
    }
}

impl std::fmt::Debug for BatchFaultStore {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("BatchFaultStore")
            .finish_non_exhaustive()
    }
}

fn fault_error(error: impl std::fmt::Display) -> DurabilityError {
    DurabilityError::ConfigError(format!("batch actor fault boundary failed: {error}"))
}

#[async_trait::async_trait]
impl DurableStore for BatchFaultStore {
    async fn apply_routed_batch(
        &self,
        _: &str,
        _: Vec<crate::durability::RoutedBatchItem>,
    ) -> Result<(), crate::durability::DurabilityError> {
        Err(crate::durability::DurabilityError::BatchUnsupported {
            store: std::any::type_name::<Self>(),
        })
    }

    async fn get_routed(
        &self,
        _: &str,
        _: &[u8],
    ) -> Result<Option<Vec<u8>>, crate::durability::DurabilityError> {
        Err(crate::durability::DurabilityError::BatchUnsupported {
            store: std::any::type_name::<Self>(),
        })
    }

    async fn append(
        &self,
        key: &str,
        payload: Vec<u8>,
        expected: u64,
    ) -> Result<u64, DurabilityError> {
        let assigned = self.inner.append(key, payload, expected).await?;
        self.finish_fault()?;
        Ok(assigned)
    }

    async fn append_all(
        &self,
        key: &str,
        payloads: Vec<Vec<u8>>,
        expected: u64,
    ) -> Result<u64, DurabilityError> {
        self.inner.append_all(key, payloads, expected).await
    }

    async fn read_from(
        &self,
        key: &str,
        offset: u64,
        limit: usize,
    ) -> Result<Vec<StoredEntry>, DurabilityError> {
        self.inner.read_from(key, offset, limit).await
    }

    async fn read_at(
        &self,
        key: &str,
        sequence: u64,
    ) -> Result<Option<StoredEntry>, DurabilityError> {
        self.inner.read_at(key, sequence).await
    }

    async fn cas(&self, key: &str, old: u64, new: u64) -> Result<(), DurabilityError> {
        self.inner.cas(key, old, new).await
    }

    async fn read_stream_next_seq(&self, stream_key: &str) -> Result<Option<u64>, DurabilityError> {
        self.inner.read_stream_next_seq(stream_key).await
    }

    async fn read_value(&self, key: &str) -> Result<Option<u64>, DurabilityError> {
        self.inner.read_value(key).await
    }

    async fn scan_page(
        &self,
        prefix: &str,
        cursor: Option<ScanCursor>,
        limit: usize,
    ) -> Result<ScanPage, DurabilityError> {
        self.inner.scan_page(prefix, cursor, limit).await
    }

    async fn flush(&self) -> Result<(), DurabilityError> {
        self.inner.flush().await?;
        self.finish_fault()
    }
}

struct OwnedSupervisor(ChannelSupervisor);

fn assert_no_batch_reappend(counts: StoreCounts) {
    assert_eq!(
        counts.append_all_calls, 0,
        "retained batch was appended again"
    );
}

fn assert_count_pins(
    committed: StoreCounts,
    retry: StoreCounts,
    refusal: StoreCounts,
    duplicate_control: StoreCounts,
) {
    // 171, derived from `MessageEnvelope::serialize` (durability/channel/codec.rs):
    // three rows of magic 4 + payload `{"generation":N}` (8 + 16) + causal
    // ABSENT 1 + timestamp 8 + "batch-probe" (8 + 11) + key ABSENT 1 = 57.
    assert_eq!(
        committed,
        StoreCounts {
            append_all_calls: 1,
            append_all_rows: 3,
            append_all_bytes: 171,
            flushes: 1,
            ..StoreCounts::default()
        }
    );
    // 41, derived the same way: the following publish of `{}` is magic 4 +
    // payload (8 + 2) + causal ABSENT 1 + timestamp 8 + "following" (8 + 9) +
    // key ABSENT 1.
    assert_eq!(
        retry,
        StoreCounts {
            append_calls: 1,
            append_rows: 1,
            append_bytes: 41,
            ..StoreCounts::default()
        }
    );
    assert_no_batch_reappend(retry);
    assert_eq!(refusal, StoreCounts::default());
    // 171: the committed batch's stored rows, appended again byte for byte.
    assert_eq!(
        duplicate_control,
        StoreCounts {
            append_all_calls: 1,
            append_all_rows: 3,
            append_all_bytes: 171,
            ..StoreCounts::default()
        }
    );
}

fn assert_single_count_pins(committed: StoreCounts, retry: StoreCounts) {
    // Derived from `MessageEnvelope::serialize` (durability/channel/codec.rs):
    // magic 4 + payload (8 + 16) + causal context + timestamp 8 +
    // "single-probe" (8 + 12) + key PRESENT 1 + "single-key" (8 + 10). The
    // causal context is 47 bytes now that the publish carries a parent
    // (PRESENT 1, parent PRESENT 1, length 8, hyphenated UUID 36, clock
    // ABSENT 1): 75 + 47 = 122. Before R3b row 13 this publish passed no
    // parent, the context was ABSENT 1, and the pin was 76.
    assert_eq!(
        committed,
        StoreCounts {
            append_calls: 1,
            append_rows: 1,
            append_bytes: 122,
            ..StoreCounts::default()
        }
    );
    // 41: the following publish, as in `assert_count_pins`.
    assert_eq!(
        retry,
        StoreCounts {
            append_calls: 1,
            append_rows: 1,
            append_bytes: 41,
            ..StoreCounts::default()
        }
    );
}

fn assert_fixture_counts(exits: &AtomicU64, fixture_builds: &AtomicU64) {
    assert_eq!(exits.load(Ordering::Acquire), 2);
    assert_eq!(fixture_builds.load(Ordering::Relaxed), 1);
    eprintln!(
        "RECEIPT_FIXTURE_COUNTS exits={} fixture_builds={}",
        exits.load(Ordering::Acquire),
        fixture_builds.load(Ordering::Relaxed)
    );
}

impl Drop for OwnedSupervisor {
    fn drop(&mut self) {
        self.0.shutdown();
    }
}

fn assert_single_receipt_bytes(
    store: &Arc<CountingStore<BatchFaultStore>>,
    fault: &Arc<Mutex<Option<ActorFault>>>,
    supervisor: &ChannelSupervisor,
) -> Result<(), Box<dyn Error>> {
    let config = ChannelConfig::new(
        "single.actor.failure".to_owned(),
        Schema::new(serde_json::json!({"type": "object"}))?,
        ChannelMode::Durable,
    );
    let channel =
        ChannelHandle::new_durable_with_supervisor(config, store.clone(), supervisor.clone())?;
    let subscription = channel.subscribe()?;
    let pid = channel.actor_pid()?;
    let scheduler = channel.scheduler()?;
    let watch = match scheduler.watch_exit(pid) {
        ExitWatchState::Live(watch) => watch,
        state => return Err(format!("actor was not live at injection: {state:?}").into()),
    };
    *fault.lock().map_err(fault_error)? = Some(ActorFault {
        scheduler,
        pid,
        watch,
    });
    let payload = br#"{"generation":4}"#;
    // Row 13 of the R3b review: the receipt path keeps the parent too.
    let parent = crate::causal::MessageId::new();
    let before = store.snapshot();
    let failure = channel
        .publish_with_key(
            payload,
            PublisherId::new("single-probe"),
            Some(crate::causal::CausalContext::from_parent_chain(vec![
                parent,
            ])),
            Some("single-key"),
            None,
        )
        .err()
        .ok_or("dead actor unexpectedly accepted the committed row")?;
    let committed = store.snapshot().since(&before);
    let LiminalError::PublishUnresolved { append } = failure else {
        return Err(format!("committed actor refusal lost receipt: {failure}").into());
    };
    assert_eq!(append.channel(), "single.actor.failure");
    assert_eq!(append.partition(), 0);
    assert_eq!(append.sequence(), 0);
    assert_eq!(append.idempotency_key(), Some("single-key"));
    assert!(
        append
            .cause()
            .starts_with("row committed; live delivery failed:")
    );
    assert!(subscription.try_next()?.is_none());
    let row = block_on(store.read_at("single.actor.failure:0", 0))??
        .ok_or("committed single row absent")?;
    assert_eq!(append.payload(), row.payload);
    let envelope = MessageEnvelope::deserialize(&row.payload)?;
    assert_eq!(envelope.payload, payload);
    assert_eq!(envelope.publisher_id, "single-probe");
    assert_eq!(envelope.idempotency_key.as_deref(), Some("single-key"));
    assert_eq!(
        envelope
            .causal_context
            .and_then(|context| context.parent_id),
        Some(parent.as_uuid().to_string()),
        "the committed row of a failed live delivery lost its parent"
    );
    let before = store.snapshot();
    let next = channel.publish_with_delivery(b"{}", PublisherId::new("following"), None)?;
    let retry = store.snapshot().since(&before);
    assert_eq!(next.durable_position(), Some(1));
    for expected in [payload.as_slice(), b"{}".as_slice()] {
        assert_eq!(
            subscription
                .try_next()?
                .ok_or("single delivery absent")?
                .payload,
            expected
        );
    }
    assert!(subscription.try_next()?.is_none());
    eprintln!(
        "SINGLE_RECEIPT_COUNTS committed={committed:?} retry={retry:?} receipt_bytes={}",
        append.payload().len()
    );
    assert_single_count_pins(committed, retry);
    Ok(())
}

#[test]
fn batch_actor_enqueue_failure_keeps_committed_receipt_and_recovers() -> Result<(), Box<dyn Error>>
{
    let fixture_builds = AtomicU64::new(0);
    let inner: Arc<dyn DurableStore> = Arc::new(open_ephemeral(1)?);
    fixture_builds.fetch_add(1, Ordering::Relaxed);
    let fault = Arc::new(Mutex::new(None));
    let exits = Arc::new(AtomicU64::new(0));
    let store = Arc::new(CountingStore::new(BatchFaultStore {
        inner,
        fault: Arc::clone(&fault),
        exits: Arc::clone(&exits),
    }));
    let supervisor = OwnedSupervisor(ChannelSupervisor::new()?);
    let config = ChannelConfig::new(
        "batch.actor.failure".to_owned(),
        Schema::new(serde_json::json!({"type": "object"}))?,
        ChannelMode::Durable,
    );
    let channel =
        ChannelHandle::new_durable_with_supervisor(config, store.clone(), supervisor.0.clone())?;
    let subscription = channel.subscribe()?;
    let pid = channel.actor_pid()?;
    let scheduler = channel.scheduler()?;
    let watch = match scheduler.watch_exit(pid) {
        ExitWatchState::Live(watch) => watch,
        state => return Err(format!("actor was not live at injection: {state:?}").into()),
    };
    *fault.lock().map_err(fault_error)? = Some(ActorFault {
        scheduler,
        pid,
        watch,
    });
    let payloads = vec![
        br#"{"generation":1}"#.to_vec(),
        br#"{"generation":2}"#.to_vec(),
        br#"{"generation":3}"#.to_vec(),
    ];
    let publisher = PublisherId::new("batch-probe");
    let before = store.snapshot();
    let failure = channel
        .publish_batch_with_delivery(&payloads, &publisher, None)
        .err()
        .ok_or("dead actor unexpectedly accepted the committed batch")?;
    let committed = store.snapshot().since(&before);
    let LiminalError::PublishUnresolved { append } = failure else {
        return Err(format!("committed actor refusal lost receipt: {failure}").into());
    };
    assert_eq!(append.channel(), "batch.actor.failure");
    assert_eq!(append.partition(), 0);
    assert_eq!(append.sequence(), 0);
    assert!(
        append
            .cause()
            .starts_with("batch committed; live enqueue failed:")
    );
    assert!(subscription.try_next()?.is_none());
    let stream = "batch.actor.failure:0";
    let mut duplicate_payloads = Vec::with_capacity(payloads.len());
    for (sequence, payload) in payloads.iter().enumerate() {
        let row = block_on(store.read_at(stream, u64::try_from(sequence)?))??
            .ok_or("committed batch row absent")?;
        let envelope = MessageEnvelope::deserialize(&row.payload)?;
        assert_eq!(envelope.payload, *payload);
        assert_eq!(envelope.publisher_id, "batch-probe");
        duplicate_payloads.push(row.payload);
    }
    let before = store.snapshot();
    let next = channel.publish_with_delivery(b"{}", PublisherId::new("following"), None)?;
    let retry = store.snapshot().since(&before);
    assert_eq!(next.durable_position(), Some(3));
    for payload in payloads.iter().chain(std::iter::once(&b"{}".to_vec())) {
        assert_eq!(
            subscription
                .try_next()?
                .ok_or("retained or following delivery absent")?
                .payload,
            *payload
        );
    }
    assert!(subscription.try_next()?.is_none());
    let before = store.snapshot();
    assert!(
        channel
            .publish_batch_with_delivery(&[b"invalid JSON".to_vec()], &publisher, None)
            .is_err()
    );
    let refusal = store.snapshot().since(&before);
    let before = store.snapshot();
    let accepted = block_on(store.append_all(stream, duplicate_payloads, 4))??;
    let duplicate_control = store.snapshot().since(&before);
    eprintln!(
        "BATCH_FAULT_COUNTS committed={committed:?} retry={retry:?} refusal={refusal:?} duplicate_control={duplicate_control:?} accepted_control_head={accepted} exits={} fixture_builds={}",
        exits.load(Ordering::Acquire),
        fixture_builds.load(Ordering::Relaxed)
    );
    assert_count_pins(committed, retry, refusal, duplicate_control);
    assert_eq!(accepted, 7);
    assert_eq!(exits.load(Ordering::Acquire), 1);
    assert_single_receipt_bytes(&store, &fault, &supervisor.0)?;
    assert_fixture_counts(&exits, &fixture_builds);
    Ok(())
}