Fix fleet queue wedge: reap expired cancel-requested leases

src/rsi/__tests__/postgres-queue.test.ts

@@ -6,7 +6,12 @@ import test from "node:test"
 import { S3RsiArtifactStore } from "../artifact-store.js"
 import { FileRsiArtifactStore } from "../artifact-store.js"
 import type { OpenRouterBrokerGrant, OpenRouterCallReservation, OpenRouterBrokerLimits, OpenRouterEndpointPriceSnapshot } from "../inference-broker.js"
-import { fleetJobMatches, fleetQueuePolicy, missingFleetResourceClasses, PostgresRsiJobQueue, requiredFleetJobLeaseMs, type FleetJob, type FleetJobIdentity, type FleetJobPayload, type FleetQueuePolicy } from "../postgres-queue.js"
+import { fleetJobMatches, fleetQueuePolicy } from "../postgres-queue.js"
+import { missingFleetResourceClasses } from "../postgres-queue.js"
+import { PostgresRsiJobQueue } from "../postgres-queue.js"
+import { requiredFleetJobLeaseMs } from "../postgres-queue.js"
+import type { FleetJob, FleetJobIdentity, FleetJobLease, FleetJobPayload,
+	FleetQueuePolicy } from "../postgres-queue.js"
 
 test("fleet job replay identity rejects stale snapshots, tasks, base commits, evaluation config, and routing", () => {
 	const expected: FleetJobIdentity = {
@@ -520,6 +525,47 @@ test("PostgreSQL queue integration: active cancellation fences lease completion
 	}
 })
 
+type CancelReapFixture = {
+	jobId: string; lease: FleetJobLease; waitingJobId: string
+}
+
+async function setupCancelReap(queue: PostgresRsiJobQueue,
+	pool: Pool): Promise<CancelReapFixture> {
+	for (const workerId of ["worker-dead", "worker-next"]) {
+		await queue.registerWorker({ workerId, gatewayId: `gateway-${workerId}`,
+			hostname: `host-${workerId}`, maxActive: 1, resourceClasses: ["LOCAL_GPU"] })
+	}
+	const job = await addJob(queue, "dead-owner-cancel")
+	const waitingJob = await addJob(queue, "waiting-for-release")
+	const lease = await queue.claim("worker-dead")
+	assert.ok(lease)
+	assert.equal(await queue.cancel(job.jobId, "supervisor stop"), true)
+	const expire = "UPDATE headlesscode_rsi_jobs " +
+		"SET lease_expires_at=clock_timestamp()-interval '1 second' " +
+		"WHERE job_id=$1"
+	await pool.query(expire, [job.jobId])
+	return { jobId: job.jobId, lease, waitingJobId: waitingJob.jobId }
+}
+
+test("PostgreSQL queue integration: expired cancel-requested lease " +
+	"resolves without its owner", { skip: !integrationReady }, async () => {
+	const { queue, pool, cleanup } = await createQueue(testPolicy(
+		`cancel-reap-${randomUUID().slice(0, 8)}`,
+		{ maxInFlight: 1, classCapacity: { LOCAL_GPU: 1 } }))
+	try {
+		const cancelled = await setupCancelReap(queue, pool)
+		const replacement = await queue.claim("worker-next")
+		assert.equal((await queue.getJob(cancelled.jobId))?.status, "cancelled")
+		const error = (await queue.getJob(cancelled.jobId))?.error
+		assert.ok(error?.includes("supervisor stop"))
+		assert.equal(replacement?.job.jobId, cancelled.waitingJobId,
+			"expired cancel-requested lease frees capacity")
+		assert.equal(await queue.acknowledgeCancellation(cancelled.lease), false)
+	} finally {
+		await cleanup()
+	}
+})
+
 test("PostgreSQL queue integration: idempotency survives process restart and conflicts fail closed", { skip: !integrationReady }, async () => {
 	const policy = testPolicy(`restart-${randomUUID().slice(0, 8)}`)
 	const first = await createQueue(policy)

src/rsi/postgres-queue.ts

@@ -480,6 +480,24 @@ export class PostgresRsiJobQueue {
 				 FROM expired WHERE jobs.job_id=expired.job_id`,
 				[this.policy.queueName],
 			)
+			await client.query(
+				`WITH abandoned AS (
+				 SELECT job_id FROM headlesscode_rsi_jobs
+				 WHERE queue_name=$1 AND status='leased'
+				 AND cancel_requested_at IS NOT NULL
+				 AND lease_expires_at <= clock_timestamp()
+				 ORDER BY lease_expires_at LIMIT 100 FOR UPDATE SKIP LOCKED
+				)
+				UPDATE headlesscode_rsi_jobs AS jobs SET status='cancelled',
+				 error=LEFT(COALESCE(jobs.cancel_reason,
+				 'cancellation requested')
+				 || ' (lease expired before teardown was acknowledged)',
+				 4000),
+				 lease_owner=NULL, lease_token=NULL,
+				 lease_expires_at=NULL, updated_at=clock_timestamp()
+				 FROM abandoned WHERE jobs.job_id=abandoned.job_id`,
+				[this.policy.queueName],
+			)
 			const capacities = await client.query(
 				`SELECT count(*)::int AS total,
 				 count(*) FILTER (WHERE lease_owner=$1)::int AS worker_total