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