Feat/lease deadlines cancellation

README.md

@@ -72,19 +72,27 @@ flowchart LR
 ```mermaid
 stateDiagram-v2
     [*] --> QUEUED : submit
-    QUEUED --> RUNNING : assigned (new attempt lease)
+    QUEUED --> RUNNING : assigned (new attempt lease + deadline)
     RUNNING --> COMPLETED : result accepted
-    RUNNING --> RETRY_WAIT : worker lost / task failed,<br/>attempts remain
+    RUNNING --> RETRY_WAIT : worker lost / task failed /<br/>deadline expired, attempts remain
     RUNNING --> FAILED : attempts exhausted
     RUNNING --> QUEUED : coordinator restart recovery
     RETRY_WAIT --> QUEUED : backoff elapsed
+    QUEUED --> CANCELLED : client cancel
+    RETRY_WAIT --> CANCELLED : client cancel
+    RUNNING --> CANCELLED : client cancel (lease revoked)
     COMPLETED --> [*]
     FAILED --> [*]
+    CANCELLED --> [*]
 ```
 
-Every assignment carries a fresh **attempt lease** (a UUID). A result is only
-accepted if it carries the job's *current* lease — results from workers that
-were declared dead and later resurface are logged and rejected. See
+Every assignment carries a fresh **attempt lease** (a UUID) that is
+**time-bounded**: it expires at a deadline measured on the coordinator's clock
+(`assignedAt + executionTimeout`). A result is only accepted if it carries the
+job's *current* lease — results from workers that were declared dead, timed
+out, or cancelled, and later resurface, are logged and rejected. When a lease
+expires the coordinator revokes it, best-effort asks the worker to stop
+(`TASK_CANCEL`), and retries the job. See
 [docs/FAILURE_MODEL.md](docs/FAILURE_MODEL.md).
 
 ## Requirements
@@ -120,6 +128,7 @@ well under a minute.
 ./scripts/client.sh submit sleep --milliseconds 30000
 ./scripts/client.sh status <job-id>
 ./scripts/client.sh wait <job-id>
+./scripts/client.sh cancel <job-id>
 ./scripts/client.sh list
 ./scripts/client.sh workers
 ```
@@ -129,6 +138,19 @@ 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.
 
+**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
+stop, and the job is retried — even if the worker is alive and heartbeating.
+This closes the gap where one wedged task would otherwise occupy a slot
+forever. To watch it, start the coordinator with a short timeout and submit a
+longer sleep:
+
+```bash
+./scripts/coordinator.sh --task-timeout-millis 3000
+./scripts/client.sh submit sleep --milliseconds 20000   # exceeds the 3s deadline
+```
+
 ## The failure-recovery demo
 
 ```bash
@@ -212,7 +234,9 @@ containers.
   multi-worker spread and true concurrency (asserted by elapsed time), worker
   failure via heartbeat silence *and* via abrupt disconnect, retry exhaustion,
   stale-result rejection in three variants, coordinator restart recovery
-  (including the final-attempt rule and worker auto-reconnect), a
+  (including the final-attempt rule, the stale-deadline rule, and worker
+  auto-reconnect), execution-deadline timeout on a healthy worker with
+  reassignment, client cancellation of queued and running jobs, a
   60-jobs/4-clients/5-workers scheduling race test asserting exactly one
   assignment per job, and malformed-bytes robustness on both ports.
 
@@ -227,6 +251,15 @@ containers.
 - **Failure suspicion, not proof.** A missed heartbeat means the worker is
   *suspected* dead under a timeout model; a slow-but-alive worker can be
   declared dead. Its late results are then rejected as stale.
+- **Execution deadline ≠ worker death.** A timed-out attempt means the *task*
+  exceeded its execution contract, not that the worker failed — the worker may
+  be healthy and heartbeating. Expiry is judged solely on the coordinator's
+  clock, so worker clocks need not be synchronized. The revoked attempt's late
+  result is rejected through the same lease check.
+- **Cancellation is cooperative.** `cancel` and deadline expiry revoke the lease
+  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.
 - **Single coordinator.** The coordinator is a single point of failure;
   durability (not availability) is what restart recovery provides. Multiple
   coordinators would require leader election/consensus — out of scope, by
