From 116ea7afca7b112bd6bac957136f39bfadb0505f Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Thu, 29 Oct 2026 14:42:50 +0000 Subject: [PATCH 1/5] feat(queue): Deliver local dev work through BullMQ on Redis Add a BullMQ queue provider behind the QueueProvider contract from #2338. JUNIOR_QUEUE_PROVIDER selects it, and the default stays vercel. pnpm dev selects bullmq with the compose.yaml Redis, so local work no longer needs Vercel Queues access or an OIDC token. bullmq and redis are optional peer dependencies. Junior imports them by a name in a variable so bundlers leave them out of the production server, and check-vercel-output.mjs fails if they reach a deployed function. Co-Authored-By: Thomas Jiang <20501235+thomasjiangcy@users.noreply.github.com> Co-Authored-By: Thomas Jiang Co-Authored-By: Thomas Jiang <20501235+thomasjiangcy@users.noreply.github.com> --- apps/example/scripts/check-vercel-output.mjs | 14 +- .../no-queue-provider-module-imports.yml | 4 +- .../content/docs/contribute/development.md | 2 + .../content/docs/reference/config-and-env.md | 9 + packages/junior/package.json | 14 +- packages/junior/src/chat/config.ts | 3 + packages/junior/src/chat/queue/README.md | 16 +- packages/junior/src/chat/queue/bullmq.ts | 204 ++++++++++++++++++ packages/junior/src/chat/queue/config.ts | 26 +++ packages/junior/src/chat/queue/provider.ts | 8 +- .../component/config/chat-config.test.ts | 24 ++- .../tests/integration/queue-bullmq.test.ts | 139 ++++++++++++ pnpm-lock.yaml | 136 ++++++++++++ pnpm-workspace.yaml | 1 + scripts/dev-server.mjs | 2 + scripts/lib/load-env-files.mjs | 13 ++ scripts/lib/load-env-files.test.mjs | 32 +++ 17 files changed, 638 insertions(+), 9 deletions(-) create mode 100644 packages/junior/src/chat/queue/bullmq.ts create mode 100644 packages/junior/src/chat/queue/config.ts create mode 100644 packages/junior/tests/integration/queue-bullmq.test.ts diff --git a/apps/example/scripts/check-vercel-output.mjs b/apps/example/scripts/check-vercel-output.mjs index 93f7572be2..ec3ca384c8 100644 --- a/apps/example/scripts/check-vercel-output.mjs +++ b/apps/example/scripts/check-vercel-output.mjs @@ -1,4 +1,4 @@ -import { existsSync, readFileSync, statSync } from "node:fs"; +import { existsSync, readdirSync, readFileSync, statSync } from "node:fs"; import path from "node:path"; import { fileURLToPath } from "node:url"; @@ -102,6 +102,17 @@ function assertFunctionHasJuniorContent(functionDir, pluginPackages) { } } +// The BullMQ queue provider is for local dev only. Junior imports it at run +// time so bundlers skip it. Fail if BullMQ or its msgpackr dependency still +// reaches a deployed function. +function assertNoDevQueuePackages(functionDir) { + for (const entry of readdirSync(functionDir, { recursive: true })) { + if (/(^|[\\/])(bullmq|msgpackr)[^\\/]*$/.test(entry)) { + fail(`${path.relative(appRoot, functionDir)} contains ${entry}`); + } + } +} + if (existsSync(path.join(appRoot, "api"))) { fail( "apps/example/api exists; Vercel would route source functions before Nitro", @@ -119,6 +130,7 @@ if (pluginPackages.length === 0) { for (const functionDir of [serverFunctionDir, queueFunctionDir]) { assertFunctionHasJuniorContent(functionDir, pluginPackages); + assertNoDevQueuePackages(functionDir); } console.log( diff --git a/ast-grep/rules/no-queue-provider-module-imports.yml b/ast-grep/rules/no-queue-provider-module-imports.yml index f5ce79df00..3d8d76c40a 100644 --- a/ast-grep/rules/no-queue-provider-module-imports.yml +++ b/ast-grep/rules/no-queue-provider-module-imports.yml @@ -1,7 +1,7 @@ id: no-queue-provider-module-imports language: TypeScript severity: error -message: Only packages/junior/src/chat/queue/provider.ts may import a queue provider module such as chat/queue/vercel.ts. Other code calls getQueueProvider(). See packages/junior/src/chat/queue/README.md. +message: Only packages/junior/src/chat/queue/provider.ts may import a queue provider module such as chat/queue/vercel.ts or chat/queue/bullmq.ts. Other code calls getQueueProvider(). See packages/junior/src/chat/queue/README.md. files: - packages/junior/src/**/*.ts - packages/junior/src/**/*.tsx @@ -13,4 +13,4 @@ rule: - kind: export_statement has: kind: string - regex: '^["''](@/chat/queue/vercel|(\.{1,2}/)+(.+/)?queue/vercel)["'']$' + regex: '^["''](@/chat/queue/(vercel|bullmq)|(\.{1,2}/)+(.+/)?queue/(vercel|bullmq))["'']$' diff --git a/packages/docs/src/content/docs/contribute/development.md b/packages/docs/src/content/docs/contribute/development.md index a365511fe7..a60e4585f4 100644 --- a/packages/docs/src/content/docs/contribute/development.md +++ b/packages/docs/src/content/docs/contribute/development.md @@ -41,6 +41,8 @@ pnpm dev This starts the example app on `http://localhost:3000` by default. It also rebuilds and watches the workspace packages that the example app consumes, so dashboard and runtime package edits are reflected without manually rebuilding first. +`pnpm dev` delivers background work through BullMQ on the Redis in `compose.yaml` (`redis://localhost:6382`). Start it with `docker compose up -d redis`. Local work then needs no Vercel Queues access. To use Vercel Queues, set `JUNIOR_QUEUE_PROVIDER=vercel` and run `pnpm dev:env` for the Vercel token. To use another Redis, set `JUNIOR_QUEUE_REDIS_URL`. + To also open the Sentry Local UI in your browser, set `OPEN_SPOTLIGHT_WEB=1`: ```bash diff --git a/packages/docs/src/content/docs/reference/config-and-env.md b/packages/docs/src/content/docs/reference/config-and-env.md index 2980e33a24..38baa8d1e3 100644 --- a/packages/docs/src/content/docs/reference/config-and-env.md +++ b/packages/docs/src/content/docs/reference/config-and-env.md @@ -51,6 +51,8 @@ related: | `JUNIOR_S3_ACCESS_KEY_ID` | Conditional | S3 access key ID. Required when the provider is `s3`. | | `JUNIOR_S3_SECRET_ACCESS_KEY` | Conditional | S3 secret access key. Required when the provider is `s3`. | | `JUNIOR_S3_REGION` | No | S3 signing region. Defaults to `us-east-1`. | +| `JUNIOR_QUEUE_PROVIDER` | No | Queue that delivers background work: `vercel` or `bullmq`. Defaults to `vercel`. Use `bullmq` only for local development. | +| `JUNIOR_QUEUE_REDIS_URL` | Conditional | Redis for the BullMQ queue, such as `redis://localhost:6382`. Required when the provider is `bullmq`. | For Vercel deployments, create a private Blob store and connect it to the project before using `sendFiles` or `publishImage`. Prefer an OIDC connection. @@ -68,6 +70,13 @@ visible after you change to another provider. Junior loses them when the process stops, and each process has its own copy. Do not use it for a deployed app. +`JUNIOR_QUEUE_PROVIDER=bullmq` delivers background work through BullMQ on +Redis, with workers in the dev server process. Local development then needs no +Vercel Queues access. Install the optional `bullmq` and `redis` packages to use +it. Junior loads them only when this provider is selected, so they never enter a +deployed bundle. Give the queue its own Redis URL. If two dev servers share one +Redis, each one can take the other's jobs. + Junior applies `JUNIOR_SQL_STATEMENT_TIMEOUT_MS` through PostgreSQL `statement_timeout` for both the Neon and node-postgres drivers. `junior upgrade` does not apply this runtime limit because schema migrations can legitimately take longer. `JUNIOR_CROSS_ACTOR_MID_RUN_MODE=follow_up` keeps another actor's mid-run ask diff --git a/packages/junior/package.json b/packages/junior/package.json index 5336150596..37aceffad4 100644 --- a/packages/junior/package.json +++ b/packages/junior/package.json @@ -109,7 +109,17 @@ "zod": "catalog:" }, "peerDependencies": { - "nitro": "3.0.260415-beta || 3.0.260429-beta || 3.0.260522-beta || 3.0.260603-beta || 3.0.260610-beta || 3.0.260903-beta" + "bullmq": "^6.5.0", + "nitro": "3.0.260415-beta || 3.0.260429-beta || 3.0.260522-beta || 3.0.260603-beta || 3.0.260610-beta || 3.0.260903-beta", + "redis": "^5.12.1" + }, + "peerDependenciesMeta": { + "bullmq": { + "optional": true + }, + "redis": { + "optional": true + } }, "devDependencies": { "@emnapi/core": "^1.10.0", @@ -123,11 +133,13 @@ "@types/pg": "^8.15.6", "@typescript/typescript6": "6.0.2", "@vitest/coverage-v8": "4.1.12", + "bullmq": "6.5.0", "dependency-cruiser": "^17.4.0", "drizzle-kit": "catalog:", "msw": "^2.14.6", "nitro": "3.0.260522-beta", "oxlint": "^1.79.0", + "redis": "5.12.1", "tsdown": "0.23.1", "typescript": "^7.0.2", "vercel": "^54.4.0", diff --git a/packages/junior/src/chat/config.ts b/packages/junior/src/chat/config.ts index 8019c1f4b6..e6891475d0 100644 --- a/packages/junior/src/chat/config.ts +++ b/packages/junior/src/chat/config.ts @@ -3,6 +3,7 @@ import { type AttachmentStorageConfig, } from "@/chat/attachments/config"; import { toOptionalTrimmed } from "@/chat/optional-string"; +import { readQueueConfig, type QueueConfig } from "@/chat/queue/config"; import { resolveGatewayModel } from "@/chat/pi/client"; import { normalizeSlackEmojiName } from "@/chat/slack/emoji"; import { logWarn } from "@/chat/logging"; @@ -131,6 +132,7 @@ export type SqlDriver = "neon" | "postgres"; export interface ChatConfig { attachmentStorage: AttachmentStorageConfig; bot: BotConfig; + queue: QueueConfig; functionMaxDurationSeconds: number; conversationWorkEnabled: boolean; conversationWorkSoftYieldAfterMs: number; @@ -612,6 +614,7 @@ export function readChatConfig( return { attachmentStorage: readAttachmentStorageConfig(env), bot: readBotConfig(env, resolvedFunctionMaxDurationSeconds), + queue: readQueueConfig(env), functionMaxDurationSeconds: resolvedFunctionMaxDurationSeconds, conversationWorkEnabled: parseConversationWorkEnabled( env.JUNIOR_CONVERSATION_WORK_ENABLED, diff --git a/packages/junior/src/chat/queue/README.md b/packages/junior/src/chat/queue/README.md index 2ace708066..09f8e53b99 100644 --- a/packages/junior/src/chat/queue/README.md +++ b/packages/junior/src/chat/queue/README.md @@ -14,11 +14,23 @@ Conversation work is different. It only wakes a conversation. Keep using ## Provider - `provider.ts` owns the `QueueProvider` contract and `getQueueProvider()`. - Queue code calls `getQueueProvider()`, not a provider module. Today it - always returns the Vercel provider. + Queue code calls `getQueueProvider()`, not a provider module. + `JUNIOR_QUEUE_PROVIDER` selects the provider. The default is `vercel`. - `vercel.ts` implements `QueueProvider`. It is the only module that imports `@vercel/queue`, and only `provider.ts` imports it. Two ast-grep rules enforce these boundaries. +- `bullmq.ts` implements `QueueProvider` with BullMQ on Redis, for local + development only. `pnpm dev` selects it. Its workers start where + `registerDevConsumer` runs, in the dev server process, and pull jobs from + Redis. Nothing pushes to the callback routes. +- `bullmq.ts` imports `bullmq` and `redis` by a name in a variable, so + bundlers do not put them into the production server. + `apps/example/scripts/check-vercel-output.mjs` fails if they reach a + deployed function. +- BullMQ is not the same as Vercel Queues. An idempotency key drops a second + send only while BullMQ still stores the first job (one hour after it + completes). BullMQ renews the lock of a running job, so it does not send a + slow job again when `visibilityTimeoutSeconds` runs out. - Shared types, such as `QueueMessageMetadata` and `QueueSendOptions`, live in `provider.ts`. - `nitro.ts` still writes the Vercel `queue/v2beta` triggers into the build diff --git a/packages/junior/src/chat/queue/bullmq.ts b/packages/junior/src/chat/queue/bullmq.ts new file mode 100644 index 0000000000..35dbb0a2f5 --- /dev/null +++ b/packages/junior/src/chat/queue/bullmq.ts @@ -0,0 +1,204 @@ +import { createHash } from "node:crypto"; +import type { Job, Queue, Worker } from "bullmq"; +import { getChatConfig } from "@/chat/config"; +import { logException } from "@/chat/logging"; +import type { + QueueConsumer, + QueueMessageMetadata, + QueueProvider, +} from "@/chat/queue/provider"; + +// Module overview: a queue provider for local development, backed by BullMQ +// on Redis. Workers run in this process, so `pnpm dev` needs no Vercel +// Queues access. Deployed apps use the Vercel provider. +// +// `bullmq` and `redis` are optional peer dependencies. Only dev installs +// need them. + +interface JobData { + payload: unknown; + retentionSeconds?: number; +} + +type BullMq = typeof import("bullmq"); +type RedisConnection = ReturnType; + +// Durable state owns the real attempt limit for conversation work +// (`maxDeliveries: null`). This cap only stops a message that always fails +// from looping for ever in dev. +const MAX_ATTEMPTS = 50; +const MAX_BACKOFF_MS = 30_000; +// BullMQ runs one job at a time unless told otherwise. One long turn would +// then block every other conversation. +const WORKER_CONCURRENCY = 20; +// A job id drops a second send only while BullMQ still stores the first job. +// So this is the dedupe window for idempotency keys. +const COMPLETED_JOB_AGE_SECONDS = 60 * 60; +const FAILED_JOB_AGE_SECONDS = 24 * 60 * 60; + +let loaded: + | Promise<{ bullmq: BullMq; connection: RedisConnection }> + | undefined; +const queues = new Map>(); + +/** + * Import a module by a name held in a variable. + * + * Bundlers follow a literal `import("bullmq")` and put BullMQ into the + * production server, even though production never selects this provider. + * Node resolves this import when the app runs instead. + */ +function importAtRuntime(specifier: string): Promise { + return import(/* @vite-ignore */ specifier); +} + +function redisUrl(): string { + const config = getChatConfig().queue; + if (config.provider !== "bullmq") { + throw new Error("The BullMQ queue provider is not configured"); + } + return config.redisUrl; +} + +/** Load BullMQ and one shared Redis client the first time they are needed. */ +function load() { + loaded ??= (async () => { + const [bullmq, redis] = (await Promise.all([ + importAtRuntime("bullmq"), + importAtRuntime("redis"), + ]).catch((error: unknown) => { + throw new Error( + "JUNIOR_QUEUE_PROVIDER=bullmq needs the bullmq and redis packages installed", + { cause: error }, + ); + })) as [BullMq, typeof import("redis")]; + const client = redis.createClient({ url: redisUrl() }); + await client.connect(); + return { bullmq, connection: bullmq.createNodeRedisClient(client) }; + })(); + return loaded; +} + +async function getQueue(topic: string): Promise> { + const { bullmq, connection } = await load(); + let queue = queues.get(topic); + if (!queue) { + queue = new bullmq.Queue(topic, { + connection, + }); + queues.set(topic, queue); + } + return queue; +} + +/** BullMQ job ids must not contain ":", and Junior's keys do. */ +function toJobId(idempotencyKey: string): string { + return `k-${createHash("sha256").update(idempotencyKey).digest("base64url")}`; +} + +function metadataFor( + job: Job, + consumer: QueueConsumer, +): QueueMessageMetadata { + return { + consumerGroup: consumer.consumerGroup, + createdAt: new Date(job.timestamp), + deliveryCount: job.attemptsMade + 1, + messageId: job.id ?? "", + topicName: consumer.topic, + }; +} + +async function processJob( + job: Job, + consumer: QueueConsumer, +): Promise { + const { payload, retentionSeconds } = job.data; + if ( + retentionSeconds !== undefined && + Date.now() - job.timestamp > retentionSeconds * 1000 + ) { + return; + } + const metadata = metadataFor(job, consumer); + try { + await consumer.handler(payload, metadata); + } catch (error) { + if (consumer.retry(error, metadata)?.acknowledge) { + return; + } + throw error; + } +} + +/** Start one in-process worker. The returned function stops it. */ +function startWorker(consumer: QueueConsumer): () => Promise { + let worker: Worker | undefined; + let stopped = false; + const started = load().then(({ bullmq, connection }) => { + if (stopped) { + return; + } + worker = new bullmq.Worker( + consumer.topic, + (job) => processJob(job, consumer), + { + concurrency: WORKER_CONCURRENCY, + connection, + // BullMQ renews this lock while the job runs. Unlike Vercel Queues, + // a running job is not sent again when the time runs out. + ...(consumer.visibilityTimeoutSeconds !== undefined && { + lockDuration: consumer.visibilityTimeoutSeconds * 1000, + }), + settings: { + backoffStrategy: (attemptsMade: number) => + Math.min(1000 * 2 ** (attemptsMade - 1), MAX_BACKOFF_MS), + }, + }, + ); + worker.on("error", (error) => { + logException(error, "queue.bullmq.worker.exception", { + "messaging.destination.name": consumer.topic, + }); + }); + }); + return async () => { + stopped = true; + await started; + await worker?.close(); + }; +} + +/** Queue provider backed by BullMQ, for local development only. */ +export const bullmqQueueProvider: QueueProvider = { + messagingSystem: "bullmq", + send: async (topic, payload, options = {}) => { + const queue = await getQueue(topic); + const job = await queue.add( + topic, + { payload, retentionSeconds: options.retentionSeconds }, + { + attempts: MAX_ATTEMPTS, + backoff: { type: "custom" }, + delay: options.delaySeconds ? options.delaySeconds * 1000 : undefined, + jobId: options.idempotencyKey + ? toJobId(options.idempotencyKey) + : undefined, + removeOnComplete: { age: COMPLETED_JOB_AGE_SECONDS }, + removeOnFail: { age: FAILED_JOB_AGE_SECONDS }, + }, + ); + return job.id ? { messageId: job.id } : {}; + }, + // Workers pull jobs from Redis, so nothing pushes to these routes. + createCallback: () => async () => + new Response("This queue provider does not push messages", { + status: 404, + }), + registerDevConsumer: (consumer) => { + const stop = startWorker(consumer); + return () => { + void stop(); + }; + }, +}; diff --git a/packages/junior/src/chat/queue/config.ts b/packages/junior/src/chat/queue/config.ts new file mode 100644 index 0000000000..eb884b474b --- /dev/null +++ b/packages/junior/src/chat/queue/config.ts @@ -0,0 +1,26 @@ +import { toOptionalTrimmed } from "@/chat/optional-string"; + +/** Which queue delivers background work. */ +export type QueueConfig = + | { provider: "vercel" } + | { provider: "bullmq"; redisUrl: string }; + +/** Read queue settings from environment variables. */ +export function readQueueConfig(env: NodeJS.ProcessEnv): QueueConfig { + const provider = toOptionalTrimmed(env.JUNIOR_QUEUE_PROVIDER) ?? "vercel"; + switch (provider) { + case "vercel": + return { provider }; + case "bullmq": { + const redisUrl = toOptionalTrimmed(env.JUNIOR_QUEUE_REDIS_URL); + if (!redisUrl) { + throw new Error( + "JUNIOR_QUEUE_REDIS_URL is required when JUNIOR_QUEUE_PROVIDER is bullmq", + ); + } + return { provider, redisUrl }; + } + default: + throw new Error("JUNIOR_QUEUE_PROVIDER must be vercel or bullmq"); + } +} diff --git a/packages/junior/src/chat/queue/provider.ts b/packages/junior/src/chat/queue/provider.ts index da01c11782..c48669bff6 100644 --- a/packages/junior/src/chat/queue/provider.ts +++ b/packages/junior/src/chat/queue/provider.ts @@ -1,9 +1,11 @@ +import { getChatConfig } from "@/chat/config"; +import { bullmqQueueProvider } from "@/chat/queue/bullmq"; import { vercelQueueProvider } from "@/chat/queue/vercel"; // Module overview: the one place that chooses a queue provider. // // Queue code calls `getQueueProvider()` and uses only this contract. -// Provider modules, such as `vercel.ts`, implement it. Only this module +// Provider modules, such as `vercel.ts` and `bullmq.ts`, implement it. Only this module // imports a provider module. /** Delivery facts that a queue provider passes with each message. */ @@ -69,5 +71,7 @@ export interface QueueProvider { /** Return the queue provider for this deployment. */ export function getQueueProvider(): QueueProvider { - return vercelQueueProvider; + return getChatConfig().queue.provider === "bullmq" + ? bullmqQueueProvider + : vercelQueueProvider; } diff --git a/packages/junior/tests/component/config/chat-config.test.ts b/packages/junior/tests/component/config/chat-config.test.ts index de32489ddd..8274fcfc4e 100644 --- a/packages/junior/tests/component/config/chat-config.test.ts +++ b/packages/junior/tests/component/config/chat-config.test.ts @@ -25,7 +25,8 @@ describe("chat config", () => { for (const name of Object.keys(process.env)) { if ( name === "JUNIOR_ATTACHMENT_STORAGE_PROVIDER" || - name.startsWith("JUNIOR_S3_") + name.startsWith("JUNIOR_S3_") || + name.startsWith("JUNIOR_QUEUE_") ) { delete process.env[name]; } @@ -409,6 +410,27 @@ describe("chat config", () => { ); }); + it("uses Vercel Queues by default and reads BullMQ settings", async () => { + expect((await loadConfig()).getChatConfig().queue).toEqual({ + provider: "vercel", + }); + + process.env.JUNIOR_QUEUE_PROVIDER = "bullmq"; + process.env.JUNIOR_QUEUE_REDIS_URL = "redis://localhost:6382"; + expect((await loadConfig()).getChatConfig().queue).toEqual({ + provider: "bullmq", + redisUrl: "redis://localhost:6382", + }); + }); + + it("throws when the BullMQ queue has no Redis URL", async () => { + process.env.JUNIOR_QUEUE_PROVIDER = "bullmq"; + + await expect(loadConfig()).rejects.toThrow( + "JUNIOR_QUEUE_REDIS_URL is required when JUNIOR_QUEUE_PROVIDER is bullmq", + ); + }); + it("enables conversation work by default", async () => { const { getChatConfig } = await loadConfig(); expect(getChatConfig().conversationWorkEnabled).toBe(true); diff --git a/packages/junior/tests/integration/queue-bullmq.test.ts b/packages/junior/tests/integration/queue-bullmq.test.ts new file mode 100644 index 0000000000..d8e8ea2e41 --- /dev/null +++ b/packages/junior/tests/integration/queue-bullmq.test.ts @@ -0,0 +1,139 @@ +import { randomUUID } from "node:crypto"; +import { afterAll, describe, expect, it, vi } from "vitest"; +import { + getQueueProvider, + type QueueConsumer, + type QueueMessageMetadata, +} from "@/chat/queue/provider"; + +// Chat config reads the queue provider when it loads, so set it first. CI +// runs Redis at REDIS_URL. Locally, compose.yaml runs it on port 6382. +vi.hoisted(() => { + process.env.JUNIOR_QUEUE_PROVIDER = "bullmq"; + process.env.JUNIOR_QUEUE_REDIS_URL = + process.env.REDIS_URL?.trim() || "redis://localhost:6382"; +}); + +const stops: Array<() => void> = []; + +afterAll(() => { + for (const stop of stops) { + stop(); + } +}); + +function uniqueTopic(name: string): string { + return `junior_test_${name}_${randomUUID()}`; +} + +function startConsumer( + topic: string, + handler: QueueConsumer["handler"], + retry: QueueConsumer["retry"] = () => undefined, +): void { + stops.push( + getQueueProvider().registerDevConsumer({ + consumerGroup: `${topic}_dev`, + handler, + retry, + topic, + }), + ); +} + +async function waitFor(check: () => boolean, timeoutMs = 10_000) { + const deadline = Date.now() + timeoutMs; + while (!check()) { + if (Date.now() > deadline) { + throw new Error("Timed out waiting for queue delivery"); + } + await new Promise((resolve) => setTimeout(resolve, 25)); + } +} + +describe("BullMQ queue provider", () => { + it("delivers each idempotency key once, in this process", async () => { + const topic = uniqueTopic("delivery"); + const deliveries: Array<{ + metadata: QueueMessageMetadata; + payload: unknown; + }> = []; + startConsumer(topic, async (payload, metadata) => { + deliveries.push({ metadata, payload }); + }); + + const provider = getQueueProvider(); + expect(provider.messagingSystem).toBe("bullmq"); + // Junior's keys contain ":", which BullMQ does not allow in job ids. + const key = `message:conversation_1:${Date.now()}`; + const first = await provider.send(topic, { n: 1 }, { idempotencyKey: key }); + const second = await provider.send( + topic, + { n: 2 }, + { idempotencyKey: key }, + ); + await waitFor(() => deliveries.length > 0); + await provider.send(topic, { n: 3 }, { idempotencyKey: key }); + await new Promise((resolve) => setTimeout(resolve, 500)); + + expect(second.messageId).toBe(first.messageId); + expect(deliveries).toEqual([ + { + metadata: expect.objectContaining({ + consumerGroup: `${topic}_dev`, + deliveryCount: 1, + messageId: first.messageId, + topicName: topic, + }), + payload: { n: 1 }, + }, + ]); + }); + + it("retries a failed message until retry() acknowledges it", async () => { + const topic = uniqueTopic("retry"); + const deliveryCounts: number[] = []; + let acknowledged = false; + startConsumer( + topic, + async (_payload, metadata) => { + deliveryCounts.push(metadata.deliveryCount); + throw new Error("temporary failure"); + }, + (_error, metadata) => { + if (metadata.deliveryCount < 3) { + return undefined; + } + acknowledged = true; + return { acknowledge: true }; + }, + ); + + await getQueueProvider().send(topic, {}); + await waitFor(() => acknowledged, 15_000); + await new Promise((resolve) => setTimeout(resolve, 500)); + + expect(deliveryCounts).toEqual([1, 2, 3]); + }, 20_000); + + it("runs slow messages at the same time", async () => { + const topic = uniqueTopic("concurrency"); + let running = 0; + let mostRunning = 0; + let finished = 0; + startConsumer(topic, async () => { + running += 1; + mostRunning = Math.max(mostRunning, running); + await new Promise((resolve) => setTimeout(resolve, 500)); + running -= 1; + finished += 1; + }); + + await Promise.all( + [1, 2, 3].map((n) => getQueueProvider().send(topic, { n })), + ); + await waitFor(() => finished === 3); + + expect(mostRunning).toBe(3); + }); +}); diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 4f2d4af056..aa146588a1 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -323,6 +323,9 @@ importers: '@vitest/coverage-v8': specifier: 4.1.12 version: 4.1.12(vitest@4.1.12) + bullmq: + specifier: 6.5.0 + version: 6.5.0(pg@8.21.0)(redis@5.12.1) dependency-cruiser: specifier: ^17.4.0 version: 17.4.0 @@ -338,6 +341,9 @@ importers: oxlint: specifier: ^1.79.0 version: 1.79.0 + redis: + specifier: 5.12.1 + version: 5.12.1 tsdown: specifier: 0.23.1 version: 0.23.1(publint@0.3.21)(tsx@4.22.3)(typescript@7.0.2) @@ -2958,6 +2964,36 @@ packages: resolution: {integrity: sha512-mQ2s0pYYiav+tzCDR05Zptem8Ey2v8s11lri5RKGhTtL4COVCvVCk5vtyRYNT+9L8qSfyOqqefF9UtnW8mC5jA==} engines: {node: '>= 20.19.0'} + '@msgpackr-extract/msgpackr-extract-darwin-arm64@3.0.4': + resolution: {integrity: sha512-LCkGo6JDfaBhgST7UpPWgNgLINpcpabaHfyz5OBx75nUYxBsaEPxjnyNjWpeb/xBup/682QnBfRBy2/LvPutZQ==} + cpu: [arm64] + os: [darwin] + + '@msgpackr-extract/msgpackr-extract-darwin-x64@3.0.4': + resolution: {integrity: sha512-zExlW9zUJKZH/tOtVMttwjKa4Xm/3KcNjnE3dPN92uCktwavMxpgCA3MoJK/DOnTWsQgo224OaST27/mPNAf+w==} + cpu: [x64] + os: [darwin] + + '@msgpackr-extract/msgpackr-extract-linux-arm64@3.0.4': + resolution: {integrity: sha512-dgX0P/9wGPJeHFBG+ZmhgE6bmtMt7NP5CRBGyyktpopdk/mW4POnrpQsSLtKI1dwpc+pPLuXHDh6vvskyQE/sw==} + cpu: [arm64] + os: [linux] + + '@msgpackr-extract/msgpackr-extract-linux-arm@3.0.4': + resolution: {integrity: sha512-Tg3yX65f5GbtXLkrYEHE5oibZG9epyYWas7FogTTEJeDEF9JlXJzKgXaNhT3UXlTOeA+AfZpYZYZ0uPj7Cfquw==} + cpu: [arm] + os: [linux] + + '@msgpackr-extract/msgpackr-extract-linux-x64@3.0.4': + resolution: {integrity: sha512-8TNXMEjJc3QEy7R/x1INhgiU+XakDAFUzBhaz7+Rbrs8NH5UQeHQxxmzsSBJGyV6I1jW79undiQm8tOI+D+8FQ==} + cpu: [x64] + os: [linux] + + '@msgpackr-extract/msgpackr-extract-win32-x64@3.0.4': + resolution: {integrity: sha512-CmCXPQrkbwExx3j946/PtHWHbYJiCRBRDl4BlkRQcJB/YOwQxJRTpoo7aTsortjgoJ1x7opzTSxn7C+ASSLVjQ==} + cpu: [x64] + os: [win32] + '@mswjs/interceptors@0.41.9': resolution: {integrity: sha512-VVPPgHyQ6ShqnrmDWuxjmUIsO9gWyOZFmuOfLd9LfBGQJwZfy0gvv9pbHSJuoFNIYC7ZDX9aoFwowjcdSC4E8w==} engines: {node: '>=18'} @@ -4107,6 +4143,14 @@ packages: '@sentry/junior@file:packages/junior': resolution: {directory: packages/junior, type: directory} + peerDependencies: + bullmq: ^6.5.0 + redis: ^5.12.1 + peerDependenciesMeta: + bullmq: + optional: true + redis: + optional: true '@sentry/nitro@11.10.0': resolution: {integrity: sha512-RuIH5u98J+a9KcFXSFb/CJ7j7ecQ7RtBB9WMr5oDPTZyLxJkkDVUMvGfeuiSsliyNBtSOP420rL9+NntOgArig==} @@ -5323,6 +5367,24 @@ packages: buffer@5.7.1: resolution: {integrity: sha512-EHcyIPBQ4BSGlvjB16k5KgAJ27CIsHY/2JBmCRReo48y9rQ3MaUzWX3KVlBa4U7MyX02HdVj0K7C3WaB3ju7FQ==} + bullmq@6.5.0: + resolution: {integrity: sha512-GDaUfEig2lWHS52aOI0h8f/W0m+5PPmHOJSahAKRMfi82BdL/2EqCuiDmJcNiwgpaZ0KHz7NxC8eNMI3bh8NQg==} + engines: {node: '>=14.18.0'} + peerDependencies: + bullmq-otel: '>=2.0.0' + ioredis: '>=5.0.0' + pg: '>=8.0.0' + redis: '>=5.0.0' + peerDependenciesMeta: + bullmq-otel: + optional: true + ioredis: + optional: true + pg: + optional: true + redis: + optional: true + bundle-name@4.1.1: resolution: {integrity: sha512-DdH81/zPLVS11EUgWq3tEu/xn+EzljlMYooDNdzWEnFha3R3NBMpMV1UqYIpjYHV/SgpFKMIX1Oh7o06SjM/oA==} engines: {node: '>=18'} @@ -5595,6 +5657,10 @@ packages: resolution: {integrity: sha512-tJtZBBHA6vjIAaF6EnIaq6laBBP9aq/Y3ouVJjEfoHbRBcHBAHYcMh/w8LDrk2PvIMMq8gmopa5D4V8RmbrxGw==} engines: {node: '>= 0.10'} + cron-parser@5.10.2: + resolution: {integrity: sha512-JfAYMY0LZaWQTuict8PhtF/ksSnq8xefSDmQUnnoqvkPYkU/y3UP0DTWzJKFBz0P1IXsDG8q0vY3zHkoFZOhvw==} + engines: {node: '>=18'} + cross-spawn@7.0.6: resolution: {integrity: sha512-uV2QOWP2nWzsy2aMp8aRibhi9dlzF5Hgh5SHaB9OiTGEyDTiJJyx0uy51QXdyWbtAHNua4XJzUKca3OzKUd3vA==} engines: {node: '>= 8'} @@ -7563,6 +7629,13 @@ packages: ms@2.1.3: resolution: {integrity: sha512-6FlzubTLZG3J2a/NVCAleEhjzq5oxgHyaCU9yYXvcLsvoVaHJq/s5xXI6/XXP6tz7R9xAOtHnSO/tXtF3WRTlA==} + msgpackr-extract@3.0.4: + resolution: {integrity: sha512-4kmO/MdyUIkLIvTPr8VHLil4AtoKIoniWPIEk5+CDy0xnWC84azhSFmuJ7PxZdsYtiP5kEeQsORAVIeMgxT+Hw==} + hasBin: true + + msgpackr@2.1.2: + resolution: {integrity: sha512-JHAdwubWNEadHQsFAib6uXOzKFxY8QOWHu9LCCksOINMC+TjgaKS2Tdrw7dW5SY0LLz8YK2/YqLSi2vigQt05w==} + msw@2.14.6: resolution: {integrity: sha512-ALe+N10S72cyx94cMcy3Zs4HhXCj35sgeAL4c+WTvKi0zWnbd8/h0lcFqv0mb2P+aSgAdD7p9HzvA0DiUPxsyg==} engines: {node: '>=18'} @@ -7651,6 +7724,9 @@ packages: resolution: {integrity: sha512-KdHvFWZjEKDf0cakgFjebl371GPsISX2oZHcuyKqM7DtogIsHrqKeLTo8wBHxaXRAQlY2PsPlZmfo+9ZCxEREQ==} engines: {node: '>=10'} + node-abort-controller@3.1.1: + resolution: {integrity: sha512-AGK2yQKIjRuqnc6VkX2Xj5d+QW8xZ87pa1UK6yA6ouUyuxfHuMP6umE5QK7UmTeOAymo+Zx1Fxiuw9rVx8taHQ==} + node-addon-api@7.1.1: resolution: {integrity: sha512-5m3bsyrjFWE1xf7nz7YXdN4udnVtXK6/Yfgn5qnahL6bCkf2yKt4k3nuTKAtT4r3IG8JNR2ncsIMdZuAzJjHQQ==} @@ -7701,6 +7777,10 @@ packages: resolution: {integrity: sha512-dRB78srN/l6gqWulah9SrxeYnxeddIG30+GOqK/9OlLVyLg3HPnr6SqOWTWOXKRwC2eGYCkZ59NNuSgvSrpgOA==} engines: {node: ^12.20.0 || ^14.13.1 || >=16.0.0} + node-gyp-build-optional-packages@5.2.2: + resolution: {integrity: sha512-s+w+rBWnpTMwSFbaE0UXsRlg7hU4FjekKU4eyAih5T8nJuNZT1nNsskXpxmeqSK9UzkBl6UgRlnKc8hz8IEqOw==} + hasBin: true + node-gyp-build@4.8.4: resolution: {integrity: sha512-LA4ZjwlnUblHVgq0oBF3Jl/6h/Nvs5fzBLwdEF4nuxnFdsfajde4WfxtJr3CaiH+F6ewcIB/q4jQ4UzPyid+CQ==} @@ -11959,6 +12039,24 @@ snapshots: prebuild-install: 7.1.3 optional: true + '@msgpackr-extract/msgpackr-extract-darwin-arm64@3.0.4': + optional: true + + '@msgpackr-extract/msgpackr-extract-darwin-x64@3.0.4': + optional: true + + '@msgpackr-extract/msgpackr-extract-linux-arm64@3.0.4': + optional: true + + '@msgpackr-extract/msgpackr-extract-linux-arm@3.0.4': + optional: true + + '@msgpackr-extract/msgpackr-extract-linux-x64@3.0.4': + optional: true + + '@msgpackr-extract/msgpackr-extract-win32-x64@3.0.4': + optional: true + '@mswjs/interceptors@0.41.9': dependencies: '@open-draft/deferred-promise': 2.2.0 @@ -14924,6 +15022,17 @@ snapshots: ieee754: 1.2.1 optional: true + bullmq@6.5.0(pg@8.21.0)(redis@5.12.1): + dependencies: + cron-parser: 5.10.2 + msgpackr: 2.1.2 + node-abort-controller: 3.1.1 + semver: 7.8.5 + tslib: 2.8.1 + optionalDependencies: + pg: 8.21.0 + redis: 5.12.1 + bundle-name@4.1.1: dependencies: run-applescript: 7.1.0 @@ -15142,6 +15251,10 @@ snapshots: object-assign: 4.1.1 vary: 1.1.2 + cron-parser@5.10.2: + dependencies: + luxon: 3.7.2 + cross-spawn@7.0.6: dependencies: path-key: 3.1.1 @@ -17516,6 +17629,22 @@ snapshots: ms@2.1.3: {} + msgpackr-extract@3.0.4: + dependencies: + node-gyp-build-optional-packages: 5.2.2 + optionalDependencies: + '@msgpackr-extract/msgpackr-extract-darwin-arm64': 3.0.4 + '@msgpackr-extract/msgpackr-extract-darwin-x64': 3.0.4 + '@msgpackr-extract/msgpackr-extract-linux-arm': 3.0.4 + '@msgpackr-extract/msgpackr-extract-linux-arm64': 3.0.4 + '@msgpackr-extract/msgpackr-extract-linux-x64': 3.0.4 + '@msgpackr-extract/msgpackr-extract-win32-x64': 3.0.4 + optional: true + + msgpackr@2.1.2: + optionalDependencies: + msgpackr-extract: 3.0.4 + msw@2.14.6(@types/node@25.9.1)(typescript@7.0.2): dependencies: '@inquirer/confirm': 6.0.13(@types/node@25.9.1) @@ -17748,6 +17877,8 @@ snapshots: semver: 7.8.5 optional: true + node-abort-controller@3.1.1: {} + node-addon-api@7.1.1: {} node-addon-api@8.8.0: @@ -17782,6 +17913,11 @@ snapshots: fetch-blob: 3.2.0 formdata-polyfill: 4.0.10 + node-gyp-build-optional-packages@5.2.2: + dependencies: + detect-libc: 2.1.2 + optional: true + node-gyp-build@4.8.4: {} node-html-markdown@2.0.0: diff --git a/pnpm-workspace.yaml b/pnpm-workspace.yaml index dee9bbed16..f8e4544726 100644 --- a/pnpm-workspace.yaml +++ b/pnpm-workspace.yaml @@ -62,6 +62,7 @@ allowBuilds: "@parcel/watcher": true agent-browser: true esbuild: true + msgpackr-extract: false msw: true node-liblzma: false protobufjs: true diff --git a/scripts/dev-server.mjs b/scripts/dev-server.mjs index f36caec7fb..f8dcbc7344 100644 --- a/scripts/dev-server.mjs +++ b/scripts/dev-server.mjs @@ -143,6 +143,8 @@ const tunnelUrl = process.env.CLOUDFLARE_TUNNEL_URL?.trim() || `http://localhost:${devPort}`; applyJuniorDevelopmentDefaults(process.env, { baseUrl: `http://localhost:${devPort}`, + // The Redis service in `compose.yaml`. + queueRedisUrl: "redis://localhost:6382", }); if ( !process.env.JUNIOR_SCHEDULER_SECRET?.trim() && diff --git a/scripts/lib/load-env-files.mjs b/scripts/lib/load-env-files.mjs index 43848ab565..2356793901 100644 --- a/scripts/lib/load-env-files.mjs +++ b/scripts/lib/load-env-files.mjs @@ -37,6 +37,19 @@ export function applyJuniorDevelopmentDefaults(env, options = {}) { if (!hasValue(env.JUNIOR_BASE_URL) && hasValue(options.baseUrl)) { env.JUNIOR_BASE_URL = options.baseUrl.trim(); } + // Only the dev server starts queue workers, so only it passes this option. + // Other local commands keep the Vercel queue. + if (hasValue(options.queueRedisUrl)) { + if (!hasValue(env.JUNIOR_QUEUE_PROVIDER)) { + env.JUNIOR_QUEUE_PROVIDER = "bullmq"; + } + if ( + env.JUNIOR_QUEUE_PROVIDER.trim() === "bullmq" && + !hasValue(env.JUNIOR_QUEUE_REDIS_URL) + ) { + env.JUNIOR_QUEUE_REDIS_URL = options.queueRedisUrl.trim(); + } + } } /** Load env files so app-local defaults override repo defaults without replacing shell env. */ diff --git a/scripts/lib/load-env-files.test.mjs b/scripts/lib/load-env-files.test.mjs index 3c1dfac42c..d0d088cf2c 100644 --- a/scripts/lib/load-env-files.test.mjs +++ b/scripts/lib/load-env-files.test.mjs @@ -116,6 +116,36 @@ describe("applyJuniorDevelopmentDefaults", () => { JUNIOR_LOCAL_DEV_HEARTBEAT_SECRET, ); assert.equal(env.JUNIOR_BASE_URL, "http://localhost:3000"); + assert.equal(env.JUNIOR_QUEUE_PROVIDER, undefined); + }); + + test("selects the BullMQ queue when the dev server gives a Redis URL", () => { + const env = { NODE_ENV: "development" }; + applyJuniorDevelopmentDefaults(env, { + queueRedisUrl: "redis://localhost:6382", + }); + assert.equal(env.JUNIOR_QUEUE_PROVIDER, "bullmq"); + assert.equal(env.JUNIOR_QUEUE_REDIS_URL, "redis://localhost:6382"); + + const vercelEnv = { + NODE_ENV: "development", + JUNIOR_QUEUE_PROVIDER: "vercel", + }; + applyJuniorDevelopmentDefaults(vercelEnv, { + queueRedisUrl: "redis://localhost:6382", + }); + assert.equal(vercelEnv.JUNIOR_QUEUE_PROVIDER, "vercel"); + assert.equal(vercelEnv.JUNIOR_QUEUE_REDIS_URL, undefined); + + const customEnv = { + NODE_ENV: "development", + JUNIOR_QUEUE_REDIS_URL: "redis://localhost:6390", + }; + applyJuniorDevelopmentDefaults(customEnv, { + queueRedisUrl: "redis://localhost:6382", + }); + assert.equal(customEnv.JUNIOR_QUEUE_PROVIDER, "bullmq"); + assert.equal(customEnv.JUNIOR_QUEUE_REDIS_URL, "redis://localhost:6390"); }); test("preserves explicit config values", () => { @@ -144,9 +174,11 @@ describe("applyJuniorDevelopmentDefaults", () => { applyJuniorDevelopmentDefaults(env, { baseUrl: "http://localhost:3000", + queueRedisUrl: "redis://localhost:6382", }); assert.equal(env.JUNIOR_SECRET, undefined); + assert.equal(env.JUNIOR_QUEUE_PROVIDER, undefined); assert.equal(env.JUNIOR_STATE_ADAPTER, undefined); assert.equal(env.JUNIOR_SCHEDULER_SECRET, undefined); assert.equal(env.JUNIOR_BASE_URL, undefined); From 2fa1f5cb477b381b843bdd861eb1fcb412f75252 Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Thu, 29 Oct 2026 16:12:28 +0000 Subject: [PATCH 2/5] fix(queue): Run local BullMQ consumers in one worker process Nitro now only sends BullMQ jobs. scripts/dev-server.mjs starts one sibling worker process (scripts/dev-queue-worker.mjs) that creates the example app and calls the new startQueueWorker(app). The worker owns its BullMQ workers, queues, Redis client, state adapter and database. On SIGINT or SIGTERM it waits up to 10 seconds for running jobs, then closes all of them. A new Junior build restarts Nitro and the worker. A change to the example app's server, plugins, dashboard or app files restarts only the worker. Nitro HMR no longer leaves old workers taking jobs. The worker is a separate process, so pnpm dev keeps state in the local queue Redis unless JUNIOR_STATE_ADAPTER is set, and refuses memory state with BullMQ. BullMQ concurrency defaults to 4 per queue topic. JUNIOR_QUEUE_WORKER_CONCURRENCY overrides it, and the worker logs the value at startup. The queue provider contract gains an optional startWorker(), and registerDevConsumer becomes optional. The Vercel provider and production behavior do not change. Co-Authored-By: Thomas Jiang <20501235+thomasjiangcy@users.noreply.github.com> --- .../content/docs/contribute/development.md | 2 + .../content/docs/reference/config-and-env.md | 8 +- packages/junior/src/app.ts | 47 ++++++++ .../junior/src/chat/plugins/task-queue.ts | 6 + packages/junior/src/chat/queue/README.md | 16 ++- packages/junior/src/chat/queue/bullmq.ts | 91 ++++++++++----- packages/junior/src/chat/queue/callback.ts | 3 +- packages/junior/src/chat/queue/config.ts | 31 ++++- packages/junior/src/chat/queue/provider.ts | 18 ++- .../src/chat/sandbox/snapshot/job-callback.ts | 10 +- .../chat/task-execution/vercel-callback.ts | 8 ++ .../component/config/chat-config.test.ts | 16 +++ .../tests/integration/queue-bullmq.test.ts | 56 +++++---- scripts/dev-queue-worker.mjs | 48 ++++++++ scripts/dev-server.mjs | 106 ++++++++++++++++-- scripts/lib/load-env-files.mjs | 23 ++-- scripts/lib/load-env-files.test.mjs | 22 ++++ 17 files changed, 435 insertions(+), 76 deletions(-) create mode 100644 scripts/dev-queue-worker.mjs diff --git a/packages/docs/src/content/docs/contribute/development.md b/packages/docs/src/content/docs/contribute/development.md index a60e4585f4..8a6b77ca23 100644 --- a/packages/docs/src/content/docs/contribute/development.md +++ b/packages/docs/src/content/docs/contribute/development.md @@ -43,6 +43,8 @@ This starts the example app on `http://localhost:3000` by default. It also rebui `pnpm dev` delivers background work through BullMQ on the Redis in `compose.yaml` (`redis://localhost:6382`). Start it with `docker compose up -d redis`. Local work then needs no Vercel Queues access. To use Vercel Queues, set `JUNIOR_QUEUE_PROVIDER=vercel` and run `pnpm dev:env` for the Vercel token. To use another Redis, set `JUNIOR_QUEUE_REDIS_URL`. +With BullMQ, `pnpm dev` starts one queue worker process next to Nitro. Nitro accepts requests and sends jobs. The worker runs them. A new Junior build restarts both processes. A change to `apps/example/server.ts`, `plugins.ts`, `dashboard.ts` or `apps/example/app/` restarts only the worker. The worker and Nitro must share state, so `pnpm dev` keeps state in that Redis unless you set `JUNIOR_STATE_ADAPTER`. `JUNIOR_STATE_ADAPTER=memory` does not work with BullMQ. Each queue topic runs up to 4 jobs at the same time. Set `JUNIOR_QUEUE_WORKER_CONCURRENCY` to change that number. + To also open the Sentry Local UI in your browser, set `OPEN_SPOTLIGHT_WEB=1`: ```bash diff --git a/packages/docs/src/content/docs/reference/config-and-env.md b/packages/docs/src/content/docs/reference/config-and-env.md index 38baa8d1e3..539d5834e1 100644 --- a/packages/docs/src/content/docs/reference/config-and-env.md +++ b/packages/docs/src/content/docs/reference/config-and-env.md @@ -53,6 +53,7 @@ related: | `JUNIOR_S3_REGION` | No | S3 signing region. Defaults to `us-east-1`. | | `JUNIOR_QUEUE_PROVIDER` | No | Queue that delivers background work: `vercel` or `bullmq`. Defaults to `vercel`. Use `bullmq` only for local development. | | `JUNIOR_QUEUE_REDIS_URL` | Conditional | Redis for the BullMQ queue, such as `redis://localhost:6382`. Required when the provider is `bullmq`. | +| `JUNIOR_QUEUE_WORKER_CONCURRENCY` | No | Jobs that the BullMQ worker runs at the same time for each queue topic. A positive integer. Defaults to `4`. | For Vercel deployments, create a private Blob store and connect it to the project before using `sendFiles` or `publishImage`. Prefer an OIDC connection. @@ -71,8 +72,11 @@ Junior loses them when the process stops, and each process has its own copy. Do not use it for a deployed app. `JUNIOR_QUEUE_PROVIDER=bullmq` delivers background work through BullMQ on -Redis, with workers in the dev server process. Local development then needs no -Vercel Queues access. Install the optional `bullmq` and `redis` packages to use +Redis. Local development then needs no Vercel Queues access. The app only +sends jobs. One separate worker process runs them. `pnpm dev` starts that +process, and the worker logs `queue.bullmq.worker.started` with its +concurrency for each queue topic. Because the worker is a separate process, +state must be in Redis, not memory. Install the optional `bullmq` and `redis` packages to use it. Junior loads them only when this provider is selected, so they never enter a deployed bundle. Give the queue its own Redis URL. If two dev servers share one Redis, each one can take the other's jobs. diff --git a/packages/junior/src/app.ts b/packages/junior/src/app.ts index 426c3b1f20..9788d596b7 100644 --- a/packages/junior/src/app.ts +++ b/packages/junior/src/app.ts @@ -77,6 +77,7 @@ import { } from "@/deployment"; import { consumeConversationQueueMessage, + conversationWorkQueueConsumer, createVercelConversationWorkCallback, registerVercelConversationWorkDevConsumer, } from "@/chat/task-execution/vercel-callback"; @@ -89,6 +90,7 @@ import { bindMessageAgent } from "@/chat/agent-invocations/message"; import { bindSpawnAgent } from "@/chat/agent-invocations/spawn"; import { createVercelPluginTaskCallback, + pluginTaskQueueConsumer, registerVercelPluginTaskDevConsumer, type PluginTaskQueueMessage, } from "@/chat/plugins/task-queue"; @@ -96,7 +98,15 @@ import { processPluginTask } from "@/chat/plugins/task-runner"; import { createVercelWorkspaceSnapshotJobCallback, registerVercelWorkspaceSnapshotJobDevConsumer, + workspaceSnapshotJobQueueConsumer, } from "@/chat/sandbox/snapshot/job-callback"; +import { + getQueueProvider, + type QueueConsumer, + type QueueWorker, +} from "@/chat/queue/provider"; +import { closeDb } from "@/chat/db"; +import { disconnectStateAdapter } from "@/chat/state/adapter"; import { createProductionConversationWorkOptions, createProductionSlackWebhookServices, @@ -410,6 +420,38 @@ function mountRoutes(app: Hono, routes: HostRouteRegistration[]): void { } } +// The queue consumers of each app from `createApp()`, for `startQueueWorker()`. +const appQueueConsumers = new WeakMap QueueConsumer[]>(); + +/** + * Pull this app's queue messages in this process until `close()`. + * + * Only for a queue provider whose consumers pull messages, such as the + * local BullMQ queue. The dev server runs it in its own worker process, so + * the Nitro process only sends messages. `close()` also closes the database + * and state connections of this process. + */ +export async function startQueueWorker(app: Hono): Promise { + const consumers = appQueueConsumers.get(app); + if (!consumers) { + throw new Error("startQueueWorker() needs an app from createApp()"); + } + const provider = getQueueProvider(); + if (!provider.startWorker) { + throw new Error( + `The ${provider.messagingSystem} queue provider has no queue worker`, + ); + } + const worker = await provider.startWorker(consumers()); + return { + close: async () => { + await worker.close(); + await disconnectStateAdapter(); + await closeDb(); + }, + }; +} + /** Create a Hono app with all Junior routes. */ export async function createApp(options?: JuniorAppOptions): Promise { const virtualConfig = await resolveVirtualConfig(); @@ -701,6 +743,11 @@ export async function createApp(options?: JuniorAppOptions): Promise { let pluginTaskPOST: | ReturnType | undefined; + appQueueConsumers.set(app, () => [ + conversationWorkQueueConsumer(getConversationWorkOptions()), + pluginTaskQueueConsumer(), + workspaceSnapshotJobQueueConsumer(), + ]); if (process.env.NODE_ENV === "development") { registerVercelConversationWorkDevConsumer(getConversationWorkOptions()); registerVercelPluginTaskDevConsumer(); diff --git a/packages/junior/src/chat/plugins/task-queue.ts b/packages/junior/src/chat/plugins/task-queue.ts index 9324fbea27..c8d1b3d92a 100644 --- a/packages/junior/src/chat/plugins/task-queue.ts +++ b/packages/junior/src/chat/plugins/task-queue.ts @@ -10,6 +10,7 @@ import { logWarn } from "@/chat/logging"; import { queueCallback } from "@/chat/queue/callback"; import { getQueueProvider, + type QueueConsumer, type QueueMessageMetadata, } from "@/chat/queue/provider"; import { @@ -139,6 +140,11 @@ export function createVercelPluginTaskCallback(): ( return pluginTaskCallback().create(); } +/** Return the queue consumer for plugin tasks, for a pulling queue worker. */ +export function pluginTaskQueueConsumer(): QueueConsumer { + return pluginTaskCallback().consumer; +} + /** Register the local-dev consumer for plugin tasks. */ export function registerVercelPluginTaskDevConsumer(): | (() => void) diff --git a/packages/junior/src/chat/queue/README.md b/packages/junior/src/chat/queue/README.md index 09f8e53b99..1e0652ac37 100644 --- a/packages/junior/src/chat/queue/README.md +++ b/packages/junior/src/chat/queue/README.md @@ -20,9 +20,19 @@ Conversation work is different. It only wakes a conversation. Keep using `@vercel/queue`, and only `provider.ts` imports it. Two ast-grep rules enforce these boundaries. - `bullmq.ts` implements `QueueProvider` with BullMQ on Redis, for local - development only. `pnpm dev` selects it. Its workers start where - `registerDevConsumer` runs, in the dev server process, and pull jobs from - Redis. Nothing pushes to the callback routes. + development only. `pnpm dev` selects it. The Nitro process only sends jobs. + `scripts/dev-queue-worker.mjs` runs in its own process, creates the app, and + calls `startQueueWorker()` from `app.ts`. That calls the provider's + `startWorker()`, which pulls jobs from Redis. Nothing pushes to the + callback routes. +- `startWorker()` returns a worker whose `close()` waits up to 10 seconds for + running jobs, then closes each BullMQ worker and queue and the Redis client. + `startQueueWorker()` then closes the state and database connections. Run + one worker in each process. `close()` releases the connections of the whole + process. +- `registerDevConsumer` is for a provider that delivers each message in the + process that sent it, like the Vercel SDK in dev. BullMQ does not implement + it. - `bullmq.ts` imports `bullmq` and `redis` by a name in a variable, so bundlers do not put them into the production server. `apps/example/scripts/check-vercel-output.mjs` fails if they reach a diff --git a/packages/junior/src/chat/queue/bullmq.ts b/packages/junior/src/chat/queue/bullmq.ts index 35dbb0a2f5..a1678e02af 100644 --- a/packages/junior/src/chat/queue/bullmq.ts +++ b/packages/junior/src/chat/queue/bullmq.ts @@ -1,16 +1,21 @@ import { createHash } from "node:crypto"; import type { Job, Queue, Worker } from "bullmq"; import { getChatConfig } from "@/chat/config"; -import { logException } from "@/chat/logging"; +import { logException, logInfo, logWarn } from "@/chat/logging"; import type { QueueConsumer, QueueMessageMetadata, QueueProvider, + QueueWorker, } from "@/chat/queue/provider"; // Module overview: a queue provider for local development, backed by BullMQ -// on Redis. Workers run in this process, so `pnpm dev` needs no Vercel -// Queues access. Deployed apps use the Vercel provider. +// on Redis, so `pnpm dev` needs no Vercel Queues access. Deployed apps use +// the Vercel provider. +// +// The Nitro process only sends jobs. The dev server starts one separate +// worker process that calls `startWorker()` and pulls the jobs. `close()` +// on that worker releases every BullMQ and Redis resource of its process. // // `bullmq` and `redis` are optional peer dependencies. Only dev installs // need them. @@ -28,9 +33,9 @@ type RedisConnection = ReturnType; // from looping for ever in dev. const MAX_ATTEMPTS = 50; const MAX_BACKOFF_MS = 30_000; -// BullMQ runs one job at a time unless told otherwise. One long turn would -// then block every other conversation. -const WORKER_CONCURRENCY = 20; +// Running jobs get this long to finish when the worker stops. After that, +// BullMQ stops waiting. Their locks expire, and the jobs run again later. +const SHUTDOWN_GRACE_MS = 10_000; // A job id drops a second send only while BullMQ still stores the first job. // So this is the dedupe window for idempotency keys. const COMPLETED_JOB_AGE_SECONDS = 60 * 60; @@ -52,12 +57,12 @@ function importAtRuntime(specifier: string): Promise { return import(/* @vite-ignore */ specifier); } -function redisUrl(): string { +function bullmqConfig() { const config = getChatConfig().queue; if (config.provider !== "bullmq") { throw new Error("The BullMQ queue provider is not configured"); } - return config.redisUrl; + return config; } /** Load BullMQ and one shared Redis client the first time they are needed. */ @@ -72,7 +77,7 @@ function load() { { cause: error }, ); })) as [BullMq, typeof import("redis")]; - const client = redis.createClient({ url: redisUrl() }); + const client = redis.createClient({ url: bullmqConfig().redisUrl }); await client.connect(); return { bullmq, connection: bullmq.createNodeRedisClient(client) }; })(); @@ -131,19 +136,46 @@ async function processJob( } } -/** Start one in-process worker. The returned function stops it. */ -function startWorker(consumer: QueueConsumer): () => Promise { - let worker: Worker | undefined; - let stopped = false; - const started = load().then(({ bullmq, connection }) => { - if (stopped) { - return; - } - worker = new bullmq.Worker( +/** Wait for running jobs, but stop waiting after the grace period. */ +async function closeWorker(worker: Worker): Promise { + let timer: NodeJS.Timeout | undefined; + const graceful = worker.close().then(() => true); + const timedOut = new Promise((resolve) => { + timer = setTimeout(() => resolve(false), SHUTDOWN_GRACE_MS); + }); + const finished = await Promise.race([graceful, timedOut]); + clearTimeout(timer); + if (!finished) { + logWarn("queue.bullmq.worker.shutdown_timeout", { + "messaging.destination.name": worker.name, + }); + await worker.close(true); + } +} + +/** Release the queues and the Redis client of this process. */ +async function closeConnections(): Promise { + const current = loaded; + loaded = undefined; + const open = [...queues.values()]; + queues.clear(); + await Promise.all(open.map((queue) => queue.close())); + if (current) { + const { connection } = await current; + await connection.quit(); + } +} + +/** Start one BullMQ worker for each consumer in this process. */ +async function startWorker(consumers: QueueConsumer[]): Promise { + const { bullmq, connection } = await load(); + const concurrency = bullmqConfig().workerConcurrency; + const workers = consumers.map((consumer) => { + const worker = new bullmq.Worker( consumer.topic, (job) => processJob(job, consumer), { - concurrency: WORKER_CONCURRENCY, + concurrency, connection, // BullMQ renews this lock while the job runs. Unlike Vercel Queues, // a running job is not sent again when the time runs out. @@ -161,11 +193,17 @@ function startWorker(consumer: QueueConsumer): () => Promise { "messaging.destination.name": consumer.topic, }); }); + return worker; }); - return async () => { - stopped = true; - await started; - await worker?.close(); + logInfo("queue.bullmq.worker.started", { + "app.queue.worker_concurrency": concurrency, + "messaging.destination.names": consumers.map((consumer) => consumer.topic), + }); + return { + close: async () => { + await Promise.all(workers.map((worker) => closeWorker(worker))); + await closeConnections(); + }, }; } @@ -195,10 +233,5 @@ export const bullmqQueueProvider: QueueProvider = { new Response("This queue provider does not push messages", { status: 404, }), - registerDevConsumer: (consumer) => { - const stop = startWorker(consumer); - return () => { - void stop(); - }; - }, + startWorker, }; diff --git a/packages/junior/src/chat/queue/callback.ts b/packages/junior/src/chat/queue/callback.ts index febebcbf86..31bc11e50e 100644 --- a/packages/junior/src/chat/queue/callback.ts +++ b/packages/junior/src/chat/queue/callback.ts @@ -98,12 +98,13 @@ export function queueCallback(options: QueueCallbackOptions) { return { /** Run a message that an in-process queue already trusts. */ consume, + consumer, create: () => getQueueProvider().createCallback(consumer), registerDev: () => { if (process.env.NODE_ENV !== "development") { return undefined; } - return getQueueProvider().registerDevConsumer(consumer); + return getQueueProvider().registerDevConsumer?.(consumer); }, }; } diff --git a/packages/junior/src/chat/queue/config.ts b/packages/junior/src/chat/queue/config.ts index eb884b474b..bd765ed2ff 100644 --- a/packages/junior/src/chat/queue/config.ts +++ b/packages/junior/src/chat/queue/config.ts @@ -3,7 +3,28 @@ import { toOptionalTrimmed } from "@/chat/optional-string"; /** Which queue delivers background work. */ export type QueueConfig = | { provider: "vercel" } - | { provider: "bullmq"; redisUrl: string }; + | { provider: "bullmq"; redisUrl: string; workerConcurrency: number }; + +/** Jobs that one BullMQ worker runs at the same time, for each queue topic. */ +const DEFAULT_BULLMQ_WORKER_CONCURRENCY = 4; + +function readWorkerConcurrency(value: string | undefined): number { + const raw = toOptionalTrimmed(value); + if (raw === undefined) { + return DEFAULT_BULLMQ_WORKER_CONCURRENCY; + } + const concurrency = Number(raw); + if ( + !/^\d+$/.test(raw) || + !Number.isSafeInteger(concurrency) || + concurrency < 1 + ) { + throw new Error( + "JUNIOR_QUEUE_WORKER_CONCURRENCY must be a positive integer", + ); + } + return concurrency; +} /** Read queue settings from environment variables. */ export function readQueueConfig(env: NodeJS.ProcessEnv): QueueConfig { @@ -18,7 +39,13 @@ export function readQueueConfig(env: NodeJS.ProcessEnv): QueueConfig { "JUNIOR_QUEUE_REDIS_URL is required when JUNIOR_QUEUE_PROVIDER is bullmq", ); } - return { provider, redisUrl }; + return { + provider, + redisUrl, + workerConcurrency: readWorkerConcurrency( + env.JUNIOR_QUEUE_WORKER_CONCURRENCY, + ), + }; } default: throw new Error("JUNIOR_QUEUE_PROVIDER must be vercel or bullmq"); diff --git a/packages/junior/src/chat/queue/provider.ts b/packages/junior/src/chat/queue/provider.ts index c48669bff6..e32ee3ed47 100644 --- a/packages/junior/src/chat/queue/provider.ts +++ b/packages/junior/src/chat/queue/provider.ts @@ -65,8 +65,22 @@ export interface QueueProvider { createCallback( consumer: QueueConsumer, ): (request: Request) => Promise; - /** Receive messages in this process during local development. */ - registerDevConsumer(consumer: QueueConsumer): () => void; + /** + * Receive messages in the process that sends them during local development. + * Only providers that deliver this way implement it. + */ + registerDevConsumer?(consumer: QueueConsumer): () => void; + /** + * Pull messages for these consumers in this process until `close()`. + * Only providers whose consumers pull messages implement it. + */ + startWorker?(consumers: QueueConsumer[]): Promise; +} + +/** Consumers that pull messages in one process. */ +export interface QueueWorker { + /** Stop taking messages, wait for running ones, and release connections. */ + close(): Promise; } /** Return the queue provider for this deployment. */ diff --git a/packages/junior/src/chat/sandbox/snapshot/job-callback.ts b/packages/junior/src/chat/sandbox/snapshot/job-callback.ts index 259da1f09b..52fb645edb 100644 --- a/packages/junior/src/chat/sandbox/snapshot/job-callback.ts +++ b/packages/junior/src/chat/sandbox/snapshot/job-callback.ts @@ -1,7 +1,10 @@ /** Receive queue messages for Workspace snapshot builds. */ import { logWarn } from "@/chat/logging"; import { queueCallback } from "@/chat/queue/callback"; -import type { QueueMessageMetadata } from "@/chat/queue/provider"; +import type { + QueueConsumer, + QueueMessageMetadata, +} from "@/chat/queue/provider"; import { processWorkspaceSnapshotJob } from "./job-runner"; import { verifyWorkspaceSnapshotJobMessage, @@ -42,6 +45,11 @@ export function createVercelWorkspaceSnapshotJobCallback(): ( return workspaceSnapshotJobCallback().create(); } +/** Return the queue consumer for snapshot builds, for a pulling queue worker. */ +export function workspaceSnapshotJobQueueConsumer(): QueueConsumer { + return workspaceSnapshotJobCallback().consumer; +} + /** Receive snapshot build messages during local development. */ export function registerVercelWorkspaceSnapshotJobDevConsumer(): | (() => void) diff --git a/packages/junior/src/chat/task-execution/vercel-callback.ts b/packages/junior/src/chat/task-execution/vercel-callback.ts index fb7e5a5b14..f6a1e77594 100644 --- a/packages/junior/src/chat/task-execution/vercel-callback.ts +++ b/packages/junior/src/chat/task-execution/vercel-callback.ts @@ -4,6 +4,7 @@ import { logWarn, withLogContext, withSpan } from "@/chat/logging"; import { queueCallback } from "@/chat/queue/callback"; import { getQueueProvider, + type QueueConsumer, type QueueMessageMetadata, } from "@/chat/queue/provider"; import type { ConversationStore } from "@/chat/conversations/store"; @@ -216,6 +217,13 @@ export function createVercelConversationWorkCallback( return conversationWorkCallback(options).create(); } +/** Return the queue consumer for conversation work, for a pulling queue worker. */ +export function conversationWorkQueueConsumer( + options: VercelConversationWorkCallbackOptions, +): QueueConsumer { + return conversationWorkCallback(options).consumer; +} + /** Register the Vercel Queue local-dev consumer for Nitro's central route dispatcher. */ export function registerVercelConversationWorkDevConsumer( options: VercelConversationWorkCallbackOptions, diff --git a/packages/junior/tests/component/config/chat-config.test.ts b/packages/junior/tests/component/config/chat-config.test.ts index 8274fcfc4e..be15a93993 100644 --- a/packages/junior/tests/component/config/chat-config.test.ts +++ b/packages/junior/tests/component/config/chat-config.test.ts @@ -420,6 +420,12 @@ describe("chat config", () => { expect((await loadConfig()).getChatConfig().queue).toEqual({ provider: "bullmq", redisUrl: "redis://localhost:6382", + workerConcurrency: 4, + }); + + process.env.JUNIOR_QUEUE_WORKER_CONCURRENCY = "2"; + expect((await loadConfig()).getChatConfig().queue).toMatchObject({ + workerConcurrency: 2, }); }); @@ -431,6 +437,16 @@ describe("chat config", () => { ); }); + it("throws when the BullMQ worker concurrency is not a positive integer", async () => { + process.env.JUNIOR_QUEUE_PROVIDER = "bullmq"; + process.env.JUNIOR_QUEUE_REDIS_URL = "redis://localhost:6382"; + process.env.JUNIOR_QUEUE_WORKER_CONCURRENCY = "0"; + + await expect(loadConfig()).rejects.toThrow( + "JUNIOR_QUEUE_WORKER_CONCURRENCY must be a positive integer", + ); + }); + it("enables conversation work by default", async () => { const { getChatConfig } = await loadConfig(); expect(getChatConfig().conversationWorkEnabled).toBe(true); diff --git a/packages/junior/tests/integration/queue-bullmq.test.ts b/packages/junior/tests/integration/queue-bullmq.test.ts index d8e8ea2e41..8e546b1c8a 100644 --- a/packages/junior/tests/integration/queue-bullmq.test.ts +++ b/packages/junior/tests/integration/queue-bullmq.test.ts @@ -1,9 +1,10 @@ import { randomUUID } from "node:crypto"; -import { afterAll, describe, expect, it, vi } from "vitest"; +import { afterEach, describe, expect, it, vi } from "vitest"; import { getQueueProvider, type QueueConsumer, type QueueMessageMetadata, + type QueueWorker, } from "@/chat/queue/provider"; // Chat config reads the queue provider when it loads, so set it first. CI @@ -14,31 +15,30 @@ vi.hoisted(() => { process.env.REDIS_URL?.trim() || "redis://localhost:6382"; }); -const stops: Array<() => void> = []; +let worker: QueueWorker | undefined; -afterAll(() => { - for (const stop of stops) { - stop(); - } +afterEach(async () => { + await worker?.close(); + worker = undefined; }); function uniqueTopic(name: string): string { return `junior_test_${name}_${randomUUID()}`; } -function startConsumer( +async function startConsumer( topic: string, handler: QueueConsumer["handler"], retry: QueueConsumer["retry"] = () => undefined, -): void { - stops.push( - getQueueProvider().registerDevConsumer({ - consumerGroup: `${topic}_dev`, - handler, - retry, - topic, - }), - ); +): Promise { + const provider = getQueueProvider(); + if (!provider.startWorker) { + throw new Error("The BullMQ provider must start workers"); + } + worker = await provider.startWorker([ + { consumerGroup: `${topic}_dev`, handler, retry, topic }, + ]); + return worker; } async function waitFor(check: () => boolean, timeoutMs = 10_000) { @@ -58,7 +58,7 @@ describe("BullMQ queue provider", () => { metadata: QueueMessageMetadata; payload: unknown; }> = []; - startConsumer(topic, async (payload, metadata) => { + await startConsumer(topic, async (payload, metadata) => { deliveries.push({ metadata, payload }); }); @@ -94,7 +94,7 @@ describe("BullMQ queue provider", () => { const topic = uniqueTopic("retry"); const deliveryCounts: number[] = []; let acknowledged = false; - startConsumer( + await startConsumer( topic, async (_payload, metadata) => { deliveryCounts.push(metadata.deliveryCount); @@ -121,7 +121,7 @@ describe("BullMQ queue provider", () => { let running = 0; let mostRunning = 0; let finished = 0; - startConsumer(topic, async () => { + await startConsumer(topic, async () => { running += 1; mostRunning = Math.max(mostRunning, running); await new Promise((resolve) => setTimeout(resolve, 500)); @@ -136,4 +136,22 @@ describe("BullMQ queue provider", () => { expect(mostRunning).toBe(3); }); + + it("lets a running message finish before close() returns", async () => { + const topic = uniqueTopic("close"); + let started = false; + let finished = false; + const running = await startConsumer(topic, async () => { + started = true; + await new Promise((resolve) => setTimeout(resolve, 500)); + finished = true; + }); + + await getQueueProvider().send(topic, {}); + await waitFor(() => started); + await running.close(); + worker = undefined; + + expect(finished).toBe(true); + }); }); diff --git a/scripts/dev-queue-worker.mjs b/scripts/dev-queue-worker.mjs new file mode 100644 index 0000000000..c7d8df0b5a --- /dev/null +++ b/scripts/dev-queue-worker.mjs @@ -0,0 +1,48 @@ +// The local queue worker process that `scripts/dev-server.mjs` starts when +// `pnpm dev` uses the BullMQ queue. Nitro only sends jobs. This process +// creates the example app, pulls its jobs, and owns every connection it +// opens. The dev server passes the environment and restarts this process +// when the Junior build or the example app's runtime files change. +import { createRequire } from "node:module"; +import path from "node:path"; +import { fileURLToPath, pathToFileURL } from "node:url"; + +const exampleDir = path.resolve( + path.dirname(fileURLToPath(import.meta.url)), + "..", + "apps", + "example", +); + +// Import `@sentry/junior` from the path that the example app resolves. +// Another copy would hold other app state and find no app consumers. +const juniorEntry = createRequire(path.join(exampleDir, "server.ts")).resolve( + "@sentry/junior", +); +const { startQueueWorker } = await import(pathToFileURL(juniorEntry).href); +const { default: app } = await import( + pathToFileURL(path.join(exampleDir, "server.ts")).href +); +const worker = await startQueueWorker(app); + +let stopping = false; +async function stop(signal) { + if (stopping) { + // A second signal stops at once, without waiting for running jobs. + process.exit(1); + } + stopping = true; + try { + await worker.close(); + } catch (error) { + console.error(`Queue worker did not close cleanly (${signal}):`, error); + process.exit(1); + } + process.exit(0); +} + +for (const signal of ["SIGINT", "SIGTERM"]) { + process.on(signal, () => { + void stop(signal); + }); +} diff --git a/scripts/dev-server.mjs b/scripts/dev-server.mjs index f8dcbc7344..603ce5704a 100644 --- a/scripts/dev-server.mjs +++ b/scripts/dev-server.mjs @@ -146,6 +146,17 @@ applyJuniorDevelopmentDefaults(process.env, { // The Redis service in `compose.yaml`. queueRedisUrl: "redis://localhost:6382", }); +// With BullMQ, one separate worker process pulls the queue jobs. +const runQueueWorker = process.env.JUNIOR_QUEUE_PROVIDER?.trim() === "bullmq"; +if ( + runQueueWorker && + process.env.JUNIOR_STATE_ADAPTER?.trim().toLowerCase() === "memory" +) { + console.error( + "JUNIOR_STATE_ADAPTER=memory does not work with the BullMQ queue: its worker runs in a separate process. Use JUNIOR_STATE_ADAPTER=redis or JUNIOR_QUEUE_PROVIDER=vercel.", + ); + process.exit(1); +} if ( !process.env.JUNIOR_SCHEDULER_SECRET?.trim() && !process.env.CRON_SECRET?.trim() @@ -264,18 +275,63 @@ function restartNitroDev() { stopChild(nitroChild, "SIGTERM"); } -function watchDistForNitroRestart() { +let queueWorkerChild; +let restartingQueueWorker = false; + +function startQueueWorker() { + const child = spawnChild( + process.execPath, + [path.join(workspaceRoot, "scripts", "dev-queue-worker.mjs")], + { cwd: exampleDir }, + ); + queueWorkerChild = child; + + // A worker that fails, for example on a syntax error, does not stop the + // dev server. The next watched change starts it again. + child.on("exit", (code, signal) => { + if (queueWorkerChild === child) { + queueWorkerChild = undefined; + } + if (!stoppedChildren.has(child)) { + console.error( + `Queue worker exited (${signal ?? `code ${code}`}). It starts again after the next change to Junior or the example app.`, + ); + } + }); +} + +function restartQueueWorker() { + if (restartingQueueWorker) { + return; + } + if (!queueWorkerChild) { + startQueueWorker(); + return; + } + + // Start the new worker only after the old one has closed, so that old + // code never takes new jobs. + restartingQueueWorker = true; + queueWorkerChild.once("exit", () => { + restartingQueueWorker = false; + startQueueWorker(); + }); + stopChild(queueWorkerChild, "SIGTERM"); +} + +function watchForRestart(paths, restart, options = {}) { + const { includeFile = () => true, recursive = false } = options; let timer; - const scheduleRestart = () => { + const scheduleRestart = (_event, filename) => { + if (filename && !includeFile(filename)) { + return; + } clearTimeout(timer); - timer = setTimeout(restartNitroDev, 1500); + timer = setTimeout(restart, 1500); }; - for (const distDir of [ - path.join(juniorPackageDir, "dist"), - path.join(dashboardPackageDir, "dist"), - ]) { - const watcher = fs.watch(distDir, scheduleRestart); + for (const watchedPath of paths) { + const watcher = fs.watch(watchedPath, { recursive }, scheduleRestart); children.add({ killed: false, kill() { @@ -287,6 +343,35 @@ function watchDistForNitroRestart() { } } +function watchForRestarts() { + // A new Junior build restarts both processes. + watchForRestart( + [ + path.join(juniorPackageDir, "dist"), + path.join(dashboardPackageDir, "dist"), + ], + () => { + restartNitroDev(); + if (runQueueWorker) { + restartQueueWorker(); + } + }, + ); + if (!runQueueWorker) { + return; + } + // Nitro reloads the example app by itself. The worker loads the app once, + // so a change to its runtime files restarts only the worker. Watch the + // folder, because editors often save a file by replacing it. + const appEntryFiles = new Set(["server.ts", "plugins.ts", "dashboard.ts"]); + watchForRestart([exampleDir], restartQueueWorker, { + includeFile: (filename) => appEntryFiles.has(filename), + }); + watchForRestart([path.join(exampleDir, "app")], restartQueueWorker, { + recursive: true, + }); +} + runRequiredChild("pnpm", ["build"], { cwd: juniorPackageDir, }); @@ -324,8 +409,11 @@ if (tunnelToken) { ]); } -watchDistForNitroRestart(); +watchForRestarts(); startNitroDev({ openLocalUi }); +if (runQueueWorker) { + startQueueWorker(); +} startLocalHeartbeat(); for (const signal of ["SIGINT", "SIGTERM"]) { diff --git a/scripts/lib/load-env-files.mjs b/scripts/lib/load-env-files.mjs index 2356793901..38b332b316 100644 --- a/scripts/lib/load-env-files.mjs +++ b/scripts/lib/load-env-files.mjs @@ -28,9 +28,6 @@ export function applyJuniorDevelopmentDefaults(env, options = {}) { if (!hasValue(env.JUNIOR_SECRET)) { env.JUNIOR_SECRET = JUNIOR_LOCAL_DEV_INTERNAL_SECRET; } - if (!hasValue(env.JUNIOR_STATE_ADAPTER)) { - env.JUNIOR_STATE_ADAPTER = "memory"; - } if (!hasValue(env.JUNIOR_SCHEDULER_SECRET) && !hasValue(env.CRON_SECRET)) { env.JUNIOR_SCHEDULER_SECRET = JUNIOR_LOCAL_DEV_HEARTBEAT_SECRET; } @@ -43,13 +40,23 @@ export function applyJuniorDevelopmentDefaults(env, options = {}) { if (!hasValue(env.JUNIOR_QUEUE_PROVIDER)) { env.JUNIOR_QUEUE_PROVIDER = "bullmq"; } - if ( - env.JUNIOR_QUEUE_PROVIDER.trim() === "bullmq" && - !hasValue(env.JUNIOR_QUEUE_REDIS_URL) - ) { - env.JUNIOR_QUEUE_REDIS_URL = options.queueRedisUrl.trim(); + if (env.JUNIOR_QUEUE_PROVIDER.trim() === "bullmq") { + if (!hasValue(env.JUNIOR_QUEUE_REDIS_URL)) { + env.JUNIOR_QUEUE_REDIS_URL = options.queueRedisUrl.trim(); + } + // BullMQ jobs run in a separate worker process, so the state must be + // in Redis, where both processes can read it. Without an explicit state + // adapter, use the local queue Redis. A `REDIS_URL` from + // `pnpm dev:env` can be shared with preview deployments. + if (!hasValue(env.JUNIOR_STATE_ADAPTER)) { + env.JUNIOR_STATE_ADAPTER = "redis"; + env.REDIS_URL = env.JUNIOR_QUEUE_REDIS_URL.trim(); + } } } + if (!hasValue(env.JUNIOR_STATE_ADAPTER)) { + env.JUNIOR_STATE_ADAPTER = "memory"; + } } /** Load env files so app-local defaults override repo defaults without replacing shell env. */ diff --git a/scripts/lib/load-env-files.test.mjs b/scripts/lib/load-env-files.test.mjs index d0d088cf2c..6edf2c3331 100644 --- a/scripts/lib/load-env-files.test.mjs +++ b/scripts/lib/load-env-files.test.mjs @@ -126,6 +126,27 @@ describe("applyJuniorDevelopmentDefaults", () => { }); assert.equal(env.JUNIOR_QUEUE_PROVIDER, "bullmq"); assert.equal(env.JUNIOR_QUEUE_REDIS_URL, "redis://localhost:6382"); + assert.equal(env.JUNIOR_STATE_ADAPTER, "redis"); + assert.equal(env.REDIS_URL, "redis://localhost:6382"); + + const sharedStateEnv = { + NODE_ENV: "development", + REDIS_URL: "redis://preview.example:6379", + }; + applyJuniorDevelopmentDefaults(sharedStateEnv, { + queueRedisUrl: "redis://localhost:6382", + }); + assert.equal(sharedStateEnv.REDIS_URL, "redis://localhost:6382"); + + const explicitStateEnv = { + NODE_ENV: "development", + JUNIOR_STATE_ADAPTER: "redis", + REDIS_URL: "redis://localhost:6390", + }; + applyJuniorDevelopmentDefaults(explicitStateEnv, { + queueRedisUrl: "redis://localhost:6382", + }); + assert.equal(explicitStateEnv.REDIS_URL, "redis://localhost:6390"); const vercelEnv = { NODE_ENV: "development", @@ -136,6 +157,7 @@ describe("applyJuniorDevelopmentDefaults", () => { }); assert.equal(vercelEnv.JUNIOR_QUEUE_PROVIDER, "vercel"); assert.equal(vercelEnv.JUNIOR_QUEUE_REDIS_URL, undefined); + assert.equal(vercelEnv.JUNIOR_STATE_ADAPTER, "memory"); const customEnv = { NODE_ENV: "development", From d57ecece50a26506ea2ed4a758a3e4c9d16ccdb1 Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Fri, 30 Oct 2026 03:33:57 +0000 Subject: [PATCH 3/5] fix(queue): Bound worker shutdown and drop BullMQ module state The dev server now kills the queue worker process if it has not exited 10 seconds after SIGTERM. BullMQ reuses its first close promise, so the in-process forced close could not bound shutdown. The BullMQ provider keeps no module state. Each send opens and closes its own Redis client, and each worker owns its client and BullMQ workers. createApp() returns a JuniorApp with startQueueWorker(), which replaces the app-to-consumers WeakMap. Worker cleanup now tries every step even when an earlier one fails. Co-Authored-By: Thomas Jiang <20501235+thomasjiangcy@users.noreply.github.com> --- packages/junior/src/app.ts | 65 +++++----- packages/junior/src/chat/queue/README.md | 18 +-- packages/junior/src/chat/queue/bullmq.ts | 145 ++++++++++------------- scripts/dev-queue-worker.mjs | 12 +- scripts/dev-server.mjs | 33 +++++- 5 files changed, 145 insertions(+), 128 deletions(-) diff --git a/packages/junior/src/app.ts b/packages/junior/src/app.ts index 9788d596b7..38c1e5a43e 100644 --- a/packages/junior/src/app.ts +++ b/packages/junior/src/app.ts @@ -420,40 +420,49 @@ function mountRoutes(app: Hono, routes: HostRouteRegistration[]): void { } } -// The queue consumers of each app from `createApp()`, for `startQueueWorker()`. -const appQueueConsumers = new WeakMap QueueConsumer[]>(); - -/** - * Pull this app's queue messages in this process until `close()`. - * - * Only for a queue provider whose consumers pull messages, such as the - * local BullMQ queue. The dev server runs it in its own worker process, so - * the Nitro process only sends messages. `close()` also closes the database - * and state connections of this process. - */ -export async function startQueueWorker(app: Hono): Promise { - const consumers = appQueueConsumers.get(app); - if (!consumers) { - throw new Error("startQueueWorker() needs an app from createApp()"); - } +/** A Hono app from `createApp()` that can also run its queue consumers. */ +export type JuniorApp = Hono & { + /** + * Pull this app's queue messages in this process until `close()`. + * + * Only for a queue provider whose consumers pull messages, such as the + * local BullMQ queue. The dev server runs it in its own worker process, so + * the Nitro process only sends messages. `close()` also closes the database + * and state connections of this process. + */ + startQueueWorker(): Promise; +}; + +async function startQueueWorker( + consumers: QueueConsumer[], +): Promise { const provider = getQueueProvider(); if (!provider.startWorker) { throw new Error( `The ${provider.messagingSystem} queue provider has no queue worker`, ); } - const worker = await provider.startWorker(consumers()); + const worker = await provider.startWorker(consumers); return { + // Try every step, so one failure does not leave the other resources open. close: async () => { - await worker.close(); - await disconnectStateAdapter(); - await closeDb(); + try { + await worker.close(); + } finally { + try { + await disconnectStateAdapter(); + } finally { + await closeDb(); + } + } }, }; } /** Create a Hono app with all Junior routes. */ -export async function createApp(options?: JuniorAppOptions): Promise { +export async function createApp( + options?: JuniorAppOptions, +): Promise { const virtualConfig = await resolveVirtualConfig(); const dashboard = options?.dashboard ?? virtualConfig?.dashboard; const configuredPlugins = options?.plugins ?? virtualConfig?.pluginSet; @@ -743,11 +752,6 @@ export async function createApp(options?: JuniorAppOptions): Promise { let pluginTaskPOST: | ReturnType | undefined; - appQueueConsumers.set(app, () => [ - conversationWorkQueueConsumer(getConversationWorkOptions()), - pluginTaskQueueConsumer(), - workspaceSnapshotJobQueueConsumer(), - ]); if (process.env.NODE_ENV === "development") { registerVercelConversationWorkDevConsumer(getConversationWorkOptions()); registerVercelPluginTaskDevConsumer(); @@ -789,5 +793,12 @@ export async function createApp(options?: JuniorAppOptions): Promise { }); }); - return app; + return Object.assign(app, { + startQueueWorker: () => + startQueueWorker([ + conversationWorkQueueConsumer(getConversationWorkOptions()), + pluginTaskQueueConsumer(), + workspaceSnapshotJobQueueConsumer(), + ]), + }); } diff --git a/packages/junior/src/chat/queue/README.md b/packages/junior/src/chat/queue/README.md index 1e0652ac37..333c2c1623 100644 --- a/packages/junior/src/chat/queue/README.md +++ b/packages/junior/src/chat/queue/README.md @@ -21,15 +21,17 @@ Conversation work is different. It only wakes a conversation. Keep using enforce these boundaries. - `bullmq.ts` implements `QueueProvider` with BullMQ on Redis, for local development only. `pnpm dev` selects it. The Nitro process only sends jobs. + Each `send()` opens a Redis client and closes it after the job is added, so + the module keeps no connection between calls. `scripts/dev-queue-worker.mjs` runs in its own process, creates the app, and - calls `startQueueWorker()` from `app.ts`. That calls the provider's - `startWorker()`, which pulls jobs from Redis. Nothing pushes to the - callback routes. -- `startWorker()` returns a worker whose `close()` waits up to 10 seconds for - running jobs, then closes each BullMQ worker and queue and the Redis client. - `startQueueWorker()` then closes the state and database connections. Run - one worker in each process. `close()` releases the connections of the whole - process. + calls `app.startQueueWorker()`. That calls the provider's `startWorker()`, + which pulls jobs from Redis. Nothing pushes to the callback routes. +- The worker from `startWorker()` owns its Redis client and BullMQ workers. + Its `close()` waits for running jobs, then closes the client, even when a + BullMQ worker fails to close. `app.startQueueWorker()` then closes the state + and database connections of the process. `close()` has no time limit. The + dev server sends `SIGTERM` to the worker process and kills it if it has not + exited after 10 seconds. - `registerDevConsumer` is for a provider that delivers each message in the process that sent it, like the Vercel SDK in dev. BullMQ does not implement it. diff --git a/packages/junior/src/chat/queue/bullmq.ts b/packages/junior/src/chat/queue/bullmq.ts index a1678e02af..b6e183c44c 100644 --- a/packages/junior/src/chat/queue/bullmq.ts +++ b/packages/junior/src/chat/queue/bullmq.ts @@ -1,11 +1,13 @@ import { createHash } from "node:crypto"; -import type { Job, Queue, Worker } from "bullmq"; +import type { Job } from "bullmq"; import { getChatConfig } from "@/chat/config"; -import { logException, logInfo, logWarn } from "@/chat/logging"; +import { logException, logInfo } from "@/chat/logging"; import type { QueueConsumer, QueueMessageMetadata, QueueProvider, + QueueSendOptions, + QueueSendResult, QueueWorker, } from "@/chat/queue/provider"; @@ -13,9 +15,10 @@ import type { // on Redis, so `pnpm dev` needs no Vercel Queues access. Deployed apps use // the Vercel provider. // -// The Nitro process only sends jobs. The dev server starts one separate -// worker process that calls `startWorker()` and pulls the jobs. `close()` -// on that worker releases every BullMQ and Redis resource of its process. +// This module keeps no state between calls. Each `send()` opens a Redis +// client, adds the job, and closes the client. The dev server starts one +// separate worker process that calls `startWorker()`. That worker owns its +// Redis client and BullMQ workers, and `close()` releases them. // // `bullmq` and `redis` are optional peer dependencies. Only dev installs // need them. @@ -33,19 +36,11 @@ type RedisConnection = ReturnType; // from looping for ever in dev. const MAX_ATTEMPTS = 50; const MAX_BACKOFF_MS = 30_000; -// Running jobs get this long to finish when the worker stops. After that, -// BullMQ stops waiting. Their locks expire, and the jobs run again later. -const SHUTDOWN_GRACE_MS = 10_000; // A job id drops a second send only while BullMQ still stores the first job. // So this is the dedupe window for idempotency keys. const COMPLETED_JOB_AGE_SECONDS = 60 * 60; const FAILED_JOB_AGE_SECONDS = 24 * 60 * 60; -let loaded: - | Promise<{ bullmq: BullMq; connection: RedisConnection }> - | undefined; -const queues = new Map>(); - /** * Import a module by a name held in a variable. * @@ -65,35 +60,23 @@ function bullmqConfig() { return config; } -/** Load BullMQ and one shared Redis client the first time they are needed. */ -function load() { - loaded ??= (async () => { - const [bullmq, redis] = (await Promise.all([ - importAtRuntime("bullmq"), - importAtRuntime("redis"), - ]).catch((error: unknown) => { - throw new Error( - "JUNIOR_QUEUE_PROVIDER=bullmq needs the bullmq and redis packages installed", - { cause: error }, - ); - })) as [BullMq, typeof import("redis")]; - const client = redis.createClient({ url: bullmqConfig().redisUrl }); - await client.connect(); - return { bullmq, connection: bullmq.createNodeRedisClient(client) }; - })(); - return loaded; -} - -async function getQueue(topic: string): Promise> { - const { bullmq, connection } = await load(); - let queue = queues.get(topic); - if (!queue) { - queue = new bullmq.Queue(topic, { - connection, - }); - queues.set(topic, queue); - } - return queue; +/** Load BullMQ and open a Redis client that the caller must close. */ +async function connect(): Promise<{ + bullmq: BullMq; + connection: RedisConnection; +}> { + const [bullmq, redis] = (await Promise.all([ + importAtRuntime("bullmq"), + importAtRuntime("redis"), + ]).catch((error: unknown) => { + throw new Error( + "JUNIOR_QUEUE_PROVIDER=bullmq needs the bullmq and redis packages installed", + { cause: error }, + ); + })) as [BullMq, typeof import("redis")]; + const client = redis.createClient({ url: bullmqConfig().redisUrl }); + await client.connect(); + return { bullmq, connection: bullmq.createNodeRedisClient(client) }; } /** BullMQ job ids must not contain ":", and Junior's keys do. */ @@ -136,39 +119,9 @@ async function processJob( } } -/** Wait for running jobs, but stop waiting after the grace period. */ -async function closeWorker(worker: Worker): Promise { - let timer: NodeJS.Timeout | undefined; - const graceful = worker.close().then(() => true); - const timedOut = new Promise((resolve) => { - timer = setTimeout(() => resolve(false), SHUTDOWN_GRACE_MS); - }); - const finished = await Promise.race([graceful, timedOut]); - clearTimeout(timer); - if (!finished) { - logWarn("queue.bullmq.worker.shutdown_timeout", { - "messaging.destination.name": worker.name, - }); - await worker.close(true); - } -} - -/** Release the queues and the Redis client of this process. */ -async function closeConnections(): Promise { - const current = loaded; - loaded = undefined; - const open = [...queues.values()]; - queues.clear(); - await Promise.all(open.map((queue) => queue.close())); - if (current) { - const { connection } = await current; - await connection.quit(); - } -} - -/** Start one BullMQ worker for each consumer in this process. */ +/** Start one BullMQ worker for each consumer, on a Redis client it owns. */ async function startWorker(consumers: QueueConsumer[]): Promise { - const { bullmq, connection } = await load(); + const { bullmq, connection } = await connect(); const concurrency = bullmqConfig().workerConcurrency; const workers = consumers.map((consumer) => { const worker = new bullmq.Worker( @@ -200,18 +153,34 @@ async function startWorker(consumers: QueueConsumer[]): Promise { "messaging.destination.names": consumers.map((consumer) => consumer.topic), }); return { + // `close()` waits for running jobs without a time limit. The dev server + // stops the worker process if it does not exit within its grace period. close: async () => { - await Promise.all(workers.map((worker) => closeWorker(worker))); - await closeConnections(); + try { + const failures = ( + await Promise.allSettled(workers.map((worker) => worker.close())) + ).flatMap((result) => + result.status === "rejected" ? [result.reason] : [], + ); + if (failures.length > 0) { + throw new AggregateError(failures, "BullMQ workers did not close"); + } + } finally { + await connection.quit(); + } }, }; } -/** Queue provider backed by BullMQ, for local development only. */ -export const bullmqQueueProvider: QueueProvider = { - messagingSystem: "bullmq", - send: async (topic, payload, options = {}) => { - const queue = await getQueue(topic); +/** Add one job on a Redis client that is open only for this send. */ +async function send( + topic: string, + payload: unknown, + options: QueueSendOptions = {}, +): Promise { + const { bullmq, connection } = await connect(); + const queue = new bullmq.Queue(topic, { connection }); + try { const job = await queue.add( topic, { payload, retentionSeconds: options.retentionSeconds }, @@ -227,7 +196,19 @@ export const bullmqQueueProvider: QueueProvider = { }, ); return job.id ? { messageId: job.id } : {}; - }, + } finally { + try { + await queue.close(); + } finally { + await connection.quit(); + } + } +} + +/** Queue provider backed by BullMQ, for local development only. */ +export const bullmqQueueProvider: QueueProvider = { + messagingSystem: "bullmq", + send, // Workers pull jobs from Redis, so nothing pushes to these routes. createCallback: () => async () => new Response("This queue provider does not push messages", { diff --git a/scripts/dev-queue-worker.mjs b/scripts/dev-queue-worker.mjs index c7d8df0b5a..27fbcfa05a 100644 --- a/scripts/dev-queue-worker.mjs +++ b/scripts/dev-queue-worker.mjs @@ -2,8 +2,8 @@ // `pnpm dev` uses the BullMQ queue. Nitro only sends jobs. This process // creates the example app, pulls its jobs, and owns every connection it // opens. The dev server passes the environment and restarts this process -// when the Junior build or the example app's runtime files change. -import { createRequire } from "node:module"; +// when the Junior build or the example app's runtime files change. If this +// process does not exit within 10 seconds of `SIGTERM`, the dev server kills it. import path from "node:path"; import { fileURLToPath, pathToFileURL } from "node:url"; @@ -14,16 +14,10 @@ const exampleDir = path.resolve( "example", ); -// Import `@sentry/junior` from the path that the example app resolves. -// Another copy would hold other app state and find no app consumers. -const juniorEntry = createRequire(path.join(exampleDir, "server.ts")).resolve( - "@sentry/junior", -); -const { startQueueWorker } = await import(pathToFileURL(juniorEntry).href); const { default: app } = await import( pathToFileURL(path.join(exampleDir, "server.ts")).href ); -const worker = await startQueueWorker(app); +const worker = await app.startQueueWorker(); let stopping = false; async function stop(signal) { diff --git a/scripts/dev-server.mjs b/scripts/dev-server.mjs index 603ce5704a..1bdcab8de4 100644 --- a/scripts/dev-server.mjs +++ b/scripts/dev-server.mjs @@ -67,7 +67,10 @@ function stopChild(child, signal) { return; } stoppedChildren.add(child); + signalChild(child, signal); +} +function signalChild(child, signal) { // Long-lived commands launch wrappers. Stop their process group so a // restarted wrapper cannot leave the server that owns its port behind. if (useProcessGroups && child.pid) { @@ -86,7 +89,11 @@ function stopChild(child, signal) { function terminateChildren(signal = "SIGTERM") { for (const child of children) { - stopChild(child, signal); + if (child === queueWorkerChild) { + stopQueueWorker(child); + } else { + stopChild(child, signal); + } } } @@ -252,6 +259,11 @@ function startNitroDev({ openLocalUi = false } = {}) { return; } + // Wait for the queue worker, which can take its stop timeout to exit. + if (queueWorkerChild) { + queueWorkerChild.once("exit", () => process.exit(code ?? 1)); + return; + } process.exit(code ?? 1); }); } @@ -277,6 +289,9 @@ function restartNitroDev() { let queueWorkerChild; let restartingQueueWorker = false; +// The worker waits for running jobs when it stops. After this time, the dev +// server kills it, so that a long job cannot block a restart or shutdown. +const queueWorkerStopTimeoutMs = 10_000; function startQueueWorker() { const child = spawnChild( @@ -316,7 +331,21 @@ function restartQueueWorker() { restartingQueueWorker = false; startQueueWorker(); }); - stopChild(queueWorkerChild, "SIGTERM"); + stopQueueWorker(queueWorkerChild); +} + +function stopQueueWorker(child) { + if (stoppedChildren.has(child)) { + return; + } + const timer = setTimeout(() => { + console.error( + `Queue worker did not stop within ${queueWorkerStopTimeoutMs / 1000} seconds. Killing it.`, + ); + signalChild(child, "SIGKILL"); + }, queueWorkerStopTimeoutMs); + child.once("exit", () => clearTimeout(timer)); + stopChild(child, "SIGTERM"); } function watchForRestart(paths, restart, options = {}) { From f5a7b3fb2a1e3ec13c87902d93237009758cbcfc Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Fri, 30 Oct 2026 04:00:25 +0000 Subject: [PATCH 4/5] fix(queue): Close the Redis client when BullMQ connect fails node-redis leaves a client open and reconnecting after a failed connect(). Destroy it before the error propagates, so a send or worker start against a Redis that is down does not leak a client. --- packages/junior/src/chat/queue/bullmq.ts | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/packages/junior/src/chat/queue/bullmq.ts b/packages/junior/src/chat/queue/bullmq.ts index b6e183c44c..b3feed3d72 100644 --- a/packages/junior/src/chat/queue/bullmq.ts +++ b/packages/junior/src/chat/queue/bullmq.ts @@ -75,8 +75,14 @@ async function connect(): Promise<{ ); })) as [BullMq, typeof import("redis")]; const client = redis.createClient({ url: bullmqConfig().redisUrl }); - await client.connect(); - return { bullmq, connection: bullmq.createNodeRedisClient(client) }; + try { + await client.connect(); + return { bullmq, connection: bullmq.createNodeRedisClient(client) }; + } catch (error) { + // A failed `connect()` leaves the client open and reconnecting. + client.destroy(); + throw error; + } } /** BullMQ job ids must not contain ":", and Junior's keys do. */ From 80d1bd810d8351255aaa23e1e424636073d32324 Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Fri, 30 Oct 2026 04:13:15 +0000 Subject: [PATCH 5/5] fix(queue): Stop dev restarts after shutdown and drain worker background work A worker or Nitro restart that waited for the old process could start a new process after shutdown began. Shutdown now blocks new starts. The queue worker now waits for waitUntil work, such as title work, before it closes state and the database. --- packages/junior/src/app.ts | 30 ++++++++++++++++++++++++------ scripts/dev-server.mjs | 12 +++++++++++- 2 files changed, 35 insertions(+), 7 deletions(-) diff --git a/packages/junior/src/app.ts b/packages/junior/src/app.ts index 38c1e5a43e..7f4ed70010 100644 --- a/packages/junior/src/app.ts +++ b/packages/junior/src/app.ts @@ -435,6 +435,7 @@ export type JuniorApp = Hono & { async function startQueueWorker( consumers: QueueConsumer[], + backgroundWork: Set>, ): Promise { const provider = getQueueProvider(); if (!provider.startWorker) { @@ -449,6 +450,11 @@ async function startQueueWorker( try { await worker.close(); } finally { + // A job can end before the work it started with `waitUntil`, such + // as title work. Let that work finish before its connections close. + while (backgroundWork.size > 0) { + await Promise.allSettled(backgroundWork); + } try { await disconnectStateAdapter(); } finally { @@ -590,6 +596,9 @@ export async function createApp( sandbox: { tracePropagation }, visionContext: { attachmentStorage }, }; + // Background work that Conversation work starts. A queue worker waits for + // it before it closes its connections. + const backgroundWork = new Set>(); let conversationWorkOptions: ConversationWorkCallbackOptions | undefined; const getConversationWorkOptions = (): ConversationWorkCallbackOptions => { conversationWorkOptions ??= @@ -601,7 +610,13 @@ export async function createApp( ? { sendPluginTask: (message) => pluginTaskQueue.send(message) } : undefined), services: runtimeServiceOverrides, - waitUntil: (task) => waitUntil(task), + waitUntil: (task) => { + backgroundWork.add(task); + void task + .finally(() => backgroundWork.delete(task)) + .catch(() => undefined); + waitUntil(task); + }, }); return conversationWorkOptions; }; @@ -795,10 +810,13 @@ export async function createApp( return Object.assign(app, { startQueueWorker: () => - startQueueWorker([ - conversationWorkQueueConsumer(getConversationWorkOptions()), - pluginTaskQueueConsumer(), - workspaceSnapshotJobQueueConsumer(), - ]), + startQueueWorker( + [ + conversationWorkQueueConsumer(getConversationWorkOptions()), + pluginTaskQueueConsumer(), + workspaceSnapshotJobQueueConsumer(), + ], + backgroundWork, + ), }); } diff --git a/scripts/dev-server.mjs b/scripts/dev-server.mjs index 1bdcab8de4..bbf5d5b34f 100644 --- a/scripts/dev-server.mjs +++ b/scripts/dev-server.mjs @@ -45,6 +45,7 @@ const openLocalUi = process.env.OPEN_SPOTLIGHT_WEB?.trim() === "1"; const children = new Set(); const stoppedChildren = new WeakSet(); +let shuttingDown = false; function spawnChild(command, args, options = {}) { const child = spawn(command, args, { @@ -88,6 +89,9 @@ function signalChild(child, signal) { } function terminateChildren(signal = "SIGTERM") { + // A restart that is waiting for an old process to exit must not start a + // new one after shutdown begins. + shuttingDown = true; for (const child of children) { if (child === queueWorkerChild) { stopQueueWorker(child); @@ -227,6 +231,9 @@ function clearExampleVercelOutput() { // Restarts reuse this function. Open the Sentry Local UI only on the first // start, so that each rebuild does not open a new browser tab. function startNitroDev({ openLocalUi = false } = {}) { + if (shuttingDown) { + return; + } const child = spawnChild("pnpm", [ "exec", "sentry", @@ -248,7 +255,7 @@ function startNitroDev({ openLocalUi = false } = {}) { child.on("exit", (code, signal) => { stopChild(child, signal ?? "SIGTERM"); - if (restartingNitro) { + if (restartingNitro && !shuttingDown) { return; } @@ -294,6 +301,9 @@ let restartingQueueWorker = false; const queueWorkerStopTimeoutMs = 10_000; function startQueueWorker() { + if (shuttingDown) { + return; + } const child = spawnChild( process.execPath, [path.join(workspaceRoot, "scripts", "dev-queue-worker.mjs")],