liminal-server 0.21.1

//! The services read the durable delivery dedup namespace for in-flight claims
//! at start only over a store that can hold them: a store this process created
//! (the ephemeral store a server without a persistence path makes for itself,
//! or a persistent store this start creates) is empty, and reading it opened
//! every shard of it on the boot path before the listener bound.
//!
//! [`ScanCountingSubsystems`] builds the store the production way and counts
//! the work done through it (stores opened, writes, flushes, reads, rows read,
//! paged scans), so the work of a start is seen where it runs. The
//! `*_START` ceilings below are those counts for each kind of start, measured;
//! a change that makes a start do more must raise them here, in the open.

use std::path::Path;
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};

use liminal::durability::ClaimEvidence;

use liminal::conversation::ConversationSupervisor;
use liminal::durability::bridge::block_on;
use liminal::durability::{
    DedupCache, DurabilityError, DurableStore, ProcessingReceipt, ScanCursor, ScanPage,
    StoredEntry, open_ephemeral_rooted,
};

use super::{
    ChannelCluster, DELIVERY_DEDUP_NAMESPACE, ProductionSubsystems, StoreHistory, SubsystemFactory,
    THROWAWAY_SHARD_COUNT, ThrowawayStoreOpener, build_connection_services_via,
    build_durable_store,
};
use crate::ServerError;
use crate::config::types::{LimitsConfig, ServerConfig, ServicesConfig};
use crate::server::connection::ConnectionServices;
use crate::test_fixture::fixture_copy;

type TestResult = Result<(), Box<dyn std::error::Error>>;

/// The work done through the stores a start builds, counted where it runs.
#[derive(Debug, Default)]
struct WorkCounters {
    /// Durable stores opened: the services' store and any throwaway store.
    stores_opened: AtomicUsize,
    /// Appends, multi-event appends and compare-and-swaps.
    writes: AtomicUsize,
    flushes: AtomicUsize,
    /// Reads of any kind: a stream page, one event, a value or a paged scan.
    reads: AtomicUsize,
    /// Events and values the reads handed back.
    rows_read: AtomicUsize,
    scans: AtomicUsize,
    /// Rows read from the durable delivery dedup namespace's key streams.
    dedup_rows: AtomicUsize,
}

/// A snapshot of [`WorkCounters`], or the ceiling one start may reach.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
struct StartWork {
    stores_opened: usize,
    /// Haematite shard directories the start created on disk.
    shards_created: usize,
    writes: usize,
    flushes: usize,
    reads: usize,
    rows_read: usize,
    scans: usize,
}

impl StartWork {
    /// What `self` did beyond `before`.
    const fn since(self, before: Self) -> Self {
        Self {
            stores_opened: self.stores_opened - before.stores_opened,
            shards_created: self.shards_created - before.shards_created,
            writes: self.writes - before.writes,
            flushes: self.flushes - before.flushes,
            reads: self.reads - before.reads,
            rows_read: self.rows_read - before.rows_read,
            scans: self.scans - before.scans,
        }
    }

    /// Fails naming every count above `ceiling`.
    fn assert_within(self, ceiling: Self, name: &str) {
        let over: Vec<String> = [
            ("stores_opened", self.stores_opened, ceiling.stores_opened),
            (
                "shards_created",
                self.shards_created,
                ceiling.shards_created,
            ),
            ("writes", self.writes, ceiling.writes),
            ("flushes", self.flushes, ceiling.flushes),
            ("reads", self.reads, ceiling.reads),
            ("rows_read", self.rows_read, ceiling.rows_read),
            ("scans", self.scans, ceiling.scans),
        ]
        .into_iter()
        .filter(|(_, counted, allowed)| counted > allowed)
        .map(|(field, counted, allowed)| format!("{field} {counted} > {allowed}"))
        .collect();
        assert!(
            over.is_empty(),
            "a start rose above {name}: {}; counted {self:?}",
            over.join(", ")
        );
    }
}

/// A missing path refuses before the store or any shard can be opened.
const MISSING_PATH_START: StartWork = StartWork {
    stores_opened: 0,
    shards_created: 0,
    writes: 0,
    flushes: 0,
    reads: 0,
    rows_read: 0,
    scans: 0,
};

