From ab3cdc96caf31758d14e90a536f6431b27563226 Mon Sep 17 00:00:00 2001
From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com>
Date: Sun, 5 Jul 2026 09:29:10 +0530
Subject: [PATCH 1/9] fix(java): guard native handles against use-after-close

---
 .../taskito/internal/JniQueueBackend.java     | 251 ++++++++++++------
 .../taskito/internal/JniWorkerControl.java    |  64 ++++-
 .../org/byteveda/taskito/worker/Worker.java   |  15 +-
 3 files changed, 241 insertions(+), 89 deletions(-)

diff --git a/sdks/java/src/main/java/org/byteveda/taskito/internal/JniQueueBackend.java b/sdks/java/src/main/java/org/byteveda/taskito/internal/JniQueueBackend.java
index 3491fb33..fd53d0c8 100644
--- a/sdks/java/src/main/java/org/byteveda/taskito/internal/JniQueueBackend.java
+++ b/sdks/java/src/main/java/org/byteveda/taskito/internal/JniQueueBackend.java
@@ -1,18 +1,29 @@
 package org.byteveda.taskito.internal;
 
 import java.util.Optional;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.function.Supplier;
 import org.byteveda.taskito.spi.QueueBackend;
 import org.byteveda.taskito.spi.WorkerBridge;
 import org.byteveda.taskito.spi.WorkerControl;
 
