From ad005a46df759a6be4e4bf8812427b41b8165f82 Mon Sep 17 00:00:00 2001 From: Chris Arderne Date: Mon, 28 Sep 2026 09:20:17 +0000 Subject: [PATCH 1/9] fix(ci): switch MinIO tests to Chainguard images (#4985) ## Summary Switches the MinIO test and local development containers to Chainguard's maintained Docker Hub images because the previous Quay images are no longer publicly pullable. ## Fix Pins the multi-platform server and client images by digest. The local stack now uses the Chainguard client entrypoint for bucket initialization, while test containers keep client state in a writable temporary directory. --- .github/workflows/e2e-webapp.yml | 2 +- .github/workflows/unit-tests-webapp.yml | 2 +- docker/docker-compose.yml | 16 ++++--- internal-packages/testcontainers/src/minio.ts | 42 +++++++++++++++---- 4 files changed, 42 insertions(+), 20 deletions(-) diff --git a/.github/workflows/e2e-webapp.yml b/.github/workflows/e2e-webapp.yml index 48cc1d6df40..f9b6b79f311 100644 --- a/.github/workflows/e2e-webapp.yml +++ b/.github/workflows/e2e-webapp.yml @@ -102,7 +102,7 @@ jobs: redis:7.2 \ testcontainers/ryuk:0.14.0 \ ghcr.io/s2-streamstore/s2:0.40.0@sha256:b26249e2ede0949755f5af8028185dc2bcfc3aa2db21eb9610543d144eb6ee9d \ - quay.io/minio/minio:RELEASE.2025-09-07T16-13-09Z@sha256:14cea493d9a34af32f524e538b8346cf79f3321eff8e708c1e2960462bd8936e + chainguard/minio:latest@sha256:6a1d0b45c8669726bba580ced0bfa4cb9fdeed1ed636dfabd81d1577beb6937b do pull "$image" & pids+=("$!") diff --git a/.github/workflows/unit-tests-webapp.yml b/.github/workflows/unit-tests-webapp.yml index f15f9991a49..45d15be4015 100644 --- a/.github/workflows/unit-tests-webapp.yml +++ b/.github/workflows/unit-tests-webapp.yml @@ -109,7 +109,7 @@ jobs: redis:7.2 \ testcontainers/ryuk:0.14.0 \ electricsql/electric:1.2.4@sha256:20da3d0b0e74926c5623392db67fd56698b9e374c4aeb6cb5cadeb8fea171c36 \ - quay.io/minio/minio:RELEASE.2025-09-07T16-13-09Z@sha256:14cea493d9a34af32f524e538b8346cf79f3321eff8e708c1e2960462bd8936e + chainguard/minio:latest@sha256:6a1d0b45c8669726bba580ced0bfa4cb9fdeed1ed636dfabd81d1577beb6937b do pull "$image" & pids+=("$!") diff --git a/docker/docker-compose.yml b/docker/docker-compose.yml index 9ca98e1d405..ac3cf2f9527 100644 --- a/docker/docker-compose.yml +++ b/docker/docker-compose.yml @@ -144,12 +144,13 @@ services: # routes to it when the OBJECT_STORE_* env vars are set (see .env.example). minio: container_name: ${CONTAINER_PREFIX:-}minio - image: quay.io/minio/minio:RELEASE.2025-09-07T16-13-09Z@sha256:14cea493d9a34af32f524e538b8346cf79f3321eff8e708c1e2960462bd8936e + image: chainguard/minio:latest@sha256:6a1d0b45c8669726bba580ced0bfa4cb9fdeed1ed636dfabd81d1577beb6937b restart: always command: server /data --console-address ":9001" environment: MINIO_ROOT_USER: minioadmin MINIO_ROOT_PASSWORD: minioadmin + MC_HOST_local: http://minioadmin:minioadmin@localhost:9000 volumes: - minio-data:/data ports: @@ -158,25 +159,22 @@ services: networks: - app_network healthcheck: - test: ["CMD", "curl", "-f", "http://localhost:9000/minio/health/live"] + test: ["CMD", "mc", "ready", "local"] interval: 5s timeout: 5s retries: 5 start_period: 5s minio-init: - image: quay.io/minio/mc:RELEASE.2025-08-13T08-35-41Z@sha256:a7fe349ef4bd8521fb8497f55c6042871b2ae640607cf99d9bede5e9bdf11727 + image: chainguard/minio-client:latest@sha256:b2bd7824d23d3e3b15bedd7e87fbc3be29d2e213307b4f901e4a1d92356dc20f depends_on: minio: condition: service_healthy networks: - app_network - entrypoint: /bin/sh - command: - - -c - - | - mc alias set local http://minio:9000 minioadmin minioadmin - mc mb -p local/packets || true + environment: + MC_HOST_local: http://minioadmin:minioadmin@minio:9000 + command: ["mb", "--ignore-existing", "local/packets"] restart: "no" electric: diff --git a/internal-packages/testcontainers/src/minio.ts b/internal-packages/testcontainers/src/minio.ts index 6d158d99de2..e96391c7713 100644 --- a/internal-packages/testcontainers/src/minio.ts +++ b/internal-packages/testcontainers/src/minio.ts @@ -17,7 +17,7 @@ export class MinIOContainer extends GenericContainer { private region = "us-east-1"; constructor( - image = "quay.io/minio/minio:RELEASE.2025-09-07T16-13-09Z@sha256:14cea493d9a34af32f524e538b8346cf79f3321eff8e708c1e2960462bd8936e" + image = "chainguard/minio:latest@sha256:6a1d0b45c8669726bba580ced0bfa4cb9fdeed1ed636dfabd81d1577beb6937b" ) { super(image); this.withExposedPorts(MINIO_PORT); @@ -49,13 +49,14 @@ export class MinIOContainer extends GenericContainer { const startedContainer = await super.start(); - // Create the "packets" bucket using MinIO client await x( "docker", [ "exec", startedContainer.getId(), "mc", + "--config-dir", + "/tmp/.mc", "alias", "set", "local", @@ -66,9 +67,11 @@ export class MinIOContainer extends GenericContainer { { throwOnError: true } ); - await x("docker", ["exec", startedContainer.getId(), "mc", "mb", "local/packets"], { - throwOnError: true, - }); + await x( + "docker", + ["exec", startedContainer.getId(), "mc", "--config-dir", "/tmp/.mc", "mb", "local/packets"], + { throwOnError: true } + ); return new StartedMinIOContainer( startedContainer, @@ -123,14 +126,35 @@ export class StartedMinIOContainer extends AbstractStartedContainer { public async resetBucket(bucket = "packets"): Promise { await x( "docker", - ["exec", this.getId(), "mc", "rm", "--recursive", "--force", `local/${bucket}`], + [ + "exec", + this.getId(), + "mc", + "--config-dir", + "/tmp/.mc", + "rm", + "--recursive", + "--force", + `local/${bucket}`, + ], { throwOnError: false, } ); - await x("docker", ["exec", this.getId(), "mc", "mb", "--ignore-existing", `local/${bucket}`], { - throwOnError: true, - }); + await x( + "docker", + [ + "exec", + this.getId(), + "mc", + "--config-dir", + "/tmp/.mc", + "mb", + "--ignore-existing", + `local/${bucket}`, + ], + { throwOnError: true } + ); } /** From 8f15fa5af2f95a998587e409f9b8ae1da2aeed8e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marcus=20Nerl=C3=B8e?= <32075361+NERLOE@users.noreply.github.com> Date: Mon, 28 Sep 2026 09:42:46 +0000 Subject: [PATCH 2/9] fix(core): accept IPC acks for void callbacks under zod 4.4+ (#4980) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closes #4979 Credit to @marcus-dk for finding and root-causing this in #4979, including the zod version bisect and the end-to-end confirmation with a patched worker. ## Summary `FLUSH` and `CANCEL` declare `callback: z.void()`, so the task run process acks them with `{ type: "ACK", id, message: undefined }`. Node's JSON IPC serialization drops the undefined `message` key. Under zod 4.4+ a `z.any()` object key is required at parse time (it was optional under zod 3 and 4.0–4.3), so `Packet.safeParse` rejects every such ack, `#handlePacket` returns silently, and `sendWithAck` waits out its full timeout. Because `zodIpc.ts` imports `zod/v4` and `@trigger.dev/core` resolves `zod` from the user's project, every project on zod ≥ 4.4 pays this on every run: ~6 s between the task finishing and the completion reaching the engine (FLUSH: 5 s + 1 s), and ~31 s on cancel. The fix makes the ACK packet's `message` optional again (`z.any().optional()`). Only the ACK packet can legitimately carry `undefined`, so the other packets are unchanged. ## ✅ Checklist - [x] I have followed every step in the [contributing guide](https://github.com/triggerdotdev/trigger.dev/blob/main/CONTRIBUTING.md) - [x] The PR title follows the convention. - [x] I ran and tested the code works --- ## Testing - Added `packages/core/test/zodIpc.test.ts`: forks a real child process (`test/fixtures/zodIpcChild.ts`) and sends `FLUSH` (void callback) and `PING` (payload callback) over Node's IPC channel, so the ack goes through the same JSON serialization as between a worker and its task run process. On `main` (zod 4.5.4 in this repo) the void-callback test fails with `sendWithAck() timeout`; with the fix it resolves immediately. - `pnpm run build --filter @trigger.dev/core` passes. - Production evidence from our self-hosted instance (`completed_at - started_at - usage_duration_ms` on successful runs): | SDK | runs | p10 gap | p50 gap | |---|---|---|---| | 4.5.12 | 2.6M | 179 ms | 244 ms | | 4.5.15 | 5.0M | 176 ms | 251 ms | | 4.6.0 | 1.6M | 6,162 ms | 6,196 ms | | 4.6.2 | 4.0M | 6,180 ms | 6,240 ms | --- ## Changelog Fix a ~6 second delay between a task finishing and its run completing (and a ~31 second delay when cancelling a run) in projects that use zod 4.4 or newer. 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 (1M context) Co-authored-by: Matt Aitken --- .changeset/void-ipc-acks.md | 5 +++ packages/core/src/v3/zodIpc.ts | 4 +- packages/core/test/fixtures/zodIpcCatalog.ts | 14 +++++++ packages/core/test/fixtures/zodIpcChild.ts | 13 ++++++ packages/core/test/zodIpc.test.ts | 44 ++++++++++++++++++++ 5 files changed, 79 insertions(+), 1 deletion(-) create mode 100644 .changeset/void-ipc-acks.md create mode 100644 packages/core/test/fixtures/zodIpcCatalog.ts create mode 100644 packages/core/test/fixtures/zodIpcChild.ts create mode 100644 packages/core/test/zodIpc.test.ts diff --git a/.changeset/void-ipc-acks.md b/.changeset/void-ipc-acks.md new file mode 100644 index 00000000000..85caba20d35 --- /dev/null +++ b/.changeset/void-ipc-acks.md @@ -0,0 +1,5 @@ +--- +"@trigger.dev/core": patch +--- + +Fix a ~6-second delay between a task finishing and its run completing (and a ~31-second delay when cancelling a run) in projects that use zod 4.4 or newer. Run cost and billed usage was not impacted by this issue. diff --git a/packages/core/src/v3/zodIpc.ts b/packages/core/src/v3/zodIpc.ts index ebd5a685df7..e2b5028b376 100644 --- a/packages/core/src/v3/zodIpc.ts +++ b/packages/core/src/v3/zodIpc.ts @@ -124,7 +124,9 @@ const Packet = discriminatedUnion("type", [ }), z.object({ type: z.literal("ACK"), - message: z.any(), + // Optional: a void callback acks with `message: undefined`, and `process.send` + // drops undefined keys when serializing, so the key is absent on arrival. + message: z.any().optional(), id: z.number(), }), z.object({ diff --git a/packages/core/test/fixtures/zodIpcCatalog.ts b/packages/core/test/fixtures/zodIpcCatalog.ts new file mode 100644 index 00000000000..6f24288b049 --- /dev/null +++ b/packages/core/test/fixtures/zodIpcCatalog.ts @@ -0,0 +1,14 @@ +import { z } from "zod/v4"; + +export const ParentToChild = { + FLUSH: { + message: z.object({ timeoutInMs: z.number() }), + callback: z.void(), + }, + PING: { + message: z.object({ value: z.string() }), + callback: z.object({ echoed: z.string() }), + }, +}; + +export const ChildToParent = {}; diff --git a/packages/core/test/fixtures/zodIpcChild.ts b/packages/core/test/fixtures/zodIpcChild.ts new file mode 100644 index 00000000000..721b534ddb9 --- /dev/null +++ b/packages/core/test/fixtures/zodIpcChild.ts @@ -0,0 +1,13 @@ +import { ZodIpcConnection } from "../../src/v3/zodIpc.js"; +import { ChildToParent, ParentToChild } from "./zodIpcCatalog.js"; + +// Forked by zodIpc.test.ts: answers the parent's messages over the real IPC channel. +new ZodIpcConnection({ + listenSchema: ParentToChild, + emitSchema: ChildToParent, + process, + handlers: { + FLUSH: async () => {}, + PING: async ({ value }) => ({ echoed: value }), + }, +}); diff --git a/packages/core/test/zodIpc.test.ts b/packages/core/test/zodIpc.test.ts new file mode 100644 index 00000000000..7b5d2669723 --- /dev/null +++ b/packages/core/test/zodIpc.test.ts @@ -0,0 +1,44 @@ +import { fork, type ChildProcess } from "node:child_process"; +import { fileURLToPath } from "node:url"; +import { afterEach, describe, expect, it } from "vitest"; +import { ZodIpcConnection } from "../src/v3/zodIpc.js"; +import { ChildToParent, ParentToChild } from "./fixtures/zodIpcCatalog.js"; + +const childPath = fileURLToPath(new URL("./fixtures/zodIpcChild.ts", import.meta.url)); + +let child: ChildProcess | undefined; + +afterEach(() => { + child?.kill(); + child = undefined; +}); + +// Node's default IPC serialization is JSON, so an ack with `message: undefined` +// arrives without the `message` key, exactly as it does between a worker and its task run process. +function forkChild() { + child = fork(childPath, { execArgv: ["--import", "tsx"], stdio: "inherit" }); + + return new ZodIpcConnection({ + listenSchema: ChildToParent, + emitSchema: ParentToChild, + process: child, + }); +} + +describe("ZodIpcConnection", () => { + it("resolves sendWithAck for a void callback", async () => { + const connection = forkChild(); + + await expect(connection.sendWithAck("FLUSH", { timeoutInMs: 1000 }, 2000)).resolves.toBe( + undefined + ); + }); + + it("resolves sendWithAck with the callback payload", async () => { + const connection = forkChild(); + + await expect(connection.sendWithAck("PING", { value: "hello" }, 2000)).resolves.toEqual({ + echoed: "hello", + }); + }); +}); From 8fe554a7a5ce4b75954d89070b160171216161cb Mon Sep 17 00:00:00 2001 From: Chris Arderne Date: Mon, 28 Sep 2026 10:13:30 +0000 Subject: [PATCH 3/9] fix(webapp): keep scheduled last run times accurate Mono-RevId: b455565ff705ccad4d1bba3fe3e02b25fbcd063e --- .server-changes/schedule-last-run.md | 6 + apps/webapp/app/v3/scheduleTimings.server.ts | 74 ++++++++---- .../createBackgroundWorker.server.test.ts | 7 ++ .../services/createBackgroundWorker.server.ts | 39 +++--- apps/webapp/test/scheduleTimings.test.ts | 113 +++++++++++++++--- 5 files changed, 188 insertions(+), 51 deletions(-) create mode 100644 .server-changes/schedule-last-run.md diff --git a/.server-changes/schedule-last-run.md b/.server-changes/schedule-last-run.md new file mode 100644 index 00000000000..b1a05c63e41 --- /dev/null +++ b/.server-changes/schedule-last-run.md @@ -0,0 +1,6 @@ +--- +area: webapp +type: fix +--- + +Keep scheduled task "Last run" times stable across unchanged deployments and align them with configured schedule windows. diff --git a/apps/webapp/app/v3/scheduleTimings.server.ts b/apps/webapp/app/v3/scheduleTimings.server.ts index 04c35ec98f9..c22af28fb8b 100644 --- a/apps/webapp/app/v3/scheduleTimings.server.ts +++ b/apps/webapp/app/v3/scheduleTimings.server.ts @@ -88,7 +88,10 @@ export function resolveScheduleTimings( { phaseSecret, includeLastRun, now = new Date() }: ResolveScheduleTimingsOptions ): ScheduleTiming[] { const nominalCache = new Map(); - const previousCache = new Map(); + const previousCache = new Map< + string, + { latestNominal: Date; previousNominal?: Date } | undefined + >(); return inputs.map((input) => { const window: NormalizedScheduleWindow | undefined = resolveScheduleWindow({ @@ -134,51 +137,80 @@ export function resolveScheduleTimings( return { nextRun: nominalAt, nextRunEffectiveAt: effectiveAt, - lastRun: includeLastRun ? resolveLastRun(input, now, previousCache) : undefined, + lastRun: includeLastRun + ? resolveLastRun(input, now, nominalAt, phase, window, previousCache) + : undefined, }; }); } /** - * Approximates "last run" from the cron's previous slot. + * Approximates "last run" from the most recent effective schedule time. * - * Skips inactive schedules — the previous slot reflects what *would* have - * fired. Skips slots that predate `updatedAt`: any config change (cron edited, - * timezone changed, deactivate/reactivate) bumps `updatedAt`, and a slot from - * before the most recent change didn't fire under the current configuration. - * - * `cron-parser` throws on malformed expressions, so this degrades to undefined - * per row rather than failing the whole list. Best-effort by design; the runs - * page is the source of truth. + * Skips inactive schedules and effective times that predate `updatedAt`. Best-effort by design; + * the runs page is the source of truth. */ function resolveLastRun( input: ScheduleTimingInput, now: Date, - cache: Map + nextNominal: Date, + phase: number, + window: NormalizedScheduleWindow | undefined, + cache: Map ): Date | undefined { if (!input.active) { return undefined; } const key = cacheKey(input.cron, input.timezone); + let nominalTimes = cache.get(key); - let previous: Date | undefined; - if (cache.has(key)) { - previous = cache.get(key); - } else { + if (!cache.has(key)) { try { - previous = previousScheduledTimestamp(input.cron, input.timezone, now); + nominalTimes = { + latestNominal: previousScheduledTimestamp( + input.cron, + input.timezone, + new Date(now.getTime() + 1) + ), + }; } catch { - previous = undefined; + nominalTimes = undefined; } - cache.set(key, previous); + cache.set(key, nominalTimes); } - if (!previous) { + if (!nominalTimes) { return undefined; } - return previous.getTime() > input.updatedAt.getTime() ? previous : undefined; + const latestEffective = calculateEffectiveScheduleTime({ + nominalAt: nominalTimes.latestNominal, + nextNominalAt: nextNominal, + schedulePhase: phase, + window, + minimumWindowDurationSeconds: input.minimumWindowDurationSeconds, + }).effectiveAt; + if (latestEffective.getTime() <= now.getTime()) { + return latestEffective.getTime() > input.updatedAt.getTime() ? latestEffective : undefined; + } + + if (!nominalTimes.previousNominal) { + nominalTimes.previousNominal = previousScheduledTimestamp( + input.cron, + input.timezone, + nominalTimes.latestNominal + ); + } + const previousEffective = calculateEffectiveScheduleTime({ + nominalAt: nominalTimes.previousNominal, + nextNominalAt: nominalTimes.latestNominal, + schedulePhase: phase, + window, + minimumWindowDurationSeconds: input.minimumWindowDurationSeconds, + }).effectiveAt; + + return previousEffective.getTime() > input.updatedAt.getTime() ? previousEffective : undefined; } /** diff --git a/apps/webapp/app/v3/services/createBackgroundWorker.server.test.ts b/apps/webapp/app/v3/services/createBackgroundWorker.server.test.ts index 675727db80b..e4618413924 100644 --- a/apps/webapp/app/v3/services/createBackgroundWorker.server.test.ts +++ b/apps/webapp/app/v3/services/createBackgroundWorker.server.test.ts @@ -394,6 +394,7 @@ describe("syncDeclarativeSchedules registration", () => { async ({ prisma, redisOptions }) => { const { project, prodEnv } = await seedProjectWithEnvs(prisma); const schedule = await makeDeclarativeSchedule(prisma, project.id, [prodEnv.id]); + const updatedAt = schedule.updatedAt; await seedScheduledTask(prisma, project.id, prodEnv.id); const engine = createTestScheduleEngine(prisma, redisOptions); @@ -412,6 +413,12 @@ describe("syncDeclarativeSchedules registration", () => { expect(before).toBeDefined(); expect(await engine.getJob(jobId)).toEqual(before); + expect( + await prisma.taskSchedule.findUniqueOrThrow({ + where: { id: schedule.id }, + select: { updatedAt: true }, + }) + ).toEqual({ updatedAt }); } finally { await engine.quit(); } diff --git a/apps/webapp/app/v3/services/createBackgroundWorker.server.ts b/apps/webapp/app/v3/services/createBackgroundWorker.server.ts index 13a24576042..fe9dd5666ec 100644 --- a/apps/webapp/app/v3/services/createBackgroundWorker.server.ts +++ b/apps/webapp/app/v3/services/createBackgroundWorker.server.ts @@ -1399,6 +1399,7 @@ async function prepareDeclarativeSchedules( minimumWindowDurationSeconds: true, instances: { select: { + id: true, environmentId: true, }, }, @@ -1537,21 +1538,29 @@ export async function syncDeclarativeSchedules( !scheduleWindowsEqual(previousWindow, nextWindow) || // Clearing the minimum changes the effective range. existingSchedule.minimumWindowDurationSeconds !== minimumWindowDurationSeconds; - const schedule = await prisma.taskSchedule.update({ - where: { - id: existingSchedule.id, - }, - data: { - generatorExpression: task.schedule.cron, - generatorDescription: cronstrue.toString(task.schedule.cron), - timezone: task.schedule.timezone, - minimumWindowDurationSeconds, - ...normalizedWindow, - }, - include: { - instances: true, - }, - }); + const persistedValuesChanged = + existingSchedule.generatorExpression !== task.schedule.cron || + existingSchedule.timezone !== task.schedule.timezone || + existingSchedule.windowDurationSeconds !== normalizedWindow.windowDurationSeconds || + existingSchedule.windowPercentage !== normalizedWindow.windowPercentage || + existingSchedule.minimumWindowDurationSeconds !== minimumWindowDurationSeconds; + const schedule = persistedValuesChanged + ? await prisma.taskSchedule.update({ + where: { + id: existingSchedule.id, + }, + data: { + generatorExpression: task.schedule.cron, + generatorDescription: cronstrue.toString(task.schedule.cron), + timezone: task.schedule.timezone, + minimumWindowDurationSeconds, + ...normalizedWindow, + }, + include: { + instances: true, + }, + }) + : existingSchedule; missingSchedules.delete(existingSchedule.id); const instances = timingChanged diff --git a/apps/webapp/test/scheduleTimings.test.ts b/apps/webapp/test/scheduleTimings.test.ts index 5a3601f9ff9..d98651697ec 100644 --- a/apps/webapp/test/scheduleTimings.test.ts +++ b/apps/webapp/test/scheduleTimings.test.ts @@ -4,9 +4,9 @@ import { SCHEDULE_PHASE_DENOMINATOR, calculateEffectiveScheduleTime, calculateSchedulePhase, + resolveScheduleWindow, } from "@internal/schedule-engine"; import { CronPattern } from "~/v3/schedules"; -import { type NormalizedScheduleWindow } from "@trigger.dev/core/v3"; import { parseExpression } from "cron-parser"; import { describe, expect, it } from "vitest"; import { @@ -77,24 +77,48 @@ function referenceResolve( deduplicationKey: input.deduplicationKey, }); - const window: NormalizedScheduleWindow | undefined = - input.windowPercentage !== null - ? { type: "percentage", percentage: input.windowPercentage } - : input.windowDurationSeconds !== null - ? { type: "duration", durationSeconds: input.windowDurationSeconds } - : undefined; + const window = resolveScheduleWindow({ + windowDurationSeconds: input.windowDurationSeconds, + windowPercentage: input.windowPercentage, + defaultWindowDurationSeconds: input.defaultWindowDurationSeconds, + }).window; const { effectiveAt } = calculateEffectiveScheduleTime({ nominalAt: nominalTimes[0], nextNominalAt: nominalTimes[1], schedulePhase: phase, window, + minimumWindowDurationSeconds: input.minimumWindowDurationSeconds, }); let lastRun: Date | undefined; if (includeLastRun && input.active) { try { - const previous = previousScheduledTimestamp(input.cron, input.timezone, now); + const latestNominal = previousScheduledTimestamp( + input.cron, + input.timezone, + new Date(now.getTime() + 1) + ); + const previousNominal = previousScheduledTimestamp( + input.cron, + input.timezone, + latestNominal + ); + const latestEffective = calculateEffectiveScheduleTime({ + nominalAt: latestNominal, + nextNominalAt: nominalTimes[0], + schedulePhase: phase, + window, + minimumWindowDurationSeconds: input.minimumWindowDurationSeconds, + }).effectiveAt; + const previousEffective = calculateEffectiveScheduleTime({ + nominalAt: previousNominal, + nextNominalAt: latestNominal, + schedulePhase: phase, + window, + minimumWindowDurationSeconds: input.minimumWindowDurationSeconds, + }).effectiveAt; + const previous = latestEffective <= now ? latestEffective : previousEffective; lastRun = previous.getTime() > input.updatedAt.getTime() ? previous : undefined; } catch { lastRun = undefined; @@ -255,15 +279,24 @@ describe("resolveScheduleTimings", () => { }); it("skips lastRun when the previous slot predates the last config change", () => { - const [stale] = resolveScheduleTimings([input({ cron: "0 0 * * *", updatedAt: now })], { - phaseSecret: PHASE_SECRET, - includeLastRun: true, - now, - }); + const [stale] = resolveScheduleTimings( + [input({ cron: "0 0 * * *", schedulePhase: 0, updatedAt: now })], + { + phaseSecret: PHASE_SECRET, + includeLastRun: true, + now, + } + ); expect(stale.lastRun).toBeUndefined(); const [fresh] = resolveScheduleTimings( - [input({ cron: "0 0 * * *", updatedAt: new Date("2020-01-01T00:00:00.000Z") })], + [ + input({ + cron: "0 0 * * *", + schedulePhase: 0, + updatedAt: new Date("2020-01-01T00:00:00.000Z"), + }), + ], { phaseSecret: PHASE_SECRET, includeLastRun: true, now } ); expect(fresh.lastRun).toEqual(new Date("2024-06-15T00:00:00.000Z")); @@ -282,7 +315,7 @@ describe("resolveScheduleTimings", () => { }); it("resolves lastRun for a valid expression", () => { - const [valid] = resolveScheduleTimings([input({ cron: "0 0 * * *" })], { + const [valid] = resolveScheduleTimings([input({ cron: "0 0 * * *", schedulePhase: 0 })], { phaseSecret: PHASE_SECRET, includeLastRun: true, now, @@ -291,6 +324,56 @@ describe("resolveScheduleTimings", () => { expect(valid.lastRun).toEqual(new Date("2024-06-15T00:00:00.000Z")); }); + it("resolves lastRun from the most recent effective window time", () => { + const [beforeCurrentWindow] = resolveScheduleTimings( + [ + input({ + cron: "0 * * * *", + schedulePhase: SCHEDULE_PHASE_DENOMINATOR / 2, + windowPercentage: 100, + }), + ], + { + phaseSecret: PHASE_SECRET, + includeLastRun: true, + now, + } + ); + const [afterCurrentWindow] = resolveScheduleTimings( + [ + input({ + cron: "0 * * * *", + schedulePhase: SCHEDULE_PHASE_DENOMINATOR / 2, + windowPercentage: 100, + }), + ], + { + phaseSecret: PHASE_SECRET, + includeLastRun: true, + now: new Date("2024-06-15T09:40:00.000Z"), + } + ); + + expect(beforeCurrentWindow.lastRun).toEqual(new Date("2024-06-15T08:30:00.000Z")); + expect(afterCurrentWindow.lastRun).toEqual(new Date("2024-06-15T09:30:00.000Z")); + }); + + it("compares updatedAt with the effective window time", () => { + const [timing] = resolveScheduleTimings( + [ + input({ + cron: "0 * * * *", + schedulePhase: SCHEDULE_PHASE_DENOMINATOR / 2, + windowPercentage: 100, + updatedAt: new Date("2024-06-15T08:15:00.000Z"), + }), + ], + { phaseSecret: PHASE_SECRET, includeLastRun: true, now } + ); + + expect(timing.lastRun).toEqual(new Date("2024-06-15T08:30:00.000Z")); + }); + it("honours a caller-supplied schedulePhase over the derived one", () => { const rows = [input({ schedulePhase: 0, windowPercentage: 100, cron: "0 * * * *" })]; From b32c1d157a2cecd76ffc10c237f9bb5b45d48bcf Mon Sep 17 00:00:00 2001 From: Saadi Myftija Date: Mon, 28 Sep 2026 11:08:42 +0000 Subject: [PATCH 4/9] fix(cli): stop percent-encoding Windows output paths in the build manifest Fix Windows deploys failing at indexing with `Cannot find module` on a percent-encoded path when the project directory contains spaces or non-ASCII characters. Mono-RevId: 570c78ee09bc121ae8066a7e70848ae4509d74e3 --- .../windows-deploy-paths-with-spaces.md | 5 +++ packages/cli-v3/src/build/buildWorker.test.ts | 31 +++++++++++++++++++ packages/cli-v3/src/build/buildWorker.ts | 21 +++++-------- packages/cli-v3/src/deploy/buildImage.test.ts | 17 ++++++++++ packages/cli-v3/src/deploy/buildImage.ts | 4 +-- 5 files changed, 62 insertions(+), 16 deletions(-) create mode 100644 .changeset/windows-deploy-paths-with-spaces.md create mode 100644 packages/cli-v3/src/build/buildWorker.test.ts diff --git a/.changeset/windows-deploy-paths-with-spaces.md b/.changeset/windows-deploy-paths-with-spaces.md new file mode 100644 index 00000000000..54fae1d759d --- /dev/null +++ b/.changeset/windows-deploy-paths-with-spaces.md @@ -0,0 +1,5 @@ +--- +"trigger.dev": patch +--- + +Fix Windows deploys failing at indexing with `Cannot find module` on a percent-encoded path when the project directory contains spaces or non-ASCII characters. diff --git a/packages/cli-v3/src/build/buildWorker.test.ts b/packages/cli-v3/src/build/buildWorker.test.ts new file mode 100644 index 00000000000..90f828c014e --- /dev/null +++ b/packages/cli-v3/src/build/buildWorker.test.ts @@ -0,0 +1,31 @@ +import { join, posix, sep, win32 } from "node:path"; +import { describe, expect, it } from "vitest"; +import { rewriteOutputPath } from "./buildWorker.js"; + +describe("rewriteOutputPath", () => { + it("defaults to the host platform's path semantics", () => { + const out = join(sep, "tmp", "out"); + expect(rewriteOutputPath(out, join(out, "src", "a b.mjs"))).toBe("/app/src/a b.mjs"); + }); + + it("maps a bundled file to its /app path on posix", () => { + expect(rewriteOutputPath("/tmp/out", "/tmp/out/src/trigger/a.mjs", posix)).toBe( + "/app/src/trigger/a.mjs" + ); + }); + + it("uses forward slashes and keeps spaces and non-ASCII characters on windows", () => { + const out = "C:\\Users\\me\\AppData\\Local\\Temp\\trigger-build"; + const file = `${out}\\Documents\\DATEN VORBEREITUNG FÜR AUKTIONEN\\trigger.config.mjs`; + + expect(rewriteOutputPath(out, file, win32)).toBe( + "/app/Documents/DATEN VORBEREITUNG FÜR AUKTIONEN/trigger.config.mjs" + ); + }); + + it("tolerates mixed separators and drive letter case on windows", () => { + expect( + rewriteOutputPath("c:\\Users\\me\\out", "C:/Users/me/out/Trigger Demo/index.mjs", win32) + ).toBe("/app/Trigger Demo/index.mjs"); + }); +}); diff --git a/packages/cli-v3/src/build/buildWorker.ts b/packages/cli-v3/src/build/buildWorker.ts index 2aee0192af7..6ad0da8ba07 100644 --- a/packages/cli-v3/src/build/buildWorker.ts +++ b/packages/cli-v3/src/build/buildWorker.ts @@ -22,15 +22,13 @@ import { import { createExternalsBuildExtension } from "./externals.js"; import { tmpdir } from "node:os"; import { mkdtemp, rm } from "node:fs/promises"; -import { join, relative, sep } from "node:path"; +import path, { join } from "node:path"; import { generateContainerfile } from "../deploy/buildImage.js"; import { writeFile } from "node:fs/promises"; import { buildManifestToJSON } from "../utilities/buildManifest.js"; import { logger } from "../utilities/logger.js"; import { readPackageJSON } from "pkg-types"; import { writeJSONFile } from "../utilities/fileSystem.js"; -import { isWindows } from "std-env"; -import { pathToFileURL } from "node:url"; import { logBuildWorkerStart } from "./buildWorkerLogging.js"; import { SdkVersionExtractor } from "./plugins.js"; import { spinner } from "../utilities/windows.js"; @@ -215,17 +213,12 @@ function cleanEntryPath(entry: string): string { return entry.split("?")[0]!; } -function rewriteOutputPath(destinationDir: string, filePath: string) { - if (isWindows) { - return `/app/${relative( - pathToFileURL(destinationDir).pathname, - pathToFileURL(filePath).pathname - ) - .split(sep) - .join("/")}`; - } else { - return `/app/${relative(destinationDir, filePath)}`; - } +export function rewriteOutputPath( + destinationDir: string, + filePath: string, + pathApi: Pick = path +) { + return `/app/${pathApi.relative(destinationDir, filePath).split(pathApi.sep).join("/")}`; } async function writeDeployFiles({ diff --git a/packages/cli-v3/src/deploy/buildImage.test.ts b/packages/cli-v3/src/deploy/buildImage.test.ts index cb7c78242ba..6b25bbf89dc 100644 --- a/packages/cli-v3/src/deploy/buildImage.test.ts +++ b/packages/cli-v3/src/deploy/buildImage.test.ts @@ -62,6 +62,23 @@ describe("generateContainerfile", () => { expect(containerfile).toContain(`FROM ${build} AS build`); }); + it.each(["node", "bun"] as BuildRuntime[])( + "runs the indexer in exec form so entry paths with spaces survive on %s", + async (runtime) => { + const indexScript = "/app/Documents/Trigger Demo/index.mjs"; + const containerfile = await generateContainerfile({ + runtime, + build: {}, + image: undefined, + indexScript, + entrypoint: "entrypoint.js", + }); + + expect(containerfile).toContain(`"${indexScript}" ]`); + expect(containerfile).not.toMatch(/^RUN (node|bun run) \//m); + } + ); + it.each(["node", "bun"] as BuildRuntime[])( "runs no package installation for uncustomized projects on %s", async (runtime) => { diff --git a/packages/cli-v3/src/deploy/buildImage.ts b/packages/cli-v3/src/deploy/buildImage.ts index 0505698e2f3..40b82717cbd 100644 --- a/packages/cli-v3/src/deploy/buildImage.ts +++ b/packages/cli-v3/src/deploy/buildImage.ts @@ -892,7 +892,7 @@ ARG BUILDPLATFORM ENV BUILDPLATFORM=$BUILDPLATFORM TARGETPLATFORM=$TARGETPLATFORM # Run the indexer -RUN bun run ${options.indexScript} +RUN [ "bun", "run", "${options.indexScript}" ] # Development or production stage builds upon the base stage FROM base AS final @@ -1001,7 +1001,7 @@ ARG BUILDPLATFORM ENV BUILDPLATFORM=$BUILDPLATFORM TARGETPLATFORM=$TARGETPLATFORM # Run the indexer -RUN node ${options.indexScript} +RUN [ "node", "${options.indexScript}" ] # Development or production stage builds upon the base stage FROM base AS final From 3fef4acdab408a66bca1562c199f9a84a2ad41c2 Mon Sep 17 00:00:00 2001 From: Chris Arderne Date: Mon, 28 Sep 2026 11:11:04 +0000 Subject: [PATCH 5/9] fix(webapp): allow schedule replacements at the quota limit Mono-RevId: 90559be02c928bd675173ec4a5ff2e8cb3319a7e --- .server-changes/declarative-schedule-quota.md | 6 + .../app/v3/services/checkSchedule.server.ts | 28 +- .../createBackgroundWorker.server.test.ts | 262 +++++++++++++ .../services/createBackgroundWorker.server.ts | 344 ++++++++++-------- 4 files changed, 486 insertions(+), 154 deletions(-) create mode 100644 .server-changes/declarative-schedule-quota.md diff --git a/.server-changes/declarative-schedule-quota.md b/.server-changes/declarative-schedule-quota.md new file mode 100644 index 00000000000..3e63083cff5 --- /dev/null +++ b/.server-changes/declarative-schedule-quota.md @@ -0,0 +1,6 @@ +--- +area: webapp +type: fix +--- + +Allow deployments to replace declarative schedules at the schedule limit when the resulting set stays within quota. diff --git a/apps/webapp/app/v3/services/checkSchedule.server.ts b/apps/webapp/app/v3/services/checkSchedule.server.ts index 946cfad080c..1458aa5f55b 100644 --- a/apps/webapp/app/v3/services/checkSchedule.server.ts +++ b/apps/webapp/app/v3/services/checkSchedule.server.ts @@ -17,7 +17,14 @@ type Schedule = { }; export class CheckScheduleService extends BaseService { - public async call(projectId: string, schedule: Schedule, environmentIds: string[]) { + public async call( + projectId: string, + schedule: Schedule, + environmentIds: string[], + quotaExclusions?: { environmentId: string; scheduleIds: string[] }, + pendingCreations = 0, + quotaLimit?: number + ) { //validate the cron expression try { CronPattern.parse(schedule.cron); @@ -115,15 +122,18 @@ export class CheckScheduleService extends BaseService { //if creating a schedule, check they're under the limits if (!schedule.friendlyId) { - const limit = await getLimit(project.organizationId, "schedules", 100_000_000); + const limit = + quotaLimit ?? (await getLimit(project.organizationId, "schedules", 100_000_000)); const schedulesCount = await CheckScheduleService.getUsedSchedulesCount({ prisma: this._prisma, projectId, + quotaExclusions, }); - if (schedulesCount >= limit) { + const projectedCount = schedulesCount + pendingCreations; + if (projectedCount >= limit) { throw new ServiceValidationError( - `You have created ${schedulesCount}/${limit} schedules so you'll need to increase your limits or delete some schedules.` + `You have created ${projectedCount}/${limit} schedules so you'll need to increase your limits or delete some schedules.` ); } } @@ -132,14 +142,24 @@ export class CheckScheduleService extends BaseService { static async getUsedSchedulesCount({ prisma, projectId, + quotaExclusions, }: { prisma: PrismaClientOrTransaction; projectId: string; + quotaExclusions?: { environmentId: string; scheduleIds: string[] }; }) { return await prisma.taskScheduleInstance.count({ where: { projectId, active: true, + ...(quotaExclusions?.scheduleIds.length + ? { + NOT: { + environmentId: quotaExclusions.environmentId, + taskScheduleId: { in: boundedIn(quotaExclusions.scheduleIds) }, + }, + } + : {}), environment: { projectId, type: { diff --git a/apps/webapp/app/v3/services/createBackgroundWorker.server.test.ts b/apps/webapp/app/v3/services/createBackgroundWorker.server.test.ts index e4618413924..11ab03545a9 100644 --- a/apps/webapp/app/v3/services/createBackgroundWorker.server.test.ts +++ b/apps/webapp/app/v3/services/createBackgroundWorker.server.test.ts @@ -499,6 +499,268 @@ describe("syncDeclarativeSchedules registration", () => { }); }); +describe("syncDeclarativeSchedules quota", () => { + containerTest( + "replaces a schedule at the limit without deleting it before validation", + async ({ prisma, redisOptions }) => { + const { project, prodEnv } = await seedProjectWithEnvs(prisma); + const old = await makeDeclarativeSchedule(prisma, project.id, [prodEnv.id], "old-task"); + await seedScheduledTask(prisma, project.id, prodEnv.id); + const engine = createTestScheduleEngine(prisma, redisOptions); + + try { + await expect( + syncDeclarativeSchedules( + declarativeTasks({ cron: "not-a-cron", timezone: "UTC" }), + noWorker, + asEnv(prodEnv), + prisma, + engine, + undefined, + 1 + ) + ).rejects.toThrow("Invalid cron expression"); + expect(await prisma.taskSchedule.findUnique({ where: { id: old.id } })).not.toBeNull(); + + await syncDeclarativeSchedules( + declarativeTasks({ cron: "0 * * * *", timezone: "UTC" }), + noWorker, + asEnv(prodEnv), + prisma, + engine, + undefined, + 1 + ); + + expect(await prisma.taskSchedule.findUnique({ where: { id: old.id } })).toBeNull(); + expect( + await prisma.taskSchedule.findFirst({ + where: { projectId: project.id, taskIdentifier: "my-task" }, + }) + ).not.toBeNull(); + expect(await prisma.taskScheduleInstance.count({ where: { projectId: project.id } })).toBe( + 1 + ); + const created = await prisma.taskSchedule.findFirstOrThrow({ + where: { projectId: project.id, taskIdentifier: "my-task" }, + include: { instances: true }, + }); + expect(await engine.getJob(scheduleJobId(created.instances[0].id))).toBeDefined(); + } finally { + await engine.quit(); + } + } + ); + + containerTest( + "rejects a later invalid schedule before writing an earlier replacement", + async ({ prisma }) => { + const { project, prodEnv } = await seedProjectWithEnvs(prisma); + const old = await makeDeclarativeSchedule(prisma, project.id, [prodEnv.id], "old-task"); + await seedScheduledTask(prisma, project.id, prodEnv.id); + const worker = await prisma.backgroundWorker.findFirstOrThrow({ + where: { projectId: project.id }, + }); + await prisma.backgroundWorkerTask.create({ + data: { + friendlyId: `task_other_${prodEnv.id}`, + slug: "other-task", + filePath: "src/trigger/other-task.ts", + workerId: worker.id, + projectId: project.id, + runtimeEnvironmentId: prodEnv.id, + triggerSource: "SCHEDULED", + }, + }); + const tasks = [ + { id: "my-task", schedule: { cron: "0 * * * *", timezone: "UTC" } }, + { id: "other-task", schedule: { cron: "0 * * * *", timezone: "Not/A_Timezone" } }, + ] as TasksArg; + + await expect( + syncDeclarativeSchedules(tasks, noWorker, asEnv(prodEnv), prisma, undefined, undefined, 1) + ).rejects.toThrow("Invalid IANA timezone"); + expect(await prisma.taskSchedule.findUnique({ where: { id: old.id } })).not.toBeNull(); + expect(await prisma.taskSchedule.count({ where: { projectId: project.id } })).toBe(1); + + tasks[1].schedule.timezone = "UTC"; + await expect( + syncDeclarativeSchedules(tasks, noWorker, asEnv(prodEnv), prisma, undefined, undefined, 1) + ).rejects.toThrow("You have created 1/1 schedules"); + expect(await prisma.taskSchedule.count({ where: { projectId: project.id } })).toBe(1); + } + ); + + containerTest( + "does not count development schedules toward projected quota", + async ({ prisma, redisOptions }) => { + const { project, devEnv } = await seedProjectWithEnvs(prisma); + await seedScheduledTask(prisma, project.id, devEnv.id); + const worker = await prisma.backgroundWorker.findFirstOrThrow({ + where: { projectId: project.id }, + }); + await prisma.backgroundWorkerTask.create({ + data: { + friendlyId: `task_other_${devEnv.id}`, + slug: "other-task", + filePath: "src/trigger/other-task.ts", + workerId: worker.id, + projectId: project.id, + runtimeEnvironmentId: devEnv.id, + triggerSource: "SCHEDULED", + }, + }); + const engine = createTestScheduleEngine(prisma, redisOptions); + try { + await syncDeclarativeSchedules( + [ + { id: "my-task", schedule: { cron: "0 * * * *", timezone: "UTC" } }, + { id: "other-task", schedule: { cron: "0 * * * *", timezone: "UTC" } }, + ] as TasksArg, + noWorker, + asEnv(devEnv), + prisma, + engine, + undefined, + 1 + ); + expect(await prisma.taskScheduleInstance.count({ where: { projectId: project.id } })).toBe( + 2 + ); + } finally { + await engine.quit(); + } + } + ); + + containerTest( + "serializes concurrent replacements at the limit", + async ({ prisma, redisOptions }) => { + const { project, prodEnv } = await seedProjectWithEnvs(prisma); + await makeDeclarativeSchedule(prisma, project.id, [prodEnv.id], "old-task"); + await seedScheduledTask(prisma, project.id, prodEnv.id); + const worker = await prisma.backgroundWorker.findFirstOrThrow({ + where: { projectId: project.id }, + }); + await prisma.backgroundWorkerTask.create({ + data: { + friendlyId: `task_other_${prodEnv.id}`, + slug: "other-task", + filePath: "src/trigger/other-task.ts", + workerId: worker.id, + projectId: project.id, + runtimeEnvironmentId: prodEnv.id, + triggerSource: "SCHEDULED", + }, + }); + const engine = createTestScheduleEngine(prisma, redisOptions); + try { + const results = await Promise.allSettled( + ["my-task", "other-task"].map((id) => + syncDeclarativeSchedules( + [{ id, schedule: { cron: "0 * * * *", timezone: "UTC" } }] as TasksArg, + noWorker, + asEnv(prodEnv), + prisma, + engine, + undefined, + 1 + ) + ) + ); + expect(results.some((result) => result.status === "fulfilled")).toBe(true); + expect(await prisma.taskScheduleInstance.count({ where: { projectId: project.id } })).toBe( + 1 + ); + expect( + await prisma.taskSchedule.findFirstOrThrow({ where: { projectId: project.id } }) + ).toMatchObject({ + type: "DECLARATIVE", + }); + } finally { + await engine.quit(); + } + } + ); + + containerTest( + "concurrent replacements with spare quota leave only the last declaration", + async ({ prisma, redisOptions }) => { + const { project, prodEnv } = await seedProjectWithEnvs(prisma); + await makeDeclarativeSchedule(prisma, project.id, [prodEnv.id], "old-task"); + await seedScheduledTask(prisma, project.id, prodEnv.id); + const worker = await prisma.backgroundWorker.findFirstOrThrow({ + where: { projectId: project.id }, + }); + await prisma.backgroundWorkerTask.create({ + data: { + friendlyId: `task_other_${prodEnv.id}`, + slug: "other-task", + filePath: "src/trigger/other-task.ts", + workerId: worker.id, + projectId: project.id, + runtimeEnvironmentId: prodEnv.id, + triggerSource: "SCHEDULED", + }, + }); + const engine = createTestScheduleEngine(prisma, redisOptions); + try { + await Promise.all( + ["my-task", "other-task"].map((id) => + syncDeclarativeSchedules( + [{ id, schedule: { cron: "0 * * * *", timezone: "UTC" } }] as TasksArg, + noWorker, + asEnv(prodEnv), + prisma, + engine, + undefined, + 2 + ) + ) + ); + const schedules = await prisma.taskSchedule.findMany({ + where: { projectId: project.id }, + select: { taskIdentifier: true }, + }); + expect(schedules).toHaveLength(1); + expect(["my-task", "other-task"]).toContain(schedules[0].taskIdentifier); + } finally { + await engine.quit(); + } + } + ); + + containerTest("does not borrow quota from another environment's instance", async ({ prisma }) => { + const { project, prodEnv, devEnv } = await seedProjectWithEnvs(prisma); + await prisma.runtimeEnvironment.update({ + where: { id: devEnv.id }, + data: { type: "PRODUCTION" }, + }); + const old = await makeDeclarativeSchedule( + prisma, + project.id, + [prodEnv.id, devEnv.id], + "old-task" + ); + await seedScheduledTask(prisma, project.id, prodEnv.id); + + await expect( + syncDeclarativeSchedules( + declarativeTasks({ cron: "0 * * * *", timezone: "UTC" }), + noWorker, + asEnv(prodEnv), + prisma, + undefined, + undefined, + 1 + ) + ).rejects.toThrow("You have created 1/1 schedules"); + + expect(await prisma.taskSchedule.findUnique({ where: { id: old.id } })).not.toBeNull(); + expect(await prisma.taskScheduleInstance.count({ where: { projectId: project.id } })).toBe(2); + }); +}); + describe("syncDeclarativeSchedules deletion path", () => { containerTest( "does not issue any instance delete when the env owns no instance of the missing schedules", diff --git a/apps/webapp/app/v3/services/createBackgroundWorker.server.ts b/apps/webapp/app/v3/services/createBackgroundWorker.server.ts index fe9dd5666ec..f61845dd7f1 100644 --- a/apps/webapp/app/v3/services/createBackgroundWorker.server.ts +++ b/apps/webapp/app/v3/services/createBackgroundWorker.server.ts @@ -1487,175 +1487,219 @@ export async function syncDeclarativeSchedules( environment: AuthenticatedEnvironment, prisma: PrismaClientOrTransaction, engine: Pick = scheduleEngine, - prepared?: Awaited> + prepared?: Awaited>, + quotaLimit?: number ) { - const { - existingDeclarativeSchedules, - preparedTasks, - defaultWindowDurationSeconds, - warnings = [], - } = prepared ?? (await prepareDeclarativeSchedules(tasks, environment, prisma)); - - //start out by assuming they're all missing - const missingSchedules = new Set( - existingDeclarativeSchedules.map((schedule) => schedule.id) - ); - - const checkSchedule = new CheckScheduleService(prisma); - - // Resource and quota checks must stay after worker resources exist and between schedule writes. - for (const { task, existingSchedule, minimumWindowDurationSeconds } of preparedTasks) { - //this throws errors if the schedule is invalid - await checkSchedule.call( - environment.projectId, - { - cron: task.schedule.cron, - timezone: task.schedule.timezone, - taskIdentifier: task.id, - friendlyId: existingSchedule?.friendlyId, - window: task.schedule.window, - }, - [environment.id] - ); + let { existingDeclarativeSchedules, preparedTasks, defaultWindowDurationSeconds, warnings } = + prepared ?? (await prepareDeclarativeSchedules(tasks, environment, prisma)); + + let missingSchedules = new Set(existingDeclarativeSchedules.map((schedule) => schedule.id)); + const declaredTaskIds = new Set(preparedTasks.map(({ task }) => task.id)); + let quotaExclusions = { + environmentId: environment.id, + scheduleIds: existingDeclarativeSchedules + .filter((schedule) => !declaredTaskIds.has(schedule.taskIdentifier)) + .map((schedule) => schedule.id), + }; - const normalizedWindow = normalizeScheduleWindow(task.schedule.window); - - if (existingSchedule) { - // Compare effective windows so equivalent explicit/default values preserve the pending job. - const previousWindow = resolveScheduleWindow({ - windowDurationSeconds: existingSchedule.windowDurationSeconds, - windowPercentage: existingSchedule.windowPercentage, - defaultWindowDurationSeconds: existingSchedule.defaultWindowDurationSeconds, - }).window; - const nextWindow = resolveScheduleWindow({ - windowDurationSeconds: normalizedWindow.windowDurationSeconds, - windowPercentage: normalizedWindow.windowPercentage, - defaultWindowDurationSeconds: existingSchedule.defaultWindowDurationSeconds, - }).window; - const timingChanged = - existingSchedule.generatorExpression !== task.schedule.cron || - existingSchedule.timezone !== task.schedule.timezone || - !scheduleWindowsEqual(previousWindow, nextWindow) || - // Clearing the minimum changes the effective range. - existingSchedule.minimumWindowDurationSeconds !== minimumWindowDurationSeconds; - const persistedValuesChanged = - existingSchedule.generatorExpression !== task.schedule.cron || - existingSchedule.timezone !== task.schedule.timezone || - existingSchedule.windowDurationSeconds !== normalizedWindow.windowDurationSeconds || - existingSchedule.windowPercentage !== normalizedWindow.windowPercentage || - existingSchedule.minimumWindowDurationSeconds !== minimumWindowDurationSeconds; - const schedule = persistedValuesChanged - ? await prisma.taskSchedule.update({ - where: { - id: existingSchedule.id, - }, - data: { - generatorExpression: task.schedule.cron, - generatorDescription: cronstrue.toString(task.schedule.cron), - timezone: task.schedule.timezone, - minimumWindowDurationSeconds, - ...normalizedWindow, - }, - include: { - instances: true, - }, - }) - : existingSchedule; - - missingSchedules.delete(existingSchedule.id); - const instances = timingChanged - ? schedule.instances - : schedule.instances.filter((instance) => instance.environmentId === environment.id); - if (instances.length === 0) { - throw new CreateDeclarativeScheduleError( - `Missing instance for declarative schedule ${schedule.id}` - ); - } - for (const instance of instances) { - await engine.registerNextTaskScheduleInstance({ - instanceId: instance.id, - preserveExistingJob: !timingChanged, - }); - } - } else { - const newSchedule = await prisma.taskSchedule.create({ - data: { - friendlyId: generateFriendlyId("sched"), - projectId: environment.projectId, - taskIdentifier: task.id, - generatorExpression: task.schedule.cron, - generatorDescription: cronstrue.toString(task.schedule.cron), + const validate = async (db: PrismaClientOrTransaction) => { + const checkSchedule = new CheckScheduleService(db); + let pendingCreations = 0; + for (const { task, existingSchedule } of preparedTasks) { + await checkSchedule.call( + environment.projectId, + { + cron: task.schedule.cron, timezone: task.schedule.timezone, - type: "DECLARATIVE", - minimumWindowDurationSeconds, - ...normalizedWindow, - defaultWindowDurationSeconds, - instances: { - create: [ - { - environmentId: environment.id, - projectId: environment.projectId, + taskIdentifier: task.id, + friendlyId: existingSchedule?.friendlyId, + window: task.schedule.window, + }, + [environment.id], + quotaExclusions, + pendingCreations, + quotaLimit + ); + if (!existingSchedule && environment.type !== "DEVELOPMENT") pendingCreations++; + } + return pendingCreations; + }; + const pendingCreations = await validate(prisma); + + const registrations: Array<{ instanceId: string; preserveExistingJob?: boolean }> = []; + const transactional = + quotaExclusions.scheduleIds.length > 0 && pendingCreations > 0 && "$transaction" in prisma; + const apply = async (db: PrismaClientOrTransaction) => { + for (const { task, existingSchedule, minimumWindowDurationSeconds } of preparedTasks) { + const normalizedWindow = normalizeScheduleWindow(task.schedule.window); + + if (existingSchedule) { + // Compare effective windows so equivalent explicit/default values preserve the pending job. + const previousWindow = resolveScheduleWindow({ + windowDurationSeconds: existingSchedule.windowDurationSeconds, + windowPercentage: existingSchedule.windowPercentage, + defaultWindowDurationSeconds: existingSchedule.defaultWindowDurationSeconds, + }).window; + const nextWindow = resolveScheduleWindow({ + windowDurationSeconds: normalizedWindow.windowDurationSeconds, + windowPercentage: normalizedWindow.windowPercentage, + defaultWindowDurationSeconds: existingSchedule.defaultWindowDurationSeconds, + }).window; + const timingChanged = + existingSchedule.generatorExpression !== task.schedule.cron || + existingSchedule.timezone !== task.schedule.timezone || + !scheduleWindowsEqual(previousWindow, nextWindow) || + // Clearing the minimum changes the effective range. + existingSchedule.minimumWindowDurationSeconds !== minimumWindowDurationSeconds; + const persistedValuesChanged = + existingSchedule.generatorExpression !== task.schedule.cron || + existingSchedule.timezone !== task.schedule.timezone || + existingSchedule.windowDurationSeconds !== normalizedWindow.windowDurationSeconds || + existingSchedule.windowPercentage !== normalizedWindow.windowPercentage || + existingSchedule.minimumWindowDurationSeconds !== minimumWindowDurationSeconds; + const schedule = persistedValuesChanged + ? await db.taskSchedule.update({ + where: { + id: existingSchedule.id, + }, + data: { + generatorExpression: task.schedule.cron, + generatorDescription: cronstrue.toString(task.schedule.cron), + timezone: task.schedule.timezone, + minimumWindowDurationSeconds, + ...normalizedWindow, }, - ], + include: { + instances: true, + }, + }) + : existingSchedule; + + missingSchedules.delete(existingSchedule.id); + const instances = timingChanged + ? schedule.instances + : schedule.instances.filter((instance) => instance.environmentId === environment.id); + if (instances.length === 0) { + throw new CreateDeclarativeScheduleError( + `Missing instance for declarative schedule ${schedule.id}` + ); + } + for (const instance of instances) { + const registration = { instanceId: instance.id, preserveExistingJob: !timingChanged }; + if (transactional) registrations.push(registration); + else await engine.registerNextTaskScheduleInstance(registration); + } + } else { + const newSchedule = await db.taskSchedule.create({ + data: { + friendlyId: generateFriendlyId("sched"), + projectId: environment.projectId, + taskIdentifier: task.id, + generatorExpression: task.schedule.cron, + generatorDescription: cronstrue.toString(task.schedule.cron), + timezone: task.schedule.timezone, + type: "DECLARATIVE", + minimumWindowDurationSeconds, + ...normalizedWindow, + defaultWindowDurationSeconds, + instances: { + create: [ + { + environmentId: environment.id, + projectId: environment.projectId, + }, + ], + }, }, - }, - include: { - instances: true, - }, - }); + include: { + instances: true, + }, + }); - const instance = newSchedule.instances.at(0); + const instance = newSchedule.instances.at(0); - if (instance) { - await engine.registerNextTaskScheduleInstance({ instanceId: instance.id }); - } else { - throw new CreateDeclarativeScheduleError( - `Missing instance for declarative schedule ${newSchedule.id}` - ); + if (instance) { + const registration = { instanceId: instance.id }; + if (transactional) registrations.push(registration); + else await engine.registerNextTaskScheduleInstance(registration); + } else { + throw new CreateDeclarativeScheduleError( + `Missing instance for declarative schedule ${newSchedule.id}` + ); + } } } - } - //Delete instances for this environment - //Delete schedules that have no instances left - const potentiallyDeletableSchedules = existingDeclarativeSchedules.filter((schedule) => - missingSchedules.has(schedule.id) - ); + //Delete instances for this environment + //Delete schedules that have no instances left + const potentiallyDeletableSchedules = existingDeclarativeSchedules.filter((schedule) => + missingSchedules.has(schedule.id) + ); - const scheduleIdsToDelete: string[] = []; - const scheduleIdsToDetachFromEnvironment: string[] = []; + const scheduleIdsToDelete: string[] = []; + const scheduleIdsToDetachFromEnvironment: string[] = []; - for (const schedule of potentiallyDeletableSchedules) { - const canDeleteSchedule = - schedule.instances.length === 0 || - schedule.instances.every((instance) => instance.environmentId === environment.id); + for (const schedule of potentiallyDeletableSchedules) { + const canDeleteSchedule = + schedule.instances.length === 0 || + schedule.instances.every((instance) => instance.environmentId === environment.id); - if (canDeleteSchedule) { - scheduleIdsToDelete.push(schedule.id); - } else if (schedule.instances.some((instance) => instance.environmentId === environment.id)) { - scheduleIdsToDetachFromEnvironment.push(schedule.id); + if (canDeleteSchedule) { + scheduleIdsToDelete.push(schedule.id); + } else if (schedule.instances.some((instance) => instance.environmentId === environment.id)) { + scheduleIdsToDetachFromEnvironment.push(schedule.id); + } } - } - if (scheduleIdsToDelete.length > 0) { - await prisma.taskSchedule.deleteMany({ - where: { - id: { - in: boundedIn(scheduleIdsToDelete), + if (scheduleIdsToDelete.length > 0) { + await db.taskSchedule.deleteMany({ + where: { + id: { + in: boundedIn(scheduleIdsToDelete), + }, }, - }, - }); - } + }); + } - if (scheduleIdsToDetachFromEnvironment.length > 0) { - await prisma.taskScheduleInstance.deleteMany({ - where: { - taskScheduleId: { - in: boundedIn(scheduleIdsToDetachFromEnvironment), + if (scheduleIdsToDetachFromEnvironment.length > 0) { + await db.taskScheduleInstance.deleteMany({ + where: { + taskScheduleId: { + in: boundedIn(scheduleIdsToDetachFromEnvironment), + }, + environmentId: environment.id, }, - environmentId: environment.id, + }); + } + }; + + if (transactional) { + await $transaction( + prisma, + "syncDeclarativeSchedules", + async (tx) => { + await tx.$queryRaw`SELECT id FROM "Project" WHERE id = ${environment.projectId} FOR UPDATE`; + ({ existingDeclarativeSchedules, preparedTasks, defaultWindowDurationSeconds, warnings } = + await prepareDeclarativeSchedules(tasks, environment, tx)); + missingSchedules = new Set(existingDeclarativeSchedules.map((schedule) => schedule.id)); + const declaredTaskIds = new Set(preparedTasks.map(({ task }) => task.id)); + quotaExclusions = { + environmentId: environment.id, + scheduleIds: existingDeclarativeSchedules + .filter((schedule) => !declaredTaskIds.has(schedule.taskIdentifier)) + .map((schedule) => schedule.id), + }; + await validate(tx); + await apply(tx); }, - }); + { timeout: 120_000 } + ); + } else { + await apply(prisma); + } + + for (const registration of registrations) { + await engine.registerNextTaskScheduleInstance(registration); } return warnings; From 7f3e6073ca76acfdfcad24ab0335b47569a413f3 Mon Sep 17 00:00:00 2001 From: Chris Arderne Date: Mon, 28 Sep 2026 11:22:19 +0000 Subject: [PATCH 6/9] fix(webapp,run-engine): recover interrupted batch completion Mono-RevId: 6f35650c0ce4a1243998ab7fbb10545b321bb4c1 --- .server-changes/batch-completion-recovery.md | 6 + .../webapp/app/v3/runEngineHandlers.server.ts | 2 - .../app/v3/runEngineHandlersShared.server.ts | 8 +- apps/webapp/test/runEngineHandlers.test.ts | 20 +--- .../src/engine/systems/batchSystem.test.ts | 106 +++++++++++++----- .../src/engine/systems/batchSystem.ts | 67 ++++++----- 6 files changed, 132 insertions(+), 77 deletions(-) create mode 100644 .server-changes/batch-completion-recovery.md diff --git a/.server-changes/batch-completion-recovery.md b/.server-changes/batch-completion-recovery.md new file mode 100644 index 00000000000..405625627de --- /dev/null +++ b/.server-changes/batch-completion-recovery.md @@ -0,0 +1,6 @@ +--- +area: webapp +type: fix +--- + +Prevent batch waits from remaining suspended when batch completion is briefly interrupted diff --git a/apps/webapp/app/v3/runEngineHandlers.server.ts b/apps/webapp/app/v3/runEngineHandlers.server.ts index 4ef52805e1e..2d3b6022940 100644 --- a/apps/webapp/app/v3/runEngineHandlers.server.ts +++ b/apps/webapp/app/v3/runEngineHandlers.server.ts @@ -9,7 +9,6 @@ import { runOpsNewReplica, runOpsLegacyReplica, runOpsNewPrismaClient, - runOpsNewReplicaClient, runOpsLegacyPrismaClient, runOpsShardHandles, } from "~/db.server"; @@ -1060,7 +1059,6 @@ export function setupBatchQueueCallbacks() { engine.setBatchCompletionCallback(async (result: CompleteBatchResult) => { await handleBatchCompletion(result, { splitEnabled: await splitEnabledPromise, - newReplica: runOpsNewReplicaClient, newWriter: runOpsNewPrismaClient, legacyWriter: runOpsLegacyPrismaClient, shards: runOpsShardHandles, diff --git a/apps/webapp/app/v3/runEngineHandlersShared.server.ts b/apps/webapp/app/v3/runEngineHandlersShared.server.ts index a20d531ba7e..a2bfbe19f3d 100644 --- a/apps/webapp/app/v3/runEngineHandlersShared.server.ts +++ b/apps/webapp/app/v3/runEngineHandlersShared.server.ts @@ -76,12 +76,12 @@ export async function readRunForEventOrThrow( * on a single run-ops DB. Length classification is INVALID here: a batch id may * be a run-ops id (cut-over orgs) or a cuid (and cuid-shaped ids can be backfilled * onto NEW), so id-shape does not reliably indicate the row's actual residency. - * The existence probe is the correct signal. + * The existence probe uses the writer because batch finalization is a read-after-write + * path and a lagging replica can incorrectly route a NEW batch to LEGACY. */ export async function resolveBatchRunOpsWriter( batchId: string, deps: { - newReplica: RunOpsPrismaClient; newWriter: RunOpsPrismaClient; legacyWriter: RunOpsPrismaClient; shards?: ReadonlyArray<{ key: string; writer: RunOpsPrismaClient }>; @@ -101,7 +101,7 @@ export async function resolveBatchRunOpsWriter( return shard.writer; } - const onNew = await deps.newReplica.batchTaskRun.findFirst({ + const onNew = await deps.newWriter.batchTaskRun.findFirst({ where: { id: batchId }, select: { id: true }, }); @@ -119,7 +119,6 @@ export const QUEUE_SIZE_LIMIT_EXCEEDED_ERROR_CODE = "QUEUE_SIZE_LIMIT_EXCEEDED"; export type BatchCompletionDeps = { splitEnabled: boolean; - newReplica: RunOpsPrismaClient; newWriter: RunOpsPrismaClient; legacyWriter: RunOpsPrismaClient; shards?: ReadonlyArray<{ key: string; writer: RunOpsPrismaClient }>; @@ -150,7 +149,6 @@ export async function handleBatchCompletion( // Always probe residency — never special-case on splitEnabled (see commit msg). const runOpsWriter = await resolveBatchRunOpsWriter(batchId, { - newReplica: deps.newReplica, newWriter: deps.newWriter, legacyWriter: deps.legacyWriter, shards: deps.shards, diff --git a/apps/webapp/test/runEngineHandlers.test.ts b/apps/webapp/test/runEngineHandlers.test.ts index 63f7e88505f..d9a25dd1135 100644 --- a/apps/webapp/test/runEngineHandlers.test.ts +++ b/apps/webapp/test/runEngineHandlers.test.ts @@ -117,7 +117,6 @@ async function seedBatch( function makeBatchDeps( overrides: { splitEnabled?: boolean; - newReplica?: PrismaClient; newWriter?: PrismaClient; legacyWriter?: PrismaClient; legacyReplica?: PrismaClient; @@ -127,7 +126,6 @@ function makeBatchDeps( const tryCompleteBatchCalls: string[] = []; return { splitEnabled: overrides.splitEnabled ?? false, - newReplica: (overrides.newReplica ?? single)!, newWriter: (overrides.newWriter ?? single)!, legacyWriter: (overrides.legacyWriter ?? single)!, tryCompleteBatch: async (batchId: string) => { @@ -508,7 +506,6 @@ describe("runEngineHandlers batch residency routing", () => { const shards = [{ key: "a", writer: prisma14 }] as const; const writer = await resolveBatchRunOpsWriter(gen2BatchId, { - newReplica: prisma17, newWriter: prisma17, legacyWriter: prisma17, shards: shards as never, @@ -526,7 +523,6 @@ describe("runEngineHandlers batch residency routing", () => { }, { splitEnabled: true, - newReplica: prisma17, newWriter: prisma17, legacyWriter: prisma17, shards: shards as never, @@ -555,13 +551,6 @@ describe("runEngineHandlers batch residency routing", () => { const shardWriter = {} as never; const writer = await resolveBatchRunOpsWriter(`${"a".repeat(24)}a2`, { - newReplica: { - batchTaskRun: { - findFirst: async () => { - throw new Error("a gen-2 batch id must never probe the NEW store"); - }, - }, - } as never, newWriter: {} as never, legacyWriter: {} as never, shards: [{ key: "a", writer: shardWriter as never }], @@ -573,7 +562,6 @@ describe("runEngineHandlers batch residency routing", () => { it("an unconfigured shard key fails loud rather than writing elsewhere", async () => { await expect( resolveBatchRunOpsWriter(`${"a".repeat(24)}z2`, { - newReplica: {} as never, newWriter: {} as never, legacyWriter: {} as never, shards: [{ key: "a", writer: {} as never }], @@ -581,7 +569,7 @@ describe("runEngineHandlers batch residency routing", () => { ).rejects.toThrow(/shard/i); }); - // True single-DB invariant: the topology's cpFallback makes newReplica and + // True single-DB invariant: the topology's cpFallback makes newWriter and // legacyWriter the SAME control-plane client, so the probe always resolves to // that one client regardless of where length-classification would guess. containerTest("true single-DB resolves to the single client", async ({ prisma }) => { @@ -594,7 +582,6 @@ describe("runEngineHandlers batch residency routing", () => { }); const writer = await resolveBatchRunOpsWriter(batchId, { - newReplica: prisma, newWriter: prisma, legacyWriter: prisma, }); @@ -616,7 +603,6 @@ describe("runEngineHandlers batch residency routing", () => { // The probe misses on new (the new DB has no such batch) and resolves the legacy writer. const writer = await resolveBatchRunOpsWriter(batchId, { - newReplica: prisma17, newWriter: prisma17, legacyWriter: prisma14, }); @@ -624,7 +610,6 @@ describe("runEngineHandlers batch residency routing", () => { const deps: BatchCompletionDeps = { splitEnabled: true, - newReplica: prisma17, newWriter: prisma17, legacyWriter: prisma14, tryCompleteBatch: async () => {}, @@ -674,7 +659,6 @@ describe("runEngineHandlers batch residency routing", () => { const deps: BatchCompletionDeps = { splitEnabled: false, - newReplica: prisma17, newWriter: prisma17, legacyWriter: prisma14, tryCompleteBatch: async () => {}, @@ -721,7 +705,6 @@ describe("runEngineHandlers batch residency routing", () => { }); const writer = await resolveBatchRunOpsWriter(batchId, { - newReplica: prisma17, newWriter: prisma17, legacyWriter: prisma14, }); @@ -729,7 +712,6 @@ describe("runEngineHandlers batch residency routing", () => { const deps: BatchCompletionDeps = { splitEnabled: true, - newReplica: prisma17, newWriter: prisma17, legacyWriter: prisma14, tryCompleteBatch: async () => {}, diff --git a/internal-packages/run-engine/src/engine/systems/batchSystem.test.ts b/internal-packages/run-engine/src/engine/systems/batchSystem.test.ts index defd36f6147..44964d46bd0 100644 --- a/internal-packages/run-engine/src/engine/systems/batchSystem.test.ts +++ b/internal-packages/run-engine/src/engine/systems/batchSystem.test.ts @@ -113,7 +113,8 @@ class CountingPostgresRunStore extends PostgresRunStore { async function driveBatchToAllChildrenComplete( engine: RunEngine, prisma: PrismaClient, - friendlyPrefix: string + friendlyPrefix: string, + completeChildren = true ) { const environment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION"); const parentTask = "parent-task"; @@ -183,7 +184,7 @@ async function driveBatchToAllChildrenComplete( queue: `task/${childTask}`, isTest: false, tags: [], - resumeParentOnCompletion: true, + resumeParentOnCompletion: completeChildren, parentTaskRunId: parentRun.id, batch: { id: batch.id, index: 0 }, }, @@ -206,39 +207,41 @@ async function driveBatchToAllChildrenComplete( queue: `task/${childTask}`, isTest: false, tags: [], - resumeParentOnCompletion: true, + resumeParentOnCompletion: completeChildren, parentTaskRunId: parentRun.id, batch: { id: batch.id, index: 1 }, }, prisma ); - for (const child of [child1, child2]) { + if (completeChildren) { + for (const child of [child1, child2]) { + await setTimeout(500); + const dequeued = await engine.dequeueFromWorkerQueue({ + consumerId: "test_consumer", + workerQueue: "main", + }); + const match = dequeued.find((d) => d.run.id === child.id) ?? dequeued[0]; + assertNonNullable(match); + const attempt = await engine.startRunAttempt({ + runId: match.run.id, + snapshotId: match.snapshot.id, + }); + await engine.completeRunAttempt({ + runId: attempt.run.id, + snapshotId: attempt.snapshot.id, + completion: { + id: attempt.run.id, + ok: true, + output: '{"foo":"bar"}', + outputType: "application/json", + }, + }); + } + await setTimeout(500); - const dequeued = await engine.dequeueFromWorkerQueue({ - consumerId: "test_consumer", - workerQueue: "main", - }); - const match = dequeued.find((d) => d.run.id === child.id) ?? dequeued[0]; - assertNonNullable(match); - const attempt = await engine.startRunAttempt({ - runId: match.run.id, - snapshotId: match.snapshot.id, - }); - await engine.completeRunAttempt({ - runId: attempt.run.id, - snapshotId: attempt.snapshot.id, - completion: { - id: attempt.run.id, - ok: true, - output: '{"foo":"bar"}', - outputType: "application/json", - }, - }); } - await setTimeout(500); - return { environment, batch, parentRun, child1, child2 }; } @@ -308,6 +311,57 @@ describe("RunEngine #tryCompleteBatch store routing", () => { } ); + containerTest( + "retries waitpoint completion for a completed batch whose parent was not resumed", + async ({ prisma, redisOptions }) => { + const engine = new RunEngine(createEngineOptions(redisOptions, prisma)); + + try { + const { batch, parentRun } = await driveBatchToAllChildrenComplete( + engine, + prisma, + "run_batch_completion_recovery", + false + ); + + const waitpoint = await prisma.waitpoint.findFirstOrThrow({ + where: { completedByBatchId: batch.id }, + }); + expect(waitpoint.status).toBe("PENDING"); + expect( + await prisma.taskRunWaitpoint.count({ + where: { taskRunId: parentRun.id, waitpointId: waitpoint.id }, + }) + ).toBe(1); + + await prisma.batchTaskRun.update({ + where: { id: batch.id }, + data: { status: "COMPLETED", resumedAt: null }, + }); + + await engine.batchSystem.performCompleteBatch({ batchId: batch.id }); + + const recoveredBatch = await prisma.batchTaskRun.findFirstOrThrow({ + where: { id: batch.id }, + }); + expect(recoveredBatch.resumedAt).not.toBeNull(); + + const recoveredWaitpoint = await prisma.waitpoint.findFirstOrThrow({ + where: { id: waitpoint.id }, + }); + expect(recoveredWaitpoint.status).toBe("COMPLETED"); + + await setTimeout(1_000); + expect(await prisma.taskRunWaitpoint.count({ where: { taskRunId: parentRun.id } })).toBe(0); + const parentExecution = await engine.getRunExecutionData({ runId: parentRun.id }); + assertNonNullable(parentExecution); + expect(parentExecution.snapshot.executionStatus).not.toBe("EXECUTING_WITH_WAITPOINTS"); + } finally { + await engine.quit(); + } + } + ); + // The member-run read is driven by batchId only and does not rely on the // BatchTaskRun.runtimeEnvironmentId FK. A second batch (distinct batchId) must not leak members // into the first batch's batchId-scoped read. diff --git a/internal-packages/run-engine/src/engine/systems/batchSystem.ts b/internal-packages/run-engine/src/engine/systems/batchSystem.ts index 6b100c5f82c..dbd9149e1ff 100644 --- a/internal-packages/run-engine/src/engine/systems/batchSystem.ts +++ b/internal-packages/run-engine/src/engine/systems/batchSystem.ts @@ -170,7 +170,17 @@ export class BatchSystem { } if (batch.status === "COMPLETED") { - this.$.logger.debug("#tryCompleteBatch: Batch already completed", { batchId }); + if (batch.resumedAt) { + this.$.logger.debug("#tryCompleteBatch: Batch already completed and resumed", { + batchId, + }); + return; + } + + this.$.logger.debug("#tryCompleteBatch: Batch completed before its parent resumed", { + batchId, + }); + await this.#resumeBatchParent(batchId); return; } @@ -230,33 +240,40 @@ export class BatchSystem { return; } - //get waitpoint (if there is one) - const waitpoint = await this.$.runStore.findWaitpoint( - { - where: { - completedByBatchId: batchId, - }, - }, - this.$.prisma - ); - - if (!waitpoint) { - this.$.logger.debug( - "RunEngine.unblockRunForBatch(): Waitpoint not found. This is ok, because only batchTriggerAndWait has waitpoints", - { - batchId, - } - ); - return; - } - - await this.waitpointSystem.completeWaitpoint({ - id: waitpoint.id, - output: { value: "Batch waitpoint completed", isError: false }, - }); + await this.#resumeBatchParent(batchId); } else { this.$.logger.debug("#tryCompleteBatch: Not all runs are completed", { batchId }); } }); } + + async #resumeBatchParent(batchId: string): Promise { + const waitpoint = await this.$.runStore.findWaitpoint( + { + where: { completedByBatchId: batchId }, + }, + this.$.prisma + ); + + if (!waitpoint) { + this.$.logger.debug( + "RunEngine.unblockRunForBatch(): Waitpoint not found. This is ok, because only batchTriggerAndWait has waitpoints", + { batchId } + ); + return; + } + + await this.waitpointSystem.completeWaitpoint({ + id: waitpoint.id, + output: { value: "Batch waitpoint completed", isError: false }, + }); + + await this.$.runStore.updateManyBatchTaskRun( + { + where: { id: batchId, resumedAt: null }, + data: { resumedAt: new Date() }, + }, + this.$.prisma + ); + } } From e6ec2c4e8d5450668c95eaf84c1567c9eb9abb87 Mon Sep 17 00:00:00 2001 From: Chris Arderne Date: Mon, 28 Sep 2026 11:54:14 +0000 Subject: [PATCH 7/9] fix(sdk): re-suspend chat agents after unmatched wakes Chat agents now return to a durable wait when session activity does not deliver a matching message. Repeated wakes preserve the original turn timeout instead of leaving the run active until its maximum duration. Mono-RevId: f9bb2238bbd2ea663eaac75ce3d92106a83c1104 --- .changeset/chat-route-spurious-wakes.md | 5 + packages/trigger-sdk/src/v3/ai.ts | 37 +++---- .../trigger-sdk/src/v3/chatRouteWait.test.ts | 101 ++++++++++++++++++ packages/trigger-sdk/src/v3/chatRouteWait.ts | 64 +++++++++++ 4 files changed, 184 insertions(+), 23 deletions(-) create mode 100644 .changeset/chat-route-spurious-wakes.md create mode 100644 packages/trigger-sdk/src/v3/chatRouteWait.test.ts create mode 100644 packages/trigger-sdk/src/v3/chatRouteWait.ts diff --git a/.changeset/chat-route-spurious-wakes.md b/.changeset/chat-route-spurious-wakes.md new file mode 100644 index 00000000000..cd247801525 --- /dev/null +++ b/.changeset/chat-route-spurious-wakes.md @@ -0,0 +1,5 @@ +--- +"@trigger.dev/sdk": patch +--- + +Chat agents now return to a durable wait when a session wake does not deliver a matching message. This prevents resumed runs from staying active until their maximum duration and preserves the configured turn timeout across repeated wakes. diff --git a/packages/trigger-sdk/src/v3/ai.ts b/packages/trigger-sdk/src/v3/ai.ts index 8113f89ee3b..0bc2e6bbb7e 100644 --- a/packages/trigger-sdk/src/v3/ai.ts +++ b/packages/trigger-sdk/src/v3/ai.ts @@ -66,6 +66,7 @@ import type { // ESM-only `ai@7` (see ../imports/ai-runtime.ts). import { type Attributes, trace } from "@opentelemetry/api"; import { traceSessionIdle } from "./sessionTracing.js"; +import { waitForChatRouteAfterIdle } from "./chatRouteWait.js"; import { tool as aiTool, convertToModelMessages, @@ -1780,31 +1781,21 @@ async function waitOnChatRoute( if (options.onSuspend) await options.onSuspend(); span.setAttribute("wait.resolved", "suspended"); - while (true) { - /** - * The floor doubles as the wake cursor: the server completes the - * waitpoint immediately if anything sits after this sequence, so a - * floor that has advanced past an unread record parks a waitpoint - * nothing will complete. Recorded on the span so a run that never woke - * can be diagnosed from its trace alone. - */ - const wakeFrom = router.resumeFloor(); - span.setAttribute("wait.lastSeqNum", wakeFrom ?? -1); - const wake = await session.in.awaitWake({ - timeout: options.timeout, - lastSeqNum: wakeFrom, - }); - if (!wake.ok) { - span.recordException(wake.error); - return { ok: false as const, error: wake.error }; - } - - const record = await router.next(route); - if (!record) continue; + const result = await waitForChatRouteAfterIdle(router, route, { + timeout: options.timeout, + wake: async (timeout, lastSeqNum) => { + span.setAttribute("wait.lastSeqNum", lastSeqNum ?? -1); + return session.in.awaitWake({ timeout, lastSeqNum }); + }, + }); - if (options.onResume) await options.onResume(); - return { ok: true as const, output: record.data as T, record }; + if (!result.ok) { + span.recordException(result.error); + return result; } + + if (options.onResume) await options.onResume(); + return { ok: true as const, output: result.record.data as T, record: result.record }; }, { attributes: { diff --git a/packages/trigger-sdk/src/v3/chatRouteWait.test.ts b/packages/trigger-sdk/src/v3/chatRouteWait.test.ts new file mode 100644 index 00000000000..5dac1ed1968 --- /dev/null +++ b/packages/trigger-sdk/src/v3/chatRouteWait.test.ts @@ -0,0 +1,101 @@ +import { SessionChannelRouter, WaitpointTimeoutError } from "@trigger.dev/core/v3"; +import { describe, expect, it } from "vitest"; +import { waitForChatRouteAfterIdle } from "./chatRouteWait.js"; + +function inputRouter() { + return new SessionChannelRouter({ + kindOf: (data) => (data as { kind?: string }).kind, + routes: [ + { name: "messages", delivery: "queue", replayable: true, kinds: ["message"] }, + { name: "control", delivery: "queue", replayable: false, kinds: ["control"] }, + ], + }); +} + +describe("chat route waits after suspension", () => { + it("re-suspends after a wake that has no record for the requested route", async () => { + const router = inputRouter(); + const timeouts: Array = []; + let now = 0; + + const result = await waitForChatRouteAfterIdle(router, "messages", { + timeout: "10s", + postWakeTimeoutMs: 0, + now: () => now, + wake: async (timeout) => { + timeouts.push(timeout); + if (timeouts.length === 1) { + router.ingest({ id: "control-1", seqNum: 1, data: { kind: "control" } }); + now = 4_000; + return { ok: true, waitpointId: "waitpoint-1" }; + } + return { ok: false, error: new WaitpointTimeoutError("Timed out") }; + }, + }); + + expect(result.ok).toBe(false); + expect(timeouts).toEqual(["10s", "6s"]); + }); + + it("does not re-wake on an unrelated queued message while waiting for another route", async () => { + const router = inputRouter(); + const cursors: Array = []; + + const result = await waitForChatRouteAfterIdle(router, "control", { + postWakeTimeoutMs: 0, + wake: async (_timeout, lastSeqNum) => { + cursors.push(lastSeqNum); + if (cursors.length === 1) { + router.ingest({ id: "message-1", seqNum: 1, data: { kind: "message" } }); + return { ok: true, waitpointId: "waitpoint-1" }; + } + if (lastSeqNum === undefined || lastSeqNum < 1) { + return { ok: true, waitpointId: "waitpoint-repeated" }; + } + return { ok: false, error: new WaitpointTimeoutError("Timed out") }; + }, + }); + + expect(result.ok).toBe(false); + expect(cursors).toEqual([undefined, 1]); + expect(router.resumeFloor()).toBe(0); + expect(router.hasPending("messages")).toBe(true); + }); + + it("expires an absolute timeout after an unmatched wake", async () => { + const router = inputRouter(); + const deadline = Date.parse("2026-09-28T10:15:00Z"); + let now = deadline - 1_000; + const timeouts: Array = []; + + const result = await waitForChatRouteAfterIdle(router, "messages", { + timeout: "2026-09-28T10:15:00Z", + postWakeTimeoutMs: 0, + now: () => now, + wake: async (timeout) => { + timeouts.push(timeout); + now = deadline + 1; + return { ok: true, waitpointId: "waitpoint-1" }; + }, + }); + + expect(result.ok).toBe(false); + expect(timeouts).toEqual(["1s"]); + }); + + it("returns a matching record delivered after the channel wakes", async () => { + const router = inputRouter(); + const message = { id: "message-1", seqNum: 1, data: { kind: "message", text: "hello" } }; + + const result = await waitForChatRouteAfterIdle(router, "messages", { + timeout: "10s", + postWakeTimeoutMs: 0, + wake: async () => { + router.ingest(message); + return { ok: true, waitpointId: "waitpoint-1" }; + }, + }); + + expect(result).toEqual({ ok: true, record: message }); + }); +}); diff --git a/packages/trigger-sdk/src/v3/chatRouteWait.ts b/packages/trigger-sdk/src/v3/chatRouteWait.ts new file mode 100644 index 00000000000..b489c25e500 --- /dev/null +++ b/packages/trigger-sdk/src/v3/chatRouteWait.ts @@ -0,0 +1,64 @@ +import { + type SessionChannelRouter, + type SessionStreamRecord, + WaitpointTimeoutError, +} from "@trigger.dev/core/v3"; +import { parseNaturalLanguageDurationInMs } from "@trigger.dev/core/v3/isomorphic"; + +const POST_WAKE_TIMEOUT_MS = 5_000; + +type ChatRouteWakeResult = { ok: true; waitpointId: string } | { ok: false; error: Error }; + +export async function waitForChatRouteAfterIdle( + router: SessionChannelRouter, + route: string, + options: { + timeout?: string; + postWakeTimeoutMs?: number; + now?: () => number; + wake: ( + timeout: string | undefined, + lastSeqNum: number | undefined + ) => Promise; + } +): Promise<{ ok: true; record: SessionStreamRecord } | { ok: false; error: Error }> { + const now = options.now ?? Date.now; + const timeoutMs = options.timeout ? parseNaturalLanguageDurationInMs(options.timeout) : undefined; + const parsedDeadline = + timeoutMs !== undefined + ? now() + timeoutMs + : options.timeout === undefined + ? undefined + : Date.parse(options.timeout); + const deadline = + parsedDeadline !== undefined && Number.isFinite(parsedDeadline) ? parsedDeadline : undefined; + const postWakeTimeoutMs = options.postWakeTimeoutMs ?? POST_WAKE_TIMEOUT_MS; + + while (true) { + const remainingMs = deadline === undefined ? undefined : deadline - now(); + if (remainingMs !== undefined && remainingMs <= 0) { + return { ok: false, error: new WaitpointTimeoutError("Timed out") }; + } + + const wakeTimeout = + remainingMs === undefined + ? options.timeout + : `${Math.max(1, Math.ceil(remainingMs / 1000))}s`; + // A queued record holds the replay floor back, but should not wake this + // waitpoint again after we have already seen it. Snapshot before checking + // the route so a record arriving during registration still wakes us. + const wakeFrom = router.appliedThrough(); + const buffered = await router.next(route, { timeoutMs: 0 }); + if (buffered) return { ok: true, record: buffered }; + + const wake = await options.wake(wakeTimeout, wakeFrom); + if (!wake.ok) return wake; + + const deliveryBudget = + deadline === undefined + ? postWakeTimeoutMs + : Math.max(0, Math.min(postWakeTimeoutMs, deadline - now())); + const record = await router.next(route, { timeoutMs: deliveryBudget }); + if (record) return { ok: true, record }; + } +} From a3f3fc09b5fa6d0409ae40a33de8a57cbc5b04f2 Mon Sep 17 00:00:00 2001 From: Chris Arderne Date: Mon, 28 Sep 2026 12:15:53 +0000 Subject: [PATCH 8/9] fix: simplify onboarding questions Simplify project creation by removing optional onboarding questions. Mono-RevId: a04b3112a96c1dcf6d20aaf9892339f026e1f3e7 --- .../simplify-onboarding-questions.md | 6 + .../onboarding/TechnologyPicker.tsx | 386 ------------------ .../route.tsx | 249 +---------- 3 files changed, 7 insertions(+), 634 deletions(-) create mode 100644 .server-changes/simplify-onboarding-questions.md delete mode 100644 apps/webapp/app/components/onboarding/TechnologyPicker.tsx diff --git a/.server-changes/simplify-onboarding-questions.md b/.server-changes/simplify-onboarding-questions.md new file mode 100644 index 00000000000..6f0153d5b51 --- /dev/null +++ b/.server-changes/simplify-onboarding-questions.md @@ -0,0 +1,6 @@ +--- +area: webapp +type: improvement +--- + +Creating a project now asks only for its name. diff --git a/apps/webapp/app/components/onboarding/TechnologyPicker.tsx b/apps/webapp/app/components/onboarding/TechnologyPicker.tsx deleted file mode 100644 index e70af0d7030..00000000000 --- a/apps/webapp/app/components/onboarding/TechnologyPicker.tsx +++ /dev/null @@ -1,386 +0,0 @@ -import * as Ariakit from "@ariakit/react"; -import { - XMarkIcon, - PlusIcon, - CubeIcon, - MagnifyingGlassIcon, - ChevronDownIcon, -} from "@heroicons/react/20/solid"; -import { useCallback, useMemo, useRef, useState } from "react"; -import { CheckboxIndicator } from "~/components/primitives/CheckboxIndicator"; -import { cn } from "~/utils/cn"; -import { matchSorter } from "match-sorter"; -import { ShortcutKey } from "~/components/primitives/ShortcutKey"; - -const pillColors = [ - "bg-green-800/40 border-green-600/50", - "bg-teal-800/40 border-teal-600/50", - "bg-blue-800/40 border-blue-600/50", - "bg-indigo-800/40 border-indigo-600/50", - "bg-violet-800/40 border-violet-600/50", - "bg-purple-800/40 border-purple-600/50", - "bg-fuchsia-800/40 border-fuchsia-600/50", - "bg-pink-800/40 border-pink-600/50", - "bg-rose-800/40 border-rose-600/50", - "bg-orange-800/40 border-orange-600/50", - "bg-amber-800/40 border-amber-600/50", - "bg-yellow-800/40 border-yellow-600/50", - "bg-lime-800/40 border-lime-600/50", - "bg-emerald-800/40 border-emerald-600/50", - "bg-cyan-800/40 border-cyan-600/50", - "bg-sky-800/40 border-sky-600/50", -]; - -function getPillColor(value: string): string { - let hash = 0; - for (let i = 0; i < value.length; i++) { - hash = (hash << 5) - hash + value.charCodeAt(i); - hash |= 0; - } - return pillColors[Math.abs(hash) % pillColors.length]; -} - -const TECHNOLOGY_OPTIONS = [ - "Airflow", - "Angular", - "Anthropic", - "Astro", - "Auth0", - "AWS", - "AWS SQS", - "Azure", - "BigQuery", - "BullMQ", - "Bun", - "Cassandra", - "Celery", - "ClickHouse", - "Clerk", - "Cloudflare", - "CockroachDB", - "Cohere", - "Convex", - "Databricks", - "Datadog", - "DeepSeek", - "Deno", - "DigitalOcean", - "Django", - "Docker", - "Drizzle", - "DynamoDB", - "Elasticsearch", - "Electron", - "Elevenlabs", - "Expo", - "Express", - "FastAPI", - "Fastify", - "Firebase", - "Flask", - "Fly.io", - "Gatsby", - "GCP", - "Go", - "Google Cloud Tasks", - "Google Gemini", - "GraphQL", - "Groq", - "Heroku", - "Hono", - "htmx", - "Hugging Face", - "Inngest", - "Kafka", - "Kubernetes", - "LangChain", - "Laravel", - "LlamaIndex", - "MariaDB", - "Midjourney", - "Mistral", - "MongoDB", - "Mongoose", - "MySQL", - "Neo4j", - "Neon", - "Nest.js", - "Netlify", - "Next.js", - "Node.js", - "Nuxt", - "Ollama", - "OpenAI", - "Perplexity", - "PHP", - "Pinecone", - "PlanetScale", - "Python", - "PostHog", - "PostgreSQL", - "Prisma", - "Pulumi", - "RabbitMQ", - "Railway", - "React", - "React Native", - "Redis", - "Redshift", - "Remix", - "Render", - "Replicate", - "Resend", - "Ruby on Rails", - "Rust", - "SendGrid", - "Sentry", - "Sidekiq", - "Snowflake", - "Solid.js", - "Spring Boot", - "SQLite", - "Stability AI", - "Stripe", - "Supabase", - "Svelte", - "SvelteKit", - "Tailwind CSS", - "Temporal", - "Terraform", - "Together AI", - "tRPC", - "Turso", - "Twilio", - "TypeORM", - "Upstash", - "Vercel", - "Vercel AI SDK", - "Vite", - "Vue", - "Weaviate", -] as const; - -type TechnologyPickerProps = { - value: string[]; - onChange: (value: string[]) => void; - customValues: string[]; - onCustomValuesChange: (values: string[]) => void; -}; - -export function TechnologyPicker({ - value, - onChange, - customValues, - onCustomValuesChange, -}: TechnologyPickerProps) { - const [open, setOpen] = useState(false); - const [searchValue, setSearchValue] = useState(""); - const [otherInputValue, setOtherInputValue] = useState(""); - const [showOtherInput, setShowOtherInput] = useState(false); - const otherInputRef = useRef(null); - - const allSelected = useMemo(() => [...value, ...customValues], [value, customValues]); - - const filteredOptions = useMemo(() => { - if (!searchValue) return TECHNOLOGY_OPTIONS; - return matchSorter([...TECHNOLOGY_OPTIONS], searchValue); - }, [searchValue]); - - const toggleOption = useCallback( - (option: string) => { - if (value.includes(option)) { - onChange(value.filter((v) => v !== option)); - } else { - onChange([...value, option]); - } - }, - [value, onChange] - ); - - const removeItem = useCallback( - (item: string) => { - if (value.includes(item)) { - onChange(value.filter((v) => v !== item)); - } else { - onCustomValuesChange(customValues.filter((v) => v !== item)); - } - }, - [value, onChange, customValues, onCustomValuesChange] - ); - - const addCustomValue = useCallback(() => { - const trimmed = otherInputValue.trim(); - if (!trimmed) return; - - const matchedOption = TECHNOLOGY_OPTIONS.find( - (opt) => opt.toLowerCase() === trimmed.toLowerCase() - ); - - if (matchedOption) { - if (!value.includes(matchedOption)) { - onChange([...value, matchedOption]); - } - } else if (!customValues.includes(trimmed) && !value.includes(trimmed)) { - onCustomValuesChange([...customValues, trimmed]); - } - - setOtherInputValue(""); - }, [otherInputValue, customValues, onCustomValuesChange, value, onChange]); - - const handleOtherKeyDown = useCallback( - (e: React.KeyboardEvent) => { - if (e.key === "Enter") { - e.preventDefault(); - e.stopPropagation(); - addCustomValue(); - } - }, - [addCustomValue] - ); - - return ( -
- {allSelected.length > 0 && ( -
- {allSelected.map((item) => ( - - {item} - - - ))} -
- )} - - { - setSearchValue(val); - }} - > - { - if (Array.isArray(v)) { - onChange(v); - } - }} - virtualFocus - > - -
- - Select your technologies… -
- - - - -
- - -
- - - {filteredOptions.map((option) => ( - { - e.preventDefault(); - toggleOption(option); - }} - > -
- - {option} -
- - ))} - - {filteredOptions.length === 0 && searchValue && ( -
- No matches for “{searchValue}” -
- )} - - -
- {showOtherInput ? ( -
- setOtherInputValue(e.target.value)} - onKeyDown={handleOtherKeyDown} - placeholder="Type and press Enter to add" - className="flex-1 border-none bg-transparent pl-2 text-2sm text-text-bright shadow-none outline-hidden ring-0 placeholder:text-text-dimmed focus:border-none focus:outline-hidden focus:ring-0" - autoFocus - /> - 0 ? "opacity-100" : "opacity-0" - )} - /> - -
- ) : ( - - )} -
- - - -
- ); -} diff --git a/apps/webapp/app/routes/_app.orgs.$organizationSlug_.projects.new/route.tsx b/apps/webapp/app/routes/_app.orgs.$organizationSlug_.projects.new/route.tsx index 397a5eeb953..550cb0d39f4 100644 --- a/apps/webapp/app/routes/_app.orgs.$organizationSlug_.projects.new/route.tsx +++ b/apps/webapp/app/routes/_app.orgs.$organizationSlug_.projects.new/route.tsx @@ -1,6 +1,6 @@ import { getFormProps, getInputProps, useForm } from "@conform-to/react"; import { parseWithZod } from "@conform-to/zod/v4"; -import { CommandLineIcon, FolderIcon } from "@heroicons/react/20/solid"; +import { FolderIcon } from "@heroicons/react/20/solid"; import { json, redirectDocument, @@ -8,15 +8,12 @@ import { type LoaderFunctionArgs, } from "@remix-run/node"; import { Form, useActionData, useNavigation } from "@remix-run/react"; -import type { Prisma } from "@trigger.dev/database"; -import React, { useEffect, useState } from "react"; import { redirect, typedjson, useTypedLoaderData } from "remix-typedjson"; import invariant from "tiny-invariant"; import { z } from "zod"; import { BackgroundWrapper } from "~/components/BackgroundWrapper"; import { Feedback } from "~/components/Feedback"; import { AppContainer, MainCenteredContainer } from "~/components/layout/AppLayout"; -import { TechnologyPicker } from "~/components/onboarding/TechnologyPicker"; import { Button, LinkButton } from "~/components/primitives/Buttons"; import { Callout } from "~/components/primitives/Callout"; import { Fieldset } from "~/components/primitives/Fieldset"; @@ -26,7 +23,6 @@ import { FormTitle } from "~/components/primitives/FormTitle"; import { Input } from "~/components/primitives/Input"; import { InputGroup } from "~/components/primitives/InputGroup"; import { Label } from "~/components/primitives/Label"; -import { Select, SelectItem } from "~/components/primitives/Select"; import { prisma } from "~/db.server"; import { featuresForRequest } from "~/features.server"; @@ -45,82 +41,6 @@ import { pageMeta } from "~/utils/pageTitle"; export const meta = pageMeta("New project"); -const WORKING_ON_OTHER = "Other/not sure yet"; -const GOALS_OTHER = "Other/not sure yet"; - -const workingOnOptions = [ - "AI agent", - "Media processing pipeline", - "Media generation with AI", - "Event-driven workflow", - "Realtime streaming", - "Internal tool or background job", - WORKING_ON_OTHER, -] as const; - -const goalOptions = [ - "Ship a production workflow", - "Prototype or explore", - "Migrate an existing system", - "Learn how Trigger works", - "Evaluate against alternatives", - GOALS_OTHER, -] as const; - -function shuffleArray(arr: T[]): T[] { - const shuffled = [...arr]; - for (let i = shuffled.length - 1; i > 0; i--) { - const j = Math.floor(Math.random() * (i + 1)); - [shuffled[i], shuffled[j]] = [shuffled[j], shuffled[i]]; - } - return shuffled; -} - -function renderMultiSelectValue(value: string[]) { - if (value.length === 0) return; - - return ( - - {value.slice(0, 2).join(", ")} - {value.length > 2 && +{value.length - 2} more} - - ); -} - -function MultiSelectField({ - value, - setValue, - items, - icon, -}: { - value: string[]; - setValue: (value: string[]) => void; - items: string[]; - icon: React.ReactNode; -}) { - return ( - - value={value} - setValue={setValue} - placeholder="Select some options" - variant="secondary/small" - dropdownIcon - icon={icon} - items={items} - className="h-8 min-w-0 border-0 bg-background-hover pl-2 text-sm text-text-dimmed ring-border-bright transition hover:bg-secondary hover:text-text-dimmed hover:ring-1" - text={renderMultiSelectValue} - > - {(items) => - items.map((item) => ( - - {item} - - )) - } - - ); -} - export async function loader({ params, request }: LoaderFunctionArgs) { const userId = await requireUserId(request); const { organizationSlug } = OrganizationParamsSchema.parse(params); @@ -171,14 +91,6 @@ export async function loader({ params, request }: LoaderFunctionArgs) { const schema = z.object({ projectName: z.string().min(3, "Project name must have at least 3 characters").max(50), projectVersion: z.enum(["v2", "v3"]), - workingOn: z.string().optional(), - workingOnOther: z.string().optional(), - technologies: z.string().optional(), - technologiesOther: z.string().optional(), - goals: z.string().optional(), - goalsOther: z.string().optional(), - workingOnPositions: z.string().optional(), - goalsPositions: z.string().optional(), }); export const action: ActionFunction = async ({ request, params }) => { @@ -198,66 +110,12 @@ export const action: ActionFunction = async ({ request, params }) => { const configurationId = url.searchParams.get("configurationId"); const next = url.searchParams.get("next"); - const stringArraySchema = z.array(z.string()); - - function safeParseStringArray(value: string | undefined): string[] | undefined { - if (!value) return undefined; - try { - const result = stringArraySchema.safeParse(JSON.parse(value)); - return result.success && result.data.length > 0 ? result.data : undefined; - } catch { - return undefined; - } - } - - const numberArraySchema = z.array(z.number()); - function safeParseNumberArray(value: string | undefined): number[] | undefined { - if (!value) return undefined; - try { - const result = numberArraySchema.safeParse(JSON.parse(value)); - return result.success && result.data.length > 0 ? result.data : undefined; - } catch { - return undefined; - } - } - - const onboardingData: Record = {}; - - const workingOn = safeParseStringArray(submission.value.workingOn); - if (workingOn) { - onboardingData.workingOn = workingOn; - const workingOnPositions = safeParseNumberArray(submission.value.workingOnPositions); - if (workingOnPositions) onboardingData.workingOnPositions = workingOnPositions; - } - - if (submission.value.workingOnOther) { - onboardingData.workingOnOther = submission.value.workingOnOther; - } - - const technologies = safeParseStringArray(submission.value.technologies); - if (technologies) onboardingData.technologies = technologies; - - const technologiesOther = safeParseStringArray(submission.value.technologiesOther); - if (technologiesOther) onboardingData.technologiesOther = technologiesOther; - - const goals = safeParseStringArray(submission.value.goals); - if (goals) { - onboardingData.goals = goals; - const goalsPositions = safeParseNumberArray(submission.value.goalsPositions); - if (goalsPositions) onboardingData.goalsPositions = goalsPositions; - } - - if (submission.value.goalsOther) { - onboardingData.goalsOther = submission.value.goalsOther; - } - try { const project = await createProject({ organizationSlug: organizationSlug, name: submission.value.projectName, userId, version: submission.value.projectVersion, - onboardingData: Object.keys(onboardingData).length > 0 ? onboardingData : undefined, }); if (code && configurationId) { @@ -348,28 +206,6 @@ export default function Page() { const navigation = useNavigation(); const isLoading = navigation.state === "submitting" || navigation.state === "loading"; - const [selectedWorkingOn, setSelectedWorkingOn] = useState([]); - const [workingOnOther, setWorkingOnOther] = useState(""); - const [selectedTechnologies, setSelectedTechnologies] = useState([]); - const [customTechnologies, setCustomTechnologies] = useState([]); - const [selectedGoals, setSelectedGoals] = useState([]); - const [goalsOther, setGoalsOther] = useState(""); - - const [shuffledWorkingOn, setShuffledWorkingOn] = useState([...workingOnOptions]); - const [shuffledGoals, setShuffledGoals] = useState([...goalOptions]); - - useEffect(() => { - const nonOther = workingOnOptions.filter((o) => o !== WORKING_ON_OTHER); - // oxlint-disable-next-line react/set-state-in-effect -- This effect intentionally synchronizes route state after an external or lifecycle change. - setShuffledWorkingOn([...shuffleArray(nonOther), WORKING_ON_OTHER]); - - const nonOtherGoals = goalOptions.filter((o) => o !== GOALS_OTHER); - setShuffledGoals([...shuffleArray(nonOtherGoals), GOALS_OTHER]); - }, []); - - const showWorkingOnOther = selectedWorkingOn.includes(WORKING_ON_OTHER); - const showGoalsOther = selectedGoals.includes(GOALS_OTHER); - return ( @@ -414,89 +250,6 @@ export default function Page() { /> )} -
- - - - shuffledWorkingOn.indexOf(v) + 1) - )} - /> - } - /> - {showWorkingOnOther && ( - <> - - setWorkingOnOther(e.target.value)} - placeholder="Tell us what you're working on" - spellCheck={false} - containerClassName="h-8" - /> - - )} - - - - - - - - - - - - - shuffledGoals.indexOf(v) + 1))} - /> - } - /> - {showGoalsOther && ( - <> - - setGoalsOther(e.target.value)} - placeholder="Tell us what you're trying to do with Trigger.dev" - spellCheck={false} - containerClassName="h-8" - /> - - )} - - From cbf421575f74e28748f37cf06e499b66291800b1 Mon Sep 17 00:00:00 2001 From: Matt Aitken Date: Mon, 28 Sep 2026 12:21:34 +0000 Subject: [PATCH 9/9] fix(webapp): stop the OTLP ingest worker pool reaping healthy workers when queued batches time out ## Summary Under a burst of telemetry, the OpenTelemetry ingest worker pool could terminate its own healthy workers and then keep terminating every replacement, leaving a server rejecting every OTLP batch until the process was restarted. Batches that wait too long in the queue are now dropped individually, workers are only terminated when they are genuinely stuck on a single batch, and every worker termination is logged. ## Root cause The pool starts a task's timeout when the batch is enqueued, but the timeout handler treated any dispatched task as evidence of a stuck worker. When queue wait alone exceeded the budget, the first timer to fire belonged to a batch a healthy worker was actively computing, so that worker was terminated. Each replacement worker was handed the oldest queued batch, whose timer was already about to fire, and was terminated moments later. With respawn backoff at its cap the pool sat at zero workers indefinitely, and the termination path logged nothing. ## Fix - Track when a task is dispatched. On timeout the caller is still rejected, but the worker is only reaped once it has spent a full timeout computing that one task; otherwise a compute timer gives it the remaining budget and the worker returns to the idle set when it replies. Deadline arithmetic uses a monotonic clock so a wall-clock step cannot trigger sheds or reaps. - Before dispatch, shed queued tasks whose remaining budget is below a small floor (recorded as `outcome=stale` on `ingest.worker_pool.tasks`) so a free worker starts on a batch it can finish. - Log every reap with reason, task age, time since dispatch, queue depth and alive workers; aggregate shed counts into at most one debug line per second. - Timeout and respawn backoff are constructor options with unchanged defaults so the overload scenario is testable. The new tests reproduce the failure on the previous code (all respawns `reason=timeout`, zero alive workers, throughput frozen) and cover the late-reply and hang-after-queue-wait paths. Mono-RevId: 69295ed642e15b72d3ee8d15c6fe734e6579b94e --- .../otlp-worker-pool-stale-task-reap.md | 6 + apps/webapp/app/v3/otlpWorkerPool.server.ts | 221 +++++++++++-- apps/webapp/test/fixtures/otlpHangWorker.cjs | 6 + apps/webapp/test/fixtures/otlpSlowWorker.cjs | 22 ++ .../webapp/test/otlpWorkerPoolTimeout.test.ts | 292 ++++++++++++++++++ 5 files changed, 525 insertions(+), 22 deletions(-) create mode 100644 .server-changes/otlp-worker-pool-stale-task-reap.md create mode 100644 apps/webapp/test/fixtures/otlpHangWorker.cjs create mode 100644 apps/webapp/test/fixtures/otlpSlowWorker.cjs create mode 100644 apps/webapp/test/otlpWorkerPoolTimeout.test.ts diff --git a/.server-changes/otlp-worker-pool-stale-task-reap.md b/.server-changes/otlp-worker-pool-stale-task-reap.md new file mode 100644 index 00000000000..f5da2aa269d --- /dev/null +++ b/.server-changes/otlp-worker-pool-stale-task-reap.md @@ -0,0 +1,6 @@ +--- +area: webapp +type: fix +--- + +Fixed a bug where a burst of telemetry could leave OpenTelemetry ingest rejecting every batch for a long time. Batches that wait too long are now dropped individually instead of restarting the processing workers, so ingest recovers as soon as the burst passes. diff --git a/apps/webapp/app/v3/otlpWorkerPool.server.ts b/apps/webapp/app/v3/otlpWorkerPool.server.ts index 51bc0bca831..0d73728d457 100644 --- a/apps/webapp/app/v3/otlpWorkerPool.server.ts +++ b/apps/webapp/app/v3/otlpWorkerPool.server.ts @@ -1,5 +1,6 @@ import { Worker } from "node:worker_threads"; import path from "node:path"; +import { performance } from "node:perf_hooks"; import { getMeter, type Counter, @@ -31,15 +32,40 @@ type Task = { // Wall-clock stamp at enqueue; the task-duration histogram measures enqueue -> terminal state // (queue wait + worker compute), so the gap from the worker-reported compute time is queue wait. enqueuedAt: number; + /** Monotonic stamps (performance.now) drive every deadline decision so a wall-clock step can't shed or reap. */ + enqueuedAtMono: number; + dispatchedAtMono?: number; }; type ReapReason = "error" | "exit" | "timeout"; +export type ReapEvent = { + reason: ReapReason; + taskAgeMs?: number; + sinceDispatchMs?: number; + queueDepth: number; + aliveWorkers: number; +}; + +export type OtlpWorkerPoolOptions = { + taskTimeoutMs?: number; + respawnBaseMs?: number; + respawnMaxMs?: number; + /** + * Observer for every reap, carrying the same fields as the warn log. It must not affect pool + * liveness: it runs after the worker is terminated and the respawn is scheduled, and any + * exception it throws or promise it rejects is logged and swallowed. + */ + onReap?: (event: ReapEvent) => void | Promise; +}; + const TASK_TIMEOUT_MS = 30_000; const MAX_QUEUE_DEPTH = 2_000; const RESPAWN_BASE_MS = 500; const RESPAWN_MAX_MS = 30_000; const SHUTDOWN_DRAIN_MS = 5_000; +const STALE_FLOOR_MAX_MS = 1_000; +const SHED_LOG_INTERVAL_MS = 1_000; // Hand-rolled worker_threads pool: one in-flight task per worker so CPU-bound transforms run // fully in parallel. The main thread stays the only DB reader and broadcasts pricing to workers. @@ -48,11 +74,21 @@ export class OtlpWorkerPool { private readonly idle: Worker[] = []; private readonly queue: number[] = []; private readonly tasks = new Map(); - private readonly busyByWorker = new Map(); + private readonly busyByWorker = new Map(); + private readonly computeTimers = new Map(); private nextId = 1; private consecutiveFailures = 0; private isShuttingDown = false; private latestPricingModels: unknown[]; + private readonly taskTimeoutMs: number; + private readonly respawnBaseMs: number; + private readonly respawnMaxMs: number; + private readonly staleFloorMs: number; + private readonly onReap?: (event: ReapEvent) => void | Promise; + private shedInWindow = 0; + private shedWindowStartMono = 0; + private lastShedMono = 0; + private shedFlushTimer?: NodeJS.Timeout; // Pre-allocated per-kind {kind} attribute objects so the per-task record path never allocates. private readonly _kindAttrs: Record = { @@ -69,9 +105,15 @@ export class OtlpWorkerPool { private readonly size: number, private readonly workerPath: string, pricingModels: unknown[], - meter?: Meter + meter?: Meter, + options?: OtlpWorkerPoolOptions ) { this.latestPricingModels = pricingModels; + this.taskTimeoutMs = options?.taskTimeoutMs ?? TASK_TIMEOUT_MS; + this.respawnBaseMs = options?.respawnBaseMs ?? RESPAWN_BASE_MS; + this.respawnMaxMs = options?.respawnMaxMs ?? RESPAWN_MAX_MS; + this.staleFloorMs = Math.min(STALE_FLOOR_MAX_MS, this.taskTimeoutMs / 10); + this.onReap = options?.onReap; this.#setupOtelMetrics(meter); for (let i = 0; i < size; i++) this.spawn(); logger.info("OtlpWorkerPool started", { size, workerPath }); @@ -131,6 +173,15 @@ export class OtlpWorkerPool { } } + /** + * A task that already timed out for its caller recorded its outcome and duration then; its + * compute time only becomes known when the worker finally replies, and skipping it would drop + * exactly the slow samples from the compute histogram. + */ + #recordLateCompute(task: Task, computeMs: number): void { + this._computeDurationHistogram?.record(computeMs, this._kindAttrs[task.message.kind]); + } + private spawn() { const worker = new Worker(this.workerPath, { workerData: { pricingModels: this.latestPricingModels }, @@ -141,7 +192,9 @@ export class OtlpWorkerPool { (msg: { id: number; ok: boolean; result?: any; error?: string; computeMs?: number }) => { if (this.workers.indexOf(worker) === -1) return; // late message from an already-reaped worker this.consecutiveFailures = 0; + const inFlight = this.busyByWorker.get(worker); this.busyByWorker.delete(worker); + this.#clearComputeTimer(worker); const task = this.tasks.get(msg.id); if (task) { clearTimeout(task.timer); @@ -153,6 +206,8 @@ export class OtlpWorkerPool { this.#recordTaskEnd(task, "error", msg.computeMs); task.reject(new Error(msg.error ?? "otlp worker error")); } + } else if (inFlight?.message.id === msg.id && msg.computeMs !== undefined) { + this.#recordLateCompute(inFlight, msg.computeMs); } this.release(worker); } @@ -182,30 +237,92 @@ export class OtlpWorkerPool { const ii = this.idle.indexOf(worker); if (ii !== -1) this.idle.splice(ii, 1); + this.#clearComputeTimer(worker); - const inFlightId = this.busyByWorker.get(worker); + const now = performance.now(); + const inFlight = this.busyByWorker.get(worker); this.busyByWorker.delete(worker); - if (inFlightId !== undefined) { - const task = this.tasks.get(inFlightId); - if (task) { - clearTimeout(task.timer); - this.tasks.delete(inFlightId); - // A timed-out task already recorded its own end + was removed from the map, so this only - // fires for a crash that killed a task mid-flight. - this.#recordTaskEnd(task, "crash"); - task.reject(error); - } + if (inFlight !== undefined && this.tasks.has(inFlight.message.id)) { + clearTimeout(inFlight.timer); + this.tasks.delete(inFlight.message.id); + this.#recordTaskEnd(inFlight, "crash"); + inFlight.reject(error); } this._respawnsCounter?.add(1, { reason }); + const event: ReapEvent = { + reason, + queueDepth: this.queue.length, + aliveWorkers: this.workers.length, + taskAgeMs: inFlight === undefined ? undefined : Math.round(now - inFlight.enqueuedAtMono), + sinceDispatchMs: + inFlight?.dispatchedAtMono === undefined + ? undefined + : Math.round(now - inFlight.dispatchedAtMono), + }; + logger.warn("OtlpWorkerPool reaped worker", { ...event, error: error.message }); void worker.terminate().catch(() => {}); this.scheduleRespawn(); + this.#notifyReap(event); + } + + #notifyReap(event: ReapEvent): void { + if (this.onReap === undefined) return; + try { + const result = this.onReap(event); + if (result !== undefined && typeof result.then === "function") { + result.then(undefined, (thrown) => this.#logObserverError(thrown)); + } + } catch (thrown) { + this.#logObserverError(thrown); + } + } + + #logObserverError(thrown: unknown): void { + logger.error("OtlpWorkerPool onReap observer threw", { + error: thrown instanceof Error ? thrown.message : String(thrown), + }); + } + + #clearComputeTimer(worker: Worker) { + const timer = this.computeTimers.get(worker); + if (timer === undefined) return; + clearTimeout(timer); + this.computeTimers.delete(worker); + } + + /** + * A dispatched task whose caller deadline passed is not evidence the worker is stuck: it may + * have sat in the queue for most of its budget. Give the worker the rest of a full compute + * budget for it, and only reap if it still hasn't replied by then. The worker's reply (or a + * crash reap) clears this timer. + */ + #armComputeTimer(worker: Worker, task: Task, delayMs: number) { + this.#clearComputeTimer(worker); + const timer = setTimeout( + () => { + this.computeTimers.delete(worker); + if (this.busyByWorker.get(worker) !== task) return; + const remainingMs = this.taskTimeoutMs - (performance.now() - task.dispatchedAtMono!); + if (remainingMs > 0) { + this.#armComputeTimer(worker, task, remainingMs); + return; + } + this.reap( + worker, + new Error(`otlp worker stuck for ${this.taskTimeoutMs}ms on a single task`), + "timeout" + ); + }, + Math.max(1, Math.ceil(delayMs)) + ); + this.computeTimers.set(worker, timer); } private scheduleRespawn() { if (this.isShuttingDown) return; if (this.workers.length >= this.size) return; - const delay = Math.min(RESPAWN_BASE_MS * 2 ** this.consecutiveFailures, RESPAWN_MAX_MS); + const delay = Math.min(this.respawnBaseMs * 2 ** this.consecutiveFailures, this.respawnMaxMs); this.consecutiveFailures++; setTimeout(() => { if (this.isShuttingDown) return; @@ -219,28 +336,83 @@ export class OtlpWorkerPool { this.drain(); } + /** + * Hand queued tasks to idle workers in FIFO order, shedding any task whose remaining budget is + * below the stale floor so a free worker starts on something it can still finish. The floor is + * a small fraction of the timeout, so only tasks whose deadline is effectively already here are + * dropped; everything else keeps its FIFO turn. + */ private drain() { + const now = performance.now(); while (this.queue.length > 0 && this.idle.length > 0) { - const worker = this.idle.pop()!; const id = this.queue.shift()!; const task = this.tasks.get(id); if (!task) continue; + const remainingMs = task.enqueuedAtMono + this.taskTimeoutMs - now; + if (remainingMs < this.staleFloorMs) { + this.#shed(task, remainingMs, now); + continue; + } + const worker = this.idle.pop()!; task.worker = worker; - this.busyByWorker.set(worker, id); + task.dispatchedAtMono = now; + this.busyByWorker.set(worker, task); worker.postMessage(task.message, task.transfer); } } + /** + * Sheds are aggregated into at most one debug line per second; each line carries the count and + * the span between the first and last shed it covers. + */ + #shed(task: Task, remainingMs: number, now: number) { + clearTimeout(task.timer); + this.tasks.delete(task.message.id); + this.#recordTaskEnd(task, "stale"); + task.reject( + new Error( + `otlp worker task shed after ${Math.round(now - task.enqueuedAtMono)}ms in queue with ${Math.max( + 0, + Math.round(remainingMs) + )}ms of ${this.taskTimeoutMs}ms budget left` + ) + ); + if (this.shedInWindow === 0) this.shedWindowStartMono = now; + this.lastShedMono = now; + this.shedInWindow++; + if (this.shedFlushTimer !== undefined) return; + this.shedFlushTimer = setTimeout(() => { + this.shedFlushTimer = undefined; + logger.debug("OtlpWorkerPool shed stale tasks", { + shed: this.shedInWindow, + windowMs: Math.round(this.lastShedMono - this.shedWindowStartMono), + queueDepth: this.queue.length, + aliveWorkers: this.workers.length, + }); + this.shedInWindow = 0; + }, SHED_LOG_INTERVAL_MS); + this.shedFlushTimer.unref(); + } + + /** + * The caller's budget (queue wait + compute) is spent, so reject it now. Whether the worker is + * at fault depends on how long it has held the task: reap only once it has had a full timeout + * of compute time on this one task, otherwise let it finish via the compute timer and return to + * the idle set on reply. The task is removed here, so neither path can double-reject. + */ private onTimeout(id: number) { const task = this.tasks.get(id); if (!task) return; this.tasks.delete(id); this.#recordTaskEnd(task, "timeout"); - const err = new Error(`otlp worker task timed out after ${TASK_TIMEOUT_MS}ms`); - if (task.worker) { - // Dispatched to a stuck worker: reap it. The task is already removed, so reap won't - // double-reject. - this.reap(task.worker, err, "timeout"); + const err = new Error(`otlp worker task timed out after ${this.taskTimeoutMs}ms`); + if (task.worker !== undefined && task.dispatchedAtMono !== undefined) { + const remainingComputeMs = this.taskTimeoutMs - (performance.now() - task.dispatchedAtMono); + if (remainingComputeMs <= 0) { + this.reap(task.worker, err, "timeout"); + } else { + this.#armComputeTimer(task.worker, task, remainingComputeMs); + } } else { const qi = this.queue.indexOf(id); if (qi !== -1) this.queue.splice(qi, 1); @@ -262,7 +434,7 @@ export class OtlpWorkerPool { } const id = this.nextId++; return new Promise((resolve, reject) => { - const timer = setTimeout(() => this.onTimeout(id), TASK_TIMEOUT_MS); + const timer = setTimeout(() => this.onTimeout(id), this.taskTimeoutMs); this.tasks.set(id, { message: { id, @@ -277,6 +449,7 @@ export class OtlpWorkerPool { reject, timer, enqueuedAt: Date.now(), + enqueuedAtMono: performance.now(), }); this.queue.push(id); this.drain(); @@ -319,6 +492,10 @@ export class OtlpWorkerPool { this.idle.length = 0; this.queue.length = 0; this.busyByWorker.clear(); + for (const timer of this.computeTimers.values()) clearTimeout(timer); + this.computeTimers.clear(); + clearTimeout(this.shedFlushTimer); + this.shedFlushTimer = undefined; // Reject anything that didn't drain within the deadline. for (const [, task] of this.tasks) { clearTimeout(task.timer); diff --git a/apps/webapp/test/fixtures/otlpHangWorker.cjs b/apps/webapp/test/fixtures/otlpHangWorker.cjs new file mode 100644 index 00000000000..d7d90376721 --- /dev/null +++ b/apps/webapp/test/fixtures/otlpHangWorker.cjs @@ -0,0 +1,6 @@ +/** + * Real worker that never replies, to exercise the pool's stuck-worker reap path. + */ +const { parentPort } = require("node:worker_threads"); + +parentPort.on("message", () => {}); diff --git a/apps/webapp/test/fixtures/otlpSlowWorker.cjs b/apps/webapp/test/fixtures/otlpSlowWorker.cjs new file mode 100644 index 00000000000..e5bd1566c9d --- /dev/null +++ b/apps/webapp/test/fixtures/otlpSlowWorker.cjs @@ -0,0 +1,22 @@ +/** + * Real worker that takes a fixed wall-clock time per task, so pool tests can drive queue wait past + * the task deadline while the worker itself stays healthy. Delay comes from + * OTLP_SLOW_WORKER_DELAY_MS (default 100). When OTLP_SLOW_WORKER_HANG_AFTER is set the worker + * answers that many tasks and then never replies again, to model a worker that wedges mid-run. + */ +const { parentPort } = require("node:worker_threads"); + +const delayMs = Number(process.env.OTLP_SLOW_WORKER_DELAY_MS ?? 100); +const hangAfter = process.env.OTLP_SLOW_WORKER_HANG_AFTER + ? Number(process.env.OTLP_SLOW_WORKER_HANG_AFTER) + : Infinity; +let answered = 0; + +parentPort.on("message", (message) => { + if (message && message.type === "pricing") return; + if (answered >= hangAfter) return; + answered++; + setTimeout(() => { + parentPort.postMessage({ id: message.id, ok: true, result: { rows: [] }, computeMs: delayMs }); + }, delayMs); +}); diff --git a/apps/webapp/test/otlpWorkerPoolTimeout.test.ts b/apps/webapp/test/otlpWorkerPoolTimeout.test.ts new file mode 100644 index 00000000000..1e4a9c6d81f --- /dev/null +++ b/apps/webapp/test/otlpWorkerPoolTimeout.test.ts @@ -0,0 +1,292 @@ +import { fileURLToPath } from "node:url"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { OtlpWorkerPool, type ReapEvent } from "~/v3/otlpWorkerPool.server"; +import { createInMemoryMetrics } from "./utils/tracing"; +import { gaugeValue, histogramCount, latestMetrics, metricSum } from "./otlpMetrics.helpers"; + +const slowWorker = fileURLToPath(new URL("./fixtures/otlpSlowWorker.cjs", import.meta.url)); +const hangWorker = fileURLToPath(new URL("./fixtures/otlpHangWorker.cjs", import.meta.url)); + +const config = { spanAttributeValueLengthLimit: 8192, defaultEventStore: "clickhouse" }; +const payload = () => new Uint8Array([1, 2, 3, 4]); + +const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); + +describe("OtlpWorkerPool task timeouts", () => { + const cleanups: Array<() => Promise> = []; + + afterEach(async () => { + for (const cleanup of cleanups.splice(0)) { + await cleanup(); + } + delete process.env.OTLP_SLOW_WORKER_DELAY_MS; + delete process.env.OTLP_SLOW_WORKER_HANG_AFTER; + }); + + it("keeps a healthy worker alive when queued tasks exceed their deadline under overload", async () => { + const taskTimeoutMs = 500; + const computeMs = 20; + const arrivalIntervalMs = 10; + const runMs = 3000; + process.env.OTLP_SLOW_WORKER_DELAY_MS = String(computeMs); + + const metrics = createInMemoryMetrics(); + const pool = new OtlpWorkerPool(1, slowWorker, [], metrics.meter, { + taskTimeoutMs, + respawnBaseMs: 50, + respawnMaxMs: 200, + }); + cleanups.push(async () => { + await pool.shutdown(); + await metrics.shutdown(); + }); + + let resolved = 0; + let rejected = 0; + const pending: Promise[] = []; + const aliveSamples: Array = []; + const okSamples: number[] = []; + + const startedAt = Date.now(); + const enqueue = setInterval(() => { + const p = pool + .runTransform("traces", payload(), config) + .then(() => resolved++) + .catch(() => rejected++); + pending.push(p); + }, arrivalIntervalMs); + + let lastSampleAt = startedAt; + while (Date.now() - startedAt < runMs) { + await sleep(25); + if (Date.now() - startedAt >= 1000 && Date.now() - lastSampleAt >= 200) { + lastSampleAt = Date.now(); + const rm = await latestMetrics(metrics); + aliveSamples.push(gaugeValue(rm, "ingest.worker_pool.workers", { state: "alive" })); + okSamples.push(metricSum(rm, "ingest.worker_pool.tasks", { outcome: "ok" })); + } + } + clearInterval(enqueue); + await Promise.all(pending); + + const rm = await latestMetrics(metrics); + const summary = { + resolved, + rejected, + respawnsTimeout: metricSum(rm, "ingest.worker_pool.respawns", { reason: "timeout" }), + respawnsError: metricSum(rm, "ingest.worker_pool.respawns", { reason: "error" }), + respawnsExit: metricSum(rm, "ingest.worker_pool.respawns", { reason: "exit" }), + crash: metricSum(rm, "ingest.worker_pool.tasks", { outcome: "crash" }), + ok: metricSum(rm, "ingest.worker_pool.tasks", { outcome: "ok" }), + timeout: metricSum(rm, "ingest.worker_pool.tasks", { outcome: "timeout" }), + stale: metricSum(rm, "ingest.worker_pool.tasks", { outcome: "stale" }), + aliveSamples, + okSamples, + }; + console.log("overload summary", JSON.stringify(summary)); + + expect(summary.timeout + summary.stale).toBeGreaterThan(0); + expect(rejected).toBeGreaterThan(0); + + expect(summary.respawnsTimeout).toBe(0); + expect(summary.respawnsError).toBe(0); + expect(summary.respawnsExit).toBe(0); + expect(summary.crash).toBe(0); + expect(aliveSamples.length).toBeGreaterThan(0); + expect(aliveSamples.every((alive) => alive === 1)).toBe(true); + expect(summary.ok).toBeGreaterThanOrEqual(50); + expect(summary.ok).toBeGreaterThan(okSamples[0]!); + + await vi.waitFor( + async () => { + const rm = await latestMetrics(metrics); + expect(gaugeValue(rm, "ingest.worker_pool.workers", { state: "alive" })).toBe(1); + }, + { timeout: 2000, interval: 50 } + ); + }, 15_000); + + it("lets a healthy worker finish a task whose caller deadline passed and reuses it", async () => { + process.env.OTLP_SLOW_WORKER_DELAY_MS = "200"; + const metrics = createInMemoryMetrics(); + const pool = new OtlpWorkerPool(1, slowWorker, [], metrics.meter, { + taskTimeoutMs: 300, + respawnBaseMs: 50, + respawnMaxMs: 200, + }); + cleanups.push(async () => { + await pool.shutdown(); + await metrics.shutdown(); + }); + await pool.runTransform("traces", payload(), config); + + const results = await Promise.allSettled([ + pool.runTransform("traces", payload(), config), + pool.runTransform("traces", payload(), config), + pool.runTransform("traces", payload(), config), + ]); + expect(results[0]!.status).toBe("fulfilled"); + expect(results[1]!.status).toBe("rejected"); + expect((results[1] as PromiseRejectedResult).reason.message).toMatch(/timed out/); + + await vi.waitFor( + async () => { + const rm = await latestMetrics(metrics); + expect(gaugeValue(rm, "ingest.worker_pool.workers", { state: "idle" })).toBe(1); + }, + { timeout: 2000, interval: 20 } + ); + await expect(pool.runTransform("traces", payload(), config)).resolves.toBeDefined(); + + await vi.waitFor( + async () => { + const rm = await latestMetrics(metrics); + expect(metricSum(rm, "ingest.worker_pool.respawns")).toBe(0); + expect(metricSum(rm, "ingest.worker_pool.tasks", { outcome: "crash" })).toBe(0); + expect(histogramCount(rm, "ingest.worker_pool.task.duration")).toBe(5); + expect(histogramCount(rm, "ingest.worker_pool.compute.duration")).toBe(4); + expect(gaugeValue(rm, "ingest.worker_pool.workers", { state: "alive" })).toBe(1); + expect(gaugeValue(rm, "ingest.worker_pool.workers", { state: "idle" })).toBe(1); + }, + { timeout: 3000, interval: 50 } + ); + }); + + it("reaps a worker that hangs only after a full compute budget, not at the caller deadline", async () => { + const taskTimeoutMs = 300; + process.env.OTLP_SLOW_WORKER_DELAY_MS = "100"; + process.env.OTLP_SLOW_WORKER_HANG_AFTER = "2"; + const reaps: ReapEvent[] = []; + const metrics = createInMemoryMetrics(); + const pool = new OtlpWorkerPool(1, slowWorker, [], metrics.meter, { + taskTimeoutMs, + respawnBaseMs: 50, + respawnMaxMs: 200, + onReap: (event) => reaps.push(event), + }); + cleanups.push(async () => { + await pool.shutdown(); + await metrics.shutdown(); + }); + await pool.runTransform("traces", payload(), config); + + const first = pool.runTransform("traces", payload(), config); + const second = pool.runTransform("traces", payload(), config); + await expect(first).resolves.toBeDefined(); + await expect(second).rejects.toThrow(/timed out/); + const rejectedAt = Date.now(); + + await vi.waitFor( + () => { + expect(reaps.some((event) => event.reason === "timeout")).toBe(true); + }, + { timeout: 2000, interval: 10 } + ); + const reapedAt = Date.now(); + const reap = reaps.find((event) => event.reason === "timeout")!; + expect(reap.sinceDispatchMs).toBeGreaterThanOrEqual(taskTimeoutMs); + expect(reap.sinceDispatchMs).toBeLessThan(taskTimeoutMs + 150); + expect(reap.taskAgeMs).toBeGreaterThan(reap.sinceDispatchMs!); + expect(reapedAt - rejectedAt).toBeGreaterThanOrEqual(50); + + await vi.waitFor( + async () => { + const rm = await latestMetrics(metrics); + expect(metricSum(rm, "ingest.worker_pool.respawns", { reason: "timeout" })).toBe(1); + expect(metricSum(rm, "ingest.worker_pool.respawns")).toBe(1); + expect(gaugeValue(rm, "ingest.worker_pool.workers", { state: "alive" })).toBe(1); + }, + { timeout: 3000, interval: 50 } + ); + }); + + it("keeps reaping and respawning when the onReap observer throws", async () => { + const metrics = createInMemoryMetrics(); + const pool = new OtlpWorkerPool(1, hangWorker, [], metrics.meter, { + taskTimeoutMs: 200, + respawnBaseMs: 50, + respawnMaxMs: 200, + onReap: () => { + throw new Error("observer boom"); + }, + }); + cleanups.push(async () => { + await pool.shutdown(); + await metrics.shutdown(); + }); + + await expect(pool.runTransform("traces", payload(), config)).rejects.toThrow(/timed out/); + + await vi.waitFor( + async () => { + const rm = await latestMetrics(metrics); + expect(metricSum(rm, "ingest.worker_pool.respawns", { reason: "timeout" })).toBe(1); + expect(gaugeValue(rm, "ingest.worker_pool.workers", { state: "alive" })).toBe(1); + expect(gaugeValue(rm, "ingest.worker_pool.workers", { state: "idle" })).toBe(1); + }, + { timeout: 3000, interval: 50 } + ); + }); + + it("keeps reaping and respawning when the onReap observer rejects asynchronously", async () => { + const metrics = createInMemoryMetrics(); + const pool = new OtlpWorkerPool(1, hangWorker, [], metrics.meter, { + taskTimeoutMs: 200, + respawnBaseMs: 50, + respawnMaxMs: 200, + onReap: async () => { + throw new Error("async observer boom"); + }, + }); + cleanups.push(async () => { + await pool.shutdown(); + await metrics.shutdown(); + }); + + await expect(pool.runTransform("traces", payload(), config)).rejects.toThrow(/timed out/); + + await vi.waitFor( + async () => { + const rm = await latestMetrics(metrics); + expect(metricSum(rm, "ingest.worker_pool.respawns", { reason: "timeout" })).toBe(1); + expect(gaugeValue(rm, "ingest.worker_pool.workers", { state: "alive" })).toBe(1); + expect(gaugeValue(rm, "ingest.worker_pool.workers", { state: "idle" })).toBe(1); + }, + { timeout: 3000, interval: 50 } + ); + }); + + it("reaps and respawns a worker that never replies", async () => { + const metrics = createInMemoryMetrics(); + const pool = new OtlpWorkerPool(1, hangWorker, [], metrics.meter, { + taskTimeoutMs: 200, + respawnBaseMs: 50, + respawnMaxMs: 200, + }); + cleanups.push(async () => { + await pool.shutdown(); + await metrics.shutdown(); + }); + + await expect(pool.runTransform("traces", payload(), config)).rejects.toThrow(/timed out/); + + await vi.waitFor( + async () => { + const rm = await latestMetrics(metrics); + expect(metricSum(rm, "ingest.worker_pool.respawns", { reason: "timeout" })).toBe(1); + expect(metricSum(rm, "ingest.worker_pool.tasks", { outcome: "timeout" })).toBe(1); + expect(metricSum(rm, "ingest.worker_pool.tasks", { outcome: "crash" })).toBe(0); + }, + { timeout: 3000, interval: 50 } + ); + + await vi.waitFor( + async () => { + const rm = await latestMetrics(metrics); + expect(gaugeValue(rm, "ingest.worker_pool.workers", { state: "alive" })).toBe(1); + expect(gaugeValue(rm, "ingest.worker_pool.workers", { state: "idle" })).toBe(1); + }, + { timeout: 3000, interval: 50 } + ); + }); +});