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