From 0ae23cd878deb3b1b02c4d5221b4563a28da9320 Mon Sep 17 00:00:00 2001 From: granbellllc-png Date: Wed, 11 Nov 2026 05:25:20 +0400 Subject: [PATCH] fix(run-engine): record system failure when dequeue recovery nack dead-letters run (#5106) In DequeueSystem.dequeueFromWorkerQueue recovery path, when the run read or resolveEnv fails (e.g. during a datastore outage or when runStore.findRun rejects), the code previously called runQueue.nackMessage without inspecting the return value. When attempt >= maxAttempts, nackMessage returns false and dead-letters the message into DLQ, leaving the run indefinitely in QUEUED/PENDING status with no failure recorded. This fix: 1. Wraps controlPlaneResolver.resolveEnv in tryCatch to prevent unhandled rejections from escaping the recovery lock. 2. Checks the boolean result of runQueue.nackMessage: if requeued === false, calls runAttemptSystem.systemFailure with dequeueMaxRetriesError (matching the behavior of the snapshot read recovery branch). 3. Adds regression test dequeueRecoveryNackDlq.test.ts verifying that dead-lettered runs transition to SYSTEM_FAILURE and snapshot to FINISHED. --- .../src/engine/systems/dequeueSystem.ts | 20 ++- .../tests/dequeueRecoveryNackDlq.test.ts | 124 ++++++++++++++++++ 2 files changed, 138 insertions(+), 6 deletions(-) create mode 100644 internal-packages/run-engine/src/engine/tests/dequeueRecoveryNackDlq.test.ts diff --git a/internal-packages/run-engine/src/engine/systems/dequeueSystem.ts b/internal-packages/run-engine/src/engine/systems/dequeueSystem.ts index 6c8fa321867..855af789a7d 100644 --- a/internal-packages/run-engine/src/engine/systems/dequeueSystem.ts +++ b/internal-packages/run-engine/src/engine/systems/dequeueSystem.ts @@ -1029,16 +1029,24 @@ export class DequeueSystem { ) ); - const env = run - ? await this.$.controlPlaneResolver.resolveEnv(run.runtimeEnvironmentId) - : null; + const [resolveEnvError, env] = run + ? await tryCatch(this.$.controlPlaneResolver.resolveEnv(run.runtimeEnvironmentId)) + : [undefined, null]; - if (findError || !run || !env) { + if (findError || resolveEnvError || !run || !env) { this.$.logger.error( "RunEngine.dequeueFromWorkerQueue(): Failed to find run, nacking directly via Redis", - { runId, orgId, findError } + { runId, orgId, findError, resolveEnvError } ); - await this.$.runQueue.nackMessage({ orgId, messageId: runId }); + const requeued = await this.$.runQueue.nackMessage({ orgId, messageId: runId }); + if (requeued === false) { + await this.runAttemptSystem.systemFailure({ + runId, + organizationId: orgId, + error: dequeueMaxRetriesError, + tx: prisma, + }); + } return; } diff --git a/internal-packages/run-engine/src/engine/tests/dequeueRecoveryNackDlq.test.ts b/internal-packages/run-engine/src/engine/tests/dequeueRecoveryNackDlq.test.ts new file mode 100644 index 00000000000..6a9d935f2ab --- /dev/null +++ b/internal-packages/run-engine/src/engine/tests/dequeueRecoveryNackDlq.test.ts @@ -0,0 +1,124 @@ +// Reproduction: a run whose queue message is dead-lettered by the dequeue recovery path is left QUEUED (#5106) +// run: cd internal-packages/run-engine && pnpm run test src/engine/tests/dequeueRecoveryNackDlq.test.ts --run +// +// Claim (dequeueSystem.ts, recovery catch, `if (findError || !run || !env)`): when dequeue preparation throws +// and the recovery's run read (runStore.findRun) also fails, the fallback `runQueue.nackMessage(...)` result +// is ignored. On the message's last attempt nackMessage dead-letters it and returns false; the branch above +// (snapshot read failed) turns that false into systemFailure(TASK_RUN_DEQUEUED_MAX_RETRIES), this one does not. +// +// Fault: a wrapper over the real PostgresRunStore whose findRun rejects, once armed, for the two reads that +// select runtimeEnvironmentId (dequeue preparation and the recovery read). Everything else is the real engine, +// Postgres and Redis. The queue retry budget is 1, so the first nack is the last. +// The test asserts the expected behaviour: a run whose message was dead-lettered must not be left QUEUED. +import { assertNonNullable, containerTest } from "@internal/testcontainers"; +import { trace } from "@internal/tracing"; +import { PostgresRunStore } from "@internal/run-store"; +import { setTimeout } from "timers/promises"; +import { describe, expect, vi } from "vitest"; +import { RunEngine } from "../index.js"; +import { setupAuthenticatedEnvironment, setupBackgroundWorker } from "./setup.js"; + +vi.setConfig({ testTimeout: 60_000 }); + +describe("dequeue recovery nack (#5106)", () => { + containerTest( + "a run whose message is dead-lettered by the recovery nack is failed, not left QUEUED", + async ({ prisma, redisOptions }) => { + const base = new PostgresRunStore({ prisma, readOnlyPrisma: prisma }); + let armed = false; + let injected = 0; + const store = new Proxy(base, { + get(target, prop) { + const value = Reflect.get(target, prop, target); + if (prop === "findRun") { + return (...args: any[]) => { + // Dequeue preparation (dequeueRunSelect) and the recovery read both select + // runtimeEnvironmentId; the snapshot helper's fallback read only selects { id }. + // The real findRun is async, so a store outage is a rejected promise, never a sync throw. + if (armed && args[1]?.select?.runtimeEnvironmentId) { + injected++; + return Promise.reject(new Error("injected run-store outage")); + } + return (value as Function).apply(target, args); + }; + } + return typeof value === "function" ? value.bind(target) : value; + }, + }); + + const engine = new RunEngine({ + prisma, + store, + worker: { redis: redisOptions, workers: 1, tasksPerWorker: 10, pollIntervalMs: 100 }, + queue: { + redis: redisOptions, + retryOptions: { maxAttempts: 1, minTimeoutInMs: 50, maxTimeoutInMs: 50 }, + masterQueueConsumersDisabled: true, + processWorkerQueueDebounceMs: 50, + }, + 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 env = await setupAuthenticatedEnvironment(prisma, "PRODUCTION"); + await setupBackgroundWorker(engine, env, "test-task"); + const run = await engine.trigger( + { + number: 1, + friendlyId: "run_dlq1", + environment: env, + taskIdentifier: "test-task", + payload: "{}", + payloadType: "application/json", + context: {}, + traceContext: {}, + traceId: "t1", + spanId: "s1", + workerQueue: "main", + queue: "task/test-task", + isTest: false, + tags: [], + }, + prisma + ); + await setTimeout(500); + + const queued = await engine.getRunExecutionData({ runId: run.id }); + assertNonNullable(queued); + expect(queued.snapshot.executionStatus).toBe("QUEUED"); + + armed = true; + const dequeued = await engine.dequeueFromWorkerQueue({ + consumerId: "c", + workerQueue: "main", + }); + armed = false; + // Long enough for any retry, requeue or heartbeat on a QUEUED run to have fired. + await setTimeout(3_000); + + const inDlq = await engine.runQueue.messageInDeadLetterQueue(env, run.id); + const after = await engine.getRunExecutionData({ runId: run.id }); + assertNonNullable(after); + + // Preconditions: both reads were hit (so the recovery branch ran) and the message was dead-lettered. + expect(injected).toBeGreaterThanOrEqual(2); + expect(dequeued.length).toBe(0); + expect(inDlq).toBe(true); + + // The claim: nothing records a failure. The expected behaviour is a failed run. + expect(after.run.status).toBe("SYSTEM_FAILURE"); + expect(after.snapshot.executionStatus).toBe("FINISHED"); + } finally { + await engine.quit(); + } + } + ); +});