runledger_runtime/catalog/
error.rs

1use runledger_core::jobs::{IdentifierValidationError, WorkflowBuildError};
2use runledger_postgres::jobs::JobDefinitionCatalogSyncError;
3use thiserror::Error;
4
5/// Error returned by [`super::JobCatalog`] validation, sync, and helper methods.
6#[non_exhaustive]
7#[derive(Debug, Error)]
8pub enum CatalogError {
9    /// A caller supplied an invalid catalog job type.
10    #[error("job type {job_type:?} is invalid: {source}")]
11    InvalidJobType {
12        /// Invalid job type value supplied by the caller.
13        job_type: String,
14        /// Identifier validation failure reported by `runledger-core`.
15        #[source]
16        source: IdentifierValidationError,
17    },
18    /// A registered handler returned an invalid job type.
19    #[error("handler job type {handler_job_type:?} is invalid: {source}")]
20    InvalidHandlerJobType {
21        /// Invalid job type returned by the handler.
22        handler_job_type: String,
23        /// Identifier validation failure reported by `runledger-core`.
24        #[source]
25        source: IdentifierValidationError,
26    },
27    /// The declared catalog job type did not match the handler's job type.
28    #[error("job type {declared} does not match handler job type {handler}")]
29    HandlerJobTypeMismatch {
30        /// Job type declared at the catalog registration site.
31        declared: String,
32        /// Job type returned by the handler.
33        handler: String,
34    },
35    /// The catalog already contains the requested job type.
36    #[error("job type {job_type} is already registered in the catalog")]
37    DuplicateJobType {
38        /// Duplicate job type.
39        job_type: String,
40    },
41    /// A catalog definition default failed validation.
42    #[error("catalog defaults are invalid: {field} must be positive")]
43    InvalidDefinitionValue {
44        /// Name of the invalid defaults field.
45        field: &'static str,
46    },
47    /// Retry-delay override failure codes must be non-empty.
48    #[error("failure code must be non-empty")]
49    InvalidFailureCode,
50    /// Retry-delay override values must be positive.
51    #[error("retry delay override must be positive")]
52    InvalidRetryDelay,
53    /// Exact sync requires a non-empty owned job-type scope.
54    #[error("exact sync scope must include at least one job type")]
55    InvalidExactSyncScope,
56    /// Exact sync scope construction received an invalid job type.
57    #[error("exact sync scope job type {job_type:?} is invalid: {source}")]
58    InvalidExactSyncScopeJobType {
59        /// Invalid job type supplied for the exact-sync scope.
60        job_type: String,
61        /// Identifier validation failure reported by `runledger-core`.
62        #[source]
63        source: IdentifierValidationError,
64    },
65    /// Exact sync cannot run against an empty catalog.
66    #[error("exact sync requires at least one catalog job")]
67    EmptyExactSyncCatalog,
68    /// A catalog job was not included in the exact-sync scope.
69    #[error("catalog job type {job_type} is outside the exact sync scope")]
70    JobTypeOutsideExactSyncScope {
71        /// Catalog job type missing from the exact-sync scope.
72        job_type: String,
73    },
74    /// An active schedule still references an enabled definition absent from the catalog.
75    #[error("active schedule {schedule_name} still references absent catalog job type {job_type}")]
76    ActiveScheduleForAbsentJobType {
77        /// Active schedule name that blocks disabling the absent definition.
78        schedule_name: String,
79        /// Absent catalog job type referenced by the active schedule.
80        job_type: String,
81    },
82    /// An active schedule still references a catalog job that would be disabled.
83    #[error(
84        "active schedule {schedule_name} still references disabled catalog job type {job_type}"
85    )]
86    ActiveScheduleForDisabledJobType {
87        /// Active schedule name that blocks disabling the catalog definition.
88        schedule_name: String,
89        /// Catalog job type referenced by the active schedule.
90        job_type: String,
91    },
92    /// The requested job type is not registered in the catalog.
93    #[error("job type {job_type} is not registered in the catalog")]
94    UnknownJobType {
95        /// Missing job type.
96        job_type: String,
97    },
98    /// The requested job type is disabled in catalog defaults.
99    #[error("job type {job_type} is disabled in the catalog")]
100    DisabledJobType {
101        /// Disabled job type.
102        job_type: String,
103    },
104    /// Workflow enqueue construction failed.
105    #[error(transparent)]
106    WorkflowBuild(#[from] WorkflowBuildError),
107    /// Starting a catalog sync transaction failed.
108    #[error("failed to start job definition sync transaction: {0}")]
109    SyncFailure(#[source] runledger_postgres::Error),
110    /// A persistence-layer catalog sync failure had no runtime-specific mapping.
111    #[error("failed to sync job definitions with an unmapped persistence error: {0}")]
112    DefinitionCatalogSyncFailure(#[source] Box<dyn std::error::Error + Send + Sync>),
113    /// Syncing a specific catalog job definition failed.
114    #[error("failed to sync job definition {job_type}: {source}")]
115    DefinitionSyncFailure {
116        /// Catalog job type whose definition failed to sync.
117        job_type: String,
118        /// Persistence-layer failure that occurred while syncing the definition.
119        #[source]
120        source: runledger_postgres::Error,
121    },
122    /// Committing a catalog sync transaction failed.
123    #[error("failed to commit job definition sync transaction: {0}")]
124    CommitFailure(#[source] sqlx::Error),
125    /// Applying the transaction-local bounds for definition-disabling sync failed.
126    #[error("failed to bound job definition sync critical section: {0}")]
127    CriticalSectionTimeoutFailure(#[source] runledger_postgres::Error),
128    /// Catalog definition sync input failed persistence-layer validation.
129    #[error("job definition sync input is invalid: {0}")]
130    DefinitionSyncValidationFailure(#[source] runledger_postgres::Error),
131    /// Locking schedules before disabling definitions failed.
132    #[error("failed to lock job schedules before disabling job definitions: {0}")]
133    ScheduleLockFailure(#[source] runledger_postgres::Error),
134    /// Locking definitions before checking and disabling definitions failed.
135    #[error("failed to lock job definitions before disabling job definitions: {0}")]
136    DefinitionLockFailure(#[source] runledger_postgres::Error),
137    /// Checking active schedules before disabling definitions failed.
138    #[error("failed to check active schedules before disabling job definitions: {0}")]
139    ScheduleCheckFailure(#[source] runledger_postgres::Error),
140    /// Inspecting existing job definitions before sync failed.
141    #[error("failed to inspect job definitions before syncing catalog: {0}")]
142    DefinitionInspectFailure(#[source] runledger_postgres::Error),
143    /// Disabling absent job definitions failed.
144    #[error("failed to disable absent job definitions: {0}")]
145    DisableAbsentFailure(#[source] runledger_postgres::Error),
146}
147
148impl CatalogError {
149    pub(crate) fn from_definition_catalog_sync_error(error: JobDefinitionCatalogSyncError) -> Self {
150        match error {
151            JobDefinitionCatalogSyncError::ActiveScheduleForAbsentJobType(reference) => {
152                Self::ActiveScheduleForAbsentJobType {
153                    schedule_name: reference.schedule_name,
154                    job_type: reference.job_type.to_string(),
155                }
156            }
157            JobDefinitionCatalogSyncError::ActiveScheduleForDisabledJobType(reference) => {
158                Self::ActiveScheduleForDisabledJobType {
159                    schedule_name: reference.schedule_name,
160                    job_type: reference.job_type.to_string(),
161                }
162            }
163            JobDefinitionCatalogSyncError::CriticalSectionTimeoutFailure(source) => {
164                Self::CriticalSectionTimeoutFailure(source)
165            }
166            JobDefinitionCatalogSyncError::ScheduleLockFailure(source) => {
167                Self::ScheduleLockFailure(source)
168            }
169            JobDefinitionCatalogSyncError::DefinitionLockFailure(source) => {
170                Self::DefinitionLockFailure(source)
171            }
172            JobDefinitionCatalogSyncError::ScheduleCheckFailure(source) => {
173                Self::ScheduleCheckFailure(source)
174            }
175            JobDefinitionCatalogSyncError::ValidationFailure(source) => {
176                Self::DefinitionSyncValidationFailure(source)
177            }
178            JobDefinitionCatalogSyncError::DefinitionInspectFailure(source) => {
179                Self::DefinitionInspectFailure(source)
180            }
181            JobDefinitionCatalogSyncError::DefinitionSyncFailure { job_type, source } => {
182                Self::DefinitionSyncFailure { job_type, source }
183            }
184            JobDefinitionCatalogSyncError::DisableAbsentFailure(source) => {
185                Self::DisableAbsentFailure(source)
186            }
187            // Fallback for future #[non_exhaustive] variants that this runtime
188            // version cannot map to a more specific CatalogError.
189            _ => Self::DefinitionCatalogSyncFailure(Box::new(error)),
190        }
191    }
192}
runledger_runtime/catalog/error.rs

runledger_runtime/catalog/
error.rs
