liminal-rs 0.16.3

use std::io::{self, Write};
use std::sync::{Arc, Mutex};

use super::bridge::block_on;
use super::head_checkpoint_stop::HeadCheckpointStop;
use super::tests_memory_store::MemoryStore;
use super::{
    CountingStore, DurableStore, RoutedBatchEntry, RoutedBatchItem, head_checkpoint_key,
    recover_partition_head,
};

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

#[derive(Clone, Default)]
struct Log(Arc<Mutex<Vec<u8>>>);
impl Write for Log {
    fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
        self.0
            .lock()
            .map_err(|_| io::Error::other("log lock poisoned"))?
            .extend_from_slice(bytes);
        Ok(bytes.len())
    }
    fn flush(&mut self) -> io::Result<()> {
        Ok(())
    }
}
impl<'a> tracing_subscriber::fmt::MakeWriter<'a> for Log {
    type Writer = Self;
    fn make_writer(&'a self) -> Self {
        self.clone()
    }
}
impl Log {
    fn capture<T>(&self, operation: impl FnOnce() -> T) -> T {
        let subscriber = tracing_subscriber::fmt()
            .with_max_level(tracing::Level::TRACE)
            .with_ansi(false)
            .with_writer(self.clone())
            .finish();
        tracing::subscriber::with_default(subscriber, operation)
    }
    fn contents(&self) -> Result<String, Box<dyn std::error::Error>> {
        Ok(String::from_utf8(
            self.0.lock().map_err(|_| "log lock poisoned")?.clone(),
        )?)
    }
}

#[test]
fn a_log_under_its_first_checkpoint_logs_no_warning_and_records_its_head() -> TestResult {
    let store = MemoryStore::default();
    block_on(store.append_all("young:0", vec![vec![1]; 12], 0))??;
    let log = Log::default();
    assert_eq!(
        log.capture(|| block_on(recover_partition_head(&store, "young:0")))??,
        12
    );
    assert_eq!(
        block_on(store.read_value(&head_checkpoint_key("young:0")))??,
        Some(12)
    );
    let text = log.contents()?;
    assert_eq!(
        text.lines().filter(|line| line.contains("WARN")).count(),
        0,
        "{text}"
    );
    assert_eq!(
        text.lines()
            .filter(|line| line.contains("INFO")
                && line.contains(
                    "durable channel head recorded for a stream under its first checkpoint"
                )
                && line.contains("rows=12"))
            .count(),
        1,
        "{text}"
    );
    Ok(())
}

#[test]
fn a_regressed_or_absent_head_still_warns_by_name() -> TestResult {
    let store = MemoryStore::default();
    block_on(store.append_all("refused:0", vec![vec![1]; 12], 0))??;
    block_on(store.cas(&head_checkpoint_key("refused:0"), 0, 13))??;
    let log = Log::default();
    assert_eq!(
        log.capture(|| block_on(recover_partition_head(&store, "refused:0")))??,
        12
    );
    let text = log.contents()?;
    assert_eq!(
        text.lines()
            .filter(|line| line.contains("WARN")
                && line.contains("the recorded head 13 claims rows the log does not hold"))
            .count(),
        1,
        "{text}"
    );
    Ok(())
}

type SeededHeads = (Arc<CountingStore<MemoryStore>>, HeadCheckpointStop);

fn seeded() -> Result<SeededHeads, Box<dyn std::error::Error>> {
    let store = Arc::new(CountingStore::new(MemoryStore::default()));
    for stream in ["young:0", "busy:0", "idle:0"] {
        block_on(store.append_all(stream, vec![vec![1]; 256], 0))??;
        block_on(store.cas(&head_checkpoint_key(stream), 0, 256))??;
    }
    let checkpoint = HeadCheckpointStop::new(store.clone());
    block_on(checkpoint.append_all("young:0", vec![vec![1]; 12], 256))??;
    block_on(checkpoint.append_all("busy:0", vec![vec![1]; 300], 256))??;
    Ok((store, checkpoint))
}

#[test]
fn a_clean_stop_records_the_head_of_every_stream_that_appended_since_its_last_head() -> TestResult {
    let (store, checkpoint) = seeded()?;
    let before = store.snapshot();
    block_on(checkpoint.checkpoint_heads())??;
    assert_eq!(
        store.snapshot().since(&before).cas_calls,
        2,
        "exactly two head CAS calls"
    );
    for (stream, next) in [("young:0", 268), ("busy:0", 556), ("idle:0", 256)] {
        assert_eq!(
            block_on(store.read_value(&head_checkpoint_key(stream)))??,
            Some(next)
        );
    }
    let before = store.snapshot();
    block_on(checkpoint.checkpoint_heads())??;
    assert_eq!(
        store.snapshot().since(&before).cas_calls,
        0,
        "idle stop writes no heads"
    );
    Ok(())
}

#[test]
fn a_start_after_a_clean_stop_walks_no_rows() -> TestResult {
    let (store, checkpoint) = seeded()?;
    block_on(checkpoint.checkpoint_heads())??;
    for (stream, next) in [("young:0", 268), ("busy:0", 556), ("idle:0", 256)] {
        let before = store.snapshot();
        assert_eq!(
            block_on(recover_partition_head(store.as_ref(), stream))??,
            next
        );
        let counts = store.snapshot().since(&before);
        assert_eq!(counts.read_at_calls, 1, "one verify row for {stream}");
        assert_eq!(counts.read_at_rows, 1);
        assert_eq!(
            counts.read_from_calls, 0,
            "zero recovery pages for {stream}"
        );
        assert_eq!(counts.read_from_rows, 0);
    }
    Ok(())
}

/// A routed entry batch with a delete (DUR-028 R3) reaches the store
/// beneath as it is: the stop wraps the production store, so the checkpoint
/// commit that removes the body it supersedes must pass through it whole.
#[test]
fn a_routed_delete_through_the_stop_reaches_the_store_beneath() -> TestResult {
    let store = Arc::new(CountingStore::new(MemoryStore::default()));
    let checkpoint = HeadCheckpointStop::new(store.clone());
    let put = |key: &[u8], value: &[u8]| RoutedBatchItem {
        key: key.to_vec(),
        expected: None,
        value: value.to_vec(),
        ttl: None,
    };
    block_on(checkpoint.apply_routed_batch("route", vec![put(b"body", b"old")]))??;
    block_on(checkpoint.apply_routed_entries(
        "route",
        vec![
            RoutedBatchEntry::Delete {
                key: b"body".to_vec(),
                expected: haematite::tree::Hash::of(b"old"),
            },
            RoutedBatchEntry::Put(put(b"locator", b"new")),
        ],
    ))??;
    assert_eq!(block_on(store.get_routed("route", b"body"))??, None);
    assert_eq!(
        block_on(store.get_routed("route", b"locator"))??,
        Some(b"new".to_vec())
    );
    Ok(())
}