-/** JNI-backed {@link QueueBackend} over a native queue handle. */
+/**
+ * JNI-backed {@link QueueBackend} over a native queue handle.
+ *
+ * <p>The native contract forbids using a handle after {@code close()} or closing
+ * concurrently with a call — violating it is a use-after-free in the native
+ * layer. Every call therefore holds the read lock while touching the handle;
+ * {@code close()} takes the write lock, so it waits out in-flight calls and any
+ * later call fails with {@link IllegalStateException} instead of crashing the JVM.
+ */
 public final class JniQueueBackend implements QueueBackend {
     private final long handle;
 
     /** Hot byte ops (enqueue/enqueueMany/getResult) route through this; FFM when available, else JNI. */
     private final NativeTransport transport;
 
-    private volatile boolean closed;
+    private final ReentrantReadWriteLock stateLock = new ReentrantReadWriteLock();
+    private boolean closed; // guarded by stateLock
 
     private JniQueueBackend(long handle) {
         this.handle = handle;
@@ -24,200 +35,228 @@ public static JniQueueBackend open(String optionsJson) {
         return new JniQueueBackend(NativeQueue.open(optionsJson));
     }
 
+    private <T> T withOpenHandle(Supplier<T> nativeCall) {
+        stateLock.readLock().lock();
+        try {
+            if (closed) {
+                throw new IllegalStateException("queue backend is closed");
+            }
+            return nativeCall.get();
+        } finally {
+            stateLock.readLock().unlock();
+        }
+    }
+
     @Override
     public String enqueue(String taskName, byte[] payload, String optionsJson) {
-        return transport.enqueue(taskName, payload, optionsJson);
+        return withOpenHandle(() -> transport.enqueue(taskName, payload, optionsJson));
     }
 
     @Override
     public String[] enqueueMany(String taskName, byte[][] payloads, String optionsJson) {
-        return transport.enqueueMany(taskName, payloads, optionsJson);
+        return withOpenHandle(() -> transport.enqueueMany(taskName, payloads, optionsJson));
     }
 
     @Override
     public Optional<String> getJobJson(String jobId) {
-        return Optional.ofNullable(NativeQueue.getJob(handle, jobId));
+        return withOpenHandle(() -> Optional.ofNullable(NativeQueue.getJob(handle, jobId)));
     }
 
     @Override
     public Optional<byte[]> getResult(String jobId) {
-        return Optional.ofNullable(transport.getResult(jobId));
+        return withOpenHandle(() -> Optional.ofNullable(transport.getResult(jobId)));
     }
 
     @Override
     public boolean cancel(String jobId) {
-        return NativeQueue.cancel(handle, jobId);
+        return withOpenHandle(() -> NativeQueue.cancel(handle, jobId));
     }
 
     @Override
     public boolean requestCancel(String jobId) {
-        return NativeQueue.requestCancel(handle, jobId);
+        return withOpenHandle(() -> NativeQueue.requestCancel(handle, jobId));
     }
 
     @Override
     public boolean isCancelRequested(String jobId) {
-        return NativeQueue.isCancelRequested(handle, jobId);
+        return withOpenHandle(() -> NativeQueue.isCancelRequested(handle, jobId));
     }
 
     @Override
     public void setProgress(String jobId, int progress) {
-        NativeQueue.setProgress(handle, jobId, progress);
+        withOpenHandle(() -> {
+            NativeQueue.setProgress(handle, jobId, progress);
+            return null;
+        });
     }
 
     @Override
     public String statsJson() {
-        return NativeQueue.stats(handle);
+        return withOpenHandle(() -> NativeQueue.stats(handle));
     }
 
     @Override
     public String statsByQueueJson(String queue) {
-        return NativeQueue.statsByQueue(handle, queue);
+        return withOpenHandle(() -> NativeQueue.statsByQueue(handle, queue));
     }
 
     @Override
     public String statsAllQueuesJson() {
-        return NativeQueue.statsAllQueues(handle);
+        return withOpenHandle(() -> NativeQueue.statsAllQueues(handle));
     }
 
     @Override
     public String listJobsJson(String filterJson) {
-        return NativeQueue.listJobs(handle, filterJson);
+        return withOpenHandle(() -> NativeQueue.listJobs(handle, filterJson));
     }
 
     @Override
     public String jobErrorsJson(String jobId) {
-        return NativeQueue.jobErrors(handle, jobId);
+        return withOpenHandle(() -> NativeQueue.jobErrors(handle, jobId));
     }
 
     @Override
     public String metricsJson(String taskNameOrNull, long sinceMs) {
-        return NativeQueue.metrics(handle, taskNameOrNull, sinceMs);
+        return withOpenHandle(() -> NativeQueue.metrics(handle, taskNameOrNull, sinceMs));
     }
 
     @Override
     public String listWorkersJson() {
-        return NativeQueue.listWorkers(handle);
+        return withOpenHandle(() -> NativeQueue.listWorkers(handle));
     }
 
     @Override
     public String listDeadJson(long limit, long offset) {
-        return NativeQueue.listDead(handle, limit, offset);
+        return withOpenHandle(() -> NativeQueue.listDead(handle, limit, offset));
     }
 
     @Override
     public String retryDead(String deadId) {
-        return NativeQueue.retryDead(handle, deadId);
+        return withOpenHandle(() -> NativeQueue.retryDead(handle, deadId));
     }
 
     @Override
     public boolean deleteDead(String deadId) {
-        return NativeQueue.deleteDead(handle, deadId);
+        return withOpenHandle(() -> NativeQueue.deleteDead(handle, deadId));
     }
 
     @Override
     public long purgeDead(long olderThanMs) {
-        return NativeQueue.purgeDead(handle, olderThanMs);
+        return withOpenHandle(() -> NativeQueue.purgeDead(handle, olderThanMs));
     }
 
     @Override
     public String listDeadByTaskJson(String taskName, long limit, long offset) {
-        return NativeQueue.listDeadByTask(handle, taskName, limit, offset);
+        return withOpenHandle(() -> NativeQueue.listDeadByTask(handle, taskName, limit, offset));
     }
 
     @Override
     public long purgeDeadByTask(String taskName) {
-        return NativeQueue.purgeDeadByTask(handle, taskName);
+        return withOpenHandle(() -> NativeQueue.purgeDeadByTask(handle, taskName));
     }
 
     @Override
     public long purgeCompleted(long olderThanMs) {
-        return NativeQueue.purgeCompleted(handle, olderThanMs);
+        return withOpenHandle(() -> NativeQueue.purgeCompleted(handle, olderThanMs));
     }
 
     @Override
     public void pauseQueue(String queue) {
-        NativeQueue.pauseQueue(handle, queue);
+        withOpenHandle(() -> {
+            NativeQueue.pauseQueue(handle, queue);
+            return null;
+        });
     }
 
     @Override
     public void resumeQueue(String queue) {
-        NativeQueue.resumeQueue(handle, queue);
+        withOpenHandle(() -> {
+            NativeQueue.resumeQueue(handle, queue);
+            return null;
+        });
     }
 
     @Override
     public String listPausedQueuesJson() {
-        return NativeQueue.listPausedQueues(handle);
+        return withOpenHandle(() -> NativeQueue.listPausedQueues(handle));
     }
 
     @Override
     public Optional<String> getSetting(String key) {
-        return Optional.ofNullable(NativeQueue.getSetting(handle, key));
+        return withOpenHandle(() -> Optional.ofNullable(NativeQueue.getSetting(handle, key)));
     }
 
     @Override
     public void setSetting(String key, String value) {
-        NativeQueue.setSetting(handle, key, value);
+        withOpenHandle(() -> {
+            NativeQueue.setSetting(handle, key, value);
+            return null;
+        });
     }
 
     @Override
     public boolean deleteSetting(String key) {
-        return NativeQueue.deleteSetting(handle, key);
+        return withOpenHandle(() -> NativeQueue.deleteSetting(handle, key));
     }
 
     @Override
     public String listSettingsJson() {
-        return NativeQueue.listSettings(handle);
+        return withOpenHandle(() -> NativeQueue.listSettings(handle));
     }
 
     @Override
     public void writeTaskLog(String jobId, String taskName, String level, String message, String extraOrNull) {
-        NativeQueue.writeTaskLog(handle, jobId, taskName, level, message, extraOrNull);
+        withOpenHandle(() -> {
+            NativeQueue.writeTaskLog(handle, jobId, taskName, level, message, extraOrNull);
+            return null;
+        });
     }
 
     @Override
     public String getTaskLogsJson(String jobId) {
-        return NativeQueue.getTaskLogs(handle, jobId);
+        return withOpenHandle(() -> NativeQueue.getTaskLogs(handle, jobId));
     }
 
     @Override
     public boolean acquireLock(String name, String ownerId, long ttlMs) {
-        return NativeQueue.acquireLock(handle, name, ownerId, ttlMs);
+        return withOpenHandle(() -> NativeQueue.acquireLock(handle, name, ownerId, ttlMs));
     }
 
     @Override
     public boolean releaseLock(String name, String ownerId) {
-        return NativeQueue.releaseLock(handle, name, ownerId);
+        return withOpenHandle(() -> NativeQueue.releaseLock(handle, name, ownerId));
     }
 
     @Override
     public boolean extendLock(String name, String ownerId, long ttlMs) {
-        return NativeQueue.extendLock(handle, name, ownerId, ttlMs);
+        return withOpenHandle(() -> NativeQueue.extendLock(handle, name, ownerId, ttlMs));
     }
 
     @Override
     public Optional<String> lockInfoJson(String name) {
-        return Optional.ofNullable(NativeQueue.getLockInfo(handle, name));
+        return withOpenHandle(() -> Optional.ofNullable(NativeQueue.getLockInfo(handle, name)));
     }
 
     @Override
     public long registerPeriodic(
             String name, String taskName, String cron, byte[] args, String queue, String timezone, boolean enabled) {
-        return NativeQueue.registerPeriodic(handle, name, taskName, cron, args, queue, timezone, enabled);
+        return withOpenHandle(
+                () -> NativeQueue.registerPeriodic(handle, name, taskName, cron, args, queue, timezone, enabled));
     }
 
     @Override
     public String listPeriodicJson() {
-        return NativeQueue.listPeriodic(handle);
+        return withOpenHandle(() -> NativeQueue.listPeriodic(handle));
     }
 
     @Override
     public boolean deletePeriodic(String name) {
-        return NativeQueue.deletePeriodic(handle, name);
+        return withOpenHandle(() -> NativeQueue.deletePeriodic(handle, name));
     }
 
     @Override
     public boolean setPeriodicEnabled(String name, boolean enabled) {
-        return NativeQueue.setPeriodicEnabled(handle, name, enabled);
+        return withOpenHandle(() -> NativeQueue.setPeriodicEnabled(handle, name, enabled));
     }
 
     @Override
@@ -232,7 +271,7 @@ public String submitWorkflow(
             String[] deferredNames,
             String parentRunId,
             String parentNodeName) {
-        return NativeWorkflows.submitWorkflow(
+        return withOpenHandle(() -> NativeWorkflows.submitWorkflow(
                 handle,
                 name,
                 version,
@@ -243,37 +282,41 @@ public String submitWorkflow(
                 paramsJson,
                 deferredNames,
                 parentRunId,
-                parentNodeName);
+                parentNodeName));
     }
 
     @Override
     public String markWorkflowNodeResult(String jobId, boolean succeeded, String error, boolean skipCascade) {
-        return NativeWorkflows.markWorkflowNodeResult(handle, jobId, succeeded, error, skipCascade);
+        return withOpenHandle(
+                () -> NativeWorkflows.markWorkflowNodeResult(handle, jobId, succeeded, error, skipCascade));
     }
 
     @Override
     public Optional<String> getWorkflowStatusJson(String runId) {
-        return Optional.ofNullable(NativeWorkflows.getWorkflowStatus(handle, runId));
+        return withOpenHandle(() -> Optional.ofNullable(NativeWorkflows.getWorkflowStatus(handle, runId)));
     }
 
     @Override
     public void cancelWorkflowRun(String runId) {
-        NativeWorkflows.cancelWorkflowRun(handle, runId);
+        withOpenHandle(() -> {
+            NativeWorkflows.cancelWorkflowRun(handle, runId);
+            return null;
+        });
     }
 
     @Override
     public Optional<String> getWorkflowPlanJson(String runId) {
-        return Optional.ofNullable(NativeWorkflows.getWorkflowPlan(handle, runId));
+        return withOpenHandle(() -> Optional.ofNullable(NativeWorkflows.getWorkflowPlan(handle, runId)));
     }
 
     @Override
     public Optional<String> workflowNodeForJobJson(String jobId) {
-        return Optional.ofNullable(NativeWorkflows.workflowNodeForJob(handle, jobId));
+        return withOpenHandle(() -> Optional.ofNullable(NativeWorkflows.workflowNodeForJob(handle, jobId)));
     }
 
     @Override
     public Optional<String> workflowNameForRun(String runId) {
-        return Optional.ofNullable(NativeWorkflows.workflowNameForRun(handle, runId));
+        return withOpenHandle(() -> Optional.ofNullable(NativeWorkflows.workflowNameForRun(handle, runId)));
     }
 
     @Override
@@ -287,13 +330,23 @@ public String[] expandFanOut(
             int maxRetries,
             long timeoutMs,
             int priority) {
-        return NativeWorkflows.expandFanOut(
-                handle, runId, parentNode, childNames, childPayloads, taskName, queue, maxRetries, timeoutMs, priority);
+        return withOpenHandle(() -> NativeWorkflows.expandFanOut(
+                handle,
+                runId,
+                parentNode,
+                childNames,
+                childPayloads,
+                taskName,
+                queue,
+                maxRetries,
+                timeoutMs,
+                priority));
     }
 
     @Override
     public Optional<String> checkFanOutCompletionJson(String runId, String parentNode) {
-        return Optional.ofNullable(NativeWorkflows.checkFanOutCompletion(handle, runId, parentNode));
+        return withOpenHandle(
+                () -> Optional.ofNullable(NativeWorkflows.checkFanOutCompletion(handle, runId, parentNode)));
     }
 
     @Override
@@ -306,98 +359,144 @@ public String createDeferredJob(
             int maxRetries,
             long timeoutMs,
             int priority) {
-        return NativeWorkflows.createDeferredJob(
-                handle, runId, nodeName, payload, taskName, queue, maxRetries, timeoutMs, priority);
+        return withOpenHandle(() -> NativeWorkflows.createDeferredJob(
+                handle, runId, nodeName, payload, taskName, queue, maxRetries, timeoutMs, priority));
     }
 
     @Override
     public void cascadeSkipPending(String runId) {
-        NativeWorkflows.cascadeSkipPending(handle, runId);
+        withOpenHandle(() -> {
+            NativeWorkflows.cascadeSkipPending(handle, runId);
+            return null;
+        });
     }
 
     @Override
     public Optional<String> finalizeRunIfTerminal(String runId) {
-        return Optional.ofNullable(NativeWorkflows.finalizeRunIfTerminal(handle, runId));
+        return withOpenHandle(() -> Optional.ofNullable(NativeWorkflows.finalizeRunIfTerminal(handle, runId)));
     }
 
     @Override
     public void setWorkflowNodeWaitingApproval(String runId, String nodeName) {
-        NativeWorkflows.setWorkflowNodeWaitingApproval(handle, runId, nodeName);
+        withOpenHandle(() -> {
+            NativeWorkflows.setWorkflowNodeWaitingApproval(handle, runId, nodeName);
+            return null;
+        });
     }
 
     @Override
     public void resolveWorkflowGate(String runId, String nodeName, boolean approved, String error) {
-        NativeWorkflows.resolveWorkflowGate(handle, runId, nodeName, approved, error);
+        withOpenHandle(() -> {
+            NativeWorkflows.resolveWorkflowGate(handle, runId, nodeName, approved, error);
+            return null;
+        });
     }
 
     @Override
     public void setWorkflowNodeRunning(String runId, String nodeName) {
-        NativeWorkflows.setWorkflowNodeRunning(handle, runId, nodeName);
+        withOpenHandle(() -> {
+            NativeWorkflows.setWorkflowNodeRunning(handle, runId, nodeName);
+            return null;
+        });
     }
 
     @Override
     public void failWorkflowNode(String runId, String nodeName, String error) {
-        NativeWorkflows.failWorkflowNode(handle, runId, nodeName, error);
+        withOpenHandle(() -> {
+            NativeWorkflows.failWorkflowNode(handle, runId, nodeName, error);
+            return null;
+        });
     }
 
     @Override
     public void skipWorkflowNode(String runId, String nodeName) {
-        NativeWorkflows.skipWorkflowNode(handle, runId, nodeName);
+        withOpenHandle(() -> {
+            NativeWorkflows.skipWorkflowNode(handle, runId, nodeName);
+            return null;
+        });
     }
 
     @Override
     public void setWorkflowNodeCacheHit(String runId, String nodeName) {
-        NativeWorkflows.setWorkflowNodeCacheHit(handle, runId, nodeName);
+        withOpenHandle(() -> {
+            NativeWorkflows.setWorkflowNodeCacheHit(handle, runId, nodeName);
+            return null;
+        });
     }
 
     @Override
     public void setWorkflowRunCompensating(String runId) {
-        NativeWorkflows.setWorkflowRunCompensating(handle, runId);
+        withOpenHandle(() -> {
+            NativeWorkflows.setWorkflowRunCompensating(handle, runId);
+            return null;
+        });
     }
 
     @Override
     public void setWorkflowRunCompensated(String runId, long completedAt) {
-        NativeWorkflows.setWorkflowRunCompensated(handle, runId, completedAt);
+        withOpenHandle(() -> {
+            NativeWorkflows.setWorkflowRunCompensated(handle, runId, completedAt);
+            return null;
+        });
     }
 
     @Override
     public void setWorkflowRunCompensationFailed(String runId, long completedAt, String error) {
-        NativeWorkflows.setWorkflowRunCompensationFailed(handle, runId, completedAt, error);
+        withOpenHandle(() -> {
+            NativeWorkflows.setWorkflowRunCompensationFailed(handle, runId, completedAt, error);
+            return null;
+        });
     }
 
     @Override
     public void setWorkflowRunCompletedWithFailures(String runId, long completedAt) {
-        NativeWorkflows.setWorkflowRunCompletedWithFailures(handle, runId, completedAt);
+        withOpenHandle(() -> {
+            NativeWorkflows.setWorkflowRunCompletedWithFailures(handle, runId, completedAt);
+            return null;
+        });
     }
 
     @Override
     public void setWorkflowNodeCompensationJob(
             String runId, String nodeName, String compensationJobId, long startedAt) {
-        NativeWorkflows.setWorkflowNodeCompensationJob(handle, runId, nodeName, compensationJobId, startedAt);
+        withOpenHandle(() -> {
+            NativeWorkflows.setWorkflowNodeCompensationJob(handle, runId, nodeName, compensationJobId, startedAt);
+            return null;
+        });
     }
 
     @Override
     public void setWorkflowNodeCompensated(String runId, String nodeName, long completedAt) {
-        NativeWorkflows.setWorkflowNodeCompensated(handle, runId, nodeName, completedAt);
+        withOpenHandle(() -> {
+            NativeWorkflows.setWorkflowNodeCompensated(handle, runId, nodeName, completedAt);
+            return null;
+        });
     }
 
     @Override
     public void setWorkflowNodeCompensationFailed(String runId, String nodeName, String error, long completedAt) {
-        NativeWorkflows.setWorkflowNodeCompensationFailed(handle, runId, nodeName, error, completedAt);
+        withOpenHandle(() -> {
+            NativeWorkflows.setWorkflowNodeCompensationFailed(handle, runId, nodeName, error, completedAt);
+            return null;
+        });
     }
 
     @Override
     public WorkerControl startWorker(WorkerBridge bridge, String optionsJson) {
-        return new JniWorkerControl(NativeQueue.runWorker(handle, bridge, optionsJson));
+        return withOpenHandle(() -> new JniWorkerControl(NativeQueue.runWorker(handle, bridge, optionsJson)));
     }
 
-    /** Idempotent: the native handle is freed exactly once, so a second {@code close()}
-     * (e.g. an explicit call plus try-with-resources) can't double-free. */
+    /** Idempotent: frees the native queue handle exactly once, after in-flight calls drain. */
     @Override
-    public synchronized void close() {
-        if (!closed) {
-            closed = true;
-            NativeQueue.close(handle);
+    public void close() {
+        stateLock.writeLock().lock();
+        try {
+            if (!closed) {
+                closed = true;
+                NativeQueue.close(handle);
+            }
+        } finally {
+            stateLock.writeLock().unlock();
         }
     }
 }
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 13f540d0..b9eebe7e 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
@@ -1,48 +1,88 @@
 package org.byteveda.taskito.internal;
 
 import java.util.Optional;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.function.Supplier;
 import org.byteveda.taskito.spi.WorkerControl;
 
-/** JNI-backed {@link WorkerControl} over a native worker handle. */
+/**
+ * JNI-backed {@link WorkerControl} over a native worker handle.
+ *
+ * <p>The native contract forbids using a handle after {@code close()} or closing
+ * concurrently with a call — violating it is a use-after-free in the native
+ * layer. Every call therefore holds the read lock while touching the handle;
+ * {@code close()} takes the write lock, so it waits out in-flight calls and any
+ * later call fails with {@link IllegalStateException} instead of crashing the JVM.
+ */
 public final class JniWorkerControl implements WorkerControl {
     private final long handle;
-    private volatile boolean closed;
+    private final ReentrantReadWriteLock stateLock = new ReentrantReadWriteLock();
+    private boolean closed; // guarded by stateLock
 
     JniWorkerControl(long handle) {
         this.handle = handle;
     }
 
+    private <T> T withOpenHandle(Supplier<T> nativeCall) {
+        stateLock.readLock().lock();
+        try {
+            if (closed) {
+                throw new IllegalStateException("worker control is closed");
+            }
+            return nativeCall.get();
+        } finally {
+            stateLock.readLock().unlock();
+        }
+    }
+
     @Override
     public void completeJob(long token, byte[] result) {
-        NativeWorker.completeJob(handle, token, result);
+        withOpenHandle(() -> {
+            NativeWorker.completeJob(handle, token, result);
+            return null;
+        });
     }
 
     @Override
     public void failJob(long token, String error) {
-        NativeWorker.failJob(handle, token, error);
+        withOpenHandle(() -> {
+            NativeWorker.failJob(handle, token, error);
+            return null;
+        });
     }
 
     @Override
     public void cancelJob(long token) {
-        NativeWorker.cancelJob(handle, token);
+        withOpenHandle(() -> {
+            NativeWorker.cancelJob(handle, token);
+            return null;
+        });
     }
 
     @Override
     public void stop() {
-        NativeWorker.stop(handle);
+        withOpenHandle(() -> {
+            NativeWorker.stop(handle);
+            return null;
+        });
     }
 
     @Override
     public Optional<String> meshClusterInfoJson() {
-        return Optional.ofNullable(NativeWorker.meshClusterInfo(handle));
+        return withOpenHandle(() -> Optional.ofNullable(NativeWorker.meshClusterInfo(handle)));
     }
 
-    /** Idempotent: frees the native worker handle exactly once. */
+    /** Idempotent: frees the native worker handle exactly once, after in-flight calls drain. */
     @Override
-    public synchronized void close() {
-        if (!closed) {
-            closed = true;
-            NativeWorker.close(handle);
+    public void close() {
+        stateLock.writeLock().lock();
+        try {
+            if (!closed) {
+                closed = true;
+                NativeWorker.close(handle);
+            }
+        } finally {
+            stateLock.writeLock().unlock();
         }
     }
 }
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 310fd1a3..729a0780 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
@@ -43,6 +43,7 @@
  * jobs.
  */
 public final class Worker implements AutoCloseable {
+    private static final System.Logger LOG = System.getLogger(Worker.class.getName());
     private static final long SHUTDOWN_TIMEOUT_SECONDS = 30;
     private static final ObjectMapper MESH_JSON = new ObjectMapper();
 
@@ -144,13 +145,25 @@ public synchronized void close() {
         executor.shutdown(); // stop accepting; let running handlers finish
         try {
             if (!executor.awaitTermination(SHUTDOWN_TIMEOUT_SECONDS, TimeUnit.SECONDS)) {
+                // Interrupt stragglers, then give them one more window: shutdownNow()
+                // only *requests* interruption, and a handler stuck in
+                // non-interruptible work can outlive it.
                 executor.shutdownNow();
+                if (!executor.awaitTermination(SHUTDOWN_TIMEOUT_SECONDS, TimeUnit.SECONDS)) {
+                    LOG.log(
+                            System.Logger.Level.WARNING,
+                            "handler threads still running after {0}s; closing the native worker anyway —"
+                                    + " their late completions will be rejected",
+                            2 * SHUTDOWN_TIMEOUT_SECONDS);
+                }
             }
         } catch (InterruptedException e) {
             Thread.currentThread().interrupt();
             executor.shutdownNow();
         }
-        control.close(); // now safe to free the native handle — no handler can touch it
+        // Safe even if a straggler survives: JniWorkerControl serializes close()
+        // against in-flight native calls and rejects any call made afterwards.
+        control.close();
         if (tracker != null) {
             tracker.close(); // stop the gate-timeout scheduler
         }

From 85ba39d3de5488ffc7e26b990173898ac1eb435d Mon Sep 17 00:00:00 2001
From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com>
Date: Sun, 5 Jul 2026 09:29:10 +0530
Subject: [PATCH 2/9] fix(java): fail workflow nodes on promotion errors

A promotion failure previously propagated into the outcome listener,
where it was swallowed silently and left the node PENDING forever.
---
 .../org/byteveda/taskito/events/Emitter.java  | 12 +++-
 .../taskito/workflows/WorkflowTracker.java    | 64 ++++++++++++++++++-
 2 files changed, 72 insertions(+), 4 deletions(-)

diff --git a/sdks/java/src/main/java/org/byteveda/taskito/events/Emitter.java b/sdks/java/src/main/java/org/byteveda/taskito/events/Emitter.java
index 8e40df08..d3838e95 100644
--- a/sdks/java/src/main/java/org/byteveda/taskito/events/Emitter.java
+++ b/sdks/java/src/main/java/org/byteveda/taskito/events/Emitter.java
@@ -8,6 +8,8 @@
 
 /** Dispatches {@link OutcomeEvent}s to registered listeners. Thread-safe. */
 public final class Emitter {
+    private static final System.Logger LOG = System.getLogger(Emitter.class.getName());
+
     private final Map<EventName, List<Consumer<OutcomeEvent>>> listeners = new EnumMap<>(EventName.class);
 
     public void on(EventName name, Consumer<OutcomeEvent> listener) {
@@ -23,8 +25,14 @@ public void emit(OutcomeEvent event) {
         for (Consumer<OutcomeEvent> listener : bound) {
             try {
                 listener.accept(event);
-            } catch (RuntimeException ignored) {
-                // A listener fault must not break event dispatch.
+            } catch (RuntimeException e) {
+                // A listener fault must not break event dispatch — but it must
+                // not vanish either: this is the only place a workflow-tracker
+                // failure would otherwise surface.
+                LOG.log(
+                        System.Logger.Level.WARNING,
+                        "listener for " + event.name + " (job " + event.jobId + ") threw",
+                        e);
             }
         }
     }
diff --git a/sdks/java/src/main/java/org/byteveda/taskito/workflows/WorkflowTracker.java b/sdks/java/src/main/java/org/byteveda/taskito/workflows/WorkflowTracker.java
index e819dbc3..29e20a3a 100644
--- a/sdks/java/src/main/java/org/byteveda/taskito/workflows/WorkflowTracker.java
+++ b/sdks/java/src/main/java/org/byteveda/taskito/workflows/WorkflowTracker.java
@@ -45,6 +45,8 @@
  * run reaches a terminal state.
  */
 public final class WorkflowTracker {
+    private static final System.Logger LOG = System.getLogger(WorkflowTracker.class.getName());
+
     private final QueueBackend backend;
     private final Serializer serializer;
     private final ObjectMapper json = new ObjectMapper();
@@ -146,7 +148,11 @@ private void onOutcome(String jobId, boolean succeeded, String error) {
             return;
         }
         if (succeeded) {
-            expandFanOutIfAny(ref.runId, ref.nodeName, jobId, plan);
+            try {
+                expandFanOutIfAny(ref.runId, ref.nodeName, jobId, plan);
+            } catch (RuntimeException e) {
+                failFanOutsOf(ref.runId, ref.nodeName, plan, e);
+            }
         }
         // Evaluate every successor's condition — running on-failure recovery nodes
         // and skip-cascading the rest. Replaces a blanket fail-fast cascade.
@@ -290,9 +296,63 @@ private void promoteOrSkip(String runId, PlanNode node, Map<String, NodeSnapshot
             }
             createDeferredJobFor(runId, node);
         } catch (RuntimeException e) {
+            // A failed promotion (native error, missing payload, bad metadata) must
+            // never leave the node wedged PENDING with no trace — outcome listeners
+            // swallow throws, so rethrowing here would silently stall the run.
+            LOG.log(
+                    System.Logger.Level.WARNING,
+                    "promoting workflow node '" + node.name + "' in run " + runId + " failed",
+                    e);
+            failNodeAndCascade(runId, node.name, promotionKey, plan, e);
+        }
+    }
+
+    /** Mark a node Failed so the run still settles; on storage failure, allow a later retry. */
+    private void failNodeAndCascade(
+            String runId, String nodeName, GateKey promotionKey, RunPlan plan, RuntimeException cause) {
+        try {
+            backend.failWorkflowNode(runId, nodeName, describe(cause));
+            promoteSuccessors(runId, nodeName, plan);
+            finalizeIfTerminal(runId);
+        } catch (RuntimeException e) {
+            // Even the failure record failed (e.g. storage down). Roll the dedupe
+            // marker back so the next outcome can retry the promotion.
             promotedNodes.remove(promotionKey);
-            throw e;
+            LOG.log(
+                    System.Logger.Level.ERROR,
+                    "failed to record failure of workflow node '" + nodeName + "' in run " + runId,
+                    e);
+        }
+    }
+
+    /**
+     * A fan-out expansion failed (undeserializable producer result, native error):
+     * fail each fan-out node instead of leaving it PENDING forever, then cascade.
+     */
+    private void failFanOutsOf(String runId, String producer, RunPlan plan, RuntimeException cause) {
+        LOG.log(
+                System.Logger.Level.WARNING,
+                "expanding fan-out(s) after node '" + producer + "' in run " + runId + " failed",
+                cause);
+        if (plan == null) {
+            return;
         }
+        for (PlanNode fanOut : plan.successorsMatching(producer, candidate -> candidate.fanOut != null)) {
+            try {
+                backend.failWorkflowNode(runId, fanOut.name, describe(cause));
+                promoteSuccessors(runId, fanOut.name, plan);
+            } catch (RuntimeException e) {
+                LOG.log(
+                        System.Logger.Level.ERROR,
+                        "failed to record failure of fan-out node '" + fanOut.name + "' in run " + runId,
+                        e);
+            }
+        }
+    }
+
+    private static String describe(RuntimeException cause) {
+        String message = cause.getMessage();
+        return cause.getClass().getSimpleName() + (message == null ? "" : ": " + message);
     }
 
     /** If a prior run cached this step's result within its TTL, mark it a cache hit instead of running it. */

From 388e641e1fc1520a37bd6e03baa245aa08da9673 Mon Sep 17 00:00:00 2001
From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com>
Date: Sun, 5 Jul 2026 09:29:10 +0530
Subject: [PATCH 3/9] chore(java): publish processor, test-support and spring

---
 sdks/java/processor/build.gradle.kts    | 29 +++++++++++++++++++++++++
 sdks/java/spring/build.gradle.kts       | 29 +++++++++++++++++++++++++
 sdks/java/test-support/build.gradle.kts | 29 +++++++++++++++++++++++++
 3 files changed, 87 insertions(+)

diff --git a/sdks/java/processor/build.gradle.kts b/sdks/java/processor/build.gradle.kts
index 4e53dc47..c1bbb822 100644
--- a/sdks/java/processor/build.gradle.kts
+++ b/sdks/java/processor/build.gradle.kts
@@ -5,11 +5,40 @@ plugins {
     `java-library`
     checkstyle
     id("com.diffplug.spotless") version "7.2.1"
+    id("com.vanniktech.maven.publish") version "0.37.0"
 }
 
 group = "org.byteveda"
 version = "0.18.0"
 
+mavenPublishing {
+    publishToMavenCentral()
+    signAllPublications()
+    coordinates(group.toString(), "taskito-processor", version.toString())
+    pom {
+        name.set("Taskito Processor")
+        description.set("Compile-time annotation processor generating task-handler bindings for the Taskito JVM SDK.")
+        url.set("https://github.com/ByteVeda/taskito")
+        licenses {
+            license {
+                name.set("MIT")
+                url.set("https://opensource.org/licenses/MIT")
+            }
+        }
+        developers {
+            developer {
+                id.set("byteveda")
+                name.set("ByteVeda")
+            }
+        }
+        scm {
+            url.set("https://github.com/ByteVeda/taskito")
+            connection.set("scm:git:https://github.com/ByteVeda/taskito.git")
+            developerConnection.set("scm:git:ssh://git@github.com/ByteVeda/taskito.git")
+        }
+    }
+}
+
 java {
     sourceCompatibility = JavaVersion.VERSION_17
     targetCompatibility = JavaVersion.VERSION_17
diff --git a/sdks/java/spring/build.gradle.kts b/sdks/java/spring/build.gradle.kts
index 8e15e366..ae028ba0 100644
--- a/sdks/java/spring/build.gradle.kts
+++ b/sdks/java/spring/build.gradle.kts
@@ -4,12 +4,41 @@ plugins {
     `java-library`
     checkstyle
     id("com.diffplug.spotless") version "7.2.1"
+    id("com.vanniktech.maven.publish") version "0.37.0"
 }
 
 group = "org.byteveda"
 
 version = "0.18.0"
 
+mavenPublishing {
+    publishToMavenCentral()
+    signAllPublications()
+    coordinates(group.toString(), "taskito-spring", version.toString())
+    pom {
+        name.set("Taskito Spring")
+        description.set("Spring Boot 3 starter that auto-configures a Taskito bean.")
+        url.set("https://github.com/ByteVeda/taskito")
+        licenses {
+            license {
+                name.set("MIT")
+                url.set("https://opensource.org/licenses/MIT")
+            }
+        }
+        developers {
+            developer {
+                id.set("byteveda")
+                name.set("ByteVeda")
+            }
+        }
+        scm {
+            url.set("https://github.com/ByteVeda/taskito")
+            connection.set("scm:git:https://github.com/ByteVeda/taskito.git")
+            developerConnection.set("scm:git:ssh://git@github.com/ByteVeda/taskito.git")
+        }
+    }
+}
+
 java {
     sourceCompatibility = JavaVersion.VERSION_17
     targetCompatibility = JavaVersion.VERSION_17
diff --git a/sdks/java/test-support/build.gradle.kts b/sdks/java/test-support/build.gradle.kts
index 033124d5..3f175f43 100644
--- a/sdks/java/test-support/build.gradle.kts
+++ b/sdks/java/test-support/build.gradle.kts
@@ -5,11 +5,40 @@ plugins {
     `java-library`
     checkstyle
     id("com.diffplug.spotless") version "7.2.1"
+    id("com.vanniktech.maven.publish") version "0.37.0"
 }
 
 group = "org.byteveda"
 version = "0.18.0"
 
+mavenPublishing {
+    publishToMavenCentral()
+    signAllPublications()
+    coordinates(group.toString(), "taskito-test", version.toString())
+    pom {
+        name.set("Taskito Test Support")
+        description.set("In-memory queue backend for testing Taskito JVM applications without the native runtime.")
+        url.set("https://github.com/ByteVeda/taskito")
+        licenses {
+            license {
+                name.set("MIT")
+                url.set("https://opensource.org/licenses/MIT")
+            }
+        }
+        developers {
+            developer {
+                id.set("byteveda")
+                name.set("ByteVeda")
+            }
+        }
+        scm {
+            url.set("https://github.com/ByteVeda/taskito")
+            connection.set("scm:git:https://github.com/ByteVeda/taskito.git")
+            developerConnection.set("scm:git:ssh://git@github.com/ByteVeda/taskito.git")
+        }
+    }
+}
+
 java {
     sourceCompatibility = JavaVersion.VERSION_17
     targetCompatibility = JavaVersion.VERSION_17

From e6db2875310aaba5fc347b5f70b1875edf4d51dd Mon Sep 17 00:00:00 2001
From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com>
Date: Sun, 5 Jul 2026 09:42:14 +0530
Subject: [PATCH 4/9] fix(java): serialize lazy workflow storage init

Concurrent first workflow calls could run the DDL migrations in
parallel and hit Postgres catalog races.
---
 crates/taskito-java/src/backend.rs | 17 ++++++++++++++++-
 1 file changed, 16 insertions(+), 1 deletion(-)

diff --git a/crates/taskito-java/src/backend.rs b/crates/taskito-java/src/backend.rs
index 94e77a99..b40addc2 100644
--- a/crates/taskito-java/src/backend.rs
+++ b/crates/taskito-java/src/backend.rs
@@ -20,6 +20,11 @@ pub struct QueueHandle {
     /// runs the workflow-table migrations). Shares this queue's connection pool.
     #[cfg(feature = "workflows")]
     pub workflow_storage: std::sync::OnceLock<taskito_workflows::WorkflowStorageBackend>,
+    /// Serializes the lazy init above: the constructor runs DDL migrations, and
+    /// concurrent `CREATE TABLE/INDEX IF NOT EXISTS` from separate sessions can
+    /// race in Postgres's catalog, failing one thread's first workflow call.
+    #[cfg(feature = "workflows")]
+    workflow_init: std::sync::Mutex<()>,
 }
 
 #[cfg(feature = "workflows")]
@@ -31,8 +36,16 @@ impl QueueHandle {
         if let Some(wf) = self.workflow_storage.get() {
             return Ok(wf.clone());
         }
+        // A poisoned lock only means another thread panicked mid-init; the
+        // OnceLock still tells the truth, so recover and retry the init.
+        let _init = self
+            .workflow_init
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner());
+        if let Some(wf) = self.workflow_storage.get() {
+            return Ok(wf.clone()); // another thread won the race while we waited
+        }
         let wf = build_workflow_storage(&self.storage)?;
-        // A racing thread's handle wraps the same pool, so either is fine.
         let _ = self.workflow_storage.set(wf.clone());
         Ok(wf)
     }
@@ -110,5 +123,7 @@ pub fn open(options: OpenOptions) -> Result<QueueHandle, BindingError> {
         namespace: options.namespace,
         #[cfg(feature = "workflows")]
         workflow_storage: std::sync::OnceLock::new(),
+        #[cfg(feature = "workflows")]
+        workflow_init: std::sync::Mutex::new(()),
     })
 }

From 07517f119f89daaa87034f06a83065bdf69d9de3 Mon Sep 17 00:00:00 2001
From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com>
Date: Sun, 5 Jul 2026 09:43:35 +0530
Subject: [PATCH 5/9] fix(java): dedup unique keys in batch enqueue

enqueue_batch is a raw insert, so an active duplicate uniqueKey
violated the partial unique index and rolled back the whole batch.
---
 crates/taskito-java/src/backend.rs            | 82 ++++++++++++++++++-
 crates/taskito-java/src/error.rs              |  9 ++
 crates/taskito-java/src/ffi_c.rs              |  6 +-
 crates/taskito-java/src/queue/mod.rs          | 10 ++-
 .../java/org/byteveda/taskito/QueueTest.java  | 28 +++++++
 5 files changed, 126 insertions(+), 9 deletions(-)

diff --git a/crates/taskito-java/src/backend.rs b/crates/taskito-java/src/backend.rs
index b40addc2..955927e2 100644
--- a/crates/taskito-java/src/backend.rs
+++ b/crates/taskito-java/src/backend.rs
@@ -1,6 +1,6 @@
 //! Construct a [`QueueHandle`] from caller options (mirrors the Node shell).
 
-use taskito_core::{SqliteStorage, StorageBackend};
+use taskito_core::{NewJob, SqliteStorage, Storage, StorageBackend};
 
 use crate::convert::OpenOptions;
 use crate::error::BindingError;
@@ -73,6 +73,40 @@ fn build_workflow_storage(
     Ok(wf)
 }
 
+/// Enqueue a batch, routing any job with a `unique_key` through the dedup path
+/// (matching single `enqueue`) instead of the raw batch insert. The raw insert
+/// would hit the partial unique index on an active duplicate and roll back the
+/// whole batch; here a duplicate resolves to the existing job's id, and only
+/// keyless jobs share the one-transaction fast path. Ids come back in input order.
+pub fn enqueue_batch_dedup(
+    storage: &StorageBackend,
+    jobs: Vec<NewJob>,
+) -> Result<Vec<String>, BindingError> {
+    let mut ids: Vec<Option<String>> = std::iter::repeat_with(|| None).take(jobs.len()).collect();
+    let mut plain_jobs = Vec::new();
+    let mut plain_slots = Vec::new();
+    for (index, job) in jobs.into_iter().enumerate() {
+        if job.unique_key.is_some() {
+            ids[index] = Some(storage.enqueue_unique(job)?.id);
+        } else {
+            plain_jobs.push(job);
+            plain_slots.push(index);
+        }
+    }
+    if !plain_jobs.is_empty() {
+        let created = storage.enqueue_batch(plain_jobs)?;
+        if created.len() != plain_slots.len() {
+            return Err(BindingError::new(
+                "storage returned a different number of jobs than were batch-enqueued",
+            ));
+        }
+        for (slot, job) in plain_slots.into_iter().zip(created) {
+            ids[slot] = Some(job.id);
+        }
+    }
+    Ok(ids.into_iter().flatten().collect())
+}
+
 /// Error for a backend that is unknown or whose cargo feature is not compiled in.
 fn unknown_backend(name: &str) -> BindingError {
     BindingError::new(format!(
@@ -127,3 +161,49 @@ pub fn open(options: OpenOptions) -> Result<QueueHandle, BindingError> {
         workflow_init: std::sync::Mutex::new(()),
     })
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn new_job(unique_key: Option<&str>) -> NewJob {
+        NewJob {
+            queue: "default".into(),
+            task_name: "task".into(),
+            payload: Vec::new(),
+            priority: 0,
+            scheduled_at: 0,
+            max_retries: 0,
+            timeout_ms: 0,
+            unique_key: unique_key.map(str::to_owned),
+            metadata: None,
+            notes: None,
+            depends_on: Vec::new(),
+            expires_at: None,
+            result_ttl_ms: None,
+            namespace: None,
+        }
+    }
+
+    /// A batch containing an active duplicate `unique_key` must dedup that job
+    /// to the existing id instead of failing the whole batch on the unique index.
+    #[test]
+    fn batch_dedup_resolves_duplicate_unique_keys() {
+        let storage = StorageBackend::Sqlite(SqliteStorage::in_memory().expect("storage"));
+        let existing = storage
+            .enqueue_unique(new_job(Some("k1")))
+            .expect("seed enqueue");
+        let ids = enqueue_batch_dedup(
+            &storage,
+            vec![new_job(Some("k1")), new_job(Some("k2")), new_job(None)],
+        )
+        .expect("batch enqueue");
+        assert_eq!(ids.len(), 3);
+        assert_eq!(
+            ids[0], existing.id,
+            "duplicate key resolves to existing job"
+        );
+        assert_ne!(ids[1], ids[0]);
+        assert_ne!(ids[2], ids[0]);
+    }
+}
diff --git a/crates/taskito-java/src/error.rs b/crates/taskito-java/src/error.rs
index 715b950f..e21872cb 100644
--- a/crates/taskito-java/src/error.rs
+++ b/crates/taskito-java/src/error.rs
@@ -8,6 +8,7 @@ pub const TASKITO_EXCEPTION: &str = "org/byteveda/taskito/TaskitoException";
 
 /// A deferred Java exception. The binding builds one of these on failure and
 /// throws it at the FFI boundary instead of unwinding a Rust panic across it.
+#[derive(Debug)]
 pub struct BindingError {
     class: &'static str,
     message: String,
@@ -29,6 +30,14 @@ impl BindingError {
     }
 }
 
+/// The message alone: the C-ABI surface (`ffi_c`) reports errors as strings,
+/// with no exception class to carry.
+impl std::fmt::Display for BindingError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.write_str(&self.message)
+    }
+}
+
 impl From<QueueError> for BindingError {
     fn from(err: QueueError) -> Self {
         Self::new(err.to_string())
diff --git a/crates/taskito-java/src/ffi_c.rs b/crates/taskito-java/src/ffi_c.rs
index ac2463bf..11c5cbd2 100644
--- a/crates/taskito-java/src/ffi_c.rs
+++ b/crates/taskito-java/src/ffi_c.rs
@@ -190,11 +190,9 @@ pub unsafe extern "C" fn taskito_ffi_enqueue_many(
             .zip(option_list)
             .map(|(payload, options)| build_new_job(task.clone(), payload, options, namespace))
             .collect();
-        let created = queue
-            .storage
-            .enqueue_batch(new_jobs)
+        let ids = crate::backend::enqueue_batch_dedup(&queue.storage, new_jobs)
             .map_err(|e| e.to_string())?;
-        let ids: Vec<Vec<u8>> = created.into_iter().map(|job| job.id.into_bytes()).collect();
+        let ids: Vec<Vec<u8>> = ids.into_iter().map(String::into_bytes).collect();
         Ok::<Vec<u8>, String>(frame(&ids))
     });
     finish(catch_unwind(work), out_data, out_len)
diff --git a/crates/taskito-java/src/queue/mod.rs b/crates/taskito-java/src/queue/mod.rs
index f162b0bc..c5fc0a70 100644
--- a/crates/taskito-java/src/queue/mod.rs
+++ b/crates/taskito-java/src/queue/mod.rs
@@ -93,8 +93,11 @@ pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_enqueue<'l
 }
 
 /// `String[] enqueueMany(long handle, String taskName, byte[][] payloads, String optionsJson)`
-/// — enqueue a batch in one storage transaction. `optionsJson` is a JSON array
-/// of per-job options, the same length as `payloads`. Returns the new job ids.
+/// — enqueue a batch. `optionsJson` is a JSON array of per-job options, the same
+/// length as `payloads`. Keyless jobs insert in one storage transaction; a job
+/// with a `uniqueKey` takes the dedup path instead (an active duplicate resolves
+/// to the existing job's id, matching single `enqueue`). Returns the job ids in
+/// input order.
 #[no_mangle]
 pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_enqueueMany<'local>(
     mut env: JNIEnv<'local>,
@@ -121,8 +124,7 @@ pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_enqueueMan
             .zip(option_list)
             .map(|(payload, options)| build_new_job(task.clone(), payload, options, namespace))
             .collect();
-        let created = queue.storage.enqueue_batch(new_jobs)?;
-        let ids: Vec<String> = created.into_iter().map(|job| job.id).collect();
+        let ids = backend::enqueue_batch_dedup(&queue.storage, new_jobs)?;
         new_string_array(env, &ids)
     })
 }
diff --git a/sdks/java/src/test/java/org/byteveda/taskito/QueueTest.java b/sdks/java/src/test/java/org/byteveda/taskito/QueueTest.java
index 410f8a3d..07242505 100644
--- a/sdks/java/src/test/java/org/byteveda/taskito/QueueTest.java
+++ b/sdks/java/src/test/java/org/byteveda/taskito/QueueTest.java
@@ -3,6 +3,7 @@
 import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 import java.nio.file.Path;
@@ -96,6 +97,33 @@ void enqueueManyAndFilteredList(@TempDir Path dir) {
         }
     }
 
+    @Test
+    void enqueueManyDedupesActiveUniqueKeys(@TempDir Path dir) {
+        try (Taskito queue = open(dir)) {
+            String first = queue.enqueue(
+                    SEND_EMAIL,
+                    Collections.singletonMap("to", "a"),
+                    EnqueueOptions.builder().uniqueKey("k1").build());
+            // The batch's duplicate key must resolve to the existing job instead
+            // of rolling back the whole batch on the unique index.
+            List<String> ids = queue.enqueueMany(
+                    SEND_EMAIL,
+                    Arrays.asList(Collections.singletonMap("to", "a"), Collections.singletonMap("to", "b")),
+                    EnqueueOptions.builder().uniqueKey("k1").build());
+            assertEquals(2, ids.size());
+            assertEquals(first, ids.get(0));
+            assertEquals(first, ids.get(1));
+            assertEquals(1, queue.stats().pending);
+        }
+    }
+
+    @Test
+    void callsAfterCloseThrowInsteadOfTouchingFreedHandle(@TempDir Path dir) {
+        Taskito queue = open(dir);
+        queue.close();
+        assertThrows(IllegalStateException.class, queue::stats);
+    }
+
     @Test
     void settingsRoundTrip(@TempDir Path dir) {
         try (Taskito queue = open(dir)) {

From 8d47bd6d7ddaa41098e1f20c3c6c72df5dc4f62c Mon Sep 17 00:00:00 2001
From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com>
Date: Sun, 5 Jul 2026 09:43:48 +0530
Subject: [PATCH 6/9] feat(java): add common leveled logger

Mirrors the Node SDK logger: stderr sink, TASKITO_LOG_LEVEL
threshold, namespaced tags, pluggable sink for tests.
---
 .../org/byteveda/taskito/events/Emitter.java  |  13 +--
 .../byteveda/taskito/logging/LogLevel.java    |  35 ++++++
 .../org/byteveda/taskito/logging/LogSink.java |   7 ++
 .../taskito/logging/TaskitoLogger.java        | 109 ++++++++++++++++++
 .../taskito/logging/package-info.java         |   6 +
 .../org/byteveda/taskito/worker/Worker.java   |  15 +--
 .../taskito/workflows/WorkflowTracker.java    |  29 ++---
 .../byteveda/taskito/TaskitoLoggerTest.java   |  56 +++++++++
 8 files changed, 233 insertions(+), 37 deletions(-)
 create mode 100644 sdks/java/src/main/java/org/byteveda/taskito/logging/LogLevel.java
 create mode 100644 sdks/java/src/main/java/org/byteveda/taskito/logging/LogSink.java
 create mode 100644 sdks/java/src/main/java/org/byteveda/taskito/logging/TaskitoLogger.java
 create mode 100644 sdks/java/src/main/java/org/byteveda/taskito/logging/package-info.java
 create mode 100644 sdks/java/src/test/java/org/byteveda/taskito/TaskitoLoggerTest.java

diff --git a/sdks/java/src/main/java/org/byteveda/taskito/events/Emitter.java b/sdks/java/src/main/java/org/byteveda/taskito/events/Emitter.java
index d3838e95..adad46c7 100644
--- a/sdks/java/src/main/java/org/byteveda/taskito/events/Emitter.java
+++ b/sdks/java/src/main/java/org/byteveda/taskito/events/Emitter.java
@@ -5,10 +5,11 @@
 import java.util.Map;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.function.Consumer;
+import org.byteveda.taskito.logging.TaskitoLogger;
 
 /** Dispatches {@link OutcomeEvent}s to registered listeners. Thread-safe. */
 public final class Emitter {
-    private static final System.Logger LOG = System.getLogger(Emitter.class.getName());
+    private static final TaskitoLogger LOG = TaskitoLogger.create("events");
 
     private final Map<EventName, List<Consumer<OutcomeEvent>>> listeners = new EnumMap<>(EventName.class);
 
@@ -26,13 +27,9 @@ public void emit(OutcomeEvent event) {
             try {
                 listener.accept(event);
             } catch (RuntimeException e) {
-                // A listener fault must not break event dispatch — but it must
-                // not vanish either: this is the only place a workflow-tracker
-                // failure would otherwise surface.
-                LOG.log(
-                        System.Logger.Level.WARNING,
-                        "listener for " + event.name + " (job " + event.jobId + ") threw",
-                        e);
+                // A listener fault must not break dispatch — but log it: this is
+                // the only place a workflow-tracker failure would surface.
+                LOG.warn("listener for " + event.name + " (job " + event.jobId + ") threw", e);
             }
         }
     }
diff --git a/sdks/java/src/main/java/org/byteveda/taskito/logging/LogLevel.java b/sdks/java/src/main/java/org/byteveda/taskito/logging/LogLevel.java
new file mode 100644
index 00000000..abd0b8e0
--- /dev/null
+++ b/sdks/java/src/main/java/org/byteveda/taskito/logging/LogLevel.java
@@ -0,0 +1,35 @@
+package org.byteveda.taskito.logging;
+
+import java.util.Locale;
+
+/** Severity, from most to least verbose. {@link #SILENT} disables all output. */
+public enum LogLevel {
+    DEBUG(10),
+    INFO(20),
+    WARN(30),
+    ERROR(40),
+    SILENT(Integer.MAX_VALUE);
+
+    private final int severity;
+
+    LogLevel(int severity) {
+        this.severity = severity;
+    }
+
+    /** Whether a message at this level clears the {@code threshold}. */
+    boolean passes(LogLevel threshold) {
+        return severity >= threshold.severity;
+    }
+
+    /** Case-insensitive parse; {@code null} when {@code raw} is not a level name. */
+    static LogLevel parseOrNull(String raw) {
+        if (raw == null) {
+            return null;
+        }
+        try {
+            return valueOf(raw.trim().toUpperCase(Locale.ROOT));
+        } catch (IllegalArgumentException e) {
+            return null;
+        }
+    }
+}
diff --git a/sdks/java/src/main/java/org/byteveda/taskito/logging/LogSink.java b/sdks/java/src/main/java/org/byteveda/taskito/logging/LogSink.java
new file mode 100644
index 00000000..a18f0fe3
--- /dev/null
+++ b/sdks/java/src/main/java/org/byteveda/taskito/logging/LogSink.java
@@ -0,0 +1,7 @@
+package org.byteveda.taskito.logging;
+
+/** Receives every formatted line that clears the level threshold. */
+@FunctionalInterface
+public interface LogSink {
+    void accept(LogLevel level, String line);
+}
diff --git a/sdks/java/src/main/java/org/byteveda/taskito/logging/TaskitoLogger.java b/sdks/java/src/main/java/org/byteveda/taskito/logging/TaskitoLogger.java
new file mode 100644
index 00000000..2680d7d4
--- /dev/null
+++ b/sdks/java/src/main/java/org/byteveda/taskito/logging/TaskitoLogger.java
@@ -0,0 +1,109 @@
+package org.byteveda.taskito.logging;
+
+import java.io.PrintWriter;
+import java.io.StringWriter;
+import java.time.Instant;
+import java.util.Objects;
+import java.util.function.Supplier;
+
+/**
+ * A namespaced leveled logger ({@code [taskito:worker]}). Obtain via
+ * {@link #create}; the level threshold and sink are global, so
+ * {@link #setLevel}/{@link #setSink} take effect everywhere immediately.
+ * Messages below the threshold are dropped (suppliers are never invoked).
+ */
+public final class TaskitoLogger {
+    private static volatile LogLevel level = envLevel();
+    private static volatile LogSink sink = TaskitoLogger::writeToStderr;
+
+    private final String tag;
+
+    private TaskitoLogger(String namespace) {
+        this.tag = namespace == null ? "taskito" : "taskito:" + namespace;
+    }
+
+    /** The root {@code [taskito]} logger. Prefer {@link #create} for a namespaced one. */
+    public static TaskitoLogger root() {
+        return new TaskitoLogger(null);
+    }
+
+    /** A namespaced logger: {@code create("worker")} tags lines {@code [taskito:worker]}. */
+    public static TaskitoLogger create(String namespace) {
+        Objects.requireNonNull(namespace, "namespace");
+        return new TaskitoLogger(namespace);
+    }
+
+    /** Set the global threshold; messages below it are dropped (and never built). */
+    public static void setLevel(LogLevel newLevel) {
+        level = Objects.requireNonNull(newLevel, "level");
+    }
+
+    /** Replace the output sink (default: stderr). Useful for capture in tests. */
+    public static void setSink(LogSink newSink) {
+        sink = Objects.requireNonNull(newSink, "sink");
+    }
+
+    public void debug(String message) {
+        emit(LogLevel.DEBUG, message, null);
+    }
+
+    public void info(String message) {
+        emit(LogLevel.INFO, message, null);
+    }
+
+    public void warn(String message) {
+        emit(LogLevel.WARN, message, null);
+    }
+
+    public void warn(String message, Throwable cause) {
+        emit(LogLevel.WARN, message, cause);
+    }
+
+    public void error(String message) {
+        emit(LogLevel.ERROR, message, null);
+    }
+
+    public void error(String message, Throwable cause) {
+        emit(LogLevel.ERROR, message, cause);
+    }
+
+    /** Lazy variant: {@code message} is built only when the level passes. */
+    public void debug(Supplier<String> message) {
+        if (LogLevel.DEBUG.passes(level)) {
+            emit(LogLevel.DEBUG, message.get(), null);
+        }
+    }
+
+    private void emit(LogLevel messageLevel, String message, Throwable cause) {
+        if (!messageLevel.passes(level)) {
+            return;
+        }
+        StringBuilder line = new StringBuilder()
+                .append(Instant.now())
+                .append(" [")
+                .append(tag)
+                .append("] ")
+                .append(messageLevel)
+                .append(' ')
+                .append(message);
+        if (cause != null) {
+            line.append(System.lineSeparator()).append(stackTraceOf(cause));
+        }
+        sink.accept(messageLevel, line.toString());
+    }
+
+    private static String stackTraceOf(Throwable cause) {
+        StringWriter buffer = new StringWriter();
+        cause.printStackTrace(new PrintWriter(buffer));
+        return buffer.toString().stripTrailing();
+    }
+
+    private static LogLevel envLevel() {
+        LogLevel parsed = LogLevel.parseOrNull(System.getenv("TASKITO_LOG_LEVEL"));
+        return parsed == null ? LogLevel.WARN : parsed;
+    }
+
+    private static void writeToStderr(LogLevel ignored, String line) {
+        System.err.println(line);
+    }
+}
diff --git a/sdks/java/src/main/java/org/byteveda/taskito/logging/package-info.java b/sdks/java/src/main/java/org/byteveda/taskito/logging/package-info.java
new file mode 100644
index 00000000..efe3b037
--- /dev/null
+++ b/sdks/java/src/main/java/org/byteveda/taskito/logging/package-info.java
@@ -0,0 +1,6 @@
+/**
+ * The SDK's zero-dependency leveled logger. Writes to stderr by default (stdout
+ * stays clean for CLI/piped output); threshold from {@code TASKITO_LOG_LEVEL},
+ * overridable via {@link org.byteveda.taskito.logging.TaskitoLogger#setLevel}.
+ */
+package org.byteveda.taskito.logging;
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 729a0780..4333c97e 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
@@ -25,6 +25,7 @@
 import org.byteveda.taskito.events.Emitter;
 import org.byteveda.taskito.events.EventName;
 import org.byteveda.taskito.events.OutcomeEvent;
+import org.byteveda.taskito.logging.TaskitoLogger;
 import org.byteveda.taskito.middleware.Middleware;
 import org.byteveda.taskito.resources.ResourceRuntime;
 import org.byteveda.taskito.serialization.PayloadCodec;
@@ -43,7 +44,7 @@
  * jobs.
  */
 public final class Worker implements AutoCloseable {
-    private static final System.Logger LOG = System.getLogger(Worker.class.getName());
+    private static final TaskitoLogger LOG = TaskitoLogger.create("worker");
     private static final long SHUTDOWN_TIMEOUT_SECONDS = 30;
     private static final ObjectMapper MESH_JSON = new ObjectMapper();
 
@@ -145,16 +146,12 @@ public synchronized void close() {
         executor.shutdown(); // stop accepting; let running handlers finish
         try {
             if (!executor.awaitTermination(SHUTDOWN_TIMEOUT_SECONDS, TimeUnit.SECONDS)) {
-                // Interrupt stragglers, then give them one more window: shutdownNow()
-                // only *requests* interruption, and a handler stuck in
-                // non-interruptible work can outlive it.
+                // shutdownNow() only *requests* interruption; give stuck handlers
+                // one more window before closing the handle out from under them.
                 executor.shutdownNow();
                 if (!executor.awaitTermination(SHUTDOWN_TIMEOUT_SECONDS, TimeUnit.SECONDS)) {
-                    LOG.log(
-                            System.Logger.Level.WARNING,
-                            "handler threads still running after {0}s; closing the native worker anyway —"
-                                    + " their late completions will be rejected",
-                            2 * SHUTDOWN_TIMEOUT_SECONDS);
+                    LOG.warn("handler threads still running after " + (2 * SHUTDOWN_TIMEOUT_SECONDS)
+                            + "s; closing the native worker — their late completions will be rejected");
                 }
             }
         } catch (InterruptedException e) {
diff --git a/sdks/java/src/main/java/org/byteveda/taskito/workflows/WorkflowTracker.java b/sdks/java/src/main/java/org/byteveda/taskito/workflows/WorkflowTracker.java
index 29e20a3a..bf81d28b 100644
--- a/sdks/java/src/main/java/org/byteveda/taskito/workflows/WorkflowTracker.java
+++ b/sdks/java/src/main/java/org/byteveda/taskito/workflows/WorkflowTracker.java
@@ -23,6 +23,7 @@
 import org.byteveda.taskito.errors.SerializationException;
 import org.byteveda.taskito.errors.WorkflowException;
 import org.byteveda.taskito.events.OutcomeEvent;
+import org.byteveda.taskito.logging.TaskitoLogger;
 import org.byteveda.taskito.serialization.Serializer;
 import org.byteveda.taskito.spi.QueueBackend;
 
@@ -45,7 +46,7 @@
  * run reaches a terminal state.
  */
 public final class WorkflowTracker {
-    private static final System.Logger LOG = System.getLogger(WorkflowTracker.class.getName());
+    private static final TaskitoLogger LOG = TaskitoLogger.create("workflows");
 
     private final QueueBackend backend;
     private final Serializer serializer;
@@ -296,13 +297,10 @@ private void promoteOrSkip(String runId, PlanNode node, Map<String, NodeSnapshot
             }
             createDeferredJobFor(runId, node);
         } catch (RuntimeException e) {
-            // A failed promotion (native error, missing payload, bad metadata) must
-            // never leave the node wedged PENDING with no trace — outcome listeners
-            // swallow throws, so rethrowing here would silently stall the run.
-            LOG.log(
-                    System.Logger.Level.WARNING,
-                    "promoting workflow node '" + node.name + "' in run " + runId + " failed",
-                    e);
+            // A failed promotion must never leave the node wedged PENDING with no
+            // trace — outcome listeners swallow throws, so rethrowing would
+            // silently stall the run.
+            LOG.warn("promoting workflow node '" + node.name + "' in run " + runId + " failed", e);
             failNodeAndCascade(runId, node.name, promotionKey, plan, e);
         }
     }
@@ -318,10 +316,7 @@ private void failNodeAndCascade(
             // Even the failure record failed (e.g. storage down). Roll the dedupe
             // marker back so the next outcome can retry the promotion.
             promotedNodes.remove(promotionKey);
-            LOG.log(
-                    System.Logger.Level.ERROR,
-                    "failed to record failure of workflow node '" + nodeName + "' in run " + runId,
-                    e);
+            LOG.error("failed to record failure of workflow node '" + nodeName + "' in run " + runId, e);
         }
     }
 
@@ -330,10 +325,7 @@ private void failNodeAndCascade(
      * fail each fan-out node instead of leaving it PENDING forever, then cascade.
      */
     private void failFanOutsOf(String runId, String producer, RunPlan plan, RuntimeException cause) {
-        LOG.log(
-                System.Logger.Level.WARNING,
-                "expanding fan-out(s) after node '" + producer + "' in run " + runId + " failed",
-                cause);
+        LOG.warn("expanding fan-out(s) after node '" + producer + "' in run " + runId + " failed", cause);
         if (plan == null) {
             return;
         }
@@ -342,10 +334,7 @@ private void failFanOutsOf(String runId, String producer, RunPlan plan, RuntimeE
                 backend.failWorkflowNode(runId, fanOut.name, describe(cause));
                 promoteSuccessors(runId, fanOut.name, plan);
             } catch (RuntimeException e) {
-                LOG.log(
-                        System.Logger.Level.ERROR,
-                        "failed to record failure of fan-out node '" + fanOut.name + "' in run " + runId,
-                        e);
+                LOG.error("failed to record failure of fan-out node '" + fanOut.name + "' in run " + runId, e);
             }
         }
     }
diff --git a/sdks/java/src/test/java/org/byteveda/taskito/TaskitoLoggerTest.java b/sdks/java/src/test/java/org/byteveda/taskito/TaskitoLoggerTest.java
new file mode 100644
index 00000000..49cc34b2
--- /dev/null
+++ b/sdks/java/src/test/java/org/byteveda/taskito/TaskitoLoggerTest.java
@@ -0,0 +1,56 @@
+package org.byteveda.taskito;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.ArrayList;
+import java.util.List;
+import org.byteveda.taskito.logging.LogLevel;
+import org.byteveda.taskito.logging.TaskitoLogger;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+class TaskitoLoggerTest {
+    private final List<String> lines = new ArrayList<>();
+
+    @AfterEach
+    void restoreDefaults() {
+        TaskitoLogger.setLevel(LogLevel.WARN);
+        TaskitoLogger.setSink((level, line) -> System.err.println(line));
+    }
+
+    @Test
+    void dropsMessagesBelowThresholdAndTagsNamespace() {
+        TaskitoLogger.setSink((level, line) -> lines.add(line));
+        TaskitoLogger.setLevel(LogLevel.WARN);
+        TaskitoLogger log = TaskitoLogger.create("worker");
+
+        log.info("dropped");
+        log.warn("kept");
+
+        assertEquals(1, lines.size());
+        assertTrue(lines.get(0).contains("[taskito:worker] WARN kept"));
+    }
+
+    @Test
+    void appendsStackTraceForThrowables() {
+        TaskitoLogger.setSink((level, line) -> lines.add(line));
+        TaskitoLogger.setLevel(LogLevel.ERROR);
+
+        TaskitoLogger.root().error("boom", new IllegalStateException("cause detail"));
+
+        assertEquals(1, lines.size());
+        assertTrue(lines.get(0).contains("[taskito] ERROR boom"));
+        assertTrue(lines.get(0).contains("IllegalStateException: cause detail"));
+    }
+
+    @Test
+    void silentDisablesEverything() {
+        TaskitoLogger.setSink((level, line) -> lines.add(line));
+        TaskitoLogger.setLevel(LogLevel.SILENT);
+
+        TaskitoLogger.root().error("never emitted");
+
+        assertTrue(lines.isEmpty());
+    }
+}

From cc0590fab3c47db8cc29956953f7923b96cecea8 Mon Sep 17 00:00:00 2001
From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com>
Date: Sun, 5 Jul 2026 09:43:48 +0530
Subject: [PATCH 7/9] docs(java): tighten handle-guard javadoc

---
 .../byteveda/taskito/internal/JniQueueBackend.java    | 11 ++++-------
 .../byteveda/taskito/internal/JniWorkerControl.java   | 11 ++++-------
 2 files changed, 8 insertions(+), 14 deletions(-)

diff --git a/sdks/java/src/main/java/org/byteveda/taskito/internal/JniQueueBackend.java b/sdks/java/src/main/java/org/byteveda/taskito/internal/JniQueueBackend.java
index fd53d0c8..47647f5a 100644
--- a/sdks/java/src/main/java/org/byteveda/taskito/internal/JniQueueBackend.java
+++ b/sdks/java/src/main/java/org/byteveda/taskito/internal/JniQueueBackend.java
@@ -8,13 +8,10 @@
 import org.byteveda.taskito.spi.WorkerControl;
 
 /**
- * JNI-backed {@link QueueBackend} over a native queue handle.
- *
- * <p>The native contract forbids using a handle after {@code close()} or closing
- * concurrently with a call — violating it is a use-after-free in the native
- * layer. Every call therefore holds the read lock while touching the handle;
- * {@code close()} takes the write lock, so it waits out in-flight calls and any
- * later call fails with {@link IllegalStateException} instead of crashing the JVM.
+ * JNI-backed {@link QueueBackend} over a native queue handle. Every call holds
+ * the read lock; {@code close()} takes the write lock, so it waits out in-flight
+ * calls and later calls throw {@link IllegalStateException} instead of touching
+ * freed native memory.
  */
 public final class JniQueueBackend implements QueueBackend {
     private final long handle;
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 b9eebe7e..3dfd8103 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
@@ -6,13 +6,10 @@
 import org.byteveda.taskito.spi.WorkerControl;
 
 /**
- * JNI-backed {@link WorkerControl} over a native worker handle.
- *
- * <p>The native contract forbids using a handle after {@code close()} or closing
- * concurrently with a call — violating it is a use-after-free in the native
- * layer. Every call therefore holds the read lock while touching the handle;
- * {@code close()} takes the write lock, so it waits out in-flight calls and any
- * later call fails with {@link IllegalStateException} instead of crashing the JVM.
+ * JNI-backed {@link WorkerControl} over a native worker handle. Every call holds
+ * the read lock; {@code close()} takes the write lock, so it waits out in-flight
+ * calls and later calls throw {@link IllegalStateException} instead of touching
+ * freed native memory.
  */
 public final class JniWorkerControl implements WorkerControl {
     private final long handle;

From 4cf56006da502da7251d2d60b54524daf9b05929 Mon Sep 17 00:00:00 2001
From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com>
Date: Sun, 5 Jul 2026 10:12:01 +0530
Subject: [PATCH 8/9] fix(java): scope fan-out failures per branch

A later branch's expansion error blanket-failed siblings whose
children were already running; storage failure during the failure
record left a stale promotion marker.
---
 .../taskito/workflows/WorkflowTracker.java    | 98 +++++++++++--------
 1 file changed, 57 insertions(+), 41 deletions(-)

diff --git a/sdks/java/src/main/java/org/byteveda/taskito/workflows/WorkflowTracker.java b/sdks/java/src/main/java/org/byteveda/taskito/workflows/WorkflowTracker.java
index bf81d28b..9c81c560 100644
--- a/sdks/java/src/main/java/org/byteveda/taskito/workflows/WorkflowTracker.java
+++ b/sdks/java/src/main/java/org/byteveda/taskito/workflows/WorkflowTracker.java
@@ -149,11 +149,7 @@ private void onOutcome(String jobId, boolean succeeded, String error) {
             return;
         }
         if (succeeded) {
-            try {
-                expandFanOutIfAny(ref.runId, ref.nodeName, jobId, plan);
-            } catch (RuntimeException e) {
-                failFanOutsOf(ref.runId, ref.nodeName, plan, e);
-            }
+            expandFanOutIfAny(ref.runId, ref.nodeName, jobId, plan);
         }
         // Evaluate every successor's condition — running on-failure recovery nodes
         // and skip-cascading the rest. Replaces a blanket fail-fast cascade.
@@ -161,38 +157,62 @@ private void onOutcome(String jobId, boolean succeeded, String error) {
         finalizeIfTerminal(ref.runId);
     }
 
-    /** If the just-completed node feeds fan-outs, split its result list into child jobs. */
+    /**
+     * If the just-completed node feeds fan-outs, split its result list into child
+     * jobs. Failures are scoped per branch: an unusable producer result fails
+     * every branch, but one branch's expansion error never touches a sibling
+     * whose children are already running.
+     */
     private void expandFanOutIfAny(String runId, String node, String jobId, RunPlan plan) {
         List<PlanNode> fanOuts = plan.successorsMatching(node, candidate -> candidate.fanOut != null);
         if (fanOuts.isEmpty()) {
             return;
         }
-        List<?> items = serializer.deserialize(fetchResult(jobId), List.class);
-        for (PlanNode fanOut : fanOuts) {
-            String[] names = new String[items.size()];
-            byte[][] payloads = new byte[items.size()][];
-            for (int i = 0; i < items.size(); i++) {
-                names[i] = fanOut.name + "[" + i + "]";
-                payloads[i] = serializer.serialize(items.get(i));
+        List<?> items;
+        try {
+            items = serializer.deserialize(fetchResult(jobId), List.class);
+        } catch (RuntimeException e) {
+            LOG.warn("reading fan-out producer result of node '" + node + "' in run " + runId + " failed", e);
+            for (PlanNode fanOut : fanOuts) {
+                failFanOut(runId, fanOut, plan, e);
             }
-            backend.expandFanOut(
-                    runId,
-                    fanOut.name,
-                    names,
-                    payloads,
-                    fanOut.taskName,
-                    fanOut.queueOrDefault(),
-                    fanOut.retries(),
-                    fanOut.timeout(),
-                    fanOut.priorityOrDefault());
-            // An empty producer list yields no child jobs, so no child outcome will
-            // ever drive the fan-in — collect it now with an empty result list.
-            if (items.isEmpty()) {
-                collectFanIns(runId, fanOut.name, List.of(), plan);
+            return;
+        }
+        for (PlanNode fanOut : fanOuts) {
+            try {
+                expandFanOut(runId, fanOut, items, plan);
+            } catch (RuntimeException e) {
+                LOG.warn("expanding fan-out node '" + fanOut.name + "' in run " + runId + " failed", e);
+                failFanOut(runId, fanOut, plan, e);
             }
         }
     }
 
+    /** Split the producer's items into one child job per element of {@code fanOut}. */
+    private void expandFanOut(String runId, PlanNode fanOut, List<?> items, RunPlan plan) {
+        String[] names = new String[items.size()];
+        byte[][] payloads = new byte[items.size()][];
+        for (int i = 0; i < items.size(); i++) {
+            names[i] = fanOut.name + "[" + i + "]";
+            payloads[i] = serializer.serialize(items.get(i));
+        }
+        backend.expandFanOut(
+                runId,
+                fanOut.name,
+                names,
+                payloads,
+                fanOut.taskName,
+                fanOut.queueOrDefault(),
+                fanOut.retries(),
+                fanOut.timeout(),
+                fanOut.priorityOrDefault());
+        // An empty producer list yields no child jobs, so no child outcome will
+        // ever drive the fan-in — collect it now with an empty result list.
+        if (items.isEmpty()) {
+            collectFanIns(runId, fanOut.name, List.of(), plan);
+        }
+    }
+
     /** When every fan-out child has settled, collect their results into the fan-in job(s). */
     private void handleFanOutChild(String runId, String childName, RunPlan plan) {
         String parent = childName.substring(0, childName.indexOf('['));
@@ -321,21 +341,17 @@ private void failNodeAndCascade(
     }
 
     /**
-     * A fan-out expansion failed (undeserializable producer result, native error):
-     * fail each fan-out node instead of leaving it PENDING forever, then cascade.
+     * Fail one fan-out node (its expansion failed) and cascade. On a storage
+     * failure the promotion marker is rolled back so the node is not treated as
+     * decided while it is still PENDING.
      */
-    private void failFanOutsOf(String runId, String producer, RunPlan plan, RuntimeException cause) {
-        LOG.warn("expanding fan-out(s) after node '" + producer + "' in run " + runId + " failed", cause);
-        if (plan == null) {
-            return;
-        }
-        for (PlanNode fanOut : plan.successorsMatching(producer, candidate -> candidate.fanOut != null)) {
-            try {
-                backend.failWorkflowNode(runId, fanOut.name, describe(cause));
-                promoteSuccessors(runId, fanOut.name, plan);
-            } catch (RuntimeException e) {
-                LOG.error("failed to record failure of fan-out node '" + fanOut.name + "' in run " + runId, e);
-            }
+    private void failFanOut(String runId, PlanNode fanOut, RunPlan plan, RuntimeException cause) {
+        try {
+            backend.failWorkflowNode(runId, fanOut.name, describe(cause));
+            promoteSuccessors(runId, fanOut.name, plan);
+        } catch (RuntimeException e) {
+            promotedNodes.remove(new GateKey(runId, fanOut.name));
+            LOG.error("failed to record failure of fan-out node '" + fanOut.name + "' in run " + runId, e);
         }
     }
 

From 0a0bb26da5070eee3e1d7fe9b0128e02a1cfb06f Mon Sep 17 00:00:00 2001
From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com>
Date: Sun, 5 Jul 2026 10:31:59 +0530
Subject: [PATCH 9/9] fix(java): retry fan-out failure recording

Nothing re-drives a fan-out after its producer settles, so a storage
failure while recording the node failure now retries on the scheduler
instead of wedging the run.
---
 .../taskito/workflows/WorkflowTracker.java    | 30 ++++++++++++++++---
 1 file changed, 26 insertions(+), 4 deletions(-)

diff --git a/sdks/java/src/main/java/org/byteveda/taskito/workflows/WorkflowTracker.java b/sdks/java/src/main/java/org/byteveda/taskito/workflows/WorkflowTracker.java
index 9c81c560..a1aa6836 100644
--- a/sdks/java/src/main/java/org/byteveda/taskito/workflows/WorkflowTracker.java
+++ b/sdks/java/src/main/java/org/byteveda/taskito/workflows/WorkflowTracker.java
@@ -47,6 +47,8 @@
  */
 public final class WorkflowTracker {
     private static final TaskitoLogger LOG = TaskitoLogger.create("workflows");
+    private static final int FAIL_RECORD_MAX_ATTEMPTS = 5;
+    private static final long FAIL_RECORD_RETRY_MS = 1_000;
 
     private final QueueBackend backend;
     private final Serializer serializer;
@@ -341,17 +343,37 @@ private void failNodeAndCascade(
     }
 
     /**
-     * Fail one fan-out node (its expansion failed) and cascade. On a storage
-     * failure the promotion marker is rolled back so the node is not treated as
-     * decided while it is still PENDING.
+     * Fail one fan-out node (its expansion failed) and cascade. Unlike deferred
+     * nodes, nothing re-drives a fan-out after its producer settles, so a storage
+     * failure while recording the node failure is retried on the scheduler
+     * (bounded) instead of relying on a later outcome.
      */
     private void failFanOut(String runId, PlanNode fanOut, RunPlan plan, RuntimeException cause) {
+        failFanOut(runId, fanOut, plan, cause, 1);
+    }
+
+    private void failFanOut(String runId, PlanNode fanOut, RunPlan plan, RuntimeException cause, int attempt) {
         try {
             backend.failWorkflowNode(runId, fanOut.name, describe(cause));
             promoteSuccessors(runId, fanOut.name, plan);
+            finalizeIfTerminal(runId);
         } catch (RuntimeException e) {
             promotedNodes.remove(new GateKey(runId, fanOut.name));
-            LOG.error("failed to record failure of fan-out node '" + fanOut.name + "' in run " + runId, e);
+            if (attempt >= FAIL_RECORD_MAX_ATTEMPTS) {
+                LOG.error(
+                        "failed to record failure of fan-out node '" + fanOut.name + "' in run " + runId + " after "
+                                + attempt + " attempts; run cannot finalize",
+                        e);
+                return;
+            }
+            LOG.warn(
+                    "recording failure of fan-out node '" + fanOut.name + "' in run " + runId + " failed (attempt "
+                            + attempt + "); retrying",
+                    e);
+            gateScheduler.schedule(
+                    () -> failFanOut(runId, fanOut, plan, cause, attempt + 1),
+                    FAIL_RECORD_RETRY_MS,
+                    TimeUnit.MILLISECONDS);
         }
     }