Skip to content

Commit 31bdbee

Browse files
myftijaTrigger.dev RepoOps
myftija
authored and
Trigger.dev RepoOps
committed
fix(run-engine): keep a run's execution snapshots in write order across engine clocks
Fix a rare case where a run could stay executing after its task finished, or a wait could never resume, when the run engine instances handling the run briefly disagreed on the current time. The run engine is now correct in the presence of clock skew between its instances. A run's execution snapshots are written by whichever instance handles each transition. Both the snapshot's `createdAt`, which the engine uses to find a run's current snapshot, and its id, which runners use to tell whether a snapshot is newer, came from that instance's clock. A snapshot written by an instance whose clock was behind could therefore sort before the snapshot it replaced. When that happened: - the runner's attempt completion was rejected; - the stall check stopped without failing the run; - a wait could suspend the run with no checkpoint to resume from. Each new snapshot is now stamped and identified no earlier than 1ms after the snapshot it follows. A run's snapshots are always ordered in the order they were written, by `createdAt` and by id, whatever each instance's clock says, and runners that are already deployed see the same order as the engine. Self-hosted deployments that run more than one webapp instance get this without any configuration change. A new `run_engine.snapshot.created_at_clamp` histogram records how far ahead of the local clock a snapshot had to be stamped, which also makes clock drift between instances visible. Mono-RevId: b4344abb1441b4c295a1ac77fb5c7e11800f5b67
1 parent f65fa00 commit 31bdbee

13 files changed

Lines changed: 870 additions & 8 deletions

