feat(java): idempotent, dependsOn, notes, circuit breakers

crates/taskito-java/src/convert.rs

@@ -5,8 +5,10 @@
 
 use serde::{Deserialize, Serialize};
 use taskito_core::job::{now_millis, Job, NewJob};
+use taskito_core::resilience::circuit_breaker::CircuitState;
 use taskito_core::storage::models::{
-    JobErrorRow, LockInfoRow, PeriodicTaskRow, TaskLogRow, TaskMetricRow, WorkerRow,
+    CircuitBreakerRow, JobErrorRow, LockInfoRow, PeriodicTaskRow, TaskLogRow, TaskMetricRow,
+    WorkerRow,
 };
 use taskito_core::storage::{DeadJob, QueueStats};
 
@@ -43,6 +45,9 @@ pub struct EnqueueOptions {
     pub unique_key: Option<String>,
     pub metadata: Option<String>,
     pub namespace: Option<String>,
+    pub depends_on: Option<Vec<String>>,
+    /// Pre-encoded canonical JSON produced by the SDK; stored verbatim.
+    pub notes: Option<String>,
 }
 
 /// Parse a JSON argument, attributing any failure to the named field.
@@ -70,8 +75,8 @@ pub fn build_new_job(
         timeout_ms: options.timeout_ms.unwrap_or(0),
         unique_key: options.unique_key,
         metadata: options.metadata,
-        notes: None,
-        depends_on: Vec::new(),
+        notes: options.notes,
+        depends_on: options.depends_on.unwrap_or_default(),
         expires_at: None,
         result_ttl_ms: None,
         namespace: options
@@ -129,6 +134,8 @@ pub struct JobView<'a> {
     pub namespace: Option<&'a str>,
     /// Opaque metadata blob (JSON the SDK sets, e.g. middleware-injected trace ids).
     pub metadata: Option<&'a str>,
+    /// Structured notes as canonical JSON; the SDK parses it back into a map.
+    pub notes: Option<&'a str>,
 }
 
 impl<'a> From<&'a Job> for JobView<'a> {
@@ -151,6 +158,41 @@ impl<'a> From<&'a Job> for JobView<'a> {
             unique_key: j.unique_key.as_deref(),
             namespace: j.namespace.as_deref(),
             metadata: j.metadata.as_deref(),
+            notes: j.notes.as_deref(),
+        }
+    }
+}
+
+/// Java-facing view of a task's circuit-breaker state. `state` is the lowercase wire
+/// string (`closed`/`open`/`half_open`); timestamps are Unix milliseconds.
+#[derive(Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct CircuitBreakerView<'a> {
+    pub task_name: &'a str,
+    pub state: &'static str,
+    pub failure_count: i32,
+    pub threshold: i32,
+    pub window_ms: i64,
+    pub cooldown_ms: i64,
+    pub opened_at: Option<i64>,
+    pub last_failure_at: Option<i64>,
+    pub half_open_max_probes: i32,
+    pub half_open_success_rate: f64,
+}
+
+impl<'a> From<&'a CircuitBreakerRow> for CircuitBreakerView<'a> {
+    fn from(r: &'a CircuitBreakerRow) -> Self {
+        Self {
+            task_name: &r.task_name,
+            state: CircuitState::from_i32(r.state).as_str(),
+            failure_count: r.failure_count,
+            threshold: r.threshold,
+            window_ms: r.window_ms,
+            cooldown_ms: r.cooldown_ms,
+            opened_at: r.opened_at,
+            last_failure_at: r.last_failure_at,
+            half_open_max_probes: r.half_open_max_probes,
+            half_open_success_rate: r.half_open_success_rate,
         }
     }
 }
@@ -182,6 +224,13 @@ pub struct TaskRetryConfig {
     pub base_delay_ms: Option<i64>,
     pub max_delay_ms: Option<i64>,
     pub custom_delays_ms: Option<Vec<i64>>,
+    /// Circuit-breaker knobs. `threshold` present ⇒ the breaker is enabled; the SDK
+    /// already converts window/cooldown to milliseconds, so these are stored as-is.
+    pub circuit_breaker_threshold: Option<i32>,
+    pub circuit_breaker_window_ms: Option<i64>,
+    pub circuit_breaker_cooldown_ms: Option<i64>,
+    pub circuit_breaker_half_open_probes: Option<i32>,
+    pub circuit_breaker_half_open_success_rate: Option<f64>,
 }
 
 /// Filter accepted by `NativeQueue.listJobs`.

crates/taskito-java/src/queue/inspect.rs

@@ -9,7 +9,8 @@ use taskito_core::Storage;
 
 use super::borrow_queue;
 use crate::convert::{
-    status_code, to_json, JobErrorView, JobFilter, JobView, MetricView, StatsView, WorkerView,
+    status_code, to_json, CircuitBreakerView, JobErrorView, JobFilter, JobView, MetricView,
+    StatsView, WorkerView,
 };
 use crate::error::BindingError;
 use crate::ffi::{guard, new_string, read_optional_string, read_string};
