liminal-server 0.25.4

//! INCARNATION-STREAM-BOUND acceptance pins 2, 4 and 5.
//!
//! Pin 1 (scale) lives in [`super::incarnation_stream_scale_tests`] and pin 3
//! (invariants) is the module's existing suite, unchanged in meaning. What is
//! left is everything that could go WRONG with bounding history by rotation:
//!
//! - **the regression pin (adversarial)** — the property the genesis-replay
//!   design bought must survive the fix. A resume point that regresses, is
//!   forged, or is stale must not be obeyed.
//! - **the compatibility pin** — a store carrying pre-fix v2 history, built by
//!   the old code path in the test itself, opens, migrates and continues, with
//!   nothing refused and nothing silently dropped.
//! - **the crash pin** — the rotation interrupted mid-flight at the store seam,
//!   with no timer anywhere: the next startup either recovers to a valid
//!   bounded state or refuses loudly, and never opens ambiguous history.

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

use liminal::durability::{
    DurabilityError, DurableStore, ScanCursor, ScanPage, StoredEntry, bridge::block_on,
    open_ephemeral,
};
use liminal_protocol::wire::ConnectionIncarnation;

use super::incarnation_stream::{
    IncarnationAllocation, IncarnationStartup, IncarnationStream, IncarnationStreamError,
    encode_allocate_event_fixture, encode_startup_event_fixture,
};

/// Signed live-reference bound used by every boot below.
const REFERENCE_BOUND: usize = 4;
/// Read limit large enough to deliver any stream these pins build.
const WHOLE_STREAM: usize = 4096;

/// One fault injected at the durable-store seam. No timers, no sleeps: each
/// arm is a deterministic refusal at a named durable operation, which is the
/// only way to interrupt a rotation at a point of our choosing.
#[derive(Clone, Copy, Debug)]
enum StoreFault {
    /// The anchor reads back as absent even though a value is stored — a
    /// store that has lost the hint but kept every durable epoch.
    HideAnchor,
    /// The anchor's compare-and-swap refuses. This is the crash window the
    /// rotation is ordered around: the epoch's Startup row is already appended
    /// and fsynced, and the anchor never comes to name it. Since SRV-048 the
    /// refusal is the settle's, after the start has published.
    RefuseAnchorSwap,
    /// Appends to one epoch's stream refuse, so the rotation dies before any
    /// durable byte of the successor exists.
    RefuseEpochAppend { server_incarnation: u64 },
}

/// A durable store with exactly one injected fault, armed until disarmed.
#[derive(Debug)]
struct FaultyStore {
    inner: Arc<dyn DurableStore>,
    fault: StoreFault,
    armed: AtomicBool,
}

