From 54d147e66a03cd1272dd083f42d5f611df741384 Mon Sep 17 00:00:00 2001 From: Reece Yang <7040416+reeceyang@users.noreply.github.com> Date: Tue, 8 Sep 2026 13:46:03 -0700 Subject: [PATCH] Let the batch worker's idle behavior be tuned per aggregate Every queued write pings the batch worker, which reads the one record holding the worker's run state. Nothing writes that record while the worker keeps running, but parking it once the queue drains and waking it back up both do, and each of those writes makes the enqueues in flight retry. Waking is worse still: it's written from inside the enqueuing mutations, so a burst arriving at a parked worker has all of them fighting over it. The worker parks a couple of seconds after the queue drains, so a workload that writes in bursts pays for a park and a wake on every burst. Polling an empty queue for longer rides out the gaps instead: bursts closer together than the cooldown never park the worker, and never contend. The right trade depends on the write pattern, so expose it as two optional env vars on the component rather than picking a default. Left unset, the batch worker's own defaults apply and nothing changes. Co-Authored-By: Claude Opus 5 (1M context) --- src/component/_generated/server.ts | 2 + src/component/convex.config.ts | 8 ++- src/component/worker.test.ts | 89 ++++++++++++++++++++++++++++++ src/component/worker.ts | 14 ++++- tsconfig.json | 1 + 5 files changed, 112 insertions(+), 2 deletions(-) diff --git a/src/component/_generated/server.ts b/src/component/_generated/server.ts index 1a7688a4..481cb3ad 100644 --- a/src/component/_generated/server.ts +++ b/src/component/_generated/server.ts @@ -39,6 +39,8 @@ import type { DataModel } from "./dataModel.js"; type Env = { readonly CONVEX_CLOUD_URL: string; readonly CONVEX_SITE_URL: string; + readonly WORKER_IDLE_COOLDOWN_MS: string | undefined; + readonly WORKER_POLL_INTERVAL_MS: string | undefined; }; /** diff --git a/src/component/convex.config.ts b/src/component/convex.config.ts index 86ec81d3..90127ea1 100644 --- a/src/component/convex.config.ts +++ b/src/component/convex.config.ts @@ -1,7 +1,13 @@ import { defineComponent } from "convex/server"; +import { v } from "convex/values"; import batchWorker from "@convex-dev/batch-worker/convex.config.js"; -const component = defineComponent("aggregate"); +const component = defineComponent("aggregate", { + env: { + WORKER_IDLE_COOLDOWN_MS: v.optional(v.string()), + WORKER_POLL_INTERVAL_MS: v.optional(v.string()), + }, +}); component.use(batchWorker); export default component; diff --git a/src/component/worker.test.ts b/src/component/worker.test.ts index b233dec7..c844139c 100644 --- a/src/component/worker.test.ts +++ b/src/component/worker.test.ts @@ -32,6 +32,7 @@ import { } from "./btree.js"; import { BATCH_MAX_OPERATIONS, + parseDuration, MAX_BYTES_PER_ENTRY, enqueueOperations, hasHeadroomToFinish, @@ -104,6 +105,13 @@ async function drainViaWorker(t: TestConvex): Promise { await t.finishAllScheduledFunctions(vi.runAllTimers); } +// The batch worker's run state for our queue, or null before it is registered. +async function workerStatus(t: TestConvex) { + return await t.run(async (ctx) => + ctx.runQuery(components.batchWorker.lib.status, { name: OPS_WORKER_NAME }), + ); +} + // The cursor the batch worker has committed for our queue, or null before it // has drained anything. async function workerCursor( @@ -901,6 +909,87 @@ describe("guards", () => { }); }); +describe("staying awake between bursts", () => { + // Like `drainViaWorker`, but without running the clock past the worker's + // next poll, so its run state stays observable mid-gap. + async function advance(t: TestConvex, ms: number) { + for (let elapsed = 0; elapsed < ms; elapsed += 200) { + vi.advanceTimersByTime(200); + await t.finishInProgressScheduledFunctions(); + } + } + + afterEach(() => { + vi.unstubAllEnvs(); + }); + + const COOLDOWN_MS = 60_000; + + function configureWorker() { + vi.stubEnv("WORKER_IDLE_COOLDOWN_MS", String(COOLDOWN_MS)); + vi.stubEnv("WORKER_POLL_INTERVAL_MS", "1000"); + } + + test("an unusable env var is left unset", () => { + expect(parseDuration("5000")).toBe(5_000); + expect(parseDuration(undefined)).toBeUndefined(); + expect(parseDuration("")).toBeUndefined(); + expect(parseDuration("soon")).toBeUndefined(); + expect(parseDuration("0")).toBeUndefined(); + expect(parseDuration("-1000")).toBeUndefined(); + }); + + test("an empty queue defers to the batch worker unless configured", async () => { + const t = initConvexTest(); + expect( + await t.query(internal.worker.getBatch, { name: OPS_WORKER_NAME }), + ).toEqual({ kind: "idle" }); + + configureWorker(); + expect( + await t.query(internal.worker.getBatch, { name: OPS_WORKER_NAME }), + ).toEqual({ kind: "idle", cooldownMs: COOLDOWN_MS, pollIntervalMs: 1_000 }); + }); + + test("the worker stays running across a gap shorter than the cooldown", async () => { + configureWorker(); + const t = initConvexTest(); + await enqueue(t, { type: "insert", key: 1, value: "a" }); + await advance(t, COOLDOWN_MS / 2); + expect(await workerStatus(t)).toEqual({ kind: "running" }); + + await enqueue(t, { type: "insert", key: 2, value: "b" }); + await advance(t, 2_000); + await t.run(async (ctx) => { + expect(await getHandler(ctx, { key: 2 })).toEqual({ + k: 2, + v: "b", + s: 0, + }); + }); + }); + + test("it still parks once the queue has been empty for the cooldown", async () => { + configureWorker(); + const t = initConvexTest(); + await enqueue(t, { type: "insert", key: 1, value: "a" }); + await advance(t, COOLDOWN_MS + 2_000); + expect(await workerStatus(t)).toEqual({ kind: "idle" }); + + await enqueue(t, { type: "insert", key: 2, value: "b" }); + expect(await workerStatus(t)).toEqual({ kind: "running" }); + await drainViaWorker(t); + await t.run(async (ctx) => { + expect(await getHandler(ctx, { key: 2 })).toEqual({ + k: 2, + v: "b", + s: 0, + }); + expect(await ctx.db.query("pendingOperations").collect()).toEqual([]); + }); + }); +}); + describe("pinging the worker", () => { beforeEach(() => { vi.mocked(ping).mockClear(); diff --git a/src/component/worker.ts b/src/component/worker.ts index 604c4b5e..7f2352d5 100644 --- a/src/component/worker.ts +++ b/src/component/worker.ts @@ -9,6 +9,7 @@ import { import type { TransactionMetrics, WithoutSystemFields } from "convex/server"; import { type DatabaseWriter, + env, internalMutation, internalQuery, type MutationCtx, @@ -35,6 +36,13 @@ export const MAX_BYTES_PER_ENTRY = 100 * 1024; export const OPS_WORKER_NAME = "ops"; +export function parseDuration(value: string | undefined): number | undefined { + const parsed = Number(value); + return value !== undefined && Number.isFinite(parsed) && parsed > 0 + ? parsed + : undefined; +} + export async function enqueueOperations( ctx: MutationCtx, operations: Operation[], @@ -144,7 +152,11 @@ export const getBatch = internalQuery({ operations += row.operations.length; } if (entries.length === 0) { - return { kind: "idle" as const }; + return { + kind: "idle" as const, + cooldownMs: parseDuration(env.WORKER_IDLE_COOLDOWN_MS), + pollIntervalMs: parseDuration(env.WORKER_POLL_INTERVAL_MS), + }; } return { kind: "work" as const, batch: { entries } }; }, diff --git a/tsconfig.json b/tsconfig.json index d2381558..be83fe91 100644 --- a/tsconfig.json +++ b/tsconfig.json @@ -6,6 +6,7 @@ "target": "ESNext", "lib": ["ES2021", "dom", "DOM.Iterable"], + "types": ["node"], "jsx": "react-jsx", "forceConsistentCasingInFileNames": true, "allowSyntheticDefaultImports": true,