@@ -146,3 +147,21 @@ pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_listWorker
         new_string(env, to_json(&views)?)
     })
 }
+
+/// `String listCircuitBreakers(long handle)` — every configured task's breaker state.
+#[no_mangle]
+pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_listCircuitBreakers<
+    'local,
+>(
+    mut env: JNIEnv<'local>,
+    _class: JClass<'local>,
+    handle: jlong,
+) -> jstring {
+    guard(&mut env, std::ptr::null_mut(), |env| {
+        let queue = unsafe { borrow_queue(handle) };
+        let breakers = queue.storage.list_circuit_breakers()?;
+        let views: Vec<CircuitBreakerView> =
+            breakers.iter().map(CircuitBreakerView::from).collect();
+        new_string(env, to_json(&views)?)
+    })
+}

crates/taskito-java/src/worker.rs

@@ -12,6 +12,7 @@ use std::time::Duration;
 use jni::objects::{GlobalRef, JByteArray, JClass, JObject, JString, JValue};
 use jni::sys::jlong;
 use jni::JNIEnv;
+use taskito_core::resilience::circuit_breaker::CircuitBreakerConfig;
 use taskito_core::resilience::retry::RetryPolicy;
 use taskito_core::scheduler::{ResultOutcome, TaskConfig};
 use taskito_core::worker::WorkerDispatcher;
@@ -181,12 +182,26 @@ fn register_task_policies(scheduler: &mut Scheduler, configs: Option<Vec<TaskRet
                 retry_policy.max_delay_ms = max;
             }
         }
+        // A present threshold enables the breaker; window/cooldown already arrive in ms
+        // from the SDK, so they pass through unscaled (defaults mirror the core's).
+        let circuit_breaker =
+            config
+                .circuit_breaker_threshold
+                .map(|threshold| CircuitBreakerConfig {
+                    threshold,
+                    window_ms: config.circuit_breaker_window_ms.unwrap_or(60_000),
+                    cooldown_ms: config.circuit_breaker_cooldown_ms.unwrap_or(300_000),
+                    half_open_max_probes: config.circuit_breaker_half_open_probes.unwrap_or(5),
+                    half_open_success_rate: config
+                        .circuit_breaker_half_open_success_rate
+                        .unwrap_or(0.8),
+                });
         scheduler.register_task(
             config.name,
             TaskConfig {
                 retry_policy,
                 rate_limit: None,
-                circuit_breaker: None,
+                circuit_breaker,
                 max_concurrent: None,
             },
         );

sdks/java/processor/src/main/java/org/byteveda/taskito/processor/TaskHandlerProcessor.java

@@ -190,9 +190,34 @@ private String options(ExecutableElement method) {
         if (priority != 0) {
             chain.append(".priority(").append(priority).append(")");
         }
+        if (boolValue(mirror, "idempotent", false)) {
+            chain.append(".idempotent(true)");
+        }
+        appendCircuitBreaker(chain, mirror);
         return chain.toString();
     }
 
