fix(runner): retry cancellation version races

Author: ALikesToCode

packages/session-core/src/runner-jobs.test.ts

@@ -183,6 +183,26 @@ class StartRaceRunnerJobRepository extends MemoryRunnerJobRepository {
   }
 }
 
+class CancelRaceRunnerJobRepository extends MemoryRunnerJobRepository {
+  private injectEventUpdate = true;
+
+  override async save(job: RunnerJob, expectedVersion: number): Promise<void> {
+    if (this.injectEventUpdate && job.status === "CANCELLED") {
+      this.injectEventUpdate = false;
+      const current = this.jobs.get(job.jobId);
+      if (current === undefined) throw new Error("runner job disappeared");
+      this.jobs.set(job.jobId, {
+        ...current,
+        eventCursor: current.eventCursor + 1,
+        jobVersion: current.jobVersion + 1,
+        updatedAt: "2026-07-15T00:00:02.000Z",
+      });
+      throw new ConcurrentRunnerJobUpdateError(job.jobId);
+    }
+    await super.save(job, expectedVersion);
+  }
+}
+
 const HASH_A = "a".repeat(64);
 const HASH_B = "b".repeat(64);
 const HASH_C = "c".repeat(64);
@@ -334,6 +354,29 @@ describe("RunnerJobService", () => {
     );
   });
 
+  it("retries cancellation after a concurrent active job update", async () => {
+    const harness = service(new CancelRaceRunnerJobRepository());
+    const queued = await harness.service.createJob(jobInput());
+    const starting = await harness.service.transition(
+      queued.jobId,
+      queued.jobVersion,
+      "STARTING",
+      { runnerIdentity: "runner-container-test" },
+    );
+
+    const cancelled = await harness.service.cancelJob(starting.jobId);
+
+    expect(cancelled).toMatchObject({
+      status: "CANCELLED",
+      eventCursor: 1,
+      jobVersion: 4,
+      error: {
+        code: "RUNNER_JOB_CANCELLED",
+        retryable: true,
+      },
+    });
+  });
+
   it("expires a stalled non-terminal job from its start deadline", async () => {
     const repository = new MemoryRunnerJobRepository();
     let now = new Date("2026-07-15T00:00:00.000Z");

packages/session-core/src/runner-jobs.ts

@@ -195,6 +195,7 @@ const REUSABLE_STARTED_JOB_STATUSES = new Set<RunnerJobStatus>([
   "REPAIRING",
 ]);
 const START_CONFLICT_RETRY_LIMIT = 3;
+const CANCEL_CONFLICT_RETRY_LIMIT = 3;
 const CALLBACK_SETTLEMENT_GRACE_MS = 60_000;
 
 function jobDeadline(job: RunnerJob): number {
@@ -358,34 +359,37 @@ export class RunnerJobService {
   }
 
   async cancelJob(jobId: string): Promise<RunnerJob> {
-    const current = await this.getJob(jobId);
-    if (current.status === "CANCELLED") return current;
-    if (TERMINAL_JOB_STATUSES.has(current.status)) {
-      throw new RunnerCallbackStateError(
-        `Runner job ${jobId} is already terminal with status ${current.status}`,
-      );
-    }
-    try {
-      return await this.transition(
-        current.jobId,
-        current.jobVersion,
-        "CANCELLED",
-        {
-          runnerIdentity:
-            current.runnerIdentity ?? "counterlab-control-plane-cancel",
-          error: {
-            code: "RUNNER_JOB_CANCELLED",
-            message: "The learner cancelled this runner job.",
-            retryable: true,
+    let current = await this.getJob(jobId);
+    let lastConflict: ConcurrentRunnerJobUpdateError | undefined;
+    for (let attempt = 0; attempt < CANCEL_CONFLICT_RETRY_LIMIT; attempt += 1) {
+      if (current.status === "CANCELLED") return current;
+      if (TERMINAL_JOB_STATUSES.has(current.status)) {
+        throw new RunnerCallbackStateError(
+          `Runner job ${jobId} is already terminal with status ${current.status}`,
+        );
+      }
+      try {
+        return await this.transition(
+          current.jobId,
+          current.jobVersion,
+          "CANCELLED",
+          {
+            runnerIdentity:
+              current.runnerIdentity ?? "counterlab-control-plane-cancel",
+            error: {
+              code: "RUNNER_JOB_CANCELLED",
+              message: "The learner cancelled this runner job.",
+              retryable: true,
+            },
           },
-        },
-      );
-    } catch (error) {
-      if (!(error instanceof ConcurrentRunnerJobUpdateError)) throw error;
-      const updated = await this.getJob(jobId);
-      if (updated.status === "CANCELLED") return updated;
-      throw error;
+        );
+      } catch (error) {
+        if (!(error instanceof ConcurrentRunnerJobUpdateError)) throw error;
+        lastConflict = error;
+        current = await this.getJob(jobId);
+      }
     }
+    throw lastConflict ?? new ConcurrentRunnerJobUpdateError(jobId);
   }
 
   async expireIfTimedOut(jobId: string): Promise<RunnerJob> {