feat: per-task dead-letter list + purge

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

@@ -93,6 +93,45 @@ macro_rules! impl_diesel_dead_letter_ops {
                 Ok(rows.into_iter().map(DeadJob::from).collect())
             }
 
+            /// List dead letter entries for a single task, newest first.
+            pub fn list_dead_by_task(
+                &self,
+                task_name: &str,
+                limit: i64,
+                offset: i64,
+            ) -> Result<Vec<DeadJob>> {
+                // Normalize pagination so every backend agrees: a non-positive
+                // limit yields no page (matches Redis), and offset never goes
+                // negative.
+                if limit <= 0 {
+                    return Ok(Vec::new());
+                }
+                let offset = offset.max(0);
+
+                let mut conn = self.conn()?;
+
+                let rows: Vec<DeadLetterRow> = dead_letter::table
+                    .filter(dead_letter::task_name.eq(task_name))
+                    .order(dead_letter::failed_at.desc())
+                    .limit(limit)
+                    .offset(offset)
+                    .select(DeadLetterRow::as_select())
+                    .load(&mut conn)?;
+
+                Ok(rows.into_iter().map(DeadJob::from).collect())
+            }
+
+            /// Delete every dead letter entry for a task. Returns the count removed.
+            pub fn purge_dead_by_task(&self, task_name: &str) -> Result<u64> {
+                let mut conn = self.conn()?;
+
+                let affected =
+                    diesel::delete(dead_letter::table.filter(dead_letter::task_name.eq(task_name)))
+                        .execute(&mut conn)?;
+
+                Ok(affected as u64)
+            }
+
             /// Re-enqueue a dead letter job. Returns the new job ID.
             ///
             /// Returns `JobNotFound` only when the dead-letter row is absent or

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

@@ -244,6 +244,17 @@ macro_rules! impl_storage {
             ) -> $crate::error::Result<Vec<$crate::storage::DeadJob>> {
                 self.list_dead(limit, offset)
             }
+            fn list_dead_by_task(
+                &self,
+                task_name: &str,
+                limit: i64,
+                offset: i64,
+            ) -> $crate::error::Result<Vec<$crate::storage::DeadJob>> {
+                self.list_dead_by_task(task_name, limit, offset)
+            }
+            fn purge_dead_by_task(&self, task_name: &str) -> $crate::error::Result<u64> {
+                self.purge_dead_by_task(task_name)
+            }
             fn retry_dead(&self, dead_id: &str) -> $crate::error::Result<String> {
                 self.retry_dead(dead_id)
             }
