refactor(run-engine): extract a WaitpointCoordinator seam around the Postgres waitpoint implementation - #4753
refactor(run-engine): extract a WaitpointCoordinator seam around the Postgres waitpoint implementation#4753d-cs wants to merge 8 commits into
Conversation
|
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Repository UI Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (3)
🚧 Files skipped from review as they are similar to previous changes (3)
Included review availability: Your plan provides up to 10 included reviews per hour; 5 remain after this review. 📜 Recent review details⏰ Context from checks skipped due to timeout. (33)
WalkthroughAdded the Merge Risk: ⚪ Minimal · up to This refactor preserves the existing waitpoint behavior and introduces no actionable merge-blocking risk; it is merge-ready after normal checks and review. 🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
🧹 Nitpick comments (1)
internal-packages/run-engine/src/engine/systems/waitpointSystem.ts (1)
45-55: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winConsider injecting the coordinator through
WaitpointSystemOptions.
WaitpointSystemconstructsLegacyPostgresWaitpointCoordinatordirectly. The seam therefore cannot be substituted by any caller, and tests cannot supply a fake coordinator. An optional option keeps the current default and completes the stated goal that a non-Postgres implementation can replace it without caller changes.♻️ Proposed optional injection
Add the field to the options type:
export type WaitpointSystemOptions = { resources: SystemResources; executionSnapshotSystem: ExecutionSnapshotSystem; enqueueSystem: EnqueueSystem; + coordinator?: WaitpointCoordinator; };Then default it in the constructor:
- this.coordinator = new LegacyPostgresWaitpointCoordinator({ - runStore: this.$.runStore, - prisma: this.$.prisma, - logger: this.$.logger, - }); + this.coordinator = + options.coordinator ?? + new LegacyPostgresWaitpointCoordinator({ + runStore: this.$.runStore, + prisma: this.$.prisma, + logger: this.$.logger, + });
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro Plus
Run ID: da984ec7-a423-4d2f-a6b0-9207cd712ff9
📒 Files selected for processing (3)
internal-packages/run-engine/src/engine/systems/waitpointSystem.tsinternal-packages/run-engine/src/engine/waitpointCoordinator/legacyPostgresCoordinator.tsinternal-packages/run-engine/src/engine/waitpointCoordinator/types.ts
Included review availability: Your plan provides up to 10 included reviews per hour; 9 remain after this review.
📜 Review details
⏰ Context from checks skipped due to timeout. (32)
- GitHub Check: webapp / 🧪 Unit Tests: Webapp (14, 24)
- GitHub Check: webapp / 🧪 Unit Tests: Webapp (17, 24)
- GitHub Check: webapp / 🧪 Unit Tests: Webapp (9, 24)
- GitHub Check: webapp / 🧪 Unit Tests: Webapp (24, 24)
- GitHub Check: webapp / 🧪 Unit Tests: Webapp (22, 24)
- GitHub Check: webapp / 🧪 Unit Tests: Webapp (21, 24)
- GitHub Check: webapp / 🧪 Unit Tests: Webapp (1, 24)
- GitHub Check: webapp / 🧪 Unit Tests: Webapp (15, 24)
- GitHub Check: webapp / 🧪 Unit Tests: Webapp (4, 24)
- GitHub Check: webapp / 🧪 Unit Tests: Webapp (8, 24)
- GitHub Check: webapp / 🧪 Unit Tests: Webapp (5, 24)
- GitHub Check: webapp / 🧪 Unit Tests: Webapp (18, 24)
- GitHub Check: webapp / 🧪 Unit Tests: Webapp (20, 24)
- GitHub Check: webapp / 🧪 Unit Tests: Webapp (19, 24)
- GitHub Check: webapp / 🧪 Unit Tests: Webapp (16, 24)
- GitHub Check: webapp / 🧪 Unit Tests: Webapp (2, 24)
- GitHub Check: webapp / 🧪 Unit Tests: Webapp (11, 24)
- GitHub Check: webapp / 🧪 Unit Tests: Webapp (23, 24)
- GitHub Check: webapp / 🧪 Unit Tests: Webapp (3, 24)
- GitHub Check: webapp / 🧪 Unit Tests: Webapp (13, 24)
- GitHub Check: webapp / 🧪 Unit Tests: Webapp (6, 24)
- GitHub Check: webapp / 🧪 Unit Tests: Webapp (7, 24)
- GitHub Check: webapp / 🧪 Unit Tests: Webapp (10, 24)
- GitHub Check: webapp / 🧪 Unit Tests: Webapp (12, 24)
- GitHub Check: e2e-webapp / 🧪 E2E Tests: Webapp (1, 2)
- GitHub Check: runops-guard / runops-guard
- GitHub Check: fk-cascade-guard / fk-cascade-guard
- GitHub Check: e2e-webapp / 🧪 E2E Tests: Webapp (2, 2)
- GitHub Check: internal / 🧪 Unit Tests: Internal
- GitHub Check: typecheck / typecheck
- GitHub Check: code-quality / code-quality
- GitHub Check: Analyze (javascript-typescript)
🧰 Additional context used
📓 Path-based instructions (5)
**/*.{ts,tsx}
📄 CodeRabbit inference engine (.github/copilot-instructions.md)
**/*.{ts,tsx}: Use types over interfaces for TypeScript
Avoid using enums; prefer string unions or const objects instead
**/*.{ts,tsx}: Prefer static imports over dynamic imports. Only use dynamicimport()when:
- Circular dependencies cannot be resolved otherwise
- Code splitting is genuinely needed for performance
- The module must be loaded conditionally at runtime
Files:
internal-packages/run-engine/src/engine/waitpointCoordinator/types.tsinternal-packages/run-engine/src/engine/systems/waitpointSystem.tsinternal-packages/run-engine/src/engine/waitpointCoordinator/legacyPostgresCoordinator.ts
**/*.{ts,tsx,js,jsx}
📄 CodeRabbit inference engine (.github/copilot-instructions.md)
Use function declarations instead of default exports
Files:
internal-packages/run-engine/src/engine/waitpointCoordinator/types.tsinternal-packages/run-engine/src/engine/systems/waitpointSystem.tsinternal-packages/run-engine/src/engine/waitpointCoordinator/legacyPostgresCoordinator.ts
**/*.ts
📄 CodeRabbit inference engine (.cursor/rules/otel-metrics.mdc)
**/*.ts: When creating or editing OTEL metrics (counters, histograms, gauges), ensure metric attributes have low cardinality by using only enums, booleans, bounded error codes, or bounded shard IDs
Do not use high-cardinality attributes in OTEL metrics such as UUIDs/IDs (envId, userId, runId, projectId, organizationId), unbounded integers (itemCount, batchSize, retryCount), timestamps (createdAt, startTime), or free-form strings (errorMessage, taskName, queueName)
When exporting OTEL metrics via OTLP to Prometheus, be aware that the exporter automatically adds unit suffixes to metric names (e.g., 'my_duration_ms' becomes 'my_duration_ms_milliseconds', 'my_counter' becomes 'my_counter_total'). Account for these transformations when writing Grafana dashboards or Prometheus queries
Files:
internal-packages/run-engine/src/engine/waitpointCoordinator/types.tsinternal-packages/run-engine/src/engine/systems/waitpointSystem.tsinternal-packages/run-engine/src/engine/waitpointCoordinator/legacyPostgresCoordinator.ts
**/*
📄 CodeRabbit inference engine (AGENTS.md)
**/*: Add crumbs as you write code — not just when debugging. Mark lines with
//@Crumbsor wrap blocks in `// `#region` `@crumbs. They stay on the
branch throughout development and are stripped byagentcrumbs strip
before merge.
Do not invent new namespaces — pick from this table or ask first.
Do not flag//@Crumbsmarkers, `#region `@crumbsblocks, or agentcrumbs
imports in reviews. These are temporary debug instrumentation stripped
before merge. Data logged in crumbs (IDs, names, values) never reaches
production.
Files:
internal-packages/run-engine/src/engine/waitpointCoordinator/types.tsinternal-packages/run-engine/src/engine/systems/waitpointSystem.tsinternal-packages/run-engine/src/engine/waitpointCoordinator/legacyPostgresCoordinator.ts
internal-packages/run-engine/src/engine/systems/**/*.ts
📄 CodeRabbit inference engine (internal-packages/run-engine/CLAUDE.md)
Integrate OpenTelemetry tracer and meter instrumentation in RunEngine systems for observability
Files:
internal-packages/run-engine/src/engine/systems/waitpointSystem.ts
🧠 Learnings (1)
📚 Learning: 2026-06-04T18:16:35.386Z
Learnt from: nicktrn
Repo: triggerdotdev/trigger.dev PR: 3836
File: apps/supervisor/src/backpressure/backpressureMonitor.ts:3-5
Timestamp: 2026-06-04T18:16:35.386Z
Learning: When reviewing TypeScript in this repo, apply the rule “prefer type aliases over interfaces” only to data/object shapes and union/intersection type modeling. If an interface is being used as a behavioral contract for collaborators to implement (e.g., method-shape interfaces that define required behavior, such as `BackpressureLogger` / `BackpressureSignalSource` in `apps/supervisor/src/backpressure/backpressureMonitor.ts`), keep it as an `interface` and do not flag it as a type-alias-vs-interface violation.
Applied to files:
internal-packages/run-engine/src/engine/waitpointCoordinator/types.ts
🔇 Additional comments (23)
internal-packages/run-engine/src/engine/waitpointCoordinator/types.ts (6)
16-32: LGTM!
34-56: LGTM!
58-78: LGTM!
80-99: LGTM!
101-133: LGTM!
135-145: 🗄️ Data Integrity & IntegrationRemove this concern
WaitpointId.generate()returns onlyidandfriendlyId, which matchAssociatedWaitpointData.> Likely an incorrect or invalid review comment.internal-packages/run-engine/src/engine/waitpointCoordinator/legacyPostgresCoordinator.ts (9)
1-44: LGTM!
46-66: LGTM!
68-83: LGTM!
85-109: LGTM!
111-183: LGTM!
185-264: LGTM!
266-377: LGTM!
413-437: LGTM!
397-411: 🗄️ Data Integrity & IntegrationNo routing change is required.
runStore.createWaitpointusesdata.completedByTaskRunIdwhen no explicit routing hint is provided, so this waitpoint co-locates with its owning run.> Likely an incorrect or invalid review comment.internal-packages/run-engine/src/engine/systems/waitpointSystem.ts (8)
1-13: LGTM!
65-67: LGTM!
83-95: LGTM!
152-172: LGTM!
200-228: LGTM!
285-296: LGTM!Also applies to: 395-405
472-472: LGTM!Also applies to: 703-706
728-728: LGTM!Also applies to: 812-814
…number The RunBlockEdge comment pointed at stale waitpointSystem.ts line numbers that no longer match the file after this branch shrank it. Name the continueRunIfUnblocked method instead so the citation can't drift again.
![devin-ai-integration[bot]](https://avatars.githubusercontent.com/in/811515?s=60&v=4){kind=small}
`BlockedRun` is only named inside types.ts, by CompleteResult. The repo's knip gate rejects unused exports, so drop the export keyword rather than add a knip.json exception — nothing outside this file needs the name yet.
3367c51 to
b2d56a9
Compare
|
Note GitHub couldn't provide a complete incremental comparison for this pull request, so CodeRabbit is performing a full review instead. This review may take a little longer. |
@trigger.dev/build
trigger.dev
@trigger.dev/core
@trigger.dev/python
@trigger.dev/react-hooks
@trigger.dev/redis-worker
@trigger.dev/rsc
@trigger.dev/schema-to-json
@trigger.dev/sdk
commit: |
1 participant
Extracts every Postgres waitpoint and edge operation out of
WaitpointSysteminto aWaitpointCoordinatorseam with one Postgres implementation, so a different coordination backend can be plugged in later without any caller changing.Pure refactor. Zero behaviour change, and zero test-file diffs — the existing engine corpus is the characterisation test.
What moved
WaitpointCoordinator(waitpointCoordinator/types.ts, declared withtype) has nine members:clearRunBlockState,readRunBlockState,registerBlocks,registerBlocksLockless,complete,createDateTimeWaitpoint,createManualWaitpoint,mintAssociatedWaitpointData,createAssociatedWaitpoint.LegacyPostgresWaitpointCoordinatorimplements them against the run-ops store. Its dependencies are{ runStore, prisma, logger }only, so it structurally cannot reach the run lock, the worker, or the event bus — orchestration stays inWaitpointSystem, which keeps all ten public signatures, all sixworker.enqueuesites, the racepoints, the snapshot transitions, and the event emissions.Two register methods rather than one with a flag, so "the batch path issues no extra query" is structural instead of conditional. Both share one private edge-write helper.
Six notes for reviewers — please read before "simplifying" any of these
nanoid(24)is called twice with different values on purpose, in each create path: once for the upsertwherekey, once forcreate.data. Hoisting either to a shared constant makes the where-key match the create-key, turning a guaranteed-miss upsert into a possible update. IncreateManualWaitpointboth calls plusWaitpointId.generate()stay inside the retry loop so each attempt tries a fresh key.The two enqueue conditions are deliberately asymmetric. DATETIME enqueues
finishWaitpointunconditionally after a non-cached create, withavailableAt: completedAfter. MANUAL enqueues only whentimeoutis set. That is existing behaviour, not an oversight. The coordinator returns a discriminated union onkindrather than a boolean so the enqueue is structurally unreachable on the cached path.One false clause was deleted from a moved comment. The old comment on the full-clear delete claimed the caller's
txis not forwarded. The code does forward it, andPostgresRunStoreusestx ?? this.prisma, so a single store joins the caller's transaction — only the routing store strips it. The rest of that comment is unchanged.The MANUAL timeout enqueue now sits outside the P2002 retry loop. Safe because the worker is Redis-backed and cannot raise
Prisma.PrismaClientKnownRequestError, so the loop never retried on it. If a Postgres-backed enqueue is ever swapped in, that equivalence breaks silently.The coordinator caches
runStore/prisma/loggerat construction, where the old code readthis.$.*per call. Equivalent only because nothing reassigns them: one assignment atengine/index.ts, and theresourcesobject is aconstthat is never mutated.Two comments in other files are now stale and were left alone —
engine/index.tsandcompleteWaitpointCrossSeamGuard.test.tsboth describe routing as the first statement ofwaitpointSystem.completeWaitpoint. Both tests still pass, because that guard sits inindex.tsbefore the delegation. Left untouched to keep this diff to three files.Preserved verbatim
The
unnestedge CTE rather than aWaitpointjoin; the pending count as a separate statement after the edge write (READ COMMITTED needs its own snapshot); completion'sfindWaitpointOnPrimaryre-read through the resolved handle while the blocked-run fan-out goes back through the router; the residency and colocate hints, with colocation objects built only in the Postgres arm and the count keeping itsrunIdargument;ON CONFLICT DO NOTHINGand the(taskRunId, waitpointId, batchIndex)multi-index edge semantics; the unreadbatchIdselect, which rides inside twologger.debugpayloads.internal-packages/run-store/is untouched, so the CTE and the conflict semantics never moved.Verification
run-enginetypechecktsc --noEmit -p tsconfig.build.jsonexits 0webapptypecheckThe webapp typecheck does not pass. The failures are pre-existing (
PrismaPgnot assignable tonever; missing@trigger.dev/rbacexports) and the sorted error lists are byte-identical to the merge base, so this branch adds none — but the criterion is genuinely unmet and needs a separate fix.No changeset and no
.server-changesnote: internal refactor with no user-visible change.Follow-ups this surfaced
buildRunAssociatedWaitpointnow mints through the coordinator, but the row is inserted nested insidecreateRun/createFailedRun. That needs its own packet before a second backend lands, or the commonest waitpoint gets split across two of them.clearRunBlockStateoverloads opposite outcomes onundefinedversus[]:undefinedclears every edge,[]clears none. Both callers are correct today; worth splitting when the file is next touched..sqlentry ininternal-packages/clickhouse/schema/breaks everycontainerTestin the repo, because the testcontainers migration readerreadFiles everyreaddirentry without filtering despite a comment claiming it filters. Hit this during setup; unrelated to this change and left for a separate fix.