liminal-rs 0.16.2

use std::time::Duration;

use crate::durability::settle::append_settled;
use crate::durability::{DurabilityError, ScanCursor, StoredEntry};

use super::codec::DedupRecord;
use super::{DedupCache, DedupEntry};

const CLOCK_SKEW_GRACE_MILLIS: u64 = 1_000;

/// Entries the sweep requests per store page.
///
/// The sweep walks the whole dedup prefix, so this bounds what one store call
/// materialises, not what the sweep ultimately considers: each page is folded
/// into the keys it finishes and dropped.
const SWEEP_SCAN_PAGE_LIMIT: usize = 1024;

/// The characters a key stream's hash suffix is written in ([`super::key_hash`]).
///
/// The walk reads one prefix per character, `namespace:` followed by it, so it
/// reads this namespace's key streams and nothing nested under the namespace
/// with another name, such as a server incarnation's ephemeral namespace.
const KEY_HASH_DIGITS: &[u8; 16] = b"0123456789abcdef";

/// Summary produced by a dedup TTL sweep.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct DedupSweepReport {
    scanned: usize,
    expired: usize,
    retained: usize,
}

impl DedupSweepReport {
    /// Number of scanned stored dedup events.
    #[must_use]
    pub const fn scanned_entries(self) -> usize {
        self.scanned
    }

    /// Number of logical cache entries tombstoned by the sweep.
    #[must_use]
    pub const fn expired_entries(self) -> usize {
        self.expired
    }

    /// Number of latest active entries retained after the sweep.
    #[must_use]
    pub const fn retained_entries(self) -> usize {
        self.retained
    }
}

/// Configurable TTL sweeper for haematite-backed dedup entries.
#[derive(Clone, Debug)]
pub struct DedupSweeper {
    cache: DedupCache,
    ttl: Duration,
    sweep_interval: Duration,
}

impl DedupSweeper {
    /// Creates a sweeper with caller-configured TTL and interval.
    #[must_use]
    pub const fn new(cache: DedupCache, ttl: Duration, sweep_interval: Duration) -> Self {
        Self {
            cache,
            ttl,
            sweep_interval,
        }
    }

    /// Returns the configured dedup TTL.
    #[must_use]
    pub const fn ttl(&self) -> Duration {
        self.ttl
    }

    /// Returns the configured sweep interval.
    #[must_use]
    pub const fn sweep_interval(&self) -> Duration {
        self.sweep_interval
    }

    /// Runs one scan-based sweep at the supplied epoch millisecond timestamp.
    ///
    /// # Errors
    ///
    /// Propagates store scan/append errors and serialized-entry decode errors.
    pub async fn sweep_once(&self, now_millis: u64) -> Result<DedupSweepReport, DurabilityError> {
        let ttl_millis = duration_millis_saturating(self.ttl);
        let mut expired_entries = 0;
        let mut retained_entries = 0;
        let mut walk = self.cache.latest_records();
        while let Some(batch) = walk.next_batch().await? {
            for candidate in batch {
                let DedupRecord::Active(entry) = candidate.record else {
                    continue;
                };
                // A claim with evidence is settled by looking its row up, never
                // by its age. A receipt, or a claim with no evidence of where
                // its row would be, ends with its TTL.
                if entry.evidence().is_none()
                    && is_expired(entry.timestamp_millis(), now_millis, ttl_millis)
                {
                    self.tombstone(&entry, candidate.sequence).await?;
                    expired_entries += 1;
                } else {
                    retained_entries += 1;
                }
            }
        }

        Ok(DedupSweepReport {
            scanned: walk.scanned(),
            expired: expired_entries,
            retained: retained_entries,
        })
    }

    async fn tombstone(&self, entry: &DedupEntry, sequence: u64) -> Result<(), DurabilityError> {
        let stream_key = self.cache.stream_key_for(entry.idempotency_key());
        let expected_seq = sequence.checked_add(1).ok_or_else(|| {
            DurabilityError::ConfigError("dedup sweep sequence overflow".to_owned())
        })?;
        let tombstone =
            DedupRecord::tombstone(entry.idempotency_key().to_owned(), entry.timestamp_millis());
        append_settled(
            self.cache.store.as_ref(),
            &stream_key,
            tombstone.serialize()?,
            expected_seq,
        )
        .await?;
        if entry.receipt().is_none() {
            self.cache
                .leave_in_flight(entry.idempotency_key(), sequence)
                .await;
        }
        Ok(())
    }
}

