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..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 @@ -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 ""; + } + } + static String sign(String secret, byte[] body) { try { Mac mac = Mac.getInstance(ALGORITHM); 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..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 @@ -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 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 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 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); } 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"); + } + } + } }