liminal-server 0.20.2

//! DUR-018: the receipt is in the checkpoint's batch before any row goes,
//! the cut follows and answers per source, a repeated cut writes no second
//! receipt, the ledger reads back proved, a corrupt receipt is a named
//! defect, and a server without the section says once that pruning is off.

use std::error::Error;
use std::io;
use std::sync::{Arc, Mutex};

use liminal::durability::bridge::block_on;
use liminal::durability::{DurableStore, RoutedBatchItem, StoredEntry};

use super::ProductionParticipantHandler;
use super::log::STREAM_PREFIX;
use super::operation_batch::OperationBatch;
use super::outbox_log::OUTBOX_STREAM_PREFIX;
use super::prune::{
    PruneLedgerError, PruneReceipt, PrunedStream, RecipientAcknowledgement, RowSource,
    StreamBounds, cut, plan, prune_receipts, receipt_key, stage, unwritten,
};
use super::restore_checkpoint_locator::checkpoint_conversation_route;
use super::routed_stream_codec::{encode_head, encode_row, head_key, row_key};
use super::tests::test_participant_config;
use super::tests_memory_store::MemoryStore;

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

const CONVERSATION: u64 = 7;
const ROWS: u64 = 300;
const KEEP: u64 = 100;

fn streams() -> [String; 2] {
    [
        format!("{STREAM_PREFIX}{CONVERSATION}"),
        format!("{OUTBOX_STREAM_PREFIX}{CONVERSATION}"),
    ]
}

/// Routed rows 0..ROWS and a head at ROWS for both of the conversation's
/// streams, as the wired-boot head writes them.
fn seeded() -> Result<MemoryStore, Box<dyn Error>> {
    let store = MemoryStore::default();
    let route = checkpoint_conversation_route(CONVERSATION);
    let mut items = Vec::new();
    for stream in streams() {
        for sequence in 0..ROWS {
            items.push(RoutedBatchItem {
                key: row_key(&stream, sequence)?,
                expected: None,
                value: encode_row(&StoredEntry {
                    payload: vec![1, 2, 3],
                    sequence,
                    timestamp: 0,
                })?,
                ttl: None,
            });
        }
        items.push(RoutedBatchItem {
            key: head_key(&stream)?,
            expected: None,
            value: encode_head(ROWS).to_vec(),
            ttl: None,
        });
    }
    block_on(store.apply_routed_batch(&route, items))??;
    Ok(store)
}

fn acknowledged_plan() -> Result<super::prune::PrunePlan, Box<dyn Error>> {
    let bounds = StreamBounds {
        checkpoint_position: ROWS,
        acknowledged_through: ROWS,
    };
    plan(
        Some(KEEP),
        CONVERSATION,
        bounds,
        bounds,
        &[RecipientAcknowledgement {
            participant_id: 1,
            acknowledged_through: ROWS,
        }],
        10_000,
    )
    .ok_or_else(|| "rows lie below the kept window".into())
}

fn commit_receipts(store: &MemoryStore, plan: &super::prune::PrunePlan) -> TestResult {
    let route = checkpoint_conversation_route(CONVERSATION);
    let mut batch = OperationBatch::new(&route);
    stage(&mut batch, plan)?;
    block_on(batch.commit(store, || ()))??;
    Ok(())
}

#[test]
fn the_receipt_is_committed_first_and_the_cut_answers_per_source() -> TestResult {
    let store = seeded()?;
    let route = checkpoint_conversation_route(CONVERSATION);
    let plan = acknowledged_plan()?;
    assert_eq!(plan.receipts.len(), 2);
    commit_receipts(&store, &plan)?;
    for receipt in &plan.receipts {
        assert_eq!(receipt.cut_before, ROWS - KEEP);
        assert!(
            block_on(store.get_routed(&route, &receipt.key()))??.is_some(),
            "the receipt is durable before any row goes"
        );
    }
    for stream in streams() {
        assert!(block_on(store.get_routed(&route, &row_key(&stream, 0)?))??.is_some());
    }

    let outcomes = block_on(cut(&store, &plan))?;
    assert_eq!(
        outcomes.len(),
        4,
        "legacy and routed, per stream: {outcomes:?}"
    );
    for outcome in &outcomes {
        match outcome.source {
            RowSource::Routed => {
                assert_eq!(outcome.refused, None, "{outcome:?}");
                assert_eq!(outcome.removed, ROWS - KEEP, "{outcome:?}");
            }
            RowSource::Legacy => {
                // The memory store has no legacy delete: refused by name,
                // the receipt stands, the cut is not claimed.
                assert_eq!(outcome.removed, 0);
                let refusal = outcome.refused.as_deref().unwrap_or_default();
                assert!(refusal.contains("cannot truncate"), "{outcome:?}");
            }
        }
    }
    for stream in streams() {
        assert!(
            block_on(store.get_routed(&route, &row_key(&stream, ROWS - KEEP - 1)?))??.is_none(),
            "the last row below the cut is gone"
        );
        assert!(
            block_on(store.get_routed(&route, &row_key(&stream, ROWS - KEEP)?))??.is_some(),
            "the first kept row stays"
        );
        assert_eq!(
            block_on(store.get_routed(&route, &head_key(&stream)?))??,
            Some(encode_head(ROWS).to_vec()),
            "the head record is outside the window"
        );
    }

    let receipts = block_on(prune_receipts(&store, CONVERSATION))??;
    assert_eq!(receipts.len(), 2);
    assert_eq!(
        receipts
            .iter()
            .map(|receipt| receipt.stream)
            .collect::<Vec<_>>(),
        [PrunedStream::Operations, PrunedStream::Outbox]
    );
    for receipt in &receipts {
        assert!(receipt.cut_before <= receipt.acknowledged_through);
        assert!(receipt.acknowledged_through <= receipt.checkpoint_position);
    }
    Ok(())
}

