From d2a4416d931a2cd261559fffe37ead373a4857af Mon Sep 17 00:00:00 2001 From: Reece Yang <7040416+reeceyang@users.noreply.github.com> Date: Tue, 8 Sep 2026 13:50:33 -0700 Subject: [PATCH] Add an env var to schedule the batch worker ping Queuing a write pings the batch worker from inside the queuing mutation, which reads the record holding the worker's run state and, when the worker is parked, writes it too. A burst arriving at a parked worker therefore has every mutation in it writing one record, and all but one losing the race. Under `WORKER_SCHEDULE_PING`, the ping is scheduled instead, so queuing a write touches the worker component not at all: no read to conflict on and nothing written. The scheduled ping runs after the mutation commits and does the same work it always did, so nothing can be stranded by racing the worker as it parks. Over a bursty workload this takes the reads made from queuing mutations from 31 to 0 and the writes from 6 to 0, leaving the same 12 writes in the worker's own transactions. Off by default: it costs one scheduled function per queued write, whether or not the worker needed waking, which is a poor trade for a deployment that can just poll for longer. Co-Authored-By: Claude Opus 5 (1M context) --- CHANGELOG.md | 6 ++ src/component/_generated/server.ts | 1 + src/component/convex.config.ts | 3 + src/component/worker.test.ts | 102 +++++++++++++++++++++++++++++ src/component/worker.ts | 16 +++++ 5 files changed, 128 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index a0b76a1e..115276fe 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,11 @@ # Changelog +## Unreleased + +- Add experimental configuration options for configuring the Aggregate's batch + worker: `WORKER_IDLE_COOLDOWN_MS`, `WORKER_POLL_INTERVAL_MS`, and + `WORKER_SCHEDULE_PING`. + ## 0.3.3 - Bump `convex-helpers` to 0.1.128, which isn’t affected by diff --git a/src/component/_generated/server.ts b/src/component/_generated/server.ts index 481cb3ad..594ee419 100644 --- a/src/component/_generated/server.ts +++ b/src/component/_generated/server.ts @@ -41,6 +41,7 @@ type Env = { readonly CONVEX_SITE_URL: string; readonly WORKER_IDLE_COOLDOWN_MS: string | undefined; readonly WORKER_POLL_INTERVAL_MS: string | undefined; + readonly WORKER_SCHEDULE_PING: "true" | "false" | undefined; }; /** diff --git a/src/component/convex.config.ts b/src/component/convex.config.ts index 90127ea1..54a96ef8 100644 --- a/src/component/convex.config.ts +++ b/src/component/convex.config.ts @@ -6,6 +6,9 @@ const component = defineComponent("aggregate", { env: { WORKER_IDLE_COOLDOWN_MS: v.optional(v.string()), WORKER_POLL_INTERVAL_MS: v.optional(v.string()), + WORKER_SCHEDULE_PING: v.optional( + v.union(v.literal("true"), v.literal("false")), + ), }, }); component.use(batchWorker); diff --git a/src/component/worker.test.ts b/src/component/worker.test.ts index c844139c..380bf200 100644 --- a/src/component/worker.test.ts +++ b/src/component/worker.test.ts @@ -990,6 +990,108 @@ describe("staying awake between bursts", () => { }); }); +describe("scheduling the ping", () => { + afterEach(() => { + vi.unstubAllEnvs(); + }); + + // Pending only: convex-test keeps completed runs in the table too. + async function pendingPings(t: TestConvex) { + return await t.run(async (ctx) => { + const scheduled = await ctx.db.system + .query("_scheduled_functions") + .collect(); + return scheduled.filter( + (f) => f.name.includes("pingWorker") && f.state.kind === "pending", + ); + }); + } + + test("the ping is inline unless asked for", async () => { + const t = initConvexTest(); + await enqueue(t, { type: "insert", key: 1, value: "a" }); + expect(await pendingPings(t)).toHaveLength(0); + expect(await workerStatus(t)).toEqual({ kind: "running" }); + }); + + test("a parked worker is woken from its own transaction", async () => { + vi.stubEnv("WORKER_SCHEDULE_PING", "true"); + const t = initConvexTest(); + await enqueue(t, { type: "insert", key: 1, value: "a" }); + // The enqueue left the worker alone; the scheduled ping registers it. + expect(await workerStatus(t)).toBeNull(); + expect(await pendingPings(t)).toHaveLength(1); + + await drainViaWorker(t); + await t.run(async (ctx) => { + expect(await getHandler(ctx, { key: 1 })).toEqual({ + k: 1, + v: "a", + s: 0, + }); + expect(await ctx.db.query("pendingOperations").collect()).toEqual([]); + }); + }); + + test("each transaction schedules one, and it no-ops on a running worker", async () => { + vi.stubEnv("WORKER_SCHEDULE_PING", "true"); + const t = initConvexTest(); + await enqueue(t, { type: "insert", key: 1, value: "a" }); + vi.advanceTimersByTime(1); + await t.finishInProgressScheduledFunctions(); + expect(await workerStatus(t)).toEqual({ kind: "running" }); + + await enqueue(t, { type: "insert", key: 2, value: "b" }); + expect(await pendingPings(t)).toHaveLength(1); + await drainViaWorker(t); + await t.run(async (ctx) => { + expect(await getHandler(ctx, { key: 2 })).toEqual({ + k: 2, + v: "b", + s: 0, + }); + }); + }); + + test("a transaction that enqueues repeatedly schedules one ping", async () => { + vi.stubEnv("WORKER_SCHEDULE_PING", "true"); + const t = initConvexTest(); + await enqueue( + t, + { type: "insert", key: 1, value: "a" }, + { type: "insert", key: 2, value: "b" }, + { type: "insert", key: 3, value: "c" }, + ); + expect(await pendingPings(t)).toHaveLength(1); + + await drainViaWorker(t); + await t.run(async (ctx) => { + expect(await getHandler(ctx, { key: 3 })).toEqual({ k: 3, v: "c", s: 0 }); + expect(await ctx.db.query("pendingOperations").collect()).toEqual([]); + }); + }); + + test("work enqueued after the worker parks still drains", async () => { + vi.stubEnv("WORKER_SCHEDULE_PING", "true"); + const t = initConvexTest(); + await enqueue(t, { type: "insert", key: 1, value: "a" }); + await drainViaWorker(t); + expect(await workerStatus(t)).toEqual({ kind: "idle" }); + + await enqueue(t, { type: "insert", key: 2, value: "b" }); + expect(await pendingPings(t)).toHaveLength(1); + 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 7f2352d5..4562ff5f 100644 --- a/src/component/worker.ts +++ b/src/component/worker.ts @@ -95,6 +95,14 @@ export async function enqueueOperations( if (newestEntry) { return; } + if (env.WORKER_SCHEDULE_PING === "true") { + await ctx.scheduler.runAfter(0, internal.worker.pingWorker, {}); + return; + } + await pingBatchWorker(ctx); +} + +async function pingBatchWorker(ctx: MutationCtx) { await ping(ctx, components.batchWorker, { // TODO: explore separate queues by namespace name: OPS_WORKER_NAME, @@ -103,6 +111,14 @@ export async function enqueueOperations( }); } +export const pingWorker = internalMutation({ + args: {}, + returns: v.null(), + handler: async (ctx) => { + await pingBatchWorker(ctx); + }, +}); + const vBatchEntry = v.object({ id: v.id("pendingOperations"), commitTs: v.int64(),