From c224eb65ee0d3c6ee4ad998202e223005b606945 Mon Sep 17 00:00:00 2001
From: Ian Macartney <366683+ianmacartney@users.noreply.github.com>
Date: Fri, 14 Aug 2026 15:11:10 -0700
Subject: [PATCH 1/9] key scheduled work by its start time and sweep commit
 order for stragglers
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

An entry scheduled for later is now keyed at its start time (rounded up to
the next whole millisecond), so it sorts above the loop's read bound until
due — no re-keying, and a bulk enqueue of scheduled work never sits in front
of ready work. Since such a key is chosen before the enqueue commits, the
entry can land behind the loop's cursor when the commit takes longer than
the delay; every wall-clock-keyed entry therefore records its commit stamp
in scheduledAt, and the loop sweeps that index in commit order — which
nothing can land behind — inspecting each entry once and directly starting
any entry the cursor already passed.

The incoming cursor's bounds are now explicit values instead of a stateful
guard: the work query computes a ceiling (the highest commit timestamp its
snapshot observed, seeded by the previous run's own stamp recorded in
internalState) and the worker mutation raises every segment it writes to at
least the cursor and advances to at most the lowest segment it wrote and the
ceiling. The ceiling is what keeps a cursor that lands on a wall-clock key
from passing a racing commit; nothing anywhere relates wall clocks to the
commit-timestamp clock.

The runAt and hasRunAt fields this branch previously introduced are no
longer written; the loop clears them from entries earlier revisions wrote as
it re-keys them. The five-minute safe-future threshold is gone: the sweep
covers any delay uniformly.
---
 CHANGELOG.md                |  25 ++-
 src/component/lib.ts        |  23 +--
 src/component/loop.test.ts  | 200 ++++++++++++++-----
 src/component/loop.ts       | 388 +++++++++++++++++++++---------------
 src/component/schema.ts     |  47 +++--
 src/component/shared.ts     |  26 +--
 src/component/stats.test.ts |   4 +-
 src/component/stats.ts      |   6 +-
 8 files changed, 447 insertions(+), 272 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 5dfe49f0..88cdba39 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -11,19 +11,24 @@
   pools.
 - The `segment` fields keep their name, but now hold nanoseconds rather than
   100ms buckets. Work scheduled to start later stores its start time there
-  directly, which sorts after everything already committed, so ready and
-  scheduled work share an ordering.
-- Work scheduled less than five minutes out can't safely store its start time at
-  enqueue, so it's held at its commit position until the loop moves it. A
-  `hasRunAt` marker puts those entries in their own lane of the `pendingStart`
-  index, moved along at a full batch per iteration in parallel with starting
-  ready work — so a bulk enqueue of near-future work never delays work that's
-  ready now.
+  directly, so ready and scheduled work share an ordering, scheduled work sorts
+  above the loop's read bound until it's due, and a bulk enqueue of scheduled
+  work never sits in front of ready work.
+- A scheduled start time can commit _behind_ the loop's cursor (when the enqueue
+  takes longer to commit than the delay), where the ordered scan would never see
+  it. Every entry keyed by a wall-clock time also records its commit timestamp
+  in a `scheduledAt` field; the loop sweeps that index in commit order — which
+  nothing can land behind — inspects each entry exactly once, and directly
+  starts the rare entry the cursor passed over. No entry is ever rewritten.
+- The loop's cursor never advances past the newest commit timestamp it has
+  observed, so a scheduled entry starting at its wall-clock time can't push the
+  cursor ahead of a racing enqueue's commit — the design makes no assumptions
+  about how wall clocks relate to the commit timestamp clock.
 - Upgrading in place is safe, including for work that hasn't come due yet. A
   `pendingStart` an older version wrote holds a 100ms bucket — eight orders of
   magnitude below a nanosecond timestamp — so the loop recognizes it, reads the
-  bucket back as the time the work should start, and either starts it or
-  rewrites it as a timestamp and leaves it alone until then. This is a one-time
+  bucket back as the time the work should start, and either starts it or re-keys
+  it as a timestamp and leaves it alone until then. This is a one-time
   re-ordering of every queued entry, 64 per loop iteration, so right after the
   push a large backlog of scheduled work adds processing overhead and can
   briefly delay work that's ready now. Queued completions and cancelations sort
diff --git a/src/component/lib.ts b/src/component/lib.ts
index a5afd208..23b08fd6 100644
--- a/src/component/lib.ts
+++ b/src/component/lib.ts
@@ -21,9 +21,8 @@ import {
   fnType,
   vOnCompleteFnContext,
   retryBehavior,
-  SAFE_FUTURE_MS,
   status as statusValidator,
-  toTimestamp,
+  dueTimestamp,
 } from "./shared.js";
 import { recordEnqueued } from "./stats.js";
 import { getOrUpdateGlobals } from "./config.js";
