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
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
//! Named storage failures and uncertain channel append outcomes.
/// Error taxonomy for haematite-backed durability operations.
#[derive(Debug, thiserror::Error)]
pub enum DurabilityError {
/// A logical operation has no staged rows to commit.
#[error("operation batch is empty")]
EmptyOperationBatch,
/// A store does not implement atomic routed storage.
#[error("{store} does not support atomic routed storage")]
BatchUnsupported {
/// The store's type.
store: &'static str,
},
/// The store did not establish whether a sent append committed.
#[error("append to {stream_key} at sequence {sequence} is unresolved: {cause}")]
AppendUnresolved {
/// Partition stream receiving the append.
stream_key: String,
/// Reserved sequence of the sent bytes.
sequence: u64,
/// Original append failure, never used to classify the outcome.
cause: String,
},
/// The authoritative read of an earlier uncertain append failed.
#[error(
"append to {stream_key} at sequence {sequence} remains unresolved: append: {append_error}; read-back: {read_error}"
)]
PendingAppendUnreadable {
/// Partition stream that must be reconciled.
stream_key: String,
/// Reserved sequence of the sent bytes.
sequence: u64,
/// Original append error, retained across every failed read.
append_error: String,
/// The latest authoritative read error.
read_error: String,
},
/// A write the store had started when its caller stopped waiting: it may
/// have been applied. It is never treated as not written: the caller
/// reads the position back to settle it (`settle`), or refuses by this
/// name.
#[error(
"write outcome unknown on shard {pid}: it may have been applied; read the position again"
)]
OutcomeUnknown {
/// The shard that had started the write.
pid: u64,
},
/// A write withdrawn before the store started it: nothing was written,
/// and it may be sent again.
#[error("write not started on shard {pid}: nothing was written")]
NotStarted {
/// The shard the write was queued for.
pid: u64,
},
/// Haematite returned a store-level failure.
///
/// The umbrella `From<haematite::ApiError>` conversion lives in
/// [`super::store`] because it routes the optimistic-concurrency variants to
/// their dedicated cases rather than wrapping them here.
#[error("haematite store error: {0}")]
StoreError(haematite::ApiError),
/// An append observed a different stream sequence than the caller expected.
#[error("sequence conflict: expected {expected}, actual {actual}")]
SequenceConflict {
/// Caller-provided expected stream sequence.
expected: u64,
/// Actual stream sequence reported by haematite.
actual: u64,
},
/// A cursor checkpoint attempted to move from a stale stored value.
#[error("cursor regression: stored {stored}, attempted {attempted}")]
CursorRegression {
/// Current value stored for the cursor.
stored: u64,
/// Value the caller attempted to checkpoint from.
attempted: u64,
},
/// A producer idempotency key collided with an existing dedup entry.
#[error("dedup key collision for key {key}")]
DedupCollision {
/// Idempotency key that collided.
key: String,
},
/// Durability configuration failed validation.
#[error("configuration error: {0}")]
ConfigError(String),
/// An ephemeral durable store could not be opened on disk.
///
/// Raised only on the construction path of a self-owned ephemeral store
/// (see [`super::EphemeralRoot::open_store`]); the guarding temporary directory
/// is already removed by the time this surfaces.
#[error("ephemeral store open failed: {0}")]
EphemeralStoreOpen(String),
/// An operation reached an ephemeral store whose backing database was
/// already detached by teardown.
///
/// Unreachable by construction: the store is detached only inside the
/// ephemeral guard's `Drop`, which cannot overlap a live handle's call.
/// The variant exists because that detachment is expressed through an
/// `Option` the compiler cannot see through, and the fallback must be a
/// typed refusal rather than a panic.
#[error("ephemeral store already detached by teardown")]
EphemeralStoreDetached,
/// A store that cannot delete entries was asked to truncate a stream.
#[error("{store} cannot truncate {stream_key} below sequence {before_seq}")]
TruncateUnsupported {
/// The store's type.
store: &'static str,
/// The stream it was asked to cut.
stream_key: String,
/// The sequence it was asked to cut below.
before_seq: u64,
},
/// A store that cannot list a routed key window was asked to scan one.
#[error("{store} cannot scan routed keys on route {route_key}")]
RoutedScanUnsupported {
/// The store's type.
store: &'static str,
/// The route it was asked to scan.
route_key: String,
},
/// A store that cannot delete a routed key window in one commit was
/// asked to remove routed rows.
///
/// Composing the window from single deletes is never done in its place
/// (DUR-018 R3): each would be its own commit.
#[error("{store} cannot remove routed rows on route {route_key} in one commit")]
RoutedRemoveUnsupported {
/// The store's type.
store: &'static str,
/// The route it was asked to cut.
route_key: String,
},
/// A store that cannot delete a key inside a routed batch was asked to
/// commit a batch with a delete (DUR-028 R3). Nothing was written: the
/// batch is one durable operation or none.
#[error("{store} cannot delete inside a routed batch on route {route_key}")]
RoutedDeleteUnsupported {
/// The store's type.
store: &'static str,
/// The route the batch was for.
route_key: String,
},
/// Persisted envelope bytes could not be encoded or decoded.
#[error("envelope serialization error: {0}")]
EnvelopeError(String),
}
impl From<haematite::SequenceConflict> for DurabilityError {
fn from(error: haematite::SequenceConflict) -> Self {
Self::SequenceConflict {
expected: error.expected,
actual: error.actual,
}
}
}
impl From<haematite::CasMismatch> for DurabilityError {
fn from(error: haematite::CasMismatch) -> Self {
// The real `CasMismatch` carries `Option<u64>` to distinguish absent
// (`None`) from stored-zero (`Some(0)`). The cursor contract treats an
// absent key as the value 0 (see `HaematiteStore::cas`), so a `None`
// collapses to 0 here when reporting a regression.
Self::CursorRegression {
stored: error.actual.unwrap_or(0),
attempted: error.expected.unwrap_or(0),
}
}
}