feat: add persistent priority scheduling

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

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<String> 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<String> 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 <n>] [--wait-timeout-millis <ms>]
                 Global options: --host <host> (default localhost), --port <port> (default 7071)
                 Submit options: --max-attempts <n> (default: coordinator setting)
+                  --priority high|normal|low      base scheduling priority (default normal;
+                                                  case-insensitive)
                   --submit-retries <n>            retries after the first attempt on a
                                                   retryable overload rejection (default 0 = off)
                   --submit-retry-base-millis <ms> base back-off (default 200)

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();
         }

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");
+    }
+}

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.
+ *
+ * <p>{@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);
+    }
 }

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.
+ *
+ * <p>Each level has an integer <em>base level</em> 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}.
+ *
+ * <p><b>Default:</b> 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");
+        };
+    }
+}

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 <em>original</em> job's id instead of
  * creating another logical job (see {@code CoordinatorCore}). A {@code null}
  * key preserves the historical behavior — every submission creates a fresh job.
- * The field is deliberately last with a delegating constructor so existing
- * call sites and older peers (which omit it, and it decodes to {@code null})
- * remain source- and wire-compatible.
+ *
+ * <p>{@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);
     }
 }