impl DedupCache {
    /// A walk of the namespace's key streams that hands back each key's latest
    /// record once the walk has passed its stream.
    pub(super) const fn latest_records(&self) -> LatestRecords<'_> {
        LatestRecords {
            cache: self,
            digit: 0,
            cursor: None,
            pending: None,
            scanned: 0,
        }
    }
}

/// One key's latest record and the sequence it was read at.
pub(super) struct SweepCandidate {
    pub(super) record: DedupRecord,
    pub(super) sequence: u64,
}

/// A paged walk of a namespace's key streams, a page per call.
///
/// Streams come in key order and every record of one idempotency key lives in
/// the one stream [`DedupCache::stream_key_for`] derives from it, so a key's
/// records arrive together: the walk holds only the key it is inside and hands
/// that key back when the next key begins. Its memory is one page.
pub(super) struct LatestRecords<'cache> {
    cache: &'cache DedupCache,
    /// Index into [`KEY_HASH_DIGITS`] of the prefix being walked.
    digit: usize,
    cursor: Option<ScanCursor>,
    pending: Option<SweepCandidate>,
    scanned: usize,
}

impl LatestRecords<'_> {
    /// Stored entries the walk has read.
    pub(super) const fn scanned(&self) -> usize {
        self.scanned
    }

    /// Reads the next page and answers the keys it finished, or `None` once
    /// the walk is done.
    ///
    /// # Errors
    ///
    /// Propagates store scan errors and serialized-entry decode errors.
    pub(super) async fn next_batch(
        &mut self,
    ) -> Result<Option<Vec<SweepCandidate>>, DurabilityError> {
        let Some(digit) = KEY_HASH_DIGITS.get(self.digit) else {
            return Ok(None);
        };
        let prefix = format!("{}{}", self.cache.scan_prefix(), char::from(*digit));
        let page = self
            .cache
            .store
            .scan_page(&prefix, self.cursor.take(), SWEEP_SCAN_PAGE_LIMIT)
            .await?;
        self.scanned = self
            .scanned
            .checked_add(page.entries.len())
            .ok_or_else(|| {
                DurabilityError::ConfigError("dedup sweep scan counter overflow".to_owned())
            })?;
        let mut finished = fold_finished_keys(page.entries, &mut self.pending)?;
        if let Some(next) = page.next {
            self.cursor = Some(next);
        } else {
            // The prefix is done, so the key it ended inside is too.
            finished.extend(self.pending.take());
            self.digit += 1;
        }
        Ok(Some(finished))
    }
}

/// Folds one scanned page into the key being walked, answering every key the
/// page moved past with its latest record.
fn fold_finished_keys(
    entries: Vec<StoredEntry>,
    pending: &mut Option<SweepCandidate>,
) -> Result<Vec<SweepCandidate>, DurabilityError> {
    let mut finished = Vec::new();
    for stored in entries {
        let record = DedupRecord::deserialize(&stored.payload)?;
        let candidate = SweepCandidate {
            record,
            sequence: stored.sequence,
        };
        match pending.take() {
            Some(previous)
                if previous.record.idempotency_key() == candidate.record.idempotency_key() =>
            {
                *pending = Some(candidate);
            }
            Some(previous) => {
                finished.push(previous);
                *pending = Some(candidate);
            }
            None => *pending = Some(candidate),
        }
    }
    Ok(finished)
}

const fn is_expired(timestamp_millis: u64, now_millis: u64, ttl_millis: u64) -> bool {
    let expiry = timestamp_millis
        .saturating_add(ttl_millis)
        .saturating_add(CLOCK_SKEW_GRACE_MILLIS);
    now_millis > expiry
}

fn duration_millis_saturating(duration: Duration) -> u64 {
    u64::try_from(duration.as_millis()).unwrap_or(u64::MAX)
}