diff --git a/README.md b/README.md index deff3ec..c5a868c 100644 --- a/README.md +++ b/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. diff --git a/client/src/main/java/io/github/achrafaittayeb/dtp/client/ClientMain.java b/client/src/main/java/io/github/achrafaittayeb/dtp/client/ClientMain.java index 7451b8d..dd350b1 100644 --- a/client/src/main/java/io/github/achrafaittayeb/dtp/client/ClientMain.java +++ b/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,6 +121,7 @@ private static void submit(CoordinatorClient client, List 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)); @@ -128,7 +129,13 @@ private static void submit(CoordinatorClient client, List positionals, A 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 after 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 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 ] [--prime-limit ] [--wait-timeout-millis ] Global options: --host (default localhost), --port (default 7071) - Submit options: --max-attempts (default: coordinator setting)"""); + Submit options: --max-attempts (default: coordinator setting) + --submit-retries retries after the first attempt on a + retryable overload rejection (default 0 = off) + --submit-retry-base-millis base back-off (default 200) + --submit-retry-max-millis back-off cap (default 5000)"""); } private static final class UsageException extends Exception { diff --git a/client/src/main/java/io/github/achrafaittayeb/dtp/client/ClientRetryPolicy.java b/client/src/main/java/io/github/achrafaittayeb/dtp/client/ClientRetryPolicy.java new file mode 100644 index 0000000..e3a8f98 --- /dev/null +++ b/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 submission 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 running 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. + * + *

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 after 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); + } +} diff --git a/client/src/main/java/io/github/achrafaittayeb/dtp/client/CoordinatorClient.java b/client/src/main/java/io/github/achrafaittayeb/dtp/client/CoordinatorClient.java index f0277d3..c8b3873 100644 --- a/client/src/main/java/io/github/achrafaittayeb/dtp/client/CoordinatorClient.java +++ b/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) { diff --git a/client/src/main/java/io/github/achrafaittayeb/dtp/client/SubmitRejectedException.java b/client/src/main/java/io/github/achrafaittayeb/dtp/client/SubmitRejectedException.java index a5169b1..4304683 100644 --- a/client/src/main/java/io/github/achrafaittayeb/dtp/client/SubmitRejectedException.java +++ b/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; } } diff --git a/client/src/main/java/io/github/achrafaittayeb/dtp/client/SubmitRetrier.java b/client/src/main/java/io/github/achrafaittayeb/dtp/client/SubmitRetrier.java new file mode 100644 index 0000000..141273c --- /dev/null +++ b/client/src/main/java/io/github/achrafaittayeb/dtp/client/SubmitRetrier.java @@ -0,0 +1,88 @@ +package io.github.achrafaittayeb.dtp.client; + +import java.io.IOException; +import java.io.InterruptedIOException; +import java.util.concurrent.ThreadLocalRandom; +import java.util.function.DoubleSupplier; + +/** + * Runs a single submission attempt under a {@link ClientRetryPolicy}, retrying + * only when the coordinator returns a typed, retryable + * {@link SubmitRejectedException} (overload). Everything else — a + * {@code retryable == false} rejection, a malformed-request error, a protocol + * violation, or any transport failure — is propagated on the first occurrence + * and never retried: those either must not be resent as-is, or are ambiguous + * about whether the job was actually created, and a blind resend could submit + * the same job twice. + * + *

This class deliberately knows nothing about sockets. The wire exchange is + * supplied as an {@link Attempt}, and the clock ({@link Sleeper}) and jitter + * source (an RNG returning {@code [0, 1)}) are injectable, so the retry + * decision and back-off timing can be unit-tested deterministically without a + * network or real sleeps. + */ +final class SubmitRetrier { + + /** One submission attempt over the wire. */ + @FunctionalInterface + interface Attempt { + String submit() throws IOException; + } + + /** Injectable sleep, so tests can observe delays instead of waiting them out. */ + @FunctionalInterface + interface Sleeper { + void sleep(long millis) throws InterruptedException; + } + + private final ClientRetryPolicy policy; + private final Sleeper sleeper; + private final DoubleSupplier jitter; + + /** Production wiring: real sleep and a per-thread RNG. */ + SubmitRetrier(ClientRetryPolicy policy) { + this(policy, Thread::sleep, () -> ThreadLocalRandom.current().nextDouble()); + } + + SubmitRetrier(ClientRetryPolicy policy, Sleeper sleeper, DoubleSupplier jitter) { + this.policy = policy; + this.sleeper = sleeper; + this.jitter = jitter; + } + + /** + * Attempts the submission, retrying retryable overload rejections up to the + * policy's attempt budget with jittered exponential back-off. + * + * @return the submitted job id + * @throws SubmitRejectedException the last rejection, if the budget is + * exhausted or the rejection is not retryable + * @throws IOException any other error or transport failure, unretried; also + * an {@link InterruptedIOException} (interrupt flag restored) if a + * back-off sleep is interrupted + */ + String submit(Attempt attempt) throws IOException { + int attemptNumber = 1; + while (true) { + try { + return attempt.submit(); + } catch (SubmitRejectedException rejected) { + if (!rejected.retryable() || attemptNumber >= policy.maxAttempts()) { + throw rejected; + } + long delayMillis = policy.backoffMillis(attemptNumber, jitter.getAsDouble()); + try { + sleeper.sleep(delayMillis); + } catch (InterruptedException interrupted) { + Thread.currentThread().interrupt(); + InterruptedIOException abort = new InterruptedIOException( + "Interrupted during submit-retry back-off after " + + attemptNumber + " attempt(s)"); + abort.initCause(rejected); + throw abort; + } + attemptNumber++; + } + } + } +} diff --git a/client/src/test/java/io/github/achrafaittayeb/dtp/client/ClientRetryPolicyTest.java b/client/src/test/java/io/github/achrafaittayeb/dtp/client/ClientRetryPolicyTest.java new file mode 100644 index 0000000..0730766 --- /dev/null +++ b/client/src/test/java/io/github/achrafaittayeb/dtp/client/ClientRetryPolicyTest.java @@ -0,0 +1,70 @@ +package io.github.achrafaittayeb.dtp.client; + +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Unit tests for the retry policy's attempt budget and jittered back-off bounds. */ +class ClientRetryPolicyTest { + + @Test + void noneIsASingleAttempt() { + ClientRetryPolicy policy = ClientRetryPolicy.none(); + assertThat(policy.maxAttempts()).isEqualTo(1); + assertThat(policy.enabled()).isFalse(); + } + + @Test + void retriesAreCountedOnTopOfTheFirstAttempt() { + // --submit-retries 3 -> 4 total attempts. + assertThat(ClientRetryPolicy.ofRetries(3, 100, 1000).maxAttempts()).isEqualTo(4); + assertThat(ClientRetryPolicy.ofRetries(0, 100, 1000).maxAttempts()).isEqualTo(1); + assertThat(ClientRetryPolicy.ofRetries(0, 100, 1000).enabled()).isFalse(); + } + + @Test + void rejectsInvalidArguments() { + assertThatThrownBy(() -> ClientRetryPolicy.ofRetries(-1, 100, 1000)) + .isInstanceOf(IllegalArgumentException.class); + assertThatThrownBy(() -> ClientRetryPolicy.ofRetries(3, 1000, 100)) // max < base + .isInstanceOf(IllegalArgumentException.class); + assertThatThrownBy(() -> ClientRetryPolicy.ofRetries(3, -1, 1000)) + .isInstanceOf(IllegalArgumentException.class); + } + + @Test + void backoffStaysWithinTheJitteredExponentialCeiling() { + ClientRetryPolicy policy = ClientRetryPolicy.ofRetries(10, 100, 5_000); + // For each attempt, the ceiling is min(maxDelay, base * 2^(n-1)); a + // full-jitter delay must land in [0, ceiling] for every random value. + for (int attempt = 1; attempt <= 10; attempt++) { + long ceiling = Math.min(5_000L, 100L << Math.min(attempt - 1, 30)); + for (double r : new double[] {0.0, 0.25, 0.5, 0.999}) { + long delay = policy.backoffMillis(attempt, r); + assertThat(delay).isBetween(0L, ceiling); + } + } + } + + @Test + void backoffGrowsThenSaturatesAtTheCap() { + ClientRetryPolicy policy = ClientRetryPolicy.ofRetries(20, 100, 800); + // With the jitter pinned high, the delay tracks the exponential ceiling + // until it saturates at maxDelay and never exceeds it. + assertThat(policy.backoffMillis(1, 0.999)).isLessThanOrEqualTo(100); + assertThat(policy.backoffMillis(2, 0.999)).isLessThanOrEqualTo(200); + assertThat(policy.backoffMillis(20, 0.999)).isLessThanOrEqualTo(800); + } + + @Test + void rejectsOutOfRangeJitter() { + ClientRetryPolicy policy = ClientRetryPolicy.ofRetries(3, 100, 1000); + assertThatThrownBy(() -> policy.backoffMillis(1, 1.0)) + .isInstanceOf(IllegalArgumentException.class); + assertThatThrownBy(() -> policy.backoffMillis(1, -0.1)) + .isInstanceOf(IllegalArgumentException.class); + assertThatThrownBy(() -> policy.backoffMillis(0, 0.5)) + .isInstanceOf(IllegalArgumentException.class); + } +} diff --git a/client/src/test/java/io/github/achrafaittayeb/dtp/client/SubmitRetrierTest.java b/client/src/test/java/io/github/achrafaittayeb/dtp/client/SubmitRetrierTest.java new file mode 100644 index 0000000..25c9403 --- /dev/null +++ b/client/src/test/java/io/github/achrafaittayeb/dtp/client/SubmitRetrierTest.java @@ -0,0 +1,165 @@ +package io.github.achrafaittayeb.dtp.client; + +import io.github.achrafaittayeb.dtp.common.net.ProtocolException; +import io.github.achrafaittayeb.dtp.common.protocol.SubmitRejected; +import org.junit.jupiter.api.Test; + +import java.io.IOException; +import java.io.InterruptedIOException; +import java.util.ArrayList; +import java.util.List; +import java.util.function.DoubleSupplier; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** + * Deterministic unit tests for the submission retry loop: no sockets, no real + * sleeps. A recording {@link SubmitRetrier.Sleeper} captures the back-off + * delays and a fixed jitter source removes randomness, so both the retry + * decision and the delay bounds are asserted exactly. + */ +class SubmitRetrierTest { + + /** Records requested sleeps instead of performing them. */ + private static final class RecordingSleeper implements SubmitRetrier.Sleeper { + final List delays = new ArrayList<>(); + boolean interruptNext; + + @Override + public void sleep(long millis) throws InterruptedException { + if (interruptNext) { + throw new InterruptedException("test interrupt"); + } + delays.add(millis); + } + } + + private static final DoubleSupplier HALF = () -> 0.5; + + private static SubmitRejectedException overloaded() { + return new SubmitRejectedException(SubmitRejected.overloaded(5, 5)); + } + + private static SubmitRejectedException nonRetryable() { + return new SubmitRejectedException(new SubmitRejected("nope", 5, 5, false)); + } + + @Test + void returnsImmediatelyWhenFirstAttemptSucceeds() throws Exception { + RecordingSleeper sleeper = new RecordingSleeper(); + SubmitRetrier retrier = new SubmitRetrier( + ClientRetryPolicy.ofRetries(3, 10, 100), sleeper, HALF); + + String id = retrier.submit(() -> "job-1"); + + assertThat(id).isEqualTo("job-1"); + assertThat(sleeper.delays).isEmpty(); + } + + @Test + void retriesRetryableRejectionsThenSucceeds() throws Exception { + RecordingSleeper sleeper = new RecordingSleeper(); + SubmitRetrier retrier = new SubmitRetrier( + ClientRetryPolicy.ofRetries(3, 10, 100), sleeper, HALF); + int[] calls = {0}; + + String id = retrier.submit(() -> { + if (++calls[0] < 3) { + throw overloaded(); + } + return "job-2"; + }); + + assertThat(id).isEqualTo("job-2"); + assertThat(calls[0]).isEqualTo(3); // two rejections + one success + assertThat(sleeper.delays).hasSize(2); // one back-off before each retry + assertThat(sleeper.delays).allSatisfy(d -> assertThat(d).isBetween(0L, 100L)); + } + + @Test + void throwsLastRejectionWhenBudgetExhausted() { + RecordingSleeper sleeper = new RecordingSleeper(); + SubmitRetrier retrier = new SubmitRetrier( + ClientRetryPolicy.ofRetries(2, 10, 100), sleeper, HALF); // 3 attempts total + int[] calls = {0}; + + assertThatThrownBy(() -> retrier.submit(() -> { + calls[0]++; + throw overloaded(); + })).isInstanceOf(SubmitRejectedException.class); + + assertThat(calls[0]).isEqualTo(3); // initial + 2 retries + assertThat(sleeper.delays).hasSize(2); // slept only between attempts + } + + @Test + void doesNotRetryNonRetryableRejection() { + RecordingSleeper sleeper = new RecordingSleeper(); + SubmitRetrier retrier = new SubmitRetrier( + ClientRetryPolicy.ofRetries(3, 10, 100), sleeper, HALF); + int[] calls = {0}; + + assertThatThrownBy(() -> retrier.submit(() -> { + calls[0]++; + throw nonRetryable(); + })).isInstanceOf(SubmitRejectedException.class); + + assertThat(calls[0]).isEqualTo(1); // no retry + assertThat(sleeper.delays).isEmpty(); + } + + @Test + void doesNotRetryTransportOrProtocolFailures() { + RecordingSleeper sleeper = new RecordingSleeper(); + SubmitRetrier retrier = new SubmitRetrier( + ClientRetryPolicy.ofRetries(3, 10, 100), sleeper, HALF); + int[] calls = {0}; + + // A plain IOException (e.g. a dropped connection) is ambiguous about + // whether the job was created, so it must propagate unretried. + assertThatThrownBy(() -> retrier.submit(() -> { + calls[0]++; + throw new ProtocolException("boom"); + })).isInstanceOf(IOException.class) + .isNotInstanceOf(SubmitRejectedException.class); + + assertThat(calls[0]).isEqualTo(1); + assertThat(sleeper.delays).isEmpty(); + } + + @Test + void neverRetriesUnderNonePolicy() { + RecordingSleeper sleeper = new RecordingSleeper(); + SubmitRetrier retrier = new SubmitRetrier(ClientRetryPolicy.none(), sleeper, HALF); + int[] calls = {0}; + + assertThatThrownBy(() -> retrier.submit(() -> { + calls[0]++; + throw overloaded(); + })).isInstanceOf(SubmitRejectedException.class); + + assertThat(calls[0]).isEqualTo(1); + assertThat(sleeper.delays).isEmpty(); + } + + @Test + void restoresInterruptAndAbortsWhenBackoffSleepInterrupted() { + RecordingSleeper sleeper = new RecordingSleeper(); + sleeper.interruptNext = true; + SubmitRetrier retrier = new SubmitRetrier( + ClientRetryPolicy.ofRetries(3, 10, 100), sleeper, HALF); + + Throwable thrown = org.assertj.core.api.Assertions.catchThrowable( + () -> retrier.submit(SubmitRetrierTest::overloaded0)); + + assertThat(thrown).isInstanceOf(InterruptedIOException.class); + assertThat(thrown.getCause()).isInstanceOf(SubmitRejectedException.class); + assertThat(Thread.interrupted()).isTrue(); // flag restored (and cleared for other tests) + } + + /** First attempt rejects (retryable), forcing the loop into a back-off sleep. */ + private static String overloaded0() throws IOException { + throw overloaded(); + } +} diff --git a/docs/DESIGN_DECISIONS.md b/docs/DESIGN_DECISIONS.md index ce1200d..9c527ff 100644 --- a/docs/DESIGN_DECISIONS.md +++ b/docs/DESIGN_DECISIONS.md @@ -207,3 +207,46 @@ What this explicitly does **not** buy: it is not fairness (a single global limit with no per-client quota or priority in what gets shed) and it does not raise throughput (the service rate is still the workers' slot ceiling) — the point is *defined* overload behavior, not a faster system. + +## 15. Optional client-side submission retry (opt-in, jittered, rejection-only) + +Decision 14 gives the client a typed, retryable `SUBMIT_REJECTED` and leaves the +back-off decision to the caller. This adds an *optional* client policy that +performs that back-off, without changing the default behavior or the wire. + +- **Off by default; a client policy, not a protocol change.** With no flag the + client makes exactly one attempt and surfaces the rejection, exactly as + before. `--submit-retries N` enables up to `N` retries *after* the first + attempt (`N + 1` total). The coordinator, the protocol, and the benchmark are + untouched — this lives entirely in `CoordinatorClient` and a small + `ClientRetryPolicy` / `SubmitRetrier` pair. +- **Retry the rejection, never an ambiguous failure.** The loop retries *only* a + `SubmitRejectedException` whose wire `retryable` flag is true. A malformed + request, a protocol violation, or any transport `IOException` is propagated on + first sight. The reason is correctness, not caution: a rejection provably + creates and persists nothing (asserted by the admission-control tests), so + resending is safe; a connection that dropped *after* the coordinator accepted + the job is ambiguous, and a blind resend would risk a duplicate submission. + This is why the exception now reads the `retryable` field off the wire instead + of assuming it — the client only retries when the server actually said it was + safe. +- **This is not the coordinator's execution retry.** Decision 5's retries re-run + a job that was *accepted* and whose attempt failed; those never re-enter + admission control. This retries *admission itself* for a job that was never + accepted. Keeping them in separate classes (`RetryPolicy` on the coordinator, + `ClientRetryPolicy` on the client) keeps the two from being conflated. +- **Full jitter, bounded.** Back-off is `random in [0, min(cap, base·2^(n-1))]`. + Full jitter (rather than a fixed exponential schedule) desynchronizes clients + that were all rejected in the same overload spike, so they do not retry in + lockstep and re-create it. A hard attempt cap bounds the worst-case wait to + `(maxAttempts − 1) · cap`. Honestly, retry adds no capacity — under sustained + overload it only shifts where load is shed; jitter and the cap keep it from + amplifying the overload. +- **Back-off holds no lock.** The client synchronizes each wire exchange so one + connection serves one request at a time, but the retry loop sleeps *outside* + that critical section, so a back-off never blocks another thread's exchange on + the same client. Sleep is injectable, so the retry decision and delay bounds + are unit-tested deterministically with no real waiting. +- **Benchmark stays one-shot.** The load generator deliberately does not retry, + so its accepted/rejected counts keep measuring raw admission shedding and stay + comparable across runs. diff --git a/docs/PROTOCOL.md b/docs/PROTOCOL.md index 70d99ba..3bb43b8 100644 --- a/docs/PROTOCOL.md +++ b/docs/PROTOCOL.md @@ -75,6 +75,14 @@ coordinator is overloaded — back off and retry". `retryable` is always `true`; are gated this way; retries of already-accepted jobs never pass through admission control. +Acting on `retryable` is a **client-side** concern, not part of the wire +contract: the reference client honors the flag (it only retries when the server +sets it) and can optionally back off and re-submit on the caller's behalf +(`--submit-retries`, off by default; see the README and design decision 15). The +protocol itself is unchanged — the coordinator sends one reply per request and +has no notion of client retry. A client must never treat an ambiguous transport +or protocol failure as retryable, since a job may already have been created. + A job snapshot contains: `jobId`, `taskType`, `state` (one of QUEUED, RUNNING, RETRY_WAIT, COMPLETED, FAILED, CANCELLED), `attempts`, `maxAttempts`, `workerId` (only while RUNNING), `result`, `error` (last attempt's error, kept diff --git a/integration-tests/src/test/java/io/github/achrafaittayeb/dtp/it/SubmitRetryIT.java b/integration-tests/src/test/java/io/github/achrafaittayeb/dtp/it/SubmitRetryIT.java new file mode 100644 index 0000000..efdd2c8 --- /dev/null +++ b/integration-tests/src/test/java/io/github/achrafaittayeb/dtp/it/SubmitRetryIT.java @@ -0,0 +1,104 @@ +package io.github.achrafaittayeb.dtp.it; + +import com.fasterxml.jackson.databind.node.JsonNodeFactory; +import io.github.achrafaittayeb.dtp.client.ClientRetryPolicy; +import io.github.achrafaittayeb.dtp.client.CoordinatorClient; +import io.github.achrafaittayeb.dtp.client.SubmitRejectedException; +import io.github.achrafaittayeb.dtp.common.model.TaskType; +import io.github.achrafaittayeb.dtp.coordinator.Coordinator; +import io.github.achrafaittayeb.dtp.coordinator.CoordinatorConfig; +import org.junit.jupiter.api.Test; + +import java.io.IOException; +import java.util.concurrent.atomic.AtomicReference; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** + * End-to-end coverage for optional client-side submission retry against a real + * coordinator over the wire (extends the admission-control scenario). All runs + * use a limit of 1 and no worker, so a single queued job pins the coordinator + * at capacity and every further submission is a genuine overload rejection. + */ +class SubmitRetryIT { + + private static String submitSha(CoordinatorClient client) throws IOException { + return client.submit(TaskType.SHA256, + JsonNodeFactory.instance.objectNode().put("text", "x"), 0); + } + + private static CoordinatorClient retryingClient(Coordinator coordinator, ClientRetryPolicy policy) + throws IOException { + return new CoordinatorClient("localhost", coordinator.clientPort(), policy); + } + + @Test + void retryingClientEventuallySucceedsWhenCapacityFrees() throws Exception { + try (Coordinator coordinator = Testbed.startCoordinator( + CoordinatorConfig.IN_MEMORY_DATABASE, 3, Testbed.TASK_TIMEOUT_MILLIS, 1); + CoordinatorClient blocker = Testbed.connectClient(coordinator)) { + + String queued = submitSha(blocker); // fills the only slot (QUEUED) + assertThat(coordinator.activeJobCount()).isEqualTo(1); + + // A retrying client submits while the coordinator is at capacity; it + // keeps re-attempting with back-off long enough for the slot to free. + AtomicReference submittedId = new AtomicReference<>(); + AtomicReference failure = new AtomicReference<>(); + Thread submitter = new Thread(() -> { + try (CoordinatorClient retrying = retryingClient(coordinator, + ClientRetryPolicy.ofRetries(100, 30, 120))) { + submittedId.set(submitSha(retrying)); + } catch (Throwable t) { + failure.set(t); + } + }, "retrying-submitter"); + submitter.start(); + + // Give the submitter time to hit at least one rejection and back off, + // then free the slot by cancelling the queued job. + Thread.sleep(100); + blocker.cancel(queued); + + submitter.join(Testbed.TERMINAL_TIMEOUT.toMillis()); + assertThat(failure.get()).isNull(); + assertThat(submittedId.get()).isNotBlank(); + assertThat(coordinator.activeJobCount()).isEqualTo(1); // the retried job now holds the slot + } + } + + @Test + void retryingClientThrowsTypedRejectionWhenCapacityNeverFrees() throws Exception { + try (Coordinator coordinator = Testbed.startCoordinator( + CoordinatorConfig.IN_MEMORY_DATABASE, 3, Testbed.TASK_TIMEOUT_MILLIS, 1); + CoordinatorClient blocker = Testbed.connectClient(coordinator); + CoordinatorClient retrying = retryingClient(coordinator, + ClientRetryPolicy.ofRetries(2, 10, 40))) { + + submitSha(blocker); // fills the only slot, never freed + assertThat(coordinator.activeJobCount()).isEqualTo(1); + + // Budget is bounded: after the retries are spent the caller still + // sees the typed rejection, and nothing extra was admitted. + assertThatThrownBy(() -> submitSha(retrying)) + .isInstanceOf(SubmitRejectedException.class) + .satisfies(t -> assertThat(((SubmitRejectedException) t).retryable()).isTrue()); + assertThat(coordinator.activeJobCount()).isEqualTo(1); + } + } + + @Test + void defaultClientStillThrowsOnFirstRejection() throws Exception { + try (Coordinator coordinator = Testbed.startCoordinator( + CoordinatorConfig.IN_MEMORY_DATABASE, 3, Testbed.TASK_TIMEOUT_MILLIS, 1); + CoordinatorClient blocker = Testbed.connectClient(coordinator); + CoordinatorClient defaultClient = Testbed.connectClient(coordinator)) { + + submitSha(blocker); + // No retry policy configured → unchanged one-shot behavior. + assertThatThrownBy(() -> submitSha(defaultClient)) + .isInstanceOf(SubmitRejectedException.class); + } + } +}