feat: add bounded client retry for admission rejections

README.md

@@ -187,6 +187,10 @@ admission check stays cheap even under heavy load. Overload semantics:
   since-lowered limit**; the recovered jobs run to completion normally, and only
   *new* submissions are rejected until the count falls back below the limit.
 
+The client **surfaces** a rejection by default (one-shot submit). It can also
+**opt in** to bounded, jittered retry of *retryable* rejections — see
+[client-side submission retry](#client-side-submission-retry) below.
+
 Watch it shed load with no workers running (jobs stay `QUEUED`, so they stay
 active):
 
@@ -195,6 +199,31 @@ active):
 ./scripts/client.sh submit sleep --milliseconds 60000   # x4; the 4th is rejected
 ```
 
+**Client-side submission retry.** By default the client makes exactly one
+submission attempt and surfaces a `SUBMIT_REJECTED` to the caller (unchanged
+behavior). Passing `--submit-retries N` lets it re-attempt admission up to `N`
+times **after** the first attempt (so `N + 1` attempts total; `N = 0` is the
+default one-shot):
+
+```bash
+./scripts/client.sh submit sha256 --text abc --submit-retries 4
+# tune the back-off: --submit-retry-base-millis 200 --submit-retry-max-millis 5000
+```
+
+- It retries **only** a typed `SUBMIT_REJECTED` whose `retryable` flag is true.
+  A malformed-request error, a protocol violation, or any transport failure is
+  **never** retried — a dropped connection is ambiguous about whether the job
+  was created, so a blind resend could submit it twice. Retrying a *rejection*
+  is safe because a rejection provably creates and persists nothing.
+- Back-off is bounded exponential with **full jitter**
+  (`random in [0, min(cap, base·2^(n-1))]`), so many clients rejected at once do
+  not retry in lockstep. When the budget is exhausted the caller still receives
+  the typed `SubmitRejectedException`.
+- This is purely a **client** convenience; the wire protocol and coordinator are
+  unchanged, and it is distinct from the coordinator's *execution* retries.
+- The **benchmark deliberately does not retry** — it measures raw admission
+  shedding, so its accepted/rejected counts stay directly comparable.
+
 **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):
 
@@ -325,7 +354,9 @@ containers.
   coordinator refuses new submissions with a typed, retryable `SUBMIT_REJECTED`
   rather than queueing them. This bounds the coordinator's memory and the
   latency of admitted work; it does *not* raise throughput, and it applies only
-  to new work — jobs already in the system always run to a terminal state.
+  to new work — jobs already in the system always run to a terminal state. The
+  client can opt in to bounded, jittered retry of these rejections
+  (`--submit-retries`), which never resends on an ambiguous transport failure.
 - **Single coordinator.** The coordinator is a single point of failure;
   durability (not availability) is what restart recovery provides. Multiple
   coordinators would require leader election/consensus — out of scope, by
@@ -342,8 +373,11 @@ More detail in [docs/FAILURE_MODEL.md](docs/FAILURE_MODEL.md) and
 - Scheduling is FIFO / least-loaded; no priorities, deadlines, or fairness.
 - Results live in the job row; large results would need external storage.
 - Admission is a single global active-job limit (`--max-active-jobs`); there is
-  no per-client fairness or priority in what gets shed, and the client does not
-  yet auto-retry rejections (it surfaces them for the caller to handle).
+  no per-client fairness or priority in what gets shed. The client surfaces
+  rejections by default and can opt in to bounded, jittered retry
+  (`--submit-retries`), but this only reschedules the client's own attempts — it
+  adds no capacity and, under sustained overload, only shifts where the load is
+  shed.
 - 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.

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

@@ -73,7 +73,7 @@ private static void run(String[] rawArgs) throws Exception {
             return;
         }
 
-        try (CoordinatorClient client = new CoordinatorClient(host, port)) {
+        try (CoordinatorClient client = new CoordinatorClient(host, port, submitRetryPolicy(options))) {
             switch (command) {
                 case "submit" -> submit(client, positionals, options);
                 case "status" -> printJob(client.status(requireJobId(positionals)));
@@ -121,14 +121,21 @@ private static void submit(CoordinatorClient client, List<String> positionals, A
             default -> throw new UsageException("Unknown task type: " + positionals.getFirst());
         };
 
+        int submitRetries = options.getInt("submit-retries", 0);
         String jobId;
         try {
             jobId = client.submit(taskType, payload, options.getInt("max-attempts", 0));
         } catch (SubmitRejectedException rejected) {
             System.err.println("Submission rejected (coordinator overloaded): " + rejected.getMessage());
             System.err.println("  Active jobs: " + rejected.activeCount()
                     + " (limit " + rejected.limit() + ")");
-            System.err.println("  Retryable: back off and submit again once load subsides.");
+            if (submitRetries > 0) {
+                System.err.println("  Still overloaded after " + submitRetries
+                        + " retr" + (submitRetries == 1 ? "y" : "ies") + "; giving up.");
+            } else {
+                System.err.println("  Retryable: back off and submit again once load subsides"
+                        + " (or pass --submit-retries N).");
+            }
             System.exit(3);
             return;
         }
@@ -219,6 +226,29 @@ private static void listWorkers(CoordinatorClient client) throws Exception {
         }
     }
 
+    /**
+     * Builds the submission-retry policy from opt-in flags. Absent (or
+     * {@code --submit-retries 0}) means no retry — the historical one-shot
+     * behavior. {@code --submit-retries N} allows N retries <em>after</em> the
+     * first attempt (N+1 attempts total), with jittered exponential back-off.
+     */
+    private static ClientRetryPolicy submitRetryPolicy(Args options) throws UsageException {
+        int retries = options.getInt("submit-retries", 0);
+        if (retries < 0) {
+            throw new UsageException("--submit-retries must be >= 0");
+        }
+        if (retries == 0) {
+            return ClientRetryPolicy.none();
+        }
+        long base = options.getLong("submit-retry-base-millis", 200);
+        long max = options.getLong("submit-retry-max-millis", 5_000);
+        try {
+            return ClientRetryPolicy.ofRetries(retries, base, max);
+        } catch (IllegalArgumentException invalid) {
+            throw new UsageException(invalid.getMessage());
+        }
+    }
+
     private static String requireJobId(List<String> positionals) throws UsageException {
         if (positionals.isEmpty()) {
             throw new UsageException("Missing job id");
@@ -259,7 +289,11 @@ private static void printUsage() {
                         [--task sha256|sleep|prime-count] [--sleep-millis <ms>]
                         [--prime-limit <n>] [--wait-timeout-millis <ms>]
                 Global options: --host <host> (default localhost), --port <port> (default 7071)
-                Submit options: --max-attempts <n> (default: coordinator setting)""");
+                Submit options: --max-attempts <n> (default: coordinator setting)
+                  --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)""");
     }
 
     private static final class UsageException extends Exception {

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

@@ -0,0 +1,91 @@
+package io.github.achrafaittayeb.dtp.client;
+
+/**
+ * Client-side policy for retrying <em>submission</em> after a typed, retryable
+ * {@link io.github.achrafaittayeb.dtp.common.protocol.SubmitRejected} (coordinator
+ * overload). This is distinct from the coordinator's execution-retry policy,
+ * which decides how a <em>running</em> job's failed attempts are re-run — see
+ * the coordinator's {@code RetryPolicy}. This one never re-runs work; it only
+ * re-attempts admission of a job that was never accepted.
+ *
+ * <p>Back-off is bounded exponential with full jitter: the wait before the
+ * retry that follows attempt {@code n} is a uniform random value in
+ * {@code [0, min(maxDelay, baseDelay * 2^(n-1))]}. Full jitter (rather than a
+ * fixed schedule) desynchronizes many clients that were all rejected at once,
+ * so they do not retry in lockstep and re-create the overload spike.
+ */
+public final class ClientRetryPolicy {
+
+    private final int maxAttempts;
+    private final long baseDelayMillis;
+    private final long maxDelayMillis;
+
+    private ClientRetryPolicy(int maxAttempts, long baseDelayMillis, long maxDelayMillis) {
+        if (maxAttempts < 1) {
+            throw new IllegalArgumentException("maxAttempts must be >= 1, got " + maxAttempts);
+        }
+        if (baseDelayMillis < 0 || maxDelayMillis < baseDelayMillis) {
+            throw new IllegalArgumentException(
+                    "Require 0 <= baseDelay <= maxDelay, got "
+                            + baseDelayMillis + "/" + maxDelayMillis);
+        }
+        this.maxAttempts = maxAttempts;
+        this.baseDelayMillis = baseDelayMillis;
+        this.maxDelayMillis = maxDelayMillis;
+    }
+
+    /** No retry: exactly one submission attempt (today's default behavior). */
+    public static ClientRetryPolicy none() {
+        return new ClientRetryPolicy(1, 0, 0);
+    }
+
+    /**
+     * {@code retries} attempts <em>after</em> the initial one, so the total
+     * attempt budget is {@code retries + 1}. {@code retries == 0} is equivalent
+     * to {@link #none()}.
+     */
+    public static ClientRetryPolicy ofRetries(int retries, long baseDelayMillis, long maxDelayMillis) {
+        if (retries < 0) {
+            throw new IllegalArgumentException("retries must be >= 0, got " + retries);
+        }
+        return new ClientRetryPolicy(retries + 1, baseDelayMillis, maxDelayMillis);
+    }
+
+    /** Total submission attempts allowed, including the first. Always {@code >= 1}. */
+    public int maxAttempts() {
+        return maxAttempts;
+    }
+
+    /** Whether this policy ever retries (i.e. allows more than one attempt). */
+    public boolean enabled() {
+        return maxAttempts > 1;
+    }
+
+    /**
+     * Full-jitter back-off before the retry that follows {@code attemptNumber}
+     * (the number of attempts already made, {@code >= 1}).
+     *
+     * @param attemptNumber attempts made so far (1 = only the first has run)
+     * @param random        a value in {@code [0, 1)}, e.g. from an RNG
+     * @return milliseconds to wait, in {@code [0, min(maxDelay, base*2^(n-1))]}
+     */
+    long backoffMillis(int attemptNumber, double random) {
+        if (attemptNumber < 1) {
+            throw new IllegalArgumentException("attemptNumber must be >= 1, got " + attemptNumber);
+        }
+        if (random < 0 || random >= 1) {
+            throw new IllegalArgumentException("random must be in [0, 1), got " + random);
+        }
+        // Shift instead of Math.pow; clamp the exponent so it can never overflow.
+        int exponent = Math.min(attemptNumber - 1, 30);
+        long ceiling = baseDelayMillis << exponent;
+        if (ceiling < 0 || ceiling > maxDelayMillis) {
+            ceiling = maxDelayMillis;
+        }
+        // Full jitter over the inclusive range [0, ceiling]; random in [0, 1)
+        // maps onto 0..ceiling. Guard the +1 against overflow at extreme caps.
+        long span = ceiling == Long.MAX_VALUE ? ceiling : ceiling + 1;
+        long delay = (long) (random * span);
+        return Math.min(delay, maxDelayMillis);
+    }
+}

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

@@ -37,24 +37,54 @@ public final class CoordinatorClient implements AutoCloseable {
     private final Socket socket;
     private final InputStream in;
     private final OutputStream out;
+    private final SubmitRetrier submitRetrier;
 
     public CoordinatorClient(String host, int port) throws IOException {
+        this(host, port, ClientRetryPolicy.none());
+    }
+
+    /**
+     * Connects with an explicit submission-retry policy. {@link ClientRetryPolicy#none()}
+     * (the default of the two-arg constructor) preserves the historical
+     * one-shot behavior; any other policy re-attempts admission after a
+     * retryable overload rejection.
+     */
+    public CoordinatorClient(String host, int port, ClientRetryPolicy retryPolicy)
+            throws IOException {
+        this(host, port, new SubmitRetrier(retryPolicy));
+    }
+
+    /** Test seam: inject a retrier with a fake sleeper/RNG for deterministic timing. */
+    CoordinatorClient(String host, int port, SubmitRetrier submitRetrier) throws IOException {
         this.socket = new Socket(host, port);
         this.socket.setTcpNoDelay(true);
         this.in = socket.getInputStream();
         this.out = socket.getOutputStream();
+        this.submitRetrier = submitRetrier;
     }
 
     /**
      * Submits a job; returns its id. {@code maxAttempts <= 0} uses the server
-     * default.
+     * default. When configured with a retrying {@link ClientRetryPolicy}, a
+     * retryable overload rejection is re-attempted with jittered back-off; the
+     * back-off sleep happens outside this instance's synchronization, so it
+     * never blocks another thread's wire exchange on the same connection. (A
+     * connection still serves one request at a time, so concurrent callers on
+     * one client are not expected; the point is that sleeping holds no monitor.)
      *
-     * @throws SubmitRejectedException if the coordinator is at its active-job
-     *     limit (a valid request refused under overload — distinct from a
-     *     malformed-request error, and safe to retry after a back-off)
-     * @throws IOException on any other error reply or transport failure
+     * @throws SubmitRejectedException if the coordinator refused the submission
+     *     under overload and the retry budget (if any) was exhausted — a valid
+     *     request refused, distinct from a malformed-request error
+     * @throws IOException on any other error reply or transport failure (never
+     *     retried, since a job may already exist)
      */
-    public synchronized String submit(TaskType taskType, JsonNode payload, int maxAttempts)
+    public String submit(TaskType taskType, JsonNode payload, int maxAttempts)
+            throws IOException {
+        return submitRetrier.submit(() -> submitOnce(taskType, payload, maxAttempts));
+    }
+
+    /** One submission exchange over the wire; synchronized like every other request. */
+    private synchronized String submitOnce(TaskType taskType, JsonNode payload, int maxAttempts)
             throws IOException {
         Message reply = exchange(new SubmitJob(taskType, payload, maxAttempts));
         if (reply instanceof JobSubmitted submitted) {

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

@@ -15,11 +15,13 @@ public final class SubmitRejectedException extends IOException {
 
     private final int activeCount;
     private final int limit;
+    private final boolean retryable;
 
     public SubmitRejectedException(SubmitRejected rejected) {
         super(rejected.reason());
         this.activeCount = rejected.activeCount();
         this.limit = rejected.limit();
+        this.retryable = rejected.retryable();
     }
 
     public int activeCount() {
@@ -30,8 +32,14 @@ public int limit() {
         return limit;
     }
 
-    /** Overload rejection is always a transient, retryable condition. */
+    /**
+     * Whether the coordinator marked this rejection retryable. Reflects the
+     * {@code retryable} field on the wire reply rather than assuming a value,
+     * so a caller (or the client's own retry loop) only retries when the server
+     * actually said it was safe to. Overload rejections are retryable today,
+     * but honoring the field keeps clients correct if that ever changes.
+     */
     public boolean retryable() {
-        return true;
+        return retryable;
     }
 }