liminal-rs 0.16.2

//! DUR-024 R2: a store's admission bound is read from the store, never
//! assumed. A `HaematiteStore` answers the queue byte limit its database was
//! opened with; a store that names no bound answers `None`; the memory store
//! answers the bound a test gave it and refuses a request over it by name.

use std::sync::Arc;

use haematite::{Database, DatabaseConfig, DatabaseOpenOptions, EventStore};

use super::store::HaematiteStore;
use super::tests_memory_store::MemoryStore;
use super::{DurabilityError, DurableStore, RoutedBatchItem};

const EIGHT_MIB: usize = 8 * 1024 * 1024;

fn database_config(dir: &std::path::Path) -> DatabaseConfig {
    DatabaseConfig {
        data_dir: dir.to_path_buf(),
        shard_count: 1,
        distributed: None,
        executor_threads: None,
        node_cache_budget: Some(haematite::NodeCacheBudget::Unlimited),
        retention: haematite::Retention::reclaim(0),
    }
}

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

#[test]
fn a_haematite_store_answers_the_bound_its_database_was_opened_with()
-> Result<(), Box<dyn std::error::Error>> {
    let dir = tempfile::tempdir()?;
    let database = Database::create_with_options(
        database_config(dir.path()),
        runtime(),
        DatabaseOpenOptions::new(EIGHT_MIB)?,
    )?;
    super::counting::record_store_opened();
    let store = HaematiteStore::new(Arc::new(EventStore::new(database)));
    assert_eq!(store.max_request_bytes(), Some(EIGHT_MIB));
    Ok(())
}

#[test]
fn a_haematite_store_opened_by_default_answers_the_default_bound()
-> Result<(), Box<dyn std::error::Error>> {
    let dir = tempfile::tempdir()?;
    let database = Database::create(database_config(dir.path()), runtime())?;
    super::counting::record_store_opened();
    let store = HaematiteStore::new(Arc::new(EventStore::new(database)));
    // Read from the open database, never a constant copied into liminal: the
    // default is whatever `DatabaseOpenOptions::default()` says it is.
    assert_eq!(
        store.max_request_bytes(),
        Some(DatabaseOpenOptions::default().queue_byte_limit().get())
    );
    Ok(())
}

#[test]
fn a_store_that_names_no_bound_answers_none() {
    assert_eq!(MemoryStore::default().max_request_bytes(), None);
}

#[test]
fn a_bounded_memory_store_answers_its_bound_and_refuses_over_it_by_name()
-> Result<(), Box<dyn std::error::Error>> {
    let store = MemoryStore::with_request_bound(64);
    assert_eq!(store.max_request_bytes(), Some(64));
    let item = |key: &[u8], bytes: usize| RoutedBatchItem {
        key: key.to_vec(),
        expected: None,
        value: vec![7; bytes],
        ttl: None,
    };
    super::bridge::block_on(
        store.apply_routed_batch("route", vec![item(b"a", 32), item(b"b", 32)]),
    )??;
    let refused = super::bridge::block_on(
        store.apply_routed_batch("route", vec![item(b"c", 33), item(b"d", 32)]),
    )?;
    match refused {
        Err(DurabilityError::ConfigError(words)) => assert_eq!(
            words,
            "memory_store_request_over_bound: request_bytes=65 bound=64"
        ),
        other => return Err(format!("a request over the bound was not refused: {other:?}").into()),
    }
    assert_eq!(
        super::bridge::block_on(store.get_routed("route", b"c"))??,
        None,
        "a refused request wrote nothing"
    );
    Ok(())
}