liminal-server 0.25.4

//! A keyed publish to a durable channel claims its key with its evidence before
//! the publish, holds the claim across an unresolved append, and a same-key
//! retry or the channel's next keyed admission looks the row up by its key, so
//! the stream never holds a second row for the key.
//!
//! [`OrdersFaultStore`] faults the durable channel "orders" by path: its append
//! switches act on `orders:0`, its resolution read fault refuses the reads of
//! `orders:0` (point and page) after a set number pass, so the reads
//! registration's recovery makes can pass while the key lookup's are refused,
//! and the `read_from` reads and the rows they answer are counted. Appends to
//! one named stream can be refused, or held at a gate until the test opens it.

use std::error::Error;
use std::io::Write;
use std::path::Path;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::mpsc::{Receiver, SyncSender};
use std::sync::{Arc, Mutex, MutexGuard};

use haematite::{Database, DatabaseConfig, EventStore};
use liminal::durability::{
    ClaimEvidence, DedupCache, DedupDecision, DedupEntry, DurabilityError, DurableStore,
    HaematiteStore, ScanCursor, ScanPage, StoredEntry, open_ephemeral,
};
use liminal::protocol::{CausalContext, MessageEnvelope, SchemaId};

use super::services::{ConnectionServices, LiminalConnectionServices, PublishOutcome};
use crate::ServerError;
use crate::config::types::{ChannelDef, ServerConfig};

pub(super) const STREAM: &str = "orders:0";
pub(super) const DEDUP_NAMESPACE: &str = "liminal:delivery-dedup";

pub(super) type TestResult = Result<(), Box<dyn Error>>;

/// A row of `orders:0`: its envelope's payload and idempotency key.
pub(super) type OrderRow = (Vec<u8>, Option<String>);

/// How an `orders:0` append fails while its switch is armed.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(super) enum AppendFault {
    /// The row is stored, then the append answers a store error.
    CommitThenError,
    /// The append answers a store error and stores nothing.
    ErrorBeforeCommit,
}

#[derive(Debug, Default)]
struct Switches {
    append: Option<AppendFault>,
    /// `orders:0` reads still let through before every later one is refused;
    /// `None` refuses nothing.
    resolution_reads: Option<usize>,
    claims: bool,
    /// A stream whose appends are refused while named.
    refused_stream: Option<String>,
}

/// A gate on the next append to `stream`: the append reports that it arrived
/// and waits until the test opens the gate.
#[derive(Debug)]
pub(super) struct AppendGate {
    stream: String,
    arrived: SyncSender<()>,
    open: Receiver<()>,
}

/// The "orders" fault double over an in-memory store.
#[derive(Debug)]
pub(super) struct OrdersFaultStore {
    inner: Arc<dyn DurableStore>,
    switches: Mutex<Switches>,
    order_appends: AtomicUsize,
    order_reads_from: AtomicUsize,
    order_rows_read: AtomicUsize,
    refused_appends: AtomicUsize,
    gate: Mutex<Option<AppendGate>>,
    append_streams: Mutex<Vec<String>>,
}

fn injected() -> DurabilityError {
    DurabilityError::StoreError(haematite::ApiError::Storage(
        haematite::DatabaseError::IoError(std::io::Error::other("injected")),
    ))
}

/// A failure of the double itself, named: a lock a panicking thread poisoned
/// or a gate whose other end the test dropped. It fails the operation rather
/// than carrying on with recovered state.
fn double_failure(what: &str) -> DurabilityError {
    DurabilityError::StoreError(haematite::ApiError::Storage(
        haematite::DatabaseError::IoError(std::io::Error::other(format!(
            "orders fault double: {what}"
        ))),
    ))
}

impl OrdersFaultStore {
    pub(super) fn new() -> Result<Arc<Self>, Box<dyn Error>> {
        Ok(Self::over(Arc::new(open_ephemeral(1)?)))
    }

