liminal-server 0.24.9

//! Prune acknowledged conversation rows behind a checkpoint (DUR-018).
//!
//! liminal is not the system of record: a row is durable until it is
//! delivered and acknowledged, then the app store owns it. At every
//! checkpoint a conversation's operation and outbox rows that lie behind the
//! checkpoint and behind every recipient's acknowledgement are cut, keeping a
//! configured, pinned number behind the checkpoint. A receipt is staged into
//! the checkpoint's own batch before any row goes, so a reader proves from
//! the receipt that nothing unacknowledged was cut. Nothing here bounds a
//! conversation log by row count or by age.

use liminal::durability::{DurabilityError, DurableStore, RoutedBatchItem};
use serde::{Deserialize, Serialize};

use super::log::{OperationLogError, STREAM_PREFIX};
use super::operation_batch::OperationBatch;
use super::outbox_log::OUTBOX_STREAM_PREFIX;
use super::restore_checkpoint_locator::checkpoint_conversation_route;
use super::routed_stream_codec;

/// Prefix of a conversation's prune ledger keys inside its checkpoint route.
pub(super) const PRUNE_RECEIPT_PREFIX: &str = "prune:";

/// Acknowledged rows kept behind the newest checkpoint when
/// `[participant.pruning]` is absent (DUR-024 R4).
///
/// Two checkpoint intervals (`CHECKPOINT_EVERY` is 256) of slack, so a
/// reader resuming from the previous checkpoint still finds every row it
/// can ask for. Waffles' number by Tom's rule of 10 October 2026; pinned in
/// `tests/ceilings/checkpoint.toml`. A configured value still governs.
pub const DEFAULT_KEEP_ACKNOWLEDGED_BEHIND_CHECKPOINT: u64 = 512;

/// The kept window a server prunes with: the configured one, or the pinned
/// default when the section is absent. Pruning is never off.
#[must_use]
pub const fn effective_keep(pruning: Option<crate::config::types::PruningConfig>) -> u64 {
    match pruning {
        Some(configured) => configured.keep_acknowledged_behind_checkpoint,
        None => DEFAULT_KEEP_ACKNOWLEDGED_BEHIND_CHECKPOINT,
    }
}

/// The key just past every receipt key: `;` follows `:` in byte order, so
/// `[PRUNE_RECEIPT_PREFIX, PRUNE_RECEIPT_PREFIX_END)` is exactly the ledger.
const PRUNE_RECEIPT_PREFIX_END: &[u8] = b"prune;";

/// The two conversation streams a cut applies to.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum PrunedStream {
    /// The operation log.
    Operations,
    /// The outbox log.
    Outbox,
}

/// A recipient's acknowledgement at the time of the cut.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct RecipientAcknowledgement {
    /// The recipient.
    pub participant_id: u64,
    /// Every delivery at or below this delivery sequence is acknowledged by
    /// it. A delivery sequence, not a row of either stream.
    pub acknowledged_through: u64,
}

/// The undelivered records the checkpoint a cut sits behind carries: every
/// record some recipient owed it has not acknowledged, body and all. Boot
/// restores them from the checkpoint and redelivery reads them from the
/// outbox, so a row cut behind the checkpoint is never the only copy of
/// anything owed.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct HeldRecords {
    /// Records held, each owed to at least one recipient.
    pub records: u64,
    /// Recipient obligations across them.
    pub obligations: u64,
    /// The lowest delivery sequence held; `None` when nothing is held.
    pub lowest_delivery: Option<u64>,
}

/// What a cut was decided on; written before the cut, read to prove it.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct PruneReceipt {
    /// The conversation.
    pub conversation_id: u64,
    /// The stream the cut applies to.
    pub stream: PrunedStream,
    /// Rows below this sequence are cut.
    pub cut_before: u64,
    /// Every row below here is acknowledged by every recipient owed it, or
    /// its record is among those `held` by the checkpoint at
    /// `checkpoint_position`.
    #[serde(alias = "acknowledged_through")]
    pub covered_through: u64,
    /// The checkpoint the cut sits behind.
    pub checkpoint_position: u64,
    /// The rows kept behind the checkpoint, from config.
    pub kept: u64,
    /// Each recipient's acknowledgement at the cut.
    pub recipients: Vec<RecipientAcknowledgement>,
    /// What the checkpoint carries undelivered at the cut.
    #[serde(default)]
    pub held: HeldRecords,
    /// Milliseconds since the Unix epoch on the server clock.
    pub at_ms: u64,
}

