feat: add durable idempotent job submissions

README.md

@@ -224,6 +224,32 @@ default one-shot):
 - The **benchmark deliberately does not retry** — it measures raw admission
   shedding, so its accepted/rejected counts stay directly comparable.
 
+**Durable idempotent submission.** A client may attach a stable
+`--idempotency-key` to a submission. The coordinator remembers `key → jobId`
+(persisted on the job row, so it survives restart), and a later submission with
+the same key returns the **original** job id instead of creating another job:
+
+```bash
+./scripts/client.sh submit sha256 --text abc --idempotency-key order-42
+./scripts/client.sh submit sha256 --text abc --idempotency-key order-42  # same job id, no new job
+```
+
+- The key is **optional**; omitting it is exactly today's behavior (every
+  submission is a fresh job). It is checked **before** admission control, so a
+  known key returns its job even at `--max-active-jobs` (it adds no active job);
+  a genuinely new key is admission-controlled as normal.
+- The mapping holds for the job's whole life, **including terminal states** — a
+  completed, failed, or cancelled key returns that same job and never re-runs it.
+  A key is therefore "burned" once its job is terminal; use a new key to run
+  again.
+- Reusing a key with a **different** task type, payload, or max attempts is a
+  conflict (`ERROR`); the existing job is left untouched.
+- This deduplicates **submission**, not execution — execution stays
+  at-least-once. It is the precondition for safely auto-retrying transport
+  failures, but that retry is intentionally **not** enabled in this milestone.
+  Because there is no client authentication, keys share one global namespace —
+  use unguessable, namespaced keys (e.g. UUIDs).
+
 **Measured overload behavior** (Apple M1 Pro, 3 workers × capacity 4 = 12 slots,
 240× 500 ms sleep jobs from 8 connections; full methodology in the doc):
 
@@ -381,6 +407,12 @@ More detail in [docs/FAILURE_MODEL.md](docs/FAILURE_MODEL.md) and
 - Cancellation is cooperative (thread interruption), not hard preemption: a
   task that ignores interruption keeps running until it finishes, though its
   lease is already revoked so its result is discarded.
+- Idempotent submission (`--idempotency-key`) deduplicates *submissions*, not
+  execution — execution remains at-least-once. Keys share one global namespace
+  (there is no client authentication to scope them), and a key is bound to its
+  job for the job's whole life, so a terminal key cannot be reused to run again.
+  Automatic retry of ambiguous transport failures is not yet enabled; the key is
+  the groundwork that would make it safe.
 
 ## Documentation
 

client/src/main/java/io/github/achrafaittayeb/dtp/client/ClientMain.java

@@ -122,9 +122,10 @@ private static void submit(CoordinatorClient client, List<String> positionals, A
         };
 
         int submitRetries = options.getInt("submit-retries", 0);
