liminal/durability/dedup/
sweep.rs

1use std::time::Duration;
2
3use crate::durability::settle::append_settled;
4use crate::durability::{DurabilityError, ScanCursor, StoredEntry};
5
6use super::codec::DedupRecord;
7use super::{DedupCache, DedupEntry};
8
9const CLOCK_SKEW_GRACE_MILLIS: u64 = 1_000;
10
11/// Entries the sweep requests per store page.
12///
13/// The sweep walks the whole dedup prefix, so this bounds what one store call
14/// materialises, not what the sweep ultimately considers: each page is folded
15/// into the keys it finishes and dropped.
16const SWEEP_SCAN_PAGE_LIMIT: usize = 1024;
17
18/// The characters a key stream's hash suffix is written in ([`super::key_hash`]).
19///
20/// The walk reads one prefix per character, `namespace:` followed by it, so it
21/// reads this namespace's key streams and nothing nested under the namespace
22/// with another name, such as a server incarnation's ephemeral namespace.
23const KEY_HASH_DIGITS: &[u8; 16] = b"0123456789abcdef";
24
25/// Summary produced by a dedup TTL sweep.
26#[derive(Clone, Copy, Debug, PartialEq, Eq)]
27pub struct DedupSweepReport {
28    scanned: usize,
29    expired: usize,
30    retained: usize,
31}
32
33impl DedupSweepReport {
34    /// Number of scanned stored dedup events.
35    #[must_use]
36    pub const fn scanned_entries(self) -> usize {
37        self.scanned
38    }
39
40    /// Number of logical cache entries tombstoned by the sweep.
41    #[must_use]
42    pub const fn expired_entries(self) -> usize {
43        self.expired
44    }
45
46    /// Number of latest active entries retained after the sweep.
47    #[must_use]
48    pub const fn retained_entries(self) -> usize {
49        self.retained
50    }
51}
52
53/// Configurable TTL sweeper for haematite-backed dedup entries.
54#[derive(Clone, Debug)]
55pub struct DedupSweeper {
56    cache: DedupCache,
57    ttl: Duration,
58    sweep_interval: Duration,
59}
60
61impl DedupSweeper {
62    /// Creates a sweeper with caller-configured TTL and interval.
63    #[must_use]
64    pub const fn new(cache: DedupCache, ttl: Duration, sweep_interval: Duration) -> Self {
65        Self {
66            cache,
67            ttl,
68            sweep_interval,
69        }
70    }
71
72    /// Returns the configured dedup TTL.
73    #[must_use]
74    pub const fn ttl(&self) -> Duration {
75        self.ttl
76    }
77
78    /// Returns the configured sweep interval.
79    #[must_use]
80    pub const fn sweep_interval(&self) -> Duration {
81        self.sweep_interval
82    }
83
84    /// Runs one scan-based sweep at the supplied epoch millisecond timestamp.
85    ///
86    /// # Errors
87    ///
88    /// Propagates store scan/append errors and serialized-entry decode errors.
89    pub async fn sweep_once(&self, now_millis: u64) -> Result<DedupSweepReport, DurabilityError> {
90        let ttl_millis = duration_millis_saturating(self.ttl);
91        let mut expired_entries = 0;
92        let mut retained_entries = 0;
93        let mut walk = self.cache.latest_records();
94        while let Some(batch) = walk.next_batch().await? {
95            for candidate in batch {
96                let DedupRecord::Active(entry) = candidate.record else {
97                    continue;
98                };
99                // A claim with evidence is settled by looking its row up, never
100                // by its age. A receipt, or a claim with no evidence of where
101                // its row would be, ends with its TTL.
102                if entry.evidence().is_none()
103                    && is_expired(entry.timestamp_millis(), now_millis, ttl_millis)
104                {
105                    self.tombstone(&entry, candidate.sequence).await?;
106                    expired_entries += 1;
107                } else {
108                    retained_entries += 1;
109                }
110            }
111        }
112
113        Ok(DedupSweepReport {
114            scanned: walk.scanned(),
115            expired: expired_entries,
116            retained: retained_entries,
117        })
118    }
119
120    async fn tombstone(&self, entry: &DedupEntry, sequence: u64) -> Result<(), DurabilityError> {
121        let stream_key = self.cache.stream_key_for(entry.idempotency_key());
122        let expected_seq = sequence.checked_add(1).ok_or_else(|| {
123            DurabilityError::ConfigError("dedup sweep sequence overflow".to_owned())
124        })?;
125        let tombstone =
126            DedupRecord::tombstone(entry.idempotency_key().to_owned(), entry.timestamp_millis());
127        append_settled(
128            self.cache.store.as_ref(),
129            &stream_key,
130            tombstone.serialize()?,
131            expected_seq,
132        )
133        .await?;
134        if entry.receipt().is_none() {
135            self.cache
136                .leave_in_flight(entry.idempotency_key(), sequence)
137                .await;
138        }
139        Ok(())
140    }
141}
142
143impl DedupCache {
144    /// A walk of the namespace's key streams that hands back each key's latest
145    /// record once the walk has passed its stream.
146    pub(super) const fn latest_records(&self) -> LatestRecords<'_> {
147        LatestRecords {
148            cache: self,
149            digit: 0,
150            cursor: None,
151            pending: None,
152            scanned: 0,
153        }
154    }
155}
156
157/// One key's latest record and the sequence it was read at.
158pub(super) struct SweepCandidate {
159    pub(super) record: DedupRecord,
160    pub(super) sequence: u64,
161}
162
163/// A paged walk of a namespace's key streams, a page per call.
164///
165/// Streams come in key order and every record of one idempotency key lives in
166/// the one stream [`DedupCache::stream_key_for`] derives from it, so a key's
167/// records arrive together: the walk holds only the key it is inside and hands
168/// that key back when the next key begins. Its memory is one page.
169pub(super) struct LatestRecords<'cache> {
170    cache: &'cache DedupCache,
171    /// Index into [`KEY_HASH_DIGITS`] of the prefix being walked.
172    digit: usize,
173    cursor: Option<ScanCursor>,
174    pending: Option<SweepCandidate>,
175    scanned: usize,
176}
177
178impl LatestRecords<'_> {
179    /// Stored entries the walk has read.
180    pub(super) const fn scanned(&self) -> usize {
181        self.scanned
182    }
183
184    /// Reads the next page and answers the keys it finished, or `None` once
185    /// the walk is done.
186    ///
187    /// # Errors
188    ///
189    /// Propagates store scan errors and serialized-entry decode errors.
190    pub(super) async fn next_batch(
191        &mut self,
192    ) -> Result<Option<Vec<SweepCandidate>>, DurabilityError> {
193        let Some(digit) = KEY_HASH_DIGITS.get(self.digit) else {
194            return Ok(None);
195        };
196        let prefix = format!("{}{}", self.cache.scan_prefix(), char::from(*digit));
197        let page = self
198            .cache
199            .store
200            .scan_page(&prefix, self.cursor.take(), SWEEP_SCAN_PAGE_LIMIT)
201            .await?;
202        self.scanned = self
203            .scanned
204            .checked_add(page.entries.len())
205            .ok_or_else(|| {
206                DurabilityError::ConfigError("dedup sweep scan counter overflow".to_owned())
207            })?;
208        let mut finished = fold_finished_keys(page.entries, &mut self.pending)?;
209        if let Some(next) = page.next {
210            self.cursor = Some(next);
211        } else {
212            // The prefix is done, so the key it ended inside is too.
213            finished.extend(self.pending.take());
214            self.digit += 1;
215        }
216        Ok(Some(finished))
217    }
218}
219
220/// Folds one scanned page into the key being walked, answering every key the
221/// page moved past with its latest record.
222fn fold_finished_keys(
223    entries: Vec<StoredEntry>,
224    pending: &mut Option<SweepCandidate>,
225) -> Result<Vec<SweepCandidate>, DurabilityError> {
226    let mut finished = Vec::new();
227    for stored in entries {
228        let record = DedupRecord::deserialize(&stored.payload)?;
229        let candidate = SweepCandidate {
230            record,
231            sequence: stored.sequence,
232        };
233        match pending.take() {
234            Some(previous)
235                if previous.record.idempotency_key() == candidate.record.idempotency_key() =>
236            {
237                *pending = Some(candidate);
238            }
239            Some(previous) => {
240                finished.push(previous);
241                *pending = Some(candidate);
242            }
243            None => *pending = Some(candidate),
244        }
245    }
246    Ok(finished)
247}
248
249const fn is_expired(timestamp_millis: u64, now_millis: u64, ttl_millis: u64) -> bool {
250    let expiry = timestamp_millis
251        .saturating_add(ttl_millis)
252        .saturating_add(CLOCK_SKEW_GRACE_MILLIS);
253    now_millis > expiry
254}
255
256fn duration_millis_saturating(duration: Duration) -> u64 {
257    u64::try_from(duration.as_millis()).unwrap_or(u64::MAX)
258}
liminal/durability/dedup/sweep.rs

liminal/durability/dedup/
sweep.rs