impl FaultyStore {
    fn new(inner: Arc<dyn DurableStore>, fault: StoreFault) -> Self {
        Self {
            inner,
            fault,
            armed: AtomicBool::new(true),
        }
    }

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

#[async_trait::async_trait]
impl DurableStore for FaultyStore {
    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> {
        if let StoreFault::RefuseEpochAppend { server_incarnation } = self.fault
            && self.armed()
            && stream_key == IncarnationStream::epoch_stream_key(server_incarnation)
        {
            return Err(DurabilityError::ConfigError(format!(
                "injected epoch append failure for incarnation {server_incarnation}"
            )));
        }
        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> {
        if let StoreFault::RefuseEpochAppend { server_incarnation } = self.fault
            && self.armed()
            && stream_key == IncarnationStream::epoch_stream_key(server_incarnation)
        {
            return Err(DurabilityError::ConfigError(format!(
                "injected epoch append failure for incarnation {server_incarnation}"
            )));
        }
        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 cas(&self, key: &str, old_value: u64, new_value: u64) -> Result<(), DurabilityError> {
        if matches!(self.fault, StoreFault::RefuseAnchorSwap)
            && self.armed()
            && key == IncarnationStream::epoch_anchor_key()
        {
            return Err(DurabilityError::ConfigError(
                "injected anchor swap failure".to_owned(),
            ));
        }
        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> {
        if matches!(self.fault, StoreFault::HideAnchor)
            && self.armed()
            && key == IncarnationStream::epoch_anchor_key()
        {
            return Ok(None);
        }
        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 store() -> Result<Arc<dyn DurableStore>, Box<dyn Error>> {
    Ok(Arc::new(open_ephemeral(1)?))
}

/// One whole server lifetime: boot, allocate one connection, stop.
///
/// Returns the incarnation this boot durably started under and the pair it
/// published, which together are what a regression would corrupt.
fn boot_and_allocate(
    store: &Arc<dyn DurableStore>,
) -> Result<(u64, ConnectionIncarnation), Box<dyn Error>> {
    let startup = block_on(IncarnationStream::new(Arc::clone(store), REFERENCE_BOUND).startup())??;
    let IncarnationStartup::Started(mut stream) = startup else {
        return Err("a boot could not start the incarnation stream".into());
    };
    let server_incarnation = stream.server_incarnation();
    let IncarnationAllocation::Allocated {
        connection_incarnation,
        ..
    } = block_on(stream.allocate(&[]))??
    else {
        return Err("a boot exhausted the connection-ordinal space".into());
    };
    // The start thread settles the anchor once admission is open (SRV-048).
    block_on(stream.settle_anchor())??;
    Ok((server_incarnation, connection_incarnation))
}

/// The boot's outcome as a typed error, for the arms that must REFUSE.
fn boot_failure(store: &Arc<dyn DurableStore>) -> Result<IncarnationStreamError, Box<dyn Error>> {
    match block_on(IncarnationStream::new(Arc::clone(store), REFERENCE_BOUND).startup())? {
        Ok(_) => Err("the boot was expected to refuse and did not".into()),
        Err(error) => Ok(error),
    }
}

fn anchor(store: &Arc<dyn DurableStore>) -> Result<Option<u64>, Box<dyn Error>> {
    Ok(block_on(
        store.read_value(IncarnationStream::epoch_anchor_key()),
    )??)
}

fn epoch_row_count(
    store: &Arc<dyn DurableStore>,
    server_incarnation: u64,
) -> Result<usize, Box<dyn Error>> {
    Ok(block_on(store.read_from(
        &IncarnationStream::epoch_stream_key(server_incarnation),
        0,
        WHOLE_STREAM,
    ))??
    .len())
}

fn genesis_rows(store: &Arc<dyn DurableStore>) -> Result<Vec<StoredEntry>, Box<dyn Error>> {
    Ok(block_on(store.read_from(
        IncarnationStream::stream_key(),
        0,
        WHOLE_STREAM,
    ))??)
}

/// Appends one row straight into a durable stream, as a hostile or torn writer
/// would. Nothing in production writes this way.
fn forge_row(
    store: &Arc<dyn DurableStore>,
    stream_key: &str,
    sequence: u64,
    payload: Vec<u8>,
) -> Result<(), Box<dyn Error>> {
    let assigned = block_on(store.append(stream_key, payload, sequence))??;
    assert_eq!(
        assigned, sequence,
        "the forged row did not land where asked"
    );
    block_on(store.flush())??;
    Ok(())
}

// ---------------------------------------------------------------------------
// Pin 2 — the regression pin (adversarial)
// ---------------------------------------------------------------------------

/// An anchor moved BACKWARDS is overruled by the epochs that exist above it.
///
/// This is the adversarial form of the property the genesis-replay design
/// bought. The anchor is the one scalar the rotation added, so it is the one
/// thing a forger would reach for; the pin drives it down by two incarnations
/// on a store with real history and shows the boot walking back up. The
/// incarnation that follows is 4 — the successor of durable truth — and never
/// 2, which would re-issue connection pairs the estate has already published.
#[test]
fn a_regressed_anchor_is_overruled_by_the_epochs_above_it() -> Result<(), Box<dyn Error>> {
    let store = store()?;
    for expected in 1..=3_u64 {
        let (server_incarnation, _) = boot_and_allocate(&store)?;
        assert_eq!(server_incarnation, expected);
    }
    assert_eq!(anchor(&store)?, Some(3));

    // The forgery: the anchor now names an incarnation two below the truth,
    // and epochs 2 and 3 are still on disk saying otherwise.
    block_on(store.cas(IncarnationStream::epoch_anchor_key(), 3, 1))??;
    assert_eq!(anchor(&store)?, Some(1));

    let (server_incarnation, connection_incarnation) = boot_and_allocate(&store)?;
    assert_eq!(
        server_incarnation, 4,
        "a regressed anchor must not re-issue an incarnation the estate has published"
    );
    assert_eq!(connection_incarnation, ConnectionIncarnation::new(4, 0));
    assert_eq!(
        anchor(&store)?,
        Some(4),
        "the walk repairs the anchor rather than leaving it wrong"
    );
    Ok(())
}

/// An epoch whose canonical inputs do not produce the incarnation its key names
/// is REFUSED, not adopted.
///
/// The forged epoch here is internally well-formed — two legal Startup rows,
/// each decoding and each a legal protocol transition — and it is the anchored
/// stream of record. What condemns it is that its key says "incarnation 2" and
/// re-executing it says 3. Because the key IS the resume seed, that
/// disagreement is the whole check: there is no header byte to compare against
/// and none to forge.
#[test]
fn an_epoch_that_disagrees_with_the_key_naming_it_is_refused() -> Result<(), Box<dyn Error>> {
    let store = store()?;
    let (first, _) = boot_and_allocate(&store)?;
    assert_eq!(first, 1);

    let epoch_two = IncarnationStream::epoch_stream_key(2);
    forge_row(&store, &epoch_two, 0, encode_startup_event_fixture()?)?;
    forge_row(&store, &epoch_two, 1, encode_startup_event_fixture()?)?;
    block_on(store.cas(IncarnationStream::epoch_anchor_key(), 1, 2))??;

    assert!(
        matches!(
            boot_failure(&store)?,
            IncarnationStreamError::EpochHeaderMismatch {
                epoch_server_incarnation: 2,
                replayed_server_incarnation: 3,
            }
        ),
        "a forged epoch must be refused by name, not silently adopted"
    );
    Ok(())
}

/// An epoch that is not a startup at all is refused by the same grammar that
/// refused it before the rotation existed.
///
/// The stale-checkpoint shape: an anchored stream whose first row is an
/// allocation, as it would be if a writer had tried to resume mid-generation.
#[test]
fn an_epoch_whose_first_row_is_not_a_startup_is_refused() -> Result<(), Box<dyn Error>> {
    let store = store()?;
    let (first, _) = boot_and_allocate(&store)?;
    assert_eq!(first, 1);

    forge_row(
        &store,
        &IncarnationStream::epoch_stream_key(2),
        0,
        encode_allocate_event_fixture(REFERENCE_BOUND, &[])?,
    )?;
    block_on(store.cas(IncarnationStream::epoch_anchor_key(), 1, 2))??;

    assert!(
        matches!(
            boot_failure(&store)?,
            IncarnationStreamError::AllocateBeforeStartup { stored_sequence: 0 }
        ),
        "an epoch with no Startup names no incarnation and must be refused"
    );
    Ok(())
}

/// A LOST anchor cannot regress the header either: it refuses loudly.
///
/// The anchor reads back absent while every durable epoch is still there, which
/// is the worst case for a design that used a stored scalar as its authority —
/// it would resume from genesis and hand out incarnation 1 a second time. Here
/// the boot instead finds epoch 1 above the hint it was given, tries to walk up
/// to it, and is refused by the store's own compare-and-swap because the real
/// anchor disagrees. The refusal is the correct outcome: it leaves durable
/// state untouched, and the same store boots normally once the hint is back.
#[test]
fn a_lost_anchor_refuses_loudly_and_never_replays_to_a_lower_header() -> Result<(), Box<dyn Error>>
{
    let inner = store()?;
    for expected in 1..=2_u64 {
        let (server_incarnation, _) = boot_and_allocate(&inner)?;
        assert_eq!(server_incarnation, expected);
    }

    let hiding = Arc::new(FaultyStore::new(Arc::clone(&inner), StoreFault::HideAnchor));
    let observed: Arc<dyn DurableStore> = Arc::clone(&hiding) as Arc<dyn DurableStore>;
    assert!(
        matches!(
            boot_failure(&observed)?,
            IncarnationStreamError::Durability(_)
        ),
        "a boot that cannot see the anchor must refuse, never resume from genesis"
    );
    assert_eq!(
        anchor(&inner)?,
        Some(2),
        "the refused boot left the durable anchor exactly where it was"
    );
    assert_eq!(
        epoch_row_count(&inner, 3)?,
        0,
        "the refused boot opened no successor epoch"
    );

    hiding.armed.store(false, Ordering::SeqCst);
    let (server_incarnation, connection_incarnation) = boot_and_allocate(&inner)?;
    assert_eq!(server_incarnation, 3);
    assert_eq!(connection_incarnation, ConnectionIncarnation::new(3, 0));
    Ok(())
}

// ---------------------------------------------------------------------------
// Pin 4 — the compatibility pin
// ---------------------------------------------------------------------------

/// A store carrying pre-fix v2 history opens, migrates, and continues bounded.
///
/// The history is built HERE by the old code path — the same canonical rows the
/// pre-rotation writer appended to the one estate-global stream — never copied
/// from an estate. After the migrating boot the genesis stream is byte-for-byte
/// what it was: nothing is dropped, nothing is rewritten, and the reclaiming of
/// that dead history belongs to the store vacuum, not to this seam.
///
/// The second boot is the point of the whole flight: it does not read the
/// genesis stream at all.
#[test]
fn pre_fix_genesis_history_opens_migrates_and_then_stops_being_read() -> Result<(), Box<dyn Error>>
{
    let store = store()?;
    // Two pre-fix server generations, each with one allocation, exactly as the
    // estate-global stream carried them.
    let pre_fix = [
        encode_startup_event_fixture()?,
        encode_allocate_event_fixture(REFERENCE_BOUND, &[])?,
        encode_startup_event_fixture()?,
        encode_allocate_event_fixture(REFERENCE_BOUND, &[])?,
    ];
    for (sequence, payload) in pre_fix.into_iter().enumerate() {
        forge_row(
            &store,
            IncarnationStream::stream_key(),
            u64::try_from(sequence)?,
            payload,
        )?;
    }
    let before_migration = genesis_rows(&store)?;
    assert_eq!(before_migration.len(), 4);
    assert_eq!(anchor(&store)?, None, "pre-fix history carries no anchor");

    // The migrating boot: it reads what exists and continues from it.
    let (migrated, connection_incarnation) = boot_and_allocate(&store)?;
    assert_eq!(
        migrated, 3,
        "the migrating boot must continue the durable incarnation, not restart it"
    );
    assert_eq!(connection_incarnation, ConnectionIncarnation::new(3, 0));
    assert_eq!(anchor(&store)?, Some(3));
    assert_eq!(
        genesis_rows(&store)?,
        before_migration,
        "migration must not drop or rewrite a single pre-fix row"
    );

    // And the boot after it never goes back. Every incarnation-namespace read
    // is recorded, and the genesis key is absent from the record.
    let recording = Arc::new(RecordingStore::new(Arc::clone(&store)));
    let observed: Arc<dyn DurableStore> = Arc::clone(&recording) as Arc<dyn DurableStore>;
    let (next, _) = boot_and_allocate(&observed)?;
    assert_eq!(next, 4);
    let read_keys = recording.read_keys()?;
    assert!(
        !read_keys.is_empty(),
        "the measured boot must have reached the store"
    );
    assert!(
        !read_keys
            .iter()
            .any(|key| key == IncarnationStream::stream_key()),
        "a migrated store must never replay pre-fix history again, read: {read_keys:?}"
    );
    Ok(())
}

/// Records which incarnation-namespace stream keys a boot actually reads.
#[derive(Debug)]
struct RecordingStore {
    inner: Arc<dyn DurableStore>,
    keys: std::sync::Mutex<Vec<String>>,
}

impl RecordingStore {
    fn new(inner: Arc<dyn DurableStore>) -> Self {
        Self {
            inner,
            keys: std::sync::Mutex::new(Vec::new()),
        }
    }

    fn read_keys(&self) -> Result<Vec<String>, Box<dyn Error>> {
        Ok(self
            .keys
            .lock()
            .map_err(|error| std::io::Error::other(error.to_string()))?
            .clone())
    }
}

#[async_trait::async_trait]
impl DurableStore for RecordingStore {
    async fn apply_routed_batch(
        &self,
        _: &str,
        _: Vec<liminal::durability::RoutedBatchItem>,
    ) -> Result<(), liminal::durability::DurabilityError> {
        Err(liminal::durability::DurabilityError::BatchUnsupported {
            store: std::any::type_name::<Self>(),
        })
    }

    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.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.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.starts_with(IncarnationStream::stream_key())
            && let Ok(mut keys) = self.keys.lock()
        {
            keys.push(stream_key.to_owned());
        }
        self.inner.read_from(stream_key, offset, limit).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
    }
}

// ---------------------------------------------------------------------------
// Pin 5 — the crash pin (outcome-gated)
// ---------------------------------------------------------------------------

/// A rotation interrupted between its epoch's fsync and its anchor swap is
/// ADOPTED by the next boot, never re-minted.
///
/// This is the window the rotation's write order exists for. The epoch's
/// Startup row is durable, so incarnation 2 has been published as far as any
/// observer is concerned; the anchor never came to name it. The next boot must
/// therefore treat 2 as spent and take 3 — handing out 2 again would re-open a
/// connection-ordinal space the estate has already used, which is the exact
/// harm the whole design is built to prevent.
#[test]
fn a_rotation_interrupted_before_its_anchor_swap_is_adopted_not_reminted()
-> Result<(), Box<dyn Error>> {
    let inner = store()?;
    let (first, _) = boot_and_allocate(&inner)?;
    assert_eq!(first, 1);

    let interrupted = Arc::new(FaultyStore::new(
        Arc::clone(&inner),
        StoreFault::RefuseAnchorSwap,
    ));
    let observed: Arc<dyn DurableStore> = Arc::clone(&interrupted) as Arc<dyn DurableStore>;
    // SRV-048: the start publishes on its flushed Startup row and the swap is
    // settled after; the refusal fails the settle, loudly, and the server's
    // start thread stops the server on it.
    let startup = block_on(IncarnationStream::new(observed, REFERENCE_BOUND).startup())??;
    let IncarnationStartup::Started(mut started) = startup else {
        return Err("the interrupted boot did not start".into());
    };
    assert_eq!(started.server_incarnation(), 2);
    assert!(
        matches!(
            block_on(started.settle_anchor())?,
            Err(IncarnationStreamError::Durability(_))
        ),
        "a rotation whose anchor swap fails must fail its settle, loudly"
    );
    assert_eq!(
        epoch_row_count(&inner, 2)?,
        1,
        "the interrupted rotation left its epoch's Startup row durable"
    );
    assert_eq!(
        anchor(&inner)?,
        Some(1),
        "and left the anchor still naming the previous epoch"
    );

    let (recovered, connection_incarnation) = boot_and_allocate(&inner)?;
    assert_eq!(
        recovered, 3,
        "the unanchored epoch is adopted; its incarnation is never handed out twice"
    );
    assert_eq!(connection_incarnation, ConnectionIncarnation::new(3, 0));
    assert_eq!(anchor(&inner)?, Some(3));
    assert_eq!(
        epoch_row_count(&inner, 2)?,
        1,
        "the adopted epoch is read, never appended to"
    );
    Ok(())
}

/// A rotation interrupted BEFORE any successor byte exists changes nothing.
///
/// The other side of the crash window, and the control that makes the pin above
/// mean something: when the epoch's own append fails there is no durable claim
/// on incarnation 2, so the next boot takes it. A design that leaked the
/// incarnation here would still "recover", just by burning a number on every
/// failed boot.
#[test]
fn a_rotation_that_never_reaches_the_store_leaves_its_incarnation_unspent()
-> Result<(), Box<dyn Error>> {
    let inner = store()?;
    let (first, _) = boot_and_allocate(&inner)?;
    assert_eq!(first, 1);

    let interrupted = Arc::new(FaultyStore::new(
        Arc::clone(&inner),
        StoreFault::RefuseEpochAppend {
            server_incarnation: 2,
        },
    ));
    let observed: Arc<dyn DurableStore> = Arc::clone(&interrupted) as Arc<dyn DurableStore>;
    assert!(
        matches!(
            boot_failure(&observed)?,
            IncarnationStreamError::Durability(_)
        ),
        "a rotation whose epoch append fails must fail the boot, loudly"
    );
    assert_eq!(
        epoch_row_count(&inner, 2)?,
        0,
        "the refused rotation wrote no durable byte of its successor"
    );
    assert_eq!(anchor(&inner)?, Some(1));

    let (recovered, connection_incarnation) = boot_and_allocate(&inner)?;
    assert_eq!(
        recovered, 2,
        "an incarnation with no durable claim on it is still available"
    );
    assert_eq!(connection_incarnation, ConnectionIncarnation::new(2, 0));
    Ok(())
}