aion_server/
error.rs

1//! `ServerError` taxonomy for server library modules.
2
3use std::borrow::Cow;
4use std::net::SocketAddr;
5
6use aion::EngineError;
7use aion_proto::WireError;
8use aion_store::StoreError;
9use thiserror::Error;
10
11/// Server-library error taxonomy.
12#[derive(Debug, Error)]
13pub enum ServerError {
14    /// Operator configuration could not be loaded or validated.
15    #[error("configuration error: {message}")]
16    Config {
17        /// Redacted, operator-facing failure message.
18        message: String,
19    },
20
21    /// A transport listener could not bind or start.
22    #[error("{transport} transport failed at {address}: {message}")]
23    TransportBind {
24        /// Transport name.
25        transport: &'static str,
26        /// Configured listener address.
27        address: SocketAddr,
28        /// Redacted, operator-facing failure message.
29        message: String,
30    },
31
32    /// A running transport task aborted: it panicked or was cancelled.
33    #[error("{transport} transport task failed: {message}")]
34    Transport {
35        /// Transport name.
36        transport: &'static str,
37        /// Redacted, operator-facing failure message.
38        message: String,
39    },
40
41    /// A termination-signal listener could not be installed or failed.
42    #[error("{listener} listener failed: {message}")]
43    SignalListener {
44        /// Listener name (`SIGTERM`, `SIGINT`, or the portable fallback).
45        listener: &'static str,
46        /// Redacted, operator-facing failure message.
47        message: String,
48    },
49
50    /// Namespace validation or authorization failed.
51    #[error("namespace error: {message}")]
52    Namespace {
53        /// Redacted namespace failure message.
54        message: String,
55    },
56
57    /// Engine call failed.
58    #[error("engine call failed: {source}")]
59    EngineCall {
60        /// Typed engine error returned by the embedded engine.
61        #[from]
62        source: EngineError,
63    },
64
65    /// Store backend call failed before an engine handle was available.
66    #[error("store backend failed: {source}")]
67    StoreBackend {
68        /// Typed store error returned by the configured backend.
69        #[from]
70        source: StoreError,
71    },
72
73    /// Streaming failure.
74    #[error("stream failure: {failure}")]
75    Stream {
76        /// Stream failure class.
77        failure: StreamFailure,
78    },
79
80    /// A scheduled activity could not be pushed to a worker.
81    #[error(
82        "worker dispatch failed for namespace {namespace}, activity type {activity_type}: {reason}"
83    )]
84    WorkerDispatch {
85        /// Namespace scoped before dispatch.
86        namespace: String,
87        /// Activity type requested by the engine.
88        activity_type: String,
89        /// Redacted dispatch failure reason.
90        reason: String,
91    },
92
93    /// A lock was poisoned and the protected state cannot be trusted.
94    #[error("{resource} lock was poisoned")]
95    LockPoisoned {
96        /// Protected resource name.
97        resource: &'static str,
98    },
99
100    /// A failure already translated into the public wire taxonomy.
101    #[error("wire error: {wire}")]
102    Wire {
103        /// Stable wire error.
104        wire: WireError,
105    },
106}
107
108/// Bounded-stream and connection failure classes.
109#[derive(Clone, Copy, Debug, Eq, Error, PartialEq)]
110pub enum StreamFailure {
111    /// Bounded per-connection buffer overflowed because the consumer lagged.
112    #[error("consumer lagged behind bounded buffer")]
113    Lagged,
114    /// Subscriber closed the connection.
115    #[error("subscriber connection closed")]
116    Closed,
117    /// Upstream engine event stream ended unexpectedly.
118    #[error("engine event stream closed")]
119    UpstreamClosed,
120}
121
122impl From<WireError> for ServerError {
123    fn from(wire: WireError) -> Self {
124        Self::Wire { wire }
125    }
126}
127
128impl ServerError {
129    /// Convert a server error that crosses a transport boundary into the stable
130    /// public wire taxonomy.
131    #[must_use]
132    pub fn to_wire_error(&self) -> WireError {
133        match self {
134            Self::Config { .. }
135            | Self::TransportBind { .. }
136            | Self::Transport { .. }
137            | Self::SignalListener { .. }
138            | Self::LockPoisoned { .. } => WireError::backend("server backend failure"),
139            Self::WorkerDispatch { .. } => WireError::backend("worker dispatch failed"),
140            Self::Namespace { message } => WireError::namespace_denied(message.clone()),
141            Self::EngineCall { source } => wire_from_engine(source),
142            Self::StoreBackend { source } => wire_from_store(source),
143            Self::Stream { failure } => match failure {
144                StreamFailure::Lagged => WireError::lagged("subscriber lagged behind"),
145                StreamFailure::Closed | StreamFailure::UpstreamClosed => {
146                    WireError::backend("event stream closed")
147                }
148            },
149            Self::Wire { wire } => wire.clone(),
150        }
151    }
152
153    /// Return true when this is an operator configuration failure.
154    #[must_use]
155    pub const fn is_config(&self) -> bool {
156        matches!(self, Self::Config { .. })
157    }
158
159    /// Construct a namespace-denied error without embedding authorization logic.
160    #[must_use]
161    pub fn namespace_denied(message: impl Into<String>) -> Self {
162        Self::Namespace {
163            message: message.into(),
164        }
165    }
166
167    /// Construct a deploy-authorization denial carried on the dedicated
168    /// `deploy_denied` wire code (deploy is not a namespace operation).
169    #[must_use]
170    pub fn deploy_denied(message: impl Into<String>) -> Self {
171        Self::Wire {
172            wire: WireError::deploy_denied(message),
173        }
174    }
175
176    /// Construct a lagged-stream error.
177    #[must_use]
178    pub const fn lagged_stream() -> Self {
179        Self::Stream {
180            failure: StreamFailure::Lagged,
181        }
182    }
183
184    /// Construct a worker-dispatch error.
185    #[must_use]
186    pub fn worker_dispatch(
187        namespace: impl Into<String>,
188        activity_type: impl Into<String>,
189        reason: impl Into<String>,
190    ) -> Self {
191        Self::WorkerDispatch {
192            namespace: namespace.into(),
193            activity_type: activity_type.into(),
194            reason: reason.into(),
195        }
196    }
197
198    /// Construct a lock-poison error at the lock boundary.
199    #[must_use]
200    pub const fn lock_poisoned(resource: &'static str) -> Self {
201        Self::LockPoisoned { resource }
202    }
203}
204
205/// Stable structured error metadata for tracing events.
206#[derive(Clone)]
207pub struct ErrorTraceFields<'a> {
208    /// Outer error type recorded in the `error_type` tracing field.
209    pub error_type: Cow<'a, str>,
210    /// Optional inner store error type for `StoreError` records.
211    pub store_error_type: Option<&'static str>,
212    /// Human-readable reason safe for operator logs.
213    pub reason: &'a dyn std::fmt::Display,
214}
215
216impl ServerError {
217    /// Return stable typed fields for structured error logging.
218    #[must_use]
219    pub fn trace_fields(&self) -> ErrorTraceFields<'_> {
220        match self {
221            Self::Config { message } => ErrorTraceFields {
222                error_type: Cow::Borrowed("Config"),
223                store_error_type: None,
224                reason: message,
225            },
226            Self::TransportBind { message, .. } => ErrorTraceFields {
227                error_type: Cow::Borrowed("TransportBind"),
228                store_error_type: None,
229                reason: message,
230            },
231            Self::Transport { message, .. } => ErrorTraceFields {
232                error_type: Cow::Borrowed("Transport"),
233                store_error_type: None,
234                reason: message,
235            },
236            Self::SignalListener { message, .. } => ErrorTraceFields {
237                error_type: Cow::Borrowed("SignalListener"),
238                store_error_type: None,
239                reason: message,
240            },
241            Self::Namespace { message } => ErrorTraceFields {
242                error_type: Cow::Borrowed("Namespace"),
243                store_error_type: None,
244                reason: message,
245            },
246            Self::EngineCall { source } => engine_trace_fields(source),
247            Self::StoreBackend { source } => store_trace_fields(source),
248            Self::Stream { failure } => ErrorTraceFields {
249                error_type: Cow::Borrowed("Stream"),
250                store_error_type: None,
251                reason: failure,
252            },
253            Self::WorkerDispatch { reason, .. } => ErrorTraceFields {
254                error_type: Cow::Borrowed("WorkerDispatch"),
255                store_error_type: None,
256                reason,
257            },
258            Self::LockPoisoned { resource } => ErrorTraceFields {
259                error_type: Cow::Borrowed("LockPoisoned"),
260                store_error_type: None,
261                reason: resource,
262            },
263            Self::Wire { wire } => ErrorTraceFields {
264                error_type: wire
265                    .error_type
266                    .as_deref()
267                    .map_or_else(|| Cow::Borrowed(wire.code.as_str()), Cow::Borrowed),
268                store_error_type: None,
269                reason: wire,
270            },
271        }
272    }
273}
274
275fn engine_trace_fields(source: &EngineError) -> ErrorTraceFields<'_> {
276    match source {
277        EngineError::WorkflowNotFound { .. } => simple_engine_fields("WorkflowNotFound", source),
278        EngineError::ScheduleNotFound { .. } => simple_engine_fields("ScheduleNotFound", source),
279        EngineError::ShuttingDown => simple_engine_fields("ShuttingDown", source),
280        EngineError::Store(store) => store_trace_fields(store),
281        EngineError::Durability(durability) => match durability {
282            aion::durability::DurabilityError::Store(store) => store_trace_fields(store),
283            aion::durability::DurabilityError::NonDeterminism(_)
284            | aion::durability::DurabilityError::HistoryShape { .. }
285            | aion::durability::DurabilityError::SearchAttribute(_) => {
286                simple_engine_fields("Durability", source)
287            }
288        },
289        EngineError::MissingStore => simple_engine_fields("MissingStore", source),
290        EngineError::MissingVisibilityStore => {
291            simple_engine_fields("MissingVisibilityStore", source)
292        }
293        EngineError::ConflictingEventPublisher => {
294            simple_engine_fields("ConflictingEventPublisher", source)
295        }
296        EngineError::EventStreaming(_) => simple_engine_fields("EventStreaming", source),
297        EngineError::Load { .. } => simple_engine_fields("Load", source),
298        EngineError::UnknownVersion { .. } => simple_engine_fields("UnknownVersion", source),
299        EngineError::VersionPinned { .. } => simple_engine_fields("VersionPinned", source),
300        EngineError::RouteActive { .. } => simple_engine_fields("RouteActive", source),
301        EngineError::ManifestMismatch { .. } => simple_engine_fields("ManifestMismatch", source),
302        EngineError::Package(_) => simple_engine_fields("Package", source),
303        EngineError::Schedule { .. } => simple_engine_fields("Schedule", source),
304        EngineError::Runtime { .. } => simple_engine_fields("Runtime", source),
305        EngineError::RegistryPoisoned => simple_engine_fields("RegistryPoisoned", source),
306        EngineError::CatalogPoisoned => simple_engine_fields("CatalogPoisoned", source),
307        EngineError::NifRegistration { .. } => simple_engine_fields("NifRegistration", source),
308        EngineError::SignalRouter(_) => simple_engine_fields("SignalRouter", source),
309        EngineError::Query(query) => simple_engine_fields(query_error_type(query), source),
310    }
311}
312
313/// Trace discriminator for live-query dispatch failures.
314fn query_error_type(source: &aion::QueryError) -> &'static str {
315    match source {
316        aion::QueryError::UnknownQuery(_) => "UnknownQuery",
317        aion::QueryError::Timeout => "QueryTimeout",
318        aion::QueryError::NotRunning(_) => "QueryNotRunning",
319        aion::QueryError::Unknown(_) => "QueryUnknownWorkflow",
320        aion::QueryError::ReplyDropped => "QueryReplyDropped",
321        aion::QueryError::HandlerFailed { .. } => "QueryFailed",
322        aion::QueryError::Engine(_) => "QueryEngine",
323    }
324}
325
326fn simple_engine_fields<'a>(
327    error_type: &'static str,
328    source: &'a EngineError,
329) -> ErrorTraceFields<'a> {
330    ErrorTraceFields {
331        error_type: Cow::Borrowed(error_type),
332        store_error_type: None,
333        reason: source,
334    }
335}
336
337fn store_trace_fields(source: &StoreError) -> ErrorTraceFields<'_> {
338    ErrorTraceFields {
339        error_type: Cow::Borrowed("StoreError"),
340        store_error_type: Some(store_error_type(source)),
341        reason: source,
342    }
343}
344
345fn store_error_type(source: &StoreError) -> &'static str {
346    match source {
347        StoreError::SequenceConflict { .. } => "SequenceConflict",
348        StoreError::NotFound { .. } => "NotFound",
349        StoreError::Backend(_) => "Backend",
350        StoreError::Serialization(_) => "Serialization",
351    }
352}
353
354fn wire_from_engine(source: &EngineError) -> WireError {
355    match source {
356        EngineError::WorkflowNotFound { .. } => {
357            WireError::not_found_with_type("WorkflowNotFound", source.to_string())
358        }
359        EngineError::ScheduleNotFound { .. } => {
360            WireError::not_found_with_type("ScheduleNotFound", source.to_string())
361        }
362        EngineError::ShuttingDown => {
363            WireError::not_running_with_type("ShuttingDown", source.to_string())
364        }
365        EngineError::Store(store) => wire_from_store(store),
366        EngineError::Durability(durability) => match durability {
367            aion::durability::DurabilityError::Store(store) => wire_from_store(store),
368            aion::durability::DurabilityError::NonDeterminism(_)
369            | aion::durability::DurabilityError::HistoryShape { .. }
370            | aion::durability::DurabilityError::SearchAttribute(_) => {
371                WireError::backend_with_type("Durability", source.to_string())
372            }
373        },
374        EngineError::MissingStore => {
375            WireError::backend_with_type("MissingStore", source.to_string())
376        }
377        EngineError::MissingVisibilityStore => {
378            WireError::backend_with_type("MissingVisibilityStore", source.to_string())
379        }
380        EngineError::ConflictingEventPublisher => {
381            WireError::backend_with_type("ConflictingEventPublisher", source.to_string())
382        }
383        EngineError::EventStreaming(_) => {
384            WireError::backend_with_type("EventStreaming", source.to_string())
385        }
386        EngineError::Load { .. } => WireError::backend_with_type("Load", source.to_string()),
387        // Deploy-surface refusals (the §2.4 mapping table): unknown
388        // `(type, version)` is not-found; route-active and pinned versions
389        // are state conflicts carried by the dedicated `version_pinned`
390        // code; a same-hash-different-manifest archive is invalid input.
391        EngineError::UnknownVersion { .. } => {
392            WireError::not_found_with_type("UnknownVersion", source.to_string())
393        }
394        EngineError::VersionPinned { .. } => {
395            WireError::version_pinned(source.to_string()).with_error_type("VersionPinned")
396        }
397        EngineError::RouteActive { .. } => {
398            WireError::version_pinned(source.to_string()).with_error_type("RouteActive")
399        }
400        EngineError::ManifestMismatch { .. } => {
401            WireError::invalid_input(source.to_string()).with_error_type("ManifestMismatch")
402        }
403        EngineError::Package(_) => WireError::backend_with_type("Package", source.to_string()),
404        EngineError::Schedule { .. } => {
405            WireError::backend_with_type("Schedule", source.to_string())
406        }
407        EngineError::Runtime { .. } => WireError::backend_with_type("Runtime", source.to_string()),
408        EngineError::CatalogPoisoned => {
409            WireError::backend_with_type("CatalogPoisoned", source.to_string())
410        }
411        EngineError::RegistryPoisoned => {
412            WireError::backend_with_type("RegistryPoisoned", source.to_string())
413        }
414        EngineError::NifRegistration { .. } => {
415            WireError::backend_with_type("NifRegistration", source.to_string())
416        }
417        EngineError::SignalRouter(_) => {
418            WireError::backend_with_type("SignalRouter", source.to_string())
419        }
420        EngineError::Query(query) => wire_from_query(query, source),
421    }
422}
423
424/// Wire mapping for live-query dispatch failures (per the #45 brief).
425///
426/// `ReplyDropped` maps to `not_running` per decision Q3: the workflow ended
427/// before answering. `HandlerFailed` maps to the dedicated `query_failed`
428/// code per decision Q1(b).
429fn wire_from_query(query: &aion::QueryError, source: &EngineError) -> WireError {
430    match query {
431        aion::QueryError::UnknownQuery(_) => WireError::unknown_query(source.to_string()),
432        aion::QueryError::Timeout => WireError::query_timeout(source.to_string()),
433        aion::QueryError::NotRunning(_) | aion::QueryError::ReplyDropped => {
434            WireError::not_running_with_type(query_error_type(query), source.to_string())
435        }
436        aion::QueryError::Unknown(_) => {
437            WireError::not_found_with_type(query_error_type(query), source.to_string())
438        }
439        aion::QueryError::HandlerFailed { .. } => {
440            WireError::query_failed(source.to_string()).with_error_type(query_error_type(query))
441        }
442        aion::QueryError::Engine(_) => {
443            WireError::backend_with_type(query_error_type(query), source.to_string())
444        }
445    }
446}
447
448fn wire_from_store(source: &StoreError) -> WireError {
449    match source {
450        StoreError::SequenceConflict { .. } => WireError::new_with_type(
451            aion_proto::WireErrorCode::SequenceConflict,
452            "SequenceConflict",
453            source.to_string(),
454        ),
455        StoreError::NotFound { .. } => {
456            WireError::not_found_with_type("NotFound", source.to_string())
457        }
458        StoreError::Backend(_) => WireError::backend_with_type("Backend", source.to_string()),
459        StoreError::Serialization(_) => {
460            WireError::backend_with_type("Serialization", source.to_string())
461        }
462    }
463}
464
465#[cfg(test)]
466mod tests {
467    use super::{ServerError, StreamFailure};
468    use aion::{EngineError, QueryError, engine_seam::EngineSeamError};
469    use aion_core::WorkflowId;
470    use aion_proto::WireErrorCode;
471
472    fn assert_send_sync<T: Send + Sync>() {}
473
474    #[test]
475    fn server_error_is_send_sync() {
476        assert_send_sync::<ServerError>();
477    }
478
479    #[test]
480    fn lagged_stream_maps_to_wire_lagged() {
481        let error = ServerError::Stream {
482            failure: StreamFailure::Lagged,
483        };
484
485        assert_eq!(error.to_wire_error().code, WireErrorCode::Lagged);
486    }
487
488    fn workflow_id() -> WorkflowId {
489        WorkflowId::new(uuid::Uuid::from_u128(7))
490    }
491
492    fn query_wire(query: QueryError) -> aion_proto::WireError {
493        ServerError::EngineCall {
494            source: EngineError::Query(query),
495        }
496        .to_wire_error()
497    }
498
499    /// Pins the wire mapping for every `QueryError` arm (#45 decisions
500    /// Q1(b)/Q3): adding a variant breaks the exhaustive list below until its
501    /// mapping is decided and pinned here.
502    #[test]
503    fn every_query_error_arm_maps_to_its_pinned_wire_code() {
504        let arms: Vec<(QueryError, WireErrorCode, Option<&str>)> = vec![
505            (
506                QueryError::UnknownQuery(String::from("state")),
507                WireErrorCode::UnknownQuery,
508                None,
509            ),
510            (QueryError::Timeout, WireErrorCode::QueryTimeout, None),
511            (
512                QueryError::NotRunning(workflow_id()),
513                WireErrorCode::NotRunning,
514                Some("QueryNotRunning"),
515            ),
516            (
517                QueryError::Unknown(workflow_id()),
518                WireErrorCode::NotFound,
519                Some("QueryUnknownWorkflow"),
520            ),
521            // Q3: the workflow ended before answering — not_running, not backend.
522            (
523                QueryError::ReplyDropped,
524                WireErrorCode::NotRunning,
525                Some("QueryReplyDropped"),
526            ),
527            // Q1(b): the dedicated query_failed wire code.
528            (
529                QueryError::HandlerFailed {
530                    message: String::from("handler raised"),
531                },
532                WireErrorCode::QueryFailed,
533                Some("QueryFailed"),
534            ),
535            (
536                QueryError::Engine(EngineSeamError::Delivery {
537                    reason: String::from("mailbox closed"),
538                }),
539                WireErrorCode::Backend,
540                Some("QueryEngine"),
541            ),
542        ];
543
544        // Count-lock: the pin list must grow with the enum. The exhaustive
545        // match below numbers every variant; a new variant breaks the match
546        // first, and updating the match without pinning the new mapping
547        // breaks this assertion.
548        let variant_count = arms
549            .iter()
550            .map(|(query, _, _)| match query {
551                QueryError::UnknownQuery(_) => 0,
552                QueryError::Timeout => 1,
553                QueryError::NotRunning(_) => 2,
554                QueryError::Unknown(_) => 3,
555                QueryError::ReplyDropped => 4,
556                QueryError::HandlerFailed { .. } => 5,
557                QueryError::Engine(_) => 6,
558            })
559            .collect::<std::collections::BTreeSet<usize>>()
560            .len();
561        assert_eq!(
562            arms.len(),
563            variant_count,
564            "every QueryError variant must appear exactly once in the pin list",
565        );
566        assert_eq!(variant_count, 7, "pin list must cover all 7 variants");
567
568        for (query, expected_code, expected_type) in arms {
569            let wire = query_wire(query.clone());
570            assert_eq!(
571                wire.code, expected_code,
572                "{query:?} must map to {expected_code:?}",
573            );
574            assert_eq!(
575                wire.error_type.as_deref(),
576                expected_type,
577                "{query:?} must carry error_type {expected_type:?}",
578            );
579        }
580    }
581
582    /// The trace discriminator for `HandlerFailed` matches the wire
583    /// `error_type` so operators can correlate logs with client branches.
584    #[test]
585    fn handler_failed_trace_fields_use_query_failed_type() {
586        let error = ServerError::EngineCall {
587            source: EngineError::Query(QueryError::HandlerFailed {
588                message: String::from("handler raised"),
589            }),
590        };
591
592        assert_eq!(error.trace_fields().error_type, "QueryFailed");
593    }
594}
aion_server/error.rs

aion_server/
error.rs
