Manage a cursor for worker automatically

README.md

@@ -56,14 +56,20 @@ component too: call `component.use(batchWorker)` in your component's
 
 ## Usage
 
-Insert work into your own table, then call `ping`. Provide a query (typed with
-`vBatchQueryArgs` / `vBatchResult`) that returns the next batch or `idle`, and a
-mutation that processes it. The query's `batch` shape must match the mutation's
-args. (`kind: "work"` is optional — returning `{ batch }` alone also works.)
+Insert work into your own table, then call `ping`. Provide a query that returns
+the next batch or `idle`, and a mutation that processes it.
+`defineBatchWorkerValidators` declares the batch shape once and hands you all
+four validators, so the two halves can't drift apart. (`kind: "work"` is
+optional — returning `{ batch }` alone also works.)
+
+Alongside the batch, the query returns a **cursor** saying how far it got. The
+component commits it with the batch and hands it back as `args.cursor` next
+time, so each scan resumes where the last one stopped. See
+[why that matters](#why-is-a-cursor-necessary).
 
 ```ts
 import { v } from "convex/values";
-import { ping, vBatchQueryArgs, vBatchResult } from "@convex-dev/batch-worker";
+import { defineBatchWorkerValidators, ping } from "@convex-dev/batch-worker";
 import { components, internal } from "./_generated/api";
 import { internalMutation, internalQuery, mutation } from "./_generated/server";
 
@@ -73,7 +79,9 @@ const BATCH_SIZE = 10;
 export const addEvent = mutation({
   args: { value: v.number() },
   handler: async (ctx, { value }) => {
-    await ctx.db.insert("events", { value });
+    // `db.vars.commitTs` resolves at commit to an int64 ordered by commit
+    // order — that's what makes it safe to cursor on. See below.
+    await ctx.db.insert("events", { value, insertedAt: ctx.db.vars.commitTs });
     await ping(ctx, components.batchWorker, {
       name: "events", // distinct names give you independent queues
       workQuery: internal.example.getBatch,
@@ -84,28 +92,38 @@ export const addEvent = mutation({
 
 const vEvents = v.array(v.object({ id: v.id("events"), value: v.number() }));
 
+// One declaration → the query's args (`{ name, cursor? }`) and returns, plus
+// the mutation's args. `cursor` defaults to a commit timestamp.
+const { vQueryArgs, vQueryReturns, vMutationArgs } =
+  defineBatchWorkerValidators({ batch: { events: vEvents } });
+
 // Return the next batch of work, or `idle` when there's nothing to do.
 export const getBatch = internalQuery({
-  args: vBatchQueryArgs, // { name } — lets one query serve multiple queues
-  returns: vBatchResult({ events: vEvents }),
-  handler: async (ctx) => {
-    // See the "Resume from a cursor" section below for an optimized approach
-    const events = await ctx.db.query("events").take(BATCH_SIZE);
+  args: vQueryArgs, // `name` lets one query serve many queues
+  returns: vQueryReturns,
+  handler: async (ctx, { cursor }) => {
+    // Resume where the last batch stopped instead of rescanning from the front.
+    const events = await ctx.db
+      .query("events")
+      .withIndex("insertedAt", (q) => q.gte("insertedAt", cursor ?? 0n))
+      .take(BATCH_SIZE);
     if (events.length === 0) {
       return { kind: "idle" as const };
       // Or, if you know when the next item is due:
       // return { kind: "idle" as const, timeoutMs: 30_000 };
     }
     return {
       kind: "work" as const,
-      batch: events.map((e) => ({ id: e._id, value: e.value })),
+      batch: { events: events.map((e) => ({ id: e._id, value: e.value })) },
+      // Rows come back in commit order, so the last one is how far we got.
+      cursor: events.at(-1)!.insertedAt,
     };
   },
 });
 
 // Process one batch. The worker owns cleanup — delete what you process!
 export const processBatch = internalMutation({
-  args: { events: vEvents },
+  args: vMutationArgs,
   handler: async (ctx, { events }) => {
     for (const { value, id } of events) {
       // ... do the work (sum, call an API, schedule downstream jobs, etc.) ...
@@ -122,9 +140,6 @@ The component **does not clean up your work for you** — your worker mutation i
 responsible for deleting (or marking complete / advancing past) the rows it
 processed, otherwise the next query will return them again.
 
-Note: the snippet above keeps that as simple as possible; for a busy worker, use
-a cursor as described [below](#use-a-cursor-to-track-worker-progress).
-
 ### Fetch work in the query, process it in the mutation
 
 Think of the work query and the worker mutation as **two separate transactions**
@@ -154,7 +169,8 @@ return {
 
 ### Steering the loop dynamically
 
-Your worker mutation may return `{ debounceMs }` to throttle the loop:
+Your worker mutation may return `{ debounceMs }` to throttle the loop (and a
+`cursor` to override the query's — see [The cursor](#the-cursor)):
 
 ```ts
 return {
@@ -214,80 +230,104 @@ await ping(ctx, components.batchWorker, {
 });
 ```
 
-### Use a cursor to track worker progress
+### The cursor
 
-For best performance, it's advised to use a cursor to track how far you've read,
-and use it to focus future queries, even if you're deleting documents as you go.
-See below for [why a cursor is necessary](#why-is-a-cursor-necessary).
+The component stores the cursor for you: whatever your work query returns
+alongside a batch comes back as `args.cursor` on the next call. It's committed
+in the same transaction as the batch, so if your worker mutation throws, the
+cursor stays where it was and the batch is handed out again.
 
-The best cursor to use is the
+The default cursor type is a
 [commit timestamp](https://docs.convex.dev/database/advanced/commit-timestamp)
-(Convex ≥ 1.43), as it avoids having to handle out-of-order commits. Do not use
-`_creationTime`. It's assigned when a mutation _starts_, so a row inserted by a
-slow mutation can land _behind_ rows that were already committed and queried by
-the worker.
-
-**Step 1**: Add a field to your schema tracking the commit timestamp for work
-you'll query, and track the cursor somewhere for the worker to access and
-update.
+(Convex ≥ 1.43), which avoids having to handle out-of-order commits. Add a
+`v.commitTs()` field to the table the worker scans, index it, and write
+`ctx.db.vars.commitTs` to it on every insert or patch:
 
 ```ts
 // convex/schema.ts
-const schema = defineSchema({
-  myWork: defineTable({
-    updatedAt: v.commitTs(),
-    ...fields
-  }).index("updatedAt", ["updatedAt"]),
-  workerState: defineTable({
-    cursor: v.int64(), // a commit timestamp resolves to a bigint
-    ...
-  })
-  ...
-});
+myWork: defineTable({
+  updatedAt: v.commitTs(), // resolves at commit to an int64 in commit order
+  ...fields
+}).index("updatedAt", ["updatedAt"]),
 ```
 
-**Step 2:** Set it when inserting or updating a document that needs processing:
+```ts
+await ctx.db.insert("myWork", { updatedAt: ctx.db.vars.commitTs, ... });
+await ctx.db.patch("myWork", id, { updatedAt: ctx.db.vars.commitTs, ... });
+```
+
+Do **not** use `_creationTime`. It's assigned when a mutation _starts_, so a row
+inserted by a slow mutation can land _behind_ rows the worker has already
+queried past.
+
+**Use `gte`, not `gt`.** Everything a single mutation inserts shares one commit
+timestamp, and a batch can end in the middle of such a tie — `gt` would skip the
+rest of it. Re-reading from the tie costs at most one mutation's worth of
+already-processed rows.
+
+An alternative is to read the remaining documents at that same commit timestamp
+into the batch so it ends on a boundary, which also ensures work added "at once"
+is processed together — if you can be sure that fits in one transaction. See
+[the aggregate example](./example/convex/aggregates.ts).
+
+#### Other cursor types
+
+The cursor can be any Convex value — for instance a `paginator` cursor from
+`convex-helpers`, which captures the index range
+`[commitTs, _creationTime, _id]` so you know exactly where you left off. Pass
+`cursor` to `defineBatchWorkerValidators` and all four validators pick it up:
 
 ```ts
-// On insert/patch, use the placeholder, (resolves to a bigint at commit time)
-await ctx.db.insert("myWork", { updatedAt: ctx.db.vars.commitTs, ... }); await
-ctx.db.patch("myWork", id, { updatedAt: ctx.db.vars.commitTs, ... });
+const { vQueryArgs, vQueryReturns, vMutationArgs, vMutationReturns } =
+  defineBatchWorkerValidators({
+    batch: { ids: v.array(v.id("myWork")) },
+    cursor: v.string(),
+  });
 ```
 
-**Step 3:** In the worker query, resume from the cursor and report how far you
-got:
+`ping` takes the cursor's type from the work query's `cursor` arg and checks the
+two return positions against it, so a mismatch points at one place.
+
+> The older `vBatchQueryArgs`, `vBatchResult` and `vWorkerResult` are
+> deprecated. They still work, but each is fixed to the commit-timestamp cursor
+> and they're declared separately enough that the query and mutation can drift
+> apart.
+
+#### Adjusting the cursor from the worker mutation
+
+The worker mutation can return a `cursor` that overrides the query's — use it
+when the batch only made partial progress and you can work out how far it
+actually got, e.g. from a per-item timestamp carried in the batch:
 
 ```ts
-const workerState = await getWorkerState(ctx, ...) // however you access it
-const work = await ctx.db
-  .query("myWork")
-  .withIndex("updatedAt", (q) => q.gte("updatedAt", workerState.cursor))
-  .take(BATCH_SIZE);
-if (events.length === 0)  return { kind: "idle" as const };
-const cursor = work.at(-1)!.updatedAt as bigint; // cast is fine here
-return { kind: "work" as const, batch: { cursor, ... } };
+return { cursor: lastProcessed.updatedAt };
 ```
 
-Note: the query uses `gte`, not `gt`, since multiple entries may have been
-inserted at the same commit timestamp (they committed transactionally). If our
-batch of work lands in the middle of them, we don't want to skip the rest.
+You can also return `ctx.db.vars.commitTs` to mean "everything up to now": the
+worker mutation shares the loop's transaction, so it resolves to the same
+timestamp this batch stamped on its own rows.
 
-An alternative here is to read the remaining documents from that same commitTs
-and include them in this batch. This has the advantage of ensuring all work
-items added at "once" get processed together, if you can guarantee you can
-process all of that work at once without exceeding transaction limits. See an
-example of that in [the aggregate example](./example/convex/aggregates.ts).
+If a batch can't be processed at all, throw — that rolls back the transaction,
+so the cursor doesn't move and the monitor retries the batch.
 
-Another option is to use `paginator` from `convex-helpers` to track a cursor
-that captures the index range `[commitTs, _creationTime, _id]`, so you know
-exactly where you left off while still being immune to out-of-order commits.
+#### Reading and resetting the cursor
 
-**Step 4:** In the worker mutation, update the cursor
+Normally the work query gets the cursor in its args and that's all you need. If
+one of your own queries wants to scan the same range — counting pending work,
+say — read it off the component, like `status`:
 
 ```ts
-await ctx.db.patch("workerState", workerStateId, { cursor: args.cursor });
+const from = ((await ctx.runQuery(components.batchWorker.lib.getCursor, {
+  name: "events",
+})) ?? 0n) as bigint;
 ```
 
+`lib.setCursor` overwrites it, and clears it when `cursor` is omitted, so the
+next scan starts from the front. It's for migrations and recovery — the loop
+writes that document every iteration, so calling it while the worker is busy is
+liable to conflict. It throws if the worker doesn't exist yet (`ping` creates
+it).
+
 #### Why is a cursor necessary?
 
 The reason is due to a technical nuance of how Convex tracks document changes.
@@ -389,10 +429,11 @@ Tips for rate-limiting LLM calls:
 
 If your work query or worker mutation throws, the loop dies and the liveness
 monitor restarts it after ~`monitorLagMs` — and since the unprocessed rows are
-still in your table, the query will hand 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 item, catch errors
-inside the worker mutation, and isolate bad docs in a table for async debugging.
+still in your table and the cursor didn't move, the query will hand 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 item, catch errors inside the worker mutation, and isolate bad docs in a
+table for async debugging.
 
 This is a low-level primitive, relative to components like Workpool or Workflow,
 so you have to handle exceptional cases yourself.
@@ -457,9 +498,8 @@ Note: Use `convex-test` (≥ 0.0.55-alpha.0) for `v.commitTs()` support.
 
 See the full working examples in the example app:
 [example.ts](./example/convex/example.ts) (basic queue),
-[aggregates.ts](./example/convex/aggregates.ts) (denormalized aggregates),
-[rateLimited.ts](./example/convex/rateLimited.ts) (async LLM batches), and
-[cursor.ts](./example/convex/cursor.ts) (the shared `commitTs` cursor helpers).
+[aggregates.ts](./example/convex/aggregates.ts) (denormalized aggregates), and
+[rateLimited.ts](./example/convex/rateLimited.ts) (async LLM batches).
 
 <!-- END: Include on https://convex.dev/components -->
 
@@ -489,7 +529,8 @@ published to `https://<deployment>.convex.site`. See
 | `workerState` | `loop` (every iteration)            | `loop`, monitor         |
 
 The high-churn loop state lives in `workerState` (generation, heartbeat, the
-scheduled runner, and the monitor), separate from the rarely-written `workers`
+scheduled runner, the monitor, and the cursor — which rides along on a patch the
+loop already does every iteration), separate from the rarely-written `workers`
 doc (which holds the handles, config, and run-status: `idle` / `running` /
 `stopped`, plus a pointer to its `workerState`). That lets `ping`/`start` —
 which you call on every insert — read `workers` and return without conflicting

example/README.md

@@ -9,9 +9,9 @@ defines three named workers and the Vite UI ([`src/`](./src)) drives each one:
 - **Rate-limited LLM batches** ([rateLimited.ts](./convex/rateLimited.ts)) —
   collect requests and spend a token budget on async calls.
 
-All three drain their queue by resuming from a `v.commitTs()` cursor
-([cursor.ts](./convex/cursor.ts)) rather than rescanning the front of the table
-each batch.
+All three drain their queue by returning a `v.commitTs()` cursor from the work
+query — the component commits it with the batch and hands it back on the next
+call — rather than rescanning the front of the table each batch.
 
 Run all commands from the **root of the repo**.
 

example/convex/_generated/api.d.ts

@@ -9,7 +9,6 @@
  */
 
 import type * as aggregates from "../aggregates.js";
-import type * as cursor from "../cursor.js";
 import type * as e2e from "../e2e.js";
 import type * as example from "../example.js";
 import type * as rateLimited from "../rateLimited.js";
@@ -22,7 +21,6 @@ import type {
 
 declare const fullApi: ApiFromModules<{
   aggregates: typeof aggregates;
-  cursor: typeof cursor;
   e2e: typeof e2e;
   example: typeof example;
   rateLimited: typeof rateLimited;

example/convex/aggregates.test.ts

@@ -1,6 +1,6 @@
 import { afterEach, beforeEach, describe, expect, test, vi } from "vitest";
 import { initConvexTest } from "./setup.test";
-import { api } from "./_generated/api";
+import { api, components } from "./_generated/api";
 
 describe("serial aggregate updates", () => {
   beforeEach(() => {
@@ -92,13 +92,14 @@ describe("serial aggregate updates", () => {
     await t.mutation(api.aggregates.recordScore, { team: "red", points: 2 });
     await t.finishAllScheduledFunctions(vi.runAllTimers);
 
-    const [scores, cursor] = await t.run(async (ctx) => [
-      await ctx.db.query("scoreEvents").withIndex("insertedAt").collect(),
-      await ctx.db
-        .query("cursors")
-        .withIndex("name", (q) => q.eq("name", "aggregates"))
-        .unique(),
-    ]);
-    expect(cursor?.commitTs).toBe(scores.at(-1)!.insertedAt);
+    const scores = await t.run((ctx) =>
+      ctx.db.query("scoreEvents").withIndex("insertedAt").collect(),
+    );
+    const cursor = await t.run((ctx) =>
+      ctx.runQuery(components.batchWorker.lib.getCursor, {
+        name: "aggregates",
+      }),
+    );
+    expect(cursor).toBe(scores.at(-1)!.insertedAt);
   });
 });