Let the batch worker's idle behavior be tuned per aggregate (#268)

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) <noreply@anthropic.com>

Author: reeceyang

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;
 };
 
 /**

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;

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<typeof schema>): Promise<void> {
   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<typeof schema>) {
+  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<typeof schema>, 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();

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 } };
   },

tsconfig.json

@@ -6,6 +6,7 @@
 
     "target": "ESNext",
     "lib": ["ES2021", "dom", "DOM.Iterable"],
+    "types": ["node"],
     "jsx": "react-jsx",
     "forceConsistentCasingInFileNames": true,
     "allowSyntheticDefaultImports": true,