From 5cf73cdc7d3bfcde87b73d236cfc2718caaaaf97 Mon Sep 17 00:00:00 2001 From: Matt Aitken Date: Tue, 18 Aug 2026 17:03:40 +0100 Subject: [PATCH] fix(run-engine): stop requeued runs with a lapsed ttl being orphaned in the queue A run triggered with a ttl registers a TTL entry that the first dequeue removes ("executing, not expired"). Nacking the run back onto the queue kept the original ttlExpiresAt in the rewritten message without re-registering that entry, so the next dequeue pass took the expired-TTL branch: it removed the run from every queue structure and deferred finalization to a TTL consumer that could never find it. The run then sat QUEUED in the database forever, invisible to dequeue, the TTL consumer, and queue repair. Two changes: - nackMessage drops ttlExpiresAt from the rewritten message. TTL only applies to runs that have never been dequeued, matching the existing includeTtl re-enqueue contract, so a requeued run is never expired or dropped by its original deadline. - The dequeue expired-TTL branches re-register the TTL entry instead of assuming it exists, so any message still carrying a lapsed ttlExpiresAt with no TTL entry (e.g. written before this fix) finalizes as EXPIRED instead of orphaning. --- .../ttl-runs-no-longer-stuck-after-requeue.md | 6 + .../src/engine/tests/ttlNackRequeue.test.ts | 528 ++++++++++++++++++ .../run-engine/src/run-queue/index.ts | 28 +- 3 files changed, 559 insertions(+), 3 deletions(-) create mode 100644 .server-changes/ttl-runs-no-longer-stuck-after-requeue.md create mode 100644 internal-packages/run-engine/src/engine/tests/ttlNackRequeue.test.ts diff --git a/.server-changes/ttl-runs-no-longer-stuck-after-requeue.md b/.server-changes/ttl-runs-no-longer-stuck-after-requeue.md new file mode 100644 index 00000000000..b167ecde8c0 --- /dev/null +++ b/.server-changes/ttl-runs-no-longer-stuck-after-requeue.md @@ -0,0 +1,6 @@ +--- +area: webapp +type: fix +--- + +Runs triggered with a `ttl` could get permanently stuck in the queued state if they started executing and were then requeued after a failure (for example a worker dying mid-run) once the TTL had already elapsed. Requeued runs now dequeue normally: a run's TTL only applies while it is waiting to start for the first time. diff --git a/internal-packages/run-engine/src/engine/tests/ttlNackRequeue.test.ts b/internal-packages/run-engine/src/engine/tests/ttlNackRequeue.test.ts new file mode 100644 index 00000000000..c41104abf4d --- /dev/null +++ b/internal-packages/run-engine/src/engine/tests/ttlNackRequeue.test.ts @@ -0,0 +1,528 @@ +import { containerTest, assertNonNullable } from "@internal/testcontainers"; +import { trace } from "@internal/tracing"; +import { expect } from "vitest"; +import { RunEngine } from "../index.js"; +import { setTimeout } from "timers/promises"; +import type { EventBusEventArgs } from "../eventBus.js"; +import { setupAuthenticatedEnvironment, setupBackgroundWorker } from "./setup.js"; + +vi.setConfig({ testTimeout: 60_000 }); + +/** + * TTL interacts with nack/requeue in a dangerous way: enqueue registers a run in a + * TTL sorted set for the TTL consumer, and the first dequeue removes that entry + * ("the run is executing, not expired"). If the run is later nacked back onto the + * queue with its original (now lapsed) ttlExpiresAt still in the message, the next + * dequeue pass takes the expired-TTL branch: it removes the run from the queue + * sorted sets and defers finalization to a TTL consumer that no longer has any + * entry for the run. The run then exists in no queue structure at all — Postgres + * says QUEUED forever, and nothing (dequeue, TTL consumer, concurrency sweeper, + * repair) can ever see it again. + * + * These tests lock in the two halves of the fix: + * 1. nack strips ttlExpiresAt — TTL only applies to runs that have never been + * dequeued (the same contract as includeTtl on re-enqueues), so a requeued run + * stays dequeuable and is never expired by its original deadline. + * 2. The dequeue expired-TTL branch re-registers the TTL entry instead of assuming + * it exists, so any message still carrying a lapsed ttlExpiresAt with no TTL + * entry (e.g. written before the fix) finalizes as EXPIRED instead of orphaning. + */ +describe("RunEngine ttl + nack/requeue", () => { + containerTest( + "Heartbeat-stalled run with a lapsed TTL is requeued and dequeued again (not orphaned)", + async ({ prisma, redisOptions }) => { + const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION"); + + const executingTimeout = 200; + + const engine = new RunEngine({ + prisma, + worker: { + redis: redisOptions, + workers: 1, + tasksPerWorker: 10, + pollIntervalMs: 100, + }, + queue: { + redis: redisOptions, + processWorkerQueueDebounceMs: 50, + masterQueueConsumersDisabled: true, + retryOptions: { + maxAttempts: 12, + minTimeoutInMs: 50, + maxTimeoutInMs: 50, + factor: 1, + randomize: false, + }, + ttlSystem: { + pollIntervalMs: 100, + batchSize: 10, + batchMaxWaitMs: 100, + }, + }, + runLock: { + redis: redisOptions, + }, + machines: { + defaultMachine: "small-1x", + machines: { + "small-1x": { + name: "small-1x" as const, + cpu: 0.5, + memory: 0.5, + centsPerMs: 0.0001, + }, + }, + baseCostInCents: 0.0001, + }, + heartbeatTimeoutsMs: { + EXECUTING: executingTimeout, + }, + tracer: trace.getTracer("test", "0.0.0"), + }); + + try { + const taskIdentifier = "test-task"; + + await setupBackgroundWorker(engine, authenticatedEnvironment, taskIdentifier); + + const expiredEvents: EventBusEventArgs<"runExpired">[0][] = []; + engine.eventBus.on("runExpired", (result) => { + expiredEvents.push(result); + }); + + const triggeredAt = Date.now(); + const run = await engine.trigger( + { + number: 1, + friendlyId: "run_stall1", + environment: authenticatedEnvironment, + taskIdentifier, + payload: "{}", + payloadType: "application/json", + context: {}, + traceContext: {}, + traceId: "t_stall1", + spanId: "s_stall1", + workerQueue: "main", + queue: "task/test-task", + isTest: false, + tags: [], + ttl: "1s", + }, + prisma + ); + + await setTimeout(500); + const dequeued = await engine.dequeueFromWorkerQueue({ + consumerId: "test_stall1", + workerQueue: "main", + }); + expect(dequeued.length).toBe(1); + + await engine.startRunAttempt({ + runId: dequeued[0].run.id, + snapshotId: dequeued[0].snapshot.id, + }); + + const executionData = await engine.getRunExecutionData({ runId: run.id }); + assertNonNullable(executionData); + expect(executionData.snapshot.executionStatus).toBe("EXECUTING"); + + await vi.waitFor( + async () => { + const data = await engine.getRunExecutionData({ runId: run.id }); + assertNonNullable(data); + expect(data.snapshot.executionStatus).toBe("QUEUED"); + }, + { timeout: 10_000, interval: 100 } + ); + + const pastDeadlineMs = triggeredAt + 1_000 + 300 - Date.now(); + if (pastDeadlineMs > 0) { + await setTimeout(pastDeadlineMs); + } + + const message = await engine.runQueue.readMessage( + authenticatedEnvironment.organization.id, + run.id + ); + assertNonNullable(message); + expect(message.ttlExpiresAt).toBeUndefined(); + + await engine.runQueue.processMasterQueueForEnvironment(authenticatedEnvironment.id, 10); + const dequeued2 = await engine.dequeueFromWorkerQueue({ + consumerId: "test_stall1", + workerQueue: "main", + blockingPopTimeoutSeconds: 1, + }); + expect(dequeued2.length).toBe(1); + expect(dequeued2[0]?.run.id).toBe(run.id); + + expect(expiredEvents.length).toBe(0); + } finally { + await engine.quit(); + } + } + ); + + containerTest( + "Requeue after a failure strips ttlExpiresAt so later dequeues do not treat the run as expired", + async ({ prisma, redisOptions }) => { + const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION"); + + const engine = new RunEngine({ + prisma, + worker: { + redis: redisOptions, + workers: 1, + tasksPerWorker: 10, + pollIntervalMs: 100, + }, + queue: { + redis: redisOptions, + processWorkerQueueDebounceMs: 50, + masterQueueConsumersDisabled: true, + retryOptions: { + maxAttempts: 12, + minTimeoutInMs: 50, + maxTimeoutInMs: 50, + factor: 1, + randomize: false, + }, + ttlSystem: { + pollIntervalMs: 100, + batchSize: 10, + batchMaxWaitMs: 100, + }, + }, + runLock: { + redis: redisOptions, + }, + machines: { + defaultMachine: "small-1x", + machines: { + "small-1x": { + name: "small-1x" as const, + cpu: 0.5, + memory: 0.5, + centsPerMs: 0.0001, + }, + }, + baseCostInCents: 0.0001, + }, + tracer: trace.getTracer("test", "0.0.0"), + }); + + try { + const taskIdentifier = "test-task"; + + await setupBackgroundWorker(engine, authenticatedEnvironment, taskIdentifier); + + const expiredEvents: EventBusEventArgs<"runExpired">[0][] = []; + engine.eventBus.on("runExpired", (result) => { + expiredEvents.push(result); + }); + + const triggeredAt = Date.now(); + const run = await engine.trigger( + { + number: 1, + friendlyId: "run_nack1", + environment: authenticatedEnvironment, + taskIdentifier, + payload: "{}", + payloadType: "application/json", + context: {}, + traceContext: {}, + traceId: "t_nack1", + spanId: "s_nack1", + workerQueue: "main", + queue: "task/test-task", + isTest: false, + tags: [], + ttl: "1s", + }, + prisma + ); + + await setTimeout(500); + const dequeued = await engine.dequeueFromWorkerQueue({ + consumerId: "test_nack1", + workerQueue: "main", + }); + expect(dequeued.length).toBe(1); + + const nackResult = await engine.runAttemptSystem.tryNackAndRequeue({ + run: { id: run.id }, + environment: { + id: authenticatedEnvironment.id, + type: authenticatedEnvironment.type, + }, + orgId: authenticatedEnvironment.organization.id, + projectId: authenticatedEnvironment.project.id, + timestamp: Date.now(), + error: { + type: "INTERNAL_ERROR", + code: "TASK_RUN_DEQUEUED_MAX_RETRIES", + message: "test requeue", + }, + }); + expect(nackResult.wasRequeued).toBe(true); + + const message = await engine.runQueue.readMessage( + authenticatedEnvironment.organization.id, + run.id + ); + assertNonNullable(message); + expect(message.ttlExpiresAt).toBeUndefined(); + + const pastDeadlineMs = triggeredAt + 1_000 + 300 - Date.now(); + if (pastDeadlineMs > 0) { + await setTimeout(pastDeadlineMs); + } + + await engine.runQueue.processMasterQueueForEnvironment(authenticatedEnvironment.id, 10); + const dequeued2 = await engine.dequeueFromWorkerQueue({ + consumerId: "test_nack1", + workerQueue: "main", + blockingPopTimeoutSeconds: 1, + }); + expect(dequeued2.length).toBe(1); + expect(dequeued2[0]?.run.id).toBe(run.id); + + expect(expiredEvents.length).toBe(0); + } finally { + await engine.quit(); + } + } + ); + + containerTest( + "Dequeue re-registers a lapsed-TTL message for the TTL consumer when its TTL entry is missing", + async ({ prisma, redisOptions }) => { + const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION"); + + const engine = new RunEngine({ + prisma, + worker: { + disabled: true, + redis: redisOptions, + workers: 1, + tasksPerWorker: 10, + pollIntervalMs: 100, + }, + queue: { + redis: redisOptions, + processWorkerQueueDebounceMs: 50, + masterQueueConsumersDisabled: true, + ttlSystem: { + pollIntervalMs: 100, + batchSize: 10, + batchMaxWaitMs: 100, + }, + }, + runLock: { + redis: redisOptions, + }, + machines: { + defaultMachine: "small-1x", + machines: { + "small-1x": { + name: "small-1x" as const, + cpu: 0.5, + memory: 0.5, + centsPerMs: 0.0001, + }, + }, + baseCostInCents: 0.0001, + }, + tracer: trace.getTracer("test", "0.0.0"), + }); + + try { + const taskIdentifier = "test-task"; + + await setupBackgroundWorker(engine, authenticatedEnvironment, taskIdentifier); + + const run = await engine.trigger( + { + number: 1, + friendlyId: "run_lostttl1", + environment: authenticatedEnvironment, + taskIdentifier, + payload: "{}", + payloadType: "application/json", + context: {}, + traceContext: {}, + traceId: "t_lost1", + spanId: "s_lost1", + workerQueue: "main", + queue: "task/test-task", + isTest: false, + tags: [], + ttl: "1s", + }, + prisma + ); + + const message = await engine.runQueue.readMessage( + authenticatedEnvironment.organization.id, + run.id + ); + assertNonNullable(message); + expect(message.ttlExpiresAt).toBeDefined(); + + const ttlMember = `${message.queue}|${run.id}|${authenticatedEnvironment.organization.id}`; + let removed = 0; + for (let shard = 0; shard < 4; shard++) { + removed += await engine.runQueue.redis.zrem( + engine.runQueue.keys.ttlQueueKeyForShard(shard), + ttlMember + ); + } + expect(removed).toBe(1); + + await setTimeout(1_300); + + await engine.runQueue.processMasterQueueForEnvironment(authenticatedEnvironment.id, 10); + + await vi.waitFor( + async () => { + const expiredRun = await prisma.taskRun.findUnique({ + where: { id: run.id }, + select: { status: true }, + }); + expect(expiredRun?.status).toBe("EXPIRED"); + }, + { timeout: 15_000, interval: 200 } + ); + + const messageExists = await engine.runQueue.messageExists( + authenticatedEnvironment.organization.id, + run.id + ); + expect(messageExists).toBe(0); + + const envConcurrency = + await engine.runQueue.currentConcurrencyOfEnvironment(authenticatedEnvironment); + expect(envConcurrency).toBe(0); + } finally { + await engine.quit(); + } + } + ); + + containerTest( + "Dequeue re-registers a lapsed-TTL message with a concurrency key when its TTL entry is missing", + async ({ prisma, redisOptions }) => { + const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION"); + + const engine = new RunEngine({ + prisma, + worker: { + disabled: true, + redis: redisOptions, + workers: 1, + tasksPerWorker: 10, + pollIntervalMs: 100, + }, + queue: { + redis: redisOptions, + processWorkerQueueDebounceMs: 50, + masterQueueConsumersDisabled: true, + ttlSystem: { + pollIntervalMs: 100, + batchSize: 10, + batchMaxWaitMs: 100, + }, + }, + runLock: { + redis: redisOptions, + }, + machines: { + defaultMachine: "small-1x", + machines: { + "small-1x": { + name: "small-1x" as const, + cpu: 0.5, + memory: 0.5, + centsPerMs: 0.0001, + }, + }, + baseCostInCents: 0.0001, + }, + tracer: trace.getTracer("test", "0.0.0"), + }); + + try { + const taskIdentifier = "test-task"; + + await setupBackgroundWorker(engine, authenticatedEnvironment, taskIdentifier); + + const run = await engine.trigger( + { + number: 1, + friendlyId: "run_lostttl2", + environment: authenticatedEnvironment, + taskIdentifier, + payload: "{}", + payloadType: "application/json", + context: {}, + traceContext: {}, + traceId: "t_lost2", + spanId: "s_lost2", + workerQueue: "main", + queue: "task/test-task", + isTest: false, + tags: [], + ttl: "1s", + concurrencyKey: "ckA", + }, + prisma + ); + + const message = await engine.runQueue.readMessage( + authenticatedEnvironment.organization.id, + run.id + ); + assertNonNullable(message); + expect(message.ttlExpiresAt).toBeDefined(); + expect(message.concurrencyKey).toBeDefined(); + + const ttlMember = `${message.queue}|${run.id}|${authenticatedEnvironment.organization.id}`; + let removed = 0; + for (let shard = 0; shard < 4; shard++) { + removed += await engine.runQueue.redis.zrem( + engine.runQueue.keys.ttlQueueKeyForShard(shard), + ttlMember + ); + } + expect(removed).toBe(1); + + await setTimeout(1_300); + + await engine.runQueue.processMasterQueueForEnvironment(authenticatedEnvironment.id, 10); + + await vi.waitFor( + async () => { + const expiredRun = await prisma.taskRun.findUnique({ + where: { id: run.id }, + select: { status: true }, + }); + expect(expiredRun?.status).toBe("EXPIRED"); + }, + { timeout: 15_000, interval: 200 } + ); + + const messageExists = await engine.runQueue.messageExists( + authenticatedEnvironment.organization.id, + run.id + ); + expect(messageExists).toBe(0); + } finally { + await engine.quit(); + } + } + ); +}); diff --git a/internal-packages/run-engine/src/run-queue/index.ts b/internal-packages/run-engine/src/run-queue/index.ts index 57cfe518f37..1edc83380e4 100644 --- a/internal-packages/run-engine/src/run-queue/index.ts +++ b/internal-packages/run-engine/src/run-queue/index.ts @@ -1104,6 +1104,12 @@ export class RunQueue { /** * Negative acknowledge a message, which will requeue the message (with an optional future date). If you pass no date it will get reattempted with exponential backoff. + + The rewritten message drops ttlExpiresAt: TTL only applies to runs that have never been + dequeued, and a nack is always post-dequeue (the run's TTL set entry was already removed + at dequeue time). Carrying a lapsed ttlExpiresAt forward would make the next dequeue pass + treat the requeued run as expired and drop it from the queue sorted sets, deferring to a + TTL consumer that has no entry for it — orphaning the run. */ public async nackMessage({ orgId, @@ -1151,6 +1157,8 @@ export class RunQueue { } } + delete message.ttlExpiresAt; + if (!skipDequeueProcessing) { // For CK queues, use wildcard dedup so all CKs share one worker queue processing job const dedupQueueKey = message.concurrencyKey @@ -4290,10 +4298,15 @@ for i = 1, #messages, 2 do -- Check if TTL has expired if ttlExpiresAt and ttlExpiresAt <= currentTime then -- TTL expired - remove from dequeue queues so it won't be retried, - -- but leave messageKey and ttlQueueKey intact for the TTL consumer - -- to discover and properly expire the run. + -- leave messageKey intact, and (re-)register the TTL entry so the + -- TTL consumer can discover and properly expire the run. The entry + -- is removed on first dequeue, so it cannot be assumed to exist. redis.call('ZREM', queueKey, messageId) redis.call('ZREM', envQueueKey, messageId) + if ttlQueueKey and ttlQueueKey ~= '' then + local ttlMember = queueName .. '|' .. messageId .. '|' .. (messageData.orgId or '') + redis.call('ZADD', ttlQueueKey, ttlExpiresAt, ttlMember) + end else -- Not expired - process normally redis.call('ZREM', queueKey, messageId) @@ -4423,9 +4436,14 @@ for _, ckQueueName in ipairs(ckQueues) do local ttlExpiresAt = messageData and messageData.ttlExpiresAt if ttlExpiresAt and ttlExpiresAt <= currentTime then - -- TTL expired - remove from queues + -- TTL expired - remove from queues and (re-)register the TTL entry + -- so the TTL consumer can discover and properly expire the run redis.call('ZREM', fullQueueKey, messageId) redis.call('ZREM', envQueueKey, messageId) + if ttlQueueKey and ttlQueueKey ~= '' then + local ttlMember = ckQueueName .. '|' .. messageId .. '|' .. (messageData.orgId or '') + redis.call('ZADD', ttlQueueKey, ttlExpiresAt, ttlMember) + end else -- Dequeue normally redis.call('ZREM', fullQueueKey, messageId) @@ -4578,6 +4596,10 @@ for _, ckQueueName in ipairs(ckQueues) do redis.call('ZREM', fullQueueKey, messageId) redis.call('ZREM', envQueueKey, messageId) decrLengthCounter() + if ttlQueueKey and ttlQueueKey ~= '' then + local ttlMember = ckQueueName .. '|' .. messageId .. '|' .. (messageData.orgId or '') + redis.call('ZADD', ttlQueueKey, ttlExpiresAt, ttlMember) + end else redis.call('ZREM', fullQueueKey, messageId) redis.call('ZREM', envQueueKey, messageId)