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

haematite_cli/serve/
receipt_wire.rs
