runledger_postgres/jobs/types/
enqueue.rs

1use std::fmt;
2
3use chrono::{DateTime, Utc};
4use runledger_core::jobs::{JobStage, JobStatus, JobType, JobTypeName};
5use serde::Serialize;
6use serde_json::Value;
7use sqlx::types::Uuid;
8
9#[derive(Clone, Debug)]
10pub struct JobEnqueue<'a> {
11    pub job_type: JobType<'a>,
12    pub organization_id: Option<Uuid>,
13    pub payload: &'a Value,
14    pub priority: Option<i32>,
15    pub max_attempts: Option<i32>,
16    pub timeout_seconds: Option<i32>,
17    /// For keyed enqueues, this value is part of the stored idempotency request
18    /// snapshot. Retries must pass the same scheduled time as the original
19    /// enqueue instead of recomputing a fresh timestamp.
20    pub next_run_at: Option<DateTime<Utc>>,
21    pub idempotency_key: Option<&'a str>,
22    pub stage: Option<JobStage>,
23}
24
25#[derive(Clone, Copy, Debug, Eq, PartialEq)]
26#[non_exhaustive]
27pub enum JobEnqueueDisposition {
28    Inserted,
29    Existing,
30}
31
32/// Stable job state returned from a transactional enqueue.
33///
34/// Keyed existing rows are held under a mutation-ready row lock until the
35/// caller's transaction ends, so `status` and `run_number` describe the row
36/// protected by that transaction rather than a later unlocked lookup. This
37/// lock composes with a later mutation of the same row in the transaction.
38#[derive(Clone, Debug, Eq, PartialEq)]
39pub struct JobEnqueueOutcome {
40    pub job_id: Uuid,
41    pub status: JobStatus,
42    pub run_number: i32,
43    pub disposition: JobEnqueueDisposition,
44}
45
46/// A durable request to enqueue a job after its definition becomes available.
47///
48/// Recording an intent does not read `job_definitions` and does not create a
49/// queue row. Every intent is strictly idempotent; retries must preserve every
50/// requested enqueue field.
51#[derive(Clone, Debug)]
52pub struct JobEnqueueIntent<'a> {
53    job_type: JobType<'a>,
54    organization_id: Option<Uuid>,
55    payload: &'a Value,
56    priority: Option<i32>,
57    max_attempts: Option<i32>,
58    timeout_seconds: Option<i32>,
59    next_run_at: Option<DateTime<Utc>>,
60    idempotency_key: &'a str,
61    stage: Option<JobStage>,
62    execution_resource_key: Option<&'a str>,
63}
64
65impl<'a> JobEnqueueIntent<'a> {
66    #[must_use]
67    pub fn new(job_type: JobType<'a>, payload: &'a Value, idempotency_key: &'a str) -> Self {
68        Self {
69            job_type,
70            organization_id: None,
71            payload,
72            priority: None,
73            max_attempts: None,
74            timeout_seconds: None,
75            next_run_at: None,
76            idempotency_key,
77            stage: None,
78            execution_resource_key: None,
79        }
80    }
81
82    #[must_use]
83    pub fn with_organization_id(mut self, organization_id: Uuid) -> Self {
84        self.organization_id = Some(organization_id);
85        self
86    }
87
88    #[must_use]
89    pub fn with_priority(mut self, priority: i32) -> Self {
90        self.priority = Some(priority);
91        self
92    }
93
94    #[must_use]
95    pub fn with_max_attempts(mut self, max_attempts: i32) -> Self {
96        self.max_attempts = Some(max_attempts);
97        self
98    }
99
100    #[must_use]
101    pub fn with_timeout_seconds(mut self, timeout_seconds: i32) -> Self {
102        self.timeout_seconds = Some(timeout_seconds);
103        self
104    }
105
106    #[must_use]
107    pub fn with_next_run_at(mut self, next_run_at: DateTime<Utc>) -> Self {
108        self.next_run_at = Some(next_run_at);
109        self
110    }
111
112    #[must_use]
113    pub fn with_stage(mut self, stage: JobStage) -> Self {
114        self.stage = Some(stage);
115        self
116    }
117
118    #[must_use]
119    pub fn with_execution_resource(mut self, execution_resource_key: &'a str) -> Self {
120        self.execution_resource_key = Some(execution_resource_key);
121        self
122    }
123
124    pub(crate) fn as_job_enqueue(&self) -> JobEnqueue<'a> {
125        JobEnqueue {
126            job_type: self.job_type,
127            organization_id: self.organization_id,
128            payload: self.payload,
129            priority: self.priority,
130            max_attempts: self.max_attempts,
131            timeout_seconds: self.timeout_seconds,
132            next_run_at: self.next_run_at,
133            idempotency_key: Some(self.idempotency_key),
134            stage: self.stage,
135        }
136    }
137
138    pub(crate) fn execution_resource_key(&self) -> Option<&'a str> {
139        self.execution_resource_key
140    }
141}
142
143/// Durable lifecycle state of a job enqueue intent.
144#[derive(Clone, Copy, Debug, Eq, PartialEq)]
145#[non_exhaustive]
146pub enum JobEnqueueIntentStatus {
147    Pending,
148    Promoted,
149    Conflicted,
150}
151
152impl JobEnqueueIntentStatus {
153    #[must_use]
154    pub const fn as_db_value(self) -> &'static str {
155        match self {
156            Self::Pending => "PENDING",
157            Self::Promoted => "PROMOTED",
158            Self::Conflicted => "CONFLICTED",
159        }
160    }
161}
162
163impl std::str::FromStr for JobEnqueueIntentStatus {
164    type Err = ();
165
166    fn from_str(value: &str) -> Result<Self, Self::Err> {
167        match value {
168            "PENDING" => Ok(Self::Pending),
169            "PROMOTED" => Ok(Self::Promoted),
170            "CONFLICTED" => Ok(Self::Conflicted),
171            _ => Err(()),
172        }
173    }
174}
175
176/// Whether an intent record call inserted a row or resolved an existing retry.
177#[derive(Clone, Copy, Debug, Eq, PartialEq)]
178#[non_exhaustive]
179pub enum JobEnqueueIntentDisposition {
180    Inserted,
181    Existing,
182}
183
184/// Lifecycle state observed by an enqueue-intent record call.
185#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)]
186#[serde(tag = "state", rename_all = "snake_case")]
187#[non_exhaustive]
188pub enum JobEnqueueIntentOutcomeState {
189    Pending,
190    Promoted { job_id: Uuid },
191    Conflicted,
192}
193
194impl JobEnqueueIntentOutcomeState {
195    #[must_use]
196    pub const fn status(self) -> JobEnqueueIntentStatus {
197        match self {
198            Self::Pending => JobEnqueueIntentStatus::Pending,
199            Self::Promoted { .. } => JobEnqueueIntentStatus::Promoted,
200            Self::Conflicted => JobEnqueueIntentStatus::Conflicted,
201        }
202    }
203
204    #[must_use]
205    pub const fn promoted_job_id(self) -> Option<Uuid> {
206        match self {
207            Self::Promoted { job_id } => Some(job_id),
208            Self::Pending | Self::Conflicted => None,
209        }
210    }
211}
212
213/// Point-in-time state observed by an intent record call.
214///
215/// An existing intent is protected from deletion while a caller-owned record
216/// transaction remains open, but promotion may still update its non-key
217/// lifecycle fields. `state` can therefore change after it is read, including
218/// before the caller transaction ends. Monitor intent metrics and use the read
219/// APIs when current lifecycle state is required.
220#[derive(Clone, Debug, Eq, PartialEq)]
221#[must_use = "callers must inspect the observed pending, promoted, or conflicted state"]
222#[non_exhaustive]
223pub struct JobEnqueueIntentOutcome {
224    pub intent_id: Uuid,
225    pub state: JobEnqueueIntentOutcomeState,
226    pub disposition: JobEnqueueIntentDisposition,
227}
228
229impl JobEnqueueIntentOutcome {
230    #[must_use]
231    pub const fn status(&self) -> JobEnqueueIntentStatus {
232        self.state.status()
233    }
234
235    #[must_use]
236    pub const fn promoted_job_id(&self) -> Option<Uuid> {
237        self.state.promoted_job_id()
238    }
239}
240
241/// Stable error metadata from an enqueue-intent promotion attempt.
242#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
243pub struct JobEnqueueIntentPromotionError {
244    code: String,
245    message: String,
246}
247
248impl JobEnqueueIntentPromotionError {
249    pub(crate) fn new(code: String, message: String) -> Self {
250        Self { code, message }
251    }
252
253    #[must_use]
254    pub fn code(&self) -> &str {
255        &self.code
256    }
257
258    #[must_use]
259    pub fn message(&self) -> &str {
260        &self.message
261    }
262}
263
264/// Correlated lifecycle fields decoded from one persisted enqueue-intent row.
265#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
266#[serde(tag = "state", rename_all = "snake_case")]
267#[non_exhaustive]
268pub enum JobEnqueueIntentState {
269    InitialPending,
270    RetryPending {
271        promotion_attempts: i32,
272        last_attempted_at: DateTime<Utc>,
273        error: JobEnqueueIntentPromotionError,
274    },
275    Promoted {
276        promotion_attempts: i32,
277        last_attempted_at: DateTime<Utc>,
278        job_id: Uuid,
279        promoted_at: DateTime<Utc>,
280    },
281    Conflicted {
282        promotion_attempts: i32,
283        last_attempted_at: DateTime<Utc>,
284        conflicted_at: DateTime<Utc>,
285        error: JobEnqueueIntentPromotionError,
286    },
287}
288
289impl JobEnqueueIntentState {
290    #[must_use]
291    pub const fn status(&self) -> JobEnqueueIntentStatus {
292        match self {
293            Self::InitialPending | Self::RetryPending { .. } => JobEnqueueIntentStatus::Pending,
294            Self::Promoted { .. } => JobEnqueueIntentStatus::Promoted,
295            Self::Conflicted { .. } => JobEnqueueIntentStatus::Conflicted,
296        }
297    }
298
299    #[must_use]
300    pub const fn promotion_attempts(&self) -> i32 {
301        match self {
302            Self::InitialPending => 0,
303            Self::RetryPending {
304                promotion_attempts, ..
305            }
306            | Self::Promoted {
307                promotion_attempts, ..
308            }
309            | Self::Conflicted {
310                promotion_attempts, ..
311            } => *promotion_attempts,
312        }
313    }
314
315    #[must_use]
316    pub const fn last_attempted_at(&self) -> Option<DateTime<Utc>> {
317        match self {
318            Self::InitialPending => None,
319            Self::RetryPending {
320                last_attempted_at, ..
321            }
322            | Self::Promoted {
323                last_attempted_at, ..
324            }
325            | Self::Conflicted {
326                last_attempted_at, ..
327            } => Some(*last_attempted_at),
328        }
329    }
330
331    #[must_use]
332    pub const fn promoted_job_id(&self) -> Option<Uuid> {
333        match self {
334            Self::Promoted { job_id, .. } => Some(*job_id),
335            Self::InitialPending | Self::RetryPending { .. } | Self::Conflicted { .. } => None,
336        }
337    }
338
339    #[must_use]
340    pub const fn promoted_at(&self) -> Option<DateTime<Utc>> {
341        match self {
342            Self::Promoted { promoted_at, .. } => Some(*promoted_at),
343            Self::InitialPending | Self::RetryPending { .. } | Self::Conflicted { .. } => None,
344        }
345    }
346
347    #[must_use]
348    pub const fn conflicted_at(&self) -> Option<DateTime<Utc>> {
349        match self {
350            Self::Conflicted { conflicted_at, .. } => Some(*conflicted_at),
351            Self::InitialPending | Self::RetryPending { .. } | Self::Promoted { .. } => None,
352        }
353    }
354
355    #[must_use]
356    pub const fn promotion_error(&self) -> Option<&JobEnqueueIntentPromotionError> {
357        match self {
358            Self::RetryPending { error, .. } | Self::Conflicted { error, .. } => Some(error),
359            Self::InitialPending | Self::Promoted { .. } => None,
360        }
361    }
362}
363
364/// Persisted enqueue intent returned by lookup and list APIs.
365#[derive(Clone, Debug, Eq, PartialEq)]
366#[non_exhaustive]
367pub struct JobEnqueueIntentRecord {
368    pub id: Uuid,
369    pub job_type: JobTypeName,
370    pub organization_id: Option<Uuid>,
371    pub payload: Value,
372    pub priority: Option<i32>,
373    pub max_attempts: Option<i32>,
374    pub timeout_seconds: Option<i32>,
375    pub next_run_at: Option<DateTime<Utc>>,
376    pub idempotency_key: String,
377    pub stage: JobStage,
378    pub enqueue_request_version: i16,
379    pub execution_resource_key: Option<String>,
380    pub next_promotion_at: DateTime<Utc>,
381    pub state: JobEnqueueIntentState,
382    pub created_at: DateTime<Utc>,
383    pub updated_at: DateTime<Utc>,
384}
385
386impl JobEnqueueIntentRecord {
387    #[must_use]
388    pub const fn status(&self) -> JobEnqueueIntentStatus {
389        self.state.status()
390    }
391
392    #[must_use]
393    pub const fn promotion_attempts(&self) -> i32 {
394        self.state.promotion_attempts()
395    }
396
397    #[must_use]
398    pub const fn last_attempted_at(&self) -> Option<DateTime<Utc>> {
399        self.state.last_attempted_at()
400    }
401
402    #[must_use]
403    pub const fn promoted_job_id(&self) -> Option<Uuid> {
404        self.state.promoted_job_id()
405    }
406
407    #[must_use]
408    pub const fn promoted_at(&self) -> Option<DateTime<Utc>> {
409        self.state.promoted_at()
410    }
411
412    #[must_use]
413    pub const fn conflicted_at(&self) -> Option<DateTime<Utc>> {
414        self.state.conflicted_at()
415    }
416
417    #[must_use]
418    pub const fn promotion_error(&self) -> Option<&JobEnqueueIntentPromotionError> {
419        self.state.promotion_error()
420    }
421}
422
423/// Bounded filters for listing durable enqueue intents.
424#[derive(Clone, Debug)]
425pub struct JobEnqueueIntentListFilter<'a> {
426    pub(crate) organization_id: Option<Uuid>,
427    pub(crate) status: Option<JobEnqueueIntentStatus>,
428    pub(crate) job_type_query: Option<&'a str>,
429    pub(crate) limit: i64,
430    pub(crate) offset: i64,
431}
432
433impl<'a> JobEnqueueIntentListFilter<'a> {
434    #[must_use]
435    pub const fn new(limit: i64, offset: i64) -> Self {
436        Self {
437            organization_id: None,
438            status: None,
439            job_type_query: None,
440            limit,
441            offset,
442        }
443    }
444
445    #[must_use]
446    pub const fn with_organization_id(mut self, organization_id: Uuid) -> Self {
447        self.organization_id = Some(organization_id);
448        self
449    }
450
451    #[must_use]
452    pub const fn with_status(mut self, status: JobEnqueueIntentStatus) -> Self {
453        self.status = Some(status);
454        self
455    }
456
457    /// Filters by a case-insensitive job-type substring.
458    ///
459    /// PostgreSQL `ILIKE` metacharacters in `job_type_query` retain their
460    /// normal wildcard meaning, matching the crate's other admin filters.
461    #[must_use]
462    pub const fn with_job_type_query(mut self, job_type_query: &'a str) -> Self {
463        self.job_type_query = Some(job_type_query);
464        self
465    }
466}
467
468/// Bounded filters for durable enqueue-intent metrics grouped by job type.
469#[derive(Clone, Debug)]
470pub struct JobEnqueueIntentMetricsFilter<'a> {
471    pub(crate) organization_id: Option<Uuid>,
472    pub(crate) job_type: Option<JobType<'a>>,
473    pub(crate) limit: i64,
474    pub(crate) offset: i64,
475}
476
477impl<'a> JobEnqueueIntentMetricsFilter<'a> {
478    #[must_use]
479    pub const fn new(limit: i64, offset: i64) -> Self {
480        Self {
481            organization_id: None,
482            job_type: None,
483            limit,
484            offset,
485        }
486    }
487
488    #[must_use]
489    pub const fn with_organization_id(mut self, organization_id: Uuid) -> Self {
490        self.organization_id = Some(organization_id);
491        self
492    }
493
494    #[must_use]
495    pub const fn with_job_type(mut self, job_type: JobType<'a>) -> Self {
496        self.job_type = Some(job_type);
497        self
498    }
499}
500
501/// Operational backlog signals for one intent job type.
502#[derive(Clone, Debug, Eq, PartialEq)]
503#[non_exhaustive]
504pub struct JobEnqueueIntentMetricsRecord {
505    /// Canonical job type shared by every population in this record.
506    pub job_type: JobTypeName,
507    /// Number of intents currently awaiting promotion.
508    pub pending_count: i64,
509    /// Number of pending intents whose promotion has already been attempted.
510    pub retrying_count: i64,
511    /// Highest promotion-attempt count among pending intents.
512    ///
513    /// This is zero when the returned job type has no pending intents.
514    pub max_promotion_attempts: i32,
515    /// Number of intents that entered the terminal conflicted state during the
516    /// preceding 24 hours.
517    pub conflicted_24h: i64,
518    /// Number of retained intents promoted during the preceding 24 hours.
519    pub promoted_24h: i64,
520    /// Creation time of the oldest pending intent, if one exists.
521    pub oldest_pending_at: Option<DateTime<Utc>>,
522}
523
524/// Results from one bounded intent promotion pass.
525#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
526#[non_exhaustive]
527pub struct JobEnqueueIntentPromotionReport {
528    /// Intents that created a new queue row.
529    pub inserted_jobs: u64,
530    /// Intents that converged on an existing idempotent queue row.
531    pub existing_jobs: u64,
532    /// Intents that reached a terminal idempotency conflict.
533    pub conflicted: u64,
534    /// Intents whose enabled definition became unavailable after selection.
535    ///
536    /// Missing, disabled, unsupported-version, and unregistered intent rows are
537    /// not eligible for a promotion pass and therefore are not counted here.
538    /// Monitor pending count and age for those populations.
539    pub definition_became_unavailable: u64,
540    /// Intents deferred after a retryable persistence failure.
541    pub retry_deferred: u64,
542    /// Sum of newly inserted and existing jobs linked to promoted intents.
543    pub total_promoted: u64,
544    batch_was_full: bool,
545}
546
547impl JobEnqueueIntentPromotionReport {
548    /// Returns whether the storage layer claimed its effective per-transaction
549    /// limit, indicating that immediately eligible work may remain.
550    #[must_use]
551    pub const fn batch_was_full(&self) -> bool {
552        self.batch_was_full
553    }
554
555    pub(in crate::jobs) fn mark_batch_size(&mut self, claimed: usize, limit: i64) {
556        self.batch_was_full = usize::try_from(limit).is_ok_and(|limit| claimed == limit);
557    }
558}
559
560/// Exact tenant scope for a job mutation.
561#[derive(Clone, Copy, Debug, Eq, PartialEq)]
562pub enum JobScope {
563    /// Match only a job whose `organization_id` is `NULL`.
564    Global,
565    /// Match only a job owned by this exact organization.
566    Organization(Uuid),
567}
568
569impl JobScope {
570    #[must_use]
571    pub const fn organization_id(self) -> Option<Uuid> {
572        match self {
573            Self::Global => None,
574            Self::Organization(organization_id) => Some(organization_id),
575        }
576    }
577}
578
579/// Terminal job statuses that may be recovered through compare-and-requeue.
580///
581/// `SUCCEEDED` is deliberately absent: replaying successful work requires a
582/// separate policy decision and cannot be requested through this type.
583#[derive(Clone, Copy, Debug, Eq, PartialEq)]
584pub enum RequeueableJobStatus {
585    DeadLettered,
586    Canceled,
587}
588
589impl RequeueableJobStatus {
590    #[must_use]
591    pub const fn as_job_status(self) -> JobStatus {
592        match self {
593            Self::DeadLettered => JobStatus::DeadLettered,
594            Self::Canceled => JobStatus::Canceled,
595        }
596    }
597
598    #[must_use]
599    pub const fn as_db_value(self) -> &'static str {
600        self.as_job_status().as_db_value()
601    }
602}
603
604/// Error returned when a job observation cannot seed compare-and-requeue.
605#[derive(Clone, Copy, Debug, Eq, PartialEq)]
606pub struct NonRequeueableJobStatusError {
607    status: JobStatus,
608}
609
610impl NonRequeueableJobStatusError {
611    /// The observed status that compare-and-requeue does not accept.
612    #[must_use]
613    pub const fn status(&self) -> JobStatus {
614        self.status
615    }
616}
617
618impl fmt::Display for NonRequeueableJobStatusError {
619    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
620        write!(
621            f,
622            "job status {} cannot be compare-and-requeued; expected CANCELED or DEAD_LETTERED",
623            self.status.as_db_value()
624        )
625    }
626}
627
628impl std::error::Error for NonRequeueableJobStatusError {}
629
630impl TryFrom<JobStatus> for RequeueableJobStatus {
631    type Error = NonRequeueableJobStatusError;
632
633    fn try_from(status: JobStatus) -> Result<Self, Self::Error> {
634        match status {
635            JobStatus::DeadLettered => Ok(Self::DeadLettered),
636            JobStatus::Canceled => Ok(Self::Canceled),
637            status => Err(NonRequeueableJobStatusError { status }),
638        }
639    }
640}
641
642/// Whether compare-and-requeue carries durable resume state into the new run.
643#[derive(Clone, Copy, Debug, Eq, PartialEq)]
644pub enum JobRequeueStatePolicy {
645    /// Keep `progress_done`, `progress_total`, and `checkpoint` so recovery can
646    /// resume from the last committed position.
647    PreserveProgressAndCheckpoint,
648    /// Clear progress and checkpoint state so the new run starts from scratch.
649    ResetProgressAndCheckpoint,
650}
651
652impl JobRequeueStatePolicy {
653    #[must_use]
654    pub const fn preserves_progress_and_checkpoint(self) -> bool {
655        matches!(self, Self::PreserveProgressAndCheckpoint)
656    }
657
658    #[must_use]
659    pub const fn as_event_value(self) -> &'static str {
660        match self {
661            Self::PreserveProgressAndCheckpoint => "preserve_progress_and_checkpoint",
662            Self::ResetProgressAndCheckpoint => "reset_progress_and_checkpoint",
663        }
664    }
665
666    pub(crate) fn from_event_value(value: &str) -> Option<Self> {
667        match value {
668            "preserve_progress_and_checkpoint" => Some(Self::PreserveProgressAndCheckpoint),
669            "reset_progress_and_checkpoint" => Some(Self::ResetProgressAndCheckpoint),
670            _ => None,
671        }
672    }
673}
674
675#[derive(Clone, Debug)]
676pub struct JobQueueRecord {
677    pub id: Uuid,
678    pub job_type: JobTypeName,
679    pub organization_id: Option<Uuid>,
680    pub payload: Value,
681    pub status: JobStatus,
682    pub priority: i32,
683    pub run_number: i32,
684    pub attempt: i32,
685    pub max_attempts: i32,
686    pub timeout_seconds: i32,
687    pub next_run_at: DateTime<Utc>,
688    pub lease_expires_at: Option<DateTime<Utc>>,
689    pub last_heartbeat_at: Option<DateTime<Utc>>,
690    pub worker_id: Option<String>,
691    pub started_at: Option<DateTime<Utc>>,
692    pub finished_at: Option<DateTime<Utc>>,
693    pub stage: JobStage,
694    pub progress_done: Option<i64>,
695    pub progress_total: Option<i64>,
696    pub progress_pct: Option<f64>,
697    pub checkpoint: Option<Value>,
698    pub output: Option<Value>,
699    pub idempotency_key: Option<String>,
700    pub status_reason: Option<String>,
701    pub last_error_code: Option<String>,
702    pub last_error_message: Option<String>,
703    pub created_at: DateTime<Utc>,
704    pub updated_at: DateTime<Utc>,
705}
706
707#[derive(Clone, Debug)]
708pub struct CompareAndRequeueJob<'a> {
709    pub scope: JobScope,
710    pub job_id: Uuid,
711    pub expected_status: RequeueableJobStatus,
712    pub expected_run_number: i32,
713    pub state_policy: JobRequeueStatePolicy,
714    pub reason: &'a str,
715}
716
717impl<'a> CompareAndRequeueJob<'a> {
718    /// Builds a compare-and-requeue request from an observed terminal job.
719    ///
720    /// The job's exact tenant scope, identifier, status, and run number are
721    /// copied into the request so callers cannot accidentally turn a scoped
722    /// observation into a wildcard or lose the optimistic-concurrency fence.
723    ///
724    /// # Errors
725    /// Returns [`NonRequeueableJobStatusError`] unless the observation is
726    /// canceled or dead-lettered. Successful jobs require a separate replay
727    /// policy and pending or leased jobs are not recovery candidates.
728    pub fn from_observed_job(
729        observed: &JobQueueRecord,
730        state_policy: JobRequeueStatePolicy,
731        reason: &'a str,
732    ) -> Result<Self, NonRequeueableJobStatusError> {
733        let expected_status = RequeueableJobStatus::try_from(observed.status)?;
734        let scope = observed
735            .organization_id
736            .map_or(JobScope::Global, JobScope::Organization);
737
738        Ok(Self {
739            scope,
740            job_id: observed.id,
741            expected_status,
742            expected_run_number: observed.run_number,
743            state_policy,
744            reason,
745        })
746    }
747}
748
749#[derive(Clone, Debug)]
750#[must_use = "callers must inspect whether the expected job was requeued"]
751#[non_exhaustive]
752pub enum CompareAndRequeueJobOutcome {
753    Requeued {
754        before: Box<JobQueueRecord>,
755        after: Box<JobQueueRecord>,
756        event_id: i64,
757    },
758    ExpectationMismatch {
759        actual: Box<JobQueueRecord>,
760    },
761    /// Cancellation fenced a live handler, but its original lease window has
762    /// not passed yet. Retrying before `retry_after` could overlap the new run
763    /// with the canceled handler's external side effects.
764    CancellationNotQuiesced {
765        actual: Box<JobQueueRecord>,
766        retry_after: DateTime<Utc>,
767    },
768    NotFound,
769}
runledger_postgres/jobs/types/enqueue.rs

runledger_postgres/jobs/types/
enqueue.rs
