liminal-server 0.20.1

//! 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:";

/// 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 sequence is acknowledged by it.
    pub acknowledged_through: 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 recipient owed a row has acknowledged through here.
    pub acknowledged_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>,
    /// Milliseconds since the Unix epoch on the server clock.
    pub at_ms: u64,
}

impl PruneReceipt {
    /// Proves the receipt's own arithmetic: nothing unacknowledged was cut.
    ///
    /// # Errors
    /// The receipt names a cut above an acknowledgement or a checkpoint.
    pub fn prove(&self) -> Result<(), PruneProofError> {
        if self.cut_before > self.acknowledged_through {
            return Err(PruneProofError::CutAboveAcknowledgement {
                cut_before: self.cut_before,
                acknowledged_through: self.acknowledged_through,
            });
        }
        if self.acknowledged_through > self.checkpoint_position {
            return Err(PruneProofError::AcknowledgementAboveCheckpoint {
                acknowledged_through: self.acknowledged_through,
                checkpoint_position: self.checkpoint_position,
            });
        }
        if let Some(recipient) = self
            .recipients
            .iter()
            .find(|recipient| recipient.acknowledged_through < self.cut_before)
        {
            return Err(PruneProofError::RecipientBelowCut {
                participant_id: recipient.participant_id,
                acknowledged_through: recipient.acknowledged_through,
                cut_before: self.cut_before,
            });
        }
        Ok(())
    }

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

/// The ledger key of a receipt for `stream` at `cut_before`: one key per
/// cut point, so a repeated cut to the same point writes no second receipt.
#[must_use]
pub(super) fn 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 acknowledged_through {acknowledged_through}")]
    CutAboveAcknowledgement {
        cut_before: u64,
        acknowledged_through: u64,
    },
    #[error(
        "acknowledged_through {acknowledged_through} is above checkpoint_position {checkpoint_position}"
    )]
    AcknowledgementAboveCheckpoint {
        acknowledged_through: u64,
        checkpoint_position: u64,
    },
    #[error(
        "recipient {participant_id} acknowledged through {acknowledged_through}, below cut_before {cut_before}"
    )]
    RecipientBelowCut {
        participant_id: u64,
        acknowledged_through: u64,
        cut_before: u64,
    },
}

/// One stream's positions a cut is planned from: where its checkpoint
/// stands, and the row every recipient has acknowledged through.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(super) struct StreamBounds {
    pub(super) checkpoint_position: u64,
    pub(super) acknowledged_through: u64,
}

/// 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 pruning is off, when no
/// recipient has acknowledged anything, or when nothing lies below the kept
/// window.
#[must_use]
pub(super) fn plan(
    keep: Option<u64>,
    conversation_id: u64,
    operations: StreamBounds,
    outbox: StreamBounds,
    recipients: &[RecipientAcknowledgement],
    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 acknowledged_through = bounds.acknowledged_through.min(checkpoint_position);
        let cut_before = acknowledged_through.saturating_sub(keep);
        if cut_before == 0 {
            continue;
        }
        receipts.push(PruneReceipt {
            conversation_id,
            stream,
            cut_before,
            acknowledged_through,
            checkpoint_position,
            kept: keep,
            recipients: recipients.to_owned(),
            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.
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_id`'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_id: u64,
) -> Result<Vec<PruneReceipt>, PruneLedgerError> {
    let route = checkpoint_conversation_route(conversation_id);
    let rows = store
        .scan_routed(
            &route,
            PRUNE_RECEIPT_PREFIX.as_bytes(),
            PRUNE_RECEIPT_PREFIX_END,
        )
        .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::{
        PruneProofError, PruneReceipt, PrunedStream, RecipientAcknowledgement, StreamBounds, plan,
    };

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

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

    #[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)]),
                0
            ),
            None
        );
    }

    #[test]
    fn the_cut_sits_the_kept_rows_behind_the_least_acknowledgement()
    -> 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)]),
            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.acknowledged_through, 900);
            assert_eq!(receipt.checkpoint_position, 2_000);
            receipt.prove()?;
        }
        Ok(())
    }

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

    #[test]
    fn the_checkpoint_bounds_the_acknowledgement() -> 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)]),
            0,
        )
        .ok_or("rows lie below the kept window")?;
        assert_eq!(planned.receipts[0].acknowledged_through, 2_048);
        assert_eq!(planned.receipts[0].cut_before, 1_792);
        Ok(())
    }

    #[test]
    fn a_receipt_that_cuts_above_an_acknowledgement_is_a_named_defect() {
        let receipt = PruneReceipt {
            conversation_id: 7,
            stream: PrunedStream::Operations,
            cut_before: 901,
            acknowledged_through: 900,
            checkpoint_position: 2_000,
            kept: 0,
            recipients: recipients(&[(1, 900)]),
            at_ms: 0,
        };
        assert_eq!(
            receipt.prove(),
            Err(PruneProofError::CutAboveAcknowledgement {
                cut_before: 901,
                acknowledged_through: 900
            })
        );
        let receipt = PruneReceipt {
            cut_before: 800,
            recipients: recipients(&[(1, 900), (2, 700)]),
            ..receipt
        };
        assert_eq!(
            receipt.prove(),
            Err(PruneProofError::RecipientBelowCut {
                participant_id: 2,
                acknowledged_through: 700,
                cut_before: 800
            })
        );
    }
}