Order pending work by commit and snapshot timestamps

CHANGELOG.md

@@ -1,5 +1,14 @@
 # Changelog
 
+## 0.5.0 (Unreleased)
+
+- Orders queues by commit timestamp. Requires 1.47 or later.
+- Recovers near-term scheduled work that commits behind the queue cursor.
+- Migrates queued work from older versions automatically, preserving scheduled
+  start times. Large queues may briefly delay new work during migration.
+- Downgrading after this upgrade is unsupported because the queue format
+  changes.
+
 ## 0.4.15
 
 - Upgrades @convex-dev/batch-worker to 0.3. Requires Convex 1.43 or later.

package.json

@@ -72,7 +72,7 @@
     }
   },
   "peerDependencies": {
-    "convex": "^1.43.0",
+    "convex": "^1.47.0",
     "convex-helpers": "^0.1.94"
   },
   "devDependencies": {
@@ -88,7 +88,7 @@
     "@vitejs/plugin-react": "5.1.3",
     "@vitest/coverage-v8": "5.0.0",
     "chokidar-cli": "3.0.0",
-    "convex": "1.46.0",
+    "convex": "1.47.0",
     "convex-helpers": "0.1.128",
     "convex-test": "0.0.64",
     "eslint": "10.0.2",

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);
+    });
+  });
+});