Skip to content

Commit 967dedc

Browse files
0ski0ski
authored
fix(run-engine): correct park deadline and snapshot state for debounced parked runs (#4708)
Two defects that surface when a run parked on an external deployment id gets pushed by a debounce key. Both were reproduced against a local instance before being fixed. ## 1. The run is expired before it is due ``` now | status | statusReason | delayUntil | expiredAt 13:57:06 | EXPIRED | EXTERNAL_DEPLOYMENT_NOT_FOUND | 14:01:37 | 13:57:02 ``` Killed 4m35s before its own scheduled start, blaming a missing deployment. **Why.** The park deadline is armed **once**, when the run is first parked, from `max(now, delayUntil) + deadline`. Debounce pushes `delayUntil` out afterwards and nothing re-arms it: - `rescheduleDelayedRun` reschedules `enqueueDelayedRun:<id>`, not `expireParkedExternalDeploymentRun:<id>` - the redis-worker reschedule is an update-only `ZADD … XX`, and a parked run has no `enqueueDelayedRun` job, so that call is a silent no-op Repeat triggers on one key walk `delayUntil` away from a deadline that no longer moves. Once it crosses, the run dies while parked and not yet due. **Fix.** The expiry job already loads `delayUntil`, so it re-arms from the current value and returns instead of expiring a run that is not due. The guard lives in the expiry job rather than the debounce path deliberately: it covers **every** caller that moves `delayUntil`, so a future call site can't reintroduce this by forgetting to re-arm. It stays bounded by the debounce max-duration contract, so a hot key can't postpone expiry indefinitely. ## 2. The run reports itself as delayed while it is parked ``` RUN_CREATED | PENDING_VERSION | Run is waiting for a deployment of 'debounce-test-2' DELAYED | DELAYED | Delayed run was rescheduled to a future date ← after one debounce push ``` The row stays `PENDING_VERSION`; the latest snapshot claims `DELAYED`, so the run page describes a parked run as delayed. Happens on the *first* push. **Fix.** `rescheduleRun` hardcoded `DELAYED`/`DELAYED`. The snapshot statuses are now supplied by the caller and **default to `DELAYED`**, so the ordinary delayed path is byte-identical, and `rescheduleDelayedRun` passes the parked statuses through when the run is parked. ## Reproducing Repeated triggers on one debounce key against an id that hasn't landed: ```bash curl … -d '{"options":{"externalDeploymentId":"x","debounce":{"key":"k","delay":"5m"}}}' ``` Three triggers correctly fold into one parked run; the defects show up on the pushes. ## Testing Two tests, each verified red before green and failing alone: - a run whose delay was pushed past the deadline stays `PENDING_VERSION` instead of expiring - a debounce push on a parked run leaves a `RUN_CREATED`/`PENDING_VERSION` snapshot, not `DELAYED` `56 passed` across parking, pendingVersion, delayedRunSystem and debounce; `43 passed` in `PostgresRunStore`. Typecheck, lint, format clean. ## Notes - Stacks on #4665, so it lands after the whole external-deployment-id series. - No changeset: this fixes unreleased behaviour introduced by the stack below it, so no user has seen it. - Both found by Devin's review on #4664, and both confirmed end to end on a local instance before fixing.
1 parent cde8919 commit 967dedc

6 files changed

Lines changed: 266 additions & 3 deletions

File tree

‎internal-packages/run-engine/src/engine/index.ts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -403,6 +403,7 @@ export class RunEngine {
403403
this.pendingVersionSystem = new PendingVersionSystem({
404404
resources,
405405
enqueueSystem: this.enqueueSystem,
406+
executionSnapshotSystem: this.executionSnapshotSystem,
406407
queueRunsPendingVersionBatchSize: options.queueRunsWaitingForWorkerBatchSize,
407408
lagRetryDelayMs: options.pendingVersionLagRetryDelayMs,
408409
lagMaxRetries: options.pendingVersionLagMaxRetries,

‎internal-packages/run-engine/src/engine/systems/delayedRunSystem.ts‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -48,6 +48,8 @@ export class DelayedRunSystem {
4848
throw new ServiceValidationError("Cannot reschedule a run that is not delayed");
4949
}
5050

51+
const isParked = snapshot.runStatus === "PENDING_VERSION";
52+
5153
const updatedRun = await this.$.runStore.rescheduleRun(
5254
runId,
5355
{
@@ -57,6 +59,13 @@ export class DelayedRunSystem {
5759
environmentType: snapshot.environmentType,
5860
projectId: snapshot.projectId,
5961
organizationId: snapshot.organizationId,
62+
...(isParked
63+
? {
64+
executionStatus: "RUN_CREATED" as const,
65+
runStatus: "PENDING_VERSION" as const,
66+
description: "Parked run was rescheduled to a future date",
67+
}
68+
: {}),
6069
},
6170
},
6271
prisma

‎internal-packages/run-engine/src/engine/systems/pendingVersionSystem.ts‎

Lines changed: 48 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@ import {
55
import type { TaskRunError } from "@trigger.dev/core/v3/schemas";
66
import type { MinimalAuthenticatedEnvironment } from "../../shared/index.js";
77
import type { EnqueueSystem } from "./enqueueSystem.js";
8+
import type { ExecutionSnapshotSystem } from "./executionSnapshotSystem.js";
89
import type { SystemResources } from "./systems.js";
910

1011
import { boundedIn } from "@trigger.dev/database";
@@ -28,6 +29,7 @@ export type PendingVersionSystemOptions = {
2829
*/
2930
lagMaxRetries?: number;
3031
externalDeploymentParkDeadlineMs?: number;
32+
executionSnapshotSystem: ExecutionSnapshotSystem;
3133
};
3234

3335
const DEFAULT_LAG_RETRY_DELAY_MS = 5_000;
@@ -60,10 +62,12 @@ export function readExternalDeploymentIdAnnotation(annotations: unknown): string
6062
export class PendingVersionSystem {
6163
private readonly $: SystemResources;
6264
private readonly enqueueSystem: EnqueueSystem;
65+
private readonly executionSnapshotSystem: ExecutionSnapshotSystem;
6366

6467
constructor(private readonly options: PendingVersionSystemOptions) {
6568
this.$ = options.resources;
6669
this.enqueueSystem = options.enqueueSystem;
70+
this.executionSnapshotSystem = options.executionSnapshotSystem;
6771
}
6872

6973
async enqueueRunsForBackgroundWorker(backgroundWorkerId: string, attempt: number = 0) {
@@ -231,6 +235,20 @@ export class PendingVersionSystem {
231235
}
232236

233237
if (stillDelayed) {
238+
await this.executionSnapshotSystem.createExecutionSnapshot(
239+
tx,
240+
{
241+
run: { id: run.id, status: "DELAYED" },
242+
snapshot: { executionStatus: "DELAYED", description: "Run is delayed" },
243+
batchId: run.batchId ?? undefined,
244+
environmentId: backgroundWorker.runtimeEnvironment.id,
245+
environmentType: backgroundWorker.runtimeEnvironment.type,
246+
projectId: backgroundWorker.runtimeEnvironment.project.id,
247+
organizationId: backgroundWorker.runtimeEnvironment.organization.id,
248+
},
249+
store
250+
);
251+
234252
return true;
235253
}
236254

@@ -470,6 +488,22 @@ export class PendingVersionSystem {
470488
);
471489
}
472490

491+
if (run.delayUntil && run.delayUntil > new Date()) {
492+
this.$.logger.info(
493+
"expireParkedExternalDeploymentRun: run is not due yet, re-arming the park deadline",
494+
{ runId, externalDeploymentId, delayUntil: run.delayUntil }
495+
);
496+
497+
await this.scheduleExternalDeploymentParkDeadline({
498+
runId,
499+
externalDeploymentId,
500+
ttl: run.ttl,
501+
delayUntil: run.delayUntil,
502+
});
503+
504+
return;
505+
}
506+
473507
const error: TaskRunError = {
474508
type: "STRING_ERROR",
475509
raw: `Run expired because no deployment with external id '${externalDeploymentId}' became available`,
@@ -574,6 +608,20 @@ export class PendingVersionSystem {
574608
}
575609

576610
if (stillDelayed) {
611+
await this.executionSnapshotSystem.createExecutionSnapshot(
612+
tx,
613+
{
614+
run: { id: run.id, status: "DELAYED" },
615+
snapshot: { executionStatus: "DELAYED", description: "Run is delayed" },
616+
batchId: run.batchId ?? undefined,
617+
environmentId: env.id,
618+
environmentType: env.type,
619+
projectId: env.project.id,
620+
organizationId: env.organization.id,
621+
},
622+
store
623+
);
624+
577625
return true;
578626
}
579627

‎internal-packages/run-engine/src/engine/tests/externalDeploymentParking.test.ts‎

Lines changed: 201 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -566,6 +566,207 @@ describe("RunEngine external deployment parking", () => {
566566
}
567567
);
568568

569+
containerTest(
570+
"a run released while still delayed records a DELAYED snapshot and no longer reports as parked",
571+
async ({ prisma, redisOptions }) => {
572+
const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
573+
const engine = createEngine(prisma, redisOptions);
574+
575+
try {
576+
const taskIdentifier = "test-task";
577+
578+
const run = await engine.trigger(
579+
{
580+
number: 1,
581+
friendlyId: "run_1234",
582+
environment: authenticatedEnvironment,
583+
taskIdentifier,
584+
payload: "{}",
585+
payloadType: "application/json",
586+
context: {},
587+
traceContext: {},
588+
traceId: "t1234",
589+
spanId: "s1234",
590+
queue: `task/${taskIdentifier}`,
591+
isTest: false,
592+
tags: [],
593+
delayUntil: new Date(Date.now() + 60 * 60 * 1000),
594+
annotations: {
595+
triggerSource: "sdk",
596+
triggerAction: "trigger",
597+
rootTriggerSource: "sdk",
598+
externalDeploymentId: "commit-released",
599+
},
600+
parkedOnExternalDeploymentId: "commit-released",
601+
},
602+
prisma
603+
);
604+
605+
const worker = await setupBackgroundWorker(
606+
engine,
607+
authenticatedEnvironment,
608+
taskIdentifier
609+
);
610+
await nameDeploymentWithExternalId(prisma, worker.worker.id, "commit-released");
611+
612+
await engine.pendingVersionSystem.enqueueRunsForBackgroundWorker(worker.worker.id);
613+
614+
const released = await prisma.taskRun.findFirstOrThrow({ where: { id: run.id } });
615+
expect(released.status).toBe("DELAYED");
616+
expect(released.lockedToVersionId).toBe(worker.worker.id);
617+
618+
const afterRelease = await prisma.taskRunExecutionSnapshot.findMany({
619+
where: { runId: run.id },
620+
orderBy: { createdAt: "asc" },
621+
select: { executionStatus: true, runStatus: true },
622+
});
623+
624+
expect(afterRelease.at(-1)?.runStatus).toBe("DELAYED");
625+
expect(afterRelease.at(-1)?.executionStatus).toBe("DELAYED");
626+
627+
// A later debounce push must not re-label an already-released run as parked.
628+
await engine.delayedRunSystem.rescheduleDelayedRun({
629+
runId: run.id,
630+
delayUntil: new Date(Date.now() + 2 * 60 * 60 * 1000),
631+
tx: prisma,
632+
});
633+
634+
const afterPush = await prisma.taskRunExecutionSnapshot.findMany({
635+
where: { runId: run.id },
636+
orderBy: { createdAt: "asc" },
637+
select: { executionStatus: true, runStatus: true },
638+
});
639+
640+
expect(afterPush.at(-1)?.runStatus).toBe("DELAYED");
641+
expect(afterPush.at(-1)?.executionStatus).toBe("DELAYED");
642+
} finally {
643+
await engine.quit();
644+
}
645+
}
646+
);
647+
648+
containerTest(
649+
"a debounce push on a parked run keeps the snapshot parked instead of reporting it delayed",
650+
async ({ prisma, redisOptions }) => {
651+
const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
652+
const engine = createEngine(prisma, redisOptions);
653+
654+
try {
655+
const taskIdentifier = "test-task";
656+
657+
const run = await engine.trigger(
658+
{
659+
number: 1,
660+
friendlyId: "run_1234",
661+
environment: authenticatedEnvironment,
662+
taskIdentifier,
663+
payload: "{}",
664+
payloadType: "application/json",
665+
context: {},
666+
traceContext: {},
667+
traceId: "t1234",
668+
spanId: "s1234",
669+
queue: `task/${taskIdentifier}`,
670+
isTest: false,
671+
tags: [],
672+
delayUntil: new Date(Date.now() + 60 * 1000),
673+
annotations: {
674+
triggerSource: "sdk",
675+
triggerAction: "trigger",
676+
rootTriggerSource: "sdk",
677+
externalDeploymentId: "commit-snapshot",
678+
},
679+
parkedOnExternalDeploymentId: "commit-snapshot",
680+
},
681+
prisma
682+
);
683+
684+
await engine.delayedRunSystem.rescheduleDelayedRun({
685+
runId: run.id,
686+
delayUntil: new Date(Date.now() + 10 * 60 * 1000),
687+
tx: prisma,
688+
});
689+
690+
const snapshots = await prisma.taskRunExecutionSnapshot.findMany({
691+
where: { runId: run.id },
692+
orderBy: { createdAt: "asc" },
693+
select: { executionStatus: true, runStatus: true },
694+
});
695+
696+
const latest = snapshots.at(-1);
697+
698+
assertNonNullable(latest);
699+
expect(latest.runStatus).toBe("PENDING_VERSION");
700+
expect(latest.executionStatus).toBe("RUN_CREATED");
701+
702+
const stillParked = await prisma.taskRun.findFirstOrThrow({ where: { id: run.id } });
703+
expect(stillParked.status).toBe("PENDING_VERSION");
704+
} finally {
705+
await engine.quit();
706+
}
707+
}
708+
);
709+
710+
containerTest(
711+
"the parking deadline re-arms instead of expiring a run whose delay was pushed past it",
712+
async ({ prisma, redisOptions }) => {
713+
const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
714+
const engine = createEngine(prisma, redisOptions);
715+
716+
try {
717+
const taskIdentifier = "test-task";
718+
719+
const run = await engine.trigger(
720+
{
721+
number: 1,
722+
friendlyId: "run_1234",
723+
environment: authenticatedEnvironment,
724+
taskIdentifier,
725+
payload: "{}",
726+
payloadType: "application/json",
727+
context: {},
728+
traceContext: {},
729+
traceId: "t1234",
730+
spanId: "s1234",
731+
queue: `task/${taskIdentifier}`,
732+
isTest: false,
733+
tags: [],
734+
delayUntil: new Date(Date.now() + 60 * 1000),
735+
annotations: {
736+
triggerSource: "sdk",
737+
triggerAction: "trigger",
738+
rootTriggerSource: "sdk",
739+
externalDeploymentId: "commit-pushed",
740+
},
741+
parkedOnExternalDeploymentId: "commit-pushed",
742+
},
743+
prisma
744+
);
745+
746+
// Stand in for a debounce push: the run's delay moves out, but nothing re-arms the
747+
// deadline that was computed when the run was first parked.
748+
const pushedDelayUntil = new Date(Date.now() + 60 * 60 * 1000);
749+
await prisma.taskRun.update({
750+
where: { id: run.id },
751+
data: { delayUntil: pushedDelayUntil },
752+
});
753+
754+
await engine.pendingVersionSystem.expireParkedExternalDeploymentRun({
755+
runId: run.id,
756+
externalDeploymentId: "commit-pushed",
757+
});
758+
759+
const stillParked = await prisma.taskRun.findFirstOrThrow({ where: { id: run.id } });
760+
761+
expect(stillParked.status).toBe("PENDING_VERSION");
762+
expect(stillParked.statusReason).toBe("EXTERNAL_DEPLOYMENT_PENDING");
763+
expect(stillParked.expiredAt).toBeNull();
764+
} finally {
765+
await engine.quit();
766+
}
767+
}
768+
);
769+
569770
containerTest(
570771
"the parking deadline expires a run whose deployment never arrived",
571772
async ({ prisma, redisOptions }) => {

‎internal-packages/run-store/src/PostgresRunStore.ts‎

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1439,9 +1439,10 @@ export class PostgresRunStore implements RunStore {
14391439
executionSnapshots: {
14401440
create: {
14411441
engine: "V2",
1442-
executionStatus: "DELAYED",
1443-
description: "Delayed run was rescheduled to a future date",
1444-
runStatus: "DELAYED",
1442+
executionStatus: data.snapshot.executionStatus ?? "DELAYED",
1443+
description:
1444+
data.snapshot.description ?? "Delayed run was rescheduled to a future date",
1445+
runStatus: data.snapshot.runStatus ?? "DELAYED",
14451446
environmentId: data.snapshot.environmentId,
14461447
environmentType: data.snapshot.environmentType,
14471448
projectId: data.snapshot.projectId,

‎internal-packages/run-store/src/types.ts‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -79,6 +79,9 @@ export type RescheduleSnapshotInput = {
7979
environmentType: RuntimeEnvironmentType;
8080
projectId: string;
8181
organizationId: string;
82+
executionStatus?: TaskRunExecutionStatus;
83+
runStatus?: TaskRunStatus;
84+
description?: string;
8285
};
8386

8487
export type LockSnapshotInput = {

0 commit comments

Comments
 (0)