From 3c0f344dfdc04171736ef43aa0075676cf65f847 Mon Sep 17 00:00:00 2001 From: Dan Sutton Date: Fri, 21 Aug 2026 13:17:12 +0100 Subject: [PATCH] refactor(run-engine): add WaitpointCoordinator seam with clearRunBlockState --- .../src/engine/systems/waitpointSystem.ts | 27 +++++----- .../legacyPostgresCoordinator.ts | 51 +++++++++++++++++++ .../src/engine/waitpointCoordinator/types.ts | 28 ++++++++++ 3 files changed, 92 insertions(+), 14 deletions(-) create mode 100644 internal-packages/run-engine/src/engine/waitpointCoordinator/legacyPostgresCoordinator.ts create mode 100644 internal-packages/run-engine/src/engine/waitpointCoordinator/types.ts diff --git a/internal-packages/run-engine/src/engine/systems/waitpointSystem.ts b/internal-packages/run-engine/src/engine/systems/waitpointSystem.ts index 5d5a80772..d43e811c6 100644 --- a/internal-packages/run-engine/src/engine/systems/waitpointSystem.ts +++ b/internal-packages/run-engine/src/engine/systems/waitpointSystem.ts @@ -7,13 +7,15 @@ import type { TaskRunExecutionStatus, Waitpoint, } from "@trigger.dev/database"; -import { Prisma, boundedIn } from "@trigger.dev/database"; +import { Prisma } from "@trigger.dev/database"; import type { RunStore } from "@internal/run-store"; import { assertNever } from "assert-never"; import { nanoid } from "nanoid"; import { UnclassifiableWaitpointId } from "../errors.js"; import { sendNotificationToWorker } from "../eventBus.js"; import { isFinalRunStatus } from "../statuses.js"; +import { LegacyPostgresWaitpointCoordinator } from "../waitpointCoordinator/legacyPostgresCoordinator.js"; +import type { WaitpointCoordinator } from "../waitpointCoordinator/types.js"; import type { EnqueueSystem } from "./enqueueSystem.js"; import type { ExecutionSnapshotSystem } from "./executionSnapshotSystem.js"; import { getLatestExecutionSnapshot } from "./executionSnapshotSystem.js"; @@ -45,11 +47,17 @@ export class WaitpointSystem { private readonly $: SystemResources; private readonly executionSnapshotSystem: ExecutionSnapshotSystem; private readonly enqueueSystem: EnqueueSystem; + private readonly coordinator: WaitpointCoordinator; constructor(private readonly options: WaitpointSystemOptions) { this.$ = options.resources; this.executionSnapshotSystem = options.executionSnapshotSystem; this.enqueueSystem = options.enqueueSystem; + this.coordinator = new LegacyPostgresWaitpointCoordinator({ + runStore: this.$.runStore, + prisma: this.$.prisma, + logger: this.$.logger, + }); } public async clearBlockingWaitpoints({ @@ -59,14 +67,7 @@ export class WaitpointSystem { runId: string; tx?: PrismaClientOrTransaction; }) { - // A run's edges co-locate with the run (the edge write routes by runId), so the router routes this - // taskRunId-keyed delete to the run's store rather than fanning out. The caller's `tx` is not - // forwarded — the delete runs on the owning store's own client (the router never threads a - // control-plane tx into a routed write). - const deleted = await this.$.runStore.deleteManyTaskRunWaitpoints( - { where: { taskRunId: runId } }, - tx - ); + const deleted = await this.coordinator.clearRunBlockState({ runId, tx }); return deleted.count; } @@ -926,11 +927,9 @@ export class WaitpointSystem { if (blockingWaitpoints.length > 0) { //5. Remove the blocking waitpoints - await this.$.runStore.deleteManyTaskRunWaitpoints({ - where: { - taskRunId: runId, - id: { in: boundedIn(blockingWaitpoints.map((b) => b.id)) }, - }, + await this.coordinator.clearRunBlockState({ + runId, + edgeIds: blockingWaitpoints.map((b) => b.id), }); this.$.logger.debug(`continueRunIfUnblocked: removed blocking waitpoints`, { diff --git a/internal-packages/run-engine/src/engine/waitpointCoordinator/legacyPostgresCoordinator.ts b/internal-packages/run-engine/src/engine/waitpointCoordinator/legacyPostgresCoordinator.ts new file mode 100644 index 000000000..b2e6aadb2 --- /dev/null +++ b/internal-packages/run-engine/src/engine/waitpointCoordinator/legacyPostgresCoordinator.ts @@ -0,0 +1,51 @@ +import type { RunStore } from "@internal/run-store"; +import type { Logger } from "@trigger.dev/core/logger"; +import type { PrismaClient } from "@trigger.dev/database"; +import { boundedIn } from "@trigger.dev/database"; +import type { ClearRunBlockStateParams, WaitpointCoordinator } from "./types.js"; + +export type LegacyPostgresWaitpointCoordinatorOptions = { + runStore: RunStore; + prisma: PrismaClient; + logger: Logger; +}; + +/** + * Waitpoint coordination against Postgres, through the run-ops store. + * + * Dependencies are deliberately narrow: no run lock, no worker, no event bus. + * That makes "this owns waitpoint state only" structural rather than a convention. + */ +export class LegacyPostgresWaitpointCoordinator implements WaitpointCoordinator { + private readonly runStore: RunStore; + private readonly prisma: PrismaClient; + private readonly logger: Logger; + + constructor(options: LegacyPostgresWaitpointCoordinatorOptions) { + this.runStore = options.runStore; + this.prisma = options.prisma; + this.logger = options.logger; + } + + async clearRunBlockState({ + runId, + edgeIds, + tx, + }: ClearRunBlockStateParams): Promise<{ count: number }> { + if (edgeIds) { + // Bounded delete of named edges, on the unblock path. No tx: that path is not inside a + // caller transaction, and boundedIn caps the id-list arity for Prisma. + return this.runStore.deleteManyTaskRunWaitpoints({ + where: { + taskRunId: runId, + id: { in: boundedIn(edgeIds) }, + }, + }); + } + + // A run's edges co-locate with the run (the edge write routes by runId), so the router routes + // this taskRunId-keyed delete to the run's store rather than fanning out. The caller's `tx` is + // passed through: a routing store strips it, and a single store joins it. + return this.runStore.deleteManyTaskRunWaitpoints({ where: { taskRunId: runId } }, tx); + } +} diff --git a/internal-packages/run-engine/src/engine/waitpointCoordinator/types.ts b/internal-packages/run-engine/src/engine/waitpointCoordinator/types.ts new file mode 100644 index 000000000..8db4b986b --- /dev/null +++ b/internal-packages/run-engine/src/engine/waitpointCoordinator/types.ts @@ -0,0 +1,28 @@ +import type { PrismaClientOrTransaction } from "@trigger.dev/database"; + +/** + * The waitpoint and edge state operations that `WaitpointSystem` delegates. + * + * Orchestration stays in `WaitpointSystem`: the run lock, snapshot transitions, + * worker-job enqueues, event emissions and racepoints. This owns waitpoint and + * edge state only, so a non-Postgres implementation can replace it without any + * caller learning that it changed. + * + * The residency hints and `tx` are opaque pass-throughs. Opaque does not mean + * type-free — a Prisma type appears here — it means a non-Postgres implementation + * never reads the value. + */ +export type WaitpointCoordinator = { + clearRunBlockState(params: ClearRunBlockStateParams): Promise<{ count: number }>; +}; + +export type ClearRunBlockStateParams = { + runId: string; + /** Edge ids to delete. Omit to clear every edge for the run. */ + edgeIds?: string[]; + /** + * Forwarded verbatim on the full-clear leg only, and never on the bounded leg + * or an edge write. A routing store strips it; a single store joins it. + */ + tx?: PrismaClientOrTransaction; +};