@@ -242,7 +275,11 @@ More detail in [docs/FAILURE_MODEL.md](docs/FAILURE_MODEL.md) and
 - One coordinator (see above); leader election is the natural next step.
 - Scheduling is FIFO / least-loaded; no priorities, deadlines, or fairness.
 - Results live in the job row; large results would need external storage.
-- No cancellation, backpressure, or per-task execution timeout yet.
+- No backpressure or bounded submission queue yet; the job set grows
+  unboundedly.
+- Cancellation is cooperative (thread interruption), not hard preemption: a
+  task that ignores interruption keeps running until it finishes, though its
+  lease is already revoked so its result is discarded.
 
 ## Documentation
 

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

@@ -77,6 +77,7 @@ private static void run(String[] rawArgs) throws Exception {
                 }
                 case "list" -> listJobs(client);
                 case "workers" -> listWorkers(client);
+                case "cancel" -> cancel(client, requireJobId(positionals));
                 default -> throw new UsageException("Unknown command: " + command);
             }
         }
@@ -152,6 +153,17 @@ private static void listJobs(CoordinatorClient client) throws Exception {
         }
     }
 
+    private static void cancel(CoordinatorClient client, String jobId) throws Exception {
+        var reply = client.cancel(jobId);
+        if (reply.cancelled()) {
+            System.out.println("Job cancelled");
+        } else {
+            System.out.println("Job already finished; nothing to cancel (state "
+                    + reply.job().state() + ")");
+        }
+        printJob(reply.job());
+    }
+
     private static void listWorkers(CoordinatorClient client) throws Exception {
         List<WorkerSnapshot> workers = client.listWorkers();
         if (workers.isEmpty()) {
@@ -199,6 +211,7 @@ private static void printUsage() {
                   submit fail --fail-until-attempt <n>
                   status <job-id>
                   wait <job-id> [--timeout-millis <ms>]
+                  cancel <job-id>
                   list
                   workers
                 Global options: --host <host> (default localhost), --port <port> (default 7071)

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

@@ -6,8 +6,10 @@
 import io.github.achrafaittayeb.dtp.common.model.WorkerSnapshot;
 import io.github.achrafaittayeb.dtp.common.net.MessageIO;
 import io.github.achrafaittayeb.dtp.common.net.ProtocolException;
+import io.github.achrafaittayeb.dtp.common.protocol.CancelJob;
 import io.github.achrafaittayeb.dtp.common.protocol.ErrorReply;
 import io.github.achrafaittayeb.dtp.common.protocol.GetJobStatus;
+import io.github.achrafaittayeb.dtp.common.protocol.JobCancelReply;
 import io.github.achrafaittayeb.dtp.common.protocol.JobListReply;
 import io.github.achrafaittayeb.dtp.common.protocol.JobStatusReply;
 import io.github.achrafaittayeb.dtp.common.protocol.JobSubmitted;
@@ -76,6 +78,21 @@ public synchronized List<WorkerSnapshot> listWorkers() throws IOException {
         throw asError(reply);
     }
 
+    /**
+     * Requests cancellation of a job. Returns the job's snapshot; check its
+     * state to see the outcome (CANCELLED if this call cancelled it, or the
+     * job's actual terminal state if it had already finished).
+     *
+     * @throws IOException if the job id is unknown
+     */
+    public synchronized JobCancelReply cancel(String jobId) throws IOException {
+        Message reply = exchange(new CancelJob(jobId));
+        if (reply instanceof JobCancelReply cancelReply) {
+            return cancelReply;
+        }
+        throw asError(reply);
+    }
+
     /**
      * Polls until the job reaches a terminal state (COMPLETED or FAILED).
      *

common/src/main/java/io/github/achrafaittayeb/dtp/common/model/JobState.java

@@ -13,7 +13,10 @@
  *    │            │ └────────▶ FAILED        (retries exhausted, or non-retryable)
  *    ├────────────┘                          (coordinator recovery requeue)
  *    │
- *    └──── RETRY_WAIT ◀─────── RUNNING       (worker lost / task failed, retry pending)
+ *    └──── RETRY_WAIT ◀─────── RUNNING       (worker lost / task failed / deadline
+ *                                             expired, retry pending)
+ *
+ * QUEUED / RETRY_WAIT / RUNNING ──▶ CANCELLED   (client-requested)
  * </pre>
  */
 public enum JobState {
@@ -26,20 +29,23 @@ public enum JobState {
     /** Terminal: a worker reported a successful result that was accepted. */
     COMPLETED,
     /** Terminal: retries were exhausted or the job was rejected permanently. */
-    FAILED;
+    FAILED,
+    /** Terminal: cancelled on client request; any in-flight attempt lease was revoked. */
+    CANCELLED;
 
     private static final Map<JobState, Set<JobState>> LEGAL_TRANSITIONS = Map.of(
-            QUEUED, EnumSet.of(RUNNING),
-            RUNNING, EnumSet.of(COMPLETED, FAILED, RETRY_WAIT, QUEUED),
-            RETRY_WAIT, EnumSet.of(QUEUED),
+            QUEUED, EnumSet.of(RUNNING, CANCELLED),
+            RUNNING, EnumSet.of(COMPLETED, FAILED, RETRY_WAIT, QUEUED, CANCELLED),
+            RETRY_WAIT, EnumSet.of(QUEUED, CANCELLED),
             COMPLETED, EnumSet.noneOf(JobState.class),
-            FAILED, EnumSet.noneOf(JobState.class));
+            FAILED, EnumSet.noneOf(JobState.class),
+            CANCELLED, EnumSet.noneOf(JobState.class));
 
     public boolean canTransitionTo(JobState target) {
         return LEGAL_TRANSITIONS.get(this).contains(target);
     }
 
     public boolean isTerminal() {
-        return this == COMPLETED || this == FAILED;
+        return this == COMPLETED || this == FAILED || this == CANCELLED;
     }
 }

common/src/main/java/io/github/achrafaittayeb/dtp/common/protocol/CancelJob.java

@@ -0,0 +1,5 @@
+package io.github.achrafaittayeb.dtp.common.protocol;
+
+/** Client request to cancel a job by id. */
+public record CancelJob(String jobId) implements Message {
+}

common/src/main/java/io/github/achrafaittayeb/dtp/common/protocol/JobCancelReply.java

@@ -0,0 +1,11 @@
+package io.github.achrafaittayeb.dtp.common.protocol;
+
+import io.github.achrafaittayeb.dtp.common.model.JobSnapshot;
+
+/**
+ * Reply to {@link CancelJob}. {@code cancelled} is true when this request moved
+ * the job into CANCELLED; it is false when the job was already terminal (its
+ * snapshot shows the actual final state). Unknown ids return {@link ErrorReply}.
+ */
+public record JobCancelReply(boolean cancelled, JobSnapshot job) implements Message {
+}

common/src/main/java/io/github/achrafaittayeb/dtp/common/protocol/Message.java

@@ -22,23 +22,26 @@
         // Coordinator -> Worker
         @JsonSubTypes.Type(value = WorkerRegistered.class, name = "WORKER_REGISTERED"),
         @JsonSubTypes.Type(value = TaskAssign.class, name = "TASK_ASSIGN"),
+        @JsonSubTypes.Type(value = TaskCancel.class, name = "TASK_CANCEL"),
         // Client -> Coordinator
         @JsonSubTypes.Type(value = SubmitJob.class, name = "SUBMIT_JOB"),
         @JsonSubTypes.Type(value = GetJobStatus.class, name = "GET_JOB_STATUS"),
         @JsonSubTypes.Type(value = ListJobs.class, name = "LIST_JOBS"),
         @JsonSubTypes.Type(value = ListWorkers.class, name = "LIST_WORKERS"),
+        @JsonSubTypes.Type(value = CancelJob.class, name = "CANCEL_JOB"),
         // Coordinator -> Client
         @JsonSubTypes.Type(value = JobSubmitted.class, name = "JOB_SUBMITTED"),
         @JsonSubTypes.Type(value = JobStatusReply.class, name = "JOB_STATUS"),
         @JsonSubTypes.Type(value = JobListReply.class, name = "JOB_LIST"),
         @JsonSubTypes.Type(value = WorkerListReply.class, name = "WORKER_LIST"),
+        @JsonSubTypes.Type(value = JobCancelReply.class, name = "JOB_CANCEL"),
         // Either direction
         @JsonSubTypes.Type(value = ErrorReply.class, name = "ERROR"),
 })
 public sealed interface Message
         permits WorkerRegister, Heartbeat, TaskResult,
-        WorkerRegistered, TaskAssign,
-        SubmitJob, GetJobStatus, ListJobs, ListWorkers,
-        JobSubmitted, JobStatusReply, JobListReply, WorkerListReply,
+        WorkerRegistered, TaskAssign, TaskCancel,
+        SubmitJob, GetJobStatus, ListJobs, ListWorkers, CancelJob,
+        JobSubmitted, JobStatusReply, JobListReply, WorkerListReply, JobCancelReply,
         ErrorReply {
 }

common/src/main/java/io/github/achrafaittayeb/dtp/common/protocol/TaskCancel.java

@@ -0,0 +1,15 @@
+package io.github.achrafaittayeb.dtp.common.protocol;
+
+/**
+ * Coordinator's request that a worker stop one specific attempt. The worker
+ * must cancel only if both {@code jobId} and {@code attemptId} match its
+ * currently executing assignment — a cancel for a superseded attempt must
+ * never touch a newer one.
+ *
+ * <p>Delivery is best-effort and cancellation is cooperative (thread
+ * interruption). Correctness never depends on it: by the time this message is
+ * sent the attempt's lease is already revoked, so a task that ignores the
+ * cancel can only ever produce a stale, rejected result.
+ */
+public record TaskCancel(String jobId, String attemptId) implements Message {
+}

common/src/test/java/io/github/achrafaittayeb/dtp/common/model/JobStateTest.java

@@ -2,6 +2,7 @@
 
 import org.junit.jupiter.api.Test;
 
+import static io.github.achrafaittayeb.dtp.common.model.JobState.CANCELLED;
 import static io.github.achrafaittayeb.dtp.common.model.JobState.COMPLETED;
 import static io.github.achrafaittayeb.dtp.common.model.JobState.FAILED;
 import static io.github.achrafaittayeb.dtp.common.model.JobState.QUEUED;
@@ -21,23 +22,36 @@ void allowsDocumentedTransitions() {
         assertThat(RETRY_WAIT.canTransitionTo(QUEUED)).isTrue();
     }
 
+    @Test
+    void cancellationIsAllowedFromEveryNonTerminalState() {
+        assertThat(QUEUED.canTransitionTo(CANCELLED)).isTrue();
+        assertThat(RETRY_WAIT.canTransitionTo(CANCELLED)).isTrue();
+        assertThat(RUNNING.canTransitionTo(CANCELLED)).isTrue();
+        assertThat(COMPLETED.canTransitionTo(CANCELLED)).isFalse();
+        assertThat(FAILED.canTransitionTo(CANCELLED)).isFalse();
+    }
+
     @Test
     void rejectsIllegalTransitions() {
         assertThat(QUEUED.canTransitionTo(COMPLETED)).isFalse();
         assertThat(QUEUED.canTransitionTo(FAILED)).isFalse();
         assertThat(RETRY_WAIT.canTransitionTo(RUNNING)).isFalse();
         assertThat(COMPLETED.canTransitionTo(RUNNING)).isFalse();
         assertThat(FAILED.canTransitionTo(QUEUED)).isFalse();
+        assertThat(CANCELLED.canTransitionTo(QUEUED)).isFalse();
+        assertThat(CANCELLED.canTransitionTo(RUNNING)).isFalse();
     }
 
     @Test
     void terminalStatesHaveNoOutgoingTransitions() {
         for (JobState target : JobState.values()) {
             assertThat(COMPLETED.canTransitionTo(target)).isFalse();
             assertThat(FAILED.canTransitionTo(target)).isFalse();
+            assertThat(CANCELLED.canTransitionTo(target)).isFalse();
         }
         assertThat(COMPLETED.isTerminal()).isTrue();
         assertThat(FAILED.isTerminal()).isTrue();
+        assertThat(CANCELLED.isTerminal()).isTrue();
         assertThat(RUNNING.isTerminal()).isFalse();
     }
 }

