liminal-rs 0.16.3

//! The recorded head of a long dedup key stream (DUR-008 A12).
//!
//! A lookup reads a key stream's first page; only a stream that fills it is
//! read past it, from the head recorded under the stream's `head_checkpoint`
//! key. The head is used only once it verifies: it lies inside the stream, the
//! row before it sits at its own position, and that row is a record of the key
//! looked up. A head that fails any of these is ignored, counted by name, and
//! the stream is walked from its first page on.
//!
//! No append records a head. A lookup records one after a walk, or when the
//! rows past a verified head reach [`HEAD_CHECKPOINT_EVERY`], and only ever
//! forward: a compare-and-swap from the head it read, retried while another
//! writer's head is still behind its own. Finding another writer's head at or
//! past its own is a lost race, which is not an error and is counted. A head
//! record that fails, as on a store that refuses writes, is logged and the
//! lookup still answers what it read.
//!
//! A refused head is provably wrong, so it is repaired: the walk's verified
//! result replaces it by one compare-and-swap whose expected value is the
//! exact refused head. A writer that moved the head in between makes that swap
//! fail; its head stands, the loss is counted as a lost race, and the repair
//! is not retried, so a repair never moves back over a newer head. Each repair
//! that lands is counted.
//!
//! The three counters are read in a running process through
//! [`dedup_head_counts`].

use std::fmt;
use std::sync::atomic::{AtomicU64, Ordering};

use super::codec::DedupRecord;
use super::{DedupCache, fold_page, len_to_u64};
use crate::durability::head_checkpoint::{HEAD_CHECKPOINT_EVERY, READ_BATCH_SIZE};
use crate::durability::{DurabilityError, StoredEntry, head_checkpoint_key};

/// How many times one head record retries after another writer moved the
/// head, still behind this one, in between.
const HEAD_RECORD_ATTEMPTS: usize = 8;

/// Head records that found another writer's head already at or past their
/// own since the process started.
static HEAD_RECORD_LOST_RACES: AtomicU64 = AtomicU64::new(0);

/// Recorded heads a lookup refused (beyond the stream, at the wrong sequence,
/// or naming another key's record) since the process started.
static HEAD_REFUSALS: AtomicU64 = AtomicU64::new(0);

/// Refused heads replaced by the walk's verified result since the process
/// started.
static HEAD_REPAIRS: AtomicU64 = AtomicU64::new(0);

/// The dedup head counters of this process, read at one moment.
///
/// Counters only grow; each is read once, without a lock.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct DedupHeadCounts {
    /// Recorded heads a lookup refused: beyond the stream, at the wrong
    /// sequence, or naming another key's record.
    pub refusals: u64,
    /// Head records and repairs that found another writer's head already in
    /// place of the one they expected.
    pub record_lost_races: u64,
    /// Refused heads replaced by the walk's verified result.
    pub repairs: u64,
}

impl DedupHeadCounts {
    /// Every counter with its name, in one fixed order.
    #[must_use]
    pub const fn named(&self) -> [(&'static str, u64); 3] {
        [
            ("liminal_dedup_head_refusals_total", self.refusals),
            (
                "liminal_dedup_head_record_lost_races_total",
                self.record_lost_races,
            ),
            ("liminal_dedup_head_repairs_total", self.repairs),
        ]
    }
}

/// The dedup head counters of this process as they stand now: refusals
/// (`HEAD_REFUSALS`), lost races (`HEAD_RECORD_LOST_RACES`) and repairs
/// (`HEAD_REPAIRS`).
#[must_use]
pub fn dedup_head_counts() -> DedupHeadCounts {
    DedupHeadCounts {
        refusals: HEAD_REFUSALS.load(Ordering::Relaxed),
        record_lost_races: HEAD_RECORD_LOST_RACES.load(Ordering::Relaxed),
        repairs: HEAD_REPAIRS.load(Ordering::Relaxed),
    }
}

/// Why a recorded head was not used.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum HeadRefusal {
    /// The head claims rows the stream does not hold.
    BeyondLog { covered: u64 },
    /// The row before the head carries another sequence than its position.
    WrongSequence { covered: u64, found: u64 },
    /// The row before the head is a record of another idempotency key.
    ForeignKey { covered: u64 },
}

impl fmt::Display for HeadRefusal {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::BeyondLog { covered } => write!(
                formatter,
                "the recorded head {covered} claims rows the stream does not hold"
            ),
            Self::WrongSequence { covered, found } => write!(
                formatter,
                "the recorded head {covered} names a row whose sequence is {found}"
            ),
            Self::ForeignKey { covered } => write!(
                formatter,
                "the recorded head {covered} names a record of another key"
            ),
        }
    }
}

