liminal/durability/channel/
pending.rs

1//! Exact append evidence survives cancellation until an authoritative read settles its outcome.
2use std::fmt;
3
4use super::DurableChannel;
5use crate::durability::DurabilityError;
6
7/// A publish sent to storage whose outcome has not yet been established.
8#[derive(Clone)]
9pub struct UnresolvedAppend {
10    channel: String,
11    partition: usize,
12    sequence: u64,
13    payload: Vec<u8>,
14    idempotency_key: Option<String>,
15    cause: String,
16}
17
18impl UnresolvedAppend {
19    pub(crate) fn new(
20        channel: String,
21        partition: usize,
22        sequence: u64,
23        payload: Vec<u8>,
24        idempotency_key: Option<String>,
25    ) -> Self {
26        Self {
27            channel,
28            partition,
29            sequence,
30            payload,
31            idempotency_key,
32            cause: "append future did not return an outcome".to_owned(),
33        }
34    }
35
36    /// The channel that owns this append.
37    #[must_use]
38    pub fn channel(&self) -> &str {
39        &self.channel
40    }
41
42    /// Partition whose next slot is reserved.
43    #[must_use]
44    pub const fn partition(&self) -> usize {
45        self.partition
46    }
47
48    /// Reserved durable sequence.
49    #[must_use]
50    pub const fn sequence(&self) -> u64 {
51        self.sequence
52    }
53
54    /// Original append error, or the fact that it returned no outcome.
55    #[must_use]
56    pub fn cause(&self) -> &str {
57        &self.cause
58    }
59
60    /// Producer's original idempotency key, if one was supplied.
61    #[must_use]
62    pub fn idempotency_key(&self) -> Option<&str> {
63        self.idempotency_key.as_deref()
64    }
65
66    pub(crate) fn set_cause(&mut self, cause: String) {
67        self.cause = cause;
68    }
69
70    pub(crate) fn payload(&self) -> &[u8] {
71        &self.payload
72    }
73}
74
75impl fmt::Display for UnresolvedAppend {
76    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
77        write!(
78            f,
79            "channel {} partition {} sequence {}: {}",
80            self.channel, self.partition, self.sequence, self.cause
81        )
82    }
83}
84
85impl fmt::Debug for UnresolvedAppend {
86    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
87        f.debug_struct("UnresolvedAppend")
88            .field("channel", &self.channel)
89            .field("partition", &self.partition)
90            .field("sequence", &self.sequence)
91            .field("byte_length", &self.payload.len())
92            .finish_non_exhaustive()
93    }
94}
95
96/// Authoritative outcome of a pending append or idempotency-key lookup.
97#[derive(Clone, Copy, Debug, PartialEq, Eq)]
98pub enum AppendResolution {
99    /// The row exists at its original durable position.
100    Committed { position: u64 },
101    /// The authoritative read established that the row is absent.
102    Absent,
103}
104
105impl DurableChannel {
106    /// Evidence retained for this channel's unresolved append.
107    #[must_use]
108    pub fn pending_append(&self) -> Option<UnresolvedAppend> {
109        self.pending.clone()
110    }
111
112    /// Resolve the exact reserved slot without writing another row.
113    ///
114    /// A row with the pending bytes at the reserved sequence is the append,
115    /// committed. No row, or another writer's row there, means the append is
116    /// absent: the store refuses an append at an occupied sequence. Another
117    /// writer's row also moves the partition head past it, as a sequence
118    /// conflict does.
119    ///
120    /// # Errors
121    /// An unreadable slot retains its original evidence and refuses by name.
122    pub async fn resolve_pending(&mut self) -> Result<Option<AppendResolution>, DurabilityError> {
123        let Some(pending) = self.pending.clone() else {
124            return Ok(None);
125        };
126        let stream_key = self.stream_key_for(pending.partition);
127        let row = self
128            .store
129            .read_at(&stream_key, pending.sequence)
130            .await
131            .map_err(|error| DurabilityError::PendingAppendUnreadable {
132                stream_key: stream_key.clone(),
133                sequence: pending.sequence,
134                append_error: pending.cause.clone(),
135                read_error: error.to_string(),
136            })?;
137        let Some(row) = row else {
138            self.pending = None;
139            return Ok(Some(AppendResolution::Absent));
140        };
141        let next = row.sequence.checked_add(1).ok_or_else(|| {
142            DurabilityError::ConfigError(format!(
143                "channel {stream_key} sequence overflow during recovery"
144            ))
145        })?;
146        if row.sequence != pending.sequence || row.payload != pending.payload {
147            self.pending = None;
148            self.refresh_sequence(pending.partition, &stream_key, pending.sequence, next)
149                .await?;
150            return Ok(Some(AppendResolution::Absent));
151        }
152        self.pending = None;
153        self.advance_partition(pending.partition, &stream_key, next)
154            .await?;
155        Ok(Some(AppendResolution::Committed {
156            position: row.sequence,
157        }))
158    }
159}
liminal/durability/channel/pending.rs

liminal/durability/channel/
pending.rs
