liminal-server 0.14.3

//! A1 §5 pin: **dedup namespaces scoped by durability.**
//!
//! `docs/design/A1-DEFER-SEMANTICS.md` §5, graft §0.2 — the panel's sharpest
//! crash hole:
//!
//! > ephemeral + keyed publish → Defer (buffered in memory) →
//! > `complete_receipt` (persisted) → crash before the consumer pops. The
//! > message dies with the process but the receipt survives and suppresses the
//! > producer's retry — silent permanent loss of a message the producer was
//! > told was admitted.
//!
//! The fix is a namespace, not an API change: ephemeral dedup records live
//! under a per-boot incarnation id so message and receipt share a lifetime;
//! durable dedup records keep the persisted namespace, where the surviving
//! receipt is truthful because the message survived too.

use std::sync::Arc;

use haematite::{Database, DatabaseConfig, EventStore};
use liminal::durability::{DurableStore, HaematiteStore};
use liminal::protocol::{CausalContext, MessageEnvelope, SchemaId};
use tempfile::TempDir;

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

fn disk_store() -> Result<(Arc<dyn DurableStore>, TempDir), Box<dyn std::error::Error>> {
    let dir = tempfile::tempdir()?;
    let database = Database::create(DatabaseConfig {
        data_dir: dir.path().join("db"),
        shard_count: 4,
        distributed: None,
        executor_threads: None,
        node_cache_budget: Some(haematite::NodeCacheBudget::Unlimited),
    })?;
    let store: Arc<dyn DurableStore> =
        Arc::new(HaematiteStore::new(Arc::new(EventStore::new(database))));
    Ok((store, dir))
}

/// One durable channel and one ephemeral channel over the SAME store, so the
/// only variable between the two halves of the pin is durability.
fn mixed_channels_config() -> Result<ServerConfig, Box<dyn std::error::Error>> {
    Ok(ServerConfig {
        listen_address: "127.0.0.1:0".parse()?,
        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,
            },
            ChannelDef {
                name: "events".to_owned(),
                schema_ref: None,
                durable: false,
                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,
    })
}

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

/// Publish `key` to `channel` and report whether a subscriber actually received
/// it — the only observation that distinguishes "re-claimed and re-delivered"
/// from "suppressed by a surviving receipt". A returned message id proves
/// nothing: a dedup-suppressed publish is still assigned one.
fn delivered(
    services: &LiminalConnectionServices,
    channel: &str,
    key: &str,
    payload: &[u8],
) -> Result<bool, Box<dyn std::error::Error>> {
    let subscription = services.subscribe_handle_for_test(channel)?;
    services.publish(channel, &envelope(payload), Some(key))?;
    let received = subscription.try_next()?.is_some();
    drop(subscription);
    Ok(received)
}