/// A start that creates its persistent store: the store is opened and
/// created, the delivery namespace's empty index start is recorded (one
/// compare-and-set, DUR-023, which creates the one shard it lands in), and
/// no index is read or walked, and no throwaway store is opened. Before
/// DUR-023 the start wrote nothing (`writes` 0, `shards_created` 0) and the next
/// start over a store anything else had been written to walked the
/// namespace once. From haematite 0.18.5 a new store boots its cold shards
/// side by side, so each of the server's eight shards has its (empty)
/// directory after the start; the one write is unchanged (`shards_created`
/// was 1 under 0.18.1).
const CREATED_PERSISTENT_START: StartWork = StartWork {
    stores_opened: 1,
    shards_created: 8,
    writes: 1,
    flushes: 0,
    reads: 0,
    rows_read: 0,
    scans: 0,
};

/// A start over a persistent store whose delivery namespace holds no claim
/// and whose empty index start an earlier start recorded: the store is
/// opened, the recorded index start is read (one read, one row) and the
/// empty index after it (one read, no row), and nothing is walked or
/// written. The one shard is the directory haematite 0.18.1 creates for the
/// index stream at that first read of it, a read of nothing that writes a
/// directory (measured on Dean, check 35: `shards_created` 1, `writes` 0).
/// Before DUR-023 such a store, if anything else had been written to it,
/// was walked once (`scans` at least 1) and its start recorded after.
const RECORDED_EMPTY_START: StartWork = StartWork {
    stores_opened: 1,
    shards_created: 1,
    writes: 0,
    flushes: 0,
    reads: 2,
    rows_read: 1,
    scans: 0,
};

/// A start over a persistent store holding [`SETTLED_EVIDENCED_CLAIMS`]
/// settled claims and none in flight: the recorded index start is read (one
/// read, one row), then the 44 index rows after it (one read) and the stream
/// of the key each names (44 reads of two rows each), and nothing is written.
/// The reads are bounded by the index window, not by the claims ever made.
const SETTLED_PERSISTENT_START: StartWork = StartWork {
    stores_opened: 1,
    shards_created: 0,
    writes: 0,
    flushes: 0,
    reads: 46,
    rows_read: 133,
    scans: 0,
};

/// Claims settled in the store [`SETTLED_PERSISTENT_START`] starts over: more
/// than one index window, so a start that read every claim ever made would
/// read more than twice what the window holds.
const SETTLED_EVIDENCED_CLAIMS: usize = 300;

/// A store that counts the work done through it.
#[derive(Debug)]
struct ScanCountingStore {
    inner: Arc<dyn DurableStore>,
    counters: Arc<WorkCounters>,
}

impl ScanCountingStore {
    fn count_read(&self, stream_key: &str, rows: usize) {
        self.counters.reads.fetch_add(1, Ordering::SeqCst);
        self.counters.rows_read.fetch_add(rows, Ordering::SeqCst);
        if stream_key.starts_with(&format!("{DELIVERY_DEDUP_NAMESPACE}:")) {
            self.counters.dedup_rows.fetch_add(rows, Ordering::SeqCst);
        }
    }

    fn count_write(&self) {
        self.counters.writes.fetch_add(1, Ordering::SeqCst);
    }
}

#[async_trait::async_trait]
impl DurableStore for ScanCountingStore {
    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.count_write();
        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.count_write();
        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> {
        let rows = self.inner.read_from(stream_key, offset, limit).await?;
        self.count_read(stream_key, rows.len());
        Ok(rows)
    }

    async fn read_at(
        &self,
        stream_key: &str,
        sequence: u64,
    ) -> Result<Option<StoredEntry>, DurabilityError> {
        let row = self.inner.read_at(stream_key, sequence).await?;
        self.count_read(stream_key, usize::from(row.is_some()));
        Ok(row)
    }

