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
//! Named storage failures and uncertain channel append outcomes.
/// Error taxonomy for haematite-backed durability operations.
#[derive(Debug, thiserror::Error)]
pub enum DurabilityError {
/// 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::store::open_ephemeral`]); 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,
},
/// 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),
}
}
}