‎.server-changes/run-snapshot-ordering.md‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,6 @@
1+
---
2+
area: webapp
3+
type: fix
4+
---
5+
6+
Fix a rare case where clock skew between run engine instances could leave a run executing after its task finished, or stop a wait from ever resuming. A run's state changes are now always ordered in the order they happened, regardless of clock skew.

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

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -441,6 +441,7 @@ export class RunEngine {
441441
this.delayedRunSystem = new DelayedRunSystem({
442442
resources,
443443
enqueueSystem: this.enqueueSystem,
444+
executionSnapshotSystem: this.executionSnapshotSystem,
444445
});
445446

446447
this.debounceSystem = new DebounceSystem({
@@ -475,6 +476,7 @@ export class RunEngine {
475476
this.ttlSystem = new TtlSystem({
476477
resources,
477478
waitpointSystem: this.waitpointSystem,
479+
executionSnapshotSystem: this.executionSnapshotSystem,
478480
finalizationGuardDelayMs: this.options.finalizationGuardDelayMs,
479481
});
480482

@@ -2768,6 +2770,7 @@ export class RunEngine {
27682770
const { wasRequeued } = await this.runAttemptSystem.tryNackAndRequeue({
27692771
run,
27702772
snapshotRoute: snapshotRouteFromSnapshot(latestSnapshot),
2773+
previousSnapshot: latestSnapshot,
27712774
environment: {
27722775
id: latestSnapshot.environmentId,
27732776
type: latestSnapshot.environmentType,

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

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -212,6 +212,7 @@ export class CheckpointSystem {
212212
metadata: snapshot.metadata,
213213
},
214214
previousSnapshotId: snapshot.id,
215+
previousSnapshot: snapshot,
215216
snapshotRoute: snapshotRouteFromSnapshot(snapshot),
216217
batchId: snapshot.batchId ?? undefined,
217218
completedWaitpoints: snapshot.completedWaitpoints.map((waitpoint) => ({
@@ -248,6 +249,7 @@ export class CheckpointSystem {
248249
metadata: snapshot.metadata,
249250
},
250251
previousSnapshotId: snapshot.id,
252+
previousSnapshot: snapshot,
251253
snapshotRoute: snapshotRouteFromSnapshot(snapshot),
252254
batchId: snapshot.batchId ?? undefined,
253255
completedWaitpoints: snapshot.completedWaitpoints.map((waitpoint) => ({
@@ -417,6 +419,7 @@ export class CheckpointSystem {
417419
description: CONTINUED_AFTER_SUSPENDED_DESCRIPTION,
418420
},
419421
previousSnapshotId: snapshot.id,
422+
previousSnapshot: snapshot,
420423
snapshotRoute: snapshotRouteFromSnapshot(snapshot),
421424
environmentId: snapshot.environmentId,
422425
environmentType: snapshot.environmentType,

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

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@ import type { PrismaClientOrTransaction, TaskRun } from "@trigger.dev/database";
55
import {
66
getLatestExecutionSnapshot,
77
getLatestExecutionSnapshotOrFinishedRun,
8+
type ExecutionSnapshotSystem,
89
} from "./executionSnapshotSystem.js";
910
import { parseNaturalLanguageDuration } from "@trigger.dev/core/v3/isomorphic";
1011
import type { EnqueueSystem } from "./enqueueSystem.js";
@@ -14,15 +15,18 @@ import { ServiceValidationError } from "../errors.js";
1415
export type DelayedRunSystemOptions = {
1516
resources: SystemResources;
1617
enqueueSystem: EnqueueSystem;
18+
executionSnapshotSystem: ExecutionSnapshotSystem;
1719
};
1820

1921
export class DelayedRunSystem {
2022
private readonly $: SystemResources;
2123
private readonly enqueueSystem: EnqueueSystem;
24+
private readonly executionSnapshotSystem: ExecutionSnapshotSystem;
2225

2326
constructor(private readonly options: DelayedRunSystemOptions) {
2427
this.$ = options.resources;
2528
this.enqueueSystem = options.enqueueSystem;
29+
this.executionSnapshotSystem = options.executionSnapshotSystem;
2630
}
2731

2832
/**
@@ -68,6 +72,7 @@ export class DelayedRunSystem {
6872
{
6973
delayUntil: delayUntil,
7074
snapshot: {
75+
...this.executionSnapshotSystem.nextSnapshotIdentity(snapshot, runId),
7176
snapshotRoute: snapshotRouteFromSnapshot(snapshot),
7277
environmentId: snapshot.environmentId,
7378
environmentType: snapshot.environmentType,
@@ -238,6 +243,7 @@ export class DelayedRunSystem {
238243
run,
239244
env,
240245
previousSnapshotId: snapshot.id,
246+
previousSnapshot: snapshot,
241247
snapshotRoute: snapshotRouteFromSnapshot(snapshot),
242248
batchId: run.batchId ?? undefined,
243249
skipRunLock: true,

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

Lines changed: 13 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,7 @@ import { startSpan } from "@internal/tracing";
22
import { assertExhaustive, tryCatch } from "@trigger.dev/core";
33
import type { DequeuedMessage, TaskRunInternalError } from "@trigger.dev/core/v3";
44
import { RetryOptions, RunAnnotations } from "@trigger.dev/core/v3";
5-
import { generateInternalId, getMaxDuration, SnapshotId } from "@trigger.dev/core/v3/isomorphic";
5+
import { getMaxDuration, SnapshotId } from "@trigger.dev/core/v3/isomorphic";
66
import { placementTag } from "@trigger.dev/core/v3/serverOnly";
77
import type {
88
BackgroundWorker,
@@ -27,6 +27,7 @@ import type { RunEngineOptions, WorkerQueueDequeueOptions } from "../types.js";
2727
import {
2828
getLatestExecutionSnapshotOrFinishedRun,
2929
type ExecutionSnapshotSystem,
30+
type PreviousSnapshot,
3031
} from "./executionSnapshotSystem.js";
3132
import type { RunAttemptSystem } from "./runAttemptSystem.js";
3233
import type { SystemResources } from "./systems.js";
@@ -254,6 +255,7 @@ export class DequeueSystem {
254255
"Tried to dequeue a run that is not in a valid state to be dequeued.",
255256
},
256257
previousSnapshotId: snapshot.id,
258+
previousSnapshot: snapshot,
257259
snapshotRoute: snapshotRouteFromSnapshot(snapshot),
258260
environmentId: snapshot.environmentId,
259261
environmentType: snapshot.environmentType,
@@ -361,6 +363,7 @@ export class DequeueSystem {
361363
description: "Run was continued, whilst still executing.",
362364
},
363365
previousSnapshotId: snapshot.id,
366+
previousSnapshot: snapshot,
364367
snapshotRoute: snapshotRouteFromSnapshot(snapshot),
365368
environmentId: snapshot.environmentId,
366369
environmentType: snapshot.environmentType,
@@ -449,6 +452,7 @@ export class DequeueSystem {
449452
orgId,
450453
runId,
451454
snapshotRoute: snapshotRouteFromSnapshot(snapshot),
455+
previousSnapshot: snapshot,
452456
reason: result.message,
453457
statusReason: result.code,
454458
tx: prisma,
@@ -489,6 +493,7 @@ export class DequeueSystem {
489493
orgId,
490494
runId,
491495
snapshotRoute: snapshotRouteFromSnapshot(snapshot),
496+
previousSnapshot: snapshot,
492497
reason: "No deployment or deployment image reference found for deployed run",
493498
statusReason: "NO_DEPLOYMENT",
494499
tx: prisma,
@@ -562,7 +567,8 @@ export class DequeueSystem {
562567
: result.task.retryConfig;
563568

564569
// Pre-generate snapshot ID so we can construct the result without an extra read
565-
const snapshotId = generateInternalId();
570+
const { id: snapshotId, createdAt: snapshotCreatedAt } =
571+
this.executionSnapshotSystem.nextSnapshotIdentity(snapshot, runId);
566572

567573
const lockedTaskRun = await this.$.runStore.lockRunToWorker(
568574
runId,
@@ -582,6 +588,7 @@ export class DequeueSystem {
582588
maxAttempts: maxAttempts ?? undefined,
583589
snapshot: {
584590
id: snapshotId,
591+
createdAt: snapshotCreatedAt,
585592
previousSnapshotId: snapshot.id,
586593
snapshotRoute: snapshotRouteFromSnapshot(snapshot),
587594
attemptNumber: result.run.attemptNumber ?? undefined,
@@ -683,8 +690,6 @@ export class DequeueSystem {
683690
// Snapshot was created as part of the taskRun.update above (single transaction).
684691
// Construct the snapshot info from data we already have and handle side effects
685692
// (heartbeat + event) manually — no extra DB read needed.
686-
const snapshotCreatedAt = new Date();
687-
688693
this.$.eventBus.emit("executionSnapshotCreated", {
689694
time: snapshotCreatedAt,
690695
run: {
@@ -880,6 +885,7 @@ export class DequeueSystem {
880885
batchId: snapshot.batchId ?? undefined,
881886
error: dequeueMaxRetriesError,
882887
snapshotRoute: snapshotRouteFromSnapshot(snapshot),
888+
previousSnapshot: snapshot,
883889
tx: prisma,
884890
});
885891
});
@@ -973,6 +979,7 @@ export class DequeueSystem {
973979
orgId,
974980
runId,
975981
snapshotRoute,
982+
previousSnapshot,
976983
workerId,
977984
runnerId,
978985
reason,
@@ -982,6 +989,7 @@ export class DequeueSystem {
982989
orgId: string;
983990
runId: string;
984991
snapshotRoute?: SnapshotRouteWire;
992+
previousSnapshot?: PreviousSnapshot;
985993
statusReason: string;
986994
workerId?: string;
987995
runnerId?: string;
@@ -1040,6 +1048,7 @@ export class DequeueSystem {
10401048
await this.executionSnapshotSystem.createExecutionSnapshot(prisma, {
10411049
run,
10421050
snapshotRoute,
1051+
previousSnapshot,
10431052
snapshot: {
10441053
executionStatus: "RUN_CREATED",
10451054
description:

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

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,7 @@ import type { MinimalAuthenticatedEnvironment } from "../../shared/index.js";
1212
import { QUEUED_SNAPSHOT_DESCRIPTION, QUEUED_SNAPSHOT_STATUS } from "../consts.js";
1313
import { parseGates } from "../gateParsing.js";
1414
import { workerQueueForPublish } from "../workerQueueRouting.js";
15-
import type { ExecutionSnapshotSystem } from "./executionSnapshotSystem.js";
15+
import type { ExecutionSnapshotSystem, PreviousSnapshot } from "./executionSnapshotSystem.js";
1616
import type { SystemResources } from "./systems.js";
1717

1818
export type EnqueueSystemOptions = {
@@ -42,6 +42,7 @@ export class EnqueueSystem {
4242
tx,
4343
snapshot,
4444
previousSnapshotId,
45+
previousSnapshot,
4546
snapshotRoute,
4647
batchId,
4748
checkpoint,
@@ -64,6 +65,7 @@ export class EnqueueSystem {
6465
metadata?: Prisma.JsonValue;
6566
};
6667
previousSnapshotId?: string;
68+
previousSnapshot?: PreviousSnapshot;
6769
/** Server-side only: never published to workers or exposed through an API. */
6870
snapshotRoute?: SnapshotRouteWire;
6971
batchId?: string;
@@ -110,6 +112,7 @@ export class EnqueueSystem {
110112
metadata: snapshot?.metadata ?? undefined,
111113
},
112114
previousSnapshotId,
115+
previousSnapshot,
113116
snapshotRoute,
114117
batchId,
115118
environmentId: env.id,

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

Lines changed: 70 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,7 @@ import {
3131
import { isFinalRunStatus } from "../statuses.js";
3232
import type { HeartbeatTimeouts } from "../types.js";
3333
import type { SystemResources } from "./systems.js";
34+
import type { Histogram } from "@internal/tracing";
3435

3536
import { boundedIn } from "@trigger.dev/database";
3637
/** Waitpoint outputs can be 100KB+, so a large fetch can exceed the NAPI string limit. */
@@ -485,13 +486,75 @@ export async function getExecutionSnapshotsSince(
485486
});
486487
}
487488

489+
const CREATED_AT_CLAMP_WARN_MS = 5;
490+
491+
export type PreviousSnapshot = { id: string; createdAt: Date };
492+
493+
export type SnapshotIdentity = { id: string; createdAt: Date };
494+
495+
const CUID_TIMESTAMP_LENGTH = 8;
496+
const CUID_TAIL_LENGTH = 16;
497+
498+
function cuidTimestampMs(id: string): number | undefined {
499+
if (id.length !== 1 + CUID_TIMESTAMP_LENGTH + CUID_TAIL_LENGTH || id[0] !== "c") {
500+
return undefined;
501+
}
502+
const timestamp = parseInt(id.slice(1, 1 + CUID_TIMESTAMP_LENGTH), 36);
503+
return Number.isFinite(timestamp) ? timestamp : undefined;
504+
}
505+
506+
function snapshotIdAt(timestampMs: number): string {
507+
const fresh = SnapshotId.generate().id;
508+
return `c${timestampMs.toString(36)}${fresh.slice(-CUID_TAIL_LENGTH)}`;
509+
}
510+
488511
export class ExecutionSnapshotSystem {
489512
private readonly $: SystemResources;
490513
private readonly heartbeatTimeouts: HeartbeatTimeouts;
514+
private readonly createdAtClampHistogram: Histogram;
491515

492516
constructor(private readonly options: ExecutionSnapshotSystemOptions) {
493517
this.$ = options.resources;
494518
this.heartbeatTimeouts = options.heartbeatTimeouts;
519+
this.createdAtClampHistogram = this.$.meter.createHistogram(
520+
"run_engine.snapshot.created_at_clamp",
521+
{
522+
description:
523+
"How far ahead of the writer's clock a snapshot was stamped to sort after the snapshot it follows",
524+
unit: "ms",
525+
}
526+
);
527+
}
528+
529+
// Engine orders snapshots by createdAt, runners by id, both from the writer's clock: keep both after the predecessor.
530+
public nextSnapshotIdentity(previous: PreviousSnapshot, runId: string): SnapshotIdentity {
531+
const now = Date.now();
532+
const floor =
533+
Math.max(
534+
previous.createdAt.getTime(),
535+
cuidTimestampMs(previous.id) ?? Number.NEGATIVE_INFINITY
536+
) + 1;
537+
if (now >= floor) {
538+
return { id: SnapshotId.generate().id, createdAt: new Date(now) };
539+
}
540+
541+
const clampMs = floor - now;
542+
if (clampMs > 1) {
543+
this.createdAtClampHistogram.record(clampMs);
544+
}
545+
if (clampMs > CREATED_AT_CLAMP_WARN_MS) {
546+
this.$.logger.warn(
547+
"Snapshot stamped ahead of this engine's clock to follow its predecessor",
548+
{
549+
runId,
550+
clampMs,
551+
previousSnapshotId: previous.id,
552+
previousCreatedAt: previous.createdAt,
553+
}
554+
);
555+
}
556+
557+
return { id: snapshotIdAt(floor), createdAt: new Date(floor) };
495558
}
496559

497560
public async createExecutionSnapshot(
@@ -501,6 +564,7 @@ export class ExecutionSnapshotSystem {
501564
run,
502565
snapshot,
503566
previousSnapshotId,
567+
previousSnapshot,
504568
snapshotRoute,
505569
batchId,
506570
environmentId,
@@ -526,6 +590,7 @@ export class ExecutionSnapshotSystem {
526590
metadata?: Prisma.JsonValue;
527591
};
528592
previousSnapshotId?: string;
593+
previousSnapshot?: PreviousSnapshot;
529594
snapshotRoute?: SnapshotRouteWire;
530595
batchId?: string;
531596
environmentId: string;
@@ -549,11 +614,15 @@ export class ExecutionSnapshotSystem {
549614
store?: RunStore
550615
) {
551616
// Minted above the store, so a retry inside the store replays the same transition idempotently.
552-
const id = snapshotId ?? SnapshotId.generate().id;
617+
const identity = previousSnapshot
618+
? this.nextSnapshotIdentity(previousSnapshot, run.id)
619+
: undefined;
620+
const id = snapshotId ?? identity?.id ?? SnapshotId.generate().id;
553621

554622
const newSnapshot = await (store ?? this.$.runStore).createExecutionSnapshot(
555623
{
556624
id,
625+
createdAt: identity?.createdAt,
557626
run,
558627
snapshot,
559628
previousSnapshotId,

0 commit comments

Comments
 (0)