haematite_cli/serve/
receipt_wire.rs

1//! Lossless typed projections retain every admitted outcome and its causes.
2
3use haematite::{BatchOutcome, BatchReceipt, DatabaseError, ShardError};
4use serde::{Deserialize, Serialize};
5
6use super::receipt_fragments::{ReceiptCodecError, encode_hex};
7
8#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
9#[serde(deny_unknown_fields)]
10pub struct WireReceipt {
11    pub outcomes: Vec<WireJobOutcome>,
12    pub submit_error: Option<WireCause>,
13    pub first_error: Option<WireCause>,
14}
15
16#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
17#[serde(deny_unknown_fields)]
18pub struct WireJobOutcome {
19    pub job_index: u64,
20    pub shard: u64,
21    pub outcome: WireOutcome,
22}
23
24#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
25#[serde(tag = "kind", deny_unknown_fields)]
26pub enum WireOutcome {
27    Committed {
28        root: [u8; 32],
29        callback_failure: Option<WireCause>,
30    },
31    SequenceRead {
32        sequences: Vec<WireSequence>,
33    },
34    RangeRead {
35        sequences: Vec<WireSequence>,
36        hit_limit: bool,
37        invalid_outside_range: u64,
38    },
39    Failed {
40        cause: WireCause,
41    },
42    Unknown {
43        cause: WireCause,
44    },
45}
46
47#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
48#[serde(deny_unknown_fields)]
49pub struct WireSequence {
50    #[serde(deserialize_with = "read_key_hex")]
51    pub key_hex: String,
52    pub sequence: u64,
53}
54
55#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
56#[serde(deny_unknown_fields)]
57pub struct WireCause {
58    pub kind: DatabaseErrorKind,
59    pub message: String,
60    pub shard: Option<WireShardCause>,
61    pub callback: Option<WireCallback>,
62    pub batch_receipt: Option<Box<WireReceipt>>,
63    pub first_error: Option<Box<Self>>,
64}
65
66#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
67#[serde(deny_unknown_fields)]
68pub struct WireShardCause {
69    pub kind: ShardErrorKind,
70    pub message: String,
71}
72
73#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
74pub enum ShardErrorKind {
75    QueueFull,
76    RequestSizeOverflow,
77    ActorUnavailable,
78    ReplyDisconnected,
79    OutcomeUnknown,
80    Wal,
81    Tree,
82    Store,
83    Spawn,
84    ReclaimClose,
85    ShutdownStorage,
86    SequenceConflict,
87    ReservedSequenceKey,
88    EmptyAppend,
89    LeafRecordCorrupt,
90    LeafLogCorrupt,
91    LeafLogNotAdmitted,
92    LeafLogTruncate,
93    LeafLogStreamDelete,
94    LeafLogTtlWrite,
95    LeafLogMergeAdopt,
96    PinWouldGoBack,
97    PinConflict,
98    PinBeyondExtent,
99    CasMismatch,
100    CasHashMismatch,
101    Fenced,
102    WriteDuringRootAdvanceEmission,
103    DatabaseClosed,
104}
105
106impl ShardErrorKind {
107    const fn project(error: &ShardError) -> Self {
108        match error {
109            ShardError::QueueFull { .. } => Self::QueueFull,
110            ShardError::RequestSizeOverflow { .. } => Self::RequestSizeOverflow,
111            ShardError::ActorUnavailable { .. } => Self::ActorUnavailable,
112            ShardError::ReplyDisconnected { .. } => Self::ReplyDisconnected,
113            ShardError::OutcomeUnknown { .. } => Self::OutcomeUnknown,
114            ShardError::Wal(_) => Self::Wal,
115            ShardError::Tree(_) => Self::Tree,
116            ShardError::Store(_) => Self::Store,
117            ShardError::Spawn(_) => Self::Spawn,
118            ShardError::ReclaimClose { .. } => Self::ReclaimClose,
119            ShardError::ShutdownStorage { .. } => Self::ShutdownStorage,
120            ShardError::SequenceConflict { .. } => Self::SequenceConflict,
121            ShardError::ReservedSequenceKey => Self::ReservedSequenceKey,
122            ShardError::EmptyAppend => Self::EmptyAppend,
123            ShardError::LeafRecordCorrupt { .. } => Self::LeafRecordCorrupt,
124            ShardError::LeafLogCorrupt { .. } => Self::LeafLogCorrupt,
125            ShardError::LeafLogNotAdmitted => Self::LeafLogNotAdmitted,
126            ShardError::LeafLogTruncate => Self::LeafLogTruncate,
127            ShardError::LeafLogStreamDelete => Self::LeafLogStreamDelete,
128            ShardError::LeafLogTtlWrite => Self::LeafLogTtlWrite,
129            ShardError::LeafLogMergeAdopt => Self::LeafLogMergeAdopt,
130            ShardError::PinWouldGoBack { .. } => Self::PinWouldGoBack,
131            ShardError::PinConflict { .. } => Self::PinConflict,
132            ShardError::PinBeyondExtent { .. } => Self::PinBeyondExtent,
133            ShardError::CasMismatch { .. } => Self::CasMismatch,
134            ShardError::CasHashMismatch { .. } => Self::CasHashMismatch,
135            ShardError::Fenced { .. } => Self::Fenced,
136            ShardError::WriteDuringRootAdvanceEmission => Self::WriteDuringRootAdvanceEmission,
137            ShardError::DatabaseClosed => Self::DatabaseClosed,
138        }
139    }
140}
141
142#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
143#[serde(deny_unknown_fields)]
144pub struct WireCallback {
145    pub job_index: u64,
146    pub shard: u64,
147    pub pid: u64,
148    pub command: String,
149    pub subscriber_index: u64,
150    pub event: WireAdvance,
151}
152
153#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
154#[serde(deny_unknown_fields)]
155pub struct WireAdvance {
156    pub shard_id: u64,
157    pub prior_root: [u8; 32],
158    pub new_root: [u8; 32],
159    pub advance_gen: u64,
160}
161
162#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
163pub enum DatabaseErrorKind {
164    BatchFailed,
165    BatchShardFailed,
166    RootCallbackPanicked,
167    BatchJobUnknown,
168    TailRepairRefused,
169    DirectoryCreate,
170    ConfigWrite,
171    ConfigRead,
172    ConfigParse,
173    InvalidShardCount,
174    InvalidLongWaitInterval,
175    InvalidQueueByteLimit,
176    ShardQueueFull,
177    ShardRequestSizeOverflow,
178    ShardSpawn,
179    SweepSpawn,
180    ShardError,
181    OutcomeUnknown,
182    SweepError,
183    SyncSchedulerSpawn,
184    SyncSchedulerError,
185    IoError,
186    InvalidSweepInterval,
187    MissingSyncTopology,
188    InvalidSyncInterval,
189    FormatVersionTooNew,
190    ReservedSequenceKey,
191    ReservedBranchRefKey,
192    SequenceConflict,
193    CasMismatch,
194    CasHashMismatch,
195    ConsistencyError,
196    Distribution,
197    LocalCommitFailed,
198    CommittedButOwnerBehind,
199    ElectionLost,
200    ElectionTimeout,
201    Fenced,
202    CasConflict,
203    ClusterMembers,
204    DataDirLocked,
205    LockFileIo,
206    DataDirAlreadyInitialised,
207    LeafLogStore,
208    EphemeralStore,
209    PathWriterRequired,
210    WriteDuringRootAdvanceEmission,
211    DuplicateShardId,
212    ExecutorThreadsInvalid,
213    MissingNodeCacheBudget,
214    InvalidNodeCacheBudget,
215    RetentionRefused,
216    StreamDeleteRefused,
217    RootNotRetained,
218    ReclaimFailed,
219    ExecutorShutdown,
220    MigrationInProgress,
221    InvalidChunkingStamp,
222    MigrationFailed,
223    MigrationRefusedFormat,
224    MigrationRefusedWalTail,
225    MigrationRefusedUncommittedMutations,
226    DatabaseClosed,
227}
228
229impl DatabaseErrorKind {
230    const fn project(error: &DatabaseError) -> Self {
231        match error {
232            DatabaseError::BatchFailed(_) => Self::BatchFailed,
233            DatabaseError::BatchShardFailed { .. } => Self::BatchShardFailed,
234            DatabaseError::RootCallbackPanicked { .. } => Self::RootCallbackPanicked,
235            DatabaseError::BatchJobUnknown { .. } => Self::BatchJobUnknown,
236            DatabaseError::TailRepairRefused { .. } => Self::TailRepairRefused,
237            DatabaseError::DirectoryCreate(_) => Self::DirectoryCreate,
238            DatabaseError::ConfigWrite(_) => Self::ConfigWrite,
239            DatabaseError::ConfigRead(_) => Self::ConfigRead,
240            DatabaseError::ConfigParse(_) => Self::ConfigParse,
241            DatabaseError::InvalidShardCount => Self::InvalidShardCount,
242            DatabaseError::InvalidLongWaitInterval { .. } => Self::InvalidLongWaitInterval,
243            DatabaseError::InvalidQueueByteLimit => Self::InvalidQueueByteLimit,
244            DatabaseError::ShardQueueFull { .. } => Self::ShardQueueFull,
245            DatabaseError::ShardRequestSizeOverflow { .. } => Self::ShardRequestSizeOverflow,
246            DatabaseError::ShardSpawn(_) => Self::ShardSpawn,
247            DatabaseError::DatabaseClosed => Self::DatabaseClosed,
248            DatabaseError::SweepSpawn(_) => Self::SweepSpawn,
249            DatabaseError::ShardError(_) => Self::ShardError,
250            DatabaseError::OutcomeUnknown { .. } => Self::OutcomeUnknown,
251            DatabaseError::SweepError(_) => Self::SweepError,
252            DatabaseError::SyncSchedulerSpawn(_) => Self::SyncSchedulerSpawn,
253            DatabaseError::SyncSchedulerError(_) => Self::SyncSchedulerError,
254            DatabaseError::IoError(_) => Self::IoError,
255            DatabaseError::InvalidSweepInterval => Self::InvalidSweepInterval,
256            DatabaseError::MissingSyncTopology => Self::MissingSyncTopology,
257            DatabaseError::InvalidSyncInterval => Self::InvalidSyncInterval,
258            DatabaseError::FormatVersionTooNew { .. } => Self::FormatVersionTooNew,
259            DatabaseError::ReservedSequenceKey => Self::ReservedSequenceKey,
260            DatabaseError::ReservedBranchRefKey => Self::ReservedBranchRefKey,
261            DatabaseError::SequenceConflict { .. } => Self::SequenceConflict,
262            DatabaseError::CasMismatch { .. } => Self::CasMismatch,
263            DatabaseError::CasHashMismatch { .. } => Self::CasHashMismatch,
264            DatabaseError::ConsistencyError(_) => Self::ConsistencyError,
265            DatabaseError::Distribution(_) => Self::Distribution,
266            DatabaseError::LocalCommitFailed(_) => Self::LocalCommitFailed,
267            DatabaseError::CommittedButOwnerBehind { .. } => Self::CommittedButOwnerBehind,
268            DatabaseError::ElectionLost { .. } => Self::ElectionLost,
269            DatabaseError::ElectionTimeout { .. } => Self::ElectionTimeout,
270            DatabaseError::Fenced { .. } => Self::Fenced,
271            DatabaseError::CasConflict { .. } => Self::CasConflict,
272            DatabaseError::ClusterMembers(_) => Self::ClusterMembers,
273            DatabaseError::DataDirLocked { .. } => Self::DataDirLocked,
274            DatabaseError::LockFileIo { .. } => Self::LockFileIo,
275            DatabaseError::DataDirAlreadyInitialised { .. } => Self::DataDirAlreadyInitialised,
276            DatabaseError::LeafLogStore { .. } => Self::LeafLogStore,
277            DatabaseError::EphemeralStore { .. } => Self::EphemeralStore,
278            DatabaseError::PathWriterRequired { .. } => Self::PathWriterRequired,
279            DatabaseError::WriteDuringRootAdvanceEmission => Self::WriteDuringRootAdvanceEmission,
280            DatabaseError::DuplicateShardId { .. } => Self::DuplicateShardId,
281            DatabaseError::ExecutorThreadsInvalid(_) => Self::ExecutorThreadsInvalid,
282            DatabaseError::MissingNodeCacheBudget => Self::MissingNodeCacheBudget,
283            DatabaseError::InvalidNodeCacheBudget(_) => Self::InvalidNodeCacheBudget,
284            DatabaseError::RetentionRefused(_) => Self::RetentionRefused,
285            DatabaseError::StreamDeleteRefused(_) => Self::StreamDeleteRefused,
286            DatabaseError::RootNotRetained { .. } => Self::RootNotRetained,
287            DatabaseError::ReclaimFailed { .. } => Self::ReclaimFailed,
288            DatabaseError::ExecutorShutdown => Self::ExecutorShutdown,
289            DatabaseError::MigrationInProgress { .. } => Self::MigrationInProgress,
290            DatabaseError::InvalidChunkingStamp(_) => Self::InvalidChunkingStamp,
291            DatabaseError::MigrationFailed(_) => Self::MigrationFailed,
292            DatabaseError::MigrationRefusedFormat { .. } => Self::MigrationRefusedFormat,
293            DatabaseError::MigrationRefusedWalTail { .. } => Self::MigrationRefusedWalTail,
294            DatabaseError::MigrationRefusedUncommittedMutations { .. } => {
295                Self::MigrationRefusedUncommittedMutations
296            }
297        }
298    }
299}
300
301fn index(value: usize) -> Result<u64, ReceiptCodecError> {
302    u64::try_from(value).map_err(|_| ReceiptCodecError::IndexOverflow)
303}
304
305fn sequences(values: &[(Vec<u8>, u64)]) -> Result<Vec<WireSequence>, ReceiptCodecError> {
306    let mut result = Vec::new();
307    result
308        .try_reserve_exact(values.len())
309        .map_err(ReceiptCodecError::allocation)?;
310    for (key, sequence) in values {
311        result.push(WireSequence {
312            key_hex: encode_hex(key)?,
313            sequence: *sequence,
314        });
315    }
316    Ok(result)
317}
318
319impl WireReceipt {
320    /// Borrow the original receipt while building its complete portable projection.
321    ///
322    /// # Errors
323    ///
324    /// Refuses an unrepresentable index or failed allocation; never returns a partial projection.
325    pub fn project(receipt: &BatchReceipt) -> Result<Self, ReceiptCodecError> {
326        let mut outcomes = Vec::new();
327        outcomes
328            .try_reserve_exact(receipt.outcomes().len())
329            .map_err(ReceiptCodecError::allocation)?;
330        for job in receipt.outcomes() {
331            let outcome = match &job.outcome {
332                BatchOutcome::Committed {
333                    root,
334                    callback_failure,
335                } => WireOutcome::Committed {
336                    root: *root.as_bytes(),
337                    callback_failure: callback_failure
338                        .as_deref()
339                        .map(WireCause::project)
340                        .transpose()?,
341                },
342                BatchOutcome::SequenceRead { sequences: values } => WireOutcome::SequenceRead {
343                    sequences: sequences(values)?,
344                },
345                BatchOutcome::RangeRead {
346                    sequences: values,
347                    hit_limit,
348                    invalid_outside_range,
349                } => WireOutcome::RangeRead {
350                    sequences: sequences(values)?,
351                    hit_limit: *hit_limit,
352                    invalid_outside_range: index(*invalid_outside_range)?,
353                },
354                BatchOutcome::Failed { cause } => WireOutcome::Failed {
355                    cause: WireCause::project(cause)?,
356                },
357                BatchOutcome::Unknown { cause } => WireOutcome::Unknown {
358                    cause: WireCause::project(cause)?,
359                },
360            };
361            outcomes.push(WireJobOutcome {
362                job_index: index(job.job_index)?,
363                shard: index(job.shard)?,
364                outcome,
365            });
366        }
367        Ok(Self {
368            outcomes,
369            submit_error: receipt
370                .submit_error()
371                .map(|cause| WireCause::project(cause))
372                .transpose()?,
373            first_error: receipt.first_error().map(WireCause::project).transpose()?,
374        })
375    }
376}
377
378impl WireCause {
379    /// Retain the exact typed kind and complete diagnostic without accessing a panic payload.
380    ///
381    /// # Errors
382    ///
383    /// Refuses unrepresentable metadata or any failed nested receipt projection.
384    pub fn project(error: &DatabaseError) -> Result<Self, ReceiptCodecError> {
385        let kind = DatabaseErrorKind::project(error);
386        let shard = if let DatabaseError::BatchShardFailed { cause } = error {
387            Some(WireShardCause {
388                kind: ShardErrorKind::project(cause),
389                message: cause.to_string(),
390            })
391        } else {
392            None
393        };
394        let callback = if let DatabaseError::RootCallbackPanicked { cause } = error {
395            let identity = cause.identity();
396            let event = cause.event();
397            Some(WireCallback {
398                job_index: index(identity.job_index)?,
399                shard: index(identity.shard)?,
400                pid: identity.pid,
401                command: identity.command.to_owned(),
402                subscriber_index: index(cause.subscriber_index())?,
403                event: WireAdvance {
404                    shard_id: index(event.shard_id)?,
405                    prior_root: *event.prior_root.as_bytes(),
406                    new_root: *event.new_root.as_bytes(),
407                    advance_gen: event.advance_gen,
408                },
409            })
410        } else {
411            None
412        };
413        let (batch_receipt, first_error) = if let DatabaseError::BatchFailed(failure) = error {
414            (
415                Some(Box::new(WireReceipt::project(failure.receipt())?)),
416                Some(Box::new(Self::project(failure.first_error())?)),
417            )
418        } else {
419            (None, None)
420        };
421        Ok(Self {
422            kind,
423            message: error.to_string(),
424            shard,
425            callback,
426            batch_receipt,
427            first_error,
428        })
429    }
430}
431
432fn read_key_hex<'de, D: serde::Deserializer<'de>>(deserializer: D) -> Result<String, D::Error> {
433    let key = String::deserialize(deserializer)?;
434    if !key.len().is_multiple_of(2) || !key.bytes().all(|byte| byte.is_ascii_hexdigit()) {
435        return Err(serde::de::Error::custom("receipt_read_key_hex_invalid"));
436    }
437    Ok(key)
438}
haematite_cli/serve/receipt_wire.rs

haematite_cli/serve/
receipt_wire.rs
