⭐
Log in to mark this repository.
Reset attempts when re-enqueueing a job, so retries can escape the DLQ
48f426e · Gabriel Morell · 2026-08-25 16:21
Message
{commit_body(@commit)}
Files changed
modified
Cargo.toml
+2
−0
@@ -68,3 +68,5 @@ minijinja-embed = "2.12"
| 68 | 68 | |
| 69 | 69 | [dev-dependencies] |
| 70 | 70 | tokio = { version = "1.47.1", features = ["macros", "rt-multi-thread"] } |
| 71 | +# "mock" lets the job-upsert tests assert on generated SQL without a live DB. | |
| 72 | +sea-orm = { workspace = true, features = ["mock"] } |
modified
src/jobs/mod.rs
+9
−1
@@ -29,7 +29,8 @@ pub const UPLOAD_STATUS_PARTIAL: &str = "partial";
| 29 | 29 | |
| 30 | 30 | /// Insert a pending row for (upload, kind), or reset an existing row to |
| 31 | 31 | /// pending (used when a job is re-enqueued — manual force, follow-up jobs, |
| 32 | −/// etc.). Idempotent. | |
| 32 | +/// etc.). Idempotent. Clears `attempts` so a retry genuinely starts the | |
| 33 | +/// WORKER_MAX_ATTEMPTS budget over rather than resuming the old count. | |
| 33 | 34 | pub async fn upsert_pending( |
| 34 | 35 | db: &DatabaseConnection, |
| 35 | 36 | upload_id: Uuid, |
@@ -61,6 +62,13 @@ pub async fn upsert_pending(
| 61 | 62 | upload_jobs::Column::StartedAt, |
| 62 | 63 | upload_jobs::Column::CompletedAt, |
| 63 | 64 | upload_jobs::Column::UpdatedAt, |
| 65 | + // Attempts must be reset here, not just in the INSERT | |
| 66 | + // branch above. A re-enqueue always conflicts onto the | |
| 67 | + // existing row, so leaving it out stranded any job that | |
| 68 | + // had already hit WORKER_MAX_ATTEMPTS: the retry flipped | |
| 69 | + // status to pending, but the worker's pre-flight check | |
| 70 | + // still saw the old count and bounced it to the DLQ. | |
| 71 | + upload_jobs::Column::Attempts, | |
| 64 | 72 | ]) |
| 65 | 73 | .to_owned(), |
| 66 | 74 | ) |
added
tests/jobs_upsert.rs
+49
−0
@@ -0,0 +1,49 @@
| 1 | +//! Regression cover for the re-enqueue path. | |
| 2 | +//! | |
| 3 | +//! `upsert_pending` always conflicts onto the existing (upload_id, kind) row | |
| 4 | +//! when a job is retried, so the ActiveModel's `attempts: Set(0)` never | |
| 5 | +//! applies -- only the columns named in DO UPDATE SET do. Omitting Attempts | |
| 6 | +//! there stranded any job that had reached WORKER_MAX_ATTEMPTS: retry flipped | |
| 7 | +//! status to pending, but the worker's pre-flight attempts check still saw | |
| 8 | +//! the exhausted count and bounced it straight to the DLQ. | |
| 9 | + | |
| 10 | +use dms::jobs::upsert_pending; | |
| 11 | +use dms::queue::JobKind; | |
| 12 | +use sea_orm::{DatabaseBackend, MockDatabase, MockExecResult}; | |
| 13 | +use uuid::Uuid; | |
| 14 | + | |
| 15 | +async fn upsert_sql(kind: JobKind) -> String { | |
| 16 | + let db = MockDatabase::new(DatabaseBackend::Postgres) | |
| 17 | + .append_exec_results([MockExecResult { last_insert_id: 0, rows_affected: 1 }]) | |
| 18 | + .into_connection(); | |
| 19 | + upsert_pending(&db, Uuid::now_v7(), kind).await.expect("upsert_pending"); | |
| 20 | + let log = db.into_transaction_log(); | |
| 21 | + assert_eq!(log.len(), 1, "expected exactly one statement"); | |
| 22 | + format!("{:?}", log[0]) | |
| 23 | +} | |
| 24 | + | |
| 25 | +/// The whole point of the fix: a retry must clear the attempts budget. | |
| 26 | +#[tokio::test] | |
| 27 | +async fn upsert_pending_resets_attempts_on_conflict() { | |
| 28 | + let sql = upsert_sql(JobKind::Describe).await; | |
| 29 | + let update = sql | |
| 30 | + .split_once("DO UPDATE SET") | |
| 31 | + .expect("statement should be an upsert with a DO UPDATE branch") | |
| 32 | + .1; | |
| 33 | + assert!( | |
| 34 | + update.contains("attempts"), | |
| 35 | + "attempts must be reset in the conflict branch, or an exhausted job \ | |
| 36 | + can never be retried out of the DLQ; got: {sql}" | |
| 37 | + ); | |
| 38 | +} | |
| 39 | + | |
| 40 | +/// Guard the columns the retry path also depends on, so a future edit to the | |
| 41 | +/// update list can't quietly drop one. | |
| 42 | +#[tokio::test] | |
| 43 | +async fn upsert_pending_resets_the_rest_of_the_retry_state() { | |
| 44 | + let sql = upsert_sql(JobKind::Ocr).await; | |
| 45 | + let update = sql.split_once("DO UPDATE SET").unwrap().1; | |
| 46 | + for col in ["status", "error", "started_at", "completed_at", "updated_at"] { | |
| 47 | + assert!(update.contains(col), "{col} missing from DO UPDATE SET; got: {sql}"); | |
| 48 | + } | |
| 49 | +} |
Parents: 45647d3