From 0d9526f33814761d2b489e6d8cf14630a7de6923 Mon Sep 17 00:00:00 2001 From: stromanni Date: Wed, 22 Jul 2026 21:09:31 +0530 Subject: [PATCH 1/4] feat(node): add a retryOn predicate to task options MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The JS callback now resolves a JsTaskOutcome instead of rejecting, so a failure can carry its retry verdict — a napi rejection is only a string. --- crates/taskito-node/src/convert/job.rs | 15 ++++ crates/taskito-node/src/convert/mod.rs | 2 +- crates/taskito-node/src/dispatcher.rs | 92 ++++++++++++++------- sdks/node/src/native.ts | 1 + sdks/node/src/types.ts | 11 +++ sdks/node/src/worker.ts | 35 ++++++-- sdks/node/test/worker/retryOn.test.ts | 109 +++++++++++++++++++++++++ 7 files changed, 225 insertions(+), 40 deletions(-) create mode 100644 sdks/node/test/worker/retryOn.test.ts diff --git a/crates/taskito-node/src/convert/job.rs b/crates/taskito-node/src/convert/job.rs index 0e4fc55e..b716318d 100644 --- a/crates/taskito-node/src/convert/job.rs +++ b/crates/taskito-node/src/convert/job.rs @@ -66,6 +66,21 @@ pub struct JsTaskInvocation { pub payload: Buffer, } +/// What the JS task callback resolves with: either a result or an error, never +/// both. Resolving (rather than rejecting) is what lets the shell say *how* a +/// failure should be treated — a rejection carries nothing but a string. +#[napi(object)] +pub struct JsTaskOutcome { + /// The serialized task result. Set only on success. + pub result: Option, + /// The failure, as the canonical cross-SDK `{errtype,message,traceback}` + /// JSON. Its presence is what marks the outcome a failure. + pub error: Option, + /// Whether the failure may be retried; absent means yes. `false` sends the + /// job straight to the dead-letter queue, whatever budget is left. + pub retryable: Option, +} + /// JS-facing view of a stored [`Job`]. `result` is the opaque result blob (or /// `null`); the shell deserializes it. #[napi(object)] diff --git a/crates/taskito-node/src/convert/mod.rs b/crates/taskito-node/src/convert/mod.rs index a25e7f27..2c9eec59 100644 --- a/crates/taskito-node/src/convert/mod.rs +++ b/crates/taskito-node/src/convert/mod.rs @@ -13,7 +13,7 @@ mod task_config; #[cfg(feature = "workflows")] mod workflow; -pub use job::{build_new_job, job_to_js, JsJob, JsJobPage, JsTaskInvocation}; +pub use job::{build_new_job, job_to_js, JsJob, JsJobPage, JsTaskInvocation, JsTaskOutcome}; pub(crate) use job::{DEFAULT_MAX_RETRIES, DEFAULT_PRIORITY, DEFAULT_TIMEOUT_MS}; pub use lock::{lock_info_to_js, JsLockInfo}; pub use log::{log_to_js, JsTaskLog}; diff --git a/crates/taskito-node/src/dispatcher.rs b/crates/taskito-node/src/dispatcher.rs index 39aec9b7..4f1c8ecc 100644 --- a/crates/taskito-node/src/dispatcher.rs +++ b/crates/taskito-node/src/dispatcher.rs @@ -2,7 +2,7 @@ //! //! Implements the core [`WorkerDispatcher`] trait. For every dequeued job it //! invokes the JS task callback via a `ThreadsafeFunction`, awaits the returned -//! `Promise`, and reports a [`JobResult`] back to the scheduler. This is +//! `Promise`, and reports a [`JobResult`] back to the scheduler. This is //! the Node mirror of the Python shell's worker pool. use std::sync::Arc; @@ -17,9 +17,9 @@ use taskito_core::worker::WorkerDispatcher; use taskito_core::{Storage, StorageBackend}; use tokio::sync::{oneshot, Semaphore}; -use crate::convert::JsTaskInvocation; +use crate::convert::{JsTaskInvocation, JsTaskOutcome}; -/// Task callback registered from JS: `(invocation) => Promise`. +/// Task callback registered from JS: `(invocation) => Promise`. type TaskCallback = ThreadsafeFunction; /// Executes jobs by dispatching them to a JavaScript callback. @@ -110,14 +110,15 @@ async fn run_one(callback: &TaskCallback, storage: &StorageBackend, mut job: Job }; // The callback runs on the JS thread and returns a Promise; bridge its - // awaited value back to this async context through a oneshot. - let (tx, rx) = oneshot::channel::>>(); + // awaited value back to this async context through a oneshot. The JS view + // is copied out (Buffer -> Vec) while still on that side. + let (tx, rx) = oneshot::channel::>(); callback.call_with_return_value( invocation, ThreadsafeFunctionCallMode::NonBlocking, - move |promise: Promise| { + move |promise: Promise| { spawn(async move { - let outcome = promise.await.map(|buffer| buffer.to_vec()); + let outcome = promise.await.map(TaskOutcome::from); let _ = tx.send(outcome); }); Ok(()) @@ -133,49 +134,80 @@ async fn run_one(callback: &TaskCallback, storage: &StorageBackend, mut job: Job }; let wall_time_ns = started.elapsed().as_nanos() as i64; match timed { - Ok(Ok(Ok(result))) => JobResult::Success { - job_id: job.id, - result: Some(result), - task_name: job.task_name, - wall_time_ns, - }, - Ok(Ok(Err(err))) => { - // A rejected task that was cancel-requested is a cancellation, not a + Ok(Ok(Ok(outcome))) => match outcome.error { + None => JobResult::Success { + job_id: job.id, + result: outcome.result, + task_name: job.task_name, + wall_time_ns, + }, + // A failed task that was cancel-requested is a cancellation, not a // failure (the JS side aborts via the cancel signal). - if storage.is_cancel_requested(&job.id).unwrap_or(false) { + Some(_) if storage.is_cancel_requested(&job.id).unwrap_or(false) => { JobResult::Cancelled { job_id: job.id, task_name: job.task_name, wall_time_ns, } - } else { - // `Error::to_string()` prepends the napi status ("GenericFailure, "); - // the bare reason is the JS error's string form, which the worker - // shapes into the cross-SDK structured-error JSON. - let reason = if err.reason.is_empty() { - err.to_string() - } else { - err.reason.clone() - }; - failure(job, reason, wall_time_ns, false) } + Some(error) => failure(job, error, wall_time_ns, false, outcome.retryable), + }, + // The promise rejected rather than resolving an outcome: the shell threw + // before it could shape one (an unregistered task, a payload it could + // not decode). Retryable — another worker may well have the task. + // `Error::to_string()` prepends the napi status ("GenericFailure, "), so + // prefer the bare reason, which is the JS error's string form. + Ok(Ok(Err(err))) => { + let reason = if err.reason.is_empty() { + err.to_string() + } else { + err.reason.clone() + }; + failure(job, reason, wall_time_ns, false, true) } Ok(Err(_)) => failure( job, "node task channel dropped".to_string(), wall_time_ns, false, + true, ), // We report the timeout, but the underlying JS promise cannot be force- // killed and keeps running in the background — same limitation as the // Python shell. Non-idempotent tasks should cooperate via the cancel // signal (`currentJob().signal`) and check it on long operations. - Err(_) => failure(job, "task timed out".to_string(), wall_time_ns, true), + Err(_) => failure(job, "task timed out".to_string(), wall_time_ns, true, true), + } +} + +/// One finished invocation, copied out of its JS view so it can cross threads. +struct TaskOutcome { + result: Option>, + error: Option, + /// Whether a failure may be retried. The shell's per-task `retryOn` + /// predicate decides; an absent flag means retry, as before it existed. + retryable: bool, +} + +impl From for TaskOutcome { + fn from(outcome: JsTaskOutcome) -> Self { + Self { + result: outcome.result.map(|buffer| buffer.to_vec()), + error: outcome.error, + retryable: outcome.retryable.unwrap_or(true), + } } } -/// Build a retryable [`JobResult::Failure`] from a job and error message. -fn failure(job: Job, error: String, wall_time_ns: i64, timed_out: bool) -> JobResult { +/// Build a [`JobResult::Failure`] from a job and error message. `should_retry` +/// false skips the retry budget entirely and dead-letters the job. +fn failure( + job: Job, + error: String, + wall_time_ns: i64, + timed_out: bool, + should_retry: bool, +) -> JobResult { JobResult::Failure { job_id: job.id, error, @@ -183,7 +215,7 @@ fn failure(job: Job, error: String, wall_time_ns: i64, timed_out: bool) -> JobRe max_retries: job.max_retries, task_name: job.task_name, wall_time_ns, - should_retry: true, + should_retry, timed_out, } } diff --git a/sdks/node/src/native.ts b/sdks/node/src/native.ts index 95723911..8d90406f 100644 --- a/sdks/node/src/native.ts +++ b/sdks/node/src/native.ts @@ -34,6 +34,7 @@ export type { JsSubscription, JsTaskInvocation, JsTaskLog, + JsTaskOutcome, JsTopic, JsTopicLogStat, JsTopicMessage, diff --git a/sdks/node/src/types.ts b/sdks/node/src/types.ts index a158e5bd..d99a50b8 100644 --- a/sdks/node/src/types.ts +++ b/sdks/node/src/types.ts @@ -126,6 +126,17 @@ export interface TaskOptions { maxRetries?: number; /** Exponential backoff bounds for retries. */ retryBackoff?: { baseMs?: number; maxMs?: number }; + /** + * Classifies a thrown error as retryable. Returning `false` dead-letters the + * job immediately, whatever retry budget is left — use it for permanent + * failures (a malformed payload, a 4xx) that no amount of retrying fixes. + * Unset retries everything, as does a predicate that throws. + * + * Synchronous by design: the job cannot settle until it answers. Applies to + * errors the handler throws — a timeout is detected outside the handler and + * always consumes a retry. + */ + retryOn?: (error: unknown) => boolean; /** Per-job timeout default (ms); enforced by the worker. */ timeoutMs?: number; /** Cap on concurrently-running jobs of this task, across the cluster. */ diff --git a/sdks/node/src/worker.ts b/sdks/node/src/worker.ts index d7172d9e..35b561ff 100644 --- a/sdks/node/src/worker.ts +++ b/sdks/node/src/worker.ts @@ -12,6 +12,7 @@ import type { Middleware, TaskContext } from "./middleware"; import type { JsOutcome, JsTaskInvocation, + JsTaskOutcome, NativeQueue, NativeWorker, WorkerOptions as NativeWorkerOptions, @@ -111,11 +112,11 @@ export class Worker { // Advance workflow runs as node-jobs settle, unless disabled or unsupported. const tracker = (run?.advanceWorkflows ?? true) ? (params.workflowTracker ?? null) : null; - const taskCallback = async (invocation: JsTaskInvocation): Promise => { + const taskCallback = async (invocation: JsTaskInvocation): Promise => { // Built-in workflow cache-return: echo the single (cached) arg as the result. if (invocation.taskName === CACHE_TASK) { const [value] = deserializeCall(serializer, invocation.payload); - return Buffer.from(serializer.serialize(value)); + return { result: Buffer.from(serializer.serialize(value)) }; } const task = tasks.get(invocation.taskName); if (!task) { @@ -186,7 +187,7 @@ export class Worker { for (const mw of chain) { await mw.after?.(ctx, result); } - return Buffer.from(serializer.serialize(result)); + return { result: Buffer.from(serializer.serialize(result)) }; } catch (error) { for (const mw of chain) { try { @@ -195,12 +196,10 @@ export class Worker { // onError hooks must not mask the original task failure. } } - // Rethrow as the canonical structured-error JSON. With an empty name, - // Error#toString() is just the message, so the native layer stores the - // bare JSON object as the job's error string. - const structured = new Error(encodeTaskError(error)); - structured.name = ""; - throw structured; + // Resolve rather than reject: a rejection carries only a string, and the + // native layer needs the retry verdict alongside the canonical + // structured-error JSON it stores as the job's error. + return { error: encodeTaskError(error), retryable: isRetryable(task, error) }; } finally { clearInterval(poller); try { @@ -336,6 +335,24 @@ export class Worker { } } +/** + * Ask a task's `retryOn` predicate whether `error` is worth retrying. No + * predicate means retry, and so does one that throws — a broken classifier must + * not silently turn transient failures into dead letters. + */ +function isRetryable(task: RegisteredTask, error: unknown): boolean { + const predicate = task.options?.retryOn; + if (!predicate) { + return true; + } + try { + return predicate(error); + } catch (predicateError) { + log.error(() => "retryOn predicate threw; retrying the failure", predicateError); + return true; + } +} + /** Start one poll loop per managed consumer; return their timers to clear on stop. */ function startLogConsumers( queue: NativeQueue, diff --git a/sdks/node/test/worker/retryOn.test.ts b/sdks/node/test/worker/retryOn.test.ts new file mode 100644 index 00000000..464cbabf --- /dev/null +++ b/sdks/node/test/worker/retryOn.test.ts @@ -0,0 +1,109 @@ +import { mkdtempSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, expect, it } from "vitest"; +import { type OutcomeEvent, Queue, type Worker } from "../../src/index"; + +let worker: Worker | undefined; + +afterEach(() => { + worker?.stop(); + worker = undefined; +}); + +function newQueue(): Queue { + return new Queue({ dbPath: join(mkdtempSync(join(tmpdir(), "taskito-retryon-")), "q.db") }); +} + +async function waitFor(predicate: () => boolean, timeoutMs = 5000): Promise { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + if (predicate()) { + return true; + } + await new Promise((resolve) => setTimeout(resolve, 20)); + } + return false; +} + +/** A permanent failure the predicate rejects. */ +class PermanentError extends Error {} + +it("dead-letters a rejected error without spending the retry budget", async () => { + const queue = newQueue(); + const dead: OutcomeEvent[] = []; + const retries: OutcomeEvent[] = []; + let attempts = 0; + + queue.on("job.dead", (event) => dead.push(event)); + queue.on("job.retrying", (event) => retries.push(event)); + queue.task( + "permanent", + () => { + attempts += 1; + throw new PermanentError("malformed input"); + }, + { maxRetries: 5, retryOn: (error) => !(error instanceof PermanentError) }, + ); + + queue.enqueue("permanent"); + worker = queue.runWorker(); + + expect(await waitFor(() => dead.length > 0)).toBe(true); + expect(attempts).toBe(1); + expect(retries).toHaveLength(0); +}); + +it("still retries an error the predicate accepts", async () => { + const queue = newQueue(); + const dead: OutcomeEvent[] = []; + let attempts = 0; + + queue.on("job.dead", (event) => dead.push(event)); + queue.task( + "transient", + () => { + attempts += 1; + throw new Error("connection reset"); + }, + { + maxRetries: 2, + retryBackoff: { baseMs: 1, maxMs: 5 }, + retryOn: (error) => !(error instanceof PermanentError), + }, + ); + + queue.enqueue("transient"); + worker = queue.runWorker(); + + expect(await waitFor(() => dead.length > 0)).toBe(true); + expect(attempts).toBe(3); // first run plus both retries +}); + +it("retries when the predicate itself throws", async () => { + const queue = newQueue(); + const dead: OutcomeEvent[] = []; + let attempts = 0; + + queue.on("job.dead", (event) => dead.push(event)); + queue.task( + "brokenPredicate", + () => { + attempts += 1; + throw new Error("boom"); + }, + { + maxRetries: 1, + retryBackoff: { baseMs: 1, maxMs: 5 }, + retryOn: () => { + throw new Error("predicate is broken"); + }, + }, + ); + + queue.enqueue("brokenPredicate"); + worker = queue.runWorker(); + + expect(await waitFor(() => dead.length > 0)).toBe(true); + expect(attempts).toBe(2); +}); From 787e354057d9f891dab2efefb612b7da8826647a Mon Sep 17 00:00:00 2001 From: stromanni Date: Wed, 22 Jul 2026 21:09:35 +0530 Subject: [PATCH 2/4] feat(java): add a retryOn predicate to tasks Predicate lives on Task, not RetryPolicy: RetryPolicy is wire-encoded into the scheduler's per-task config, which a lambda cannot cross. --- crates/taskito-java/src/dispatcher.rs | 32 +++++-- crates/taskito-java/src/worker.rs | 12 ++- .../taskito/internal/JniWorkerControl.java | 4 +- .../taskito/internal/NativeWorker.java | 2 +- .../byteveda/taskito/spi/WorkerControl.java | 3 +- .../java/org/byteveda/taskito/task/Task.java | 68 +++++++++++--- .../taskito/worker/RegisteredTask.java | 12 ++- .../org/byteveda/taskito/worker/Worker.java | 9 +- .../taskito/worker/WorkerDispatchBridge.java | 22 ++++- .../taskito/worker/RetryPredicateTest.java | 91 +++++++++++++++++++ .../taskito/test/InMemoryQueueBackend.java | 11 ++- 11 files changed, 225 insertions(+), 41 deletions(-) create mode 100644 sdks/java/src/test/java/org/byteveda/taskito/worker/RetryPredicateTest.java diff --git a/crates/taskito-java/src/dispatcher.rs b/crates/taskito-java/src/dispatcher.rs index 52a15546..cc69323d 100644 --- a/crates/taskito-java/src/dispatcher.rs +++ b/crates/taskito-java/src/dispatcher.rs @@ -23,10 +23,11 @@ use tokio::sync::oneshot; use crate::jvm; -/// The outcome Java reports for a submitted job. +/// The outcome Java reports for a submitted job. A failure carries whether it +/// may be retried — the task's `retryOn` predicate runs on the Java side. pub enum TaskOutcome { Success(Vec), - Failure(String), + Failure(String, bool), Cancelled, } @@ -109,7 +110,7 @@ async fn run_one( if let Err(err) = submit_to_java(callbacks, token, &job) { registry.forget(token); - return failure(job, err, started.elapsed().as_nanos() as i64, false); + return failure(job, err, started.elapsed().as_nanos() as i64, false, true); } // `timeout_ms <= 0` means no limit. On timeout we drop the pending entry; a @@ -120,7 +121,7 @@ async fn run_one( Err(_) => { registry.forget(token); let wall = started.elapsed().as_nanos() as i64; - return failure(job, "task timed out".to_string(), wall, true); + return failure(job, "task timed out".to_string(), wall, true, true); } } } else { @@ -136,16 +137,22 @@ async fn run_one( wall_time_ns: wall, }, Ok(TaskOutcome::Cancelled) => cancelled(job, wall), - Ok(TaskOutcome::Failure(error)) => { + Ok(TaskOutcome::Failure(error, retryable)) => { // A failure on a cancel-requested job is a cancellation, not a fault. if storage.is_cancel_requested(&job.id).unwrap_or(false) { cancelled(job, wall) } else { - failure(job, error, wall, false) + failure(job, error, wall, false, retryable) } } // The oneshot sender dropped without completing — the Java side died. - Err(_) => failure(job, "java task channel dropped".to_string(), wall, false), + Err(_) => failure( + job, + "java task channel dropped".to_string(), + wall, + false, + true, + ), } } @@ -193,7 +200,14 @@ fn cancelled(job: Job, wall_time_ns: i64) -> JobResult { } } -fn failure(job: Job, error: String, wall_time_ns: i64, timed_out: bool) -> JobResult { +/// `should_retry` false skips the retry budget entirely and dead-letters the job. +fn failure( + job: Job, + error: String, + wall_time_ns: i64, + timed_out: bool, + should_retry: bool, +) -> JobResult { JobResult::Failure { job_id: job.id, error, @@ -201,7 +215,7 @@ fn failure(job: Job, error: String, wall_time_ns: i64, timed_out: bool) -> JobRe max_retries: job.max_retries, task_name: job.task_name, wall_time_ns, - should_retry: true, + should_retry, timed_out, } } diff --git a/crates/taskito-java/src/worker.rs b/crates/taskito-java/src/worker.rs index 37a01d20..91d1f9ca 100644 --- a/crates/taskito-java/src/worker.rs +++ b/crates/taskito-java/src/worker.rs @@ -10,7 +10,7 @@ use std::sync::Arc; use std::time::Duration; use jni::objects::{GlobalRef, JByteArray, JClass, JObject, JString, JValue}; -use jni::sys::jlong; +use jni::sys::{jboolean, jlong, JNI_FALSE}; use jni::JNIEnv; use taskito_core::resilience::circuit_breaker::CircuitBreakerConfig; use taskito_core::resilience::rate_limiter::RateLimitConfig; @@ -600,7 +600,7 @@ pub extern "system" fn Java_org_byteveda_taskito_internal_NativeWorker_completeJ }) } -/// `void failJob(long workerHandle, long token, String error)`. +/// `void failJob(long workerHandle, long token, String error, boolean retryable)`. #[no_mangle] pub extern "system" fn Java_org_byteveda_taskito_internal_NativeWorker_failJob<'local>( mut env: JNIEnv<'local>, @@ -608,13 +608,15 @@ pub extern "system" fn Java_org_byteveda_taskito_internal_NativeWorker_failJob<' handle: jlong, token: jlong, error: JString<'local>, + retryable: jboolean, ) { guard(&mut env, (), |env| { let worker = unsafe { borrow_worker(handle) }; let message = read_string(env, &error)?; - worker - .registry - .complete(token as u64, TaskOutcome::Failure(message)); + worker.registry.complete( + token as u64, + TaskOutcome::Failure(message, retryable != JNI_FALSE), + ); Ok(()) }) } diff --git a/sdks/java/src/main/java/org/byteveda/taskito/internal/JniWorkerControl.java b/sdks/java/src/main/java/org/byteveda/taskito/internal/JniWorkerControl.java index 3dfd8103..1d6b7b94 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/internal/JniWorkerControl.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/internal/JniWorkerControl.java @@ -41,9 +41,9 @@ public void completeJob(long token, byte[] result) { } @Override - public void failJob(long token, String error) { + public void failJob(long token, String error, boolean retryable) { withOpenHandle(() -> { - NativeWorker.failJob(handle, token, error); + NativeWorker.failJob(handle, token, error, retryable); return null; }); } diff --git a/sdks/java/src/main/java/org/byteveda/taskito/internal/NativeWorker.java b/sdks/java/src/main/java/org/byteveda/taskito/internal/NativeWorker.java index 6586a9f0..7bda29fc 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/internal/NativeWorker.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/internal/NativeWorker.java @@ -16,7 +16,7 @@ private NativeWorker() {} public static native void completeJob(long handle, long token, byte[] result); - public static native void failJob(long handle, long token, String error); + public static native void failJob(long handle, long token, String error, boolean retryable); public static native void cancelJob(long handle, long token); diff --git a/sdks/java/src/main/java/org/byteveda/taskito/spi/WorkerControl.java b/sdks/java/src/main/java/org/byteveda/taskito/spi/WorkerControl.java index dc6b7f53..5d6f6292 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/spi/WorkerControl.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/spi/WorkerControl.java @@ -6,7 +6,8 @@ public interface WorkerControl extends AutoCloseable { void completeJob(long token, byte[] result); - void failJob(long token, String error); + /** Fail a job. {@code retryable} false dead-letters it whatever budget is left. */ + void failJob(long token, String error, boolean retryable); void cancelJob(long token); diff --git a/sdks/java/src/main/java/org/byteveda/taskito/task/Task.java b/sdks/java/src/main/java/org/byteveda/taskito/task/Task.java index b429ea87..ac0514a5 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/task/Task.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/task/Task.java @@ -6,6 +6,7 @@ import java.util.Arrays; import java.util.List; import java.util.Objects; +import java.util.function.Predicate; /** * Typed task descriptor: a name, its payload type, and default enqueue options. @@ -26,6 +27,7 @@ public final class Task { private final String retryBudget; private final Integer maxConcurrent; private final Integer maxInFlightPerTask; + private final Predicate retryOn; private Task( String name, @@ -38,7 +40,8 @@ private Task( String rateLimit, String retryBudget, Integer maxConcurrent, - Integer maxInFlightPerTask) { + Integer maxInFlightPerTask, + Predicate retryOn) { this.name = Objects.requireNonNull(name, "task name must not be null"); if (name.trim().isEmpty()) { throw new IllegalArgumentException("task name must not be blank"); @@ -53,12 +56,13 @@ private Task( this.retryBudget = retryBudget; this.maxConcurrent = maxConcurrent; this.maxInFlightPerTask = maxInFlightPerTask; + this.retryOn = retryOn; } /** A task whose payload deserializes to {@code payloadType}. */ public static Task of(String name, Class payloadType) { return new Task<>( - name, payloadType, EnqueueOptions.none(), null, List.of(), false, null, null, null, null, null); + name, payloadType, EnqueueOptions.none(), null, List.of(), false, null, null, null, null, null, null); } /** A task whose payload deserializes to a generic type, e.g. {@code new TypeReference>(){}}. */ @@ -74,6 +78,7 @@ public static Task of(String name, TypeReference payloadType) { null, null, null, + null, null); } @@ -90,7 +95,8 @@ public Task withOptions(EnqueueOptions options) { rateLimit, retryBudget, maxConcurrent, - maxInFlightPerTask); + maxInFlightPerTask, + retryOn); } /** @@ -110,7 +116,35 @@ public Task retryPolicy(RetryPolicy retryPolicy) { rateLimit, retryBudget, maxConcurrent, - maxInFlightPerTask); + maxInFlightPerTask, + retryOn); + } + + /** + * A copy of this task that retries a failure only when {@code retryOn} accepts + * the thrown exception. Returning {@code false} dead-letters the job at once, + * whatever retry budget is left — for permanent failures (a malformed payload, + * a 4xx) that no amount of retrying fixes. Unset retries every exception, and + * so does a predicate that itself throws. + * + *

