feat: add a retry predicate to the Node and Java SDKs

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<u8>),
-    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,15 +200,22 @@ 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,
         retry_count: job.retry_count,
         max_retries: job.max_retries,
         task_name: job.task_name,
         wall_time_ns,
-        should_retry: true,
+        should_retry,
         timed_out,
     }
 }

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,21 +600,23 @@ 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>,
     _class: JClass<'local>,
     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(())
     })
 }

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<Buffer>,
+    /// The failure, as the canonical cross-SDK `{errtype,message,traceback}`
+    /// JSON. Its presence is what marks the outcome a failure.
+    pub error: Option<String>,
+    /// 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<bool>,
+}
+
 /// JS-facing view of a stored [`Job`]. `result` is the opaque result blob (or
 /// `null`); the shell deserializes it.
 #[napi(object)]

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};

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<Buffer>`, and reports a [`JobResult`] back to the scheduler. This is
+//! `Promise<JsTaskOutcome>`, 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<Buffer>`.
+/// Task callback registered from JS: `(invocation) => Promise<JsTaskOutcome>`.
 type TaskCallback = ThreadsafeFunction<JsTaskInvocation, ErrorStrategy::Fatal>;
 
 /// 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::<napi::Result<Vec<u8>>>();
+    // 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::<napi::Result<TaskOutcome>>();
     callback.call_with_return_value(
         invocation,
         ThreadsafeFunctionCallMode::NonBlocking,
-        move |promise: Promise<Buffer>| {
+        move |promise: Promise<JsTaskOutcome>| {
             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,57 +134,88 @@ 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<Vec<u8>>,
+    error: Option<String>,
+    /// 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<JsTaskOutcome> 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,
         retry_count: job.retry_count,
         max_retries: job.max_retries,
         task_name: job.task_name,
         wall_time_ns,
-        should_retry: true,
+        should_retry,
         timed_out,
     }
 }

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<Throwable>)` | 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. |

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`. |