liminal-server 0.24.9

//! A store that tallies what it is asked, by the family of key it is asked
//! for, and the ceilings those tallies are held under: the real one-shard
//! store DUR-027 R7 drives, the crossing DUR-028 R3 counts and the
//! migration DUR-028 R4 counts.

use std::collections::BTreeSet;
use std::error::Error;
use std::path::Path;
use std::sync::{Arc, Mutex};

use haematite::{Database, DatabaseConfig, EventStore, NodeCacheBudget, Retention};
use liminal::durability::{
    DurabilityError, DurableStore, HaematiteStore, RoutedBatchEntry, RoutedBatchItem, ScanCursor,
    ScanPage, StoredEntry,
};

use super::log::{GENERATION_STREAM_PREFIX, STREAM_PREFIX};
use super::observer::OBSERVER_STREAM_KEY;
use super::outbox_log::OUTBOX_STREAM_PREFIX;
use super::registry::REGISTRY_STREAM_KEY;

type TestResult<T = ()> = Result<T, Box<dyn Error>>;

/// What a boot asked of the store, by the family of key it asked for.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub(super) struct Asked {
    pub(super) read_from_calls: u64,
    /// Rows `read_from` answered on the observer stream.
    pub(super) observer_rows: u64,
    /// Rows `read_from` answered on the registry stream.
    pub(super) registry_rows: u64,
    /// Rows `read_from` answered on conversation streams, and the streams.
    pub(super) conversation_rows: u64,
    pub(super) conversation_streams: BTreeSet<String>,
    pub(super) get_routed_calls: u64,
    pub(super) scan_routed_calls: u64,
    /// Routed batches (of items or of entries) alone.
    pub(super) apply_routed_batch_calls: u64,
    pub(super) remove_routed_range_calls: u64,
    pub(super) read_value_calls: u64,
    /// Durable commits: routed batches, appends, CAS moves and cuts.
    pub(super) commits: u64,
}

/// A real store that tallies what it is asked, by key family.
#[derive(Debug)]
pub(super) struct Tally {
    pub(super) inner: Arc<dyn DurableStore>,
    asked: Mutex<Asked>,
    /// The commit (counted from the last `take`) this store refuses once,
    /// before forwarding it: a crash between two batches, by arrangement.
    refuse_commit: Mutex<Option<u64>>,
}

fn is_conversation_stream(stream_key: &str) -> bool {
    stream_key.starts_with(STREAM_PREFIX)
        || stream_key.starts_with(GENERATION_STREAM_PREFIX)
        || stream_key.starts_with(OUTBOX_STREAM_PREFIX)
}

impl Tally {
    pub(super) fn new(inner: Arc<dyn DurableStore>) -> Arc<Self> {
        Arc::new(Self {
            inner,
            asked: Mutex::new(Asked::default()),
            refuse_commit: Mutex::new(None),
        })
    }

    /// Arranges for the `nth` commit after the next `take` to be refused
    /// once, before it reaches the store.
    pub(super) fn refuse_commit(&self, nth: u64) -> TestResult {
        *self.refuse_commit.lock().map_err(|_| "tally poisoned")? = Some(nth);
        Ok(())
    }

    pub(super) fn shared(self: &Arc<Self>) -> Arc<dyn DurableStore> {
        Arc::clone(self) as Arc<dyn DurableStore>
    }

    pub(super) fn take(&self) -> TestResult<Asked> {
        Ok(std::mem::take(
            &mut *self.asked.lock().map_err(|_| "tally poisoned")?,
        ))
    }

    fn note(&self, work: impl FnOnce(&mut Asked)) -> Result<(), DurabilityError> {
        let mut asked = self
            .asked
            .lock()
            .map_err(|_| DurabilityError::ConfigError("tally poisoned".to_owned()))?;
        work(&mut asked);
        drop(asked);
        Ok(())
    }

    fn commit(&self) -> Result<(), DurabilityError> {
        self.note(|asked| asked.commits += 1)?;
        let commits = self
            .asked
            .lock()
            .map_err(|_| DurabilityError::ConfigError("tally poisoned".to_owned()))?
            .commits;
        let mut refuse = self
            .refuse_commit
            .lock()
            .map_err(|_| DurabilityError::ConfigError("tally poisoned".to_owned()))?;
        let refused = *refuse == Some(commits);
        if refused {
            *refuse = None;
        }
        drop(refuse);
        if refused {
            return Err(DurabilityError::ConfigError(format!(
                "commit {commits} refused by the fixture, as arranged"
            )));
        }
        Ok(())
    }
}

#[async_trait::async_trait]
impl DurableStore for Tally {
    fn reclaim_spine(&self) -> Option<Arc<haematite::EventStore>> {
        self.inner.reclaim_spine()
    }