    async fn cas(&self, key: &str, old_value: u64, new_value: u64) -> Result<(), DurabilityError> {
        self.count_write();
        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> {
        let value = self.inner.read_value(key).await?;
        self.count_read(key, usize::from(value.is_some()));
        Ok(value)
    }

    async fn scan_page(
        &self,
        prefix: &str,
        cursor: Option<ScanCursor>,
        limit: usize,
    ) -> Result<ScanPage, DurabilityError> {
        self.counters.scans.fetch_add(1, Ordering::SeqCst);
        let page = self.inner.scan_page(prefix, cursor, limit).await?;
        self.count_read(prefix, page.entries.len());
        Ok(page)
    }

    async fn flush(&self) -> Result<(), DurabilityError> {
        self.counters.flushes.fetch_add(1, Ordering::SeqCst);
        self.inner.flush().await
    }
}

/// The production subsystems, with the durable store built by the production
/// branch logic (ephemeral stores rooted in an isolated directory) and wrapped
/// to count the work done through it.
struct ScanCountingSubsystems {
    ephemeral_root: tempfile::TempDir,
    counters: Arc<WorkCounters>,
}

impl ScanCountingSubsystems {
    fn new() -> Result<Self, std::io::Error> {
        Ok(Self {
            ephemeral_root: tempfile::tempdir()?,
            counters: Arc::new(WorkCounters::default()),
        })
    }

    fn scans(&self) -> usize {
        self.counters.scans.load(Ordering::SeqCst)
    }

    fn dedup_rows(&self) -> usize {
        self.counters.dedup_rows.load(Ordering::SeqCst)
    }

    /// The work counted so far, with the shard directories that exist under
    /// the ephemeral root and under `persistence_path`.
    fn work(&self, persistence_path: Option<&Path>) -> Result<StartWork, std::io::Error> {
        let mut shards = shard_directories(self.ephemeral_root.path(), 2)?;
        if let Some(path) = persistence_path {
            shards += shard_directories(path, 2)?;
        }
        Ok(StartWork {
            stores_opened: self.counters.stores_opened.load(Ordering::SeqCst),
            shards_created: shards,
            writes: self.counters.writes.load(Ordering::SeqCst),
            flushes: self.counters.flushes.load(Ordering::SeqCst),
            reads: self.counters.reads.load(Ordering::SeqCst),
            rows_read: self.counters.rows_read.load(Ordering::SeqCst),
            scans: self.scans(),
        })
    }
}

/// Haematite shard directories (`shard-<n>`) at most `depth` levels below
/// `root`: a database directory sits one level down, its shards one more.
fn shard_directories(root: &Path, depth: usize) -> Result<usize, std::io::Error> {
    if !root.is_dir() {
        return Ok(0);
    }
    let mut found = 0;
    for entry in std::fs::read_dir(root)? {
        let entry = entry?;
        if !entry.file_type()?.is_dir() {
            continue;
        }
        if entry.file_name().to_string_lossy().starts_with("shard-") {
            found += 1;
        } else if depth > 1 {
            found += shard_directories(&entry.path(), depth - 1)?;
        }
    }
    Ok(found)
}

impl SubsystemFactory for ScanCountingSubsystems {
    fn channel_cluster(
        &self,
        cluster_config: Option<&crate::config::types::ClusterConfig>,
    ) -> Result<ChannelCluster, ServerError> {
        ProductionSubsystems.channel_cluster(cluster_config)
    }

    fn conversation_supervisor(&self) -> Result<Arc<ConversationSupervisor>, ServerError> {
        ProductionSubsystems.conversation_supervisor()
    }

    fn durable_store(&self, persistence_path: &Path) -> Result<Arc<dyn DurableStore>, ServerError> {
        let inner = build_durable_store(persistence_path)?;
        self.counters.stores_opened.fetch_add(1, Ordering::SeqCst);
        Ok(Arc::new(ScanCountingStore {
            inner,
            counters: Arc::clone(&self.counters),
        }))
    }

    fn throwaway_store(&self) -> ThrowawayStoreOpener {
        let root = self.ephemeral_root.path().to_path_buf();
        let counters = Arc::clone(&self.counters);
        Box::new(move || {
            counters.stores_opened.fetch_add(1, Ordering::SeqCst);
            open_ephemeral_rooted(&root, THROWAWAY_SHARD_COUNT)
        })
    }
}

/// A channel-free full-profile config over `persistence_path`.
fn config(persistence_path: Option<&Path>) -> 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::new(),
        routing_rules: Vec::new(),
        persistence_path: persistence_path.map(Path::to_path_buf),
        cluster: None,
        auth: None,
        services: ServicesConfig::default(),
        limits: LimitsConfig::default(),
        participant: None,
        websocket: None,
    })
}