+    /** Emit a {@code .circuitBreaker(...)} call when the annotation configures a threshold. */
+    private void appendCircuitBreaker(StringBuilder chain, AnnotationMirror mirror) {
+        long threshold = longValue(mirror, "circuitBreakerThreshold", 0);
+        if (threshold <= 0) {
+            return;
+        }
+        // Fully qualified so the generated companion needs no extra import.
+        chain.append(".circuitBreaker(org.byteveda.taskito.task.CircuitBreakerConfig.builder(")
+                .append(threshold)
+                .append(")")
+                .append(".windowSeconds(")
+                .append(longValue(mirror, "circuitBreakerWindowSeconds", 60))
+                .append("L).cooldownSeconds(")
+                .append(longValue(mirror, "circuitBreakerCooldownSeconds", 300))
+                .append("L).halfOpenProbes(")
+                .append(longValue(mirror, "circuitBreakerHalfOpenProbes", 5))
+                .append(").halfOpenSuccessRate(")
+                .append(doubleValue(mirror, "circuitBreakerHalfOpenSuccessRate", 0.8))
+                .append(").build())");
+    }
+
     /**
      * The {@code TaskFunction} expression for a handler: a method reference when it
      * takes only the payload, otherwise a lambda that resolves each {@code @Resource}
@@ -285,6 +310,16 @@ private long longValue(AnnotationMirror mirror, String key, long fallback) {
         return value instanceof Number ? ((Number) value).longValue() : fallback;
     }
 
+    private boolean boolValue(AnnotationMirror mirror, String key, boolean fallback) {
+        Object value = rawValue(mirror, key);
+        return value instanceof Boolean ? (Boolean) value : fallback;
+    }
+
+    private double doubleValue(AnnotationMirror mirror, String key, double fallback) {
+        Object value = rawValue(mirror, key);
+        return value instanceof Number ? ((Number) value).doubleValue() : fallback;
+    }
+
     private Object rawValue(AnnotationMirror mirror, String key) {
         if (mirror == null) {
             return null;

sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java

@@ -24,10 +24,12 @@
 import org.byteveda.taskito.errors.WorkflowException;
 import org.byteveda.taskito.interception.Interception;
 import org.byteveda.taskito.interception.Interceptor;
+import org.byteveda.taskito.internal.IdempotencyKeys;
 import org.byteveda.taskito.locks.Lock;
 import org.byteveda.taskito.locks.LockInfo;
 import org.byteveda.taskito.middleware.EnqueueContext;
 import org.byteveda.taskito.middleware.Middleware;
+import org.byteveda.taskito.model.CircuitBreakerState;
 import org.byteveda.taskito.model.DeadJob;
 import org.byteveda.taskito.model.Job;
 import org.byteveda.taskito.model.JobError;
@@ -167,12 +169,13 @@ public <T> String enqueue(Task<T> task, T payload) {
 
     @Override
     public <T> String enqueue(Task<T> task, T payload, EnqueueOptions options) {
-        return required(dispatchEnqueue(task.name(), payload, options, task.codecNames()), task.name());
+        return required(
+                dispatchEnqueue(task.name(), payload, options, task.codecNames(), task.idempotent()), task.name());
     }
 
     @Override
     public String enqueue(String taskName, Object payload) {
-        return required(dispatchEnqueue(taskName, payload, EnqueueOptions.none(), List.of()), taskName);
+        return required(dispatchEnqueue(taskName, payload, EnqueueOptions.none(), List.of(), false), taskName);
     }
 
     @Override
@@ -182,12 +185,12 @@ public <T> Optional<String> tryEnqueue(Task<T> task, T payload) {
 
     @Override
     public <T> Optional<String> tryEnqueue(Task<T> task, T payload, EnqueueOptions options) {
-        return dispatchEnqueue(task.name(), payload, options, task.codecNames());
+        return dispatchEnqueue(task.name(), payload, options, task.codecNames(), task.idempotent());
     }
 
     @Override
     public Optional<String> tryEnqueue(String taskName, Object payload) {
-        return dispatchEnqueue(taskName, payload, EnqueueOptions.none(), List.of());
+        return dispatchEnqueue(taskName, payload, EnqueueOptions.none(), List.of(), false);
     }
 
     /** Unwrap a dispatch result, turning a gate {@code Skip} into an {@link EnqueueSkippedException}. */