impl DedupCache {
    /// The latest record and next sequence of a stream whose first page,
    /// `first_page`, was full.
    ///
    /// # Errors
    /// Propagates store reads, malformed records, and a record of another key
    /// ([`DurabilityError::DedupCollision`]) among the rows read.
    pub(super) async fn latest_past_first_page(
        &self,
        stream_key: &str,
        idempotency_key: &str,
        first_page: &[StoredEntry],
    ) -> Result<(Option<DedupRecord>, u64), DurabilityError> {
        let recorded = self
            .store
            .read_value(&head_checkpoint_key(stream_key))
            .await?
            .unwrap_or(0);
        let mut head_refused = false;
        if recorded > 0 {
            match self
                .verified_head(stream_key, idempotency_key, recorded)
                .await?
            {
                Ok(record) => {
                    let (latest, next) = self
                        .fold_from(stream_key, idempotency_key, recorded)
                        .await?;
                    if next.saturating_sub(recorded) >= HEAD_CHECKPOINT_EVERY {
                        self.record_head(stream_key, recorded, next).await;
                    }
                    return Ok((latest.or(Some(record)), next));
                }
                Err(refusal) => {
                    HEAD_REFUSALS.fetch_add(1, Ordering::Relaxed);
                    tracing::warn!(
                        stream = stream_key,
                        reason = %refusal,
                        "dedup stream head refused; the stream is walked from its first page"
                    );
                    head_refused = true;
                }
            }
        }
        let first = fold_page(stream_key, idempotency_key, first_page, 0)?;
        let (rest, next) = self
            .fold_from(stream_key, idempotency_key, len_to_u64(first_page.len())?)
            .await?;
        if head_refused {
            self.repair_head(stream_key, recorded, next).await;
        } else {
            self.record_head(stream_key, recorded, next).await;
        }
        Ok((rest.or(first), next))
    }

    /// Checks the head `covered` against the stream: the row at `covered - 1`
    /// exists, sits at its position and is a record of `idempotency_key`.
    /// Answers that record, or why the head is refused.
    async fn verified_head(
        &self,
        stream_key: &str,
        idempotency_key: &str,
        covered: u64,
    ) -> Result<Result<DedupRecord, HeadRefusal>, DurabilityError> {
        let last = covered.saturating_sub(1);
        let Some(stored) = self.store.read_at(stream_key, last).await? else {
            return Ok(Err(HeadRefusal::BeyondLog { covered }));
        };
        if stored.sequence != last {
            return Ok(Err(HeadRefusal::WrongSequence {
                covered,
                found: stored.sequence,
            }));
        }
        let record = DedupRecord::deserialize(&stored.payload)?;
        if record.idempotency_key() != idempotency_key {
            return Ok(Err(HeadRefusal::ForeignKey { covered }));
        }
        Ok(Ok(record))
    }

    /// Reads the stream from `offset` to its end, checking every row, and
    /// answers the last record read and the stream's next sequence.
    async fn fold_from(
        &self,
        stream_key: &str,
        idempotency_key: &str,
        offset: u64,
    ) -> Result<(Option<DedupRecord>, u64), DurabilityError> {
        let mut latest = None;
        let mut next = offset;
        loop {
            let page = self
                .store
                .read_from(stream_key, next, READ_BATCH_SIZE)
                .await?;
            if let Some(last) = fold_page(stream_key, idempotency_key, &page, next)? {
                latest = Some(last);
            }
            next = next.checked_add(len_to_u64(page.len())?).ok_or_else(|| {
                DurabilityError::ConfigError(format!("dedup stream {stream_key} offset overflow"))
            })?;
            if page.len() < READ_BATCH_SIZE {
                return Ok((latest, next));
            }
        }
    }

    /// Records `next` as the stream's head, forward only, from `recorded`,
    /// the head this lookup read. A head at or past `next` already stored by
    /// another writer is a lost race, counted and not an error; any other
    /// failure is logged, since the lookup has already read what it answers.
    async fn record_head(&self, stream_key: &str, recorded: u64, next: u64) {
        let key = head_checkpoint_key(stream_key);
        let mut stored = recorded;
        for _ in 0..HEAD_RECORD_ATTEMPTS {
            if stored >= next {
                return;
            }
            match self.store.cas(&key, stored, next).await {
                Ok(()) => return,
                Err(DurabilityError::CursorRegression {
                    stored: observed, ..
                }) if observed >= next => {
                    HEAD_RECORD_LOST_RACES.fetch_add(1, Ordering::Relaxed);
                    return;
                }
                Err(DurabilityError::CursorRegression {
                    stored: observed, ..
                }) if observed != stored => stored = observed,
                Err(error) => {
                    tracing::warn!(
                        stream = stream_key,
                        %error,
                        "dedup stream head was not recorded; the next lookup walks again"
                    );
                    return;
                }
            }
        }
    }

    /// Replaces the refused head `refused` with `next`, the walk's verified
    /// result, by one compare-and-swap expecting exactly `refused`. Another
    /// writer's head found in its place stands: the loss is counted and the
    /// repair not retried. Any other failure is logged, since the lookup has
    /// already read what it answers.
    async fn repair_head(&self, stream_key: &str, refused: u64, next: u64) {
        if refused == next {
            return;
        }
        if next == 0 {
            // The value store keeps no physical zero, so an empty stream's
            // head cannot be recorded; the refused head stays and is refused
            // again by name.
            tracing::warn!(
                stream = stream_key,
                refused,
                "a refused dedup stream head cannot be repaired while its stream is empty"
            );
            return;
        }
        match self
            .store
            .cas(&head_checkpoint_key(stream_key), refused, next)
            .await
        {
            Ok(()) => {
                HEAD_REPAIRS.fetch_add(1, Ordering::Relaxed);
            }
            Err(DurabilityError::CursorRegression { .. }) => {
                HEAD_RECORD_LOST_RACES.fetch_add(1, Ordering::Relaxed);
            }
            Err(error) => {
                tracing::warn!(
                    stream = stream_key,
                    %error,
                    "refused dedup stream head was not repaired; the next lookup walks again"
                );
            }
        }
    }
}