    async fn apply_routed_batch(
        &self,
        route_key: &str,
        items: Vec<RoutedBatchItem>,
    ) -> Result<(), DurabilityError> {
        self.note(|asked| asked.apply_routed_batch_calls += 1)?;
        self.commit()?;
        self.inner.apply_routed_batch(route_key, items).await
    }

    async fn apply_routed_entries(
        &self,
        route_key: &str,
        entries: Vec<RoutedBatchEntry>,
    ) -> Result<(), DurabilityError> {
        self.note(|asked| asked.apply_routed_batch_calls += 1)?;
        self.commit()?;
        self.inner.apply_routed_entries(route_key, entries).await
    }

    async fn get_routed(
        &self,
        route_key: &str,
        key: &[u8],
    ) -> Result<Option<Vec<u8>>, DurabilityError> {
        self.note(|asked| asked.get_routed_calls += 1)?;
        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.commit()?;
        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.commit()?;
        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?;
        let count = u64::try_from(rows.len()).unwrap_or(u64::MAX);
        self.note(|asked| {
            asked.read_from_calls += 1;
            if stream_key == OBSERVER_STREAM_KEY {
                asked.observer_rows += count;
            } else if stream_key == REGISTRY_STREAM_KEY {
                asked.registry_rows += count;
            } else if is_conversation_stream(stream_key) {
                asked.conversation_rows += count;
                asked.conversation_streams.insert(stream_key.to_owned());
            }
        })?;
        Ok(rows)
    }

    async fn cas(&self, key: &str, old_value: u64, new_value: u64) -> Result<(), DurabilityError> {
        self.commit()?;
        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.note(|asked| asked.read_value_calls += 1)?;
        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
    }

    async fn truncate_before(
        &self,
        stream_key: &str,
        before_seq: u64,
    ) -> Result<u64, DurabilityError> {
        self.commit()?;
        self.inner.truncate_before(stream_key, before_seq).await
    }

    async fn scan_routed(
        &self,
        route_key: &str,
        from: &[u8],
        to: &[u8],
    ) -> Result<Vec<(Vec<u8>, Vec<u8>)>, DurabilityError> {
        self.note(|asked| asked.scan_routed_calls += 1)?;
        self.inner.scan_routed(route_key, from, to).await
    }

    async fn remove_routed_range(
        &self,
        route_key: &str,
        from: &[u8],
        to: &[u8],
    ) -> Result<u64, DurabilityError> {
        self.note(|asked| asked.remove_routed_range_calls += 1)?;
        self.commit()?;
        self.inner.remove_routed_range(route_key, from, to).await
    }
}

fn runtime() -> haematite::RuntimeOptions {
    haematite::RuntimeOptions {
        long_wait_interval: std::time::Duration::from_secs(5),
        reclaim_start: haematite::ReclaimStart::AtOpen,
    }
}

/// A real on-disk store of `shard_count` shards at `path`, opened when one
/// is there and created otherwise, behind a tally.
pub(super) fn disk_store(path: &Path, shard_count: usize) -> TestResult<Arc<Tally>> {
    let database = if path.join("config.json").exists() {
        Database::open(path, runtime())?
    } else {
        Database::create(
            DatabaseConfig {
                data_dir: path.to_path_buf(),
                shard_count,
                distributed: None,
                executor_threads: None,
                node_cache_budget: Some(NodeCacheBudget::Unlimited),
                retention: Retention::reclaim(0),
            },
            runtime(),
        )?
    };
    liminal::durability::record_store_opened();
    Ok(Tally::new(Arc::new(HaematiteStore::new(Arc::new(
        EventStore::new(database),
    )))))
}

/// The ceilings every store-count table is read from.
const CEILINGS: &str = include_str!("../../../../tests/ceilings/store-counts.toml");

/// The ceiling `[table] counter` names in `tests/ceilings/store-counts.toml`.
pub(super) fn ceiling(table: &str, counter: &str) -> TestResult<u64> {
    let tables: toml::Table = CEILINGS.parse()?;
    let value = tables
        .get(table)
        .and_then(toml::Value::as_table)
        .and_then(|found| found.get(counter))
        .and_then(toml::Value::as_integer)
        .ok_or_else(|| format!("store-counts.toml has no [{table}] {counter}"))?;
    Ok(u64::try_from(value)?)
}

/// `measured` against its ceiling: over fails with all the figures, more
/// than ten per cent under says the ceiling can drop.
pub(super) fn under(table: &str, counter: &str, measured: u64) -> TestResult {
    let ceiling = ceiling(table, counter)?;
    if measured > ceiling {
        return Err(format!(
            "{table}.{counter} is over its ceiling: ceiling {ceiling}, measured {measured}"
        )
        .into());
    }
    if measured < ceiling.saturating_sub(ceiling / 10) {
        println!("ceiling can drop: {table}.{counter} {ceiling} -> {measured}");
    }
    Ok(())
}