From 8be53b0243daae5ab60a3efc09c127e0d586a487 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Sat, 27 Jun 2026 14:58:37 +0530 Subject: [PATCH 1/7] feat(core): per-task dead-letter list + purge list_dead_by_task / purge_dead_by_task across SQLite/Postgres/Redis. Diesel filters by task_name; Redis scans dlq:all and filters (no secondary index). --- .../src/storage/diesel_common/dead_letter.rs | 31 ++++++++ crates/taskito-core/src/storage/mod.rs | 17 +++++ .../src/storage/redis_backend/dead_letter.rs | 75 +++++++++++++++++++ crates/taskito-core/src/storage/traits.rs | 4 + .../taskito-core/tests/rust/storage_tests.rs | 33 ++++++++ 5 files changed, 160 insertions(+) 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 19962019..d3c32572 100644 --- a/crates/taskito-core/src/storage/diesel_common/dead_letter.rs +++ b/crates/taskito-core/src/storage/diesel_common/dead_letter.rs @@ -93,6 +93,37 @@ 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> { + let mut conn = self.conn()?; + + let rows: Vec = 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 { + 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 diff --git a/crates/taskito-core/src/storage/mod.rs b/crates/taskito-core/src/storage/mod.rs index 7c8a3f46..30722867 100644 --- a/crates/taskito-core/src/storage/mod.rs +++ b/crates/taskito-core/src/storage/mod.rs @@ -244,6 +244,17 @@ macro_rules! impl_storage { ) -> $crate::error::Result> { self.list_dead(limit, offset) } + fn list_dead_by_task( + &self, + task_name: &str, + limit: i64, + offset: i64, + ) -> $crate::error::Result> { + self.list_dead_by_task(task_name, limit, offset) + } + fn purge_dead_by_task(&self, task_name: &str) -> $crate::error::Result { + self.purge_dead_by_task(task_name) + } fn retry_dead(&self, dead_id: &str) -> $crate::error::Result { self.retry_dead(dead_id) } @@ -767,6 +778,12 @@ impl Storage for StorageBackend { fn list_dead(&self, limit: i64, offset: i64) -> Result> { delegate!(self, list_dead, limit, offset) } + fn list_dead_by_task(&self, task_name: &str, limit: i64, offset: i64) -> Result> { + delegate!(self, list_dead_by_task, task_name, limit, offset) + } + fn purge_dead_by_task(&self, task_name: &str) -> Result { + delegate!(self, purge_dead_by_task, task_name) + } fn retry_dead(&self, dead_id: &str) -> Result { delegate!(self, retry_dead, dead_id) } 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 9716de9f..27e8dbf2 100644 --- a/crates/taskito-core/src/storage/redis_backend/dead_letter.rs +++ b/crates/taskito-core/src/storage/redis_backend/dead_letter.rs @@ -157,6 +157,81 @@ impl RedisStorage { Ok(results) } + pub fn list_dead_by_task( + &self, + task_name: &str, + limit: i64, + offset: i64, + ) -> Result> { + // 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 = 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 = 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. + if matches.len() >= offset + limit { + break; + } + } + } + } + + Ok(matches.into_iter().skip(offset).take(limit).collect()) + } + + pub fn purge_dead_by_task(&self, task_name: &str) -> Result { + 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 = 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 = conn.get(&dlq_key).map_err(map_err)?; + if let Some(d) = data { + if let Ok(entry) = serde_json::from_str::(&d) { + 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 { let mut conn = self.conn()?; let dlq_key = self.key(&["dlq", dead_id]); diff --git a/crates/taskito-core/src/storage/traits.rs b/crates/taskito-core/src/storage/traits.rs index 34b9696e..51519b7e 100644 --- a/crates/taskito-core/src/storage/traits.rs +++ b/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>; + /// Dead-letter entries for one task, newest first, paginated. + fn list_dead_by_task(&self, task_name: &str, limit: i64, offset: i64) -> Result>; + /// Delete every dead-letter entry for a task. Returns the number removed. + fn purge_dead_by_task(&self, task_name: &str) -> 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; diff --git a/crates/taskito-core/tests/rust/storage_tests.rs b/crates/taskito-core/tests/rust/storage_tests.rs index 9cdee058..1fa114e6 100644 --- a/crates/taskito-core/tests/rust/storage_tests.rs +++ b/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); From 80add6ffef5c6f92acf4fea5b2d44eb8a9297e79 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Sat, 27 Jun 2026 14:58:48 +0530 Subject: [PATCH 2/7] feat(python): expose per-task dead-letter list + purge --- .../taskito-python/src/py_queue/inspection.rs | 44 +++++++++++++++++++ sdks/python/taskito/_taskito.pyi | 4 ++ 2 files changed, 48 insertions(+) diff --git a/crates/taskito-python/src/py_queue/inspection.rs b/crates/taskito-python/src/py_queue/inspection.rs index 34220f5e..8fcd42a8 100644 --- a/crates/taskito-python/src/py_queue/inspection.rs +++ b/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>> { + 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 { + 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 { self.storage diff --git a/sdks/python/taskito/_taskito.pyi b/sdks/python/taskito/_taskito.pyi index ea43dfc4..4a4a8fdf 100644 --- a/sdks/python/taskito/_taskito.pyi +++ b/sdks/python/taskito/_taskito.pyi @@ -164,6 +164,10 @@ class PyQueue: namespace: str | None = None, ) -> list[PyJob]: ... def dead_letters(self, limit: int = 10, offset: int = 0) -> list[dict[str, Any]]: ... + def dead_letters_by_task( + self, task_name: str, limit: int = 10, offset: int = 0 + ) -> list[dict[str, Any]]: ... + def purge_dead_by_task(self, task_name: str) -> int: ... 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: ... From e4820f2ece0697ea17d233278e268a85d4d9d176 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Sat, 27 Jun 2026 14:59:01 +0530 Subject: [PATCH 3/7] feat(java): per-task dead-letter list + purge Queue.listDeadByTask / purgeDeadByTask over the new core ops, with an in-memory backend impl. Live-verified by DeadLetterByTaskTest. --- crates/taskito-java/src/queue/admin.rs | 37 ++++++++++++ .../org/byteveda/taskito/DefaultQueue.java | 10 ++++ .../main/java/org/byteveda/taskito/Queue.java | 6 ++ .../taskito/internal/JniQueueBackend.java | 10 ++++ .../taskito/internal/NativeQueue.java | 6 ++ .../byteveda/taskito/spi/QueueBackend.java | 10 ++++ .../taskito/DeadLetterByTaskTest.java | 57 +++++++++++++++++++ .../taskito/test/InMemoryQueueBackend.java | 20 +++++++ 8 files changed, 156 insertions(+) create mode 100644 sdks/java/src/test/java/org/byteveda/taskito/DeadLetterByTaskTest.java diff --git a/crates/taskito-java/src/queue/admin.rs b/crates/taskito-java/src/queue/admin.rs index 91b6a479..5aa1f356 100644 --- a/crates/taskito-java/src/queue/admin.rs +++ b/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 = 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>( diff --git a/sdks/java/src/main/java/org/byteveda/taskito/DefaultQueue.java b/sdks/java/src/main/java/org/byteveda/taskito/DefaultQueue.java index 285c42f5..0441d5d3 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/DefaultQueue.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/DefaultQueue.java @@ -201,6 +201,16 @@ public List listDead(long limit, long offset) { return decodeList(backend.listDeadJson(limit, offset), DeadJob.class); } + @Override + public List 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); diff --git a/sdks/java/src/main/java/org/byteveda/taskito/Queue.java b/sdks/java/src/main/java/org/byteveda/taskito/Queue.java index c05e26ad..5a5258a1 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/Queue.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/Queue.java @@ -87,6 +87,12 @@ public interface Queue extends AutoCloseable { List listDead(long limit, long offset); + /** Dead-letter entries for a single task, newest first. */ + List 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); diff --git a/sdks/java/src/main/java/org/byteveda/taskito/internal/JniQueueBackend.java b/sdks/java/src/main/java/org/byteveda/taskito/internal/JniQueueBackend.java index 077ef919..768e3bb4 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/internal/JniQueueBackend.java +++ b/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); diff --git a/sdks/java/src/main/java/org/byteveda/taskito/internal/NativeQueue.java b/sdks/java/src/main/java/org/byteveda/taskito/internal/NativeQueue.java index 39843c8d..b43c7424 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/internal/NativeQueue.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/internal/NativeQueue.java @@ -62,6 +62,12 @@ private NativeQueue() {} public static native long purgeDead(long handle, long olderThanMs); + /** Dead-letter entries for one task, as a JSON array. */ + public static native String listDeadByTask(long handle, String taskName, long limit, long offset); + + /** Delete every dead-letter entry for a task; returns the number removed. */ + public static native long purgeDeadByTask(long handle, String taskName); + public static native long purgeCompleted(long handle, long olderThanMs); public static native void pauseQueue(long handle, String queue); diff --git a/sdks/java/src/main/java/org/byteveda/taskito/spi/QueueBackend.java b/sdks/java/src/main/java/org/byteveda/taskito/spi/QueueBackend.java index 80151ebb..c9913e27 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/spi/QueueBackend.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/spi/QueueBackend.java @@ -61,6 +61,16 @@ public interface QueueBackend extends AutoCloseable { long purgeDead(long olderThanMs); + /** A JSON array of dead-letter entries for one task. */ + default String listDeadByTaskJson(String taskName, long limit, long offset) { + throw new UnsupportedOperationException("per-task dead-letter queries not supported by this backend"); + } + + /** Delete every dead-letter entry for a task; returns the number removed. */ + default long purgeDeadByTask(String taskName) { + throw new UnsupportedOperationException("per-task dead-letter queries not supported by this backend"); + } + long purgeCompleted(long olderThanMs); void pauseQueue(String queue); diff --git a/sdks/java/src/test/java/org/byteveda/taskito/DeadLetterByTaskTest.java b/sdks/java/src/test/java/org/byteveda/taskito/DeadLetterByTaskTest.java new file mode 100644 index 00000000..2e91e56b --- /dev/null +++ b/sdks/java/src/test/java/org/byteveda/taskito/DeadLetterByTaskTest.java @@ -0,0 +1,57 @@ +package org.byteveda.taskito; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.nio.file.Path; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import org.byteveda.taskito.events.EventName; +import org.byteveda.taskito.task.Task; +import org.byteveda.taskito.worker.Worker; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; +import org.junit.jupiter.api.io.TempDir; + +/** Dead-letter entries can be listed and purged scoped to a single task. */ +class DeadLetterByTaskTest { + + @Test + @Timeout(30) + void listAndPurgeByTask(@TempDir Path dir) throws Exception { + // maxRetries(0) → a thrown handler dead-letters immediately. + Task alpha = Task.of("alpha", String.class).maxRetries(0); + Task beta = Task.of("beta", String.class).maxRetries(0); + + try (Queue queue = Taskito.builder() + .backend("sqlite") + .url(dir.resolve("t.db").toString()) + .open()) { + queue.enqueue(alpha, "1"); + queue.enqueue(alpha, "2"); + queue.enqueue(beta, "3"); + + CountDownLatch dead = new CountDownLatch(3); + try (Worker worker = queue.worker() + .handle(alpha, (String p) -> { + throw new IllegalStateException("boom"); + }) + .handle(beta, (String p) -> { + throw new IllegalStateException("boom"); + }) + .on(EventName.DEAD, event -> dead.countDown()) + .start()) { + assertTrue(dead.await(20, TimeUnit.SECONDS), "all three should dead-letter"); + } + + assertEquals(2, queue.listDeadByTask("alpha", 10, 0).size()); + assertEquals(1, queue.listDeadByTask("beta", 10, 0).size()); + // Pagination applies within the task's own entries. + assertEquals(1, queue.listDeadByTask("alpha", 1, 1).size()); + + assertEquals(2, queue.purgeDeadByTask("alpha")); + assertEquals(0, queue.listDeadByTask("alpha", 10, 0).size()); + assertEquals(1, queue.listDead(10, 0).size(), "beta entry survives"); + } + } +} diff --git a/sdks/java/test-support/src/main/java/org/byteveda/taskito/test/InMemoryQueueBackend.java b/sdks/java/test-support/src/main/java/org/byteveda/taskito/test/InMemoryQueueBackend.java index 4358659c..f6f389d7 100644 --- a/sdks/java/test-support/src/main/java/org/byteveda/taskito/test/InMemoryQueueBackend.java +++ b/sdks/java/test-support/src/main/java/org/byteveda/taskito/test/InMemoryQueueBackend.java @@ -250,6 +250,26 @@ public long purgeDead(long olderThanMs) { return before - dead.size(); } + @Override + public String listDeadByTaskJson(String taskName, long limit, long offset) { + List> matches = new ArrayList<>(); + for (Map entry : dead) { + if (taskName.equals(entry.get("taskName"))) { + matches.add(entry); + } + } + int from = (int) Math.min(offset, matches.size()); + int to = (int) Math.min(from + limit, matches.size()); + return toJson(new ArrayList<>(matches.subList(from, to))); + } + + @Override + public long purgeDeadByTask(String taskName) { + int before = dead.size(); + dead.removeIf(entry -> taskName.equals(entry.get("taskName"))); + return before - dead.size(); + } + @Override public long purgeCompleted(long olderThanMs) { List remove = new ArrayList<>(); From 121c1928c9f1e07533fc90018da5b94558e41b6a Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Sat, 27 Jun 2026 14:59:03 +0530 Subject: [PATCH 4/7] feat(node): per-task dead-letter list + purge Queue.deadLettersByTask / purgeDeadByTask over the new core ops. --- crates/taskito-node/src/queue/admin.rs | 26 +++++++++++++++++++++++++ sdks/node/src/queue.ts | 10 ++++++++++ sdks/node/test/core/inspection.test.ts | 27 ++++++++++++++++++++++++++ 3 files changed, 63 insertions(+) diff --git a/crates/taskito-node/src/queue/admin.rs b/crates/taskito-node/src/queue/admin.rs index 7f2ab45e..48ee0712 100644 --- a/crates/taskito-node/src/queue/admin.rs +++ b/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, + offset: Option, + ) -> Result> { + 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 { + 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 { diff --git a/sdks/node/src/queue.ts b/sdks/node/src/queue.ts index f5f51eac..0bc9cd73 100644 --- a/sdks/node/src/queue.ts +++ b/sdks/node/src/queue.ts @@ -454,6 +454,16 @@ export class Queue { return this.native.deadLetters(limit, offset); } + /** List dead-letter entries for a single task (paginated, newest first). */ + deadLettersByTask(taskName: string, limit?: number, offset?: number): DeadJob[] { + return this.native.deadLettersByTask(taskName, limit, offset); + } + + /** Delete every dead-letter entry for a task. Returns the count removed. */ + purgeDeadByTask(taskName: string): number { + return this.native.purgeDeadByTask(taskName); + } + /** Re-enqueue a dead-letter entry. Returns the new job id. */ retryDead(deadId: string): string { return this.native.retryDead(deadId); diff --git a/sdks/node/test/core/inspection.test.ts b/sdks/node/test/core/inspection.test.ts index 97dfeea6..46f1de8b 100644 --- a/sdks/node/test/core/inspection.test.ts +++ b/sdks/node/test/core/inspection.test.ts @@ -64,6 +64,33 @@ it("dead-letters a failing job and can retry it", async () => { expect(typeof queue.retryDead(entry.id)).toBe("string"); }); +it("lists and purges dead-letter entries by task", async () => { + const queue = newQueue(); + const boom = () => { + throw new Error("nope"); + }; + queue.task("alpha", boom, { maxRetries: 0 }); + queue.task("beta", boom, { maxRetries: 0 }); + queue.enqueue("alpha"); + queue.enqueue("alpha"); + queue.enqueue("beta"); + worker = queue.runWorker(); + + const deadline = Date.now() + 3000; + while (Date.now() < deadline && queue.deadLetters().length < 3) { + await new Promise((resolve) => setTimeout(resolve, 25)); + } + + expect(queue.deadLettersByTask("alpha")).toHaveLength(2); + expect(queue.deadLettersByTask("beta")).toHaveLength(1); + // Pagination applies within the task's own entries. + expect(queue.deadLettersByTask("alpha", 1, 1)).toHaveLength(1); + + expect(queue.purgeDeadByTask("alpha")).toBe(2); + expect(queue.deadLettersByTask("alpha")).toHaveLength(0); + expect(queue.deadLetters()).toHaveLength(1); +}); + it("pauses and resumes a queue", () => { const queue = newQueue(); expect(queue.listPausedQueues()).not.toContain("default"); From c2db4f97892f2756947b8d403f1714e7b754cdf9 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Sat, 27 Jun 2026 15:15:34 +0530 Subject: [PATCH 5/7] fix(core): normalize DLQ-by-task pagination, fail on corrupt entries Diesel now clamps limit/offset like Redis (non-positive limit = empty page); Redis purge propagates a deserialize error instead of silently skipping (which under-reported the count); saturating page target avoids i64->usize overflow. --- .../src/storage/diesel_common/dead_letter.rs | 8 ++++++++ .../src/storage/redis_backend/dead_letter.rs | 13 ++++++++----- 2 files changed, 16 insertions(+), 5 deletions(-) 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 d3c32572..70763b46 100644 --- a/crates/taskito-core/src/storage/diesel_common/dead_letter.rs +++ b/crates/taskito-core/src/storage/diesel_common/dead_letter.rs @@ -100,6 +100,14 @@ macro_rules! impl_diesel_dead_letter_ops { limit: i64, offset: i64, ) -> Result> { + // 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 = dead_letter::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 27e8dbf2..9c519b30 100644 --- a/crates/taskito-core/src/storage/redis_backend/dead_letter.rs +++ b/crates/taskito-core/src/storage/redis_backend/dead_letter.rs @@ -188,7 +188,8 @@ impl RedisStorage { if entry.task_name == task_name { matches.push(DeadJob::from(entry)); // Stop once we have enough matches to satisfy this page. - if matches.len() >= offset + limit { + // saturating: offset/limit are public i64 inputs. + if matches.len() >= offset.saturating_add(limit) { break; } } @@ -211,10 +212,12 @@ impl RedisStorage { 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.task_name == task_name { - to_delete.push(id); - } + // 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); } } } From a4bfa6ea0bffd60ad0a6919b878ce48163e975d9 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Sat, 27 Jun 2026 15:15:40 +0530 Subject: [PATCH 6/7] fix(java): count purgeDeadByTask removals, not a size diff --- .../byteveda/taskito/test/InMemoryQueueBackend.java | 13 ++++++++++--- 1 file changed, 10 insertions(+), 3 deletions(-) diff --git a/sdks/java/test-support/src/main/java/org/byteveda/taskito/test/InMemoryQueueBackend.java b/sdks/java/test-support/src/main/java/org/byteveda/taskito/test/InMemoryQueueBackend.java index f6f389d7..9a8e1296 100644 --- a/sdks/java/test-support/src/main/java/org/byteveda/taskito/test/InMemoryQueueBackend.java +++ b/sdks/java/test-support/src/main/java/org/byteveda/taskito/test/InMemoryQueueBackend.java @@ -265,9 +265,16 @@ public String listDeadByTaskJson(String taskName, long limit, long offset) { @Override public long purgeDeadByTask(String taskName) { - int before = dead.size(); - dead.removeIf(entry -> taskName.equals(entry.get("taskName"))); - return before - dead.size(); + // Count the entries actually removed (not a before/after size diff, which + // a concurrent onFail append could skew). + List> matches = new ArrayList<>(); + for (Map entry : dead) { + if (taskName.equals(entry.get("taskName"))) { + matches.add(entry); + } + } + dead.removeAll(matches); + return matches.size(); } @Override From 234682b0e1d82d5d19fa70801af1667068133695 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Sat, 27 Jun 2026 15:15:40 +0530 Subject: [PATCH 7/7] test(node): widen dead-letter wait budget for slow CI --- sdks/node/test/core/inspection.test.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/sdks/node/test/core/inspection.test.ts b/sdks/node/test/core/inspection.test.ts index 46f1de8b..db7ba6c8 100644 --- a/sdks/node/test/core/inspection.test.ts +++ b/sdks/node/test/core/inspection.test.ts @@ -76,7 +76,7 @@ it("lists and purges dead-letter entries by task", async () => { queue.enqueue("beta"); worker = queue.runWorker(); - const deadline = Date.now() + 3000; + const deadline = Date.now() + 10000; while (Date.now() < deadline && queue.deadLetters().length < 3) { await new Promise((resolve) => setTimeout(resolve, 25)); } @@ -100,7 +100,7 @@ it("pauses and resumes a queue", () => { expect(queue.listPausedQueues()).not.toContain("default"); }); -async function waitForDead(queue: Queue, timeoutMs = 3000): Promise { +async function waitForDead(queue: Queue, timeoutMs = 10000): Promise { const deadline = Date.now() + timeoutMs; while (Date.now() < deadline) { const dead = queue.deadLetters();