    /// The double over the single-shard database in `data_dir`, created there
    /// when `create` and opened as it stands otherwise.
    fn on_disk(data_dir: &Path, create: bool) -> Result<Arc<Self>, Box<dyn Error>> {
        let database = if create {
            Database::create(
                DatabaseConfig {
                    data_dir: data_dir.to_path_buf(),
                    shard_count: 1,
                    distributed: None,
                    executor_threads: None,
                    node_cache_budget: Some(haematite::NodeCacheBudget::Unlimited),
                    retention: haematite::Retention::KeepAll,
                },
                haematite::RuntimeOptions {
                    long_wait_interval: std::time::Duration::from_secs(5),
                    reclaim_start: haematite::ReclaimStart::AtOpen,
                },
            )?
        } else {
            Database::open(
                data_dir,
                haematite::RuntimeOptions {
                    long_wait_interval: std::time::Duration::from_secs(5),
                    reclaim_start: haematite::ReclaimStart::AtOpen,
                },
            )?
        };
        liminal::durability::record_store_opened();
        Ok(Self::over(Arc::new(HaematiteStore::new(Arc::new(
            EventStore::new(database),
        )))))
    }

    fn over(inner: Arc<dyn DurableStore>) -> Arc<Self> {
        Arc::new(Self {
            inner,
            switches: Mutex::new(Switches::default()),
            order_appends: AtomicUsize::new(0),
            order_reads_from: AtomicUsize::new(0),
            order_rows_read: AtomicUsize::new(0),
            refused_appends: AtomicUsize::new(0),
            gate: Mutex::new(None),
            append_streams: Mutex::new(Vec::new()),
        })
    }

    fn switches(&self) -> Result<MutexGuard<'_, Switches>, DurabilityError> {
        self.switches
            .lock()
            .map_err(|_| double_failure("the switches lock was poisoned"))
    }

    pub(super) fn arm_append(&self, fault: AppendFault) -> Result<(), DurabilityError> {
        self.switches()?.append = Some(fault);
        Ok(())
    }

    pub(super) fn lift_append(&self) -> Result<(), DurabilityError> {
        self.switches()?.append = None;
        Ok(())
    }

    /// Refuses every read of `orders:0` while armed.
    pub(super) fn arm_resolution_reads(&self, armed: bool) -> Result<(), DurabilityError> {
        self.switches()?.resolution_reads = armed.then_some(0);
        Ok(())
    }

    /// Lets the next `passing` reads of `orders:0` through, then refuses
    /// every later one until disarmed.
    pub(super) fn arm_resolution_reads_after(&self, passing: usize) -> Result<(), DurabilityError> {
        self.switches()?.resolution_reads = Some(passing);
        Ok(())
    }

    /// Whether this read of `orders:0` is refused, spending one passing read.
    fn refuses_order_read(&self) -> Result<bool, DurabilityError> {
        let mut switches = self.switches()?;
        Ok(match switches.resolution_reads {
            None => false,
            Some(0) => true,
            Some(passing) => {
                switches.resolution_reads = Some(passing - 1);
                false
            }
        })
    }

    /// Refuses every append to the durable dedup namespace while armed.
    pub(super) fn arm_claims(&self, armed: bool) -> Result<(), DurabilityError> {
        self.switches()?.claims = armed;
        Ok(())
    }

    /// Refuses every append to `stream` while named; `None` lifts it.
    pub(super) fn refuse_stream(&self, stream: Option<String>) -> Result<(), DurabilityError> {
        self.switches()?.refused_stream = stream;
        Ok(())
    }

    /// Appends the named stream refused.
    pub(super) fn refused_appends(&self) -> usize {
        self.refused_appends.load(Ordering::SeqCst)
    }

    /// Holds the next append to `stream` at a gate. Answers the receiver
    /// that hears the append arrive and the sender that opens the gate.
    pub(super) fn gate_next_append(
        &self,
        stream: String,
    ) -> Result<(Receiver<()>, SyncSender<()>), DurabilityError> {
        let (arrived, heard) = std::sync::mpsc::sync_channel(1);
        let (opener, open) = std::sync::mpsc::sync_channel(1);
        *self.gate_slot()? = Some(AppendGate {
            stream,
            arrived,
            open,
        });
        Ok((heard, opener))
    }

    fn gate_slot(&self) -> Result<MutexGuard<'_, Option<AppendGate>>, DurabilityError> {
        self.gate
            .lock()
            .map_err(|_| double_failure("the append gate lock was poisoned"))
    }

    /// Waits at the gate when this append is the one it holds. A test that
    /// dropped either end of the gate fails the append by name.
    fn pass_gate(&self, stream_key: &str) -> Result<(), DurabilityError> {
        let gate = {
            let mut slot = self.gate_slot()?;
            if slot.as_ref().is_some_and(|gate| gate.stream == stream_key) {
                slot.take()
            } else {
                None
            }
        };
        if let Some(gate) = gate {
            gate.arrived
                .send(())
                .map_err(|_| double_failure("the test stopped listening for the gated append"))?;
            gate.open
                .recv()
                .map_err(|_| double_failure("the test dropped the gate without opening it"))?;
        }
        Ok(())
    }

    /// Rows the reads of `orders:0` have answered.
    pub(super) fn order_rows_read(&self) -> usize {
        self.order_rows_read.load(Ordering::SeqCst)
    }

    pub(super) fn order_appends(&self) -> usize {
        self.order_appends.load(Ordering::SeqCst)
    }

    pub(super) fn order_reads_from(&self) -> usize {
        self.order_reads_from.load(Ordering::SeqCst)
    }

    /// The stream keys appended to, in order.
    pub(super) fn append_streams(&self) -> Result<Vec<String>, DurabilityError> {
        Ok(self
            .append_streams
            .lock()
            .map_err(|_| double_failure("the append-streams lock was poisoned"))?
            .clone())
    }

    /// The rows of `orders:0`.
    pub(super) fn order_rows(&self) -> Result<Vec<OrderRow>, Box<dyn Error>> {
        let rows = liminal::durability::bridge::block_on(self.inner.read_from(STREAM, 0, 64))??;
        rows.iter()
            .map(|row| {
                let envelope = liminal::durability::MessageEnvelope::deserialize(&row.payload)?;
                Ok((envelope.payload, envelope.idempotency_key))
            })
            .collect()
    }

    /// The durable delivery dedup cache over the same store.
    pub(super) fn dedup(self: &Arc<Self>) -> DedupCache {
        DedupCache::new(Arc::clone(self) as Arc<dyn DurableStore>, DEDUP_NAMESPACE)
    }
}

