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..8a6b77ca23 100644 --- a/packages/docs/src/content/docs/contribute/development.md +++ b/packages/docs/src/content/docs/contribute/development.md @@ -41,6 +41,10 @@ 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`. + +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 2980e33a24..539d5834e1 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,9 @@ 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`. | +| `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. @@ -68,6 +71,16 @@ 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. 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. + 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/app.ts b/packages/junior/src/app.ts index 426c3b1f20..7f4ed70010 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,8 +420,55 @@ function mountRoutes(app: Hono, routes: HostRouteRegistration[]): void { } } +/** 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[], + backgroundWork: Set>, +): 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); + return { + // Try every step, so one failure does not leave the other resources open. + close: async () => { + 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 { + 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; @@ -539,6 +596,9 @@ export async function createApp(options?: JuniorAppOptions): Promise { 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 ??= @@ -550,7 +610,13 @@ export async function createApp(options?: JuniorAppOptions): Promise { ? { 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; }; @@ -742,5 +808,15 @@ export async function createApp(options?: JuniorAppOptions): Promise { }); }); - return app; + return Object.assign(app, { + startQueueWorker: () => + startQueueWorker( + [ + conversationWorkQueueConsumer(getConversationWorkOptions()), + pluginTaskQueueConsumer(), + workspaceSnapshotJobQueueConsumer(), + ], + backgroundWork, + ), + }); } 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/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 2ace708066..333c2c1623 100644 --- a/packages/junior/src/chat/queue/README.md +++ b/packages/junior/src/chat/queue/README.md @@ -14,11 +14,35 @@ 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. 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 `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. +- `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..b3feed3d72 --- /dev/null +++ b/packages/junior/src/chat/queue/bullmq.ts @@ -0,0 +1,224 @@ +import { createHash } from "node:crypto"; +import type { Job } from "bullmq"; +import { getChatConfig } from "@/chat/config"; +import { logException, logInfo } from "@/chat/logging"; +import type { + QueueConsumer, + QueueMessageMetadata, + QueueProvider, + QueueSendOptions, + QueueSendResult, + QueueWorker, +} from "@/chat/queue/provider"; + +// Module overview: a queue provider for local development, backed by BullMQ +// on Redis, so `pnpm dev` needs no Vercel Queues access. Deployed apps use +// the Vercel provider. +// +// 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. + +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; +// 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; + +/** + * 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 bullmqConfig() { + const config = getChatConfig().queue; + if (config.provider !== "bullmq") { + throw new Error("The BullMQ queue provider is not configured"); + } + return config; +} + +/** 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 }); + 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. */ +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 BullMQ worker for each consumer, on a Redis client it owns. */ +async function startWorker(consumers: QueueConsumer[]): Promise { + const { bullmq, connection } = await connect(); + const concurrency = bullmqConfig().workerConcurrency; + const workers = consumers.map((consumer) => { + const worker = new bullmq.Worker( + consumer.topic, + (job) => processJob(job, consumer), + { + 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 worker; + }); + logInfo("queue.bullmq.worker.started", { + "app.queue.worker_concurrency": concurrency, + "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 () => { + 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(); + } + }, + }; +} + +/** 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 }, + { + 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 } : {}; + } 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", { + status: 404, + }), + 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 new file mode 100644 index 0000000000..bd765ed2ff --- /dev/null +++ b/packages/junior/src/chat/queue/config.ts @@ -0,0 +1,53 @@ +import { toOptionalTrimmed } from "@/chat/optional-string"; + +/** Which queue delivers background work. */ +export type QueueConfig = + | { provider: "vercel" } + | { 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 { + 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, + 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 da01c11782..e32ee3ed47 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. */ @@ -63,11 +65,27 @@ 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. */ export function getQueueProvider(): QueueProvider { - return vercelQueueProvider; + return getChatConfig().queue.provider === "bullmq" + ? bullmqQueueProvider + : vercelQueueProvider; } 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 de32489ddd..be15a93993 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,43 @@ 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", + workerConcurrency: 4, + }); + + process.env.JUNIOR_QUEUE_WORKER_CONCURRENCY = "2"; + expect((await loadConfig()).getChatConfig().queue).toMatchObject({ + workerConcurrency: 2, + }); + }); + + 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("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 new file mode 100644 index 0000000000..8e546b1c8a --- /dev/null +++ b/packages/junior/tests/integration/queue-bullmq.test.ts @@ -0,0 +1,157 @@ +import { randomUUID } from "node:crypto"; +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 +// 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"; +}); + +let worker: QueueWorker | undefined; + +afterEach(async () => { + await worker?.close(); + worker = undefined; +}); + +function uniqueTopic(name: string): string { + return `junior_test_${name}_${randomUUID()}`; +} + +async function startConsumer( + topic: string, + handler: QueueConsumer["handler"], + retry: QueueConsumer["retry"] = () => undefined, +): 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) { + 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; + }> = []; + await 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; + await 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; + await 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); + }); + + 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/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-queue-worker.mjs b/scripts/dev-queue-worker.mjs new file mode 100644 index 0000000000..27fbcfa05a --- /dev/null +++ b/scripts/dev-queue-worker.mjs @@ -0,0 +1,42 @@ +// 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. 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"; + +const exampleDir = path.resolve( + path.dirname(fileURLToPath(import.meta.url)), + "..", + "apps", + "example", +); + +const { default: app } = await import( + pathToFileURL(path.join(exampleDir, "server.ts")).href +); +const worker = await app.startQueueWorker(); + +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 f36caec7fb..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, { @@ -67,7 +68,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) { @@ -85,8 +89,15 @@ function stopChild(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) { - stopChild(child, signal); + if (child === queueWorkerChild) { + stopQueueWorker(child); + } else { + stopChild(child, signal); + } } } @@ -143,7 +154,20 @@ 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", }); +// 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() @@ -207,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", @@ -228,7 +255,7 @@ function startNitroDev({ openLocalUi = false } = {}) { child.on("exit", (code, signal) => { stopChild(child, signal ?? "SIGTERM"); - if (restartingNitro) { + if (restartingNitro && !shuttingDown) { return; } @@ -239,6 +266,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); }); } @@ -262,18 +294,83 @@ function restartNitroDev() { stopChild(nitroChild, "SIGTERM"); } -function watchDistForNitroRestart() { +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() { + if (shuttingDown) { + return; + } + 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(); + }); + 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 = {}) { + 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() { @@ -285,6 +382,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, }); @@ -322,8 +448,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 43848ab565..38b332b316 100644 --- a/scripts/lib/load-env-files.mjs +++ b/scripts/lib/load-env-files.mjs @@ -28,15 +28,35 @@ 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; } 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") { + 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 3c1dfac42c..6edf2c3331 100644 --- a/scripts/lib/load-env-files.test.mjs +++ b/scripts/lib/load-env-files.test.mjs @@ -116,6 +116,58 @@ 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"); + 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", + 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); + assert.equal(vercelEnv.JUNIOR_STATE_ADAPTER, "memory"); + + 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 +196,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);