awa_worker/
events.rs

1use awa_model::JobRow;
2use chrono::{DateTime, Utc};
3use std::future::Future;
4use std::pin::Pin;
5use std::sync::Arc;
6use std::time::Duration;
7
8/// Typed lifecycle event for a specific job argument type.
9#[derive(Debug, Clone)]
10pub enum JobEvent<T> {
11    /// Job execution started after the claim committed.
12    Started { args: T, job: JobRow },
13    /// Job parked waiting for an external callback. `job.callback_id` and
14    /// `job.callback_timeout_at` identify the pending callback.
15    WaitingForCallback { args: T, job: JobRow },
16    /// Job completed successfully.
17    ///
18    /// For a callback-driven completion `duration` is [`Duration::ZERO`], since
19    /// no handler ran during the parked phase.
20    Completed {
21        args: T,
22        job: JobRow,
23        duration: Duration,
24    },
25    /// Job failed but will be retried later.
26    Retried {
27        args: T,
28        job: JobRow,
29        error: String,
30        attempt: i16,
31        next_run_at: DateTime<Utc>,
32    },
33    /// Job exhausted its retry budget and moved to `failed`.
34    Exhausted {
35        args: T,
36        job: JobRow,
37        error: String,
38        attempt: i16,
39    },
40    /// Job was cancelled by the handler.
41    Cancelled {
42        args: T,
43        job: JobRow,
44        reason: String,
45    },
46    /// Job was rescued by maintenance — either its callback wait expired,
47    /// its heartbeat went stale (presumed crashed worker), or its deadline
48    /// was exceeded. `reason` identifies the rescue path. The post-rescue
49    /// `job.state` is `retryable` when retries remain or `failed` when the
50    /// attempt that was rescued exhausted its retry budget.
51    Rescued {
52        args: T,
53        job: JobRow,
54        reason: RescueReason,
55    },
56}
57
58/// Why maintenance rescued the job.
59#[derive(Debug, Clone, Copy, PartialEq, Eq)]
60pub enum RescueReason {
61    /// The external callback's `callback_timeout_at` elapsed without resolution.
62    ExpiredCallback,
63    /// The job's heartbeat went stale (worker presumed crashed or stuck).
64    StaleHeartbeat,
65    /// The job's `deadline_at` elapsed.
66    DeadlineExceeded,
67}
68
69impl RescueReason {
70    /// Stable identifier suitable for logs / spec dispatch context.
71    pub fn as_str(self) -> &'static str {
72        match self {
73            RescueReason::ExpiredCallback => "expired_callback",
74            RescueReason::StaleHeartbeat => "stale_heartbeat",
75            RescueReason::DeadlineExceeded => "deadline_exceeded",
76        }
77    }
78}
79
80impl<T> JobEvent<T> {
81    /// The job snapshot associated with this event.
82    pub fn job(&self) -> &JobRow {
83        match self {
84            JobEvent::Started { job, .. }
85            | JobEvent::WaitingForCallback { job, .. }
86            | JobEvent::Completed { job, .. }
87            | JobEvent::Retried { job, .. }
88            | JobEvent::Exhausted { job, .. }
89            | JobEvent::Cancelled { job, .. }
90            | JobEvent::Rescued { job, .. } => job,
91        }
92    }
93}
94
95/// Untyped lifecycle event keyed only by job kind.
96#[derive(Debug, Clone)]
97pub enum UntypedJobEvent {
98    /// Job execution started after the claim committed.
99    Started { job: JobRow },
100    /// Job parked waiting for an external callback.
101    WaitingForCallback { job: JobRow },
102    /// Job completed successfully.
103    Completed { job: JobRow, duration: Duration },
104    /// Job failed but will be retried later.
105    Retried {
106        job: JobRow,
107        error: String,
108        attempt: i16,
109        next_run_at: DateTime<Utc>,
110    },
111    /// Job exhausted its retry budget and moved to `failed`.
112    Exhausted {
113        job: JobRow,
114        error: String,
115        attempt: i16,
116    },
117    /// Job was cancelled by the handler.
118    Cancelled { job: JobRow, reason: String },
119    /// Job was rescued by maintenance. See [`JobEvent::Rescued`].
120    Rescued { job: JobRow, reason: RescueReason },
121}
122
123impl UntypedJobEvent {
124    /// The job snapshot associated with this event.
125    pub fn job(&self) -> &JobRow {
126        match self {
127            UntypedJobEvent::Started { job, .. }
128            | UntypedJobEvent::WaitingForCallback { job, .. }
129            | UntypedJobEvent::Completed { job, .. }
130            | UntypedJobEvent::Retried { job, .. }
131            | UntypedJobEvent::Exhausted { job, .. }
132            | UntypedJobEvent::Cancelled { job, .. }
133            | UntypedJobEvent::Rescued { job, .. } => job,
134        }
135    }
136
137    pub(crate) fn into_typed<T>(self, args: T) -> JobEvent<T> {
138        match self {
139            UntypedJobEvent::Started { job } => JobEvent::Started { args, job },
140            UntypedJobEvent::WaitingForCallback { job } => {
141                JobEvent::WaitingForCallback { args, job }
142            }
143            UntypedJobEvent::Completed { job, duration } => JobEvent::Completed {
144                args,
145                job,
146                duration,
147            },
148            UntypedJobEvent::Retried {
149                job,
150                error,
151                attempt,
152                next_run_at,
153            } => JobEvent::Retried {
154                args,
155                job,
156                error,
157                attempt,
158                next_run_at,
159            },
160            UntypedJobEvent::Exhausted {
161                job,
162                error,
163                attempt,
164            } => JobEvent::Exhausted {
165                args,
166                job,
167                error,
168                attempt,
169            },
170            UntypedJobEvent::Cancelled { job, reason } => JobEvent::Cancelled { args, job, reason },
171            UntypedJobEvent::Rescued { job, reason } => JobEvent::Rescued { args, job, reason },
172        }
173    }
174}
175
176pub(crate) type BoxedLifecycleFuture = Pin<Box<dyn Future<Output = ()> + Send + 'static>>;
177pub(crate) type BoxedUntypedEventHandler =
178    Arc<dyn Fn(UntypedJobEvent) -> BoxedLifecycleFuture + Send + Sync + 'static>;
awa_worker/events.rs

awa_worker/
events.rs
