Clarify snapshot consistency and revalidate example batches

README.md

@@ -90,12 +90,12 @@ export const addEvent = mutation({
   },
 });
 
-const vEvents = v.array(v.object({ id: v.id("events"), value: v.number() }));
+const vEventIds = v.array(v.id("events"));
 
 // One declaration gives 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 } });
+  defineBatchWorkerValidators({ batch: { ids: vEventIds } });
 
 // Return the next batch of work, or `idle` when there's nothing to do.
 export const getBatch = internalQuery({
@@ -114,7 +114,7 @@ export const getBatch = internalQuery({
     }
     return {
       kind: "work" as const,
-      batch: { events: events.map((e) => ({ id: e._id, value: e.value })) },
+      batch: { ids: events.map((e) => e._id) },
       // Rows come back in commit order, so the last one is how far we got.
       cursor: events.at(-1)!.insertedAt,
     };
@@ -124,9 +124,12 @@ export const getBatch = internalQuery({
 // Process one batch. The worker owns cleanup — delete what you process!
 export const processBatch = internalMutation({
   args: vMutationArgs,
-  handler: async (ctx, { events }) => {
-    for (const { value, id } of events) {
-      // ... do the work (sum, call an API, schedule downstream jobs, etc.) ...
+  handler: async (ctx, { ids }) => {
+    for (const id of ids) {
+      // Re-fetch if other mutations may edit or delete queued events.
+      const event = await ctx.db.get("events", id);
+      if (!event) continue; // deleted by another mutation
+      // ... do the work using event.value (sum, schedule downstream jobs, etc.) ...
 
       // Clean up the work you processed.
       await ctx.db.delete("events", id);
@@ -140,33 +143,61 @@ 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.
 
-### Fetch work in the query, process it in the mutation
+### Find work in the query, process it in the mutation
 
-Think of the work query and the worker mutation as **two separate transactions**
-(under the hood they run against the same database snapshot, but they behave
-independently for conflict purposes). This split enables concurrent processing
-without database conflicts on work being added while being fetched.
+Treat the work query and the worker mutation as **two separate transactions**.
+The query may read an older database snapshot than the mutation, but **each
+round's query sees the committed writes from this worker's previous rounds**. If
+a batch deletes rows or marks them started, the next query observes those
+changes, so it can exclude that work from its results.
+
+Other mutations can still change, delete, or claim rows between the query's
+snapshot and the worker mutation's snapshot. Revalidate in the mutation when
+processing depends on those concurrent changes.
 
 - The **work query** runs as a snapshot read that takes **no read
   dependencies**. It can scan the queue (`.take(BATCH_SIZE)` over an index)
   alongside concurrent inserts.
 - The **worker mutation** is a normal transaction: every range it reads becomes
   a read dependency. If it re-queries the range of work in the queue, concurrent
-  inserts would conflict, so do range reads in the work query only.
-
-So: **fetch the batch in the query, and pass everything the mutation needs
-through `batch`**, so the mutation doesn't need to re-fetch it. Note: when the
-mutation updates rows by `_id`, it will depend on those documents, so it is
-protected against concurrent changes to those documents.
+  inserts would conflict. Keep queue scans in the query and use point reads by
+  `_id` when selected rows need revalidation in the mutation.
+
+If other writers can change the work, **return IDs from the query, then re-fetch
+those rows in the mutation**. Check that each row still exists and is eligible
+for work (for example, `state === "pending"`), and compute updates from its
+current values. These point reads participate in the mutation's conflict
+detection without adding a dependency on the whole queue.
+
+You can **pass values directly through `batch`** when they cannot change before
+processing, or when your application can safely process older values. For
+example, if events are immutable until this worker processes and deletes them,
+and no other consumer edits or deletes them, there is no need to re-fetch just
+to check for deletes from earlier rounds. If events are retained, an exclusive
+cursor can track which ones have been processed. Document these assumptions.
+
+Patching a row by `_id` takes a dependency in the mutation's snapshot; it **does
+not validate values from the query's older snapshot** or establish that another
+writer has left the row eligible for processing.
 
 ```ts
-// ✅ Query scans; mutation gets the data handed to it.
+// When other writers can change the work, return IDs for revalidation.
 return {
   kind: "work" as const,
-  batch: { events: rows.map((e) => ({ id: e._id, value: e.value })) },
+  batch: { ids: rows.map((row) => row._id) },
 };
 ```
 
+```ts
+// Mutation: check for concurrent changes before applying effects.
+for (const id of ids) {
+  const row = await ctx.db.get("tasks", id);
+  if (!row || row.state !== "pending") continue;
+  await ctx.db.patch("tasks", id, { state: "started" });
+  // Schedule work using the current row's values.
+}
+```
+
 ### Steering the loop dynamically
 
 Your worker mutation may return `{ debounceMs }` to throttle the loop, and a
@@ -235,7 +266,8 @@ await ping(ctx, components.batchWorker, {
 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.
+cursor stays where it was and the query retries from that position. The retry
+may return a different batch as the query's snapshot changes.
 
 The default cursor type is a
 [commit timestamp](https://docs.convex.dev/database/advanced/commit-timestamp)
@@ -364,9 +396,14 @@ the value. If you need the fully-up-to-date value, you have a couple options:
 
 The worker mutation is a transaction, so it can't call external APIs itself. To
 batch work that requires `fetch`— calling an LLM, hitting a third-party API —
-have the worker mutation _claim_ a batch (mark the rows started so the query
-won't hand them out again) and then schedule the action, or enqueue it in a
-Workpool. The action does the work and calls a mutation to commit the results
+have the worker mutation _claim_ a batch: re-fetch the candidate IDs, check that
+each row is still pending, and mark those rows started. Then schedule the
+action, or enqueue it in a Workpool, using only the rows actually claimed. The
+next round's query sees this worker's claims and excludes those rows from its
+pending results. Re-fetching protects against other mutations changing or
+claiming requests between snapshots; if this worker is the only consumer and
+pending requests cannot change, it can claim them using values passed through
+`batch`. The action does the work and calls a mutation to commit the results
 back.
 
 Full working code is in [rateLimited.ts](./example/convex/rateLimited.ts). It
@@ -420,12 +457,14 @@ 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
-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
-item, catch errors inside the worker mutation, and isolate bad docs in a table
-for async debugging.
+monitor restarts it after ~`monitorLagMs`. The failed mutation's writes and
+cursor advance roll back, so the query retries from the previous cursor. Its
+next batch may differ because other mutations can change the work before the
+retry. Successful batches' committed writes remain visible to later rounds, so
+their cleanup is not lost on retry. Re-check rows if other writers can change
+them. 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.

example/README.md

@@ -13,6 +13,14 @@ 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, so each scan resumes where the last one stopped.
 
+The work query may read an older snapshot than the worker mutation, but each
+round sees this worker's committed writes from previous rounds, including
+deletes and claims. The queue and LLM examples return IDs and re-fetch current
+rows to handle edits, deletes, or claims by other mutations between snapshots.
+If only this worker can change or remove queued work, values can be passed
+directly through the batch. The scoreboard does this because its score events
+are immutable and its exclusive cursor tracks processed events.
+
 Run all commands from the **root of the repo**.
 
 ## Develop

example/convex/aggregates.ts

@@ -14,6 +14,11 @@ import {
 // batches of them into per-team totals. Because exactly one loop runs at a
 // time, those hot aggregate rows have a single writer — so they never OCC
 // against each other or against incoming scores, no matter the rate.
+//
+// This example treats scoreEvents as immutable, append-only events. Passing
+// their values through the batch is safe under that assumption; an exclusive
+// cursor skips events already folded into the totals. If events can be edited
+// or deleted, return IDs and re-check their current state in the mutation.
 
 const WORKER = "aggregates";
 

example/convex/e2e.ts

@@ -54,9 +54,7 @@ export const enqueue = mutation({
 const { vQueryArgs, vQueryReturns, vMutationArgs } =
   defineBatchWorkerValidators({
     batch: {
-      items: v.array(
-        v.object({ id: v.id("e2eEvents"), creationTime: v.number() }),
-      ),
+      ids: v.array(v.id("e2eEvents")),
     },
   });
 
@@ -76,10 +74,7 @@ export const getBatch = internalQuery({
     return {
       kind: "work" as const,
       batch: {
-        items: events.map((e) => ({
-          id: e._id,
-          creationTime: e._creationTime,
-        })),
+        ids: events.map((e) => e._id),
       },
       cursor: events.at(-1)!.updatedAt,
     };
@@ -88,17 +83,23 @@ export const getBatch = internalQuery({
 
 export const processBatch = internalMutation({
   args: vMutationArgs,
-  handler: async (ctx, { items }) => {
+  handler: async (ctx, { ids }) => {
+    // Match the queue example: measure only rows still present in this snapshot.
+    const candidates = await Promise.all(
+      ids.map((id) => ctx.db.get("e2eEvents", id)),
+    );
+    const items = candidates.filter((item) => item !== null);
+    if (items.length === 0) return;
     const now = Date.now();
-    const latencies = items.map((b) => now - b.creationTime);
+    const latencies = items.map((b) => now - b._creationTime);
     await ctx.db.insert("e2eSamples", {
       processedAt: now,
       batchSize: items.length,
       oldestLatencyMs: Math.max(...latencies),
       newestLatencyMs: Math.min(...latencies),
     });
     for (const b of items) {
-      await ctx.db.delete("e2eEvents", b.id);
+      await ctx.db.delete("e2eEvents", b._id);
     }
   },
 });

example/convex/example.test.ts

@@ -1,6 +1,6 @@
 import { afterEach, beforeEach, describe, expect, test, vi } from "vitest";
 import { initConvexTest } from "./setup.test";
-import { api, components } from "./_generated/api";
+import { api, components, internal } from "./_generated/api";
 
 describe("example worker", () => {
   beforeEach(() => {
@@ -21,6 +21,35 @@ describe("example worker", () => {
     expect(totals).toEqual({ total: 5, count: 1, pending: 0 });
   });
 
+  test("re-fetches changed events and skips deleted or already processed candidates", async () => {
+    const t = initConvexTest();
+    const ids = await t.run(async (ctx) => {
+      const first = await ctx.db.insert("events", {
+        value: 1,
+        insertedAt: ctx.db.vars.commitTs,
+      });
+      const second = await ctx.db.insert("events", {
+        value: 2,
+        insertedAt: ctx.db.vars.commitTs,
+      });
+      return [first, second];
+    });
+    const batch = await t.query(internal.example.getBatch, { name: "events" });
+    if (!batch.batch) throw new Error("Expected work");
+    // Explicitly create the gap between the query and mutation snapshots.
+    await t.run(async (ctx) => {
+      await ctx.db.patch("events", ids[0], { value: 10 });
+      await ctx.db.delete("events", ids[1]);
+    });
+    await t.mutation(internal.example.processBatch, batch.batch);
+    await t.mutation(internal.example.processBatch, batch.batch);
+    expect(await t.query(api.example.getTotals, {})).toEqual({
+      total: 10,
+      count: 1,
+      pending: 0,
+    });
+  });
+
   test("batches many events across iterations", async () => {
     const t = initConvexTest();
     // 25 events => 3 batches of 10/10/5.

example/convex/example.ts

@@ -33,10 +33,8 @@ export const addEvent = mutation({
   },
 });
 
-const vEvent = v.object({ id: v.id("events"), value: v.number() });
-
 const { vQueryArgs, vQueryReturns, vMutationArgs, vMutationReturns } =
-  defineBatchWorkerValidators({ batch: { events: v.array(vEvent) } });
+  defineBatchWorkerValidators({ batch: { ids: v.array(v.id("events")) } });
 
 /**
  * The work query: returns the next batch of work, or `idle` when the queue is
@@ -59,7 +57,7 @@ export const getBatch = internalQuery({
     }
     return {
       kind: "work" as const,
-      batch: { events: events.map((e) => ({ id: e._id, value: e.value })) },
+      batch: { ids: events.map((e) => e._id) },
       // Rows come back in commit order, so the last one is how far we got.
       // The component commits this with the batch and hands it back above.
       cursor: events.at(-1)!.insertedAt,
@@ -75,7 +73,14 @@ export const getBatch = internalQuery({
 export const processBatch = internalMutation({
   args: vMutationArgs,
   returns: vMutationReturns,
-  handler: async (ctx, { events }) => {
+  handler: async (ctx, { ids }) => {
+    // Re-fetch to handle edits or deletes by other mutations between snapshots.
+    // The query already sees this worker's deletes from previous rounds.
+    const candidates = await Promise.all(
+      ids.map((id) => ctx.db.get("events", id)),
+    );
+    const events = candidates.filter((event) => event !== null);
+    if (events.length === 0) return null;
     const sum = events.reduce((a, e) => a + e.value, 0);
     const totals = await ctx.db
       .query("totals")
@@ -93,8 +98,8 @@ export const processBatch = internalMutation({
         count: events.length,
       });
     }
-    for (const { id } of events) {
-      await ctx.db.delete("events", id);
+    for (const { _id } of events) {
+      await ctx.db.delete("events", _id);
     }
   },
 });

example/convex/rateLimited.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, internal } from "./_generated/api";
 
 describe("rate-limited LLM batches", () => {
   beforeEach(() => {
@@ -27,6 +27,55 @@ describe("rate-limited LLM batches", () => {
     expect(requests.every((r) => r.response?.startsWith("Echo:"))).toBe(true);
   });
 
+  test("re-checks stale candidates before reserving tokens and scheduling work", async () => {
+    const t = initConvexTest();
+    const ids = await t.run(async (ctx) => {
+      const ids = [];
+      for (let i = 0; i < 3; i++) {
+        ids.push(
+          await ctx.db.insert("llmRequests", {
+            prompt: `old ${i}`,
+            inputTokens: 1,
+            state: "pending",
+            updatedAt: ctx.db.vars.commitTs,
+          }),
+        );
+      }
+      return ids;
+    });
+    const batch = await t.query(internal.rateLimited.getBatch, { name: "llm" });
+    if (!batch.batch) throw new Error("Expected work");
+    await t.run(async (ctx) => {
+      await ctx.db.patch("llmRequests", ids[0], {
+        prompt: "current prompt",
+        inputTokens: 2,
+      });
+      await ctx.db.patch("llmRequests", ids[1], {
+        state: "finished",
+        response: "already done",
+      });
+      await ctx.db.delete("llmRequests", ids[2]);
+    });
+    await t.mutation(internal.rateLimited.startBatch, batch.batch);
+    const scheduled = await t.run((ctx) =>
+      ctx.db.system.query("_scheduled_functions").collect(),
+    );
+    await t.mutation(internal.rateLimited.startBatch, batch.batch);
+    expect(
+      await t.run((ctx) =>
+        ctx.db.system.query("_scheduled_functions").collect(),
+      ),
+    ).toEqual(scheduled);
+    expect(scheduled).toHaveLength(1);
+    expect(scheduled[0].args).toEqual([
+      { prompts: [{ id: ids[0], prompt: "current prompt" }] },
+    ]);
+    await t.finishAllScheduledFunctions(vi.runAllTimers);
+    expect(
+      await t.run((ctx) => ctx.db.get("llmRequests", ids[1])),
+    ).toMatchObject({ response: "already done" });
+  });
+
   test("claims every request once when one transaction's requests span batches", async () => {
     const t = initConvexTest();
     // 11 requests in one transaction — one commit timestamp, split by