ursula_runtime/
error.rs

1use ursula_shard::CoreId;
2use ursula_shard::RaftGroupId;
3use ursula_shard::ShardMapError;
4use ursula_shard::ShardPlacement;
5use ursula_stream::StreamErrorCode;
6use ursula_stream::StreamErrorContext;
7
8use crate::engine::GroupEngineError;
9use crate::engine::GroupLeaderHint;
10
11#[derive(Debug, Clone, Copy, PartialEq, Eq)]
12pub enum ErrorStatus {
13    Permanent,
14    Temporary,
15    // Persistent is reserved for non-retryable service-side failures. HTTP currently
16    // treats it like Permanent; keeping it distinct leaves room for logging/alerting.
17    Persistent,
18}
19
20#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
21pub enum RuntimeError {
22    #[error("invalid shard runtime config: {0}")]
23    InvalidConfig(#[from] ShardMapError),
24    #[error("raft group {} is outside configured range 0..{raft_group_count}", .raft_group_id.0)]
25    InvalidRaftGroup {
26        raft_group_id: RaftGroupId,
27        raft_group_count: u32,
28    },
29    #[error(
30        "snapshot placement for raft group {} is core {}, expected core {}",
31        .actual.raft_group_id.0,
32        .actual.core_id.0,
33        .expected.core_id.0
34    )]
35    SnapshotPlacementMismatch {
36        expected: ShardPlacement,
37        actual: ShardPlacement,
38    },
39    #[error("append payload must be non-empty")]
40    EmptyAppend,
41    #[error("invalid cold store config: {message}")]
42    ColdStoreConfig { message: String },
43    #[error("invalid static Raft membership config: {message}")]
44    StaticMembershipConfig { message: String },
45    #[error("cold store IO error: {message}")]
46    ColdStoreIo { message: String },
47    #[error(
48        "core {} live read waiters at {current_waiters} would exceed limit {limit}",
49        .core_id.0
50    )]
51    LiveReadBackpressure {
52        core_id: CoreId,
53        current_waiters: u64,
54        limit: u64,
55    },
56    #[error("core {} does not host raft group {}", .core_id.0, .raft_group_id.0)]
57    GroupNotHosted {
58        core_id: CoreId,
59        raft_group_id: RaftGroupId,
60    },
61    #[error(
62        "core {} raft group {} operation failed: {}",
63        .core_id.0,
64        .raft_group_id.0,
65        .error.message()
66    )]
67    GroupEngine {
68        core_id: CoreId,
69        raft_group_id: RaftGroupId,
70        error: GroupEngineError,
71    },
72    #[error("core {} mailbox is closed", .core_id.0)]
73    MailboxClosed { core_id: CoreId },
74    #[error("core {} dropped append response", .core_id.0)]
75    ResponseDropped { core_id: CoreId },
76    #[error("failed to spawn core {} thread: {message}", .core_id.0)]
77    SpawnCoreThread { core_id: CoreId, message: String },
78}
79
80impl RuntimeError {
81    pub(crate) fn group_engine(placement: ShardPlacement, err: GroupEngineError) -> Self {
82        Self::GroupEngine {
83            core_id: placement.core_id,
84            raft_group_id: placement.raft_group_id,
85            error: err,
86        }
87    }
88
89    pub fn stream_error_code(&self) -> Option<StreamErrorCode> {
90        match self {
91            Self::GroupEngine { error, .. } => error.code(),
92            _ => None,
93        }
94    }
95
96    pub fn stream_next_offset(&self) -> Option<u64> {
97        match self {
98            Self::GroupEngine { error, .. } => error.next_offset(),
99            _ => None,
100        }
101    }
102
103    pub fn stream_error_context(&self) -> &[StreamErrorContext] {
104        match self {
105            Self::GroupEngine { error, .. } => error.context(),
106            _ => &[],
107        }
108    }
109
110    pub fn leader_hint(&self) -> Option<&GroupLeaderHint> {
111        match self {
112            Self::GroupEngine { error, .. } => error.leader_hint(),
113            _ => None,
114        }
115    }
116
117    pub fn status(&self) -> ErrorStatus {
118        match self {
119            Self::LiveReadBackpressure { .. } | Self::GroupNotHosted { .. } => {
120                ErrorStatus::Temporary
121            }
122            Self::GroupEngine { error, .. } if error.leader_hint().is_some() => {
123                ErrorStatus::Temporary
124            }
125            Self::GroupEngine { error, .. } if error.is_backpressure() => ErrorStatus::Temporary,
126            Self::GroupEngine { error, .. } if error.code().is_some() => ErrorStatus::Permanent,
127            Self::GroupEngine { .. } => ErrorStatus::Persistent,
128            _ => ErrorStatus::Permanent,
129        }
130    }
131}
ursula_runtime/error.rs

ursula_runtime/
error.rs
