aion_server/
error.rs

1//! `ServerError` taxonomy for server library modules.
2
3use std::borrow::Cow;
4use std::net::SocketAddr;
5use std::path::PathBuf;
6
7use aion::EngineError;
8use aion_proto::WireError;
9use aion_store::StoreError;
10use thiserror::Error;
11
12/// Server-library error taxonomy.
13#[derive(Debug, Error)]
14pub enum ServerError {
15    /// Operator configuration could not be loaded or validated.
16    #[error("configuration error: {message}")]
17    Config {
18        /// Redacted, operator-facing failure message.
19        message: String,
20    },
21
22    /// A path-ambient store backend was configured beneath a renameable directory.
23    #[error(
24        "unsafe store.data_dir `{}`: ancestor `{}` is not owner-controlled: {reason}; \
25         move store.data_dir beneath the private Aion home (`$AION_HOME`, default `~/.aion`) \
26         and keep its ancestor chain owner-only",
27        .data_root.display(),
28        .component.display()
29    )]
30    UnsafeDataRootAncestor {
31        /// Descriptor-resolved data root that the backend would use by pathname.
32        data_root: PathBuf,
33        /// First unsafe component in the resolved root's ancestor chain.
34        component: PathBuf,
35        /// Ownership, mode, or inspection failure that made the component unsafe.
36        reason: String,
37    },
38
39    /// A transport listener could not bind or start.
40    #[error("{transport} transport failed at {address}: {message}")]
41    TransportBind {
42        /// Transport name.
43        transport: &'static str,
44        /// Configured listener address.
45        address: SocketAddr,
46        /// Redacted, operator-facing failure message.
47        message: String,
48    },
49
50    /// A running transport task aborted: it panicked or was cancelled.
51    #[error("{transport} transport task failed: {message}")]
52    Transport {
53        /// Transport name.
54        transport: &'static str,
55        /// Redacted, operator-facing failure message.
56        message: String,
57    },
58
59    /// A termination-signal listener could not be installed or failed.
60    #[error("{listener} listener failed: {message}")]
61    SignalListener {
62        /// Listener name (`SIGTERM`, `SIGINT`, or the portable fallback).
63        listener: &'static str,
64        /// Redacted, operator-facing failure message.
65        message: String,
66    },
67
68    /// Namespace validation or authorization failed.
69    #[error("namespace error: {message}")]
70    Namespace {
71        /// Redacted namespace failure message.
72        message: String,
73    },
74
75    /// Engine call failed.
76    #[error("engine call failed: {source}")]
77    EngineCall {
78        /// Typed engine error returned by the embedded engine.
79        #[from]
80        source: EngineError,
81    },
82
83    /// Store backend call failed before an engine handle was available.
84    #[error("store backend failed: {source}")]
85    StoreBackend {
86        /// Typed store error returned by the configured backend.
87        #[from]
88        source: StoreError,
89    },
90
91    /// Streaming failure.
92    #[error("stream failure: {failure}")]
93    Stream {
94        /// Stream failure class.
95        failure: StreamFailure,
96    },
97
98    /// A scheduled activity could not be pushed to a worker.
99    #[error(
100        "worker dispatch failed for namespace {namespace}, activity type {activity_type}: {reason}"
101    )]
102    WorkerDispatch {
103        /// Namespace scoped before dispatch.
104        namespace: String,
105        /// Activity type requested by the engine.
106        activity_type: String,
107        /// Redacted dispatch failure reason.
108        reason: String,
109    },
110
111    /// The worker connection chosen for a dispatch was lost mid-flight: the
112    /// connection was already gone at push time, or it closed before the worker
113    /// sent its correlated push reply.
114    ///
115    /// This is DISTINCT from [`Self::WorkerDispatch`]: a `WorkerDispatch` covers a
116    /// genuine reply timeout (the worker is alive but slow), a no-worker-available
117    /// selection failure, or any other dispatch fault, all of which keep the
118    /// outbox's normal exponential backoff. A `WorkerConnectionLost` instead means
119    /// the chosen worker is gone (and has already been deregistered by liminal's
120    /// `on_worker_unregistered`), so the row can be re-armed for IMMEDIATE re-claim
121    /// to fail over to a live worker without waiting out the backoff. The outbox
122    /// dispatcher keys its fast-failover decision on this variant.
123    #[error("worker connection lost during dispatch on {channel}: {detail}")]
124    WorkerConnectionLost {
125        /// Row-derived dispatch channel for operator diagnostics.
126        channel: String,
127        /// Redacted, operator-facing description of how the connection was lost.
128        detail: String,
129    },
130
131    /// A lock was poisoned and the protected state cannot be trusted.
132    #[error("{resource} lock was poisoned")]
133    LockPoisoned {
134        /// Protected resource name.
135        resource: &'static str,
136    },
137
138    /// A failure already translated into the public wire taxonomy.
139    #[error("wire error: {wire}")]
140    Wire {
141        /// Stable wire error.
142        wire: WireError,
143    },
144}
145
146/// Bounded-stream and connection failure classes.
147#[derive(Clone, Copy, Debug, Eq, Error, PartialEq)]
148pub enum StreamFailure {
149    /// Bounded per-connection buffer overflowed because the consumer lagged.
150    #[error("consumer lagged behind bounded buffer")]
151    Lagged,
152    /// Subscriber closed the connection.
153    #[error("subscriber connection closed")]
154    Closed,
155    /// Upstream engine event stream ended unexpectedly.
156    #[error("engine event stream closed")]
157    UpstreamClosed,
158}
159
160impl From<WireError> for ServerError {
161    fn from(wire: WireError) -> Self {
162        Self::Wire { wire }
163    }
164}
165
166impl ServerError {
167    /// Convert a server error that crosses a transport boundary into the stable
168    /// public wire taxonomy.
169    #[must_use]
170    pub fn to_wire_error(&self) -> WireError {
171        match self {
172            Self::Config { .. }
173            | Self::UnsafeDataRootAncestor { .. }
174            | Self::TransportBind { .. }
175            | Self::Transport { .. }
176            | Self::SignalListener { .. }
177            | Self::LockPoisoned { .. } => WireError::backend("server backend failure"),
178            Self::WorkerDispatch { .. } => WireError::backend("worker dispatch failed"),
179            Self::WorkerConnectionLost { .. } => {
180                WireError::backend("worker connection lost during dispatch")
181            }
182            Self::Namespace { message } => WireError::namespace_denied(message.clone()),
183            Self::EngineCall { source } => wire_from_engine(source),
184            Self::StoreBackend { source } => wire_from_store(source),
185            Self::Stream { failure } => match failure {
186                StreamFailure::Lagged => WireError::lagged("subscriber lagged behind"),
187                StreamFailure::Closed | StreamFailure::UpstreamClosed => {
188                    WireError::backend("event stream closed")
189                }
190            },
191            Self::Wire { wire } => wire.clone(),
192        }
193    }
194
195    /// Return true when this is an operator configuration failure.
196    #[must_use]
197    pub const fn is_config(&self) -> bool {
198        matches!(
199            self,
200            Self::Config { .. } | Self::UnsafeDataRootAncestor { .. }
201        )
202    }
203
204    /// Construct a namespace-denied error without embedding authorization logic.
205    #[must_use]
206    pub fn namespace_denied(message: impl Into<String>) -> Self {
207        Self::Namespace {
208            message: message.into(),
209        }
210    }
211
212    /// Construct the loud, whole-registration rejection when a worker's advertised
213    /// `node` violates a `Pinned{L}` namespace's placement (Control-Plane Phase 2,
214    /// P2-I1). Names the offending namespace, the worker's advertised node (or
215    /// "none"), and the required label set, so the operator sees exactly why the
216    /// registration was refused. Carried on the namespace-denied wire code — a
217    /// registration refused on isolation grounds is a namespace-authorization
218    /// failure, not a transient dispatch error.
219    #[must_use]
220    pub fn placement_admission_denied(
221        namespace: &str,
222        worker_node: Option<&str>,
223        required: &std::collections::BTreeSet<String>,
224    ) -> Self {
225        let node = worker_node.unwrap_or("none");
226        let required = required
227            .iter()
228            .map(String::as_str)
229            .collect::<Vec<_>>()
230            .join(", ");
231        Self::namespace_denied(format!(
232            "worker registration rejected: namespace {namespace} is Pinned to node label(s) \
233             [{required}] but the worker advertises node {node}, which is not in the required set"
234        ))
235    }
236
237    /// Construct a deploy-authorization denial carried on the dedicated
238    /// `deploy_denied` wire code (deploy is not a namespace operation).
239    #[must_use]
240    pub fn deploy_denied(message: impl Into<String>) -> Self {
241        Self::Wire {
242            wire: WireError::deploy_denied(message),
243        }
244    }
245
246    /// Construct a lagged-stream error.
247    #[must_use]
248    pub const fn lagged_stream() -> Self {
249        Self::Stream {
250            failure: StreamFailure::Lagged,
251        }
252    }
253
254    /// Construct a worker-dispatch error.
255    #[must_use]
256    pub fn worker_dispatch(
257        namespace: impl Into<String>,
258        activity_type: impl Into<String>,
259        reason: impl Into<String>,
260    ) -> Self {
261        Self::WorkerDispatch {
262            namespace: namespace.into(),
263            activity_type: activity_type.into(),
264            reason: reason.into(),
265        }
266    }
267
268    /// Construct a worker-connection-lost error for a dispatch whose chosen
269    /// worker connection was gone at push time or closed before replying.
270    #[must_use]
271    pub fn worker_connection_lost(channel: impl Into<String>, detail: impl Into<String>) -> Self {
272        Self::WorkerConnectionLost {
273            channel: channel.into(),
274            detail: detail.into(),
275        }
276    }
277
278    /// Return true when this is a lost-worker-connection dispatch failure.
279    ///
280    /// The outbox dispatcher keys its fast cross-node failover on this: a lost
281    /// connection means the worker is gone (already deregistered), so the row is
282    /// re-armed for immediate re-claim instead of waiting out the retry backoff.
283    #[must_use]
284    pub const fn is_worker_connection_lost(&self) -> bool {
285        matches!(self, Self::WorkerConnectionLost { .. })
286    }
287
288    /// Construct a lock-poison error at the lock boundary.
289    #[must_use]
290    pub const fn lock_poisoned(resource: &'static str) -> Self {
291        Self::LockPoisoned { resource }
292    }
293}
294
295/// Stable structured error metadata for tracing events.
296#[derive(Clone)]
297pub struct ErrorTraceFields<'a> {
298    /// Outer error type recorded in the `error_type` tracing field.
299    pub error_type: Cow<'a, str>,
300    /// Optional inner store error type for `StoreError` records.
301    pub store_error_type: Option<&'static str>,
302    /// Human-readable reason safe for operator logs.
303    pub reason: &'a dyn std::fmt::Display,
304}
305
306impl ServerError {
307    /// Return stable typed fields for structured error logging.
308    #[must_use]
309    pub fn trace_fields(&self) -> ErrorTraceFields<'_> {
310        match self {
311            Self::Config { message } => ErrorTraceFields {
312                error_type: Cow::Borrowed("Config"),
313                store_error_type: None,
314                reason: message,
315            },
316            Self::UnsafeDataRootAncestor { reason, .. } => ErrorTraceFields {
317                error_type: Cow::Borrowed("UnsafeDataRootAncestor"),
318                store_error_type: None,
319                reason,
320            },
321            Self::TransportBind { message, .. } => ErrorTraceFields {
322                error_type: Cow::Borrowed("TransportBind"),
323                store_error_type: None,
324                reason: message,
325            },
326            Self::Transport { message, .. } => ErrorTraceFields {
327                error_type: Cow::Borrowed("Transport"),
328                store_error_type: None,
329                reason: message,
330            },
331            Self::SignalListener { message, .. } => ErrorTraceFields {
332                error_type: Cow::Borrowed("SignalListener"),
333                store_error_type: None,
334                reason: message,
335            },
336            Self::Namespace { message } => ErrorTraceFields {
337                error_type: Cow::Borrowed("Namespace"),
338                store_error_type: None,
339                reason: message,
340            },
341            Self::EngineCall { source } => engine_trace_fields(source),
342            Self::StoreBackend { source } => store_trace_fields(source),
343            Self::Stream { failure } => ErrorTraceFields {
344                error_type: Cow::Borrowed("Stream"),
345                store_error_type: None,
346                reason: failure,
347            },
348            Self::WorkerDispatch { reason, .. } => ErrorTraceFields {
349                error_type: Cow::Borrowed("WorkerDispatch"),
350                store_error_type: None,
351                reason,
352            },
353            Self::WorkerConnectionLost { detail, .. } => ErrorTraceFields {
354                error_type: Cow::Borrowed("WorkerConnectionLost"),
355                store_error_type: None,
356                reason: detail,
357            },
358            Self::LockPoisoned { resource } => ErrorTraceFields {
359                error_type: Cow::Borrowed("LockPoisoned"),
360                store_error_type: None,
361                reason: resource,
362            },
363            Self::Wire { wire } => ErrorTraceFields {
364                error_type: wire
365                    .error_type
366                    .as_deref()
367                    .map_or_else(|| Cow::Borrowed(wire.code.as_str()), Cow::Borrowed),
368                store_error_type: None,
369                reason: wire,
370            },
371        }
372    }
373}
374
375fn engine_trace_fields(source: &EngineError) -> ErrorTraceFields<'_> {
376    match source {
377        EngineError::WorkflowNotFound { .. } => simple_engine_fields("WorkflowNotFound", source),
378        EngineError::InvalidState { .. } => simple_engine_fields("InvalidState", source),
379        EngineError::ScheduleNotFound { .. } => simple_engine_fields("ScheduleNotFound", source),
380        EngineError::ShuttingDown => simple_engine_fields("ShuttingDown", source),
381        EngineError::Store(store) => store_trace_fields(store),
382        EngineError::Durability(durability) => match durability {
383            aion::durability::DurabilityError::Store(store) => store_trace_fields(store),
384            aion::durability::DurabilityError::NonDeterminism(_)
385            | aion::durability::DurabilityError::HistoryShape { .. }
386            | aion::durability::DurabilityError::SearchAttribute(_) => {
387                simple_engine_fields("Durability", source)
388            }
389        },
390        EngineError::MissingStore => simple_engine_fields("MissingStore", source),
391        EngineError::MissingVisibilityStore => {
392            simple_engine_fields("MissingVisibilityStore", source)
393        }
394        EngineError::ConflictingEventPublisher => {
395            simple_engine_fields("ConflictingEventPublisher", source)
396        }
397        EngineError::EventStreaming(_) => simple_engine_fields("EventStreaming", source),
398        EngineError::Load { .. } => simple_engine_fields("Load", source),
399        EngineError::UnknownVersion { .. } => simple_engine_fields("UnknownVersion", source),
400        EngineError::VersionPinned { .. } => simple_engine_fields("VersionPinned", source),
401        EngineError::RouteActive { .. } => simple_engine_fields("RouteActive", source),
402        EngineError::ManifestMismatch { .. } => simple_engine_fields("ManifestMismatch", source),
403        EngineError::Package(_) => simple_engine_fields("Package", source),
404        EngineError::Schedule { .. } => simple_engine_fields("Schedule", source),
405        EngineError::Runtime { .. } => simple_engine_fields("Runtime", source),
406        EngineError::RegistryPoisoned => simple_engine_fields("RegistryPoisoned", source),
407        EngineError::CatalogPoisoned => simple_engine_fields("CatalogPoisoned", source),
408        EngineError::NifRegistration { .. } => simple_engine_fields("NifRegistration", source),
409        EngineError::SignalRouter(_) => simple_engine_fields("SignalRouter", source),
410        EngineError::Query(query) => simple_engine_fields(query_error_type(query), source),
411    }
412}
413
414/// Trace discriminator for live-query dispatch failures.
415fn query_error_type(source: &aion::QueryError) -> &'static str {
416    match source {
417        aion::QueryError::UnknownQuery(_) => "UnknownQuery",
418        aion::QueryError::Timeout => "QueryTimeout",
419        aion::QueryError::NotRunning(_) => "QueryNotRunning",
420        aion::QueryError::Unknown(_) => "QueryUnknownWorkflow",
421        aion::QueryError::ReplyDropped => "QueryReplyDropped",
422        aion::QueryError::HandlerFailed { .. } => "QueryFailed",
423        aion::QueryError::Engine(_) => "QueryEngine",
424    }
425}
426
427fn simple_engine_fields<'a>(
428    error_type: &'static str,
429    source: &'a EngineError,
430) -> ErrorTraceFields<'a> {
431    ErrorTraceFields {
432        error_type: Cow::Borrowed(error_type),
433        store_error_type: None,
434        reason: source,
435    }
436}
437
438fn store_trace_fields(source: &StoreError) -> ErrorTraceFields<'_> {
439    ErrorTraceFields {
440        error_type: Cow::Borrowed("StoreError"),
441        store_error_type: Some(store_error_type(source)),
442        reason: source,
443    }
444}
445
446fn store_error_type(source: &StoreError) -> &'static str {
447    match source {
448        StoreError::SequenceConflict { .. } => "SequenceConflict",
449        StoreError::NotFound { .. } => "NotFound",
450        StoreError::NotOwner { .. } => "NotOwner",
451        StoreError::Backend(_) => "Backend",
452        StoreError::Serialization(_) => "Serialization",
453    }
454}
455
456fn wire_from_engine(source: &EngineError) -> WireError {
457    match source {
458        EngineError::WorkflowNotFound { .. } => {
459            WireError::not_found_with_type("WorkflowNotFound", source.to_string())
460        }
461        // Reopen precondition failure (AD-012): the run is not a reopenable
462        // terminal. Carried on the dedicated `invalid_state` wire code (gRPC
463        // FailedPrecondition / HTTP 409), distinct from NotFound and Backend.
464        EngineError::InvalidState { reason } => {
465            WireError::invalid_state_with_type("InvalidState", reason.clone())
466        }
467        EngineError::ScheduleNotFound { .. } => {
468            WireError::not_found_with_type("ScheduleNotFound", source.to_string())
469        }
470        EngineError::ShuttingDown => {
471            WireError::not_running_with_type("ShuttingDown", source.to_string())
472        }
473        EngineError::Store(store) => wire_from_store(store),
474        EngineError::Durability(durability) => match durability {
475            aion::durability::DurabilityError::Store(store) => wire_from_store(store),
476            aion::durability::DurabilityError::NonDeterminism(_)
477            | aion::durability::DurabilityError::HistoryShape { .. }
478            | aion::durability::DurabilityError::SearchAttribute(_) => {
479                WireError::backend_with_type("Durability", source.to_string())
480            }
481        },
482        EngineError::MissingStore => {
483            WireError::backend_with_type("MissingStore", source.to_string())
484        }
485        EngineError::MissingVisibilityStore => {
486            WireError::backend_with_type("MissingVisibilityStore", source.to_string())
487        }
488        EngineError::ConflictingEventPublisher => {
489            WireError::backend_with_type("ConflictingEventPublisher", source.to_string())
490        }
491        EngineError::EventStreaming(_) => {
492            WireError::backend_with_type("EventStreaming", source.to_string())
493        }
494        EngineError::Load { .. } => WireError::backend_with_type("Load", source.to_string()),
495        // Deploy-surface refusals (the §2.4 mapping table): unknown
496        // `(type, version)` is not-found; route-active and pinned versions
497        // are state conflicts carried by the dedicated `version_pinned`
498        // code; a same-hash-different-manifest archive is invalid input.
499        EngineError::UnknownVersion { .. } => {
500            WireError::not_found_with_type("UnknownVersion", source.to_string())
501        }
502        EngineError::VersionPinned { .. } => {
503            WireError::version_pinned(source.to_string()).with_error_type("VersionPinned")
504        }
505        EngineError::RouteActive { .. } => {
506            WireError::version_pinned(source.to_string()).with_error_type("RouteActive")
507        }
508        EngineError::ManifestMismatch { .. } => {
509            WireError::invalid_input(source.to_string()).with_error_type("ManifestMismatch")
510        }
511        EngineError::Package(_) => WireError::backend_with_type("Package", source.to_string()),
512        EngineError::Schedule { .. } => {
513            WireError::backend_with_type("Schedule", source.to_string())
514        }
515        EngineError::Runtime { .. } => WireError::backend_with_type("Runtime", source.to_string()),
516        EngineError::CatalogPoisoned => {
517            WireError::backend_with_type("CatalogPoisoned", source.to_string())
518        }
519        EngineError::RegistryPoisoned => {
520            WireError::backend_with_type("RegistryPoisoned", source.to_string())
521        }
522        EngineError::NifRegistration { .. } => {
523            WireError::backend_with_type("NifRegistration", source.to_string())
524        }
525        EngineError::SignalRouter(_) => {
526            WireError::backend_with_type("SignalRouter", source.to_string())
527        }
528        EngineError::Query(query) => wire_from_query(query, source),
529    }
530}
531
532/// Wire mapping for live-query dispatch failures (per the #45 brief).
533///
534/// `ReplyDropped` maps to `not_running` per decision Q3: the workflow ended
535/// before answering. `HandlerFailed` maps to the dedicated `query_failed`
536/// code per decision Q1(b).
537fn wire_from_query(query: &aion::QueryError, source: &EngineError) -> WireError {
538    match query {
539        aion::QueryError::UnknownQuery(_) => WireError::unknown_query(source.to_string()),
540        aion::QueryError::Timeout => WireError::query_timeout(source.to_string()),
541        aion::QueryError::NotRunning(_) | aion::QueryError::ReplyDropped => {
542            WireError::not_running_with_type(query_error_type(query), source.to_string())
543        }
544        aion::QueryError::Unknown(_) => {
545            WireError::not_found_with_type(query_error_type(query), source.to_string())
546        }
547        aion::QueryError::HandlerFailed { .. } => {
548            WireError::query_failed(source.to_string()).with_error_type(query_error_type(query))
549        }
550        aion::QueryError::Engine(_) => {
551            WireError::backend_with_type(query_error_type(query), source.to_string())
552        }
553    }
554}
555
556fn wire_from_store(source: &StoreError) -> WireError {
557    match source {
558        StoreError::SequenceConflict { .. } => WireError::new_with_type(
559            aion_proto::WireErrorCode::SequenceConflict,
560            "SequenceConflict",
561            source.to_string(),
562        ),
563        StoreError::NotFound { .. } => {
564            WireError::not_found_with_type("NotFound", source.to_string())
565        }
566        StoreError::NotOwner { .. } => {
567            WireError::not_owner(source.to_string()).with_error_type("NotOwner")
568        }
569        StoreError::Backend(_) => WireError::backend_with_type("Backend", source.to_string()),
570        StoreError::Serialization(_) => {
571            WireError::backend_with_type("Serialization", source.to_string())
572        }
573    }
574}
575
576#[cfg(test)]
577mod tests {
578    use super::{ServerError, StreamFailure};
579    use aion::{EngineError, QueryError, engine_seam::EngineSeamError};
580    use aion_core::WorkflowId;
581    use aion_proto::WireErrorCode;
582
583    fn assert_send_sync<T: Send + Sync>() {}
584
585    #[test]
586    fn server_error_is_send_sync() {
587        assert_send_sync::<ServerError>();
588    }
589
590    #[test]
591    fn lagged_stream_maps_to_wire_lagged() {
592        let error = ServerError::Stream {
593            failure: StreamFailure::Lagged,
594        };
595
596        assert_eq!(error.to_wire_error().code, WireErrorCode::Lagged);
597    }
598
599    /// R-0: a fenced quorum write (`StoreError::NotOwner`) must surface as the
600    /// typed, retryable `NotOwner` wire code with a `NotOwner` `error_type`, NOT
601    /// the opaque `Backend` it used to collapse into.
602    #[test]
603    fn not_owner_store_error_maps_to_wire_not_owner() {
604        let error = ServerError::StoreBackend {
605            source: aion_store::StoreError::NotOwner { shard: 3 },
606        };
607        let wire = error.to_wire_error();
608        assert_eq!(wire.code, WireErrorCode::NotOwner);
609        assert_eq!(wire.error_type.as_deref(), Some("NotOwner"));
610    }
611
612    fn workflow_id() -> WorkflowId {
613        WorkflowId::new(uuid::Uuid::from_u128(7))
614    }
615
616    fn query_wire(query: QueryError) -> aion_proto::WireError {
617        ServerError::EngineCall {
618            source: EngineError::Query(query),
619        }
620        .to_wire_error()
621    }
622
623    /// Pins the wire mapping for every `QueryError` arm (#45 decisions
624    /// Q1(b)/Q3): adding a variant breaks the exhaustive list below until its
625    /// mapping is decided and pinned here.
626    #[test]
627    fn every_query_error_arm_maps_to_its_pinned_wire_code() {
628        let arms: Vec<(QueryError, WireErrorCode, Option<&str>)> = vec![
629            (
630                QueryError::UnknownQuery(String::from("state")),
631                WireErrorCode::UnknownQuery,
632                None,
633            ),
634            (QueryError::Timeout, WireErrorCode::QueryTimeout, None),
635            (
636                QueryError::NotRunning(workflow_id()),
637                WireErrorCode::NotRunning,
638                Some("QueryNotRunning"),
639            ),
640            (
641                QueryError::Unknown(workflow_id()),
642                WireErrorCode::NotFound,
643                Some("QueryUnknownWorkflow"),
644            ),
645            // Q3: the workflow ended before answering — not_running, not backend.
646            (
647                QueryError::ReplyDropped,
648                WireErrorCode::NotRunning,
649                Some("QueryReplyDropped"),
650            ),
651            // Q1(b): the dedicated query_failed wire code.
652            (
653                QueryError::HandlerFailed {
654                    message: String::from("handler raised"),
655                },
656                WireErrorCode::QueryFailed,
657                Some("QueryFailed"),
658            ),
659            (
660                QueryError::Engine(EngineSeamError::Delivery {
661                    reason: String::from("mailbox closed"),
662                }),
663                WireErrorCode::Backend,
664                Some("QueryEngine"),
665            ),
666        ];
667
668        // Count-lock: the pin list must grow with the enum. The exhaustive
669        // match below numbers every variant; a new variant breaks the match
670        // first, and updating the match without pinning the new mapping
671        // breaks this assertion.
672        let variant_count = arms
673            .iter()
674            .map(|(query, _, _)| match query {
675                QueryError::UnknownQuery(_) => 0,
676                QueryError::Timeout => 1,
677                QueryError::NotRunning(_) => 2,
678                QueryError::Unknown(_) => 3,
679                QueryError::ReplyDropped => 4,
680                QueryError::HandlerFailed { .. } => 5,
681                QueryError::Engine(_) => 6,
682            })
683            .collect::<std::collections::BTreeSet<usize>>()
684            .len();
685        assert_eq!(
686            arms.len(),
687            variant_count,
688            "every QueryError variant must appear exactly once in the pin list",
689        );
690        assert_eq!(variant_count, 7, "pin list must cover all 7 variants");
691
692        for (query, expected_code, expected_type) in arms {
693            let wire = query_wire(query.clone());
694            assert_eq!(
695                wire.code, expected_code,
696                "{query:?} must map to {expected_code:?}",
697            );
698            assert_eq!(
699                wire.error_type.as_deref(),
700                expected_type,
701                "{query:?} must carry error_type {expected_type:?}",
702            );
703        }
704    }
705
706    /// The trace discriminator for `HandlerFailed` matches the wire
707    /// `error_type` so operators can correlate logs with client branches.
708    #[test]
709    fn handler_failed_trace_fields_use_query_failed_type() {
710        let error = ServerError::EngineCall {
711            source: EngineError::Query(QueryError::HandlerFailed {
712                message: String::from("handler raised"),
713            }),
714        };
715
716        assert_eq!(error.trace_fields().error_type, "QueryFailed");
717    }
718}
aion_server/error.rs

aion_server/
error.rs
