expose env vars for setting batch worker's idle behavior

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,