#[test]
fn a_cut_repeated_to_the_same_point_writes_no_second_receipt() -> TestResult {
    let store = seeded()?;
    let plan = acknowledged_plan()?;
    let first = block_on(unwritten(&store, CONVERSATION, plan.clone()))??;
    assert_eq!(
        first.as_ref(),
        Some(&plan),
        "nothing written yet: the whole plan"
    );
    commit_receipts(&store, &plan)?;
    // The crash between receipt and cut: the next checkpoint plans the same
    // cut and finds every receipt already in the ledger.
    let again = block_on(unwritten(&store, CONVERSATION, plan.clone()))??;
    assert_eq!(again, None);
    let cut_again = block_on(cut(&store, &plan))?;
    let routed: Vec<_> = cut_again
        .iter()
        .filter(|outcome| outcome.source == RowSource::Routed)
        .collect();
    assert_eq!(routed.len(), 2);
    assert_eq!(
        routed[0].removed,
        ROWS - KEEP,
        "the first cut removes the rows"
    );
    let third = block_on(cut(&store, &plan))?;
    for outcome in third
        .iter()
        .filter(|outcome| outcome.source == RowSource::Routed)
    {
        assert_eq!(
            outcome.removed, 0,
            "a repeated cut removes nothing: {outcome:?}"
        );
        assert_eq!(outcome.refused, None);
    }
    assert_eq!(block_on(prune_receipts(&store, CONVERSATION))??.len(), 2);
    Ok(())
}

#[test]
fn a_hand_corrupted_receipt_reads_back_as_a_named_defect() -> TestResult {
    let store = seeded()?;
    let route = checkpoint_conversation_route(CONVERSATION);
    let corrupt = PruneReceipt {
        conversation_id: CONVERSATION,
        stream: PrunedStream::Operations,
        cut_before: 250,
        acknowledged_through: 200,
        checkpoint_position: ROWS,
        kept: KEEP,
        recipients: vec![RecipientAcknowledgement {
            participant_id: 1,
            acknowledged_through: 200,
        }],
        at_ms: 1,
    };
    block_on(store.apply_routed_batch(
        &route,
        vec![RoutedBatchItem {
            key: receipt_key(PrunedStream::Operations, 250),
            expected: None,
            value: serde_json::to_vec(&corrupt)?,
            ttl: None,
        }],
    ))??;
    let read = block_on(prune_receipts(&store, CONVERSATION))?;
    match read {
        Err(PruneLedgerError::Defect { key, error }) => {
            assert_eq!(key, "prune:operations:00000000000000000250");
            assert!(
                error.to_string().contains("above acknowledged_through 200"),
                "{error}"
            );
        }
        other => return Err(format!("a corrupt receipt was accepted: {other:?}").into()),
    }
    block_on(store.apply_routed_batch(
        &route,
        vec![RoutedBatchItem {
            key: receipt_key(PrunedStream::Outbox, 1),
            expected: None,
            value: b"not a receipt".to_vec(),
            ttl: None,
        }],
    ))??;
    // The first defect in key order is the one named: operations before outbox.
    assert!(matches!(
        block_on(prune_receipts(&store, CONVERSATION))?,
        Err(PruneLedgerError::Defect { .. })
    ));
    Ok(())
}

#[derive(Clone, Default)]
struct Captured(Arc<Mutex<Vec<u8>>>);

impl io::Write for Captured {
    fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
        self.0
            .lock()
            .map_err(|_| io::Error::other("captured log poisoned"))?
            .extend_from_slice(bytes);
        Ok(bytes.len())
    }

    fn flush(&mut self) -> io::Result<()> {
        Ok(())
    }
}

impl<'a> tracing_subscriber::fmt::MakeWriter<'a> for Captured {
    type Writer = Self;

    fn make_writer(&'a self) -> Self {
        self.clone()
    }
}

#[test]
fn a_server_without_the_section_says_once_that_pruning_is_off() -> TestResult {
    let captured = Captured::default();
    let subscriber = tracing_subscriber::fmt()
        .without_time()
        .with_ansi(false)
        .with_max_level(tracing::Level::INFO)
        .with_writer(captured.clone())
        .finish();
    let config = test_participant_config();
    assert_eq!(config.pruning, None, "the test config has no section");
    let handler = tracing::subscriber::with_default(subscriber, || {
        ProductionParticipantHandler::new(Arc::new(MemoryStore::default()), config)
    })?;
    drop(handler);
    let log = String::from_utf8(
        captured
            .0
            .lock()
            .map_err(|_| "captured log poisoned")?
            .clone(),
    )?;
    assert_eq!(
        log.matches("pruning is off").count(),
        1,
        "said once by name: {log}"
    );
    assert!(log.contains("keep_acknowledged_behind_checkpoint"), "{log}");
    Ok(())
}