From 008b7c11cc191b95390943ade3a29d29678e41fc Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Sun, 5 Jul 2026 18:53:43 +0530 Subject: [PATCH 1/4] fix(java): make zero defer override task delay --- .../org/byteveda/taskito/DefaultTaskito.java | 10 +++++++--- .../byteveda/taskito/EnqueueDecisionTest.java | 18 ++++++++++++++++++ 2 files changed, 25 insertions(+), 3 deletions(-) diff --git a/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java b/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java index 94a02532..53e4a963 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java @@ -242,10 +242,14 @@ private Optional dispatchEnqueue( return Optional.of(backend.enqueue(taskName, data, encode(finalOptions))); } - /** Apply a defer delay to the options (overriding any delay already set). */ + /** + * Apply a defer delay to the options, overriding any delay already set — + * {@code Defer(Duration.ZERO)} (e.g. {@code deferUntil} of a past instant) + * means "enqueue now", not "keep the task's baked-in delay". + */ private static EnqueueOptions withDelay(EnqueueOptions options, Duration delay) { - long delayMs = delay.toMillis(); - return delayMs <= 0 ? options : options.toBuilder().delayMs(delayMs).build(); + long delayMs = Math.max(delay.toMillis(), 0); + return options.toBuilder().delayMs(delayMs).build(); } /** Apply a task's payload codecs in order; throws if a name is not registered. */ diff --git a/sdks/java/src/test/java/org/byteveda/taskito/EnqueueDecisionTest.java b/sdks/java/src/test/java/org/byteveda/taskito/EnqueueDecisionTest.java index c0e0d53c..46a4ca99 100644 --- a/sdks/java/src/test/java/org/byteveda/taskito/EnqueueDecisionTest.java +++ b/sdks/java/src/test/java/org/byteveda/taskito/EnqueueDecisionTest.java @@ -12,6 +12,7 @@ import org.byteveda.taskito.errors.PredicateRejectedException; import org.byteveda.taskito.model.Job; import org.byteveda.taskito.predicates.EnqueueDecision; +import org.byteveda.taskito.task.EnqueueOptions; import org.byteveda.taskito.task.Task; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; @@ -53,6 +54,23 @@ void deferDelaysTheScheduledTime(@TempDir Path dir) throws Exception { } } + @Test + void zeroDeferOverridesTaskDelay(@TempDir Path dir) throws Exception { + try (Taskito queue = open(dir, "deferzero")) { + queue.gate(TASK.name(), context -> EnqueueDecision.defer(Duration.ZERO)); + + Task delayed = + TASK.withOptions(EnqueueOptions.builder().delayMs(60_000).build()); + Optional id = queue.tryEnqueue(delayed, "now"); + assertTrue(id.isPresent()); + Job job = queue.getJob(id.get()).orElseThrow(); + // Defer(ZERO) means "enqueue now" — the task's baked-in delay must not survive. + assertTrue( + job.scheduledAt - job.createdAt < 1_000, + "expected immediate schedule, got " + (job.scheduledAt - job.createdAt) + "ms"); + } + } + @Test void allowEnqueuesNormally(@TempDir Path dir) throws Exception { try (Taskito queue = open(dir, "allow")) { From 94678ab5454828cf1eda1d013eda0c6b609227c7 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Sun, 5 Jul 2026 18:53:43 +0530 Subject: [PATCH 2/4] docs(java): correct gzip codec chain order MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Javadoc told readers to place gzip after the signing codec, which decodes to decompress-before-verify — the insecure order. --- .../java/org/byteveda/taskito/serialization/GzipCodec.java | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/sdks/java/src/main/java/org/byteveda/taskito/serialization/GzipCodec.java b/sdks/java/src/main/java/org/byteveda/taskito/serialization/GzipCodec.java index 4ef718c2..50f95e70 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/serialization/GzipCodec.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/serialization/GzipCodec.java @@ -10,8 +10,9 @@ /** * A {@link PayloadCodec} that gzip-compresses payloads. Decompression is bounded * by {@code maxDecompressedBytes} so a small malicious payload can't expand to - * exhaust worker memory (zip bomb). Order {@code GzipCodec} after a - * signing/encryption codec in the chain to verify integrity before decompressing. + * exhaust worker memory (zip bomb). Order {@code GzipCodec} before a + * signing/encryption codec in the list (e.g. {@code List.of(gzip, hmac)}): + * codecs decode in reverse, so integrity is verified before decompressing. */ public final class GzipCodec implements PayloadCodec { /** Default cap on decompressed output: 64 MiB. */ From d03983042d007a7e55ba59f514c58931739d45fc Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Sun, 5 Jul 2026 18:53:43 +0530 Subject: [PATCH 3/4] fix(java): reject primitive task handler payloads --- .../byteveda/taskito/processor/TaskHandlerProcessor.java | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/sdks/java/processor/src/main/java/org/byteveda/taskito/processor/TaskHandlerProcessor.java b/sdks/java/processor/src/main/java/org/byteveda/taskito/processor/TaskHandlerProcessor.java index 514877c3..95bd28af 100644 --- a/sdks/java/processor/src/main/java/org/byteveda/taskito/processor/TaskHandlerProcessor.java +++ b/sdks/java/processor/src/main/java/org/byteveda/taskito/processor/TaskHandlerProcessor.java @@ -69,6 +69,12 @@ private boolean validate(ExecutableElement method) { error(method, "the first @TaskHandler parameter is the payload and must not be annotated @Resource"); return false; } + if (params.get(0).asType().getKind().isPrimitive()) { + // A primitive payload would generate `new TypeReference() {}` — + // invalid Java that only surfaces as a cryptic error in the generated file. + error(method, "@TaskHandler payload must be a reference type (use the boxed type, e.g. Integer)"); + return false; + } for (int i = 1; i < params.size(); i++) { if (resourceName(params.get(i)) == null) { error(method, "@TaskHandler parameters after the payload must be annotated @Resource"); From 7835c1207866d90183518159b017c15659133407 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Sun, 5 Jul 2026 18:53:43 +0530 Subject: [PATCH 4/4] fix(java): dequeue FIFO within priority in test backend Production orders priority DESC then scheduled_at ASC; the in-memory backend kept hash-map iteration order on ties. --- .../taskito/test/InMemoryQueueBackend.java | 20 +++++++++++-- .../test/InMemoryQueueBackendTest.java | 29 +++++++++++++++++++ 2 files changed, 46 insertions(+), 3 deletions(-) diff --git a/sdks/java/test-support/src/main/java/org/byteveda/taskito/test/InMemoryQueueBackend.java b/sdks/java/test-support/src/main/java/org/byteveda/taskito/test/InMemoryQueueBackend.java index dae38bee..80bf8e90 100644 --- a/sdks/java/test-support/src/main/java/org/byteveda/taskito/test/InMemoryQueueBackend.java +++ b/sdks/java/test-support/src/main/java/org/byteveda/taskito/test/InMemoryQueueBackend.java @@ -73,7 +73,8 @@ private synchronized String enqueueOne(String taskName, byte[] payload, JsonNode } } JobRec job = new JobRec(); - job.id = "im-" + seq.incrementAndGet(); + job.enqueueSeq = seq.incrementAndGet(); + job.id = "im-" + job.enqueueSeq; job.taskName = taskName; job.payload = payload; job.queue = optText(opts, "queue", DEFAULT_QUEUE); @@ -215,7 +216,8 @@ public synchronized String retryDead(String deadId) { JobRec source = jobs.get((String) entry.get("originalJobId")); dead.remove(entry); JobRec job = new JobRec(); - job.id = "im-" + seq.incrementAndGet(); + job.enqueueSeq = seq.incrementAndGet(); + job.id = "im-" + job.enqueueSeq; job.taskName = (String) entry.get("taskName"); job.queue = (String) entry.get("queue"); job.payload = source == null ? new byte[0] : source.payload; @@ -563,7 +565,7 @@ private synchronized JobRec claimNext(java.util.Set queues) { if (!queues.isEmpty() && !queues.contains(job.queue)) { continue; } - if (best == null || job.priority > best.priority) { + if (best == null || claimsBefore(job, best)) { best = job; } } @@ -574,6 +576,17 @@ private synchronized JobRec claimNext(java.util.Set queues) { return best; } + /** Production dequeue order: priority DESC, then FIFO (scheduled_at ASC, insertion order on ties). */ + private static boolean claimsBefore(JobRec candidate, JobRec current) { + if (candidate.priority != current.priority) { + return candidate.priority > current.priority; + } + if (candidate.scheduledAt != current.scheduledAt) { + return candidate.scheduledAt < current.scheduledAt; + } + return candidate.enqueueSeq < current.enqueueSeq; + } + private synchronized void onComplete(JobRec job, byte[] result) { job.result = result; job.status = "complete"; @@ -690,6 +703,7 @@ private static long optLong(JsonNode node, String field, long fallback) { /** A stored job. */ private static final class JobRec { String id; + long enqueueSeq; String queue = DEFAULT_QUEUE; String taskName; byte[] payload; diff --git a/sdks/java/test-support/src/test/java/org/byteveda/taskito/test/InMemoryQueueBackendTest.java b/sdks/java/test-support/src/test/java/org/byteveda/taskito/test/InMemoryQueueBackendTest.java index c35b9c79..a7b91538 100644 --- a/sdks/java/test-support/src/test/java/org/byteveda/taskito/test/InMemoryQueueBackendTest.java +++ b/sdks/java/test-support/src/test/java/org/byteveda/taskito/test/InMemoryQueueBackendTest.java @@ -4,6 +4,9 @@ import static org.junit.jupiter.api.Assertions.assertTrue; import java.time.Duration; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; import org.byteveda.taskito.Taskito; import org.byteveda.taskito.model.Job; import org.byteveda.taskito.model.JobStatus; @@ -28,6 +31,32 @@ void runsJobToCompletion() throws Exception { } } + @Test + @Timeout(20) + void samePriorityJobsRunInEnqueueOrder() throws Exception { + Task order = Task.of("im.order", Integer.class); + List seen = Collections.synchronizedList(new ArrayList<>()); + try (Taskito queue = InMemoryTaskito.open()) { + String last = null; + for (int i = 0; i < 5; i++) { + last = queue.enqueue(order, i); + } + // Single-threaded worker so claim order is observable as run order. + try (Worker worker = queue.worker() + .concurrency(1) + .handle(order, p -> { + seen.add(p); + return p; + }) + .start()) { + queue.awaitJob(last, Duration.ofSeconds(10)).orElseThrow(); + // Production dequeues FIFO within a priority tier; the in-memory + // backend must match, not follow hash-map iteration order. + assertEquals(List.of(0, 1, 2, 3, 4), seen); + } + } + } + @Test @Timeout(20) void retriesThenDeadLetters() throws Exception {