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("death note error: {message}")]
79 DeathNote {
80 message: String,
82 },
83
84 #[error("namespace error: {message}")]
86 Namespace {
87 message: String,
89 },
90
91 #[error("engine call failed: {source}")]
93 EngineCall {
94 #[from]
96 source: EngineError,
97 },
98
99 #[error("store backend failed: {source}")]
101 StoreBackend {
102 #[from]
104 source: StoreError,
105 },
106
107 #[error("stream failure: {failure}")]
109 Stream {
110 failure: StreamFailure,
112 },
113
114 #[error(
116 "worker dispatch failed for namespace {namespace}, activity type {activity_type}: {reason}"
117 )]
118 WorkerDispatch {
119 namespace: String,
121 activity_type: String,
123 reason: String,
125 },
126
127 #[error("worker connection lost during dispatch on {channel}: {detail}")]
140 WorkerConnectionLost {
141 channel: String,
143 detail: String,
145 },
146
147 #[error("worker connection busy during dispatch on {channel}: {detail}")]
158 WorkerBusy {
159 channel: String,
161 detail: String,
163 },
164
165 #[error(
168 "pending activity collision for workflow {workflow_id}, activity {activity_id}: \
169 a live responder already owns this execution site"
170 )]
171 PendingActivityCollision {
172 workflow_id: WorkflowId,
174 activity_id: ActivityId,
176 },
177
178 #[error(
181 "activity completion rejected for workflow {workflow_id}, activity {activity_id}: {reason}"
182 )]
183 ActivityCompletionRejected {
184 workflow_id: WorkflowId,
186 activity_id: ActivityId,
188 reason: CompletionRejectionReason,
190 },
191
192 #[error("{resource} lock was poisoned")]
194 LockPoisoned {
195 resource: &'static str,
197 },
198
199 #[error("wire error: {wire}")]
201 Wire {
202 wire: WireError,
204 },
205}
206
207#[derive(Clone, Copy, Debug, Eq, Error, PartialEq)]
209pub enum CompletionRejectionReason {
210 #[error("completion token is missing (worker registration era is incompatible)")]
212 MissingCompletionToken,
213 #[error("no execution generation is currently accepting completion")]
215 NoCurrentGeneration,
216 #[error("completion token belongs to a stale execution generation")]
218 StaleGeneration,
219}
220
221#[derive(Clone, Copy, Debug, Eq, Error, PartialEq)]
223pub enum StreamFailure {
224 #[error("consumer lagged behind bounded buffer")]
226 Lagged,
227 #[error("subscriber connection closed")]
229 Closed,
230 #[error("engine event stream closed")]
232 UpstreamClosed,
233}
234
235impl From<WireError> for ServerError {
236 fn from(wire: WireError) -> Self {
237 Self::Wire { wire }
238 }
239}
240
241impl ServerError {
242 #[must_use]
245 pub fn to_wire_error(&self) -> WireError {
246 match self {
247 Self::Config { .. }
248 | Self::UnsafeDataRootAncestor { .. }
249 | Self::TransportBind { .. }
250 | Self::Transport { .. }
251 | Self::SignalListener { .. }
252 | Self::DeathNote { .. }
253 | Self::LockPoisoned { .. } => WireError::backend("server backend failure"),
254 Self::ActivityCompletionRejected { .. } => {
255 WireError::backend("stale activity completion rejected")
256 }
257 Self::PendingActivityCollision { .. } => {
258 WireError::backend("pending activity collision")
259 }
260 Self::WorkerDispatch { .. } => WireError::backend("worker dispatch failed"),
261 Self::WorkerConnectionLost { .. } => {
262 WireError::backend("worker connection lost during dispatch")
263 }
264 Self::WorkerBusy { .. } => WireError::backend("worker connection busy during dispatch"),
265 Self::Namespace { message } => WireError::namespace_denied(message.clone()),
266 Self::EngineCall { source } => wire_from_engine(source),
267 Self::StoreBackend { source } => wire_from_store(source),
268 Self::Stream { failure } => match failure {
269 StreamFailure::Lagged => WireError::lagged("subscriber lagged behind"),
270 StreamFailure::Closed | StreamFailure::UpstreamClosed => {
271 WireError::backend("event stream closed")
272 }
273 },
274 Self::Wire { wire } => wire.clone(),
275 }
276 }
277
278 #[must_use]
280 pub const fn is_config(&self) -> bool {
281 matches!(
282 self,
283 Self::Config { .. } | Self::UnsafeDataRootAncestor { .. }
284 )
285 }
286
287 #[must_use]
289 pub fn namespace_denied(message: impl Into<String>) -> Self {
290 Self::Namespace {
291 message: message.into(),
292 }
293 }
294
295 #[must_use]
303 pub fn placement_admission_denied(
304 namespace: &str,
305 worker_node: Option<&str>,
306 required: &std::collections::BTreeSet<String>,
307 ) -> Self {
308 let node = worker_node.unwrap_or("none");
309 let required = required
310 .iter()
311 .map(String::as_str)
312 .collect::<Vec<_>>()
313 .join(", ");
314 Self::namespace_denied(format!(
315 "worker registration rejected: namespace {namespace} is Pinned to node label(s) \
316 [{required}] but the worker advertises node {node}, which is not in the required set"
317 ))
318 }
319
320 #[must_use]
323 pub fn deploy_denied(message: impl Into<String>) -> Self {
324 Self::Wire {
325 wire: WireError::deploy_denied(message),
326 }
327 }
328
329 #[must_use]
331 pub const fn lagged_stream() -> Self {
332 Self::Stream {
333 failure: StreamFailure::Lagged,
334 }
335 }
336
337 #[must_use]
339 pub fn worker_dispatch(
340 namespace: impl Into<String>,
341 activity_type: impl Into<String>,
342 reason: impl Into<String>,
343 ) -> Self {
344 Self::WorkerDispatch {
345 namespace: namespace.into(),
346 activity_type: activity_type.into(),
347 reason: reason.into(),
348 }
349 }
350
351 #[must_use]
354 pub fn worker_connection_lost(channel: impl Into<String>, detail: impl Into<String>) -> Self {
355 Self::WorkerConnectionLost {
356 channel: channel.into(),
357 detail: detail.into(),
358 }
359 }
360
361 #[must_use]
367 pub const fn is_worker_connection_lost(&self) -> bool {
368 matches!(self, Self::WorkerConnectionLost { .. })
369 }
370
371 #[must_use]
374 pub fn worker_busy(channel: impl Into<String>, detail: impl Into<String>) -> Self {
375 Self::WorkerBusy {
376 channel: channel.into(),
377 detail: detail.into(),
378 }
379 }
380
381 #[must_use]
388 pub const fn is_worker_busy(&self) -> bool {
389 matches!(self, Self::WorkerBusy { .. })
390 }
391
392 #[must_use]
394 pub const fn lock_poisoned(resource: &'static str) -> Self {
395 Self::LockPoisoned { resource }
396 }
397}
398
399#[derive(Clone)]
401pub struct ErrorTraceFields<'a> {
402 pub error_type: Cow<'a, str>,
404 pub store_error_type: Option<&'static str>,
406 pub reason: &'a dyn std::fmt::Display,
408}
409
410impl ServerError {
411 #[must_use]
413 pub fn trace_fields(&self) -> ErrorTraceFields<'_> {
414 match self {
415 Self::Config { message } => ErrorTraceFields {
416 error_type: Cow::Borrowed("Config"),
417 store_error_type: None,
418 reason: message,
419 },
420 Self::UnsafeDataRootAncestor { reason, .. } => ErrorTraceFields {
421 error_type: Cow::Borrowed("UnsafeDataRootAncestor"),
422 store_error_type: None,
423 reason,
424 },
425 Self::TransportBind { message, .. } => ErrorTraceFields {
426 error_type: Cow::Borrowed("TransportBind"),
427 store_error_type: None,
428 reason: message,
429 },
430 Self::Transport { message, .. } => ErrorTraceFields {
431 error_type: Cow::Borrowed("Transport"),
432 store_error_type: None,
433 reason: message,
434 },
435 Self::SignalListener { message, .. } => ErrorTraceFields {
436 error_type: Cow::Borrowed("SignalListener"),
437 store_error_type: None,
438 reason: message,
439 },
440 Self::DeathNote { message } => ErrorTraceFields {
441 error_type: Cow::Borrowed("DeathNote"),
442 store_error_type: None,
443 reason: message,
444 },
445 Self::Namespace { message } => ErrorTraceFields {
446 error_type: Cow::Borrowed("Namespace"),
447 store_error_type: None,
448 reason: message,
449 },
450 Self::EngineCall { source } => engine_trace_fields(source),
451 Self::StoreBackend { source } => store_trace_fields(source),
452 Self::Stream { failure } => ErrorTraceFields {
453 error_type: Cow::Borrowed("Stream"),
454 store_error_type: None,
455 reason: failure,
456 },
457 Self::WorkerDispatch { reason, .. } => ErrorTraceFields {
458 error_type: Cow::Borrowed("WorkerDispatch"),
459 store_error_type: None,
460 reason,
461 },
462 Self::WorkerConnectionLost { detail, .. } => ErrorTraceFields {
463 error_type: Cow::Borrowed("WorkerConnectionLost"),
464 store_error_type: None,
465 reason: detail,
466 },
467 Self::WorkerBusy { detail, .. } => ErrorTraceFields {
468 error_type: Cow::Borrowed("WorkerBusy"),
469 store_error_type: None,
470 reason: detail,
471 },
472 Self::PendingActivityCollision { activity_id, .. } => ErrorTraceFields {
473 error_type: Cow::Borrowed("PendingActivityCollision"),
474 store_error_type: None,
475 reason: activity_id,
476 },
477 Self::ActivityCompletionRejected { reason, .. } => ErrorTraceFields {
478 error_type: Cow::Borrowed("ActivityCompletionRejected"),
479 store_error_type: None,
480 reason,
481 },
482 Self::LockPoisoned { resource } => ErrorTraceFields {
483 error_type: Cow::Borrowed("LockPoisoned"),
484 store_error_type: None,
485 reason: resource,
486 },
487 Self::Wire { wire } => ErrorTraceFields {
488 error_type: wire
489 .error_type
490 .as_deref()
491 .map_or_else(|| Cow::Borrowed(wire.code.as_str()), Cow::Borrowed),
492 store_error_type: None,
493 reason: wire,
494 },
495 }
496 }
497}
498
499fn never_alive_error_type(source: &EngineError) -> &'static str {
511 match source {
512 EngineError::TerminalWriterUnavailable { .. } => "TerminalWriterUnavailable",
513 EngineError::TerminalWriterHeld { .. } => "TerminalWriterHeld",
514 EngineError::RunIsRecoverable { .. } => "RunIsRecoverable",
515 EngineError::NoResidencyVerdict { .. } => "NoResidencyVerdict",
516 _ => "EngineError",
517 }
518}
519
520fn durability_trace_fields<'a>(
526 durability: &'a aion::durability::DurabilityError,
527 source: &'a EngineError,
528) -> ErrorTraceFields<'a> {
529 match durability {
530 aion::durability::DurabilityError::Store(store) => store_trace_fields(store),
531 aion::durability::DurabilityError::NonDeterminism(_)
532 | aion::durability::DurabilityError::HistoryShape { .. }
533 | aion::durability::DurabilityError::SearchAttribute(_) => {
534 simple_engine_fields("Durability", source)
535 }
536 aion::durability::DurabilityError::EngineTaskEpochClosed { .. } => {
542 simple_engine_fields("EngineTaskEpochClosed", source)
543 }
544 }
545}
546
547fn engine_trace_fields(source: &EngineError) -> ErrorTraceFields<'_> {
548 match source {
549 EngineError::WorkflowNotFound { .. } => simple_engine_fields("WorkflowNotFound", source),
550 EngineError::TerminalWriterUnavailable { .. }
552 | EngineError::TerminalWriterHeld { .. }
553 | EngineError::RunIsRecoverable { .. }
554 | EngineError::NoResidencyVerdict { .. } => {
555 simple_engine_fields(never_alive_error_type(source), source)
556 }
557 EngineError::InvalidState { .. } => simple_engine_fields("InvalidState", source),
558 EngineError::ScheduleNotFound { .. } => simple_engine_fields("ScheduleNotFound", source),
559 EngineError::ShuttingDown => simple_engine_fields("ShuttingDown", source),
560 EngineError::EngineTaskEpochClosed { .. } => {
561 simple_engine_fields("EngineTaskEpochClosed", source)
562 }
563 EngineError::Store(store) => store_trace_fields(store),
564 EngineError::Durability(durability) => durability_trace_fields(durability, source),
565 EngineError::MissingStore => simple_engine_fields("MissingStore", source),
566 EngineError::MissingVisibilityStore => {
567 simple_engine_fields("MissingVisibilityStore", source)
568 }
569 EngineError::ConflictingEventPublisher => {
570 simple_engine_fields("ConflictingEventPublisher", source)
571 }
572 EngineError::EventStreaming(_) => simple_engine_fields("EventStreaming", source),
573 EngineError::Load { .. } => simple_engine_fields("Load", source),
574 EngineError::UnenforceableContract { .. } => {
575 simple_engine_fields("UnenforceableContract", source)
576 }
577 EngineError::UnknownVersion { .. } => simple_engine_fields("UnknownVersion", source),
578 EngineError::VersionPinned { .. } => simple_engine_fields("VersionPinned", source),
579 EngineError::RouteActive { .. } => simple_engine_fields("RouteActive", source),
580 EngineError::ManifestMismatch { .. } => simple_engine_fields("ManifestMismatch", source),
581 EngineError::Package(_) => simple_engine_fields("Package", source),
582 EngineError::ContractIdentity { .. } => simple_engine_fields("ContractIdentity", source),
583 EngineError::NoQueueDeclaration { .. } => {
584 simple_engine_fields("NoQueueDeclaration", source)
585 }
586 EngineError::StartInputRefused { .. } => simple_engine_fields("StartInputRefused", source),
587 EngineError::SignalRefused { .. } => simple_engine_fields("SignalRefused", source),
588 EngineError::Schedule { .. } => simple_engine_fields("Schedule", source),
589 EngineError::Runtime { .. } => simple_engine_fields("Runtime", source),
590 EngineError::Gate3BifReplacementMissing { .. } => {
591 simple_engine_fields("Gate3BifReplacementMissing", source)
592 }
593 EngineError::StartupRecoveryNotDeferred => {
594 simple_engine_fields("StartupRecoveryNotDeferred", source)
595 }
596 EngineError::StartupRecoveryAlreadyRan => {
597 simple_engine_fields("StartupRecoveryAlreadyRan", source)
598 }
599 EngineError::StartupRecoverySlotPoisoned => {
600 simple_engine_fields("StartupRecoverySlotPoisoned", source)
601 }
602 EngineError::CleanupExecutorPoisoned => {
603 simple_engine_fields("CleanupExecutorPoisoned", source)
604 }
605 EngineError::CleanupExecutorShutdownTimedOut { .. } => {
606 simple_engine_fields("CleanupExecutorShutdownTimedOut", source)
607 }
608 EngineError::ProcessExitRegistryPoisoned => {
609 simple_engine_fields("ProcessExitRegistryPoisoned", source)
610 }
611 EngineError::ProcessExitOwnershipPoisoned { .. } => {
612 simple_engine_fields("ProcessExitOwnershipPoisoned", source)
613 }
614 EngineError::ProcessExitStatePoisoned { .. }
615 | EngineError::ProcessExitSubscriptionUnavailable
616 | EngineError::ProcessExitDrainerSpawn { .. }
617 | EngineError::ProcessExitDrainerPoisoned
618 | EngineError::ProcessExitOutcomeMissingAfterEvent { .. }
619 | EngineError::ProcessExitEventStreamDisconnected
620 | EngineError::ProcessExitDrainerShutdownTimedOut { .. }
621 | EngineError::ProcessExitDrainerPanicked => process_exit::drainer_trace(source),
622 EngineError::ProcessExitCallbackDispatcherPoisoned
623 | EngineError::ProcessExitCallbackDispatcherUnavailable
624 | EngineError::ProcessExitCallbackDispatcherShutdownTimedOut { .. } => {
625 process_exit::callback_trace(source)
626 }
627 EngineError::ProcessExitAlreadyTerminal { .. } => {
628 simple_engine_fields("ProcessExitAlreadyTerminal", source)
629 }
630 EngineError::ActivityDeliveryPoisoned { .. } => {
631 simple_engine_fields("ActivityDeliveryPoisoned", source)
632 }
633 EngineError::RegistryPoisoned => simple_engine_fields("RegistryPoisoned", source),
634 EngineError::CatalogPoisoned => simple_engine_fields("CatalogPoisoned", source),
635 EngineError::NifRegistration { .. } => simple_engine_fields("NifRegistration", source),
636 EngineError::SignalRouter(_) => simple_engine_fields("SignalRouter", source),
637 EngineError::Query(query) => simple_engine_fields(engine::query_error_type(query), source),
638 }
639}
640
641fn simple_engine_fields<'a>(
642 error_type: &'static str,
643 source: &'a EngineError,
644) -> ErrorTraceFields<'a> {
645 ErrorTraceFields {
646 error_type: Cow::Borrowed(error_type),
647 store_error_type: None,
648 reason: source,
649 }
650}
651
652fn store_trace_fields(source: &StoreError) -> ErrorTraceFields<'_> {
653 ErrorTraceFields {
654 error_type: Cow::Borrowed("StoreError"),
655 store_error_type: Some(engine::store_error_type(source)),
656 reason: source,
657 }
658}
659
660fn wire_from_engine(source: &EngineError) -> WireError {
661 use EngineError as E;
662 use engine::backend_wire as backend;
663
664 match source {
665 EngineError::WorkflowNotFound { .. } => {
666 WireError::not_found_with_type("WorkflowNotFound", source.to_string())
667 }
668 EngineError::InvalidState { reason } => engine::invalid_state_wire(reason),
670 E::TerminalWriterUnavailable { .. }
676 | E::TerminalWriterHeld { .. }
677 | E::RunIsRecoverable { .. }
678 | E::NoResidencyVerdict { .. } => engine::invalid_state_wire(&source.to_string())
679 .with_error_type(never_alive_error_type(source)),
680 EngineError::ScheduleNotFound { .. } => {
681 WireError::not_found_with_type("ScheduleNotFound", source.to_string())
682 }
683 EngineError::ShuttingDown => {
684 WireError::not_running_with_type("ShuttingDown", source.to_string())
685 }
686 E::EngineTaskEpochClosed { .. } => backend("EngineTaskEpochClosed", source),
707 EngineError::Store(store) => wire_from_store(store),
708 EngineError::Durability(durability) => engine::durability_wire(durability, source),
709 E::MissingStore => backend("MissingStore", source),
710 E::MissingVisibilityStore => backend("MissingVisibilityStore", source),
711 E::ConflictingEventPublisher => backend("ConflictingEventPublisher", source),
712 E::EventStreaming(_) => backend("EventStreaming", source),
713 E::Load { .. } => backend("Load", source),
714 EngineError::UnenforceableContract { .. } => {
717 WireError::invalid_input(source.to_string()).with_error_type("UnenforceableContract")
718 }
719 EngineError::UnknownVersion { .. } => {
721 WireError::not_found_with_type("UnknownVersion", source.to_string())
722 }
723 EngineError::VersionPinned { .. } => {
724 WireError::version_pinned(source.to_string()).with_error_type("VersionPinned")
725 }
726 EngineError::RouteActive { .. } => {
727 WireError::version_pinned(source.to_string()).with_error_type("RouteActive")
728 }
729 EngineError::ManifestMismatch { .. } => {
730 WireError::invalid_input(source.to_string()).with_error_type("ManifestMismatch")
731 }
732 E::Package(_) => backend("Package", source),
733 E::ContractIdentity { .. } => engine::contract_refusal_wire("ContractIdentity", source),
734 E::NoQueueDeclaration { .. } => engine::contract_refusal_wire("NoQueueDeclaration", source),
735 E::StartInputRefused { .. } => engine::declared_contract_wire("StartInputRefused", source),
736 E::SignalRefused { .. } => engine::declared_contract_wire("SignalRefused", source),
737 EngineError::Schedule { .. } => backend("Schedule", source),
738 E::Runtime { .. } => backend("Runtime", source),
739 E::Gate3BifReplacementMissing { .. } => backend("Gate3BifReplacementMissing", source),
740 E::StartupRecoveryNotDeferred => backend("StartupRecoveryNotDeferred", source),
743 E::StartupRecoveryAlreadyRan => backend("StartupRecoveryAlreadyRan", source),
744 E::StartupRecoverySlotPoisoned => backend("StartupRecoverySlotPoisoned", source),
745 E::CleanupExecutorPoisoned => backend("CleanupExecutorPoisoned", source),
746 E::CleanupExecutorShutdownTimedOut { .. } => {
747 backend("CleanupExecutorShutdownTimedOut", source)
748 }
749 E::ProcessExitRegistryPoisoned => backend("ProcessExitRegistryPoisoned", source),
750 E::ProcessExitOwnershipPoisoned { .. } => backend("ProcessExitOwnershipPoisoned", source),
751 EngineError::ProcessExitStatePoisoned { .. }
752 | EngineError::ProcessExitSubscriptionUnavailable
753 | EngineError::ProcessExitDrainerSpawn { .. }
754 | EngineError::ProcessExitDrainerPoisoned
755 | EngineError::ProcessExitOutcomeMissingAfterEvent { .. }
756 | EngineError::ProcessExitEventStreamDisconnected
757 | EngineError::ProcessExitDrainerShutdownTimedOut { .. }
758 | EngineError::ProcessExitDrainerPanicked => process_exit::drainer_wire(source),
759 EngineError::ProcessExitCallbackDispatcherPoisoned
760 | EngineError::ProcessExitCallbackDispatcherUnavailable
761 | EngineError::ProcessExitCallbackDispatcherShutdownTimedOut { .. } => {
762 process_exit::callback_wire(source)
763 }
764 E::ProcessExitAlreadyTerminal { .. } => backend("ProcessExitAlreadyTerminal", source),
765 E::ActivityDeliveryPoisoned { .. } => backend("ActivityDeliveryPoisoned", source),
766 E::CatalogPoisoned => backend("CatalogPoisoned", source),
767 E::RegistryPoisoned => backend("RegistryPoisoned", source),
768 E::NifRegistration { .. } => backend("NifRegistration", source),
769 E::SignalRouter(_) => backend("SignalRouter", source),
770 EngineError::Query(query) => engine::query_wire(query, source),
771 }
772}
773
774fn wire_from_store(source: &StoreError) -> WireError {
775 match source {
776 StoreError::SequenceConflict { .. } => WireError::new_with_type(
777 aion_proto::WireErrorCode::SequenceConflict,
778 "SequenceConflict",
779 source.to_string(),
780 ),
781 StoreError::NotFound { .. } => {
782 WireError::not_found_with_type("NotFound", source.to_string())
783 }
784 StoreError::NotOwner { .. } => {
785 WireError::not_owner(source.to_string()).with_error_type("NotOwner")
786 }
787 StoreError::Backend(_) => WireError::backend_with_type("Backend", source.to_string()),
788 StoreError::Serialization(_) => {
789 WireError::backend_with_type("Serialization", source.to_string())
790 }
791 }
792}
793
794#[cfg(test)]
795#[path = "error_tests.rs"]
796mod tests;