@@ -202,7 +205,11 @@ private static String required(Optional<String> jobId, String taskName) {
      * the enqueue; throws when a gate rejects it.
      */
     private Optional<String> dispatchEnqueue(
-            String taskName, Object payload, EnqueueOptions options, List<String> codecNames) {
+            String taskName,
+            Object payload,
+            EnqueueOptions options,
+            List<String> codecNames,
+            boolean taskIdempotentDefault) {
         String originalTaskName = taskName;
         for (Interceptor interceptor : interceptors) {
             Interception outcome = interceptor.intercept(taskName, payload);
@@ -249,10 +256,40 @@ private Optional<String> dispatchEnqueue(
                     .metadata(encode(context.metadata()))
                     .build();
         }
-        byte[] data = encodeCodecs(serializer.serialize(context.payload()), codecNames);
+        // Serialize before codec-encoding so the idempotency key hashes the deterministic
+        // pre-codec payload — a non-deterministic codec (e.g. an AES-GCM nonce) must not
+        // change the dedup key.
+        byte[] payloadBytes = serializer.serialize(context.payload());
+        String uniqueKey = resolveUniqueKey(taskName, payloadBytes, finalOptions, taskIdempotentDefault);
+        if (uniqueKey != null && !uniqueKey.equals(finalOptions.uniqueKey())) {
+            finalOptions = finalOptions.toBuilder().uniqueKey(uniqueKey).build();
+        }
+        byte[] data = encodeCodecs(payloadBytes, codecNames);
         return Optional.of(backend.enqueue(taskName, data, encode(finalOptions)));
     }
 
+    /**
+     * Resolve the effective dedup key: an explicit {@code uniqueKey} wins, then an explicit
+     * {@code idempotencyKey}; a per-enqueue {@code idempotent(false)} opts out; otherwise a
+     * per-enqueue {@code idempotent(true)} or the task-level default auto-derives one from the
+     * payload. Returns {@code null} when the enqueue should not be deduped.
+     */
+    private static String resolveUniqueKey(
+            String taskName, byte[] preCodecPayload, EnqueueOptions options, boolean taskIdempotentDefault) {
+        if (options.uniqueKey() != null) {
+            return options.uniqueKey();
+        }
+        if (options.idempotencyKey() != null) {
+            return options.idempotencyKey();
+        }
+        Boolean idempotent = options.idempotent();
+        if (Boolean.FALSE.equals(idempotent)) {
+            return null;
+        }
+        boolean enabled = Boolean.TRUE.equals(idempotent) || taskIdempotentDefault;
+        return enabled ? IdempotencyKeys.autoKey(taskName, preCodecPayload) : null;
+    }
+
     /**
      * Apply a defer delay to the options, overriding any delay already set —
      * {@code Defer(Duration.ZERO)} (e.g. {@code deferUntil} of a past instant)
@@ -315,8 +352,18 @@ public <T> List<String> enqueueMany(Task<T> task, List<T> payloads, EnqueueOptio
                 throw new EnqueueSkippedException("enqueue of '" + task.name()
                         + "' was skipped or deferred by a gate; batch enqueue needs Allow");
             }
-            bytes[i] = encodeCodecs(serializer.serialize(payload), task.codecNames());
-            perJob.add(options);
+            // Resolve the dedup key per payload so bulk enqueue honors idempotent
+            // auto-derive, mirroring the single-enqueue path (explicit
+            // uniqueKey/idempotencyKey precedence preserved). Key hashes the
+            // deterministic pre-codec bytes.
+            byte[] payloadBytes = serializer.serialize(payload);
+            EnqueueOptions jobOptions = options;
+            String uniqueKey = resolveUniqueKey(task.name(), payloadBytes, jobOptions, task.idempotent());
+            if (uniqueKey != null && !uniqueKey.equals(jobOptions.uniqueKey())) {
+                jobOptions = jobOptions.toBuilder().uniqueKey(uniqueKey).build();
+            }
+            bytes[i] = encodeCodecs(payloadBytes, task.codecNames());
+            perJob.add(jobOptions);
         }
         return Arrays.asList(backend.enqueueMany(task.name(), bytes, encode(perJob)));
     }
@@ -427,6 +474,11 @@ public List<WorkerInfo> listWorkers() {
         return decodeList(backend.listWorkersJson(), WorkerInfo.class);
     }
 
+    @Override
+    public List<CircuitBreakerState> listCircuitBreakers() {
+        return decodeList(backend.listCircuitBreakersJson(), CircuitBreakerState.class);
+    }
+
     // ── Admin ───────────────────────────────────────────────────────
 
     @Override

sdks/java/src/main/java/org/byteveda/taskito/Taskito.java

@@ -21,6 +21,7 @@
 import org.byteveda.taskito.locks.Lock;
 import org.byteveda.taskito.locks.LockInfo;
 import org.byteveda.taskito.middleware.Middleware;
+import org.byteveda.taskito.model.CircuitBreakerState;
 import org.byteveda.taskito.model.DeadJob;
 import org.byteveda.taskito.model.Job;
 import org.byteveda.taskito.model.JobError;
@@ -178,6 +179,9 @@ static Builder builder() {
 
     List<WorkerInfo> listWorkers();
 
+    /** Every configured task's circuit-breaker state. */
+    List<CircuitBreakerState> listCircuitBreakers();
+
     // ── Admin ───────────────────────────────────────────────────────
 
     List<DeadJob> listDead(long limit, long offset);

sdks/java/src/main/java/org/byteveda/taskito/annotation/TaskHandler.java

@@ -32,4 +32,22 @@
 
     /** Priority; 0 (default) is left unset. */
     int priority() default 0;
+
+    /** Auto-derive an idempotency {@code uniqueKey} from the payload on every enqueue. */
+    boolean idempotent() default false;
+
+    /** Circuit-breaker failure threshold; 0 (default) leaves the breaker off. */
+    int circuitBreakerThreshold() default 0;
+
+    /** Rolling window, in seconds, over which failures count toward the threshold. */
+    long circuitBreakerWindowSeconds() default 60;
+
+    /** How long, in seconds, the breaker stays open before admitting half-open probes. */
+    long circuitBreakerCooldownSeconds() default 300;
+
+    /** Probe runs admitted while half-open. */
+    int circuitBreakerHalfOpenProbes() default 5;
+
+    /** Probe success rate (0.0–1.0) required to re-close the breaker. */
+    double circuitBreakerHalfOpenSuccessRate() default 0.8;
 }

sdks/java/src/main/java/org/byteveda/taskito/dashboard/Contract.java

@@ -44,6 +44,7 @@ static Map<String, Object> job(Job j) {
         m.put("error", j.error);
         m.put("unique_key", j.uniqueKey);
         m.put("namespace", j.namespace);
+        m.put("notes", j.notes);
         return m;
     }