@@ -112,20 +111,18 @@ export async function enqueueHandler(
   // Store the work item
   const workId = await ctx.db.insert("work", workItem);
 
-  // Ready now: order it by this transaction's commit timestamp. Far enough out
-  // that no commit latency could put it behind the loop's cursor: order it by
-  // its start time directly. In between: order it by the commit timestamp so
-  // it can't be lost, and let the loop move it forward once it sees the `runAt`
-  // (see `promoteScheduled`) — one extra write, only for near-future work.
-  // `hasRunAt` puts that last case in its own index lane, so entries waiting
-  // to be moved never sit in front of ready work.
+  // Ready now: order it by this transaction's commit timestamp, which can't
+  // land behind the loop's cursor. Scheduled for later: order it by its start
+  // time, so it stays invisible to the loop until it's due. A scheduled entry
+  // could commit *behind* the cursor (if this enqueue takes longer to commit
+  // than the delay), so it also records its commit stamp in `scheduledAt`;
+  // the loop sweeps that index in commit order and starts anything the cursor
+  // passed over.
   const delayMs = runAt - Date.now();
   await ctx.db.insert("pendingStart", {
     workId,
-    segment:
-      delayMs > SAFE_FUTURE_MS ? toTimestamp(runAt) : ctx.db.vars.commitTs,
-    ...(delayMs > 0 ? { runAt } : {}),
-    ...(delayMs > 0 && delayMs <= SAFE_FUTURE_MS ? { hasRunAt: true } : {}),
+    segment: delayMs > 0 ? dueTimestamp(runAt) : ctx.db.vars.commitTs,
+    ...(delayMs > 0 ? { scheduledAt: ctx.db.vars.commitTs } : {}),
   });
   recordEnqueued(console, { workId, fnName: workArgs.fnName, runAt });
   return workId;
diff --git a/src/component/loop.test.ts b/src/component/loop.test.ts
index cd3bdffd..565645ed 100644
--- a/src/component/loop.test.ts
+++ b/src/component/loop.test.ts
@@ -17,7 +17,7 @@ import { setupTest } from "./setup.test.js";
 import {
   DEFAULT_MAX_PARALLELISM,
   fromSegment,
-  SAFE_FUTURE_MS,
+  fromTimestamp,
   toSegment,
   toTimestamp,
   WORKER_NAME,
@@ -448,6 +448,7 @@ describe("loop", () => {
       expect(o.running).toHaveLength(0);
       expect(o.pendingStart[0].segment).toBe(toTimestamp(startsAt));
 
+      await runLoop(); // the sweep verifies the re-keyed entry, once
       const idle = await runLoop();
       assert(idle.kind === "idle");
       expect(idle.timeoutMs).toBe(startsAt - Date.now());
@@ -550,6 +551,7 @@ describe("loop", () => {
       await runLoop();
       await simulateCompletion(workId, { kind: "failed", error: "boom" }, 0);
       await runLoop();
+      await runLoop(); // the sweep verifies the retry's entry, once
 
       // Still held back by the backoff: nothing to start yet.
       expect((await runLoop()).kind).toBe("idle");
@@ -675,9 +677,10 @@ describe("loop", () => {
 
     it("idles with a timeoutMs when only far-future work remains", async () => {
       await initialize();
-      // Beyond SAFE_FUTURE_MS, so the enqueue ordered it by its start time and
-      // the loop never has to look at it.
-      await enqueueWork({}, { runAt: Date.now() + 2 * SAFE_FUTURE_MS });
+      // Ordered by its start time, so beyond the sweep's one-time
+      // verification the loop never has to look at it.
+      await enqueueWork({}, { runAt: Date.now() + 10 * MINUTE });
+      await runLoop(); // the sweep verifies the entry, once
 
       const result = await t.query(internal.loop.getBatch, {
         name: WORKER_NAME,
@@ -686,33 +689,32 @@ describe("loop", () => {
       expect(result.timeoutMs).toBeGreaterThan(0);
     });
 
-    it("moves near-future work forward once, then idles until it's due", async () => {
+    it("sweeps near-future work once, then idles until it's due", async () => {
       await initialize();
       const runAt = Date.now() + 100 * SECOND;
-      // Inside SAFE_FUTURE_MS, so the enqueue had to order it by its commit
-      // timestamp, in the scheduled lane; the loop sees it there exactly once, to
-      // move it.
+      // Keyed at its start time directly; the sweep just verifies its enqueue
+      // committed in order, once, and never rewrites it.
       const workId = await enqueueWork({}, { runAt });
 
       const first = await runLoop();
       assert(first.kind === "work");
       expect(first.batch.starts).toHaveLength(0);
-      expect(first.batch.scheduled?.map((s) => s.workId)).toEqual([workId]);
+      expect(first.batch.sweep).toHaveLength(1);
+      expect(first.batch.sweep![0].starts).toHaveLength(0);
 
       let o = await observe();
-      expect(o.running).toHaveLength(0); // moved, not started
+      expect(o.running).toHaveLength(0);
       expect(o.pendingStart[0].segment).toBe(toTimestamp(runAt));
-      // Moved out of the scheduled lane, and that lane's cursor moved past where it
-      // used to sit, so it won't come back.
-      expect(o.pendingStart[0].hasRunAt).toBeUndefined();
-      expect(o.segmentCursors!.scheduled).toBeGreaterThan(0n);
-      expect(o.segmentCursors!.scheduled).toBeLessThan(toTimestamp(runAt));
+      // The sweep cursor covers its commit stamp now, so it's never re-read.
+      expect(o.segmentCursors!.scheduled).toBe(
+        o.pendingStart[0].scheduledAt as bigint,
+      );
 
       const second = await runLoop();
       assert(second.kind === "idle");
       expect(second.timeoutMs).toBe(100 * SECOND);
 
-      // It starts when it comes due, without another move.
+      // It starts when it comes due, untouched in the meantime.
       vi.advanceTimersByTime(100 * SECOND);
       await runLoop();
       o = await observe();
@@ -726,7 +728,8 @@ describe("loop", () => {
       await runLoop(); // start it: running=1, lastRecovery=now
 
       // Future work well past the ~1min recovery period.
-      await enqueueWork({}, { runAt: Date.now() + 2 * SAFE_FUTURE_MS });
+      await enqueueWork({}, { runAt: Date.now() + 10 * MINUTE });
+      await runLoop(); // the sweep verifies the entry, once
 
       const result = await t.query(internal.loop.getBatch, {
         name: WORKER_NAME,
@@ -741,15 +744,13 @@ describe("loop", () => {
   });
 
   // ────────────────────────────────────────────────────────────────────
-  // The scheduled lane: near-future work waiting to be moved to its start time
+  // Scheduled work: keyed at its start time, invisible until due
   // ────────────────────────────────────────────────────────────────────
 
-  describe("scheduled lane", () => {
-    it("never lets held work sit in front of ready work", async () => {
-      // A bulk enqueue of near-future work lands ahead of anything enqueued
-      // after it, and only MAIN_BATCH_SIZE entries move per iteration. In its
-      // own lane it drains in parallel; in the incoming lane it would starve the
-      // ready item below for ceil(70/64) iterations.
+  describe("scheduled work", () => {
+    it("never lets a scheduled backlog sit in front of ready work", async () => {
+      // A bulk enqueue of near-future work sorts above the eligibility bound,
+      // so it costs the ready item below nothing — no re-keying, no waiting.
       await initialize({ maxParallelism: 3 });
       const runAt = Date.now() + 100 * SECOND;
       for (let i = 0; i < 70; i++) await enqueueWork({}, { runAt });
@@ -759,37 +760,41 @@ describe("loop", () => {
 
       const o = await observe();
       expect(o.running.map((r) => r.workId)).toEqual([readyId]);
-      const moved = o.pendingStart.filter((p) => p.hasRunAt === undefined);
-      const held = o.pendingStart.filter((p) => p.hasRunAt === true);
-      expect(moved).toHaveLength(64); // one full batch moved alongside the start
-      expect(held).toHaveLength(6);
-      for (const p of moved) expect(p.segment).toBe(toTimestamp(runAt));
+      expect(o.pendingStart).toHaveLength(70);
+      for (const p of o.pendingStart) {
+        expect(p.segment).toBe(toTimestamp(runAt));
+        expect(p.scheduledAt).toBeDefined();
+      }
+      // The sweep verified all 70 in one pass; nothing left to do before the
+      // sooner of their due time and the next stuck-job recovery check.
+      const idle = await runLoop();
+      assert(idle.kind === "idle");
+      expect(idle.timeoutMs).toBeLessThanOrEqual(100 * SECOND);
     });
 
-    it("gives due held work start slots before fetching more ready work", async () => {
+    it("starts due scheduled work and ready work in eligibility order", async () => {
       await initialize({ maxParallelism: 1 });
-      const heldId = await enqueueWork({}, { runAt: Date.now() + SECOND });
+      const scheduledId = await enqueueWork({}, { runAt: Date.now() + SECOND });
       const readyId = await enqueueWork();
       vi.advanceTimersByTime(SECOND);
 
-      // The due held entry claims the only slot, so no ready work is fetched.
+      // The ready item became eligible first: its commit precedes the runAt.
       await runLoop();
       let o = await observe();
-      expect(o.running.map((r) => r.workId)).toEqual([heldId]);
-      expect(o.pendingStart.map((p) => p.workId)).toEqual([readyId]);
+      expect(o.running.map((r) => r.workId)).toEqual([readyId]);
 
       await simulateCompletion(
-        heldId,
+        readyId,
         { kind: "success", returnValue: null },
         0,
       );
       await runLoop();
       o = await observe();
       expect(o.pendingStart).toHaveLength(0);
-      expect(o.running.map((r) => r.workId)).toEqual([readyId]);
+      expect(o.running.map((r) => r.workId)).toEqual([scheduledId]);
     });
 
-    it("starts held work directly once it's due, without moving it", async () => {
+    it("starts scheduled work once it's due, without ever rewriting it", async () => {
       await initialize();
       const workId = await enqueueWork({}, { runAt: Date.now() + SECOND });
 
@@ -801,19 +806,95 @@ describe("loop", () => {
       expect(o.running.map((r) => r.workId)).toEqual([workId]);
     });
 
-    it("leaves due held work for the iteration that has capacity for it", async () => {
+    it("holds due work while saturated, and starts it as capacity frees", async () => {
       await initialize({ maxParallelism: 1 });
       const firstId = await enqueueWork();
       await runLoop(); // fills the only slot
-      const heldId = await enqueueWork({}, { runAt: Date.now() + SECOND });
+      const dueId = await enqueueWork({}, { runAt: Date.now() + SECOND });
       vi.advanceTimersByTime(2 * SECOND);
 
-      // Saturated: nothing can start, so the scheduled lane isn't even read.
+      await runLoop(); // the sweep verifies the new entry, once
       const idle = await runLoop();
-      expect(idle.kind).toBe("idle");
+      expect(idle.kind).toBe("idle"); // saturated: nothing can start
       expect((await observe()).pendingStart).toHaveLength(1);
 
-      // A completion frees the slot; the same iteration starts the held work.
+      // A completion frees the slot; the same iteration starts the work.
+      await simulateCompletion(
+        firstId,
+        { kind: "success", returnValue: null },
+        0,
+      );
+      await runLoop();
+      const o = await observe();
+      expect(o.pendingStart).toHaveLength(0);
+      expect(o.running.map((r) => r.workId)).toEqual([dueId]);
+    });
+  });
+
+  // ────────────────────────────────────────────────────────────────────
+  // The out-of-order sweep: entries whose enqueue lost the race
+  // ────────────────────────────────────────────────────────────────────
+
+  describe("out-of-order sweep", () => {
+    /**
+     * A near-future enqueue whose commit took longer than its delay: the
+     * entry appears with a `segment` the cursor already passed, where the
+     * segment scan will never see it. Its `scheduledAt` commit stamp is how
+     * the sweep finds it.
+     */
+    async function enqueueBehindCursor(cursor: bigint): Promise<Id<"work">> {
+      return t.run(async (ctx) => {
+        const workId = await ctx.db.insert("work", {
+          fnType: "action",
+          fnHandle: "test_handle",
+          fnName: "test_handle",
+          fnArgs: {},
+          attempts: 0,
+        });
+        await ctx.db.insert("pendingStart", {
+          workId,
+          segment: cursor - 1n,
+          runAt: fromTimestamp(cursor - 1n),
+          scheduledAt: ctx.db.vars.commitTs,
+        });
+        return workId;
+      });
+    }
+
+    it("starts an entry whose enqueue lost the race with the cursor", async () => {
+      await initialize({ maxParallelism: 2 });
+      // Run something so the incoming cursor is somewhere real.
+      const readyId = await enqueueWork();
+      await runLoop();
+      const cursor = (await observe()).segmentCursors!.incoming;
+      expect(cursor).toBeGreaterThan(0n);
+
+      const lostId = await enqueueBehindCursor(cursor);
+
+      await runLoop();
+      const o = await observe();
+      expect(o.running.map((r) => r.workId)).toContain(lostId);
+      expect(o.pendingStart).toHaveLength(0);
+      // Its commit stamp is covered, so the sweep never re-reads it.
+      expect(o.segmentCursors!.scheduled).toBeGreaterThan(0n);
+      expect(readyId).toBeDefined();
+    });
+
+    it("waits for capacity to retire an out-of-order entry, without losing it", async () => {
+      await initialize({ maxParallelism: 1 });
+      const firstId = await enqueueWork();
+      await runLoop(); // fills the only slot
+      const cursor = (await observe()).segmentCursors!.incoming;
+      const lostId = await enqueueBehindCursor(cursor);
+
+      // Saturated: the entry can't start, so the sweep leaves its stamp
+      // uncovered rather than passing it.
+      const idle = await runLoop();
+      expect(idle.kind).toBe("idle");
+      expect((await observe()).pendingStart.map((p) => p.workId)).toEqual([
+        lostId,
+      ]);
+
       await simulateCompletion(
         firstId,
         { kind: "success", returnValue: null },
@@ -822,7 +903,23 @@ describe("loop", () => {
       await runLoop();
       const o = await observe();
       expect(o.pendingStart).toHaveLength(0);
-      expect(o.running.map((r) => r.workId)).toEqual([heldId]);
+      expect(o.running.map((r) => r.workId)).toEqual([lostId]);
+    });
+
+    it("caps the cursor at observed commit stamps when starting wall-keyed work", async () => {
+      await initialize();
+      const runAt = Date.now() + SECOND;
+      const workId = await enqueueWork({}, { runAt });
+      await runLoop(); // sweep verifies the entry
+      vi.advanceTimersByTime(SECOND);
+      await runLoop(); // starts it, keyed at its start time
+
+      const o = await observe();
+      expect(o.running.map((r) => r.workId)).toEqual([workId]);
+      // The cursor stops at the freshest commit stamp this run had seen, not
+      // at the wall-clock key: a commit racing this iteration could land
+      // below that key, and nothing relates the two clocks.
+      expect(o.segmentCursors!.incoming).toBeLessThan(toTimestamp(runAt));
     });
   });
 
@@ -863,20 +960,21 @@ describe("loop", () => {
       expect(o.running.map((r) => r.workId)).toContain(workId);
     });
 
-    it("keeps a sub-millisecond runAt readable when it moves out of the scheduled lane", async () => {
+    it("keys a sub-millisecond runAt to the next whole millisecond", async () => {
       await initialize();
-      // Truncating this `runAt` to whole milliseconds lands it at or below the
-      // cursor, which by now holds a commit timestamp from this millisecond.
-      const workId = await enqueueWork({}, { runAt: Date.now() + 0.5 });
+      // Truncated to whole milliseconds this `runAt` would be readable while
+      // `isDue` still says no; rounded up, it's invisible until the first
+      // millisecond the clock can call it due.
+      const runAt = Date.now() + 0.5;
+      const workId = await enqueueWork({}, { runAt });
       const readyId = await enqueueWork();
 
       await runLoop();
       const o = await observe();
       expect(o.running.map((r) => r.workId)).toEqual([readyId]);
-      const moved = o.pendingStart.find((p) => p.workId === workId);
-      assert(moved);
-      expect(moved.hasRunAt).toBeUndefined();
-      expect(moved.segment).toBeGreaterThanOrEqual(o.segmentCursors!.incoming);
+      const entry = o.pendingStart.find((p) => p.workId === workId);
+      assert(entry);
+      expect(entry.segment).toBe(toTimestamp(Math.ceil(runAt)));
 
       vi.advanceTimersByTime(1);
       await runLoop();
diff --git a/src/component/loop.ts b/src/component/loop.ts
index 7010f9f0..58f8e16f 100644
--- a/src/component/loop.ts
+++ b/src/component/loop.ts
@@ -29,7 +29,7 @@ import {
   SECOND,
   type RunResult,
   toSegment,
-  toTimestamp,
+  dueTimestamp,
   vResult,
 } from "./shared.js";
 import { generateReport, recordCompleted, recordStarted } from "./stats.js";
@@ -103,6 +103,18 @@ const vStart = v.object({
 });
 type Start = Infer<typeof vStart>;
 
+/**
+ * One commit stamp's worth of the out-of-order sweep, in commit order. The
+ * sweep cursor may advance to `scheduledAt` once every entry in `starts` has
+ * been retired; entries that shared the stamp but weren't behind the incoming
+ * cursor need nothing (the segment scan will reach them) and aren't carried.
+ */
+const vSweepStep = v.object({
+  scheduledAt: v.int64(),
+  starts: v.array(vStart),
+});
+type SweepStep = Infer<typeof vSweepStep>;
+
 /** The shape `getBatch` hands to `run`. */
 const batchFields = {
   // The segment at query time — what "now" was when the batch was built.
@@ -112,10 +124,16 @@ const batchFields = {
   completions: v.array(vCompletion),
   cancelations: v.array(vCancelation),
   starts: v.array(vStart),
-  // Entries from the `hasRunAt: true` lane: near-future work still ordered by
-  // its enqueue's commit timestamp, waiting to be moved to its start time.
-  // Optional so a batch built just before this field existed still validates.
+  // @deprecated Held-lane entries from a batch an unreleased revision built.
+  // Ignored: those rows stay put, and the segment scan re-reads them.
   scheduled: v.optional(v.array(vStart)),
+  // The out-of-order sweep's findings. Optional so a batch built just before
+  // this field existed still validates.
+  sweep: v.optional(v.array(vSweepStep)),
+  // The highest commit timestamp observed while building the batch — the
+  // frontier the incoming cursor may advance to, but not beyond. Optional for
+  // the same deploy-boundary reason; `run` falls back to its own last stamp.
+  cursorCeiling: v.optional(v.int64()),
 };
 type Batch = Infer<ReturnType<typeof v.object<typeof batchFields>>>;
 
@@ -146,24 +164,25 @@ export const getBatch = internalQuery({
     const isRecoveryIter =
       running.length > 0 && segment - lastRecovery >= RECOVERY_PERIOD_SEGMENTS;
 
-    const { starts, scheduled, cancelations, completions } = await queryPending(
-      ctx,
-      {
+    const { starts, sweep, cancelations, completions, cursorCeiling } =
+      await queryPending(ctx, {
         completionCursor: cursors.completion,
         cancelationCursor: cursors.cancelation,
         incomingCursor: cursors.incoming,
-        scheduledCursor: cursors.scheduled ?? 0n,
+        sweepCursor: cursors.scheduled ?? 0n,
+        lastCommitTs: (state?.lastCommitTs as bigint | undefined) ?? 0n,
         maxParallelism: globals.maxParallelism,
         runningCount: running.length,
         eligibleBefore,
-      },
-    );
+      });
 
+    // Every sweep step is progress — an entry to start, or a cursor advance —
+    // so its presence counts as work.
     const hasWork =
       completions.length > 0 ||
       cancelations.length > 0 ||
       starts.length > 0 ||
-      scheduled.length > 0 ||
+      sweep.length > 0 ||
       isRecoveryIter;
 
     if (hasWork) {
@@ -183,22 +202,20 @@ export const getBatch = internalQuery({
           segment: c.segment as bigint,
         })),
         starts,
-        scheduled,
+        sweep,
+        cursorCeiling,
       };
       return { kind: "work" as const, batch };
     }
 
     // Nothing to do now. Figure out when to wake up next: the sooner of the
     // earliest future-scheduled start and (if jobs are running) the next
-    // recovery scan. A ping still wakes us sooner. No need to check the
-    // `hasRunAt: true` lane: this iteration read it and it was empty. (The
-    // read is only skipped when the pool is saturated, and then jobs are
-    // running, so the recovery wait applies and completions ping us.)
+    // recovery scan. A ping still wakes us sooner. Work the sweep couldn't
+    // retire (a due entry with no capacity) is covered too: capacity implies
+    // jobs are running, so the recovery wait applies and completions ping us.
     const futureStart = await ctx.db
       .query("pendingStart")
-      .withIndex("hasRunAt_segment", (q) =>
-        q.eq("hasRunAt", undefined).gte("segment", eligibleBefore),
-      )
+      .withIndex("segment", (q) => q.gte("segment", eligibleBefore))
       .first();
     const waits: number[] = [];
     if (futureStart) {
@@ -237,17 +254,29 @@ export const run = internalMutation({
     const globals = await getGlobals(ctx);
     const console = createLogger(globals.logLevel);
     const segment = getCurrentSegment();
-    const incomingGuard = cursorGuard(state.segmentCursors.incoming);
-    const scheduledGuard = cursorGuard(state.segmentCursors.scheduled ?? 0n);
+    const sweep = batch.sweep ?? [];
+    // The bounds on this run's cursor movement, both plain values:
+    // - `floor`: the cursor as it stands. Every `segment` this run writes is
+    //   raised to at least the floor, so nothing lands where the loop already
+    //   read past; the cursor never moves backwards past it.
+    // - `ceiling`: the highest commit timestamp the batch's snapshot had
+    //   observed (computed alongside the batch in `queryPending`). The cursor
+    //   never advances beyond it: a wall-clock key can exceed every commit
+    //   stamp that exists, and a commit racing this run would land behind a
+    //   cursor set there. Nothing relates the two clocks; observed commit
+    //   stamps are the only safe frontier.
+    const floor = state.segmentCursors.incoming;
+    const ceiling =
+      batch.cursorCeiling ?? (state.lastCommitTs as bigint | undefined) ?? 0n;
 
     const compLabel = `[main] pendingCompletion(${batch.completions.length})`;
     console.time(compLabel);
-    const toCancel = await handleCompletions(
+    const { toCancel, lowestRetry } = await handleCompletions(
       ctx,
       state,
       batch.completions,
       console,
-      incomingGuard,
+      floor,
     );
     console.timeEnd(compLabel);
 
@@ -269,18 +298,20 @@ export const run = internalMutation({
     }
 
     // ── Start new work ──
-    // Entries whose `runAt` hasn't arrived were only visible because we
-    // couldn't safely write that time into `segment` at enqueue. We can here:
-    // this transaction also writes the cursors, so the guard can place the
-    // entry relative to them. Move them and they stop coming back.
+    // The segment scan's entries are all due (scheduled ones only become
+    // visible at their start time), except entries written by older versions,
+    // which the loop re-keys to their start time. Re-keying is safe here where
+    // it wasn't at enqueue: this transaction also writes the cursor, so the
+    // key can be placed relative to it. The sweep's entries are the opposite —
+    // committed *behind* the cursor, so the scan will never reach them; they
+    // start from here.
     const now = Date.now();
     const isDue = (s: Start) => s.runAt === undefined || s.runAt <= now;
-    const scheduled = batch.scheduled ?? [];
-    const all = [...batch.starts, ...scheduled];
+    const swept = sweep.flatMap((step) => step.starts);
+    const all = [...batch.starts, ...swept];
     const notYet = all.filter((s) => !isDue(s));
-    // Oldest first across both lanes: the incoming lane's `segment` is when an
-    // entry became eligible, and a due entry from the scheduled lane has been
-    // eligible since its `runAt`.
+    // Oldest first: `segment` is when an entry became eligible, and a swept
+    // entry has been eligible since its (past) start time.
     const eligible = all
       .filter(isDue)
       .sort(
@@ -288,10 +319,11 @@ export const run = internalMutation({
           (a.runAt ?? fromTimestamp(a.segment)) -
           (b.runAt ?? fromTimestamp(b.segment)),
       );
+    let lowestPromoted: bigint | undefined;
     if (notYet.length > 0) {
       const promoteLabel = `[main] promote(${notYet.length})`;
       console.time(promoteLabel);
-      await promoteScheduled(ctx, notYet, incomingGuard);
+      lowestPromoted = await promoteScheduled(ctx, notYet, floor);
       console.timeEnd(promoteLabel);
     }
 
@@ -335,25 +367,38 @@ export const run = internalMutation({
     if (batch.cancelations.length > 0) {
       state.segmentCursors.cancelation = batch.cancelations.at(-1)!.segment;
     }
-    // Capacity can cut the starts short, so only advance each lane's cursor
-    // over the leading run we finished with — started or moved forward.
-    // Stopping at the first entry we left alone is what keeps it from being
-    // skipped. The guards keep the advance safe against everything placed
-    // this transaction (only the incoming lane receives placements — promoting
-    // an entry is what takes it out of the scheduled lane).
+    // Capacity can cut the starts short, so only advance each cursor over the
+    // leading run we finished with — started or re-keyed. Stopping at the
+    // first entry we left alone is what keeps it from being skipped. The
+    // incoming cursor additionally never passes the ceiling, nor any segment
+    // this run wrote — a retry or re-keyed entry lands at or above the floor,
+    // but the cursor's final position isn't known when those writes happen.
+    // Equality is fine throughout: the index is read with `gte`.
     const handled = new Set([...pending, ...notYet].map((s) => s._id));
-    let incoming = state.segmentCursors.incoming;
+    let incoming = floor;
     for (const start of batch.starts) {
       if (!handled.has(start._id)) break;
       incoming = start.segment;
     }
-    state.segmentCursors.incoming = incomingGuard.advance(incoming);
-    let scheduledCursor = state.segmentCursors.scheduled ?? 0n;
-    for (const entry of scheduled) {
-      if (!handled.has(entry._id)) break;
-      scheduledCursor = entry.segment;
+    for (const cap of [ceiling, lowestRetry, lowestPromoted]) {
+      if (cap !== undefined && cap < incoming) incoming = cap;
     }
-    state.segmentCursors.scheduled = scheduledGuard.advance(scheduledCursor);
+    state.segmentCursors.incoming = maxTimestamp(floor, incoming);
+    // The sweep cursor moves a whole commit stamp at a time, and only past
+    // stamps whose behind-the-cursor entries were all retired. Its keys are
+    // observed commit stamps, so no ceiling is needed.
+    let sweepCursor = state.segmentCursors.scheduled ?? 0n;
+    for (const step of sweep) {
+      if (step.starts.some((s) => !handled.has(s._id))) break;
+      sweepCursor = step.scheduledAt;
+    }
+    state.segmentCursors.scheduled = maxTimestamp(
+      state.segmentCursors.scheduled ?? 0n,
+      sweepCursor,
+    );
+    // Record this run's own commit stamp: the next run reads this document,
+    // so its snapshot is at least this recent, and the mark seeds its ceiling.
+    state.lastCommitTs = ctx.db.vars.commitTs;
 
     await ctx.db.replace("internalState", state._id, state);
     // Return null: batch-worker re-runs `getBatch` immediately to drain, and
@@ -362,6 +407,13 @@ export const run = internalMutation({
   },
 });
 
+// How many `scheduledAt` entries the sweep inspects per iteration. Inspection
+// is read-only and each entry is inspected once ever, so this mainly bounds
+// per-iteration reads. It must exceed the most entries one transaction can
+// stamp identically (a maximal batch enqueue), or the cursor could never
+// clear that stamp.
+const SWEEP_BATCH = 2048;
+
 /** Read the three pending tables the loop processes. */
 async function queryPending(
   ctx: QueryCtx,
@@ -369,7 +421,8 @@ async function queryPending(
     completionCursor,
     cancelationCursor,
     incomingCursor,
-    scheduledCursor,
+    sweepCursor,
+    lastCommitTs,
     maxParallelism,
     runningCount,
     eligibleBefore,
@@ -377,7 +430,8 @@ async function queryPending(
     completionCursor: bigint;
     cancelationCursor: bigint;
     incomingCursor: bigint;
-    scheduledCursor: bigint;
+    sweepCursor: bigint;
+    lastCommitTs: bigint;
     maxParallelism: number;
     runningCount: number;
     eligibleBefore: bigint;
@@ -402,48 +456,97 @@ async function queryPending(
     ...completions.map((c) => c.workId),
     ...cancelations.map((c) => c.workId),
   ];
-  // Near-future work, still at its enqueue's commit position; `run` moves each
-  // entry to its start time (or starts it, if it came due first). Reading this
-  // lane separately means a backlog of it never sits in front of ready work —
-  // it drains at MAIN_BATCH_SIZE per iteration alongside the starts. Skipped
-  // when nothing can start: an already-due entry only leaves the lane by
-  // starting, so reading it with no capacity would spin the loop.
-  const scheduled =
-    startLimit === 0
-      ? []
-      : await ctx.db
-          .query("pendingStart")
-          .withIndex("hasRunAt_segment", (q) =>
-            q.eq("hasRunAt", true).gte("segment", scheduledCursor),
-          )
-          // eslint-disable-next-line @convex-dev/no-filter-in-query
-          .filter((q) =>
-            q.and(...excludedIds.map((id) => q.neq(q.field("workId"), id))),
-          )
-          .take(MAIN_BATCH_SIZE);
-  // Held entries that are already due take start slots, so only fetch ready
-  // work for the slots left over. Everything eligible, oldest first; work
-  // scheduled for later sorts above the bound, so it's left alone without
-  // pinning the cursor.
-  const now = Date.now();
-  const dueHeld = scheduled.filter((s) => (s.runAt ?? 0) <= now).length;
-  const readyLimit = Math.max(0, startLimit - dueHeld);
+
+  // ── The out-of-order sweep ──
+  // A scheduled enqueue writes its start time as its `segment`, but commits
+  // later — so if the commit took longer than the delay, the entry can land
+  // behind the incoming cursor, where the segment scan will never see it.
+  // Every wall-clock-keyed entry also records its commit stamp in
+  // `scheduledAt`, and commit order is the one order nothing can land behind:
+  // walking it from a cursor inspects every entry exactly once. An entry
+  // whose `segment` is still at or above the incoming cursor can be passed
+  // forever — the scan can only get past it by reading it. One behind the
+  // cursor is due (the cursor never passes the current time) and must start
+  // from here; those claim start slots first. The cursor moves a whole commit
+  // stamp at a time (`gt` — a batch enqueue shares one stamp, and a bare
+  // stamp can't split the group), and only past stamps whose stragglers all
+  // got retired.
+  const scanned = await ctx.db
+    .query("pendingStart")
+    .withIndex("scheduledAt", (q) => q.gt("scheduledAt", sweepCursor))
+    // eslint-disable-next-line @convex-dev/no-filter-in-query
+    .filter((q) =>
+      q.and(...excludedIds.map((id) => q.neq(q.field("workId"), id))),
+    )
+    .take(SWEEP_BATCH);
+  const exhausted = scanned.length < SWEEP_BATCH;
+  const sweep: SweepStep[] = [];
+  let oooBudget = startLimit;
+  for (let i = 0; i < scanned.length; ) {
+    const stamp = scanned[i].scheduledAt as bigint;
+    const group = [];
+    while (i < scanned.length && (scanned[i].scheduledAt as bigint) === stamp) {
+      group.push(scanned[i]);
+      i++;
+    }
+    // If the scan was cut mid-stamp, rows beyond it may share this stamp, so
+    // it can't be cleared yet. (Reachable only if one transaction stamped
+    // more entries than SWEEP_BATCH.)
+    if (i === scanned.length && !exhausted) break;
+    const behind = group.filter((r) => (r.segment as bigint) < incomingCursor);
+    if (behind.length > oooBudget) break;
+    oooBudget -= behind.length;
+    const starts = behind.map((r) => ({
+      _id: r._id,
+      workId: r.workId,
+      segment: r.segment as bigint,
+      runAt: r.runAt,
+    }));
+    // Fold stamp-only advances into the previous step to keep batches small;
+    // a step is only worth carrying for its starts or as the furthest bound.
+    const previous = sweep.at(-1);
+    if (starts.length === 0 && previous && previous.starts.length === 0) {
+      previous.scheduledAt = stamp;
+    } else {
+      sweep.push({ scheduledAt: stamp, starts });
+    }
+  }
+
+  // Entries the sweep starts take slots first; only fetch ready work for the
+  // slots left over. Everything eligible, oldest first; work scheduled for
+  // later sorts above the bound, so it's left alone without pinning the
+  // cursor.
+  const sweptCount = sweep.reduce((n, s) => n + s.starts.length, 0);
+  const readyLimit = Math.max(0, startLimit - sweptCount);
   const starts =
     readyLimit === 0
       ? []
       : await ctx.db
           .query("pendingStart")
-          .withIndex("hasRunAt_segment", (q) =>
-            q
-              .eq("hasRunAt", undefined)
-              .gte("segment", incomingCursor)
-              .lt("segment", eligibleBefore),
+          .withIndex("segment", (q) =>
+            q.gte("segment", incomingCursor).lt("segment", eligibleBefore),
           )
           // eslint-disable-next-line @convex-dev/no-filter-in-query
           .filter((q) =>
             q.and(...excludedIds.map((id) => q.neq(q.field("workId"), id))),
           )
           .take(readyLimit);
+  // The highest commit timestamp observed while building this batch. Every
+  // source is the stamp of a transaction visible to this snapshot, so
+  // anything committing later is stamped above it — the frontier the cursor
+  // may advance to. Entries with `scheduledAt` are wall-clock-keyed and don't
+  // count (their stamps do); entries without it carry an observed commit
+  // stamp in `segment` (an older version's tiny buckets can't win a max).
+  const cursorCeiling = [
+    lastCommitTs,
+    ...completions.map((c) => c.segment as bigint),
+    ...cancelations.map((c) => c.segment as bigint),
+    ...starts
+      .filter((s) => s.scheduledAt === undefined)
+      .map((s) => s.segment as bigint),
+    ...scanned.map((r) => r.scheduledAt as bigint),
+  ].reduce(maxTimestamp);
+
   return {
     completions,
     cancelations,
@@ -454,33 +557,33 @@ async function queryPending(
         workId: s.workId,
         segment,
         // An entry from before commit-timestamp ordering keeps its start time
-        // in `segment`; recovering it here means `run` handles it like any
-        // other scheduled entry and rewrites it as a timestamp.
+        // in `segment` (or in its deprecated `runAt`); recovering it here
+        // means `run` handles it like any other not-yet-due entry and re-keys
+        // it as a timestamp.
         runAt: s.runAt ?? legacyRunAt(segment),
       };
     }) satisfies Start[],
-    scheduled: scheduled.map((s) => ({
-      _id: s._id,
-      workId: s.workId,
-      segment: s.segment as bigint,
-      runAt: s.runAt,
-    })) satisfies Start[],
+    sweep,
+    cursorCeiling,
   };
 }
 
 /**
  * Handles the completion of pending completions.
  * This only processes work that succeeded or failed, not canceled.
+ * Returns the lowest `segment` written for a retry, which bounds how far the
+ * incoming cursor may advance this iteration.
  */
 async function handleCompletions(
   ctx: MutationCtx,
   state: Doc<"internalState">,
   completed: Completion[],
   console: Logger,
-  guard: CursorGuard,
+  floor: bigint,
 ) {
   // Completions that were going to be retried but have since been canceled.
   const toCancel: CompleteJob[] = [];
+  let lowestRetry: bigint | undefined;
   await Promise.all(
     completed.map(async (c) => {
       await ctx.db.delete("pendingCompletion", c._id);
@@ -499,8 +602,11 @@ async function handleCompletions(
           console.warn(`[main] ${c.workId} is gone, but trying to complete`);
           return;
         }
-        const retried = await rescheduleJob(ctx, work, console, guard);
-        if (retried) {
+        const retriedAt = await rescheduleJob(ctx, work, console, floor);
+        if (retriedAt !== undefined) {
+          if (lowestRetry === undefined || retriedAt < lowestRetry) {
+            lowestRetry = retriedAt;
+          }
           state.report.retries++;
           recordCompleted(console, work, "retrying", undefined);
         } else {
@@ -529,7 +635,7 @@ async function handleCompletions(
   const numCompleted = before - state.running.length;
   state.report.completed += numCompleted;
   console.debug(`[main] completed ${numCompleted} work`);
-  return toCancel;
+  return { toCancel, lowestRetry };
 }
 
 /**
@@ -642,72 +748,39 @@ function maxTimestamp(a: bigint, b: bigint) {
 }
 
 /**
- * Nothing rescans a lane behind its cursor, so an entry below the cursor is
- * lost for good. This guard is how the loop keeps the two sides of that
- * invariant consistent within one transaction:
- *
- * - `place` raises a `segment` we're about to write to at least the cursor, so
- *   we never put an entry somewhere the loop has already read past.
- * - `advance` moves the cursor as far as the batch earned, but never past a
- *   segment placed this transaction — the cursor isn't final when those writes
- *   happen (a retry is written while completions are handled, before we know
- *   which starts we'll get through) — and never backwards, since the batch was
- *   built from a snapshot that may predate the cursor as read here.
- *
- * Placing against the cursor rather than against the clock is what makes the
- * requirement local: nothing that writes a `segment` has to know where the
- * eligibility bound sits or how a commit timestamp compares to `Date.now()`.
- * Equality is fine throughout — lanes are read with `gte`.
- */
-function cursorGuard(cursor: bigint) {
-  let lowestPlaced: bigint | undefined;
-  return {
-    place(segment: bigint): bigint {
-      const placed = maxTimestamp(segment, cursor);
-      if (lowestPlaced === undefined || placed < lowestPlaced) {
-        lowestPlaced = placed;
-      }
-      return placed;
-    },
-    advance(candidate: bigint): bigint {
-      const capped =
-        lowestPlaced !== undefined && lowestPlaced < candidate
-          ? lowestPlaced
-          : candidate;
-      return maxTimestamp(cursor, capped);
-    },
-  };
-}
-type CursorGuard = ReturnType<typeof cursorGuard>;
-
-/**
- * Moves entries that aren't due yet to sort at their `runAt` instead of at the
- * commit timestamp they were enqueued with, so the cursor can pass them and
- * they don't come back until they're actually due. Safe to compute from the
- * clock here, unlike at enqueue: this transaction writes the cursors too, so
- * the guard can hold the entry above them.
+ * Re-keys entries that are visible but not due to sort at their start time,
+ * so the cursor can pass them and they don't come back until they're actually
+ * due. New-format entries are keyed at their start time and only become
+ * visible once due, so this handles what older versions wrote — 100ms
+ * buckets, and an unreleased revision's entries held at their commit
+ * position — clearing that revision's fields as it goes. Safe to compute from
+ * the clock here, unlike at enqueue: this transaction writes the cursor too,
+ * and raising the key to at least `floor` keeps the entry readable. Returns
+ * the lowest key written, which bounds how far the cursor may advance.
  */
 async function promoteScheduled(
   ctx: MutationCtx,
   notYet: Start[],
-  guard: CursorGuard,
-) {
+  floor: bigint,
+): Promise<bigint | undefined> {
+  let lowest: bigint | undefined;
   await Promise.all(
     notYet.map(async ({ _id, runAt }) => {
       // A concurrent cancelation may have removed it.
       if (!(await ctx.db.get("pendingStart", _id))) return;
+      const segment = maxTimestamp(dueTimestamp(runAt!), floor);
+      if (lowest === undefined || segment < lowest) lowest = segment;
       await ctx.db.patch("pendingStart", _id, {
-        // Round a fractional `runAt` up to the next whole millisecond — the
-        // first one the clock can call it due. Rounding down would leave it
-        // readable while `isDue` still says no, so it'd be re-read and
-        // re-moved until the clock leaves the current millisecond.
-        segment: guard.place(toTimestamp(Math.ceil(runAt!))),
-        // Now ordered by its start time, so it leaves the scheduled lane and
-        // waits in the incoming one like any far-future enqueue.
+        segment,
+        // The key is now a wall-clock time, and `scheduledAt` is what records
+        // that (its absence marks a key as an observed commit stamp).
+        scheduledAt: ctx.db.vars.commitTs,
+        runAt: undefined,
         hasRunAt: undefined,
       });
     }),
   );
+  return lowest;
 }
 
 /**
@@ -843,14 +916,14 @@ async function beginWorkBatch(
 /**
  * Reschedules a job for retry.
  * If it's been canceled in the mean time, don't retry.
- * @returns true if the job was rescheduled, false if it was not.
+ * @returns the `segment` it was rescheduled at, or undefined if it was not.
  */
 async function rescheduleJob(
   ctx: MutationCtx,
   work: Doc<"work">,
   console: Logger,
-  guard: CursorGuard,
-): Promise<boolean> {
+  floor: bigint,
+): Promise<bigint | undefined> {
   const pendingCancelation = await ctx.db
     .query("pendingCancelation")
     .withIndex("workId", (q) => q.eq("workId", work._id))
@@ -858,14 +931,14 @@ async function rescheduleJob(
   if (pendingCancelation) {
     // If there's an un-processed cancelation request, don't retry.
     console.warn(`[main] ${work._id} in pendingCancelation so not retrying`);
-    return false;
+    return undefined;
   }
   if (work.canceled) {
-    return false;
+    return undefined;
   }
   if (!work.retryBehavior) {
     console.warn(`[main] ${work._id} has no retryBehavior so not retrying`);
-    return false;
+    return undefined;
   }
   const existing = await ctx.db
     .query("pendingStart")
@@ -874,21 +947,24 @@ async function rescheduleJob(
   if (existing) {
     // Not sure why this would ever happen, but ensure uniqueness explicitly.
     console.error(`[main] ${work._id} already in pendingStart so not retrying`);
-    return false;
+    return undefined;
   }
   const backoffMs =
     work.retryBehavior.initialBackoffMs *
     Math.pow(work.retryBehavior.base, work.attempts - 1);
   const nextAttempt = withJitter(backoffMs);
   // The backoff can go straight into `segment` however short it is: we're in
-  // the transaction that writes the cursor, so the guard can keep the entry
-  // above it even for a zero backoff. No `runAt` needed — `segment` already is
-  // the start time.
+  // the transaction that writes the cursor, and raising the key to at least
+  // `floor` keeps the entry readable even for a zero backoff. The key is a
+  // wall-clock time, so it records its commit stamp in `scheduledAt` like any
+  // scheduled enqueue (its absence marks a key as an observed commit stamp).
+  const segment = maxTimestamp(dueTimestamp(Date.now() + nextAttempt), floor);
   await ctx.db.insert("pendingStart", {
     workId: work._id,
-    segment: guard.place(toTimestamp(Date.now() + nextAttempt)),
+    segment,
+    scheduledAt: ctx.db.vars.commitTs,
   });
-  return true;
+  return segment;
 }
 
 export function withJitter(delay: number) {
diff --git a/src/component/schema.ts b/src/component/schema.ts
index 9c5f7191..61df53d0 100644
--- a/src/component/schema.ts
+++ b/src/component/schema.ts
@@ -18,11 +18,12 @@ import {
 // is assigned when the mutation *starts*, so a slow mutation's row can land
 // behind rows that already committed and were read.)
 //
-// Entries that shouldn't be processed until later store that time directly,
-// which sorts after everything committed before then. `v.commitTs()` accepts any
-// int64, so both live in one field and one index.
+// Entries that shouldn't be processed until later store that time directly.
+// `v.commitTs()` accepts any int64, so both clocks live in one field and one
+// index, and scheduled work sorts above the loop's read bound until it's due.
 const segment = v.commitTs();
-// A cursor into a `segment` index. Reads of the field come back as a plain int64.
+// A cursor into an index on a commit-timestamp field. Reads of such a field
+// come back as a plain int64.
 const timestamp = v.int64();
 
 export default defineSchema({
@@ -41,10 +42,16 @@ export default defineSchema({
       incoming: timestamp,
       completion: timestamp,
       cancelation: timestamp,
-      // Cursor into pendingStart's `hasRunAt: true` lane. Optional because it
-      // didn't exist before that lane did; absent means "from the beginning".
+      // Cursor into pendingStart's `scheduledAt` index — the out-of-order
+      // sweep. Optional because it didn't always exist; absent means "from the
+      // beginning".
       scheduled: v.optional(timestamp),
     }),
+    // The commit timestamp of the previous `run`. Everything this run read is
+    // at least this recent, so the `incoming` cursor may advance up to the
+    // highest commit timestamp observed — but no further, or a commit racing
+    // this run could land behind it. See `run`'s cursor advance.
+    lastCommitTs: v.optional(v.commitTs()),
     // Unlike the cursors, this stays a 100ms "segment": it only paces how often
     // the loop checks for stuck jobs.
     lastRecovery: v.int64(),
@@ -85,23 +92,25 @@ export default defineSchema({
   pendingStart: defineTable({
     workId: v.id("work"),
     segment,
-    // Only set when the work shouldn't start yet. `segment` already holds that
-    // time whenever we could safely write it there; when we couldn't — a
-    // caller-supplied `runAt` too near to be sure it lands ahead of the loop's
-    // cursor — `segment` is the commit timestamp and this is what tells the
-    // loop to move the entry forward instead of starting it.
+    // The commit timestamp, recorded on every entry whose `segment` is a
+    // wall-clock start time rather than a commit timestamp. Such an entry can
+    // commit *behind* the loop's `segment` cursor (when the enqueue takes
+    // longer to commit than the delay); the sweep walks this index in commit
+    // order — which nothing can land behind — and starts any entry whose
+    // `segment` the cursor already passed. Its absence also tells the loop
+    // that `segment` is a commit timestamp it has observed, which is what
+    // bounds how far the cursor may advance. See `queryPending`.
+    scheduledAt: v.optional(v.commitTs()),
+    // @deprecated The exact start time of an entry written by an unreleased
+    // revision. The loop re-keys such entries and clears this on first read.
     runAt: v.optional(v.number()),
-    // Set (to true) on exactly the entries described above: held at their
-    // commit position awaiting a move to their start time. It splits the index
-    // into two lanes so those entries never sit in front of ready work — the
-    // loop drains this lane in parallel with starting work from the other.
-    // Entries whose `segment` is already their start time don't need it, and
-    // its absence puts entries older versions wrote in the incoming lane, where
-    // the legacy handling lives.
+    // @deprecated An unreleased revision held near-future entries in a
+    // separate index lane marked by this field; cleared like `runAt`.
     hasRunAt: v.optional(v.boolean()),
   })
     .index("workId", ["workId"])
-    .index("hasRunAt_segment", ["hasRunAt", "segment"]),
+    .index("segment", ["segment"])
+    .index("scheduledAt", ["scheduledAt"]),
 
   // Written by complete, read & deleted by `main`.
   pendingCompletion: defineTable({
diff --git a/src/component/shared.ts b/src/component/shared.ts
index 6e860d51..4cc21f66 100644
--- a/src/component/shared.ts
+++ b/src/component/shared.ts
@@ -55,6 +55,16 @@ export function fromTimestamp(timestamp: bigint): number {
   return Number(timestamp / NS_PER_MS);
 }
 
+/**
+ * A start time as an ordering value: rounded up to the next whole millisecond,
+ * the first one the clock can call it due. Rounding down would make an entry
+ * readable while `isDue` still says no, so the loop would re-read it until the
+ * clock leaves the current millisecond.
+ */
+export function dueTimestamp(runAt: number): bigint {
+  return toTimestamp(Math.ceil(runAt));
+}
+
 /**
  * The exclusive upper bound on entries eligible at `ms` — the end of that
  * millisecond, not the start of it.
@@ -83,22 +93,6 @@ export function legacyRunAt(segment: bigint): number | undefined {
   return segment < MIN_TIMESTAMP ? fromSegment(segment) : undefined;
 }
 
-/**
- * How far ahead a caller-supplied `runAt` has to be before we can write it
- * straight into the ordering field.
- *
- * Writing a wall-clock time there is only safe if the value is guaranteed to
- * land after every commit timestamp the loop has already read. The time is
- * chosen when the enqueue *starts*, so the margin has to cover however long
- * that transaction then takes to commit. Five minutes is far longer than any
- * mutation can run, so a `runAt` beyond it cannot land in the past. Anything
- * nearer is ordered by `db.vars.commitTs` instead, keeps its `runAt` in a
- * separate field, and is marked `hasRunAt` so it sits in its own index lane.
- * The loop then moves it to its start time itself, which is safe there; see
- * `promoteScheduled` in loop.ts.
- */
-export const SAFE_FUTURE_MS = 5 * MINUTE;
-
 export const vConfig = v.object({
   maxParallelism: v.number(),
   logLevel,
diff --git a/src/component/stats.test.ts b/src/component/stats.test.ts
index 773d25d8..693de339 100644
--- a/src/component/stats.test.ts
+++ b/src/component/stats.test.ts
@@ -305,9 +305,7 @@ describe("stats", () => {
       const cursor = await t.run(async (ctx) => {
         return await paginator(ctx.db, schema)
           .query("pendingStart")
-          .withIndex("hasRunAt_segment", (q) =>
-            q.eq("hasRunAt", undefined).gte("segment", 0n),
-          )
+          .withIndex("segment", (q) => q.gte("segment", 0n))
           .paginate({
             numItems: 1,
             cursor: null,
diff --git a/src/component/stats.ts b/src/component/stats.ts
index 5bdef9d7..5e256d01 100644
--- a/src/component/stats.ts
+++ b/src/component/stats.ts
@@ -80,13 +80,11 @@ export async function generateReport(
   }
   const currentSegment = getCurrentSegment();
   // Backlog is work that's eligible now; anything scheduled for later sorts
-  // above the current time — or sits in the held `hasRunAt` lane — and isn't
-  // waiting on us.
+  // above the current time and isn't waiting on us.
   const pendingStart = await paginator(ctx.db, schema)
     .query("pendingStart")
-    .withIndex("hasRunAt_segment", (q) =>
+    .withIndex("segment", (q) =>
       q
-        .eq("hasRunAt", undefined)
         .gte("segment", state.segmentCursors.incoming)
         .lt("segment", endOfMs(Date.now())),
     )

From 3b22d9af7e2a11b145f26f41e842b374b48be7c8 Mon Sep 17 00:00:00 2001
From: Ian Macartney <366683+ianmacartney@users.noreply.github.com>
Date: Fri, 14 Aug 2026 15:16:22 -0700
Subject: [PATCH 2/9] replace the pendingStart workId index with a pointer on
 work
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Every site that looked up a pendingStart by workId has the work document in
hand, so the work doc now carries pendingStartId and the index is gone,
leaving pendingStart with two indexes (segment, scheduledAt). The pointer
can be stale — the entry starts and is deleted — so readers check the entry
still exists.

Entries the pointer can't reach (work deleted out from under them, or
written by a version without the pointer) are dropped reactively: the
segment scan deletes an entry whose work is gone, and now also one whose
work was canceled, finishing the cancelation the pointer couldn't. Canceling
pre-pointer work therefore marks it immediately but clears its queue entry
only when the entry comes due.
---
 src/component/danger.ts            | 10 +++---
 src/component/lib.ts               | 11 ++++---
 src/component/loop.test.ts         | 20 +++++++++++
 src/component/loop.ts              | 53 +++++++++++++++++++-----------
 src/component/schema.ts            |  7 +++-
 src/component/stateMachine.test.ts | 37 +++++++++++++--------
 6 files changed, 95 insertions(+), 43 deletions(-)

diff --git a/src/component/danger.ts b/src/component/danger.ts
index bf91148b..60c31c77 100644
--- a/src/component/danger.ts
+++ b/src/component/danger.ts
@@ -70,10 +70,12 @@ export const clearOldWork = internalMutation({
       .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();
+      // The pointer can be stale or missing on entries older versions wrote;
+      // a pendingStart missed here is dropped when the loop reads it and
+      // finds the work gone.
+      const pendingStart = entry.pendingStartId
+        ? await ctx.db.get("pendingStart", entry.pendingStartId)
+        : null;
       const pendingCompletion = await ctx.db
         .query("pendingCompletion")
         .withIndex("workId", (q) => q.eq("workId", entry._id))
diff --git a/src/component/lib.ts b/src/component/lib.ts
index 23b08fd6..a10868e8 100644
--- a/src/component/lib.ts
+++ b/src/component/lib.ts
@@ -119,11 +119,12 @@ export async function enqueueHandler(
   // the loop sweeps that index in commit order and starts anything the cursor
   // passed over.
   const delayMs = runAt - Date.now();
-  await ctx.db.insert("pendingStart", {
+  const pendingStartId = await ctx.db.insert("pendingStart", {
     workId,
     segment: delayMs > 0 ? dueTimestamp(runAt) : ctx.db.vars.commitTs,
     ...(delayMs > 0 ? { scheduledAt: ctx.db.vars.commitTs } : {}),
   });
+  await ctx.db.patch("work", workId, { pendingStartId });
   recordEnqueued(console, { workId, fnName: workArgs.fnName, runAt });
   return workId;
 }
@@ -214,10 +215,10 @@ async function statusHandler(ctx: QueryCtx, { id }: { id: Id<"work"> }) {
   if (!work) {
     return { state: "finished" } as const;
   }
-  const pendingStart = await ctx.db
-    .query("pendingStart")
-    .withIndex("workId", (q) => q.eq("workId", id))
-    .unique();
+  // The pointer can be stale after the work starts; check the entry exists.
+  const pendingStart = work.pendingStartId
+    ? await ctx.db.get("pendingStart", work.pendingStartId)
+    : null;
   if (pendingStart) {
     return { state: "pending", previousAttempts: work.attempts } as const;
   }
diff --git a/src/component/loop.test.ts b/src/component/loop.test.ts
index 565645ed..8457268a 100644
--- a/src/component/loop.test.ts
+++ b/src/component/loop.test.ts
@@ -459,6 +459,26 @@ describe("loop", () => {
       expect(o.pendingStart).toHaveLength(0);
       expect(o.running.map((r) => r.workId)).toEqual([workId]);
     });
+
+    it("drops the entry of canceled work that predates the pendingStartId pointer", async () => {
+      await initialize();
+      const runAt = Date.now() + 100 * SECOND;
+      const workId = await enqueueLegacyWork(runAt);
+      await runLoop(); // re-keys it to its start time
+      await runLoop(); // the sweep verifies the re-keyed entry
+
+      // Cancelation can't reach the entry (no pointer on old work docs), so
+      // it only marks the work; the scan drops the entry when it comes due.
+      await t.mutation(api.lib.cancel, { id: workId });
+      await runLoop();
+      expect((await observe()).pendingStart).toHaveLength(1);
+
+      vi.advanceTimersByTime(100 * SECOND);
+      await runLoop();
+      const o = await observe();
+      expect(o.pendingStart).toHaveLength(0);
+      expect(o.running).toHaveLength(0);
+    });
   });
 
   describe("capacity", () => {
diff --git a/src/component/loop.ts b/src/component/loop.ts
index 58f8e16f..9d9b179a 100644
--- a/src/component/loop.ts
+++ b/src/component/loop.ts
@@ -668,23 +668,19 @@ async function handleCancelation(
           }
           const work = await ctx.db.get("work", workId);
           if (!work) {
+            // A pendingStart left pointing at it, if any, is dropped when the
+            // segment scan reads it and finds the work gone.
             console.warn(`[main] ${workId} is gone, but trying to cancel`);
-            // Drop any pendingStart left pointing at it. Nothing rescans that
-            // lane, so a row left behind here would never be read again.
-            const orphan = await ctx.db
-              .query("pendingStart")
-              .withIndex("workId", (q) => q.eq("workId", workId))
-              .unique();
-            if (orphan) await ctx.db.delete("pendingStart", orphan._id);
             return null;
           }
-          // Ensure it doesn't retry.
+          // Ensure it doesn't retry — and doesn't start, if its pendingStart
+          // is only reachable through the scan (the pointer can be missing on
+          // entries older versions wrote; `handleStart` checks this flag).
           await ctx.db.patch("work", workId, { canceled: true });
-          // Ensure it doesn't start.
-          const pendingStart = await ctx.db
-            .query("pendingStart")
-            .withIndex("workId", (q) => q.eq("workId", workId))
-            .unique();
+          // Ensure it doesn't start. The pointer can be stale; check it.
+          const pendingStart = work.pendingStartId
+            ? await ctx.db.get("pendingStart", work.pendingStartId)
+            : null;
           if (pendingStart && !canceledWork.has(workId)) {
             state.report.canceled++;
             await ctx.db.delete("pendingStart", pendingStart._id);
@@ -816,6 +812,23 @@ async function handleStart(
           console.error(`Trying to start, but work not found: ${workId}`);
           return null;
         }
+        if (work.canceled) {
+          // Canceled while its entry was only reachable through this scan
+          // (no `pendingStartId` pointer — written by an older version).
+          // Finish the cancelation that couldn't find the entry then.
+          console.debug(`[main] ${workId} was canceled (not starting)`);
+          state.report.canceled++;
+          await ctx.scheduler.runAfter(0, internal.complete.complete, {
+            jobs: [
+              {
+                workId,
+                runResult: { kind: "canceled" as const },
+                attempt: work.attempts,
+              },
+            ],
+          });
+          return null;
+        }
         return {
           work,
           // `segment` is when this became eligible: the time it was scheduled
@@ -940,11 +953,12 @@ async function rescheduleJob(
     console.warn(`[main] ${work._id} has no retryBehavior so not retrying`);
     return undefined;
   }
-  const existing = await ctx.db
-    .query("pendingStart")
-    .withIndex("workId", (q) => q.eq("workId", work._id))
-    .first();
-  if (existing) {
+  // The pointer can be stale (its entry started); check before declaring a
+  // duplicate.
+  if (
+    work.pendingStartId &&
+    (await ctx.db.get("pendingStart", work.pendingStartId))
+  ) {
     // Not sure why this would ever happen, but ensure uniqueness explicitly.
     console.error(`[main] ${work._id} already in pendingStart so not retrying`);
     return undefined;
@@ -959,11 +973,12 @@ async function rescheduleJob(
   // wall-clock time, so it records its commit stamp in `scheduledAt` like any
   // scheduled enqueue (its absence marks a key as an observed commit stamp).
   const segment = maxTimestamp(dueTimestamp(Date.now() + nextAttempt), floor);
-  await ctx.db.insert("pendingStart", {
+  const pendingStartId = await ctx.db.insert("pendingStart", {
     workId: work._id,
     segment,
     scheduledAt: ctx.db.vars.commitTs,
   });
+  await ctx.db.patch("work", work._id, { pendingStartId });
   return segment;
 }
 
diff --git a/src/component/schema.ts b/src/component/schema.ts
index 61df53d0..ae1620b2 100644
--- a/src/component/schema.ts
+++ b/src/component/schema.ts
@@ -86,6 +86,12 @@ export default defineSchema({
     onComplete: v.optional(vOnCompleteFnContext),
     retryBehavior: v.optional(retryBehavior),
     canceled: v.optional(v.boolean()),
+    // The queue entry waiting to start this work, if any — replacing a
+    // `workId` index on pendingStart. May be stale (the entry started and was
+    // deleted, or predates this pointer), so readers check the entry still
+    // exists. A pendingStart unreachable through this pointer is dropped
+    // reactively: the loop reads it and finds its work gone or canceled.
+    pendingStartId: v.optional(v.id("pendingStart")),
   }),
 
   // Written on enqueue & rescheduled for retry, read & deleted by `main`.
@@ -108,7 +114,6 @@ export default defineSchema({
     // separate index lane marked by this field; cleared like `runAt`.
     hasRunAt: v.optional(v.boolean()),
   })
-    .index("workId", ["workId"])
     .index("segment", ["segment"])
     .index("scheduledAt", ["scheduledAt"]),
 
diff --git a/src/component/stateMachine.test.ts b/src/component/stateMachine.test.ts
index ddace386..16e0e782 100644
--- a/src/component/stateMachine.test.ts
+++ b/src/component/stateMachine.test.ts
@@ -266,10 +266,13 @@ describe("state machine", () => {
 
       // pendingStart
       if (state.pendingStart) {
-        await ctx.db.insert("pendingStart", {
+        const pendingStartId = await ctx.db.insert("pendingStart", {
           workId: wId,
           segment: seg,
         });
+        if (await ctx.db.get("work", wId)) {
+          await ctx.db.patch("work", wId, { pendingStartId });
+        }
       }
 
       // pendingCompletion
@@ -305,10 +308,10 @@ describe("state machine", () => {
   async function observeState(workId: Id<"work">): Promise<ObservedState> {
     return t.run(async (ctx) => {
       const work = await ctx.db.get("work", workId);
-      const ps = await ctx.db
-        .query("pendingStart")
-        .withIndex("workId", (q) => q.eq("workId", workId))
-        .first();
+      const ps =
+        (await ctx.db.query("pendingStart").collect()).find(
+          (p) => p.workId === workId,
+        ) ?? null;
       const state = await ctx.db.query("internalState").unique();
       const inRunning =
         state?.running.some((r) => r.workId === workId) ?? false;
@@ -961,9 +964,11 @@ describe("state machine", () => {
           fnArgs: {},
           attempts: 0,
         });
-        await ctx.db.insert("pendingStart", {
-          workId: wId,
-          segment: seg,
+        await ctx.db.patch("work", wId, {
+          pendingStartId: await ctx.db.insert("pendingStart", {
+            workId: wId,
+            segment: seg,
+          }),
         });
         await ctx.db.insert("pendingCancelation", {
           workId: wId,
@@ -1155,17 +1160,21 @@ describe("state machine", () => {
           retry: false,
           runResult: { kind: "success", returnValue: null },
         });
-        await ctx.db.insert("pendingStart", {
-          workId: w2,
-          segment: seg,
+        await ctx.db.patch("work", w2, {
+          pendingStartId: await ctx.db.insert("pendingStart", {
+            workId: w2,
+            segment: seg,
+          }),
         });
         await ctx.db.insert("pendingCancelation", {
           workId: w2,
           segment: seg,
         });
-        await ctx.db.insert("pendingStart", {
-          workId: w3,
-          segment: seg,
+        await ctx.db.patch("work", w3, {
+          pendingStartId: await ctx.db.insert("pendingStart", {
+            workId: w3,
+            segment: seg,
+          }),
         });
 
         return { w1, w2, w3 };

From 80aad373519cda999dd707b2977f7840fb281d7d Mon Sep 17 00:00:00 2001
From: Ian Macartney <366683+ianmacartney@users.noreply.github.com>
Date: Mon, 17 Aug 2026 10:45:06 -0700
Subject: [PATCH 3/9] drop the runAt and hasRunAt fields

Only unreleased revisions ever wrote them, no deployment still holds them
(schema validation on push confirms), and the current design doesn't need
them: an entry's segment is its start time, scheduledAt marks it as
wall-clock-keyed, and due-ness is visibility. The only remaining upgrade
shape is 0.4.9's 100ms bucket, recovered from segment by magnitude.
---
 src/component/loop.test.ts | 37 ++-----------------------------------
 src/component/loop.ts      | 24 +++++++++++-------------
 src/component/schema.ts    |  6 ------
 3 files changed, 13 insertions(+), 54 deletions(-)

diff --git a/src/component/loop.test.ts b/src/component/loop.test.ts
index 8457268a..71548989 100644
--- a/src/component/loop.test.ts
+++ b/src/component/loop.test.ts
@@ -17,7 +17,6 @@ import { setupTest } from "./setup.test.js";
 import {
   DEFAULT_MAX_PARALLELISM,
   fromSegment,
-  fromTimestamp,
   toSegment,
   toTimestamp,
   WORKER_NAME,
@@ -401,38 +400,6 @@ describe("loop", () => {
       expect(o.running.map((r) => r.workId)).toEqual([workId]);
     });
 
-    it("moves near-future work a pre-lane version left in the incoming lane", async () => {
-      await initialize();
-      const runAt = Date.now() + 100 * SECOND;
-      // Held at its commit position with `runAt` but no `hasRunAt`: the shape
-      // written before the scheduled lane existed.
-      const workId = await t.run(async (ctx) => {
-        const wid = await ctx.db.insert("work", {
-          fnType: "action",
-          fnHandle: "test_handle",
-          fnName: "test_handle",
-          fnArgs: {},
-          attempts: 0,
-        });
-        await ctx.db.insert("pendingStart", {
-          workId: wid,
-          segment: ctx.db.vars.commitTs,
-          runAt,
-        });
-        return wid;
-      });
-
-      await runLoop();
-
-      const o = await observe();
-      expect(o.running).toHaveLength(0);
-      expect(o.pendingStart[0].segment).toBe(toTimestamp(runAt));
-
-      vi.advanceTimersByTime(100 * SECOND);
-      await runLoop();
-      expect((await observe()).running.map((r) => r.workId)).toEqual([workId]);
-    });
-
     it("keeps holding work scheduled for later, rather than starting it early", async () => {
       await initialize();
       const runAt = Date.now() + 100 * SECOND;
@@ -871,12 +838,12 @@ describe("loop", () => {
           fnArgs: {},
           attempts: 0,
         });
-        await ctx.db.insert("pendingStart", {
+        const pendingStartId = await ctx.db.insert("pendingStart", {
           workId,
           segment: cursor - 1n,
-          runAt: fromTimestamp(cursor - 1n),
           scheduledAt: ctx.db.vars.commitTs,
         });
+        await ctx.db.patch("work", workId, { pendingStartId });
         return workId;
       });
     }
diff --git a/src/component/loop.ts b/src/component/loop.ts
index 9d9b179a..4112dd31 100644
--- a/src/component/loop.ts
+++ b/src/component/loop.ts
@@ -99,6 +99,9 @@ const vStart = v.object({
   _id: v.id("pendingStart"),
   workId: v.id("work"),
   segment: v.int64(),
+  // Not a stored field: the start time recovered from an entry an older
+  // version wrote (its `segment` is a 100ms bucket, not a timestamp). Absent
+  // on anything written by this version — such entries are due when visible.
   runAt: v.optional(v.number()),
 });
 type Start = Infer<typeof vStart>;
@@ -500,7 +503,6 @@ async function queryPending(
       _id: r._id,
       workId: r.workId,
       segment: r.segment as bigint,
-      runAt: r.runAt,
     }));
     // Fold stamp-only advances into the previous step to keep batches small;
     // a step is only worth carrying for its starts or as the furthest bound.
@@ -557,10 +559,9 @@ async function queryPending(
         workId: s.workId,
         segment,
         // An entry from before commit-timestamp ordering keeps its start time
-        // in `segment` (or in its deprecated `runAt`); recovering it here
-        // means `run` handles it like any other not-yet-due entry and re-keys
-        // it as a timestamp.
-        runAt: s.runAt ?? legacyRunAt(segment),
+        // in `segment`; recovering it here means `run` handles it like any
+        // other not-yet-due entry and re-keys it as a timestamp.
+        runAt: legacyRunAt(segment),
       };
     }) satisfies Start[],
     sweep,
@@ -747,12 +748,11 @@ function maxTimestamp(a: bigint, b: bigint) {
  * Re-keys entries that are visible but not due to sort at their start time,
  * so the cursor can pass them and they don't come back until they're actually
  * due. New-format entries are keyed at their start time and only become
- * visible once due, so this handles what older versions wrote — 100ms
- * buckets, and an unreleased revision's entries held at their commit
- * position — clearing that revision's fields as it goes. Safe to compute from
- * the clock here, unlike at enqueue: this transaction writes the cursor too,
- * and raising the key to at least `floor` keeps the entry readable. Returns
- * the lowest key written, which bounds how far the cursor may advance.
+ * visible once due, so this only handles the 100ms buckets older versions
+ * wrote. Safe to compute from the clock here, unlike at enqueue: this
+ * transaction writes the cursor too, and raising the key to at least `floor`
+ * keeps the entry readable. Returns the lowest key written, which bounds how
+ * far the cursor may advance.
  */
 async function promoteScheduled(
   ctx: MutationCtx,
@@ -771,8 +771,6 @@ async function promoteScheduled(
         // The key is now a wall-clock time, and `scheduledAt` is what records
         // that (its absence marks a key as an observed commit stamp).
         scheduledAt: ctx.db.vars.commitTs,
-        runAt: undefined,
-        hasRunAt: undefined,
       });
     }),
   );
diff --git a/src/component/schema.ts b/src/component/schema.ts
index ae1620b2..5346a714 100644
--- a/src/component/schema.ts
+++ b/src/component/schema.ts
@@ -107,12 +107,6 @@ export default defineSchema({
     // that `segment` is a commit timestamp it has observed, which is what
     // bounds how far the cursor may advance. See `queryPending`.
     scheduledAt: v.optional(v.commitTs()),
-    // @deprecated The exact start time of an entry written by an unreleased
-    // revision. The loop re-keys such entries and clears this on first read.
-    runAt: v.optional(v.number()),
-    // @deprecated An unreleased revision held near-future entries in a
-    // separate index lane marked by this field; cleared like `runAt`.
-    hasRunAt: v.optional(v.boolean()),
   })
     .index("segment", ["segment"])
     .index("scheduledAt", ["scheduledAt"]),

From 93530d4f8520967fcbb74749487c40bf4041319a Mon Sep 17 00:00:00 2001
From: Ian Macartney <366683+ianmacartney@users.noreply.github.com>
Date: Mon, 17 Aug 2026 11:02:00 -0700
Subject: [PATCH 4/9] treat work without a pendingStartId pointer as pending

Work docs written before the pointer existed could be queued or mid-attempt,
and status can't tell which without the old index. Queued is the longer-lived
of the two (it can last until a far-future start; mid-attempt lasts one
attempt, after which the work finishes or retries and sets the pointer), so
report pending rather than running.
---
 CHANGELOG.md               |  5 +++++
 src/component/lib.ts       | 13 +++++++++----
 src/component/loop.test.ts |  8 ++++++++
 3 files changed, 22 insertions(+), 4 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 88cdba39..b59a993c 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -33,6 +33,11 @@
   push a large backlog of scheduled work adds processing overhead and can
   briefly delay work that's ready now. Queued completions and cancelations sort
   first and drain immediately, which is what they want.
+- Work docs now carry a `pendingStartId` pointer to their queue entry, replacing
+  an index. Work enqueued by an older version has no pointer, so until it next
+  starts, `status` reports it as `"pending"` — even if it's mid-attempt (queued
+  is the longer-lived of the two states it could be in) — and canceling it takes
+  effect immediately but only clears its queue entry when that entry comes due.
 - This change is not backwards compatible. It requires `convex` 1.43 or later,
   and downgrading a workpool that has run this version is not supported: an
   older version would read a nanosecond timestamp as a 100ms bucket far in the
diff --git a/src/component/lib.ts b/src/component/lib.ts
index a10868e8..fbae5c2d 100644
--- a/src/component/lib.ts
+++ b/src/component/lib.ts
@@ -215,10 +215,15 @@ async function statusHandler(ctx: QueryCtx, { id }: { id: Id<"work"> }) {
   if (!work) {
     return { state: "finished" } as const;
   }
-  // The pointer can be stale after the work starts; check the entry exists.
-  const pendingStart = work.pendingStartId
-    ? await ctx.db.get("pendingStart", work.pendingStartId)
-    : null;
+  if (work.pendingStartId === undefined) {
+    // Written before the pointer existed (every enqueue sets it now), so this
+    // could be queued or mid-attempt. Report the longer-lived state: queued
+    // can last until a far-future start time, mid-attempt lasts one attempt —
+    // after which the work finishes (correct) or retries (sets the pointer).
+    return { state: "pending", previousAttempts: work.attempts } as const;
+  }
+  // The pointer goes stale when the work starts; check the entry exists.
+  const pendingStart = await ctx.db.get("pendingStart", work.pendingStartId);
   if (pendingStart) {
     return { state: "pending", previousAttempts: work.attempts } as const;
   }
diff --git a/src/component/loop.test.ts b/src/component/loop.test.ts
index 71548989..7ca5c5dc 100644
--- a/src/component/loop.test.ts
+++ b/src/component/loop.test.ts
@@ -427,6 +427,14 @@ describe("loop", () => {
       expect(o.running.map((r) => r.workId)).toEqual([workId]);
     });
 
+    it("reports work that predates the pendingStartId pointer as pending", async () => {
+      await initialize();
+      const workId = await enqueueLegacyWork(Date.now() + 100 * SECOND);
+      // Without the pointer, queued and mid-attempt are indistinguishable;
+      // pending is the longer-lived state, so it's the safer report.
+      expect(await statusOf(workId)).toMatchObject({ state: "pending" });
+    });
+
     it("drops the entry of canceled work that predates the pendingStartId pointer", async () => {
       await initialize();
       const runAt = Date.now() + 100 * SECOND;

From 9f8a528ee1a38c86429719cf86bae99620f58882 Mon Sep 17 00:00:00 2001
From: Ian Macartney <366683+ianmacartney@users.noreply.github.com>
Date: Mon, 17 Aug 2026 11:02:00 -0700
Subject: [PATCH 5/9] update cursor-strategy.md to the implemented protocol

Replaces the two-lane proposal with what shipped: start-time keys with a
commit-ordered scheduledAt sweep, the floor/ceiling cursor rules, the
pendingStartId pointer, and the upgrade behavior.
---
 src/component/cursor-strategy.md | 150 +++++++++++++++++++++++++++++++
 1 file changed, 150 insertions(+)
 create mode 100644 src/component/cursor-strategy.md

diff --git a/src/component/cursor-strategy.md b/src/component/cursor-strategy.md
new file mode 100644
index 00000000..0a179e77
--- /dev/null
+++ b/src/component/cursor-strategy.md
@@ -0,0 +1,150 @@
+# Commit-timestamp cursors for scheduled work
+
+Workpool drains pending work by scanning an ordered index with a cursor. The
+index is ordered by a hybrid of commit timestamp and desired execution time,
+avoiding both tombstone re-scans and out-of-order commits while running work in
+a reasonable order.
+
+## Benefit
+
+Using a benchmark of 5000 mutation tasks (20ms each) with max parallelism of
+200, this had ~20% better throughput and ~20% lower latency than the previous
+rewind-and-rescan approach, with per-iteration cost no longer scaling with
+recent load.
+
+## The tombstone problem
+
+Continuously scanning for the first entries in a table, then deleting them,
+produces a mountain of tombstones (markers of deleted documents) that need to be
+scanned on subsequent queries. We solve this by keeping a cursor per index
+region we walk, jumping ahead to where new entries actually are.
+
+## The out-of-order commit problem
+
+A cursor is only safe if nothing can appear behind it. A sort key chosen from
+the wall clock is chosen when the writing transaction starts, but the row only
+appears when it commits — a slow enqueue can insert a row keyed behind a cursor
+that already moved past it, orphaning the row. The previous design compensated
+by scanning 15 seconds behind the cursor on every query and rescanning the table
+once a minute.
+
+A commit timestamp can't do this: any row not yet visible to a read must have
+committed later, so its key is above any cursor derived from that read. But a
+commit timestamp says nothing about when the work should run.
+
+## The approach
+
+One column, `segment` (nanoseconds), holds either clock, chosen at enqueue:
+
+| When should it run     | `segment`          | `scheduledAt`      |
+| ---------------------- | ------------------ | ------------------ |
+| Now (`runAfter <= 0`)  | `db.vars.commitTs` | `-`                |
+| Later (`runAfter > 0`) | the start time     | `db.vars.commitTs` |
+
+Scheduled entries are keyed at their start time (rounded up to a whole
+millisecond), so they sort above the loop's read bound until due — a bulk
+enqueue of delayed work costs ready work nothing, and no entry is ever
+rewritten. `scheduledAt` records the commit stamp of every wall-clock-keyed
+entry; its absence marks a key as an observed commit timestamp. Retries written
+by the loop get the same shape.
+
+Two indexes: `[segment]` and `[scheduledAt]`. (A third `[workId]` index was
+replaced by a `pendingStartId` pointer on the work document; it can be stale, so
+readers check the entry still exists, and unreachable entries are dropped
+reactively when the scan reads them and finds their work gone or canceled.)
+
+## Reads per iteration
+
+**The sweep** walks `[scheduledAt]` in commit order — the one order nothing can
+land behind — inspecting each entry exactly once, up to 2048 per iteration:
+
+- An entry whose `segment` is at or above the incoming cursor is safe forever:
+  the segment scan can only get past it by reading it. Pass.
+- An entry whose `segment` is behind the cursor was committed by an enqueue that
+  lost the race (commit latency exceeded its delay). The scan will never see it,
+  and it is necessarily due (the cursor never passes the current time), so it
+  starts from here, taking start slots first.
+
+The sweep cursor advances a whole commit stamp at a time (`gt` — a batch enqueue
+shares one stamp, and a bare stamp can't split the group), and only past stamps
+whose behind-the-cursor entries were all retired. Out-of-order entries are rare
+— they require the enqueue's commit latency to exceed its delay — so the common
+case is a read-only pass.
+
+**The segment scan** reads `[segment]` from the incoming cursor up to the end of
+the current millisecond, taking however many start slots the sweep left.
+Everything it returns is due: scheduled entries only become visible at their
+start time. (The one exception is entries written by 0.4.9 and earlier, whose
+`segment` is a 100ms wall-clock bucket — recognized by magnitude, eight orders
+below any nanosecond timestamp — and re-keyed to their start time on first
+read.)
+
+## Cursor rules
+
+The incoming cursor's bounds are plain values, computed where the data is:
+
+- **Floor**: the cursor as read. Every `segment` the loop writes (retries,
+  re-keyed legacy entries) is raised to at least the floor, so nothing lands
+  where the loop already read past. Safe because the write and the cursor
+  advance share a transaction; at enqueue time this comparison would be racy.
+- **Ceiling**: the highest commit timestamp observed while building the batch —
+  completion/cancelation keys, ready entries' keys, sweep stamps, and the
+  previous run's own commit stamp (recorded in `internalState.lastCommitTs`; the
+  next run reads that document, so its snapshot is at least that recent). The
+  cursor never advances past the ceiling: a wall-clock key can exceed every
+  commit stamp that exists, and a commit racing this run would land behind a
+  cursor set there. Nothing anywhere relates wall clocks to the commit clock.
+- **Advance**: to the last entry of the leading run the iteration fully handled
+  (stopping at the first entry left alone, e.g. by capacity), capped by the
+  ceiling and by the lowest `segment` written this transaction (the cursor's
+  final position isn't known when a retry is written), never backwards.
+
+Cursors are inclusive (`gte`): a commit timestamp is unique per transaction, not
+per row, so a batch enqueue writes N rows sharing one key, and an exclusive
+bound would drop the rest of the group whenever capacity cut it short.
+
+The ceiling means a cursor that starts wall-keyed work rests at the freshest
+observed commit stamp rather than the wall key. It catches up one iteration
+later from the loop's own recorded stamp; the lag window contains only
+tombstones of entries already started, so it costs a few re-reads, never delayed
+work.
+
+## Scenarios
+
+**Burst of delayed work.** 10k jobs enqueued with a 60s delay, then one ready
+job. The ready job starts on the first iteration; the delayed jobs sit above the
+read bound, unrewritten, while the sweep verifies them ~2048 per iteration. A
+1s-delayed task enqueued after the burst surfaces at its own start time,
+unaffected by the backlog.
+
+**Saturated pool, scheduled work comes due.** The entry becomes visible to the
+segment scan at its start time and competes in eligibility order as capacity
+frees. The sweep keeps verifying new enqueues even while saturated; only an
+out-of-order entry (which needs a start slot) can hold the sweep cursor, and the
+pool being saturated means the incoming cursor isn't moving, so nothing new
+becomes out-of-order behind it.
+
+**Self-scheduling hourly cron.** Writes its start time as `segment`; the sweep
+reads it once to verify it committed in order, and it's untouched until due.
+
+**Small action retry backoff.** Retries are written from the loop's own
+transaction, raised to the cursor floor, so even a zero-millisecond backoff
+lands readable and starts on the next iteration.
+
+**An enqueue loses the race.** `runAfter(1000)` on an enqueue that takes 1.5s to
+commit lands its entry behind the cursor. The sweep finds it by its commit
+stamp, sees its `segment` below the cursor, and starts it directly — no rewrite,
+at most one sweep-batch behind.
+
+## Upgrading
+
+- Entries written by 0.4.9 hold 100ms buckets, recognized by magnitude and
+  re-keyed on first read: a one-time re-ordering of the queued backlog, 64 per
+  iteration, right after the push.
+- Work docs written by older versions lack `pendingStartId`. Their work runs
+  normally (the scan doesn't use the pointer), but until it next starts,
+  `status` reports it as `"pending"` — queued is the longer-lived of the two
+  states it could be in — and canceling it takes effect immediately while its
+  queue entry is only cleared when it comes due.
+- Downgrading is not supported: an older version reads a nanosecond timestamp as
+  a bucket far in the future and never starts the work.

From 067e734a3a9290c8dd7b10b9f2c90f4ff2aa71a8 Mon Sep 17 00:00:00 2001
From: Ian Macartney <366683+ianmacartney@users.noreply.github.com>
Date: Mon, 17 Aug 2026 19:01:04 -0700
Subject: [PATCH 6/9] 1024

---
 src/component/loop.ts | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/src/component/loop.ts b/src/component/loop.ts
index 4112dd31..190ef0d5 100644
--- a/src/component/loop.ts
+++ b/src/component/loop.ts
@@ -415,7 +415,7 @@ export const run = internalMutation({
 // per-iteration reads. It must exceed the most entries one transaction can
 // stamp identically (a maximal batch enqueue), or the cursor could never
 // clear that stamp.
-const SWEEP_BATCH = 2048;
+const SWEEP_BATCH = 1024;
 
 /** Read the three pending tables the loop processes. */
 async function queryPending(

From 44d7c345010cd6dbbf3dcad9645441a234bc393f Mon Sep 17 00:00:00 2001
From: Ian Macartney <366683+ianmacartney@users.noreply.github.com>
Date: Mon, 17 Aug 2026 19:12:05 -0700
Subject: [PATCH 7/9] page the sweep through a commit stamp larger than one
 batch

The sweep cursor can't split a commit stamp, so a batch enqueue bigger than
one sweep batch could otherwise pin it forever. Paging to the end of the
boundary stamp within the iteration (continuing on the index's implicit
_creationTime) makes the batch size a pure read budget, with no relation to
transaction write limits.
---
 src/component/loop.test.ts | 35 ++++++++++++++++++++++
 src/component/loop.ts      | 60 +++++++++++++++++++++++++++++++-------
 2 files changed, 84 insertions(+), 11 deletions(-)

diff --git a/src/component/loop.test.ts b/src/component/loop.test.ts
index 7ca5c5dc..1346fe02 100644
--- a/src/component/loop.test.ts
+++ b/src/component/loop.test.ts
@@ -901,6 +901,41 @@ describe("loop", () => {
       expect(o.running.map((r) => r.workId)).toEqual([lostId]);
     });
 
+    it("pages through a commit stamp larger than one sweep batch", async () => {
+      await initialize();
+      const runAt = Date.now() + 100 * SECOND;
+      // One transaction stamps every entry identically, and the sweep cursor
+      // can't split a stamp — so it pages to the end of the group (larger
+      // than SWEEP_BATCH = 1024) within one iteration.
+      await t.run(async (ctx) => {
+        const workId = await ctx.db.insert("work", {
+          fnType: "action",
+          fnHandle: "test_handle",
+          fnName: "test_handle",
+          fnArgs: {},
+          attempts: 0,
+        });
+        for (let i = 0; i < 1030; i++) {
+          await ctx.db.insert("pendingStart", {
+            workId,
+            segment: toTimestamp(runAt),
+            scheduledAt: ctx.db.vars.commitTs,
+          });
+        }
+      });
+
+      const first = await runLoop();
+      assert(first.kind === "work");
+      const idle = await runLoop();
+      assert(idle.kind === "idle");
+
+      const o = await observe();
+      expect(o.pendingStart).toHaveLength(1030); // verified, never rewritten
+      expect(o.segmentCursors!.scheduled).toBe(
+        o.pendingStart[0].scheduledAt as bigint,
+      );
+    });
+
     it("caps the cursor at observed commit stamps when starting wall-keyed work", async () => {
       await initialize();
       const runAt = Date.now() + SECOND;
diff --git a/src/component/loop.ts b/src/component/loop.ts
index 190ef0d5..487450e2 100644
--- a/src/component/loop.ts
+++ b/src/component/loop.ts
@@ -410,11 +410,14 @@ export const run = internalMutation({
   },
 });
 
-// How many `scheduledAt` entries the sweep inspects per iteration. Inspection
-// is read-only and each entry is inspected once ever, so this mainly bounds
-// per-iteration reads. It must exceed the most entries one transaction can
-// stamp identically (a maximal batch enqueue), or the cursor could never
-// clear that stamp.
+// How many `scheduledAt` entries the sweep inspects per iteration — a read
+// budget, not a work bound: inspection is read-only and each entry is
+// inspected once ever, because the cursor advances a whole commit stamp at a
+// time and never revisits a cleared stamp. (Advancing inclusively instead
+// would re-read the boundary stamp's group every iteration for as long as its
+// entries wait to come due — unbounded for far-future work.) A stamp group
+// larger than one page is paged through in the same iteration, so this needn't
+// exceed the largest batch enqueue.
 const SWEEP_BATCH = 1024;
 
 /** Read the three pending tables the loop processes. */
@@ -482,7 +485,35 @@ async function queryPending(
       q.and(...excludedIds.map((id) => q.neq(q.field("workId"), id))),
     )
     .take(SWEEP_BATCH);
-  const exhausted = scanned.length < SWEEP_BATCH;
+  let exhausted = scanned.length < SWEEP_BATCH;
+  // A single (very large) batch enqueue can stamp more entries than one page.
+  // The cursor can't split a stamp, so page through to the end of the
+  // boundary stamp within this iteration; the extra reads are bounded by
+  // transaction write limits and paid once ever, like the rest of the sweep.
+  while (
+    !exhausted &&
+    (scanned[0].scheduledAt as bigint) ===
+      (scanned.at(-1)!.scheduledAt as bigint)
+  ) {
+    const last = scanned.at(-1)!;
+    const more = await ctx.db
+      .query("pendingStart")
+      .withIndex("scheduledAt", (q) =>
+        q
+          .eq("scheduledAt", last.scheduledAt)
+          .gt("_creationTime", last._creationTime),
+      )
+      // eslint-disable-next-line @convex-dev/no-filter-in-query
+      .filter((q) =>
+        q.and(...excludedIds.map((id) => q.neq(q.field("workId"), id))),
+      )
+      .take(SWEEP_BATCH);
+    scanned.push(...more);
+    if (more.length < SWEEP_BATCH) {
+      // The boundary stamp is complete; later stamps wait for next iteration.
+      exhausted = true;
+    }
+  }
   const sweep: SweepStep[] = [];
   let oooBudget = startLimit;
   for (let i = 0; i < scanned.length; ) {
@@ -829,8 +860,11 @@ async function handleStart(
         }
         return {
           work,
-          // `segment` is when this became eligible: the time it was scheduled
-          // for, or the commit timestamp of the enqueue if it was ready then.
+          // `segment` round-trips to when this became eligible, in whole
+          // milliseconds: the scheduled start time, or the commit timestamp
+          // of the enqueue if it was ready then. `runAt` only exists for an
+          // entry an older version wrote, whose `segment` is a 100ms bucket
+          // rather than a timestamp.
           lagMs: Date.now() - (runAt ?? fromTimestamp(segment)),
         };
       }),
@@ -967,9 +1001,12 @@ async function rescheduleJob(
   const nextAttempt = withJitter(backoffMs);
   // The backoff can go straight into `segment` however short it is: we're in
   // the transaction that writes the cursor, and raising the key to at least
-  // `floor` keeps the entry readable even for a zero backoff. The key is a
-  // wall-clock time, so it records its commit stamp in `scheduledAt` like any
-  // scheduled enqueue (its absence marks a key as an observed commit stamp).
+  // `floor` keeps the entry readable even for a zero backoff — so unlike an
+  // enqueue, this can't commit out of order and doesn't need the sweep.
+  // `scheduledAt` is still set, for the other thing it means: its absence
+  // marks a `segment` as an observed commit stamp when computing the cursor
+  // ceiling, and this key is a wall-clock time. The sweep just verifies it
+  // once and passes.
   const segment = maxTimestamp(dueTimestamp(Date.now() + nextAttempt), floor);
   const pendingStartId = await ctx.db.insert("pendingStart", {
     workId: work._id,
@@ -996,6 +1033,7 @@ async function getGlobals(ctx: QueryCtx) {
 }
 
 async function getOrCreateState(ctx: MutationCtx) {
+  // TODO: this might be very slow and improved by .order("desc").first() instead.
   const state = await ctx.db.query("internalState").unique();
   if (state) return state;
   const globals = await getGlobals(ctx);

From 69796b0ef41775fd7d43cca24f28eb59e5749b21 Mon Sep 17 00:00:00 2001
From: Ian Macartney <366683+ianmacartney@users.noreply.github.com>
Date: Mon, 17 Aug 2026 19:12:05 -0700
Subject: [PATCH 8/9] document the cursor bounds and scope the segment-bucket
 helpers

Answers from review: retries carry scheduledAt not for the sweep (they can't
commit out of order) but so their wall-clock keys aren't counted as observed
commit stamps in the cursor ceiling; lastCommitTs only bounds cursor
advancement, and without it a pool running only scheduled work would re-read
started entries' tombstones until each completion's commit stamp arrived;
keys stay whole-millisecond because that's the clock's resolution and a
fractional key would start work before any clock reading equals its start
time.

Also scopes the 100ms bucket helpers to their two remaining uses (recovery
pacing and decoding what older versions wrote), moves getNextSegment into
the one test that uses it, and stops passing the ignored segment range to
calculateBacklogAndReport.
---
 src/component/cursor-strategy.md   | 119 ++++++-----
 src/component/loop.test.ts         |  40 ++--
 src/component/loop.ts              | 333 +++++++++++++----------------
 src/component/schema.ts            |  20 +-
 src/component/shared.ts            |  37 ++--
 src/component/stateMachine.test.ts |   6 +-
 src/component/stats.ts             |  10 +-
 7 files changed, 290 insertions(+), 275 deletions(-)

diff --git a/src/component/cursor-strategy.md b/src/component/cursor-strategy.md
index 0a179e77..da0315eb 100644
--- a/src/component/cursor-strategy.md
+++ b/src/component/cursor-strategy.md
@@ -1,9 +1,10 @@
 # Commit-timestamp cursors for scheduled work
 
-Workpool drains pending work by scanning an ordered index with a cursor. The
-index is ordered by a hybrid of commit timestamp and desired execution time,
-avoiding both tombstone re-scans and out-of-order commits while running work in
-a reasonable order.
+Workpool drains pending work by scanning an ordered index with a cursor. This
+change avoids tombstone re-scans without losing out-of-order commits, while
+running work in a reasonable order. It orders the index by a hybrid of commit
+timestamp and desired execution time, with a second cursored index on commits
+using wall-clock time to check for out-of-order entries.
 
 ## Benefit
 
@@ -16,68 +17,78 @@ recent load.
 
 Continuously scanning for the first entries in a table, then deleting them,
 produces a mountain of tombstones (markers of deleted documents) that need to be
-scanned on subsequent queries. We solve this by keeping a cursor per index
-region we walk, jumping ahead to where new entries actually are.
+scanned on subsequent queries.
+
+We solve this by keeping a cursor per index region we walk, jumping ahead to
+where new entries actually are.
 
 ## The out-of-order commit problem
 
 A cursor is only safe if nothing can appear behind it. A sort key chosen from
 the wall clock is chosen when the writing transaction starts, but the row only
-appears when it commits — a slow enqueue can insert a row keyed behind a cursor
-that already moved past it, orphaning the row. The previous design compensated
-by scanning 15 seconds behind the cursor on every query and rescanning the table
-once a minute.
+appears when it commits, so a slow enqueue can insert a row keyed behind a
+cursor that already moved past it, orphaning the row.
+
+The previous design compensated by scanning 15 seconds behind the cursor on
+every query and rescanning the table once a minute.
 
-A commit timestamp can't do this: any row not yet visible to a read must have
-committed later, so its key is above any cursor derived from that read. But a
-commit timestamp says nothing about when the work should run.
+A commit timestamp, on the other hand, is guaranteed to always move forward: any
+row not yet visible to a read must have committed later, so its key is above any
+cursor derived from rows being read.
 
-## The approach
+But a commit timestamp doesn’t always correspond directly to when the work
+should run.
 
-One column, `segment` (nanoseconds), holds either clock, chosen at enqueue:
+## Blending commit time and target time
+
+One column, `segment` (nanoseconds), holds either clock, chosen at enqueue time:
 
 | When should it run     | `segment`          | `scheduledAt`      |
 | ---------------------- | ------------------ | ------------------ |
 | Now (`runAfter <= 0`)  | `db.vars.commitTs` | `-`                |
 | Later (`runAfter > 0`) | the start time     | `db.vars.commitTs` |
-
-Scheduled entries are keyed at their start time (rounded up to a whole
-millisecond), so they sort above the loop's read bound until due — a bulk
-enqueue of delayed work costs ready work nothing, and no entry is ever
-rewritten. `scheduledAt` records the commit stamp of every wall-clock-keyed
-entry; its absence marks a key as an observed commit timestamp. Retries written
-by the loop get the same shape.
-
-Two indexes: `[segment]` and `[scheduledAt]`. (A third `[workId]` index was
-replaced by a `pendingStartId` pointer on the work document; it can be stale, so
-readers check the entry still exists, and unreachable entries are dropped
-reactively when the scan reads them and finds their work gone or canceled.)
-
-## Reads per iteration
-
-**The sweep** walks `[scheduledAt]` in commit order — the one order nothing can
-land behind — inspecting each entry exactly once, up to 2048 per iteration:
-
-- An entry whose `segment` is at or above the incoming cursor is safe forever:
-  the segment scan can only get past it by reading it. Pass.
-- An entry whose `segment` is behind the cursor was committed by an enqueue that
-  lost the race (commit latency exceeded its delay). The scan will never see it,
-  and it is necessarily due (the cursor never passes the current time), so it
-  starts from here, taking start slots first.
-
-The sweep cursor advances a whole commit stamp at a time (`gt` — a batch enqueue
-shares one stamp, and a bare stamp can't split the group), and only past stamps
-whose behind-the-cursor entries were all retired. Out-of-order entries are rare
-— they require the enqueue's commit latency to exceed its delay — so the common
-case is a read-only pass.
-
-**The segment scan** reads `[segment]` from the incoming cursor up to the end of
-the current millisecond, taking however many start slots the sweep left.
-Everything it returns is due: scheduled entries only become visible at their
-start time. (The one exception is entries written by 0.4.9 and earlier, whose
-`segment` is a 100ms wall-clock bucket — recognized by magnitude, eight orders
-below any nanosecond timestamp — and re-keyed to their start time on first
-read.)
+| After retry backoff    | the start time     | `db.vars.commitTs` |
+
+Scheduled entries are keyed at their start time, so they sort above the loop's
+read bound until due. `scheduledAt` records the commit stamp of every
+wall-clock-keyed entry; its absence marks a key as an observed commit timestamp
+(see the ceiling below). A bulk enqueue of delayed work naturally sorts after
+current work, and no entry needs to be rewritten.
+
+Retries get written by the loop, so they can write the start time
+transactionally with advancing the cursor, ensuring they will never be missed by
+writing them ≥ the cursor value — they can't commit out of order, so they don't
+need the sweep's protection. They still carry `scheduledAt` for its other
+meaning: their key is a wall-clock time, and without the stamp it would be
+counted as an observed commit timestamp when computing the cursor ceiling. The
+sweep verifies each retry once and passes.
+
+## Reading jobs in order
+
+Two indexes: `[segment]` (when it’s due) and `[scheduledAt]` (when a future job
+was enqueued).
+
+**The sweep** walks `[scheduledAt]` in commit order, inspecting each
+wall-clock-based entry, up to 1024 per iteration. Out-of-order entries are rare,
+so the common case is a single pass:
+
+- `segment >= segment cursor`: safe, the **segment scan** will read it.
+- `segment < segment cursor`: missed by cursor. Prioritized to start next. It
+  was committed by an enqueue that lost the race (commit latency exceeded its
+  delay). It is necessarily overdue, as the cursor never passes the current
+  time, and was ordered before anything in the segment scan.
+
+Each entry is inspected once ever: the sweep cursor advances a whole commit
+stamp at a time (`gt` — a batch enqueue shares one stamp, and a bare stamp can't
+split the group), and only past stamps whose missed entries were all started.
+Inclusive advancement would instead re-read the boundary stamp's group every
+iteration until its entries come due — unbounded for far-future work. A stamp
+group larger than one batch (a huge batch enqueue) is paged through to its end
+within the same iteration, so the batch size is purely a read budget.
+
+**The segment scan** reads the `[segment]` index for “due” items, from the
+incoming cursor until “now.” Skipped if all available slots were taken by the
+sweep, otherwise does a `take` on for remaining slots.
 
 ## Cursor rules
 
@@ -113,7 +124,7 @@ work.
 
 **Burst of delayed work.** 10k jobs enqueued with a 60s delay, then one ready
 job. The ready job starts on the first iteration; the delayed jobs sit above the
-read bound, unrewritten, while the sweep verifies them ~2048 per iteration. A
+read bound, unrewritten, while the sweep verifies them ~1024 per iteration. A
 1s-delayed task enqueued after the burst surfaces at its own start time,
 unaffected by the backlog.
 
diff --git a/src/component/loop.test.ts b/src/component/loop.test.ts
index 1346fe02..62c05a60 100644
--- a/src/component/loop.test.ts
+++ b/src/component/loop.test.ts
@@ -695,7 +695,7 @@ describe("loop", () => {
       assert(first.kind === "work");
       expect(first.batch.starts).toHaveLength(0);
       expect(first.batch.sweep).toHaveLength(1);
-      expect(first.batch.sweep![0].starts).toHaveLength(0);
+      expect(first.batch.sweep[0].start).toBeUndefined();
 
       let o = await observe();
       expect(o.running).toHaveLength(0);
@@ -704,6 +704,9 @@ describe("loop", () => {
       expect(o.segmentCursors!.scheduled).toBe(
         o.pendingStart[0].scheduledAt as bigint,
       );
+      expect(o.segmentCursors!.scheduledCreationTime).toBe(
+        o.pendingStart[0]._creationTime,
+      );
 
       const second = await runLoop();
       assert(second.kind === "idle");
@@ -901,12 +904,12 @@ describe("loop", () => {
       expect(o.running.map((r) => r.workId)).toEqual([lostId]);
     });
 
-    it("pages through a commit stamp larger than one sweep batch", async () => {
+    it("resumes inside a commit stamp larger than one sweep batch", async () => {
       await initialize();
       const runAt = Date.now() + 100 * SECOND;
-      // One transaction stamps every entry identically, and the sweep cursor
-      // can't split a stamp — so it pages to the end of the group (larger
-      // than SWEEP_BATCH = 1024) within one iteration.
+      // One transaction stamps every entry identically. The cursor's creation
+      // time component lets it rest inside the group (larger than
+      // SWEEP_BATCH = 1024) and resume, inspecting each entry exactly once.
       await t.run(async (ctx) => {
         const workId = await ctx.db.insert("work", {
           fnType: "action",
@@ -924,8 +927,14 @@ describe("loop", () => {
         }
       });
 
-      const first = await runLoop();
+      const first = await runLoop(); // inspects the first 1024
       assert(first.kind === "work");
+      const mid = await observe();
+      expect(mid.segmentCursors!.scheduledCreationTime).toBeLessThan(
+        mid.pendingStart.at(-1)!._creationTime,
+      );
+      const second = await runLoop(); // the remaining 6
+      assert(second.kind === "work");
       const idle = await runLoop();
       assert(idle.kind === "idle");
 
@@ -934,6 +943,9 @@ describe("loop", () => {
       expect(o.segmentCursors!.scheduled).toBe(
         o.pendingStart[0].scheduledAt as bigint,
       );
+      expect(o.segmentCursors!.scheduledCreationTime).toBe(
+        o.pendingStart.at(-1)!._creationTime,
+      );
     });
 
     it("caps the cursor at observed commit stamps when starting wall-keyed work", async () => {
@@ -956,11 +968,10 @@ describe("loop", () => {
   // ────────────────────────────────────────────────────────────────────
   // Cursor invariant: an entry is never left behind the cursor
   //
-  // Nothing rescans a lane behind its cursor, so an entry written below it is
-  // lost for good. Both cases below are ones where the time the loop wants to
-  // write is behind a cursor that holds a commit timestamp from the same
-  // millisecond — the two clocks aren't the same, so the loop places its
-  // writes against the cursor rather than against `Date.now()`.
+  // Nothing rescans the segment index behind its cursor, so a readable entry
+  // written below it would be lost for good. The loop's own writes stay ahead
+  // of the cursor by being strictly in the future — at least the end of the
+  // current millisecond, which the cursor never reaches.
   // ────────────────────────────────────────────────────────────────────
 
   describe("cursor invariant", () => {
@@ -972,8 +983,8 @@ describe("loop", () => {
       await runLoop();
       await simulateCompletion(workId, { kind: "failed", error: "boom" }, 0);
       // A ready entry enqueued after the completion: its commit timestamp is
-      // above the retry's `Date.now()`-derived one, so an unclamped cursor
-      // would advance past the retry in the same iteration that wrote it.
+      // above the retry's `Date.now()`-derived one, so a cursor advancing to
+      // it in the same iteration would strand a same-millisecond retry key.
       const laterId = await enqueueWork();
 
       await runLoop();
@@ -983,7 +994,8 @@ describe("loop", () => {
       assert(retry);
       expect(retry.segment).toBeGreaterThanOrEqual(o.segmentCursors!.incoming);
 
-      // Still readable, so the next iteration starts it.
+      // Keyed at the next millisecond, so it starts as soon as that arrives.
+      vi.advanceTimersByTime(1);
       await runLoop();
       o = await observe();
       expect(o.pendingStart).toHaveLength(0);
diff --git a/src/component/loop.ts b/src/component/loop.ts
index 487450e2..dc5e3d9f 100644
--- a/src/component/loop.ts
+++ b/src/component/loop.ts
@@ -102,22 +102,27 @@ const vStart = v.object({
   // Not a stored field: the start time recovered from an entry an older
   // version wrote (its `segment` is a 100ms bucket, not a timestamp). Absent
   // on anything written by this version — such entries are due when visible.
-  runAt: v.optional(v.number()),
+  legacyStartTime: v.optional(v.number()),
 });
 type Start = Infer<typeof vStart>;
 
 /**
- * One commit stamp's worth of the out-of-order sweep, in commit order. The
- * sweep cursor may advance to `scheduledAt` once every entry in `starts` has
- * been retired; entries that shared the stamp but weren't behind the incoming
- * cursor need nothing (the segment scan will reach them) and aren't carried.
+ * One inspection point of the out-of-order sweep, in commit order. The sweep
+ * cursor may advance to (`scheduledAt`, `creationTime`) once the step's
+ * `start` — an entry whose enqueue committed behind the incoming cursor — has
+ * been retired; runs of in-order entries, which need nothing, fold into a
+ * single start-less step.
  */
 const vSweepStep = v.object({
   scheduledAt: v.int64(),
-  starts: v.array(vStart),
+  creationTime: v.number(),
+  start: v.optional(vStart),
 });
 type SweepStep = Infer<typeof vSweepStep>;
 
+// batch-worker runs `getBatch` and `run` in the same transaction, so a batch
+// never crosses a deploy or a snapshot: `run` sees exactly the state the
+// batch was built from.
 /** The shape `getBatch` hands to `run`. */
 const batchFields = {
   // The segment at query time — what "now" was when the batch was built.
@@ -127,16 +132,11 @@ const batchFields = {
   completions: v.array(vCompletion),
   cancelations: v.array(vCancelation),
   starts: v.array(vStart),
-  // @deprecated Held-lane entries from a batch an unreleased revision built.
-  // Ignored: those rows stay put, and the segment scan re-reads them.
-  scheduled: v.optional(v.array(vStart)),
-  // The out-of-order sweep's findings. Optional so a batch built just before
-  // this field existed still validates.
-  sweep: v.optional(v.array(vSweepStep)),
+  // The out-of-order sweep's findings.
+  sweep: v.array(vSweepStep),
   // The highest commit timestamp observed while building the batch — the
-  // frontier the incoming cursor may advance to, but not beyond. Optional for
-  // the same deploy-boundary reason; `run` falls back to its own last stamp.
-  cursorCeiling: v.optional(v.int64()),
+  // frontier the incoming cursor may advance to, but not beyond.
+  cursorCeiling: v.int64(),
 };
 type Batch = Infer<ReturnType<typeof v.object<typeof batchFields>>>;
 
@@ -173,6 +173,7 @@ export const getBatch = internalQuery({
         cancelationCursor: cursors.cancelation,
         incomingCursor: cursors.incoming,
         sweepCursor: cursors.scheduled ?? 0n,
+        sweepCreationTime: cursors.scheduledCreationTime,
         lastCommitTs: (state?.lastCommitTs as bigint | undefined) ?? 0n,
         maxParallelism: globals.maxParallelism,
         runningCount: running.length,
@@ -257,29 +258,15 @@ export const run = internalMutation({
     const globals = await getGlobals(ctx);
     const console = createLogger(globals.logLevel);
     const segment = getCurrentSegment();
-    const sweep = batch.sweep ?? [];
-    // The bounds on this run's cursor movement, both plain values:
-    // - `floor`: the cursor as it stands. Every `segment` this run writes is
-    //   raised to at least the floor, so nothing lands where the loop already
-    //   read past; the cursor never moves backwards past it.
-    // - `ceiling`: the highest commit timestamp the batch's snapshot had
-    //   observed (computed alongside the batch in `queryPending`). The cursor
-    //   never advances beyond it: a wall-clock key can exceed every commit
-    //   stamp that exists, and a commit racing this run would land behind a
-    //   cursor set there. Nothing relates the two clocks; observed commit
-    //   stamps are the only safe frontier.
-    const floor = state.segmentCursors.incoming;
-    const ceiling =
-      batch.cursorCeiling ?? (state.lastCommitTs as bigint | undefined) ?? 0n;
+    const sweep = batch.sweep;
 
     const compLabel = `[main] pendingCompletion(${batch.completions.length})`;
     console.time(compLabel);
-    const { toCancel, lowestRetry } = await handleCompletions(
+    const toCancel = await handleCompletions(
       ctx,
       state,
       batch.completions,
       console,
-      floor,
     );
     console.timeEnd(compLabel);
 
@@ -303,14 +290,13 @@ export const run = internalMutation({
     // ── Start new work ──
     // The segment scan's entries are all due (scheduled ones only become
     // visible at their start time), except entries written by older versions,
-    // which the loop re-keys to their start time. Re-keying is safe here where
-    // it wasn't at enqueue: this transaction also writes the cursor, so the
-    // key can be placed relative to it. The sweep's entries are the opposite —
-    // committed *behind* the cursor, so the scan will never reach them; they
-    // start from here.
+    // which the loop re-keys to their start time. The sweep's entries are the
+    // opposite — committed *behind* the cursor, so the scan will never reach
+    // them; they start from here.
     const now = Date.now();
-    const isDue = (s: Start) => s.runAt === undefined || s.runAt <= now;
-    const swept = sweep.flatMap((step) => step.starts);
+    const isDue = (s: Start) =>
+      s.legacyStartTime === undefined || s.legacyStartTime <= now;
+    const swept = sweep.flatMap((step) => (step.start ? [step.start] : []));
     const all = [...batch.starts, ...swept];
     const notYet = all.filter((s) => !isDue(s));
     // Oldest first: `segment` is when an entry became eligible, and a swept
@@ -319,14 +305,13 @@ export const run = internalMutation({
       .filter(isDue)
       .sort(
         (a, b) =>
-          (a.runAt ?? fromTimestamp(a.segment)) -
-          (b.runAt ?? fromTimestamp(b.segment)),
+          (a.legacyStartTime ?? fromTimestamp(a.segment)) -
+          (b.legacyStartTime ?? fromTimestamp(b.segment)),
       );
-    let lowestPromoted: bigint | undefined;
     if (notYet.length > 0) {
       const promoteLabel = `[main] promote(${notYet.length})`;
       console.time(promoteLabel);
-      lowestPromoted = await promoteScheduled(ctx, notYet, floor);
+      await promoteScheduled(ctx, notYet);
       console.timeEnd(promoteLabel);
     }
 
@@ -373,32 +358,35 @@ export const run = internalMutation({
     // Capacity can cut the starts short, so only advance each cursor over the
     // leading run we finished with — started or re-keyed. Stopping at the
     // first entry we left alone is what keeps it from being skipped. The
-    // incoming cursor additionally never passes the ceiling, nor any segment
-    // this run wrote — a retry or re-keyed entry lands at or above the floor,
-    // but the cursor's final position isn't known when those writes happen.
-    // Equality is fine throughout: the index is read with `gte`.
+    // incoming cursor additionally never passes `cursorCeiling`, the highest
+    // commit timestamp observed while building the batch: a wall-clock key
+    // can exceed every commit stamp that exists, and a commit racing this run
+    // would land behind a cursor set there. (The keys this run itself writes
+    // need no accounting — they're all at or above the end of the current
+    // millisecond, which no cursor reaches: the scan's bound is exclusive and
+    // the ceiling only lowers.) Equality is fine throughout: the index is
+    // read with `gte`.
     const handled = new Set([...pending, ...notYet].map((s) => s._id));
-    let incoming = floor;
+    let incoming = state.segmentCursors.incoming;
     for (const start of batch.starts) {
       if (!handled.has(start._id)) break;
       incoming = start.segment;
     }
-    for (const cap of [ceiling, lowestRetry, lowestPromoted]) {
-      if (cap !== undefined && cap < incoming) incoming = cap;
-    }
-    state.segmentCursors.incoming = maxTimestamp(floor, incoming);
-    // The sweep cursor moves a whole commit stamp at a time, and only past
-    // stamps whose behind-the-cursor entries were all retired. Its keys are
-    // observed commit stamps, so no ceiling is needed.
-    let sweepCursor = state.segmentCursors.scheduled ?? 0n;
+    // Never backwards: a batch can carry a ceiling below the cursor when it
+    // observed no commit stamps at all (e.g. a recovery-only iteration).
+    state.segmentCursors.incoming = maxTimestamp(
+      state.segmentCursors.incoming,
+      incoming < batch.cursorCeiling ? incoming : batch.cursorCeiling,
+    );
+    // The sweep cursor rests at the last inspected entry — (commit stamp,
+    // creation time), since a batch enqueue shares one stamp — and only moves
+    // past an out-of-order entry once it was retired. Its components are read
+    // straight off inspected entries, so no ceiling is needed.
     for (const step of sweep) {
-      if (step.starts.some((s) => !handled.has(s._id))) break;
-      sweepCursor = step.scheduledAt;
+      if (step.start && !handled.has(step.start._id)) break;
+      state.segmentCursors.scheduled = step.scheduledAt;
+      state.segmentCursors.scheduledCreationTime = step.creationTime;
     }
-    state.segmentCursors.scheduled = maxTimestamp(
-      state.segmentCursors.scheduled ?? 0n,
-      sweepCursor,
-    );
     // Record this run's own commit stamp: the next run reads this document,
     // so its snapshot is at least this recent, and the mark seeds its ceiling.
     state.lastCommitTs = ctx.db.vars.commitTs;
@@ -428,6 +416,7 @@ async function queryPending(
     cancelationCursor,
     incomingCursor,
     sweepCursor,
+    sweepCreationTime,
     lastCommitTs,
     maxParallelism,
     runningCount,
@@ -437,6 +426,7 @@ async function queryPending(
     cancelationCursor: bigint;
     incomingCursor: bigint;
     sweepCursor: bigint;
+    sweepCreationTime: number | undefined;
     lastCommitTs: bigint;
     maxParallelism: number;
     runningCount: number;
@@ -473,75 +463,71 @@ async function queryPending(
   // whose `segment` is still at or above the incoming cursor can be passed
   // forever — the scan can only get past it by reading it. One behind the
   // cursor is due (the cursor never passes the current time) and must start
-  // from here; those claim start slots first. The cursor moves a whole commit
-  // stamp at a time (`gt` — a batch enqueue shares one stamp, and a bare
-  // stamp can't split the group), and only past stamps whose stragglers all
-  // got retired.
-  const scanned = await ctx.db
-    .query("pendingStart")
-    .withIndex("scheduledAt", (q) => q.gt("scheduledAt", sweepCursor))
-    // eslint-disable-next-line @convex-dev/no-filter-in-query
-    .filter((q) =>
-      q.and(...excludedIds.map((id) => q.neq(q.field("workId"), id))),
-    )
-    .take(SWEEP_BATCH);
-  let exhausted = scanned.length < SWEEP_BATCH;
-  // A single (very large) batch enqueue can stamp more entries than one page.
-  // The cursor can't split a stamp, so page through to the end of the
-  // boundary stamp within this iteration; the extra reads are bounded by
-  // transaction write limits and paid once ever, like the rest of the sweep.
-  while (
-    !exhausted &&
-    (scanned[0].scheduledAt as bigint) ===
-      (scanned.at(-1)!.scheduledAt as bigint)
-  ) {
-    const last = scanned.at(-1)!;
-    const more = await ctx.db
-      .query("pendingStart")
-      .withIndex("scheduledAt", (q) =>
-        q
-          .eq("scheduledAt", last.scheduledAt)
-          .gt("_creationTime", last._creationTime),
-      )
+  // from here; those claim start slots first. The cursor is a (commit stamp,
+  // creation time) pair — a batch enqueue shares one stamp, and the creation
+  // time (the index's tiebreak) resumes inside it — so iteration can stop at
+  // any point: at the read budget, or as soon as an out-of-order entry
+  // exceeds the start slots, leaving later entries uninspected rather than
+  // reading work that couldn't start anyway.
+  const sweep: SweepStep[] = [];
+  {
+    let oooBudget = startLimit;
+    let inspected = 0;
+    // Resume inside the cursor's stamp if it stopped mid-stamp, then continue
+    // with later stamps. The first query is empty when the stamp is done.
+    const streams = [
+      ...(sweepCreationTime === undefined
+        ? []
+        : [
+            ctx.db
+              .query("pendingStart")
+              .withIndex("scheduledAt", (q) =>
+                q
+                  .eq("scheduledAt", sweepCursor)
+                  .gt("_creationTime", sweepCreationTime),
+              ),
+          ]),
+      ctx.db
+        .query("pendingStart")
+        .withIndex("scheduledAt", (q) => q.gt("scheduledAt", sweepCursor)),
+    ];
+    scan: for (const stream of streams) {
       // eslint-disable-next-line @convex-dev/no-filter-in-query
-      .filter((q) =>
+      const filtered = stream.filter((q) =>
         q.and(...excludedIds.map((id) => q.neq(q.field("workId"), id))),
-      )
-      .take(SWEEP_BATCH);
-    scanned.push(...more);
-    if (more.length < SWEEP_BATCH) {
-      // The boundary stamp is complete; later stamps wait for next iteration.
-      exhausted = true;
-    }
-  }
-  const sweep: SweepStep[] = [];
-  let oooBudget = startLimit;
-  for (let i = 0; i < scanned.length; ) {
-    const stamp = scanned[i].scheduledAt as bigint;
-    const group = [];
-    while (i < scanned.length && (scanned[i].scheduledAt as bigint) === stamp) {
-      group.push(scanned[i]);
-      i++;
-    }
-    // If the scan was cut mid-stamp, rows beyond it may share this stamp, so
-    // it can't be cleared yet. (Reachable only if one transaction stamped
-    // more entries than SWEEP_BATCH.)
-    if (i === scanned.length && !exhausted) break;
-    const behind = group.filter((r) => (r.segment as bigint) < incomingCursor);
-    if (behind.length > oooBudget) break;
-    oooBudget -= behind.length;
-    const starts = behind.map((r) => ({
-      _id: r._id,
-      workId: r.workId,
-      segment: r.segment as bigint,
-    }));
-    // Fold stamp-only advances into the previous step to keep batches small;
-    // a step is only worth carrying for its starts or as the furthest bound.
-    const previous = sweep.at(-1);
-    if (starts.length === 0 && previous && previous.starts.length === 0) {
-      previous.scheduledAt = stamp;
-    } else {
-      sweep.push({ scheduledAt: stamp, starts });
+      );
+      for await (const row of filtered) {
+        if (inspected >= SWEEP_BATCH) break scan;
+        const behind = (row.segment as bigint) < incomingCursor;
+        if (behind) {
+          if (oooBudget === 0) break scan;
+          oooBudget--;
+        }
+        inspected++;
+        const step: SweepStep = {
+          scheduledAt: row.scheduledAt as bigint,
+          creationTime: row._creationTime,
+          ...(behind
+            ? {
+                start: {
+                  _id: row._id,
+                  workId: row.workId,
+                  segment: row.segment as bigint,
+                },
+              }
+            : {}),
+        };
+        // Fold runs of in-order entries into one step to keep batches small;
+        // a step is only worth carrying for its start or as the furthest
+        // inspection point.
+        const previous = sweep.at(-1);
+        if (!behind && previous && previous.start === undefined) {
+          previous.scheduledAt = step.scheduledAt;
+          previous.creationTime = step.creationTime;
+        } else {
+          sweep.push(step);
+        }
+      }
     }
   }
 
@@ -549,7 +535,7 @@ async function queryPending(
   // slots left over. Everything eligible, oldest first; work scheduled for
   // later sorts above the bound, so it's left alone without pinning the
   // cursor.
-  const sweptCount = sweep.reduce((n, s) => n + s.starts.length, 0);
+  const sweptCount = sweep.filter((s) => s.start !== undefined).length;
   const readyLimit = Math.max(0, startLimit - sweptCount);
   const starts =
     readyLimit === 0
@@ -568,8 +554,9 @@ async function queryPending(
   // source is the stamp of a transaction visible to this snapshot, so
   // anything committing later is stamped above it — the frontier the cursor
   // may advance to. Entries with `scheduledAt` are wall-clock-keyed and don't
-  // count (their stamps do); entries without it carry an observed commit
-  // stamp in `segment` (an older version's tiny buckets can't win a max).
+  // count (their stamps do, via the sweep); entries without it carry an
+  // observed commit stamp in `segment` (an older version's tiny buckets can't
+  // win a max).
   const cursorCeiling = [
     lastCommitTs,
     ...completions.map((c) => c.segment as bigint),
@@ -577,7 +564,7 @@ async function queryPending(
     ...starts
       .filter((s) => s.scheduledAt === undefined)
       .map((s) => s.segment as bigint),
-    ...scanned.map((r) => r.scheduledAt as bigint),
+    ...sweep.map((s) => s.scheduledAt),
   ].reduce(maxTimestamp);
 
   return {
@@ -592,7 +579,7 @@ async function queryPending(
         // An entry from before commit-timestamp ordering keeps its start time
         // in `segment`; recovering it here means `run` handles it like any
         // other not-yet-due entry and re-keys it as a timestamp.
-        runAt: legacyRunAt(segment),
+        legacyStartTime: legacyRunAt(segment),
       };
     }) satisfies Start[],
     sweep,
@@ -603,19 +590,15 @@ async function queryPending(
 /**
  * Handles the completion of pending completions.
  * This only processes work that succeeded or failed, not canceled.
- * Returns the lowest `segment` written for a retry, which bounds how far the
- * incoming cursor may advance this iteration.
  */
 async function handleCompletions(
   ctx: MutationCtx,
   state: Doc<"internalState">,
   completed: Completion[],
   console: Logger,
-  floor: bigint,
 ) {
   // Completions that were going to be retried but have since been canceled.
   const toCancel: CompleteJob[] = [];
-  let lowestRetry: bigint | undefined;
   await Promise.all(
     completed.map(async (c) => {
       await ctx.db.delete("pendingCompletion", c._id);
@@ -634,11 +617,7 @@ async function handleCompletions(
           console.warn(`[main] ${c.workId} is gone, but trying to complete`);
           return;
         }
-        const retriedAt = await rescheduleJob(ctx, work, console, floor);
-        if (retriedAt !== undefined) {
-          if (lowestRetry === undefined || retriedAt < lowestRetry) {
-            lowestRetry = retriedAt;
-          }
+        if (await rescheduleJob(ctx, work, console)) {
           state.report.retries++;
           recordCompleted(console, work, "retrying", undefined);
         } else {
@@ -667,7 +646,7 @@ async function handleCompletions(
   const numCompleted = before - state.running.length;
   state.report.completed += numCompleted;
   console.debug(`[main] completed ${numCompleted} work`);
-  return { toCancel, lowestRetry };
+  return toCancel;
 }
 
 /**
@@ -780,32 +759,23 @@ function maxTimestamp(a: bigint, b: bigint) {
  * so the cursor can pass them and they don't come back until they're actually
  * due. New-format entries are keyed at their start time and only become
  * visible once due, so this only handles the 100ms buckets older versions
- * wrote. Safe to compute from the clock here, unlike at enqueue: this
- * transaction writes the cursor too, and raising the key to at least `floor`
- * keeps the entry readable. Returns the lowest key written, which bounds how
- * far the cursor may advance.
+ * wrote. The new key needs no clamping: a not-yet-due start time rounds up to
+ * at least the end of the current millisecond, which no cursor ever reaches
+ * (the scan's bound is exclusive, and the ceiling can only pull it lower).
  */
-async function promoteScheduled(
-  ctx: MutationCtx,
-  notYet: Start[],
-  floor: bigint,
-): Promise<bigint | undefined> {
-  let lowest: bigint | undefined;
+async function promoteScheduled(ctx: MutationCtx, notYet: Start[]) {
   await Promise.all(
-    notYet.map(async ({ _id, runAt }) => {
+    notYet.map(async ({ _id, legacyStartTime }) => {
       // A concurrent cancelation may have removed it.
       if (!(await ctx.db.get("pendingStart", _id))) return;
-      const segment = maxTimestamp(dueTimestamp(runAt!), floor);
-      if (lowest === undefined || segment < lowest) lowest = segment;
       await ctx.db.patch("pendingStart", _id, {
-        segment,
+        segment: dueTimestamp(legacyStartTime!),
         // The key is now a wall-clock time, and `scheduledAt` is what records
         // that (its absence marks a key as an observed commit stamp).
         scheduledAt: ctx.db.vars.commitTs,
       });
     }),
   );
-  return lowest;
 }
 
 /**
@@ -821,7 +791,7 @@ async function handleStart(
   console.debug(`[main] scheduling ${pending.length} pending work`);
   const starts = (
     await Promise.all(
-      pending.map(async ({ _id, workId, segment, runAt }) => {
+      pending.map(async ({ _id, workId, segment, legacyStartTime }) => {
         if (state.running.some((r) => r.workId === workId)) {
           console.error(`[main] ${workId} already running (skipping start)`);
           // The row is spurious, and nothing rescans the lane behind the
@@ -860,12 +830,12 @@ async function handleStart(
         }
         return {
           work,
-          // `segment` round-trips to when this became eligible, in whole
-          // milliseconds: the scheduled start time, or the commit timestamp
-          // of the enqueue if it was ready then. `runAt` only exists for an
-          // entry an older version wrote, whose `segment` is a 100ms bucket
-          // rather than a timestamp.
-          lagMs: Date.now() - (runAt ?? fromTimestamp(segment)),
+          // `segment` round-trips to when this became eligible: the
+          // scheduled start time, or the commit timestamp of the enqueue if
+          // it was ready then. `legacyStartTime` only exists for an entry an
+          // older version wrote, whose `segment` is a 100ms bucket rather
+          // than a timestamp.
+          lagMs: Date.now() - (legacyStartTime ?? fromTimestamp(segment)),
         };
       }),
     )
@@ -961,14 +931,13 @@ async function beginWorkBatch(
 /**
  * Reschedules a job for retry.
  * If it's been canceled in the mean time, don't retry.
- * @returns the `segment` it was rescheduled at, or undefined if it was not.
+ * @returns true if the job was rescheduled, false if it was not.
  */
 async function rescheduleJob(
   ctx: MutationCtx,
   work: Doc<"work">,
   console: Logger,
-  floor: bigint,
-): Promise<bigint | undefined> {
+): Promise<boolean> {
   const pendingCancelation = await ctx.db
     .query("pendingCancelation")
     .withIndex("workId", (q) => q.eq("workId", work._id))
@@ -976,14 +945,14 @@ async function rescheduleJob(
   if (pendingCancelation) {
     // If there's an un-processed cancelation request, don't retry.
     console.warn(`[main] ${work._id} in pendingCancelation so not retrying`);
-    return undefined;
+    return false;
   }
   if (work.canceled) {
-    return undefined;
+    return false;
   }
   if (!work.retryBehavior) {
     console.warn(`[main] ${work._id} has no retryBehavior so not retrying`);
-    return undefined;
+    return false;
   }
   // The pointer can be stale (its entry started); check before declaring a
   // duplicate.
@@ -993,28 +962,28 @@ async function rescheduleJob(
   ) {
     // Not sure why this would ever happen, but ensure uniqueness explicitly.
     console.error(`[main] ${work._id} already in pendingStart so not retrying`);
-    return undefined;
+    return false;
   }
   const backoffMs =
     work.retryBehavior.initialBackoffMs *
     Math.pow(work.retryBehavior.base, work.attempts - 1);
   const nextAttempt = withJitter(backoffMs);
-  // The backoff can go straight into `segment` however short it is: we're in
-  // the transaction that writes the cursor, and raising the key to at least
-  // `floor` keeps the entry readable even for a zero backoff — so unlike an
-  // enqueue, this can't commit out of order and doesn't need the sweep.
-  // `scheduledAt` is still set, for the other thing it means: its absence
-  // marks a `segment` as an observed commit stamp when computing the cursor
-  // ceiling, and this key is a wall-clock time. The sweep just verifies it
-  // once and passes.
-  const segment = maxTimestamp(dueTimestamp(Date.now() + nextAttempt), floor);
+  // The key needs no clamping against the cursor: it's strictly in the
+  // future, so it rounds up to at least the end of the current millisecond,
+  // which no cursor ever reaches — the scan's bound is exclusive and the
+  // ceiling can only pull it lower. So unlike an enqueue, this can't land
+  // behind the cursor and doesn't need the sweep. `scheduledAt` is still set,
+  // for the other thing it means: its absence marks a `segment` as an
+  // observed commit stamp when computing the cursor ceiling, and this key is
+  // a wall-clock time. The sweep just verifies it once and passes.
+  const segment = dueTimestamp(Date.now() + Math.max(nextAttempt, 1));
   const pendingStartId = await ctx.db.insert("pendingStart", {
     workId: work._id,
     segment,
     scheduledAt: ctx.db.vars.commitTs,
   });
   await ctx.db.patch("work", work._id, { pendingStartId });
-  return segment;
+  return true;
 }
 
 export function withJitter(delay: number) {
diff --git a/src/component/schema.ts b/src/component/schema.ts
index 5346a714..f2211610 100644
--- a/src/component/schema.ts
+++ b/src/component/schema.ts
@@ -46,11 +46,23 @@ export default defineSchema({
       // sweep. Optional because it didn't always exist; absent means "from the
       // beginning".
       scheduled: v.optional(timestamp),
+      // The `_creationTime` of the last entry the sweep inspected at the
+      // `scheduled` stamp, letting the cursor rest inside a stamp shared by a
+      // whole batch enqueue (`_creationTime` is the index's tiebreak). Absent
+      // means the stamp was fully inspected.
+      scheduledCreationTime: v.optional(v.number()),
     }),
-    // The commit timestamp of the previous `run`. Everything this run read is
-    // at least this recent, so the `incoming` cursor may advance up to the
-    // highest commit timestamp observed — but no further, or a commit racing
-    // this run could land behind it. See `run`'s cursor advance.
+    // The commit timestamp of the previous `run`. It says nothing about what
+    // that run had read (commits can land between its snapshot and its
+    // commit) — but the *next* run reads this document, so its snapshot
+    // includes every transaction stamped at or below this value, making it a
+    // safe addition to that run's cursor ceiling (see `run`). It only bounds
+    // cursor advancement. Without it the ceiling would come only from commit
+    // stamps observed in the batch itself, which still converges — every
+    // start eventually produces a commit-stamped completion — but a pool
+    // running only scheduled (wall-clock-keyed) work would leave its cursor
+    // behind each entry it starts, re-reading those tombstones until the
+    // completion arrives. This closes that gap by the very next iteration.
     lastCommitTs: v.optional(v.commitTs()),
     // Unlike the cursors, this stays a 100ms "segment": it only paces how often
     // the loop checks for stuck jobs.
diff --git a/src/component/shared.ts b/src/component/shared.ts
index 4cc21f66..04ec80ca 100644
--- a/src/component/shared.ts
+++ b/src/component/shared.ts
@@ -19,6 +19,13 @@ export const HOUR = 60 * MINUTE;
 export const DAY = 24 * HOUR;
 export const YEAR = 365 * DAY;
 
+// ── 100ms segment buckets ────────────────────────────────────────────────
+// The unit versions ≤ 0.4.9 ordered the pending queues by. Kept for exactly
+// two purposes: pacing the periodic stuck-job check (`lastRecovery`), and
+// decoding the buckets those versions left in `pendingStart.segment` (see
+// `legacyRunAt`). Never use these for ordering keys — keys are nanoseconds on
+// the commit-timestamp scale (see `toTimestamp`).
+
 export function toSegment(ms: number): bigint {
   return BigInt(Math.floor(ms / SEGMENT_MS));
 }
@@ -27,10 +34,6 @@ export function getCurrentSegment(): bigint {
   return toSegment(Date.now());
 }
 
-export function getNextSegment(): bigint {
-  return toSegment(Date.now()) + 1n;
-}
-
 export function fromSegment(segment: bigint): number {
   return Number(segment) * SEGMENT_MS;
 }
@@ -41,25 +44,31 @@ export function fromSegment(segment: bigint): number {
 const NS_PER_MS = 1_000_000n;
 
 /**
- * A wall-clock time on the commit-timestamp scale. `Date.now()` is whole
- * milliseconds in Convex, so nothing is lost converting it; the floor only
- * rounds a caller-supplied fractional `runAt`, and never upward, so work can't
- * be held past its time.
+ * A wall-clock time on the commit-timestamp scale, preserving any fractional
+ * milliseconds exactly: the whole and fractional parts convert separately, so
+ * no precision is lost multiplying a large float. Round-trips through
+ * `fromTimestamp`.
  */
 export function toTimestamp(ms: number): bigint {
-  return BigInt(Math.floor(ms)) * NS_PER_MS;
+  const whole = Math.floor(ms);
+  return BigInt(whole) * NS_PER_MS + BigInt(Math.round((ms - whole) * 1e6));
 }
 
-/** Back to whole milliseconds, the resolution `Date.now()` reports. */
+/**
+ * Back to (possibly fractional) milliseconds, dividing the whole and
+ * remainder parts separately so large values don't lose precision.
+ */
 export function fromTimestamp(timestamp: bigint): number {
-  return Number(timestamp / NS_PER_MS);
+  return Number(timestamp / NS_PER_MS) + Number(timestamp % NS_PER_MS) / 1e6;
 }
 
 /**
  * A start time as an ordering value: rounded up to the next whole millisecond,
- * the first one the clock can call it due. Rounding down would make an entry
- * readable while `isDue` still says no, so the loop would re-read it until the
- * clock leaves the current millisecond.
+ * the first one in which the work is due for the whole millisecond's duration.
+ * Rounding down would start a fractionally-scheduled entry before its time
+ * (due-ness is visibility, and the read bound moves in whole milliseconds).
+ * If sub-millisecond earliness stops mattering — e.g. once near-future starts
+ * are clamped to "now" — this could carry the fraction instead.
  */
 export function dueTimestamp(runAt: number): bigint {
   return toTimestamp(Math.ceil(runAt));
diff --git a/src/component/stateMachine.test.ts b/src/component/stateMachine.test.ts
index 16e0e782..1ab1fb75 100644
--- a/src/component/stateMachine.test.ts
+++ b/src/component/stateMachine.test.ts
@@ -15,7 +15,6 @@ import {
   DEFAULT_MAX_PARALLELISM,
   fromSegment,
   getCurrentSegment,
-  getNextSegment,
   SECOND,
   WORKER_NAME,
 } from "./shared.js";
@@ -167,6 +166,11 @@ const S12_CANCELED_AWAITING_COMPLETE: CompositeState = {
 
 const ACTION_RECOVERY_THRESHOLD_MS = 5 * 60 * 1000;
 
+/** The bucket after the current one, as older versions computed it. */
+function getNextSegment(): bigint {
+  return getCurrentSegment() + 1n;
+}
+
 describe("state machine", () => {
   let t: ReturnType<typeof setupTest>;
 
diff --git a/src/component/stats.ts b/src/component/stats.ts
index 5e256d01..76fa86dc 100644
--- a/src/component/stats.ts
+++ b/src/component/stats.ts
@@ -9,7 +9,6 @@ import {
   type Config,
   DEFAULT_MAX_PARALLELISM,
   endOfMs,
-  getCurrentSegment,
   WORKER_NAME,
 } from "./shared.js";
 import { createLogger, type Logger, logLevel, shouldLog } from "./logging.js";
@@ -78,7 +77,6 @@ export async function generateReport(
     // Don't waste time if we're not going to log.
     return;
   }
-  const currentSegment = getCurrentSegment();
   // Backlog is work that's eligible now; anything scheduled for later sorts
   // above the current time and isn't waiting on us.
   const pendingStart = await paginator(ctx.db, schema)
@@ -100,8 +98,6 @@ export async function generateReport(
     });
   } else {
     await ctx.scheduler.runAfter(0, internal.stats.calculateBacklogAndReport, {
-      startSegment: 0n,
-      endSegment: currentSegment,
       cursor: pendingStart.continueCursor,
       report: state.report,
       running: state.running.length,
@@ -112,8 +108,10 @@ export async function generateReport(
 
 export const calculateBacklogAndReport = internalMutation({
   args: {
-    startSegment: v.int64(),
-    endSegment: v.int64(),
+    // @deprecated Unused; accepted so in-flight calls from older versions
+    // still validate.
+    startSegment: v.optional(v.int64()),
+    endSegment: v.optional(v.int64()),
     cursor: v.string(),
     report: schema.tables.internalState.validator.fields.report,
     running: v.number(),

From 24c895d5979146c70636e83f7697dd23fa8704b5 Mon Sep 17 00:00:00 2001
From: Ian Macartney <366683+ianmacartney@users.noreply.github.com>
Date: Tue, 18 Aug 2026 00:38:56 -0700
Subject: [PATCH 9/9] update doc

---
 src/component/cursor-strategy.md | 58 ++++++++++++++++++++------------
 1 file changed, 36 insertions(+), 22 deletions(-)

diff --git a/src/component/cursor-strategy.md b/src/component/cursor-strategy.md
index da0315eb..a8d505d8 100644
--- a/src/component/cursor-strategy.md
+++ b/src/component/cursor-strategy.md
@@ -55,13 +55,15 @@ wall-clock-keyed entry; its absence marks a key as an observed commit timestamp
 (see the ceiling below). A bulk enqueue of delayed work naturally sorts after
 current work, and no entry needs to be rewritten.
 
-Retries get written by the loop, so they can write the start time
-transactionally with advancing the cursor, ensuring they will never be missed by
-writing them ≥ the cursor value — they can't commit out of order, so they don't
-need the sweep's protection. They still carry `scheduledAt` for its other
-meaning: their key is a wall-clock time, and without the stamp it would be
-counted as an observed commit timestamp when computing the cursor ceiling. The
-sweep verifies each retry once and passes.
+Retries get written by the loop, so they can't be missed without any explicit
+comparison against the cursor: their key is strictly in the future (a backoff
+rounds up to at least the end of the current millisecond), and no cursor ever
+reaches the end of the current millisecond — the scan's bound is exclusive and
+the ceiling only pulls it lower. So unlike an enqueue, a retry can't commit out
+of order and doesn't need the sweep's protection. It still carries `scheduledAt`
+for its other meaning: its key is a wall-clock time, and without the stamp it
+would be counted as an observed commit timestamp when computing the cursor
+ceiling. The sweep verifies each retry once and passes.
 
 ## Reading jobs in order
 
@@ -78,13 +80,12 @@ so the common case is a single pass:
   delay). It is necessarily overdue, as the cursor never passes the current
   time, and was ordered before anything in the segment scan.
 
-Each entry is inspected once ever: the sweep cursor advances a whole commit
-stamp at a time (`gt` — a batch enqueue shares one stamp, and a bare stamp can't
-split the group), and only past stamps whose missed entries were all started.
-Inclusive advancement would instead re-read the boundary stamp's group every
-iteration until its entries come due — unbounded for far-future work. A stamp
-group larger than one batch (a huge batch enqueue) is paged through to its end
-within the same iteration, so the batch size is purely a read budget.
+Each entry is inspected once ever: the sweep cursor is a (commit stamp, creation
+time) pair — a batch enqueue shares one stamp, and the creation time (the
+index's implicit tiebreak) resumes inside it — so already-inspected entries are
+never re-read, and iteration can stop at any point: at the read budget, or as
+soon as a missed entry exceeds the available start slots, rather than reading
+entries that couldn't start anyway.
 
 **The segment scan** reads the `[segment]` index for “due” items, from the
 incoming cursor until “now.” Skipped if all available slots were taken by the
@@ -94,10 +95,12 @@ sweep, otherwise does a `take` on for remaining slots.
 
 The incoming cursor's bounds are plain values, computed where the data is:
 
-- **Floor**: the cursor as read. Every `segment` the loop writes (retries,
-  re-keyed legacy entries) is raised to at least the floor, so nothing lands
-  where the loop already read past. Safe because the write and the cursor
-  advance share a transaction; at enqueue time this comparison would be racy.
+- **Loop writes stay ahead by construction**: every `segment` the loop writes
+  (retries, re-keyed legacy entries) is a strictly-future start time, rounded up
+  to at least the end of the current millisecond — which no cursor ever reaches,
+  since the scan's bound is exclusive and the ceiling only pulls it lower. No
+  comparison against the cursor is needed; at enqueue time even that wouldn't be
+  available.
 - **Ceiling**: the highest commit timestamp observed while building the batch —
   completion/cancelation keys, ready entries' keys, sweep stamps, and the
   previous run's own commit stamp (recorded in `internalState.lastCommitTs`; the
@@ -107,8 +110,8 @@ The incoming cursor's bounds are plain values, computed where the data is:
   cursor set there. Nothing anywhere relates wall clocks to the commit clock.
 - **Advance**: to the last entry of the leading run the iteration fully handled
   (stopping at the first entry left alone, e.g. by capacity), capped by the
-  ceiling and by the lowest `segment` written this transaction (the cursor's
-  final position isn't known when a retry is written), never backwards.
+  ceiling, and never backwards (a batch that observed no commit stamps at all —
+  e.g. a recovery-only iteration — carries a ceiling below the cursor).
 
 Cursors are inclusive (`gte`): a commit timestamp is unique per transaction, not
 per row, so a batch enqueue writes N rows sharing one key, and an exclusive
@@ -139,14 +142,25 @@ becomes out-of-order behind it.
 reads it once to verify it committed in order, and it's untouched until due.
 
 **Small action retry backoff.** Retries are written from the loop's own
-transaction, raised to the cursor floor, so even a zero-millisecond backoff
-lands readable and starts on the next iteration.
+transaction, keyed at least one millisecond in the future, so even a
+zero-millisecond backoff lands ahead of the cursor and starts as soon as its
+millisecond arrives.
 
 **An enqueue loses the race.** `runAfter(1000)` on an enqueue that takes 1.5s to
 commit lands its entry behind the cursor. The sweep finds it by its commit
 stamp, sees its `segment` below the cursor, and starts it directly — no rewrite,
 at most one sweep-batch behind.
 
+## Future ideas
+
+- Clamp near-future starts (e.g. within ~100–250ms) to "now": they'd be
+  commit-keyed with no `scheduledAt` to sweep, at the cost of starting up to
+  that much early (or paired with an execution-side delay to keep exact start
+  times).
+- Pack many pending starts sharing a commit stamp into one document (each worker
+  patches itself out on start, found via `pendingStartId`), cutting write/delete
+  churn and document counts for large batch enqueues.
+
 ## Upgrading
 
 - Entries written by 0.4.9 hold 100ms buckets, recognized by magnitude and