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/// The failure transition that was durably committed before observer delivery.
127///
128/// This is authoritative for the effective retry schedule. The timing retained
129/// on [`JobFailedEvent::failure`] is the handler's request, which may be ignored
130/// when the failure is dead-lettered.
131#[derive(Debug, Clone, PartialEq, Eq)]
132#[non_exhaustive]
133pub enum JobFailureDisposition {
134    /// Another attempt was scheduled from a relative delay.
135    RetryScheduled {
136        /// Persisted positive delay, rounded up to millisecond precision.
137        retry_delay_ms: i32,
138        /// Effective claim time calculated from the PostgreSQL completion clock.
139        next_run_at: DateTime<Utc>,
140    },
141    /// The handler's lower bound selected the effective retry schedule.
142    RetryScheduledAt {
143        /// Handler not-before time, rounded up to PostgreSQL microsecond
144        /// precision.
145        requested_retry_at: DateTime<Utc>,
146        /// Effective claim time. This is never earlier than policy backoff.
147        next_run_at: DateTime<Utc>,
148    },
149    /// No retry was scheduled and the job was dead-lettered.
150    DeadLettered { reason: JobDeadLetterReason },
151    /// A future persistence disposition unknown to this runtime version.
152    Unknown,
153}
154
155#[derive(Debug, Clone)]
156#[non_exhaustive]
157pub struct JobFailedEvent {
158    pub job: ObservedJob,
159    pub duration: Duration,
160    /// Handler failure, including any requested retry timing.
161    pub failure: JobFailure,
162    /// Authoritative post-commit retry or dead-letter outcome.
163    pub disposition: JobFailureDisposition,
164}
165
166impl JobFailedEvent {
167    #[must_use]
168    pub fn new(
169        job: ObservedJob,
170        duration: Duration,
171        failure: JobFailure,
172        disposition: JobFailureDisposition,
173    ) -> Self {
174        Self {
175            job,
176            duration,
177            failure,
178            disposition,
179        }
180    }
181}
182
183#[derive(Debug, Clone, Copy, PartialEq, Eq)]
184#[non_exhaustive]
185pub enum JobCompletionPersistenceOperation {
186    /// Terminal-success persistence failed.
187    Success,
188    /// Successful handler-continuation persistence failed.
189    Continuation,
190    /// Failure/retry/dead-letter persistence failed.
191    Failure,
192}
193
194#[derive(Debug, Clone)]
195#[non_exhaustive]
196pub struct JobCompletionPersistFailedEvent {
197    pub job: ObservedJob,
198    pub duration: Duration,
199    pub operation: JobCompletionPersistenceOperation,
200    pub error: String,
201}
202
203impl JobCompletionPersistFailedEvent {
204    #[must_use]
205    pub fn new(
206        job: ObservedJob,
207        duration: Duration,
208        operation: JobCompletionPersistenceOperation,
209        error: impl Into<String>,
210    ) -> Self {
211        Self {
212            job,
213            duration,
214            operation,
215            error: error.into(),
216        }
217    }
218}
219
220#[derive(Debug, Clone, PartialEq, Eq)]
221#[non_exhaustive]
222pub enum JobLeaseReapedDisposition {
223    ReleasedToPending,
224    RetryScheduled {
225        retry_delay_ms: i32,
226        next_run_at: DateTime<Utc>,
227    },
228    DeadLettered {
229        reason: JobDeadLetterReason,
230    },
231    Unknown,
232}
233
234#[derive(Debug, Clone)]
235#[non_exhaustive]
236pub struct JobLeaseLostEvent {
237    pub job: ObservedJob,
238    pub duration: Duration,
239    pub failure: JobFailure,
240}
241
242impl JobLeaseLostEvent {
243    #[must_use]
244    pub fn new(job: ObservedJob, duration: Duration, failure: JobFailure) -> Self {
245        Self {
246            job,
247            duration,
248            failure,
249        }
250    }
251}
252
253#[derive(Debug, Clone)]
254#[non_exhaustive]
255pub struct JobLeaseReapedEvent {
256    pub job: ObservedJob,
257    pub failure: JobFailure,
258    pub started_without_renewal_heartbeat: bool,
259    pub disposition: JobLeaseReapedDisposition,
260}
261
262impl JobLeaseReapedEvent {
263    #[must_use]
264    pub fn new(
265        job: ObservedJob,
266        failure: JobFailure,
267        started_without_renewal_heartbeat: bool,
268        disposition: JobLeaseReapedDisposition,
269    ) -> Self {
270        Self {
271            job,
272            failure,
273            started_without_renewal_heartbeat,
274            disposition,
275        }
276    }
277}
278
279#[async_trait]
280pub trait JobLifecycleObserver: Send + Sync {
281    async fn on_job_running(&self, _event: JobRunningEvent) {}
282
283    async fn on_job_continued(&self, _event: JobContinuedEvent) {}
284
285    async fn on_job_succeeded(&self, _event: JobSucceededEvent) {}
286
287    async fn on_job_failed(&self, _event: JobFailedEvent) {}
288
289    async fn on_job_completion_persist_failed(&self, _event: JobCompletionPersistFailedEvent) {}
290
291    async fn on_job_lease_lost(&self, _event: JobLeaseLostEvent) {}
292
293    async fn on_job_lease_reaped(&self, _event: JobLeaseReapedEvent) {}
294}
295
296#[derive(Clone, Default)]
297pub struct JobLifecycleObservers {
298    observers: Arc<Vec<Arc<dyn JobLifecycleObserver>>>,
299}
300
301impl JobLifecycleObservers {
302    #[must_use]
303    pub fn empty() -> Self {
304        Self::default()
305    }
306
307    #[must_use]
308    pub fn from_observer(observer: impl JobLifecycleObserver + 'static) -> Self {
309        Self {
310            observers: Arc::new(vec![Arc::new(observer)]),
311        }
312    }
313
314    #[must_use]
315    pub fn from_arc_observers(observers: Vec<Arc<dyn JobLifecycleObserver>>) -> Self {
316        Self {
317            observers: Arc::new(observers),
318        }
319    }
320
321    pub(crate) fn is_empty(&self) -> bool {
322        self.observers.is_empty()
323    }
324
325    pub(crate) async fn job_running(&self, event: JobRunningEvent) {
326        let job = event.job.clone();
327        self.notify_all_observers(
328            "on_job_running",
329            event,
330            &job,
331            |observer, event| async move {
332                observer.on_job_running(event).await;
333            },
334        )
335        .await;
336    }
337
338    pub(crate) async fn job_succeeded(&self, event: JobSucceededEvent) {
339        let job = event.job.clone();
340        self.notify_all_observers(
341            "on_job_succeeded",
342            event,
343            &job,
344            |observer, event| async move {
345                observer.on_job_succeeded(event).await;
346            },
347        )
348        .await;
349    }
350
351    pub(crate) async fn job_continued(&self, event: JobContinuedEvent) {
352        let job = event.job.clone();
353        self.notify_all_observers(
354            "on_job_continued",
355            event,
356            &job,
357            |observer, event| async move {
358                observer.on_job_continued(event).await;
359            },
360        )
361        .await;
362    }
363
364    pub(crate) async fn job_failed(&self, event: JobFailedEvent) {
365        let job = event.job.clone();
366        self.notify_all_observers("on_job_failed", event, &job, |observer, event| async move {
367            observer.on_job_failed(event).await;
368        })
369        .await;
370    }
371
372    pub(crate) async fn job_completion_persist_failed(
373        &self,
374        event: JobCompletionPersistFailedEvent,
375    ) {
376        let job = event.job.clone();
377        self.notify_all_observers(
378            "on_job_completion_persist_failed",
379            event,
380            &job,
381            |observer, event| async move {
382                observer.on_job_completion_persist_failed(event).await;
383            },
384        )
385        .await;
386    }
387
388    pub(crate) async fn job_lease_lost(&self, event: JobLeaseLostEvent) {
389        let job = event.job.clone();
390        self.notify_all_observers(
391            "on_job_lease_lost",
392            event,
393            &job,
394            |observer, event| async move {
395                observer.on_job_lease_lost(event).await;
396            },
397        )
398        .await;
399    }
400
401    pub(crate) async fn job_lease_reaped(&self, event: JobLeaseReapedEvent) {
402        let job = event.job.clone();
403        self.notify_all_observers(
404            "on_job_lease_reaped",
405            event,
406            &job,
407            |observer, event| async move {
408                observer.on_job_lease_reaped(event).await;
409            },
410        )
411        .await;
412    }
413
414    async fn notify_all_observers<E, F, Fut>(
415        &self,
416        callback_name: &'static str,
417        event: E,
418        job: &ObservedJob,
419        notify: F,
420    ) where
421        E: Clone,
422        F: Fn(Arc<dyn JobLifecycleObserver>, E) -> Fut,
423        Fut: Future<Output = ()> + Send,
424    {
425        let mut pending = FuturesUnordered::new();
426
427        for observer in self.observers.iter() {
428            let observer = Arc::clone(observer);
429            let job = ObserverJobLogContext::from(job);
430            let event = event.clone();
431            pending.push(notify_observer(callback_name, job, notify(observer, event)));
432        }
433
434        while pending.next().await.is_some() {}
435    }
436}
437
438#[derive(Debug)]
439struct ObserverJobLogContext {
440    job_id: Uuid,
441    job_type: String,
442    organization_id: Option<Uuid>,
443    run_number: i32,
444    attempt: i32,
445    max_attempts: i32,
446    worker_id: String,
447}
448
449impl From<&ObservedJob> for ObserverJobLogContext {
450    fn from(job: &ObservedJob) -> Self {
451        Self {
452            job_id: job.job_id,
453            job_type: job.job_type.to_string(),
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.clone(),
459        }
460    }
461}
462
463async fn notify_observer<F>(callback_name: &'static str, job: ObserverJobLogContext, future: F)
464where
465    F: Future<Output = ()> + Send,
466{
467    match tokio::time::timeout(OBSERVER_TIMEOUT, AssertUnwindSafe(future).catch_unwind()).await {
468        Ok(Ok(())) => {}
469        Ok(Err(panic_payload)) => {
470            let panic_message = panic_payload_message(&*panic_payload);
471            warn!(
472                callback_name,
473                job_id = %job.job_id,
474                job_type = %job.job_type,
475                organization_id = ?job.organization_id,
476                run_number = job.run_number,
477                attempt = job.attempt,
478                max_attempts = job.max_attempts,
479                worker_id = %job.worker_id,
480                panic = %panic_message,
481                "job lifecycle observer panicked"
482            );
483        }
484        Err(_) => {
485            warn!(
486                callback_name,
487                job_id = %job.job_id,
488                job_type = %job.job_type,
489                organization_id = ?job.organization_id,
490                run_number = job.run_number,
491                attempt = job.attempt,
492                max_attempts = job.max_attempts,
493                worker_id = %job.worker_id,
494                timeout_ms = OBSERVER_TIMEOUT.as_millis(),
495                "job lifecycle observer timed out"
496            );
497        }
498    }
499}
500
501fn panic_payload_message(panic_payload: &(dyn Any + Send)) -> String {
502    if let Some(message) = panic_payload.downcast_ref::<String>() {
503        return message.clone();
504    }
505
506    if let Some(message) = panic_payload.downcast_ref::<&'static str>() {
507        return (*message).to_string();
508    }
509
510    "non-string panic payload".to_string()
511}
runledger_runtime/observer.rs

runledger_runtime/
observer.rs