impl PruneReceipt {
    /// Proves the receipt's own arithmetic: nothing was cut that is not
    /// either acknowledged or held by the checkpoint the cut sits behind.
    ///
    /// # Errors
    /// The receipt names a cut above its cover, a cover above its
    /// checkpoint, or a held summary that does not add up.
    pub const fn prove(&self) -> Result<(), PruneProofError> {
        if self.cut_before > self.covered_through {
            return Err(PruneProofError::CutAboveCover {
                cut_before: self.cut_before,
                covered_through: self.covered_through,
            });
        }
        if self.covered_through > self.checkpoint_position {
            return Err(PruneProofError::CoverAboveCheckpoint {
                covered_through: self.covered_through,
                checkpoint_position: self.checkpoint_position,
            });
        }
        let held = self.held;
        if (held.records == 0) != held.lowest_delivery.is_none() || held.obligations < held.records
        {
            return Err(PruneProofError::HeldInconsistent { held });
        }
        Ok(())
    }

    /// The ledger key of this receipt inside the conversation's route.
    #[must_use]
    pub fn key(&self) -> Vec<u8> {
        receipt_key(self.conversation_id, self.stream, self.cut_before)
    }
}

/// A receipt's ledger key under the namespace (DUR-028 R2): one per cut point, so no second.
#[must_use]
pub(super) fn receipt_key(conversation: u64, stream: PrunedStream, cut_before: u64) -> Vec<u8> {
    super::route_keys::namespaced(conversation, &legacy_receipt_key(stream, cut_before))
}

/// The bare key a store written before DUR-028 holds the receipt under.
pub(super) fn legacy_receipt_key(stream: PrunedStream, cut_before: u64) -> Vec<u8> {
    let name = match stream {
        PrunedStream::Operations => "operations",
        PrunedStream::Outbox => "outbox",
    };
    format!("{PRUNE_RECEIPT_PREFIX}{name}:{cut_before:020}").into_bytes()
}

/// A receipt that does not prove its own cut.
#[derive(Debug, PartialEq, Eq, thiserror::Error)]
pub enum PruneProofError {
    #[error("cut_before {cut_before} is above covered_through {covered_through}")]
    CutAboveCover {
        cut_before: u64,
        covered_through: u64,
    },
    #[error("covered_through {covered_through} is above checkpoint_position {checkpoint_position}")]
    CoverAboveCheckpoint {
        covered_through: u64,
        checkpoint_position: u64,
    },
    #[error(
        "the held summary does not add up: {} records, {} obligations, lowest delivery {:?}",
        .held.records,
        .held.obligations,
        .held.lowest_delivery
    )]
    HeldInconsistent { held: HeldRecords },
}

/// One stream's positions a cut is planned from: where its checkpoint
/// stands, and the row below which every row is acknowledged or held by
/// that checkpoint.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(super) struct StreamBounds {
    pub(super) checkpoint_position: u64,
    pub(super) covered_through: u64,
}

impl StreamBounds {
    /// A stream whose checkpoint holds every record not yet acknowledged:
    /// covered through the checkpoint itself. A recipient that never
    /// acknowledges holds its own records in the checkpoint, never the cut
    /// (DUR-018 R2 as amended 12 October 2026).
    pub(super) const fn behind_checkpoint(checkpoint_position: u64) -> Self {
        Self {
            checkpoint_position,
            covered_through: checkpoint_position,
        }
    }
}

/// The cut decided at one checkpoint: staged as receipts in the checkpoint's
/// batch, applied to the store only after that batch committed.
#[derive(Clone, Debug, PartialEq, Eq)]
pub(super) struct PrunePlan {
    pub(super) receipts: Vec<PruneReceipt>,
}

/// Decides the cut for one checkpoint. `None` when nothing lies below the
/// kept window. `held` is what the checkpoint carries undelivered. `keep`
/// is [`effective_keep`]'s answer: a server never passes `None` (DUR-024
/// R4), and the option remains for a caller that stages no cut at all.
#[must_use]
pub(super) fn plan(
    keep: Option<u64>,
    conversation_id: u64,
    operations: StreamBounds,
    outbox: StreamBounds,
    recipients: &[RecipientAcknowledgement],
    held: HeldRecords,
    at_ms: u64,
) -> Option<PrunePlan> {
    let keep = keep?;
    let mut receipts = Vec::new();
    for (stream, bounds) in [
        (PrunedStream::Operations, operations),
        (PrunedStream::Outbox, outbox),
    ] {
        let checkpoint_position = bounds.checkpoint_position;
        let covered_through = bounds.covered_through.min(checkpoint_position);
        let cut_before = covered_through.saturating_sub(keep);
        if cut_before == 0 {
            continue;
        }
        receipts.push(PruneReceipt {
            conversation_id,
            stream,
            cut_before,
            covered_through,
            checkpoint_position,
            kept: keep,
            recipients: recipients.to_owned(),
            held,
            at_ms,
        });
    }
    (!receipts.is_empty()).then_some(PrunePlan { receipts })
}

