From 2cfcf5ae54898e993a69e53aa677a4b059e8f00d Mon Sep 17 00:00:00 2001 From: Reece Yang <7040416+reeceyang@users.noreply.github.com> Date: Wed, 9 Sep 2026 14:49:19 -0700 Subject: [PATCH 1/5] Only ping the batch worker on the first enqueue in a transaction A mutation that writes to the aggregate several times calls `enqueueOperations` once per write, and each call pings. Every ping after the first is redundant: they share a transaction, so they read the same worker record and any wake the first one did is already visible to them. The lookup that packs operations into the transaction's existing row already tells us. Rows written by this transaction still carry an unresolved commitTs, so finding one means an earlier call in this transaction has already pinged. A transaction with ten aggregate writes goes from ten pings to one. Co-Authored-By: Claude Opus 5 (1M context) --- src/component/worker.test.ts | 29 +++++++++++++++++++++++++++++ src/component/worker.ts | 6 ++++++ 2 files changed, 35 insertions(+) diff --git a/src/component/worker.test.ts b/src/component/worker.test.ts index e2d6e7de..c1259200 100644 --- a/src/component/worker.test.ts +++ b/src/component/worker.test.ts @@ -893,3 +893,32 @@ describe("guards", () => { ).rejects.toThrow(/no progress/); }); }); + +describe("pinging the worker", () => { + // Only the first enqueue in a transaction pings, so the ping still has to + // land when a transaction enqueues more than once. + test("a transaction that enqueues repeatedly still wakes the worker", async () => { + const t = initConvexTest(); + await enqueue(t, { type: "insert", key: 1, value: "a" }); + await t.finishAllScheduledFunctions(vi.runAllTimers); + expect( + await t.run(async (ctx) => + ctx.runQuery(components.batchWorker.lib.status, { + name: OPS_WORKER_NAME, + }), + ), + ).toEqual({ kind: "idle" }); + + await enqueue( + t, + { type: "insert", key: 2, value: "b" }, + { type: "insert", key: 3, value: "c" }, + ); + await t.finishAllScheduledFunctions(vi.runAllTimers); + await t.run(async (ctx) => { + expect(await getHandler(ctx, { key: 2 })).toEqual({ k: 2, v: "b", s: 0 }); + expect(await getHandler(ctx, { key: 3 })).toEqual({ k: 3, v: "c", s: 0 }); + expect(await ctx.db.query("pendingOperations").collect()).toEqual([]); + }); + }); +}); diff --git a/src/component/worker.ts b/src/component/worker.ts index 291fa171..3acb0c87 100644 --- a/src/component/worker.ts +++ b/src/component/worker.ts @@ -83,6 +83,12 @@ export async function enqueueOperations( } await ctx.db.insert("pendingOperations", entry); } + // Rows written by this transaction still carry an unresolved commitTs, so + // one turning up in the lookup above means an earlier `enqueueOperations` + // already pinged for it. + if (newestEntry) { + return; + } await ping(ctx, components.batchWorker, { // TODO: explore separate queues by namespace name: OPS_WORKER_NAME, From f4f0af4d151741f5c1fb6f86a0148cbe377823ab Mon Sep 17 00:00:00 2001 From: Reece Yang <7040416+reeceyang@users.noreply.github.com> Date: Wed, 9 Sep 2026 17:37:15 -0700 Subject: [PATCH 2/5] tweak comment --- src/component/worker.ts | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/src/component/worker.ts b/src/component/worker.ts index 3acb0c87..604c4b5e 100644 --- a/src/component/worker.ts +++ b/src/component/worker.ts @@ -83,9 +83,7 @@ export async function enqueueOperations( } await ctx.db.insert("pendingOperations", entry); } - // Rows written by this transaction still carry an unresolved commitTs, so - // one turning up in the lookup above means an earlier `enqueueOperations` - // already pinged for it. + // Only ping when `enqueueOperations` is called for the first time if (newestEntry) { return; } From dca4d7ba290e066deb8c27cb9c0653474e34a86e Mon Sep 17 00:00:00 2001 From: Reece Yang <7040416+reeceyang@users.noreply.github.com> Date: Wed, 9 Sep 2026 17:44:15 -0700 Subject: [PATCH 3/5] Test the ping count rather than the drain MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The previous test asserted that a transaction enqueuing more than once still drains, which existing tests already cover, and which holds whether or not the guard is there — removing the guard entirely leaves the whole suite green, since a redundant ping in the same transaction reads the state the first one wrote and no-ops. Count the calls into the component instead, which is the thing that changed. Verified it fails with the guard removed. Co-Authored-By: Claude Opus 5 (1M context) --- src/component/worker.test.ts | 62 +++++++++++++++++++++++++----------- 1 file changed, 43 insertions(+), 19 deletions(-) diff --git a/src/component/worker.test.ts b/src/component/worker.test.ts index c1259200..e4fc9635 100644 --- a/src/component/worker.test.ts +++ b/src/component/worker.test.ts @@ -15,6 +15,7 @@ import { getDocumentSize, } from "convex/values"; import schema, { type Operation } from "./schema.js"; +import type { MutationCtx } from "./_generated/server.js"; import { initConvexTest } from "./setup.test.js"; import { aggregateBetweenHandler, @@ -895,30 +896,53 @@ describe("guards", () => { }); describe("pinging the worker", () => { - // Only the first enqueue in a transaction pings, so the ping still has to - // land when a transaction enqueues more than once. - test("a transaction that enqueues repeatedly still wakes the worker", async () => { + // `ping` is the only thing `enqueueOperations` calls into the component for, + // so counting those counts pings. + function countingPings(ctx: MutationCtx) { + const calls = { count: 0 }; + const spy = new Proxy(ctx, { + get(target, prop) { + if (prop === "runMutation") { + return (...args: unknown[]) => { + calls.count++; + return (target.runMutation as (...a: unknown[]) => unknown)( + ...args, + ); + }; + } + const value = Reflect.get(target, prop); + return typeof value === "function" ? value.bind(target) : value; + }, + }); + return { spy, calls }; + } + + test("only the first enqueue in a transaction pings", async () => { const t = initConvexTest(); - await enqueue(t, { type: "insert", key: 1, value: "a" }); - await t.finishAllScheduledFunctions(vi.runAllTimers); - expect( - await t.run(async (ctx) => - ctx.runQuery(components.batchWorker.lib.status, { - name: OPS_WORKER_NAME, - }), - ), - ).toEqual({ kind: "idle" }); + await t.run(async (ctx) => { + const { spy, calls } = countingPings(ctx); + await enqueueOperations(spy, [{ type: "insert", key: 1, value: "a" }]); + expect(calls.count).toBe(1); + await enqueueOperations(spy, [{ type: "insert", key: 2, value: "b" }]); + await enqueueOperations(spy, [{ type: "insert", key: 3, value: "c" }]); + expect(calls.count).toBe(1); + }); - await enqueue( - t, - { type: "insert", key: 2, value: "b" }, - { type: "insert", key: 3, value: "c" }, - ); - await t.finishAllScheduledFunctions(vi.runAllTimers); + // The one ping still gets everything the transaction enqueued drained. + await drainViaWorker(t); await t.run(async (ctx) => { - expect(await getHandler(ctx, { key: 2 })).toEqual({ k: 2, v: "b", s: 0 }); expect(await getHandler(ctx, { key: 3 })).toEqual({ k: 3, v: "c", s: 0 }); expect(await ctx.db.query("pendingOperations").collect()).toEqual([]); }); }); + + test("a later transaction pings again", async () => { + const t = initConvexTest(); + await enqueue(t, { type: "insert", key: 1, value: "a" }); + await t.run(async (ctx) => { + const { spy, calls } = countingPings(ctx); + await enqueueOperations(spy, [{ type: "insert", key: 2, value: "b" }]); + expect(calls.count).toBe(1); + }); + }); }); From 64fd0bb0d7aae8d83f32156bf88afe455cde7e3f Mon Sep 17 00:00:00 2001 From: Reece Yang <7040416+reeceyang@users.noreply.github.com> Date: Wed, 9 Sep 2026 17:49:49 -0700 Subject: [PATCH 4/5] Spy on ping directly rather than counting component calls Co-Authored-By: Claude Opus 5 (1M context) --- src/component/worker.test.ts | 49 +++++++++++++----------------------- 1 file changed, 17 insertions(+), 32 deletions(-) diff --git a/src/component/worker.test.ts b/src/component/worker.test.ts index e4fc9635..7baa6356 100644 --- a/src/component/worker.test.ts +++ b/src/component/worker.test.ts @@ -15,8 +15,14 @@ import { getDocumentSize, } from "convex/values"; import schema, { type Operation } from "./schema.js"; -import type { MutationCtx } from "./_generated/server.js"; +import { ping } from "@convex-dev/batch-worker"; import { initConvexTest } from "./setup.test.js"; + +vi.mock("@convex-dev/batch-worker", async (importOriginal) => { + const actual = + await importOriginal(); + return { ...actual, ping: vi.fn(actual.ping) }; +}); import { aggregateBetweenHandler, getHandler, @@ -896,36 +902,18 @@ describe("guards", () => { }); describe("pinging the worker", () => { - // `ping` is the only thing `enqueueOperations` calls into the component for, - // so counting those counts pings. - function countingPings(ctx: MutationCtx) { - const calls = { count: 0 }; - const spy = new Proxy(ctx, { - get(target, prop) { - if (prop === "runMutation") { - return (...args: unknown[]) => { - calls.count++; - return (target.runMutation as (...a: unknown[]) => unknown)( - ...args, - ); - }; - } - const value = Reflect.get(target, prop); - return typeof value === "function" ? value.bind(target) : value; - }, - }); - return { spy, calls }; - } + beforeEach(() => { + vi.mocked(ping).mockClear(); + }); test("only the first enqueue in a transaction pings", async () => { const t = initConvexTest(); await t.run(async (ctx) => { - const { spy, calls } = countingPings(ctx); - await enqueueOperations(spy, [{ type: "insert", key: 1, value: "a" }]); - expect(calls.count).toBe(1); - await enqueueOperations(spy, [{ type: "insert", key: 2, value: "b" }]); - await enqueueOperations(spy, [{ type: "insert", key: 3, value: "c" }]); - expect(calls.count).toBe(1); + await enqueueOperations(ctx, [{ type: "insert", key: 1, value: "a" }]); + expect(ping).toHaveBeenCalledTimes(1); + await enqueueOperations(ctx, [{ type: "insert", key: 2, value: "b" }]); + await enqueueOperations(ctx, [{ type: "insert", key: 3, value: "c" }]); + expect(ping).toHaveBeenCalledTimes(1); }); // The one ping still gets everything the transaction enqueued drained. @@ -939,10 +927,7 @@ describe("pinging the worker", () => { test("a later transaction pings again", async () => { const t = initConvexTest(); await enqueue(t, { type: "insert", key: 1, value: "a" }); - await t.run(async (ctx) => { - const { spy, calls } = countingPings(ctx); - await enqueueOperations(spy, [{ type: "insert", key: 2, value: "b" }]); - expect(calls.count).toBe(1); - }); + await enqueue(t, { type: "insert", key: 2, value: "b" }); + expect(ping).toHaveBeenCalledTimes(2); }); }); From fd08d30897efd2362a14d8ed59bc7d98fb45e7e8 Mon Sep 17 00:00:00 2001 From: Reece Yang <7040416+reeceyang@users.noreply.github.com> Date: Wed, 9 Sep 2026 17:51:58 -0700 Subject: [PATCH 5/5] Drop the redundant later-transaction ping test Co-Authored-By: Claude Opus 5 (1M context) --- src/component/worker.test.ts | 7 ------- 1 file changed, 7 deletions(-) diff --git a/src/component/worker.test.ts b/src/component/worker.test.ts index 7baa6356..b233dec7 100644 --- a/src/component/worker.test.ts +++ b/src/component/worker.test.ts @@ -923,11 +923,4 @@ describe("pinging the worker", () => { expect(await ctx.db.query("pendingOperations").collect()).toEqual([]); }); }); - - test("a later transaction pings again", async () => { - const t = initConvexTest(); - await enqueue(t, { type: "insert", key: 1, value: "a" }); - await enqueue(t, { type: "insert", key: 2, value: "b" }); - expect(ping).toHaveBeenCalledTimes(2); - }); });