feat(java): per-task retry backoff via core RetryPolicy

crates/taskito-java/src/convert.rs

@@ -162,6 +162,21 @@ pub struct WorkerOptions {
     pub queues: Option<Vec<String>>,
     pub channel_capacity: Option<u32>,
     pub batch_size: Option<u32>,
+    /// Per-task retry-backoff policies, registered with the scheduler at start.
+    /// The core owns the retry engine; this only feeds it the backoff curve.
+    pub task_configs: Option<Vec<TaskRetryConfig>>,
+}
+
+/// A task's retry-backoff curve. Fields left unset fall back to the core's
+/// `RetryPolicy` defaults; the per-job retry budget travels via `maxRetries` on
+/// enqueue, so it is deliberately absent here.
+#[derive(Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct TaskRetryConfig {
+    pub name: String,
+    pub base_delay_ms: Option<i64>,
+    pub max_delay_ms: Option<i64>,
+    pub custom_delays_ms: Option<Vec<i64>>,
 }
 
 /// Filter accepted by `NativeQueue.listJobs`.

crates/taskito-java/src/worker.rs

@@ -12,13 +12,14 @@ use std::time::Duration;
 use jni::objects::{GlobalRef, JByteArray, JClass, JObject, JString, JValue};
 use jni::sys::jlong;
 use jni::JNIEnv;
-use taskito_core::scheduler::ResultOutcome;
+use taskito_core::resilience::retry::RetryPolicy;
+use taskito_core::scheduler::{ResultOutcome, TaskConfig};
 use taskito_core::worker::WorkerDispatcher;
 use taskito_core::{Scheduler, SchedulerConfig, Storage, StorageBackend};
 use tokio::sync::Notify;
 
 use crate::backend::QueueHandle;
-use crate::convert::{parse_json, WorkerOptions};
+use crate::convert::{parse_json, TaskRetryConfig, WorkerOptions};
 use crate::dispatcher::{JavaDispatcher, Registry, TaskOutcome};
 use crate::ffi::{guard, read_bytes, read_string};
 use crate::handle::{self, drop_handle, into_handle};
@@ -70,7 +71,9 @@ fn start_worker(
     let queues_csv = queues.join(",");
     let worker_id = format!("java-{}", uuid::Uuid::now_v7());
 
-    let scheduler = Arc::new(Scheduler::new(storage, queues, config, namespace));
+    let mut scheduler = Scheduler::new(storage, queues, config, namespace);
+    register_task_policies(&mut scheduler, options.task_configs);
+    let scheduler = Arc::new(scheduler);
     let shutdown = scheduler.shutdown_handle();
     let heartbeat_stop = Arc::new(Notify::new());
 
@@ -118,6 +121,41 @@ fn start_worker(
     })
 }
 