/// Stages every receipt of `plan` into the checkpoint's batch.
///
/// # Errors
/// A receipt that does not prove its own cut, by name
/// ([`OperationLogError::PruneReceipt`]); a receipt that did not encode; or
/// the batch refusing the key.
pub(super) fn stage(batch: &mut OperationBatch, plan: &PrunePlan) -> Result<(), OperationLogError> {
    for receipt in &plan.receipts {
        receipt
            .prove()
            .map_err(|source| OperationLogError::PruneReceipt {
                key: String::from_utf8_lossy(&receipt.key()).into_owned(),
                source,
            })?;
        let value = serde_json::to_vec(receipt)?;
        batch.stage(RoutedBatchItem {
            key: receipt.key(),
            expected: None,
            value,
            ttl: None,
        })?;
    }
    Ok(())
}

/// Where a conversation's rows live; a cut applies to both.
///
/// The legacy stream is what the migration and the older readers use; the
/// routed rows are what the wired-boot head writes under the conversation's
/// checkpoint route.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum RowSource {
    /// `DurableStore::truncate_before` on the stream key.
    Legacy,
    /// `DurableStore::remove_routed_range` over the row-key window on the
    /// conversation's route; the head record lies outside the window.
    Routed,
}

/// What one cut came to.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct PruneOutcome {
    /// The stream.
    pub stream: PrunedStream,
    /// Which rows of it.
    pub source: RowSource,
    /// Rows below this were asked to be cut.
    pub cut_before: u64,
    /// Rows the store reports removed.
    pub removed: u64,
    /// Named when the store refused to cut; the receipt stands, the cut is
    /// not claimed.
    pub refused: Option<String>,
}

/// The stream key a receipt's rows live under, legacy and routed alike.
fn stream_key(receipt: &PruneReceipt) -> String {
    match receipt.stream {
        PrunedStream::Operations => format!("{STREAM_PREFIX}{}", receipt.conversation_id),
        PrunedStream::Outbox => format!("{OUTBOX_STREAM_PREFIX}{}", receipt.conversation_id),
    }
}

/// Removes the routed rows of `receipt`'s stream below its cut.
///
/// The window runs from the stream's first row key to the key of
/// `cut_before`; the head record and every later row lie outside it.
async fn cut_routed(store: &dyn DurableStore, receipt: &PruneReceipt) -> Result<u64, String> {
    let route = checkpoint_conversation_route(receipt.conversation_id);
    let stream = stream_key(receipt);
    let from = routed_stream_codec::row_key(&stream, 0).map_err(|error| error.to_string())?;
    let to = routed_stream_codec::row_key(&stream, receipt.cut_before)
        .map_err(|error| error.to_string())?;
    store
        .remove_routed_range(&route, &from, &to)
        .await
        .map_err(|error| error.to_string())
}

/// Applies a committed plan to the store, each source answered on its own.
///
/// The legacy stream goes through `truncate_before`, the routed rows through
/// `remove_routed_range`. A refusal is named and the cut is not claimed.
/// Runs the cut `plan` promised and says what each source came to: a cut
/// by name, a refusal by name, never a claim (DUR-018). The receipts are
/// durable with the checkpoint before this runs.
pub(super) async fn report_cut(store: &dyn DurableStore, conversation_id: u64, plan: &PrunePlan) {
    for outcome in cut(store, plan).await {
        match outcome.refused {
            None => tracing::info!(
                conversation_id,
                stream = ?outcome.stream,
                source = ?outcome.source,
                cut_before = outcome.cut_before,
                removed = outcome.removed,
                "pruned acknowledged rows behind the checkpoint"
            ),
            Some(refusal) => tracing::warn!(
                conversation_id,
                stream = ?outcome.stream,
                source = ?outcome.source,
                cut_before = outcome.cut_before,
                refusal = %refusal,
                "prune receipt written, cut refused by the store"
            ),
        }
    }
}

