Order pending work by commit and snapshot timestamps

.changeset/commit-ordered-queues.md

@@ -0,0 +1,14 @@
+---
+"@convex-dev/workpool": minor
+---
+
+Order queues by commit timestamp and recover near-term scheduled work that
+commits behind the queue cursor. Requires Convex 1.47 or later.
+
+Queued work from older versions migrates automatically, preserving scheduled
+start times. Large queues may briefly delay new work during migration. Before
+upgrading from a version earlier than 0.4.15, deploy 0.4.15 and run
+`npx convex run --component workpool migrations:backfillPendingStartIds`,
+replacing `workpool` with each component instance's name.
+
+Downgrading after this upgrade is unsupported because the queue format changes.

package.json

@@ -71,7 +71,7 @@
     }
   },
   "peerDependencies": {
-    "convex": "^1.43.0",
+    "convex": "^1.47.0",
     "convex-helpers": "^0.1.94"
   },
   "devDependencies": {

src/component/README.md

@@ -1,22 +1,13 @@
 # Workpool: implementation notes and high-level architecture
 
-Concepts:
-
-- `segment`: A slice of time to process work. All work is bucketed into one.
-  This enables us to batch work and avoid database conflicts.
-- `generation`: A monotonically increasing counter owned by batch-worker to
-  ensure only one loop chain runs for a worker at a time. The old workpool
-  generation field is deprecated and only accepted for pre-migration docs.
-- "Retention" is used to refer to situations where a query might have to read
-  over a lot of "tombstones" - deleted data that hasn't been vacuumed from the
-  underlying database yet. If there are frequent deletions, scanning across them
-  can delay a query. Because of our delete-heavy queuing strategy, we have to be
-  careful. Strategies are below.
-- Cursors: A pointer to the last processed place in a table. In our case, they
-  might allow data to be written before them if out-of-order writes happen, so
-  we need to account for finding those "missed" writes on some granularity. We
-  choose to wait until there isn't any immediate work to do before those scans.
-  They help avoid retention issues.
+- `segment`: a nanosecond timestamp. Immediate work uses its enqueue commit
+  timestamp; scheduled work uses its start time.
+- Cursors: inclusive positions in the pending queues that skip deleted rows. The
+  incoming cursor never passes the snapshot. `segmentCursors.sweep` walks the
+  `scanTs` index to recover near-term scheduled enqueues that commit behind it.
+- `generation`: batch-worker's counter that permits only one active loop chain.
+- Tombstones: deleted rows that remain in storage until vacuumed. Advancing
+  cursors avoids repeatedly reading them.
 
 ## Data state machine
 
@@ -52,14 +43,12 @@ a 10-second idle cooldown instead of the normal 2 seconds. This keeps
 batch-worker's status `running` for longer at full throttle, so most enqueue
 pings are no-ops rather than racing an idle transition.
 
-## Retention optimization strategy
+## Storage and reads
 
-- Producers (Client, Worker, Recovery) write to a future "segment".
-- Consumers (`run`) read the current segment.
-  - On conflicts, producers will write to progressively higher segments, while
-    the main loop will continue to read the segment originally called with. This
-    means conflicts are less likely on each retry.
-- Patch singletons to avoid tombstones.
-- Use segements & cursors to bound reads to latest data.
-  - Do scans outside of the critical path (during load).
-- Do point reads otherwise.
+- Store one queue document per work item and delete it when the work starts or
+  cancels.
+- Read queues through bounded index ranges and point-read work documents.
+- Keep the incoming cursor at or below the transaction snapshot so later commits
+  remain reachable. Convert legacy buckets before resetting the cursors.
+- Limit the sweep's boundary read to entries behind the incoming cursor, so
+  large batches of future work are not repeatedly scanned.

src/component/complete.test.ts

@@ -79,8 +79,11 @@ describe("complete", () => {
         },
       );
       await ctx.db.insert("internalState", {
-        generation: 0n,
-        segmentCursors: { incoming: 0n, completion: 0n, cancelation: 0n },
+        segmentCursors: {
+          incoming: 0n,
+          completion: 0n,
+          cancelation: 0n,
+        },
         lastRecovery: 0n,
         report: {
           completed: 0,

src/component/complete.ts

@@ -5,12 +5,7 @@ import { internal } from "./_generated/api.js";
 import { internalMutation, type MutationCtx } from "./_generated/server.js";
 import { kickMainLoop } from "./kick.js";
 import { createLogger } from "./logging.js";
-import {
-  getCurrentSegment,
-  type OnCompleteArgs,
-  type RunResult,
-  vResult,
-} from "./shared.js";
+import { type OnCompleteArgs, type RunResult, vResult } from "./shared.js";
 import { recordCompleted } from "./stats.js";
 import { assert } from "convex-helpers";
 
@@ -214,12 +209,11 @@ export async function completeHandler(
         ? "retry"
         : "complete",
     );
-    const segment = getCurrentSegment();
     await Promise.all(
       pendingCompletions.map((completion) =>
         ctx.db.insert("pendingCompletion", {
           ...completion,
-          segment,
+          segment: ctx.db.vars.commitTs,
         }),
       ),
     );

src/component/danger.test.ts

@@ -0,0 +1,114 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+import type { TransactionLimits } from "convex/server";
+import { internal } from "./_generated/api.js";
+import type { Id } from "./_generated/dataModel.js";
+import { INITIAL_STATE } from "./loop.js";
+import { setupTest } from "./setup.test.js";
+import { toSegment, toTimestamp } from "./shared.js";
+
+describe("danger cleanup", () => {
+  let t: ReturnType<typeof setupTest>;
+  beforeEach(() => {
+    vi.useFakeTimers();
+    t = setupTest();
+  });
+  afterEach(() => vi.useRealTimers());
+
+  async function enqueueWork(count: number, payloadBytes = 10) {
+    return t.run(async (ctx) => {
+      const workIds: Id<"work">[] = [];
+      for (let i = 0; i < count; i++) {
+        const payloadId = await ctx.db.insert("payload", {
+          args: { data: "x".repeat(payloadBytes) },
+        });
+        const workId = await ctx.db.insert("work", {
+          fnType: "action",
+          fnHandle: "test",
+          fnName: "test",
+          attempts: 0,
+          payloadId,
+          payloadSize: payloadBytes,
+        });
+        workIds.push(workId);
+        const pendingStartId = await ctx.db.insert("pendingStart", {
+          workId,
+          segment: toTimestamp(Date.now()),
+        });
+        await ctx.db.patch("work", workId, { pendingStartId });
+      }
+      return workIds;
+    });
+  }
+
+  it.each([
+    [250, 10, { documentsRead: 100 }],
+    [250, 10, { documentsWritten: 100 }],
+    [8, 800_000, { bytesRead: 4_000_000 }],
+  ])(
+    "resumes cleanup when a transaction budget is reached (%i works)",
+    async (count, payloadBytes, transactionLimits) => {
+      const ids = await enqueueWork(count, payloadBytes);
+      const before = Date.now() + 1;
+      vi.advanceTimersByTime(10);
+      const newerIds = await enqueueWork(1);
+
+      await t.run((ctx) =>
+        ctx.runMutation(
+          internal.danger.clearPending,
+          { before },
+          {
+            transactionLimits: transactionLimits as TransactionLimits,
+          },
+        ),
+      );
+      await t.run(async (ctx) => {
+        const remaining = (await ctx.db.query("pendingStart").collect())
+          .map((p) => p.workId)
+          .filter((id) => ids.includes(id));
+        expect(remaining.length).toBeGreaterThan(0);
+        expect(remaining.length).toBeLessThan(count);
+        for (const id of remaining)
+          expect(await ctx.db.get("work", id)).not.toBeNull();
+      });
+      await t.finishAllScheduledFunctions(vi.runAllTimers);
+      await t.run(async (ctx) => {
+        expect(
+          (await ctx.db.query("work").collect()).map((w) => w._id),
+        ).toEqual(newerIds);
+        expect((await ctx.db.query("pendingStart").unique())!.workId).toBe(
+          newerIds[0],
+        );
+        expect(await ctx.db.query("payload").collect()).toHaveLength(1);
+      });
+    },
+  );
+
+  it("removes legacy queue entries even below a persisted cursor", async () => {
+    await t.run(async (ctx) => {
+      const workId = await ctx.db.insert("work", {
+        fnType: "action",
+        fnHandle: "test",
+        fnName: "test",
+        attempts: 0,
+      });
+      await ctx.db.insert("pendingStart", {
+        workId,
+        segment: toSegment(Date.now()),
+      });
+      await ctx.db.insert("internalState", {
+        ...INITIAL_STATE,
+        segmentCursors: {
+          incoming: toSegment(Date.now() + 60_000),
+          completion: 0n,
+          cancelation: 0n,
+        },
+      });
+    });
+    await t.mutation(internal.danger.clearOldWork, { before: Date.now() + 1 });
+    await t.finishAllScheduledFunctions(vi.runAllTimers);
+    await t.run(async (ctx) => {
+      expect(await ctx.db.query("work").collect()).toHaveLength(0);
+      expect(await ctx.db.query("pendingStart").collect()).toHaveLength(0);
+    });
+  });
+});

src/component/danger.ts

@@ -1,10 +1,10 @@
-import { v, getConvexSize } from "convex/values";
+import { v } from "convex/values";
+import { kickMainLoop } from "./kick.js";
 import { internal } from "./_generated/api.js";
-import { internalMutation } from "./_generated/server.js";
+import { internalMutation, type MutationCtx } from "./_generated/server.js";
+import { findPendingStart } from "./pendingStart.js";
 
 const DEFAULT_OLDER_THAN = 1000 * 60 * 60 * 24;
-const MAX_ROWS_READ = 100;
-const MAX_BYTES_READ = 4_000_000;
 
 export const clearPending = internalMutation({
   args: {
@@ -14,41 +14,25 @@ export const clearPending = internalMutation({
   handler: async (ctx, args) => {
     const time =
       args.before ?? Date.now() - (args.olderThan ?? DEFAULT_OLDER_THAN);
-    let i = 0,
-      totalBytes = 0,
-      hasMore = false,
-      nextTime;
     console.log("Clearing pending before", new Date(time).toUTCString());
     for await (const entry of ctx.db
       .query("pendingStart")
       .withIndex("by_creation_time", (q) => q.lte("_creationTime", time))
       .order("desc")) {
-      i++;
       const work = await ctx.db.get("work", entry.workId);
-      totalBytes +=
-        getConvexSize(entry) + getConvexSize(work) + (work?.payloadSize ?? 0);
-      if (i > MAX_ROWS_READ || totalBytes > MAX_BYTES_READ) {
-        hasMore = true;
-        nextTime = entry._creationTime;
-        console.log(`Continuing after ${i} entries, ${totalBytes} bytes`);
-        break;
-      }
-      await ctx.db.delete("pendingStart", entry._id);
       if (work) {
-        // Clean up any large data stored separately
-        if (work.payloadId) {
-          await ctx.db.delete("payload", work.payloadId);
-        }
+        if (work.payloadId) await ctx.db.delete("payload", work.payloadId);
         await ctx.db.delete("work", work._id);
       }
+      await ctx.db.delete("pendingStart", entry._id);
+      if (await usedHalfTransactionBudget(ctx)) {
+        await ctx.scheduler.runAfter(0, internal.danger.clearPending, {
+          before: entry._creationTime,
+        });
+        return;
+      }
     }
-    if (hasMore) {
-      await ctx.scheduler.runAfter(0, internal.danger.clearPending, {
-        before: nextTime,
-      });
-    } else {
-      console.log(`Done clearing pending entries. ${i} in the last batch.`);
-    }
+    console.log("Done clearing pending entries.");
   },
 });
 
@@ -60,20 +44,16 @@ export const clearOldWork = internalMutation({
   handler: async (ctx, args) => {
     const time =
       args.before ?? Date.now() - (args.olderThan ?? DEFAULT_OLDER_THAN);
-    let i = 0,
-      totalBytes = 0,
-      hasMore = false,
-      nextTime;
     console.log("Clearing old work before", new Date(time).toUTCString());
     for await (const entry of ctx.db
       .query("work")
       .withIndex("by_creation_time", (q) => q.lte("_creationTime", time))
       .order("desc")) {
-      i++;
-      const pendingStart = await ctx.db
-        .query("pendingStart")
-        .withIndex("workId", (q) => q.eq("workId", entry._id))
-        .unique();
+      if (entry.pendingStartId === undefined) {
+        // The upgrade pass removes orphaned legacy queue entries.
+        await kickMainLoop(ctx, "kick");
+      }
+      const pendingStart = await findPendingStart(ctx, entry);
       const pendingCompletion = await ctx.db
         .query("pendingCompletion")
         .withIndex("workId", (q) => q.eq("workId", entry._id))
@@ -82,18 +62,6 @@ export const clearOldWork = internalMutation({
         .query("pendingCancelation")
         .withIndex("workId", (q) => q.eq("workId", entry._id))
         .unique();
-      totalBytes +=
-        getConvexSize(entry) +
-        getConvexSize(pendingStart) +
-        getConvexSize(pendingCompletion) +
-        getConvexSize(pendingCancelation) +
-        (entry.payloadSize ?? 0);
-      if (i > MAX_ROWS_READ || totalBytes > MAX_BYTES_READ) {
-        hasMore = true;
-        nextTime = entry._creationTime;
-        console.log(`Continuing after ${i} entries, ${totalBytes} bytes`);
-        break;
-      }
       if (pendingStart) {
         await ctx.db.delete("pendingStart", pendingStart._id);
       }
@@ -118,13 +86,21 @@ export const clearOldWork = internalMutation({
           .join(", ")})`,
       );
       await ctx.db.delete("work", entry._id);
+      if (await usedHalfTransactionBudget(ctx)) {
+        await ctx.scheduler.runAfter(0, internal.danger.clearOldWork, {
+          before: entry._creationTime,
+        });
+        return;
+      }
     }
-    if (hasMore) {
-      await ctx.scheduler.runAfter(0, internal.danger.clearOldWork, {
-        before: nextTime,
-      });
-    } else {
-      console.log(`Done clearing old work. ${i} in the last batch.`);
-    }
+    console.log("Done clearing old work.");
   },
 });
+
+/** Leave headroom to finish the current document and schedule another batch. */
+async function usedHalfTransactionBudget(ctx: MutationCtx) {
+  const metrics = await ctx.meta.getTransactionMetrics();
+  return Object.values(metrics).some(
+    ({ used, remaining }) => used >= remaining,
+  );
+}

src/component/kick.test.ts

@@ -33,8 +33,11 @@ describe("kickMainLoop", () => {
         },
       );
       await ctx.db.insert("internalState", {
-        generation: 0n,
-        segmentCursors: { incoming: 0n, completion: 0n, cancelation: 0n },
+        segmentCursors: {
+          incoming: 0n,
+          completion: 0n,
+          cancelation: 0n,
+        },
         lastRecovery: 0n,
         report: {
           completed: 0,