liminal-server 0.25.4

//! 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 a_claim_with_no_evidence_is_released_at_start_and_the_retry_writes_its_row() -> TestResult {
    // What a binary before the evidence lane left: a claim with no evidence
    // and no receipt. At the base the start held it as uncheckable (ERROR
    // "in-flight claim cannot be checked") and the retry was refused with no
    // row written. DUR-022: the start releases it, says so by key and age,
    // and the retry writes its row.
    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 released = events(
        &start_log,
        "in-flight claim released: no evidence",
        &["WARN", "idempotency_key=\"k9\"", "claimed_at_millis=1"],
    );
    assert_eq!(released.len(), 1, "{start_log}");
    let finished = events(
        &start_log,
        "in-flight claims resolve finished",
        &["claims=1", "released_unchecked=1"],
    );
    assert_eq!(finished.len(), 1, "{start_log}");
    assert!(
        events(&start_log, "in-flight claim cannot be checked", &[]).is_empty(),
        "{start_log}"
    );
    let after_start = block_on(store.dedup().lookup("k9"))??;
    assert!(
        !matches!(
            after_start,
            Some(DedupDecision::InFlight | DedupDecision::Completed(_))
        ),
        "released, never completed: {after_start:?}"
    );
    let outcome = publish(&services, C, "k9")?;
    assert!(outcome.durable_position.is_some(), "{outcome:?}");
    assert_eq!(store.order_rows()?, [(C.to_vec(), Some("k9".to_owned()))]);
    let listed = block_on(store.dedup().in_flight_claims())??;
    assert!(
        listed.iter().all(|claim| claim.idempotency_key != "k9"),
        "{listed:?}"
    );
    Ok(())
}

/// The door's mailbox start spent 1,098 of its 1,172 ms construction resolving
/// claims it found none of (12 October 2026). Building the services now reads
/// no claim: the resolve runs behind the listening door, and a keyed publish
/// that comes before it runs it first, so a crashed claim is still looked up
/// by its key before its retry is judged, and the retry writes one row.
#[test]
fn building_the_services_reads_no_claim_and_the_first_keyed_publish_resolves_first() -> TestResult {
    let store = OrdersFaultStore::new()?;
    block_on(
        store
            .dedup()
            .claim_with_evidence("k7", ClaimEvidence::new("orders", vec![0]), 1),
    )??;
    let config = super::services_unresolved_tests::orders_config()?;
    let (built, build_log) = captured_log(|| {
        super::LiminalConnectionServices::from_config_with_store(
            &config,
            Arc::clone(&store) as Arc<dyn DurableStore>,
        )
    })?;
    let built = built?;
    assert!(
        events(&build_log, "in-flight claims resolve started", &[]).is_empty(),
        "building the services resolves nothing: {build_log}"
    );
    assert_eq!(
        block_on(store.dedup().lookup("k7"))??,
        Some(DedupDecision::InFlight),
        "the claim is untouched until the resolve runs"
    );
    let (published, publish_log) = captured_log(|| publish(&built, C, "k7"))?;
    published?;
    assert_eq!(
        events(
            &publish_log,
            "unresolved publish resolved",
            &["idempotency_key=\"k7\"", "outcome=\"absent\""],
        )
        .len(),
        1,
        "the first keyed publish resolves the start's claims before it is judged: {publish_log}"
    );
    assert_eq!(store.order_rows()?, [(C.to_vec(), Some("k7".to_owned()))]);
    Ok(())
}

/// The listener's thread resolves the start's claims, and the keyed publish
/// after it finds nothing left to resolve.
#[test]
fn the_listening_thread_resolves_the_start_s_claims() -> TestResult {
    let store = OrdersFaultStore::new()?;
    block_on(
        store
            .dedup()
            .claim_with_evidence("k8", ClaimEvidence::new("orders", vec![0]), 1),
    )??;
    let built = Arc::new(super::LiminalConnectionServices::from_config_with_store(
        &super::services_unresolved_tests::orders_config()?,
        Arc::clone(&store) as Arc<dyn DurableStore>,
    )?);
    let resolving = super::ConnectionServices::listening(Arc::clone(&built))
        .ok_or("the listening services started no resolve")?;
    resolving
        .join()
        .map_err(|_| "the resolve thread panicked")?;
    // The listener's thread looked the claim up and released it: no row is
    // behind it.
    assert_eq!(block_on(store.dedup().lookup("k8"))??, None);
    publish(&built, C, "k8")?;
    assert_eq!(store.order_rows()?, [(C.to_vec(), Some("k8".to_owned()))]);
    Ok(())
}