liminal/durability/
dedup.rs

1use crate::durability::settle::append_settled;
2use std::fmt;
3use std::sync::Arc;
4use std::time::{SystemTime, UNIX_EPOCH};
5
6use super::head_checkpoint::READ_BATCH_SIZE;
7use super::{DurabilityError, DurableStore, ProcessingReceipt, StoredEntry};
8use codec::DedupRecord;
9pub use head::{DedupHeadCounts, dedup_head_counts};
10use in_flight::InFlightIndex;
11pub use sweep::{DedupSweepReport, DedupSweeper};
12
13#[cfg(test)]
14mod ceiling_in_flight;
15mod codec;
16mod head;
17#[cfg(test)]
18mod head_tests;
19mod in_flight;
20#[cfg(test)]
21mod in_flight_tests;
22#[cfg(test)]
23mod lookup_tests;
24#[cfg(test)]
25mod start_count_tests;
26mod sweep;
27#[cfg(test)]
28mod tests;
29mod unresolved;
30
31/// Hashes a producer idempotency key into a deterministic stream-key suffix.
32#[must_use]
33pub fn key_hash(idempotency_key: &str) -> String {
34    let mut hash = 0xcbf2_9ce4_8422_2325_u64;
35    for byte in idempotency_key.as_bytes() {
36        hash ^= u64::from(*byte);
37        hash = hash.wrapping_mul(0x0000_0100_0000_01b3);
38    }
39    format!("{hash:016x}")
40}
41
42/// Where an in-flight claim's row would be, recorded with the claim.
43///
44/// The claim is written before its publish appends, so the row, if it exists,
45/// sits at or after the position each partition of the channel stood at when
46/// the claim was made: a lookup starts there and never reads the history
47/// before it.
48#[derive(Clone, Debug, PartialEq, Eq)]
49pub struct ClaimEvidence {
50    channel: String,
51    positions: Vec<u64>,
52}
53
54impl ClaimEvidence {
55    /// Evidence naming the durable `channel` and, per partition, the sequence
56    /// the channel stood at when the claim was made.
57    #[must_use]
58    pub fn new(channel: impl Into<String>, positions: Vec<u64>) -> Self {
59        Self {
60            channel: channel.into(),
61            positions,
62        }
63    }
64
65    /// Durable channel whose log can establish the claim's outcome.
66    #[must_use]
67    pub fn channel(&self) -> &str {
68        &self.channel
69    }
70
71    /// Per partition, the first sequence the claim's row can sit at.
72    #[must_use]
73    pub fn positions(&self) -> &[u64] {
74        &self.positions
75    }
76}
77
78/// Persisted dedup cache entry for one producer idempotency key.
79#[derive(Clone, PartialEq, Eq)]
80pub struct DedupEntry {
81    idempotency_key: String,
82    receipt: Option<Vec<u8>>,
83    timestamp_millis: u64,
84    evidence: Option<ClaimEvidence>,
85}
86
87impl DedupEntry {
88    /// Builds an entry containing the original key, optional receipt bytes, and epoch millis.
89    #[must_use]
90    pub fn new(
91        idempotency_key: impl Into<String>,
92        receipt: Option<Vec<u8>>,
93        timestamp_millis: u64,
94    ) -> Self {
95        Self {
96            idempotency_key: idempotency_key.into(),
97            receipt,
98            timestamp_millis,
99            evidence: None,
100        }
101    }
102
103    /// Returns the original idempotency key stored with this entry.
104    #[must_use]
105    pub fn idempotency_key(&self) -> &str {
106        &self.idempotency_key
107    }
108
109    /// Durable channel whose log can establish this in-flight claim's outcome.
110    #[must_use]
111    pub fn evidence_channel(&self) -> Option<&str> {
112        self.evidence.as_ref().map(ClaimEvidence::channel)
113    }
114
115    /// Where this in-flight claim's row would be, when the claim recorded it.
116    #[must_use]
117    pub const fn evidence(&self) -> Option<&ClaimEvidence> {
118        self.evidence.as_ref()
119    }
120
121    /// Returns the stored opaque receipt bytes, when processing has completed.
122    #[must_use]
123    pub fn receipt(&self) -> Option<&[u8]> {
124        self.receipt.as_deref()
125    }
126
127    /// Returns the entry timestamp in epoch milliseconds.
128    #[must_use]
129    pub const fn timestamp_millis(&self) -> u64 {
130        self.timestamp_millis
131    }
132
133    /// Serializes this active cache entry into deterministic storage bytes.
134    ///
135    /// # Errors
136    ///
137    /// Returns [`DurabilityError::EnvelopeError`] when a field length cannot be encoded.
138    pub fn serialize(&self) -> Result<Vec<u8>, DurabilityError> {
139        DedupRecord::Active(self.clone()).serialize()
140    }
141
142    /// Deserializes an active cache entry previously produced by [`Self::serialize`].
143    ///
144    /// # Errors
145    ///
146    /// Returns [`DurabilityError::EnvelopeError`] when bytes are malformed or contain a tombstone.
147    pub fn deserialize(bytes: &[u8]) -> Result<Self, DurabilityError> {
148        match DedupRecord::deserialize(bytes)? {
149            DedupRecord::Active(entry) => Ok(entry),
150            DedupRecord::Tombstone { .. } => Err(DurabilityError::EnvelopeError(
151                "dedup tombstone is not an active entry".to_owned(),
152            )),
153        }
154    }
155}
156
157impl fmt::Debug for DedupEntry {
158    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
159        formatter
160            .debug_struct("DedupEntry")
161            .field("idempotency_key", &self.idempotency_key)
162            .field("receipt_bytes", &self.receipt.as_ref().map(Vec::len))
163            .field("timestamp_millis", &self.timestamp_millis)
164            .field("evidence", &self.evidence)
165            .finish()
166    }
167}
168
169/// Result of checking or claiming an idempotency key.
170#[derive(Clone, Debug, PartialEq, Eq)]
171pub enum DedupDecision {
172    /// The caller claimed the key and may deliver the message for processing.
173    Claimed,
174    /// The key was already completed; return this receipt without re-delivery.
175    Completed(ProcessingReceipt),
176    /// The key is already being processed and delivery must be deferred.
177    InFlight,
178}
179
180/// Haematite-backed idempotency-key cache for the lightweight dedup contract.
181///
182/// Clones share one in-flight index (`dedup/in_flight.rs`). The index assumes
183/// it is the only one kept for its namespace in the process, as the store
184/// assumes one writing process: a second cache built over the same namespace
185/// is a second process's view, as a restart would be.
186#[derive(Clone)]
187pub struct DedupCache {
188    store: Arc<dyn DurableStore>,
189    namespace: String,
190    in_flight: Arc<InFlightIndex>,
191}
192
193impl DedupCache {
194    /// Creates a dedup cache over the given durable store and namespace prefix.
195    #[must_use]
196    pub fn new(store: Arc<dyn DurableStore>, namespace: impl Into<String>) -> Self {
197        Self {
198            store,
199            namespace: namespace.into(),
200            in_flight: Arc::new(InFlightIndex::default()),
201        }
202    }
203
204    /// Creates a dedup cache over a namespace nothing has written to: one in
205    /// a store this process created, or one no earlier process used.
206    ///
207    /// Its in-flight index starts empty without reading the store, so
208    /// [`Self::in_flight_claims`] reads nothing and no claim walks the
209    /// namespace. The first claim records the index start before it writes,
210    /// so a later process opening the namespace with [`Self::new`] reads the
211    /// index and never walks it. A namespace that may hold claims is opened
212    /// with [`Self::new`].
213    #[must_use]
214    pub fn new_empty(store: Arc<dyn DurableStore>, namespace: impl Into<String>) -> Self {
215        Self {
216            store,
217            namespace: namespace.into(),
218            in_flight: Arc::new(InFlightIndex::empty()),
219        }
220    }
221
222    /// Returns the configured dedup namespace.
223    #[must_use]
224    pub fn namespace(&self) -> &str {
225        &self.namespace
226    }
227
228    /// Formats the haematite stream key for an idempotency key.
229    #[must_use]
230    pub fn stream_key_for(&self, idempotency_key: &str) -> String {
231        format!("{}:{}", self.namespace, key_hash(idempotency_key))
232    }
233
234    /// Looks up an existing key without appending a record. A stream of
235    /// fewer than one page of records is read in one page, as before DUR-008
236    /// A12; a longer one is read from its verified recorded head, and one
237    /// whose head was never recorded is walked once and its head recorded.
238    ///
239    /// # Errors
240    ///
241    /// Propagates store read errors and returns [`DurabilityError::DedupCollision`] when
242    /// the hashed stream contains a different original idempotency key.
243    pub async fn lookup(
244        &self,
245        idempotency_key: &str,
246    ) -> Result<Option<DedupDecision>, DurabilityError> {
247        let stream_key = self.stream_key_for(idempotency_key);
248        let snapshot = self.load_snapshot(&stream_key, idempotency_key).await?;
249        Ok(snapshot.current.as_ref().map(decision_for_entry))
250    }
251
252    /// Claims a new key or returns the existing completed/in-flight decision.
253    ///
254    /// Callers must only deliver the message when this returns [`DedupDecision::Claimed`].
255    ///
256    /// # Errors
257    ///
258    /// Propagates store errors and serialization errors. A concurrent first claim that wins
259    /// the append race is converted into the duplicate completed/in-flight decision.
260    pub async fn claim_or_get(
261        &self,
262        idempotency_key: &str,
263        timestamp_millis: u64,
264    ) -> Result<DedupDecision, DurabilityError> {
265        let stream_key = self.stream_key_for(idempotency_key);
266        let snapshot = self.load_snapshot(&stream_key, idempotency_key).await?;
267        if let Some(entry) = snapshot.current.as_ref() {
268            return Ok(decision_for_entry(entry));
269        }
270
271        let entry = DedupEntry::new(idempotency_key, None, timestamp_millis);
272        self.record_empty_start().await?;
273        match append_settled(
274            self.store.as_ref(),
275            &stream_key,
276            entry.serialize()?,
277            snapshot.next_seq,
278        )
279        .await
280        {
281            Ok(_) => Ok(DedupDecision::Claimed),
282            Err(DurabilityError::SequenceConflict { expected, actual }) => {
283                self.decision_after_conflict(&stream_key, idempotency_key, expected, actual)
284                    .await
285            }
286            Err(error) => Err(error),
287        }
288    }
289
290    /// Stores a processing receipt by appending a completed entry for an existing key.
291    ///
292    /// # Errors
293    ///
294    /// Returns [`DurabilityError::ConfigError`] when the current system time cannot be encoded.
295    /// Returns [`DurabilityError::DedupCollision`] when the key is missing or collides with
296    /// another original key, and propagates serialization/store append errors.
297    pub async fn complete_receipt(
298        &self,
299        idempotency_key: &str,
300        receipt: ProcessingReceipt,
301    ) -> Result<(), DurabilityError> {
302        self.complete_receipt_at(idempotency_key, receipt, current_epoch_millis()?)
303            .await
304    }
305
306    /// Stores a processing receipt with an explicit receipt timestamp.
307    ///
308    /// This is useful for deterministic tests and for callers that already have a trusted
309    /// processing-completion timestamp. The timestamp is the TTL anchor for the stored receipt.
310    ///
311    /// # Errors
312    ///
313    /// Returns [`DurabilityError::DedupCollision`] when the key is missing, collides with
314    /// another original key, or has already completed with different receipt bytes. Propagates
315    /// serialization/store append errors.
316    async fn complete_receipt_at(
317        &self,
318        idempotency_key: &str,
319        receipt: ProcessingReceipt,
320        timestamp_millis: u64,
321    ) -> Result<(), DurabilityError> {
322        let stream_key = self.stream_key_for(idempotency_key);
323        let snapshot = self.load_snapshot(&stream_key, idempotency_key).await?;
324        let Some(entry) = snapshot.current.as_ref() else {
325            return Err(DurabilityError::DedupCollision {
326                key: idempotency_key.to_owned(),
327            });
328        };
329
330        let receipt_bytes = receipt.into_bytes();
331        if let Some(existing_receipt) = entry.receipt() {
332            if existing_receipt == receipt_bytes.as_slice() {
333                self.leave_in_flight(idempotency_key, snapshot.claim_sequence()?)
334                    .await;
335                return Ok(());
336            }
337            return Err(DurabilityError::DedupCollision {
338                key: idempotency_key.to_owned(),
339            });
340        }
341
342        let completed = DedupEntry::new(
343            entry.idempotency_key().to_owned(),
344            Some(receipt_bytes.clone()),
345            timestamp_millis,
346        );
347        match append_settled(
348            self.store.as_ref(),
349            &stream_key,
350            completed.serialize()?,
351            snapshot.next_seq,
352        )
353        .await
354        {
355            Ok(_) => {}
356            Err(DurabilityError::SequenceConflict { expected, actual }) => {
357                self.confirm_matching_receipt(
358                    &stream_key,
359                    idempotency_key,
360                    &receipt_bytes,
361                    expected,
362                    actual,
363                )
364                .await?;
365            }
366            Err(error) => return Err(error),
367        }
368        self.leave_in_flight(idempotency_key, snapshot.claim_sequence()?)
369            .await;
370        Ok(())
371    }
372
373    /// Releases an in-flight claim so the key becomes re-claimable.
374    ///
375    /// Callers use this on the publish failure path: a key was claimed
376    /// ([`DedupDecision::Claimed`]) but the downstream delivery failed before a
377    /// receipt could be recorded, leaving the key dangling [`DedupDecision::InFlight`]
378    /// forever and permanently suppressing every re-publish. Releasing appends a
379    /// tombstone so the next claim succeeds.
380    ///
381    /// This NEVER clobbers a stored receipt: an already-completed key (receipt
382    /// present) and an absent key are both no-ops, preserving the at-most-once
383    /// guarantee. Only a current in-flight entry (active, no receipt) is tombstoned.
384    ///
385    /// # Errors
386    ///
387    /// Propagates store read/append and serialization errors. A [`DurabilityError::SequenceConflict`]
388    /// on the tombstone append is re-checked: if the latest state is now completed
389    /// or already a tombstone the release is treated as successful, otherwise the
390    /// conflict is propagated.
391    pub async fn release_claim(&self, idempotency_key: &str) -> Result<(), DurabilityError> {
392        self.release_claim_at(idempotency_key, current_epoch_millis()?)
393            .await
394    }
395
396    async fn release_claim_at(
397        &self,
398        idempotency_key: &str,
399        timestamp_millis: u64,
400    ) -> Result<(), DurabilityError> {
401        let stream_key = self.stream_key_for(idempotency_key);
402        let snapshot = self.load_snapshot(&stream_key, idempotency_key).await?;
403        // No-op when there is nothing in flight: absent key, or a completed key
404        // whose receipt must never be clobbered (guards at-most-once).
405        let Some(entry) = snapshot.current.as_ref() else {
406            return Ok(());
407        };
408        if entry.receipt().is_some() {
409            return Ok(());
410        }
411
412        let tombstone = DedupRecord::tombstone(idempotency_key.to_owned(), timestamp_millis);
413        match append_settled(
414            self.store.as_ref(),
415            &stream_key,
416            tombstone.serialize()?,
417            snapshot.next_seq,
418        )
419        .await
420        {
421            Ok(_) => {}
422            Err(DurabilityError::SequenceConflict { expected, actual }) => {
423                self.confirm_release_after_conflict(&stream_key, idempotency_key, expected, actual)
424                    .await?;
425            }
426            Err(error) => return Err(error),
427        }
428        self.leave_in_flight(idempotency_key, snapshot.claim_sequence()?)
429            .await;
430        Ok(())
431    }
432
433    async fn confirm_release_after_conflict(
434        &self,
435        stream_key: &str,
436        idempotency_key: &str,
437        expected: u64,
438        actual: u64,
439    ) -> Result<(), DurabilityError> {
440        // A concurrent writer advanced the stream after our snapshot. Re-load and
441        // re-check the latest record directly (not via `into_active`, so a fresh
442        // tombstone is distinguishable): if it is now completed (receipt present)
443        // or already a tombstone, the in-flight claim is gone and the release goal
444        // is met. A still-active no-receipt entry means a legitimate re-claim won
445        // the race, so we must not clobber it -- propagate the conflict.
446        let latest = self.latest_record(stream_key, idempotency_key).await?;
447        match latest {
448            Some(DedupRecord::Tombstone { .. }) => Ok(()),
449            Some(DedupRecord::Active(entry)) if entry.receipt().is_some() => Ok(()),
450            _ => Err(DurabilityError::SequenceConflict { expected, actual }),
451        }
452    }
453
454    async fn latest_record(
455        &self,
456        stream_key: &str,
457        idempotency_key: &str,
458    ) -> Result<Option<DedupRecord>, DurabilityError> {
459        Ok(self.latest(stream_key, idempotency_key).await?.0)
460    }
461
462    /// The key stream's latest record and its next append sequence.
463    ///
464    /// A lookup reads the stream's first page. A page short of
465    /// [`READ_BATCH_SIZE`] rows is the whole stream, so a short stream, the
466    /// common case, costs one `read_from` and nothing else, as it did before
467    /// DUR-008 A12, and every record's key is checked. Only a stream that
468    /// fills its first page is read past it from its verified recorded head
469    /// (`dedup/head.rs`), at a cost bounded by that page plus the rows past
470    /// the head, never the key's whole history. No append records a head.
471    async fn latest(
472        &self,
473        stream_key: &str,
474        idempotency_key: &str,
475    ) -> Result<(Option<DedupRecord>, u64), DurabilityError> {
476        // ast-grep-allow: no-read-from-start: one bounded page is the whole of a short key stream (the J5 path, no head read); a stream that fills it is read past it from its verified head
477        let page = self.store.read_from(stream_key, 0, READ_BATCH_SIZE).await?;
478        if page.len() < READ_BATCH_SIZE {
479            let latest = fold_page(stream_key, idempotency_key, &page, 0)?;
480            return Ok((latest, len_to_u64(page.len())?));
481        }
482        self.latest_past_first_page(stream_key, idempotency_key, &page)
483            .await
484    }
485
486    fn scan_prefix(&self) -> String {
487        format!("{}:", self.namespace)
488    }
489
490    async fn decision_after_conflict(
491        &self,
492        stream_key: &str,
493        idempotency_key: &str,
494        expected: u64,
495        actual: u64,
496    ) -> Result<DedupDecision, DurabilityError> {
497        let snapshot = self.load_snapshot(stream_key, idempotency_key).await?;
498        snapshot.current.as_ref().map_or(
499            Err(DurabilityError::SequenceConflict { expected, actual }),
500            |entry| Ok(decision_for_entry(entry)),
501        )
502    }
503
504    async fn confirm_matching_receipt(
505        &self,
506        stream_key: &str,
507        idempotency_key: &str,
508        receipt_bytes: &[u8],
509        expected: u64,
510        actual: u64,
511    ) -> Result<(), DurabilityError> {
512        let snapshot = self.load_snapshot(stream_key, idempotency_key).await?;
513        if snapshot
514            .current
515            .as_ref()
516            .and_then(DedupEntry::receipt)
517            .is_some_and(|bytes| bytes == receipt_bytes)
518        {
519            Ok(())
520        } else {
521            Err(DurabilityError::SequenceConflict { expected, actual })
522        }
523    }
524
525    async fn load_snapshot(
526        &self,
527        stream_key: &str,
528        idempotency_key: &str,
529    ) -> Result<StreamSnapshot, DurabilityError> {
530        let (latest, next_seq) = self.latest(stream_key, idempotency_key).await?;
531        Ok(StreamSnapshot {
532            current: latest.and_then(DedupRecord::into_active),
533            next_seq,
534        })
535    }
536}
537
538impl fmt::Debug for DedupCache {
539    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
540        formatter
541            .debug_struct("DedupCache")
542            .field("namespace", &self.namespace)
543            .field("store", &self.store)
544            .field("in_flight", &self.in_flight)
545            .finish()
546    }
547}
548
549struct StreamSnapshot {
550    current: Option<DedupEntry>,
551    next_seq: u64,
552}
553
554impl StreamSnapshot {
555    /// The sequence of the stream's latest record, the claim a completion or
556    /// release written after this snapshot settles.
557    fn claim_sequence(&self) -> Result<u64, DurabilityError> {
558        self.next_seq.checked_sub(1).ok_or_else(|| {
559            DurabilityError::ConfigError("dedup stream has no record to settle".to_owned())
560        })
561    }
562}
563
564fn decision_for_entry(entry: &DedupEntry) -> DedupDecision {
565    entry.receipt().map_or(DedupDecision::InFlight, |bytes| {
566        DedupDecision::Completed(ProcessingReceipt::new(bytes.to_vec()))
567    })
568}
569
570/// Decodes a key stream's record, refusing one written for another original
571/// key.
572fn checked_record(payload: &[u8], idempotency_key: &str) -> Result<DedupRecord, DurabilityError> {
573    let record = DedupRecord::deserialize(payload)?;
574    if record.idempotency_key() != idempotency_key {
575        return Err(DurabilityError::DedupCollision {
576            key: idempotency_key.to_owned(),
577        });
578    }
579    Ok(record)
580}
581
582/// Checks one page of a key stream read from `offset`: each row sits at its
583/// own position and holds a record of `idempotency_key`. Answers the page's
584/// last record.
585fn fold_page(
586    stream_key: &str,
587    idempotency_key: &str,
588    page: &[StoredEntry],
589    offset: u64,
590) -> Result<Option<DedupRecord>, DurabilityError> {
591    let mut latest = None;
592    let mut expected = offset;
593    for stored in page {
594        if stored.sequence != expected {
595            return Err(DurabilityError::ConfigError(format!(
596                "dedup stream {stream_key} answered sequence {} where {expected} was read",
597                stored.sequence
598            )));
599        }
600        latest = Some(checked_record(&stored.payload, idempotency_key)?);
601        expected = expected.checked_add(1).ok_or_else(|| {
602            DurabilityError::ConfigError(format!("dedup stream {stream_key} sequence overflow"))
603        })?;
604    }
605    Ok(latest)
606}
607
608fn len_to_u64(len: usize) -> Result<u64, DurabilityError> {
609    u64::try_from(len).map_err(|error| {
610        DurabilityError::ConfigError(format!("dedup entry count cannot fit u64: {error}"))
611    })
612}
613
614fn current_epoch_millis() -> Result<u64, DurabilityError> {
615    let duration = SystemTime::now()
616        .duration_since(UNIX_EPOCH)
617        .map_err(|error| {
618            DurabilityError::ConfigError(format!("system clock is before Unix epoch: {error}"))
619        })?;
620    u64::try_from(duration.as_millis()).map_err(|error| {
621        DurabilityError::ConfigError(format!(
622            "current epoch millis cannot fit u64 for dedup receipt: {error}"
623        ))
624    })
625}
liminal/durability/dedup.rs

liminal/durability/
dedup.rs
