liminal/durability/channel/
presence.rs

1//! Authoritative idempotency-key lookup reads a channel's partitions page by page,
2//! without changing admission state and without holding the channel's append lock.
3use std::sync::Arc;
4
5use super::{AppendResolution, DurableChannel, MessageEnvelope};
6use crate::durability::head_checkpoint::READ_BATCH_SIZE;
7use crate::durability::{DurabilityError, DurableStore};
8
9/// The streams a key lookup reads, taken from a channel so the lookup can run
10/// after the channel's append lock is released.
11#[derive(Debug)]
12pub struct KeyLookup {
13    stream_keys: Vec<String>,
14    store: Arc<dyn DurableStore>,
15}
16
17impl DurableChannel {
18    /// The streams that answer for this channel's idempotency keys.
19    ///
20    /// A key is looked up only once its pending append is settled, and a held
21    /// key is never appended again, so rows appended while the lookup reads
22    /// cannot carry it: the lookup needs no lock.
23    #[must_use]
24    pub fn key_lookup(&self) -> KeyLookup {
25        KeyLookup {
26            stream_keys: (0..self.partition_count())
27                .map(|partition| self.stream_key_for(partition))
28                .collect(),
29            store: Arc::clone(&self.store),
30        }
31    }
32
33    /// Find the original durable position of a producer's key across
34    /// partitions, reading each partition from its position in `from`.
35    ///
36    /// # Errors
37    /// Propagates authoritative read and envelope decode errors. A failed read
38    /// never establishes absence.
39    pub async fn find_idempotency_key(
40        &self,
41        key: &str,
42        from: &[u64],
43    ) -> Result<AppendResolution, DurabilityError> {
44        self.key_lookup().find(key, from).await
45    }
46}
47
48impl KeyLookup {
49    /// Find the original durable position of a producer's key, reading each
50    /// partition from its position in `from` in pages of the recovery read
51    /// size.
52    ///
53    /// `from` is where each partition stood when the key was claimed, which
54    /// is before the key's row could be appended, so no row before it can
55    /// carry the key. A partition `from` does not name is read from its first
56    /// row.
57    ///
58    /// # Errors
59    /// Propagates authoritative read and envelope decode errors. A failed read
60    /// never establishes absence.
61    pub async fn find(&self, key: &str, from: &[u64]) -> Result<AppendResolution, DurabilityError> {
62        for (partition, stream_key) in self.stream_keys.iter().enumerate() {
63            let mut sequence = from.get(partition).copied().unwrap_or(0);
64            loop {
65                let page = self
66                    .store
67                    .read_from(stream_key, sequence, READ_BATCH_SIZE)
68                    .await?;
69                for row in &page {
70                    if row.sequence != sequence {
71                        return Err(DurabilityError::ConfigError(format!(
72                            "stream {stream_key} read at {sequence} returned sequence {}",
73                            row.sequence
74                        )));
75                    }
76                    let envelope = MessageEnvelope::deserialize(&row.payload)?;
77                    if envelope.idempotency_key.as_deref() == Some(key) {
78                        return Ok(AppendResolution::Committed {
79                            position: row.sequence,
80                        });
81                    }
82                    sequence = sequence.checked_add(1).ok_or_else(|| {
83                        DurabilityError::ConfigError(format!(
84                            "stream {stream_key} sequence overflow during key lookup"
85                        ))
86                    })?;
87                }
88                if page.len() < READ_BATCH_SIZE {
89                    break;
90                }
91            }
92        }
93        Ok(AppendResolution::Absent)
94    }
95}
liminal/durability/channel/presence.rs

liminal/durability/channel/
presence.rs
