Skip to content

Commit 420328d

Browse files
d-csd-cs
committed
feat(run-engine,run-store): write the completed-waitpoint record set at the resume appends
Carries an envelope per distinct id from the resume path into the wait cycle's key, filling the hole the snapshot store left for this lane. The records ride the mint only: a copy-forward writes no key and needs none. continueRunIfUnblocked builds the set once and passes it at both appends. The build is gated on id shape, so a wait with no store-resident half supplies no records and a Postgres-resident resume is byte-identical to before. Nothing mints a store-format waitpoint yet, so every live path supplies none today. The existing waitpoint corpus passes unmodified.
1 parent f4f5df2 commit 420328d

6 files changed

Lines changed: 319 additions & 13 deletions

File tree

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

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,7 @@ import type {
44
TaskRun,
55
TaskRunExecutionStatus,
66
} from "@trigger.dev/database";
7-
import type { RunStore } from "@internal/run-store";
7+
import type { CompletedWaitpointRecord, RunStore } from "@internal/run-store";
88
import { parseNaturalLanguageDuration } from "@trigger.dev/core/v3/isomorphic";
99
import type { MinimalAuthenticatedEnvironment } from "../../shared/index.js";
1010
import { QUEUED_SNAPSHOT_DESCRIPTION, QUEUED_SNAPSHOT_STATUS } from "../consts.js";
@@ -34,6 +34,7 @@ export class EnqueueSystem {
3434
batchId,
3535
checkpointId,
3636
completedWaitpoints,
37+
completedWaitpointRecords,
3738
workerId,
3839
runnerId,
3940
skipRunLock,
@@ -57,6 +58,7 @@ export class EnqueueSystem {
5758
id: string;
5859
index?: number;
5960
}[];
61+
completedWaitpointRecords?: CompletedWaitpointRecord[];
6062
workerId?: string;
6163
runnerId?: string;
6264
skipRunLock?: boolean;
@@ -108,6 +110,7 @@ export class EnqueueSystem {
108110
organizationId: env.organization.id,
109111
checkpointId,
110112
completedWaitpoints,
113+
completedWaitpointRecords,
111114
workerId,
112115
runnerId,
113116
},

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

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,7 @@ import type {
1010
TaskRunStatus,
1111
Waitpoint,
1212
} from "@trigger.dev/database";
13-
import type { RunStore } from "@internal/run-store";
13+
import type { CompletedWaitpointRecord, RunStore } from "@internal/run-store";
1414
import { ExecutionSnapshotNotFoundError, ServiceValidationError } from "../errors.js";
1515
import type { HeartbeatTimeouts } from "../types.js";
1616
import type { SystemResources } from "./systems.js";
@@ -449,6 +449,7 @@ export class ExecutionSnapshotSystem {
449449
workerId,
450450
runnerId,
451451
completedWaitpoints,
452+
completedWaitpointRecords,
452453
error,
453454
}: {
454455
run: { id: string; status: TaskRunStatus; attemptNumber?: number | null };
@@ -470,6 +471,7 @@ export class ExecutionSnapshotSystem {
470471
id: string;
471472
index?: number;
472473
}[];
474+
completedWaitpointRecords?: CompletedWaitpointRecord[];
473475
error?: string;
474476
},
475477
// When set (inside runStore.runInTransaction), the snapshot write goes through the owning store
@@ -492,6 +494,7 @@ export class ExecutionSnapshotSystem {
492494
workerId,
493495
runnerId,
494496
completedWaitpoints,
497+
completedWaitpointRecords,
495498
error,
496499
},
497500
prisma

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

Lines changed: 46 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,6 @@
11
import { timeoutError } from "@trigger.dev/core/v3";
2+
import { parseWaitpointId } from "@trigger.dev/core/v3/isomorphic";
3+
import type { CompletedWaitpointRecord } from "@internal/run-store";
24
import type {
35
PrismaClientOrTransaction,
46
TaskRun,
@@ -10,7 +12,8 @@ import { assertNever } from "assert-never";
1012
import { sendNotificationToWorker } from "../eventBus.js";
1113
import { isFinalRunStatus } from "../statuses.js";
1214
import { LegacyPostgresWaitpointCoordinator } from "../waitpointCoordinator/legacyPostgresCoordinator.js";
13-
import type { WaitpointCoordinator } from "../waitpointCoordinator/types.js";
15+
import { buildCompletedWaitpointRecords } from "../waitpointCoordinator/completedWaitpointRecords.js";
16+
import type { RunBlockEdge, WaitpointCoordinator } from "../waitpointCoordinator/types.js";
1417
import type { EnqueueSystem } from "./enqueueSystem.js";
1518
import type { ExecutionSnapshotSystem } from "./executionSnapshotSystem.js";
1619
import { getLatestExecutionSnapshot } from "./executionSnapshotSystem.js";
@@ -484,6 +487,15 @@ export class WaitpointSystem {
484487
};
485488
}
486489

490+
// The record set rides the wait cycle's key once per resume, so build it here rather
491+
// than at each append site. Nothing mints a store-format waitpoint yet, so
492+
// #completedWaitpointRecordsFor returns undefined on every live path today.
493+
const completedWaitpointRecords = await this.#completedWaitpointRecordsFor(
494+
runId,
495+
blockingWaitpoints
496+
);
497+
498+
487499
// 3. Get the run (run-ops scalars) + resolve its environment via the control-plane resolver,
488500
// so the run-ops DB can split without a cross-provider join.
489501
const run = await this.$.runStore.findRun(
@@ -623,6 +635,7 @@ export class WaitpointSystem {
623635
id: b.waitpoint.id,
624636
index: b.batchIndex ?? undefined,
625637
})),
638+
...(completedWaitpointRecords && { completedWaitpointRecords }),
626639
}
627640
);
628641

