From 52c15b7eb16db8878a114569252cec79fb2267f7 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Sun, 5 Jul 2026 18:05:53 +0530 Subject: [PATCH 1/5] fix(java): isolate outcome middleware faults --- .../taskito/worker/WorkerDispatchBridge.java | 9 +++++- .../java/org/byteveda/taskito/WorkerTest.java | 30 +++++++++++++++++++ 2 files changed, 38 insertions(+), 1 deletion(-) diff --git a/sdks/java/src/main/java/org/byteveda/taskito/worker/WorkerDispatchBridge.java b/sdks/java/src/main/java/org/byteveda/taskito/worker/WorkerDispatchBridge.java index a650f6fa..69175a35 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/worker/WorkerDispatchBridge.java +++ b/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 = new TypeReference>() {}; @@ -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); + } } } diff --git a/sdks/java/src/test/java/org/byteveda/taskito/WorkerTest.java b/sdks/java/src/test/java/org/byteveda/taskito/WorkerTest.java index ea963a75..831d734d 100644 --- a/sdks/java/src/test/java/org/byteveda/taskito/WorkerTest.java +++ b/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> 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"); + } + } + } } From 7341d7d571f6db4f39cc1d1087298f6776732f46 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Sun, 5 Jul 2026 18:05:53 +0530 Subject: [PATCH 2/5] fix(java): back off webhook delivery retries Zero-delay retries burst a down endpoint; now 500ms doubling to 30s cap, matching the Node SDK, with a warning on exhaustion. --- .../byteveda/taskito/webhooks/Deliverer.java | 34 ++++++++++++++++--- 1 file changed, 29 insertions(+), 5 deletions(-) diff --git a/sdks/java/src/main/java/org/byteveda/taskito/webhooks/Deliverer.java b/sdks/java/src/main/java/org/byteveda/taskito/webhooks/Deliverer.java index 342aa188..ef8c5938 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/webhooks/Deliverer.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/webhooks/Deliverer.java @@ -6,18 +6,25 @@ 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; 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 +34,35 @@ 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; + } + long delayMs = Math.min(MAX_BACKOFF_MS, BASE_BACKOFF_MS << attempt); + CompletableFuture.delayedExecutor(delayMs, TimeUnit.MILLISECONDS) + .execute(() -> sendWithRetry(request, attempt + 1, maxRetries)); }); } + /** Strip credentials and query string from a URI before logging it. */ + private static String redact(URI uri) { + try { + return new URI(uri.getScheme(), null, uri.getHost(), uri.getPort(), uri.getPath(), null, null).toString(); + } catch (Exception e) { + return ""; + } + } + static String sign(String secret, byte[] body) { try { Mac mac = Mac.getInstance(ALGORITHM); From 72dfb3b2cef62d20d6949a32640f002e1f761d8e Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Sun, 5 Jul 2026 18:06:25 +0530 Subject: [PATCH 3/5] fix(java): isolate webhook delivery faults per hook --- .../org/byteveda/taskito/webhooks/WebhookManager.java | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/sdks/java/src/main/java/org/byteveda/taskito/webhooks/WebhookManager.java b/sdks/java/src/main/java/org/byteveda/taskito/webhooks/WebhookManager.java index 6eee78b1..3331d3f1 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/webhooks/WebhookManager.java +++ b/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,6 +20,7 @@ * 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(); private final WebhookStore store; @@ -98,7 +100,12 @@ private void dispatch(OutcomeEvent event) { byte[] body = payload(event, wire); for (Webhook hook : store.load()) { 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.warn("webhook " + hook.id + " delivery failed", e); + } } } } From b9455c8b02e4f3d8c2c93d4af1149d8ed2b7a2b3 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Sun, 5 Jul 2026 18:06:25 +0530 Subject: [PATCH 4/5] perf(java): cache webhook list for outcome dispatch store.load() did a blocking native read + JSON parse per job outcome on the single drain thread, even with zero webhooks. --- .../taskito/webhooks/WebhookManager.java | 27 ++++++++++++++++++- 1 file changed, 26 insertions(+), 1 deletion(-) diff --git a/sdks/java/src/main/java/org/byteveda/taskito/webhooks/WebhookManager.java b/sdks/java/src/main/java/org/byteveda/taskito/webhooks/WebhookManager.java index 3331d3f1..3251daa7 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/webhooks/WebhookManager.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/webhooks/WebhookManager.java @@ -22,9 +22,15 @@ 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); @@ -55,6 +61,7 @@ public synchronized Webhook create(Webhook.Builder spec) { List all = store.load(); all.add(hook); store.save(all); + cached = null; return hook; } @@ -71,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; } @@ -96,9 +104,13 @@ public void onCancel(OutcomeEvent event) { } private void dispatch(OutcomeEvent event) { + List 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)) { try { deliverer.deliver(hook, body); @@ -110,6 +122,19 @@ private void dispatch(OutcomeEvent event) { } } + private List 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 hooks, long loadedAt) {} + private static boolean matches(String filter, String taskName) { return filter == null || filter.equals(taskName); } From 847314458dba63d17fcd63fac028313df73f6d75 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Sun, 5 Jul 2026 18:16:48 +0530 Subject: [PATCH 5/5] fix(java): clamp webhook backoff, tighten log redaction --- .../java/org/byteveda/taskito/webhooks/Deliverer.java | 10 +++++++--- .../org/byteveda/taskito/webhooks/WebhookManager.java | 6 ++++-- 2 files changed, 11 insertions(+), 5 deletions(-) diff --git a/sdks/java/src/main/java/org/byteveda/taskito/webhooks/Deliverer.java b/sdks/java/src/main/java/org/byteveda/taskito/webhooks/Deliverer.java index ef8c5938..3a8c7d15 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/webhooks/Deliverer.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/webhooks/Deliverer.java @@ -21,6 +21,8 @@ final class Deliverer { // 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(); @@ -48,16 +50,18 @@ private void sendWithRetry(HttpRequest request, int attempt, int maxRetries) { + " attempt(s)" + (error != null ? ": " + error : ": HTTP " + response.statusCode())); return; } - long delayMs = Math.min(MAX_BACKOFF_MS, BASE_BACKOFF_MS << attempt); + // 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)); }); } - /** Strip credentials and query string from a URI before logging it. */ + /** 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(), uri.getPath(), null, null).toString(); + return new URI(uri.getScheme(), null, uri.getHost(), uri.getPort(), null, null, null).toString(); } catch (Exception e) { return ""; } diff --git a/sdks/java/src/main/java/org/byteveda/taskito/webhooks/WebhookManager.java b/sdks/java/src/main/java/org/byteveda/taskito/webhooks/WebhookManager.java index 3251daa7..08ea8162 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/webhooks/WebhookManager.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/webhooks/WebhookManager.java @@ -115,8 +115,10 @@ private void dispatch(OutcomeEvent event) { try { deliverer.deliver(hook, body); } catch (RuntimeException e) { - // A bad hook (e.g. malformed URL) must not block the rest. - LOG.warn("webhook " + hook.id + " delivery failed", 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()); } } }