+        String idempotencyKey = options.get("idempotency-key", null);
         String jobId;
         try {
-            jobId = client.submit(taskType, payload, options.getInt("max-attempts", 0));
+            jobId = client.submit(taskType, payload, options.getInt("max-attempts", 0), idempotencyKey);
         } catch (SubmitRejectedException rejected) {
             System.err.println("Submission rejected (coordinator overloaded): " + rejected.getMessage());
             System.err.println("  Active jobs: " + rejected.activeCount()
@@ -293,7 +294,9 @@ private static void printUsage() {
                   --submit-retries <n>            retries after the first attempt on a
                                                   retryable overload rejection (default 0 = off)
                   --submit-retry-base-millis <ms> base back-off (default 200)
-                  --submit-retry-max-millis <ms>  back-off cap (default 5000)""");
+                  --submit-retry-max-millis <ms>  back-off cap (default 5000)
+                  --idempotency-key <key>         dedup key: resubmitting the same key returns
+                                                  the original job instead of creating a new one""");
     }
 
     private static final class UsageException extends Exception {

client/src/main/java/io/github/achrafaittayeb/dtp/client/CoordinatorClient.java

@@ -80,13 +80,27 @@ public CoordinatorClient(String host, int port, ClientRetryPolicy retryPolicy)
      */
     public String submit(TaskType taskType, JsonNode payload, int maxAttempts)
             throws IOException {
-        return submitRetrier.submit(() -> submitOnce(taskType, payload, maxAttempts));
+        return submit(taskType, payload, maxAttempts, null);
     }
 
-    /** One submission exchange over the wire; synchronized like every other request. */
-    private synchronized String submitOnce(TaskType taskType, JsonNode payload, int maxAttempts)
+    /**
+     * Submits with an optional idempotency key. A {@code null} key is the
+     * historical behavior (always a fresh job). A non-null, stable key lets the
+     * coordinator deduplicate: resubmitting the same key returns the original
+     * job's id instead of creating another logical job. Resubmitting a key that
+     * the coordinator bound to a <em>different</em> request (task type, payload,
+     * or max attempts) is a conflict, surfaced as a plain {@link IOException} —
+     * not a {@link SubmitRejectedException}, so the retry loop never retries it.
+     */
+    public String submit(TaskType taskType, JsonNode payload, int maxAttempts, String idempotencyKey)
             throws IOException {
-        Message reply = exchange(new SubmitJob(taskType, payload, maxAttempts));
+        return submitRetrier.submit(() -> submitOnce(taskType, payload, maxAttempts, idempotencyKey));
+    }
+
+    /** One submission exchange over the wire; synchronized like every other request. */
+    private synchronized String submitOnce(TaskType taskType, JsonNode payload, int maxAttempts,
+                                           String idempotencyKey) throws IOException {
+        Message reply = exchange(new SubmitJob(taskType, payload, maxAttempts, idempotencyKey));
         if (reply instanceof JobSubmitted submitted) {
             return submitted.jobId();
         }

common/src/main/java/io/github/achrafaittayeb/dtp/common/protocol/SubmitJob.java

@@ -6,6 +6,21 @@
 /**
  * Client request to enqueue a new job. {@code maxAttempts <= 0} means
  * "use the coordinator's configured default".
+ *
+ * <p>{@code idempotencyKey} is optional (nullable). When a client supplies a
+ * stable key, the coordinator deduplicates: a second {@code SubmitJob} carrying
+ * a key it has already seen returns the <em>original</em> job's id instead of
+ * creating another logical job (see {@code CoordinatorCore}). A {@code null}
+ * key preserves the historical behavior — every submission creates a fresh job.
+ * The field is deliberately last with a delegating constructor so existing
+ * call sites and older peers (which omit it, and it decodes to {@code null})
+ * remain source- and wire-compatible.
  */
-public record SubmitJob(TaskType taskType, JsonNode payload, int maxAttempts) implements Message {
+public record SubmitJob(TaskType taskType, JsonNode payload, int maxAttempts, String idempotencyKey)
+        implements Message {
+
+    /** Back-compatible constructor for a submission without an idempotency key. */
+    public SubmitJob(TaskType taskType, JsonNode payload, int maxAttempts) {
+        this(taskType, payload, maxAttempts, null);
+    }
 }

common/src/test/java/io/github/achrafaittayeb/dtp/common/protocol/MessageSerializationTest.java

@@ -99,4 +99,32 @@ void submitRejectedCarriesTypeDiscriminator() throws IOException {
                 StandardCharsets.UTF_8);
         assertThat(json).contains("\"type\":\"SUBMIT_REJECTED\"");
     }
+
+    @Test
+    void roundTripsSubmitJobWithAndWithoutIdempotencyKey() throws IOException {
+        ObjectNode payload = JsonNodeFactory.instance.objectNode().put("text", "abc");
+
+        SubmitJob withKey = new SubmitJob(TaskType.SHA256, payload, 2, "key-123");
+        SubmitJob decodedWithKey = (SubmitJob) roundTrip(withKey);
+        assertThat(decodedWithKey).isEqualTo(withKey);
+        assertThat(decodedWithKey.idempotencyKey()).isEqualTo("key-123");
+
+        SubmitJob noKey = new SubmitJob(TaskType.SHA256, payload, 2);
+        SubmitJob decodedNoKey = (SubmitJob) roundTrip(noKey);
+        assertThat(decodedNoKey).isEqualTo(noKey);
+        assertThat(decodedNoKey.idempotencyKey()).isNull();
+    }
+
+    @Test
+    void decodesLegacySubmitJobWithoutIdempotencyKeyField() throws IOException {
+        // A frame from an older client that predates the field must still decode,
+        // with the key defaulting to null.
+        byte[] legacy = ("{\"type\":\"SUBMIT_JOB\",\"taskType\":\"SHA256\","
+                + "\"payload\":{\"text\":\"abc\"},\"maxAttempts\":0}")
+                .getBytes(StandardCharsets.UTF_8);
+        SubmitJob decoded = (SubmitJob) MessageIO.decode(legacy);
+        assertThat(decoded.taskType()).isEqualTo(TaskType.SHA256);
+        assertThat(decoded.maxAttempts()).isZero();
+        assertThat(decoded.idempotencyKey()).isNull();
+    }
 }

coordinator/src/main/java/io/github/achrafaittayeb/dtp/coordinator/CoordinatorCore.java

@@ -56,6 +56,18 @@ public final class CoordinatorCore implements AutoCloseable {
     private final RetryPolicy retryPolicy;
     private final WorkerRegistry registry = new WorkerRegistry();
     private final Map<String, Job> jobs = new HashMap<>();
+
+    /**
+     * Maps a client-supplied idempotency key to the id of the one logical job it
+     * created, so a resubmission of the same key returns the original job rather
+     * than creating a duplicate. Confined to the core thread like {@link #jobs},
+     * so the check-then-create is atomic without locking. Rebuilt from persisted
+     * job rows on recovery, so deduplication survives a restart. A job keeps its
+     * entry for its whole lifetime, including terminal states — resubmitting a
+     * completed/failed/cancelled key returns that same terminal job.
+     */
+    private final Map<String, String> jobIdByKey = new HashMap<>();
+
     private final ScheduledExecutorService coreThread;
 
     /**
@@ -131,49 +143,120 @@ public void onTaskResult(TaskResult result) {
     // Requests from client connections (block until the core thread answers)
     // ------------------------------------------------------------------
 
-    /** Outcome of a submission: either accepted (with a job id) or rejected because the coordinator is at its active-job limit. */
-    public record SubmitOutcome(boolean accepted, String jobId, int activeCount, int limit) {
+    /**
+     * Outcome of a submission. {@code ACCEPTED} carries the job id (a fresh job,
+     * or the original job for an idempotent duplicate); {@code REJECTED} means
+     * the coordinator is at its active-job limit; {@code CONFLICT} means the
+     * idempotency key is already bound to a materially different submission and
+     * {@code conflictMessage} explains why (surfaced to the client as an error).
+     */
+    public record SubmitOutcome(Status status, String jobId, int activeCount, int limit,
+                                String conflictMessage) {
+
+        public enum Status { ACCEPTED, REJECTED, CONFLICT }
 
         static SubmitOutcome accepted(String jobId, int activeCount, int limit) {
-            return new SubmitOutcome(true, jobId, activeCount, limit);
+            return new SubmitOutcome(Status.ACCEPTED, jobId, activeCount, limit, null);
         }
 
         static SubmitOutcome rejected(int activeCount, int limit) {
-            return new SubmitOutcome(false, null, activeCount, limit);
+            return new SubmitOutcome(Status.REJECTED, null, activeCount, limit, null);
+        }
+
+        static SubmitOutcome conflict(String message) {
+            return new SubmitOutcome(Status.CONFLICT, null, 0, 0, message);
+        }
+
+        public boolean accepted() {
+            return status == Status.ACCEPTED;
         }
     }
 
+    /** Overload {@link #submitJob(TaskType, JsonNode, int, String)} without an idempotency key. */
+    public SubmitOutcome submitJob(TaskType taskType, JsonNode payload, int requestedMaxAttempts)
+            throws InvalidPayloadException {
+        return submitJob(taskType, payload, requestedMaxAttempts, null);
+    }
+
     /**
      * Admission point for new work. A malformed payload is rejected up front
-     * (before the core thread) as {@link InvalidPayloadException}. A well-formed
-     * request is admitted only if the active-job count is below the configured
-     * limit; otherwise it is shed with a {@link SubmitOutcome#rejected} outcome —
-     * the coordinator refuses new work rather than buffering it unboundedly.
-     * Only fresh submissions pass through here; retries never do.
+     * (before the core thread) as {@link InvalidPayloadException}. On the core
+     * thread the request is handled in this order:
+     *
+     * <ol>
+     *   <li><b>Deduplication first.</b> If a non-null {@code idempotencyKey} is
+     *       already known, the original job's id is returned. This happens
+     *       <em>before</em> admission control and so bypasses the active-job
+     *       limit: a known key creates no new job, so there is nothing to admit —
+     *       exactly like a retry of already-accepted work. If the key is known
+     *       but the request differs materially (task type, payload, or effective
+     *       max attempts), it is a {@link SubmitOutcome#conflict}: the existing
+     *       job is left untouched and no new job is created.</li>
+     *   <li><b>Admission.</b> A genuinely new submission (no key, or an unseen
+     *       key) is admitted only if the active-job count is below the limit;
+     *       otherwise it is shed with {@link SubmitOutcome#rejected}.</li>
+     * </ol>
+     *
+     * <p>Deduplication prevents a <em>duplicate logical submission</em>; it does
+     * not change execution semantics, which remain at-least-once. Only fresh
+     * submissions pass through admission; execution retries never do.
      */
-    public SubmitOutcome submitJob(TaskType taskType, JsonNode payload, int requestedMaxAttempts)
-            throws InvalidPayloadException {
+    public SubmitOutcome submitJob(TaskType taskType, JsonNode payload, int requestedMaxAttempts,
+                                   String idempotencyKey) throws InvalidPayloadException {
         TaskPayloads.validate(taskType, payload);
         int maxAttempts = requestedMaxAttempts > 0 ? requestedMaxAttempts : config.defaultMaxAttempts();
         return askCore(() -> {
+            if (idempotencyKey != null) {
+                String existingId = jobIdByKey.get(idempotencyKey);
+                if (existingId != null) {
+                    Job existing = jobs.get(existingId);
+                    if (!matchesSubmission(existing, taskType, payload, maxAttempts)) {
+                        log.warn("Idempotency conflict: key={} bound to jobId={} but resubmission "
+                                + "differs (type/payload/maxAttempts)", idempotencyKey, existingId);
+                        return SubmitOutcome.conflict("Idempotency key '" + idempotencyKey
+                                + "' is already bound to a different submission (job " + existingId + ")");
+                    }
+                    log.info("Idempotent duplicate: key={} returning original jobId={} (state {})",
+                            idempotencyKey, existingId, existing.state());
+                    return SubmitOutcome.accepted(existingId, activeJobCount, config.maxActiveJobs());
+                }
+            }
             int limit = config.maxActiveJobs();
             if (activeJobCount >= limit) {
                 log.warn("Job submission rejected (overloaded): activeJobs={} limit={} type={}",
                         activeJobCount, limit, taskType);
                 return SubmitOutcome.rejected(activeJobCount, limit);
             }
             Job job = Job.createQueued(taskType, payload, maxAttempts,
-                    config.taskTimeoutMillis(), now());
+                    config.taskTimeoutMillis(), now(), idempotencyKey);
             jobs.put(job.id(), job);
+            if (idempotencyKey != null) {
+                jobIdByKey.put(idempotencyKey, job.id());
+            }
             activeJobCount++;
             repository.save(job);
-            log.info("Job submitted: jobId={} type={} maxAttempts={} activeJobs={}/{}",
-                    job.id(), taskType, maxAttempts, activeJobCount, limit);
+            log.info("Job submitted: jobId={} type={} maxAttempts={} activeJobs={}/{} idempotencyKey={}",
+                    job.id(), taskType, maxAttempts, activeJobCount, limit, idempotencyKey);
             scheduleQueuedJobs();
             return SubmitOutcome.accepted(job.id(), activeJobCount, limit);
         });
     }
 
+    /**
+     * Whether a resubmission under a known key describes the same logical work as
+     * the job the key already created. Payload equality is structural
+     * ({@link JsonNode#equals}), and the stored job holds the already-resolved
+     * effective max attempts, so it is compared against the incoming effective
+     * value.
+     */
+    private boolean matchesSubmission(Job existing, TaskType taskType, JsonNode payload,
+                                      int effectiveMaxAttempts) {
+        return existing != null
+                && existing.taskType() == taskType
+                && existing.maxAttempts() == effectiveMaxAttempts
+                && existing.payload().equals(payload);
+    }
+
     /** Observability/test hook: current number of active (non-terminal) jobs, read on the core thread. */
     public int activeJobCount() {
         return askCore(() -> activeJobCount);
@@ -447,6 +530,12 @@ private void recoverFromRepository() {
                 repository.save(job);
             }
             jobs.put(job.id(), job);
+            // Rebuild the dedup map so idempotent submission survives restart. A
+            // job keeps its key in every state, including terminal, so a resend
+            // of a completed/failed/cancelled key still returns the same job.
+            if (job.idempotencyKey() != null) {
+                jobIdByKey.put(job.idempotencyKey(), job.id());
+            }
             // Seed the active-job counter from each job's post-recovery state, so
             // admission control resumes with an exact count. A restart may leave
             // activeJobCount above a since-lowered limit; that is intended — the