Only ping the batch worker on the first enqueue in a transaction

src/component/worker.test.ts

@@ -15,7 +15,14 @@ import {
   getDocumentSize,
 } from "convex/values";
 import schema, { type Operation } from "./schema.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<typeof import("@convex-dev/batch-worker")>();
+  return { ...actual, ping: vi.fn(actual.ping) };
+});
 import {
   aggregateBetweenHandler,
   getHandler,
@@ -893,3 +900,27 @@ describe("guards", () => {
     ).rejects.toThrow(/no progress/);
   });
 });
+
+describe("pinging the worker", () => {
+  beforeEach(() => {
+    vi.mocked(ping).mockClear();
+  });
+
+  test("only the first enqueue in a transaction pings", async () => {
+    const t = initConvexTest();
+    await t.run(async (ctx) => {
+      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.
+    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([]);
+    });
+  });
+});

src/component/worker.ts

@@ -83,6 +83,10 @@ export async function enqueueOperations(
     }
     await ctx.db.insert("pendingOperations", entry);
   }
+  // Only ping when `enqueueOperations` is called for the first time
+  if (newestEntry) {
+    return;
+  }
   await ping(ctx, components.batchWorker, {
     // TODO: explore separate queues by namespace
     name: OPS_WORKER_NAME,