fix(java): harden outcome dispatch and webhook delivery

sdks/java/src/main/java/org/byteveda/taskito/webhooks/Deliverer.java

@@ -6,18 +6,27 @@
 import java.net.http.HttpResponse;
 import java.nio.charset.StandardCharsets;
 import java.time.Duration;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
 import javax.crypto.Mac;
 import javax.crypto.spec.SecretKeySpec;
 import org.byteveda.taskito.errors.WebhookException;
+import org.byteveda.taskito.logging.TaskitoLogger;
 
 /** Delivers webhook payloads over HTTP with an optional HMAC signature and retries. */
 final class Deliverer {
+    private static final TaskitoLogger LOG = TaskitoLogger.create("webhooks");
     private static final String ALGORITHM = "HmacSHA256";
     private static final char[] HEX = "0123456789abcdef".toCharArray();
+    // Exponential backoff matching the Node SDK: 500ms, 1s, 2s, ... capped at 30s.
+    private static final long BASE_BACKOFF_MS = 500;
+    private static final long MAX_BACKOFF_MS = 30_000;
+    // 500ms << 6 = 32s > cap, so larger shifts can never change the delay.
+    private static final int MAX_BACKOFF_SHIFT = 6;
 
     private final HttpClient client = HttpClient.newHttpClient();
 
-    /** Send {@code body} to the hook (non-blocking). Retries server errors. */
+    /** Send {@code body} to the hook (non-blocking). Retries server errors with backoff. */
     void deliver(Webhook hook, byte[] body) {
         HttpRequest.Builder request = HttpRequest.newBuilder(URI.create(hook.url))
                 .timeout(Duration.ofMillis(hook.timeoutMs))
@@ -27,18 +36,37 @@ void deliver(Webhook hook, byte[] body) {
             request.header("X-Taskito-Signature", "sha256=" + sign(hook.secret, body));
         }
         hook.headers.forEach(request::header);
-        sendWithRetry(request.build(), hook.maxRetries);
+        sendWithRetry(request.build(), 0, hook.maxRetries);
     }
 
-    private void sendWithRetry(HttpRequest request, int attemptsLeft) {
+    private void sendWithRetry(HttpRequest request, int attempt, int maxRetries) {
         client.sendAsync(request, HttpResponse.BodyHandlers.discarding()).whenComplete((response, error) -> {
             boolean failed = error != null || response.statusCode() >= 500;
-            if (failed && attemptsLeft > 0) {
-                sendWithRetry(request, attemptsLeft - 1);
+            if (!failed) {
+                return;
             }
+            if (attempt >= maxRetries) {
+                LOG.warn("webhook delivery to " + redact(request.uri()) + " failed after " + (attempt + 1)
+                        + " attempt(s)" + (error != null ? ": " + error : ": HTTP " + response.statusCode()));
+                return;
+            }
+            // Clamp the exponent before shifting: an unbounded shift overflows and
+            // would schedule negative delays for very large maxRetries.
+            long delayMs = Math.min(MAX_BACKOFF_MS, BASE_BACKOFF_MS << Math.min(attempt, MAX_BACKOFF_SHIFT));
+            CompletableFuture.delayedExecutor(delayMs, TimeUnit.MILLISECONDS)
+                    .execute(() -> sendWithRetry(request, attempt + 1, maxRetries));
         });
     }
 
+    /** Origin only (scheme://host:port) — path segments and queries can carry tokens. */
+    private static String redact(URI uri) {
+        try {
+            return new URI(uri.getScheme(), null, uri.getHost(), uri.getPort(), null, null, null).toString();
+        } catch (Exception e) {
+            return "<invalid-url>";
+        }
+    }
+
     static String sign(String secret, byte[] body) {
         try {
             Mac mac = Mac.getInstance(ALGORITHM);

sdks/java/src/main/java/org/byteveda/taskito/webhooks/WebhookManager.java

@@ -11,6 +11,7 @@
 import org.byteveda.taskito.Taskito;
 import org.byteveda.taskito.errors.WebhookException;
 import org.byteveda.taskito.events.OutcomeEvent;
+import org.byteveda.taskito.logging.TaskitoLogger;
 import org.byteveda.taskito.middleware.Middleware;
 
 /**
@@ -19,10 +20,17 @@
  * delivered automatically. Persisted via the queue's settings store.
  */
 public final class WebhookManager implements Middleware {
+    private static final TaskitoLogger LOG = TaskitoLogger.create("webhooks");
     private static final ObjectMapper JSON = new ObjectMapper();
+    // Dispatch runs on the worker's single outcome-drain thread for every job, so
+    // it must not re-read + re-parse the settings blob per outcome. Local
+    // create/delete invalidate immediately; the TTL bounds staleness from writers
+    // in other processes.
+    private static final long CACHE_TTL_MS = 30_000;
 
     private final WebhookStore store;
     private final Deliverer deliverer = new Deliverer();
+    private volatile CachedHooks cached;
 
     private WebhookManager(Taskito queue) {
         this.store = new WebhookStore(queue);
@@ -53,6 +61,7 @@ public synchronized Webhook create(Webhook.Builder spec) {
         List<Webhook> all = store.load();
         all.add(hook);
         store.save(all);
+        cached = null;
         return hook;
     }
 
@@ -69,6 +78,7 @@ public synchronized boolean delete(String id) {
         boolean removed = all.removeIf(hook -> hook.id.equals(id));
         if (removed) {
             store.save(all);
+            cached = null;
         }
         return removed;
     }
@@ -94,15 +104,39 @@ public void onCancel(OutcomeEvent event) {
     }
 
     private void dispatch(OutcomeEvent event) {
+        List<Webhook> hooks = activeHooks();
+        if (hooks.isEmpty()) {
+            return;
+        }
         String wire = event.name.name().toLowerCase(Locale.ROOT);
         byte[] body = payload(event, wire);
-        for (Webhook hook : store.load()) {
+        for (Webhook hook : hooks) {
             if (hook.enabled && hook.events.contains(wire) && matches(hook.taskFilter, event.taskName)) {
-                deliverer.deliver(hook, body);
+                try {
+                    deliverer.deliver(hook, body);
+                } catch (RuntimeException e) {
+                    // A bad hook (e.g. malformed URL) must not block the rest. Log the
+                    // class only — URI parse messages can echo the URL's tokens.
+                    LOG.warn("webhook " + hook.id + " delivery failed: "
+                            + e.getClass().getSimpleName());
+                }
             }
         }
     }
 
+    private List<Webhook> activeHooks() {
+        CachedHooks snapshot = cached;
+        long now = System.currentTimeMillis();
+        if (snapshot == null || now - snapshot.loadedAt() > CACHE_TTL_MS) {
+            snapshot = new CachedHooks(List.copyOf(store.load()), now);
+            cached = snapshot;
+        }
+        return snapshot.hooks();
+    }
+
+    /** An immutable webhook list plus when it was read from the store. */
+    private record CachedHooks(List<Webhook> hooks, long loadedAt) {}
+
     private static boolean matches(String filter, String taskName) {
         return filter == null || filter.equals(taskName);
     }

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

@@ -11,6 +11,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.JobInfo;
 import org.byteveda.taskito.middleware.Middleware;
 import org.byteveda.taskito.middleware.TaskContext;
@@ -30,6 +31,7 @@
  * and event listeners.
  */
 final class WorkerDispatchBridge implements WorkerBridge {
+    private static final TaskitoLogger LOG = TaskitoLogger.create("worker");
     private static final ObjectMapper JSON = new ObjectMapper();
     private static final TypeReference<Map<String, Object>> MAP = new TypeReference<Map<String, Object>>() {};
 
@@ -129,7 +131,12 @@ public void onOutcome(String kind, String jobId, String taskName, String error,
         OutcomeEvent event = new OutcomeEvent(name, jobId, taskName, error, retryCount, timedOut);
         emitter.emit(event);
         for (Middleware m : middleware) {
-            dispatch(m, name, event);
+            try {
+                dispatch(m, name, event);
+            } catch (RuntimeException e) {
+                // One faulty middleware must not starve the rest of this outcome.
+                LOG.warn("middleware " + m.getClass().getName() + " threw on " + name + " (job " + jobId + ")", e);
+            }
         }
     }
 

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

@@ -11,6 +11,8 @@
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import org.byteveda.taskito.events.EventName;
+import org.byteveda.taskito.events.OutcomeEvent;
+import org.byteveda.taskito.middleware.Middleware;
 import org.byteveda.taskito.task.Task;
 import org.byteveda.taskito.worker.Worker;
 import org.junit.jupiter.api.Test;
@@ -53,4 +55,32 @@ void runsTaskToCompletion(@TempDir Path dir) throws Exception {
             }
         }
     }
+
+    @Test
+    @Timeout(30)
+    void middlewareFaultDoesNotStarveOthers(@TempDir Path dir) throws Exception {
+        Task<Map<String, Object>> noop = Task.of("noop", mapType());
+        try (Taskito queue = Taskito.builder()
+                .backend("sqlite")
+                .url(dir.resolve("t.db").toString())
+                .open()) {
+            CountDownLatch secondSaw = new CountDownLatch(1);
+            queue.use(new Middleware() {
+                @Override
+                public void onCompleted(OutcomeEvent event) {
+                    throw new IllegalStateException("boom");
+                }
+            });
+            queue.use(new Middleware() {
+                @Override
+                public void onCompleted(OutcomeEvent event) {
+                    secondSaw.countDown();
+                }
+            });
+            queue.enqueue(noop, new HashMap<>());
+            try (Worker worker = queue.worker().handle(noop, p -> "ok").start()) {
+                assertTrue(secondSaw.await(20, TimeUnit.SECONDS), "second middleware must still see the outcome");
+            }
+        }
+    }
 }