pub(super) async fn cut(store: &dyn DurableStore, plan: &PrunePlan) -> Vec<PruneOutcome> {
    let mut outcomes = Vec::with_capacity(plan.receipts.len().saturating_mul(2));
    for receipt in &plan.receipts {
        let legacy = store
            .truncate_before(&stream_key(receipt), receipt.cut_before)
            .await
            .map_err(|error| error.to_string());
        let routed = cut_routed(store, receipt).await;
        for (source, answer) in [(RowSource::Legacy, legacy), (RowSource::Routed, routed)] {
            let (removed, refused) = match answer {
                Ok(removed) => (removed, None),
                Err(refusal) => (0, Some(refusal)),
            };
            outcomes.push(PruneOutcome {
                stream: receipt.stream,
                source,
                cut_before: receipt.cut_before,
                removed,
                refused,
            });
        }
    }
    outcomes
}

/// Keeps only the receipts of `plan` not already in the ledger.
///
/// A cut repeated to the same point after a crash between receipt and cut
/// then writes no second receipt (DUR-018 R3). `None` when every receipt is
/// already written.
///
/// # Errors
/// The store could not be read.
pub(super) async fn unwritten(
    store: &dyn DurableStore,
    conversation_id: u64,
    plan: PrunePlan,
) -> Result<Option<PrunePlan>, DurabilityError> {
    let route = checkpoint_conversation_route(conversation_id);
    let mut receipts = Vec::with_capacity(plan.receipts.len());
    for receipt in plan.receipts {
        if store.get_routed(&route, &receipt.key()).await?.is_none() {
            receipts.push(receipt);
        }
    }
    Ok((!receipts.is_empty()).then_some(PrunePlan { receipts }))
}

/// A prune ledger that cannot be read whole.
#[derive(Debug, thiserror::Error)]
pub enum PruneLedgerError {
    /// The store refused the read.
    #[error("prune ledger read refused: {0}")]
    Store(#[from] DurabilityError),
    /// A ledger key holds bytes that are not a receipt.
    #[error("prune receipt {key} is unreadable: {detail}")]
    Unreadable {
        /// The ledger key.
        key: String,
        /// Why.
        detail: String,
    },
    /// A receipt that does not prove its own cut: a named defect, never
    /// silently accepted.
    #[error("prune receipt {key} does not prove its cut: {error}")]
    Defect {
        /// The ledger key.
        key: String,
        /// The proof that failed.
        #[source]
        error: PruneProofError,
    },
}

/// Reads every receipt of `conversation`'s prune ledger, proved.
///
/// They come in key order; a reader learns from them exactly which rows
/// were cut and that every one was acknowledged by every recipient owed it.
///
/// # Errors
/// The store refused the read, a key holds bytes that are not a receipt, or
/// a receipt fails its own proof.
pub async fn prune_receipts(
    store: &dyn DurableStore,
    conversation: u64,
) -> Result<Vec<PruneReceipt>, PruneLedgerError> {
    let from = super::route_keys::namespaced(conversation, PRUNE_RECEIPT_PREFIX.as_bytes());
    let to = super::route_keys::namespaced(conversation, PRUNE_RECEIPT_PREFIX_END);
    let route = checkpoint_conversation_route(conversation);
    let rows = store.scan_routed(&route, &from, &to).await?;
    let mut receipts = Vec::with_capacity(rows.len());
    for (key, value) in rows {
        let key = String::from_utf8_lossy(&key).into_owned();
        let receipt: PruneReceipt =
            serde_json::from_slice(&value).map_err(|error| PruneLedgerError::Unreadable {
                key: key.clone(),
                detail: error.to_string(),
            })?;
        receipt
            .prove()
            .map_err(|error| PruneLedgerError::Defect { key, error })?;
        receipts.push(receipt);
    }
    Ok(receipts)
}

#[cfg(test)]
mod tests {
    use super::{
        HeldRecords, PruneProofError, PruneReceipt, PrunedStream, RecipientAcknowledgement,
        StreamBounds, plan,
    };

    const fn bounds(checkpoint_position: u64, covered_through: u64) -> StreamBounds {
        StreamBounds {
            checkpoint_position,
            covered_through,
        }
    }

    fn recipients(acks: &[(u64, u64)]) -> Vec<RecipientAcknowledgement> {
        acks.iter()
            .map(
                |&(participant_id, acknowledged_through)| RecipientAcknowledgement {
                    participant_id,
                    acknowledged_through,
                },
            )
            .collect()
    }

    const NOTHING_HELD: HeldRecords = HeldRecords {
        records: 0,
        obligations: 0,
        lowest_delivery: None,
    };

    #[test]
    fn no_configured_bound_means_no_cut() {
        assert_eq!(
            plan(
                None,
                7,
                bounds(2_048, 2_048),
                bounds(2_048, 2_048),
                &recipients(&[(1, 2_048)]),
                NOTHING_HELD,
                0
            ),
            None
        );
    }