Evaluated by the worker that ran the handler, so unlike the backoff curve + * it never reaches the scheduler. It sees exceptions the handler threw; a + * timeout is detected outside the handler and always consumes a retry. + */ + public Task retryOn(Predicate retryOn) { + return new Task<>( + name, + payloadType, + options, + retryPolicy, + codecs, + idempotent, + circuitBreaker, + rateLimit, + retryBudget, + maxConcurrent, + maxInFlightPerTask, + retryOn); } /** @@ -131,7 +165,8 @@ public Task codecs(String... codecs) { rateLimit, retryBudget, maxConcurrent, - maxInFlightPerTask); + maxInFlightPerTask, + retryOn); } /** @@ -151,7 +186,8 @@ public Task idempotent(boolean idempotent) { rateLimit, retryBudget, maxConcurrent, - maxInFlightPerTask); + maxInFlightPerTask, + retryOn); } /** @@ -171,7 +207,8 @@ public Task circuitBreaker(CircuitBreakerConfig circuitBreaker) { rateLimit, retryBudget, maxConcurrent, - maxInFlightPerTask); + maxInFlightPerTask, + retryOn); } /** @@ -191,7 +228,8 @@ public Task rateLimit(String rateLimit) { rateLimit, retryBudget, maxConcurrent, - maxInFlightPerTask); + maxInFlightPerTask, + retryOn); } /** @@ -214,7 +252,8 @@ public Task maxConcurrent(Integer maxConcurrent) { rateLimit, retryBudget, maxConcurrent, - maxInFlightPerTask); + maxInFlightPerTask, + retryOn); } /** @@ -236,7 +275,8 @@ public Task retryBudget(String retryBudget) { rateLimit, retryBudget, maxConcurrent, - maxInFlightPerTask); + maxInFlightPerTask, + retryOn); } /** @@ -259,7 +299,8 @@ public Task maxInFlightPerTask(Integer maxInFlightPerTask) { rateLimit, retryBudget, maxConcurrent, - maxInFlightPerTask); + maxInFlightPerTask, + retryOn); } /** A cap of zero is the annotation's "unset" sentinel, never a literal zero. */ @@ -352,4 +393,9 @@ public Integer maxConcurrent() { public Integer maxInFlightPerTask() { return maxInFlightPerTask; } + + /** Predicate deciding whether a thrown exception is retryable, or {@code null} to retry all. */ + public Predicate retryOn() { + return retryOn; + } } diff --git a/sdks/java/src/main/java/org/byteveda/taskito/worker/RegisteredTask.java b/sdks/java/src/main/java/org/byteveda/taskito/worker/RegisteredTask.java index 01566fc8..1df253ba 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/worker/RegisteredTask.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/worker/RegisteredTask.java @@ -2,17 +2,25 @@ import java.lang.reflect.Type; import java.util.List; +import java.util.function.Predicate; import org.byteveda.taskito.task.TaskFunction; -/** A worker-registered handler: payload type, the function to run, and any payload codecs. */ +/** + * A worker-registered handler: payload type, the function to run, any payload + * codecs, and the predicate classifying its failures as retryable ({@code null} + * retries every exception). + */ final class RegisteredTask { final Type payloadType; final TaskFunction handler; final List codecs; + final Predicate retryOn; - RegisteredTask(Type payloadType, TaskFunction handler, List codecs) { + RegisteredTask( + Type payloadType, TaskFunction handler, List codecs, Predicate retryOn) { this.payloadType = payloadType; this.handler = handler; this.codecs = codecs; + this.retryOn = retryOn; } } diff --git a/sdks/java/src/main/java/org/byteveda/taskito/worker/Worker.java b/sdks/java/src/main/java/org/byteveda/taskito/worker/Worker.java index e5d87cad..9a308728 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/worker/Worker.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/worker/Worker.java @@ -243,12 +243,14 @@ public static final class Builder { } public Builder handle(String taskName, Class payloadType, TaskFunction handler) { - handlers.put(taskName, new RegisteredTask(payloadType, cast(handler), List.of())); + handlers.put(taskName, new RegisteredTask(payloadType, cast(handler), List.of(), null)); return this; } public Builder handle(Task task, TaskFunction handler) { - handlers.put(task.name(), new RegisteredTask(task.payloadType(), cast(handler), task.codecNames())); + handlers.put( + task.name(), + new RegisteredTask(task.payloadType(), cast(handler), task.codecNames(), task.retryOn())); capturePolicy(task); return this; } @@ -266,7 +268,8 @@ public Builder register(Handler handler) { new RegisteredTask( handler.task().payloadType(), cast(handler.function()), - handler.task().codecNames())); + handler.task().codecNames(), + handler.task().retryOn())); capturePolicy(handler.task()); return this; } diff --git a/sdks/java/src/main/java/org/byteveda/taskito/worker/WorkerDispatchBridge.java b/sdks/java/src/main/java/org/byteveda/taskito/worker/WorkerDispatchBridge.java index 114def0e..a465bacd 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/worker/WorkerDispatchBridge.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/worker/WorkerDispatchBridge.java @@ -82,7 +82,8 @@ private void runJob(long token, String jobId, String taskName, byte[] payload) { WorkerControl bound = control.join(); RegisteredTask task = handlers.get(taskName); if (task == null) { - bound.failJob(token, "no handler registered for task '" + taskName + "'"); + // Retryable: another worker in the fleet may well have it registered. + bound.failJob(token, "no handler registered for task '" + taskName + "'", true); return; } JobInfo job = new JobInfo(jobId, taskName, () -> loadMetadata(jobId)); @@ -117,7 +118,7 @@ private void runJob(long token, String jobId, String taskName, byte[] payload) { m.onError(context, t); } // Canonical structured error (middleware above saw the live Throwable). - bound.failJob(token, TaskErrors.encode(t)); + bound.failJob(token, TaskErrors.encode(t), isRetryable(task, t)); } finally { if (scope != null) { Resources.exit(scope); // unbind the thread + dispose task-scoped resources (LIFO) @@ -125,6 +126,23 @@ private void runJob(long token, String jobId, String taskName, byte[] payload) { } } + /** + * Ask a task's {@code retryOn} predicate whether {@code error} is worth retrying. + * No predicate means retry, and so does one that throws — a broken classifier must + * not silently turn transient failures into dead letters. + */ + private static boolean isRetryable(RegisteredTask task, Throwable error) { + if (task.retryOn == null) { + return true; + } + try { + return task.retryOn.test(error); + } catch (RuntimeException e) { + LOG.warn("retryOn predicate threw; retrying the failure", e); + return true; + } + } + /** Reverse a task's payload codecs (last applied, first undone). */ private byte[] decodePayload(byte[] payload, List codecNames) { byte[] bytes = payload; diff --git a/sdks/java/src/test/java/org/byteveda/taskito/worker/RetryPredicateTest.java b/sdks/java/src/test/java/org/byteveda/taskito/worker/RetryPredicateTest.java new file mode 100644 index 00000000..1271a4bc --- /dev/null +++ b/sdks/java/src/test/java/org/byteveda/taskito/worker/RetryPredicateTest.java @@ -0,0 +1,91 @@ +package org.byteveda.taskito.worker; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.nio.file.Path; +import java.time.Duration; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import org.byteveda.taskito.Taskito; +import org.byteveda.taskito.events.EventName; +import org.byteveda.taskito.task.RetryPolicy; +import org.byteveda.taskito.task.Task; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; +import org.junit.jupiter.api.io.TempDir; + +/** + * A task's {@code retryOn} predicate classifies its failures: a rejected + * exception dead-letters at once, an accepted one still spends the retry budget. + */ +class RetryPredicateTest { + + private static final RetryPolicy FAST = RetryPolicy.delays(Duration.ofMillis(10), Duration.ofMillis(10)); + + @Test + @Timeout(30) + void rejectedExceptionDeadLettersWithoutSpendingTheBudget(@TempDir Path dir) throws Exception { + Task permanent = Task.of("permanent", String.class) + .maxRetries(3) + .retryPolicy(FAST) + .retryOn(error -> !(error instanceof IllegalArgumentException)); + + try (Taskito queue = Taskito.builder() + .backend("sqlite") + .url(dir.resolve("t.db").toString()) + .open()) { + queue.enqueue(permanent, "go"); + + AtomicInteger attempts = new AtomicInteger(); + AtomicInteger retries = new AtomicInteger(); + CountDownLatch dead = new CountDownLatch(1); + + try (Worker worker = queue.worker() + .handle(permanent, (String payload) -> { + attempts.incrementAndGet(); + throw new IllegalArgumentException("malformed input"); + }) + .on(EventName.RETRY, event -> retries.incrementAndGet()) + .on(EventName.DEAD, event -> dead.countDown()) + .start()) { + assertTrue(dead.await(25, TimeUnit.SECONDS), "task should dead-letter"); + + assertEquals(1, attempts.get(), "a rejected exception must not be retried"); + assertEquals(0, retries.get(), "no RETRY outcome should be emitted"); + } + } + } + + @Test + @Timeout(30) + void acceptedExceptionStillRetries(@TempDir Path dir) throws Exception { + Task transientFailure = Task.of("transient", String.class) + .maxRetries(2) + .retryPolicy(FAST) + .retryOn(error -> !(error instanceof IllegalArgumentException)); + + try (Taskito queue = Taskito.builder() + .backend("sqlite") + .url(dir.resolve("t.db").toString()) + .open()) { + queue.enqueue(transientFailure, "go"); + + AtomicInteger attempts = new AtomicInteger(); + CountDownLatch dead = new CountDownLatch(1); + + try (Worker worker = queue.worker() + .handle(transientFailure, (String payload) -> { + attempts.incrementAndGet(); + throw new IllegalStateException("connection reset"); + }) + .on(EventName.DEAD, event -> dead.countDown()) + .start()) { + assertTrue(dead.await(25, TimeUnit.SECONDS), "task should exhaust its budget"); + + assertEquals(3, attempts.get(), "first run plus both retries"); + } + } + } +} 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 2d5539ee..3e18f945 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 @@ -965,8 +965,9 @@ private synchronized void onComplete(JobRec job, byte[] result) { job.completedAt = now(); } - private synchronized void onFail(JobRec job, String error) { - if (job.retryCount < job.maxRetries) { + /** Mirrors the core: a non-retryable failure skips the budget and dead-letters. */ + private synchronized void onFail(JobRec job, String error, boolean retryable) { + if (retryable && job.retryCount < job.maxRetries) { job.retryCount++; job.status = "pending"; job.startedAt = null; @@ -1223,14 +1224,14 @@ public void completeJob(long token, byte[] result) { } @Override - public void failJob(long token, String error) { + public void failJob(long token, String error, boolean retryable) { String jobId = inFlight.remove(token); JobRec job = jobs.get(jobId); if (job == null) { return; } - boolean willRetry = job.retryCount < job.maxRetries; - onFail(job, error); + boolean willRetry = retryable && job.retryCount < job.maxRetries; + onFail(job, error, retryable); bridge.onOutcome(willRetry ? "retry" : "dead", jobId, job.taskName, error, job.retryCount, false); } From 78d4761e4329dc405a35c04f486061b5c6394e6e Mon Sep 17 00:00:00 2001 From: stromanni Date: Wed, 22 Jul 2026 21:09:38 +0530 Subject: [PATCH 3/4] docs: document the Node and Java retry predicate --- docs/content/docs/java/api-reference/task.mdx | 1 + docs/content/docs/node/api-reference/task.mdx | 1 + .../shared/guides/reliability/retries.mdx | 73 ++++++++++++++++--- 3 files changed, 66 insertions(+), 9 deletions(-) diff --git a/docs/content/docs/java/api-reference/task.mdx b/docs/content/docs/java/api-reference/task.mdx index f81627d8..6984c473 100644 --- a/docs/content/docs/java/api-reference/task.mdx +++ b/docs/content/docs/java/api-reference/task.mdx @@ -29,6 +29,7 @@ enqueue options. The fluent option methods each return a new descriptor. | `timeoutMs(long)` / `timeout(Duration)` | Per-attempt timeout. | | `delayMs(long)` / `delay(Duration)` | Schedule after a delay. | | `retryPolicy(RetryPolicy)` | Backoff curve — registered with the worker on `start()`. | +| `retryOn(Predicate)` | Classifies a thrown exception; `false` dead-letters it immediately. | | `codecs(String...)` | Named [payload codecs](/java/api-reference/serializers) applied to this task. | | `circuitBreaker(CircuitBreakerConfig)` | Trip the task after repeated failures; the worker registers the breaker on `start()`. | | `withOptions(EnqueueOptions)` | Replace the default options wholesale. | diff --git a/docs/content/docs/node/api-reference/task.mdx b/docs/content/docs/node/api-reference/task.mdx index 0c381726..5d31ce57 100644 --- a/docs/content/docs/node/api-reference/task.mdx +++ b/docs/content/docs/node/api-reference/task.mdx @@ -18,6 +18,7 @@ replaces the function and options. |---|---|---| | `maxRetries` | `number` | Attempts before dead-lettering. | | `retryBackoff` | `{ baseMs, maxMs }` | Exponential backoff bounds. | +| `retryOn` | `(error) => boolean` | Classifies a thrown error; `false` dead-letters it immediately. | | `timeoutMs` | `number` | Per-attempt timeout. | | `maxConcurrent` | `number` | Max simultaneous running jobs of this task. | | `rateLimit` | `` `"100/m"` `` | Rate limit as `count/unit`, unit `s` \| `m` \| `h`. | diff --git a/docs/content/docs/shared/guides/reliability/retries.mdx b/docs/content/docs/shared/guides/reliability/retries.mdx index 3cc46413..a9ba65f1 100644 --- a/docs/content/docs/shared/guides/reliability/retries.mdx +++ b/docs/content/docs/shared/guides/reliability/retries.mdx @@ -202,23 +202,62 @@ In practice, use one or the other. -There's no exception-type filter on the Node binding today — every error -triggers a retry until the budget is exhausted. Distinguish terminal errors -inside the task and either swallow them (return normally) or throw so the -job dead-letters immediately by setting `maxRetries: 0` for that call: +`retryOn` classifies a thrown error: return `false` and the job dead-letters +immediately, whatever budget is left. One predicate covers both directions — +test for the errors worth retrying, or negate a test for the permanent ones: ```ts -queue.enqueue("parseData", [raw], { maxRetries: 0 }); // no retry for this job +// Whitelist: retry only these; everything else skips straight to DLQ. +queue.task("fetchData", fetchData, { + maxRetries: 5, + retryOn: (error) => error instanceof ConnectionError || error instanceof TimeoutError, +}); + +// Blacklist: retry everything except a permanent parse failure. +queue.task("parseData", parseData, { + maxRetries: 5, + retryOn: (error) => !(error instanceof SyntaxError), +}); ``` +| Option | Description | +|---|---| +| `retryOn` | `(error: unknown) => boolean`. `true` retries as usual; `false` dead-letters at once. Unset retries every error. | + +The predicate is synchronous — the job cannot settle until it answers — and +runs on the worker that caught the error, so it sees the live `Error` object, +not just its type name. One that throws is treated as `true`: a broken +classifier must not silently turn transient failures into dead letters. + -There's no exception-type filter on the Java binding today — every thrown -exception triggers a retry until the budget is exhausted. Distinguish -terminal errors inside the handler, or set `maxRetries(0)` for calls that -should dead-letter immediately on any failure. +`retryOn` classifies a thrown exception: return `false` and the job +dead-letters immediately, whatever budget is left. One predicate covers both +directions — test for the exceptions worth retrying, or negate a test for the +permanent ones: + +```java +// Whitelist: retry only these; everything else skips straight to DLQ. +Task FETCH = Task.of("fetch", String.class) + .maxRetries(5) + .retryOn(error -> error instanceof IOException); + +// Blacklist: retry everything except a permanent parse failure. +Task PARSE = Task.of("parse", String.class) + .maxRetries(5) + .retryOn(error -> !(error instanceof IllegalArgumentException)); +``` + +| Method | Description | +|---|---| +| `retryOn(Predicate)` | `true` retries as usual; `false` dead-letters at once. Unset retries every exception. | + +Unlike the backoff curve, the predicate never reaches the scheduler — it runs +on the worker that caught the exception, so it sees the live `Throwable`. One +that throws is treated as `true`: a broken classifier must not silently turn +transient failures into dead letters. @@ -279,6 +318,22 @@ straight through to the retry-budget check. + + +The "Exception passes filter?" step only applies filtering when `retryOn` is +set on the task — otherwise every error passes straight through to the +retry-budget check. + + + + + +The "Exception passes filter?" step only applies filtering when `retryOn` is +set on the task — otherwise every exception passes straight through to the +retry-budget check. + + + ## Timeouts count as retries A task that exceeds its timeout is treated as a failure by the same From b9a2442bea9a058e8fe3e718a56877655309a88e Mon Sep 17 00:00:00 2001 From: stromanni Date: Wed, 22 Jul 2026 21:23:20 +0530 Subject: [PATCH 4/4] docs: widen retryOn's documented scope past the handler MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The predicate also sees middleware, decode and serialization failures — a whitelist dead-letters those too. --- .../content/docs/shared/guides/reliability/retries.mdx | 10 ++++++++++ .../src/main/java/org/byteveda/taskito/task/Task.java | 7 +++++-- sdks/node/src/types.ts | 9 ++++++--- 3 files changed, 21 insertions(+), 5 deletions(-) diff --git a/docs/content/docs/shared/guides/reliability/retries.mdx b/docs/content/docs/shared/guides/reliability/retries.mdx index a9ba65f1..f9c51f68 100644 --- a/docs/content/docs/shared/guides/reliability/retries.mdx +++ b/docs/content/docs/shared/guides/reliability/retries.mdx @@ -229,6 +229,11 @@ runs on the worker that caught the error, so it sees the live `Error` object, not just its type name. One that throws is treated as `true`: a broken classifier must not silently turn transient failures into dead letters. +It sees every error raised while running the task, not only the handler's: a +`before`/`after` middleware hook or result serialization can fail too, and a +whitelist like the first example above dead-letters those as well. Payload +decoding fails earlier and always retries, as does a timeout. + @@ -259,6 +264,11 @@ on the worker that caught the exception, so it sees the live `Throwable`. One that throws is treated as `true`: a broken classifier must not silently turn transient failures into dead letters. +It sees every exception raised while running the task, not only the handler's: +`before`/`after` middleware, payload decoding and result serialization can fail +too, and a whitelist like the first example above dead-letters those as well. A +timeout is detected outside the handler and always retries. + ## Per-job overrides diff --git a/sdks/java/src/main/java/org/byteveda/taskito/task/Task.java b/sdks/java/src/main/java/org/byteveda/taskito/task/Task.java index ac0514a5..74eaca90 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/task/Task.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/task/Task.java @@ -128,8 +128,11 @@ public Task retryPolicy(RetryPolicy retryPolicy) { * so does a predicate that itself throws. * *

Evaluated by the worker that ran the handler, so unlike the backoff curve - * it never reaches the scheduler. It sees exceptions the handler threw; a - * timeout is detected outside the handler and always consumes a retry. + * it never reaches the scheduler. It sees every exception raised while running + * the task, not only the handler's: {@code before}/{@code after} middleware, + * payload decoding and result serialization can fail too, so a whitelist + * predicate dead-letters those as well. A timeout is detected outside the + * handler and always consumes a retry. */ public Task retryOn(Predicate retryOn) { return new Task<>( diff --git a/sdks/node/src/types.ts b/sdks/node/src/types.ts index d99a50b8..8ffee168 100644 --- a/sdks/node/src/types.ts +++ b/sdks/node/src/types.ts @@ -132,9 +132,12 @@ export interface TaskOptions { * failures (a malformed payload, a 4xx) that no amount of retrying fixes. * Unset retries everything, as does a predicate that throws. * - * Synchronous by design: the job cannot settle until it answers. Applies to - * errors the handler throws — a timeout is detected outside the handler and - * always consumes a retry. + * Synchronous by design: the job cannot settle until it answers. It sees + * every error raised while running the task, not only the handler's: a + * `before`/`after` middleware hook or result serialization can fail too, so a + * whitelist predicate dead-letters those as well. Payload decoding fails + * earlier and always retries, as does a timeout (detected outside the + * handler). */ retryOn?: (error: unknown) => boolean; /** Per-job timeout default (ms); enforced by the worker. */