From 1fdf1eef4a220013f66efa695045bad20eeb47fb Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Sun, 30 Aug 2026 14:07:27 -0700 Subject: [PATCH 1/2] Add tests for pending seed listDue paging Outstanding rows past the scan limit currently starve behind an unordered LIMIT. These tests pin oldest-due order, keyset continuation, and DrainReport.truncated versus a complete drain. --- packages/onboarding/src/pending-seed.test.ts | 58 +++++++++++++-- .../test/bench-provisioning.test.ts | 72 +++++++++++++++++-- 2 files changed, 122 insertions(+), 8 deletions(-) diff --git a/packages/onboarding/src/pending-seed.test.ts b/packages/onboarding/src/pending-seed.test.ts index a90d1ce5d..0985f9b0d 100644 --- a/packages/onboarding/src/pending-seed.test.ts +++ b/packages/onboarding/src/pending-seed.test.ts @@ -136,8 +136,10 @@ describe("createInMemoryPendingSeedStore", () => { const due = await store.listDue({}); - expect(due).toHaveLength(2); - expect(due).toEqual( + expect(due.truncated).toBe(false); + expect(due).not.toHaveProperty("next"); + expect(due.seeds).toHaveLength(2); + expect(due.seeds).toEqual( expect.arrayContaining([ SEED, { @@ -162,7 +164,10 @@ describe("createInMemoryPendingSeedStore", () => { clock = PENDING_SEED_TTL_MS + 1; const due = await store.listDue({ now: () => clock }); - expect(due).toEqual([{ ...SEED, userId: "user_2", tenantId: "ten_2" }]); + expect(due.truncated).toBe(false); + expect(due.seeds).toEqual([ + { ...SEED, userId: "user_2", tenantId: "ten_2" }, + ]); // Swept, not merely skipped — the expired row is gone for good. expect( await store.read({ userId: "user_1", tenantId: "ten_1", now: () => 0 }), @@ -181,7 +186,52 @@ describe("createInMemoryPendingSeedStore", () => { const due = await store.listDue({ limit: 2 }); - expect(due).toHaveLength(2); + expect(due.seeds).toHaveLength(2); + expect(due.truncated).toBe(true); + expect(due.next).toEqual({ + expiresAt: expect.any(Date), + userId: expect.any(String), + tenantId: expect.any(String), + }); + }); + + test("listDue returns oldest-due first and continues from an (expiresAt, userId, tenantId) cursor", async () => { + const store = createInMemoryPendingSeedStore(testCipher()); + for (let index = 0; index < 5; index += 1) { + await store.put( + { + ...SEED, + userId: `user_${index}`, + tenantId: `ten_${index}`, + }, + { ttlMs: 5_000 - index }, + ); + } + + const first = await store.listDue({ limit: 2 }); + expect(first.seeds.map((seed) => seed.userId)).toEqual([ + "user_4", + "user_3", + ]); + expect(first.truncated).toBe(true); + + const second = await store.listDue({ + limit: 2, + ...(first.next !== undefined ? { after: first.next } : {}), + }); + expect(second.seeds.map((seed) => seed.userId)).toEqual([ + "user_2", + "user_1", + ]); + expect(second.truncated).toBe(true); + + const third = await store.listDue({ + limit: 2, + ...(second.next !== undefined ? { after: second.next } : {}), + }); + expect(third.seeds.map((seed) => seed.userId)).toEqual(["user_0"]); + expect(third.truncated).toBe(false); + expect(third).not.toHaveProperty("next"); }); test("cleared on successful seed — the row is gone after clear", async () => { diff --git a/packages/onboarding/test/bench-provisioning.test.ts b/packages/onboarding/test/bench-provisioning.test.ts index aa8ffa70a..ac595086e 100644 --- a/packages/onboarding/test/bench-provisioning.test.ts +++ b/packages/onboarding/test/bench-provisioning.test.ts @@ -16,6 +16,7 @@ import { } from "../src/bench-provisioning"; import { createInMemoryPendingSeedStore, + PENDING_SEED_SCAN_LIMIT, type PendingSeed, type PendingSeedStore, } from "../src/pending-seed"; @@ -95,7 +96,7 @@ describe("createBenchProvisioner", () => { expect(calls.ensureSeeded).toBe(1); expect(deployedByTenant.get("ten_1")).toEqual(ALL_WORKFLOWS); - expect(report).toMatchObject({ converged: 1 }); + expect(report).toMatchObject({ converged: 1, truncated: false }); expect( await store.read({ userId: "user_1", tenantId: "ten_1" }), ).toBeUndefined(); @@ -122,7 +123,7 @@ describe("createBenchProvisioner", () => { const report = await provisioner.drainOnce(); expect(calls.ensureSeeded).toBe(0); - expect(report).toMatchObject({ converged: 1 }); + expect(report).toMatchObject({ converged: 1, truncated: false }); expect( await store.read({ userId: "user_1", tenantId: "ten_1" }), ).toBeUndefined(); @@ -211,7 +212,7 @@ describe("createBenchProvisioner", () => { expect(calls.ensureSeeded).toBe(1); expect(deployedByTenant.get("ten_1")).toEqual(ALL_WORKFLOWS); - expect(report).toMatchObject({ converged: 1 }); + expect(report).toMatchObject({ converged: 1, truncated: false }); }); test("overlapping drains never double-deploy the same bench", async () => { @@ -259,6 +260,69 @@ describe("createBenchProvisioner", () => { const report = await provisioner.drainOnce(); expect(calls.ensureSeeded).toBe(2); - expect(report).toMatchObject({ converged: 2 }); + expect(report).toMatchObject({ converged: 2, truncated: false }); + }); + + test("DrainReport.truncated is true when more due rows remain behind this tick's page", async () => { + const seen = new Set(); + const { provisioner, store } = harness({ + ensureSeededFn: async (args) => { + seen.add(args.tenant.tenantId); + return { + kind: "seeded-pending-agents", + deployed: [], + pending: ALL_WORKFLOWS, + message: "agents pending", + }; + }, + }); + for (let index = 0; index < PENDING_SEED_SCAN_LIMIT + 3; index += 1) { + await store.put({ + ...SEED, + userId: `user_${index}`, + tenantId: `ten_${index}`, + }); + } + + const first = await provisioner.drainOnce(); + expect(first.truncated).toBe(true); + expect(first.pending).toBe(PENDING_SEED_SCAN_LIMIT); + expect(seen.size).toBe(PENDING_SEED_SCAN_LIMIT); + + const second = await provisioner.drainOnce(); + expect(second.truncated).toBe(false); + expect(second.pending).toBe(3); + expect(seen.size).toBe(PENDING_SEED_SCAN_LIMIT + 3); + }); + + test("rows past the scan limit still get a drain pass across ticks, even when the first page never converges", async () => { + const seen = new Set(); + const { provisioner, store } = harness({ + ensureSeededFn: async (args) => { + seen.add(args.tenant.tenantId); + return { + kind: "seeded-pending-agents", + deployed: [], + pending: ALL_WORKFLOWS, + message: "agents pending", + }; + }, + }); + const total = PENDING_SEED_SCAN_LIMIT + 3; + for (let index = 0; index < total; index += 1) { + await store.put({ + ...SEED, + userId: `user_${index}`, + tenantId: `ten_${index}`, + }); + } + + await provisioner.drainOnce(); + await provisioner.drainOnce(); + + expect(seen.size).toBe(total); + for (let index = 0; index < total; index += 1) { + expect(seen.has(`ten_${index}`)).toBe(true); + } }); }); From f415459a5e6cec83a5493991a539a6cbc516ce43 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Sun, 30 Aug 2026 14:07:32 -0700 Subject: [PATCH 2/2] Page pending seed drains past the scan limit listDue now returns an ordered page (expiresAt, userId, tenantId) with a keyset cursor instead of an unordered LIMIT from the start of the table. drainOnce keeps that cursor across ticks so a full first page of never-expiring rows cannot starve everyone else, and DrainReport.truncated says when more work remains. --- packages/onboarding/src/bench-provisioning.ts | 28 +++- packages/onboarding/src/index.ts | 2 + packages/onboarding/src/pending-seed.ts | 130 +++++++++++++++--- 3 files changed, 141 insertions(+), 19 deletions(-) diff --git a/packages/onboarding/src/bench-provisioning.ts b/packages/onboarding/src/bench-provisioning.ts index 32d00d60c..069a9df01 100644 --- a/packages/onboarding/src/bench-provisioning.ts +++ b/packages/onboarding/src/bench-provisioning.ts @@ -37,6 +37,7 @@ import { isFullySeeded } from "./provision"; import { PENDING_SEED_SCAN_LIMIT, type PendingSeed, + type PendingSeedListCursor, type PendingSeedStore, } from "./pending-seed"; @@ -81,6 +82,11 @@ export type DrainReport = { /** Benches skipped this tick because a previous failure's backoff has * not elapsed, or because a pass over them is still running. */ readonly deferred: number; + /** True when this tick stopped because its scan page filled the limit + * with more rows still in the table. The next tick continues after this + * page's cursor rather than restarting at the oldest row. False when + * the scan reached the end of the table. */ + readonly truncated: boolean; }; export type BenchProvisioner = { @@ -113,6 +119,7 @@ export function createBenchProvisioner( const retryAfter = new Map(); const failureCount = new Map(); let timer: ReturnType | undefined; + let scanAfter: PendingSeedListCursor | undefined; function holdOff(key: string): void { const failures = (failureCount.get(key) ?? 0) + 1; @@ -217,13 +224,16 @@ export function createBenchProvisioner( async function drainOnce( args: { ignoreBackoff?: boolean } = {}, ): Promise { - const due = await deps.store.listDue({ limit: PENDING_SEED_SCAN_LIMIT }); + const page = await deps.store.listDue({ + limit: PENDING_SEED_SCAN_LIMIT, + ...(scanAfter !== undefined ? { after: scanAfter } : {}), + }); let converged = 0; let pending = 0; let failed = 0; let deferred = 0; - for (const seed of due) { + for (const seed of page.seeds) { const key = benchKey(seed); const heldUntil = retryAfter.get(key); if ( @@ -243,7 +253,19 @@ export function createBenchProvisioner( else deferred += 1; } - return { converged, pending, failed, deferred }; + if (page.truncated && page.next !== undefined) { + scanAfter = page.next; + } else { + scanAfter = undefined; + } + + return { + converged, + pending, + failed, + deferred, + truncated: page.truncated, + }; } function wake(): void { diff --git a/packages/onboarding/src/index.ts b/packages/onboarding/src/index.ts index b4575a25d..997e7d81b 100644 --- a/packages/onboarding/src/index.ts +++ b/packages/onboarding/src/index.ts @@ -40,6 +40,8 @@ export { export type { PendingSeed, PendingSeedDb, + PendingSeedDuePage, + PendingSeedListCursor, PendingSeedStore, } from "./pending-seed"; export { diff --git a/packages/onboarding/src/pending-seed.ts b/packages/onboarding/src/pending-seed.ts index 265f7376d..258f6e43c 100644 --- a/packages/onboarding/src/pending-seed.ts +++ b/packages/onboarding/src/pending-seed.ts @@ -57,7 +57,7 @@ // plaintext key forward at all — this entire store, table, and AEAD // dance goes away. -import { and, eq } from "drizzle-orm"; +import { and, asc, eq, gt, or } from "drizzle-orm"; import type { PostgresJsDatabase } from "drizzle-orm/postgres-js"; import { type } from "arktype"; import type { CredentialCipher } from "@intx/types"; @@ -145,15 +145,24 @@ export interface PendingSeedStore { now?: () => number; }): Promise; /** - * Every unexpired row, for the background drain that has no request — - * and therefore no (userId, tenantId) — to scope itself by. This is - * what makes provisioning survive a hub restart: the rows a crashed - * process was mid-way through are still here, and the next boot's - * first tick picks them straight back up. Expired rows are deleted on - * the way past, the same read-time sweep `read` performs, so a dead - * key is never handed to a drain. `limit` bounds one tick's work. + * One ordered page of unexpired rows, for the background drain that has + * no request — and therefore no (userId, tenantId) — to scope itself by. + * This is what makes provisioning survive a hub restart: the rows a + * crashed process was mid-way through are still here, and the next + * boot's first tick picks them straight back up. Expired rows are + * deleted on the way past, the same read-time sweep `read` performs, so + * a dead key is never handed to a drain. + * + * Rows are ordered oldest-due first (`expiresAt`, then `userId`, + * `tenantId`). `limit` bounds one page so a tick cannot scan the whole + * table; pass `after` the previous page's `next` cursor to continue. + * `truncated` is true when more rows remain behind this page. */ - listDue(args: { now?: () => number; limit?: number }): Promise; + listDue(args: { + now?: () => number; + limit?: number; + after?: PendingSeedListCursor; + }): Promise; /** Deletes the row for (userId, tenantId), if any. Called once the * pending seed has done its job (seeded successfully) or once the * bench already reads as fully seeded some other way. */ @@ -162,6 +171,19 @@ export interface PendingSeedStore { export const PENDING_SEED_SCAN_LIMIT = 50; +/** Keyset cursor for `listDue` — the last row of a truncated page. */ +export type PendingSeedListCursor = { + readonly expiresAt: Date; + readonly userId: string; + readonly tenantId: string; +}; + +export type PendingSeedDuePage = { + readonly seeds: PendingSeed[]; + readonly truncated: boolean; + readonly next?: PendingSeedListCursor; +}; + interface StoredRow { provider: string; payload: string; @@ -177,7 +199,23 @@ interface RowAccess { get(userId: string, tenantId: string): Promise; put(userId: string, tenantId: string, row: StoredRow): Promise; delete(userId: string, tenantId: string): Promise; - list(limit: number): Promise; + list(args: { + limit: number; + after?: PendingSeedListCursor; + }): Promise; +} + +function comparePendingSeedCursor( + a: PendingSeedListCursor, + b: PendingSeedListCursor, +): number { + const byExpires = a.expiresAt.getTime() - b.expiresAt.getTime(); + if (byExpires !== 0) return byExpires; + if (a.userId < b.userId) return -1; + if (a.userId > b.userId) return 1; + if (a.tenantId < b.tenantId) return -1; + if (a.tenantId > b.tenantId) return 1; + return 0; } function createPendingSeedStore( @@ -263,13 +301,32 @@ function createPendingSeedStore( async listDue(args) { const now = args.now ?? Date.now; const nowMs = now(); - const rows = await access.list(args.limit ?? PENDING_SEED_SCAN_LIMIT); + const limit = args.limit ?? PENDING_SEED_SCAN_LIMIT; + const fetched = await access.list({ + limit: limit + 1, + ...(args.after !== undefined ? { after: args.after } : {}), + }); + const truncated = fetched.length > limit; + const rows = truncated ? fetched.slice(0, limit) : fetched; const due: PendingSeed[] = []; for (const row of rows) { const seed = await decodeRow(row, nowMs); if (seed !== undefined) due.push(seed); } - return due; + const last = rows[rows.length - 1]; + return { + seeds: due, + truncated, + ...(truncated && last !== undefined + ? { + next: { + expiresAt: last.expiresAt, + userId: last.userId, + tenantId: last.tenantId, + }, + } + : {}), + }; }, async clear(args) { @@ -330,8 +387,41 @@ export function createDrizzlePendingSeedStore< ), ); }, - async list(limit) { - const rows = await db.select().from(pendingSeed).limit(limit); + async list({ limit, after }) { + const rows = + after === undefined + ? await db + .select() + .from(pendingSeed) + .orderBy( + asc(pendingSeed.expiresAt), + asc(pendingSeed.userId), + asc(pendingSeed.tenantId), + ) + .limit(limit) + : await db + .select() + .from(pendingSeed) + .where( + or( + gt(pendingSeed.expiresAt, after.expiresAt), + and( + eq(pendingSeed.expiresAt, after.expiresAt), + gt(pendingSeed.userId, after.userId), + ), + and( + eq(pendingSeed.expiresAt, after.expiresAt), + eq(pendingSeed.userId, after.userId), + gt(pendingSeed.tenantId, after.tenantId), + ), + ), + ) + .orderBy( + asc(pendingSeed.expiresAt), + asc(pendingSeed.userId), + asc(pendingSeed.tenantId), + ) + .limit(limit); return rows.map((row) => ({ userId: row.userId, tenantId: row.tenantId, @@ -364,11 +454,19 @@ export function createInMemoryPendingSeedStore( async delete(userId, tenantId) { rows.delete(keyOf(userId, tenantId)); }, - async list(limit) { - return [...rows.entries()].slice(0, limit).map(([key, row]) => { + async list({ limit, after }) { + const identified = [...rows.entries()].map(([key, row]) => { const [userId = "", tenantId = ""] = key.split(":"); return { ...row, userId, tenantId }; }); + identified.sort(comparePendingSeedCursor); + const filtered = + after === undefined + ? identified + : identified.filter( + (row) => comparePendingSeedCursor(row, after) > 0, + ); + return filtered.slice(0, limit); }, }, cipher,