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