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    ReservedBranchRefKey,
190    SequenceConflict,
191    CasMismatch,
192    CasHashMismatch,
193    ConsistencyError,
194    Distribution,
195    LocalCommitFailed,
196    CommittedButOwnerBehind,
197    ElectionLost,
198    ElectionTimeout,
199    Fenced,
200    CasConflict,
201    ClusterMembers,
202    DataDirLocked,
203    LockFileIo,
204    DataDirAlreadyInitialised,
205    LeafLogStore,
206    EphemeralStore,
207    PathWriterRequired,
208    WriteDuringRootAdvanceEmission,
209    DuplicateShardId,
210    ExecutorThreadsInvalid,
211    MissingNodeCacheBudget,
212    InvalidNodeCacheBudget,
213    RetentionRefused,
214    StreamDeleteRefused,
215    RootNotRetained,
216    ReclaimFailed,
217    ExecutorShutdown,
218    MigrationInProgress,
219    InvalidChunkingStamp,
220    MigrationFailed,
221    MigrationRefusedFormat,
222    MigrationRefusedWalTail,
223    MigrationRefusedUncommittedMutations,
224}
225
226impl DatabaseErrorKind {
227    const fn project(error: &DatabaseError) -> Self {
228        match error {
229            DatabaseError::BatchFailed(_) => Self::BatchFailed,
230            DatabaseError::BatchShardFailed { .. } => Self::BatchShardFailed,
231            DatabaseError::RootCallbackPanicked { .. } => Self::RootCallbackPanicked,
232            DatabaseError::BatchJobUnknown { .. } => Self::BatchJobUnknown,
233            DatabaseError::TailRepairRefused { .. } => Self::TailRepairRefused,
234            DatabaseError::DirectoryCreate(_) => Self::DirectoryCreate,
235            DatabaseError::ConfigWrite(_) => Self::ConfigWrite,
236            DatabaseError::ConfigRead(_) => Self::ConfigRead,
237            DatabaseError::ConfigParse(_) => Self::ConfigParse,
238            DatabaseError::InvalidShardCount => Self::InvalidShardCount,
239            DatabaseError::InvalidLongWaitInterval { .. } => Self::InvalidLongWaitInterval,
240            DatabaseError::InvalidQueueByteLimit => Self::InvalidQueueByteLimit,
241            DatabaseError::ShardQueueFull { .. } => Self::ShardQueueFull,
242            DatabaseError::ShardRequestSizeOverflow { .. } => Self::ShardRequestSizeOverflow,
243            DatabaseError::ShardSpawn(_) => Self::ShardSpawn,
244            DatabaseError::SweepSpawn(_) => Self::SweepSpawn,
245            DatabaseError::ShardError(_) => Self::ShardError,
246            DatabaseError::OutcomeUnknown { .. } => Self::OutcomeUnknown,
247            DatabaseError::SweepError(_) => Self::SweepError,
248            DatabaseError::SyncSchedulerSpawn(_) => Self::SyncSchedulerSpawn,
249            DatabaseError::SyncSchedulerError(_) => Self::SyncSchedulerError,
250            DatabaseError::IoError(_) => Self::IoError,
251            DatabaseError::InvalidSweepInterval => Self::InvalidSweepInterval,
252            DatabaseError::MissingSyncTopology => Self::MissingSyncTopology,
253            DatabaseError::InvalidSyncInterval => Self::InvalidSyncInterval,
254            DatabaseError::FormatVersionTooNew { .. } => Self::FormatVersionTooNew,
255            DatabaseError::ReservedSequenceKey => Self::ReservedSequenceKey,
256            DatabaseError::ReservedBranchRefKey => Self::ReservedBranchRefKey,
257            DatabaseError::SequenceConflict { .. } => Self::SequenceConflict,
258            DatabaseError::CasMismatch { .. } => Self::CasMismatch,
259            DatabaseError::CasHashMismatch { .. } => Self::CasHashMismatch,
260            DatabaseError::ConsistencyError(_) => Self::ConsistencyError,
261            DatabaseError::Distribution(_) => Self::Distribution,
262            DatabaseError::LocalCommitFailed(_) => Self::LocalCommitFailed,
263            DatabaseError::CommittedButOwnerBehind { .. } => Self::CommittedButOwnerBehind,
264            DatabaseError::ElectionLost { .. } => Self::ElectionLost,
265            DatabaseError::ElectionTimeout { .. } => Self::ElectionTimeout,
266            DatabaseError::Fenced { .. } => Self::Fenced,
267            DatabaseError::CasConflict { .. } => Self::CasConflict,
268            DatabaseError::ClusterMembers(_) => Self::ClusterMembers,
269            DatabaseError::DataDirLocked { .. } => Self::DataDirLocked,
270            DatabaseError::LockFileIo { .. } => Self::LockFileIo,
271            DatabaseError::DataDirAlreadyInitialised { .. } => Self::DataDirAlreadyInitialised,
272            DatabaseError::LeafLogStore { .. } => Self::LeafLogStore,
273            DatabaseError::EphemeralStore { .. } => Self::EphemeralStore,
274            DatabaseError::PathWriterRequired { .. } => Self::PathWriterRequired,
275            DatabaseError::WriteDuringRootAdvanceEmission => Self::WriteDuringRootAdvanceEmission,
276            DatabaseError::DuplicateShardId { .. } => Self::DuplicateShardId,
277            DatabaseError::ExecutorThreadsInvalid(_) => Self::ExecutorThreadsInvalid,
278            DatabaseError::MissingNodeCacheBudget => Self::MissingNodeCacheBudget,
279            DatabaseError::InvalidNodeCacheBudget(_) => Self::InvalidNodeCacheBudget,
280            DatabaseError::RetentionRefused(_) => Self::RetentionRefused,
281            DatabaseError::StreamDeleteRefused(_) => Self::StreamDeleteRefused,
282            DatabaseError::RootNotRetained { .. } => Self::RootNotRetained,
283            DatabaseError::ReclaimFailed { .. } => Self::ReclaimFailed,
284            DatabaseError::ExecutorShutdown => Self::ExecutorShutdown,
285            DatabaseError::MigrationInProgress { .. } => Self::MigrationInProgress,
286            DatabaseError::InvalidChunkingStamp(_) => Self::InvalidChunkingStamp,
287            DatabaseError::MigrationFailed(_) => Self::MigrationFailed,
288            DatabaseError::MigrationRefusedFormat { .. } => Self::MigrationRefusedFormat,
289            DatabaseError::MigrationRefusedWalTail { .. } => Self::MigrationRefusedWalTail,
290            DatabaseError::MigrationRefusedUncommittedMutations { .. } => {
291                Self::MigrationRefusedUncommittedMutations
292            }
293        }
294    }
295}
296
297fn index(value: usize) -> Result<u64, ReceiptCodecError> {
298    u64::try_from(value).map_err(|_| ReceiptCodecError::IndexOverflow)
299}
300
301fn sequences(values: &[(Vec<u8>, u64)]) -> Result<Vec<WireSequence>, ReceiptCodecError> {
302    let mut result = Vec::new();
303    result
304        .try_reserve_exact(values.len())
305        .map_err(ReceiptCodecError::allocation)?;
306    for (key, sequence) in values {
307        result.push(WireSequence {
308            key_hex: encode_hex(key)?,
309            sequence: *sequence,
310        });
311    }
312    Ok(result)
313}
314
315impl WireReceipt {
316    /// Borrow the original receipt while building its complete portable projection.
317    ///
318    /// # Errors
319    ///
320    /// Refuses an unrepresentable index or failed allocation; never returns a partial projection.
321    pub fn project(receipt: &BatchReceipt) -> Result<Self, ReceiptCodecError> {
322        let mut outcomes = Vec::new();
323        outcomes
324            .try_reserve_exact(receipt.outcomes().len())
325            .map_err(ReceiptCodecError::allocation)?;
326        for job in receipt.outcomes() {
327            let outcome = match &job.outcome {
328                BatchOutcome::Committed {
329                    root,
330                    callback_failure,
331                } => WireOutcome::Committed {
332                    root: *root.as_bytes(),
333                    callback_failure: callback_failure
334                        .as_deref()
335                        .map(WireCause::project)
336                        .transpose()?,
337                },
338                BatchOutcome::SequenceRead { sequences: values } => WireOutcome::SequenceRead {
339                    sequences: sequences(values)?,
340                },
341                BatchOutcome::RangeRead {
342                    sequences: values,
343                    hit_limit,
344                    invalid_outside_range,
345                } => WireOutcome::RangeRead {
346                    sequences: sequences(values)?,
347                    hit_limit: *hit_limit,
348                    invalid_outside_range: index(*invalid_outside_range)?,
349                },
350                BatchOutcome::Failed { cause } => WireOutcome::Failed {
351                    cause: WireCause::project(cause)?,
352                },
353                BatchOutcome::Unknown { cause } => WireOutcome::Unknown {
354                    cause: WireCause::project(cause)?,
355                },
356            };
357            outcomes.push(WireJobOutcome {
358                job_index: index(job.job_index)?,
359                shard: index(job.shard)?,
360                outcome,
361            });
362        }
363        Ok(Self {
364            outcomes,
365            submit_error: receipt
366                .submit_error()
367                .map(|cause| WireCause::project(cause))
368                .transpose()?,
369            first_error: receipt.first_error().map(WireCause::project).transpose()?,
370        })
371    }
372}
373
374impl WireCause {
375    /// Retain the exact typed kind and complete diagnostic without accessing a panic payload.
376    ///
377    /// # Errors
378    ///
379    /// Refuses unrepresentable metadata or any failed nested receipt projection.
380    pub fn project(error: &DatabaseError) -> Result<Self, ReceiptCodecError> {
381        let kind = DatabaseErrorKind::project(error);
382        let shard = if let DatabaseError::BatchShardFailed { cause } = error {
383            Some(WireShardCause {
384                kind: ShardErrorKind::project(cause),
385                message: cause.to_string(),
386            })
387        } else {
388            None
389        };
390        let callback = if let DatabaseError::RootCallbackPanicked { cause } = error {
391            let identity = cause.identity();
392            let event = cause.event();
393            Some(WireCallback {
394                job_index: index(identity.job_index)?,
395                shard: index(identity.shard)?,
396                pid: identity.pid,
397                command: identity.command.to_owned(),
398                subscriber_index: index(cause.subscriber_index())?,
399                event: WireAdvance {
400                    shard_id: index(event.shard_id)?,
401                    prior_root: *event.prior_root.as_bytes(),
402                    new_root: *event.new_root.as_bytes(),
403                    advance_gen: event.advance_gen,
404                },
405            })
406        } else {
407            None
408        };
409        let (batch_receipt, first_error) = if let DatabaseError::BatchFailed(failure) = error {
410            (
411                Some(Box::new(WireReceipt::project(failure.receipt())?)),
412                Some(Box::new(Self::project(failure.first_error())?)),
413            )
414        } else {
415            (None, None)
416        };
417        Ok(Self {
418            kind,
419            message: error.to_string(),
420            shard,
421            callback,
422            batch_receipt,
423            first_error,
424        })
425    }
426}
427
428fn read_key_hex<'de, D: serde::Deserializer<'de>>(deserializer: D) -> Result<String, D::Error> {
429    let key = String::deserialize(deserializer)?;
430    if !key.len().is_multiple_of(2) || !key.bytes().all(|byte| byte.is_ascii_hexdigit()) {
431        return Err(serde::de::Error::custom("receipt_read_key_hex_invalid"));
432    }
433    Ok(key)
434}
haematite_cli/serve/receipt_wire.rs

haematite_cli/serve/
receipt_wire.rs