#[async_trait::async_trait]
impl DurableStore for OrdersFaultStore {
    async fn apply_routed_batch(
        &self,
        route_key: &str,
        items: Vec<liminal::durability::RoutedBatchItem>,
    ) -> Result<(), liminal::durability::DurabilityError> {
        self.inner.apply_routed_batch(route_key, items).await
    }

    async fn get_routed(
        &self,
        route_key: &str,
        key: &[u8],
    ) -> Result<Option<Vec<u8>>, liminal::durability::DurabilityError> {
        self.inner.get_routed(route_key, key).await
    }

    async fn append(
        &self,
        stream_key: &str,
        payload: Vec<u8>,
        expected_seq: u64,
    ) -> Result<u64, DurabilityError> {
        self.append_streams
            .lock()
            .map_err(|_| double_failure("the append-streams lock was poisoned"))?
            .push(stream_key.to_owned());
        if stream_key.starts_with(DEDUP_NAMESPACE) && self.switches()?.claims {
            return Err(injected());
        }
        if self.switches()?.refused_stream.as_deref() == Some(stream_key) {
            self.refused_appends.fetch_add(1, Ordering::SeqCst);
            return Err(injected());
        }
        self.pass_gate(stream_key)?;
        if stream_key != STREAM {
            return self.inner.append(stream_key, payload, expected_seq).await;
        }
        self.order_appends.fetch_add(1, Ordering::SeqCst);
        let fault = self.switches()?.append;
        match fault {
            Some(AppendFault::ErrorBeforeCommit) => Err(injected()),
            Some(AppendFault::CommitThenError) => {
                self.inner.append(stream_key, payload, expected_seq).await?;
                Err(injected())
            }
            None => self.inner.append(stream_key, payload, expected_seq).await,
        }
    }

