diff --git a/crates/taskito-core/src/scheduler/maintenance.rs b/crates/taskito-core/src/scheduler/maintenance.rs index afb3f304..5b4c7be5 100644 --- a/crates/taskito-core/src/scheduler/maintenance.rs +++ b/crates/taskito-core/src/scheduler/maintenance.rs @@ -57,9 +57,8 @@ impl Scheduler { let cutoff = now_millis().saturating_sub(ttl); - // Use per-job TTL aware cleanup let completed = self.storage.purge_completed_with_ttl(cutoff)?; - let dead = self.storage.purge_dead(cutoff)?; + let dead = self.storage.purge_dead_with_ttl(cutoff)?; let errors = self.storage.purge_job_errors(cutoff)?; let metrics = self.storage.purge_metrics(cutoff).unwrap_or_else(|e| { warn!("purge_metrics failed: {e}"); @@ -131,6 +130,49 @@ impl Scheduler { Ok(()) } + /// Auto-retry eligible DLQ entries that have aged past the configured delay. + pub(super) fn auto_retry_dlq(&self) -> Result<()> { + let delay = match self.config.dlq_auto_retry_delay_ms { + Some(d) => d, + None => return Ok(()), + }; + + let now = now_millis(); + let cutoff = now.saturating_sub(delay); + let max_retries = self.config.dlq_auto_retry_max; + + let candidates = self.storage.list_dead_for_retry(cutoff, max_retries, 50)?; + + if candidates.is_empty() { + return Ok(()); + } + + let mut retried = 0u64; + for entry in &candidates { + match self.storage.retry_dead(&entry.id) { + Ok(new_id) => { + info!( + "dlq auto-retry: {} -> {} (attempt {}/{})", + entry.id, + new_id, + entry.dlq_retry_count + 1, + max_retries + ); + retried += 1; + } + Err(e) => { + warn!("dlq auto-retry failed for {}: {e}", entry.id); + } + } + } + + if retried > 0 { + info!("dlq auto-retry: retried {retried} entries"); + } + + Ok(()) + } + /// Build a cloudpickle-compatible payload from stored args. fn build_periodic_payload(args: &Option>) -> Vec { match args { diff --git a/crates/taskito-core/src/scheduler/mod.rs b/crates/taskito-core/src/scheduler/mod.rs index 47a9b41f..472958c6 100644 --- a/crates/taskito-core/src/scheduler/mod.rs +++ b/crates/taskito-core/src/scheduler/mod.rs @@ -39,6 +39,12 @@ pub struct SchedulerConfig { /// preserves the original one-job-per-round-trip behavior; values above /// `1` enable batch claiming for higher throughput. pub batch_size: usize, + /// Minimum age (ms) before a DLQ entry is auto-retried. `None` disables. + pub dlq_auto_retry_delay_ms: Option, + /// Max DLQ auto-retries per entry before giving up. + pub dlq_auto_retry_max: i32, + /// Check DLQ auto-retry every N ticks. + pub dlq_auto_retry_interval: u32, } impl Default for SchedulerConfig { @@ -51,6 +57,9 @@ impl Default for SchedulerConfig { cleanup_interval: 1200, result_ttl_ms: None, batch_size: 1, + dlq_auto_retry_delay_ms: None, + dlq_auto_retry_max: 1, + dlq_auto_retry_interval: 60, } } } @@ -157,6 +166,7 @@ struct TickCounters { reap: u32, periodic: u32, cleanup: u32, + dlq_auto_retry: u32, } impl Scheduler { @@ -316,6 +326,16 @@ impl Scheduler { } } + counters.dlq_auto_retry += 1; + if counters + .dlq_auto_retry + .is_multiple_of(self.config.dlq_auto_retry_interval) + { + if let Err(e) = self.auto_retry_dlq() { + error!("dlq auto-retry error: {e}"); + } + } + had_maintenance } } diff --git a/crates/taskito-core/src/storage/diesel_common/dead_letter.rs b/crates/taskito-core/src/storage/diesel_common/dead_letter.rs index 59f7a183..f1c86ba1 100644 --- a/crates/taskito-core/src/storage/diesel_common/dead_letter.rs +++ b/crates/taskito-core/src/storage/diesel_common/dead_letter.rs @@ -19,6 +19,13 @@ macro_rules! impl_diesel_dead_letter_ops { // Write-priority transaction: this reads the job row then writes // it to `archived_jobs`, which would deadlock under SQLite's // deferred lock-upgrade. See `archive_transaction`. + let dlq_retry_count = job + .metadata + .as_deref() + .and_then(|m| serde_json::from_str::(m).ok()) + .and_then(|v| v.get("__dlq_retry_count")?.as_i64()) + .unwrap_or(0) as i32; + self.archive_transaction(|conn| { let dlq_row = NewDeadLetterRow { id: &dlq_id, @@ -36,6 +43,7 @@ macro_rules! impl_diesel_dead_letter_ops { timeout_ms: job.timeout_ms, result_ttl_ms: job.result_ttl_ms, namespace: job.namespace.as_deref(), + dlq_retry_count, }; diesel::insert_into(dead_letter::table) @@ -106,6 +114,23 @@ macro_rules! impl_diesel_dead_letter_ops { Err(err) => return Err(err.into()), }; + let retry_metadata = { + let next_count = dead_row.dlq_retry_count + 1; + let mut obj = dead_row + .metadata + .as_deref() + .and_then(|m| { + serde_json::from_str::>(m) + .ok() + }) + .unwrap_or_default(); + obj.insert( + "__dlq_retry_count".to_string(), + serde_json::Value::from(next_count), + ); + Some(serde_json::to_string(&serde_json::Value::Object(obj)).unwrap_or_default()) + }; + let new_job = NewJob { queue: dead_row.queue, task_name: dead_row.task_name, @@ -115,7 +140,7 @@ macro_rules! impl_diesel_dead_letter_ops { max_retries: dead_row.max_retries, timeout_ms: dead_row.timeout_ms, unique_key: None, - metadata: dead_row.metadata, + metadata: retry_metadata, notes: dead_row.notes, depends_on: vec![], expires_at: None, @@ -194,6 +219,59 @@ macro_rules! impl_diesel_dead_letter_ops { Ok(affected as u64) } + + /// Delete a single dead letter entry. Returns true if it existed. + pub fn delete_dead(&self, dead_id: &str) -> Result { + let mut conn = self.conn()?; + let affected = + diesel::delete(dead_letter::table.find(dead_id)).execute(&mut conn)?; + Ok(affected > 0) + } + + /// Purge dead letter entries respecting per-entry TTL overrides. + pub fn purge_dead_with_ttl(&self, global_cutoff_ms: i64) -> Result { + let now = now_millis(); + let mut conn = self.conn()?; + + let global_count = diesel::delete( + dead_letter::table + .filter(dead_letter::result_ttl_ms.is_null()) + .filter(dead_letter::failed_at.lt(global_cutoff_ms)), + ) + .execute(&mut conn)?; + + let per_entry_count = diesel::delete( + dead_letter::table + .filter(dead_letter::result_ttl_ms.is_not_null()) + .filter( + (dead_letter::failed_at + dead_letter::result_ttl_ms.assume_not_null()) + .lt(now), + ), + ) + .execute(&mut conn)?; + + Ok((global_count + per_entry_count) as u64) + } + + /// List DLQ entries eligible for auto-retry. + pub fn list_dead_for_retry( + &self, + cutoff_ms: i64, + max_retries: i32, + limit: i64, + ) -> Result> { + let mut conn = self.conn()?; + + let rows: Vec = dead_letter::table + .filter(dead_letter::failed_at.le(cutoff_ms)) + .filter(dead_letter::dlq_retry_count.lt(max_retries)) + .order(dead_letter::failed_at.asc()) + .limit(limit) + .select(DeadLetterRow::as_select()) + .load(&mut conn)?; + + Ok(rows.into_iter().map(DeadJob::from).collect()) + } } }; } diff --git a/crates/taskito-core/src/storage/diesel_common/migrations.rs b/crates/taskito-core/src/storage/diesel_common/migrations.rs index 7465edc5..48a0c6be 100644 --- a/crates/taskito-core/src/storage/diesel_common/migrations.rs +++ b/crates/taskito-core/src/storage/diesel_common/migrations.rs @@ -360,6 +360,8 @@ pub fn alter_statements(d: &Dialect) -> Vec { // has_deps fast-path flag: lets dequeue skip the dependency lookup for // the common no-dependency job. Defaults false; backfilled below. format!("ALTER TABLE jobs ADD COLUMN {ife}has_deps {bool_false}"), + // DLQ auto-retry counter: tracks how many times an entry was auto-retried + format!("ALTER TABLE dead_letter ADD COLUMN {ife}dlq_retry_count INTEGER NOT NULL DEFAULT 0"), ] } diff --git a/crates/taskito-core/src/storage/mod.rs b/crates/taskito-core/src/storage/mod.rs index 42edd8c3..f095e7a6 100644 --- a/crates/taskito-core/src/storage/mod.rs +++ b/crates/taskito-core/src/storage/mod.rs @@ -52,6 +52,7 @@ pub struct DeadJob { pub timeout_ms: i64, pub result_ttl_ms: Option, pub namespace: Option, + pub dlq_retry_count: i32, } impl From for DeadJob { @@ -72,6 +73,7 @@ impl From for DeadJob { timeout_ms: row.timeout_ms, result_ttl_ms: row.result_ttl_ms, namespace: row.namespace, + dlq_retry_count: row.dlq_retry_count, } } } @@ -245,6 +247,20 @@ macro_rules! impl_storage { fn purge_dead(&self, older_than_ms: i64) -> $crate::error::Result { self.purge_dead(older_than_ms) } + fn delete_dead(&self, dead_id: &str) -> $crate::error::Result { + self.delete_dead(dead_id) + } + fn purge_dead_with_ttl(&self, global_cutoff_ms: i64) -> $crate::error::Result { + self.purge_dead_with_ttl(global_cutoff_ms) + } + fn list_dead_for_retry( + &self, + cutoff_ms: i64, + max_retries: i32, + limit: i64, + ) -> $crate::error::Result> { + self.list_dead_for_retry(cutoff_ms, max_retries, limit) + } fn get_rate_limit( &self, key: &str, @@ -734,6 +750,20 @@ impl Storage for StorageBackend { fn purge_dead(&self, older_than_ms: i64) -> Result { delegate!(self, purge_dead, older_than_ms) } + fn delete_dead(&self, dead_id: &str) -> Result { + delegate!(self, delete_dead, dead_id) + } + fn purge_dead_with_ttl(&self, global_cutoff_ms: i64) -> Result { + delegate!(self, purge_dead_with_ttl, global_cutoff_ms) + } + fn list_dead_for_retry( + &self, + cutoff_ms: i64, + max_retries: i32, + limit: i64, + ) -> Result> { + delegate!(self, list_dead_for_retry, cutoff_ms, max_retries, limit) + } fn get_rate_limit(&self, key: &str) -> Result> { delegate!(self, get_rate_limit, key) } diff --git a/crates/taskito-core/src/storage/models.rs b/crates/taskito-core/src/storage/models.rs index 349aaf01..294cbba0 100644 --- a/crates/taskito-core/src/storage/models.rs +++ b/crates/taskito-core/src/storage/models.rs @@ -130,6 +130,7 @@ pub struct DeadLetterRow { pub timeout_ms: i64, pub result_ttl_ms: Option, pub namespace: Option, + pub dlq_retry_count: i32, } /// Insertable struct for dead letter entries. @@ -151,6 +152,7 @@ pub struct NewDeadLetterRow<'a> { pub timeout_ms: i64, pub result_ttl_ms: Option, pub namespace: Option<&'a str>, + pub dlq_retry_count: i32, } /// A row in the `rate_limits` table. diff --git a/crates/taskito-core/src/storage/redis_backend/dead_letter.rs b/crates/taskito-core/src/storage/redis_backend/dead_letter.rs index b2644201..da51cc58 100644 --- a/crates/taskito-core/src/storage/redis_backend/dead_letter.rs +++ b/crates/taskito-core/src/storage/redis_backend/dead_letter.rs @@ -25,6 +25,8 @@ struct DeadJobEntry { pub result_ttl_ms: Option, #[serde(default)] pub namespace: Option, + #[serde(default)] + pub dlq_retry_count: i32, } impl From for DeadJob { @@ -45,6 +47,7 @@ impl From for DeadJob { timeout_ms: e.timeout_ms, result_ttl_ms: e.result_ttl_ms, namespace: e.namespace, + dlq_retry_count: e.dlq_retry_count, } } } @@ -55,6 +58,13 @@ impl RedisStorage { let dlq_id = uuid::Uuid::now_v7().to_string(); let mut conn = self.conn()?; + let dlq_retry_count = job + .metadata + .as_deref() + .and_then(|m| serde_json::from_str::(m).ok()) + .and_then(|v| v.get("__dlq_retry_count")?.as_i64()) + .unwrap_or(0) as i32; + let entry = DeadJobEntry { id: dlq_id.clone(), original_job_id: job.id.clone(), @@ -71,6 +81,7 @@ impl RedisStorage { timeout_ms: job.timeout_ms, result_ttl_ms: job.result_ttl_ms, namespace: job.namespace.clone(), + dlq_retry_count, }; let json = serde_json::to_string(&entry).map_err(|e| QueueError::Other(e.to_string()))?; @@ -140,6 +151,22 @@ impl RedisStorage { let entry: DeadJobEntry = serde_json::from_str(&data).map_err(|e| QueueError::Other(e.to_string()))?; + let retry_metadata = { + let next_count = entry.dlq_retry_count + 1; + let mut obj = entry + .metadata + .as_deref() + .and_then(|m| { + serde_json::from_str::>(m).ok() + }) + .unwrap_or_default(); + obj.insert( + "__dlq_retry_count".to_string(), + serde_json::Value::from(next_count), + ); + Some(serde_json::to_string(&serde_json::Value::Object(obj)).unwrap_or_default()) + }; + let new_job = NewJob { queue: entry.queue, task_name: entry.task_name, @@ -149,7 +176,7 @@ impl RedisStorage { max_retries: entry.max_retries, timeout_ms: entry.timeout_ms, unique_key: None, - metadata: entry.metadata, + metadata: retry_metadata, notes: entry.notes, depends_on: vec![], expires_at: None, @@ -192,4 +219,92 @@ impl RedisStorage { Ok(ids.len() as u64) } + + pub fn delete_dead(&self, dead_id: &str) -> Result { + let mut conn = self.conn()?; + let dlq_key = self.key(&["dlq", dead_id]); + let dlq_all = self.key(&["dlq", "all"]); + + let existed: bool = conn.exists(&dlq_key).map_err(map_err)?; + if !existed { + return Ok(false); + } + + let pipe = &mut redis::pipe(); + pipe.del(&dlq_key); + pipe.zrem(&dlq_all, dead_id); + pipe.query::<()>(&mut conn).map_err(map_err)?; + Ok(true) + } + + pub fn purge_dead_with_ttl(&self, global_cutoff_ms: i64) -> Result { + let mut conn = self.conn()?; + let dlq_all = self.key(&["dlq", "all"]); + let now = now_millis(); + + let ids: Vec = conn + .zrangebyscore(&dlq_all, "-inf", "+inf") + .map_err(map_err)?; + + let mut to_delete = Vec::new(); + for id in &ids { + let dlq_key = self.key(&["dlq", id]); + let data: Option = conn.get(&dlq_key).map_err(map_err)?; + if let Some(d) = data { + if let Ok(entry) = serde_json::from_str::(&d) { + let expired = match entry.result_ttl_ms { + Some(ttl) => entry.failed_at + ttl <= now, + None => entry.failed_at < global_cutoff_ms, + }; + if expired { + to_delete.push(id.clone()); + } + } + } + } + + if to_delete.is_empty() { + return Ok(0); + } + + let pipe = &mut redis::pipe(); + for id in &to_delete { + pipe.del(self.key(&["dlq", id])); + pipe.zrem(&dlq_all, id.as_str()); + } + pipe.query::<()>(&mut conn).map_err(map_err)?; + Ok(to_delete.len() as u64) + } + + pub fn list_dead_for_retry( + &self, + cutoff_ms: i64, + max_retries: i32, + limit: i64, + ) -> Result> { + let mut conn = self.conn()?; + let dlq_all = self.key(&["dlq", "all"]); + + let ids: Vec = conn + .zrangebyscore(&dlq_all, "-inf", cutoff_ms as f64) + .map_err(map_err)?; + + let mut results = Vec::new(); + for id in ids { + if results.len() >= limit as usize { + break; + } + let dlq_key = self.key(&["dlq", &id]); + let data: Option = conn.get(&dlq_key).map_err(map_err)?; + if let Some(d) = data { + if let Ok(entry) = serde_json::from_str::(&d) { + if entry.dlq_retry_count < max_retries { + results.push(DeadJob::from(entry)); + } + } + } + } + + Ok(results) + } } diff --git a/crates/taskito-core/src/storage/schema.rs b/crates/taskito-core/src/storage/schema.rs index 0c1483fa..b9ec8edc 100644 --- a/crates/taskito-core/src/storage/schema.rs +++ b/crates/taskito-core/src/storage/schema.rs @@ -44,6 +44,7 @@ diesel::table! { timeout_ms -> BigInt, result_ttl_ms -> Nullable, namespace -> Nullable, + dlq_retry_count -> Integer, } } diff --git a/crates/taskito-core/src/storage/sqlite/tests.rs b/crates/taskito-core/src/storage/sqlite/tests.rs index 77ba5573..c831a0ee 100644 --- a/crates/taskito-core/src/storage/sqlite/tests.rs +++ b/crates/taskito-core/src/storage/sqlite/tests.rs @@ -1002,3 +1002,111 @@ fn test_side_table_stays_one_to_one_with_live_jobs() { assert_eq!(jobs_row_count(&storage, &job.id), 0); assert_eq!(payload_row_count(&storage, &job.id), 0); } + +// -- DLQ policies -- + +#[test] +fn test_delete_dead_existing() { + let storage = test_storage(); + let job = storage.enqueue(make_job("del_dead")).unwrap(); + storage + .dequeue("default", now_millis() + 1000, None) + .unwrap(); + let running = storage.get_job(&job.id).unwrap().unwrap(); + storage.move_to_dlq(&running, "err", None).unwrap(); + + let dead = storage.list_dead(10, 0).unwrap(); + assert_eq!(dead.len(), 1); + + assert!(storage.delete_dead(&dead[0].id).unwrap()); + assert!(storage.list_dead(10, 0).unwrap().is_empty()); +} + +#[test] +fn test_delete_dead_nonexistent() { + let storage = test_storage(); + assert!(!storage.delete_dead("nope").unwrap()); +} + +#[test] +fn test_purge_dead_with_ttl_global() { + let storage = test_storage(); + let now = now_millis(); + + // Create a DLQ entry with no per-entry TTL + let job = storage.enqueue(make_job("ttl_global")).unwrap(); + storage.dequeue("default", now + 1000, None).unwrap(); + let running = storage.get_job(&job.id).unwrap().unwrap(); + storage.move_to_dlq(&running, "err", None).unwrap(); + + // Cutoff in the future purges it + let purged = storage.purge_dead_with_ttl(now + 5000).unwrap(); + assert_eq!(purged, 1); +} + +#[test] +fn test_purge_dead_with_ttl_per_entry() { + let storage = test_storage(); + let now = now_millis(); + + // Create a job with per-entry TTL + let mut new_job = make_job("ttl_entry"); + new_job.result_ttl_ms = Some(1); // 1ms TTL + let job = storage.enqueue(new_job).unwrap(); + storage.dequeue("default", now + 1000, None).unwrap(); + let running = storage.get_job(&job.id).unwrap().unwrap(); + storage.move_to_dlq(&running, "err", None).unwrap(); + + std::thread::sleep(std::time::Duration::from_millis(5)); + + // Global cutoff in the past — only per-entry TTL should purge + let purged = storage.purge_dead_with_ttl(0).unwrap(); + assert_eq!(purged, 1); +} + +#[test] +fn test_list_dead_for_retry() { + let storage = test_storage(); + let now = now_millis(); + + let job = storage.enqueue(make_job("retry_cand")).unwrap(); + storage.dequeue("default", now + 1000, None).unwrap(); + let running = storage.get_job(&job.id).unwrap().unwrap(); + storage.move_to_dlq(&running, "err", None).unwrap(); + + // Cutoff in the future, max_retries=3 — should find it + let cands = storage.list_dead_for_retry(now + 5000, 3, 10).unwrap(); + assert_eq!(cands.len(), 1); + assert_eq!(cands[0].dlq_retry_count, 0); + + // max_retries=0 — should find nothing + let cands = storage.list_dead_for_retry(now + 5000, 0, 10).unwrap(); + assert!(cands.is_empty()); + + // Cutoff in the past — should find nothing + let cands = storage.list_dead_for_retry(0, 3, 10).unwrap(); + assert!(cands.is_empty()); +} + +#[test] +fn test_dlq_retry_count_round_trip() { + let storage = test_storage(); + let now = now_millis(); + + // Enqueue → dequeue → DLQ (count=0) → retry → dequeue → DLQ (count=1) + let job = storage.enqueue(make_job("count_rt")).unwrap(); + storage.dequeue("default", now + 1000, None).unwrap(); + let running = storage.get_job(&job.id).unwrap().unwrap(); + storage.move_to_dlq(&running, "err1", None).unwrap(); + + let dead = storage.list_dead(10, 0).unwrap(); + assert_eq!(dead[0].dlq_retry_count, 0); + + let new_id = storage.retry_dead(&dead[0].id).unwrap(); + storage.dequeue("default", now + 2000, None).unwrap(); + let running2 = storage.get_job(&new_id).unwrap().unwrap(); + storage.move_to_dlq(&running2, "err2", None).unwrap(); + + let dead2 = storage.list_dead(10, 0).unwrap(); + assert_eq!(dead2[0].dlq_retry_count, 1); +} diff --git a/crates/taskito-core/src/storage/traits.rs b/crates/taskito-core/src/storage/traits.rs index 134e1f3c..0fec1290 100644 --- a/crates/taskito-core/src/storage/traits.rs +++ b/crates/taskito-core/src/storage/traits.rs @@ -75,6 +75,14 @@ pub trait Storage: Send + Sync + Clone { fn list_dead(&self, limit: i64, offset: i64) -> Result>; fn retry_dead(&self, dead_id: &str) -> Result; fn purge_dead(&self, older_than_ms: i64) -> Result; + fn delete_dead(&self, dead_id: &str) -> Result; + fn purge_dead_with_ttl(&self, global_cutoff_ms: i64) -> Result; + fn list_dead_for_retry( + &self, + cutoff_ms: i64, + max_retries: i32, + limit: i64, + ) -> Result>; // ── Rate limit operations ─────────────────────────────────────── diff --git a/crates/taskito-core/tests/rust/storage_tests.rs b/crates/taskito-core/tests/rust/storage_tests.rs index 6d719e7b..a94de696 100644 --- a/crates/taskito-core/tests/rust/storage_tests.rs +++ b/crates/taskito-core/tests/rust/storage_tests.rs @@ -220,6 +220,47 @@ fn test_dead_letter_queue(s: &impl Storage) { assert!(!dead.is_empty()); } +fn test_delete_dead(s: &impl Storage) { + let q = "q-del-dead"; + let job = s.enqueue(make_job(q, "del_dead_task")).unwrap(); + s.dequeue(q, now_millis() + 1000, None).unwrap(); + let running = s.get_job(&job.id).unwrap().unwrap(); + s.move_to_dlq(&running, "err", None).unwrap(); + + let dead = s.list_dead(100, 0).unwrap(); + let entry = dead + .iter() + .find(|d| d.original_job_id == job.id) + .expect("our DLQ entry should exist"); + let dead_id = entry.id.clone(); + + assert!(s.delete_dead(&dead_id).unwrap()); + assert!(!s.delete_dead(&dead_id).unwrap()); +} + +fn test_list_dead_for_retry(s: &impl Storage) { + let q = "q-dlq-retry"; + let job = s.enqueue(make_job(q, "dlq_retry_task")).unwrap(); + s.dequeue(q, now_millis() + 1000, None).unwrap(); + let running = s.get_job(&job.id).unwrap().unwrap(); + s.move_to_dlq(&running, "err", None).unwrap(); + + let now = now_millis(); + let cands = s.list_dead_for_retry(now + 5000, 3, 100).unwrap(); + let ours = cands + .iter() + .find(|d| d.original_job_id == job.id) + .expect("our entry should be eligible"); + assert_eq!(ours.dlq_retry_count, 0); + + // max_retries=0 should exclude everything + let empty = s.list_dead_for_retry(now + 5000, 0, 100).unwrap(); + assert!( + empty.iter().all(|d| d.original_job_id != job.id), + "max_retries=0 should exclude our entry" + ); +} + fn test_progress_tracking(s: &impl Storage) { let job = s.enqueue(make_job("q-progress", "progress_task")).unwrap(); s.update_progress(&job.id, 50).unwrap(); @@ -544,6 +585,8 @@ fn run_storage_tests(s: &impl Storage) { test_unique_key_dedup(s); test_enqueue_batch(s); test_dead_letter_queue(s); + test_delete_dead(s); + test_list_dead_for_retry(s); test_progress_tracking(s); test_record_and_get_errors(s); test_workers(s); diff --git a/crates/taskito-python/src/py_queue/inspection.rs b/crates/taskito-python/src/py_queue/inspection.rs index df387af9..2a2926f1 100644 --- a/crates/taskito-python/src/py_queue/inspection.rs +++ b/crates/taskito-python/src/py_queue/inspection.rs @@ -103,6 +103,7 @@ impl PyQueue { dict.set_item("retry_count", d.retry_count)?; dict.set_item("failed_at", d.failed_at)?; dict.set_item("metadata", d.metadata)?; + dict.set_item("dlq_retry_count", d.dlq_retry_count)?; result.push(dict.into()); } Ok(result) @@ -124,6 +125,13 @@ impl PyQueue { .map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(e.to_string())) } + /// Delete a single dead letter entry without re-enqueueing. + pub fn delete_dead(&self, dead_id: &str) -> PyResult { + self.storage + .delete_dead(dead_id) + .map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(e.to_string())) + } + /// Get error history for a job. pub fn get_job_errors(&self, job_id: &str) -> PyResult> { let errors = self diff --git a/crates/taskito-python/src/py_queue/mod.rs b/crates/taskito-python/src/py_queue/mod.rs index cd371c53..64b429dc 100644 --- a/crates/taskito-python/src/py_queue/mod.rs +++ b/crates/taskito-python/src/py_queue/mod.rs @@ -40,6 +40,8 @@ pub struct PyQueue { pub(crate) scheduler_reap_interval: u32, pub(crate) scheduler_cleanup_interval: u32, pub(crate) scheduler_batch_size: usize, + pub(crate) dlq_auto_retry_delay_ms: Option, + pub(crate) dlq_auto_retry_max: i32, pub(crate) namespace: Option, /// Opt-in event-driven dispatch. Honored only when the crate is built with /// the `push-dispatch` cargo feature; otherwise accepted and ignored. @@ -64,7 +66,7 @@ pub struct PyQueue { )] impl PyQueue { #[new] - #[pyo3(signature = (db_path=".taskito/taskito.db", workers=0, default_retry=3, default_timeout=300, default_priority=0, result_ttl=None, backend="sqlite", db_url=None, schema="taskito", pool_size=None, scheduler_poll_interval_ms=50, scheduler_reap_interval=100, scheduler_cleanup_interval=1200, scheduler_batch_size=1, namespace=None, push_dispatch=false))] + #[pyo3(signature = (db_path=".taskito/taskito.db", workers=0, default_retry=3, default_timeout=300, default_priority=0, result_ttl=None, backend="sqlite", db_url=None, schema="taskito", pool_size=None, scheduler_poll_interval_ms=50, scheduler_reap_interval=100, scheduler_cleanup_interval=1200, scheduler_batch_size=1, namespace=None, push_dispatch=false, dlq_auto_retry_delay=None, dlq_auto_retry_max=1))] #[allow(clippy::too_many_arguments)] pub fn new( py: Python<'_>, @@ -84,7 +86,32 @@ impl PyQueue { scheduler_batch_size: usize, namespace: Option, push_dispatch: bool, + dlq_auto_retry_delay: Option, + dlq_auto_retry_max: i32, ) -> PyResult { + if let Some(delay) = dlq_auto_retry_delay { + if delay < 0 { + return Err(pyo3::exceptions::PyValueError::new_err( + "dlq_auto_retry_delay must be non-negative", + )); + } + } + if dlq_auto_retry_max < 0 { + return Err(pyo3::exceptions::PyValueError::new_err( + "dlq_auto_retry_max must be non-negative", + )); + } + + let dlq_auto_retry_delay_ms = dlq_auto_retry_delay + .map(|s| { + s.checked_mul(1000).ok_or_else(|| { + pyo3::exceptions::PyValueError::new_err( + "dlq_auto_retry_delay too large, would overflow", + ) + }) + }) + .transpose()?; + // Storage init blocks on connection-pool builders that may emit // `log::*` records from worker threads. With the pyo3-log bridge // active, those records need the GIL to deliver to Python — so we @@ -158,6 +185,8 @@ impl PyQueue { scheduler_reap_interval, scheduler_cleanup_interval, scheduler_batch_size: scheduler_batch_size.max(1), + dlq_auto_retry_delay_ms, + dlq_auto_retry_max, namespace, push_dispatch, dispatcher: Arc::new(Mutex::new(None)), diff --git a/crates/taskito-python/src/py_queue/worker.rs b/crates/taskito-python/src/py_queue/worker.rs index 500fabb7..5623ba64 100644 --- a/crates/taskito-python/src/py_queue/worker.rs +++ b/crates/taskito-python/src/py_queue/worker.rs @@ -277,6 +277,8 @@ impl PyQueue { cleanup_interval: self.scheduler_cleanup_interval, result_ttl_ms: self.result_ttl_ms, batch_size: self.scheduler_batch_size, + dlq_auto_retry_delay_ms: self.dlq_auto_retry_delay_ms, + dlq_auto_retry_max: self.dlq_auto_retry_max, ..SchedulerConfig::default() }; let mut scheduler = Scheduler::new( diff --git a/crates/taskito-python/src/py_queue/workflow_ops/test_helpers.rs b/crates/taskito-python/src/py_queue/workflow_ops/test_helpers.rs index 56854a79..fc963dd7 100644 --- a/crates/taskito-python/src/py_queue/workflow_ops/test_helpers.rs +++ b/crates/taskito-python/src/py_queue/workflow_ops/test_helpers.rs @@ -55,6 +55,8 @@ pub(crate) fn make_test_pyqueue() -> PyQueue { scheduler_reap_interval: 100, scheduler_cleanup_interval: 1200, scheduler_batch_size: 1, + dlq_auto_retry_delay_ms: None, + dlq_auto_retry_max: 1, namespace: None, push_dispatch: false, dispatcher: Arc::new(Mutex::new(None)), diff --git a/dashboard/src/features/dead-letters/api.ts b/dashboard/src/features/dead-letters/api.ts index da8b748f..c96cbf9e 100644 --- a/dashboard/src/features/dead-letters/api.ts +++ b/dashboard/src/features/dead-letters/api.ts @@ -23,3 +23,7 @@ export function retryDeadLetter(id: string): Promise<{ new_job_id: string }> { export function purgeDeadLetters(): Promise<{ purged: number }> { return api.post<{ purged: number }>("/api/dead-letters/purge"); } + +export function deleteDeadLetter(id: string): Promise<{ deleted: boolean }> { + return api.delete<{ deleted: boolean }>(`/api/dead-letters/${encodeURIComponent(id)}`); +} diff --git a/dashboard/src/features/dead-letters/components/dead-letter-row.tsx b/dashboard/src/features/dead-letters/components/dead-letter-row.tsx index e4a445de..f8b3cd9d 100644 --- a/dashboard/src/features/dead-letters/components/dead-letter-row.tsx +++ b/dashboard/src/features/dead-letters/components/dead-letter-row.tsx @@ -1,9 +1,9 @@ import { Link } from "@tanstack/react-router"; -import { RotateCcw } from "lucide-react"; +import { RotateCcw, Trash2 } from "lucide-react"; import { Badge, Button } from "@/components/ui"; import type { DeadLetter } from "@/lib/api-types"; import { formatRelative } from "@/lib/time"; -import { useRetryDeadLetter } from "../hooks"; +import { useDeleteDeadLetter, useRetryDeadLetter } from "../hooks"; interface DeadLetterRowProps { item: DeadLetter; @@ -11,6 +11,7 @@ interface DeadLetterRowProps { export function DeadLetterRow({ item }: DeadLetterRowProps) { const retry = useRetryDeadLetter(); + const discard = useDeleteDeadLetter(); return (
@@ -30,6 +31,9 @@ export function DeadLetterRow({ item }: DeadLetterRowProps) { {item.retry_count} {item.retry_count === 1 ? "retry" : "retries"} ) : null} + {item.dlq_retry_count > 0 ? ( + DLQ retry #{item.dlq_retry_count} + ) : null} {formatRelative(item.failed_at)} @@ -40,14 +44,24 @@ export function DeadLetterRow({ item }: DeadLetterRowProps) { ) : null}
- +
+ + +
); } diff --git a/dashboard/src/features/dead-letters/hooks.ts b/dashboard/src/features/dead-letters/hooks.ts index 8a163cff..b1223815 100644 --- a/dashboard/src/features/dead-letters/hooks.ts +++ b/dashboard/src/features/dead-letters/hooks.ts @@ -7,7 +7,7 @@ import { } from "@tanstack/react-query"; import { toast } from "sonner"; import { useRefreshInterval } from "@/providers"; -import { fetchDeadLetters, purgeDeadLetters, retryDeadLetter } from "./api"; +import { deleteDeadLetter, fetchDeadLetters, purgeDeadLetters, retryDeadLetter } from "./api"; const KEY = { all: ["dead-letters"] as const, @@ -48,6 +48,29 @@ export function useRetryDeadLetter() { }); } +export function useDeleteDeadLetter() { + const qc = useQueryClient(); + return useMutation({ + mutationFn: (id: string) => deleteDeadLetter(id), + onError: (error) => { + toast.error("Couldn't discard dead letter", { + description: error instanceof Error ? error.message : String(error), + }); + }, + onSuccess: (result) => { + if (result.deleted) { + toast.success("Dead letter discarded"); + } else { + toast.message("Dead letter already removed"); + } + }, + onSettled: () => { + qc.invalidateQueries({ queryKey: KEY.all }); + qc.invalidateQueries({ queryKey: ["stats"] }); + }, + }); +} + export function usePurgeDeadLetters() { const qc = useQueryClient(); return useMutation({ diff --git a/dashboard/src/features/dead-letters/utils.test.ts b/dashboard/src/features/dead-letters/utils.test.ts index 5e95d2ba..dd626285 100644 --- a/dashboard/src/features/dead-letters/utils.test.ts +++ b/dashboard/src/features/dead-letters/utils.test.ts @@ -11,6 +11,7 @@ function dl(overrides: Partial): DeadLetter { error: overrides.error ?? null, retry_count: overrides.retry_count ?? 3, failed_at: overrides.failed_at ?? 1_700_000_000_000, + dlq_retry_count: overrides.dlq_retry_count ?? 0, }; } diff --git a/dashboard/src/lib/api-types.ts b/dashboard/src/lib/api-types.ts index f4216b0b..ea0104cc 100644 --- a/dashboard/src/lib/api-types.ts +++ b/dashboard/src/lib/api-types.ts @@ -114,6 +114,7 @@ export interface DeadLetter { retry_count: number; /** Unix milliseconds. */ failed_at: number; + dlq_retry_count: number; } export interface TaskMetrics { diff --git a/py_src/taskito/_taskito.pyi b/py_src/taskito/_taskito.pyi index 9174a25d..ce26f4c8 100644 --- a/py_src/taskito/_taskito.pyi +++ b/py_src/taskito/_taskito.pyi @@ -94,6 +94,8 @@ class PyQueue: scheduler_batch_size: int = 1, namespace: str | None = None, push_dispatch: bool = False, + dlq_auto_retry_delay: int | None = None, + dlq_auto_retry_max: int = 1, ) -> None: ... def request_shutdown(self) -> None: ... def enqueue( @@ -164,6 +166,7 @@ class PyQueue: def dead_letters(self, limit: int = 10, offset: int = 0) -> list[dict[str, Any]]: ... def retry_dead(self, dead_id: str) -> str: ... def purge_dead(self, older_than_seconds: int) -> int: ... + def delete_dead(self, dead_id: str) -> bool: ... def register_periodic( self, name: str, diff --git a/py_src/taskito/app.py b/py_src/taskito/app.py index 53f9cb0d..dc9309d4 100644 --- a/py_src/taskito/app.py +++ b/py_src/taskito/app.py @@ -142,6 +142,8 @@ def __init__( scheduler_batch_size: int = 1, namespace: str | None = None, push_dispatch: bool = False, + dlq_auto_retry_delay: int | None = None, + dlq_auto_retry_max: int = 1, ): """Initialize a new task queue. @@ -196,6 +198,10 @@ def __init__( poll. Honored only when the native module was built with the ``push-dispatch`` cargo feature; otherwise accepted and ignored (polling is kept). Defaults to ``False``. + dlq_auto_retry_delay: Minimum age in seconds before a DLQ entry + is automatically retried. ``None`` disables auto-retry. + dlq_auto_retry_max: Maximum number of DLQ auto-retries per entry + before giving up. Defaults to 1. """ if backend == "sqlite": # Ensure parent directory exists for SQLite @@ -220,6 +226,8 @@ def __init__( scheduler_batch_size=scheduler_batch_size, namespace=namespace, push_dispatch=push_dispatch, + dlq_auto_retry_delay=dlq_auto_retry_delay, + dlq_auto_retry_max=dlq_auto_retry_max, ) self._backend = backend self._namespace = namespace diff --git a/py_src/taskito/async_support/mixins.py b/py_src/taskito/async_support/mixins.py index 4f06bb81..5d9ca50e 100644 --- a/py_src/taskito/async_support/mixins.py +++ b/py_src/taskito/async_support/mixins.py @@ -80,6 +80,7 @@ def query_logs( ) -> list[dict]: ... def dead_letters(self, limit: int = ..., offset: int = ...) -> list[dict]: ... def retry_dead(self, dead_id: str) -> str: ... + def delete_dead(self, dead_id: str) -> bool: ... def replay(self, job_id: str) -> JobResult: ... def replay_history(self, job_id: str) -> list[dict]: ... def circuit_breakers(self) -> list[dict]: ... @@ -145,6 +146,10 @@ async def aretry_dead(self, dead_id: str) -> str: """Async version of :meth:`retry_dead`.""" return await self._run_sync(self.retry_dead, dead_id) + async def adelete_dead(self, dead_id: str) -> bool: + """Async version of :meth:`delete_dead`.""" + return await self._run_sync(self.delete_dead, dead_id) + async def areplay(self, job_id: str) -> JobResult: """Async version of :meth:`replay`.""" return await self._run_sync(self.replay, job_id) diff --git a/py_src/taskito/dashboard/routes.py b/py_src/taskito/dashboard/routes.py index 8fa1871b..4c44c2a9 100644 --- a/py_src/taskito/dashboard/routes.py +++ b/py_src/taskito/dashboard/routes.py @@ -259,6 +259,10 @@ def is_public_path(path: str) -> bool: (re.compile(r"^/api/tasks/([^/]+)/override$"), handle_delete_task_override), (re.compile(r"^/api/queues/([^/]+)/override$"), handle_delete_queue_override), (re.compile(r"^/api/tasks/([^/]+)/middleware$"), handle_delete_task_middleware), + ( + re.compile(r"^/api/dead-letters/([^/]+)$"), + lambda q, did: {"deleted": q.delete_dead(did)}, + ), ] diff --git a/py_src/taskito/mixins/operations.py b/py_src/taskito/mixins/operations.py index 7d59c118..e9c86b4b 100644 --- a/py_src/taskito/mixins/operations.py +++ b/py_src/taskito/mixins/operations.py @@ -28,6 +28,10 @@ def purge_dead(self, older_than: int = 86400) -> int: """Delete dead letter entries older than a given age.""" return self._inner.purge_dead(older_than) # type: ignore[no-any-return] + def delete_dead(self, dead_id: str) -> bool: + """Discard a single dead letter entry without re-enqueueing.""" + return self._inner.delete_dead(dead_id) # type: ignore[no-any-return] + # -- Replay -- def replay(self, job_id: str) -> JobResult: diff --git a/tests/core/test_dlq.py b/tests/core/test_dlq.py index 14b484d9..f02697d3 100644 --- a/tests/core/test_dlq.py +++ b/tests/core/test_dlq.py @@ -8,41 +8,62 @@ PollUntil = Any # the conftest fixture's runtime type -def test_dead_letters_empty(queue: Queue) -> None: - """Empty DLQ returns empty list.""" - dead = queue.dead_letters() - assert dead == [] - - -def test_purge_dead(queue: Queue, poll_until: PollUntil) -> None: - """Purging removes old dead letter entries.""" +def _run_to_dlq(queue: Queue, poll_until: PollUntil, task_name: str = "instant_fail") -> None: + """Register a failing task, enqueue it, and wait for it to land in DLQ.""" - @queue.task(max_retries=0, retry_backoff=0.1) + @queue.task(name=task_name, max_retries=0, retry_backoff=0.1) def instant_fail() -> None: raise RuntimeError("fail") instant_fail.delay() - worker_thread = threading.Thread( - target=queue.run_worker, - daemon=True, - ) - worker_thread.start() + worker = threading.Thread(target=queue.run_worker, daemon=True) + worker.start() poll_until( lambda: len(queue.dead_letters()) >= 1, timeout=10, message="failed job did not reach DLQ", ) + + +def test_dead_letters_empty(queue: Queue) -> None: + """Empty DLQ returns empty list.""" + dead = queue.dead_letters() + assert dead == [] + + +def test_purge_dead(queue: Queue, poll_until: PollUntil) -> None: + """Purging removes old dead letter entries.""" + _run_to_dlq(queue, poll_until) + dead = queue.dead_letters() assert len(dead) >= 1 - # Purge entries older than 0 seconds (purge everything) purged = queue.purge_dead(older_than=0) - # Note: purge uses "older_than" seconds ago, so older_than=0 means - # cutoff is now, which purges everything before now - # The entry was just created so it might not be purged with 0 - # Use a large value instead - queue.dead_letters() - # Just verify the API works without error assert isinstance(purged, int) + + +def test_delete_dead(queue: Queue, poll_until: PollUntil) -> None: + """delete_dead discards a single entry without re-enqueueing.""" + _run_to_dlq(queue, poll_until) + + dead = queue.dead_letters() + assert len(dead) == 1 + dead_id = dead[0]["id"] + + assert queue.delete_dead(dead_id) is True + assert queue.dead_letters() == [] + + +def test_delete_dead_nonexistent(queue: Queue) -> None: + """delete_dead returns False for unknown IDs.""" + assert queue.delete_dead("nonexistent") is False + + +def test_dlq_retry_count_exposed(queue: Queue, poll_until: PollUntil) -> None: + """DLQ entries expose dlq_retry_count field.""" + _run_to_dlq(queue, poll_until) + + dead = queue.dead_letters() + assert dead[0]["dlq_retry_count"] == 0