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)]
21pub enum RuntimeError {
22    InvalidConfig(ShardMapError),
23    InvalidRaftGroup {
24        raft_group_id: RaftGroupId,
25        raft_group_count: u32,
26    },
27    SnapshotPlacementMismatch {
28        expected: ShardPlacement,
29        actual: ShardPlacement,
30    },
31    EmptyAppend,
32    ColdStoreConfig {
33        message: String,
34    },
35    StaticMembershipConfig {
36        message: String,
37    },
38    ColdStoreIo {
39        message: String,
40    },
41    LiveReadBackpressure {
42        core_id: CoreId,
43        current_waiters: u64,
44        limit: u64,
45    },
46    GroupNotHosted {
47        core_id: CoreId,
48        raft_group_id: RaftGroupId,
49    },
50    GroupEngine {
51        core_id: CoreId,
52        raft_group_id: RaftGroupId,
53        error: GroupEngineError,
54    },
55    MailboxClosed {
56        core_id: CoreId,
57    },
58    ResponseDropped {
59        core_id: CoreId,
60    },
61    SpawnCoreThread {
62        core_id: CoreId,
63        message: String,
64    },
65}
66
67impl RuntimeError {
68    pub(crate) fn group_engine(placement: ShardPlacement, err: GroupEngineError) -> Self {
69        Self::GroupEngine {
70            core_id: placement.core_id,
71            raft_group_id: placement.raft_group_id,
72            error: err,
73        }
74    }
75
76    pub fn stream_error_code(&self) -> Option<StreamErrorCode> {
77        match self {
78            Self::GroupEngine { error, .. } => error.code(),
79            _ => None,
80        }
81    }
82
83    pub fn stream_next_offset(&self) -> Option<u64> {
84        match self {
85            Self::GroupEngine { error, .. } => error.next_offset(),
86            _ => None,
87        }
88    }
89
90    pub fn stream_error_context(&self) -> &[StreamErrorContext] {
91        match self {
92            Self::GroupEngine { error, .. } => error.context(),
93            _ => &[],
94        }
95    }
96
97    pub fn leader_hint(&self) -> Option<&GroupLeaderHint> {
98        match self {
99            Self::GroupEngine { error, .. } => error.leader_hint(),
100            _ => None,
101        }
102    }
103
104    pub fn status(&self) -> ErrorStatus {
105        match self {
106            Self::LiveReadBackpressure { .. } | Self::GroupNotHosted { .. } => {
107                ErrorStatus::Temporary
108            }
109            Self::GroupEngine { error, .. } if error.leader_hint().is_some() => {
110                ErrorStatus::Temporary
111            }
112            Self::GroupEngine { error, .. } if error.is_backpressure() => ErrorStatus::Temporary,
113            Self::GroupEngine { error, .. } if error.code().is_some() => ErrorStatus::Permanent,
114            Self::GroupEngine { .. } => ErrorStatus::Persistent,
115            _ => ErrorStatus::Permanent,
116        }
117    }
118}
119
120impl std::fmt::Display for RuntimeError {
121    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
122        match self {
123            Self::InvalidConfig(err) => write!(f, "invalid shard runtime config: {err}"),
124            Self::InvalidRaftGroup {
125                raft_group_id,
126                raft_group_count,
127            } => write!(
128                f,
129                "raft group {} is outside configured range 0..{}",
130                raft_group_id.0, raft_group_count
131            ),
132            Self::SnapshotPlacementMismatch { expected, actual } => write!(
133                f,
134                "snapshot placement for raft group {} is core {}, expected core {}",
135                actual.raft_group_id.0, actual.core_id.0, expected.core_id.0
136            ),
137            Self::EmptyAppend => f.write_str("append payload must be non-empty"),
138            Self::ColdStoreConfig { message } => {
139                write!(f, "invalid cold store config: {message}")
140            }
141            Self::StaticMembershipConfig { message } => {
142                write!(f, "invalid static Raft membership config: {message}")
143            }
144            Self::ColdStoreIo { message } => write!(f, "cold store IO error: {message}"),
145            Self::LiveReadBackpressure {
146                core_id,
147                current_waiters,
148                limit,
149            } => write!(
150                f,
151                "core {} live read waiters at {} would exceed limit {}",
152                core_id.0, current_waiters, limit
153            ),
154            Self::GroupNotHosted {
155                core_id,
156                raft_group_id,
157            } => write!(
158                f,
159                "core {} does not host raft group {}",
160                core_id.0, raft_group_id.0
161            ),
162            Self::GroupEngine {
163                core_id,
164                raft_group_id,
165                error,
166                ..
167            } => write!(
168                f,
169                "core {} raft group {} operation failed: {}",
170                core_id.0,
171                raft_group_id.0,
172                error.message()
173            ),
174            Self::MailboxClosed { core_id } => {
175                write!(f, "core {} mailbox is closed", core_id.0)
176            }
177            Self::ResponseDropped { core_id } => {
178                write!(f, "core {} dropped append response", core_id.0)
179            }
180            Self::SpawnCoreThread { core_id, message } => {
181                write!(f, "failed to spawn core {} thread: {message}", core_id.0)
182            }
183        }
184    }
185}
186
187impl std::error::Error for RuntimeError {}
188
189impl From<ShardMapError> for RuntimeError {
190    fn from(value: ShardMapError) -> Self {
191        Self::InvalidConfig(value)
192    }
193}
194
195pub(crate) fn map_fork_source_ref_error(
196    err: RuntimeError,
197    placement: ShardPlacement,
198) -> RuntimeError {
199    if err.stream_error_code() == Some(StreamErrorCode::StreamGone) {
200        return RuntimeError::group_engine(
201            placement,
202            GroupEngineError::stream(
203                StreamErrorCode::StreamAlreadyExistsConflict,
204                "source stream is gone and cannot be forked",
205            ),
206        );
207    }
208    err
209}
ursula_runtime/error.rs

ursula_runtime/
error.rs
