use std::error::Error;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use beamr::process::ExitReason;
use beamr::scheduler::{ExitWatch, ExitWatchState, Scheduler};
use super::{ChannelConfig, ChannelHandle, ChannelMode, ChannelSupervisor, Schema};
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;
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:?}")));
}
if fault.scheduler.process_table().get(fault.pid).is_some() {
return Err(fault_error("exit signal preceded actor removal"));
}
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,
) {
assert_eq!(
committed,
StoreCounts {
append_all_calls: 1,
append_all_rows: 3,
append_all_bytes: 171,
flushes: 1,
..StoreCounts::default()
}
);
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());
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) {
assert_eq!(
committed,
StoreCounts {
append_calls: 1,
append_rows: 1,
append_bytes: 76,
..StoreCounts::default()
}
);
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}"#;
let before = store.snapshot();
let failure = channel
.publish_with_key(
payload,
PublisherId::new("single-probe"),
None,
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"));
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(())
}