runledger_runtime/
observer.rs

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)]
94#[non_exhaustive]
95pub struct JobContinuedEvent {
96    /// Identity of the successfully completed run slice.
97    pub job: ObservedJob,
98    pub duration: Duration,
99    pub next_run_number: i32,
100    pub next_run_at: DateTime<Utc>,
101    pub progress_done: Option<i64>,
102    pub progress_total: Option<i64>,
103}
104
105impl JobContinuedEvent {
106    #[must_use]
107    pub fn new(
108        job: ObservedJob,
109        duration: Duration,
110        next_run_number: i32,
111        next_run_at: DateTime<Utc>,
112        progress_done: Option<i64>,
113        progress_total: Option<i64>,
114    ) -> Self {
115        Self {
116            job,
117            duration,
118            next_run_number,
119            next_run_at,
120            progress_done,
121            progress_total,
122        }
123    }
124}
125
126#[derive(Debug, Clone, PartialEq, Eq)]
127#[non_exhaustive]
128pub enum JobFailureDisposition {
129    RetryScheduled {
130        retry_delay_ms: i32,
131        next_run_at: DateTime<Utc>,
132    },
133    DeadLettered {
134        reason: JobDeadLetterReason,
135    },
136    Unknown,
137}
138
139#[derive(Debug, Clone)]
140#[non_exhaustive]
141pub struct JobFailedEvent {
142    pub job: ObservedJob,
143    pub duration: Duration,
144    pub failure: JobFailure,
145    pub disposition: JobFailureDisposition,
146}
147
148impl JobFailedEvent {
149    #[must_use]
150    pub fn new(
151        job: ObservedJob,
152        duration: Duration,
153        failure: JobFailure,
154        disposition: JobFailureDisposition,
155    ) -> Self {
156        Self {
157            job,
158            duration,
159            failure,
160            disposition,
161        }
162    }
163}
164
165#[derive(Debug, Clone, Copy, PartialEq, Eq)]
166#[non_exhaustive]
167pub enum JobCompletionPersistenceOperation {
168    Success,
169    Continuation,
170    Failure,
171}
172
173#[derive(Debug, Clone)]
174#[non_exhaustive]
175pub struct JobCompletionPersistFailedEvent {
176    pub job: ObservedJob,
177    pub duration: Duration,
178    pub operation: JobCompletionPersistenceOperation,
179    pub error: String,
180}
181
182impl JobCompletionPersistFailedEvent {
183    #[must_use]
184    pub fn new(
185        job: ObservedJob,
186        duration: Duration,
187        operation: JobCompletionPersistenceOperation,
188        error: impl Into<String>,
189    ) -> Self {
190        Self {
191            job,
192            duration,
193            operation,
194            error: error.into(),
195        }
196    }
197}
198
199#[derive(Debug, Clone, PartialEq, Eq)]
200#[non_exhaustive]
201pub enum JobLeaseReapedDisposition {
202    ReleasedToPending,
203    RetryScheduled {
204        retry_delay_ms: i32,
205        next_run_at: DateTime<Utc>,
206    },
207    DeadLettered {
208        reason: JobDeadLetterReason,
209    },
210    Unknown,
211}
212
213#[derive(Debug, Clone)]
214#[non_exhaustive]
215pub struct JobLeaseLostEvent {
216    pub job: ObservedJob,
217    pub duration: Duration,
218    pub failure: JobFailure,
219}
220
221impl JobLeaseLostEvent {
222    #[must_use]
223    pub fn new(job: ObservedJob, duration: Duration, failure: JobFailure) -> Self {
224        Self {
225            job,
226            duration,
227            failure,
228        }
229    }
230}
231
232#[derive(Debug, Clone)]
233#[non_exhaustive]
234pub struct JobLeaseReapedEvent {
235    pub job: ObservedJob,
236    pub failure: JobFailure,
237    pub started_without_renewal_heartbeat: bool,
238    pub disposition: JobLeaseReapedDisposition,
239}
240
241impl JobLeaseReapedEvent {
242    #[must_use]
243    pub fn new(
244        job: ObservedJob,
245        failure: JobFailure,
246        started_without_renewal_heartbeat: bool,
247        disposition: JobLeaseReapedDisposition,
248    ) -> Self {
249        Self {
250            job,
251            failure,
252            started_without_renewal_heartbeat,
253            disposition,
254        }
255    }
256}
257
258#[async_trait]
259pub trait JobLifecycleObserver: Send + Sync {
260    async fn on_job_running(&self, _event: JobRunningEvent) {}
261
262    async fn on_job_continued(&self, _event: JobContinuedEvent) {}
263
264    async fn on_job_succeeded(&self, _event: JobSucceededEvent) {}
265
266    async fn on_job_failed(&self, _event: JobFailedEvent) {}
267
268    async fn on_job_completion_persist_failed(&self, _event: JobCompletionPersistFailedEvent) {}
269
270    async fn on_job_lease_lost(&self, _event: JobLeaseLostEvent) {}
271
272    async fn on_job_lease_reaped(&self, _event: JobLeaseReapedEvent) {}
273}
274
275#[derive(Clone, Default)]
276pub struct JobLifecycleObservers {
277    observers: Arc<Vec<Arc<dyn JobLifecycleObserver>>>,
278}
279
280impl JobLifecycleObservers {
281    #[must_use]
282    pub fn empty() -> Self {
283        Self::default()
284    }
285
286    #[must_use]
287    pub fn from_observer(observer: impl JobLifecycleObserver + 'static) -> Self {
288        Self {
289            observers: Arc::new(vec![Arc::new(observer)]),
290        }
291    }
292
293    #[must_use]
294    pub fn from_arc_observers(observers: Vec<Arc<dyn JobLifecycleObserver>>) -> Self {
295        Self {
296            observers: Arc::new(observers),
297        }
298    }
299
300    pub(crate) fn is_empty(&self) -> bool {
301        self.observers.is_empty()
302    }
303
304    pub(crate) async fn job_running(&self, event: JobRunningEvent) {
305        let job = event.job.clone();
306        self.notify_all_observers(
307            "on_job_running",
308            event,
309            &job,
310            |observer, event| async move {
311                observer.on_job_running(event).await;
312            },
313        )
314        .await;
315    }
316
317    pub(crate) async fn job_succeeded(&self, event: JobSucceededEvent) {
318        let job = event.job.clone();
319        self.notify_all_observers(
320            "on_job_succeeded",
321            event,
322            &job,
323            |observer, event| async move {
324                observer.on_job_succeeded(event).await;
325            },
326        )
327        .await;
328    }
329
330    pub(crate) async fn job_continued(&self, event: JobContinuedEvent) {
331        let job = event.job.clone();
332        self.notify_all_observers(
333            "on_job_continued",
334            event,
335            &job,
336            |observer, event| async move {
337                observer.on_job_continued(event).await;
338            },
339        )
340        .await;
341    }
342
343    pub(crate) async fn job_failed(&self, event: JobFailedEvent) {
344        let job = event.job.clone();
345        self.notify_all_observers("on_job_failed", event, &job, |observer, event| async move {
346            observer.on_job_failed(event).await;
347        })
348        .await;
349    }
350
351    pub(crate) async fn job_completion_persist_failed(
352        &self,
353        event: JobCompletionPersistFailedEvent,
354    ) {
355        let job = event.job.clone();
356        self.notify_all_observers(
357            "on_job_completion_persist_failed",
358            event,
359            &job,
360            |observer, event| async move {
361                observer.on_job_completion_persist_failed(event).await;
362            },
363        )
364        .await;
365    }
366
367    pub(crate) async fn job_lease_lost(&self, event: JobLeaseLostEvent) {
368        let job = event.job.clone();
369        self.notify_all_observers(
370            "on_job_lease_lost",
371            event,
372            &job,
373            |observer, event| async move {
374                observer.on_job_lease_lost(event).await;
375            },
376        )
377        .await;
378    }
379
380    pub(crate) async fn job_lease_reaped(&self, event: JobLeaseReapedEvent) {
381        let job = event.job.clone();
382        self.notify_all_observers(
383            "on_job_lease_reaped",
384            event,
385            &job,
386            |observer, event| async move {
387                observer.on_job_lease_reaped(event).await;
388            },
389        )
390        .await;
391    }
392
393    async fn notify_all_observers<E, F, Fut>(
394        &self,
395        callback_name: &'static str,
396        event: E,
397        job: &ObservedJob,
398        notify: F,
399    ) where
400        E: Clone,
401        F: Fn(Arc<dyn JobLifecycleObserver>, E) -> Fut,
402        Fut: Future<Output = ()> + Send,
403    {
404        let mut pending = FuturesUnordered::new();
405
406        for observer in self.observers.iter() {
407            let observer = Arc::clone(observer);
408            let job = ObserverJobLogContext::from(job);
409            let event = event.clone();
410            pending.push(notify_observer(callback_name, job, notify(observer, event)));
411        }
412
413        while pending.next().await.is_some() {}
414    }
415}
416
417#[derive(Debug)]
418struct ObserverJobLogContext {
419    job_id: Uuid,
420    job_type: String,
421    organization_id: Option<Uuid>,
422    run_number: i32,
423    attempt: i32,
424    max_attempts: i32,
425    worker_id: String,
426}
427
428impl From<&ObservedJob> for ObserverJobLogContext {
429    fn from(job: &ObservedJob) -> Self {
430        Self {
431            job_id: job.job_id,
432            job_type: job.job_type.to_string(),
433            organization_id: job.organization_id,
434            run_number: job.run_number,
435            attempt: job.attempt,
436            max_attempts: job.max_attempts,
437            worker_id: job.worker_id.clone(),
438        }
439    }
440}
441
442async fn notify_observer<F>(callback_name: &'static str, job: ObserverJobLogContext, future: F)
443where
444    F: Future<Output = ()> + Send,
445{
446    match tokio::time::timeout(OBSERVER_TIMEOUT, AssertUnwindSafe(future).catch_unwind()).await {
447        Ok(Ok(())) => {}
448        Ok(Err(panic_payload)) => {
449            let panic_message = panic_payload_message(&*panic_payload);
450            warn!(
451                callback_name,
452                job_id = %job.job_id,
453                job_type = %job.job_type,
454                organization_id = ?job.organization_id,
455                run_number = job.run_number,
456                attempt = job.attempt,
457                max_attempts = job.max_attempts,
458                worker_id = %job.worker_id,
459                panic = %panic_message,
460                "job lifecycle observer panicked"
461            );
462        }
463        Err(_) => {
464            warn!(
465                callback_name,
466                job_id = %job.job_id,
467                job_type = %job.job_type,
468                organization_id = ?job.organization_id,
469                run_number = job.run_number,
470                attempt = job.attempt,
471                max_attempts = job.max_attempts,
472                worker_id = %job.worker_id,
473                timeout_ms = OBSERVER_TIMEOUT.as_millis(),
474                "job lifecycle observer timed out"
475            );
476        }
477    }
478}
479
480fn panic_payload_message(panic_payload: &(dyn Any + Send)) -> String {
481    if let Some(message) = panic_payload.downcast_ref::<String>() {
482        return message.clone();
483    }
484
485    if let Some(message) = panic_payload.downcast_ref::<&'static str>() {
486        return (*message).to_string();
487    }
488
489    "non-string panic payload".to_string()
490}
runledger_runtime/observer.rs

runledger_runtime/
observer.rs
