feat: dlq policies — auto-retry, per-entry TTL, delete

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<u8>>) -> Vec<u8> {
         match args {

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<i64>,
+    /// 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
     }
 }

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::<serde_json::Value>(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::<serde_json::Map<String, serde_json::Value>>(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<bool> {
+                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<u64> {
+                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<Vec<DeadJob>> {
+                let mut conn = self.conn()?;
+
+                let rows: Vec<DeadLetterRow> = 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())
+            }
         }
     };
 }

crates/taskito-core/src/storage/diesel_common/migrations.rs

@@ -360,6 +360,8 @@ pub fn alter_statements(d: &Dialect) -> Vec<String> {
         // 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"),
     ]
 }
 

crates/taskito-core/src/storage/mod.rs

@@ -52,6 +52,7 @@ pub struct DeadJob {
     pub timeout_ms: i64,
     pub result_ttl_ms: Option<i64>,
     pub namespace: Option<String>,
+    pub dlq_retry_count: i32,
 }
 
 impl From<models::DeadLetterRow> for DeadJob {
@@ -72,6 +73,7 @@ impl From<models::DeadLetterRow> 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<u64> {
                 self.purge_dead(older_than_ms)
             }
+            fn delete_dead(&self, dead_id: &str) -> $crate::error::Result<bool> {
+                self.delete_dead(dead_id)
+            }
+            fn purge_dead_with_ttl(&self, global_cutoff_ms: i64) -> $crate::error::Result<u64> {
+                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<Vec<$crate::storage::DeadJob>> {
+                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<u64> {
         delegate!(self, purge_dead, older_than_ms)
     }
+    fn delete_dead(&self, dead_id: &str) -> Result<bool> {
+        delegate!(self, delete_dead, dead_id)
+    }
+    fn purge_dead_with_ttl(&self, global_cutoff_ms: i64) -> Result<u64> {
+        delegate!(self, purge_dead_with_ttl, global_cutoff_ms)
+    }
+    fn list_dead_for_retry(
+        &self,
+        cutoff_ms: i64,
+        max_retries: i32,
+        limit: i64,
+    ) -> Result<Vec<DeadJob>> {
+        delegate!(self, list_dead_for_retry, cutoff_ms, max_retries, limit)
+    }
     fn get_rate_limit(&self, key: &str) -> Result<Option<models::RateLimitRow>> {
         delegate!(self, get_rate_limit, key)
     }

crates/taskito-core/src/storage/models.rs

@@ -130,6 +130,7 @@ pub struct DeadLetterRow {
     pub timeout_ms: i64,
     pub result_ttl_ms: Option<i64>,
     pub namespace: Option<String>,
+    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<i64>,
     pub namespace: Option<&'a str>,
+    pub dlq_retry_count: i32,
 }
 
 /// A row in the `rate_limits` table.