diff --git a/README.md b/README.md index c5a868c..07cba8c 100644 --- a/README.md +++ b/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 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 dd350b1..4099071 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 @@ -122,9 +122,10 @@ private static void submit(CoordinatorClient client, List 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 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)"""); + --submit-retry-max-millis back-off cap (default 5000) + --idempotency-key dedup key: resubmitting the same key returns + the original job instead of creating a new one"""); } private static final class UsageException extends Exception { 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 c8b3873..0bbba9d 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 @@ -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 different 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(); } diff --git a/common/src/main/java/io/github/achrafaittayeb/dtp/common/protocol/SubmitJob.java b/common/src/main/java/io/github/achrafaittayeb/dtp/common/protocol/SubmitJob.java index 33e374a..ee02e81 100644 --- a/common/src/main/java/io/github/achrafaittayeb/dtp/common/protocol/SubmitJob.java +++ b/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". + * + *

{@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 original 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); + } } diff --git a/common/src/test/java/io/github/achrafaittayeb/dtp/common/protocol/MessageSerializationTest.java b/common/src/test/java/io/github/achrafaittayeb/dtp/common/protocol/MessageSerializationTest.java index 683f8a4..b494d78 100644 --- a/common/src/test/java/io/github/achrafaittayeb/dtp/common/protocol/MessageSerializationTest.java +++ b/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(); + } } diff --git a/coordinator/src/main/java/io/github/achrafaittayeb/dtp/coordinator/CoordinatorCore.java b/coordinator/src/main/java/io/github/achrafaittayeb/dtp/coordinator/CoordinatorCore.java index 0f16a40..fdd45cb 100644 --- a/coordinator/src/main/java/io/github/achrafaittayeb/dtp/coordinator/CoordinatorCore.java +++ b/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 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 jobIdByKey = new HashMap<>(); + private final ScheduledExecutorService coreThread; /** @@ -131,31 +143,84 @@ 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: + * + *

    + *
  1. Deduplication first. If a non-null {@code idempotencyKey} is + * already known, the original job's id is returned. This happens + * before 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.
    2. + *
  2. Admission. 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}.
    3. + *
+ * + *

Deduplication prevents a duplicate logical submission; 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={}", @@ -163,17 +228,35 @@ public SubmitOutcome submitJob(TaskType taskType, JsonNode payload, int requeste 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 diff --git a/coordinator/src/main/java/io/github/achrafaittayeb/dtp/coordinator/job/Job.java b/coordinator/src/main/java/io/github/achrafaittayeb/dtp/coordinator/job/Job.java index 11d01ce..cf6d60b 100644 --- a/coordinator/src/main/java/io/github/achrafaittayeb/dtp/coordinator/job/Job.java +++ b/coordinator/src/main/java/io/github/achrafaittayeb/dtp/coordinator/job/Job.java @@ -23,6 +23,7 @@ public final class Job { private final int maxAttempts; private final long executionTimeoutMillis; private final long createdAtMillis; + private final String idempotencyKey; private JobState state; private int attempts; @@ -34,10 +35,17 @@ public final class Job { private String error; private long updatedAtMillis; + /** Creates a new QUEUED job with no idempotency key. */ public static Job createQueued(TaskType taskType, JsonNode payload, int maxAttempts, long executionTimeoutMillis, long now) { + return createQueued(taskType, payload, maxAttempts, executionTimeoutMillis, now, null); + } + + public static Job createQueued(TaskType taskType, JsonNode payload, int maxAttempts, + long executionTimeoutMillis, long now, String idempotencyKey) { return new Job(UUID.randomUUID().toString(), taskType, payload, maxAttempts, - executionTimeoutMillis, JobState.QUEUED, 0, null, null, 0, 0, null, null, now, now); + executionTimeoutMillis, JobState.QUEUED, 0, null, null, 0, 0, null, null, now, now, + idempotencyKey); } /** Full-field constructor used when rehydrating from persistent storage. */ @@ -45,7 +53,7 @@ public Job(String id, TaskType taskType, JsonNode payload, int maxAttempts, long executionTimeoutMillis, JobState state, int attempts, String currentAttemptId, String assignedWorkerId, long deadlineMillis, long nextEligibleTimeMillis, String result, String error, - long createdAtMillis, long updatedAtMillis) { + long createdAtMillis, long updatedAtMillis, String idempotencyKey) { this.id = id; this.taskType = taskType; this.payload = payload; @@ -61,6 +69,7 @@ public Job(String id, TaskType taskType, JsonNode payload, int maxAttempts, this.error = error; this.createdAtMillis = createdAtMillis; this.updatedAtMillis = updatedAtMillis; + this.idempotencyKey = idempotencyKey; } /** @@ -177,6 +186,11 @@ public String id() { return id; } + /** Client-supplied submission-deduplication key, or {@code null} if none was provided. */ + public String idempotencyKey() { + return idempotencyKey; + } + public TaskType taskType() { return taskType; } diff --git a/coordinator/src/main/java/io/github/achrafaittayeb/dtp/coordinator/job/SqliteJobRepository.java b/coordinator/src/main/java/io/github/achrafaittayeb/dtp/coordinator/job/SqliteJobRepository.java index fab6601..3a8c2f3 100644 --- a/coordinator/src/main/java/io/github/achrafaittayeb/dtp/coordinator/job/SqliteJobRepository.java +++ b/coordinator/src/main/java/io/github/achrafaittayeb/dtp/coordinator/job/SqliteJobRepository.java @@ -32,12 +32,15 @@ public final class SqliteJobRepository implements JobRepository { private static final Logger log = LoggerFactory.getLogger(SqliteJobRepository.class); + // idempotency_key is only ever set on INSERT (it is immutable identity for the + // logical submission); the ON CONFLICT UPDATE path deliberately leaves it alone + // so re-saving a job on each state change never rewrites its key. private static final String UPSERT = """ INSERT INTO jobs (id, task_type, payload, max_attempts, execution_timeout, state, attempts, current_attempt_id, assigned_worker_id, deadline, next_eligible_time, result, error, - created_at, updated_at) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + created_at, updated_at, idempotency_key) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(id) DO UPDATE SET state = excluded.state, attempts = excluded.attempts, @@ -84,13 +87,25 @@ CREATE TABLE IF NOT EXISTS jobs ( result TEXT, error TEXT, created_at INTEGER NOT NULL, - updated_at INTEGER NOT NULL + updated_at INTEGER NOT NULL, + idempotency_key TEXT )"""); } // Databases created before execution deadlines existed lack these two // columns; add them in place so old job history remains loadable. ensureColumn("execution_timeout", "INTEGER NOT NULL DEFAULT 600000"); ensureColumn("deadline", "INTEGER NOT NULL DEFAULT 0"); + // Databases created before idempotent submission lack this column; it is + // nullable, so existing rows migrate to a NULL (un-keyed) key. + ensureColumn("idempotency_key", "TEXT"); + // Durable backstop for deduplication: two different jobs can never share a + // non-null key. The partial predicate lets any number of NULL (un-keyed) + // jobs coexist. The in-core check is the primary guard; this catches bugs. + try (Statement statement = connection.createStatement()) { + statement.execute(""" + CREATE UNIQUE INDEX IF NOT EXISTS idx_jobs_idempotency_key + ON jobs(idempotency_key) WHERE idempotency_key IS NOT NULL"""); + } } private void ensureColumn(String column, String definition) throws SQLException { @@ -126,6 +141,7 @@ public void save(Job job) { statement.setString(13, job.error()); statement.setLong(14, job.createdAtMillis()); statement.setLong(15, job.updatedAtMillis()); + statement.setString(16, job.idempotencyKey()); statement.executeUpdate(); } catch (SQLException | JsonProcessingException e) { // Losing durability silently would break recovery guarantees; fail loudly. @@ -164,7 +180,8 @@ private static Job rowToJob(ResultSet row) throws SQLException, IOException { row.getString("result"), row.getString("error"), row.getLong("created_at"), - row.getLong("updated_at")); + row.getLong("updated_at"), + row.getString("idempotency_key")); } @Override diff --git a/coordinator/src/main/java/io/github/achrafaittayeb/dtp/coordinator/net/ClientServer.java b/coordinator/src/main/java/io/github/achrafaittayeb/dtp/coordinator/net/ClientServer.java index 3bf2b0d..1a912f9 100644 --- a/coordinator/src/main/java/io/github/achrafaittayeb/dtp/coordinator/net/ClientServer.java +++ b/coordinator/src/main/java/io/github/achrafaittayeb/dtp/coordinator/net/ClientServer.java @@ -108,10 +108,14 @@ private Message handle(Message request) { return switch (request) { case SubmitJob submit -> { CoordinatorCore.SubmitOutcome outcome = core.submitJob( - submit.taskType(), submit.payload(), submit.maxAttempts()); - yield outcome.accepted() - ? new JobSubmitted(outcome.jobId()) - : SubmitRejected.overloaded(outcome.activeCount(), outcome.limit()); + submit.taskType(), submit.payload(), submit.maxAttempts(), + submit.idempotencyKey()); + yield switch (outcome.status()) { + case ACCEPTED -> new JobSubmitted(outcome.jobId()); + case REJECTED -> SubmitRejected.overloaded( + outcome.activeCount(), outcome.limit()); + case CONFLICT -> new ErrorReply(outcome.conflictMessage()); + }; } case GetJobStatus status -> core.getJob(status.jobId()) .map(JobStatusReply::new) diff --git a/coordinator/src/test/java/io/github/achrafaittayeb/dtp/coordinator/job/SqliteJobRepositoryTest.java b/coordinator/src/test/java/io/github/achrafaittayeb/dtp/coordinator/job/SqliteJobRepositoryTest.java index fdf520b..3897936 100644 --- a/coordinator/src/test/java/io/github/achrafaittayeb/dtp/coordinator/job/SqliteJobRepositoryTest.java +++ b/coordinator/src/test/java/io/github/achrafaittayeb/dtp/coordinator/job/SqliteJobRepositoryTest.java @@ -92,6 +92,43 @@ INSERT INTO jobs VALUES ('legacy-job', 'SHA256', '{"text":"abc"}', assertThat(loaded.result()).isEqualTo("old-result"); assertThat(loaded.executionTimeoutMillis()).isEqualTo(600_000L); // migration default assertThat(loaded.deadlineMillis()).isZero(); + assertThat(loaded.idempotencyKey()).isNull(); // migrated column defaults to NULL + + // The migrated database is fully usable: a new keyed job saves and loads. + Job keyed = Job.createQueued(TaskType.SHA256, + JsonNodeFactory.instance.objectNode().put("text", "x"), 3, 5_000L, 30L, "key-1"); + migrated.save(keyed); + assertThat(migrated.loadAll()) + .filteredOn(j -> "key-1".equals(j.idempotencyKey())) + .singleElement() + .satisfies(j -> assertThat(j.id()).isEqualTo(keyed.id())); + } + } + + @Test + void persistsAndReloadsIdempotencyKey() { + Job keyed = Job.createQueued(TaskType.WORD_COUNT, + JsonNodeFactory.instance.objectNode().put("text", "a b"), 3, 5_000L, 42L, "submit-key-7"); + + try (SqliteJobRepository repository = new SqliteJobRepository(dbPath())) { + repository.save(keyed); + } + try (SqliteJobRepository reopened = new SqliteJobRepository(dbPath())) { + Job loaded = reopened.loadAll().getFirst(); + assertThat(loaded.idempotencyKey()).isEqualTo("submit-key-7"); + } + } + + @Test + void reSavingAKeyedJobKeepsItsKeyImmutable() { + try (SqliteJobRepository repository = new SqliteJobRepository(dbPath())) { + Job keyed = Job.createQueued(TaskType.SLEEP, + JsonNodeFactory.instance.objectNode().put("durationMillis", 10), 1, 5_000L, 1L, "k"); + repository.save(keyed); + keyed.assignTo("w", 2L); // a later state change re-saves the row + repository.save(keyed); + + assertThat(repository.loadAll().getFirst().idempotencyKey()).isEqualTo("k"); } } diff --git a/docs/DESIGN_DECISIONS.md b/docs/DESIGN_DECISIONS.md index 9c527ff..2022014 100644 --- a/docs/DESIGN_DECISIONS.md +++ b/docs/DESIGN_DECISIONS.md @@ -250,3 +250,53 @@ performs that back-off, without changing the default behavior or the wire. - **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. + +## 16. Durable idempotent submission (optional client key, dedup before admission) + +Decision 15 stops short of retrying *transport* failures because a resend after +an ambiguous drop could create a duplicate job — the coordinator may have +accepted and persisted the job before the connection broke. This decision adds +the missing primitive: an optional, client-supplied idempotency key that makes a +resend return the *same* logical job instead of a new one. + +- **Client-generated, optional, off by default.** The key is the client's, not + the coordinator's: only the client can reconstruct the identical request after + a lost response, so a coordinator-generated key would defeat the purpose. It is + a nullable field on `SUBMIT_JOB`; omitting it preserves today's behavior + exactly (every submission is a fresh job). Jackson decodes a missing field to + `null`, so old and new peers interoperate without a protocol version bump. +- **Dedup before admission.** A known key is resolved before the active-job + check and so bypasses `--max-active-jobs`: it creates no new job, so there is + nothing to admit — the same reasoning by which execution retries never + re-enter admission (decision 14). This does **not** weaken admission: a + genuinely new key is new work and is gated normally. A rejected submission + persists nothing (including no key), so a later resend of that key is correctly + treated as fresh. +- **Race-free by construction.** The `key → jobId` map is core-thread-confined + like all other state, so check-then-create is atomic without locks: the first + of N concurrent duplicates creates the job and the rest dedup to it. A partial + unique index on the SQLite column (`WHERE idempotency_key IS NOT NULL`) is a + durable backstop, not the primary guard. +- **A key stays bound for the job's whole life, including terminal.** Resubmitting + a completed/failed/cancelled key returns that same terminal job rather than + re-running it. This is the honest semantics — a key identifies a *submission*, + not a fresh execution. The corollary is a footgun: a key is "burned" once its + job is terminal; to run again, use a new key. +- **Conflict is an error, not a silent replacement.** Reusing a key with a + different task type, payload, or effective max attempts returns `ERROR` and + leaves the existing job untouched, so a retry that accidentally mutated the + request is caught rather than mapped to the wrong job. This milestone reuses + `ERROR` deliberately rather than growing the protocol with a new type. +- **Persistence rides the existing job row.** The key is one nullable column + added with the same in-place `ensureColumn` migration used for the deadline + columns; recovery rebuilds the in-memory map from the loaded rows. No new + table and no new config flag — the feature is inert unless a key is supplied. +- **What it does and does not buy.** It makes duplicate *logical submission* + preventable, which is the precondition for safely auto-retrying transport + failures — but that retry is intentionally **not** enabled here (it is a + separate, opt-in follow-up). It does **not** provide exactly-once *execution*: + a single logical job can still run more than once under worker failure. And + because the system has no client authentication, keys share one global + namespace — two clients choosing the same string collide. Proper per-client + isolation needs identity the system deliberately does not yet have; until then, + callers should use unguessable, namespaced keys (e.g. UUIDs). diff --git a/docs/PROTOCOL.md b/docs/PROTOCOL.md index 3bb43b8..f06a05e 100644 --- a/docs/PROTOCOL.md +++ b/docs/PROTOCOL.md @@ -58,7 +58,7 @@ connection. | Request | Reply | Notes | |---|---|---| -| `SUBMIT_JOB` (`taskType`, `payload`, `maxAttempts`) | `JOB_SUBMITTED` (`jobId`) or `SUBMIT_REJECTED` | `maxAttempts ≤ 0` → server default; payload validated before the job exists; rejected if the coordinator is at its active-job limit (see below) | +| `SUBMIT_JOB` (`taskType`, `payload`, `maxAttempts`, optional `idempotencyKey`) | `JOB_SUBMITTED` (`jobId`) or `SUBMIT_REJECTED` or `ERROR` | `maxAttempts ≤ 0` → server default; payload validated before the job exists; rejected if the coordinator is at its active-job limit (see below); optional `idempotencyKey` deduplicates submissions (see below) | | `GET_JOB_STATUS` (`jobId`) | `JOB_STATUS` (job snapshot) | unknown id → `ERROR` | | `LIST_JOBS` | `JOB_LIST` (snapshots, newest first) | | | `LIST_WORKERS` | `WORKER_LIST` (live workers) | | @@ -83,6 +83,30 @@ 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. +**`idempotencyKey`** (optional, nullable) on `SUBMIT_JOB` requests durable +submission deduplication. When present, the coordinator remembers the mapping +`key → jobId` (persisted on the job row, so it survives restart). Behavior: + +- **First time seen:** a normal new job is created and the key recorded; reply + is `JOB_SUBMITTED (jobId)`. +- **Seen again, same logical request** (identical `taskType`, `payload`, and + effective `maxAttempts`): the coordinator returns the **original** `jobId` in a + `JOB_SUBMITTED` — indistinguishable from the first success, and it creates no + new job. This holds even after the job has completed, failed, or been + cancelled: the key stays bound to that one job and never triggers a re-run. +- **Seen again, different request** (key reused with a different task type, + payload, or max attempts): `ERROR` — the existing job is left untouched. This + milestone deliberately reuses `ERROR` rather than adding a new response type. +- A known-key duplicate is resolved **before** admission control, so it is + returned even when the coordinator is at `--max-active-jobs` (it adds no active + job). A genuinely new key is normal new work and is admission-controlled. + +Omitting the field (older clients, or callers that don't want dedup) preserves +the historical behavior exactly: every submission creates a fresh job. The field +decodes to `null` when absent, so it is backward-compatible in both directions. +Deduplication concerns **submission identity only** — it does not change +execution semantics, which remain at-least-once. + 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/IdempotentSubmissionIT.java b/integration-tests/src/test/java/io/github/achrafaittayeb/dtp/it/IdempotentSubmissionIT.java new file mode 100644 index 0000000..06a98cb --- /dev/null +++ b/integration-tests/src/test/java/io/github/achrafaittayeb/dtp/it/IdempotentSubmissionIT.java @@ -0,0 +1,275 @@ +package io.github.achrafaittayeb.dtp.it; + +import com.fasterxml.jackson.databind.node.JsonNodeFactory; +import io.github.achrafaittayeb.dtp.client.CoordinatorClient; +import io.github.achrafaittayeb.dtp.client.SubmitRejectedException; +import io.github.achrafaittayeb.dtp.common.model.JobState; +import io.github.achrafaittayeb.dtp.common.model.TaskType; +import io.github.achrafaittayeb.dtp.coordinator.Coordinator; +import io.github.achrafaittayeb.dtp.coordinator.CoordinatorConfig; +import io.github.achrafaittayeb.dtp.worker.Worker; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.io.IOException; +import java.nio.file.Path; +import java.util.Set; +import java.util.concurrent.CopyOnWriteArraySet; +import java.util.concurrent.CountDownLatch; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** + * Durable idempotent submission: a client-supplied key deduplicates logical + * submissions. Deduplication is checked before admission control (a known key + * creates no job, so it bypasses the active-job limit), the mapping survives a + * coordinator restart, and a key stays bound to its job through terminal states. + * This dedups submission only — execution stays at-least-once. + */ +class IdempotentSubmissionIT { + + @TempDir + Path tempDir; + + private static String submit(CoordinatorClient client, String text, int maxAttempts, String key) + throws IOException { + return client.submit(TaskType.SHA256, + JsonNodeFactory.instance.objectNode().put("text", text), maxAttempts, key); + } + + private static String submitShaKeyed(CoordinatorClient client, String key) throws IOException { + return submit(client, "x", 0, key); + } + + @Test + void sameKeyTwiceReturnsSameJobId() throws Exception { + try (Coordinator coordinator = Testbed.startCoordinator( + CoordinatorConfig.IN_MEMORY_DATABASE, 3, Testbed.TASK_TIMEOUT_MILLIS, 100); + CoordinatorClient client = Testbed.connectClient(coordinator)) { + + String first = submitShaKeyed(client, "dup-key"); + String second = submitShaKeyed(client, "dup-key"); + + assertThat(second).isEqualTo(first); + assertThat(client.listJobs()).hasSize(1); + assertThat(coordinator.activeJobCount()).isEqualTo(1); + } + } + + @Test + void submissionsWithoutKeyCreateDistinctJobs() throws Exception { + try (Coordinator coordinator = Testbed.startCoordinator( + CoordinatorConfig.IN_MEMORY_DATABASE, 3, Testbed.TASK_TIMEOUT_MILLIS, 100); + CoordinatorClient client = Testbed.connectClient(coordinator)) { + + String first = submitShaKeyed(client, null); + String second = submitShaKeyed(client, null); + + assertThat(second).isNotEqualTo(first); + assertThat(client.listJobs()).hasSize(2); + assertThat(coordinator.activeJobCount()).isEqualTo(2); + } + } + + @Test + void knownKeyAtAdmissionLimitReturnsOriginalIdWithoutRejection() throws Exception { + try (Coordinator coordinator = Testbed.startCoordinator( + CoordinatorConfig.IN_MEMORY_DATABASE, 3, Testbed.TASK_TIMEOUT_MILLIS, 1); + CoordinatorClient client = Testbed.connectClient(coordinator)) { + + String original = submitShaKeyed(client, "at-limit"); // fills the only slot + assertThat(coordinator.activeJobCount()).isEqualTo(1); + + // Resubmitting the known key bypasses admission (creates no new job). + String again = submitShaKeyed(client, "at-limit"); + assertThat(again).isEqualTo(original); + assertThat(coordinator.activeJobCount()).isEqualTo(1); + assertThat(client.listJobs()).hasSize(1); + } + } + + @Test + void newKeyAtAdmissionLimitIsRejected() throws Exception { + try (Coordinator coordinator = Testbed.startCoordinator( + CoordinatorConfig.IN_MEMORY_DATABASE, 3, Testbed.TASK_TIMEOUT_MILLIS, 1); + CoordinatorClient client = Testbed.connectClient(coordinator)) { + + submitShaKeyed(client, "first"); // fills the only slot + assertThat(coordinator.activeJobCount()).isEqualTo(1); + + // A genuinely new key is new work and still passes through admission. + assertThatThrownBy(() -> submitShaKeyed(client, "second")) + .isInstanceOf(SubmitRejectedException.class); + assertThat(client.listJobs()).hasSize(1); + } + } + + @Test + void sameKeyDifferentPayloadIsConflictAndLeavesOriginalUnchanged() throws Exception { + try (Coordinator coordinator = Testbed.startCoordinator( + CoordinatorConfig.IN_MEMORY_DATABASE, 3, Testbed.TASK_TIMEOUT_MILLIS, 100); + CoordinatorClient client = Testbed.connectClient(coordinator)) { + + String original = submit(client, "original-text", 0, "conflict-key"); + + assertThatThrownBy(() -> submit(client, "different-text", 0, "conflict-key")) + .isInstanceOf(IOException.class) + .isNotInstanceOf(SubmitRejectedException.class) + .hasMessageContaining("Idempotency key"); + + // The original job is untouched: same id, still the only job, unchanged payload. + assertThat(client.listJobs()).hasSize(1); + assertThat(client.status(original).state()).isEqualTo(JobState.QUEUED); + assertThat(coordinator.activeJobCount()).isEqualTo(1); + } + } + + @Test + void sameKeyDifferentTaskTypeIsConflict() throws Exception { + try (Coordinator coordinator = Testbed.startCoordinator( + CoordinatorConfig.IN_MEMORY_DATABASE, 3, Testbed.TASK_TIMEOUT_MILLIS, 100); + CoordinatorClient client = Testbed.connectClient(coordinator)) { + + client.submit(TaskType.SHA256, + JsonNodeFactory.instance.objectNode().put("text", "x"), 0, "type-key"); + + assertThatThrownBy(() -> client.submit(TaskType.WORD_COUNT, + JsonNodeFactory.instance.objectNode().put("text", "x"), 0, "type-key")) + .isInstanceOf(IOException.class) + .isNotInstanceOf(SubmitRejectedException.class) + .hasMessageContaining("Idempotency key"); + assertThat(client.listJobs()).hasSize(1); + } + } + + @Test + void sameKeyDifferentMaxAttemptsIsConflict() throws Exception { + try (Coordinator coordinator = Testbed.startCoordinator( + CoordinatorConfig.IN_MEMORY_DATABASE, 3, Testbed.TASK_TIMEOUT_MILLIS, 100); + CoordinatorClient client = Testbed.connectClient(coordinator)) { + + submit(client, "x", 2, "attempts-key"); + + assertThatThrownBy(() -> submit(client, "x", 5, "attempts-key")) + .isInstanceOf(IOException.class) + .isNotInstanceOf(SubmitRejectedException.class) + .hasMessageContaining("Idempotency key"); + assertThat(client.listJobs()).hasSize(1); + } + } + + @Test + void keyMappingSurvivesRestartAndRecovery() throws Exception { + String database = tempDir.resolve("idempotency-recovery.db").toString(); + + String originalId; + try (Coordinator first = Testbed.startCoordinator(database, 3, Testbed.TASK_TIMEOUT_MILLIS, 100)) { + try (CoordinatorClient client = Testbed.connectClient(first)) { + originalId = submitShaKeyed(client, "persist-key"); // no worker → stays QUEUED + } + first.close(); + } + + // Same database, fresh coordinator: the dedup map is rebuilt from storage. + try (Coordinator second = Testbed.startCoordinator(database, 3, Testbed.TASK_TIMEOUT_MILLIS, 100)) { + try (CoordinatorClient client = Testbed.connectClient(second)) { + String afterRestart = submitShaKeyed(client, "persist-key"); + assertThat(afterRestart).isEqualTo(originalId); // deduped to the pre-restart job + assertThat(client.listJobs()).hasSize(1); + } + } + } + + @Test + void completedJobStillDeduplicates() throws Exception { + try (Coordinator coordinator = Testbed.startCoordinator( + CoordinatorConfig.IN_MEMORY_DATABASE, 3, Testbed.TASK_TIMEOUT_MILLIS, 100); + Worker worker = Testbed.startWorker(coordinator, "worker-1", 2); + CoordinatorClient client = Testbed.connectClient(coordinator)) { + + String id = submitShaKeyed(client, "done-key"); + assertThat(client.awaitTerminal(id, Testbed.POLL, Testbed.TERMINAL_TIMEOUT).state()) + .isEqualTo(JobState.COMPLETED); + assertThat(coordinator.activeJobCount()).isZero(); + + // Resubmitting the key returns the completed job — it does NOT re-run. + String again = submitShaKeyed(client, "done-key"); + assertThat(again).isEqualTo(id); + assertThat(client.status(again).state()).isEqualTo(JobState.COMPLETED); + assertThat(coordinator.activeJobCount()).isZero(); + assertThat(client.listJobs()).hasSize(1); + } + } + + @Test + void cancelledJobStillDeduplicates() throws Exception { + try (Coordinator coordinator = Testbed.startCoordinator( + CoordinatorConfig.IN_MEMORY_DATABASE, 3, Testbed.TASK_TIMEOUT_MILLIS, 100); + CoordinatorClient client = Testbed.connectClient(coordinator)) { + + String id = submitShaKeyed(client, "cancel-key"); // no worker → QUEUED + client.cancel(id); + assertThat(client.status(id).state()).isEqualTo(JobState.CANCELLED); + + String again = submitShaKeyed(client, "cancel-key"); + assertThat(again).isEqualTo(id); + assertThat(client.status(again).state()).isEqualTo(JobState.CANCELLED); + assertThat(client.listJobs()).hasSize(1); + } + } + + @Test + void failedJobStillDeduplicates() throws Exception { + try (Coordinator coordinator = Testbed.startCoordinator( + CoordinatorConfig.IN_MEMORY_DATABASE, 1, Testbed.TASK_TIMEOUT_MILLIS, 100); + Worker worker = Testbed.startWorker(coordinator, "worker-1", 2); + CoordinatorClient client = Testbed.connectClient(coordinator)) { + + String id = client.submit(TaskType.FAIL, + JsonNodeFactory.instance.objectNode().put("failUntilAttempt", 999), 0, "fail-key"); + assertThat(client.awaitTerminal(id, Testbed.POLL, Testbed.TERMINAL_TIMEOUT).state()) + .isEqualTo(JobState.FAILED); + + String again = client.submit(TaskType.FAIL, + JsonNodeFactory.instance.objectNode().put("failUntilAttempt", 999), 0, "fail-key"); + assertThat(again).isEqualTo(id); + assertThat(client.status(again).state()).isEqualTo(JobState.FAILED); + assertThat(client.listJobs()).hasSize(1); + } + } + + @Test + void concurrentDuplicateSubmissionsCreateExactlyOneJob() throws Exception { + int clients = 8; + Set returnedIds = new CopyOnWriteArraySet<>(); + Set failures = new CopyOnWriteArraySet<>(); + + try (Coordinator coordinator = Testbed.startCoordinator( + CoordinatorConfig.IN_MEMORY_DATABASE, 3, Testbed.TASK_TIMEOUT_MILLIS, 100)) { + + CountDownLatch startGate = new CountDownLatch(1); + java.util.List threads = new java.util.ArrayList<>(); + for (int i = 0; i < clients; i++) { + Thread thread = new Thread(() -> { + try (CoordinatorClient client = Testbed.connectClient(coordinator)) { + startGate.await(); + returnedIds.add(submitShaKeyed(client, "race-key")); + } catch (Throwable t) { + failures.add(t); + } + }, "submitter-" + i); + threads.add(thread); + thread.start(); + } + startGate.countDown(); + for (Thread thread : threads) { + thread.join(); + } + + assertThat(failures).isEmpty(); + assertThat(returnedIds).hasSize(1); // every client saw the same single job + assertThat(coordinator.activeJobCount()).isEqualTo(1); + } + } +}