use std::borrow::Cow;
use std::net::SocketAddr;
use std::path::PathBuf;
use aion::EngineError;
use aion_core::{ActivityId, WorkflowId};
use aion_proto::WireError;
use aion_store::StoreError;
use thiserror::Error;
#[path = "error_engine.rs"]
mod engine;
#[path = "error_process_exit.rs"]
mod process_exit;
#[derive(Debug, Error)]
pub enum ServerError {
#[error("configuration error: {message}")]
Config {
message: String,
},
#[error(
"unsafe store.data_dir `{}`: ancestor `{}` is not owner-controlled: {reason}; \
leave store.data_dir unset so it defaults beneath the private Aion home \
(`$AION_HOME`, default `$HOME/.aion`), or set a path whose ancestor chain is \
owner-only (a leading `~` expands against $HOME; a relative path resolves \
against the server's working directory)",
.data_root.display(),
.component.display()
)]
UnsafeDataRootAncestor {
data_root: PathBuf,
component: PathBuf,
reason: String,
},
#[error("{transport} transport failed at {address}: {message}")]
TransportBind {
transport: &'static str,
address: SocketAddr,
message: String,
},
#[error("{transport} transport task failed: {message}")]
Transport {
transport: &'static str,
message: String,
},
#[error("{listener} listener failed: {message}")]
SignalListener {
listener: &'static str,
message: String,
},
#[error("death note error: {message}")]
DeathNote {
message: String,
},
#[error("namespace error: {message}")]
Namespace {
message: String,
},
#[error("engine call failed: {source}")]
EngineCall {
#[from]
source: EngineError,
},
#[error("store backend failed: {source}")]
StoreBackend {
#[from]
source: StoreError,
},
#[error("stream failure: {failure}")]
Stream {
failure: StreamFailure,
},
#[error(
"worker dispatch failed for namespace {namespace}, activity type {activity_type}: {reason}"
)]
WorkerDispatch {
namespace: String,
activity_type: String,
reason: String,
},
#[error("worker connection lost during dispatch on {channel}: {detail}")]
WorkerConnectionLost {
channel: String,
detail: String,
},
#[error("worker connection busy during dispatch on {channel}: {detail}")]
WorkerBusy {
channel: String,
detail: String,
},
#[error(
"pending activity collision for workflow {workflow_id}, activity {activity_id}: \
a live responder already owns this execution site"
)]
PendingActivityCollision {
workflow_id: WorkflowId,
activity_id: ActivityId,
},
#[error(
"activity completion rejected for workflow {workflow_id}, activity {activity_id}: {reason}"
)]
ActivityCompletionRejected {
workflow_id: WorkflowId,
activity_id: ActivityId,
reason: CompletionRejectionReason,
},
#[error(
"declared command for workflow {workflow_id}, activity {activity_id} attempt {attempt} \
is already executing at this server"
)]
DeclaredAttemptCollision {
workflow_id: WorkflowId,
activity_id: ActivityId,
attempt: u32,
},
#[error("{resource} lock was poisoned")]
LockPoisoned {
resource: &'static str,
},
#[error("wire error: {wire}")]
Wire {
wire: WireError,
},
}
#[derive(Clone, Copy, Debug, Eq, Error, PartialEq)]
pub enum CompletionRejectionReason {
#[error("completion token is missing (worker registration era is incompatible)")]
MissingCompletionToken,
#[error("no execution generation is currently accepting completion")]
NoCurrentGeneration,
#[error("completion token belongs to a stale execution generation")]
StaleGeneration,
}
#[derive(Clone, Copy, Debug, Eq, Error, PartialEq)]
pub enum StreamFailure {
#[error("consumer lagged behind bounded buffer")]
Lagged,
#[error("subscriber connection closed")]
Closed,
#[error("engine event stream closed")]
UpstreamClosed,
}
impl From<WireError> for ServerError {
fn from(wire: WireError) -> Self {
Self::Wire { wire }
}
}
impl ServerError {
#[must_use]
pub fn to_wire_error(&self) -> WireError {
match self {
Self::Config { .. }
| Self::UnsafeDataRootAncestor { .. }
| Self::TransportBind { .. }
| Self::Transport { .. }
| Self::SignalListener { .. }
| Self::DeathNote { .. }
| Self::LockPoisoned { .. } => WireError::backend("server backend failure"),
Self::ActivityCompletionRejected { .. } => {
WireError::backend("stale activity completion rejected")
}
Self::PendingActivityCollision { .. } => {
WireError::backend("pending activity collision")
}
Self::DeclaredAttemptCollision { .. } => {
WireError::backend("declared command attempt collision")
}
Self::WorkerDispatch { .. } => WireError::backend("worker dispatch failed"),
Self::WorkerConnectionLost { .. } => {
WireError::backend("worker connection lost during dispatch")
}
Self::WorkerBusy { .. } => WireError::backend("worker connection busy during dispatch"),
Self::Namespace { message } => WireError::namespace_denied(message.clone()),
Self::EngineCall { source } => wire_from_engine(source),
Self::StoreBackend { source } => wire_from_store(source),
Self::Stream { failure } => match failure {
StreamFailure::Lagged => WireError::lagged("subscriber lagged behind"),
StreamFailure::Closed | StreamFailure::UpstreamClosed => {
WireError::backend("event stream closed")
}
},
Self::Wire { wire } => wire.clone(),
}
}
#[must_use]
pub const fn is_config(&self) -> bool {
matches!(
self,
Self::Config { .. } | Self::UnsafeDataRootAncestor { .. }
)
}
#[must_use]
pub fn namespace_denied(message: impl Into<String>) -> Self {
Self::Namespace {
message: message.into(),
}
}
#[must_use]
pub fn placement_admission_denied(
namespace: &str,
worker_node: Option<&str>,
required: &std::collections::BTreeSet<String>,
) -> Self {
let node = worker_node.unwrap_or("none");
let required = required
.iter()
.map(String::as_str)
.collect::<Vec<_>>()
.join(", ");
Self::namespace_denied(format!(
"worker registration rejected: namespace {namespace} is Pinned to node label(s) \
[{required}] but the worker advertises node {node}, which is not in the required set"
))
}
#[must_use]
pub fn deploy_denied(message: impl Into<String>) -> Self {
Self::Wire {
wire: WireError::deploy_denied(message),
}
}
#[must_use]
pub const fn lagged_stream() -> Self {
Self::Stream {
failure: StreamFailure::Lagged,
}
}
#[must_use]
pub fn worker_dispatch(
namespace: impl Into<String>,
activity_type: impl Into<String>,
reason: impl Into<String>,
) -> Self {
Self::WorkerDispatch {
namespace: namespace.into(),
activity_type: activity_type.into(),
reason: reason.into(),
}
}
#[must_use]
pub fn worker_connection_lost(channel: impl Into<String>, detail: impl Into<String>) -> Self {
Self::WorkerConnectionLost {
channel: channel.into(),
detail: detail.into(),
}
}
#[must_use]
pub const fn is_worker_connection_lost(&self) -> bool {
matches!(self, Self::WorkerConnectionLost { .. })
}
#[must_use]
pub fn worker_busy(channel: impl Into<String>, detail: impl Into<String>) -> Self {
Self::WorkerBusy {
channel: channel.into(),
detail: detail.into(),
}
}
#[must_use]
pub const fn is_worker_busy(&self) -> bool {
matches!(self, Self::WorkerBusy { .. })
}
#[must_use]
pub const fn lock_poisoned(resource: &'static str) -> Self {
Self::LockPoisoned { resource }
}
}
#[derive(Clone)]
pub struct ErrorTraceFields<'a> {
pub error_type: Cow<'a, str>,
pub store_error_type: Option<&'static str>,
pub reason: &'a dyn std::fmt::Display,
}
impl ServerError {
#[must_use]
pub fn trace_fields(&self) -> ErrorTraceFields<'_> {
match self {
Self::Config { message } => ErrorTraceFields {
error_type: Cow::Borrowed("Config"),
store_error_type: None,
reason: message,
},
Self::UnsafeDataRootAncestor { reason, .. } => ErrorTraceFields {
error_type: Cow::Borrowed("UnsafeDataRootAncestor"),
store_error_type: None,
reason,
},
Self::TransportBind { message, .. } => ErrorTraceFields {
error_type: Cow::Borrowed("TransportBind"),
store_error_type: None,
reason: message,
},
Self::Transport { message, .. } => ErrorTraceFields {
error_type: Cow::Borrowed("Transport"),
store_error_type: None,
reason: message,
},
Self::SignalListener { message, .. } => ErrorTraceFields {
error_type: Cow::Borrowed("SignalListener"),
store_error_type: None,
reason: message,
},
Self::DeathNote { message } => ErrorTraceFields {
error_type: Cow::Borrowed("DeathNote"),
store_error_type: None,
reason: message,
},
Self::Namespace { message } => ErrorTraceFields {
error_type: Cow::Borrowed("Namespace"),
store_error_type: None,
reason: message,
},
Self::EngineCall { source } => engine_trace_fields(source),
Self::StoreBackend { source } => store_trace_fields(source),
Self::Stream { failure } => ErrorTraceFields {
error_type: Cow::Borrowed("Stream"),
store_error_type: None,
reason: failure,
},
Self::WorkerDispatch { reason, .. } => ErrorTraceFields {
error_type: Cow::Borrowed("WorkerDispatch"),
store_error_type: None,
reason,
},
Self::WorkerConnectionLost { detail, .. } => ErrorTraceFields {
error_type: Cow::Borrowed("WorkerConnectionLost"),
store_error_type: None,
reason: detail,
},
Self::WorkerBusy { detail, .. } => ErrorTraceFields {
error_type: Cow::Borrowed("WorkerBusy"),
store_error_type: None,
reason: detail,
},
Self::PendingActivityCollision { activity_id, .. } => ErrorTraceFields {
error_type: Cow::Borrowed("PendingActivityCollision"),
store_error_type: None,
reason: activity_id,
},
Self::DeclaredAttemptCollision { activity_id, .. } => ErrorTraceFields {
error_type: Cow::Borrowed("DeclaredAttemptCollision"),
store_error_type: None,
reason: activity_id,
},
Self::ActivityCompletionRejected { reason, .. } => ErrorTraceFields {
error_type: Cow::Borrowed("ActivityCompletionRejected"),
store_error_type: None,
reason,
},
Self::LockPoisoned { resource } => ErrorTraceFields {
error_type: Cow::Borrowed("LockPoisoned"),
store_error_type: None,
reason: resource,
},
Self::Wire { wire } => ErrorTraceFields {
error_type: wire
.error_type
.as_deref()
.map_or_else(|| Cow::Borrowed(wire.code.as_str()), Cow::Borrowed),
store_error_type: None,
reason: wire,
},
}
}
}
fn never_alive_error_type(source: &EngineError) -> &'static str {
match source {
EngineError::TerminalWriterUnavailable { .. } => "TerminalWriterUnavailable",
EngineError::TerminalWriterHeld { .. } => "TerminalWriterHeld",
EngineError::RunIsRecoverable { .. } => "RunIsRecoverable",
EngineError::NoResidencyVerdict { .. } => "NoResidencyVerdict",
_ => "EngineError",
}
}
fn durability_trace_fields<'a>(
durability: &'a aion::durability::DurabilityError,
source: &'a EngineError,
) -> ErrorTraceFields<'a> {
match durability {
aion::durability::DurabilityError::Store(store) => store_trace_fields(store),
aion::durability::DurabilityError::NonDeterminism(_)
| aion::durability::DurabilityError::HistoryShape { .. }
| aion::durability::DurabilityError::SearchAttribute(_) => {
simple_engine_fields("Durability", source)
}
aion::durability::DurabilityError::EngineTaskEpochClosed { .. } => {
simple_engine_fields("EngineTaskEpochClosed", source)
}
}
}
fn engine_trace_fields(source: &EngineError) -> ErrorTraceFields<'_> {
match source {
EngineError::WorkflowNotFound { .. } => simple_engine_fields("WorkflowNotFound", source),
EngineError::TerminalWriterUnavailable { .. }
| EngineError::TerminalWriterHeld { .. }
| EngineError::RunIsRecoverable { .. }
| EngineError::NoResidencyVerdict { .. } => {
simple_engine_fields(never_alive_error_type(source), source)
}
EngineError::InvalidState { .. } => simple_engine_fields("InvalidState", source),
EngineError::ScheduleNotFound { .. } => simple_engine_fields("ScheduleNotFound", source),
EngineError::ShuttingDown => simple_engine_fields("ShuttingDown", source),
EngineError::EngineTaskEpochClosed { .. } => {
simple_engine_fields("EngineTaskEpochClosed", source)
}
EngineError::Store(store) => store_trace_fields(store),
EngineError::Durability(durability) => durability_trace_fields(durability, source),
EngineError::MissingStore => simple_engine_fields("MissingStore", source),
EngineError::MissingVisibilityStore => {
simple_engine_fields("MissingVisibilityStore", source)
}
EngineError::ConflictingEventPublisher => {
simple_engine_fields("ConflictingEventPublisher", source)
}
EngineError::EventStreaming(_) => simple_engine_fields("EventStreaming", source),
EngineError::Load { .. } => simple_engine_fields("Load", source),
EngineError::UnenforceableContract { .. } => {
simple_engine_fields("UnenforceableContract", source)
}
EngineError::UnknownVersion { .. } => simple_engine_fields("UnknownVersion", source),
EngineError::VersionPinned { .. } => simple_engine_fields("VersionPinned", source),
EngineError::RouteActive { .. } => simple_engine_fields("RouteActive", source),
EngineError::ManifestMismatch { .. } => simple_engine_fields("ManifestMismatch", source),
EngineError::Package(_) => simple_engine_fields("Package", source),
EngineError::ContractIdentity { .. } => simple_engine_fields("ContractIdentity", source),
EngineError::NoQueueDeclaration { .. } => {
simple_engine_fields("NoQueueDeclaration", source)
}
EngineError::StartInputRefused { .. } => simple_engine_fields("StartInputRefused", source),
EngineError::SignalRefused { .. } => simple_engine_fields("SignalRefused", source),
EngineError::Schedule { .. } => simple_engine_fields("Schedule", source),
EngineError::Runtime { .. } => simple_engine_fields("Runtime", source),
EngineError::Gate3BifReplacementMissing { .. } => {
simple_engine_fields("Gate3BifReplacementMissing", source)
}
EngineError::StartupRecoveryNotDeferred => {
simple_engine_fields("StartupRecoveryNotDeferred", source)
}
EngineError::StartupRecoveryAlreadyRan => {
simple_engine_fields("StartupRecoveryAlreadyRan", source)
}
EngineError::StartupCatchupBeforeWorkflowRecovery => {
simple_engine_fields("StartupCatchupBeforeWorkflowRecovery", source)
}
EngineError::StartupRecoverySlotPoisoned => {
simple_engine_fields("StartupRecoverySlotPoisoned", source)
}
EngineError::CleanupExecutorPoisoned => {
simple_engine_fields("CleanupExecutorPoisoned", source)
}
EngineError::CleanupExecutorShutdownTimedOut { .. } => {
simple_engine_fields("CleanupExecutorShutdownTimedOut", source)
}
EngineError::RunNotInHistory { .. } => simple_engine_fields("RunNotInHistory", source),
EngineError::ProcessExitRegistryPoisoned => {
simple_engine_fields("ProcessExitRegistryPoisoned", source)
}
EngineError::ProcessExitOwnershipPoisoned { .. } => {
simple_engine_fields("ProcessExitOwnershipPoisoned", source)
}
EngineError::ProcessExitStatePoisoned { .. }
| EngineError::ProcessExitSubscriptionUnavailable
| EngineError::ProcessExitDrainerSpawn { .. }
| EngineError::ProcessExitDrainerPoisoned
| EngineError::ProcessExitOutcomeMissingAfterEvent { .. }
| EngineError::ProcessExitEventStreamDisconnected
| EngineError::ProcessExitDrainerShutdownTimedOut { .. }
| EngineError::ProcessExitDrainerPanicked => process_exit::drainer_trace(source),
EngineError::ProcessExitCallbackDispatcherPoisoned
| EngineError::ProcessExitCallbackDispatcherUnavailable
| EngineError::ProcessExitCallbackDispatcherShutdownTimedOut { .. } => {
process_exit::callback_trace(source)
}
EngineError::ProcessExitAlreadyTerminal { .. } => {
simple_engine_fields("ProcessExitAlreadyTerminal", source)
}
EngineError::ActivityDeliveryPoisoned { .. } => {
simple_engine_fields("ActivityDeliveryPoisoned", source)
}
EngineError::RegistryPoisoned => simple_engine_fields("RegistryPoisoned", source),
EngineError::CatalogPoisoned => simple_engine_fields("CatalogPoisoned", source),
EngineError::NifRegistration { .. } => simple_engine_fields("NifRegistration", source),
EngineError::SignalRouter(_) => simple_engine_fields("SignalRouter", source),
EngineError::Query(query) => simple_engine_fields(engine::query_error_type(query), source),
}
}
fn simple_engine_fields<'a>(
error_type: &'static str,
source: &'a EngineError,
) -> ErrorTraceFields<'a> {
ErrorTraceFields {
error_type: Cow::Borrowed(error_type),
store_error_type: None,
reason: source,
}
}
fn store_trace_fields(source: &StoreError) -> ErrorTraceFields<'_> {
ErrorTraceFields {
error_type: Cow::Borrowed("StoreError"),
store_error_type: Some(engine::store_error_type(source)),
reason: source,
}
}
fn wire_from_engine(source: &EngineError) -> WireError {
use EngineError as E;
use engine::backend_wire as backend;
match source {
EngineError::WorkflowNotFound { .. } => {
WireError::not_found_with_type("WorkflowNotFound", source.to_string())
}
EngineError::InvalidState { reason } => engine::invalid_state_wire(reason),
E::TerminalWriterUnavailable { .. }
| E::TerminalWriterHeld { .. }
| E::RunIsRecoverable { .. }
| E::NoResidencyVerdict { .. } => engine::invalid_state_wire(&source.to_string())
.with_error_type(never_alive_error_type(source)),
EngineError::ScheduleNotFound { .. } => {
WireError::not_found_with_type("ScheduleNotFound", source.to_string())
}
EngineError::ShuttingDown => {
WireError::not_running_with_type("ShuttingDown", source.to_string())
}
E::EngineTaskEpochClosed { .. } => backend("EngineTaskEpochClosed", source),
EngineError::Store(store) => wire_from_store(store),
EngineError::Durability(durability) => engine::durability_wire(durability, source),
E::MissingStore => backend("MissingStore", source),
E::MissingVisibilityStore => backend("MissingVisibilityStore", source),
E::ConflictingEventPublisher => backend("ConflictingEventPublisher", source),
E::EventStreaming(_) => backend("EventStreaming", source),
E::Load { .. } => backend("Load", source),
EngineError::UnenforceableContract { .. } => {
WireError::invalid_input(source.to_string()).with_error_type("UnenforceableContract")
}
EngineError::UnknownVersion { .. } => {
WireError::not_found_with_type("UnknownVersion", source.to_string())
}
EngineError::VersionPinned { .. } => {
WireError::version_pinned(source.to_string()).with_error_type("VersionPinned")
}
EngineError::RouteActive { .. } => {
WireError::version_pinned(source.to_string()).with_error_type("RouteActive")
}
EngineError::ManifestMismatch { .. } => {
WireError::invalid_input(source.to_string()).with_error_type("ManifestMismatch")
}
E::Package(_) => backend("Package", source),
E::ContractIdentity { .. } => engine::contract_refusal_wire("ContractIdentity", source),
E::NoQueueDeclaration { .. } => engine::contract_refusal_wire("NoQueueDeclaration", source),
E::StartInputRefused { .. } => engine::declared_contract_wire("StartInputRefused", source),
E::SignalRefused { .. } => engine::declared_contract_wire("SignalRefused", source),
EngineError::Schedule { .. } => backend("Schedule", source),
E::Runtime { .. } => backend("Runtime", source),
E::Gate3BifReplacementMissing { .. } => backend("Gate3BifReplacementMissing", source),
E::StartupRecoveryNotDeferred => backend("StartupRecoveryNotDeferred", source),
E::StartupRecoveryAlreadyRan => backend("StartupRecoveryAlreadyRan", source),
E::StartupCatchupBeforeWorkflowRecovery => {
backend("StartupCatchupBeforeWorkflowRecovery", source)
}
E::StartupRecoverySlotPoisoned => backend("StartupRecoverySlotPoisoned", source),
E::CleanupExecutorPoisoned => backend("CleanupExecutorPoisoned", source),
E::CleanupExecutorShutdownTimedOut { .. } => {
backend("CleanupExecutorShutdownTimedOut", source)
}
E::RunNotInHistory { .. } => backend("RunNotInHistory", source),
E::ProcessExitRegistryPoisoned => backend("ProcessExitRegistryPoisoned", source),
E::ProcessExitOwnershipPoisoned { .. } => backend("ProcessExitOwnershipPoisoned", source),
EngineError::ProcessExitStatePoisoned { .. }
| EngineError::ProcessExitSubscriptionUnavailable
| EngineError::ProcessExitDrainerSpawn { .. }
| EngineError::ProcessExitDrainerPoisoned
| EngineError::ProcessExitOutcomeMissingAfterEvent { .. }
| EngineError::ProcessExitEventStreamDisconnected
| EngineError::ProcessExitDrainerShutdownTimedOut { .. }
| EngineError::ProcessExitDrainerPanicked => process_exit::drainer_wire(source),
EngineError::ProcessExitCallbackDispatcherPoisoned
| EngineError::ProcessExitCallbackDispatcherUnavailable
| EngineError::ProcessExitCallbackDispatcherShutdownTimedOut { .. } => {
process_exit::callback_wire(source)
}
E::ProcessExitAlreadyTerminal { .. } => backend("ProcessExitAlreadyTerminal", source),
E::ActivityDeliveryPoisoned { .. } => backend("ActivityDeliveryPoisoned", source),
E::CatalogPoisoned => backend("CatalogPoisoned", source),
E::RegistryPoisoned => backend("RegistryPoisoned", source),
E::NifRegistration { .. } => backend("NifRegistration", source),
E::SignalRouter(_) => backend("SignalRouter", source),
EngineError::Query(query) => engine::query_wire(query, source),
}
}
fn wire_from_store(source: &StoreError) -> WireError {
match source {
StoreError::SequenceConflict { .. } => WireError::new_with_type(
aion_proto::WireErrorCode::SequenceConflict,
"SequenceConflict",
source.to_string(),
),
StoreError::NotFound { .. } => {
WireError::not_found_with_type("NotFound", source.to_string())
}
StoreError::NotOwner { .. } => {
WireError::not_owner(source.to_string()).with_error_type("NotOwner")
}
StoreError::Backend(_) => WireError::backend_with_type("Backend", source.to_string()),
StoreError::Serialization(_) => {
WireError::backend_with_type("Serialization", source.to_string())
}
}
}
#[cfg(test)]
#[path = "error_tests.rs"]
mod tests;