    async fn append_all(
        &self,
        stream_key: &str,
        payloads: Vec<Vec<u8>>,
        expected_seq: u64,
    ) -> Result<u64, DurabilityError> {
        self.append_streams
            .lock()
            .map_err(|_| double_failure("the append-streams lock was poisoned"))?
            .push(stream_key.to_owned());
        if stream_key.starts_with(DEDUP_NAMESPACE) && self.switches()?.claims {
            return Err(injected());
        }
        if self.switches()?.refused_stream.as_deref() == Some(stream_key) {
            self.refused_appends.fetch_add(1, Ordering::SeqCst);
            return Err(injected());
        }
        self.pass_gate(stream_key)?;
        if stream_key != STREAM {
            return self
                .inner
                .append_all(stream_key, payloads, expected_seq)
                .await;
        }
        self.order_appends.fetch_add(1, Ordering::SeqCst);
        let fault = self.switches()?.append;
        match fault {
            Some(AppendFault::ErrorBeforeCommit) => Err(injected()),
            Some(AppendFault::CommitThenError) => {
                self.inner
                    .append_all(stream_key, payloads, expected_seq)
                    .await?;
                Err(injected())
            }
            None => {
                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> {
        if stream_key == STREAM {
            self.order_reads_from.fetch_add(1, Ordering::SeqCst);
            if self.refuses_order_read()? {
                return Err(injected());
            }
            let rows = self.inner.read_from(stream_key, offset, limit).await?;
            self.order_rows_read.fetch_add(rows.len(), Ordering::SeqCst);
            return Ok(rows);
        }
        self.inner.read_from(stream_key, offset, limit).await
    }

    async fn read_at(
        &self,
        stream_key: &str,
        sequence: u64,
    ) -> Result<Option<StoredEntry>, DurabilityError> {
        if stream_key == STREAM && self.refuses_order_read()? {
            return Err(injected());
        }
        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
    }

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

/// One durable channel, "orders".
pub(super) fn orders_config() -> Result<ServerConfig, Box<dyn Error>> {
    Ok(ServerConfig {
        listen_address: "127.0.0.1:0".parse()?,
        extra_listen_addresses: Vec::new(),
        health_listen_address: "127.0.0.1:0".parse()?,
        drain_timeout_ms: 30_000,
        channels: vec![ChannelDef {
            name: "orders".to_owned(),
            schema_ref: None,
            durable: true,
            loaded_schema: None,
        }],
        routing_rules: Vec::new(),
        persistence_path: None,
        cluster: None,
        auth: None,
        services: crate::config::types::ServicesConfig::default(),
        limits: crate::config::types::LimitsConfig::default(),
        participant: None,
        websocket: None,
    })
}

pub(super) fn services(
    store: &Arc<OrdersFaultStore>,
) -> Result<LiminalConnectionServices, ServerError> {
    let services = LiminalConnectionServices::from_config_with_store(
        &orders_config().map_err(|error| ServerError::ListenerAccept {
            message: error.to_string(),
        })?,
        Arc::clone(store) as Arc<dyn DurableStore>,
    )?;
    // What the listener's thread runs once it is bound.
    services.ensure_claims_resolved()?;
    Ok(services)
}

pub(super) fn envelope(payload: &[u8]) -> MessageEnvelope {
    MessageEnvelope::new(
        SchemaId::new([0_u8; SchemaId::WIRE_LEN]),
        CausalContext::independent(),
        payload.to_vec(),
    )
}

pub(super) fn publish(
    services: &LiminalConnectionServices,
    payload: &[u8],
    key: &str,
) -> Result<PublishOutcome, ServerError> {
    services.publish(7, "orders", &envelope(payload), Some(key))
}

/// Publishes `payload` under `key` with `fault` armed and lifts it, answering
/// the unresolved publish's sequence.
pub(super) fn unresolved_publish(
    store: &OrdersFaultStore,
    services: &LiminalConnectionServices,
    fault: AppendFault,
    payload: &[u8],
    key: &str,
) -> Result<Option<u64>, Box<dyn Error>> {
    store.arm_append(fault)?;
    let outcome = publish(services, payload, key);
    store.lift_append()?;
    match outcome {
        Err(ServerError::PublishUnresolved {
            channel,
            idempotency_key,
            sequence,
            ..
        }) if channel == "orders" && idempotency_key.as_deref() == Some(key) => Ok(sequence),
        other => Err(format!("expected an unresolved publish, got {other:?}").into()),
    }
}

/// Runs `body` with this thread's INFO-and-above log events in a buffer.
pub(super) fn captured_log<T>(body: impl FnOnce() -> T) -> Result<(T, String), Box<dyn Error>> {
    let buffer = Arc::new(Mutex::new(Vec::new()));
    let writer = Arc::clone(&buffer);
    let subscriber = tracing_subscriber::fmt()
        .with_writer(move || CaptureWriter(Arc::clone(&writer)))
        .with_max_level(tracing::Level::INFO)
        .with_ansi(false)
        .finish();
    let value = tracing::subscriber::with_default(subscriber, body);
    let bytes = buffer
        .lock()
        .map_err(|_| "the captured log's buffer lock was poisoned")?
        .clone();
    Ok((value, String::from_utf8_lossy(&bytes).into_owned()))
}

/// The captured lines carrying `message`, asserting each names `fields`.
pub(super) fn events<'log>(log: &'log str, message: &str, fields: &[&str]) -> Vec<&'log str> {
    let lines: Vec<&str> = log.lines().filter(|line| line.contains(message)).collect();
    for line in &lines {
        for field in fields {
            assert!(line.contains(field), "{line} lacks {field}");
        }
    }
    lines
}

struct CaptureWriter(Arc<Mutex<Vec<u8>>>);

impl Write for CaptureWriter {
    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
        self.0
            .lock()
            .map_err(|_| std::io::Error::other("the captured log's buffer lock was poisoned"))?
            .extend_from_slice(buf);
        Ok(buf.len())
    }

    fn flush(&mut self) -> std::io::Result<()> {
        Ok(())
    }
}

#[test]
fn the_claim_and_its_evidence_are_written_before_the_channel_append() -> TestResult {
    let store = OrdersFaultStore::new()?;
    let services = services(&store)?;
    publish(&services, br#"{"row":"c"}"#, "k1")?;
    let dedup = store.dedup();
    let claim_stream = dedup.stream_key_for("k1");
    let streams = store.append_streams()?;
    let claim_at = streams.iter().position(|stream| *stream == claim_stream);
    let row_at = streams.iter().position(|stream| stream == STREAM);
    assert!(matches!((claim_at, row_at), (Some(claim), Some(row)) if claim < row));
    let claims =
        liminal::durability::bridge::block_on(store.inner.read_from(&claim_stream, 0, 8))??;
    let first = claims.first().ok_or("no claim was written")?;
    assert!(first.payload.starts_with(b"LDE2"));
    assert_eq!(
        DedupEntry::deserialize(&first.payload)?.evidence_channel(),
        Some("orders")
    );
    Ok(())
}

#[test]
fn an_unresolved_publish_keeps_its_evidenced_claim_in_flight() -> TestResult {
    let store = OrdersFaultStore::new()?;
    let services = services(&store)?;
    unresolved_publish(&store, &services, AppendFault::CommitThenError, b"{}", "k1")?;
    let dedup = store.dedup();
    assert_eq!(
        liminal::durability::bridge::block_on(dedup.lookup("k1"))??,
        Some(DedupDecision::InFlight)
    );
    assert_eq!(
        by_key(&liminal::durability::bridge::block_on(
            dedup.in_flight_claims()
        )??),
        [("k1".to_owned(), Some(ClaimEvidence::new("orders", vec![0])))]
    );
    Ok(())
}

#[test]
fn a_failed_claim_write_publishes_nothing_and_the_retry_publishes_once() -> TestResult {
    let store = OrdersFaultStore::new()?;
    let services = services(&store)?;
    store.arm_claims(true)?;
    assert!(publish(&services, br#"{"row":"c"}"#, "k1").is_err());
    assert_eq!(store.order_appends(), 0);
    store.arm_claims(false)?;
    publish(&services, br#"{"row":"c"}"#, "k1")?;
    assert_eq!(store.order_rows()?.len(), 1);
    Ok(())
}

/// A fresh keyed row on a durable channel is appended at a store-assigned
/// position the publish is told, so its outcome carries that position: the
/// same position a same-key retry after an unresolved publish answers with.
#[test]
fn a_fresh_admitted_row_carries_its_durable_position() -> TestResult {
    let store = OrdersFaultStore::new()?;
    let services = services(&store)?;
    let outcome = publish(&services, br#"{"row":"c"}"#, "k1")?;
    assert_eq!(outcome.durable_position, Some(0));
    Ok(())
}

#[test]
fn same_key_retry_after_committed_unresolved_answers_the_original_position() -> TestResult {
    let store = OrdersFaultStore::new()?;
    let services = services(&store)?;
    let sequence = unresolved_publish(
        &store,
        &services,
        AppendFault::CommitThenError,
        br#"{"row":"c"}"#,
        "k1",
    )?;
    assert_eq!(sequence, Some(0));
    let outcome = publish(&services, br#"{"row":"c"}"#, "k1")?;
    assert_eq!(outcome.durable_position, Some(0));
    assert!(!outcome.delivered);
    assert_eq!(store.order_rows()?.len(), 1);
    Ok(())
}

#[test]
fn same_key_retry_after_absent_unresolved_publishes() -> TestResult {
    let store = OrdersFaultStore::new()?;
    let services = services(&store)?;
    let sequence = unresolved_publish(
        &store,
        &services,
        AppendFault::ErrorBeforeCommit,
        br#"{"row":"c"}"#,
        "k1",
    )?;
    assert_eq!(sequence, Some(0));
    publish(&services, br#"{"row":"c"}"#, "k1")?;
    assert_eq!(
        store.order_rows()?,
        [(br#"{"row":"c"}"#.to_vec(), Some("k1".to_owned()))]
    );
    Ok(())
}

#[test]
fn same_key_retry_while_unreadable_answers_unresolved() -> TestResult {
    let store = OrdersFaultStore::new()?;
    let services = services(&store)?;
    unresolved_publish(
        &store,
        &services,
        AppendFault::CommitThenError,
        br#"{"row":"c"}"#,
        "k1",
    )?;
    store.arm_resolution_reads(true)?;
    let appends = store.order_appends();
    assert!(matches!(
        publish(&services, br#"{"row":"c"}"#, "k1"),
        Err(ServerError::PublishUnresolved { ref idempotency_key, .. })
            if idempotency_key.as_deref() == Some("k1")
    ));
    assert_eq!(store.order_appends(), appends);
    store.arm_resolution_reads(false)?;
    let outcome = publish(&services, br#"{"row":"c"}"#, "k1")?;
    assert_eq!(outcome.durable_position, Some(0));
    assert_eq!(store.order_rows()?.len(), 1);
    Ok(())
}

#[test]
fn unresolved_key_resolves_at_the_next_admission_on_the_channel() -> TestResult {
    let store = OrdersFaultStore::new()?;
    let services = services(&store)?;
    unresolved_publish(
        &store,
        &services,
        AppendFault::CommitThenError,
        br#"{"row":"c"}"#,
        "k1",
    )?;
    let (admitted, log) = captured_log(|| publish(&services, br#"{"row":"e"}"#, "k2"))?;
    admitted?;
    let resolved = events(
        &log,
        "unresolved publish resolved",
        &[
            "channel=\"orders\"",
            "idempotency_key=\"k1\"",
            "outcome=\"committed\"",
        ],
    );
    assert_eq!(resolved.len(), 1, "{log}");
    let retry = publish(&services, br#"{"row":"c"}"#, "k1")?;
    assert_eq!(retry.durable_position, Some(0));
    let payloads: Vec<Vec<u8>> = store
        .order_rows()?
        .into_iter()
        .map(|(payload, _)| payload)
        .collect();
    assert_eq!(
        payloads,
        [br#"{"row":"c"}"#.to_vec(), br#"{"row":"e"}"#.to_vec()]
    );
    Ok(())
}

#[test]
fn an_unkeyed_unresolved_publish_is_answered_as_unresolved() -> TestResult {
    let store = OrdersFaultStore::new()?;
    let services = services(&store)?;
    store.arm_append(AppendFault::CommitThenError)?;
    let outcome = services.publish(7, "orders", &envelope(br#"{"row":"c"}"#), None);
    store.lift_append()?;
    assert!(matches!(
        outcome,
        Err(ServerError::PublishUnresolved {
            idempotency_key: None,
            sequence: Some(0),
            ..
        })
    ));
    Ok(())
}

/// Keyed rows on "orders" ahead of the publish a test looks up: enough that a
/// lookup reading the channel from its first row is unmistakable.
const HISTORY_ROWS: usize = 300;

/// Publishes `HISTORY_ROWS` keyed rows to "orders".
fn publish_history(services: &LiminalConnectionServices) -> TestResult {
    for index in 0..HISTORY_ROWS {
        publish(services, br#"{"row":"h"}"#, &format!("history-{index}"))?;
    }
    Ok(())
}

/// Writes "orders" with its `HISTORY_ROWS` keyed rows into a new database in
/// `data_dir`, through the services as a running server would.
fn build_history(data_dir: &Path) -> crate::test_fixture::FixtureResult<()> {
    let store = OrdersFaultStore::on_disk(data_dir, true)?;
    let services = services(&store)?;
    publish_history(&services)?;
    drop(services);
    liminal::durability::bridge::block_on(store.flush())??;
    Ok(())
}

/// A private copy of the "orders" history, built once per test binary, with
/// the double and the services over it. The copy's directory lives as long as
/// the answer's first member.
fn history() -> Result<
    (
        tempfile::TempDir,
        Arc<OrdersFaultStore>,
        LiminalConnectionServices,
    ),
    Box<dyn Error>,
> {
    let copy = crate::test_fixture::fixture_copy("orders-keyed-history-300", build_history)?;
    let store = OrdersFaultStore::on_disk(copy.path(), false)?;
    let services = services(&store)?;
    Ok((copy, store, services))
}

#[test]
fn a_same_key_retry_reads_only_the_rows_appended_since_its_claim() -> TestResult {
    let (_copy, store, services) = history()?;
    let sequence = unresolved_publish(
        &store,
        &services,
        AppendFault::CommitThenError,
        br#"{"row":"c"}"#,
        "late",
    )?;
    assert_eq!(sequence, Some(u64::try_from(HISTORY_ROWS)?));
    let rows_read = store.order_rows_read();

    let outcome = publish(&services, br#"{"row":"c"}"#, "late")?;

    assert_eq!(outcome.durable_position, sequence);
    let read = store.order_rows_read() - rows_read;
    assert!(
        read <= 2,
        "the lookup starts where the channel stood when the key was claimed; it read {read} rows"
    );
    crate::test_fixture::pin_store_opens(
        module_path!(),
        "a_same_key_retry_reads_only_the_rows_appended_since_its_claim",
    )?;
    Ok(())
}

#[test]
fn a_held_key_that_cannot_be_settled_does_not_make_every_publish_pay_for_it() -> TestResult {
    const LATER_PUBLISHES: usize = 32;
    let (_copy, store, services) = history()?;
    unresolved_publish(
        &store,
        &services,
        AppendFault::CommitThenError,
        br#"{"row":"c"}"#,
        "late",
    )?;
    // The row is found; the completion that would settle the claim is refused.
    store.refuse_stream(Some(store.dedup().stream_key_for("late")))?;
    let rows_read = store.order_rows_read();

    for index in 0..LATER_PUBLISHES {
        publish(&services, br#"{"row":"e"}"#, &format!("later-{index}"))?;
    }

    let read = store.order_rows_read() - rows_read;
    assert!(
        read <= 4,
        "a held key's row is looked up once, not again by every publish; {read} rows read"
    );
    let attempts = store.refused_appends();
    assert!(
        attempts <= 8,
        "a claim that keeps failing to settle is retried less and less often; \
         {attempts} attempts over {LATER_PUBLISHES} publishes"
    );
    store.refuse_stream(None)?;
    let outcome = publish(&services, br#"{"row":"c"}"#, "late")?;
    assert_eq!(outcome.durable_position, Some(u64::try_from(HISTORY_ROWS)?));
    crate::test_fixture::pin_store_opens(
        module_path!(),
        "a_held_key_that_cannot_be_settled_does_not_make_every_publish_pay_for_it",
    )?;
    Ok(())
}

#[test]
fn a_retry_meeting_a_claim_being_settled_as_absent_is_answered_unresolved() -> TestResult {
    let store = OrdersFaultStore::new()?;
    let services = services(&store)?;
    unresolved_publish(
        &store,
        &services,
        AppendFault::ErrorBeforeCommit,
        br#"{"row":"c"}"#,
        "k1",
    )?;
    // The next admission settles k1 as absent; its release waits at the gate.
    let (arrived, open) = store.gate_next_append(store.dedup().stream_key_for("k1"))?;
    std::thread::scope(|scope| -> TestResult {
        let admission = scope.spawn(|| publish(&services, br#"{"row":"e"}"#, "k2"));
        arrived.recv()?;

        let retry = publish(&services, br#"{"row":"c"}"#, "k1");

        open.send(())?;
        admission.join().map_err(|_| "the admission panicked")??;
        assert!(
            matches!(
                retry,
                Err(ServerError::PublishUnresolved { ref idempotency_key, .. })
                    if idempotency_key.as_deref() == Some("k1")
            ),
            "a retry meeting a claim its owner is settling as absent is never accepted: {retry:?}"
        );
        Ok(())
    })?;
    assert_eq!(
        store.order_rows()?,
        [(br#"{"row":"e"}"#.to_vec(), Some("k2".to_owned()))]
    );
    publish(&services, br#"{"row":"c"}"#, "k1")?;
    assert_eq!(
        store.order_rows()?,
        [
            (br#"{"row":"e"}"#.to_vec(), Some("k2".to_owned())),
            (br#"{"row":"c"}"#.to_vec(), Some("k1".to_owned())),
        ]
    );
    Ok(())
}

#[test]
fn a_claim_with_no_evidence_is_released_at_start_and_its_retry_publishes() -> TestResult {
    // What a binary before the evidence lane left (liminal-server 0.16.0 and
    // earlier): a claim written by claim_or_get, with no evidence and no
    // receipt. At the base the start held it as uncheckable and every publish
    // of its key was refused: "the claim carries no evidence of where its row
    // would be". DUR-022: the start releases it, and the retry publishes.
    let store = OrdersFaultStore::new()?;
    let legacy = store.dedup();
    liminal::durability::bridge::block_on(legacy.claim_or_get("old", 1))??;
    assert_eq!(
        liminal::durability::bridge::block_on(legacy.lookup("old"))??,
        Some(DedupDecision::InFlight)
    );

    let services = services(&store)?;
    // Released at the start, never completed: no receipt vouches for a row
    // nothing can find.
    let after_start = liminal::durability::bridge::block_on(store.dedup().lookup("old"))??;
    assert!(
        !matches!(
            after_start,
            Some(DedupDecision::InFlight | DedupDecision::Completed(_))
        ),
        "{after_start:?}"
    );
    let outcome = publish(&services, br#"{"row":"o"}"#, "old")?;
    assert!(
        outcome.durable_position.is_some(),
        "the retry publishes and its row lands: {outcome:?}"
    );
    // The next start finds nothing in flight for the key.
    let listed = liminal::durability::bridge::block_on(store.dedup().in_flight_claims())??;
    assert!(
        listed.iter().all(|claim| claim.idempotency_key != "old"),
        "{listed:?}"
    );
    Ok(())
}

/// The keys and evidence of `claims`, the shape a pin compares when the
/// claim time is not its subject.
fn by_key(
    claims: &[liminal::durability::InFlightClaim],
) -> Vec<(String, Option<liminal::durability::ClaimEvidence>)> {
    claims
        .iter()
        .map(|claim| (claim.idempotency_key.clone(), claim.evidence.clone()))
        .collect()
}