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    Success,
187    Continuation,
188    Failure,
189}
190
191#[derive(Debug, Clone)]
192#[non_exhaustive]
193pub struct JobCompletionPersistFailedEvent {
194    pub job: ObservedJob,
195    pub duration: Duration,
196    pub operation: JobCompletionPersistenceOperation,
197    pub error: String,
198}
199
200impl JobCompletionPersistFailedEvent {
201    #[must_use]
202    pub fn new(
203        job: ObservedJob,
204        duration: Duration,
205        operation: JobCompletionPersistenceOperation,
206        error: impl Into<String>,
207    ) -> Self {
208        Self {
209            job,
210            duration,
211            operation,
212            error: error.into(),
213        }
214    }
215}
216
217#[derive(Debug, Clone, PartialEq, Eq)]
218#[non_exhaustive]
219pub enum JobLeaseReapedDisposition {
220    ReleasedToPending,
221    RetryScheduled {
222        retry_delay_ms: i32,
223        next_run_at: DateTime<Utc>,
224    },
225    DeadLettered {
226        reason: JobDeadLetterReason,
227    },
228    Unknown,
229}
230
231#[derive(Debug, Clone)]
232#[non_exhaustive]
233pub struct JobLeaseLostEvent {
234    pub job: ObservedJob,
235    pub duration: Duration,
236    pub failure: JobFailure,
237}
238
239impl JobLeaseLostEvent {
240    #[must_use]
241    pub fn new(job: ObservedJob, duration: Duration, failure: JobFailure) -> Self {
242        Self {
243            job,
244            duration,
245            failure,
246        }
247    }
248}
249
250#[derive(Debug, Clone)]
251#[non_exhaustive]
252pub struct JobLeaseReapedEvent {
253    pub job: ObservedJob,
254    pub failure: JobFailure,
255    pub started_without_renewal_heartbeat: bool,
256    pub disposition: JobLeaseReapedDisposition,
257}
258
259impl JobLeaseReapedEvent {
260    #[must_use]
261    pub fn new(
262        job: ObservedJob,
263        failure: JobFailure,
264        started_without_renewal_heartbeat: bool,
265        disposition: JobLeaseReapedDisposition,
266    ) -> Self {
267        Self {
268            job,
269            failure,
270            started_without_renewal_heartbeat,
271            disposition,
272        }
273    }
274}
275
276#[async_trait]
277pub trait JobLifecycleObserver: Send + Sync {
278    async fn on_job_running(&self, _event: JobRunningEvent) {}
279
280    async fn on_job_continued(&self, _event: JobContinuedEvent) {}
281
282    async fn on_job_succeeded(&self, _event: JobSucceededEvent) {}
283
284    async fn on_job_failed(&self, _event: JobFailedEvent) {}
285
286    async fn on_job_completion_persist_failed(&self, _event: JobCompletionPersistFailedEvent) {}
287
288    async fn on_job_lease_lost(&self, _event: JobLeaseLostEvent) {}
289
290    async fn on_job_lease_reaped(&self, _event: JobLeaseReapedEvent) {}
291}
292
293#[derive(Clone, Default)]
294pub struct JobLifecycleObservers {
295    observers: Arc<Vec<Arc<dyn JobLifecycleObserver>>>,
296}
297
298impl JobLifecycleObservers {
299    #[must_use]
300    pub fn empty() -> Self {
301        Self::default()
302    }
303
304    #[must_use]
305    pub fn from_observer(observer: impl JobLifecycleObserver + 'static) -> Self {
306        Self {
307            observers: Arc::new(vec![Arc::new(observer)]),
308        }
309    }
310
311    #[must_use]
312    pub fn from_arc_observers(observers: Vec<Arc<dyn JobLifecycleObserver>>) -> Self {
313        Self {
314            observers: Arc::new(observers),
315        }
316    }
317
318    pub(crate) fn is_empty(&self) -> bool {
319        self.observers.is_empty()
320    }
321
322    pub(crate) async fn job_running(&self, event: JobRunningEvent) {
323        let job = event.job.clone();
324        self.notify_all_observers(
325            "on_job_running",
326            event,
327            &job,
328            |observer, event| async move {
329                observer.on_job_running(event).await;
330            },
331        )
332        .await;
333    }
334
335    pub(crate) async fn job_succeeded(&self, event: JobSucceededEvent) {
336        let job = event.job.clone();
337        self.notify_all_observers(
338            "on_job_succeeded",
339            event,
340            &job,
341            |observer, event| async move {
342                observer.on_job_succeeded(event).await;
343            },
344        )
345        .await;
346    }
347
348    pub(crate) async fn job_continued(&self, event: JobContinuedEvent) {
349        let job = event.job.clone();
350        self.notify_all_observers(
351            "on_job_continued",
352            event,
353            &job,
354            |observer, event| async move {
355                observer.on_job_continued(event).await;
356            },
357        )
358        .await;
359    }
360
361    pub(crate) async fn job_failed(&self, event: JobFailedEvent) {
362        let job = event.job.clone();
363        self.notify_all_observers("on_job_failed", event, &job, |observer, event| async move {
364            observer.on_job_failed(event).await;
365        })
366        .await;
367    }
368
369    pub(crate) async fn job_completion_persist_failed(
370        &self,
371        event: JobCompletionPersistFailedEvent,
372    ) {
373        let job = event.job.clone();
374        self.notify_all_observers(
375            "on_job_completion_persist_failed",
376            event,
377            &job,
378            |observer, event| async move {
379                observer.on_job_completion_persist_failed(event).await;
380            },
381        )
382        .await;
383    }
384
385    pub(crate) async fn job_lease_lost(&self, event: JobLeaseLostEvent) {
386        let job = event.job.clone();
387        self.notify_all_observers(
388            "on_job_lease_lost",
389            event,
390            &job,
391            |observer, event| async move {
392                observer.on_job_lease_lost(event).await;
393            },
394        )
395        .await;
396    }
397
398    pub(crate) async fn job_lease_reaped(&self, event: JobLeaseReapedEvent) {
399        let job = event.job.clone();
400        self.notify_all_observers(
401            "on_job_lease_reaped",
402            event,
403            &job,
404            |observer, event| async move {
405                observer.on_job_lease_reaped(event).await;
406            },
407        )
408        .await;
409    }
410
411    async fn notify_all_observers<E, F, Fut>(
412        &self,
413        callback_name: &'static str,
414        event: E,
415        job: &ObservedJob,
416        notify: F,
417    ) where
418        E: Clone,
419        F: Fn(Arc<dyn JobLifecycleObserver>, E) -> Fut,
420        Fut: Future<Output = ()> + Send,
421    {
422        let mut pending = FuturesUnordered::new();
423
424        for observer in self.observers.iter() {
425            let observer = Arc::clone(observer);
426            let job = ObserverJobLogContext::from(job);
427            let event = event.clone();
428            pending.push(notify_observer(callback_name, job, notify(observer, event)));
429        }
430
431        while pending.next().await.is_some() {}
432    }
433}
434
435#[derive(Debug)]
436struct ObserverJobLogContext {
437    job_id: Uuid,
438    job_type: String,
439    organization_id: Option<Uuid>,
440    run_number: i32,
441    attempt: i32,
442    max_attempts: i32,
443    worker_id: String,
444}
445
446impl From<&ObservedJob> for ObserverJobLogContext {
447    fn from(job: &ObservedJob) -> Self {
448        Self {
449            job_id: job.job_id,
450            job_type: job.job_type.to_string(),
451            organization_id: job.organization_id,
452            run_number: job.run_number,
453            attempt: job.attempt,
454            max_attempts: job.max_attempts,
455            worker_id: job.worker_id.clone(),
456        }
457    }
458}
459
460async fn notify_observer<F>(callback_name: &'static str, job: ObserverJobLogContext, future: F)
461where
462    F: Future<Output = ()> + Send,
463{
464    match tokio::time::timeout(OBSERVER_TIMEOUT, AssertUnwindSafe(future).catch_unwind()).await {
465        Ok(Ok(())) => {}
466        Ok(Err(panic_payload)) => {
467            let panic_message = panic_payload_message(&*panic_payload);
468            warn!(
469                callback_name,
470                job_id = %job.job_id,
471                job_type = %job.job_type,
472                organization_id = ?job.organization_id,
473                run_number = job.run_number,
474                attempt = job.attempt,
475                max_attempts = job.max_attempts,
476                worker_id = %job.worker_id,
477                panic = %panic_message,
478                "job lifecycle observer panicked"
479            );
480        }
481        Err(_) => {
482            warn!(
483                callback_name,
484                job_id = %job.job_id,
485                job_type = %job.job_type,
486                organization_id = ?job.organization_id,
487                run_number = job.run_number,
488                attempt = job.attempt,
489                max_attempts = job.max_attempts,
490                worker_id = %job.worker_id,
491                timeout_ms = OBSERVER_TIMEOUT.as_millis(),
492                "job lifecycle observer timed out"
493            );
494        }
495    }
496}
497
498fn panic_payload_message(panic_payload: &(dyn Any + Send)) -> String {
499    if let Some(message) = panic_payload.downcast_ref::<String>() {
500        return message.clone();
501    }
502
503    if let Some(message) = panic_payload.downcast_ref::<&'static str>() {
504        return (*message).to_string();
505    }
506
507    "non-string panic payload".to_string()
508}
runledger_runtime/observer.rs

runledger_runtime/
observer.rs
