fix(java): java sdk gaps

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<int>() {}` —
+            // 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");

sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java

@@ -242,10 +242,14 @@ private Optional<String> 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. */

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} <em>after</em> a
- * signing/encryption codec in the chain to verify integrity before decompressing.
+ * exhaust worker memory (zip bomb). Order {@code GzipCodec} <em>before</em> 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. */

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<String> delayed =
+                    TASK.withOptions(EnqueueOptions.builder().delayMs(60_000).build());
+            Optional<String> 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")) {

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<String> 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<String> 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;

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<Integer> order = Task.of("im.order", Integer.class);
+        List<Integer> 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 {