diff --git a/crates/taskito-java/src/convert.rs b/crates/taskito-java/src/convert.rs index 7bb3e3ec..f7c3b08b 100644 --- a/crates/taskito-java/src/convert.rs +++ b/crates/taskito-java/src/convert.rs @@ -162,6 +162,21 @@ pub struct WorkerOptions { pub queues: Option>, pub channel_capacity: Option, pub batch_size: Option, + /// 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>, +} + +/// 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, + pub max_delay_ms: Option, + pub custom_delays_ms: Option>, } /// Filter accepted by `NativeQueue.listJobs`. diff --git a/crates/taskito-java/src/worker.rs b/crates/taskito-java/src/worker.rs index 0baa5bd3..11548b11 100644 --- a/crates/taskito-java/src/worker.rs +++ b/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>) { + 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, diff --git a/sdks/java/src/main/java/org/byteveda/taskito/task/RetryPolicy.java b/sdks/java/src/main/java/org/byteveda/taskito/task/RetryPolicy.java new file mode 100644 index 00000000..b2d08e59 --- /dev/null +++ b/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. + * + *

The retry budget (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. + * + *

Instances are immutable. + */ +public final class RetryPolicy { + private final Duration baseDelay; + private final Duration maxDelay; + private final List customDelays; + + private RetryPolicy(Duration baseDelay, Duration maxDelay, List 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 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 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; + } +} 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 8fe64cb1..7293c41c 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 @@ -16,29 +16,40 @@ public final class Task { 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 Task of(String name, Class 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>(){}}. */ public static Task of(String name, TypeReference 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 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 retryPolicy(RetryPolicy retryPolicy) { + return new Task<>(name, payloadType, options, retryPolicy); } public Task 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; + } } 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 8524d38d..4dae12ab 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 @@ -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; private final Map handlers = new HashMap<>(); + private final Map taskPolicies = new HashMap<>(); private final Map>> listeners = new EnumMap<>(EventName.class); private List queues; private int concurrency; @@ -111,6 +113,7 @@ public Builder handle(String taskName, Class payloadType, TaskFunction public Builder handle(Task task, TaskFunction handler) { handlers.put(task.name(), new RegisteredTask(task.payloadType(), cast(handler))); + capturePolicy(task); return this; } @@ -124,9 +127,18 @@ public Builder apply(Consumer 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,6 +202,9 @@ 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) { @@ -197,6 +212,28 @@ private String encodeOptions() { } } + /** Serialize each captured retry policy into the wire shape the binding reads. */ + private List> encodeTaskConfigs() { + List> configs = new ArrayList<>(taskPolicies.size()); + taskPolicies.forEach((name, policy) -> { + Map 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 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 TaskFunction cast(TaskFunction handler) { return (TaskFunction) handler; diff --git a/sdks/java/src/test/java/org/byteveda/taskito/RetryPolicyTest.java b/sdks/java/src/test/java/org/byteveda/taskito/RetryPolicyTest.java new file mode 100644 index 00000000..f8713653 --- /dev/null +++ b/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 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 result = queue.getResult(id); + assertTrue(result.isPresent()); + assertEquals("42", new String(result.get(), StandardCharsets.UTF_8)); + } + } + } +}