@@ -682,6 +695,7 @@ export class WaitpointSystem {
682695
id: b.waitpoint.id,
683696
index: b.batchIndex ?? undefined,
684697
})),
698+
...(completedWaitpointRecords && { completedWaitpointRecords }),
685699
checkpointId: snapshot.checkpointId ?? undefined,
686700
});
687701

@@ -728,6 +742,37 @@ export class WaitpointSystem {
728742
return this.coordinator.mintAssociatedWaitpointData({ projectId, environmentId });
729743
}
730744

745+
/**
746+
* The record set for one resume, or undefined when this wait has no store-resident half.
747+
*
748+
* The classification gate is what keeps this inert. `parseWaitpointId` reports legacy for
749+
* every id minted today, so no live resume reads an envelope or writes a record until a
750+
* waitpoint mints in store format.
751+
*/
752+
async #completedWaitpointRecordsFor(
753+
runId: string,
754+
blockingWaitpoints: RunBlockEdge[]
755+
): Promise<CompletedWaitpointRecord[] | undefined> {
756+
const storeResidentIds = [
757+
...new Set(
758+
blockingWaitpoints
759+
.map((b) => b.waitpoint.id)
760+
.filter((id) => parseWaitpointId(id).format === "b32hexW")
761+
),
762+
];
763+
764+
if (storeResidentIds.length === 0) {
765+
return undefined;
766+
}
767+
768+
const sources = await this.coordinator.readCompletionEnvelopes({
769+
runId,
770+
waitpointIds: storeResidentIds,
771+
});
772+
773+
return buildCompletedWaitpointRecords(sources);
774+
}
775+
731776
/**
732777
* Builds the waitpoint output payload from a completed run's stored output/error.
733778
*/

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

