feat(sdk,core): offload large trigger payloads via object storage - #4470
feat(sdk,core): offload large trigger payloads via object storage#4470deepshekhardas wants to merge 20 commits into
Conversation
…t build server failures (triggerdotdev#2913)
- Include reproduction scripts for Sentry (triggerdotdev#2900) and engine strictness (triggerdotdev#2913) - Include PR body drafts for consolidated tracking
- Include reproduction scripts for Sentry (triggerdotdev#2900) and engine strictness (triggerdotdev#2913) - Include PR body drafts for consolidated tracking
When the underlying logical-replication client errored (e.g. after a Postgres failover), the runs and sessions replication services logged the error and left the stream stopped. The host process kept running, the WAL backed up, and ClickHouse silently fell behind. Both services now run a configurable recovery strategy on stream errors, defaulting to in-process reconnect with exponential backoff so a fresh self-hosted setup heals on its own: - "reconnect" (default) re-subscribes via the existing subscribe(lastLsn) path with exponential backoff (1s -> 60s cap, unlimited attempts), which re-validates the publication, re-acquires the leader lock, and resumes from the last acknowledged LSN. - "exit" calls process.exit after a short flush window so a host's supervisor (Docker restart=always, systemd, k8s, etc.) can replace the process. - "log" preserves the historical behaviour. Per-service strategy + exit knobs are env-driven via RUN_REPLICATION_ERROR_STRATEGY / SESSION_REPLICATION_ERROR_STRATEGY plus matching *_EXIT_DELAY_MS / *_EXIT_CODE. Reconnect tuning is shared across both services via REPLICATION_RECONNECT_INITIAL_DELAY_MS / _MAX_DELAY_MS / _MAX_ATTEMPTS (0 = unlimited).
Addresses PR review feedback:
- LogicalReplicationClient.subscribe() can throw before its internal
"error" listener is wired up (notably when pg client.connect() fails
mid-failover). The reconnect strategy's catch block only logged, so
recovery silently stopped. Now also calls scheduleReconnect(err) — the
pendingReconnect guard makes it idempotent if an error event was also
emitted.
- Reject negative values for the new replication-recovery env vars and
cap exit codes at 255.
- Convert the new ReplicationErrorRecovery{Deps,} interfaces to type
aliases to match the repo's TypeScript style.
- Tighten the reconnect dep comment to drop a stale "lastAcknowledgedLsn"
reference (the wrapper-tracked resume LSN is what callers actually pass).
- Restore process.exit after service.shutdown() in the exit-strategy
test so a delayed exit timer can't terminate the test worker.
LogicalReplicationClient.subscribe() can resolve without throwing or emitting an "error" event when leader-lock acquisition fails — it just calls this.stop() and returns. The reconnect callback now checks isStopped after subscribe() and throws so the recovery handler can schedule the next attempt instead of silently giving up.
…rough handle() The previous post-subscribe() isStopped check was always true on the happy path: subscribe() calls stop() up front (setting _isStopped=true) and only resets the flag inside the replicationStart event, which fires asynchronously after subscribe() returns. So the check threw on every successful reconnect, the catch rescheduled, the next attempt tore down the just-built client, and the cycle continued — replication briefly worked between teardowns, which is why the integration test passed. Replace it with the correct nudge: subscribe to leaderElection and call the recovery handler on isLeader=false. That's the only subscribe() exit path that doesn't either throw or emit an "error" event (the other silent-return paths emit "error" first via createPublication/createSlot failures).
The previous commit routed leaderElection(false) through handle(), which under the exit strategy schedules process.exit. In a multi-instance deployment that turns lost leader election — a normal operational state — into a restart loop: exit, supervisor restarts, election fails again, exit, and so on. Add a dedicated notifyLeaderElectionLost() on ReplicationErrorRecovery that the reconnect strategy treats as another retry trigger, while exit and log strategies no-op. Wire the wrapper services through the new method.
fix(webapp): auto-recover replication services after stream errors
…riggerdotdev#3785) - Add ioSerialization utilities for payload serialization - Add api-type.test.ts for new API types - Update trigger payload schema with IOPacket support - Update shared.ts to import IOPacket type - Add changeset for @trigger.dev/core and @trigger.dev/sdk Closes triggerdotdev#3785
🦋 Changeset detectedLatest commit: d95d2bc The changes in this PR will be included in the next version bump. Not sure what this means? Click here to learn what changesets are. Click here if you're a maintainer who wants to add another changeset to this PR |
|
Warning Review limit reached
Next review available in: 13 minutes Enable usage-based reviews in Billing to review now. Otherwise, wait until the next included review is available. How can I continue?After more reviews become available, a review can be triggered using the To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based reviews. How do review limits work?CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability. For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window. Please refer docs for additional details. Review details⚙️ Run configurationConfiguration used: Repository UI Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (34)
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 |
|
Hi @deepshekhardas, thanks for your interest in contributing! This project requires that pull request authors are vouched, and you are not in the list of vouched users. This PR will be closed automatically. See https://github.com/triggerdotdev/trigger.dev/blob/main/CONTRIBUTING.md for more details. |
![devin-ai-integration[bot]](https://avatars.githubusercontent.com/in/811515?s=60&v=4){kind=small}
| } finally { | ||
| await watcher?.stop(); | ||
| process.off("SIGINT", signalHandler); | ||
| process.off("SIGTERM", signalHandler); | ||
| await cleanup(); |
There was a problem hiding this comment.
🔴 Local development mode shuts itself down immediately after starting
The local dev session is torn down (await cleanup() at packages/cli-v3/src/commands/dev.ts:217) as soon as startup finishes, because the call that is supposed to keep the command alive returns instantly, so tasks stop running right after trigger dev starts.
Impact: Developers running the dev command lose their local task worker immediately and cannot execute tasks.
Why the wait never blocks and what the new cleanup tears down
startDev returns waitUntilExit = async () => { } (packages/cli-v3/src/commands/dev.ts:312), a no-op. So await devInstance.waitUntilExit() resolves immediately and control falls into the new finally block, which now calls cleanup() → devInstance.stop(). That stop() (packages/cli-v3/src/commands/dev.ts:316-320) calls the dev-session stop() which removes the build destination, stops bundling, and calls runtime.shutdown() (packages/cli-v3/src/dev/devSession.ts:247-255), plus stops the config watcher and deletes the lockfile.
Before this PR the finally block only stopped the config watcher (await watcher?.stop()), leaving the dev session and its worker runtime alive, so the command kept working. Cleanup should only run on actual exit/signal, or waitUntilExit must resolve only when the session ends.
Prompt for agents
In packages/cli-v3/src/commands/dev.ts, devCommand now awaits devInstance.waitUntilExit() and then unconditionally runs cleanup() in a finally block. However startDev returns waitUntilExit as an empty async function (a no-op), so the await resolves immediately and cleanup() tears down the dev session (devSession stop -> runtime.shutdown, bundling stop, watcher stop, lockfile removal) right after boot. Previously the finally block only stopped the config watcher, so the session survived. Either make waitUntilExit return a promise that only resolves when the dev session actually ends (e.g. resolved by the signal handler or session exit), or don't run the teardown from the finally path until the session has genuinely finished.
Was this helpful? React with 👍 or 👎 to provide feedback.
| const signalHandler = async (signal: string) => { | ||
| logger.debug(`Received ${signal}, cleaning up...`); | ||
| await cleanup(); | ||
| process.exit(0); | ||
| }; | ||
|
|
||
| try { | ||
| const devInstance = await startDev({ ...options, cwd: process.cwd(), login: authorization }); | ||
| watcher = devInstance.watcher; | ||
| process.on("SIGINT", signalHandler); | ||
| process.on("SIGTERM", signalHandler); |
There was a problem hiding this comment.
🟡 Ctrl+C cleanup handler never runs because an earlier handler exits the process first
The new interrupt handler that is supposed to clean up child processes (process.on("SIGINT", signalHandler) at packages/cli-v3/src/commands/dev.ts:209) is registered after a global handler that exits the process synchronously, so the cleanup never executes and worker processes can still be left behind.
Impact: Pressing Ctrl+C can still leave orphaned worker processes running, which is the exact problem the change intends to fix.
Handler ordering
installExitHandler() runs at CLI startup (packages/cli-v3/src/cli/index.ts:46) and registers process.on("SIGINT", () => process.exit(0)) (packages/cli-v3/src/cli/common.ts:88-95). Node invokes signal listeners in registration order; the first listener calls process.exit(0) synchronously, terminating the process before the later async signalHandler in devCommand is invoked (and before any awaited cleanup could complete anyway).
A fix requires either removing/avoiding the global exit handler for the dev command, or performing cleanup from within that handler.
Was this helpful? React with 👍 or 👎 to provide feedback.
| export const UpdateCommandOptions = CommonCommandOptions.pick({ | ||
| logLevel: true, | ||
| skipTelemetry: true, | ||
| ignoreEngines: true, |
There was a problem hiding this comment.
🔴 Option to ignore Node engine checks during install has no effect
The new setting that is supposed to relax Node version checks during dependency installation (ignoreEngines: true at packages/cli-v3/src/commands/update.ts:21) is never read when packages are installed, so deployments still fail when the project declares a stricter Node version.
Impact: Deployments on build servers with a mismatched Node version keep failing, and the newly added tests asserting the flags are passed will fail.
Missing plumbing into installDependencies
UpdateCommandOptions now picks ignoreEngines (packages/cli-v3/src/commands/update.ts:18-21) and deployCommand passes { ...options, ignoreEngines: true } (packages/cli-v3/src/commands/deploy.ts:262), but updateTriggerPackages still calls await installDependencies({ cwd: projectPath, silent: true }) (packages/cli-v3/src/commands/update.ts:261) with no args. The new test file packages/cli-v3/src/commands/update.test.ts:74-112 asserts args: ["--no-engine-strict"] (npm), ["--config.engine-strict=false"] (pnpm), ["--ignore-engines"] (yarn) and [] otherwise — none of which the implementation produces.
Prompt for agents
packages/cli-v3/src/commands/update.ts accepts a new ignoreEngines option (picked into UpdateCommandOptions, set to true by deployCommand) but never uses it. The call to installDependencies({ cwd: projectPath, silent: true }) needs to pass package-manager-specific args derived from the detected package manager when options.ignoreEngines is true: npm -> --no-engine-strict, pnpm -> --config.engine-strict=false, yarn -> --ignore-engines, otherwise an empty array. See the expectations in packages/cli-v3/src/commands/update.test.ts.
Was this helpful? React with 👍 or 👎 to provide feedback.
| async function prepareTriggerPayload( | ||
| payload: unknown, | ||
| apiClient: ApiClient, | ||
| taskId: string | ||
| ): Promise<IOPacket> { | ||
| const payloadPacket = await stringifyIO(payload); | ||
| return await conditionallyExportPacket( | ||
| payloadPacket, | ||
| createTriggerPayloadPathPrefix(taskId), | ||
| undefined, | ||
| apiClient | ||
| ); | ||
| } |
There was a problem hiding this comment.
🟡 Large trigger payloads now fail outright when the upload step is unavailable or unauthorized
Oversized payloads are uploaded to storage before triggering (conditionallyExportPacket(...) at packages/trigger-sdk/src/v3/shared.ts:3077-3082) with no fallback to sending them inline, so a trigger that used to succeed now throws whenever the upload step is rejected.
Impact: Triggering a task with a large payload fails against older servers, or when using a public/JWT access token, even though the trigger request itself would have been accepted.
Two concrete rejection paths
exportPacket calls client.createUploadPayloadUrl(filename) which hits PUT /api/v2/packets/... (packages/core/src/v3/apiClient/index.ts:579-590) and throws if storagePath is missing (packages/core/src/v3/utils/ioSerialization.ts:172-176).
- Older self-hosted servers have no
/api/v2/packetsroute, so the presign 404s and the trigger throws — previously the server offloaded the payload itself inDefaultPayloadProcessor(apps/webapp/app/runEngine/concerns/payloads.server.ts:20-45). - The v2 packets route calls
authenticateApiRequest(request)with no options (apps/webapp/app/routes/api.v2.packets.$.ts:21), which rejects public keys and public JWTs (apps/webapp/app/services/apiAuth.server.ts:119-125), while the trigger route itself setsallowJWT: true. So SDK triggers authenticated with a public token now 401 on payloads ≥128KB.
Consider catching upload failures and falling back to sending the packet inline (the server still offloads it), or restricting offload to clients using secret keys.
Was this helpful? React with 👍 or 👎 to provide feedback.
| this._isShuttingDown = true; | ||
| this._errorRecovery.dispose(); |
There was a problem hiding this comment.
🟡 Stopped replication service can resurrect itself through a pending reconnect timer
A scheduled retry of the database replication stream is only cancelled during full shutdown (this._errorRecovery.dispose() at apps/webapp/app/services/runsReplicationService.server.ts:311), so stopping the service without shutting it down lets a queued retry re-open the stream afterwards.
Impact: A replication service that was explicitly stopped can silently reconnect and keep consuming the replication slot.
Paths that don't dispose
stop() (apps/webapp/app/services/runsReplicationService.server.ts:336-344) and teardown() (:346-354) call this._replicationClient.stop()/teardown() and clear the ack interval, but never call this._errorRecovery.dispose(), and they don't set _isShuttingDown, so the recovery helper's isShuttingDown() guard (apps/webapp/app/services/replicationErrorRecovery.server.ts:92) is false when the pending timer fires and reconnect() calls subscribe() again. The same applies to SessionsReplicationService.stop()/teardown() (apps/webapp/app/services/sessionsReplicationService.server.ts:320-344).
Prompt for agents
RunsReplicationService.stop()/teardown() and SessionsReplicationService.stop()/teardown() stop the underlying LogicalReplicationClient but do not dispose the new error-recovery helper, and they do not set the shutting-down flag. A reconnect timer scheduled before the stop will therefore fire and call subscribe() again, resurrecting a service that was intentionally stopped. Call _errorRecovery.dispose() (or otherwise mark the service as not-running so isShuttingDown() returns true) from both stop() and teardown() in both services.
Was this helpful? React with 👍 or 👎 to provide feedback.
| // Mock dependencies | ||
| vi.mock("nypm"); | ||
| vi.mock("pkg-types"); | ||
| vi.mock("node:fs/promises"); | ||
| vi.mock("@clack/prompts"); | ||
| vi.mock("std-env", () => ({ | ||
| hasTTY: true, | ||
| isCI: false, | ||
| })); |
There was a problem hiding this comment.
🟡 New tests rely on mocks instead of the required real-dependency test approach
The added test suites replace real modules with fakes (vi.mock(...) at packages/cli-v3/src/commands/update.test.ts:11-51), which the repository's testing rules explicitly forbid.
Impact: These tests violate the mandated testing conventions and validate mock wiring rather than real behaviour.
Rule and locations
CLAUDE.md: "We use vitest exclusively. Never mock anything - use testcontainers instead." AGENTS.md: "Tests should avoid mocks or stubs and use the helpers from @internal/testcontainers when Redis or Postgres are needed."
Violations: packages/cli-v3/src/commands/update.test.ts:11-51 mocks nypm, pkg-types, node:fs/promises, @clack/prompts, std-env, and several internal modules; packages/cli-v3/src/utilities/sourceMaps.test.ts:6-10 mocks source-map-support and stubs process.setSourceMapsEnabled.
Was this helpful? React with 👍 or 👎 to provide feedback.
| # Consolidated Bug Fixes | ||
|
|
||
| This PR combines fixes for several independent issues identified in the codebase, covering CLI stability, deployment/build reliability, and runtime correctness. | ||
|
|
||
| ## Fixes | ||
|
|
||
| | Issue / Feature | Description | | ||
| |-----------------|-------------| | ||
| | **Orphaned Workers** | Fixes `trigger dev` leaving orphaned `trigger-dev-run-worker` processes by ensuring graceful shutdown on `SIGINT`/`SIGTERM` and robust process cleanup. | | ||
| | **Sentry Interception** | Fixes `ConsoleInterceptor` swallowing logs when Sentry (or other monkey-patchers) are present by delegating to the original preserved console methods. | | ||
| | **Engine Strictness** | Fixes deployment failures on GitHub Integration when `engines.node` is strict (e.g. "22") by passing `--no-engine-strict` (and equivalents) during the `trigger deploy` build phase. | |
There was a problem hiding this comment.
🟡 Change bundles many unrelated fixes and ships a stray planning document
A summary document listing nine independent fixes is committed to the repository root (consolidated_pr_body.md:1-11), confirming that this change bundles multiple unrelated issues, which the contribution rules disallow.
Impact: The change is hard to review and revert, and an internal planning file is published in the repository.
Rule
CONTRIBUTING.md: "Important: We only accept PRs that address a single issue. Please do not submit PRs containing multiple unrelated fixes or features. If you have multiple contributions, open a separate PR for each one."
The diff mixes SDK payload offloading, webapp replication error recovery, CLI dev/deploy/update changes, Docker Hub login, source-map handling and console interception, alongside consolidated_pr_body.md, which should not be committed at all.
Was this helpful? React with 👍 or 👎 to provide feedback.
| function createTriggerPayloadPathPrefix(taskId: string): string { | ||
| const safeTaskId = encodeURIComponent(taskId); | ||
| return `trigger/${safeTaskId}/${Date.now()}-${Math.random().toString(36).slice(2)}/payload`; | ||
| } |
There was a problem hiding this comment.
🔍 Payload is uploaded before idempotency/dedupe is evaluated, leaving orphaned objects
prepareTriggerPayload runs before the trigger request, so a large payload is always uploaded even when the trigger is deduplicated by an idempotency key, debounced, or when the request subsequently fails. The generated key (trigger/<taskId>/<timestamp>-<random>/payload.<ext>) is not tied to a run id, so nothing later can associate or clean up these objects (server-side offloads use ${friendlyId}/payload.json, see apps/webapp/app/runEngine/concerns/payloads.server.ts:32). Expect slow growth of unreferenced objects in the packets bucket; a retention policy or run-scoped key may be needed.
Was this helpful? React with 👍 or 👎 to provide feedback.
| this.logger.info("Leader election", { isLeader }); | ||
| if (!isLeader) { | ||
| // Failed leader election doesn't throw or emit an "error" event — | ||
| // subscribe() just emits leaderElection(false), calls stop(), and | ||
| // returns. Route through a dedicated handler so only the reconnect | ||
| // strategy acts; the exit strategy must not restart-loop when | ||
| // another instance holds the lock. | ||
| this._errorRecovery.notifyLeaderElectionLost( | ||
| new Error("Failed to acquire replication leader lock") | ||
| ); | ||
| } |
There was a problem hiding this comment.
🔍 Losing the leader lock triggers an endless reconnect loop by design — verify the log volume
leaderElection(false) now routes into scheduleReconnect, which retries forever by default (REPLICATION_RECONNECT_MAX_ATTEMPTS default 0 = unlimited) and logs each attempt at error level (apps/webapp/app/services/replicationErrorRecovery.server.ts:84-88). In a normal multi-instance deployment every non-leader instance will emit an error log every up-to-60s indefinitely, which is a meaningful change from the previous silent behaviour. Consider logging leader-lock contention at a lower level than error.
Was this helpful? React with 👍 or 👎 to provide feedback.
| this._errorRecovery = createReplicationErrorRecovery({ | ||
| strategy: options.errorRecovery ?? { type: "reconnect" }, | ||
| logger: this.logger, | ||
| reconnect: async () => { | ||
| await this._replicationClient.subscribe(this._latestCommitEndLsn ?? undefined); | ||
| }, | ||
| isShuttingDown: () => this._isShuttingDown || this._isShutDownComplete, | ||
| }); |
There was a problem hiding this comment.
🔍 Reconnect does not restart the acknowledge interval or reset transaction state
The reconnect callback only calls this._replicationClient.subscribe(this._latestCommitEndLsn ?? undefined), unlike start() which also creates the acknowledge interval and starts the flush scheduler. That is fine when recovery follows a mid-life stream error (both are already running), but if the very first start() fails at leader election, the recovery path resubscribes without those having been (re)initialised in the failure case, and any partially accumulated _currentTransaction from the dropped stream is not cleared before the new stream replays from the last commit LSN — worth confirming the client emits a fresh begin before any further events.
Was this helpful? React with 👍 or 👎 to provide feedback.
2 participants
Offloads large trigger payloads via object storage. See original PR #3785.