+/// Register each task's retry-backoff curve with the scheduler. Only the curve
+/// is set here — the core resolves the retry budget from the job's `max_retries`
+/// — so unset fields keep the core [`RetryPolicy`] defaults.
+fn register_task_policies(scheduler: &mut Scheduler, configs: Option<Vec<TaskRetryConfig>>) {
+    let Some(configs) = configs else { return };
+    for config in configs {
+        let mut retry_policy = RetryPolicy::default();
+        if let Some(custom) = config.custom_delays_ms {
+            // Explicit per-attempt delays are honored exactly. The core derives
+            // its jitter and post-exhaustion fallback from base_delay_ms, so
+            // zeroing it leaves the listed delays jitter-free; once the list is
+            // spent, further retries fire immediately (callers list enough).
+            retry_policy.base_delay_ms = 0;
+            retry_policy.max_delay_ms = 0;
+            retry_policy.custom_delays_ms = Some(custom);
+        } else {
+            if let Some(base) = config.base_delay_ms {
+                retry_policy.base_delay_ms = base;
+            }
+            if let Some(max) = config.max_delay_ms {
+                retry_policy.max_delay_ms = max;
+            }
+        }
+        scheduler.register_task(
+            config.name,
+            TaskConfig {
+                retry_policy,
+                rate_limit: None,
+                circuit_breaker: None,
+                max_concurrent: None,
+            },
+        );
+    }
+}
+
 /// Drain results: apply each to storage and invoke `WorkerBridge.onOutcome`.
 fn drain_results(
     result_rx: crossbeam_channel::Receiver<taskito_core::scheduler::JobResult>,

sdks/java/src/main/java/org/byteveda/taskito/task/RetryPolicy.java

@@ -0,0 +1,78 @@
+package org.byteveda.taskito.task;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+
+/**
+ * A task's retry-backoff curve: how long to wait between attempts.
+ *
+ * <p>The retry <em>budget</em> (how many attempts) is set per enqueue via
+ * {@link Task#maxRetries}. The core scheduler owns retry execution, so retries
+ * stay durable, survive worker crashes, and behave identically to the other
+ * Taskito SDKs — this type only supplies the timing the scheduler applies.
+ *
+ * <p>Instances are immutable.
+ */
+public final class RetryPolicy {
+    private final Duration baseDelay;
+    private final Duration maxDelay;
+    private final List<Duration> customDelays;
+
+    private RetryPolicy(Duration baseDelay, Duration maxDelay, List<Duration> customDelays) {
+        this.baseDelay = baseDelay;
+        this.maxDelay = maxDelay;
+        this.customDelays = customDelays;
+    }
+
+    /**
+     * Exponential backoff: retry N waits about {@code base · 2^N}, capped at
+     * {@code max}, plus a random jitter of up to {@code base} (spreads retries so
+     * a batch of failures doesn't retry in lockstep).
+     */
+    public static RetryPolicy exponential(Duration base, Duration max) {
+        return new RetryPolicy(nonNegative(base, "base"), nonNegative(max, "max"), Collections.emptyList());
+    }
+
+    /**
+     * Explicit per-attempt delays, applied exactly (no jitter): retry N waits
+     * {@code delays[N]}. The list is authoritative for the retries it covers, so
+     * supply at least as many delays as the task's {@code maxRetries} — once the
+     * list is exhausted any further retries fire immediately.
+     */
+    public static RetryPolicy delays(Duration... delays) {
+        if (delays.length == 0) {
+            throw new IllegalArgumentException("at least one delay is required");
+        }
+        List<Duration> copy = new ArrayList<>(delays.length);
+        for (Duration delay : delays) {
+            copy.add(nonNegative(delay, "delay"));
+        }
+        return new RetryPolicy(null, null, Collections.unmodifiableList(copy));
+    }
+
+    /** Exponential base delay, or {@code null} when using {@link #delays}. */
+    public Duration baseDelay() {
+        return baseDelay;
+    }
+
+    /** Backoff cap, or {@code null} when using {@link #delays}. */
+    public Duration maxDelay() {
+        return maxDelay;
+    }
+
+    /** Explicit per-attempt delays, or an empty list when using exponential backoff. */
+    public List<Duration> customDelays() {
+        return customDelays;
+    }
+
+    private static Duration nonNegative(Duration value, String what) {
+        Objects.requireNonNull(value, what + " must not be null");
+        if (value.isNegative()) {
+            throw new IllegalArgumentException(what + " must not be negative");
+        }
+        return value;
+    }
+}

sdks/java/src/main/java/org/byteveda/taskito/task/Task.java

@@ -16,29 +16,40 @@ public final class Task<T> {
     private final String name;
     private final Type payloadType;
     private final EnqueueOptions options;
+    private final RetryPolicy retryPolicy;
 
-    private Task(String name, Type payloadType, EnqueueOptions options) {
+    private Task(String name, Type payloadType, EnqueueOptions options, RetryPolicy retryPolicy) {
         this.name = Objects.requireNonNull(name, "task name must not be null");
         if (name.trim().isEmpty()) {
             throw new IllegalArgumentException("task name must not be blank");
         }
         this.payloadType = Objects.requireNonNull(payloadType, "payloadType must not be null");
         this.options = Objects.requireNonNull(options, "options must not be null");
+        this.retryPolicy = retryPolicy;
     }
 
     /** A task whose payload deserializes to {@code payloadType}. */
     public static <T> Task<T> of(String name, Class<T> payloadType) {
-        return new Task<>(name, payloadType, EnqueueOptions.none());
+        return new Task<>(name, payloadType, EnqueueOptions.none(), null);
     }
 
     /** A task whose payload deserializes to a generic type, e.g. {@code new TypeReference<List<Foo>>(){}}. */
     public static <T> Task<T> of(String name, TypeReference<T> payloadType) {
-        return new Task<>(name, payloadType.getType(), EnqueueOptions.none());
+        return new Task<>(name, payloadType.getType(), EnqueueOptions.none(), null);
     }
 
     /** A copy of this task with the given default options. */
     public Task<T> withOptions(EnqueueOptions options) {
-        return new Task<>(name, payloadType, options);
+        return new Task<>(name, payloadType, options, retryPolicy);
+    }
+
+    /**
+     * A copy of this task whose retries use {@code retryPolicy}'s backoff curve.
+     * Registered with the worker on {@code start()}; the retry budget still comes
+     * from {@link #maxRetries}.
+     */
+    public Task<T> retryPolicy(RetryPolicy retryPolicy) {
+        return new Task<>(name, payloadType, options, retryPolicy);
     }
 
     public Task<T> queue(String queue) {
@@ -86,4 +97,9 @@ public Type payloadType() {
     public EnqueueOptions options() {
         return options;
     }
+
+    /** The retry-backoff curve for this task, or {@code null} for the core defaults. */
+    public RetryPolicy retryPolicy() {
+        return retryPolicy;
+    }
 }

sdks/java/src/main/java/org/byteveda/taskito/worker/Worker.java

@@ -21,6 +21,7 @@
 import org.byteveda.taskito.serialization.Serializer;
 import org.byteveda.taskito.spi.QueueBackend;
 import org.byteveda.taskito.spi.WorkerControl;
+import org.byteveda.taskito.task.RetryPolicy;
 import org.byteveda.taskito.task.Task;
 import org.byteveda.taskito.task.TaskFunction;
 import org.byteveda.taskito.workflows.WorkflowTracker;
@@ -92,6 +93,7 @@ public static final class Builder {
         private final Serializer serializer;
         private final List<Middleware> middleware;
         private final Map<String, RegisteredTask> handlers = new HashMap<>();
+        private final Map<String, RetryPolicy> taskPolicies = new HashMap<>();
         private final Map<EventName, List<Consumer<OutcomeEvent>>> listeners = new EnumMap<>(EventName.class);
         private List<String> queues;
         private int concurrency;
@@ -111,6 +113,7 @@ public <T, R> Builder handle(String taskName, Class<T> payloadType, TaskFunction
 
         public <T, R> Builder handle(Task<T> task, TaskFunction<T, R> handler) {
             handlers.put(task.name(), new RegisteredTask(task.payloadType(), cast(handler)));
+            capturePolicy(task);
             return this;
         }
 
@@ -124,9 +127,18 @@ public Builder apply(Consumer<Builder> customizer) {
         public Builder register(Handler<?, ?> handler) {
             handlers.put(
                     handler.task().name(), new RegisteredTask(handler.task().payloadType(), cast(handler.function())));
+            capturePolicy(handler.task());
             return this;
         }
 
+        /** Remember a task's retry-backoff curve so {@code start()} registers it. */
+        private void capturePolicy(Task<?> task) {
+            RetryPolicy policy = task.retryPolicy();
+            if (policy != null) {
+                taskPolicies.put(task.name(), policy);
+            }
+        }
+
         /** Register every handler in a {@link HandlerRegistry} (e.g. a generated {@code XxxTasks.handlers}). */
         public Builder register(HandlerRegistry registry) {
             registry.handlers().forEach(this::register);
@@ -190,13 +202,38 @@ private String encodeOptions() {
             if (batchSize != null) {
                 options.put("batchSize", batchSize);
             }
+            if (!taskPolicies.isEmpty()) {
+                options.put("taskConfigs", encodeTaskConfigs());
+            }
             try {
                 return JSON.writeValueAsString(options);
             } catch (Exception e) {
                 throw new TaskitoException("failed to encode worker options", e);
             }
         }
 
+        /** Serialize each captured retry policy into the wire shape the binding reads. */
+        private List<Map<String, Object>> encodeTaskConfigs() {
+            List<Map<String, Object>> configs = new ArrayList<>(taskPolicies.size());
+            taskPolicies.forEach((name, policy) -> {
+                Map<String, Object> config = new LinkedHashMap<>();
+                config.put("name", name);
+                if (policy.baseDelay() != null) {
+                    config.put("baseDelayMs", policy.baseDelay().toMillis());
+                }
+                if (policy.maxDelay() != null) {
+                    config.put("maxDelayMs", policy.maxDelay().toMillis());
+                }
+                if (!policy.customDelays().isEmpty()) {
+                    List<Long> delaysMs = new ArrayList<>(policy.customDelays().size());
+                    policy.customDelays().forEach(delay -> delaysMs.add(delay.toMillis()));
+                    config.put("customDelaysMs", delaysMs);
+                }
+                configs.add(config);
+            });
+            return configs;
+        }
+
         @SuppressWarnings("unchecked")
         private static <T, R> TaskFunction<Object, Object> cast(TaskFunction<T, R> handler) {
             return (TaskFunction<Object, Object>) handler;

sdks/java/src/test/java/org/byteveda/taskito/RetryPolicyTest.java

@@ -0,0 +1,66 @@
+package org.byteveda.taskito;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Path;
+import java.time.Duration;
+import java.util.Optional;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.byteveda.taskito.events.EventName;
+import org.byteveda.taskito.task.RetryPolicy;
+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;
+
+/**
+ * A handler that fails twice then succeeds must be retried by the core scheduler
+ * — proving the per-task {@link RetryPolicy} is wired through to the native retry
+ * engine (RETRY outcomes fire; no Java-side re-enqueue emulation).
+ */
+class RetryPolicyTest {
+
+    @Test
+    @Timeout(30)
+    void failingTaskIsRetriedUntilItSucceeds(@TempDir Path dir) throws Exception {
+        Task<String> flaky = Task.of("flaky", String.class)
+                .maxRetries(3)
+                .retryPolicy(RetryPolicy.delays(Duration.ofMillis(10), Duration.ofMillis(10)));
+
+        try (Queue queue = Taskito.builder()
+                .backend("sqlite")
+                .url(dir.resolve("t.db").toString())
+                .open()) {
+            String id = queue.enqueue(flaky, "go");
+
+            AtomicInteger attempts = new AtomicInteger();
+            AtomicInteger retries = new AtomicInteger();
+            CountDownLatch done = new CountDownLatch(1);
+
+            try (Worker worker = queue.worker()
+                    .handle(flaky, (String payload) -> {
+                        if (attempts.incrementAndGet() < 3) {
+                            throw new IllegalStateException("transient failure");
+                        }
+                        return 42;
+                    })
+                    .on(EventName.RETRY, event -> retries.incrementAndGet())
+                    .on(EventName.SUCCESS, event -> done.countDown())
+                    .start()) {
+                assertTrue(done.await(25, TimeUnit.SECONDS), "task should eventually succeed");
+
+                assertEquals(3, attempts.get(), "should run three times (two failures, one success)");
+                assertEquals(2, retries.get(), "core should emit a RETRY outcome per failure");
+
+                Optional<byte[]> result = queue.getResult(id);
+                assertTrue(result.isPresent());
+                assertEquals("42", new String(result.get(), StandardCharsets.UTF_8));
+            }
+        }
+    }
+}