feat(workpool): support shared batch execution and failure recovery

README.md

@@ -230,6 +230,69 @@ await ping(ctx, components.batchWorker, {
 });
 ```
 
+### Sharing a Workpool to limit parallelism
+
+Pass an optional `Workpool` instance to `ping` to limit concurrent batch
+mutations across named workers. Mount the pool alongside BatchWorker in your
+app's `convex.config.ts`:
+
+```ts
+import workpool from "@convex-dev/workpool/convex.config.js";
+
+app.use(workpool);
+```
+
+This requires Workpool's `onFailure` option, currently available in the
+[PR #231 build](https://github.com/get-convex/workpool/pull/231):
+
+```sh
+npm i https://pkg.pr.new/get-convex/workpool/@convex-dev/workpool@231
+```
+
+```ts
+import { Workpool } from "@convex-dev/workpool";
+import { ping } from "@convex-dev/batch-worker";
+
+const pool = new Workpool(components.workpool, { maxParallelism: 10 });
+
+// Inside the mutation that inserts work:
+await ping(ctx, components.batchWorker, {
+  name: `user:${userId}`,
+  workQuery: internal.jobs.getBatch,
+  workerMutation: internal.jobs.processBatch,
+  workpool: pool,
+});
+```
+
+Each pool task represents one loop iteration, carrying only the worker name
+and generation. It fetches its batch when capacity is available, so work added
+while waiting can join the batch. All workers using the same mounted pool
+share its capacity with any other jobs in that pool. Each named worker still
+processes batches serially. The limit covers the batch mutation, not actions
+it schedules separately; `status: running` also includes workers queued for
+capacity.
+
+Pass the pool consistently on every `ping`. Omitting it selects the scheduler.
+Changing pools moves the current continuation and preserves the cursor and
+debounce eligibility. `start` and `kick` reuse the stored pool. Only enqueue
+and cancel function handles and the pool's concurrency/logging options are
+stored; BatchWorker has no runtime dependency on Workpool.
+
+Pooled workers use `onFailure` instead of a monitor or status polling. A failed
+iteration retries after `config.monitorLagMs` (default 60 seconds, minimum 10
+seconds). Workpool's action retry options do not apply to these mutations.
+`stop` invalidates the current generation and requests cancellation, so even
+an admitted stale mutation cannot commit another batch after the stop commits.
+
+Canceling a job directly through the Workpool API does **not** invoke
+`onFailure`. Recover that worker manually:
+
+```ts
+await ctx.runMutation(components.batchWorker.lib.kick, { name: `user:${userId}` });
+```
+
+See [pooled.ts](./example/convex/pooled.ts) for a complete per-user queue example.
+
 ### The cursor
 
 The component stores the cursor for you: whatever your work query returns
@@ -420,7 +483,8 @@ Tips for rate-limiting LLM calls:
 ### Failure handling
 
 If your work query or worker mutation throws, the loop dies and the liveness
-monitor restarts it after ~`monitorLagMs`. The unprocessed rows are still in
+monitor (or Workpool's `onFailure` callback) restarts it after ~`monitorLagMs`.
+The unprocessed rows are still in
 your table and the cursor stayed where it was, so the query hands out the **same
 batch again**. That gives you at-least-once processing, but it also means one
 poison item that always throws can wedge the queue. For work that can fail per
@@ -513,6 +577,25 @@ The demo is hosted on Convex itself via
 published to `https://<deployment>.convex.site`. See
 [example/README.md](./example/README.md#deploy) for the deploy commands.
 
+### Benchmarking execution cadence
+
+For a live comparison of scheduler and Workpool iteration/polling cadence,
+deploy the example to a development deployment and run:
+
+```sh
+npm run build
+npx convex dev --once
+node execution-benchmark.mjs --duration-ms 10000 --trials 3
+```
+
+The benchmark uses a dedicated pool and internal functions. It compares busy
+iterations and empty-queue polling at 0 and 200 ms, with pool capacities of 1
+and 10, and includes ten-worker throughput. Server execution logs provide the
+timings; instrumentation adds no per-iteration database writes. Results,
+raw logs, and a report go to `.context/execution-benchmark/`. For a shorter run,
+append `--pilot --duration-ms 3000 --trials 1`. Run one benchmark at a time:
+its worker names are reused between trials.
+
 ### How it works
 
 | Table         | Written by                          | Read by                 |

example/convex/_generated/api.d.ts

@@ -11,6 +11,8 @@
 import type * as aggregates from "../aggregates.js";
 import type * as e2e from "../e2e.js";
 import type * as example from "../example.js";
+import type * as executionBenchmark from "../executionBenchmark.js";
+import type * as pooled from "../pooled.js";
 import type * as rateLimited from "../rateLimited.js";
 
 import type {
@@ -23,6 +25,8 @@ declare const fullApi: ApiFromModules<{
   aggregates: typeof aggregates;
   e2e: typeof e2e;
   example: typeof example;
+  executionBenchmark: typeof executionBenchmark;
+  pooled: typeof pooled;
   rateLimited: typeof rateLimited;
 }>;
 
@@ -54,6 +58,8 @@ export declare const internal: FilterApi<
 
 export declare const components: {
   batchWorker: import("@convex-dev/batch-worker/_generated/component.js").ComponentApi<"batchWorker">;
+  workpool: import("@convex-dev/workpool/_generated/component.js").ComponentApi<"workpool">;
+  benchmarkPool: import("@convex-dev/workpool/_generated/component.js").ComponentApi<"benchmarkPool">;
   rateLimiter: import("@convex-dev/rate-limiter/_generated/component.js").ComponentApi<"rateLimiter">;
   staticHosting: import("@convex-dev/static-hosting/_generated/component.js").ComponentApi<"staticHosting">;
 };

example/convex/convex.config.ts

@@ -2,11 +2,16 @@ import { defineApp } from "convex/server";
 import batchWorker from "@convex-dev/batch-worker/convex.config.js";
 import rateLimiter from "@convex-dev/rate-limiter/convex.config.js";
 import staticHosting from "@convex-dev/static-hosting/convex.config.js";
+import workpool from "@convex-dev/workpool/convex.config.js";
 
 // The static-hosting component owns `/` and serves the built demo UI; any app
 // HTTP routes would move under `/api`. This is its fastest serving mode.
 const app = defineApp({ httpPrefix: "/api" });
 app.use(batchWorker);
+// Shared by the per-user workers in pooled.ts.
+app.use(workpool);
+// Isolated from example workloads while comparing execution cadence.
+app.use(workpool, { name: "benchmarkPool" });
 // Used by the rate-limited worker example in rateLimited.ts.
 app.use(rateLimiter);
 // Serves the Vite app in example/src — see `npm run deploy`.

example/convex/executionBenchmark.ts

@@ -0,0 +1,156 @@
+import { Workpool } from "@convex-dev/workpool";
+import { defineBatchWorkerValidators, ping } from "@convex-dev/batch-worker";
+import { v } from "convex/values";
+import { components, internal } from "./_generated/api.js";
+import { internalMutation, internalQuery } from "./_generated/server.js";
+
+// Console markers instrument the real query path without adding per-iteration
+// writes or turning an idle poll into a batch mutation. The Node runner reads
+// successful execution logs and retains their execution timestamps as well.
+const validators = defineBatchWorkerValidators({
+  cursor: v.number(),
+  batch: { iteration: v.number() },
+});
+
+export const start = internalMutation({
+  args: {
+    runId: v.string(),
+    backend: v.union(v.literal("scheduler"), v.literal("workpool")),
+    mode: v.union(v.literal("busy"), v.literal("poll")),
+    durationMs: v.number(),
+    pollIntervalMs: v.number(),
+    workers: v.number(),
+    maxParallelism: v.number(),
+  },
+  returns: v.object({
+    names: v.array(v.string()),
+    startedAt: v.number(),
+    deadline: v.number(),
+  }),
+  handler: async (ctx, args) => {
+    if (
+      !Number.isFinite(args.durationMs) ||
+      args.durationMs < 1000 ||
+      args.durationMs > 60_000 ||
+      !Number.isInteger(args.workers) ||
+      args.workers < 1 ||
+      args.workers > 10 ||
+      !Number.isInteger(args.maxParallelism) ||
+      args.maxParallelism < 1 ||
+      args.maxParallelism > 10 ||
+      !Number.isFinite(args.pollIntervalMs) ||
+      args.pollIntervalMs < 0 ||
+      args.pollIntervalMs > 1000
+    ) {
+      throw new Error(
+        "Benchmark limits: 1–60s, 1–10 workers/slots, 0–1000ms polls",
+      );
+    }
+    const startedAt = Date.now();
+    const deadline = startedAt + args.durationMs;
+    const pool = new Workpool(components.benchmarkPool, {
+      maxParallelism: args.maxParallelism,
+      logLevel: "ERROR",
+    });
+    const names: string[] = [];
+    for (let i = 0; i < args.workers; i++) {
+      // Reuse names between trials to bound benchmark metadata growth.
+      const name = `benchmark:${args.backend}:${args.mode}:${args.pollIntervalMs}:${i}`;
+      names.push(name);
+      const previous = await ctx.db
+        .query("executionBenchmarks")
+        .withIndex("name", (q) => q.eq("name", name))
+        .unique();
+      const config = {
+        name,
+        runId: args.runId,
+        mode: args.mode,
+        startedAt,
+        deadline,
+        pollIntervalMs: args.pollIntervalMs,
+      };
+      if (previous)
+        await ctx.db.replace("executionBenchmarks", previous._id, config);
+      else await ctx.db.insert("executionBenchmarks", config);
+      await ctx.runMutation(components.batchWorker.lib.stop, { name });
+      await ping(ctx, components.batchWorker, {
+        name,
+        workQuery: internal.executionBenchmark.getBatch,
+        workerMutation: internal.executionBenchmark.processBatch,
+        ...(args.backend === "workpool" ? { workpool: pool } : {}),
+      });
+      await ctx.runMutation(components.batchWorker.lib.setCursor, { name });
+      await ctx.runMutation(components.batchWorker.lib.start, { name });
+    }
+    return { names, startedAt, deadline };
+  },
+});
+
+export const getBatch = internalQuery({
+  args: validators.vQueryArgs,
+  returns: validators.vQueryReturns,
+  handler: async (ctx, { name, cursor }) => {
+    const config = await ctx.db
+      .query("executionBenchmarks")
+      .withIndex("name", (q) => q.eq("name", name))
+      .unique();
+    if (!config || Date.now() >= config.deadline || (cursor ?? 0) >= 10_000) {
+      return { kind: "idle" as const, cooldownMs: 0 };
+    }
+    const phase =
+      config.mode === "poll"
+        ? cursor === undefined
+          ? "bootstrap"
+          : "poll"
+        : "iteration";
+    console.log(
+      "BATCH_BENCH",
+      JSON.stringify({
+        runId: config.runId,
+        name,
+        phase,
+        atMs: Date.now(),
+        iteration: cursor ?? 0,
+      }),
+    );
+    if (phase === "poll") {
+      return {
+        kind: "idle" as const,
+        cooldownMs: config.deadline - config.startedAt + 1000,
+        pollIntervalMs: config.pollIntervalMs,
+      };
+    }
+    // A single bootstrap batch starts the cooldown clock for polling trials.
+    return { batch: { iteration: cursor ?? 0 }, cursor: (cursor ?? 0) + 1 };
+  },
+});
+
+export const processBatch = internalMutation({
+  args: validators.vMutationArgs,
+  returns: validators.vMutationReturns,
+  handler: async () => null,
+});
+
+export const status = internalQuery({
+  args: { names: v.array(v.string()) },
+  returns: v.boolean(),
+  handler: async (ctx, { names }) => {
+    for (const name of names) {
+      const state = await ctx.runQuery(components.batchWorker.lib.status, {
+        name,
+      });
+      if (state?.kind === "running") return false;
+    }
+    return true;
+  },
+});
+
+export const stop = internalMutation({
+  args: { names: v.array(v.string()) },
+  returns: v.null(),
+  handler: async (ctx, { names }) => {
+    for (const name of names)
+      await ctx.runMutation(components.batchWorker.lib.stop, { name });
+    return null;
+  },
+});

example/convex/pooled.ts

@@ -0,0 +1,88 @@
+import { Workpool } from "@convex-dev/workpool";
+import { defineBatchWorkerValidators, ping } from "@convex-dev/batch-worker";
+import { v } from "convex/values";
+import { components, internal } from "./_generated/api.js";
+import { internalMutation, internalQuery } from "./_generated/server.js";
+
+const pool = new Workpool(components.workpool, { maxParallelism: 10 });
+const validators = defineBatchWorkerValidators({
+  batch: {
+    user: v.string(),
+    events: v.array(v.object({ id: v.id("userEvents"), value: v.number() })),
+  },
+});
+
+// Internal demo entry point. A public endpoint should derive the user from auth.
+// npx convex run pooled:addEvents '{"user":"alice","values":[1,2,3]}'
+export const addEvents = internalMutation({
+  args: { user: v.string(), values: v.array(v.number()) },
+  returns: v.null(),
+  handler: async (ctx, { user, values }) => {
+    for (const value of values) {
+      await ctx.db.insert("userEvents", {
+        user,
+        value,
+        insertedAt: ctx.db.vars.commitTs,
+      });
+    }
+    await ping(ctx, components.batchWorker, {
+      name: `user:${user}`,
+      workQuery: internal.pooled.getBatch,
+      workerMutation: internal.pooled.processBatch,
+      workpool: pool,
+    });
+    return null;
+  },
+});
+
+export const getBatch = internalQuery({
+  args: validators.vQueryArgs,
+  returns: validators.vQueryReturns,
+  handler: async (ctx, { name, cursor }) => {
+    const user = name.slice("user:".length);
+    const events = await ctx.db
+      .query("userEvents")
+      .withIndex("user_insertedAt", (q) =>
+        q.eq("user", user).gte("insertedAt", cursor ?? 0n),
+      )
+      .take(100);
+    if (!events.length) return { kind: "idle" as const };
+    return {
+      batch: {
+        user,
+        events: events.map((event) => ({ id: event._id, value: event.value })),
+      },
+      cursor: events.at(-1)!.insertedAt,
+    };
+  },
+});
+
+export const processBatch = internalMutation({
+  args: validators.vMutationArgs,
+  returns: validators.vMutationReturns,
+  handler: async (ctx, { user, events }) => {
+    const previous = await ctx.db
+      .query("userTotals")
+      .withIndex("user", (q) => q.eq("user", user))
+      .unique();
+    const total =
+      (previous?.total ?? 0) +
+      events.reduce((sum, event) => sum + event.value, 0);
+    if (previous) await ctx.db.patch("userTotals", previous._id, { total });
+    else await ctx.db.insert("userTotals", { user, total });
+    for (const event of events) await ctx.db.delete("userEvents", event.id);
+    return null;
+  },
+});
+
+export const getTotal = internalQuery({
+  args: { user: v.string() },
+  returns: v.number(),
+  handler: async (ctx, { user }) => {
+    const total = await ctx.db
+      .query("userTotals")
+      .withIndex("user", (q) => q.eq("user", user))
+      .unique();
+    return total?.total ?? 0;
+  },
+});