laminar_db/subscription/
error.rs

1//! Stable failures exposed by the committed cluster-subscription backend.
2
3use laminar_core::checkpoint::{OutputPartitionId, PartitionSequence};
4
5/// Structured terminal or admission error for a cluster subscription.
6#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
7pub enum ClusterSubscriptionError {
8    /// The final operator has no stable recoverable output distribution.
9    #[error("cluster subscription is unsupported for this plan: {reason}")]
10    UnsupportedPlan {
11        /// Planner-owned rejection reason.
12        reason: String,
13    },
14    /// History or a cursor belongs to a different stream incarnation.
15    #[error("subscription stream generation does not match the current catalog object")]
16    GenerationMismatch,
17    /// The requested epoch has not committed.
18    #[error("subscription epoch {requested} is not committed")]
19    EpochNotCommitted {
20        /// Requested checkpoint epoch.
21        requested: u64,
22    },
23    /// The requested epoch is older than retained history.
24    #[error("subscription epoch {requested} is no longer retained")]
25    ReplayPruned {
26        /// Requested checkpoint epoch.
27        requested: u64,
28    },
29    /// Authoritative committed metadata is malformed.
30    #[error("committed subscription manifest is corrupt: {reason}")]
31    ManifestCorrupt {
32        /// Bounded validation detail.
33        reason: String,
34    },
35    /// A segment referenced by committed authority is unavailable.
36    #[error("committed subscription segment is missing for partition {partition:?} at {first:?}")]
37    SegmentMissing {
38        /// Affected output partition.
39        partition: OutputPartitionId,
40        /// First missing sequence.
41        first: PartitionSequence,
42    },
43    /// A referenced segment failed metadata, length, or digest validation.
44    #[error("committed subscription segment is corrupt for partition {partition:?} at {first:?}")]
45    SegmentCorrupt {
46        /// Affected output partition.
47        partition: OutputPartitionId,
48        /// First corrupt sequence.
49        first: PartitionSequence,
50    },
51    /// User schema differs from the certified output schema.
52    #[error("subscription output schema does not match its distribution certificate")]
53    SchemaMismatch,
54    /// A committed partition range is discontinuous.
55    #[error(
56        "subscription partition {partition:?} expected sequence {expected:?}, found {actual:?}"
57    )]
58    PartitionSequenceGap {
59        /// Affected output partition.
60        partition: OutputPartitionId,
61        /// Required next sequence.
62        expected: PartitionSequence,
63        /// Observed sequence.
64        actual: PartitionSequence,
65    },
66    /// The same frame identity names different content.
67    #[error("subscription frame identity has conflicting immutable content")]
68    ConflictingDuplicateSequence,
69    /// Writer authority no longer covers the target vnode.
70    #[error("subscription output writer is stale")]
71    StaleOutputWriter,
72    /// Assignment changed while an operation was being fenced.
73    #[error("subscription output assignment changed")]
74    AssignmentChanged,
75    /// Shared storage or committed-index authority cannot currently be reached.
76    #[error("committed subscription backend is unavailable")]
77    BackendUnavailable,
78    /// Bounded connection buffers could not preserve every frame.
79    #[error("subscription reader exceeded its bounded lag allowance")]
80    SubscriberLagged,
81    /// Token is malformed, tampered with, or bound to another stream.
82    #[error("subscription resume token is invalid")]
83    ResumeTokenInvalid,
84    /// Token is valid but outside its accepted lifetime.
85    #[error("subscription resume token has expired")]
86    ResumeTokenExpired,
87    /// Required history disappeared after it was admitted for replay.
88    #[error("subscription retention no longer covers the required committed range")]
89    RetentionLost,
90    /// Decoded envelope uses an unsupported version.
91    #[error("subscription protocol version {actual} is unsupported")]
92    ProtocolVersion {
93        /// Decoded protocol value.
94        actual: u16,
95    },
96}
97
98impl ClusterSubscriptionError {
99    /// Stable `LaminarDB` error code.
100    #[must_use]
101    pub const fn code(&self) -> &'static str {
102        use laminar_core::error_codes as codes;
103
104        match self {
105            Self::UnsupportedPlan { .. } => codes::SUBSCRIPTION_PLAN_UNSUPPORTED,
106            Self::GenerationMismatch => codes::SUBSCRIPTION_GENERATION_MISMATCH,
107            Self::EpochNotCommitted { .. } => codes::SUBSCRIPTION_EPOCH_NOT_COMMITTED,
108            Self::ReplayPruned { .. } => codes::SUBSCRIPTION_REPLAY_PRUNED,
109            Self::ManifestCorrupt { .. } => codes::SUBSCRIPTION_MANIFEST_CORRUPT,
110            Self::SegmentMissing { .. } => codes::SUBSCRIPTION_SEGMENT_MISSING,
111            Self::SegmentCorrupt { .. } => codes::SUBSCRIPTION_SEGMENT_CORRUPT,
112            Self::SchemaMismatch => codes::SUBSCRIPTION_SCHEMA_MISMATCH,
113            Self::PartitionSequenceGap { .. } => codes::SUBSCRIPTION_SEQUENCE_GAP,
114            Self::ConflictingDuplicateSequence => codes::SUBSCRIPTION_CONFLICTING_DUPLICATE,
115            Self::StaleOutputWriter => codes::SUBSCRIPTION_STALE_WRITER,
116            Self::AssignmentChanged => codes::SUBSCRIPTION_ASSIGNMENT_CHANGED,
117            Self::BackendUnavailable => codes::SUBSCRIPTION_BACKEND_UNAVAILABLE,
118            Self::SubscriberLagged => codes::SUBSCRIPTION_LAGGED,
119            Self::ResumeTokenInvalid => codes::SUBSCRIPTION_RESUME_TOKEN_INVALID,
120            Self::ResumeTokenExpired => codes::SUBSCRIPTION_RESUME_TOKEN_EXPIRED,
121            Self::RetentionLost => codes::SUBSCRIPTION_RETENTION_LOST,
122            Self::ProtocolVersion { .. } => codes::SUBSCRIPTION_PROTOCOL_UNSUPPORTED,
123        }
124    }
125}
126
127#[cfg(test)]
128mod tests {
129    use super::*;
130
131    #[test]
132    fn correctness_failures_have_distinct_stable_codes() {
133        let errors = [
134            ClusterSubscriptionError::GenerationMismatch,
135            ClusterSubscriptionError::ManifestCorrupt {
136                reason: "digest".into(),
137            },
138            ClusterSubscriptionError::ConflictingDuplicateSequence,
139            ClusterSubscriptionError::StaleOutputWriter,
140            ClusterSubscriptionError::SubscriberLagged,
141            ClusterSubscriptionError::ResumeTokenInvalid,
142        ];
143        let codes = errors.map(|error| error.code());
144        assert!(codes.iter().all(|code| code.starts_with("LDB-")));
145        for (index, code) in codes.iter().enumerate() {
146            assert!(!codes[..index].contains(code));
147        }
148    }
149}
laminar_db/subscription/error.rs

laminar_db/subscription/
error.rs