/// The services as a start leaves them: built, then listening, with the
/// resolve the listener starts run to its end.
fn started_via(
    config: &ServerConfig,
    subsystems: &dyn SubsystemFactory,
) -> Result<Arc<dyn ConnectionServices>, ServerError> {
    let services = build_connection_services_via(config, subsystems)?;
    if let Some(resolving) = Arc::clone(&services).listening() {
        resolving.join().map_err(|_| ServerError::ListenerAccept {
            message: "the claims resolve thread panicked".to_owned(),
        })?;
    }
    Ok(services)
}

#[test]
fn a_missing_path_refuses_without_store_work() -> TestResult {
    let subsystems = ScanCountingSubsystems::new()?;

    let refused = started_via(&config(None)?, &subsystems);
    assert!(
        matches!(refused, Err(ServerError::ConfigValidation { message }) if message.contains("no store is opened"))
    );
    subsystems
        .work(None)?
        .assert_within(MISSING_PATH_START, "MISSING_PATH_START");
    Ok(())
}

#[test]
fn a_start_that_creates_its_persistent_store_stays_within_its_ceiling() -> TestResult {
    let home = tempfile::tempdir()?;
    let subsystems = ScanCountingSubsystems::new()?;

    let services = started_via(&config(Some(home.path()))?, &subsystems)?;

    subsystems
        .work(Some(home.path()))?
        .assert_within(CREATED_PERSISTENT_START, "CREATED_PERSISTENT_START");
    drop(services);
    Ok(())
}

/// A store an earlier start created and nothing but its index start was
/// written to holds no claim: the next start over it reads the recorded start
/// and the empty index and walks nothing.
#[test]
fn a_restart_over_a_store_only_the_index_start_was_written_to_reads_the_index_and_walks_nothing()
-> TestResult {
    let home = tempfile::tempdir()?;
    let subsystems = ScanCountingSubsystems::new()?;
    drop(started_via(&config(Some(home.path()))?, &subsystems)?);

    let before = subsystems.work(Some(home.path()))?;
    let services = started_via(&config(Some(home.path()))?, &subsystems)?;

    subsystems
        .work(Some(home.path()))?
        .since(before)
        .assert_within(RECORDED_EMPTY_START, "RECORDED_EMPTY_START");
    drop(services);
    Ok(())
}

/// A store an earlier start created and wrote to, with no keyed publish in
/// it (every seeded fixture, every fresh install until its first keyed
/// publish): the next start over it is a start over a recorded store whose
/// delivery namespace is empty. It reads the index start the creating start
/// recorded and the empty index, and walks nothing (DUR-023). Before DUR-023
/// no start was recorded, and this start walked the namespace once, shard by
/// shard, to find nothing: 150 ms of the door's 420 ms resolve over an empty
/// namespace.
#[test]
fn a_restart_over_a_seeded_store_with_no_claim_walks_nothing() -> TestResult {
    let home = tempfile::tempdir()?;
    let subsystems = ScanCountingSubsystems::new()?;
    drop(started_via(&config(Some(home.path()))?, &subsystems)?);
    let store = subsystems.durable_store(home.path())?;
    block_on(store.append("seed/row", b"seed".to_vec(), 0))??;
    drop(store);

    let before = subsystems.work(Some(home.path()))?;
    let services = started_via(&config(Some(home.path()))?, &subsystems)?;

    subsystems
        .work(Some(home.path()))?
        .since(before)
        .assert_within(RECORDED_EMPTY_START, "RECORDED_EMPTY_START");
    assert_eq!(
        subsystems.scans(),
        0,
        "no start over this store walked the namespace"
    );
    drop(services);
    Ok(())
}

/// The namespace a running server leaves over the store its first start
/// created: every claim made with evidence, then completed or released. Built
/// once per test binary; each test starts over its own copy.
fn build_settled_evidenced_home(home: &Path) -> crate::test_fixture::FixtureResult<()> {
    let subsystems = ScanCountingSubsystems::new()?;
    drop(started_via(&config(Some(home))?, &subsystems)?);
    let store = subsystems.durable_store(home)?;
    let dedup = StoreHistory::Created.delivery_dedup(Arc::clone(&store));
    for index in 0..SETTLED_EVIDENCED_CLAIMS {
        let key = format!("settled-{index}");
        block_on(dedup.claim_with_evidence(&key, ClaimEvidence::new("orders", vec![0]), 1))??;
        if index % 2 == 0 {
            block_on(dedup.complete_receipt(&key, ProcessingReceipt::for_position(1)))??;
        } else {
            block_on(dedup.release_claim(&key))??;
        }
    }
    Ok(())
}

