liminal/durability/dedup/
unresolved.rs

1//! Durable producer claims carry their recovery evidence in the same record write.
2use super::{ClaimEvidence, DedupCache, DedupDecision, DedupEntry, decision_for_entry};
3use crate::durability::DurabilityError;
4use crate::durability::settle::append_settled;
5
6impl DedupCache {
7    /// Claim a producer key and persist where its row would be in one append.
8    ///
9    /// The key is first named in the namespace's in-flight index
10    /// (`dedup/in_flight.rs`), so a start finds the claim without walking the
11    /// namespace; a crash between the two writes leaves an index row that the
12    /// next start drops, never a claim the index does not name.
13    ///
14    /// # Errors
15    /// Propagates storage and codec failures; conflicting keys retain the normal
16    /// dedup collision refusal. A failed write must be resolved before publishing.
17    pub async fn claim_with_evidence(
18        &self,
19        key: &str,
20        evidence: ClaimEvidence,
21        timestamp_millis: u64,
22    ) -> Result<DedupDecision, DurabilityError> {
23        let stream = self.stream_key_for(key);
24        let snapshot = self.load_snapshot(&stream, key).await?;
25        if let Some(entry) = snapshot.current.as_ref() {
26            return Ok(decision_for_entry(entry));
27        }
28        let mut entry = DedupEntry::new(key, None, timestamp_millis);
29        entry.evidence = Some(evidence);
30        let payload = entry.serialize()?;
31        let row = self.join_in_flight(key, snapshot.next_seq).await?;
32        match append_settled(self.store.as_ref(), &stream, payload, snapshot.next_seq).await {
33            Ok(_) => Ok(DedupDecision::Claimed),
34            Err(DurabilityError::SequenceConflict { expected, actual }) => {
35                self.unjoin_in_flight(key, row);
36                self.decision_after_conflict(&stream, key, expected, actual)
37                    .await
38            }
39            // The claim may have landed: the join stays, and the next start
40            // looks the key up.
41            Err(error) => Err(error),
42        }
43    }
44}
liminal/durability/dedup/unresolved.rs

liminal/durability/dedup/
unresolved.rs