/// **PIN — dedup namespaces are scoped by durability (§5, graft §0.2).**
///
/// One store, two incarnations, two channels. The assertion that matters is the
/// pair: after the "restart" the EPHEMERAL key re-delivers and the DURABLE key
/// stays suppressed. Either half alone is satisfiable by a build with no
/// namespacing at all (drop dedup entirely and both re-deliver; keep one
/// namespace and both suppress), so the discriminating power is in their
/// disagreement.
#[test]
fn ephemeral_dedup_dies_with_the_incarnation_while_durable_dedup_survives()
-> Result<(), Box<dyn std::error::Error>> {
    let (store, _dir) = disk_store()?;
    let config = mixed_channels_config()?;

    // ---- incarnation A -----------------------------------------------------
    {
        let services =
            LiminalConnectionServices::from_config_with_store(&config, Arc::clone(&store))?;

        assert!(
            delivered(&services, "events", "shared-key", br#"{"n":1}"#)?,
            "a first keyed publish to an ephemeral channel is delivered"
        );
        assert!(
            !delivered(&services, "events", "shared-key", br#"{"n":1}"#)?,
            "a SAME-INCARNATION retry is suppressed: the bus still owns the message"
        );

        assert!(
            delivered(&services, "orders", "shared-key", br#"{"n":1}"#)?,
            "a first keyed publish to a durable channel is delivered"
        );
        assert!(
            !delivered(&services, "orders", "shared-key", br#"{"n":1}"#)?,
            "a same-incarnation durable retry is suppressed too"
        );
    }

    // ---- incarnation B: same store, new boot -------------------------------
    let services = LiminalConnectionServices::from_config_with_store(&config, Arc::clone(&store))?;

    assert!(
        delivered(&services, "events", "shared-key", br#"{"n":1}"#)?,
        "EPHEMERAL: the incarnation-scoped receipt died with its message, so the \
         producer's retry re-claims and re-delivers — at-least-once, which is \
         exactly ephemeral's contract"
    );
    assert!(
        !delivered(&services, "orders", "shared-key", br#"{"n":1}"#)?,
        "DURABLE: the message survived, so the surviving receipt is truthful and \
         must keep suppressing"
    );

    Ok(())
}

/// **PIN — a keyed publish the bus DROPPED must not complete its receipt
/// (§5's `Reject => release_claim`, reached through the §7 `is_admitted()`
/// predicate at `services.rs::publish`).**
///
/// The consequence half of verifier finding 2, asserted where it bites. When
/// every matching subscriber refuses through the §5 budget/fairness path, the
/// aggregate used to answer with its zero-subscriber sentinel `Accept { 0, 0 }`
/// — so `is_admitted()` read true, `complete_receipt` ran for a message nobody
/// received, and the producer's retry with the same key was suppressed
/// FOREVER. Silent permanent loss.
///
/// The observation is a delivery to a HEALTHY subscriber added afterwards: if
/// the claim was released, the retry re-claims and that subscriber receives the
/// message; if the receipt was completed, the retry is suppressed and the
/// message is gone. A returned message id proves nothing either way.
#[test]
fn a_keyed_publish_dropped_by_every_subscriber_releases_its_claim()
-> Result<(), Box<dyn std::error::Error>> {
    let (store, _dir) = disk_store()?;
    let config = mixed_channels_config()?;
    let services = LiminalConnectionServices::from_config_with_store(&config, store)?;

    // The only subscriber on the channel, under a one-byte connection budget:
    // no envelope encodes that small, so §5 refuses the charge and the publish
    // reaches nobody. Its A1 window is wide, so this is a §5 refusal and not an
    // A1 Reject — the two verbs must not be confusable here.
    let starved = services.subscribe_handle_for_test_with_install(
        "events",
        Some(liminal::channel::InboxInstall {
            budget: liminal::channel::ConnectionInboxBudget::new(1),
            depth_cap: usize::MAX,
            notifier: None,
            capacity: Some(liminal::pressure::ConsumerCapacity::new(64, 64)?),
        }),
    )?;

    let dropped = services.publish("events", &envelope(br#"{"n":1}"#), Some("dropped-key"))?;
    assert!(
        !dropped.delivered,
        "the fixture must actually drop it: a delivered publish proves nothing"
    );
    assert!(
        starved.is_overflowed(),
        "and it must be dropped by the §5 door this pin is about"
    );

    // A healthy subscriber joins, and the producer retries the SAME key.
    let healthy = services.subscribe_handle_for_test("events")?;
    services.publish("events", &envelope(br#"{"n":1}"#), Some("dropped-key"))?;
    assert!(
        healthy.try_next()?.is_some(),
        "the dropped publish released its claim, so the retry re-claims and \
         delivers; a completed receipt would have suppressed it forever"
    );

    // The other direction, so this cannot pass on a build with no dedup at all:
    // a publish that WAS admitted still completes, and its retry is suppressed.
    let kept = services.subscribe_handle_for_test("events")?;
    services.publish("events", &envelope(br#"{"n":2}"#), Some("kept-key"))?;
    assert!(kept.try_next()?.is_some(), "the first keyed publish lands");
    services.publish("events", &envelope(br#"{"n":2}"#), Some("kept-key"))?;
    assert!(
        kept.try_next()?.is_none(),
        "an admitted publish completed its receipt, so the retry is suppressed"
    );

    drop(starved);
    Ok(())
}

/// **PIN (round 3) — the same claim on a DURABLE channel, where it was still
/// false.**
///
/// The pin above uses the ephemeral channel of this fixture, and that is exactly
/// as far as round 1's fix reached. On the durable channel the aggregate
/// resolved the same Reject and `defer_after_append` rewrote it to Defer because
/// the row had been appended, so `is_admitted()` read true, `complete_receipt`
/// ran, and the producer's same-key retry was suppressed FOREVER for a message
/// no live consumer can ever receive. The row is in the log — a deliberate
/// `replay_from` read still finds it — but no subscription is ever offered it
/// again (a §5 shed arms no auto-catch-up, and a fresh subscription seeds its
/// replay cursor at the CURRENT head), so this is a live-custody loss and the
/// dedup mapping must treat it as one.
///
/// The observation is the same as the ephemeral half's, and for the same reason:
/// a returned message id proves nothing, so the pin asks whether a healthy
/// subscriber added afterwards actually receives the retry.
#[test]
fn a_keyed_publish_dropped_on_a_durable_channel_releases_its_claim()
-> Result<(), Box<dyn std::error::Error>> {
    let (store, _dir) = disk_store()?;
    let config = mixed_channels_config()?;
    let services = LiminalConnectionServices::from_config_with_store(&config, store)?;

    // "orders" is the DURABLE channel of this fixture; everything else about the
    // setup is byte-for-byte the ephemeral pin's, so durability is the only
    // variable between them.
    let starved = services.subscribe_handle_for_test_with_install(
        "orders",
        Some(liminal::channel::InboxInstall {
            budget: liminal::channel::ConnectionInboxBudget::new(1),
            depth_cap: usize::MAX,
            notifier: None,
            capacity: Some(liminal::pressure::ConsumerCapacity::new(64, 64)?),
        }),
    )?;

    let dropped = services.publish("orders", &envelope(br#"{"n":1}"#), Some("dropped-key"))?;
    assert!(
        !dropped.delivered,
        "the fixture must actually drop it: a delivered publish proves nothing"
    );
    assert!(
        starved.is_overflowed(),
        "and it must be dropped by the §5 door this pin is about"
    );

    let healthy = services.subscribe_handle_for_test("orders")?;
    services.publish("orders", &envelope(br#"{"n":1}"#), Some("dropped-key"))?;
    assert!(
        healthy.try_next()?.is_some(),
        "the dropped publish released its claim, so the retry re-claims and \
         delivers; a completed receipt would have suppressed it forever"
    );

    // The other direction, so this cannot pass on a build with no dedup at all.
    let kept = services.subscribe_handle_for_test("orders")?;
    services.publish("orders", &envelope(br#"{"n":2}"#), Some("kept-key"))?;
    assert!(kept.try_next()?.is_some(), "the first keyed publish lands");
    services.publish("orders", &envelope(br#"{"n":2}"#), Some("kept-key"))?;
    assert!(
        kept.try_next()?.is_none(),
        "an admitted publish completed its receipt, so the retry is suppressed"
    );

    drop(starved);
    Ok(())
}

/// The same key on the two channel classes does not collide: the namespaces are
/// disjoint prefixes, so an ephemeral claim can never satisfy a durable one or
/// vice versa. Without this, the first test's durable half could pass by
/// reading the ephemeral half's receipt.
#[test]
fn the_two_namespaces_do_not_share_a_key() -> Result<(), Box<dyn std::error::Error>> {
    let (store, _dir) = disk_store()?;
    let config = mixed_channels_config()?;
    let services = LiminalConnectionServices::from_config_with_store(&config, store)?;

    assert!(
        delivered(&services, "events", "same-key", br#"{"n":1}"#)?,
        "the ephemeral publish claims the key in ITS namespace"
    );
    assert!(
        delivered(&services, "orders", "same-key", br#"{"n":1}"#)?,
        "the durable publish claims the SAME key in its own namespace, uncontested"
    );
    Ok(())
}