liminal-rs 0.14.1

//! A write the store applied but answered `OutcomeUnknown` for, in the
//! library's own durable types (classes 12, 14 and 15 of
//! `docs/design/durability/OUTCOME-UNKNOWN-WRITE-SITES-2026-10-07.md`).
//!
//! haematite answers a write whose caller stopped waiting after the shard had
//! started it with `OutcomeUnknown`: it may have been applied, so the position
//! must be read again. Each pin applies one write, answers that, and asserts
//! the caller settles it by reading the position: the write is answered as it
//! landed, and the next write succeeds.

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

use super::bridge::block_on;
use super::tests_memory_store::MemoryStore;
use super::{
    ConsumerCursor, DedupCache, DedupDecision, DurabilityError, DurableConversation, DurableStore,
    ScanCursor, ScanPage, StoredEntry,
};

/// Which kind of write the store answers `OutcomeUnknown` for.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum Write {
    Append,
    Cas,
}

/// Applies every write, and answers the first armed write of one kind to keys
/// under one prefix with `OutcomeUnknown` once it has been applied.
#[derive(Debug)]
struct UnknownOutcomeStore {
    inner: Arc<dyn DurableStore>,
    write: Write,
    prefix: String,
    armed: AtomicBool,
}

impl UnknownOutcomeStore {
    fn armed(inner: &Arc<dyn DurableStore>, write: Write, prefix: &str) -> Arc<Self> {
        Arc::new(Self {
            inner: Arc::clone(inner),
            write,
            prefix: prefix.to_owned(),
            armed: AtomicBool::new(true),
        })
    }

    fn strikes(&self, write: Write, key: &str) -> bool {
        write == self.write
            && key.starts_with(&self.prefix)
            && self.armed.swap(false, Ordering::SeqCst)
    }

    fn struck(&self) -> bool {
        !self.armed.load(Ordering::SeqCst)
    }
}

fn outcome_unknown() -> DurabilityError {
    haematite::ApiError::Storage(haematite::DatabaseError::OutcomeUnknown { pid: 1 }).into()
}

#[async_trait::async_trait]
impl DurableStore for UnknownOutcomeStore {
    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,
        stream_key: &str,
        payload: Vec<u8>,
        expected_seq: u64,
    ) -> Result<u64, DurabilityError> {
        let applied = self.inner.append(stream_key, payload, expected_seq).await?;
        if self.strikes(Write::Append, stream_key) {
            return Err(outcome_unknown());
        }
        Ok(applied)
    }

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

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

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

    async fn cas(&self, key: &str, old_value: u64, new_value: u64) -> Result<(), DurabilityError> {
        self.inner.cas(key, old_value, new_value).await?;
        if self.strikes(Write::Cas, key) {
            return Err(outcome_unknown());
        }
        Ok(())
    }

    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
    }
}

fn memory() -> Arc<dyn DurableStore> {
    Arc::new(MemoryStore::default())
}

/// Class 12: a claim that landed under an unknown outcome is the caller's
/// claim, so the message is delivered, and a retry of the key is told so.
#[test]
fn a_dedup_claim_landed_under_an_unknown_outcome_is_claimed() -> Result<(), Box<dyn Error>> {
    let inner = memory();
    let fault = UnknownOutcomeStore::armed(&inner, Write::Append, "dedup");
    let store: Arc<dyn DurableStore> = fault.clone();
    let cache = DedupCache::new(store, "dedup");
    let claimed = block_on(cache.claim_or_get("key-a", 42))?;
    assert!(fault.struck(), "the claim wrote no row");
    let claimed = claimed.map_err(|error| error.to_string());
    assert_eq!(
        claimed,
        Ok(DedupDecision::Claimed),
        "a claim the store holds was answered {claimed:?}, so its message is never delivered"
    );
    Ok(())
}

/// Class 14: a cursor checkpoint that landed under an unknown outcome moves
/// the cursor, and the next checkpoint succeeds.
#[test]
fn a_cursor_checkpoint_landed_under_an_unknown_outcome_moves_the_cursor()
-> Result<(), Box<dyn Error>> {
    let inner = memory();
    let mut cursor = ConsumerCursor::new("consumer-a", "channel-a:0");
    let fault = UnknownOutcomeStore::armed(&inner, Write::Cas, &cursor.cursor_key());
    let first = block_on(cursor.checkpoint(fault.as_ref(), 5))?;
    assert!(fault.struck(), "the checkpoint wrote no cursor");
    let next = block_on(cursor.checkpoint(fault.as_ref(), 7))?;
    let (first, next) = (
        first.map_err(|error| error.to_string()),
        next.map_err(|error| error.to_string()),
    );
    assert_eq!(
        (first, next, cursor.current_offset()),
        (Ok(()), Ok(()), 7),
        "a checkpoint the store holds left the cursor behind it"
    );
    Ok(())
}

/// Class 15: a conversation event that landed under an unknown outcome is
/// answered at its sequence, and the next event appends after it.
#[test]
fn a_conversation_event_landed_under_an_unknown_outcome_is_answered_at_its_sequence()
-> Result<(), Box<dyn Error>> {
    let inner = memory();
    let fault = UnknownOutcomeStore::armed(&inner, Write::Append, "conversation-a");
    let store: Arc<dyn DurableStore> = fault.clone();
    let mut conversation = DurableConversation::new("conversation-a", store);
    let first = block_on(conversation.record_message_received("message-1", 1))?;
    assert!(fault.struck(), "the event wrote no row");
    let next = block_on(conversation.record_processing_started("message-1"))?;
    let (first, next) = (
        first.map_err(|error| error.to_string()),
        next.map_err(|error| error.to_string()),
    );
    assert_eq!(
        (first, next),
        (Ok(0), Ok(1)),
        "an event the store holds was answered as not written"
    );
    Ok(())
}