liminal-server 0.24.9

//! 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. DUR-024 R4: a server without the section prunes with the pinned
//! default and says so once, the default cuts acknowledged rows behind the
//! checkpoint and nothing unacknowledged, and a configured value governs.

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::{
    DEFAULT_KEEP_ACKNOWLEDGED_BEHIND_CHECKPOINT, HeldRecords, PruneLedgerError, PruneReceipt,
    PrunedStream, RecipientAcknowledgement, RowSource, StreamBounds, cut, effective_keep, 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>> {
    seeded_rows(ROWS)
}

/// Routed rows `0..rows` and a head at `rows` for both streams.
fn seeded_rows(rows: u64) -> 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,
        covered_through: ROWS,
    };
    plan(
        Some(KEEP),
        CONVERSATION,
        bounds,
        bounds,
        &[RecipientAcknowledgement {
            participant_id: 1,
            acknowledged_through: ROWS,
        }],
        HeldRecords::default(),
        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.covered_through);
        assert!(receipt.covered_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,
        covered_through: 200,
        checkpoint_position: ROWS,
        kept: KEEP,
        recipients: vec![RecipientAcknowledgement {
            participant_id: 1,
            acknowledged_through: 200,
        }],
        held: HeldRecords::default(),
        at_ms: 1,
    };
    block_on(store.apply_routed_batch(
        &route,
        vec![RoutedBatchItem {
            key: receipt_key(CONVERSATION, 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 }) => {
            let expected = receipt_key(CONVERSATION, PrunedStream::Operations, 250);
            assert_eq!(key, String::from_utf8_lossy(&expected));
            assert!(
                key.ends_with("prune:operations:00000000000000000250"),
                "{key}"
            );
            assert!(
                error.to_string().contains("above covered_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(CONVERSATION, 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 pruning_is_on_by_default_with_the_pinned_slack() -> 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 on by default: 512 rows behind the checkpoint")
            .count(),
        1,
        "said once by name: {log}"
    );
    assert_eq!(
        log.matches("pruning is off").count(),
        0,
        "pruning is never off: {log}"
    );
    assert_eq!(
        effective_keep(None),
        DEFAULT_KEEP_ACKNOWLEDGED_BEHIND_CHECKPOINT
    );
    assert_eq!(DEFAULT_KEEP_ACKNOWLEDGED_BEHIND_CHECKPOINT, 512);
    Ok(())
}

/// The cut the default window decides over `rows` rows at the checkpoint,
/// with every recipient acknowledged through `acknowledged_through`.
fn default_plan(
    rows: u64,
    acknowledged_through: u64,
) -> Result<super::prune::PrunePlan, Box<dyn Error>> {
    let bounds = StreamBounds {
        checkpoint_position: rows,
        covered_through: acknowledged_through,
    };
    plan(
        Some(effective_keep(None)),
        CONVERSATION,
        bounds,
        bounds,
        &[RecipientAcknowledgement {
            participant_id: 1,
            acknowledged_through,
        }],
        HeldRecords::default(),
        10_000,
    )
    .ok_or_else(|| "rows lie below the kept window".into())
}

/// Routed rows present in `[0, rows)` on the operations stream.
fn present_rows(store: &MemoryStore, rows: u64) -> Result<u64, Box<dyn Error>> {
    let route = checkpoint_conversation_route(CONVERSATION);
    let stream = format!("{STREAM_PREFIX}{CONVERSATION}");
    let mut present = 0;
    for sequence in 0..rows {
        if block_on(store.get_routed(&route, &row_key(&stream, sequence)?))??.is_some() {
            present += 1;
        }
    }
    Ok(present)
}

#[test]
fn default_prune_cuts_acknowledged_rows_behind_the_checkpoint() -> TestResult {
    const ROWS_AT_CHECKPOINT: u64 = 2_000;
    let store = seeded_rows(ROWS_AT_CHECKPOINT)?;
    let plan = default_plan(ROWS_AT_CHECKPOINT, ROWS_AT_CHECKPOINT)?;
    commit_receipts(&store, &plan)?;
    let outcomes = block_on(cut(&store, &plan))?;
    let routed: Vec<_> = outcomes
        .iter()
        .filter(|outcome| outcome.source == RowSource::Routed)
        .collect();
    assert_eq!(routed.len(), 2, "{outcomes:?}");
    for outcome in &routed {
        assert_eq!(outcome.refused, None, "{outcome:?}");
        assert_eq!(outcome.cut_before, ROWS_AT_CHECKPOINT - 512, "{outcome:?}");
        assert_eq!(outcome.removed, ROWS_AT_CHECKPOINT - 512, "{outcome:?}");
    }
    assert_eq!(
        present_rows(&store, ROWS_AT_CHECKPOINT)?,
        512,
        "the 512 rows behind the checkpoint remain; everything acknowledged below them is gone"
    );
    let receipts = block_on(prune_receipts(&store, CONVERSATION))??;
    assert_eq!(receipts.len(), 2);
    for receipt in &receipts {
        assert_eq!(receipt.cut_before, ROWS_AT_CHECKPOINT - 512);
        assert_eq!(receipt.covered_through, ROWS_AT_CHECKPOINT);
        assert_eq!(receipt.checkpoint_position, ROWS_AT_CHECKPOINT);
        assert_eq!(receipt.kept, 512);
    }

    // A row past the cover is kept: covered through 1,000 of 2,000, the
    // cut stops 512 below the cover, never below the checkpoint.
    let store = seeded_rows(ROWS_AT_CHECKPOINT)?;
    let plan = default_plan(ROWS_AT_CHECKPOINT, 1_000)?;
    commit_receipts(&store, &plan)?;
    let outcomes = block_on(cut(&store, &plan))?;
    for outcome in outcomes
        .iter()
        .filter(|outcome| outcome.source == RowSource::Routed)
    {
        assert_eq!(outcome.cut_before, 1_000 - 512, "{outcome:?}");
    }
    assert_eq!(
        present_rows(&store, ROWS_AT_CHECKPOINT)?,
        ROWS_AT_CHECKPOINT - (1_000 - 512),
        "every row past the cover stays, with the 512 covered rows behind it"
    );
    Ok(())
}

#[test]
fn configured_value_still_governs() -> TestResult {
    let configured = crate::config::types::PruningConfig {
        keep_acknowledged_behind_checkpoint: 64,
    };
    assert_eq!(effective_keep(Some(configured)), 64);
    let store = seeded()?;
    let bounds = StreamBounds {
        checkpoint_position: ROWS,
        covered_through: ROWS,
    };
    let plan = plan(
        Some(effective_keep(Some(configured))),
        CONVERSATION,
        bounds,
        bounds,
        &[RecipientAcknowledgement {
            participant_id: 1,
            acknowledged_through: ROWS,
        }],
        HeldRecords::default(),
        10_000,
    )
    .ok_or("rows lie below the kept window")?;
    commit_receipts(&store, &plan)?;
    let outcomes = block_on(cut(&store, &plan))?;
    for outcome in outcomes
        .iter()
        .filter(|outcome| outcome.source == RowSource::Routed)
    {
        assert_eq!(outcome.cut_before, ROWS - 64, "{outcome:?}");
    }
    assert_eq!(present_rows(&store, ROWS)?, 64);
    Ok(())
}