    #[test]
    fn the_cut_sits_the_kept_rows_behind_the_cover() -> Result<(), Box<dyn std::error::Error>> {
        let planned = plan(
            Some(100),
            7,
            bounds(2_000, 900),
            bounds(2_000, 900),
            &recipients(&[(1, 900), (2, 1_200), (3, 1_500)]),
            NOTHING_HELD,
            5,
        )
        .ok_or("rows lie below the kept window")?;
        assert_eq!(planned.receipts.len(), 2);
        for receipt in &planned.receipts {
            assert_eq!(receipt.cut_before, 800);
            assert_eq!(receipt.covered_through, 900);
            assert_eq!(receipt.checkpoint_position, 2_000);
            receipt.prove()?;
        }
        Ok(())
    }

    /// The 11 October finding: one recipient that never acknowledged held
    /// every cut at 0. Its records are held by the checkpoint, so the cut
    /// sits the kept rows behind the checkpoint and the receipt names what
    /// is held.
    #[test]
    fn a_recipient_that_never_acknowledged_no_longer_holds_the_cut_at_zero()
    -> Result<(), Box<dyn std::error::Error>> {
        let held = HeldRecords {
            records: 3,
            obligations: 3,
            lowest_delivery: Some(1),
        };
        let planned = plan(
            Some(256),
            7,
            StreamBounds::behind_checkpoint(2_048),
            StreamBounds::behind_checkpoint(1_024),
            &recipients(&[(1, 0), (2, 2_048)]),
            held,
            0,
        )
        .ok_or("rows lie below the kept window")?;
        assert_eq!(planned.receipts.len(), 2);
        assert_eq!(planned.receipts[0].cut_before, 1_792);
        assert_eq!(planned.receipts[1].cut_before, 768);
        for receipt in &planned.receipts {
            assert_eq!(receipt.held, held);
            receipt.prove()?;
        }
        Ok(())
    }

    #[test]
    fn the_checkpoint_bounds_the_cover() -> Result<(), Box<dyn std::error::Error>> {
        let planned = plan(
            Some(256),
            7,
            bounds(2_048, 9_000),
            bounds(2_048, 9_000),
            &recipients(&[(1, 9_000)]),
            NOTHING_HELD,
            0,
        )
        .ok_or("rows lie below the kept window")?;
        assert_eq!(planned.receipts[0].covered_through, 2_048);
        assert_eq!(planned.receipts[0].cut_before, 1_792);
        Ok(())
    }

    #[test]
    fn a_receipt_that_does_not_add_up_is_a_named_defect() {
        let receipt = PruneReceipt {
            conversation_id: 7,
            stream: PrunedStream::Operations,
            cut_before: 901,
            covered_through: 900,
            checkpoint_position: 2_000,
            kept: 0,
            recipients: recipients(&[(1, 900)]),
            held: NOTHING_HELD,
            at_ms: 0,
        };
        assert_eq!(
            receipt.prove(),
            Err(PruneProofError::CutAboveCover {
                cut_before: 901,
                covered_through: 900
            })
        );
        let receipt = PruneReceipt {
            cut_before: 800,
            covered_through: 2_001,
            ..receipt
        };
        assert_eq!(
            receipt.prove(),
            Err(PruneProofError::CoverAboveCheckpoint {
                covered_through: 2_001,
                checkpoint_position: 2_000
            })
        );
        let held = HeldRecords {
            records: 2,
            obligations: 1,
            lowest_delivery: Some(4),
        };
        let receipt = PruneReceipt {
            covered_through: 900,
            held,
            ..receipt
        };
        assert_eq!(
            receipt.prove(),
            Err(PruneProofError::HeldInconsistent { held })
        );
    }

    /// A receipt written before the amendment reads back: its
    /// `acknowledged_through` is the cover, and it held nothing it named.
    #[test]
    fn a_receipt_written_before_the_amendment_reads_back() -> Result<(), Box<dyn std::error::Error>>
    {
        let receipt: PruneReceipt = serde_json::from_value(serde_json::json!({
            "conversation_id": 7, "stream": "operations", "cut_before": 800,
            "acknowledged_through": 900, "checkpoint_position": 2_000, "kept": 100,
            "recipients": [{"participant_id": 1, "acknowledged_through": 900}], "at_ms": 5
        }))?;
        assert_eq!(receipt.covered_through, 900);
        assert_eq!(receipt.held, NOTHING_HELD);
        receipt.prove()?;
        Ok(())
    }
}