> 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;
}
}
diff --git a/common/src/main/java/io/github/achrafaittayeb/dtp/common/protocol/CancelJob.java b/common/src/main/java/io/github/achrafaittayeb/dtp/common/protocol/CancelJob.java
new file mode 100644
index 0000000..898ae84
--- /dev/null
+++ b/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 {
+}
diff --git a/common/src/main/java/io/github/achrafaittayeb/dtp/common/protocol/JobCancelReply.java b/common/src/main/java/io/github/achrafaittayeb/dtp/common/protocol/JobCancelReply.java
new file mode 100644
index 0000000..639da6f
--- /dev/null
+++ b/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 {
+}
diff --git a/common/src/main/java/io/github/achrafaittayeb/dtp/common/protocol/Message.java b/common/src/main/java/io/github/achrafaittayeb/dtp/common/protocol/Message.java
index 6d7f295..8c9332e 100644
--- a/common/src/main/java/io/github/achrafaittayeb/dtp/common/protocol/Message.java
+++ b/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 {
}
diff --git a/common/src/main/java/io/github/achrafaittayeb/dtp/common/protocol/TaskCancel.java b/common/src/main/java/io/github/achrafaittayeb/dtp/common/protocol/TaskCancel.java
new file mode 100644
index 0000000..93c7bfa
--- /dev/null
+++ b/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.
+ *
+ * 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 {
+}
diff --git a/common/src/test/java/io/github/achrafaittayeb/dtp/common/model/JobStateTest.java b/common/src/test/java/io/github/achrafaittayeb/dtp/common/model/JobStateTest.java
index 35f37d8..050382b 100644
--- a/common/src/test/java/io/github/achrafaittayeb/dtp/common/model/JobStateTest.java
+++ b/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,6 +22,15 @@ 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();
@@ -28,6 +38,8 @@ void rejectsIllegalTransitions() {
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
@@ -35,9 +47,11 @@ 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();
}
}
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 3b451c0..f7b0c2a 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
@@ -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);
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 f4af872..215f915 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
@@ -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");
}
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 c659402..068c5a5 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
@@ -6,6 +6,7 @@
import io.github.achrafaittayeb.dtp.common.model.TaskType;
import io.github.achrafaittayeb.dtp.common.model.WorkerSnapshot;
import io.github.achrafaittayeb.dtp.common.protocol.TaskAssign;
+import io.github.achrafaittayeb.dtp.common.protocol.TaskCancel;
import io.github.achrafaittayeb.dtp.common.protocol.TaskResult;
import io.github.achrafaittayeb.dtp.common.task.InvalidPayloadException;
import io.github.achrafaittayeb.dtp.common.task.TaskPayloads;
@@ -119,7 +120,8 @@ public String submitJob(TaskType taskType, JsonNode payload, int requestedMaxAtt
TaskPayloads.validate(taskType, payload);
int maxAttempts = requestedMaxAttempts > 0 ? requestedMaxAttempts : config.defaultMaxAttempts();
return askCore(() -> {
- Job job = Job.createQueued(taskType, payload, maxAttempts, now());
+ Job job = Job.createQueued(taskType, payload, maxAttempts,
+ config.taskTimeoutMillis(), now());
jobs.put(job.id(), job);
repository.save(job);
log.info("Job submitted: jobId={} type={} maxAttempts={}",
@@ -133,6 +135,34 @@ public Optional getJob(String jobId) {
return askCore(() -> Optional.ofNullable(jobs.get(jobId)).map(Job::snapshot));
}
+ /** Outcome of a cancellation request: whether the job is present, and whether this call cancelled it. */
+ public record CancelOutcome(boolean found, boolean cancelledNow, JobSnapshot job) {
+ }
+
+ /**
+ * Cancels a job on client request. QUEUED / RETRY_WAIT / RUNNING → CANCELLED
+ * (a RUNNING attempt's lease is revoked and a best-effort TASK_CANCEL is
+ * sent); already-terminal jobs are left untouched and reported as such.
+ */
+ public CancelOutcome cancelJob(String jobId) {
+ return askCore(() -> {
+ Job job = jobs.get(jobId);
+ if (job == null) {
+ return new CancelOutcome(false, false, null);
+ }
+ if (job.state().isTerminal()) {
+ return new CancelOutcome(true, false, job.snapshot());
+ }
+ String attemptId = job.currentAttemptId();
+ String workerId = job.assignedWorkerId();
+ job.cancel(now());
+ repository.save(job);
+ revokeAttemptOnWorker(jobId, attemptId, workerId, "cancelled by client");
+ log.info("Job cancelled by client: jobId={} previousWorker={}", jobId, workerId);
+ return new CancelOutcome(true, true, job.snapshot());
+ });
+ }
+
public List listJobs() {
return askCore(() -> jobs.values().stream()
.sorted(Comparator.comparingLong(Job::createdAtMillis).reversed())
@@ -148,10 +178,11 @@ public List listWorkers() {
// Core-thread logic
// ------------------------------------------------------------------
- /** Periodic maintenance: failure detection, retry promotion, scheduling. */
+ /** Periodic maintenance: failure detection, deadline enforcement, retry promotion, scheduling. */
private void sweep() {
try {
detectDeadWorkers();
+ detectExpiredDeadlines();
promoteDueRetries();
scheduleQueuedJobs();
} catch (RuntimeException unexpected) {
@@ -193,6 +224,59 @@ private void invalidateAssignments(WorkerSession session, String reason) {
}
}
+ /**
+ * A worker being alive does not imply every task on it is making progress.
+ * Each RUNNING attempt carries a coordinator-clock deadline; once it
+ * expires the attempt's lease is revoked and the job retried, exactly as
+ * if the worker had been lost — except the worker stays registered and a
+ * best-effort {@code TASK_CANCEL} asks it to stop the wasted work. Expiry
+ * is suspicion that the task exceeded its execution contract, not proof of
+ * failure: if the task finishes anyway, its result fails the lease check.
+ */
+ private void detectExpiredDeadlines() {
+ long now = now();
+ for (Job job : jobs.values()) {
+ if (!job.isDeadlineExpired(now)) {
+ continue;
+ }
+ // Capture lease identity before retryOrFail clears it.
+ String attemptId = job.currentAttemptId();
+ String workerId = job.assignedWorkerId();
+ long overdueBy = now - job.deadlineMillis();
+ log.warn("Task execution deadline expired: jobId={} workerId={} attempt={}/{} "
+ + "timeoutMillis={} overdueMillis={}",
+ job.id(), workerId, job.attempts(), job.maxAttempts(),
+ job.executionTimeoutMillis(), overdueBy);
+ revokeAttemptOnWorker(job.id(), attemptId, workerId, "deadline exceeded");
+ retryOrFail(job, "execution deadline exceeded ("
+ + job.executionTimeoutMillis() + " ms) on worker " + workerId);
+ }
+ }
+
+ /**
+ * Best-effort request that {@code workerId} stop executing {@code attemptId},
+ * and release the worker's capacity slot for it. Safe to call even if the
+ * send fails: the caller revokes the lease anyway, so any result the task
+ * still produces is stale-rejected. Shared by deadline expiry and
+ * client-requested cancellation.
+ */
+ private void revokeAttemptOnWorker(String jobId, String attemptId, String workerId, String reason) {
+ if (workerId == null || attemptId == null) {
+ return;
+ }
+ registry.find(workerId).ifPresent(session -> {
+ session.removeActiveJob(jobId);
+ try {
+ session.send(new TaskCancel(jobId, attemptId));
+ log.info("Sent TASK_CANCEL: jobId={} workerId={} attemptId={} reason=\"{}\"",
+ jobId, workerId, attemptId, reason);
+ } catch (IOException sendFailed) {
+ log.debug("TASK_CANCEL to {} failed (harmless, lease already revoked): {}",
+ workerId, sendFailed.getMessage());
+ }
+ });
+ }
+
private void retryOrFail(Job job, String attemptError) {
long now = now();
if (job.hasAttemptsLeft()) {
diff --git a/coordinator/src/main/java/io/github/achrafaittayeb/dtp/coordinator/CoordinatorMain.java b/coordinator/src/main/java/io/github/achrafaittayeb/dtp/coordinator/CoordinatorMain.java
index cfd9b8b..8934bac 100644
--- a/coordinator/src/main/java/io/github/achrafaittayeb/dtp/coordinator/CoordinatorMain.java
+++ b/coordinator/src/main/java/io/github/achrafaittayeb/dtp/coordinator/CoordinatorMain.java
@@ -25,6 +25,7 @@ public static void main(String[] args) throws Exception {
--client-port default 7071
--heartbeat-timeout-millis default 6000
--sweep-interval-millis default 500
+ --task-timeout-millis default 600000 (per-attempt execution deadline)
--max-attempts default 3
--retry-base-delay-millis default 1000
--retry-max-delay-millis default 30000
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 66073eb..11d01ce 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
@@ -21,35 +21,41 @@ public final class Job {
private final TaskType taskType;
private final JsonNode payload;
private final int maxAttempts;
+ private final long executionTimeoutMillis;
private final long createdAtMillis;
private JobState state;
private int attempts;
private String currentAttemptId;
private String assignedWorkerId;
+ private long deadlineMillis;
private long nextEligibleTimeMillis;
private String result;
private String error;
private long updatedAtMillis;
- public static Job createQueued(TaskType taskType, JsonNode payload, int maxAttempts, long now) {
+ public static Job createQueued(TaskType taskType, JsonNode payload, int maxAttempts,
+ long executionTimeoutMillis, long now) {
return new Job(UUID.randomUUID().toString(), taskType, payload, maxAttempts,
- JobState.QUEUED, 0, null, null, 0, null, null, now, now);
+ executionTimeoutMillis, JobState.QUEUED, 0, null, null, 0, 0, null, null, now, now);
}
/** Full-field constructor used when rehydrating from persistent storage. */
public Job(String id, TaskType taskType, JsonNode payload, int maxAttempts,
- JobState state, int attempts, String currentAttemptId, String assignedWorkerId,
- long nextEligibleTimeMillis, String result, String error,
+ long executionTimeoutMillis, JobState state, int attempts,
+ String currentAttemptId, String assignedWorkerId,
+ long deadlineMillis, long nextEligibleTimeMillis, String result, String error,
long createdAtMillis, long updatedAtMillis) {
this.id = id;
this.taskType = taskType;
this.payload = payload;
this.maxAttempts = maxAttempts;
+ this.executionTimeoutMillis = executionTimeoutMillis;
this.state = state;
this.attempts = attempts;
this.currentAttemptId = currentAttemptId;
this.assignedWorkerId = assignedWorkerId;
+ this.deadlineMillis = deadlineMillis;
this.nextEligibleTimeMillis = nextEligibleTimeMillis;
this.result = result;
this.error = error;
@@ -59,13 +65,16 @@ public Job(String id, TaskType taskType, JsonNode payload, int maxAttempts,
/**
* QUEUED → RUNNING. Consumes one attempt and issues a fresh attempt lease;
- * only a result carrying this lease id will ever be accepted.
+ * only a result carrying this lease id will ever be accepted. The lease is
+ * time-bounded: it expires at {@code now + executionTimeoutMillis}, measured
+ * exclusively on the coordinator's clock.
*/
public String assignTo(String workerId, long now) {
transition(JobState.RUNNING, now);
attempts++;
currentAttemptId = UUID.randomUUID().toString();
assignedWorkerId = workerId;
+ deadlineMillis = now + executionTimeoutMillis;
return currentAttemptId;
}
@@ -107,10 +116,31 @@ public void requeueForRecovery(long now) {
clearLease();
}
+ /**
+ * QUEUED / RETRY_WAIT / RUNNING → CANCELLED. Client-requested termination;
+ * any active attempt lease is discarded, so a late result from a worker
+ * that keeps computing fails the lease check and is rejected.
+ */
+ public void cancel(long now) {
+ transition(JobState.CANCELLED, now);
+ error = "cancelled by client request";
+ nextEligibleTimeMillis = 0;
+ clearLease();
+ }
+
public boolean hasAttemptsLeft() {
return attempts < maxAttempts;
}
+ /**
+ * True when this job's current attempt has outlived its execution deadline.
+ * Expiry is suspicion that the task is stuck, not proof the worker died —
+ * the worker may be healthy and heartbeating while one task wedges.
+ */
+ public boolean isDeadlineExpired(long now) {
+ return state == JobState.RUNNING && now > deadlineMillis;
+ }
+
/** True when a reported result belongs to this job's currently leased attempt. */
public boolean isCurrentAttempt(String attemptId) {
return state == JobState.RUNNING
@@ -135,6 +165,7 @@ private void transition(JobState target, long now) {
private void clearLease() {
currentAttemptId = null;
assignedWorkerId = null;
+ deadlineMillis = 0;
}
public JobSnapshot snapshot() {
@@ -158,6 +189,15 @@ public int maxAttempts() {
return maxAttempts;
}
+ public long executionTimeoutMillis() {
+ return executionTimeoutMillis;
+ }
+
+ /** Coordinator-clock instant at which the current attempt's lease expires; 0 when not RUNNING. */
+ public long deadlineMillis() {
+ return deadlineMillis;
+ }
+
public JobState state() {
return state;
}
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 c3183e3..fab6601 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
@@ -33,15 +33,17 @@ public final class SqliteJobRepository implements JobRepository {
private static final Logger log = LoggerFactory.getLogger(SqliteJobRepository.class);
private static final String UPSERT = """
- INSERT INTO jobs (id, task_type, payload, max_attempts, state, attempts,
- current_attempt_id, assigned_worker_id, next_eligible_time,
- result, error, created_at, updated_at)
- VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+ INSERT INTO jobs (id, task_type, payload, max_attempts, execution_timeout,
+ state, attempts, current_attempt_id, assigned_worker_id,
+ deadline, next_eligible_time, result, error,
+ created_at, updated_at)
+ VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(id) DO UPDATE SET
state = excluded.state,
attempts = excluded.attempts,
current_attempt_id = excluded.current_attempt_id,
assigned_worker_id = excluded.assigned_worker_id,
+ deadline = excluded.deadline,
next_eligible_time = excluded.next_eligible_time,
result = excluded.result,
error = excluded.error,
@@ -72,10 +74,12 @@ CREATE TABLE IF NOT EXISTS jobs (
task_type TEXT NOT NULL,
payload TEXT NOT NULL,
max_attempts INTEGER NOT NULL,
+ execution_timeout INTEGER NOT NULL DEFAULT 600000,
state TEXT NOT NULL,
attempts INTEGER NOT NULL,
current_attempt_id TEXT,
assigned_worker_id TEXT,
+ deadline INTEGER NOT NULL DEFAULT 0,
next_eligible_time INTEGER NOT NULL,
result TEXT,
error TEXT,
@@ -83,6 +87,25 @@ CREATE TABLE IF NOT EXISTS jobs (
updated_at INTEGER NOT NULL
)""");
}
+ // Databases created before execution deadlines existed lack these two
+ // columns; add them in place so old job history remains loadable.
+ ensureColumn("execution_timeout", "INTEGER NOT NULL DEFAULT 600000");
+ ensureColumn("deadline", "INTEGER NOT NULL DEFAULT 0");
+ }
+
+ private void ensureColumn(String column, String definition) throws SQLException {
+ try (Statement statement = connection.createStatement();
+ ResultSet columns = statement.executeQuery("PRAGMA table_info(jobs)")) {
+ while (columns.next()) {
+ if (column.equals(columns.getString("name"))) {
+ return;
+ }
+ }
+ }
+ try (Statement statement = connection.createStatement()) {
+ statement.execute("ALTER TABLE jobs ADD COLUMN " + column + " " + definition);
+ log.info("Migrated jobs table: added column {}", column);
+ }
}
@Override
@@ -92,15 +115,17 @@ public void save(Job job) {
statement.setString(2, job.taskType().name());
statement.setString(3, MessageIO.mapper().writeValueAsString(job.payload()));
statement.setInt(4, job.maxAttempts());
- statement.setString(5, job.state().name());
- statement.setInt(6, job.attempts());
- statement.setString(7, job.currentAttemptId());
- statement.setString(8, job.assignedWorkerId());
- statement.setLong(9, job.nextEligibleTimeMillis());
- statement.setString(10, job.result());
- statement.setString(11, job.error());
- statement.setLong(12, job.createdAtMillis());
- statement.setLong(13, job.updatedAtMillis());
+ statement.setLong(5, job.executionTimeoutMillis());
+ statement.setString(6, job.state().name());
+ statement.setInt(7, job.attempts());
+ statement.setString(8, job.currentAttemptId());
+ statement.setString(9, job.assignedWorkerId());
+ statement.setLong(10, job.deadlineMillis());
+ statement.setLong(11, job.nextEligibleTimeMillis());
+ statement.setString(12, job.result());
+ statement.setString(13, job.error());
+ statement.setLong(14, job.createdAtMillis());
+ statement.setLong(15, job.updatedAtMillis());
statement.executeUpdate();
} catch (SQLException | JsonProcessingException e) {
// Losing durability silently would break recovery guarantees; fail loudly.
@@ -129,10 +154,12 @@ private static Job rowToJob(ResultSet row) throws SQLException, IOException {
TaskType.valueOf(row.getString("task_type")),
payload,
row.getInt("max_attempts"),
+ row.getLong("execution_timeout"),
JobState.valueOf(row.getString("state")),
row.getInt("attempts"),
row.getString("current_attempt_id"),
row.getString("assigned_worker_id"),
+ row.getLong("deadline"),
row.getLong("next_eligible_time"),
row.getString("result"),
row.getString("error"),
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 023633e..a8ff639 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
@@ -2,8 +2,10 @@
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;
@@ -110,6 +112,12 @@ private Message handle(Message request) {
.orElseGet(() -> new ErrorReply("Unknown job: " + status.jobId()));
case ListJobs ignored -> new JobListReply(core.listJobs());
case ListWorkers ignored -> new WorkerListReply(core.listWorkers());
+ case CancelJob cancel -> {
+ CoordinatorCore.CancelOutcome outcome = core.cancelJob(cancel.jobId());
+ yield outcome.found()
+ ? new JobCancelReply(outcome.cancelledNow(), outcome.job())
+ : new ErrorReply("Unknown job: " + cancel.jobId());
+ }
default -> new ErrorReply("Unsupported request: "
+ request.getClass().getSimpleName());
};
diff --git a/coordinator/src/test/java/io/github/achrafaittayeb/dtp/coordinator/job/JobTest.java b/coordinator/src/test/java/io/github/achrafaittayeb/dtp/coordinator/job/JobTest.java
index c47388b..320cd9b 100644
--- a/coordinator/src/test/java/io/github/achrafaittayeb/dtp/coordinator/job/JobTest.java
+++ b/coordinator/src/test/java/io/github/achrafaittayeb/dtp/coordinator/job/JobTest.java
@@ -10,10 +10,12 @@
class JobTest {
+ private static final long TIMEOUT = 60_000L;
+
private static Job newJob(int maxAttempts) {
return Job.createQueued(TaskType.SLEEP,
JsonNodeFactory.instance.objectNode().put("durationMillis", 100),
- maxAttempts, 1_000L);
+ maxAttempts, TIMEOUT, 1_000L);
}
@Test
@@ -83,6 +85,57 @@ void attemptBudgetIsEnforced() {
assertThat(job.hasAttemptsLeft()).isFalse();
}
+ @Test
+ void assignmentSetsCoordinatorClockDeadline() {
+ Job job = newJob(3);
+ job.assignTo("worker-1", 2_000L);
+
+ assertThat(job.deadlineMillis()).isEqualTo(2_000L + TIMEOUT);
+ assertThat(job.isDeadlineExpired(2_000L + TIMEOUT)).isFalse();
+ assertThat(job.isDeadlineExpired(2_001L + TIMEOUT)).isTrue();
+
+ job.complete("done", 3_000L);
+ assertThat(job.deadlineMillis()).isZero(); // deadline dies with the lease
+ assertThat(job.isDeadlineExpired(Long.MAX_VALUE)).isFalse();
+ }
+
+ @Test
+ void cancellationRevokesLeaseFromAnyNonTerminalState() {
+ Job queued = newJob(3);
+ queued.cancel(2_000L);
+ assertThat(queued.state()).isEqualTo(JobState.CANCELLED);
+ assertThat(queued.error()).contains("cancelled");
+
+ Job running = newJob(3);
+ String lease = running.assignTo("worker-1", 2_000L);
+ running.cancel(3_000L);
+ assertThat(running.state()).isEqualTo(JobState.CANCELLED);
+ assertThat(running.isCurrentAttempt(lease)).isFalse();
+ assertThat(running.assignedWorkerId()).isNull();
+ assertThat(running.deadlineMillis()).isZero();
+
+ Job retryWait = newJob(3);
+ retryWait.assignTo("worker-1", 2_000L);
+ retryWait.scheduleRetry("boom", 9_000L, 2_500L);
+ retryWait.cancel(3_000L);
+ assertThat(retryWait.state()).isEqualTo(JobState.CANCELLED);
+ assertThat(retryWait.isEligibleToRun(Long.MAX_VALUE)).isFalse();
+ }
+
+ @Test
+ void terminalJobsCannotBeCancelled() {
+ Job completed = newJob(3);
+ completed.assignTo("w", 1L);
+ completed.complete("ok", 2L);
+ assertThatThrownBy(() -> completed.cancel(3L))
+ .isInstanceOf(IllegalStateException.class);
+
+ Job cancelled = newJob(3);
+ cancelled.cancel(1L);
+ assertThatThrownBy(() -> cancelled.cancel(2L))
+ .isInstanceOf(IllegalStateException.class);
+ }
+
@Test
void illegalTransitionsThrow() {
Job queued = newJob(3);
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 71aab9b..fdf520b 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
@@ -22,7 +22,7 @@ private String dbPath() {
@Test
void savedJobsSurviveRepositoryReopen() {
Job job = Job.createQueued(TaskType.WORD_COUNT,
- JsonNodeFactory.instance.objectNode().put("text", "a b c"), 3, 42L);
+ JsonNodeFactory.instance.objectNode().put("text", "a b c"), 3, 5_000L, 42L);
job.assignTo("worker-9", 50L);
try (SqliteJobRepository repository = new SqliteJobRepository(dbPath())) {
@@ -38,6 +38,8 @@ void savedJobsSurviveRepositoryReopen() {
assertThat(loaded.attempts()).isEqualTo(1);
assertThat(loaded.assignedWorkerId()).isEqualTo("worker-9");
assertThat(loaded.currentAttemptId()).isEqualTo(job.currentAttemptId());
+ assertThat(loaded.executionTimeoutMillis()).isEqualTo(5_000L);
+ assertThat(loaded.deadlineMillis()).isEqualTo(50L + 5_000L);
assertThat(loaded.payload().get("text").asText()).isEqualTo("a b c");
assertThat(loaded.createdAtMillis()).isEqualTo(42L);
}
@@ -47,7 +49,7 @@ void savedJobsSurviveRepositoryReopen() {
void saveIsAnUpsertReflectingLatestState() {
try (SqliteJobRepository repository = new SqliteJobRepository(dbPath())) {
Job job = Job.createQueued(TaskType.SLEEP,
- JsonNodeFactory.instance.objectNode().put("durationMillis", 10), 1, 1L);
+ JsonNodeFactory.instance.objectNode().put("durationMillis", 10), 1, 5_000L, 1L);
repository.save(job);
job.assignTo("w", 2L);
repository.save(job);
@@ -62,13 +64,44 @@ void saveIsAnUpsertReflectingLatestState() {
}
}
+ @Test
+ void migratesLegacyDatabaseWithoutDeadlineColumns() throws Exception {
+ // A database written by the pre-deadline schema must remain loadable:
+ // the migration adds execution_timeout/deadline with safe defaults.
+ try (java.sql.Connection legacy =
+ java.sql.DriverManager.getConnection("jdbc:sqlite:" + dbPath());
+ java.sql.Statement statement = legacy.createStatement()) {
+ statement.execute("""
+ CREATE TABLE jobs (
+ id TEXT PRIMARY KEY, task_type TEXT NOT NULL, payload TEXT NOT NULL,
+ max_attempts INTEGER NOT NULL, state TEXT NOT NULL,
+ attempts INTEGER NOT NULL, current_attempt_id TEXT,
+ assigned_worker_id TEXT, next_eligible_time INTEGER NOT NULL,
+ result TEXT, error TEXT,
+ created_at INTEGER NOT NULL, updated_at INTEGER NOT NULL
+ )""");
+ statement.execute("""
+ INSERT INTO jobs VALUES ('legacy-job', 'SHA256', '{"text":"abc"}',
+ 3, 'COMPLETED', 1, NULL, NULL, 0, 'old-result', NULL, 10, 20)""");
+ }
+
+ try (SqliteJobRepository migrated = new SqliteJobRepository(dbPath())) {
+ Job loaded = migrated.loadAll().getFirst();
+ assertThat(loaded.id()).isEqualTo("legacy-job");
+ assertThat(loaded.state()).isEqualTo(JobState.COMPLETED);
+ assertThat(loaded.result()).isEqualTo("old-result");
+ assertThat(loaded.executionTimeoutMillis()).isEqualTo(600_000L); // migration default
+ assertThat(loaded.deadlineMillis()).isZero();
+ }
+ }
+
@Test
void loadsJobsInCreationOrder() {
try (SqliteJobRepository repository = new SqliteJobRepository(dbPath())) {
Job second = Job.createQueued(TaskType.SLEEP,
- JsonNodeFactory.instance.objectNode().put("durationMillis", 1), 1, 200L);
+ JsonNodeFactory.instance.objectNode().put("durationMillis", 1), 1, 5_000L, 200L);
Job first = Job.createQueued(TaskType.SLEEP,
- JsonNodeFactory.instance.objectNode().put("durationMillis", 1), 1, 100L);
+ JsonNodeFactory.instance.objectNode().put("durationMillis", 1), 1, 5_000L, 100L);
repository.save(second);
repository.save(first);
diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md
index e1c656f..d703366 100644
--- a/docs/ARCHITECTURE.md
+++ b/docs/ARCHITECTURE.md
@@ -27,7 +27,8 @@ and every `Job`. It runs on one dedicated thread (a single-thread
- client connection handlers submit queries and block for the answer
(`askCore`);
- the periodic **sweep** (default every 500 ms) runs on the core thread
- itself: detect dead workers → promote due retries → schedule queued jobs.
+ itself: detect dead workers → detect expired execution deadlines → promote
+ due retries → schedule queued jobs.
**Invariant:** because scheduling, result handling, failure detection, and
recovery are all serialized on one thread, no interleaving can assign one job
@@ -54,6 +55,12 @@ DESIGN_DECISIONS.
session removed and closed, each of its RUNNING jobs retried or failed.
A dropped connection triggers the same path immediately via the reader
thread's disconnect event.
+- **Deadline enforcer** (`detectExpiredDeadlines`) — a RUNNING job past its
+ per-attempt deadline (`assignedAt + task-timeout-millis`, coordinator clock)
+ has its lease revoked, a best-effort `TASK_CANCEL` sent to the owning worker
+ (`revokeAttemptOnWorker`), and is retried or failed. This is the only
+ detector that catches a stuck task on a *live* worker; it shares the retry
+ and stale-result machinery with worker-loss handling.
- **RetryPolicy** — delay = `base × 2^(attempts-1)`, capped (defaults 1 s
base, 30 s cap). The attempt *budget* lives on the job (`maxAttempts`,
default 3, settable per job).
diff --git a/docs/DESIGN_DECISIONS.md b/docs/DESIGN_DECISIONS.md
index 173ec30..929b890 100644
--- a/docs/DESIGN_DECISIONS.md
+++ b/docs/DESIGN_DECISIONS.md
@@ -116,3 +116,47 @@ backoff, and attempt budgets are all flags with documented defaults.
Integration tests shrink them (100 ms heartbeats, 600 ms timeout) so the full
failure suite runs in seconds without fake clocks — the same code paths run in
tests and in the live demo, just faster.
+
+## 13. Time-bounded attempt leases (execution deadlines + cooperative cancellation)
+
+The attempt lease started as an identity (a UUID that fences stale results).
+It is now also a *time-bounded contract*: each assignment carries a deadline,
+and an expired lease is revoked exactly like a lost worker. This is the
+canonical distributed-systems lease (Gray & Cheriton) — the identity and the
+expiry are two faces of one primitive — so it deepened the system's strongest
+existing idea rather than adding a separate subsystem.
+
+Why it was needed: heartbeat detection is worker-granular. A worker can be
+perfectly alive and heartbeating while one task wedges, holding a capacity slot
+forever. Nothing else in the system could observe that. The deadline is the
+per-task progress signal heartbeats can't provide.
+
+Deliberate choices and their trade-offs:
+
+- **Coordinator clock only.** Expiry is judged against `System.currentTimeMillis`
+ on the coordinator; worker clocks are never read or compared. This avoids a
+ clock-synchronization dependency entirely, at the cost of the deadline
+ measuring wall-clock-since-assignment (including transit and queueing), not
+ pure on-worker execution time. For a timeout whose job is to catch "stuck",
+ that over-approximation is the safe direction.
+- **Reuse, not new machinery.** Deadline expiry funnels into the existing
+ `retryOrFail`, and a revoked lease is stale-rejected by the existing result
+ check. The only genuinely new code is a sweep predicate, a `TASK_CANCEL`
+ message, and worker-side interruption. No new rejection path, no new retry
+ path — fewer states to reason about.
+- **Cooperative cancellation, not preemption.** `TASK_CANCEL` interrupts the
+ task's thread; a task that ignores interruption keeps running. Hard
+ preemption (`Thread.stop`) is unsafe and was rejected. Correctness never
+ depends on the worker obeying — the lease is revoked coordinator-side first,
+ so an uncooperative task can only produce a stale, rejected result. This is
+ why the cancel can be honestly documented as best-effort without weakening
+ any guarantee.
+- **Recovery discards deadlines.** A persisted deadline is a dead process's
+ clock reading; recovery requeues RUNNING jobs and lets the fresh assignment
+ set a fresh deadline, so a stale deadline can never instant-expire a
+ recovered job.
+
+What this explicitly does **not** buy: it is not exactly-once (a timed-out task
+that actually finished still causes a retry — the at-least-once window is
+unchanged), and it is not a real-time guarantee (detection latency is bounded
+only by the sweep interval).
diff --git a/docs/FAILURE_MODEL.md b/docs/FAILURE_MODEL.md
index 91b38ac..dee6ac3 100644
--- a/docs/FAILURE_MODEL.md
+++ b/docs/FAILURE_MODEL.md
@@ -33,6 +33,33 @@ and takes the same invalidation path. Two honest caveats:
reports, its `attemptId` no longer matches and the result is logged
(`Stale result rejected`) and dropped.
+## Stuck task on a healthy worker (execution deadline)
+
+A worker being alive does **not** mean every task on it is making progress: a
+task can wedge (an infinite loop, a stuck I/O call) while the worker keeps
+heartbeating normally. Heartbeat detection cannot catch this — the worker is
+not dead. Execution deadlines close the gap.
+
+Each attempt is leased with a deadline of `assignedAt + executionTimeout`
+(`--task-timeout-millis`, default 10 min), measured **only on the
+coordinator's clock** — worker clocks are never consulted, so no clock
+synchronization is required. When the sweep finds a RUNNING job past its
+deadline it:
+
+1. revokes the lease (so any later result is stale-rejected);
+2. sends a best-effort `TASK_CANCEL` to the owning worker and frees its slot;
+3. retries the job (backoff) or fails it if the attempt budget is spent.
+
+Honest caveats, mirroring heartbeat detection:
+
+- A deadline is a *contract*, and expiry is *suspicion* that the task exceeded
+ it — not proof the task is broken. Set the timeout above legitimate task
+ durations; too low means healthy long tasks get retried forever and end
+ FAILED with an "execution deadline exceeded" error.
+- Cancellation is cooperative (thread interruption). A task that ignores
+ interruption keeps consuming CPU until it finishes, but its lease is already
+ gone, so it cannot affect job state.
+
## The classic at-least-once window
Worker finishes the task → dies before `TASK_RESULT` is read by the
@@ -54,7 +81,22 @@ fails to hold. Covered cases (all integration-tested):
- result for an attempt superseded by a retry (zombie worker returns);
- duplicate result for an attempt that already completed;
- result for a job the coordinator no longer considers running;
-- result for an unknown job id.
+- result for an unknown job id;
+- result from an attempt whose deadline expired and was reassigned;
+- result from an attempt whose job was cancelled.
+
+The last two are the same mechanism reached through the new triggers, and are
+integration-tested.
+
+## Client cancellation
+
+A client may cancel any non-terminal job. QUEUED and RETRY_WAIT jobs move
+straight to the terminal `CANCELLED` state. A RUNNING job additionally has its
+lease revoked and a best-effort `TASK_CANCEL` sent to the worker, then becomes
+CANCELLED. Cancelling an already-terminal job is a no-op that reports the job's
+real final state; cancelling an unknown id is an error. Because cancellation
+revokes the lease, a worker that keeps running the cancelled task and later
+reports success is stale-rejected — a cancelled job never comes back.
## Worker re-registration
@@ -76,6 +118,13 @@ Job state is written through to SQLite before it is client-visible, so:
fails them if it was the final attempt, with an explicit error saying so.
This is the at-least-once trade-off again, applied across restarts.
+The persisted `deadline` is a coordinator-clock instant from the **dead**
+process, so it means nothing to the new one. Recovery never treats a loaded
+`RUNNING` job as still-leased: it requeues (discarding the stale deadline
+along with the lease), and the fresh assignment gets a fresh deadline. A stale
+deadline can therefore never "instantly expire" a recovered job — verified by
+`CoordinatorRestartIT.recoveryIgnoresStaleDeadlineAndRequeuesInsteadOfExpiring`.
+
Workers are not persisted (their sessions cannot survive anyway); they
reconnect and re-register on their own retry loop. A worker that finished a
task during the outage will have its result rejected as stale, because
@@ -104,7 +153,8 @@ While the coordinator is down the system is unavailable — see below.
|---|---|---|
| worker `--heartbeat-interval-millis` | 2000 | heartbeat period |
| coordinator `--heartbeat-timeout-millis` | 6000 | silence before suspected dead |
-| coordinator `--sweep-interval-millis` | 500 | failure scan + retry promotion + scheduling period |
+| 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 `--retry-base-delay-millis` / `--retry-max-delay-millis` | 1000 / 30000 | exponential backoff bounds |
diff --git a/docs/PROTOCOL.md b/docs/PROTOCOL.md
index a4824da..2fe11f5 100644
--- a/docs/PROTOCOL.md
+++ b/docs/PROTOCOL.md
@@ -36,6 +36,7 @@ protocol error that closes *that connection only*.
| `WORKER_REGISTERED` | C→W | `workerId` | acknowledgement |
| `HEARTBEAT` | W→C | `workerId` | every 2 s by default |
| `TASK_ASSIGN` | C→W | `jobId`, `attemptId`, `attemptNumber`, `taskType`, `payload` | `attemptId` is the lease |
+| `TASK_CANCEL` | C→W | `jobId`, `attemptId` | cancel that specific attempt; ignored unless both ids match the running one |
| `TASK_RESULT` | W→C | `workerId`, `jobId`, `attemptId`, `success`, `result`, `error` | must echo the lease |
| `ERROR` | C→W | `message` | e.g. registering wrong; connection then closes |
@@ -43,6 +44,13 @@ There is no explicit `TASK_ACCEPTED`: assignment rides a healthy TCP
connection, the coordinator tracks capacity itself, and a send failure or
connection loss invalidates the attempt anyway (see DESIGN_DECISIONS).
+`TASK_CANCEL` is sent when an attempt's lease is revoked — either its execution
+deadline expired or a client cancelled the job. It is best-effort and advisory:
+the coordinator has already revoked the lease, so whether or not the worker
+acts on it, any result the attempt later produces is stale-rejected. The worker
+cancels only if **both** `jobId` and `attemptId` match its currently executing
+assignment, so a cancel for a superseded attempt can never interrupt a newer one.
+
### Client port (default 7071)
Strict request/response: one reply frame per request frame, many requests per
@@ -54,12 +62,13 @@ connection.
| `GET_JOB_STATUS` (`jobId`) | `JOB_STATUS` (job snapshot) | unknown id → `ERROR` |
| `LIST_JOBS` | `JOB_LIST` (snapshots, newest first) | |
| `LIST_WORKERS` | `WORKER_LIST` (live workers) | |
+| `CANCEL_JOB` (`jobId`) | `JOB_CANCEL` (`cancelled`, job snapshot) | `cancelled=true` if this call moved it to CANCELLED; `false` if already terminal (snapshot shows real state); unknown id → `ERROR` |
| any invalid | `ERROR` (`message`) | malformed frames get a best-effort `ERROR`, then close |
-A job snapshot contains: `jobId`, `taskType`, `state`, `attempts`,
-`maxAttempts`, `workerId` (only while RUNNING), `result`, `error` (last
-attempt's error, kept for history even after later success), `createdAtMillis`,
-`updatedAtMillis`.
+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`.
## 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 806238d..f02f016 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
@@ -106,6 +106,48 @@ void recoveryFailsJobWhoseFinalAttemptWasInterrupted() throws Exception {
}
}
+ @Test
+ void recoveryIgnoresStaleDeadlineAndRequeuesInsteadOfExpiring() throws Exception {
+ // A RUNNING job's deadline is a coordinator-clock instant from the old
+ // process. Recovery must NOT resurrect that assignment and let its
+ // deadline "expire" it — it requeues (lease discarded) like any other
+ // in-flight job, and the fresh assignment gets a fresh deadline.
+ String database = tempDir.resolve("stale-deadline.db").toString();
+ String jobId;
+
+ // Tiny task timeout: if recovery wrongly kept the old deadline, the job
+ // would already be "expired" the instant it loads.
+ Coordinator first = Testbed.startCoordinator(database, 3, 200);
+ ScriptedWorker worker = ScriptedWorker.register(first.workerPort(), "w1", 1);
+ try (CoordinatorClient client = Testbed.connectClient(first)) {
+ jobId = client.submit(TaskType.SLEEP,
+ JsonNodeFactory.instance.objectNode().put("durationMillis", 50), 3);
+ worker.awaitAssignment();
+ assertThat(client.status(jobId).state()).isEqualTo(JobState.RUNNING);
+ } finally {
+ first.close();
+ worker.close();
+ }
+
+ // Let real time pass the old deadline before the new coordinator starts.
+ Thread.sleep(400);
+
+ try (Coordinator second = Testbed.startCoordinator(database, 3, 200);
+ CoordinatorClient client = Testbed.connectClient(second)) {
+ // Requeued by the recovery rule, not failed by a stale deadline.
+ JobSnapshot recovered = client.status(jobId);
+ assertThat(recovered.state()).isEqualTo(JobState.QUEUED);
+ assertThat(recovered.attempts()).isEqualTo(1);
+
+ try (Worker realWorker = Testbed.startWorker(second, "w2", 1)) {
+ JobSnapshot done = client.awaitTerminal(
+ jobId, Testbed.POLL, Testbed.TERMINAL_TIMEOUT);
+ assertThat(done.state()).isEqualTo(JobState.COMPLETED);
+ assertThat(done.attempts()).isEqualTo(2);
+ }
+ }
+ }
+
@Test
void workerAutomaticallyReconnectsToRestartedCoordinatorOnSamePorts() throws Exception {
String database = tempDir.resolve("reconnect.db").toString();
@@ -131,7 +173,7 @@ void workerAutomaticallyReconnectsToRestartedCoordinatorOnSamePorts() throws Exc
new io.github.achrafaittayeb.dtp.coordinator.CoordinatorConfig(
workerPort, clientPort,
Testbed.HEARTBEAT_TIMEOUT_MILLIS, Testbed.SWEEP_INTERVAL_MILLIS,
- 3, 50, 200, database))) {
+ Testbed.TASK_TIMEOUT_MILLIS, 3, 50, 200, 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/DeadlineAndCancellationIT.java b/integration-tests/src/test/java/io/github/achrafaittayeb/dtp/it/DeadlineAndCancellationIT.java
new file mode 100644
index 0000000..e0535b8
--- /dev/null
+++ b/integration-tests/src/test/java/io/github/achrafaittayeb/dtp/it/DeadlineAndCancellationIT.java
@@ -0,0 +1,234 @@
+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.JobSnapshot;
+import io.github.achrafaittayeb.dtp.common.model.JobState;
+import io.github.achrafaittayeb.dtp.common.model.TaskType;
+import io.github.achrafaittayeb.dtp.common.protocol.JobCancelReply;
+import io.github.achrafaittayeb.dtp.common.protocol.TaskAssign;
+import io.github.achrafaittayeb.dtp.common.protocol.TaskCancel;
+import io.github.achrafaittayeb.dtp.coordinator.Coordinator;
+import io.github.achrafaittayeb.dtp.coordinator.CoordinatorConfig;
+import io.github.achrafaittayeb.dtp.worker.Worker;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/**
+ * Time-bounded leases: execution deadlines close the gap where a task wedges on
+ * a worker that stays alive and heartbeating, and client cancellation lets a
+ * job be revoked on demand. Both reuse the existing attempt-lease machinery, so
+ * these tests also re-assert stale-result rejection under the new triggers.
+ */
+class DeadlineAndCancellationIT {
+
+ // Short execution timeout so deadline expiry is fast; still well above the
+ // sweep interval so the timing is deterministic, not a race.
+ private static final long SHORT_TASK_TIMEOUT = 400;
+
+ // ---- A. Execution deadlines --------------------------------------------
+
+ @Test
+ void taskExceedingDeadlineOnHealthyWorkerIsReassignedAndCompletes() throws Exception {
+ try (Coordinator coordinator = Testbed.startCoordinator(
+ CoordinatorConfig.IN_MEMORY_DATABASE, 3, SHORT_TASK_TIMEOUT);
+ CoordinatorClient client = Testbed.connectClient(coordinator)) {
+
+ // A scripted worker that keeps heartbeating (stays "alive") but never
+ // reports a result: exactly a wedged task on a healthy worker.
+ ScriptedWorker stuckButAlive = ScriptedWorker.register(
+ coordinator.workerPort(), "stuck-worker", 1);
+ String jobId = client.submit(TaskType.SLEEP,
+ JsonNodeFactory.instance.objectNode().put("durationMillis", 50), 0);
+ TaskAssign firstAttempt = stuckButAlive.awaitAssignment();
+ assertThat(client.status(jobId).state()).isEqualTo(JobState.RUNNING);
+
+ // The deadline expires while the worker is still heartbeating; the
+ // coordinator must send TASK_CANCEL and reassign.
+ TaskCancel cancel = stuckButAlive.awaitCancel();
+ assertThat(cancel.jobId()).isEqualTo(jobId);
+ assertThat(cancel.attemptId()).isEqualTo(firstAttempt.attemptId());
+
+ try (Worker healthy = Testbed.startWorker(coordinator, "healthy-worker", 1)) {
+ JobSnapshot done = client.awaitTerminal(jobId, Testbed.POLL, Testbed.TERMINAL_TIMEOUT);
+ assertThat(done.state()).isEqualTo(JobState.COMPLETED);
+ assertThat(done.result()).isEqualTo("slept 50 ms");
+ assertThat(done.attempts()).isEqualTo(2);
+ assertThat(done.error()).contains("execution deadline exceeded");
+ }
+ stuckButAlive.close();
+ }
+ }
+
+ @Test
+ void lateResultFromTimedOutAttemptIsRejectedAsStale() throws Exception {
+ try (Coordinator coordinator = Testbed.startCoordinator(
+ CoordinatorConfig.IN_MEMORY_DATABASE, 3, SHORT_TASK_TIMEOUT);
+ CoordinatorClient client = Testbed.connectClient(coordinator)) {
+
+ ScriptedWorker slow = ScriptedWorker.register(coordinator.workerPort(), "slow", 1);
+ String jobId = client.submit(TaskType.SLEEP,
+ JsonNodeFactory.instance.objectNode().put("durationMillis", 50), 0);
+ TaskAssign timedOut = slow.awaitAssignment();
+ slow.awaitCancel(); // deadline fired
+
+ // A healthy worker takes attempt 2 and completes it.
+ try (Worker healthy = Testbed.startWorker(coordinator, "healthy", 1)) {
+ JobSnapshot done = client.awaitTerminal(jobId, Testbed.POLL, Testbed.TERMINAL_TIMEOUT);
+ assertThat(done.state()).isEqualTo(JobState.COMPLETED);
+ assertThat(done.attempts()).isEqualTo(2);
+
+ // Now the timed-out attempt finally reports success — must be rejected.
+ slow.sendSuccess(timedOut, "late result from the wedged attempt");
+ Thread.sleep(300);
+
+ JobSnapshot after = client.status(jobId);
+ assertThat(after.state()).isEqualTo(JobState.COMPLETED);
+ assertThat(after.result()).isEqualTo("slept 50 ms");
+ assertThat(after.attempts()).isEqualTo(2);
+ }
+ slow.close();
+ }
+ }
+
+ @Test
+ void deadlineExpiryOnFinalAttemptFailsJobWithTimeoutError() throws Exception {
+ try (Coordinator coordinator = Testbed.startCoordinator(
+ CoordinatorConfig.IN_MEMORY_DATABASE, 1, SHORT_TASK_TIMEOUT);
+ CoordinatorClient client = Testbed.connectClient(coordinator);
+ ScriptedWorker stuck = ScriptedWorker.register(coordinator.workerPort(), "stuck", 1)) {
+
+ String jobId = client.submit(TaskType.SLEEP,
+ JsonNodeFactory.instance.objectNode().put("durationMillis", 50), 1);
+ stuck.awaitAssignment();
+
+ JobSnapshot done = client.awaitTerminal(jobId, Testbed.POLL, Testbed.TERMINAL_TIMEOUT);
+ assertThat(done.state()).isEqualTo(JobState.FAILED);
+ assertThat(done.attempts()).isEqualTo(1);
+ assertThat(done.error())
+ .contains("execution deadline exceeded")
+ .contains("all 1 attempts used");
+ }
+ }
+
+ // ---- B. Client-requested cancellation ----------------------------------
+
+ @Test
+ void cancellingQueuedJobPreventsExecution() throws Exception {
+ try (Coordinator coordinator = Testbed.startCoordinator(CoordinatorConfig.IN_MEMORY_DATABASE, 3);
+ CoordinatorClient client = Testbed.connectClient(coordinator)) {
+
+ // No workers online, so the job stays QUEUED.
+ String jobId = client.submit(TaskType.SHA256,
+ JsonNodeFactory.instance.objectNode().put("text", "never runs"), 0);
+ assertThat(client.status(jobId).state()).isEqualTo(JobState.QUEUED);
+
+ JobCancelReply reply = client.cancel(jobId);
+ assertThat(reply.cancelled()).isTrue();
+ assertThat(reply.job().state()).isEqualTo(JobState.CANCELLED);
+
+ // Even after a worker joins, a CANCELLED job is never scheduled.
+ try (Worker worker = Testbed.startWorker(coordinator, "worker-1", 1)) {
+ Thread.sleep(300);
+ assertThat(client.status(jobId).state()).isEqualTo(JobState.CANCELLED);
+ }
+ }
+ }
+
+ @Test
+ void cancellingRunningJobInterruptsRealWorkerAndRevokesLease() throws Exception {
+ try (Coordinator coordinator = Testbed.startCoordinator(CoordinatorConfig.IN_MEMORY_DATABASE, 3);
+ CoordinatorClient client = Testbed.connectClient(coordinator);
+ Worker worker = Testbed.startWorker(coordinator, "worker-1", 1)) {
+
+ Testbed.waitUntil("worker online", Testbed.TERMINAL_TIMEOUT,
+ () -> uncheckedWorkers(client) == 1);
+
+ // A long SLEEP that would otherwise run for 30 s.
+ String jobId = client.submit(TaskType.SLEEP,
+ JsonNodeFactory.instance.objectNode().put("durationMillis", 30_000), 0);
+ Testbed.waitUntil("job running", Testbed.TERMINAL_TIMEOUT,
+ () -> uncheckedStatus(client, jobId).state() == JobState.RUNNING);
+
+ long before = System.nanoTime();
+ JobCancelReply reply = client.cancel(jobId);
+ assertThat(reply.cancelled()).isTrue();
+ assertThat(reply.job().state()).isEqualTo(JobState.CANCELLED);
+
+ // The real worker's SLEEP was interrupted well before its 30 s runtime,
+ // and its capacity was freed (proven by a follow-up job completing).
+ long elapsedMillis = (System.nanoTime() - before) / 1_000_000;
+ assertThat(elapsedMillis).isLessThan(5_000);
+
+ String followUp = client.submit(TaskType.WORD_COUNT,
+ JsonNodeFactory.instance.objectNode().put("text", "one two"), 0);
+ assertThat(client.awaitTerminal(followUp, Testbed.POLL, Testbed.TERMINAL_TIMEOUT).state())
+ .isEqualTo(JobState.COMPLETED);
+
+ assertThat(client.status(jobId).state()).isEqualTo(JobState.CANCELLED);
+ }
+ }
+
+ @Test
+ void staleResultCannotResurrectACancelledJob() throws Exception {
+ try (Coordinator coordinator = Testbed.startCoordinator(CoordinatorConfig.IN_MEMORY_DATABASE, 3);
+ CoordinatorClient client = Testbed.connectClient(coordinator);
+ ScriptedWorker worker = ScriptedWorker.register(coordinator.workerPort(), "worker-1", 1)) {
+
+ String jobId = client.submit(TaskType.SLEEP,
+ JsonNodeFactory.instance.objectNode().put("durationMillis", 50), 0);
+ TaskAssign assignment = worker.awaitAssignment();
+
+ JobCancelReply reply = client.cancel(jobId);
+ assertThat(reply.cancelled()).isTrue();
+ worker.awaitCancel();
+
+ // The worker ignores the cancel and reports success anyway.
+ worker.sendSuccess(assignment, "result after cancel");
+ Thread.sleep(300);
+
+ JobSnapshot after = client.status(jobId);
+ assertThat(after.state()).isEqualTo(JobState.CANCELLED);
+ assertThat(after.result()).isNull();
+ }
+ }
+
+ @Test
+ void cancellingTerminalJobsIsReportedNotApplied() throws Exception {
+ try (Coordinator coordinator = Testbed.startCoordinator(CoordinatorConfig.IN_MEMORY_DATABASE, 3);
+ CoordinatorClient client = Testbed.connectClient(coordinator);
+ Worker worker = Testbed.startWorker(coordinator, "worker-1", 1)) {
+
+ String jobId = client.submit(TaskType.SHA256,
+ JsonNodeFactory.instance.objectNode().put("text", "abc"), 0);
+ client.awaitTerminal(jobId, Testbed.POLL, Testbed.TERMINAL_TIMEOUT);
+
+ // Cancelling a COMPLETED job is a no-op that reports the real state.
+ JobCancelReply reply = client.cancel(jobId);
+ assertThat(reply.cancelled()).isFalse();
+ assertThat(reply.job().state()).isEqualTo(JobState.COMPLETED);
+
+ // Cancelling an unknown job is an error.
+ assertThatThrownBy(() -> client.cancel("no-such-job"))
+ .hasMessageContaining("Unknown job");
+ }
+ }
+
+ private static int uncheckedWorkers(CoordinatorClient client) {
+ try {
+ return client.listWorkers().size();
+ } catch (java.io.IOException e) {
+ throw new AssertionError(e);
+ }
+ }
+
+ private static JobSnapshot uncheckedStatus(CoordinatorClient client, String jobId) {
+ try {
+ return client.status(jobId);
+ } catch (java.io.IOException e) {
+ throw new AssertionError(e);
+ }
+ }
+}
diff --git a/integration-tests/src/test/java/io/github/achrafaittayeb/dtp/it/ScriptedWorker.java b/integration-tests/src/test/java/io/github/achrafaittayeb/dtp/it/ScriptedWorker.java
index 3c60bed..4392ead 100644
--- a/integration-tests/src/test/java/io/github/achrafaittayeb/dtp/it/ScriptedWorker.java
+++ b/integration-tests/src/test/java/io/github/achrafaittayeb/dtp/it/ScriptedWorker.java
@@ -4,6 +4,7 @@
import io.github.achrafaittayeb.dtp.common.protocol.Heartbeat;
import io.github.achrafaittayeb.dtp.common.protocol.Message;
import io.github.achrafaittayeb.dtp.common.protocol.TaskAssign;
+import io.github.achrafaittayeb.dtp.common.protocol.TaskCancel;
import io.github.achrafaittayeb.dtp.common.protocol.TaskResult;
import io.github.achrafaittayeb.dtp.common.protocol.WorkerRegister;
import io.github.achrafaittayeb.dtp.common.protocol.WorkerRegistered;
@@ -31,6 +32,7 @@ final class ScriptedWorker implements AutoCloseable {
private final Socket socket;
private final OutputStream out;
private final BlockingQueue assignments = new LinkedBlockingQueue<>();
+ private final BlockingQueue cancels = new LinkedBlockingQueue<>();
private final ScheduledExecutorService heartbeater =
Executors.newSingleThreadScheduledExecutor();
private final Thread readerThread;
@@ -66,6 +68,8 @@ private void readLoop(InputStream in) {
while ((message = MessageIO.receive(in)) != null) {
if (message instanceof TaskAssign assign) {
assignments.add(assign);
+ } else if (message instanceof TaskCancel cancel) {
+ cancels.add(cancel);
}
}
} catch (IOException endOfConnection) {
@@ -101,6 +105,15 @@ java.util.List drainAssignments() {
return drained;
}
+ /** Blocks until a TASK_CANCEL arrives, or fails the test after the timeout. */
+ TaskCancel awaitCancel() throws InterruptedException {
+ TaskCancel cancel = cancels.poll(10, TimeUnit.SECONDS);
+ if (cancel == null) {
+ throw new AssertionError("Scripted worker " + workerId + " received no TASK_CANCEL");
+ }
+ return cancel;
+ }
+
/** Simulates a hung process: the TCP connection stays open but liveness stops. */
void stopHeartbeats() {
heartbeating = false;
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 316ce83..a6fafc6 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
@@ -20,6 +20,8 @@ final class Testbed {
static final long HEARTBEAT_TIMEOUT_MILLIS = 600;
static final long SWEEP_INTERVAL_MILLIS = 50;
+ /** Generous default so execution deadlines never fire in tests that aren't about them. */
+ static final long TASK_TIMEOUT_MILLIS = 60_000;
static final long WORKER_HEARTBEAT_MILLIS = 100;
static final Duration POLL = Duration.ofMillis(50);
static final Duration TERMINAL_TIMEOUT = Duration.ofSeconds(15);
@@ -29,9 +31,15 @@ private Testbed() {
/** Coordinator on ephemeral ports; {@code database} may be {@code :memory:} or a temp file. */
static Coordinator startCoordinator(String database, int maxAttempts) throws IOException {
+ return startCoordinator(database, maxAttempts, TASK_TIMEOUT_MILLIS);
+ }
+
+ static Coordinator startCoordinator(String database, int maxAttempts, long taskTimeoutMillis)
+ throws IOException {
Coordinator coordinator = new Coordinator(new CoordinatorConfig(
0, 0,
HEARTBEAT_TIMEOUT_MILLIS, SWEEP_INTERVAL_MILLIS,
+ taskTimeoutMillis,
maxAttempts,
50, 200,
database));
diff --git a/worker/src/main/java/io/github/achrafaittayeb/dtp/worker/InFlightTasks.java b/worker/src/main/java/io/github/achrafaittayeb/dtp/worker/InFlightTasks.java
new file mode 100644
index 0000000..caaee01
--- /dev/null
+++ b/worker/src/main/java/io/github/achrafaittayeb/dtp/worker/InFlightTasks.java
@@ -0,0 +1,53 @@
+package io.github.achrafaittayeb.dtp.worker;
+
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Future;
+
+/**
+ * Tracks the tasks a worker is currently executing, keyed by job id, each
+ * tagged with the attempt id it belongs to. The attempt tag is what lets
+ * cancellation target one specific attempt: a TASK_CANCEL for a superseded
+ * attempt must never interrupt a newer assignment of the same job.
+ *
+ * Thread-safe: the connection thread inserts, task-pool threads remove on
+ * completion, and the connection thread also cancels. A {@link ConcurrentHashMap}
+ * with attempt-matched compute operations keeps these races correct.
+ */
+final class InFlightTasks {
+
+ private record Entry(String attemptId, Future future) {
+ }
+
+ private final ConcurrentHashMap byJobId = new ConcurrentHashMap<>();
+
+ void put(String jobId, String attemptId, Future future) {
+ byJobId.put(jobId, new Entry(attemptId, future));
+ }
+
+ /** Removes the entry only if it still belongs to {@code attemptId}. */
+ void remove(String jobId, String attemptId) {
+ byJobId.computeIfPresent(jobId, (id, entry) ->
+ entry.attemptId().equals(attemptId) ? null : entry);
+ }
+
+ /**
+ * Interrupts the tracked task iff both ids match, then forgets it.
+ *
+ * @return true if a matching task was found and cancellation was requested
+ */
+ boolean cancelIfMatches(String jobId, String attemptId) {
+ Entry entry = byJobId.get(jobId);
+ if (entry == null || !entry.attemptId().equals(attemptId)) {
+ return false;
+ }
+ entry.future().cancel(true);
+ byJobId.remove(jobId, entry);
+ return true;
+ }
+
+ /** Cancels every tracked task; used when the connection drops or the worker stops. */
+ void cancelAll() {
+ byJobId.values().forEach(entry -> entry.future().cancel(true));
+ byJobId.clear();
+ }
+}
diff --git a/worker/src/main/java/io/github/achrafaittayeb/dtp/worker/Worker.java b/worker/src/main/java/io/github/achrafaittayeb/dtp/worker/Worker.java
index e97cd35..695321e 100644
--- a/worker/src/main/java/io/github/achrafaittayeb/dtp/worker/Worker.java
+++ b/worker/src/main/java/io/github/achrafaittayeb/dtp/worker/Worker.java
@@ -5,6 +5,7 @@
import io.github.achrafaittayeb.dtp.common.protocol.Heartbeat;
import io.github.achrafaittayeb.dtp.common.protocol.Message;
import io.github.achrafaittayeb.dtp.common.protocol.TaskAssign;
+import io.github.achrafaittayeb.dtp.common.protocol.TaskCancel;
import io.github.achrafaittayeb.dtp.common.protocol.TaskResult;
import io.github.achrafaittayeb.dtp.common.protocol.WorkerRegister;
import io.github.achrafaittayeb.dtp.common.protocol.WorkerRegistered;
@@ -18,8 +19,6 @@
import java.io.InputStream;
import java.io.OutputStream;
import java.net.Socket;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
@@ -51,7 +50,7 @@ public final class Worker implements AutoCloseable {
private final ExecutorService taskPool;
private final ScheduledExecutorService heartbeatScheduler;
private final Thread connectionThread;
- private final Map> inFlightTasks = new ConcurrentHashMap<>();
+ private final InFlightTasks inFlightTasks = new InFlightTasks();
private volatile boolean running = true;
private volatile Socket currentSocket;
@@ -131,10 +130,11 @@ private ScheduledFuture startHeartbeats(OutputStream out, Socket socket) {
private void assignmentLoop(InputStream in, OutputStream out) throws IOException {
Message message;
while ((message = MessageIO.receive(in)) != null) {
- if (message instanceof TaskAssign assign) {
- executeAsync(assign, out);
- } else {
- log.warn("Ignoring unexpected message: {}", message.getClass().getSimpleName());
+ switch (message) {
+ case TaskAssign assign -> executeAsync(assign, out);
+ case TaskCancel cancel -> cancelTask(cancel);
+ default -> log.warn("Ignoring unexpected message: {}",
+ message.getClass().getSimpleName());
}
}
log.info("Coordinator closed the connection");
@@ -145,7 +145,9 @@ private void executeAsync(TaskAssign assign, OutputStream out) {
assign.jobId(), assign.taskType(), assign.attemptNumber());
Future future = taskPool.submit(() -> {
TaskResult result = execute(assign);
- inFlightTasks.remove(assign.jobId());
+ // Remove only if this attempt is still the tracked one; a cancel or a
+ // newer assignment for the same job must not be clobbered.
+ inFlightTasks.remove(assign.jobId(), assign.attemptId());
try {
sendLocked(out, result);
log.info("Task result sent: jobId={} success={}", assign.jobId(), result.success());
@@ -154,7 +156,25 @@ private void executeAsync(TaskAssign assign, OutputStream out) {
log.warn("Could not report result for jobId={}; connection is gone", assign.jobId());
}
});
- inFlightTasks.put(assign.jobId(), future);
+ inFlightTasks.put(assign.jobId(), assign.attemptId(), future);
+ }
+
+ /**
+ * Cooperatively cancels a running attempt. Only cancels if both jobId and
+ * attemptId match the tracked assignment, so a cancel for a superseded
+ * attempt cannot interrupt a newer one. Interruption is best-effort: tasks
+ * that don't respond keep running, but their lease is already revoked at the
+ * coordinator, so any result they produce is stale-rejected.
+ */
+ private void cancelTask(TaskCancel cancel) {
+ boolean cancelled = inFlightTasks.cancelIfMatches(cancel.jobId(), cancel.attemptId());
+ if (cancelled) {
+ log.info("Cancelled task on request: jobId={} attemptId={}",
+ cancel.jobId(), cancel.attemptId());
+ } else {
+ log.debug("Ignoring TASK_CANCEL for untracked attempt: jobId={} attemptId={}",
+ cancel.jobId(), cancel.attemptId());
+ }
}
private TaskResult execute(TaskAssign assign) {
@@ -184,8 +204,7 @@ private static void sendLocked(OutputStream out, Message message) throws IOExcep
}
private void cancelInFlightTasks() {
- inFlightTasks.values().forEach(task -> task.cancel(true));
- inFlightTasks.clear();
+ inFlightTasks.cancelAll();
}
/** Graceful stop: close the connection, interrupt tasks, shut down pools. */