diff --git a/CHANGELOG.md b/CHANGELOG.md index 5dfe49f0..b59a993c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,23 +11,33 @@ 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 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/cursor-strategy.md b/src/component/cursor-strategy.md new file mode 100644 index 00000000..a8d505d8 --- /dev/null +++ b/src/component/cursor-strategy.md @@ -0,0 +1,175 @@ +# Commit-timestamp cursors for scheduled work + +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 + +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, 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, 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. + +But a commit timestamp doesn’t always correspond directly to when the work +should run. + +## 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` | +| 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'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 + +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 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 +sweep, otherwise does a `take` on for remaining slots. + +## Cursor rules + +The incoming cursor's bounds are plain values, computed where the data is: + +- **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 + 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 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 +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 ~1024 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, 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 + 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. 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 a5afd208..fbae5c2d 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,21 +111,20 @@ 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", { + const pendingStartId = 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 } : {}), }); + await ctx.db.patch("work", workId, { pendingStartId }); recordEnqueued(console, { workId, fnName: workArgs.fnName, runAt }); return workId; } @@ -217,10 +215,15 @@ 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(); + 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 cd3bdffd..62c05a60 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, - SAFE_FUTURE_MS, 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; @@ -448,6 +415,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()); @@ -458,6 +426,34 @@ describe("loop", () => { expect(o.pendingStart).toHaveLength(0); 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; + 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", () => { @@ -550,6 +546,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 +672,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 +684,35 @@ 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].start).toBeUndefined(); 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, + ); + expect(o.segmentCursors!.scheduledCreationTime).toBe( + o.pendingStart[0]._creationTime, + ); 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 +726,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 +742,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 +758,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 +804,19 @@ 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 }, @@ -822,18 +825,153 @@ 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([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> { + return t.run(async (ctx) => { + const workId = await ctx.db.insert("work", { + fnType: "action", + fnHandle: "test_handle", + fnName: "test_handle", + fnArgs: {}, + attempts: 0, + }); + const pendingStartId = await ctx.db.insert("pendingStart", { + workId, + segment: cursor - 1n, + scheduledAt: ctx.db.vars.commitTs, + }); + await ctx.db.patch("work", workId, { pendingStartId }); + 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 }, + 0, + ); + await runLoop(); + const o = await observe(); + expect(o.pendingStart).toHaveLength(0); + expect(o.running.map((r) => r.workId)).toEqual([lostId]); + }); + + 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. 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", + 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(); // 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"); + + const o = await observe(); + expect(o.pendingStart).toHaveLength(1030); // verified, never rewritten + 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 () => { + 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)); }); }); // ──────────────────────────────────────────────────────────────────── // 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", () => { @@ -845,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(); @@ -856,27 +994,29 @@ 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); 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..dc5e3d9f 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"; @@ -99,10 +99,30 @@ const vStart = v.object({ _id: v.id("pendingStart"), workId: v.id("work"), segment: v.int64(), - runAt: v.optional(v.number()), + // 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. + legacyStartTime: v.optional(v.number()), }); type Start = Infer; +/** + * 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(), + creationTime: v.number(), + start: v.optional(vStart), +}); +type SweepStep = Infer; + +// 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. @@ -112,10 +132,11 @@ 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. - scheduled: v.optional(v.array(vStart)), + // 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. + cursorCeiling: v.int64(), }; type Batch = Infer>>; @@ -146,24 +167,26 @@ 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, + sweepCreationTime: cursors.scheduledCreationTime, + 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 +206,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,8 +258,7 @@ 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; const compLabel = `[main] pendingCompletion(${batch.completions.length})`; console.time(compLabel); @@ -247,7 +267,6 @@ export const run = internalMutation({ state, batch.completions, console, - incomingGuard, ); console.timeEnd(compLabel); @@ -269,29 +288,30 @@ 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. 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 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 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( (a, b) => - (a.runAt ?? fromTimestamp(a.segment)) - - (b.runAt ?? fromTimestamp(b.segment)), + (a.legacyStartTime ?? fromTimestamp(a.segment)) - + (b.legacyStartTime ?? fromTimestamp(b.segment)), ); if (notYet.length > 0) { const promoteLabel = `[main] promote(${notYet.length})`; console.time(promoteLabel); - await promoteScheduled(ctx, notYet, incomingGuard); + await promoteScheduled(ctx, notYet); console.timeEnd(promoteLabel); } @@ -335,25 +355,41 @@ 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 `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 = state.segmentCursors.incoming; 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; + // 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.start && !handled.has(step.start._id)) break; + state.segmentCursors.scheduled = step.scheduledAt; + state.segmentCursors.scheduledCreationTime = step.creationTime; } - state.segmentCursors.scheduled = scheduledGuard.advance(scheduledCursor); + // 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 +398,16 @@ export const run = internalMutation({ }, }); +// 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. */ async function queryPending( ctx: QueryCtx, @@ -369,7 +415,9 @@ async function queryPending( completionCursor, cancelationCursor, incomingCursor, - scheduledCursor, + sweepCursor, + sweepCreationTime, + lastCommitTs, maxParallelism, runningCount, eligibleBefore, @@ -377,7 +425,9 @@ async function queryPending( completionCursor: bigint; cancelationCursor: bigint; incomingCursor: bigint; - scheduledCursor: bigint; + sweepCursor: bigint; + sweepCreationTime: number | undefined; + lastCommitTs: bigint; maxParallelism: number; runningCount: number; eligibleBefore: bigint; @@ -402,48 +452,121 @@ 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 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 + const filtered = stream.filter((q) => + q.and(...excludedIds.map((id) => q.neq(q.field("workId"), id))), + ); + 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); + } + } + } + } + + // 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.filter((s) => s.start !== undefined).length; + 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, 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), + ...cancelations.map((c) => c.segment as bigint), + ...starts + .filter((s) => s.scheduledAt === undefined) + .map((s) => s.segment as bigint), + ...sweep.map((s) => s.scheduledAt), + ].reduce(maxTimestamp); + return { completions, cancelations, @@ -455,16 +578,12 @@ async function queryPending( 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. - runAt: s.runAt ?? legacyRunAt(segment), + // other not-yet-due entry and re-keys it as a timestamp. + legacyStartTime: 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, }; } @@ -477,7 +596,6 @@ async function handleCompletions( state: Doc<"internalState">, completed: Completion[], console: Logger, - guard: CursorGuard, ) { // Completions that were going to be retried but have since been canceled. const toCancel: CompleteJob[] = []; @@ -499,8 +617,7 @@ 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) { + if (await rescheduleJob(ctx, work, console)) { state.report.retries++; recordCompleted(console, work, "retrying", undefined); } else { @@ -562,23 +679,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); @@ -642,69 +755,24 @@ 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`. + * 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 only handles the 100ms buckets older versions + * 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). */ -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; - -/** - * 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. - */ -async function promoteScheduled( - ctx: MutationCtx, - notYet: Start[], - guard: CursorGuard, -) { +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; 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. - hasRunAt: undefined, + 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, }); }), ); @@ -723,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 @@ -743,11 +811,31 @@ 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 - // for, or the commit timestamp of the enqueue if it was ready then. - 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)), }; }), ) @@ -849,7 +937,6 @@ async function rescheduleJob( ctx: MutationCtx, work: Doc<"work">, console: Logger, - guard: CursorGuard, ): Promise { const pendingCancelation = await ctx.db .query("pendingCancelation") @@ -867,11 +954,12 @@ async function rescheduleJob( console.warn(`[main] ${work._id} has no retryBehavior so not retrying`); return false; } - 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 false; @@ -880,14 +968,21 @@ async function rescheduleJob( 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. - await ctx.db.insert("pendingStart", { + // 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: guard.place(toTimestamp(Date.now() + nextAttempt)), + segment, + scheduledAt: ctx.db.vars.commitTs, }); + await ctx.db.patch("work", work._id, { pendingStartId }); return true; } @@ -907,6 +1002,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); diff --git a/src/component/schema.ts b/src/component/schema.ts index 9c5f7191..f2211610 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,28 @@ 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 `_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`. 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. lastRecovery: v.int64(), @@ -79,29 +98,30 @@ 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`. 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. - 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. - hasRunAt: v.optional(v.boolean()), + // 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()), }) - .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..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,18 +44,34 @@ 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 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)); } /** @@ -83,22 +102,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/stateMachine.test.ts b/src/component/stateMachine.test.ts index ddace386..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; @@ -266,10 +270,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 +312,10 @@ describe("state machine", () => { async function observeState(workId: Id<"work">): Promise { 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 +968,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 +1164,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 }; 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..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,15 +77,12 @@ 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 — 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())), ) @@ -102,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, @@ -114,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(),