runledger_runtime/
error.rs

1use thiserror::Error;
2
3use crate::config::JobsConfigValidationError;
4
5pub type Result<T> = std::result::Result<T, Error>;
6
7#[derive(Debug, Error)]
8pub enum Error {
9    #[error(transparent)]
10    Scheduler(#[from] SchedulerError),
11    #[error(transparent)]
12    Worker(#[from] WorkerError),
13    #[error(transparent)]
14    Reaper(#[from] ReaperError),
15    #[error(transparent)]
16    Runtime(#[from] RuntimeError),
17}
18
19#[derive(Debug, Error)]
20pub enum SchedulerError {
21    #[error("failed to begin scheduler transaction")]
22    BeginTransaction { source: runledger_postgres::Error },
23    #[error("failed to commit scheduler transaction")]
24    CommitTransaction { source: runledger_postgres::Error },
25    #[error("failed to claim due schedules")]
26    ClaimDueSchedules { source: runledger_postgres::Error },
27    #[error("failed to create savepoint using `{statement}`")]
28    SavepointCreate {
29        statement: &'static str,
30        source: runledger_postgres::Error,
31    },
32    #[error("failed to rollback savepoint using `{statement}`")]
33    SavepointRollback {
34        statement: &'static str,
35        source: runledger_postgres::Error,
36    },
37    #[error("failed to release savepoint using `{statement}`")]
38    SavepointRelease {
39        statement: &'static str,
40        source: runledger_postgres::Error,
41    },
42    #[error("failed deferring failed schedule `{schedule_id}`")]
43    DeferFailedSchedule {
44        schedule_id: uuid::Uuid,
45        source: runledger_postgres::Error,
46    },
47    #[error(
48        "invalid cron expression for schedule `{schedule_name}` ({schedule_id}): `{cron_expr}`"
49    )]
50    InvalidCronExpression {
51        schedule_id: uuid::Uuid,
52        schedule_name: String,
53        cron_expr: String,
54    },
55    #[error("failed enqueueing scheduled job `{job_type}` from schedule `{schedule_id}`")]
56    EnqueueScheduledJob {
57        schedule_id: uuid::Uuid,
58        job_type: String,
59        source: runledger_postgres::Error,
60    },
61    #[error("failed marking schedule `{schedule_id}` as fired")]
62    MarkScheduleFired {
63        schedule_id: uuid::Uuid,
64        source: runledger_postgres::Error,
65    },
66    /// A schedule row returned by the runtime claim query disappeared before the
67    /// scheduler could advance or defer its next fire cursor.
68    #[error("claimed schedule `{schedule_id}` was missing while {operation}")]
69    ClaimedScheduleMissing {
70        schedule_id: uuid::Uuid,
71        operation: &'static str,
72    },
73}
74
75#[derive(Debug, Error)]
76pub enum WorkerError {
77    #[error("failed claiming jobs for worker `{worker_id}`")]
78    ClaimJobs {
79        worker_id: String,
80        source: runledger_postgres::Error,
81    },
82    #[error("failed setting running progress for job `{job_id}` attempt `{attempt}`")]
83    SetRunningProgress {
84        job_id: uuid::Uuid,
85        attempt: i32,
86        source: runledger_postgres::Error,
87    },
88    #[error("failed releasing unstarted claim for job `{job_id}` attempt `{attempt}`")]
89    ReleaseUnstartedClaim {
90        job_id: uuid::Uuid,
91        attempt: i32,
92        source: runledger_postgres::Error,
93    },
94    #[error("failed completing job `{job_id}` attempt `{attempt}` as success")]
95    CompleteSuccess {
96        job_id: uuid::Uuid,
97        attempt: i32,
98        source: runledger_postgres::Error,
99    },
100    #[error("failed continuing job `{job_id}` attempt `{attempt}` after successful handler slice")]
101    CompleteContinuation {
102        job_id: uuid::Uuid,
103        attempt: i32,
104        source: runledger_postgres::Error,
105    },
106    #[error("failed completing job `{job_id}` attempt `{attempt}` as failure")]
107    CompleteFailure {
108        job_id: uuid::Uuid,
109        attempt: i32,
110        source: runledger_postgres::Error,
111    },
112    #[error("failed heartbeat for job `{job_id}` attempt `{attempt}`")]
113    Heartbeat {
114        job_id: uuid::Uuid,
115        attempt: i32,
116        source: runledger_postgres::Error,
117    },
118}
119
120#[derive(Debug, Error)]
121pub enum ReaperError {
122    #[error(
123        "failed reaping expired leases with batch_size `{batch_size}` and retry_delay_ms `{retry_delay_ms}`"
124    )]
125    ReapExpiredLeases {
126        batch_size: i64,
127        retry_delay_ms: i32,
128        source: runledger_postgres::Error,
129    },
130}
131
132#[derive(Debug, Error)]
133#[non_exhaustive]
134pub enum RuntimeError {
135    /// The supervisor builder received an invalid [`JobsConfig`] value. This
136    /// usually means the config was constructed directly instead of through
137    /// [`JobsConfig::from_env`]. Validation is strict for the whole supervisor
138    /// config, even if the loop that would use an invalid field is disabled.
139    ///
140    /// [`JobsConfig`]: crate::config::JobsConfig
141    /// [`JobsConfig::from_env`]: crate::config::JobsConfig::from_env
142    #[error("invalid jobs runtime configuration: {source}")]
143    InvalidJobsConfig {
144        #[source]
145        source: JobsConfigValidationError,
146    },
147    /// The supervisor builder received both direct registry and catalog
148    /// registration sources. Choose either [`SupervisorBuilder::with_registry`]
149    /// or [`SupervisorBuilder::with_catalog`] for a single builder.
150    ///
151    /// [`SupervisorBuilder::with_catalog`]: crate::supervisor::SupervisorBuilder::with_catalog
152    /// [`SupervisorBuilder::with_registry`]: crate::supervisor::SupervisorBuilder::with_registry
153    #[error(
154        "supervisor builder received both a job registry and a job catalog; choose one registration source"
155    )]
156    MixedRegistrySources,
157    /// The supervisor builder requires a job registry when the worker or reaper
158    /// loop is enabled, but none was provided. Call
159    /// [`SupervisorBuilder::with_registry`] before [`SupervisorBuilder::build`].
160    ///
161    /// [`SupervisorBuilder::with_registry`]: crate::supervisor::SupervisorBuilder::with_registry
162    /// [`SupervisorBuilder::build`]: crate::supervisor::SupervisorBuilder::build
163    #[error(
164        "supervisor builder requires a job registry when worker or reaper loops are enabled \
165         (worker_enabled={worker_enabled}, reaper_enabled={reaper_enabled})"
166    )]
167    MissingRegistry {
168        worker_enabled: bool,
169        reaper_enabled: bool,
170    },
171    /// The supervisor builder must be called from within an active Tokio runtime
172    /// context. Ensure you are calling it inside a `#[tokio::main]`, `#[tokio::test]`,
173    /// or within a spawned task inside an existing runtime.
174    #[error("supervisor builder requires an active Tokio runtime")]
175    MissingTokioRuntime {
176        #[source]
177        source: tokio::runtime::TryCurrentError,
178    },
179    /// A supervised runtime task exited cleanly before shutdown was requested.
180    /// This is treated as an error because long-running loops should only exit
181    /// in response to a shutdown signal. Investigate logs for the specific task
182    /// that exited unexpectedly.
183    #[error("jobs runtime task `{task}` exited unexpectedly before shutdown")]
184    TaskExitedUnexpectedly { task: &'static str },
185    /// A supervised task panicked or failed to join cleanly. Examine logs and
186    /// process panic output for more details about the underlying task failure.
187    #[error("failed joining jobs runtime task `{task}`")]
188    TaskJoin {
189        task: &'static str,
190        #[source]
191        source: tokio::task::JoinError,
192    },
193    /// The supervisor did not complete shutdown within the requested timeout.
194    /// Some tasks may not have received or responded to the shutdown signal.
195    /// Consider increasing the timeout or investigating why tasks are shutting
196    /// down slowly.
197    #[error("jobs runtime shutdown exceeded timeout {timeout:?}")]
198    ShutdownTimeout { timeout: std::time::Duration },
199    /// The requested shutdown timeout is too large to represent as a deadline.
200    /// Use a smaller timeout value.
201    #[error("jobs runtime shutdown timeout {timeout:?} is too large to represent")]
202    ShutdownTimeoutTooLarge { timeout: std::time::Duration },
203    /// A supervised task failed (panicked or exited unexpectedly) and the
204    /// remaining tasks could not be shut down within the timeout. The original
205    /// task failure is available through the source error.
206    #[error(
207        "jobs runtime shutdown exceeded timeout {timeout:?} while draining after earlier task failure"
208    )]
209    ShutdownTimeoutAfterTaskError {
210        timeout: std::time::Duration,
211        #[source]
212        source: Box<RuntimeError>,
213    },
214}
215
216#[cfg(test)]
217mod tests {
218    use super::*;
219
220    #[test]
221    fn scheduler_invalid_cron_variant_contains_schedule_metadata() {
222        let schedule_id = uuid::Uuid::nil();
223        let error = SchedulerError::InvalidCronExpression {
224            schedule_id,
225            schedule_name: "nightly sync".to_string(),
226            cron_expr: "not cron".to_string(),
227        };
228
229        match error {
230            SchedulerError::InvalidCronExpression {
231                schedule_id: actual_id,
232                schedule_name,
233                cron_expr,
234            } => {
235                assert_eq!(actual_id, schedule_id);
236                assert_eq!(schedule_name, "nightly sync");
237                assert_eq!(cron_expr, "not cron");
238            }
239            other => panic!("expected invalid cron variant, got: {other:?}"),
240        }
241    }
242}
runledger_runtime/error.rs

runledger_runtime/
error.rs
