1use std::borrow::Cow;
4use std::net::SocketAddr;
5
6use aion::EngineError;
7use aion_proto::WireError;
8use aion_store::StoreError;
9use thiserror::Error;
10
11#[derive(Debug, Error)]
13pub enum ServerError {
14 #[error("configuration error: {message}")]
16 Config {
17 message: String,
19 },
20
21 #[error("{transport} transport failed at {address}: {message}")]
23 TransportBind {
24 transport: &'static str,
26 address: SocketAddr,
28 message: String,
30 },
31
32 #[error("{transport} transport task failed: {message}")]
34 Transport {
35 transport: &'static str,
37 message: String,
39 },
40
41 #[error("{listener} listener failed: {message}")]
43 SignalListener {
44 listener: &'static str,
46 message: String,
48 },
49
50 #[error("namespace error: {message}")]
52 Namespace {
53 message: String,
55 },
56
57 #[error("engine call failed: {source}")]
59 EngineCall {
60 #[from]
62 source: EngineError,
63 },
64
65 #[error("store backend failed: {source}")]
67 StoreBackend {
68 #[from]
70 source: StoreError,
71 },
72
73 #[error("stream failure: {failure}")]
75 Stream {
76 failure: StreamFailure,
78 },
79
80 #[error(
82 "worker dispatch failed for namespace {namespace}, activity type {activity_type}: {reason}"
83 )]
84 WorkerDispatch {
85 namespace: String,
87 activity_type: String,
89 reason: String,
91 },
92
93 #[error("worker connection lost during dispatch on {channel}: {detail}")]
106 WorkerConnectionLost {
107 channel: String,
109 detail: String,
111 },
112
113 #[error("{resource} lock was poisoned")]
115 LockPoisoned {
116 resource: &'static str,
118 },
119
120 #[error("wire error: {wire}")]
122 Wire {
123 wire: WireError,
125 },
126}
127
128#[derive(Clone, Copy, Debug, Eq, Error, PartialEq)]
130pub enum StreamFailure {
131 #[error("consumer lagged behind bounded buffer")]
133 Lagged,
134 #[error("subscriber connection closed")]
136 Closed,
137 #[error("engine event stream closed")]
139 UpstreamClosed,
140}
141
142impl From<WireError> for ServerError {
143 fn from(wire: WireError) -> Self {
144 Self::Wire { wire }
145 }
146}
147
148impl ServerError {
149 #[must_use]
152 pub fn to_wire_error(&self) -> WireError {
153 match self {
154 Self::Config { .. }
155 | Self::TransportBind { .. }
156 | Self::Transport { .. }
157 | Self::SignalListener { .. }
158 | Self::LockPoisoned { .. } => WireError::backend("server backend failure"),
159 Self::WorkerDispatch { .. } => WireError::backend("worker dispatch failed"),
160 Self::WorkerConnectionLost { .. } => {
161 WireError::backend("worker connection lost during dispatch")
162 }
163 Self::Namespace { message } => WireError::namespace_denied(message.clone()),
164 Self::EngineCall { source } => wire_from_engine(source),
165 Self::StoreBackend { source } => wire_from_store(source),
166 Self::Stream { failure } => match failure {
167 StreamFailure::Lagged => WireError::lagged("subscriber lagged behind"),
168 StreamFailure::Closed | StreamFailure::UpstreamClosed => {
169 WireError::backend("event stream closed")
170 }
171 },
172 Self::Wire { wire } => wire.clone(),
173 }
174 }
175
176 #[must_use]
178 pub const fn is_config(&self) -> bool {
179 matches!(self, Self::Config { .. })
180 }
181
182 #[must_use]
184 pub fn namespace_denied(message: impl Into<String>) -> Self {
185 Self::Namespace {
186 message: message.into(),
187 }
188 }
189
190 #[must_use]
193 pub fn deploy_denied(message: impl Into<String>) -> Self {
194 Self::Wire {
195 wire: WireError::deploy_denied(message),
196 }
197 }
198
199 #[must_use]
201 pub const fn lagged_stream() -> Self {
202 Self::Stream {
203 failure: StreamFailure::Lagged,
204 }
205 }
206
207 #[must_use]
209 pub fn worker_dispatch(
210 namespace: impl Into<String>,
211 activity_type: impl Into<String>,
212 reason: impl Into<String>,
213 ) -> Self {
214 Self::WorkerDispatch {
215 namespace: namespace.into(),
216 activity_type: activity_type.into(),
217 reason: reason.into(),
218 }
219 }
220
221 #[must_use]
224 pub fn worker_connection_lost(channel: impl Into<String>, detail: impl Into<String>) -> Self {
225 Self::WorkerConnectionLost {
226 channel: channel.into(),
227 detail: detail.into(),
228 }
229 }
230
231 #[must_use]
237 pub const fn is_worker_connection_lost(&self) -> bool {
238 matches!(self, Self::WorkerConnectionLost { .. })
239 }
240
241 #[must_use]
243 pub const fn lock_poisoned(resource: &'static str) -> Self {
244 Self::LockPoisoned { resource }
245 }
246}
247
248#[derive(Clone)]
250pub struct ErrorTraceFields<'a> {
251 pub error_type: Cow<'a, str>,
253 pub store_error_type: Option<&'static str>,
255 pub reason: &'a dyn std::fmt::Display,
257}
258
259impl ServerError {
260 #[must_use]
262 pub fn trace_fields(&self) -> ErrorTraceFields<'_> {
263 match self {
264 Self::Config { message } => ErrorTraceFields {
265 error_type: Cow::Borrowed("Config"),
266 store_error_type: None,
267 reason: message,
268 },
269 Self::TransportBind { message, .. } => ErrorTraceFields {
270 error_type: Cow::Borrowed("TransportBind"),
271 store_error_type: None,
272 reason: message,
273 },
274 Self::Transport { message, .. } => ErrorTraceFields {
275 error_type: Cow::Borrowed("Transport"),
276 store_error_type: None,
277 reason: message,
278 },
279 Self::SignalListener { message, .. } => ErrorTraceFields {
280 error_type: Cow::Borrowed("SignalListener"),
281 store_error_type: None,
282 reason: message,
283 },
284 Self::Namespace { message } => ErrorTraceFields {
285 error_type: Cow::Borrowed("Namespace"),
286 store_error_type: None,
287 reason: message,
288 },
289 Self::EngineCall { source } => engine_trace_fields(source),
290 Self::StoreBackend { source } => store_trace_fields(source),
291 Self::Stream { failure } => ErrorTraceFields {
292 error_type: Cow::Borrowed("Stream"),
293 store_error_type: None,
294 reason: failure,
295 },
296 Self::WorkerDispatch { reason, .. } => ErrorTraceFields {
297 error_type: Cow::Borrowed("WorkerDispatch"),
298 store_error_type: None,
299 reason,
300 },
301 Self::WorkerConnectionLost { detail, .. } => ErrorTraceFields {
302 error_type: Cow::Borrowed("WorkerConnectionLost"),
303 store_error_type: None,
304 reason: detail,
305 },
306 Self::LockPoisoned { resource } => ErrorTraceFields {
307 error_type: Cow::Borrowed("LockPoisoned"),
308 store_error_type: None,
309 reason: resource,
310 },
311 Self::Wire { wire } => ErrorTraceFields {
312 error_type: wire
313 .error_type
314 .as_deref()
315 .map_or_else(|| Cow::Borrowed(wire.code.as_str()), Cow::Borrowed),
316 store_error_type: None,
317 reason: wire,
318 },
319 }
320 }
321}
322
323fn engine_trace_fields(source: &EngineError) -> ErrorTraceFields<'_> {
324 match source {
325 EngineError::WorkflowNotFound { .. } => simple_engine_fields("WorkflowNotFound", source),
326 EngineError::ScheduleNotFound { .. } => simple_engine_fields("ScheduleNotFound", source),
327 EngineError::ShuttingDown => simple_engine_fields("ShuttingDown", source),
328 EngineError::Store(store) => store_trace_fields(store),
329 EngineError::Durability(durability) => match durability {
330 aion::durability::DurabilityError::Store(store) => store_trace_fields(store),
331 aion::durability::DurabilityError::NonDeterminism(_)
332 | aion::durability::DurabilityError::HistoryShape { .. }
333 | aion::durability::DurabilityError::SearchAttribute(_) => {
334 simple_engine_fields("Durability", source)
335 }
336 },
337 EngineError::MissingStore => simple_engine_fields("MissingStore", source),
338 EngineError::MissingVisibilityStore => {
339 simple_engine_fields("MissingVisibilityStore", source)
340 }
341 EngineError::ConflictingEventPublisher => {
342 simple_engine_fields("ConflictingEventPublisher", source)
343 }
344 EngineError::EventStreaming(_) => simple_engine_fields("EventStreaming", source),
345 EngineError::Load { .. } => simple_engine_fields("Load", source),
346 EngineError::UnknownVersion { .. } => simple_engine_fields("UnknownVersion", source),
347 EngineError::VersionPinned { .. } => simple_engine_fields("VersionPinned", source),
348 EngineError::RouteActive { .. } => simple_engine_fields("RouteActive", source),
349 EngineError::ManifestMismatch { .. } => simple_engine_fields("ManifestMismatch", source),
350 EngineError::Package(_) => simple_engine_fields("Package", source),
351 EngineError::Schedule { .. } => simple_engine_fields("Schedule", source),
352 EngineError::Runtime { .. } => simple_engine_fields("Runtime", source),
353 EngineError::RegistryPoisoned => simple_engine_fields("RegistryPoisoned", source),
354 EngineError::CatalogPoisoned => simple_engine_fields("CatalogPoisoned", source),
355 EngineError::NifRegistration { .. } => simple_engine_fields("NifRegistration", source),
356 EngineError::SignalRouter(_) => simple_engine_fields("SignalRouter", source),
357 EngineError::Query(query) => simple_engine_fields(query_error_type(query), source),
358 }
359}
360
361fn query_error_type(source: &aion::QueryError) -> &'static str {
363 match source {
364 aion::QueryError::UnknownQuery(_) => "UnknownQuery",
365 aion::QueryError::Timeout => "QueryTimeout",
366 aion::QueryError::NotRunning(_) => "QueryNotRunning",
367 aion::QueryError::Unknown(_) => "QueryUnknownWorkflow",
368 aion::QueryError::ReplyDropped => "QueryReplyDropped",
369 aion::QueryError::HandlerFailed { .. } => "QueryFailed",
370 aion::QueryError::Engine(_) => "QueryEngine",
371 }
372}
373
374fn simple_engine_fields<'a>(
375 error_type: &'static str,
376 source: &'a EngineError,
377) -> ErrorTraceFields<'a> {
378 ErrorTraceFields {
379 error_type: Cow::Borrowed(error_type),
380 store_error_type: None,
381 reason: source,
382 }
383}
384
385fn store_trace_fields(source: &StoreError) -> ErrorTraceFields<'_> {
386 ErrorTraceFields {
387 error_type: Cow::Borrowed("StoreError"),
388 store_error_type: Some(store_error_type(source)),
389 reason: source,
390 }
391}
392
393fn store_error_type(source: &StoreError) -> &'static str {
394 match source {
395 StoreError::SequenceConflict { .. } => "SequenceConflict",
396 StoreError::NotFound { .. } => "NotFound",
397 StoreError::NotOwner { .. } => "NotOwner",
398 StoreError::Backend(_) => "Backend",
399 StoreError::Serialization(_) => "Serialization",
400 }
401}
402
403fn wire_from_engine(source: &EngineError) -> WireError {
404 match source {
405 EngineError::WorkflowNotFound { .. } => {
406 WireError::not_found_with_type("WorkflowNotFound", source.to_string())
407 }
408 EngineError::ScheduleNotFound { .. } => {
409 WireError::not_found_with_type("ScheduleNotFound", source.to_string())
410 }
411 EngineError::ShuttingDown => {
412 WireError::not_running_with_type("ShuttingDown", source.to_string())
413 }
414 EngineError::Store(store) => wire_from_store(store),
415 EngineError::Durability(durability) => match durability {
416 aion::durability::DurabilityError::Store(store) => wire_from_store(store),
417 aion::durability::DurabilityError::NonDeterminism(_)
418 | aion::durability::DurabilityError::HistoryShape { .. }
419 | aion::durability::DurabilityError::SearchAttribute(_) => {
420 WireError::backend_with_type("Durability", source.to_string())
421 }
422 },
423 EngineError::MissingStore => {
424 WireError::backend_with_type("MissingStore", source.to_string())
425 }
426 EngineError::MissingVisibilityStore => {
427 WireError::backend_with_type("MissingVisibilityStore", source.to_string())
428 }
429 EngineError::ConflictingEventPublisher => {
430 WireError::backend_with_type("ConflictingEventPublisher", source.to_string())
431 }
432 EngineError::EventStreaming(_) => {
433 WireError::backend_with_type("EventStreaming", source.to_string())
434 }
435 EngineError::Load { .. } => WireError::backend_with_type("Load", source.to_string()),
436 EngineError::UnknownVersion { .. } => {
441 WireError::not_found_with_type("UnknownVersion", source.to_string())
442 }
443 EngineError::VersionPinned { .. } => {
444 WireError::version_pinned(source.to_string()).with_error_type("VersionPinned")
445 }
446 EngineError::RouteActive { .. } => {
447 WireError::version_pinned(source.to_string()).with_error_type("RouteActive")
448 }
449 EngineError::ManifestMismatch { .. } => {
450 WireError::invalid_input(source.to_string()).with_error_type("ManifestMismatch")
451 }
452 EngineError::Package(_) => WireError::backend_with_type("Package", source.to_string()),
453 EngineError::Schedule { .. } => {
454 WireError::backend_with_type("Schedule", source.to_string())
455 }
456 EngineError::Runtime { .. } => WireError::backend_with_type("Runtime", source.to_string()),
457 EngineError::CatalogPoisoned => {
458 WireError::backend_with_type("CatalogPoisoned", source.to_string())
459 }
460 EngineError::RegistryPoisoned => {
461 WireError::backend_with_type("RegistryPoisoned", source.to_string())
462 }
463 EngineError::NifRegistration { .. } => {
464 WireError::backend_with_type("NifRegistration", source.to_string())
465 }
466 EngineError::SignalRouter(_) => {
467 WireError::backend_with_type("SignalRouter", source.to_string())
468 }
469 EngineError::Query(query) => wire_from_query(query, source),
470 }
471}
472
473fn wire_from_query(query: &aion::QueryError, source: &EngineError) -> WireError {
479 match query {
480 aion::QueryError::UnknownQuery(_) => WireError::unknown_query(source.to_string()),
481 aion::QueryError::Timeout => WireError::query_timeout(source.to_string()),
482 aion::QueryError::NotRunning(_) | aion::QueryError::ReplyDropped => {
483 WireError::not_running_with_type(query_error_type(query), source.to_string())
484 }
485 aion::QueryError::Unknown(_) => {
486 WireError::not_found_with_type(query_error_type(query), source.to_string())
487 }
488 aion::QueryError::HandlerFailed { .. } => {
489 WireError::query_failed(source.to_string()).with_error_type(query_error_type(query))
490 }
491 aion::QueryError::Engine(_) => {
492 WireError::backend_with_type(query_error_type(query), source.to_string())
493 }
494 }
495}
496
497fn wire_from_store(source: &StoreError) -> WireError {
498 match source {
499 StoreError::SequenceConflict { .. } => WireError::new_with_type(
500 aion_proto::WireErrorCode::SequenceConflict,
501 "SequenceConflict",
502 source.to_string(),
503 ),
504 StoreError::NotFound { .. } => {
505 WireError::not_found_with_type("NotFound", source.to_string())
506 }
507 StoreError::NotOwner { .. } => {
508 WireError::not_owner(source.to_string()).with_error_type("NotOwner")
509 }
510 StoreError::Backend(_) => WireError::backend_with_type("Backend", source.to_string()),
511 StoreError::Serialization(_) => {
512 WireError::backend_with_type("Serialization", source.to_string())
513 }
514 }
515}
516
517#[cfg(test)]
518mod tests {
519 use super::{ServerError, StreamFailure};
520 use aion::{EngineError, QueryError, engine_seam::EngineSeamError};
521 use aion_core::WorkflowId;
522 use aion_proto::WireErrorCode;
523
524 fn assert_send_sync<T: Send + Sync>() {}
525
526 #[test]
527 fn server_error_is_send_sync() {
528 assert_send_sync::<ServerError>();
529 }
530
531 #[test]
532 fn lagged_stream_maps_to_wire_lagged() {
533 let error = ServerError::Stream {
534 failure: StreamFailure::Lagged,
535 };
536
537 assert_eq!(error.to_wire_error().code, WireErrorCode::Lagged);
538 }
539
540 #[test]
544 fn not_owner_store_error_maps_to_wire_not_owner() {
545 let error = ServerError::StoreBackend {
546 source: aion_store::StoreError::NotOwner { shard: 3 },
547 };
548 let wire = error.to_wire_error();
549 assert_eq!(wire.code, WireErrorCode::NotOwner);
550 assert_eq!(wire.error_type.as_deref(), Some("NotOwner"));
551 }
552
553 fn workflow_id() -> WorkflowId {
554 WorkflowId::new(uuid::Uuid::from_u128(7))
555 }
556
557 fn query_wire(query: QueryError) -> aion_proto::WireError {
558 ServerError::EngineCall {
559 source: EngineError::Query(query),
560 }
561 .to_wire_error()
562 }
563
564 #[test]
568 fn every_query_error_arm_maps_to_its_pinned_wire_code() {
569 let arms: Vec<(QueryError, WireErrorCode, Option<&str>)> = vec![
570 (
571 QueryError::UnknownQuery(String::from("state")),
572 WireErrorCode::UnknownQuery,
573 None,
574 ),
575 (QueryError::Timeout, WireErrorCode::QueryTimeout, None),
576 (
577 QueryError::NotRunning(workflow_id()),
578 WireErrorCode::NotRunning,
579 Some("QueryNotRunning"),
580 ),
581 (
582 QueryError::Unknown(workflow_id()),
583 WireErrorCode::NotFound,
584 Some("QueryUnknownWorkflow"),
585 ),
586 (
588 QueryError::ReplyDropped,
589 WireErrorCode::NotRunning,
590 Some("QueryReplyDropped"),
591 ),
592 (
594 QueryError::HandlerFailed {
595 message: String::from("handler raised"),
596 },
597 WireErrorCode::QueryFailed,
598 Some("QueryFailed"),
599 ),
600 (
601 QueryError::Engine(EngineSeamError::Delivery {
602 reason: String::from("mailbox closed"),
603 }),
604 WireErrorCode::Backend,
605 Some("QueryEngine"),
606 ),
607 ];
608
609 let variant_count = arms
614 .iter()
615 .map(|(query, _, _)| match query {
616 QueryError::UnknownQuery(_) => 0,
617 QueryError::Timeout => 1,
618 QueryError::NotRunning(_) => 2,
619 QueryError::Unknown(_) => 3,
620 QueryError::ReplyDropped => 4,
621 QueryError::HandlerFailed { .. } => 5,
622 QueryError::Engine(_) => 6,
623 })
624 .collect::<std::collections::BTreeSet<usize>>()
625 .len();
626 assert_eq!(
627 arms.len(),
628 variant_count,
629 "every QueryError variant must appear exactly once in the pin list",
630 );
631 assert_eq!(variant_count, 7, "pin list must cover all 7 variants");
632
633 for (query, expected_code, expected_type) in arms {
634 let wire = query_wire(query.clone());
635 assert_eq!(
636 wire.code, expected_code,
637 "{query:?} must map to {expected_code:?}",
638 );
639 assert_eq!(
640 wire.error_type.as_deref(),
641 expected_type,
642 "{query:?} must carry error_type {expected_type:?}",
643 );
644 }
645 }
646
647 #[test]
650 fn handler_failed_trace_fields_use_query_failed_type() {
651 let error = ServerError::EngineCall {
652 source: EngineError::Query(QueryError::HandlerFailed {
653 message: String::from("handler raised"),
654 }),
655 };
656
657 assert_eq!(error.trace_fields().error_type, "QueryFailed");
658 }
659}