#[test]
fn a_start_over_settled_claims_stays_within_its_ceiling() -> TestResult {
    let home = fixture_copy("settled-evidenced-claims-300", build_settled_evidenced_home)?;
    let subsystems = ScanCountingSubsystems::new()?;

    let before = subsystems.work(Some(home.path()))?;
    let services = started_via(&config(Some(home.path()))?, &subsystems)?;

    subsystems
        .work(Some(home.path()))?
        .since(before)
        .assert_within(SETTLED_PERSISTENT_START, "SETTLED_PERSISTENT_START");
    drop(services);
    crate::test_fixture::pin_store_opens(
        module_path!(),
        "a_start_over_settled_claims_stays_within_its_ceiling",
    )?;
    Ok(())
}

/// What liminal-server 0.16.0 leaves: a persistent store holding claims and no
/// in-flight index. The first start over it walks the namespace once.
#[test]
fn a_store_written_before_the_index_is_walked_once_at_start() -> TestResult {
    let home = tempfile::tempdir()?;
    let subsystems = ScanCountingSubsystems::new()?;
    let store = subsystems.durable_store(home.path())?;
    // A claim without evidence names no key in the index, as 0.16.0's did not.
    let legacy = DedupCache::new(Arc::clone(&store), DELIVERY_DEDUP_NAMESPACE);
    block_on(legacy.claim_or_get("stranded", 1))??;
    drop(legacy);
    drop(store);

    let scans = subsystems.scans();
    let services = started_via(&config(Some(home.path()))?, &subsystems)?;
    assert!(
        subsystems.scans() > scans,
        "a store with claims and no index is walked to build it"
    );
    drop(services);

    let scans = subsystems.scans();
    let services = started_via(&config(Some(home.path()))?, &subsystems)?;
    assert_eq!(
        subsystems.scans(),
        scans,
        "the walk is recorded and not made again"
    );
    drop(services);
    Ok(())
}

/// Settled keys this test leaves in the durable dedup namespace between two
/// starts: enough that a start reading them is unmistakable.
const SETTLED_KEYS: usize = 600;

/// The namespace a long-running server leaves behind over the store its first
/// start created: every key completed or released, nothing in flight. Built
/// once per test binary; each test starts over its own copy.
fn build_settled_keys_home(home: &Path) -> crate::test_fixture::FixtureResult<()> {
    let subsystems = ScanCountingSubsystems::new()?;
    drop(started_via(&config(Some(home))?, &subsystems)?);
    let store = subsystems.durable_store(home)?;
    let dedup = StoreHistory::Created.delivery_dedup(Arc::clone(&store));
    for index in 0..SETTLED_KEYS {
        let key = format!("settled-{index}");
        block_on(dedup.claim_or_get(&key, 1))??;
        if index % 2 == 0 {
            block_on(dedup.complete_receipt(&key, ProcessingReceipt::for_position(1)))??;
        } else {
            block_on(dedup.release_claim(&key))??;
        }
    }
    Ok(())
}

#[test]
fn a_restart_reads_no_settled_claim_and_does_not_walk_the_namespace_again() -> TestResult {
    let home = fixture_copy("settled-keys-600", build_settled_keys_home)?;
    let subsystems = ScanCountingSubsystems::new()?;

    let scans = subsystems.scans();
    let rows = subsystems.dedup_rows();
    let services = started_via(&config(Some(home.path()))?, &subsystems)?;

    assert_eq!(
        subsystems.scans(),
        scans,
        "a start after the first reads the recorded in-flight claims, never the namespace"
    );
    assert_eq!(
        subsystems.dedup_rows(),
        rows,
        "a start reads no dedup record of a key that is not in flight"
    );
    drop(services);
    crate::test_fixture::pin_store_opens(
        module_path!(),
        "a_restart_reads_no_settled_claim_and_does_not_walk_the_namespace_again",
    )?;
    Ok(())
}