1use std::any::Any;
2use std::future::Future;
3use std::panic::AssertUnwindSafe;
4use std::sync::Arc;
5use std::time::Duration;
6
7use async_trait::async_trait;
8use chrono::{DateTime, Utc};
9use futures_util::{FutureExt, StreamExt, stream::FuturesUnordered};
10use runledger_core::jobs::{JobDeadLetterReason, JobFailure, JobTypeName};
11use tracing::warn;
12use uuid::Uuid;
13
14#[cfg(test)]
15const OBSERVER_TIMEOUT: Duration = Duration::from_millis(100);
16#[cfg(not(test))]
17const OBSERVER_TIMEOUT: Duration = Duration::from_secs(10);
18
19#[derive(Debug, Clone)]
20#[non_exhaustive]
21pub struct ObservedJob {
22 pub job_id: Uuid,
23 pub job_type: JobTypeName,
24 pub organization_id: Option<Uuid>,
25 pub run_number: i32,
26 pub attempt: i32,
27 pub max_attempts: i32,
28 pub worker_id: String,
29}
30
31impl ObservedJob {
32 #[must_use]
33 pub fn new(
34 job_id: Uuid,
35 job_type: JobTypeName,
36 organization_id: Option<Uuid>,
37 run_number: i32,
38 attempt: i32,
39 max_attempts: i32,
40 worker_id: impl Into<String>,
41 ) -> Self {
42 Self {
43 job_id,
44 job_type,
45 organization_id,
46 run_number,
47 attempt,
48 max_attempts,
49 worker_id: worker_id.into(),
50 }
51 }
52}
53
54#[derive(Debug, Clone)]
55#[non_exhaustive]
56pub struct JobRunningEvent {
57 pub job: ObservedJob,
58}
59
60impl JobRunningEvent {
61 #[must_use]
62 pub fn new(job: ObservedJob) -> Self {
63 Self { job }
64 }
65}
66
67#[derive(Debug, Clone)]
68#[non_exhaustive]
69pub struct JobSucceededEvent {
70 pub job: ObservedJob,
71 pub duration: Duration,
72 pub progress_done: Option<i64>,
73 pub progress_total: Option<i64>,
74}
75
76impl JobSucceededEvent {
77 #[must_use]
78 pub fn new(
79 job: ObservedJob,
80 duration: Duration,
81 progress_done: Option<i64>,
82 progress_total: Option<i64>,
83 ) -> Self {
84 Self {
85 job,
86 duration,
87 progress_done,
88 progress_total,
89 }
90 }
91}
92
93#[derive(Debug, Clone, PartialEq, Eq)]
94#[non_exhaustive]
95pub enum JobFailureDisposition {
96 RetryScheduled {
97 retry_delay_ms: i32,
98 next_run_at: DateTime<Utc>,
99 },
100 DeadLettered {
101 reason: JobDeadLetterReason,
102 },
103 Unknown,
104}
105
106#[derive(Debug, Clone)]
107#[non_exhaustive]
108pub struct JobFailedEvent {
109 pub job: ObservedJob,
110 pub duration: Duration,
111 pub failure: JobFailure,
112 pub disposition: JobFailureDisposition,
113}
114
115impl JobFailedEvent {
116 #[must_use]
117 pub fn new(
118 job: ObservedJob,
119 duration: Duration,
120 failure: JobFailure,
121 disposition: JobFailureDisposition,
122 ) -> Self {
123 Self {
124 job,
125 duration,
126 failure,
127 disposition,
128 }
129 }
130}
131
132#[derive(Debug, Clone, Copy, PartialEq, Eq)]
133#[non_exhaustive]
134pub enum JobCompletionPersistenceOperation {
135 Success,
136 Failure,
137}
138
139#[derive(Debug, Clone)]
140#[non_exhaustive]
141pub struct JobCompletionPersistFailedEvent {
142 pub job: ObservedJob,
143 pub duration: Duration,
144 pub operation: JobCompletionPersistenceOperation,
145 pub error: String,
146}
147
148impl JobCompletionPersistFailedEvent {
149 #[must_use]
150 pub fn new(
151 job: ObservedJob,
152 duration: Duration,
153 operation: JobCompletionPersistenceOperation,
154 error: impl Into<String>,
155 ) -> Self {
156 Self {
157 job,
158 duration,
159 operation,
160 error: error.into(),
161 }
162 }
163}
164
165#[derive(Debug, Clone, PartialEq, Eq)]
166#[non_exhaustive]
167pub enum JobLeaseReapedDisposition {
168 ReleasedToPending,
169 RetryScheduled {
170 retry_delay_ms: i32,
171 next_run_at: DateTime<Utc>,
172 },
173 DeadLettered {
174 reason: JobDeadLetterReason,
175 },
176 Unknown,
177}
178
179#[derive(Debug, Clone)]
180#[non_exhaustive]
181pub struct JobLeaseLostEvent {
182 pub job: ObservedJob,
183 pub duration: Duration,
184 pub failure: JobFailure,
185}
186
187impl JobLeaseLostEvent {
188 #[must_use]
189 pub fn new(job: ObservedJob, duration: Duration, failure: JobFailure) -> Self {
190 Self {
191 job,
192 duration,
193 failure,
194 }
195 }
196}
197
198#[derive(Debug, Clone)]
199#[non_exhaustive]
200pub struct JobLeaseReapedEvent {
201 pub job: ObservedJob,
202 pub failure: JobFailure,
203 pub started_without_renewal_heartbeat: bool,
204 pub disposition: JobLeaseReapedDisposition,
205}
206
207impl JobLeaseReapedEvent {
208 #[must_use]
209 pub fn new(
210 job: ObservedJob,
211 failure: JobFailure,
212 started_without_renewal_heartbeat: bool,
213 disposition: JobLeaseReapedDisposition,
214 ) -> Self {
215 Self {
216 job,
217 failure,
218 started_without_renewal_heartbeat,
219 disposition,
220 }
221 }
222}
223
224#[async_trait]
225pub trait JobLifecycleObserver: Send + Sync {
226 async fn on_job_running(&self, _event: JobRunningEvent) {}
227
228 async fn on_job_succeeded(&self, _event: JobSucceededEvent) {}
229
230 async fn on_job_failed(&self, _event: JobFailedEvent) {}
231
232 async fn on_job_completion_persist_failed(&self, _event: JobCompletionPersistFailedEvent) {}
233
234 async fn on_job_lease_lost(&self, _event: JobLeaseLostEvent) {}
235
236 async fn on_job_lease_reaped(&self, _event: JobLeaseReapedEvent) {}
237}
238
239#[derive(Clone, Default)]
240pub struct JobLifecycleObservers {
241 observers: Arc<Vec<Arc<dyn JobLifecycleObserver>>>,
242}
243
244impl JobLifecycleObservers {
245 #[must_use]
246 pub fn empty() -> Self {
247 Self::default()
248 }
249
250 #[must_use]
251 pub fn from_observer(observer: impl JobLifecycleObserver + 'static) -> Self {
252 Self {
253 observers: Arc::new(vec![Arc::new(observer)]),
254 }
255 }
256
257 #[must_use]
258 pub fn from_arc_observers(observers: Vec<Arc<dyn JobLifecycleObserver>>) -> Self {
259 Self {
260 observers: Arc::new(observers),
261 }
262 }
263
264 pub(crate) fn is_empty(&self) -> bool {
265 self.observers.is_empty()
266 }
267
268 pub(crate) async fn job_running(&self, event: JobRunningEvent) {
269 let job = event.job.clone();
270 self.notify_all_observers(
271 "on_job_running",
272 event,
273 &job,
274 |observer, event| async move {
275 observer.on_job_running(event).await;
276 },
277 )
278 .await;
279 }
280
281 pub(crate) async fn job_succeeded(&self, event: JobSucceededEvent) {
282 let job = event.job.clone();
283 self.notify_all_observers(
284 "on_job_succeeded",
285 event,
286 &job,
287 |observer, event| async move {
288 observer.on_job_succeeded(event).await;
289 },
290 )
291 .await;
292 }
293
294 pub(crate) async fn job_failed(&self, event: JobFailedEvent) {
295 let job = event.job.clone();
296 self.notify_all_observers("on_job_failed", event, &job, |observer, event| async move {
297 observer.on_job_failed(event).await;
298 })
299 .await;
300 }
301
302 pub(crate) async fn job_completion_persist_failed(
303 &self,
304 event: JobCompletionPersistFailedEvent,
305 ) {
306 let job = event.job.clone();
307 self.notify_all_observers(
308 "on_job_completion_persist_failed",
309 event,
310 &job,
311 |observer, event| async move {
312 observer.on_job_completion_persist_failed(event).await;
313 },
314 )
315 .await;
316 }
317
318 pub(crate) async fn job_lease_lost(&self, event: JobLeaseLostEvent) {
319 let job = event.job.clone();
320 self.notify_all_observers(
321 "on_job_lease_lost",
322 event,
323 &job,
324 |observer, event| async move {
325 observer.on_job_lease_lost(event).await;
326 },
327 )
328 .await;
329 }
330
331 pub(crate) async fn job_lease_reaped(&self, event: JobLeaseReapedEvent) {
332 let job = event.job.clone();
333 self.notify_all_observers(
334 "on_job_lease_reaped",
335 event,
336 &job,
337 |observer, event| async move {
338 observer.on_job_lease_reaped(event).await;
339 },
340 )
341 .await;
342 }
343
344 async fn notify_all_observers<E, F, Fut>(
345 &self,
346 callback_name: &'static str,
347 event: E,
348 job: &ObservedJob,
349 notify: F,
350 ) where
351 E: Clone,
352 F: Fn(Arc<dyn JobLifecycleObserver>, E) -> Fut,
353 Fut: Future<Output = ()> + Send,
354 {
355 let mut pending = FuturesUnordered::new();
356
357 for observer in self.observers.iter() {
358 let observer = Arc::clone(observer);
359 let job = ObserverJobLogContext::from(job);
360 let event = event.clone();
361 pending.push(notify_observer(callback_name, job, notify(observer, event)));
362 }
363
364 while pending.next().await.is_some() {}
365 }
366}
367
368#[derive(Debug)]
369struct ObserverJobLogContext {
370 job_id: Uuid,
371 job_type: String,
372 organization_id: Option<Uuid>,
373 run_number: i32,
374 attempt: i32,
375 max_attempts: i32,
376 worker_id: String,
377}
378
379impl From<&ObservedJob> for ObserverJobLogContext {
380 fn from(job: &ObservedJob) -> Self {
381 Self {
382 job_id: job.job_id,
383 job_type: job.job_type.to_string(),
384 organization_id: job.organization_id,
385 run_number: job.run_number,
386 attempt: job.attempt,
387 max_attempts: job.max_attempts,
388 worker_id: job.worker_id.clone(),
389 }
390 }
391}
392
393async fn notify_observer<F>(callback_name: &'static str, job: ObserverJobLogContext, future: F)
394where
395 F: Future<Output = ()> + Send,
396{
397 match tokio::time::timeout(OBSERVER_TIMEOUT, AssertUnwindSafe(future).catch_unwind()).await {
398 Ok(Ok(())) => {}
399 Ok(Err(panic_payload)) => {
400 let panic_message = panic_payload_message(&*panic_payload);
401 warn!(
402 callback_name,
403 job_id = %job.job_id,
404 job_type = %job.job_type,
405 organization_id = ?job.organization_id,
406 run_number = job.run_number,
407 attempt = job.attempt,
408 max_attempts = job.max_attempts,
409 worker_id = %job.worker_id,
410 panic = %panic_message,
411 "job lifecycle observer panicked"
412 );
413 }
414 Err(_) => {
415 warn!(
416 callback_name,
417 job_id = %job.job_id,
418 job_type = %job.job_type,
419 organization_id = ?job.organization_id,
420 run_number = job.run_number,
421 attempt = job.attempt,
422 max_attempts = job.max_attempts,
423 worker_id = %job.worker_id,
424 timeout_ms = OBSERVER_TIMEOUT.as_millis(),
425 "job lifecycle observer timed out"
426 );
427 }
428 }
429}
430
431fn panic_payload_message(panic_payload: &(dyn Any + Send)) -> String {
432 if let Some(message) = panic_payload.downcast_ref::<String>() {
433 return message.clone();
434 }
435
436 if let Some(message) = panic_payload.downcast_ref::<&'static str>() {
437 return (*message).to_string();
438 }
439
440 "non-string panic payload".to_string()
441}