Lines changed: 34 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,7 @@ import { Logger } from "@trigger.dev/core/logger";
1515
import { generateInternalId } from "@trigger.dev/core/v3/isomorphic";
1616
import { DelegatingRunStore } from "./delegatingRunStore.js";
1717
import type {
18+
CompletedWaitpointRecord,
1819
CompletedWaitpointRef,
1920
RedisSnapshotStore,
2021
SnapshotEntryInput,
@@ -114,6 +115,7 @@ export type StagedAppend = {
114115
*/
115116
expectedCur?: string;
116117
completedWaitpoints?: CompletedWaitpointRef[];
118+
completedWaitpointRecords?: CompletedWaitpointRecord[];
117119
};
118120

119121
export class TaskRunExecutionSnapshotStore extends DelegatingRunStore {
@@ -177,7 +179,8 @@ export class TaskRunExecutionSnapshotStore extends DelegatingRunStore {
177179
"runInTransaction",
178180
item.entry,
179181
item.expectedCur,
180-
item.completedWaitpoints
182+
item.completedWaitpoints,
183+
item.completedWaitpointRecords
181184
);
182185
}
183186

@@ -420,7 +423,8 @@ export class TaskRunExecutionSnapshotStore extends DelegatingRunStore {
420423
"createExecutionSnapshot",
421424
entryFromCreateExecutionSnapshot(ctx, input),
422425
input.previousSnapshotId,
423-
input.completedWaitpoints
426+
input.completedWaitpoints,
427+
input.completedWaitpointRecords
424428
);
425429
return created;
426430
}
@@ -503,7 +507,8 @@ export class TaskRunExecutionSnapshotStore extends DelegatingRunStore {
503507
site: string,
504508
entry: SnapshotEntryInput,
505509
expectedCur?: string,
506-
completedWaitpoints?: CompletedWaitpointRef[]
510+
completedWaitpoints?: CompletedWaitpointRef[],
511+
completedWaitpointRecords?: CompletedWaitpointRecord[]
507512
): Promise<void> {
508513
if (this.staging) {
509514
// Inside a transaction the append cannot run until the Postgres side commits, or a rollback
@@ -512,6 +517,7 @@ export class TaskRunExecutionSnapshotStore extends DelegatingRunStore {
512517
entry,
513518
...(expectedCur !== undefined && { expectedCur }),
514519
...(completedWaitpoints && { completedWaitpoints }),
520+
...(completedWaitpointRecords && { completedWaitpointRecords }),
515521
});
516522
return;
517523
}
@@ -523,7 +529,11 @@ export class TaskRunExecutionSnapshotStore extends DelegatingRunStore {
523529
snapshotId: entry.id,
524530
});
525531

526-
const cycle = await this.#resolveCycle(entry.runId, completedWaitpoints);
532+
const cycle = await this.#resolveCycle(
533+
entry.runId,
534+
completedWaitpoints,
535+
completedWaitpointRecords
536+
);
527537

528538
const result = await this.redis.append({
529539
entry,
@@ -572,15 +582,28 @@ export class TaskRunExecutionSnapshotStore extends DelegatingRunStore {
572582
* The extra read only happens for an append that actually carries waitpoints, which is the resume
573583
* path rather than the hot path.
574584
*
575-
* `records` is deliberately left unset. The record envelope belongs to the waitpoint lane and
576-
* ships empty in this build, so dual-write never re-versions the entry when it arrives.
585+
* `records` rides every arm that can mint. A carryForward normally writes no key, but the
586+
* store may refuse the pointer and mint a replacement inside the same call, and that
587+
* replacement needs the records or the resolver's coverage check rejects the cycle later.
588+
* A legacy-only wait supplies none at all, which is what keeps a Postgres-resident resume
589+
* byte-identical to before.
577590
*/
578591
async #resolveCycle(
579592
runId: string,
580-
completedWaitpoints?: CompletedWaitpointRef[]
593+
completedWaitpoints?: CompletedWaitpointRef[],
594+
records?: CompletedWaitpointRecord[]
581595
): Promise<
582-
| { kind: "new"; completedWaitpoints: CompletedWaitpointRef[] }
583-
| { kind: "carryForward"; cycleSeq: number; completedWaitpoints: CompletedWaitpointRef[] }
596+
| {
597+
kind: "new";
598+
completedWaitpoints: CompletedWaitpointRef[];
599+
records?: CompletedWaitpointRecord[];
600+
}
601+
| {
602+
kind: "carryForward";
603+
cycleSeq: number;
604+
completedWaitpoints: CompletedWaitpointRef[];
605+
records?: CompletedWaitpointRecord[];
606+
}
584607
| undefined
585608
> {
586609
if (!completedWaitpoints || completedWaitpoints.length === 0) {
@@ -607,6 +630,7 @@ export class TaskRunExecutionSnapshotStore extends DelegatingRunStore {
607630
kind: "carryForward",
608631
cycleSeq: head.cycle.cycleSeq,
609632
completedWaitpoints,
633+
...(records && { records }),
610634
};
611635
}
612636
} catch (error) {
@@ -616,7 +640,7 @@ export class TaskRunExecutionSnapshotStore extends DelegatingRunStore {
616640
this.logger.warn("snapshot cycle probe failed, minting a new cycle", { runId, error });
617641
}
618642

619-
return { kind: "new", completedWaitpoints };
643+
return { kind: "new", completedWaitpoints, ...(records && { records }) };
620644
}
621645

622646
/**

0 commit comments

Comments
 (0)