@@ -767,6 +778,12 @@ impl Storage for StorageBackend {
     fn list_dead(&self, limit: i64, offset: i64) -> Result<Vec<DeadJob>> {
         delegate!(self, list_dead, limit, offset)
     }
+    fn list_dead_by_task(&self, task_name: &str, limit: i64, offset: i64) -> Result<Vec<DeadJob>> {
+        delegate!(self, list_dead_by_task, task_name, limit, offset)
+    }
+    fn purge_dead_by_task(&self, task_name: &str) -> Result<u64> {
+        delegate!(self, purge_dead_by_task, task_name)
+    }
     fn retry_dead(&self, dead_id: &str) -> Result<String> {
         delegate!(self, retry_dead, dead_id)
     }

crates/taskito-core/src/storage/redis_backend/dead_letter.rs

@@ -157,6 +157,84 @@ impl RedisStorage {
         Ok(results)
     }
 
+    pub fn list_dead_by_task(
+        &self,
+        task_name: &str,
+        limit: i64,
+        offset: i64,
+    ) -> Result<Vec<DeadJob>> {
+        // Mirror `list_dead`'s non-positive-limit convention (Redis LIMIT with a
+        // count of 0 returns nothing): no page to build.
+        if limit <= 0 {
+            return Ok(Vec::new());
+        }
+
+        let mut conn = self.conn()?;
+        let dlq_all = self.key(&["dlq", "all"]);
+        let offset = offset.max(0) as usize;
+        let limit = limit as usize;
+
+        // No per-task index by design: scan the global DLQ set newest-first and
+        // filter on `task_name`, paginating after the filter.
+        let ids: Vec<String> = conn.zrevrange(&dlq_all, 0, -1).map_err(map_err)?;
+
+        let mut matches = Vec::new();
+        for id in ids {
+            let dlq_key = self.key(&["dlq", &id]);
+            let data: Option<String> = conn.get(&dlq_key).map_err(map_err)?;
+            if let Some(d) = data {
+                let entry: DeadJobEntry =
+                    serde_json::from_str(&d).map_err(|e| QueueError::Other(e.to_string()))?;
+                if entry.task_name == task_name {
+                    matches.push(DeadJob::from(entry));
+                    // Stop once we have enough matches to satisfy this page.
+                    // saturating: offset/limit are public i64 inputs.
+                    if matches.len() >= offset.saturating_add(limit) {
+                        break;
+                    }
+                }
+            }
+        }
+
+        Ok(matches.into_iter().skip(offset).take(limit).collect())
+    }
+
+    pub fn purge_dead_by_task(&self, task_name: &str) -> Result<u64> {
+        let mut conn = self.conn()?;
+        let dlq_all = self.key(&["dlq", "all"]);
+
+        // No per-task index by design: scan the global DLQ set and collect the
+        // ids whose entry matches `task_name`.
+        let ids: Vec<String> = conn.zrevrange(&dlq_all, 0, -1).map_err(map_err)?;
+
+        let mut to_delete = Vec::new();
+        for id in ids {
+            let dlq_key = self.key(&["dlq", &id]);
+            let data: Option<String> = conn.get(&dlq_key).map_err(map_err)?;
+            if let Some(d) = data {
+                // Propagate (don't skip) on a corrupt entry: silently ignoring it
+                // would leave a task-owned row behind and under-report the count.
+                let entry: DeadJobEntry =
+                    serde_json::from_str(&d).map_err(|e| QueueError::Other(e.to_string()))?;
+                if entry.task_name == task_name {
+                    to_delete.push(id);
+                }
+            }
+        }
+
+        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 retry_dead(&self, dead_id: &str) -> Result<String> {
         let mut conn = self.conn()?;
         let dlq_key = self.key(&["dlq", dead_id]);

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

@@ -78,6 +78,10 @@ pub trait Storage: Send + Sync + Clone {
 
     fn move_to_dlq(&self, job: &Job, error: &str, metadata: Option<&str>) -> Result<()>;
     fn list_dead(&self, limit: i64, offset: i64) -> Result<Vec<DeadJob>>;
+    /// Dead-letter entries for one task, newest first, paginated.
+    fn list_dead_by_task(&self, task_name: &str, limit: i64, offset: i64) -> Result<Vec<DeadJob>>;
+    /// Delete every dead-letter entry for a task. Returns the number removed.
+    fn purge_dead_by_task(&self, task_name: &str) -> Result<u64>;
     fn retry_dead(&self, dead_id: &str) -> Result<String>;
     fn purge_dead(&self, older_than_ms: i64) -> Result<u64>;
     fn delete_dead(&self, dead_id: &str) -> Result<bool>;

crates/taskito-core/tests/rust/storage_tests.rs

@@ -251,6 +251,38 @@ fn test_dead_letter_queue(s: &impl Storage) {
     assert!(!dead.is_empty());
 }
 
+fn test_dead_letter_by_task(s: &impl Storage) {
+    let q = "q-dlq-by-task";
+
+    // Move 2x "task_a" and 1x "task_b" to the DLQ.
+    let move_to_dlq = |task_name: &str| {
+        let job = s.enqueue(make_job(q, task_name)).unwrap();
+        s.dequeue(q, now_millis() + 1000, None).unwrap();
+        let running = s.get_job(&job.id).unwrap().unwrap();
+        s.move_to_dlq(&running, "boom", None).unwrap();
+    };
+    move_to_dlq("task_a");
+    move_to_dlq("task_a");
+    move_to_dlq("task_b");
+
+    let task_a = s.list_dead_by_task("task_a", 10, 0).unwrap();
+    assert_eq!(task_a.len(), 2);
+    assert!(task_a.iter().all(|d| d.task_name == "task_a"));
+
+    // Pagination: one entry per page.
+    let page = s.list_dead_by_task("task_a", 1, 1).unwrap();
+    assert_eq!(page.len(), 1);
+    assert_eq!(page[0].task_name, "task_a");
+
+    // Purge removes only the matching task's entries.
+    assert_eq!(s.purge_dead_by_task("task_a").unwrap(), 2);
+    assert!(s.list_dead_by_task("task_a", 10, 0).unwrap().is_empty());
+
+    let task_b = s.list_dead_by_task("task_b", 10, 0).unwrap();
+    assert_eq!(task_b.len(), 1);
+    assert_eq!(task_b[0].task_name, "task_b");
+}
+
 fn test_delete_dead(s: &impl Storage) {
     let q = "q-del-dead";
     let job = s.enqueue(make_job(q, "del_dead_task")).unwrap();
@@ -692,6 +724,7 @@ fn run_storage_tests(s: &impl Storage) {
     test_enqueue_unique_validates_deps(s);
     test_enqueue_batch(s);
     test_dead_letter_queue(s);
+    test_dead_letter_by_task(s);
     test_delete_dead(s);
     test_list_dead_for_retry(s);
     test_progress_tracking(s);

crates/taskito-java/src/queue/admin.rs

@@ -70,6 +70,43 @@ pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_purgeDead<
     })
 }
 
+/// `String listDeadByTask(long handle, String taskName, long limit, long offset)`
+/// — dead-letter entries for one task, as a JSON array.
+#[no_mangle]
+pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_listDeadByTask<'local>(
+    mut env: JNIEnv<'local>,
+    _class: JClass<'local>,
+    handle: jlong,
+    task_name: JString<'local>,
+    limit: jlong,
+    offset: jlong,
+) -> jstring {
+    guard(&mut env, std::ptr::null_mut(), |env| {
+        let queue = unsafe { borrow_queue(handle) };
+        let task = read_string(env, &task_name)?;
+        let dead = queue
+            .storage
+            .list_dead_by_task(&task, limit.max(0), offset.max(0))?;
+        let views: Vec<DeadJobView> = dead.iter().map(DeadJobView::from).collect();
+        new_string(env, to_json(&views)?)
+    })
+}
+
+/// `long purgeDeadByTask(long handle, String taskName)` — returns rows removed.
+#[no_mangle]
+pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_purgeDeadByTask<'local>(
+    mut env: JNIEnv<'local>,
+    _class: JClass<'local>,
+    handle: jlong,
+    task_name: JString<'local>,
+) -> jlong {
+    guard(&mut env, 0, |env| {
+        let queue = unsafe { borrow_queue(handle) };
+        let task = read_string(env, &task_name)?;
+        Ok(queue.storage.purge_dead_by_task(&task)? as jlong)
+    })
+}
+
 /// `long purgeCompleted(long handle, long olderThanMs)` — returns rows removed.
 #[no_mangle]
 pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_purgeCompleted<'local>(

crates/taskito-node/src/queue/admin.rs

@@ -23,6 +23,32 @@ impl JsQueue {
         Ok(dead.into_iter().map(dead_job_to_js).collect())
     }
 
+    /// List dead-letter entries for a single task (paginated, newest first).
+    #[napi]
+    pub fn dead_letters_by_task(
+        &self,
+        task_name: String,
+        limit: Option<i64>,
+        offset: Option<i64>,
+    ) -> Result<Vec<JsDeadJob>> {
+        let limit = non_negative(limit.unwrap_or(DEFAULT_LIMIT), "limit")?;
+        let offset = non_negative(offset.unwrap_or(0), "offset")?;
+        let dead = self
+            .storage
+            .list_dead_by_task(&task_name, limit, offset)
+            .map_err(to_napi_err)?;
+        Ok(dead.into_iter().map(dead_job_to_js).collect())
+    }
+
+    /// Delete every dead-letter entry for a task. Returns the count removed.
+    #[napi]
+    pub fn purge_dead_by_task(&self, task_name: String) -> Result<i64> {
+        self.storage
+            .purge_dead_by_task(&task_name)
+            .map(|n| n as i64)
+            .map_err(to_napi_err)
+    }
+
     /// Re-enqueue a dead-letter entry. Returns the new job id.
     #[napi]
     pub fn retry_dead(&self, dead_id: String) -> Result<String> {

crates/taskito-python/src/py_queue/inspection.rs

@@ -110,6 +110,50 @@ impl PyQueue {
         })
     }
 
