liminal-server 0.19.0

//! A restart or a crash between an idempotency claim and its resolution never
//! leaves a second row for the key: the services rebuilt over the same store
//! look every in-flight claim up by its key before judging a publish.
//!
//! Each test drops the services and builds new ones over the SAME store to
//! stand for a restart or crash; a crash state is seeded directly through the
//! store and the durable delivery dedup namespace where a test names one.

use std::sync::Arc;

use liminal::durability::bridge::block_on;
use liminal::durability::{ClaimEvidence, DedupDecision, DurableStore, MessageEnvelope};

use super::services_unresolved_tests::{
    AppendFault, OrdersFaultStore, STREAM, TestResult, captured_log, events, publish, services,
    unresolved_publish,
};
use crate::ServerError;

const C: &[u8] = br#"{"row":"c"}"#;

/// Seeds a row at `orders:0` sequence 0 whose envelope carries `key`.
fn seed_row(store: &Arc<OrdersFaultStore>, key: &str) -> TestResult {
    let envelope = MessageEnvelope {
        payload: C.to_vec(),
        causal_context: None,
        timestamp: 1,
        publisher_id: "crashed-writer".to_owned(),
        idempotency_key: Some(key.to_owned()),
    };
    block_on(store.append(STREAM, envelope.serialize()?, 0))??;
    Ok(())
}

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

#[test]
fn restart_with_absent_unresolved_publishes_once() -> TestResult {
    let store = OrdersFaultStore::new()?;
    unresolved_publish(
        &store,
        &services(&store)?,
        AppendFault::ErrorBeforeCommit,
        C,
        "k1",
    )?;
    let services = services(&store)?;
    publish(&services, C, "k1")?;
    assert_eq!(store.order_rows()?, [(C.to_vec(), Some("k1".to_owned()))]);
    let retry = publish(&services, C, "k1")?;
    assert!(!retry.delivered);
    assert_eq!(store.order_rows()?.len(), 1);
    Ok(())
}

#[test]
fn restart_with_unreadable_record_keeps_the_claim() -> TestResult {
    let store = OrdersFaultStore::new()?;
    unresolved_publish(
        &store,
        &services(&store)?,
        AppendFault::CommitThenError,
        C,
        "k1",
    )?;
    // Registration's recovery walks the one row of `orders:0` in one read;
    // the key lookup's reads after it are refused.
    store.arm_resolution_reads_after(1)?;
    let reads_before = store.order_reads_from();
    let services = services(&store)?;
    assert!(store.order_reads_from() > reads_before);
    assert_eq!(
        block_on(store.dedup().lookup("k1"))??,
        Some(DedupDecision::InFlight)
    );
    let appends = store.order_appends();
    assert!(matches!(
        publish(&services, 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, C, "k1")?;
    assert_eq!(outcome.durable_position, Some(0));
    assert_eq!(store.order_rows()?.len(), 1);
    Ok(())
}

#[test]
fn crash_after_committed_append_before_completion_gives_no_second_row() -> TestResult {
    let store = OrdersFaultStore::new()?;
    block_on(
        store
            .dedup()
            .claim_with_evidence("k5", ClaimEvidence::new("orders", vec![0]), 1),
    )??;
    seed_row(&store, "k5")?;
    let (services, log) = captured_log(|| services(&store))?;
    let services = services?;
    let resolved = events(
        &log,
        "unresolved publish resolved",
        &[
            "channel=\"orders\"",
            "idempotency_key=\"k5\"",
            "outcome=\"committed\"",
        ],
    );
    assert_eq!(resolved.len(), 1, "{log}");
    let Some(DedupDecision::Completed(receipt)) = block_on(store.dedup().lookup("k5"))?? else {
        return Err("the claim was not completed at start".into());
    };
    assert_eq!(receipt.durable_position(), Some(0));
    let outcome = publish(&services, C, "k5")?;
    assert!(!outcome.delivered);
    assert_eq!(outcome.durable_position, Some(0));
    assert_eq!(store.order_rows()?.len(), 1);
    Ok(())
}

#[test]
fn crash_before_append_then_retry_publishes_once() -> TestResult {
    let store = OrdersFaultStore::new()?;
    block_on(
        store
            .dedup()
            .claim_with_evidence("k6", ClaimEvidence::new("orders", vec![0]), 1),
    )??;
    let (services, log) = captured_log(|| services(&store))?;
    let services = services?;
    let resolved = events(
        &log,
        "unresolved publish resolved",
        &["idempotency_key=\"k6\"", "outcome=\"absent\""],
    );
    assert_eq!(resolved.len(), 1, "{log}");
    assert_eq!(block_on(store.dedup().lookup("k6"))??, None);
    publish(&services, C, "k6")?;
    assert_eq!(store.order_rows()?, [(C.to_vec(), Some("k6".to_owned()))]);
    Ok(())
}

#[test]
fn uncheckable_claim_answers_unresolved_and_writes_no_row() -> TestResult {
    let store = OrdersFaultStore::new()?;
    block_on(store.dedup().claim_or_get("k9", 1))??;
    let (services, start_log) = captured_log(|| services(&store))?;
    let services = services?;
    let at_start = events(
        &start_log,
        "in-flight claim cannot be checked",
        &["ERROR", "idempotency_key=\"k9\""],
    );
    assert_eq!(at_start.len(), 1, "{start_log}");
    assert_eq!(
        block_on(store.dedup().lookup("k9"))??,
        Some(DedupDecision::InFlight)
    );
    let (outcome, log) = captured_log(|| publish(&services, C, "k9"))?;
    assert!(matches!(
        outcome,
        Err(ServerError::PublishUnresolved { ref idempotency_key, sequence: None, .. })
            if idempotency_key.as_deref() == Some("k9")
    ));
    let refused = events(
        &log,
        "in-flight claim cannot be checked",
        &["ERROR", "idempotency_key=\"k9\""],
    );
    assert_eq!(refused.len(), 1, "{log}");
    assert!(store.order_rows()?.is_empty());
    Ok(())
}