common/src/test/java/io/github/achrafaittayeb/dtp/common/protocol/MessageSerializationTest.java

@@ -52,6 +52,19 @@ void roundTripsClientMessages() throws IOException {
                 .isEqualTo(new JobListReply(List.of(snapshot)));
     }
 
+    @Test
+    void roundTripsCancellationMessages() throws IOException {
+        assertThat(roundTrip(new TaskCancel("job-1", "att-1")))
+                .isEqualTo(new TaskCancel("job-1", "att-1"));
+        assertThat(roundTrip(new CancelJob("job-1")))
+                .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);
+        assertThat(roundTrip(new JobCancelReply(true, cancelled)))
+                .isEqualTo(new JobCancelReply(true, cancelled));
+    }
+
     @Test
     void encodedMessagesCarryTypeDiscriminator() throws IOException {
         String json = new String(MessageIO.encode(new Heartbeat("w")), StandardCharsets.UTF_8);

coordinator/src/main/java/io/github/achrafaittayeb/dtp/coordinator/CoordinatorConfig.java

@@ -16,6 +16,7 @@ public record CoordinatorConfig(
         int clientPort,
         long heartbeatTimeoutMillis,
         long sweepIntervalMillis,
+        long taskTimeoutMillis,
         int defaultMaxAttempts,
         long retryBaseDelayMillis,
         long retryMaxDelayMillis,
@@ -25,6 +26,10 @@ public record CoordinatorConfig(
     public static final int DEFAULT_CLIENT_PORT = 7071;
     public static final long DEFAULT_HEARTBEAT_TIMEOUT_MILLIS = 6_000;
     public static final long DEFAULT_SWEEP_INTERVAL_MILLIS = 500;
+    /** Default per-attempt execution deadline; matches the largest built-in task bound (10 min). */
+    public static final long DEFAULT_TASK_TIMEOUT_MILLIS = 10 * 60 * 1_000;
+    /** Upper bound for per-job execution-timeout overrides (24 h). */
+    public static final long MAX_TASK_TIMEOUT_MILLIS = 24 * 60 * 60 * 1_000;
     public static final int DEFAULT_MAX_ATTEMPTS = 3;
     public static final long DEFAULT_RETRY_BASE_DELAY_MILLIS = 1_000;
     public static final long DEFAULT_RETRY_MAX_DELAY_MILLIS = 30_000;
@@ -40,6 +45,7 @@ public static CoordinatorConfig fromArgs(String[] args) {
                 parsed.getInt("client-port", DEFAULT_CLIENT_PORT),
                 parsed.getLong("heartbeat-timeout-millis", DEFAULT_HEARTBEAT_TIMEOUT_MILLIS),
                 parsed.getLong("sweep-interval-millis", DEFAULT_SWEEP_INTERVAL_MILLIS),
+                parsed.getLong("task-timeout-millis", DEFAULT_TASK_TIMEOUT_MILLIS),
                 parsed.getInt("max-attempts", DEFAULT_MAX_ATTEMPTS),
                 parsed.getLong("retry-base-delay-millis", DEFAULT_RETRY_BASE_DELAY_MILLIS),
                 parsed.getLong("retry-max-delay-millis", DEFAULT_RETRY_MAX_DELAY_MILLIS),
@@ -50,6 +56,10 @@ public static CoordinatorConfig fromArgs(String[] args) {
         if (heartbeatTimeoutMillis <= 0 || sweepIntervalMillis <= 0) {
             throw new IllegalArgumentException("Timing intervals must be positive");
         }
+        if (taskTimeoutMillis <= 0 || taskTimeoutMillis > MAX_TASK_TIMEOUT_MILLIS) {
+            throw new IllegalArgumentException(
+                    "task-timeout-millis must be in 1.." + MAX_TASK_TIMEOUT_MILLIS);
+        }
         if (defaultMaxAttempts < 1) {
             throw new IllegalArgumentException("max-attempts must be >= 1");
         }