1use std::borrow::Cow;
4use std::net::SocketAddr;
5use std::path::PathBuf;
6
7use aion::EngineError;
8use aion_core::{ActivityId, WorkflowId};
9use aion_proto::WireError;
10use aion_store::StoreError;
11use thiserror::Error;
12
13#[path = "error_engine.rs"]
14mod engine;
15#[path = "error_process_exit.rs"]
16mod process_exit;
17
18#[derive(Debug, Error)]
20pub enum ServerError {
21 #[error("configuration error: {message}")]
23 Config {
24 message: String,
26 },
27
28 #[error(
30 "unsafe store.data_dir `{}`: ancestor `{}` is not owner-controlled: {reason}; \
31 move store.data_dir beneath the private Aion home (`$AION_HOME`, default `~/.aion`) \
32 and keep its ancestor chain owner-only",
33 .data_root.display(),
34 .component.display()
35 )]
36 UnsafeDataRootAncestor {
37 data_root: PathBuf,
39 component: PathBuf,
41 reason: String,
43 },
44
45 #[error("{transport} transport failed at {address}: {message}")]
47 TransportBind {
48 transport: &'static str,
50 address: SocketAddr,
52 message: String,
54 },
55
56 #[error("{transport} transport task failed: {message}")]
58 Transport {
59 transport: &'static str,
61 message: String,
63 },
64
65 #[error("{listener} listener failed: {message}")]
67 SignalListener {
68 listener: &'static str,
70 message: String,
72 },
73
74 #[error("namespace error: {message}")]
76 Namespace {
77 message: String,
79 },
80
81 #[error("engine call failed: {source}")]
83 EngineCall {
84 #[from]
86 source: EngineError,
87 },
88
89 #[error("store backend failed: {source}")]
91 StoreBackend {
92 #[from]
94 source: StoreError,
95 },
96
97 #[error("stream failure: {failure}")]
99 Stream {
100 failure: StreamFailure,
102 },
103
104 #[error(
106 "worker dispatch failed for namespace {namespace}, activity type {activity_type}: {reason}"
107 )]
108 WorkerDispatch {
109 namespace: String,
111 activity_type: String,
113 reason: String,
115 },
116
117 #[error("worker connection lost during dispatch on {channel}: {detail}")]
130 WorkerConnectionLost {
131 channel: String,
133 detail: String,
135 },
136
137 #[error("worker connection busy during dispatch on {channel}: {detail}")]
148 WorkerBusy {
149 channel: String,
151 detail: String,
153 },
154
155 #[error(
158 "activity completion rejected for workflow {workflow_id}, activity {activity_id}: {reason}"
159 )]
160 ActivityCompletionRejected {
161 workflow_id: WorkflowId,
163 activity_id: ActivityId,
165 reason: CompletionRejectionReason,
167 },
168
169 #[error("{resource} lock was poisoned")]
171 LockPoisoned {
172 resource: &'static str,
174 },
175
176 #[error("wire error: {wire}")]
178 Wire {
179 wire: WireError,
181 },
182}
183
184#[derive(Clone, Copy, Debug, Eq, Error, PartialEq)]
186pub enum CompletionRejectionReason {
187 #[error("completion token is missing (worker registration era is incompatible)")]
189 MissingCompletionToken,
190 #[error("no execution generation is currently accepting completion")]
192 NoCurrentGeneration,
193 #[error("completion token belongs to a stale execution generation")]
195 StaleGeneration,
196}
197
198#[derive(Clone, Copy, Debug, Eq, Error, PartialEq)]
200pub enum StreamFailure {
201 #[error("consumer lagged behind bounded buffer")]
203 Lagged,
204 #[error("subscriber connection closed")]
206 Closed,
207 #[error("engine event stream closed")]
209 UpstreamClosed,
210}
211
212impl From<WireError> for ServerError {
213 fn from(wire: WireError) -> Self {
214 Self::Wire { wire }
215 }
216}
217
218impl ServerError {
219 #[must_use]
222 pub fn to_wire_error(&self) -> WireError {
223 match self {
224 Self::Config { .. }
225 | Self::UnsafeDataRootAncestor { .. }
226 | Self::TransportBind { .. }
227 | Self::Transport { .. }
228 | Self::SignalListener { .. }
229 | Self::LockPoisoned { .. } => WireError::backend("server backend failure"),
230 Self::ActivityCompletionRejected { .. } => {
231 WireError::backend("stale activity completion rejected")
232 }
233 Self::WorkerDispatch { .. } => WireError::backend("worker dispatch failed"),
234 Self::WorkerConnectionLost { .. } => {
235 WireError::backend("worker connection lost during dispatch")
236 }
237 Self::WorkerBusy { .. } => WireError::backend("worker connection busy during dispatch"),
238 Self::Namespace { message } => WireError::namespace_denied(message.clone()),
239 Self::EngineCall { source } => wire_from_engine(source),
240 Self::StoreBackend { source } => wire_from_store(source),
241 Self::Stream { failure } => match failure {
242 StreamFailure::Lagged => WireError::lagged("subscriber lagged behind"),
243 StreamFailure::Closed | StreamFailure::UpstreamClosed => {
244 WireError::backend("event stream closed")
245 }
246 },
247 Self::Wire { wire } => wire.clone(),
248 }
249 }
250
251 #[must_use]
253 pub const fn is_config(&self) -> bool {
254 matches!(
255 self,
256 Self::Config { .. } | Self::UnsafeDataRootAncestor { .. }
257 )
258 }
259
260 #[must_use]
262 pub fn namespace_denied(message: impl Into<String>) -> Self {
263 Self::Namespace {
264 message: message.into(),
265 }
266 }
267
268 #[must_use]
276 pub fn placement_admission_denied(
277 namespace: &str,
278 worker_node: Option<&str>,
279 required: &std::collections::BTreeSet<String>,
280 ) -> Self {
281 let node = worker_node.unwrap_or("none");
282 let required = required
283 .iter()
284 .map(String::as_str)
285 .collect::<Vec<_>>()
286 .join(", ");
287 Self::namespace_denied(format!(
288 "worker registration rejected: namespace {namespace} is Pinned to node label(s) \
289 [{required}] but the worker advertises node {node}, which is not in the required set"
290 ))
291 }
292
293 #[must_use]
296 pub fn deploy_denied(message: impl Into<String>) -> Self {
297 Self::Wire {
298 wire: WireError::deploy_denied(message),
299 }
300 }
301
302 #[must_use]
304 pub const fn lagged_stream() -> Self {
305 Self::Stream {
306 failure: StreamFailure::Lagged,
307 }
308 }
309
310 #[must_use]
312 pub fn worker_dispatch(
313 namespace: impl Into<String>,
314 activity_type: impl Into<String>,
315 reason: impl Into<String>,
316 ) -> Self {
317 Self::WorkerDispatch {
318 namespace: namespace.into(),
319 activity_type: activity_type.into(),
320 reason: reason.into(),
321 }
322 }
323
324 #[must_use]
327 pub fn worker_connection_lost(channel: impl Into<String>, detail: impl Into<String>) -> Self {
328 Self::WorkerConnectionLost {
329 channel: channel.into(),
330 detail: detail.into(),
331 }
332 }
333
334 #[must_use]
340 pub const fn is_worker_connection_lost(&self) -> bool {
341 matches!(self, Self::WorkerConnectionLost { .. })
342 }
343
344 #[must_use]
347 pub fn worker_busy(channel: impl Into<String>, detail: impl Into<String>) -> Self {
348 Self::WorkerBusy {
349 channel: channel.into(),
350 detail: detail.into(),
351 }
352 }
353
354 #[must_use]
361 pub const fn is_worker_busy(&self) -> bool {
362 matches!(self, Self::WorkerBusy { .. })
363 }
364
365 #[must_use]
367 pub const fn lock_poisoned(resource: &'static str) -> Self {
368 Self::LockPoisoned { resource }
369 }
370}
371
372#[derive(Clone)]
374pub struct ErrorTraceFields<'a> {
375 pub error_type: Cow<'a, str>,
377 pub store_error_type: Option<&'static str>,
379 pub reason: &'a dyn std::fmt::Display,
381}
382
383impl ServerError {
384 #[must_use]
386 pub fn trace_fields(&self) -> ErrorTraceFields<'_> {
387 match self {
388 Self::Config { message } => ErrorTraceFields {
389 error_type: Cow::Borrowed("Config"),
390 store_error_type: None,
391 reason: message,
392 },
393 Self::UnsafeDataRootAncestor { reason, .. } => ErrorTraceFields {
394 error_type: Cow::Borrowed("UnsafeDataRootAncestor"),
395 store_error_type: None,
396 reason,
397 },
398 Self::TransportBind { message, .. } => ErrorTraceFields {
399 error_type: Cow::Borrowed("TransportBind"),
400 store_error_type: None,
401 reason: message,
402 },
403 Self::Transport { message, .. } => ErrorTraceFields {
404 error_type: Cow::Borrowed("Transport"),
405 store_error_type: None,
406 reason: message,
407 },
408 Self::SignalListener { message, .. } => ErrorTraceFields {
409 error_type: Cow::Borrowed("SignalListener"),
410 store_error_type: None,
411 reason: message,
412 },
413 Self::Namespace { message } => ErrorTraceFields {
414 error_type: Cow::Borrowed("Namespace"),
415 store_error_type: None,
416 reason: message,
417 },
418 Self::EngineCall { source } => engine_trace_fields(source),
419 Self::StoreBackend { source } => store_trace_fields(source),
420 Self::Stream { failure } => ErrorTraceFields {
421 error_type: Cow::Borrowed("Stream"),
422 store_error_type: None,
423 reason: failure,
424 },
425 Self::WorkerDispatch { reason, .. } => ErrorTraceFields {
426 error_type: Cow::Borrowed("WorkerDispatch"),
427 store_error_type: None,
428 reason,
429 },
430 Self::WorkerConnectionLost { detail, .. } => ErrorTraceFields {
431 error_type: Cow::Borrowed("WorkerConnectionLost"),
432 store_error_type: None,
433 reason: detail,
434 },
435 Self::WorkerBusy { detail, .. } => ErrorTraceFields {
436 error_type: Cow::Borrowed("WorkerBusy"),
437 store_error_type: None,
438 reason: detail,
439 },
440 Self::ActivityCompletionRejected { reason, .. } => ErrorTraceFields {
441 error_type: Cow::Borrowed("ActivityCompletionRejected"),
442 store_error_type: None,
443 reason,
444 },
445 Self::LockPoisoned { resource } => ErrorTraceFields {
446 error_type: Cow::Borrowed("LockPoisoned"),
447 store_error_type: None,
448 reason: resource,
449 },
450 Self::Wire { wire } => ErrorTraceFields {
451 error_type: wire
452 .error_type
453 .as_deref()
454 .map_or_else(|| Cow::Borrowed(wire.code.as_str()), Cow::Borrowed),
455 store_error_type: None,
456 reason: wire,
457 },
458 }
459 }
460}
461
462fn never_alive_error_type(source: &EngineError) -> &'static str {
474 match source {
475 EngineError::TerminalWriterUnavailable { .. } => "TerminalWriterUnavailable",
476 EngineError::TerminalWriterHeld { .. } => "TerminalWriterHeld",
477 EngineError::RunIsRecoverable { .. } => "RunIsRecoverable",
478 EngineError::NoResidencyVerdict { .. } => "NoResidencyVerdict",
479 _ => "EngineError",
480 }
481}
482
483fn durability_trace_fields<'a>(
489 durability: &'a aion::durability::DurabilityError,
490 source: &'a EngineError,
491) -> ErrorTraceFields<'a> {
492 match durability {
493 aion::durability::DurabilityError::Store(store) => store_trace_fields(store),
494 aion::durability::DurabilityError::NonDeterminism(_)
495 | aion::durability::DurabilityError::HistoryShape { .. }
496 | aion::durability::DurabilityError::SearchAttribute(_) => {
497 simple_engine_fields("Durability", source)
498 }
499 aion::durability::DurabilityError::EngineTaskEpochClosed { .. } => {
505 simple_engine_fields("EngineTaskEpochClosed", source)
506 }
507 }
508}
509
510fn engine_trace_fields(source: &EngineError) -> ErrorTraceFields<'_> {
511 match source {
512 EngineError::WorkflowNotFound { .. } => simple_engine_fields("WorkflowNotFound", source),
513 EngineError::TerminalWriterUnavailable { .. }
515 | EngineError::TerminalWriterHeld { .. }
516 | EngineError::RunIsRecoverable { .. }
517 | EngineError::NoResidencyVerdict { .. } => {
518 simple_engine_fields(never_alive_error_type(source), source)
519 }
520 EngineError::InvalidState { .. } => simple_engine_fields("InvalidState", source),
521 EngineError::ScheduleNotFound { .. } => simple_engine_fields("ScheduleNotFound", source),
522 EngineError::ShuttingDown => simple_engine_fields("ShuttingDown", source),
523 EngineError::EngineTaskEpochClosed { .. } => {
524 simple_engine_fields("EngineTaskEpochClosed", source)
525 }
526 EngineError::Store(store) => store_trace_fields(store),
527 EngineError::Durability(durability) => durability_trace_fields(durability, source),
528 EngineError::MissingStore => simple_engine_fields("MissingStore", source),
529 EngineError::MissingVisibilityStore => {
530 simple_engine_fields("MissingVisibilityStore", source)
531 }
532 EngineError::ConflictingEventPublisher => {
533 simple_engine_fields("ConflictingEventPublisher", source)
534 }
535 EngineError::EventStreaming(_) => simple_engine_fields("EventStreaming", source),
536 EngineError::Load { .. } => simple_engine_fields("Load", source),
537 EngineError::UnenforceableContract { .. } => {
538 simple_engine_fields("UnenforceableContract", source)
539 }
540 EngineError::UnknownVersion { .. } => simple_engine_fields("UnknownVersion", source),
541 EngineError::VersionPinned { .. } => simple_engine_fields("VersionPinned", source),
542 EngineError::RouteActive { .. } => simple_engine_fields("RouteActive", source),
543 EngineError::ManifestMismatch { .. } => simple_engine_fields("ManifestMismatch", source),
544 EngineError::Package(_) => simple_engine_fields("Package", source),
545 EngineError::ContractIdentity { .. } => simple_engine_fields("ContractIdentity", source),
546 EngineError::NoQueueDeclaration { .. } => {
547 simple_engine_fields("NoQueueDeclaration", source)
548 }
549 EngineError::StartInputRefused { .. } => simple_engine_fields("StartInputRefused", source),
550 EngineError::SignalRefused { .. } => simple_engine_fields("SignalRefused", source),
551 EngineError::Schedule { .. } => simple_engine_fields("Schedule", source),
552 EngineError::Runtime { .. } => simple_engine_fields("Runtime", source),
553 EngineError::Gate3BifReplacementMissing { .. } => {
554 simple_engine_fields("Gate3BifReplacementMissing", source)
555 }
556 EngineError::CleanupExecutorPoisoned => {
557 simple_engine_fields("CleanupExecutorPoisoned", source)
558 }
559 EngineError::CleanupExecutorShutdownTimedOut { .. } => {
560 simple_engine_fields("CleanupExecutorShutdownTimedOut", source)
561 }
562 EngineError::ProcessExitRegistryPoisoned => {
563 simple_engine_fields("ProcessExitRegistryPoisoned", source)
564 }
565 EngineError::ProcessExitOwnershipPoisoned { .. } => {
566 simple_engine_fields("ProcessExitOwnershipPoisoned", source)
567 }
568 EngineError::ProcessExitStatePoisoned { .. } => {
569 process_exit::trace("ProcessExitStatePoisoned", source)
570 }
571 EngineError::ProcessExitSubscriptionUnavailable => {
572 process_exit::trace("ProcessExitSubscriptionUnavailable", source)
573 }
574 EngineError::ProcessExitDrainerSpawn { .. } => {
575 process_exit::trace("ProcessExitDrainerSpawn", source)
576 }
577 EngineError::ProcessExitDrainerPoisoned => {
578 process_exit::trace("ProcessExitDrainerPoisoned", source)
579 }
580 EngineError::ProcessExitOutcomeMissingAfterEvent { .. } => {
581 process_exit::trace("ProcessExitOutcomeMissingAfterEvent", source)
582 }
583 EngineError::ProcessExitEventStreamDisconnected => {
584 process_exit::trace("ProcessExitEventStreamDisconnected", source)
585 }
586 EngineError::ProcessExitDrainerShutdownTimedOut { .. } => {
587 process_exit::trace("ProcessExitDrainerShutdownTimedOut", source)
588 }
589 EngineError::ProcessExitDrainerPanicked => {
590 process_exit::trace("ProcessExitDrainerPanicked", source)
591 }
592 EngineError::ProcessExitCallbackDispatcherPoisoned
593 | EngineError::ProcessExitCallbackDispatcherUnavailable
594 | EngineError::ProcessExitCallbackDispatcherShutdownTimedOut { .. } => {
595 process_exit::callback_trace(source)
596 }
597 EngineError::ProcessExitAlreadyTerminal { .. } => {
598 simple_engine_fields("ProcessExitAlreadyTerminal", source)
599 }
600 EngineError::ActivityDeliveryPoisoned { .. } => {
601 simple_engine_fields("ActivityDeliveryPoisoned", source)
602 }
603 EngineError::RegistryPoisoned => simple_engine_fields("RegistryPoisoned", source),
604 EngineError::CatalogPoisoned => simple_engine_fields("CatalogPoisoned", source),
605 EngineError::NifRegistration { .. } => simple_engine_fields("NifRegistration", source),
606 EngineError::SignalRouter(_) => simple_engine_fields("SignalRouter", source),
607 EngineError::Query(query) => simple_engine_fields(engine::query_error_type(query), source),
608 }
609}
610
611fn simple_engine_fields<'a>(
612 error_type: &'static str,
613 source: &'a EngineError,
614) -> ErrorTraceFields<'a> {
615 ErrorTraceFields {
616 error_type: Cow::Borrowed(error_type),
617 store_error_type: None,
618 reason: source,
619 }
620}
621
622fn store_trace_fields(source: &StoreError) -> ErrorTraceFields<'_> {
623 ErrorTraceFields {
624 error_type: Cow::Borrowed("StoreError"),
625 store_error_type: Some(engine::store_error_type(source)),
626 reason: source,
627 }
628}
629
630fn wire_from_engine(source: &EngineError) -> WireError {
631 use EngineError as E;
632 use engine::backend_wire as backend;
633
634 match source {
635 EngineError::WorkflowNotFound { .. } => {
636 WireError::not_found_with_type("WorkflowNotFound", source.to_string())
637 }
638 EngineError::InvalidState { reason } => engine::invalid_state_wire(reason),
640 E::TerminalWriterUnavailable { .. }
646 | E::TerminalWriterHeld { .. }
647 | E::RunIsRecoverable { .. }
648 | E::NoResidencyVerdict { .. } => engine::invalid_state_wire(&source.to_string())
649 .with_error_type(never_alive_error_type(source)),
650 EngineError::ScheduleNotFound { .. } => {
651 WireError::not_found_with_type("ScheduleNotFound", source.to_string())
652 }
653 EngineError::ShuttingDown => {
654 WireError::not_running_with_type("ShuttingDown", source.to_string())
655 }
656 E::EngineTaskEpochClosed { .. } => backend("EngineTaskEpochClosed", source),
677 EngineError::Store(store) => wire_from_store(store),
678 EngineError::Durability(durability) => engine::durability_wire(durability, source),
679 E::MissingStore => backend("MissingStore", source),
680 E::MissingVisibilityStore => backend("MissingVisibilityStore", source),
681 E::ConflictingEventPublisher => backend("ConflictingEventPublisher", source),
682 E::EventStreaming(_) => backend("EventStreaming", source),
683 E::Load { .. } => backend("Load", source),
684 EngineError::UnenforceableContract { .. } => {
687 WireError::invalid_input(source.to_string()).with_error_type("UnenforceableContract")
688 }
689 EngineError::UnknownVersion { .. } => {
691 WireError::not_found_with_type("UnknownVersion", source.to_string())
692 }
693 EngineError::VersionPinned { .. } => {
694 WireError::version_pinned(source.to_string()).with_error_type("VersionPinned")
695 }
696 EngineError::RouteActive { .. } => {
697 WireError::version_pinned(source.to_string()).with_error_type("RouteActive")
698 }
699 EngineError::ManifestMismatch { .. } => {
700 WireError::invalid_input(source.to_string()).with_error_type("ManifestMismatch")
701 }
702 E::Package(_) => backend("Package", source),
703 E::ContractIdentity { .. } => engine::contract_refusal_wire("ContractIdentity", source),
704 E::NoQueueDeclaration { .. } => engine::contract_refusal_wire("NoQueueDeclaration", source),
705 E::StartInputRefused { .. } => engine::declared_contract_wire("StartInputRefused", source),
706 E::SignalRefused { .. } => engine::declared_contract_wire("SignalRefused", source),
707 EngineError::Schedule { .. } => backend("Schedule", source),
708 E::Runtime { .. } => backend("Runtime", source),
709 E::Gate3BifReplacementMissing { .. } => backend("Gate3BifReplacementMissing", source),
710 E::CleanupExecutorPoisoned => backend("CleanupExecutorPoisoned", source),
711 E::CleanupExecutorShutdownTimedOut { .. } => {
712 backend("CleanupExecutorShutdownTimedOut", source)
713 }
714 E::ProcessExitRegistryPoisoned => backend("ProcessExitRegistryPoisoned", source),
715 E::ProcessExitOwnershipPoisoned { .. } => backend("ProcessExitOwnershipPoisoned", source),
716 EngineError::ProcessExitStatePoisoned { .. } => {
717 process_exit::wire("ProcessExitStatePoisoned", source)
718 }
719 EngineError::ProcessExitSubscriptionUnavailable => {
720 process_exit::wire("ProcessExitSubscriptionUnavailable", source)
721 }
722 EngineError::ProcessExitDrainerSpawn { .. } => {
723 process_exit::wire("ProcessExitDrainerSpawn", source)
724 }
725 EngineError::ProcessExitDrainerPoisoned => {
726 process_exit::wire("ProcessExitDrainerPoisoned", source)
727 }
728 EngineError::ProcessExitOutcomeMissingAfterEvent { .. } => {
729 process_exit::wire("ProcessExitOutcomeMissingAfterEvent", source)
730 }
731 EngineError::ProcessExitEventStreamDisconnected => {
732 process_exit::wire("ProcessExitEventStreamDisconnected", source)
733 }
734 EngineError::ProcessExitDrainerShutdownTimedOut { .. } => {
735 process_exit::wire("ProcessExitDrainerShutdownTimedOut", source)
736 }
737 EngineError::ProcessExitDrainerPanicked => {
738 process_exit::wire("ProcessExitDrainerPanicked", source)
739 }
740 EngineError::ProcessExitCallbackDispatcherPoisoned
741 | EngineError::ProcessExitCallbackDispatcherUnavailable
742 | EngineError::ProcessExitCallbackDispatcherShutdownTimedOut { .. } => {
743 process_exit::callback_wire(source)
744 }
745 E::ProcessExitAlreadyTerminal { .. } => backend("ProcessExitAlreadyTerminal", source),
746 E::ActivityDeliveryPoisoned { .. } => backend("ActivityDeliveryPoisoned", source),
747 E::CatalogPoisoned => backend("CatalogPoisoned", source),
748 E::RegistryPoisoned => backend("RegistryPoisoned", source),
749 E::NifRegistration { .. } => backend("NifRegistration", source),
750 E::SignalRouter(_) => backend("SignalRouter", source),
751 EngineError::Query(query) => engine::query_wire(query, source),
752 }
753}
754
755fn wire_from_store(source: &StoreError) -> WireError {
756 match source {
757 StoreError::SequenceConflict { .. } => WireError::new_with_type(
758 aion_proto::WireErrorCode::SequenceConflict,
759 "SequenceConflict",
760 source.to_string(),
761 ),
762 StoreError::NotFound { .. } => {
763 WireError::not_found_with_type("NotFound", source.to_string())
764 }
765 StoreError::NotOwner { .. } => {
766 WireError::not_owner(source.to_string()).with_error_type("NotOwner")
767 }
768 StoreError::Backend(_) => WireError::backend_with_type("Backend", source.to_string()),
769 StoreError::Serialization(_) => {
770 WireError::backend_with_type("Serialization", source.to_string())
771 }
772 }
773}
774
775#[cfg(test)]
776#[path = "error_tests.rs"]
777mod tests;