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
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
use std::fmt;
use std::io;
/// Errors surfaced by the top-level database handle.
#[derive(Debug)]
pub enum DatabaseError {
DirectoryCreate(io::Error),
ConfigWrite(io::Error),
ConfigRead(io::Error),
ConfigParse(String),
InvalidShardCount,
ShardSpawn(String),
SweepSpawn(String),
ShardError(String),
SweepError(String),
SyncSchedulerSpawn(String),
SyncSchedulerError(String),
IoError(io::Error),
MissingSweepInterval,
InvalidSweepInterval,
MissingSyncTopology,
InvalidSyncInterval,
SequenceConflict {
expected: u64,
actual: u64,
},
CasMismatch {
expected: Option<u64>,
actual: Option<u64>,
},
ConsistencyError(String),
/// A live distribution-endpoint operation failed (no endpoint attached, a
/// transport send/connect failure, or a disconnected inbound drain).
Distribution(String),
/// A replicated write reached peer-quorum but the proposer could not durably
/// apply its OWN committed value locally (see [`crate::db::Database::replicate_write`]).
///
/// This is reported, never swallowed: a committed write that is absent on its
/// own writer is a correctness hazard (it reopens the heal-mid-write
/// split-brain hole). Under single-owner-per-key (the step-3 epoch fence) the
/// local CAS can never mismatch, so this only ever surfaces a genuine local
/// storage/IO fault.
LocalCommitFailed(String),
/// An [`crate::db::Database::acquire_shard`] election lost: a strictly higher
/// ballot was promised elsewhere on every attempt. The candidate is NOT the
/// owner and recorded no `owner_epoch`. Carries the highest competing counter
/// seen so a caller could retry above it later. This is a clean, safe loss —
/// the unique-ballot / majority invariants were never relaxed.
ElectionLost {
highest_seen: u64,
},
/// An [`crate::db::Database::acquire_shard`] election could not collect a
/// majority of promises within the timeout on any attempt (e.g. a minority of
/// nodes was reachable). The candidate is NOT the owner — never a false win.
ElectionTimeout {
attempts: u32,
},
/// A replicated CAS write was deterministically out-voted by the cluster: a
/// stale/deposed owner's proposal collected enough rejects that a quorum of
/// accepts is no longer reachable, so the writer is fenced and NOTHING was
/// applied (the typed twin of [`crate::ConsistencyError::Fenced`]). Surfaced
/// as its own variant — distinct from a generic [`Self::ConsistencyError`]
/// string — so a consumer (e.g. an aion shard owner) can match the fence
/// directly and re-resolve ownership rather than parsing a Display message.
Fenced {
required: usize,
possible_accepts: usize,
},
/// A replicated CAS write was deterministically out-voted by *value-CAS
/// mismatches alone* — the writer is still the live owner, but enough replicas
/// refused the precondition that a quorum of accepts became unreachable (the
/// typed twin of [`crate::ConsistencyError::CasConflict`]). Distinct from
/// [`Self::Fenced`] (a higher-ballot owner deposed us, requiring ownership
/// re-resolution): a `CasConflict` caller may simply re-read and re-CAS.
CasConflict {
required: usize,
possible_accepts: usize,
},
}
impl fmt::Display for DatabaseError {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::DirectoryCreate(error) => {
write!(formatter, "failed to create database directory: {error}")
}
Self::ConfigWrite(error) => {
write!(formatter, "failed to write database config: {error}")
}
Self::ConfigRead(error) => write!(formatter, "failed to read database config: {error}"),
Self::ConfigParse(message) => {
write!(formatter, "failed to parse database config: {message}")
}
Self::InvalidShardCount => write!(formatter, "database shard_count must be at least 1"),
Self::ShardSpawn(message) => {
write!(formatter, "failed to spawn shard actor: {message}")
}
Self::SweepSpawn(message) => {
write!(formatter, "failed to spawn sweep actor: {message}")
}
Self::ShardError(message) => write!(formatter, "shard operation failed: {message}"),
Self::SweepError(message) => write!(formatter, "sweep operation failed: {message}"),
Self::SyncSchedulerSpawn(message) => {
write!(formatter, "failed to spawn sync scheduler: {message}")
}
Self::SyncSchedulerError(message) => {
write!(formatter, "sync scheduler failed: {message}")
}
Self::IoError(error) => write!(formatter, "database I/O error: {error}"),
Self::MissingSweepInterval => write!(formatter, "ttl writes require sweep_interval"),
Self::InvalidSweepInterval => {
write!(formatter, "sweep_interval must be greater than zero")
}
Self::MissingSyncTopology => {
write!(formatter, "distributed database requires sync topology")
}
Self::InvalidSyncInterval => {
write!(formatter, "sync_interval must be greater than zero")
}
Self::SequenceConflict { expected, actual } => write!(
formatter,
"sequence conflict on append: expected {expected}, actual {actual}"
),
Self::CasMismatch { expected, actual } => write!(
formatter,
"cas mismatch: expected {expected:?}, actual {actual:?}"
),
Self::ConsistencyError(message) => {
write!(formatter, "consistency requirement failed: {message}")
}
Self::Distribution(message) => {
write!(formatter, "distribution endpoint error: {message}")
}
Self::LocalCommitFailed(message) => write!(
formatter,
"replicated write reached quorum but local durable commit failed: {message}"
),
Self::ElectionLost { highest_seen } => write!(
formatter,
"shard election lost: a higher ballot (counter {highest_seen}) was promised elsewhere"
),
Self::ElectionTimeout { attempts } => write!(
formatter,
"shard election timed out without a majority after {attempts} attempts"
),
Self::Fenced {
required,
possible_accepts,
} => write!(
formatter,
"fenced by CAS rejects: required {required} accepts, only {possible_accepts} still possible"
),
Self::CasConflict {
required,
possible_accepts,
} => write!(
formatter,
"lost CAS by value mismatch: required {required} accepts, only {possible_accepts} still possible"
),
}
}
}
impl std::error::Error for DatabaseError {
fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
match self {
Self::DirectoryCreate(error)
| Self::ConfigWrite(error)
| Self::ConfigRead(error)
| Self::IoError(error) => Some(error),
Self::ConfigParse(_)
| Self::InvalidShardCount
| Self::ShardSpawn(_)
| Self::SweepSpawn(_)
| Self::ShardError(_)
| Self::SweepError(_)
| Self::SyncSchedulerSpawn(_)
| Self::SyncSchedulerError(_)
| Self::MissingSweepInterval
| Self::InvalidSweepInterval
| Self::MissingSyncTopology
| Self::InvalidSyncInterval
| Self::SequenceConflict { .. }
| Self::CasMismatch { .. }
| Self::ConsistencyError(_)
| Self::Distribution(_)
| Self::LocalCommitFailed(_)
| Self::ElectionLost { .. }
| Self::ElectionTimeout { .. }
| Self::Fenced { .. }
| Self::CasConflict { .. } => None,
}
}
}
impl From<io::Error> for DatabaseError {
fn from(error: io::Error) -> Self {
Self::IoError(error)
}
}
impl From<crate::sync::ConsistencyError> for DatabaseError {
/// Preserve the deterministic CAS fence as the typed [`Self::Fenced`] so
/// consumers can match it; every other consistency failure keeps its existing
/// stringified [`Self::ConsistencyError`] form (behaviour unchanged).
fn from(error: crate::sync::ConsistencyError) -> Self {
match error {
crate::sync::ConsistencyError::Fenced {
required,
possible_accepts,
} => Self::Fenced {
required,
possible_accepts,
},
crate::sync::ConsistencyError::CasConflict {
required,
possible_accepts,
} => Self::CasConflict {
required,
possible_accepts,
},
other => Self::ConsistencyError(other.to_string()),
}
}
}
#[cfg(test)]
mod tests {
use super::DatabaseError;
use crate::sync::ConsistencyError;
#[test]
fn consistency_fence_maps_to_typed_fenced() {
let mapped = DatabaseError::from(ConsistencyError::Fenced {
required: 3,
possible_accepts: 1,
});
assert!(
matches!(
mapped,
DatabaseError::Fenced {
required: 3,
possible_accepts: 1
}
),
"the deterministic CAS fence must survive as the typed DatabaseError::Fenced"
);
}
#[test]
fn consistency_cas_conflict_maps_to_typed_cas_conflict() {
// The value-CAS loss must survive as the typed DatabaseError::CasConflict
// (preserving required/possible_accepts), distinct from the typed Fenced and
// from the stringified fallback.
let mapped = DatabaseError::from(ConsistencyError::CasConflict {
required: 3,
possible_accepts: 1,
});
assert!(
matches!(
mapped,
DatabaseError::CasConflict {
required: 3,
possible_accepts: 1
}
),
"the value-CAS loss must survive as the typed DatabaseError::CasConflict"
);
}
#[test]
fn other_consistency_failures_stay_stringified() {
// A non-fence consistency failure must NOT be misclassified as a fence; it
// keeps its existing stringified ConsistencyError form (behaviour unchanged).
for error in [
ConsistencyError::QuorumUnavailable {
required: 2,
possible: 1,
},
ConsistencyError::TransportUnavailable,
ConsistencyError::AckFailed,
] {
let display = error.to_string();
let mapped = DatabaseError::from(error);
assert!(
matches!(mapped, DatabaseError::ConsistencyError(ref message) if *message == display),
"non-fence consistency failures must remain DatabaseError::ConsistencyError"
);
}
}
}