+    /// List dead letter queue entries for a single task, newest first.
+    #[pyo3(signature = (task_name, limit=10, offset=0))]
+    pub fn dead_letters_by_task(
+        &self,
+        task_name: &str,
+        limit: i64,
+        offset: i64,
+    ) -> PyResult<Vec<Py<PyAny>>> {
+        if limit < 0 || offset < 0 {
+            return Err(pyo3::exceptions::PyValueError::new_err(
+                "limit and offset must be non-negative",
+            ));
+        }
+        let dead = self
+            .storage
+            .list_dead_by_task(task_name, limit, offset)
+            .map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(e.to_string()))?;
+
+        Python::attach(|py| {
+            let mut result = Vec::with_capacity(dead.len());
+            for d in dead {
+                let dict = PyDict::new(py);
+                dict.set_item("id", d.id)?;
+                dict.set_item("original_job_id", d.original_job_id)?;
+                dict.set_item("queue", d.queue)?;
+                dict.set_item("task_name", d.task_name)?;
+                dict.set_item("error", d.error)?;
+                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)
+        })
+    }
+
+    /// Purge every dead letter entry for a task. Returns the count removed.
+    pub fn purge_dead_by_task(&self, task_name: &str) -> PyResult<u64> {
+        self.storage
+            .purge_dead_by_task(task_name)
+            .map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(e.to_string()))
+    }
+
     /// Re-enqueue a dead letter job. Returns new job ID.
     pub fn retry_dead(&self, dead_id: &str) -> PyResult<String> {
         self.storage

sdks/java/src/main/java/org/byteveda/taskito/DefaultQueue.java

@@ -201,6 +201,16 @@ public List<DeadJob> listDead(long limit, long offset) {
         return decodeList(backend.listDeadJson(limit, offset), DeadJob.class);
     }
 
+    @Override
+    public List<DeadJob> listDeadByTask(String taskName, long limit, long offset) {
+        return decodeList(backend.listDeadByTaskJson(taskName, limit, offset), DeadJob.class);
+    }
+
+    @Override
+    public long purgeDeadByTask(String taskName) {
+        return backend.purgeDeadByTask(taskName);
+    }
+
     @Override
     public String retryDead(String deadId) {
         return backend.retryDead(deadId);

sdks/java/src/main/java/org/byteveda/taskito/Queue.java

@@ -87,6 +87,12 @@ public interface Queue extends AutoCloseable {
 
     List<DeadJob> listDead(long limit, long offset);
 
+    /** Dead-letter entries for a single task, newest first. */
+    List<DeadJob> listDeadByTask(String taskName, long limit, long offset);
+
+    /** Delete every dead-letter entry for a task; returns the number removed. */
+    long purgeDeadByTask(String taskName);
+
     /** Re-enqueue a dead-letter entry; returns the new job id. */
     String retryDead(String deadId);
 

sdks/java/src/main/java/org/byteveda/taskito/internal/JniQueueBackend.java

@@ -114,6 +114,16 @@ public long purgeDead(long olderThanMs) {
         return NativeQueue.purgeDead(handle, olderThanMs);
     }
 
+    @Override
+    public String listDeadByTaskJson(String taskName, long limit, long offset) {
+        return NativeQueue.listDeadByTask(handle, taskName, limit, offset);
+    }
+
+    @Override
+    public long purgeDeadByTask(String taskName) {
+        return NativeQueue.purgeDeadByTask(handle, taskName);
+    }
+
     @Override
     public long purgeCompleted(long olderThanMs) {
         return NativeQueue.purgeCompleted(handle, olderThanMs);