From 3d259d226df6d203354fa79d18b4890100cccf19 Mon Sep 17 00:00:00 2001 From: Achraf Ait Tayeb <2023521460132@stu.scu.edu.cn> Date: Sat, 5 Sep 2026 12:08:08 +0800 Subject: [PATCH 1/4] feat: add cancelled state and execution deadline to job model - CANCELLED terminal state, reachable from QUEUED, RETRY_WAIT, and RUNNING; enforced by the state machine like every other transition - Attempt leases are now time-bounded: assignTo stamps a deadline (coordinator clock + configurable execution timeout, default 10 min, --task-timeout-millis); the deadline is cleared with the lease - Job.cancel revokes any active lease; Job.isDeadlineExpired documents that expiry is suspicion of a stuck task, not proof of worker death - SQLite schema gains execution_timeout and deadline columns, with an in-place migration so pre-deadline databases remain loadable - Unit tests: CANCELLED transitions, deadline stamping/clearing, cancellation from each state, legacy-database migration --- .../dtp/common/model/JobState.java | 20 ++++--- .../dtp/common/model/JobStateTest.java | 14 +++++ .../dtp/coordinator/CoordinatorConfig.java | 10 ++++ .../dtp/coordinator/CoordinatorCore.java | 3 +- .../dtp/coordinator/CoordinatorMain.java | 1 + .../dtp/coordinator/job/Job.java | 50 +++++++++++++++-- .../coordinator/job/SqliteJobRepository.java | 53 +++++++++++++----- .../dtp/coordinator/job/JobTest.java | 55 ++++++++++++++++++- .../job/SqliteJobRepositoryTest.java | 41 ++++++++++++-- .../dtp/it/CoordinatorRestartIT.java | 2 +- .../github/achrafaittayeb/dtp/it/Testbed.java | 8 +++ 11 files changed, 225 insertions(+), 32 deletions(-) diff --git a/common/src/main/java/io/github/achrafaittayeb/dtp/common/model/JobState.java b/common/src/main/java/io/github/achrafaittayeb/dtp/common/model/JobState.java index 4f16cc2..90d1785 100644 --- a/common/src/main/java/io/github/achrafaittayeb/dtp/common/model/JobState.java +++ b/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) * */ 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> 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/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/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..8843581 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 @@ -119,7 +119,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={}", 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/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/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..c39dd71 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 @@ -131,7 +131,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/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)); From aa1068eac29719e803f8de5f8de7a683e12b8c72 Mon Sep 17 00:00:00 2001 From: Achraf Ait Tayeb <2023521460132@stu.scu.edu.cn> Date: Sat, 5 Sep 2026 12:30:45 +0800 Subject: [PATCH 2/4] feat: enforce execution deadlines and add cooperative cancellation Closes the liveness gap where a task wedges on a worker that stays alive and keeps heartbeating: worker liveness no longer implies task progress. Coordinator (single-writer core, unchanged model): - the sweep gains detectExpiredDeadlines(): a RUNNING job past its coordinator-clock deadline has its lease revoked and is retried or failed via the existing retry policy, exactly like a lost worker - revokeAttemptOnWorker() sends a best-effort TASK_CANCEL and frees the worker's capacity slot; a send failure is harmless because the lease is already revoked, so any late result is stale-rejected as before - cancelJob() supports client cancellation: QUEUED/RETRY_WAIT/RUNNING -> CANCELLED (revoking any lease), terminal jobs reported unchanged Protocol (additive, unknown-type-tolerant): - TASK_CANCEL (coordinator->worker: jobId + attemptId) - CANCEL_JOB / JOB_CANCEL (client request/reply) Worker: - cooperative cancellation via thread interruption; InFlightTasks tracks each future by (jobId, attemptId) so a cancel for a superseded attempt can never interrupt a newer one. No hard preemption; a task that ignores interruption still has its lease revoked coordinator-side. Client CLI: new 'cancel ' command. --- .../achrafaittayeb/dtp/client/ClientMain.java | 13 +++ .../dtp/client/CoordinatorClient.java | 17 ++++ .../dtp/common/protocol/CancelJob.java | 5 ++ .../dtp/common/protocol/JobCancelReply.java | 11 +++ .../dtp/common/protocol/Message.java | 9 +- .../dtp/common/protocol/TaskCancel.java | 15 ++++ .../dtp/coordinator/CoordinatorCore.java | 85 ++++++++++++++++++- .../dtp/coordinator/net/ClientServer.java | 8 ++ .../dtp/worker/InFlightTasks.java | 53 ++++++++++++ .../achrafaittayeb/dtp/worker/Worker.java | 41 ++++++--- 10 files changed, 242 insertions(+), 15 deletions(-) create mode 100644 common/src/main/java/io/github/achrafaittayeb/dtp/common/protocol/CancelJob.java create mode 100644 common/src/main/java/io/github/achrafaittayeb/dtp/common/protocol/JobCancelReply.java create mode 100644 common/src/main/java/io/github/achrafaittayeb/dtp/common/protocol/TaskCancel.java create mode 100644 worker/src/main/java/io/github/achrafaittayeb/dtp/worker/InFlightTasks.java diff --git a/client/src/main/java/io/github/achrafaittayeb/dtp/client/ClientMain.java b/client/src/main/java/io/github/achrafaittayeb/dtp/client/ClientMain.java index 77bc617..a57eadd 100644 --- a/client/src/main/java/io/github/achrafaittayeb/dtp/client/ClientMain.java +++ b/client/src/main/java/io/github/achrafaittayeb/dtp/client/ClientMain.java @@ -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 workers = client.listWorkers(); if (workers.isEmpty()) { @@ -199,6 +211,7 @@ private static void printUsage() { submit fail --fail-until-attempt status wait [--timeout-millis ] + cancel list workers Global options: --host (default localhost), --port (default 7071) diff --git a/client/src/main/java/io/github/achrafaittayeb/dtp/client/CoordinatorClient.java b/client/src/main/java/io/github/achrafaittayeb/dtp/client/CoordinatorClient.java index 4546d64..dbf6866 100644 --- a/client/src/main/java/io/github/achrafaittayeb/dtp/client/CoordinatorClient.java +++ b/client/src/main/java/io/github/achrafaittayeb/dtp/client/CoordinatorClient.java @@ -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 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). * 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/coordinator/src/main/java/io/github/achrafaittayeb/dtp/coordinator/CoordinatorCore.java b/coordinator/src/main/java/io/github/achrafaittayeb/dtp/coordinator/CoordinatorCore.java index 8843581..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; @@ -134,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()) @@ -149,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) { @@ -194,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/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/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. */ From 00f33c3435c031a39869cb07a302cd047e92c482 Mon Sep 17 00:00:00 2001 From: Achraf Ait Tayeb <2023521460132@stu.scu.edu.cn> Date: Sat, 5 Sep 2026 12:34:06 +0800 Subject: [PATCH 3/4] test: cover execution deadlines, cancellation, and stale timed-out results MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit New DeadlineAndCancellationIT (7 tests): - task exceeds its deadline on a healthy, still-heartbeating worker -> TASK_CANCEL sent, job reassigned, completes on attempt 2 - late success from a timed-out attempt is stale-rejected - deadline expiry on the final attempt -> FAILED with a timeout error - cancel a QUEUED job -> CANCELLED, never scheduled even once a worker joins - cancel a RUNNING SLEEP -> real worker interrupted (finishes in <5s not 30s), capacity freed, state CANCELLED - a worker that ignores the cancel and reports success cannot resurrect a CANCELLED job - cancelling terminal/unknown jobs is reported, not misapplied CoordinatorRestartIT: a restart discards the old wall-clock deadline and requeues the in-flight job (rather than letting a stale deadline expire it immediately) — the fresh assignment gets a fresh deadline. ScriptedWorker records received TASK_CANCEL messages (awaitCancel); MessageSerializationTest round-trips the three new message types. --- .../protocol/MessageSerializationTest.java | 13 + .../dtp/it/CoordinatorRestartIT.java | 42 ++++ .../dtp/it/DeadlineAndCancellationIT.java | 234 ++++++++++++++++++ .../achrafaittayeb/dtp/it/ScriptedWorker.java | 13 + 4 files changed, 302 insertions(+) create mode 100644 integration-tests/src/test/java/io/github/achrafaittayeb/dtp/it/DeadlineAndCancellationIT.java 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/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 c39dd71..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(); 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; From 3a5fc9fdd84b9b9dab09e59aef0bb9e85ed7e0b7 Mon Sep 17 00:00:00 2001 From: Achraf Ait Tayeb <2023521460132@stu.scu.edu.cn> Date: Sat, 5 Sep 2026 12:38:20 +0800 Subject: [PATCH 4/4] docs: document time-bounded leases and cancellation semantics MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - README: CANCELLED in the state diagram, time-bounded lease explanation, execution-deadline run instructions, cancel command, updated semantics and limitations - PROTOCOL.md: TASK_CANCEL, CANCEL_JOB/JOB_CANCEL, CANCELLED in snapshots - FAILURE_MODEL.md: stuck-task-on-healthy-worker scenario, client cancellation, stale-deadline recovery rule, tuning entry - DESIGN_DECISIONS.md: ADR 13 on time-bounded leases — coordinator-clock expiry, reuse over new machinery, cooperative (not preemptive) cancel, and the explicit non-claims (still at-least-once, not real-time) - ARCHITECTURE.md: deadline enforcer in the sweep and core components --- README.md | 51 +++++++++++++++++++++++++++++++------ docs/ARCHITECTURE.md | 9 ++++++- docs/DESIGN_DECISIONS.md | 44 ++++++++++++++++++++++++++++++++ docs/FAILURE_MODEL.md | 54 ++++++++++++++++++++++++++++++++++++++-- docs/PROTOCOL.md | 17 ++++++++++--- 5 files changed, 161 insertions(+), 14 deletions(-) diff --git a/README.md b/README.md index a274996..d06c72c 100644 --- a/README.md +++ b/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,
attempts remain + RUNNING --> RETRY_WAIT : worker lost / task failed /
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 ./scripts/client.sh wait +./scripts/client.sh cancel ./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 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