From e4894633061e4872fb1c2b5bf6caa023828e0061 Mon Sep 17 00:00:00 2001 From: Achraf Ait Tayeb <2023521460132@stu.scu.edu.cn> Date: Mon, 14 Sep 2026 14:17:02 +0800 Subject: [PATCH] feat: add persistent priority scheduling --- README.md | 30 +++- .../achrafaittayeb/dtp/client/ClientMain.java | 28 ++- .../dtp/client/CoordinatorClient.java | 21 ++- .../dtp/client/ClientPriorityOptionTest.java | 43 +++++ .../dtp/common/model/JobSnapshot.java | 26 ++- .../dtp/common/model/TaskPriority.java | 66 +++++++ .../dtp/common/protocol/SubmitJob.java | 23 ++- .../dtp/common/model/TaskPriorityTest.java | 48 +++++ .../protocol/MessageSerializationTest.java | 95 +++++++++- .../dtp/coordinator/Coordinator.java | 12 +- .../dtp/coordinator/CoordinatorConfig.java | 13 ++ .../dtp/coordinator/CoordinatorCore.java | 140 +++++++++++++-- .../dtp/coordinator/job/Job.java | 71 +++++++- .../coordinator/job/SqliteJobRepository.java | 30 +++- .../dtp/coordinator/net/ClientServer.java | 2 +- .../dtp/coordinator/net/WorkerServer.java | 2 +- .../coordinator/CoordinatorConfigTest.java | 28 +++ .../CoordinatorCoreSubmissionTest.java | 169 ++++++++++++++++++ .../CoordinatorSchedulingOrderTest.java | 140 +++++++++++++++ .../dtp/coordinator/job/JobPriorityTest.java | 106 +++++++++++ .../job/SqliteJobRepositoryTest.java | 51 +++++- docs/ARCHITECTURE.md | 24 ++- docs/DESIGN_DECISIONS.md | 57 ++++++ docs/FAILURE_MODEL.md | 1 + docs/PROTOCOL.md | 35 +++- .../dtp/it/CoordinatorRestartIT.java | 2 +- .../achrafaittayeb/dtp/it/MutableClock.java | 30 ++++ .../dtp/it/PriorityRecoveryIT.java | 81 +++++++++ .../dtp/it/PrioritySchedulingIT.java | 147 +++++++++++++++ .../github/achrafaittayeb/dtp/it/Testbed.java | 37 +++- 30 files changed, 1495 insertions(+), 63 deletions(-) create mode 100644 client/src/test/java/io/github/achrafaittayeb/dtp/client/ClientPriorityOptionTest.java create mode 100644 common/src/main/java/io/github/achrafaittayeb/dtp/common/model/TaskPriority.java create mode 100644 common/src/test/java/io/github/achrafaittayeb/dtp/common/model/TaskPriorityTest.java create mode 100644 coordinator/src/test/java/io/github/achrafaittayeb/dtp/coordinator/CoordinatorCoreSubmissionTest.java create mode 100644 coordinator/src/test/java/io/github/achrafaittayeb/dtp/coordinator/CoordinatorSchedulingOrderTest.java create mode 100644 coordinator/src/test/java/io/github/achrafaittayeb/dtp/coordinator/job/JobPriorityTest.java create mode 100644 integration-tests/src/test/java/io/github/achrafaittayeb/dtp/it/MutableClock.java create mode 100644 integration-tests/src/test/java/io/github/achrafaittayeb/dtp/it/PriorityRecoveryIT.java create mode 100644 integration-tests/src/test/java/io/github/achrafaittayeb/dtp/it/PrioritySchedulingIT.java diff --git a/README.md b/README.md index 07cba8c..6541333 100644 --- a/README.md +++ b/README.md @@ -139,6 +139,18 @@ Task types: `sleep --milliseconds N`, `word-count --text "..."`, `fail --fail-until-attempt N` (deliberately fails early attempts, for watching retries). Run any command without arguments to see usage. +**Priority.** `submit` accepts `--priority high|normal|low` (case-insensitive, +default `normal`). Higher-priority jobs are assigned ahead of lower-priority +ones while they wait in the queue; equal priorities keep FIFO order; and a job's +effective priority rises the longer it waits (every `--aging-step-millis` on the +coordinator, default 60 s, capped at `high`) so low-priority work never starves. +Priority is shown by `status` and `list`. + +```bash +./scripts/coordinator.sh --aging-step-millis 30000 # promote a level every 30s +./scripts/client.sh submit sha256 --text "urgent" --priority high +``` + **Execution deadlines.** Each attempt has a coordinator-enforced execution deadline (`--task-timeout-millis`, default 10 minutes). A task that outlives it is treated like a lost attempt — the lease is revoked, the worker is asked to @@ -376,6 +388,20 @@ containers. immediately and ask the worker to stop via thread interruption; a task that ignores interruption keeps running but can no longer affect job state. This is not guaranteed preemption, and it does not make execution exactly-once. +- **Priority orders the queue; it does not preempt.** A submission carries a + base priority — `HIGH`, `NORMAL`, or `LOW` (`--priority`, default `NORMAL`; + omitting it is identical to `NORMAL`). Among `QUEUED` jobs the scheduler picks + the highest *effective* level first, breaking ties by a persistent monotonic + submission sequence so equal-priority jobs run in true FIFO order — even when + submitted within the same millisecond. To prevent starvation, a job's + effective level rises by one for every whole `--aging-step-millis` (default + 60 s) it has waited since submission, capped at `HIGH`; an old `LOW` job thus + eventually competes as `HIGH` and, having a lower sequence, runs ahead of + newer `HIGH` submissions. Priority never touches admission control, worker + selection, retries, deadlines, cancellation, or running work — it only orders + what is still queued, and there is no preemption. A retry keeps the job's + original priority, sequence, and creation time. Priority and sequence are + persisted, so ordering is recovered exactly after a coordinator restart. - **Overload is shed, not absorbed.** Beyond `--max-active-jobs` the coordinator refuses new submissions with a typed, retryable `SUBMIT_REJECTED` rather than queueing them. This bounds the coordinator's memory and the @@ -396,7 +422,9 @@ More detail in [docs/FAILURE_MODEL.md](docs/FAILURE_MODEL.md) and - Plaintext TCP with no authentication — production use would need TLS and worker authentication. - One coordinator (see above); leader election is the natural next step. -- Scheduling is FIFO / least-loaded; no priorities, deadlines, or fairness. +- Scheduling is priority-ordered with FIFO tiebreaking and age-based promotion + (see below); it is not preemptive — priority only orders jobs that are still + QUEUED, never work already running on a worker. - 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. The client surfaces 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 4099071..556f83a 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 @@ -3,6 +3,7 @@ import com.fasterxml.jackson.databind.node.JsonNodeFactory; import com.fasterxml.jackson.databind.node.ObjectNode; import io.github.achrafaittayeb.dtp.common.model.JobSnapshot; +import io.github.achrafaittayeb.dtp.common.model.TaskPriority; import io.github.achrafaittayeb.dtp.common.model.TaskType; import io.github.achrafaittayeb.dtp.common.model.WorkerSnapshot; import io.github.achrafaittayeb.dtp.common.util.Args; @@ -123,9 +124,16 @@ private static void submit(CoordinatorClient client, List positionals, A int submitRetries = options.getInt("submit-retries", 0); String idempotencyKey = options.get("idempotency-key", null); + TaskPriority priority; + try { + priority = TaskPriority.fromString(options.get("priority", null)); + } catch (IllegalArgumentException invalid) { + throw new UsageException(invalid.getMessage()); + } String jobId; try { - jobId = client.submit(taskType, payload, options.getInt("max-attempts", 0), idempotencyKey); + jobId = client.submit(taskType, payload, options.getInt("max-attempts", 0), + idempotencyKey, priority); } catch (SubmitRejectedException rejected) { System.err.println("Submission rejected (coordinator overloaded): " + rejected.getMessage()); System.err.println(" Active jobs: " + rejected.activeCount() @@ -141,9 +149,10 @@ private static void submit(CoordinatorClient client, List positionals, A return; } System.out.println("Job submitted"); - System.out.println(" ID: " + jobId); - System.out.println(" Type: " + taskType); - System.out.println(" State: QUEUED"); + System.out.println(" ID: " + jobId); + System.out.println(" Type: " + taskType); + System.out.println(" Priority: " + priority); + System.out.println(" State: QUEUED"); } private static void bench(String host, int port, Args options) throws Exception { @@ -172,6 +181,7 @@ private static void bench(String host, int port, Args options) throws Exception private static void printJob(JobSnapshot job) { System.out.println("Job " + job.jobId()); System.out.println(" Type: " + job.taskType()); + System.out.println(" Priority: " + job.priorityOrDefault()); System.out.println(" State: " + job.state()); System.out.println(" Attempt: " + job.attempts() + "/" + job.maxAttempts()); if (job.workerId() != null) { @@ -192,11 +202,11 @@ private static void listJobs(CoordinatorClient client) throws Exception { System.out.println("No jobs"); return; } - System.out.printf("%-36s %-11s %-10s %-7s %-10s%n", - "JOB ID", "TYPE", "STATE", "ATTEMPT", "WORKER"); + System.out.printf("%-36s %-11s %-8s %-10s %-7s %-10s%n", + "JOB ID", "TYPE", "PRIORITY", "STATE", "ATTEMPT", "WORKER"); for (JobSnapshot job : jobs) { - System.out.printf("%-36s %-11s %-10s %-7s %-10s%n", - job.jobId(), job.taskType(), job.state(), + System.out.printf("%-36s %-11s %-8s %-10s %-7s %-10s%n", + job.jobId(), job.taskType(), job.priorityOrDefault(), job.state(), job.attempts() + "/" + job.maxAttempts(), job.workerId() == null ? "-" : job.workerId()); } @@ -291,6 +301,8 @@ private static void printUsage() { [--prime-limit ] [--wait-timeout-millis ] Global options: --host (default localhost), --port (default 7071) Submit options: --max-attempts (default: coordinator setting) + --priority high|normal|low base scheduling priority (default normal; + case-insensitive) --submit-retries retries after the first attempt on a retryable overload rejection (default 0 = off) --submit-retry-base-millis base back-off (default 200) 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 0bbba9d..ab42b7b 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 @@ -2,6 +2,7 @@ import com.fasterxml.jackson.databind.JsonNode; import io.github.achrafaittayeb.dtp.common.model.JobSnapshot; +import io.github.achrafaittayeb.dtp.common.model.TaskPriority; import io.github.achrafaittayeb.dtp.common.model.TaskType; import io.github.achrafaittayeb.dtp.common.model.WorkerSnapshot; import io.github.achrafaittayeb.dtp.common.net.MessageIO; @@ -94,13 +95,27 @@ public String submit(TaskType taskType, JsonNode payload, int maxAttempts) */ public String submit(TaskType taskType, JsonNode payload, int maxAttempts, String idempotencyKey) throws IOException { - return submitRetrier.submit(() -> submitOnce(taskType, payload, maxAttempts, idempotencyKey)); + return submit(taskType, payload, maxAttempts, idempotencyKey, null); + } + + /** + * Submits with an optional idempotency key and an optional base priority. A + * {@code null} priority is sent as-is and the coordinator treats it as + * {@link TaskPriority#NORMAL}, so omitting priority is indistinguishable from + * an explicit NORMAL. Priority is part of the idempotent-submission identity: + * reusing a key with a different priority is a conflict. + */ + public String submit(TaskType taskType, JsonNode payload, int maxAttempts, String idempotencyKey, + TaskPriority priority) throws IOException { + return submitRetrier.submit( + () -> submitOnce(taskType, payload, maxAttempts, idempotencyKey, priority)); } /** 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)); + String idempotencyKey, TaskPriority priority) + throws IOException { + Message reply = exchange(new SubmitJob(taskType, payload, maxAttempts, idempotencyKey, priority)); if (reply instanceof JobSubmitted submitted) { return submitted.jobId(); } diff --git a/client/src/test/java/io/github/achrafaittayeb/dtp/client/ClientPriorityOptionTest.java b/client/src/test/java/io/github/achrafaittayeb/dtp/client/ClientPriorityOptionTest.java new file mode 100644 index 0000000..0f1c478 --- /dev/null +++ b/client/src/test/java/io/github/achrafaittayeb/dtp/client/ClientPriorityOptionTest.java @@ -0,0 +1,43 @@ +package io.github.achrafaittayeb.dtp.client; + +import io.github.achrafaittayeb.dtp.common.model.TaskPriority; +import io.github.achrafaittayeb.dtp.common.util.Args; +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** + * Locks the CLI's {@code --priority} contract: the flag name, the default when + * omitted, and the validation the {@code submit} command applies. It exercises + * the exact expression {@code ClientMain} uses to turn the option into a + * {@link TaskPriority}. + */ +class ClientPriorityOptionTest { + + private static TaskPriority parsePriority(String... args) { + return TaskPriority.fromString(Args.parse(args).get("priority", null)); + } + + @Test + void acceptsValidPriorityValuesCaseInsensitively() { + assertThat(parsePriority("--priority", "high")).isEqualTo(TaskPriority.HIGH); + assertThat(parsePriority("--priority", "HIGH")).isEqualTo(TaskPriority.HIGH); + assertThat(parsePriority("--priority", "Low")).isEqualTo(TaskPriority.LOW); + assertThat(parsePriority("--priority", "normal")).isEqualTo(TaskPriority.NORMAL); + } + + @Test + void defaultsToNormalWhenPriorityOmitted() { + assertThat(parsePriority()).isEqualTo(TaskPriority.NORMAL); + assertThat(parsePriority("--max-attempts", "2")).isEqualTo(TaskPriority.NORMAL); + } + + @Test + void invalidPriorityProducesAClearError() { + assertThatThrownBy(() -> parsePriority("--priority", "urgent")) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("urgent") + .hasMessageContaining("high, normal, low"); + } +} diff --git a/common/src/main/java/io/github/achrafaittayeb/dtp/common/model/JobSnapshot.java b/common/src/main/java/io/github/achrafaittayeb/dtp/common/model/JobSnapshot.java index 8d7d15a..de1bf2f 100644 --- a/common/src/main/java/io/github/achrafaittayeb/dtp/common/model/JobSnapshot.java +++ b/common/src/main/java/io/github/achrafaittayeb/dtp/common/model/JobSnapshot.java @@ -3,6 +3,12 @@ /** * Immutable, client-facing view of a job's current state. This is what status * queries return; it never exposes coordinator-internal objects. + * + *

{@code priority} is the job's base (submitted) priority. Effective (aged) + * priority is intentionally not exposed: it changes with wall-clock time, so + * surfacing it in a point-in-time snapshot would be misleading. A {@code null} + * priority decodes as {@link TaskPriority#NORMAL} for compatibility with older + * peers that predate the field. */ public record JobSnapshot( String jobId, @@ -14,5 +20,23 @@ public record JobSnapshot( String result, String error, long createdAtMillis, - long updatedAtMillis) { + long updatedAtMillis, + TaskPriority priority) { + + /** + * Back-compatible constructor for callers that predate the priority field; + * priority defaults to {@code null}, which reads as {@link TaskPriority#NORMAL} + * via {@link #priorityOrDefault()}. Keeps existing call sites source-compatible. + */ + public JobSnapshot(String jobId, TaskType taskType, JobState state, int attempts, int maxAttempts, + String workerId, String result, String error, + long createdAtMillis, long updatedAtMillis) { + this(jobId, taskType, state, attempts, maxAttempts, workerId, result, error, + createdAtMillis, updatedAtMillis, null); + } + + /** Normalized base priority: {@code null} (from an older peer) reads as {@link TaskPriority#NORMAL}. */ + public TaskPriority priorityOrDefault() { + return TaskPriority.normalize(priority); + } } diff --git a/common/src/main/java/io/github/achrafaittayeb/dtp/common/model/TaskPriority.java b/common/src/main/java/io/github/achrafaittayeb/dtp/common/model/TaskPriority.java new file mode 100644 index 0000000..7ca4789 --- /dev/null +++ b/common/src/main/java/io/github/achrafaittayeb/dtp/common/model/TaskPriority.java @@ -0,0 +1,66 @@ +package io.github.achrafaittayeb.dtp.common.model; + +import java.util.Locale; + +/** + * Base scheduling priority a client attaches to a submission. Exactly three + * levels are exposed on the wire — the protocol deliberately carries these + * names, never raw numbers, so the public contract cannot drift as internal + * scheduling math changes. + * + *

Each level has an integer base level that the coordinator's + * scheduler uses as the starting point for age-based promotion: + * {@code HIGH=2}, {@code NORMAL=1}, {@code LOW=0}. Effective (aged) level is + * derived by the coordinator at scheduling time and is never persisted or put + * on the wire — see {@code Job#effectiveLevel} and {@code CoordinatorCore}. + * + *

Default: a missing priority — an old client, an old persisted row, + * or any call site that omits it — always means {@link #NORMAL}. Use + * {@link #normalize(TaskPriority)} to collapse {@code null} to {@code NORMAL} + * so "omitted" and "explicit NORMAL" are indistinguishable everywhere. + */ +public enum TaskPriority { + /** Lowest base level (0); still eligible for aging promotion. */ + LOW(0), + /** Default base level (1); the level a missing/omitted priority maps to. */ + NORMAL(1), + /** Highest base level (2); also the ceiling that aging can promote toward. */ + HIGH(2); + + /** Ceiling for effective level: no job ever schedules above {@link #HIGH}. */ + public static final int MAX_LEVEL = 2; + + private final int baseLevel; + + TaskPriority(int baseLevel) { + this.baseLevel = baseLevel; + } + + /** Scheduler starting level before any age-based promotion is applied. */ + public int baseLevel() { + return baseLevel; + } + + /** Collapses a nullable priority to a concrete one: {@code null} → {@link #NORMAL}. */ + public static TaskPriority normalize(TaskPriority priority) { + return priority == null ? NORMAL : priority; + } + + /** + * Parses a case-insensitive priority name ({@code high}/{@code normal}/{@code low}). + * A {@code null} or blank input defaults to {@link #NORMAL}; any other + * unrecognized value throws {@link IllegalArgumentException} with guidance. + */ + public static TaskPriority fromString(String value) { + if (value == null || value.isBlank()) { + return NORMAL; + } + return switch (value.trim().toLowerCase(Locale.ROOT)) { + case "high" -> HIGH; + case "normal" -> NORMAL; + case "low" -> LOW; + default -> throw new IllegalArgumentException( + "Invalid priority '" + value + "'; expected one of: high, normal, low"); + }; + } +} 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 ee02e81..5b29cb6 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 @@ -1,6 +1,7 @@ package io.github.achrafaittayeb.dtp.common.protocol; import com.fasterxml.jackson.databind.JsonNode; +import io.github.achrafaittayeb.dtp.common.model.TaskPriority; import io.github.achrafaittayeb.dtp.common.model.TaskType; /** @@ -12,15 +13,25 @@ * 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. + * + *

{@code priority} is also optional (nullable). A {@code null} priority + * decodes and executes as {@link TaskPriority#NORMAL}, so old clients and older + * peers that omit the field are unchanged. Both optional fields are deliberately + * trailing with delegating constructors so existing call sites stay source- + * compatible and older wire peers (which omit them, decoding to {@code null}) + * stay wire-compatible. */ -public record SubmitJob(TaskType taskType, JsonNode payload, int maxAttempts, String idempotencyKey) +public record SubmitJob(TaskType taskType, JsonNode payload, int maxAttempts, + String idempotencyKey, TaskPriority priority) implements Message { - /** Back-compatible constructor for a submission without an idempotency key. */ + /** Back-compatible constructor for a submission without a key or priority. */ public SubmitJob(TaskType taskType, JsonNode payload, int maxAttempts) { - this(taskType, payload, maxAttempts, null); + this(taskType, payload, maxAttempts, null, null); + } + + /** Back-compatible constructor for a keyed submission without an explicit priority. */ + public SubmitJob(TaskType taskType, JsonNode payload, int maxAttempts, String idempotencyKey) { + this(taskType, payload, maxAttempts, idempotencyKey, null); } } diff --git a/common/src/test/java/io/github/achrafaittayeb/dtp/common/model/TaskPriorityTest.java b/common/src/test/java/io/github/achrafaittayeb/dtp/common/model/TaskPriorityTest.java new file mode 100644 index 0000000..a4d8940 --- /dev/null +++ b/common/src/test/java/io/github/achrafaittayeb/dtp/common/model/TaskPriorityTest.java @@ -0,0 +1,48 @@ +package io.github.achrafaittayeb.dtp.common.model; + +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +class TaskPriorityTest { + + @Test + void baseLevelsAreHighestForHigh() { + assertThat(TaskPriority.HIGH.baseLevel()).isEqualTo(2); + assertThat(TaskPriority.NORMAL.baseLevel()).isEqualTo(1); + assertThat(TaskPriority.LOW.baseLevel()).isEqualTo(0); + assertThat(TaskPriority.MAX_LEVEL).isEqualTo(2); + assertThat(TaskPriority.HIGH.baseLevel()).isEqualTo(TaskPriority.MAX_LEVEL); + } + + @Test + void normalizeMapsNullToNormal() { + assertThat(TaskPriority.normalize(null)).isEqualTo(TaskPriority.NORMAL); + assertThat(TaskPriority.normalize(TaskPriority.HIGH)).isEqualTo(TaskPriority.HIGH); + assertThat(TaskPriority.normalize(TaskPriority.LOW)).isEqualTo(TaskPriority.LOW); + } + + @Test + void fromStringIsCaseInsensitive() { + assertThat(TaskPriority.fromString("high")).isEqualTo(TaskPriority.HIGH); + assertThat(TaskPriority.fromString("HIGH")).isEqualTo(TaskPriority.HIGH); + assertThat(TaskPriority.fromString("Normal")).isEqualTo(TaskPriority.NORMAL); + assertThat(TaskPriority.fromString(" low ")).isEqualTo(TaskPriority.LOW); + } + + @Test + void fromStringDefaultsToNormalWhenNullOrBlank() { + assertThat(TaskPriority.fromString(null)).isEqualTo(TaskPriority.NORMAL); + assertThat(TaskPriority.fromString("")).isEqualTo(TaskPriority.NORMAL); + assertThat(TaskPriority.fromString(" ")).isEqualTo(TaskPriority.NORMAL); + } + + @Test + void fromStringRejectsUnknownValueWithGuidance() { + assertThatThrownBy(() -> TaskPriority.fromString("urgent")) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("urgent") + .hasMessageContaining("high, normal, low"); + } +} 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 b494d78..4937760 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 @@ -4,6 +4,7 @@ import com.fasterxml.jackson.databind.node.ObjectNode; import io.github.achrafaittayeb.dtp.common.model.JobSnapshot; import io.github.achrafaittayeb.dtp.common.model.JobState; +import io.github.achrafaittayeb.dtp.common.model.TaskPriority; import io.github.achrafaittayeb.dtp.common.model.TaskType; import io.github.achrafaittayeb.dtp.common.net.MessageIO; import io.github.achrafaittayeb.dtp.common.net.ProtocolException; @@ -45,7 +46,7 @@ void roundTripsTaskAssignWithPayload() throws IOException { @Test void roundTripsClientMessages() throws IOException { JobSnapshot snapshot = new JobSnapshot("job-1", TaskType.SHA256, JobState.COMPLETED, - 1, 3, "worker-2", "abc123", null, 100L, 200L); + 1, 3, "worker-2", "abc123", null, 100L, 200L, TaskPriority.HIGH); assertThat(roundTrip(new JobStatusReply(snapshot))) .isEqualTo(new JobStatusReply(snapshot)); assertThat(roundTrip(new JobListReply(List.of(snapshot)))) @@ -60,7 +61,7 @@ void roundTripsCancellationMessages() throws IOException { .isEqualTo(new CancelJob("job-1")); JobSnapshot cancelled = new JobSnapshot("job-1", TaskType.SLEEP, JobState.CANCELLED, - 1, 3, null, null, "cancelled by client request", 100L, 200L); + 1, 3, null, null, "cancelled by client request", 100L, 200L, TaskPriority.NORMAL); assertThat(roundTrip(new JobCancelReply(true, cancelled))) .isEqualTo(new JobCancelReply(true, cancelled)); } @@ -127,4 +128,94 @@ void decodesLegacySubmitJobWithoutIdempotencyKeyField() throws IOException { assertThat(decoded.maxAttempts()).isZero(); assertThat(decoded.idempotencyKey()).isNull(); } + + @Test + void roundTripsSubmitJobWithExplicitPriority() throws IOException { + ObjectNode payload = JsonNodeFactory.instance.objectNode().put("text", "abc"); + SubmitJob highPriority = new SubmitJob(TaskType.SHA256, payload, 2, "key-1", TaskPriority.HIGH); + + SubmitJob decoded = (SubmitJob) roundTrip(highPriority); + assertThat(decoded).isEqualTo(highPriority); + assertThat(decoded.priority()).isEqualTo(TaskPriority.HIGH); + } + + @Test + void explicitPriorityIsPresentInEncodedJson() throws IOException { + ObjectNode payload = JsonNodeFactory.instance.objectNode().put("text", "abc"); + String json = new String( + MessageIO.encode(new SubmitJob(TaskType.SHA256, payload, 0, null, TaskPriority.LOW)), + StandardCharsets.UTF_8); + assertThat(json).contains("\"priority\":\"LOW\""); + } + + @Test + void submitJobWithoutPriorityDecodesToNullThenNormalizesToNormal() throws IOException { + ObjectNode payload = JsonNodeFactory.instance.objectNode().put("text", "abc"); + + // The two back-compatible constructors both omit an explicit priority. + SubmitJob noPriority = new SubmitJob(TaskType.SHA256, payload, 0); + SubmitJob decoded = (SubmitJob) roundTrip(noPriority); + assertThat(decoded.priority()).isNull(); + assertThat(TaskPriority.normalize(decoded.priority())).isEqualTo(TaskPriority.NORMAL); + } + + @Test + void decodesLegacySubmitJobWithoutPriorityField() throws IOException { + // A frame from a client that predates the priority field must still decode, + // with priority absent (null) and normalizing to NORMAL. + byte[] legacy = ("{\"type\":\"SUBMIT_JOB\",\"taskType\":\"SHA256\"," + + "\"payload\":{\"text\":\"abc\"},\"maxAttempts\":0,\"idempotencyKey\":\"k\"}") + .getBytes(StandardCharsets.UTF_8); + SubmitJob decoded = (SubmitJob) MessageIO.decode(legacy); + assertThat(decoded.idempotencyKey()).isEqualTo("k"); + assertThat(decoded.priority()).isNull(); + assertThat(TaskPriority.normalize(decoded.priority())).isEqualTo(TaskPriority.NORMAL); + } + + @Test + void omittedPriorityAndExplicitNormalAreWireEquivalent() throws IOException { + ObjectNode payload = JsonNodeFactory.instance.objectNode().put("text", "abc"); + SubmitJob omitted = (SubmitJob) roundTrip(new SubmitJob(TaskType.SHA256, payload, 0)); + SubmitJob explicitNormal = (SubmitJob) roundTrip( + new SubmitJob(TaskType.SHA256, payload, 0, null, TaskPriority.NORMAL)); + + assertThat(TaskPriority.normalize(omitted.priority())) + .isEqualTo(TaskPriority.normalize(explicitNormal.priority())) + .isEqualTo(TaskPriority.NORMAL); + } + + @Test + void decodesLegacyJobSnapshotWithoutPriorityField() throws IOException { + // A JOB_STATUS reply from a peer that predates the snapshot priority field + // must still decode; priority is absent (null) and reads as NORMAL. + byte[] legacy = ("{\"type\":\"JOB_STATUS\",\"job\":{" + + "\"jobId\":\"job-1\",\"taskType\":\"SHA256\",\"state\":\"COMPLETED\"," + + "\"attempts\":1,\"maxAttempts\":3,\"workerId\":\"w2\",\"result\":\"abc\"," + + "\"error\":null,\"createdAtMillis\":100,\"updatedAtMillis\":200}}") + .getBytes(StandardCharsets.UTF_8); + JobStatusReply decoded = (JobStatusReply) MessageIO.decode(legacy); + assertThat(decoded.job().priority()).isNull(); + assertThat(decoded.job().priorityOrDefault()).isEqualTo(TaskPriority.NORMAL); + } + + @Test + void jobSnapshotBackCompatConstructorDefaultsPriorityToNormal() { + // The 10-arg constructor (no priority) keeps older call sites compiling. + JobSnapshot legacy = new JobSnapshot("job-1", TaskType.SHA256, JobState.QUEUED, + 0, 3, null, null, null, 100L, 200L); + assertThat(legacy.priority()).isNull(); + assertThat(legacy.priorityOrDefault()).isEqualTo(TaskPriority.NORMAL); + } + + @Test + void unknownFieldsRemainIgnoredAlongsidePriority() throws IOException { + // Forward-compat: an unknown trailing field must not break decoding, and a + // known priority must still be read. + byte[] withUnknown = ("{\"type\":\"SUBMIT_JOB\",\"taskType\":\"SHA256\"," + + "\"payload\":{\"text\":\"abc\"},\"maxAttempts\":0," + + "\"priority\":\"HIGH\",\"somethingNew\":123}") + .getBytes(StandardCharsets.UTF_8); + SubmitJob decoded = (SubmitJob) MessageIO.decode(withUnknown); + assertThat(decoded.priority()).isEqualTo(TaskPriority.HIGH); + } } diff --git a/coordinator/src/main/java/io/github/achrafaittayeb/dtp/coordinator/Coordinator.java b/coordinator/src/main/java/io/github/achrafaittayeb/dtp/coordinator/Coordinator.java index 45eb3b6..59938e1 100644 --- a/coordinator/src/main/java/io/github/achrafaittayeb/dtp/coordinator/Coordinator.java +++ b/coordinator/src/main/java/io/github/achrafaittayeb/dtp/coordinator/Coordinator.java @@ -9,6 +9,7 @@ import org.slf4j.LoggerFactory; import java.io.IOException; +import java.util.function.LongSupplier; /** * Wires the coordinator together and owns its lifecycle: @@ -25,8 +26,17 @@ public final class Coordinator implements AutoCloseable { private final ClientServer clientServer; public Coordinator(CoordinatorConfig config) throws IOException { + this(config, System::currentTimeMillis); + } + + /** + * Test seam: identical wiring but with an injectable clock threaded into the + * core, so integration tests can make time-dependent behavior (notably + * priority aging) deterministic. Production always uses the system clock. + */ + public Coordinator(CoordinatorConfig config, LongSupplier clock) throws IOException { JobRepository repository = openRepository(config.databasePath()); - this.core = new CoordinatorCore(config, repository); + this.core = new CoordinatorCore(config, repository, clock); this.workerServer = new WorkerServer(config.workerPort(), core); this.clientServer = new ClientServer(config.clientPort(), core); } diff --git a/coordinator/src/main/java/io/github/achrafaittayeb/dtp/coordinator/CoordinatorConfig.java b/coordinator/src/main/java/io/github/achrafaittayeb/dtp/coordinator/CoordinatorConfig.java index 47be81c..23582a5 100644 --- a/coordinator/src/main/java/io/github/achrafaittayeb/dtp/coordinator/CoordinatorConfig.java +++ b/coordinator/src/main/java/io/github/achrafaittayeb/dtp/coordinator/CoordinatorConfig.java @@ -21,6 +21,7 @@ public record CoordinatorConfig( long retryBaseDelayMillis, long retryMaxDelayMillis, int maxActiveJobs, + long agingStepMillis, String databasePath) { public static final int DEFAULT_WORKER_PORT = 7070; @@ -42,6 +43,14 @@ public record CoordinatorConfig( * observe load shedding. */ public static final int DEFAULT_MAX_ACTIVE_JOBS = 10_000; + /** + * Age-based promotion interval for priority scheduling: for each whole + * interval a QUEUED job has waited since submission, its effective scheduling + * level rises by one (capped at HIGH). The 60 s default keeps low-priority + * work from starving under sustained higher-priority load without letting a + * brief backlog reorder normal traffic. See {@code Job#effectiveLevel}. + */ + public static final long DEFAULT_AGING_STEP_MILLIS = 60_000; public static final String DEFAULT_DATABASE_PATH = "data/coordinator.db"; /** Path {@code :memory:} selects the non-durable in-memory repository. */ @@ -59,6 +68,7 @@ public static CoordinatorConfig fromArgs(String[] args) { parsed.getLong("retry-base-delay-millis", DEFAULT_RETRY_BASE_DELAY_MILLIS), parsed.getLong("retry-max-delay-millis", DEFAULT_RETRY_MAX_DELAY_MILLIS), parsed.getInt("max-active-jobs", DEFAULT_MAX_ACTIVE_JOBS), + parsed.getLong("aging-step-millis", DEFAULT_AGING_STEP_MILLIS), parsed.get("database", DEFAULT_DATABASE_PATH)); } @@ -76,5 +86,8 @@ public static CoordinatorConfig fromArgs(String[] args) { if (maxActiveJobs < 1) { throw new IllegalArgumentException("max-active-jobs must be >= 1"); } + if (agingStepMillis <= 0) { + throw new IllegalArgumentException("aging-step-millis must be >= 1"); + } } } 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 fdd45cb..06f6f7a 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 @@ -3,6 +3,7 @@ import com.fasterxml.jackson.databind.JsonNode; import io.github.achrafaittayeb.dtp.common.model.JobSnapshot; import io.github.achrafaittayeb.dtp.common.model.JobState; +import io.github.achrafaittayeb.dtp.common.model.TaskPriority; import io.github.achrafaittayeb.dtp.common.model.TaskType; import io.github.achrafaittayeb.dtp.common.model.WorkerSnapshot; import io.github.achrafaittayeb.dtp.common.protocol.TaskAssign; @@ -29,6 +30,7 @@ import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; +import java.util.function.LongSupplier; import java.util.function.Supplier; /** @@ -86,9 +88,40 @@ public final class CoordinatorCore implements AutoCloseable { */ private int activeJobCount; + /** + * Source of the monotonic {@code submissionSequence} stamped on each + * genuinely new job — the deterministic FIFO tiebreak among jobs of equal + * effective priority. Confined to the core thread, so read-then-increment is + * race-free without locking, exactly like {@link #activeJobCount}. + * + *

Only a fresh accepted submission consumes a value; idempotent + * duplicates and admission rejections never do. On recovery it is seeded to + * one past the largest persisted sequence (see {@link #recoverFromRepository}), + * so every post-restart submission still sorts strictly after every job that + * already existed. Starts at 1, leaving 0 as the reserved value migrated + * legacy rows carry. + */ + private long nextSubmissionSequence = 1; + + /** + * The one wall-clock source for all coordinator timing — assignment + * deadlines, retry backoff, heartbeat-liveness, and priority aging. Every + * time-dependent decision reads it through {@link #now()} so nothing calls + * {@link System#currentTimeMillis()} directly. Production uses the system + * clock; tests inject a controllable one to make aging deterministic without + * sleeping. Injecting it does not change any observable behavior. + */ + private final LongSupplier clock; + public CoordinatorCore(CoordinatorConfig config, JobRepository repository) { + this(config, repository, System::currentTimeMillis); + } + + /** Test seam: same as the public constructor but with an injectable clock. */ + CoordinatorCore(CoordinatorConfig config, JobRepository repository, LongSupplier clock) { this.config = config; this.repository = repository; + this.clock = clock; this.retryPolicy = new RetryPolicy(config.retryBaseDelayMillis(), config.retryMaxDelayMillis()); this.coreThread = Executors.newSingleThreadScheduledExecutor( runnable -> new Thread(runnable, "coordinator-core")); @@ -172,10 +205,16 @@ public boolean accepted() { } } - /** Overload {@link #submitJob(TaskType, JsonNode, int, String)} without an idempotency key. */ + /** Overload without an idempotency key or explicit priority (defaults NORMAL). */ public SubmitOutcome submitJob(TaskType taskType, JsonNode payload, int requestedMaxAttempts) throws InvalidPayloadException { - return submitJob(taskType, payload, requestedMaxAttempts, null); + return submitJob(taskType, payload, requestedMaxAttempts, null, null); + } + + /** Overload with an optional idempotency key but no explicit priority (defaults NORMAL). */ + public SubmitOutcome submitJob(TaskType taskType, JsonNode payload, int requestedMaxAttempts, + String idempotencyKey) throws InvalidPayloadException { + return submitJob(taskType, payload, requestedMaxAttempts, idempotencyKey, null); } /** @@ -199,25 +238,34 @@ public SubmitOutcome submitJob(TaskType taskType, JsonNode payload, int requeste * *

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. + * submissions pass through admission; execution retries never do. A fresh + * accepted job is the only thing that consumes a submission sequence. + * + *

{@code priority} is normalized so a {@code null} (omitted) priority is + * indistinguishable from an explicit {@link TaskPriority#NORMAL}, and it is + * part of the logical-submission identity for idempotency (see + * {@link #matchesSubmission}). */ public SubmitOutcome submitJob(TaskType taskType, JsonNode payload, int requestedMaxAttempts, - String idempotencyKey) throws InvalidPayloadException { + String idempotencyKey, TaskPriority priority) + throws InvalidPayloadException { TaskPayloads.validate(taskType, payload); int maxAttempts = requestedMaxAttempts > 0 ? requestedMaxAttempts : config.defaultMaxAttempts(); + TaskPriority effectivePriority = TaskPriority.normalize(priority); return askCore(() -> { if (idempotencyKey != null) { String existingId = jobIdByKey.get(idempotencyKey); if (existingId != null) { Job existing = jobs.get(existingId); - if (!matchesSubmission(existing, taskType, payload, maxAttempts)) { + if (!matchesSubmission(existing, taskType, payload, maxAttempts, effectivePriority)) { log.warn("Idempotency conflict: key={} bound to jobId={} but resubmission " - + "differs (type/payload/maxAttempts)", idempotencyKey, existingId); + + "differs (type/payload/maxAttempts/priority)", 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()); + // A duplicate creates no job, so it consumes no submission sequence. return SubmitOutcome.accepted(existingId, activeJobCount, config.maxActiveJobs()); } } @@ -225,18 +273,22 @@ public SubmitOutcome submitJob(TaskType taskType, JsonNode payload, int requeste if (activeJobCount >= limit) { log.warn("Job submission rejected (overloaded): activeJobs={} limit={} type={}", activeJobCount, limit, taskType); + // A rejection creates no persisted job, so it consumes no sequence. return SubmitOutcome.rejected(activeJobCount, limit); } + long sequence = nextSubmissionSequence++; Job job = Job.createQueued(taskType, payload, maxAttempts, - config.taskTimeoutMillis(), now(), idempotencyKey); + config.taskTimeoutMillis(), now(), idempotencyKey, effectivePriority, sequence); 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={}/{} idempotencyKey={}", - job.id(), taskType, maxAttempts, activeJobCount, limit, idempotencyKey); + log.info("Job submitted: jobId={} type={} priority={} seq={} maxAttempts={} " + + "activeJobs={}/{} idempotencyKey={}", + job.id(), taskType, effectivePriority, sequence, maxAttempts, + activeJobCount, limit, idempotencyKey); scheduleQueuedJobs(); return SubmitOutcome.accepted(job.id(), activeJobCount, limit); }); @@ -245,15 +297,17 @@ public SubmitOutcome submitJob(TaskType taskType, JsonNode payload, int requeste /** * 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. + * ({@link JsonNode#equals}); the stored job holds the already-resolved + * effective max attempts and normalized priority, so both are compared + * against the incoming effective values. Priority is part of the identity, so + * reusing a key with a different priority is a conflict, not a duplicate. */ private boolean matchesSubmission(Job existing, TaskType taskType, JsonNode payload, - int effectiveMaxAttempts) { + int effectiveMaxAttempts, TaskPriority effectivePriority) { return existing != null && existing.taskType() == taskType && existing.maxAttempts() == effectiveMaxAttempts + && existing.priority() == effectivePriority && existing.payload().equals(payload); } @@ -437,11 +491,19 @@ private void promoteDueRetries() { } } - /** Assigns QUEUED jobs (oldest first) to workers with free capacity (least loaded first). */ + /** + * Assigns QUEUED jobs to workers with free capacity (least loaded first), + * in the priority order defined by {@link #queuedOrder(long)}: higher + * effective (aged) priority first, ties broken by submission sequence (FIFO). + * The scheduling clock is read exactly once here and threaded into the + * comparator, so every pairwise comparison in a single pass sees one + * consistent {@code now} and the ordering is deterministic. + */ private void scheduleQueuedJobs() { + long now = now(); List queued = jobs.values().stream() .filter(job -> job.state() == JobState.QUEUED) - .sorted(Comparator.comparingLong(Job::createdAtMillis)) + .sorted(queuedOrder(now)) .toList(); for (Job job : queued) { if (job.state() != JobState.QUEUED) { @@ -455,6 +517,32 @@ private void scheduleQueuedJobs() { } } + /** + * The scheduling order for QUEUED jobs, evaluated against a single captured + * {@code now}: + * + *

    + *
  1. higher effective level first (base priority promoted by age, capped + * at HIGH — see {@link Job#effectiveLevel});
    2. + *
  2. then lower submission sequence first, the deterministic FIFO tiebreak + * among equal effective levels;
    3. + *
  3. then created-at, then job id — a stable, total fallback that only + * matters for migrated legacy rows, which all share sequence 0 (their + * FIFO order is therefore best-effort by creation time).
    4. + *
+ * + *

Isolated as its own method so the ordering policy is explicit and unit + * testable without a running coordinator. + */ + Comparator queuedOrder(long now) { + long step = config.agingStepMillis(); + return Comparator + .comparingInt((Job job) -> job.effectiveLevel(now, step)).reversed() + .thenComparingLong(Job::submissionSequence) + .thenComparingLong(Job::createdAtMillis) + .thenComparing(Job::id); + } + private void assign(Job job, WorkerSession worker) { String attemptId = job.assignTo(worker.workerId(), now()); worker.addActiveJob(job.id()); @@ -515,7 +603,9 @@ private void recoverFromRepository() { } log.info("Recovery started: {} persisted jobs", stored.size()); int requeued = 0; + long maxSequence = 0; for (Job job : stored) { + maxSequence = Math.max(maxSequence, job.submissionSequence()); if (job.state() == JobState.RUNNING) { if (job.hasAttemptsLeft()) { job.requeueForRecovery(now()); @@ -545,8 +635,12 @@ private void recoverFromRepository() { activeJobCount++; } } - log.info("Recovery completed: {} jobs loaded, {} requeued, {} active", - stored.size(), requeued, activeJobCount); + // Resume the sequence strictly above every persisted value so post-restart + // submissions always sort after pre-restart jobs. Legacy rows carry 0, so + // this still yields 1 for the first new submission on a migrated database. + nextSubmissionSequence = maxSequence + 1; + log.info("Recovery completed: {} jobs loaded, {} requeued, {} active, next sequence {}", + stored.size(), requeued, activeJobCount, nextSubmissionSequence); } // ------------------------------------------------------------------ @@ -554,7 +648,17 @@ private void recoverFromRepository() { // ------------------------------------------------------------------ private long now() { - return System.currentTimeMillis(); + return clock.getAsLong(); + } + + /** + * The coordinator's current clock reading (epoch millis; the system clock in + * production). Exposed so connection servers stamp worker sessions with the + * same clock the core uses for liveness, keeping every timestamp consistent + * when a test injects a controlled clock. + */ + public long nowMillis() { + return now(); } /** 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 cf6d60b..9b51654 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 @@ -3,6 +3,7 @@ import com.fasterxml.jackson.databind.JsonNode; import io.github.achrafaittayeb.dtp.common.model.JobSnapshot; import io.github.achrafaittayeb.dtp.common.model.JobState; +import io.github.achrafaittayeb.dtp.common.model.TaskPriority; import io.github.achrafaittayeb.dtp.common.model.TaskType; import java.util.UUID; @@ -25,6 +26,21 @@ public final class Job { private final long createdAtMillis; private final String idempotencyKey; + /** + * Base (submitted) priority. Immutable for the job's whole life, so a retry + * — which mutates state in place on the same instance — automatically keeps + * it. Normalized to non-null at construction. See {@link #effectiveLevel}. + */ + private final TaskPriority priority; + + /** + * Coordinator-assigned monotonic submission order, the deterministic FIFO + * tiebreak among jobs of equal effective level. Immutable like {@link #priority} + * (retries and recovery keep it), assigned once when a genuinely new job is + * accepted, and persisted so ordering survives restart. + */ + private final long submissionSequence; + private JobState state; private int attempts; private String currentAttemptId; @@ -35,17 +51,30 @@ public final class Job { private String error; private long updatedAtMillis; - /** Creates a new QUEUED job with no idempotency key. */ + /** Creates a new QUEUED job with no idempotency key, NORMAL priority, and sequence 0. */ public static Job createQueued(TaskType taskType, JsonNode payload, int maxAttempts, long executionTimeoutMillis, long now) { return createQueued(taskType, payload, maxAttempts, executionTimeoutMillis, now, null); } + /** Creates a new QUEUED job with an optional key, NORMAL priority, and sequence 0. */ public static Job createQueued(TaskType taskType, JsonNode payload, int maxAttempts, long executionTimeoutMillis, long now, String idempotencyKey) { + return createQueued(taskType, payload, maxAttempts, executionTimeoutMillis, now, + idempotencyKey, TaskPriority.NORMAL, 0); + } + + /** + * Creates a new QUEUED job. The coordinator supplies the {@code priority} + * (normalized to non-null) and the {@code submissionSequence} it assigned on + * its single-writer path. + */ + public static Job createQueued(TaskType taskType, JsonNode payload, int maxAttempts, + long executionTimeoutMillis, long now, String idempotencyKey, + TaskPriority priority, long submissionSequence) { return new Job(UUID.randomUUID().toString(), taskType, payload, maxAttempts, executionTimeoutMillis, JobState.QUEUED, 0, null, null, 0, 0, null, null, now, now, - idempotencyKey); + idempotencyKey, priority, submissionSequence); } /** Full-field constructor used when rehydrating from persistent storage. */ @@ -53,7 +82,8 @@ 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, String idempotencyKey) { + long createdAtMillis, long updatedAtMillis, String idempotencyKey, + TaskPriority priority, long submissionSequence) { this.id = id; this.taskType = taskType; this.payload = payload; @@ -70,6 +100,8 @@ public Job(String id, TaskType taskType, JsonNode payload, int maxAttempts, this.createdAtMillis = createdAtMillis; this.updatedAtMillis = updatedAtMillis; this.idempotencyKey = idempotencyKey; + this.priority = TaskPriority.normalize(priority); + this.submissionSequence = submissionSequence; } /** @@ -162,6 +194,27 @@ public boolean isEligibleToRun(long now) { || (state == JobState.RETRY_WAIT && nextEligibleTimeMillis <= now); } + /** + * The job's aged scheduling level at instant {@code now}: + * {@code min(MAX_LEVEL, baseLevel + floor(waitingAge / agingStepMillis))}. + * + *

Aging is measured from {@link #createdAtMillis} (original submission), + * so a retried job keeps accumulating fairness rather than resetting. The + * waiting age is floored at zero, so a backwards clock reading can never push + * a job below its base level. Pure function of its inputs: given the + * same {@code now} it always returns the same value, which is what keeps the + * scheduler deterministic and testable with a controlled clock. + * + * @param now the single time captured for this scheduling pass + * @param agingStepMillis the promotion interval; must be positive + */ + public int effectiveLevel(long now, long agingStepMillis) { + long waitingAge = Math.max(0L, now - createdAtMillis); + long promotions = waitingAge / agingStepMillis; + long level = priority.baseLevel() + promotions; + return (int) Math.min(TaskPriority.MAX_LEVEL, level); + } + private void transition(JobState target, long now) { if (!state.canTransitionTo(target)) { throw new IllegalStateException( @@ -179,7 +232,7 @@ private void clearLease() { public JobSnapshot snapshot() { return new JobSnapshot(id, taskType, state, attempts, maxAttempts, - assignedWorkerId, result, error, createdAtMillis, updatedAtMillis); + assignedWorkerId, result, error, createdAtMillis, updatedAtMillis, priority); } public String id() { @@ -191,6 +244,16 @@ public String idempotencyKey() { return idempotencyKey; } + /** Base (submitted) priority; never null. */ + public TaskPriority priority() { + return priority; + } + + /** Coordinator-assigned monotonic submission order, the deterministic FIFO tiebreak. */ + public long submissionSequence() { + return submissionSequence; + } + 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 3a8c2f3..5d8b813 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 @@ -3,6 +3,7 @@ import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.JsonNode; import io.github.achrafaittayeb.dtp.common.model.JobState; +import io.github.achrafaittayeb.dtp.common.model.TaskPriority; import io.github.achrafaittayeb.dtp.common.model.TaskType; import io.github.achrafaittayeb.dtp.common.net.MessageIO; import org.slf4j.Logger; @@ -35,12 +36,17 @@ public final class SqliteJobRepository implements JobRepository { // 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. + // idempotency_key, priority, and submission_sequence are only ever set on + // INSERT (they are immutable identity/ordering facts for the logical + // submission); the ON CONFLICT UPDATE path deliberately leaves them alone so + // re-saving a job on each state change never rewrites them. 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, idempotency_key) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + created_at, updated_at, idempotency_key, + priority, submission_sequence) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(id) DO UPDATE SET state = excluded.state, attempts = excluded.attempts, @@ -88,7 +94,9 @@ CREATE TABLE IF NOT EXISTS jobs ( error TEXT, created_at INTEGER NOT NULL, updated_at INTEGER NOT NULL, - idempotency_key TEXT + idempotency_key TEXT, + priority TEXT NOT NULL DEFAULT 'NORMAL', + submission_sequence INTEGER NOT NULL DEFAULT 0 )"""); } // Databases created before execution deadlines existed lack these two @@ -98,6 +106,16 @@ CREATE TABLE IF NOT EXISTS jobs ( // 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"); + // Databases created before priority scheduling lack these two columns. + // Priority defaults to NORMAL, matching the "omitted priority == NORMAL" + // rule. submission_sequence defaults to 0 for every legacy row: the + // coordinator seeds its next sequence above the persisted maximum, so all + // new submissions still receive strictly increasing, unique sequences and + // true FIFO ordering. Legacy rows share sequence 0 and are therefore + // FIFO-ordered among themselves only best-effort (by created_at, then id, + // in the scheduler's comparator) — the honest migration limitation. + ensureColumn("priority", "TEXT NOT NULL DEFAULT 'NORMAL'"); + ensureColumn("submission_sequence", "INTEGER NOT NULL DEFAULT 0"); // 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. @@ -142,6 +160,8 @@ public void save(Job job) { statement.setLong(14, job.createdAtMillis()); statement.setLong(15, job.updatedAtMillis()); statement.setString(16, job.idempotencyKey()); + statement.setString(17, job.priority().name()); + statement.setLong(18, job.submissionSequence()); statement.executeUpdate(); } catch (SQLException | JsonProcessingException e) { // Losing durability silently would break recovery guarantees; fail loudly. @@ -181,7 +201,9 @@ private static Job rowToJob(ResultSet row) throws SQLException, IOException { row.getString("error"), row.getLong("created_at"), row.getLong("updated_at"), - row.getString("idempotency_key")); + row.getString("idempotency_key"), + TaskPriority.valueOf(row.getString("priority")), + row.getLong("submission_sequence")); } @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 1a912f9..96b2993 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 @@ -109,7 +109,7 @@ private Message handle(Message request) { case SubmitJob submit -> { CoordinatorCore.SubmitOutcome outcome = core.submitJob( submit.taskType(), submit.payload(), submit.maxAttempts(), - submit.idempotencyKey()); + submit.idempotencyKey(), submit.priority()); yield switch (outcome.status()) { case ACCEPTED -> new JobSubmitted(outcome.jobId()); case REJECTED -> SubmitRejected.overloaded( diff --git a/coordinator/src/main/java/io/github/achrafaittayeb/dtp/coordinator/net/WorkerServer.java b/coordinator/src/main/java/io/github/achrafaittayeb/dtp/coordinator/net/WorkerServer.java index 684ecc7..56c5c7f 100644 --- a/coordinator/src/main/java/io/github/achrafaittayeb/dtp/coordinator/net/WorkerServer.java +++ b/coordinator/src/main/java/io/github/achrafaittayeb/dtp/coordinator/net/WorkerServer.java @@ -118,7 +118,7 @@ private WorkerSession awaitRegistration(Socket socket, InputStream in) throws IO throw new ProtocolException("Invalid registration: " + register); } WorkerSession session = new WorkerSession(register.workerId(), register.capacity(), - socket.getOutputStream(), socket, System.currentTimeMillis()); + socket.getOutputStream(), socket, core.nowMillis()); core.onWorkerRegistered(session); session.send(new WorkerRegistered(register.workerId())); return session; diff --git a/coordinator/src/test/java/io/github/achrafaittayeb/dtp/coordinator/CoordinatorConfigTest.java b/coordinator/src/test/java/io/github/achrafaittayeb/dtp/coordinator/CoordinatorConfigTest.java index 330731c..456ae8e 100644 --- a/coordinator/src/test/java/io/github/achrafaittayeb/dtp/coordinator/CoordinatorConfigTest.java +++ b/coordinator/src/test/java/io/github/achrafaittayeb/dtp/coordinator/CoordinatorConfigTest.java @@ -33,4 +33,32 @@ void rejectsNegativeMaxActiveJobs() { .isInstanceOf(IllegalArgumentException.class) .hasMessageContaining("max-active-jobs"); } + + @Test + void defaultsAgingStepMillis() { + CoordinatorConfig config = CoordinatorConfig.fromArgs(new String[0]); + assertThat(config.agingStepMillis()).isEqualTo(CoordinatorConfig.DEFAULT_AGING_STEP_MILLIS); + assertThat(config.agingStepMillis()).isEqualTo(60_000L); + } + + @Test + void parsesAgingStepMillisFlag() { + CoordinatorConfig config = CoordinatorConfig.fromArgs( + new String[]{"--aging-step-millis", "1000"}); + assertThat(config.agingStepMillis()).isEqualTo(1000L); + } + + @Test + void rejectsZeroAgingStepMillis() { + assertThatThrownBy(() -> CoordinatorConfig.fromArgs(new String[]{"--aging-step-millis", "0"})) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("aging-step-millis"); + } + + @Test + void rejectsNegativeAgingStepMillis() { + assertThatThrownBy(() -> CoordinatorConfig.fromArgs(new String[]{"--aging-step-millis", "-5"})) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("aging-step-millis"); + } } diff --git a/coordinator/src/test/java/io/github/achrafaittayeb/dtp/coordinator/CoordinatorCoreSubmissionTest.java b/coordinator/src/test/java/io/github/achrafaittayeb/dtp/coordinator/CoordinatorCoreSubmissionTest.java new file mode 100644 index 0000000..d269330 --- /dev/null +++ b/coordinator/src/test/java/io/github/achrafaittayeb/dtp/coordinator/CoordinatorCoreSubmissionTest.java @@ -0,0 +1,169 @@ +package io.github.achrafaittayeb.dtp.coordinator; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.node.JsonNodeFactory; +import io.github.achrafaittayeb.dtp.common.model.JobState; +import io.github.achrafaittayeb.dtp.common.model.TaskPriority; +import io.github.achrafaittayeb.dtp.common.model.TaskType; +import io.github.achrafaittayeb.dtp.common.task.InvalidPayloadException; +import io.github.achrafaittayeb.dtp.coordinator.job.InMemoryJobRepository; +import io.github.achrafaittayeb.dtp.coordinator.job.Job; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Submission-path semantics that are not observable through the client snapshot: + * how idempotent submissions, conflicts, admission rejection, and the monotonic + * submission sequence interact. The submission sequence is read back off the + * repository, which stores the live {@link Job} instances. + */ +class CoordinatorCoreSubmissionTest { + + private InMemoryJobRepository repository; + private CoordinatorCore core; + + private CoordinatorConfig config(int maxActiveJobs) { + return new CoordinatorConfig(0, 0, 6_000, 500, 60_000, 3, 50, 200, + maxActiveJobs, 60_000, CoordinatorConfig.IN_MEMORY_DATABASE); + } + + @BeforeEach + void setUp() { + repository = new InMemoryJobRepository(); + core = new CoordinatorCore(config(100), repository); + core.start(); + } + + @AfterEach + void tearDown() { + core.close(); + } + + private static JsonNode text(String value) { + return JsonNodeFactory.instance.objectNode().put("text", value); + } + + private long sequenceOf(String jobId) { + return repository.loadAll().stream() + .filter(job -> job.id().equals(jobId)) + .map(Job::submissionSequence) + .findFirst() + .orElseThrow(); + } + + @Test + void sameKeyAndEquivalentPriorityReturnsOriginalJob() throws InvalidPayloadException { + CoordinatorCore.SubmitOutcome first = + core.submitJob(TaskType.SHA256, text("abc"), 0, "K", TaskPriority.HIGH); + CoordinatorCore.SubmitOutcome duplicate = + core.submitJob(TaskType.SHA256, text("abc"), 0, "K", TaskPriority.HIGH); + + assertThat(first.accepted()).isTrue(); + assertThat(duplicate.accepted()).isTrue(); + assertThat(duplicate.jobId()).isEqualTo(first.jobId()); + } + + @Test + void omittedPriorityAndExplicitNormalAreIdempotentlyEquivalent() throws InvalidPayloadException { + // First submission omits priority (null → NORMAL); the resubmission states + // NORMAL explicitly. They describe the same logical work, so no conflict. + CoordinatorCore.SubmitOutcome omitted = + core.submitJob(TaskType.SHA256, text("abc"), 0, "K2", null); + CoordinatorCore.SubmitOutcome explicitNormal = + core.submitJob(TaskType.SHA256, text("abc"), 0, "K2", TaskPriority.NORMAL); + + assertThat(explicitNormal.accepted()).isTrue(); + assertThat(explicitNormal.jobId()).isEqualTo(omitted.jobId()); + } + + @Test + void sameKeyWithDifferentPriorityIsAConflict() throws InvalidPayloadException { + CoordinatorCore.SubmitOutcome first = + core.submitJob(TaskType.SHA256, text("abc"), 0, "K3", TaskPriority.NORMAL); + CoordinatorCore.SubmitOutcome conflicting = + core.submitJob(TaskType.SHA256, text("abc"), 0, "K3", TaskPriority.HIGH); + + assertThat(first.accepted()).isTrue(); + assertThat(conflicting.status()).isEqualTo(CoordinatorCore.SubmitOutcome.Status.CONFLICT); + assertThat(conflicting.conflictMessage()).contains("different submission"); + } + + @Test + void idempotentDuplicateDoesNotConsumeASequence() throws InvalidPayloadException { + CoordinatorCore.SubmitOutcome first = + core.submitJob(TaskType.SHA256, text("abc"), 0, "K4", TaskPriority.NORMAL); + // A duplicate creates no job and must not advance the sequence counter. + core.submitJob(TaskType.SHA256, text("abc"), 0, "K4", TaskPriority.NORMAL); + CoordinatorCore.SubmitOutcome next = + core.submitJob(TaskType.SHA256, text("xyz"), 0, null, TaskPriority.NORMAL); + + assertThat(sequenceOf(next.jobId())).isEqualTo(sequenceOf(first.jobId()) + 1); + } + + @Test + void rejectedSubmissionDoesNotConsumeASequence() throws InvalidPayloadException { + core.close(); // replace the default core with a limit-of-1 one + core = new CoordinatorCore(config(1), repository); + core.start(); + + CoordinatorCore.SubmitOutcome first = + core.submitJob(TaskType.SHA256, text("abc"), 0, null, TaskPriority.NORMAL); + CoordinatorCore.SubmitOutcome rejected = + core.submitJob(TaskType.SHA256, text("def"), 0, null, TaskPriority.NORMAL); + assertThat(rejected.status()).isEqualTo(CoordinatorCore.SubmitOutcome.Status.REJECTED); + + // Cancel the first to free an admission slot, then submit again. The + // rejected submission must not have burned a sequence, so the next job's + // sequence is exactly one past the first's. + core.cancelJob(first.jobId()); + CoordinatorCore.SubmitOutcome next = + core.submitJob(TaskType.SHA256, text("ghi"), 0, null, TaskPriority.NORMAL); + + assertThat(sequenceOf(next.jobId())).isEqualTo(sequenceOf(first.jobId()) + 1); + } + + @Test + void nextSequenceAfterRestartExceedsHighestPersistedTerminalJobSequence() + throws InvalidPayloadException { + // The @BeforeEach core already recovered from the empty repository; discard + // it and seed the repository as if a prior lifetime had run. + core.close(); + + // A COMPLETED (terminal) job holds the highest persisted sequence — higher + // than any active job would. The recovery seed must consider it, not just + // the non-terminal jobs loaded into the active set. + Job terminal = Job.createQueued(TaskType.SHA256, text("done"), 3, 60_000, 1_000L, + null, TaskPriority.HIGH, 500L); + terminal.assignTo("worker", 1_000L); + terminal.complete("hash", 2_000L); + assertThat(terminal.state()).isEqualTo(JobState.COMPLETED); + repository.save(terminal); + + // "Restart": a fresh core recovers from the same repository. + core = new CoordinatorCore(config(100), repository); + core.start(); + + CoordinatorCore.SubmitOutcome next = + core.submitJob(TaskType.SHA256, text("after-restart"), 0, null, TaskPriority.NORMAL); + assertThat(sequenceOf(next.jobId())) + .isGreaterThan(terminal.submissionSequence()) + .isEqualTo(501L); + } + + @Test + void freshSubmissionsReceiveStrictlyIncreasingSequences() throws InvalidPayloadException { + CoordinatorCore.SubmitOutcome a = + core.submitJob(TaskType.SHA256, text("a"), 0, null, TaskPriority.LOW); + CoordinatorCore.SubmitOutcome b = + core.submitJob(TaskType.SHA256, text("b"), 0, null, TaskPriority.HIGH); + CoordinatorCore.SubmitOutcome c = + core.submitJob(TaskType.SHA256, text("c"), 0, null, TaskPriority.NORMAL); + + long seqA = sequenceOf(a.jobId()); + assertThat(sequenceOf(b.jobId())).isEqualTo(seqA + 1); + assertThat(sequenceOf(c.jobId())).isEqualTo(seqA + 2); + } +} diff --git a/coordinator/src/test/java/io/github/achrafaittayeb/dtp/coordinator/CoordinatorSchedulingOrderTest.java b/coordinator/src/test/java/io/github/achrafaittayeb/dtp/coordinator/CoordinatorSchedulingOrderTest.java new file mode 100644 index 0000000..7796571 --- /dev/null +++ b/coordinator/src/test/java/io/github/achrafaittayeb/dtp/coordinator/CoordinatorSchedulingOrderTest.java @@ -0,0 +1,140 @@ +package io.github.achrafaittayeb.dtp.coordinator; + +import com.fasterxml.jackson.databind.node.JsonNodeFactory; +import io.github.achrafaittayeb.dtp.common.model.TaskPriority; +import io.github.achrafaittayeb.dtp.common.model.TaskType; +import io.github.achrafaittayeb.dtp.coordinator.job.InMemoryJobRepository; +import io.github.achrafaittayeb.dtp.coordinator.job.Job; +import org.junit.jupiter.api.Test; + +import java.util.Comparator; +import java.util.List; +import java.util.stream.Stream; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Unit tests for the isolated queued-job ordering policy ({@code queuedOrder}). + * No coordinator is started; the comparator is a pure function of a captured + * {@code now}, so ordering is exercised deterministically without any clock or + * threads. + */ +class CoordinatorSchedulingOrderTest { + + private static final long AGING_STEP = 100L; + private static final long CREATED = 1_000L; + + private CoordinatorCore core() { + CoordinatorConfig config = new CoordinatorConfig( + 0, 0, 6_000, 500, 60_000, 3, 50, 200, 10_000, AGING_STEP, + CoordinatorConfig.IN_MEMORY_DATABASE); + // Constructed but never started: queuedOrder needs only the config. + return new CoordinatorCore(config, new InMemoryJobRepository()); + } + + private static Job job(TaskPriority priority, long sequence, long createdAt) { + return Job.createQueued(TaskType.SLEEP, + JsonNodeFactory.instance.objectNode().put("durationMillis", 1), + 3, 60_000, createdAt, null, priority, sequence); + } + + private List sequencesInOrder(long now, Job... jobs) { + Comparator order = core().queuedOrder(now); + return Stream.of(jobs).sorted(order).map(Job::submissionSequence).toList(); + } + + @Test + void higherBasePriorityRunsFirstWhenNotAged() { + Job high = job(TaskPriority.HIGH, 3, CREATED); + Job normal = job(TaskPriority.NORMAL, 2, CREATED); + Job low = job(TaskPriority.LOW, 1, CREATED); + + assertThat(sequencesInOrder(CREATED, low, normal, high)) + .containsExactly(3L, 2L, 1L); // HIGH(seq3), NORMAL(seq2), LOW(seq1) + } + + @Test + void fifoBySequenceAmongEqualEffectivePriority() { + Job first = job(TaskPriority.NORMAL, 1, CREATED); + Job second = job(TaskPriority.NORMAL, 2, CREATED); + Job third = job(TaskPriority.NORMAL, 3, CREATED); + + // Submitted out of order into the comparator; sequence restores FIFO. + assertThat(sequencesInOrder(CREATED, third, first, second)) + .containsExactly(1L, 2L, 3L); + } + + @Test + void sameMillisecondSubmissionsStillFollowSequenceOrder() { + // All three share the exact same createdAt: only the sequence breaks the tie, + // so ordering is deterministic even at identical millisecond timestamps. + Job a = job(TaskPriority.NORMAL, 10, CREATED); + Job b = job(TaskPriority.NORMAL, 11, CREATED); + Job c = job(TaskPriority.NORMAL, 12, CREATED); + + assertThat(sequencesInOrder(CREATED, c, a, b)) + .containsExactly(10L, 11L, 12L); + } + + @Test + void agedLowTiesHighAndBeatsNewerHighBySequence() { + // An old LOW job (seq 1) that has aged to effective HIGH must run before a + // freshly submitted HIGH job (seq 100) because its sequence is lower. + Job oldLow = job(TaskPriority.LOW, 1, CREATED); + Job newHigh = job(TaskPriority.HIGH, 100, CREATED + 10 * AGING_STEP); + + long now = CREATED + 10 * AGING_STEP; // oldLow aged well past HIGH cap + assertThat(oldLow.effectiveLevel(now, AGING_STEP)).isEqualTo(TaskPriority.MAX_LEVEL); + assertThat(sequencesInOrder(now, newHigh, oldLow)) + .containsExactly(1L, 100L); // aged LOW first, by sequence + } + + @Test + void notYetAgedLowStillLosesToHigh() { + // Before the LOW job has aged, HIGH wins regardless of sequence. + Job low = job(TaskPriority.LOW, 1, CREATED); + Job high = job(TaskPriority.HIGH, 100, CREATED); + + assertThat(sequencesInOrder(CREATED, low, high)) + .containsExactly(100L, 1L); // HIGH first + } + + @Test + void extremeAgeCapsAtHighAndOrderingFallsToSequence() { + // Very old jobs of every base priority all cap at effective HIGH; with equal + // effective level the ordering is total and decided purely by sequence. + Job oldHigh = job(TaskPriority.HIGH, 30, 0L); + Job oldNormal = job(TaskPriority.NORMAL, 20, 0L); + Job oldLow = job(TaskPriority.LOW, 10, 0L); + long farFuture = 1_000_000_000L; // astronomically many aging steps → all capped at 2 + + assertThat(oldHigh.effectiveLevel(farFuture, AGING_STEP)).isEqualTo(TaskPriority.MAX_LEVEL); + assertThat(oldLow.effectiveLevel(farFuture, AGING_STEP)).isEqualTo(TaskPriority.MAX_LEVEL); + assertThat(sequencesInOrder(farFuture, oldNormal, oldHigh, oldLow)) + .containsExactly(10L, 20L, 30L); // pure sequence order + } + + @Test + void negativeWaitingAgeKeepsBasePriorityOrder() { + // A clock reading before every job's creation must not reorder below base: + // no promotion happens, so HIGH > NORMAL > LOW, then sequence. + Job high = job(TaskPriority.HIGH, 3, CREATED); + Job normal = job(TaskPriority.NORMAL, 2, CREATED); + Job low = job(TaskPriority.LOW, 1, CREATED); + + assertThat(sequencesInOrder(CREATED - 5_000, low, normal, high)) + .containsExactly(3L, 2L, 1L); // HIGH, NORMAL, LOW by base level + } + + @Test + void legacyZeroSequenceRowsOrderDeterministicallyByCreatedAt() { + // Migrated legacy rows all carry sequence 0; the created-at fallback keeps + // their ordering deterministic (best-effort FIFO). + Job older = job(TaskPriority.NORMAL, 0, CREATED); + Job newer = job(TaskPriority.NORMAL, 0, CREATED + 5); + + Comparator order = core().queuedOrder(CREATED + 100); + List sorted = Stream.of(newer, older).sorted(order).toList(); + assertThat(sorted).containsExactly(older, newer); + } +} diff --git a/coordinator/src/test/java/io/github/achrafaittayeb/dtp/coordinator/job/JobPriorityTest.java b/coordinator/src/test/java/io/github/achrafaittayeb/dtp/coordinator/job/JobPriorityTest.java new file mode 100644 index 0000000..3c36d37 --- /dev/null +++ b/coordinator/src/test/java/io/github/achrafaittayeb/dtp/coordinator/job/JobPriorityTest.java @@ -0,0 +1,106 @@ +package io.github.achrafaittayeb.dtp.coordinator.job; + +import com.fasterxml.jackson.databind.node.JsonNodeFactory; +import io.github.achrafaittayeb.dtp.common.model.TaskPriority; +import io.github.achrafaittayeb.dtp.common.model.TaskType; +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Priority, submission-sequence, and aging behavior of the {@link Job} model. */ +class JobPriorityTest { + + private static final long TIMEOUT = 60_000L; + private static final long CREATED_AT = 1_000L; + + private static Job newJob(TaskPriority priority, long sequence) { + return Job.createQueued(TaskType.SLEEP, + JsonNodeFactory.instance.objectNode().put("durationMillis", 100), + 3, TIMEOUT, CREATED_AT, null, priority, sequence); + } + + @Test + void newJobKeepsItsBasePriorityAndSequence() { + Job job = newJob(TaskPriority.HIGH, 42); + assertThat(job.priority()).isEqualTo(TaskPriority.HIGH); + assertThat(job.submissionSequence()).isEqualTo(42); + } + + @Test + void defaultFactoriesUseNormalPriorityAndZeroSequence() { + Job shortForm = Job.createQueued(TaskType.SLEEP, + JsonNodeFactory.instance.objectNode().put("durationMillis", 100), + 3, TIMEOUT, CREATED_AT); + assertThat(shortForm.priority()).isEqualTo(TaskPriority.NORMAL); + assertThat(shortForm.submissionSequence()).isZero(); + + Job keyedForm = Job.createQueued(TaskType.SLEEP, + JsonNodeFactory.instance.objectNode().put("durationMillis", 100), + 3, TIMEOUT, CREATED_AT, "key"); + assertThat(keyedForm.priority()).isEqualTo(TaskPriority.NORMAL); + assertThat(keyedForm.submissionSequence()).isZero(); + } + + @Test + void nullPriorityNormalizesToNormalAtConstruction() { + Job job = newJob(null, 1); + assertThat(job.priority()).isEqualTo(TaskPriority.NORMAL); + } + + @Test + void retryAndRequeueRetainPriorityAndSequence() { + Job job = newJob(TaskPriority.LOW, 7); + job.assignTo("worker-1", 2_000L); + job.scheduleRetry("boom", 3_000L, 2_500L); + job.requeueAfterRetryWait(3_100L); + + assertThat(job.priority()).isEqualTo(TaskPriority.LOW); + assertThat(job.submissionSequence()).isEqualTo(7); + + // A second attempt then a recovery requeue: still unchanged. + job.assignTo("worker-2", 3_200L); + job.requeueForRecovery(4_000L); + assertThat(job.priority()).isEqualTo(TaskPriority.LOW); + assertThat(job.submissionSequence()).isEqualTo(7); + } + + @Test + void effectiveLevelStartsAtBaseLevelWhenNotYetAged() { + assertThat(newJob(TaskPriority.LOW, 1).effectiveLevel(CREATED_AT, 100)).isEqualTo(0); + assertThat(newJob(TaskPriority.NORMAL, 1).effectiveLevel(CREATED_AT, 100)).isEqualTo(1); + assertThat(newJob(TaskPriority.HIGH, 1).effectiveLevel(CREATED_AT, 100)).isEqualTo(2); + } + + @Test + void effectiveLevelRisesOnePerCompletedAgingInterval() { + Job low = newJob(TaskPriority.LOW, 1); + assertThat(low.effectiveLevel(CREATED_AT + 99, 100)).isEqualTo(0); // interval not complete + assertThat(low.effectiveLevel(CREATED_AT + 100, 100)).isEqualTo(1); // one interval + assertThat(low.effectiveLevel(CREATED_AT + 200, 100)).isEqualTo(2); // two intervals + } + + @Test + void agedNormalReachesHigh() { + Job normal = newJob(TaskPriority.NORMAL, 1); + assertThat(normal.effectiveLevel(CREATED_AT + 100, 100)).isEqualTo(2); + } + + @Test + void effectiveLevelIsCappedAtHigh() { + Job low = newJob(TaskPriority.LOW, 1); + assertThat(low.effectiveLevel(CREATED_AT + 10_000, 100)).isEqualTo(TaskPriority.MAX_LEVEL); + + Job high = newJob(TaskPriority.HIGH, 1); + assertThat(high.effectiveLevel(CREATED_AT + 10_000, 100)).isEqualTo(TaskPriority.MAX_LEVEL); + } + + @Test + void negativeWaitingAgeNeverReducesBelowBaseLevel() { + // A clock that reads before the job's creation time must not demote it. + Job normal = newJob(TaskPriority.NORMAL, 1); + assertThat(normal.effectiveLevel(CREATED_AT - 5_000, 100)).isEqualTo(1); + + Job low = newJob(TaskPriority.LOW, 1); + assertThat(low.effectiveLevel(CREATED_AT - 5_000, 100)).isEqualTo(0); + } +} 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 3897936..20f19f2 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 @@ -2,6 +2,7 @@ import com.fasterxml.jackson.databind.node.JsonNodeFactory; import io.github.achrafaittayeb.dtp.common.model.JobState; +import io.github.achrafaittayeb.dtp.common.model.TaskPriority; import io.github.achrafaittayeb.dtp.common.model.TaskType; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; @@ -93,8 +94,22 @@ INSERT INTO jobs VALUES ('legacy-job', 'SHA256', '{"text":"abc"}', 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. + assertThat(loaded.priority()).isEqualTo(TaskPriority.NORMAL); // migrated column default + assertThat(loaded.submissionSequence()).isZero(); // legacy rows carry sequence 0 + + // Re-saving the migrated legacy job (as a live state change would) must + // not corrupt or replace its immutable identity/ordering fields: the + // upsert leaves priority, submission_sequence, and idempotency_key alone. + migrated.save(loaded); + Job reloaded = migrated.loadAll().stream() + .filter(j -> "legacy-job".equals(j.id())).findFirst().orElseThrow(); + assertThat(reloaded.priority()).isEqualTo(TaskPriority.NORMAL); + assertThat(reloaded.submissionSequence()).isZero(); + assertThat(reloaded.idempotencyKey()).isNull(); + assertThat(reloaded.result()).isEqualTo("old-result"); + + // The migrated database is fully usable and its idempotency unique index + // still applies: 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); @@ -132,6 +147,38 @@ void reSavingAKeyedJobKeepsItsKeyImmutable() { } } + @Test + void persistsAndReloadsPriorityAndSequence() { + Job job = Job.createQueued(TaskType.WORD_COUNT, + JsonNodeFactory.instance.objectNode().put("text", "a b"), 3, 5_000L, 42L, + null, TaskPriority.HIGH, 77L); + + try (SqliteJobRepository repository = new SqliteJobRepository(dbPath())) { + repository.save(job); + } + try (SqliteJobRepository reopened = new SqliteJobRepository(dbPath())) { + Job loaded = reopened.loadAll().getFirst(); + assertThat(loaded.priority()).isEqualTo(TaskPriority.HIGH); + assertThat(loaded.submissionSequence()).isEqualTo(77L); + } + } + + @Test + void reSavingKeepsPriorityAndSequenceImmutable() { + try (SqliteJobRepository repository = new SqliteJobRepository(dbPath())) { + Job job = Job.createQueued(TaskType.SLEEP, + JsonNodeFactory.instance.objectNode().put("durationMillis", 10), 3, 5_000L, 1L, + null, TaskPriority.LOW, 5L); + repository.save(job); + job.assignTo("w", 2L); // a later state change re-saves the row + repository.save(job); + + Job loaded = repository.loadAll().getFirst(); + assertThat(loaded.priority()).isEqualTo(TaskPriority.LOW); + assertThat(loaded.submissionSequence()).isEqualTo(5L); + } + } + @Test void loadsJobsInCreationOrder() { try (SqliteJobRepository repository = new SqliteJobRepository(dbPath())) { diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 68da0a7..09d40f1 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -47,9 +47,18 @@ DESIGN_DECISIONS. - **WorkerRegistry** — live sessions by worker id, heartbeat timestamps, capacity accounting (`capacity - activeJobs`). Plain `HashMap`s, safe by thread confinement. -- **Scheduler** (`scheduleQueuedJobs`) — oldest `QUEUED` job first, assigned - to the least-loaded worker with free capacity. Runs after every event that - could create work or capacity, plus each sweep. +- **Scheduler** (`scheduleQueuedJobs` / `queuedOrder`) — orders `QUEUED` jobs + by highest *effective* priority first, then by a persistent monotonic + submission sequence (deterministic FIFO within a level), and assigns each to + the least-loaded worker with free capacity. Effective level is base priority + (`HIGH`=2, `NORMAL`=1, `LOW`=0) promoted by `floor(age / aging-step-millis)` + and capped at `HIGH`, where `age` is measured from submission time; this + age-based promotion is what prevents starvation of low-priority work. The + scheduling clock is read exactly once per pass and threaded into the + comparator, so a pass is deterministic and the ordering policy is unit-tested + in isolation. Runs after every event that could create work or capacity, plus + each sweep. Priority affects *only* this ordering — never admission, worker + selection, or running work, and there is no preemption. - **Failure detector** (`detectDeadWorkers`) — a session whose last heartbeat is older than `heartbeat-timeout-millis` (default 6000) is suspected dead: session removed and closed, each of its RUNNING jobs retried or failed. @@ -87,7 +96,14 @@ On startup (`recoverFromRepository`), before any network listener opens: - `RUNNING` jobs (their worker sessions are gone, attempt outcome unknown) are requeued if budget remains, else failed with an explicit "restarted during final attempt" error; -- `RETRY_WAIT`/`QUEUED`/terminal jobs load as-is. +- `RETRY_WAIT`/`QUEUED`/terminal jobs load as-is; +- each job's base priority and submission sequence load with it, and the + submission-sequence counter resumes at one past the largest persisted value, + so post-restart submissions always sort after recovered jobs and scheduling + order is reproduced exactly. Databases created before this milestone migrate + their rows to `NORMAL` priority and sequence `0`; those legacy rows share + sequence `0` and so fall back to creation-time ordering among themselves, + while every new submission still gets a unique increasing sequence. `Coordinator.close()` shuts the core down *first*, freezing durable state, then closes sockets. A graceful stop therefore leaves exactly the same diff --git a/docs/DESIGN_DECISIONS.md b/docs/DESIGN_DECISIONS.md index 2022014..c1e9342 100644 --- a/docs/DESIGN_DECISIONS.md +++ b/docs/DESIGN_DECISIONS.md @@ -300,3 +300,60 @@ resend return the *same* logical job instead of a new one. 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). + +## 17. Persistent priority scheduling with age-based starvation prevention + +The scheduler was strict FIFO / least-loaded. This decision adds a base priority +to submissions and orders the queue by it, while keeping FIFO fairness within a +level and guaranteeing that low-priority work cannot starve. + +- **Three named levels, never numbers on the wire.** Priority is `HIGH`, + `NORMAL`, or `LOW` (`TaskPriority`), carried by name in `SUBMIT_JOB` and the + job snapshot. Keeping raw numbers out of the protocol means the internal + scheduling math can change without a wire break. The field is nullable and + trailing with delegating constructors — the same back-compat pattern as the + idempotency key (decision 16): an omitted priority decodes to `null` and is + normalized to `NORMAL`, so old clients and old persisted rows are unchanged and + "omitted" is indistinguishable from an explicit `NORMAL` everywhere, including + idempotency identity. +- **Effective priority = base level promoted by age, capped at HIGH.** + `effectiveLevel = min(2, baseLevel + floor(waitingAge / aging-step-millis))` + with `HIGH`=2, `NORMAL`=1, `LOW`=0 and `waitingAge = max(0, now - createdAt)`. + Age is measured from original submission, so a job that has waited long enough + is promoted regardless of base level; the floor at zero means a backwards + clock reading can never demote a job below its base. The default step is 60 s. + This is the whole starvation-prevention mechanism: a `LOW` job that waits two + steps competes as `HIGH`. +- **Deterministic FIFO via a persistent submission sequence.** Ordering is + effective-level-desc, then a coordinator-assigned monotonic `submissionSequence` + ascending. `createdAtMillis` alone is not enough — two jobs can share a + millisecond and `HashMap` iteration order is undefined — so a dedicated + sequence is the tiebreak. It is assigned on the single-writer path only when a + genuinely new job is accepted (idempotent duplicates and rejected submissions + consume none), persisted on the job row, and on recovery the counter resumes at + one past the largest persisted value, so post-restart submissions always sort + after recovered ones. This is not a distributed sequence generator — the + coordinator is already the sole writer, so a plain counter suffices. +- **The clock is read once per pass.** `scheduleQueuedJobs` captures `now` a + single time and threads it into the comparator (`queuedOrder(now)`), so every + comparison in a pass sees a consistent time and the ordering is a pure, + unit-testable function — no per-comparison clock calls, no flaky ordering. +- **Priority orders only what is queued.** It deliberately does not touch + admission control, worker selection, retry budgets, deadlines, cancellation, + or stale-result handling, and it adds no preemption: a running job is never + interrupted because something higher-priority arrived. A retry keeps the job's + original priority, sequence, and creation time — bounded attempts, not + de-prioritization, are what stop an endlessly failing job. +- **Persistence rides the existing job row.** Two columns (`priority`, + `submission_sequence`) are added with the same in-place `ensureColumn` + migration used for the deadline and idempotency columns, and are written only + on insert (immutable identity/ordering facts). Only base facts are stored; + effective priority is always recomputed and never persisted, since it changes + with time. Legacy rows migrate to `NORMAL`/sequence `0`: an honest limitation + is that pre-existing rows share sequence `0` and so order among themselves only + by creation time (best-effort FIFO), while every new submission gets true FIFO. +- **What it is not.** This is not weighted fair queuing, not per-client fairness + (there is still no client identity — see decision 16), and not production-grade + QoS. It is a single deterministic comparator with an aging term: enough to + order work by importance and guarantee no queue member waits forever, and no + more. diff --git a/docs/FAILURE_MODEL.md b/docs/FAILURE_MODEL.md index 1a3fbca..7c58300 100644 --- a/docs/FAILURE_MODEL.md +++ b/docs/FAILURE_MODEL.md @@ -180,6 +180,7 @@ the measured before/after in [MEASURED_BEHAVIOR.md](MEASURED_BEHAVIOR.md)). | coordinator `--sweep-interval-millis` | 500 | failure scan + deadline scan + retry promotion + scheduling period | | coordinator `--task-timeout-millis` | 600000 | per-attempt execution deadline (coordinator clock) | | coordinator `--max-attempts` | 3 | default attempt budget per job | +| coordinator `--aging-step-millis` | 60000 | queued-job wait per +1 effective priority level (starvation prevention) | | coordinator `--retry-base-delay-millis` / `--retry-max-delay-millis` | 1000 / 30000 | exponential backoff bounds | Lower timeouts detect failures faster but raise false positives; the 3× diff --git a/docs/PROTOCOL.md b/docs/PROTOCOL.md index f06a05e..f2267ec 100644 --- a/docs/PROTOCOL.md +++ b/docs/PROTOCOL.md @@ -58,7 +58,7 @@ connection. | Request | Reply | Notes | |---|---|---| -| `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) | +| `SUBMIT_JOB` (`taskType`, `payload`, `maxAttempts`, optional `idempotencyKey`, optional `priority`) | `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); optional `priority` is `HIGH`/`NORMAL`/`LOW`, default `NORMAL` (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) | | @@ -107,10 +107,41 @@ 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. +**`priority`** (optional, nullable) is the job's base scheduling priority, one +of `HIGH`, `NORMAL`, or `LOW`. It is carried on the wire as one of those three +names — never a raw number — so the public contract is stable. An absent field +decodes to `null` and is treated as `NORMAL`, so omitting priority is +indistinguishable from sending `NORMAL`; older clients that predate the field +are unchanged. Scheduling behavior: + +- Among `QUEUED` jobs the coordinator assigns the highest **effective** priority + first. Effective priority is the base level (`HIGH`=2, `NORMAL`=1, `LOW`=0) + raised by one for every whole `--aging-step-millis` (default 60 000) the job + has waited since submission, capped at `HIGH`. Age-based promotion prevents + low-priority jobs from starving under sustained higher-priority load. +- Ties at equal effective priority are broken by a coordinator-assigned, + persistent, strictly increasing **submission sequence** — true FIFO, even for + jobs submitted within the same millisecond. +- Priority is part of an idempotent submission's identity: reusing an + `idempotencyKey` with a *different* priority (where `null`/omitted equals + `NORMAL`) is a conflict and returns `ERROR`, exactly like a differing task + type, payload, or max attempts. +- Priority orders queued work only. It does not affect admission control, which + worker is chosen, retry budgets, deadlines, cancellation, or stale-result + handling, and there is no preemption of running work. A retry keeps the job's + original priority and submission sequence. + +Priority and submission sequence are persisted, so scheduling order is +reconstructed after a coordinator restart. Databases created before this +milestone migrate to `NORMAL` priority and sequence `0` (see the migration note +in the architecture doc). + 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 -for history even after later success), `createdAtMillis`, `updatedAtMillis`. +for history even after later success), `createdAtMillis`, `updatedAtMillis`, +and `priority` (base priority; effective/aged priority is intentionally not +exposed, since it changes with time). ## Task payloads diff --git a/integration-tests/src/test/java/io/github/achrafaittayeb/dtp/it/CoordinatorRestartIT.java b/integration-tests/src/test/java/io/github/achrafaittayeb/dtp/it/CoordinatorRestartIT.java index 2640934..4c919f3 100644 --- a/integration-tests/src/test/java/io/github/achrafaittayeb/dtp/it/CoordinatorRestartIT.java +++ b/integration-tests/src/test/java/io/github/achrafaittayeb/dtp/it/CoordinatorRestartIT.java @@ -174,7 +174,7 @@ void workerAutomaticallyReconnectsToRestartedCoordinatorOnSamePorts() throws Exc workerPort, clientPort, Testbed.HEARTBEAT_TIMEOUT_MILLIS, Testbed.SWEEP_INTERVAL_MILLIS, Testbed.TASK_TIMEOUT_MILLIS, 3, 50, 200, - Testbed.UNLIMITED_ACTIVE_JOBS, database))) { + Testbed.UNLIMITED_ACTIVE_JOBS, Testbed.AGING_STEP_MILLIS, database))) { second.start(); try (CoordinatorClient client = new CoordinatorClient("localhost", clientPort)) { Testbed.waitUntil("worker re-registered after restart", Testbed.TERMINAL_TIMEOUT, diff --git a/integration-tests/src/test/java/io/github/achrafaittayeb/dtp/it/MutableClock.java b/integration-tests/src/test/java/io/github/achrafaittayeb/dtp/it/MutableClock.java new file mode 100644 index 0000000..0874145 --- /dev/null +++ b/integration-tests/src/test/java/io/github/achrafaittayeb/dtp/it/MutableClock.java @@ -0,0 +1,30 @@ +package io.github.achrafaittayeb.dtp.it; + +import java.util.concurrent.atomic.AtomicLong; +import java.util.function.LongSupplier; + +/** + * A test clock the coordinator reads instead of the system clock, so tests that + * exercise time-dependent behavior (priority aging) can advance time explicitly + * and deterministically rather than waiting real milliseconds. Backed by an + * {@link AtomicLong} because the coordinator core thread reads it while the test + * thread advances it. + */ +final class MutableClock implements LongSupplier { + + private final AtomicLong millis; + + MutableClock(long startMillis) { + this.millis = new AtomicLong(startMillis); + } + + /** Moves time forward by {@code delta} milliseconds (delta must be >= 0). */ + void advance(long delta) { + millis.addAndGet(delta); + } + + @Override + public long getAsLong() { + return millis.get(); + } +} diff --git a/integration-tests/src/test/java/io/github/achrafaittayeb/dtp/it/PriorityRecoveryIT.java b/integration-tests/src/test/java/io/github/achrafaittayeb/dtp/it/PriorityRecoveryIT.java new file mode 100644 index 0000000..625778a --- /dev/null +++ b/integration-tests/src/test/java/io/github/achrafaittayeb/dtp/it/PriorityRecoveryIT.java @@ -0,0 +1,81 @@ +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.common.model.JobState; +import io.github.achrafaittayeb.dtp.common.model.TaskPriority; +import io.github.achrafaittayeb.dtp.common.model.TaskType; +import io.github.achrafaittayeb.dtp.common.protocol.TaskAssign; +import io.github.achrafaittayeb.dtp.coordinator.Coordinator; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.io.IOException; +import java.nio.file.Path; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Durable priority scheduling. Priorities and submission sequences persisted by + * one coordinator lifetime must drive scheduling correctly after a restart, and + * the sequence counter must resume strictly above every persisted value so new + * submissions still sort after recovered ones. + */ +class PriorityRecoveryIT { + + @TempDir + Path tempDir; + + private static String submit(CoordinatorClient client, TaskPriority priority, String text) + throws IOException { + return client.submit(TaskType.SHA256, + JsonNodeFactory.instance.objectNode().put("text", text), 0, null, priority); + } + + @Test + void priorityAndSequenceSurviveRestartAndDriveOrdering() throws Exception { + String database = tempDir.resolve("priority-restart.db").toString(); + String low; + String normal; + String high; + + // --- First lifetime: no workers, so everything stays QUEUED and persists. + try (Coordinator first = Testbed.startCoordinator(database, 3); + CoordinatorClient client = Testbed.connectClient(first)) { + low = submit(client, TaskPriority.LOW, "low"); // sequence 1 + normal = submit(client, TaskPriority.NORMAL, "normal"); // sequence 2 + high = submit(client, TaskPriority.HIGH, "high"); // sequence 3 + assertThat(client.status(low).state()).isEqualTo(JobState.QUEUED); + } + + // --- Second lifetime on the same database --------------------------- + try (Coordinator second = Testbed.startCoordinator(database, 3); + CoordinatorClient client = Testbed.connectClient(second)) { + + // Base priority survived the restart and is visible in the snapshot. + assertThat(client.status(low).priorityOrDefault()).isEqualTo(TaskPriority.LOW); + assertThat(client.status(normal).priorityOrDefault()).isEqualTo(TaskPriority.NORMAL); + assertThat(client.status(high).priorityOrDefault()).isEqualTo(TaskPriority.HIGH); + + // A brand-new submission after restart must receive a sequence strictly + // above every persisted one, so among equal HIGH priority it sorts + // AFTER the recovered HIGH job. + String newHigh = submit(client, TaskPriority.HIGH, "new-high"); // sequence 4 + + // Register a capacity-1 worker only now, so the recovered + new backlog + // drains through the scheduler in a single deterministic order. + try (ScriptedWorker worker = ScriptedWorker.register(second.workerPort(), "w1", 1)) { + assertThat(nextAssignedJobId(worker)).isEqualTo(high); // HIGH, seq 3 + assertThat(nextAssignedJobId(worker)).isEqualTo(newHigh); // HIGH, seq 4 (post-restart) + assertThat(nextAssignedJobId(worker)).isEqualTo(normal); // NORMAL, seq 2 + assertThat(nextAssignedJobId(worker)).isEqualTo(low); // LOW, seq 1 + } + } + } + + private static String nextAssignedJobId(ScriptedWorker worker) throws Exception { + TaskAssign assign = worker.awaitAssignment(); + worker.sendSuccess(assign, "ok"); + return assign.jobId(); + } +} diff --git a/integration-tests/src/test/java/io/github/achrafaittayeb/dtp/it/PrioritySchedulingIT.java b/integration-tests/src/test/java/io/github/achrafaittayeb/dtp/it/PrioritySchedulingIT.java new file mode 100644 index 0000000..040652e --- /dev/null +++ b/integration-tests/src/test/java/io/github/achrafaittayeb/dtp/it/PrioritySchedulingIT.java @@ -0,0 +1,147 @@ +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.common.model.JobState; +import io.github.achrafaittayeb.dtp.common.model.TaskPriority; +import io.github.achrafaittayeb.dtp.common.model.TaskType; +import io.github.achrafaittayeb.dtp.common.protocol.TaskAssign; +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 static org.assertj.core.api.Assertions.assertThat; + +/** + * End-to-end priority scheduling. A single scripted worker with capacity 1 is + * kept busy by a blocker job so several submissions pile up in QUEUED; the order + * in which they are assigned once the slot frees is fully determined by the + * scheduler, which lets these tests assert priority, FIFO, and aging behavior + * without racing real task execution. + */ +class PrioritySchedulingIT { + + private static String submit(CoordinatorClient client, TaskPriority priority, String text) + throws IOException { + return client.submit(TaskType.SHA256, + JsonNodeFactory.instance.objectNode().put("text", text), 0, null, priority); + } + + /** Occupies the worker's single slot and returns the still-open blocker assignment. */ + private static TaskAssign occupyWorker(CoordinatorClient client, ScriptedWorker worker) + throws Exception { + submit(client, TaskPriority.NORMAL, "blocker"); + return worker.awaitAssignment(); // now RUNNING; the one capacity slot is busy + } + + private static void awaitQueued(CoordinatorClient client, String jobId) throws Exception { + Testbed.waitUntil("job " + jobId + " queued", Testbed.TERMINAL_TIMEOUT, + () -> { + try { + return client.status(jobId).state() == JobState.QUEUED; + } catch (IOException e) { + return false; + } + }); + } + + @Test + void higherPriorityIsAssignedFirst() throws Exception { + try (Coordinator coordinator = Testbed.startCoordinator(CoordinatorConfig.IN_MEMORY_DATABASE, 3); + ScriptedWorker worker = ScriptedWorker.register(coordinator.workerPort(), "w1", 1); + CoordinatorClient client = Testbed.connectClient(coordinator)) { + + TaskAssign blocker = occupyWorker(client, worker); + + // Submit in ascending priority order so submission order can't explain + // the result — only priority can. + String low = submit(client, TaskPriority.LOW, "low"); + String normal = submit(client, TaskPriority.NORMAL, "normal"); + String high = submit(client, TaskPriority.HIGH, "high"); + awaitQueued(client, low); + awaitQueued(client, normal); + awaitQueued(client, high); + + // Free the slot; the scheduler now drains the backlog in priority order. + worker.sendSuccess(blocker, "done"); + + assertThat(nextAssignedJobId(worker)).isEqualTo(high); + assertThat(nextAssignedJobId(worker)).isEqualTo(normal); + assertThat(nextAssignedJobId(worker)).isEqualTo(low); + } + } + + @Test + void fifoIsPreservedAmongEqualPriority() throws Exception { + try (Coordinator coordinator = Testbed.startCoordinator(CoordinatorConfig.IN_MEMORY_DATABASE, 3); + ScriptedWorker worker = ScriptedWorker.register(coordinator.workerPort(), "w1", 1); + CoordinatorClient client = Testbed.connectClient(coordinator)) { + + TaskAssign blocker = occupyWorker(client, worker); + + String first = submit(client, TaskPriority.NORMAL, "first"); + String second = submit(client, TaskPriority.NORMAL, "second"); + String third = submit(client, TaskPriority.NORMAL, "third"); + awaitQueued(client, first); + awaitQueued(client, second); + awaitQueued(client, third); + + worker.sendSuccess(blocker, "done"); + + // Equal priority: submission order (sequence) decides. + assertThat(nextAssignedJobId(worker)).isEqualTo(first); + assertThat(nextAssignedJobId(worker)).isEqualTo(second); + assertThat(nextAssignedJobId(worker)).isEqualTo(third); + } + } + + @Test + void agedLowPriorityJobBeatsNewerHighPriorityJobs() throws Exception { + // Deterministic aging via an injected clock: time only moves when this + // test advances it, so the result never depends on real elapsed time or a + // Thread.sleep. The aging step is tiny and the advance is a handful of + // milliseconds — far below the heartbeat and deadline timeouts — so the + // fake advance promotes the LOW job without disturbing worker liveness. + long agingStep = 10; + MutableClock clock = new MutableClock(1_000_000); + try (Coordinator coordinator = Testbed.startCoordinator( + CoordinatorConfig.IN_MEMORY_DATABASE, 3, Testbed.TASK_TIMEOUT_MILLIS, + Testbed.UNLIMITED_ACTIVE_JOBS, agingStep, clock); + ScriptedWorker worker = ScriptedWorker.register(coordinator.workerPort(), "w1", 1); + CoordinatorClient client = Testbed.connectClient(coordinator)) { + + TaskAssign blocker = occupyWorker(client, worker); + + // Submit the LOW job first (created at the current fake time), then + // advance the clock past two promotion intervals. LOW starts at level 0, + // so two promotions raise its effective level to HIGH (2). + String agedLow = submit(client, TaskPriority.LOW, "aged-low"); + awaitQueued(client, agedLow); + clock.advance(2 * agingStep + 5); + + // Newer HIGH submissions: created "now" (no aging), higher base priority + // but higher sequence — so the aged LOW must still win the tie. + String newHigh1 = submit(client, TaskPriority.HIGH, "new-high-1"); + String newHigh2 = submit(client, TaskPriority.HIGH, "new-high-2"); + awaitQueued(client, newHigh1); + awaitQueued(client, newHigh2); + + worker.sendSuccess(blocker, "done"); + + // Starvation prevented: the long-waiting LOW job runs before the newer + // HIGH jobs; the HIGH jobs then follow in submission order. + assertThat(nextAssignedJobId(worker)).isEqualTo(agedLow); + assertThat(nextAssignedJobId(worker)).isEqualTo(newHigh1); + assertThat(nextAssignedJobId(worker)).isEqualTo(newHigh2); + } + } + + /** Captures the next assignment and immediately completes it so the next can flow. */ + private static String nextAssignedJobId(ScriptedWorker worker) throws Exception { + TaskAssign assign = worker.awaitAssignment(); + worker.sendSuccess(assign, "ok"); + return assign.jobId(); + } +} diff --git a/integration-tests/src/test/java/io/github/achrafaittayeb/dtp/it/Testbed.java b/integration-tests/src/test/java/io/github/achrafaittayeb/dtp/it/Testbed.java index cac3ca1..b61bc4c 100644 --- a/integration-tests/src/test/java/io/github/achrafaittayeb/dtp/it/Testbed.java +++ b/integration-tests/src/test/java/io/github/achrafaittayeb/dtp/it/Testbed.java @@ -32,6 +32,8 @@ private Testbed() { /** Coordinator on ephemeral ports; {@code database} may be {@code :memory:} or a temp file. */ /** Large enough that admission control never fires unless a test asks it to. */ static final int UNLIMITED_ACTIVE_JOBS = 1_000_000; + /** Long enough that age-based promotion never fires unless a test asks it to. */ + static final long AGING_STEP_MILLIS = 60_000; static Coordinator startCoordinator(String database, int maxAttempts) throws IOException { return startCoordinator(database, maxAttempts, TASK_TIMEOUT_MILLIS); @@ -44,16 +46,43 @@ static Coordinator startCoordinator(String database, int maxAttempts, long taskT static Coordinator startCoordinator(String database, int maxAttempts, long taskTimeoutMillis, int maxActiveJobs) throws IOException { - Coordinator coordinator = new Coordinator(new CoordinatorConfig( + return startCoordinator(database, maxAttempts, taskTimeoutMillis, maxActiveJobs, + AGING_STEP_MILLIS); + } + + static Coordinator startCoordinator(String database, int maxAttempts, long taskTimeoutMillis, + int maxActiveJobs, long agingStepMillis) throws IOException { + Coordinator coordinator = new Coordinator(config( + database, maxAttempts, taskTimeoutMillis, maxActiveJobs, agingStepMillis)); + coordinator.start(); + return coordinator; + } + + /** + * Coordinator driven by an injected {@code clock} instead of the system + * clock, so a test can advance time deterministically (used for priority + * aging). All other timing is identical to the standard test coordinator. + */ + static Coordinator startCoordinator(String database, int maxAttempts, long taskTimeoutMillis, + int maxActiveJobs, long agingStepMillis, + java.util.function.LongSupplier clock) throws IOException { + Coordinator coordinator = new Coordinator(config( + database, maxAttempts, taskTimeoutMillis, maxActiveJobs, agingStepMillis), clock); + coordinator.start(); + return coordinator; + } + + private static CoordinatorConfig config(String database, int maxAttempts, long taskTimeoutMillis, + int maxActiveJobs, long agingStepMillis) { + return new CoordinatorConfig( 0, 0, HEARTBEAT_TIMEOUT_MILLIS, SWEEP_INTERVAL_MILLIS, taskTimeoutMillis, maxAttempts, 50, 200, maxActiveJobs, - database)); - coordinator.start(); - return coordinator; + agingStepMillis, + database); } static Worker startWorker(Coordinator coordinator, String workerId, int capacity) {