awa_model/
dlq.rs

1//! Dead-letter queue helpers for the queue_storage backend.
2//!
3//! The DLQ is part of the queue_storage engine. The public API resolves the
4//! active queue_storage schema from `awa.runtime_storage_backends` and operates
5//! against that backend.
6
7use crate::error::AwaError;
8use crate::job::JobRow;
9use crate::queue_storage::QueueStorage;
10use chrono::{DateTime, Utc};
11use serde::{Deserialize, Serialize};
12use sqlx::PgPool;
13use std::time::Duration;
14
15#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
16pub struct DlqMetadata {
17    pub reason: String,
18    pub dlq_at: DateTime<Utc>,
19    pub original_run_lease: i64,
20}
21
22#[derive(Debug, Clone, Serialize, Deserialize)]
23pub struct DlqRow {
24    #[serde(flatten)]
25    pub job: JobRow,
26    #[serde(rename = "dlq_reason")]
27    pub reason: String,
28    pub dlq_at: DateTime<Utc>,
29    pub original_run_lease: i64,
30}
31
32impl DlqRow {
33    pub fn metadata(&self) -> DlqMetadata {
34        DlqMetadata {
35            reason: self.reason.clone(),
36            dlq_at: self.dlq_at,
37            original_run_lease: self.original_run_lease,
38        }
39    }
40
41    pub fn into_parts(self) -> (JobRow, DlqMetadata) {
42        let metadata = DlqMetadata {
43            reason: self.reason,
44            dlq_at: self.dlq_at,
45            original_run_lease: self.original_run_lease,
46        };
47        (self.job, metadata)
48    }
49}
50
51#[derive(Debug, Clone, Default, Serialize, Deserialize)]
52pub struct ListDlqFilter {
53    pub kind: Option<String>,
54    pub queue: Option<String>,
55    pub tag: Option<String>,
56    pub before_id: Option<i64>,
57    pub before_dlq_at: Option<DateTime<Utc>>,
58    pub limit: Option<i64>,
59}
60
61#[derive(Debug, Clone, Default, Serialize, Deserialize)]
62pub struct RetryFromDlqOpts {
63    pub run_at: Option<DateTime<Utc>>,
64    pub priority: Option<i16>,
65    pub queue: Option<String>,
66}
67
68#[derive(Debug, Clone, sqlx::FromRow)]
69struct QueueStorageDlqRow {
70    job_id: i64,
71    kind: String,
72    queue: String,
73    args: serde_json::Value,
74    state: crate::JobState,
75    priority: i16,
76    attempt: i16,
77    run_lease: i64,
78    max_attempts: i16,
79    run_at: DateTime<Utc>,
80    attempted_at: Option<DateTime<Utc>>,
81    finalized_at: DateTime<Utc>,
82    created_at: DateTime<Utc>,
83    unique_key: Option<Vec<u8>>,
84    payload: serde_json::Value,
85    dlq_reason: String,
86    dlq_at: DateTime<Utc>,
87    original_run_lease: i64,
88}
89
90#[derive(Debug, Clone, Serialize, Deserialize)]
91struct QueueStoragePayload {
92    #[serde(default = "default_queue_storage_payload_metadata")]
93    metadata: serde_json::Value,
94    #[serde(default)]
95    tags: Vec<String>,
96    #[serde(default)]
97    errors: Vec<serde_json::Value>,
98    #[serde(default)]
99    progress: Option<serde_json::Value>,
100}
101
102fn default_queue_storage_payload_metadata() -> serde_json::Value {
103    serde_json::json!({})
104}
105
106impl Default for QueueStoragePayload {
107    fn default() -> Self {
108        Self {
109            metadata: serde_json::json!({}),
110            tags: Vec::new(),
111            errors: Vec::new(),
112            progress: None,
113        }
114    }
115}
116
117impl QueueStorageDlqRow {
118    fn into_dlq_row(self) -> Result<DlqRow, AwaError> {
119        let payload: QueueStoragePayload = serde_json::from_value(self.payload)?;
120        Ok(DlqRow {
121            job: JobRow {
122                id: self.job_id,
123                kind: self.kind,
124                queue: self.queue,
125                args: self.args,
126                state: self.state,
127                priority: self.priority,
128                attempt: self.attempt,
129                run_lease: self.run_lease,
130                max_attempts: self.max_attempts,
131                run_at: self.run_at,
132                heartbeat_at: None,
133                deadline_at: None,
134                attempted_at: self.attempted_at,
135                finalized_at: Some(self.finalized_at),
136                created_at: self.created_at,
137                errors: (!payload.errors.is_empty()).then_some(payload.errors),
138                metadata: payload.metadata,
139                tags: payload.tags,
140                unique_key: self.unique_key,
141                unique_states: None,
142                callback_id: None,
143                callback_timeout_at: None,
144                callback_filter: None,
145                callback_on_complete: None,
146                callback_on_fail: None,
147                callback_transform: None,
148                progress: payload.progress,
149            },
150            reason: self.dlq_reason,
151            dlq_at: self.dlq_at,
152            original_run_lease: self.original_run_lease,
153        })
154    }
155}
156
157async fn active_queue_storage(pool: &PgPool) -> Result<QueueStorage, AwaError> {
158    let schema = QueueStorage::active_schema(pool).await?.ok_or_else(|| {
159        AwaError::Validation(
160            "DLQ APIs currently require an active queue_storage backend".to_string(),
161        )
162    })?;
163    QueueStorage::from_existing_schema(schema)
164}
165
166fn filter_requires_scope(filter: &ListDlqFilter) -> bool {
167    filter.kind.is_none() && filter.queue.is_none() && filter.tag.is_none()
168}
169
170pub async fn move_failed_to_dlq(
171    pool: &PgPool,
172    job_id: i64,
173    reason: &str,
174) -> Result<Option<DlqRow>, AwaError> {
175    let store = active_queue_storage(pool).await?;
176    if store
177        .move_failed_to_dlq(pool, job_id, reason)
178        .await?
179        .is_none()
180    {
181        return Ok(None);
182    }
183    get_dlq_job(pool, job_id).await
184}
185
186/// Bulk-move failed terminal rows into the DLQ.
187///
188/// Returns the number of rows moved. Requires at least one of `kind` or
189/// `queue` unless `allow_all` is `true`, which is an explicit opt-in to move
190/// every failed row currently stored in queue_storage.
191///
192/// Preserves the source row's progress snapshot and error history so manual
193/// operator moves do not discard the last known execution state.
194pub async fn bulk_move_failed_to_dlq(
195    pool: &PgPool,
196    kind: Option<&str>,
197    queue: Option<&str>,
198    reason: &str,
199    allow_all: bool,
200) -> Result<u64, AwaError> {
201    if !allow_all && kind.is_none() && queue.is_none() {
202        return Err(AwaError::Validation(
203            "bulk_move_failed_to_dlq requires at least one of kind or queue (or allow_all=true)"
204                .into(),
205        ));
206    }
207
208    let store = active_queue_storage(pool).await?;
209    store
210        .bulk_move_failed_to_dlq(pool, kind, queue, reason)
211        .await
212}
213
214pub async fn list_dlq(pool: &PgPool, filter: &ListDlqFilter) -> Result<Vec<DlqRow>, AwaError> {
215    let store = active_queue_storage(pool).await?;
216    let schema = store.schema();
217    let rows: Vec<QueueStorageDlqRow> = sqlx::query_as(&format!(
218        r#"
219        SELECT
220            job_id,
221            kind,
222            queue,
223            args,
224            state,
225            priority,
226            attempt,
227            run_lease,
228            max_attempts,
229            run_at,
230            attempted_at,
231            finalized_at,
232            created_at,
233            unique_key,
234            payload,
235            dlq_reason,
236            dlq_at,
237            original_run_lease
238        FROM {schema}.dlq_entries
239        WHERE ($1::text IS NULL OR kind = $1)
240          AND ($2::text IS NULL OR queue = $2)
241          AND ($3::text IS NULL OR payload -> 'tags' ? $3)
242          AND (
243              ($4::bigint IS NULL AND $5::timestamptz IS NULL)
244              OR ($4::bigint IS NOT NULL AND $5::timestamptz IS NULL AND job_id < $4)
245              OR ($4::bigint IS NULL AND $5::timestamptz IS NOT NULL AND dlq_at < $5)
246              OR (
247                  $4::bigint IS NOT NULL
248                  AND $5::timestamptz IS NOT NULL
249                  AND (dlq_at, job_id) < ($5, $4)
250              )
251          )
252        ORDER BY dlq_at DESC, job_id DESC
253        LIMIT $6
254        "#
255    ))
256    .bind(&filter.kind)
257    .bind(&filter.queue)
258    .bind(&filter.tag)
259    .bind(filter.before_id)
260    .bind(filter.before_dlq_at)
261    .bind(filter.limit.unwrap_or(100))
262    .fetch_all(pool)
263    .await?;
264
265    rows.into_iter()
266        .map(QueueStorageDlqRow::into_dlq_row)
267        .collect()
268}
269
270pub async fn get_dlq_job(pool: &PgPool, job_id: i64) -> Result<Option<DlqRow>, AwaError> {
271    let store = active_queue_storage(pool).await?;
272    let schema = store.schema();
273    let row: Option<QueueStorageDlqRow> = sqlx::query_as(&format!(
274        r#"
275        SELECT
276            job_id,
277            kind,
278            queue,
279            args,
280            state,
281            priority,
282            attempt,
283            run_lease,
284            max_attempts,
285            run_at,
286            attempted_at,
287            finalized_at,
288            created_at,
289            unique_key,
290            payload,
291            dlq_reason,
292            dlq_at,
293            original_run_lease
294        FROM {schema}.dlq_entries
295        WHERE job_id = $1
296        ORDER BY dlq_at DESC
297        LIMIT 1
298        "#
299    ))
300    .bind(job_id)
301    .fetch_optional(pool)
302    .await?;
303
304    row.map(QueueStorageDlqRow::into_dlq_row).transpose()
305}
306
307pub async fn dlq_depth(pool: &PgPool, queue: Option<&str>) -> Result<i64, AwaError> {
308    let store = active_queue_storage(pool).await?;
309    sqlx::query_scalar::<_, i64>(&format!(
310        "SELECT count(*)::bigint FROM {}.dlq_entries WHERE ($1::text IS NULL OR queue = $1)",
311        store.schema()
312    ))
313    .bind(queue)
314    .fetch_one(pool)
315    .await
316    .map_err(Into::into)
317}
318
319pub async fn dlq_depth_by_queue(pool: &PgPool) -> Result<Vec<(String, i64)>, AwaError> {
320    let store = active_queue_storage(pool).await?;
321    sqlx::query_as::<_, (String, i64)>(&format!(
322        r#"
323        SELECT queue, count(*)::bigint
324        FROM {}.dlq_entries
325        GROUP BY queue
326        ORDER BY count(*) DESC, queue ASC
327        "#,
328        store.schema()
329    ))
330    .fetch_all(pool)
331    .await
332    .map_err(Into::into)
333}
334
335/// Retry a single DLQ row back into live queue storage.
336///
337/// Atomic: deletes the DLQ row and inserts a fresh ready/deferred row in one
338/// transaction. Resets `attempt = 0`, `run_lease = 0`, and clears any
339/// per-attempt progress snapshot because a revived attempt starts from zero.
340/// Error history is preserved in payload metadata for post-mortem visibility.
341///
342/// If the revived row conflicts with a live unique claim, this returns
343/// [`AwaError::UniqueConflict`] and the DLQ row remains in place.
344pub async fn retry_from_dlq(
345    pool: &PgPool,
346    job_id: i64,
347    opts: &RetryFromDlqOpts,
348) -> Result<Option<JobRow>, AwaError> {
349    let store = active_queue_storage(pool).await?;
350    store.retry_from_dlq(pool, job_id, opts).await
351}
352
353/// Bulk-retry DLQ rows matching the filter.
354///
355/// Requires at least one of `kind`, `queue`, or `tag` unless `allow_all` is
356/// `true`. This guard prevents an empty payload from reviving the entire DLQ
357/// by accident.
358///
359/// Like single-row retry, unique-claim conflicts abort the transaction and
360/// leave the affected DLQ rows untouched.
361pub async fn bulk_retry_from_dlq(
362    pool: &PgPool,
363    filter: &ListDlqFilter,
364    allow_all: bool,
365) -> Result<u64, AwaError> {
366    if !allow_all && filter_requires_scope(filter) {
367        return Err(AwaError::Validation(
368            "bulk_retry_from_dlq requires at least one of kind, queue, or tag (or allow_all=true)"
369                .into(),
370        ));
371    }
372
373    let store = active_queue_storage(pool).await?;
374    store.bulk_retry_from_dlq(pool, filter).await
375}
376
377/// Purge DLQ rows matching the filter.
378///
379/// Requires at least one of `kind`, `queue`, or `tag` unless `allow_all` is
380/// `true`, which is the explicit "yes, purge the whole DLQ" escape hatch for
381/// operator tooling.
382pub async fn purge_dlq(
383    pool: &PgPool,
384    filter: &ListDlqFilter,
385    allow_all: bool,
386) -> Result<u64, AwaError> {
387    if !allow_all && filter_requires_scope(filter) {
388        return Err(AwaError::Validation(
389            "purge_dlq requires at least one of kind, queue, or tag (or allow_all=true)".into(),
390        ));
391    }
392
393    let store = active_queue_storage(pool).await?;
394    let result = sqlx::query(&format!(
395        r#"
396        DELETE FROM {}.dlq_entries
397        WHERE ($1::text IS NULL OR kind = $1)
398          AND ($2::text IS NULL OR queue = $2)
399          AND ($3::text IS NULL OR payload -> 'tags' ? $3)
400          AND (
401              ($4::bigint IS NULL AND $5::timestamptz IS NULL)
402              OR ($4::bigint IS NOT NULL AND $5::timestamptz IS NULL AND job_id < $4)
403              OR ($4::bigint IS NULL AND $5::timestamptz IS NOT NULL AND dlq_at < $5)
404              OR (
405                  $4::bigint IS NOT NULL
406                  AND $5::timestamptz IS NOT NULL
407                  AND (dlq_at, job_id) < ($5, $4)
408              )
409          )
410        "#,
411        store.schema()
412    ))
413    .bind(&filter.kind)
414    .bind(&filter.queue)
415    .bind(&filter.tag)
416    .bind(filter.before_id)
417    .bind(filter.before_dlq_at)
418    .execute(pool)
419    .await?;
420    Ok(result.rows_affected())
421}
422
423pub async fn purge_dlq_job(pool: &PgPool, job_id: i64) -> Result<bool, AwaError> {
424    let store = active_queue_storage(pool).await?;
425    let result = sqlx::query(&format!(
426        "DELETE FROM {}.dlq_entries WHERE job_id = $1",
427        store.schema()
428    ))
429    .bind(job_id)
430    .execute(pool)
431    .await?;
432    Ok(result.rows_affected() > 0)
433}
434
435/// Purge DLQ rows older than the configured retention horizon.
436///
437/// Deletes up to `batch_size` rows per call, ordered by oldest-first
438/// `(dlq_at, job_id)`, optionally restricted to a single queue. Intended for
439/// maintenance loops rather than interactive operator use.
440pub async fn cleanup_dlq(
441    pool: &PgPool,
442    retention: Duration,
443    batch_size: i64,
444    queue: Option<&str>,
445) -> Result<u64, AwaError> {
446    let store = active_queue_storage(pool).await?;
447    let retention_secs = retention.as_secs().min(i64::MAX as u64) as i64;
448    let result = sqlx::query(&format!(
449        r#"
450        DELETE FROM {}.dlq_entries
451        WHERE job_id IN (
452            SELECT job_id
453            FROM {}.dlq_entries
454            WHERE dlq_at < now() - make_interval(secs => $1::bigint)
455              AND ($3::text IS NULL OR queue = $3)
456            ORDER BY dlq_at ASC, job_id ASC
457            LIMIT $2
458        )
459        "#,
460        store.schema(),
461        store.schema()
462    ))
463    .bind(retention_secs)
464    .bind(batch_size)
465    .bind(queue)
466    .execute(pool)
467    .await?;
468    Ok(result.rows_affected())
469}
awa_model/dlq.rs

awa_model/
dlq.rs
