diff --git a/.changeset/driver-conformance-gate.md b/.changeset/driver-conformance-gate.md new file mode 100644 index 0000000000..352d66b589 --- /dev/null +++ b/.changeset/driver-conformance-gate.md @@ -0,0 +1,38 @@ +--- +"@objectstack/driver-sqlite-wasm": patch +--- + +test(drivers): the "held to by a gate" claim now has a gate behind it (#4363) + +Three changesets — filter combinator semantics (#3774), temporal storage form +(ADR-0053), deterministic paged reads (objectui#3106 / #4363) — each introduced +a shared case-set in `@objectstack/spec/data` with some version of the claim +that a future driver "is held to this by a gate rather than by remembering it". + +There was no gate. The case-sets are exports sitting in a package; nothing +obliged a driver to import them. Measured on `main`, the matrix had three holes: +`driver-sqlite-wasm` ran neither pagination case-set, and neither it nor +`driver-mongodb` ran the filter-logic one — including a hole in the very +case-set whose changeset made the claim. + +`scripts/check-driver-conformance.mjs` (`pnpm check:driver-conformance`, wired +into lint.yml's required job) makes the hole the failure. Every +(driver × case-set) cell is covered — some file under the package's `src/` +imports *and drives* the case-set's marker export — or carries a measured +DEBT/EXEMPT entry, reconciled in both directions. A third direction, CLASSIFIED, +holds the other end: a new `*-conformance.ts` fixture nobody classified fails +the run rather than starting life uncovered, which is the direction that +actually rots (#4203). It caught an unclassified `TEMPORAL_TIME_CASES` on its +first run. + +`driver-sqlite-wasm` gains the pagination suite the gate found missing. It +inherits `SqlDriver`'s ORDER BY construction, so nothing is re-implemented — +what the suite pins is that the clause survives a different *engine*: this +driver swaps knex's transport for a custom sql.js dialect that compiles, +executes and marshals every row through its own path, and a dialect that +reordered or dropped the trailing `ORDER BY id` would fail in no other suite. + +The two filter-logic holes are ledgered as DEBT rather than fixed here, with +their reasons printed on every run and tracked in #4405. The mongodb row is the +substantive one: `translateFilter` is an independent FilterCondition backend — +the fifth, and the one #3774 never enrolled when it counted "the four". diff --git a/.changeset/keyset-batch-walks.md b/.changeset/keyset-batch-walks.md new file mode 100644 index 0000000000..52d323738a --- /dev/null +++ b/.changeset/keyset-batch-walks.md @@ -0,0 +1,55 @@ +--- +"@objectstack/types": patch +"@objectstack/objectql": patch +"@objectstack/service-storage": patch +"@objectstack/plugin-approvals": patch +"@objectstack/plugin-pinyin-search": patch +--- + +fix(batch): the background walks seek instead of counting, so they stop skipping rows (#4363) + +#4363 made a single paged read a partition of its result set. It could not make +a *walk* one: seven background scans paged with a growing `offset` while writing +to the very rows they were reading, and an offset counts into a set those writes +are changing. Rows slide past the cursor and are never visited. + +That is not a slow page in any of these — it is a wrong answer wearing the shape +of a clean run: + +- **`rebuildApproverIndex`** built its desired state by walking + `sys_approval_request WHERE status = 'pending'` with no `orderBy` at all, then + **deleted** every index row that state did not explain. A skipped request + meant an approver silently dropped from someone's queue. (The loop beside it + ordered by `created_at` — not unique, so its pages were never a partition + either.) +- **`verifyFileReferences`** decides which files nothing references. A record it + never visits is reported as an unreferenced file. +- **`backfillFileReferences`** and the **pinyin companion backfill** rewrite + each row they read, so their own writes were shifting the set out from under + the cursor. Records were left unconverted and unsearchable by a run that + reported success. +- **`scanValueShapes`** exists to vouch that no stored value is off-shape, and + it opens a migration gate on that evidence. + +All of them now go through `keysetWalk` (`@objectstack/types`): order by a +unique key, and seek past the last one instead of counting from the start. A +row's key does not move when the row is updated, and cannot be shifted when +another is deleted, so the walk is stable under exactly the mutation these +functions perform. It is also O(n) rather than O(n²/page) — measured on +Postgres over 2M rows, deep pages cost ~1.1 s by offset against ~0.09 s by seek. + +One deliberate non-conversion: the REST **export** stream keeps its offset. It +honors a caller-chosen sort, and a keyset walk would have to re-order the export +by `id` to seek — changing what the user asked for to fix a cost. Its pages are +already a partition since #4363; only the depth cost remains. + +`keysetWalk` merges the cursor with `$and` rather than spreading it into the +caller's filter, so a walk whose own `where` constrains the key column +(`{ id: { $in: [...] } }`) keeps that constraint instead of having it silently +overwritten. When a `max` cap is set it reads one row beyond the cap to tell +"the cap stopped us" from "the source ended exactly there" — without that, a +walk that read everything still reports `truncated`, and a caller acting on it +goes looking for rows that were never withheld. + +The storage suites' fake engines now **throw** on an `offset` instead of serving +one, so the conversion is pinned rather than merely passing. diff --git a/.github/workflows/lint.yml b/.github/workflows/lint.yml index d5a569b81e..151af2c049 100644 --- a/.github/workflows/lint.yml +++ b/.github/workflows/lint.yml @@ -263,6 +263,19 @@ jobs: - name: Check every package is type-check covered or ledgered run: pnpm check:type-check-coverage + # Conformance meta-gate (#4363): `spec/src/data/*-conformance.ts` holds the + # shared case-sets that exist so the independent drivers answer ONE + # standard, and three separate changesets claimed a future driver "is held + # to this by a gate rather than by remembering it". There was no gate — + # nothing obliged a driver to import them — and the matrix had three holes, + # one of them in the case-set whose changeset made the claim. Same shape as + # the ratchet above: every (driver × case-set) cell is covered or carries a + # measured DEBT/EXEMPT entry, reconciled in both directions, plus a + # CLASSIFIED direction so a NEW shared fixture cannot start life uncovered. + # Reads source files only; no build, sub-second. + - name: Check every driver runs the shared conformance cases + run: pnpm check:driver-conformance + - name: Type check (@objectstack/spec) run: pnpm --filter @objectstack/spec exec tsc --noEmit diff --git a/package.json b/package.json index 9aceac6f4c..5d32bf2e79 100644 --- a/package.json +++ b/package.json @@ -44,7 +44,8 @@ "check:release-notes": "node scripts/check-release-notes.mjs", "check:node-version": "node scripts/check-node-version.mjs", "check:published-files": "node scripts/check-published-files.mjs --self-test && node scripts/check-published-files.mjs", - "check:type-check-coverage": "node scripts/check-type-check-coverage.mjs --self-test && node scripts/check-type-check-coverage.mjs" + "check:type-check-coverage": "node scripts/check-type-check-coverage.mjs --self-test && node scripts/check-type-check-coverage.mjs", + "check:driver-conformance": "node scripts/check-driver-conformance.mjs --self-test && node scripts/check-driver-conformance.mjs" }, "keywords": [ "objectstack", diff --git a/packages/objectql/src/validation/scan-value-shapes.ts b/packages/objectql/src/validation/scan-value-shapes.ts index 79bd6d6cb7..aa0db9e06d 100644 --- a/packages/objectql/src/validation/scan-value-shapes.ts +++ b/packages/objectql/src/validation/scan-value-shapes.ts @@ -31,6 +31,8 @@ * fact the validator does not recognise. */ +import { keysetWalk } from '@objectstack/types'; + import { valueShapeViolation, isScannableValueShapeField, @@ -125,31 +127,21 @@ export async function scanValueShapes( for (const object of scannedObjects) { const fields = scannableFieldsOf(engine, object); - let offset = 0; - - for (;;) { - let page: Array>; - try { - page = await engine.find(object, { - fields: ['id', ...fields.map((f) => f.name)], - limit: SCAN_PAGE_SIZE, - offset, - context: { ...SYSTEM_CTX }, - }); - } catch (err) { - // An object we cannot read is not an object we may vouch for. Record it - // AND truncate: the gate's claim is about all the data, so a hole in - // the evidence has to close it. - logger.warn( - `[value-shape] scan: cannot read ${object} (${(err as Error)?.message ?? err}) — ` + - 'reported as unreadable; the gate stays closed', - ); - unreadableObjects.push(object); - truncated = true; - break; - } - if (!page || page.length === 0) break; + // Seek by `id` (#4363) rather than counting from the start. This scan + // exists to vouch that no stored value is off-shape, and an offset walk + // cannot promise it visited every row — the same reason an unreadable + // object below closes the gate rather than being skipped quietly. + const walk = keysetWalk>( + (q) => engine.find(object, { + ...q, + fields: ['id', ...fields.map((f) => f.name)], + context: { ...SYSTEM_CTX }, + }), + { pageSize: SCAN_PAGE_SIZE, max: maxPerObject }, + ); + try { + for await (const page of walk.pages()) { for (const record of page) { scannedRecords++; for (const { name, def } of fields) { @@ -173,14 +165,20 @@ export async function scanValueShapes( } } } - - offset += page.length; - if (page.length < SCAN_PAGE_SIZE) break; - if (offset >= maxPerObject) { - truncated = true; - break; } + } catch (err) { + // An object we cannot read is not an object we may vouch for. Record it + // AND truncate: the gate's claim is about all the data, so a hole in + // the evidence has to close it. + logger.warn( + `[value-shape] scan: cannot read ${object} (${(err as Error)?.message ?? err}) — ` + + 'reported as unreadable; the gate stays closed', + ); + unreadableObjects.push(object); + truncated = true; + continue; } + if (walk.truncated) truncated = true; } const list = [...findings.values()].sort((a, b) => b.count - a.count); diff --git a/packages/plugins/driver-sqlite-wasm/src/sqlite-wasm-pagination-conformance.test.ts b/packages/plugins/driver-sqlite-wasm/src/sqlite-wasm-pagination-conformance.test.ts new file mode 100644 index 0000000000..2f16426328 --- /dev/null +++ b/packages/plugins/driver-sqlite-wasm/src/sqlite-wasm-pagination-conformance.test.ts @@ -0,0 +1,108 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * Deterministic paged reads for the wasm driver (objectui#3106, + * objectstack#4363) — the contract on `IDataDriver.find`, run against the + * shared `@objectstack/spec/data` cases like every other driver. + * + * `SqliteWasmDriver extends SqlDriver`, so the ORDER BY is *built* by inherited + * code and nothing here re-implements it. What this pins is the other half: + * the clause has to survive a different **engine**. This driver swaps knex's + * transport for a custom sql.js dialect (`Client_WasmSqlite`), which compiles + * and executes the statement and marshals every row back through its own path. + * A dialect that reordered, dropped or mis-bound the trailing `ORDER BY id` + * would produce exactly the failure the contract exists to rule out — full + * pages, real rows, one served twice and one never served — and it would fail + * in no other suite in the repo. + * + * That is the same reason the temporal suite next door exists: inheritance + * makes it easy to assume the behavior comes along for free, and the dialect + * seam is precisely where it might not. + */ + +import { describe, it, expect, beforeAll, afterAll } from 'vitest'; +import { + PAGINATION_ALL_IDS, + PAGINATION_CASES, + PAGINATION_ROWS, + PAGINATION_UNORDERED_CASES, +} from '@objectstack/spec/data'; +import { SqliteWasmDriver } from './index.js'; + +describe('driver-sqlite-wasm — paged reads are a partition of the result set', () => { + let driver: SqliteWasmDriver; + + beforeAll(async () => { + driver = new SqliteWasmDriver({ filename: ':memory:' }); + await driver.initObjects([ + { + name: 'ticket', + fields: { + status: { type: 'string' }, + rank: { type: 'integer' }, + name: { type: 'string' }, + }, + }, + ]); + for (const row of PAGINATION_ROWS) { + await driver.create('ticket', { ...row }, { bypassTenantAudit: true } as any); + } + }); + + afterAll(async () => { + await driver.disconnect(); + }); + + const walk = async (pageSize: number, orderBy?: ReadonlyArray<{ field: string; order: 'asc' | 'desc' }>) => { + const seen: string[] = []; + for (let offset = 0; offset < PAGINATION_ROWS.length; offset += pageSize) { + const page = await driver.find( + 'ticket', + { ...(orderBy ? { orderBy: [...orderBy] } : {}), limit: pageSize, offset } as any, + { bypassTenantAudit: true } as any, + ); + seen.push(...page.map((r: any) => String(r.id))); + } + return seen; + }; + + for (const testCase of PAGINATION_CASES) { + it(`visits every row exactly once — ${testCase.name}`, async () => { + const seen = await walk(testCase.pageSize, testCase.orderBy); + expect(seen).toHaveLength(PAGINATION_ALL_IDS.length); + expect(new Set(seen).size).toBe(PAGINATION_ALL_IDS.length); + expect([...seen].sort()).toEqual([...PAGINATION_ALL_IDS].sort()); + }); + + it(`page boundaries are invisible — ${testCase.name}`, async () => { + const paged = await walk(testCase.pageSize, testCase.orderBy); + const whole = await driver.find( + 'ticket', + { orderBy: [...testCase.orderBy] } as any, + { bypassTenantAudit: true } as any, + ); + expect(paged).toEqual(whole.map((r: any) => String(r.id))); + }); + } + + for (const testCase of PAGINATION_UNORDERED_CASES) { + it(`visits every row exactly once with NO orderBy at all — ${testCase.name}`, async () => { + const seen = await walk(testCase.pageSize); + expect(seen).toHaveLength(PAGINATION_ALL_IDS.length); + expect(new Set(seen).size).toBe(PAGINATION_ALL_IDS.length); + expect([...seen].sort()).toEqual([...PAGINATION_ALL_IDS].sort()); + }); + + it(`walks an unsorted read in id order — ${testCase.name}`, async () => { + // The fixture's ids are shuffled relative to insertion order, so this + // distinguishes "the inherited ORDER BY reached the wasm engine" from + // "sql.js happened to hand back insertion order". + expect(await walk(testCase.pageSize)).toEqual([...PAGINATION_ALL_IDS].sort()); + }); + } + + it('leaves an UNPAGED unordered read alone — no sort is imposed on a caller who asked for none', async () => { + const rows = await driver.find('ticket', {} as any, { bypassTenantAudit: true } as any); + expect(rows.map((r: any) => r.id)).toEqual(PAGINATION_ROWS.map((r) => r.id)); + }); +}); diff --git a/packages/plugins/plugin-approvals/package.json b/packages/plugins/plugin-approvals/package.json index c260b3efe8..b1d360e276 100644 --- a/packages/plugins/plugin-approvals/package.json +++ b/packages/plugins/plugin-approvals/package.json @@ -22,7 +22,8 @@ "@objectstack/formula": "workspace:*", "@objectstack/metadata-core": "workspace:*", "@objectstack/platform-objects": "workspace:*", - "@objectstack/spec": "workspace:*" + "@objectstack/spec": "workspace:*", + "@objectstack/types": "workspace:*" }, "devDependencies": { "@objectstack/objectql": "workspace:*", diff --git a/packages/plugins/plugin-approvals/src/approval-service.ts b/packages/plugins/plugin-approvals/src/approval-service.ts index 7613dd5daa..87e9647d88 100644 --- a/packages/plugins/plugin-approvals/src/approval-service.ts +++ b/packages/plugins/plugin-approvals/src/approval-service.ts @@ -9,6 +9,7 @@ import { type ApprovalNodeConfig, } from '@objectstack/spec/automation'; import { ExpressionEngine, collectCelRootIdentifiers } from '@objectstack/formula'; +import { keysetWalk } from '@objectstack/types'; import { ADMIN_FULL_ACCESS, ORGANIZATION_ADMIN_GRANTS, @@ -3112,37 +3113,41 @@ export class ApprovalService implements IApprovalService { * the number of *pending* requests, not the request history. */ async rebuildApproverIndex(): Promise<{ requests: number; inserted: number; deleted: number }> { - // Desired state: every pending request's CSV entries. + // Both walks seek by `id` rather than counting from the start (#4363). + // The desired-state walk fed the deletion pass below, so a row it skipped + // was not a slow page — it was an approver silently dropped from someone's + // queue. An offset walk over `status = 'pending'` has no such guarantee: + // it slices an arrangement nothing holds steady, and this method is itself + // a writer against the same tables. const desired = new Map; org: string | null }>(); const PAGE = 500; - for (let offset = 0; ; offset += PAGE) { - const batch = await this.engine.find('sys_approval_request', { - where: { status: 'pending' }, + const requests = keysetWalk>( + (q) => this.engine.find('sys_approval_request', { + ...q, fields: ['id', 'pending_approvers', 'organization_id'], - limit: PAGE, offset, context: SYSTEM_CTX, - }); - const rows: any[] = Array.isArray(batch) ? batch : []; + context: SYSTEM_CTX, + }), + { where: { status: 'pending' }, pageSize: PAGE }, + ); + for await (const rows of requests.pages()) { for (const r of rows) { desired.set(String(r.id), { approvers: new Set(csvSplit(r.pending_approvers)), org: r.organization_id ?? null, }); } - if (rows.length < PAGE) break; } - // Current state: read the whole index first (bounded by the live work - // queue), THEN mutate — deleting while paginating would shift the cursor. + // Current state: read the whole index first, THEN mutate. The seek makes + // that separation a belt rather than the only brace — `created_at` is not + // unique, so the previous ORDER BY could not make these pages a partition + // even before the deletes below shifted the rows underneath them. const indexRows: any[] = []; - for (let offset = 0; ; offset += PAGE) { - const batch = await this.engine.find('sys_approval_approver', { - orderBy: [{ field: 'created_at', order: 'asc' }], - limit: PAGE, offset, context: SYSTEM_CTX, - }); - const rows: any[] = Array.isArray(batch) ? batch : []; - indexRows.push(...rows); - if (rows.length < PAGE) break; - } + const index = keysetWalk>( + (q) => this.engine.find('sys_approval_approver', { ...q, context: SYSTEM_CTX }), + { pageSize: PAGE }, + ); + for await (const rows of index.pages()) indexRows.push(...rows); let inserted = 0; let deleted = 0; const seen = new Map>(); for (const row of indexRows) { diff --git a/packages/plugins/plugin-pinyin-search/src/companion-projection.test.ts b/packages/plugins/plugin-pinyin-search/src/companion-projection.test.ts index ace93a414a..17e4d21567 100644 --- a/packages/plugins/plugin-pinyin-search/src/companion-projection.test.ts +++ b/packages/plugins/plugin-pinyin-search/src/companion-projection.test.ts @@ -56,10 +56,22 @@ function makeEngine(schemas: any[]) { async trigger(event: string, ctx: any) { for (const { handler } of hooks[event] ?? []) await handler(ctx); }, + // Executes the KEYSET walk the backfill emits since #4363: `orderBy` on + // the key plus a `$gt` seek. `offset` throws rather than being served — + // this walk updates the rows it is reading, so an offset counts into a set + // its own writes are changing and rows slide past the cursor unconverted. async find(o: string, opts?: any) { - const rows = tables[o] ?? []; - const offset = opts?.offset ?? 0; - return rows.slice(offset, offset + (opts?.limit ?? rows.length)); + if (opts?.offset !== undefined) { + throw new Error('offset paging skips rows a mutating walk must visit — expected a keyset seek (#4363)'); + } + const key = opts?.orderBy?.[0]?.field; + const seek = opts?.where?.[key]?.$gt ?? opts?.where?.$and?.[1]?.[key]?.$gt; + let rows: any[] = tables[o] ?? []; + if (seek !== undefined) rows = rows.filter((r: any) => String(r[key]) > String(seek)); + const ordered = key + ? [...rows].sort((a: any, b: any) => String(a[key]).localeCompare(String(b[key]))) + : rows; + return ordered.slice(0, opts?.limit ?? ordered.length); }, async insert(o: string, data: Row) { const ctx = { object: o, event: 'beforeInsert', input: { data } }; diff --git a/packages/plugins/plugin-pinyin-search/src/companion-projection.ts b/packages/plugins/plugin-pinyin-search/src/companion-projection.ts index dce617d2ac..710969a896 100644 --- a/packages/plugins/plugin-pinyin-search/src/companion-projection.ts +++ b/packages/plugins/plugin-pinyin-search/src/companion-projection.ts @@ -21,6 +21,7 @@ import { resolveSearchCompanionSources, containsCJK, } from '@objectstack/objectql'; +import { keysetWalk } from '@objectstack/types'; import { computeSearchCompanionValue } from './pinyin.js'; const SYSTEM_CTX = { isSystem: true, positions: [], permissions: [] } as const; @@ -145,21 +146,21 @@ export async function backfillSearchCompanion( if (sources.length === 0) continue; result.objects++; - let offset = 0; - for (;;) { - let rows: any[] = []; - try { - rows = await engine.find(schema.name, { - fields: ['id', ...sources, SEARCH_COMPANION_FIELD], - limit: batchSize, - offset, - context: SYSTEM_CTX, - }); - } catch (err: any) { - logger?.warn?.('[pinyin-search] backfill scan failed', { object: schema.name, error: err?.message }); - break; - } - if (!rows?.length) break; + // Seek by `id` (#4363). This walk UPDATES the rows it reads, and an offset + // counts into a set those writes are changing, so rows slide past the + // cursor and never get their companion blob — a backfill that reports a + // clean pass while leaving records unsearchable. + const walk = keysetWalk( + (q) => engine.find(schema.name, { + ...q, + fields: ['id', ...sources, SEARCH_COMPANION_FIELD], + context: SYSTEM_CTX, + }), + { pageSize: batchSize }, + ); + + try { + for await (const rows of walk.pages()) { result.scanned += rows.length; for (const row of rows) { @@ -187,9 +188,10 @@ export async function backfillSearchCompanion( }); } } - - if (rows.length < batchSize) break; - offset += batchSize; + } + } catch (err: any) { + logger?.warn?.('[pinyin-search] backfill scan failed', { object: schema.name, error: err?.message }); + continue; } } diff --git a/packages/services/service-storage/src/backfill-file-references.test.ts b/packages/services/service-storage/src/backfill-file-references.test.ts index 330c15095a..10f7edaa84 100644 --- a/packages/services/service-storage/src/backfill-file-references.test.ts +++ b/packages/services/service-storage/src/backfill-file-references.test.ts @@ -27,11 +27,22 @@ function fakeEngine(tables: Record>>) { const engine: BackfillEngine & { tables: typeof tables; calls: typeof calls } = { getObject: (name) => REGISTRY[name], getConfigs: () => REGISTRY, + // Executes the KEYSET walk the backfill emits since #4363: `orderBy` on + // the key plus a `$gt` seek. `offset` THROWS rather than being served — + // this walk rewrites the rows it is reading, and an offset counts into a + // set its own writes are changing, so rows slide past the cursor and never + // get converted. A fake that honored an offset would let that back in with + // every assertion still green. async find(object, options: any) { - const rows = tables[object] ?? []; - const start = typeof options?.offset === 'number' ? options.offset : 0; - const end = typeof options?.limit === 'number' ? start + options.limit : undefined; - return rows.slice(start, end); + if (options?.offset !== undefined) { + throw new Error('offset paging skips rows a mutating walk must visit — expected a keyset seek (#4363)'); + } + const key = options?.orderBy?.[0]?.field; + const seek = (options?.where as any)?.[key]?.$gt ?? (options?.where as any)?.$and?.[1]?.[key]?.$gt; + let rows = tables[object] ?? []; + if (seek !== undefined) rows = rows.filter((r) => String(r[key]) > String(seek)); + const ordered = key ? [...rows].sort((a, b) => String(a[key]).localeCompare(String(b[key]))) : rows; + return typeof options?.limit === 'number' ? ordered.slice(0, options.limit) : ordered; }, async insert(object, data: any) { calls.push({ op: 'insert', object, arg: data }); diff --git a/packages/services/service-storage/src/backfill-file-references.ts b/packages/services/service-storage/src/backfill-file-references.ts index 31aa6c45be..f8092b9602 100644 --- a/packages/services/service-storage/src/backfill-file-references.ts +++ b/packages/services/service-storage/src/backfill-file-references.ts @@ -3,6 +3,7 @@ import { randomUUID } from 'node:crypto'; import { FILE_REFERENCE_TYPES, isFileIdToken, RAW_FILE_VALUES_CONTEXT_KEY } from '@objectstack/spec/data'; import type { IStorageService } from '@objectstack/spec/contracts'; +import { keysetWalk } from '@objectstack/types'; /** * Legacy file-value backfill (ADR-0104 D3 wave 2). @@ -228,25 +229,18 @@ export async function backfillFileReferences( for (const object of scannedObjects) { const fileFields = fileFieldsOf(engine, object); - let offset = 0; - - for (;;) { - let page: Array>; - try { - page = await engine.find(object, { - fields: ['id', ...fileFields], - limit: SCAN_PAGE_SIZE, - offset, - context: { ...SYSTEM_CTX }, - }); - } catch (err) { - logger.warn( - `[storage] backfill: cannot read ${object} (${(err as Error)?.message ?? err}) — skipped`, - ); - break; - } - if (!page || page.length === 0) break; + // Seek by `id` (#4363). This walk WRITES to the rows it is reading — the + // rewrite below updates each record in place — and an offset counts into a + // set the writes are changing underneath it, so rows slide past the cursor + // and are never converted. The key does not move when a row is updated, so + // the seek is not affected by the very thing this function does. + const walk = keysetWalk>( + (q) => engine.find(object, { ...q, fields: ['id', ...fileFields], context: { ...SYSTEM_CTX } }), + { pageSize: SCAN_PAGE_SIZE, max: maxPerObject }, + ); + try { + for await (const page of walk.pages()) { for (const record of page) { const recordId = record?.id; if (recordId == null) continue; @@ -376,14 +370,14 @@ export async function backfillFileReferences( } } } - - offset += page.length; - if (page.length < SCAN_PAGE_SIZE) break; - if (offset >= maxPerObject) { - truncated = true; - break; } + } catch (err) { + logger.warn( + `[storage] backfill: cannot read ${object} (${(err as Error)?.message ?? err}) — skipped`, + ); + continue; } + if (walk.truncated) truncated = true; } const count = (k: BackfillActionKind) => actions.filter((a) => a.kind === k).length; diff --git a/packages/services/service-storage/src/verify-file-references.test.ts b/packages/services/service-storage/src/verify-file-references.test.ts index 76d5be555b..e39b5c8ec8 100644 --- a/packages/services/service-storage/src/verify-file-references.test.ts +++ b/packages/services/service-storage/src/verify-file-references.test.ts @@ -21,20 +21,38 @@ const REGISTRY: Record = { tag: { fields: { id: { type: 'text' }, label: { type: 'text' } } }, }; +/** + * Executes the subset of the query language this scan emits. Since #4363 that + * is a KEYSET walk — `orderBy` on the key plus a `$gt` seek AND-ed onto the + * caller's filter — so the fake grew `$and` and `$gt`, and `offset` is gone. + * + * `offset` now THROWS rather than being served: this scan decides which files + * are still referenced, and an offset walk cannot promise it visited every row. + * A fake that quietly honored one would let that regression back in with every + * assertion still green. + */ function fakeEngine(tables: Record>>): VerifyReferencesEngine { - const matches = (row: Record, where: Record) => - Object.entries(where).every(([k, v]) => { + const matches = (row: Record, where: unknown): boolean => { + if (where == null) return true; + const w = where as Record; + if (Array.isArray(w.$and)) return w.$and.every((c: unknown) => matches(row, c)); + return Object.entries(w).every(([k, v]) => { if (v && typeof v === 'object' && '$ne' in (v as any)) return row[k] !== (v as any).$ne; + if (v && typeof v === 'object' && '$gt' in (v as any)) return String(row[k]) > String((v as any).$gt); return row[k] === v; }); + }; return { getObject: (name) => REGISTRY[name], getConfigs: () => REGISTRY, async find(object, options: any) { - const rows = (tables[object] ?? []).filter((r) => matches(r, options?.where ?? {})); - const start = typeof options?.offset === 'number' ? options.offset : 0; - const end = typeof options?.limit === 'number' ? start + options.limit : undefined; - return rows.slice(start, end); + if (options?.offset !== undefined) { + throw new Error('offset paging cannot promise a full walk — expected a keyset seek (#4363)'); + } + const key = options?.orderBy?.[0]?.field; + const rows = (tables[object] ?? []).filter((r) => matches(r, options?.where)); + const ordered = key ? [...rows].sort((a, b) => String(a[key]).localeCompare(String(b[key]))) : rows; + return typeof options?.limit === 'number' ? ordered.slice(0, options.limit) : ordered; }, }; } @@ -191,6 +209,9 @@ describe('verifyFileReferences (ADR-0104 D3 wave 2 — R4 gate)', () => { }); // ── Mechanics ───────────────────────────────────────────────────── + // The fake engine refuses an `offset`, so this also asserts the walk seeks + // (#4363) rather than counting from the start — which is what makes "every + // row was visited" a promise rather than a hope. it('pages through records rather than reading one unbounded page', async () => { const many = Array.from({ length: 1200 }, (_, i) => ({ id: `p${i}`, image: `file_${i}` })); const engine = fakeEngine({ diff --git a/packages/services/service-storage/src/verify-file-references.ts b/packages/services/service-storage/src/verify-file-references.ts index 7b70b41097..4ee1179f7b 100644 --- a/packages/services/service-storage/src/verify-file-references.ts +++ b/packages/services/service-storage/src/verify-file-references.ts @@ -1,6 +1,7 @@ // Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license. import { FILE_REFERENCE_TYPES, isFileIdToken, RAW_FILE_VALUES_CONTEXT_KEY } from '@objectstack/spec/data'; +import { keysetWalk } from '@objectstack/types'; /** * File-reference reconciliation (ADR-0104 D3 wave 2) — the executable form of @@ -144,71 +145,65 @@ export async function verifyFileReferences( for (const object of scannedObjects) { const fileFields = fileFieldsOf(engine, object); - let offset = 0; - for (;;) { - let page: Array>; - try { - page = await engine.find(object, { - fields: ['id', ...fileFields], - limit: SCAN_PAGE_SIZE, - offset, - context: { ...SYSTEM_CTX }, - }); - } catch { - break; // unreadable object — skip rather than abort the whole scan - } - if (!page || page.length === 0) break; - for (const record of page) { - const recordId = record?.id; - if (recordId == null) continue; - scannedRecords++; - for (const field of fileFields) { - for (const fileId of new Set(idTokensIn(record[field]))) { - const list = held.get(fileId) ?? []; - list.push({ object, recordId: String(recordId), field }); - held.set(fileId, list); + // Seek by `id` rather than counting from the start (#4363). This scan + // decides which files are still referenced, and a row it never visits is + // reported as an unreferenced file — the wrong answer with the shape of a + // clean run. An offset walk cannot promise it visits every row. + const walk = keysetWalk>( + (q) => engine.find(object, { ...q, fields: ['id', ...fileFields], context: { ...SYSTEM_CTX } }), + { pageSize: SCAN_PAGE_SIZE, max: maxPerObject }, + ); + try { + for await (const page of walk.pages()) { + for (const record of page) { + const recordId = record?.id; + if (recordId == null) continue; + scannedRecords++; + for (const field of fileFields) { + for (const fileId of new Set(idTokensIn(record[field]))) { + const list = held.get(fileId) ?? []; + list.push({ object, recordId: String(recordId), field }); + held.set(fileId, list); + } } } } - offset += page.length; - if (page.length < SCAN_PAGE_SIZE) break; - if (offset >= maxPerObject) { - truncated = true; - break; - } + } catch { + // Unreadable object — skip rather than abort the whole scan. An object + // with no `id` column (a federated table, ADR-0015) lands here too, and + // skipping it is the honest outcome: it cannot be walked exhaustively, + // so it must not contribute a "nothing holds this file" conclusion. + continue; } + if (walk.truncated) truncated = true; } // ── 2. Recorded ownership ───────────────────────────────────────── /** fileId → the slot recorded as its owner. */ const owned = new Map(); - let ownerOffset = 0; - for (;;) { - let page: Array>; - try { - page = await engine.find('sys_file', { - where: { ref_object: { $ne: null } }, - fields: ['id', 'ref_object', 'ref_id', 'ref_field', 'status'], - limit: SCAN_PAGE_SIZE, - offset: ownerOffset, - context: { ...SYSTEM_CTX }, - }); - } catch { - break; - } - if (!page || page.length === 0) break; - for (const row of page) { - // A driver without `$ne` support may hand back unowned rows too — filter - // here rather than trusting the predicate round-tripped. - if (row?.id == null || !row.ref_object || row.ref_id == null) continue; - owned.set(String(row.id), { - object: String(row.ref_object), - recordId: String(row.ref_id), - field: String(row.ref_field ?? ''), - }); + const owners = keysetWalk>( + (q) => engine.find('sys_file', { + ...q, + fields: ['id', 'ref_object', 'ref_id', 'ref_field', 'status'], + context: { ...SYSTEM_CTX }, + }), + { where: { ref_object: { $ne: null } }, pageSize: SCAN_PAGE_SIZE }, + ); + try { + for await (const page of owners.pages()) { + for (const row of page) { + // A driver without `$ne` support may hand back unowned rows too — filter + // here rather than trusting the predicate round-tripped. + if (row?.id == null || !row.ref_object || row.ref_id == null) continue; + owned.set(String(row.id), { + object: String(row.ref_object), + recordId: String(row.ref_id), + field: String(row.ref_field ?? ''), + }); + } } - ownerOffset += page.length; - if (page.length < SCAN_PAGE_SIZE) break; + } catch { + // Same posture as the record scan: an unreadable ledger is not evidence. } // ── 3. Diff ─────────────────────────────────────────────────────── @@ -277,35 +272,31 @@ export async function verifyFileReferences( // ── 4. Advisory: committed files nothing points at ──────────────── if (options.includeUnreferenced) { - let offset = 0; - for (;;) { - let page: Array>; - try { - page = await engine.find('sys_file', { - where: { status: 'committed' }, - fields: ['id', 'ref_object', 'scope'], - limit: SCAN_PAGE_SIZE, - offset, - context: { ...SYSTEM_CTX }, - }); - } catch { - break; - } - if (!page || page.length === 0) break; - for (const row of page) { - const id = row?.id == null ? null : String(row.id); - // attachments-scope files are governed by sys_attachment, not by the - // ownership columns — not this scan's business. - if (!id || row.ref_object || row.scope === 'attachments') continue; - if (held.has(id)) continue; - issues.push({ - kind: 'unreferenced_file', - fileId: id, - detail: 'committed but neither owned nor held by any field — storage cost, not data risk', - }); + const files = keysetWalk>( + (q) => engine.find('sys_file', { + ...q, + fields: ['id', 'ref_object', 'scope'], + context: { ...SYSTEM_CTX }, + }), + { where: { status: 'committed' }, pageSize: SCAN_PAGE_SIZE }, + ); + try { + for await (const page of files.pages()) { + for (const row of page) { + const id = row?.id == null ? null : String(row.id); + // attachments-scope files are governed by sys_attachment, not by the + // ownership columns — not this scan's business. + if (!id || row.ref_object || row.scope === 'attachments') continue; + if (held.has(id)) continue; + issues.push({ + kind: 'unreferenced_file', + fileId: id, + detail: 'committed but neither owned nor held by any field — storage cost, not data risk', + }); + } } - offset += page.length; - if (page.length < SCAN_PAGE_SIZE) break; + } catch { + // Same posture as the record scan above. } } diff --git a/packages/types/src/index.ts b/packages/types/src/index.ts index 6904af9903..2e0f1b996b 100644 --- a/packages/types/src/index.ts +++ b/packages/types/src/index.ts @@ -3,6 +3,9 @@ export * from './degraded-boot.js'; export * from './env.js'; export * from './error-leak.js'; +// Seek-based pagination for batch walks — the offset alternative that neither +// skips rows when the walk mutates as it goes, nor costs O(n²/p) (#4363). +export * from './keyset-walk.js'; export * from './module-not-found.js'; export * from './response-envelope.js'; diff --git a/packages/types/src/keyset-walk.test.ts b/packages/types/src/keyset-walk.test.ts new file mode 100644 index 0000000000..008281ec54 --- /dev/null +++ b/packages/types/src/keyset-walk.test.ts @@ -0,0 +1,230 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * `keysetWalk` — the seek-based batch walk (#4363). + * + * The load-bearing cases are the two an offset walk gets wrong, and they are + * asserted against a fake store that behaves the way a database does: + * + * 1. **Mutation during the walk.** A backfill updates every row it reads and a + * rebuild deletes; with an offset those shift rows past the cursor and they + * are never visited. Here the store is mutated mid-walk and every row must + * still be seen exactly once. + * 2. **The cursor must compose with the caller's filter, not overwrite it.** A + * caller whose own `where` constrains the key column must not have that + * constraint replaced by the seek predicate — the walk would then return + * rows the caller excluded. + * + * The rest pins the boundaries a partial scan hides behind: short final page, + * `max` truncation reported rather than silently passed off as completion, and + * a keyless row stopping the walk instead of spinning on the same page. + */ + +import { describe, it, expect } from 'vitest'; +import { keysetWalk, type KeysetPageQuery } from './keyset-walk.js'; + +interface Row extends Record { + id: string; + status: string; +} + +/** Ids are deliberately not in insertion order, so "id order" is visibly chosen. */ +const seed = (): Row[] => [ + { id: 'r07', status: 'open' }, + { id: 'r03', status: 'done' }, + { id: 'r11', status: 'open' }, + { id: 'r01', status: 'open' }, + { id: 'r09', status: 'done' }, + { id: 'r05', status: 'open' }, + { id: 'r12', status: 'done' }, + { id: 'r02', status: 'open' }, +]; + +/** + * A store that executes the subset of the filter language the walk emits: + * the caller's flat equality `where`, `$and`, and `{ key: { $gt } }`. + */ +function makeStore(rows: Row[]) { + const matches = (row: Row, where: unknown): boolean => { + if (where == null) return true; + const w = where as Record; + if (Array.isArray(w.$and)) return w.$and.every((c: unknown) => matches(row, c)); + return Object.entries(w).every(([field, cond]) => { + if (cond && typeof cond === 'object' && '$gt' in cond) return String(row[field]) > String(cond.$gt); + if (cond && typeof cond === 'object' && '$in' in cond) return (cond.$in as unknown[]).includes(row[field]); + return row[field] === cond; + }); + }; + + return { + rows, + read: async (q: KeysetPageQuery): Promise => { + const key = q.orderBy[0]!.field; + return rows + .filter((r) => matches(r, q.where)) + .sort((a, b) => String(a[key]).localeCompare(String(b[key]))) + .slice(0, q.limit); + }, + }; +} + +const collect = async (walk: { pages(): AsyncGenerator }) => { + const seen: string[] = []; + for await (const page of walk.pages()) seen.push(...page.map((r) => r.id)); + return seen; +}; + +describe('keysetWalk', () => { + it('visits every row exactly once, in key order', async () => { + const store = makeStore(seed()); + const walk = keysetWalk(store.read, { pageSize: 3 }); + expect(await collect(walk)).toEqual([...seed()].map((r) => r.id).sort()); + expect(walk.scanned).toBe(8); + expect(walk.truncated).toBe(false); + }); + + it('applies the caller filter to every page', async () => { + const store = makeStore(seed()); + const walk = keysetWalk(store.read, { where: { status: 'open' }, pageSize: 2 }); + expect(await collect(walk)).toEqual(['r01', 'r02', 'r05', 'r07', 'r11']); + }); + + it('keeps every row when the walk DELETES as it goes — the offset failure', async () => { + // An index rebuild deletes what it has processed. With `offset` the + // remaining rows shift left under the cursor and roughly half are never + // visited. Seeking past the last key cannot be shifted. + const store = makeStore(seed()); + const seen: string[] = []; + const walk = keysetWalk(store.read, { pageSize: 2 }); + for await (const page of walk.pages()) { + for (const row of page) { + seen.push(row.id); + const at = store.rows.findIndex((r) => r.id === row.id); + if (at >= 0) store.rows.splice(at, 1); + } + } + expect(seen.sort()).toEqual([...seed()].map((r) => r.id).sort()); + expect(store.rows).toHaveLength(0); + }); + + it('keeps every row when the walk UPDATES as it goes', async () => { + // A backfill rewrites each row it reads. The key does not move, so the + // walk does not either. + const store = makeStore(seed()); + const seen: string[] = []; + const walk = keysetWalk(store.read, { pageSize: 3 }); + for await (const page of walk.pages()) { + for (const row of page) { + seen.push(row.id); + row.status = 'processed'; + } + } + expect(seen.sort()).toEqual([...seed()].map((r) => r.id).sort()); + expect(store.rows.every((r) => r.status === 'processed')).toBe(true); + }); + + it('composes the cursor with a caller filter on the SAME column, never overwriting it', async () => { + // The subtle one: spreading the seek into the caller's object would drop + // this `$in` and hand back rows the caller excluded. + const store = makeStore(seed()); + const walk = keysetWalk(store.read, { + where: { id: { $in: ['r01', 'r05', 'r11'] } }, + pageSize: 2, + }); + expect(await collect(walk)).toEqual(['r01', 'r05', 'r11']); + }); + + it('reports `max` truncation instead of passing a partial scan off as complete', async () => { + const store = makeStore(seed()); + const walk = keysetWalk(store.read, { pageSize: 3, max: 5 }); + const seen = await collect(walk); + expect(seen).toHaveLength(5); + expect(walk.scanned).toBe(5); + expect(walk.truncated).toBe(true); + }); + + it('is NOT truncated when the source ends at exactly `max`', async () => { + // The source was exhausted at exactly the cap; nothing was withheld, so + // reporting truncation here would send a caller looking for rows that do + // not exist. This is what the one-row overshoot probe buys. + const store = makeStore(seed()); + const walk = keysetWalk(store.read, { pageSize: 3, max: 8 }); + expect(await collect(walk)).toHaveLength(8); + expect(walk.truncated).toBe(false); + }); + + it('is truncated when `max` lands exactly on a page boundary with rows left', async () => { + const store = makeStore(seed()); + const walk = keysetWalk(store.read, { pageSize: 3, max: 6 }); + expect(await collect(walk)).toHaveLength(6); + expect(walk.truncated).toBe(true); + }); + + it('stops instead of spinning when the reader ignores the seek', async () => { + // A reader that drops the `$gt` hands back the same page forever. Before + // this guard the walk hung — and a hang is the one failure nobody can read + // off a CI log. It now stops and says it was truncated. + let calls = 0; + const walk = keysetWalk( + async (q) => { + calls++; + if (calls > 50) throw new Error('walk did not terminate'); + return seed().slice(0, q.limit); // seek ignored + }, + { pageSize: 2 }, + ); + const seen = await collect(walk); + expect(seen).toHaveLength(4); // two pages, then the cursor failed to advance + expect(walk.truncated).toBe(true); + expect(calls).toBeLessThan(5); + }); + + it('stops instead of spinning when a row has no key', async () => { + const store = makeStore([{ id: 'r01', status: 'open' }, { id: undefined as any, status: 'open' }]); + const walk = keysetWalk( + async (q) => store.rows.slice(0, q.limit), + { pageSize: 2 }, + ); + const seen = await collect(walk); + expect(seen).toHaveLength(2); + expect(walk.truncated).toBe(true); + }); + + it('issues the seek the driver can serve from an index', async () => { + // The emitted query is the point: ascending on the key, and a `$gt` on it + // from page two onward. An implementation that kept an offset would still + // return the right rows here and cost O(n²) doing it, so the query shape + // is what has to be asserted. + const store = makeStore(seed()); + const queries: KeysetPageQuery[] = []; + const walk = keysetWalk( + async (q) => { + queries.push(q); + return store.read(q); + }, + { pageSize: 3 }, + ); + await collect(walk); + + expect(queries[0]!.orderBy).toEqual([{ field: 'id', order: 'asc' }]); + expect(queries[0]!.where).toBeUndefined(); + expect(queries[1]!.where).toEqual({ id: { $gt: 'r03' } }); + expect(queries[2]!.where).toEqual({ id: { $gt: 'r09' } }); + expect(queries.every((q) => !('offset' in q))).toBe(true); + }); + + it('seeks on a caller-chosen key column', async () => { + const store = makeStore(seed()); + const queries: KeysetPageQuery[] = []; + const walk = keysetWalk( + async (q) => { + queries.push(q); + return store.read(q); + }, + { pageSize: 3, key: 'status' }, + ); + await collect(walk); + expect(queries[0]!.orderBy).toEqual([{ field: 'status', order: 'asc' }]); + expect(queries[1]!.where).toEqual({ status: { $gt: 'done' } }); + }); +}); diff --git a/packages/types/src/keyset-walk.ts b/packages/types/src/keyset-walk.ts new file mode 100644 index 0000000000..cac22cee8c --- /dev/null +++ b/packages/types/src/keyset-walk.ts @@ -0,0 +1,190 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * Seek-based (keyset) pagination for the batch walks that read a whole object. + * + * # Why this exists rather than `limit`/`offset` + * + * A background walk that pages with a growing `offset` — rebuild an index, + * verify file references, backfill a projection — is wrong in two ways that a + * seek fixes at once. + * + * **It can skip rows.** `LIMIT n OFFSET k` is a slice of an arrangement, and + * the arrangement has to be the *same* one on every page for the slices to + * partition the set. Drivers now guarantee that for a single read + * (objectstack#4363), but not across a walk that *mutates as it goes*: a + * backfill that updates each page, or a rebuild that deletes, changes the very + * set the next offset counts into. Rows shift past the cursor and are never + * visited. For a verifier that decides which files are still referenced, or an + * index rebuild that deletes what it did not see, a skipped row is not a slow + * page — it is a wrong answer that looks like a clean run. A seek predicate + * carries the position *in the data* instead of counting from the start, so an + * update cannot move a row past it and a delete cannot shift one under it. + * + * **It is quadratic.** The database must produce and discard every skipped row + * to honor an offset, so walking n rows in pages of p costs O(n²/p). On a + * 2M-row table the last pages were measured at ~1.1 s each against ~0.09 s for + * the first. A seek starts each page at the cursor, so every page costs the + * same: O(n) for the walk, and index-served throughout. + * + * # What it requires + * + * A column that is **unique and orderable** — `id` by default, which every + * object this driver-managed platform creates carries. An object without one + * (a federated table, ADR-0015) cannot be walked this way; callers that scan + * arbitrary registry objects already skip what they cannot read, and that is + * the correct outcome here too rather than a silent partial scan. + * + * # Shape + * + * `read` is the caller's own query — this owns the loop, the cursor and the + * `where` merge, and nothing else. Deliberately one implementation rather than + * the six hand-rolled copies it replaces: the cursor merge is the part that is + * easy to get subtly wrong (an object whose own `where` already constrains the + * key), and six copies of it drift silently. + * + * @example + * const walk = keysetWalk( + * (q) => engine.find('sys_approval_request', { ...q, fields: ['id'], context: SYSTEM_CTX }), + * { where: { status: 'pending' }, pageSize: 500 }, + * ); + * for await (const page of walk.pages()) { … } + * if (walk.truncated) { … } + */ + +/** The query a {@link keysetWalk} hands its reader: the caller's `where`, narrowed by the cursor. */ +export interface KeysetPageQuery { + /** The caller's `where`, AND-ed with the seek predicate once the walk has a cursor. */ + where?: unknown; + /** Always ascending on the key column — the walk's order IS the seek order. */ + orderBy: Array<{ field: string; order: 'asc' }>; + /** Page size. */ + limit: number; +} + +export interface KeysetWalkOptions { + /** The caller's filter, applied to every page. */ + where?: unknown; + /** Rows per page. */ + pageSize: number; + /** + * Stop after this many rows and set {@link KeysetWalk.truncated}. Omit for an + * unbounded walk. A cap is not a failure — it is how a scan bounds its own + * cost — but it must be reported, or a partial scan reads as a complete one. + */ + max?: number; + /** Unique, orderable column to seek on. Defaults to `id`. */ + key?: string; +} + +export interface KeysetWalk { + /** Pages, in key order, until the source is exhausted or `max` is reached. */ + pages(): AsyncGenerator; + /** Rows yielded so far. */ + readonly scanned: number; + /** True when `max` stopped the walk before the source was exhausted. */ + readonly truncated: boolean; +} + +/** + * AND the seek predicate onto the caller's filter. + * + * Uses `$and` rather than spreading the key into the same object: a caller + * whose own `where` already constrains the key column (`{ id: { $in: [...] } }`) + * would otherwise have that constraint silently overwritten by the cursor, and + * the walk would return rows the caller excluded. `$and` composes instead of + * colliding, and every driver executes it. + */ +function withCursor(where: unknown, key: string, cursor: unknown): unknown { + const seek = { [key]: { $gt: cursor } }; + if (where == null) return seek; + if (typeof where === 'object' && Object.keys(where as object).length === 0) return seek; + return { $and: [where, seek] }; +} + +/** + * Walk an object by seeking past the last key rather than counting from the + * start. See the module comment for why every batch scan should. + * + * `read` receives a {@link KeysetPageQuery} and returns the page; the caller + * owns everything else about the query (projection, context, object name). + */ +export function keysetWalk>( + read: (query: KeysetPageQuery) => Promise, + options: KeysetWalkOptions, +): KeysetWalk { + const key = options.key ?? 'id'; + const pageSize = options.pageSize; + let scanned = 0; + let truncated = false; + + async function* pages(): AsyncGenerator { + let cursor: unknown = undefined; + for (;;) { + const want = options.max == null ? pageSize : Math.min(pageSize, options.max - scanned); + if (want <= 0) { + truncated = true; + return; + } + + // When `max` clips this page, ask for ONE more row than we will yield. + // That extra row is the difference between "the cap stopped us" and "the + // source ended at exactly the cap" — without it a walk that read + // everything still reports `truncated`, and a caller acting on that goes + // looking for rows that were never withheld. + const clipped = options.max != null && want < pageSize; + const page = await read({ + where: cursor === undefined ? options.where : withCursor(options.where, key, cursor), + orderBy: [{ field: key, order: 'asc' }], + limit: clipped ? want + 1 : want, + }); + if (!Array.isArray(page) || page.length === 0) return; + + const overflow = clipped && page.length > want; + const emit = overflow ? page.slice(0, want) : page; + scanned += emit.length; + yield emit; + + if (overflow) { + truncated = true; + return; + } + + const last = emit[emit.length - 1]?.[key]; + // A row without the key column cannot advance the cursor, and continuing + // would re-read the same page forever. Stop and report it as truncation + // rather than spin: a walk that cannot seek is not a walk that finished. + if (last === undefined || last === null) { + truncated = true; + return; + } + // The same stop for a reader that did not APPLY the seek — the cursor + // comes back no further along than it went in, so the next page would be + // this page again, forever. Production drivers execute the predicate; + // a test double or a future reader that quietly drops it would otherwise + // hang rather than fail, and a hang is the one failure nobody can read. + if (cursor !== undefined && !(String(last) > String(cursor))) { + truncated = true; + return; + } + cursor = last; + + // A short page means the source is exhausted. + if (emit.length < want) return; + // Reaching the cap on a full, unclipped page: more rows may remain, and + // the next iteration's `want <= 0` reports that as truncation. + if (options.max != null && scanned >= options.max && !clipped) continue; + if (options.max != null && scanned >= options.max) return; + } + } + + return { + pages, + get scanned() { + return scanned; + }, + get truncated() { + return truncated; + }, + }; +} diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 6920f0b2f5..6c7903a29f 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -1264,6 +1264,9 @@ importers: '@objectstack/spec': specifier: workspace:* version: link:../../spec + '@objectstack/types': + specifier: workspace:* + version: link:../../types devDependencies: '@objectstack/objectql': specifier: workspace:* diff --git a/scripts/check-driver-conformance.mjs b/scripts/check-driver-conformance.mjs new file mode 100644 index 0000000000..db58e49acd --- /dev/null +++ b/scripts/check-driver-conformance.mjs @@ -0,0 +1,388 @@ +#!/usr/bin/env node +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. +// +// check-driver-conformance -- every driver runs every shared conformance +// case-set, or its absence is a recorded, tracked decision (#4363). +// +// `packages/spec/src/data/*-conformance.ts` holds the case-sets that exist so +// the independent driver implementations answer ONE standard rather than each +// having its own idea: filter combinator semantics (#3774), temporal storage +// form (ADR-0053), deterministic paged reads (objectui#3106 / #4363). Each was +// introduced with some version of the claim that a future driver "is held to +// this by a gate rather than by remembering it". +// +// There was no gate. The case-sets are exports sitting in a package; nothing +// obliged a driver to import them, and the coverage matrix had three holes on +// the day this script was written -- one of them in the very case-set whose +// changeset made the claim. That is the declared-not-enforced shape Prime +// Directive #10 is about, so the fix is the gate rather than a correction to +// the sentence. +// +// node scripts/check-driver-conformance.mjs +// node scripts/check-driver-conformance.mjs --self-test +// +// ## Scope: the IDataDriver implementers +// +// `packages/plugins/driver-*` -- discovered from disk, never listed here, so a +// new driver package is in scope the moment it exists. Other consumers of the +// same case-sets (`packages/formula`'s `matchesFilter`, service-analytics' +// native-SQL strategy) are DELIBERATELY out of scope: they are not drivers, +// they implement a different subset, and enrolling them means answering "which +// case-sets even apply" per consumer -- a question this gate would have to +// guess at. Their coverage is asserted by their own suites today. If that rots, +// it wants its own gate, not a looser one here. +// +// ## Invariants +// +// CONSUMED every (driver x case-set) cell is either covered -- some file +// under the package's `src/` imports the case-set's marker export +// from `@objectstack/spec/data` -- or carries a DEBT/EXEMPT entry +// below. A new driver must arrive covered. +// CLASSIFIED every case-set exported by `spec/src/data/*-conformance.ts` is +// named in CASE_SETS. A new shared fixture nobody classified +// fails this run rather than silently dropping out of coverage -- +// the #4203 lesson, applied in the direction that actually rots. +// RECONCILED in both directions: a DEBT/EXEMPT entry for a cell that is now +// covered, for a driver that no longer exists, or for a case-set +// that no longer exists, is an error. A ledger that can only +// accrete rots into a list nobody trusts. +// +// ## What "covered" means, and what it deliberately does not +// +// This checks that the shared fixture is IMPORTED AND REFERENCED, not that the +// assertions over it are good. A gate cannot judge assertion quality, and one +// that tried would be the kind of verifier that reports success while degrading +// (route-ownership rule 3). What it can do is make the absence loud, which is +// the failure mode that actually happened: three drivers silently not running a +// standard three changesets said they were held to. +// +// ## DEBT is frozen debt, not a permission slip +// +// Every entry was measured against `main`. To clear one: write the suite, then +// delete the entry in the same PR. Deleting without the suite fails CONSUMED; +// keeping the entry alongside the suite fails RECONCILED. + +import { mkdirSync, readdirSync, readFileSync, rmSync, statSync, writeFileSync } from 'node:fs'; +import { join, dirname } from 'node:path'; +import { fileURLToPath } from 'node:url'; + +const ROOT = join(dirname(fileURLToPath(import.meta.url)), '..'); +const DRIVERS_DIR = join(ROOT, 'packages', 'plugins'); +const CASE_SETS_DIR = join(ROOT, 'packages', 'spec', 'src', 'data'); + +// ── The case-sets ─────────────────────────────────────────────────────────── +// +// `marker` is the export a suite cannot run the case-set without naming, so its +// presence is evidence the fixture is actually driven rather than re-declared +// locally. Reconciled against the files on disk by CLASSIFIED below. + +const CASE_SETS = [ + { + file: 'filter-logic-conformance.ts', + marker: 'FILTER_LOGIC_CASES', + what: 'filter combinator semantics ($and/$or/$not nesting) — #3774', + }, + { + file: 'temporal-conformance.ts', + marker: 'TEMPORAL_CASES', + what: 'temporal storage form and comparand coercion — ADR-0053', + }, + { + file: 'temporal-conformance.ts', + marker: 'TEMPORAL_TIME_CASES', + what: 'canonical `Field.time` storage and comparison — #3994', + }, + { + file: 'pagination-conformance.ts', + marker: 'PAGINATION_CASES', + what: 'a sorted paged read is a partition — objectui#3106', + }, + { + file: 'pagination-conformance.ts', + marker: 'PAGINATION_UNORDERED_CASES', + what: 'an UNSORTED paged read is a partition too — #4363', + }, +]; + +// ── The ledger ────────────────────────────────────────────────────────────── +// +// One entry per uncovered (driver x case-set) cell. `kind` is DEBT (should be +// covered, is not yet) or EXEMPT (cannot meaningfully apply). Both are measured +// claims; neither is a default. + +const LEDGER = [ + { + driver: 'driver-mongodb', + marker: 'FILTER_LOGIC_CASES', + kind: 'DEBT', + issue: 'https://github.com/objectstack-ai/objectstack/issues/4405', + why: + "`mongodb-filter.ts`'s `translateFilter` is an independent FilterCondition " + + 'backend — the fifth, and the one #3774 never enrolled when it named "the four". ' + + 'Its $and/$or/$not translation shares no code with the SQL or in-memory paths.', + }, + { + driver: 'driver-sqlite-wasm', + marker: 'FILTER_LOGIC_CASES', + kind: 'DEBT', + issue: 'https://github.com/objectstack-ai/objectstack/issues/4405', + why: + 'Inherits SqlDriver\'s filter compiler, so the risk is the sql.js dialect executing ' + + 'the compiled predicate, not the predicate being built wrong — the same risk its ' + + 'temporal and pagination suites already cover for their clauses. Lower value than ' + + 'the mongodb row above, tracked with it rather than exempted, because "inherits, ' + + 'therefore fine" is exactly the assumption those two suites exist to disprove.', + }, +]; + +// ── Discovery ─────────────────────────────────────────────────────────────── + +const listDir = (dir) => { + try { + return readdirSync(dir); + } catch { + return []; + } +}; + +/** Driver packages, from disk — never a hardcoded list. */ +function discoverDrivers() { + return listDir(DRIVERS_DIR) + .filter((name) => name.startsWith('driver-')) + .filter((name) => { + try { + return statSync(join(DRIVERS_DIR, name, 'package.json')).isFile(); + } catch { + return false; + } + }) + .sort(); +} + +/** Every `*-conformance.ts` under spec/src/data, and the case-set exports in it. */ +function discoverCaseSets() { + const found = []; + for (const file of listDir(CASE_SETS_DIR).filter((f) => f.endsWith('-conformance.ts'))) { + const src = readFileSync(join(CASE_SETS_DIR, file), 'utf8'); + // A case-set is an exported `_CASES` const: the thing a suite iterates. + // `_ROWS` / `_ALL_IDS` are its fixture data, driven through the cases. + for (const m of src.matchAll(/^export const ([A-Z0-9_]*_CASES)\b/gm)) { + found.push({ file, marker: m[1] }); + } + } + return found; +} + +/** Every `.ts` file under a directory, recursively. */ +function walkTs(dir, out = []) { + for (const entry of listDir(dir)) { + if (entry === 'node_modules' || entry === 'dist') continue; + const full = join(dir, entry); + let s; + try { + s = statSync(full); + } catch { + continue; + } + if (s.isDirectory()) walkTs(full, out); + else if (entry.endsWith('.ts')) out.push(full); + } + return out; +} + +/** + * Does this driver package drive `marker`? + * + * Requires BOTH an import naming it from `@objectstack/spec/data` and a + * reference outside that import — an unused import is not coverage, and it is + * the shape a half-finished suite leaves behind. + */ +function consumes(driverDir, marker) { + for (const file of walkTs(join(driverDir, 'src'))) { + const src = readFileSync(file, 'utf8'); + if (!src.includes(marker)) continue; + const imported = new RegExp( + `import[\\s\\S]*?\\b${marker}\\b[\\s\\S]*?from\\s+['"]@objectstack/spec/data['"]`, + ).test(src); + if (!imported) continue; + // Count references outside the import statement(s). + const withoutImports = src.replace(/import[\s\S]*?from\s+['"][^'"]+['"];?/g, ''); + if (new RegExp(`\\b${marker}\\b`).test(withoutImports)) return file; + } + return null; +} + +// ── The run ───────────────────────────────────────────────────────────────── + +function audit() { + const drivers = discoverDrivers(); + const errors = []; + const rows = []; + + // CLASSIFIED — both directions between CASE_SETS and the files on disk. + const onDisk = discoverCaseSets(); + const classified = new Set(CASE_SETS.map((c) => c.marker)); + for (const { file, marker } of onDisk) { + if (!classified.has(marker)) { + errors.push( + `CLASSIFIED: ${file} exports ${marker}, which no row of CASE_SETS names. ` + + 'Classify it (and say which drivers must run it) rather than letting a new shared ' + + 'standard start life uncovered.', + ); + } + } + const onDiskMarkers = new Set(onDisk.map((c) => c.marker)); + for (const c of CASE_SETS) { + if (!onDiskMarkers.has(c.marker)) { + errors.push(`CLASSIFIED: CASE_SETS names ${c.marker}, which ${c.file} no longer exports.`); + } + } + + // CONSUMED + collect the matrix. + const ledgerHit = new Set(); + for (const driver of drivers) { + const dir = join(DRIVERS_DIR, driver); + for (const c of CASE_SETS) { + const where = consumes(dir, c.marker); + const entry = LEDGER.find((l) => l.driver === driver && l.marker === c.marker); + if (entry) ledgerHit.add(`${driver}::${c.marker}`); + + if (where && entry) { + errors.push( + `RECONCILED: ${driver} now runs ${c.marker} (${where.slice(ROOT.length + 1)}), ` + + `but the ledger still carries a ${entry.kind} entry for it. Delete the entry.`, + ); + rows.push({ driver, marker: c.marker, state: 'covered' }); + } else if (where) { + rows.push({ driver, marker: c.marker, state: 'covered' }); + } else if (entry) { + rows.push({ driver, marker: c.marker, state: entry.kind.toLowerCase() }); + } else { + errors.push( + `CONSUMED: ${driver} does not run ${c.marker} (${c.what}). Add a suite that drives ` + + 'the shared cases, or add a measured DEBT/EXEMPT entry to the ledger in ' + + 'scripts/check-driver-conformance.mjs saying why not.', + ); + rows.push({ driver, marker: c.marker, state: 'MISSING' }); + } + } + } + + // RECONCILED — ledger rows pointing at things that no longer exist. + for (const entry of LEDGER) { + if (!drivers.includes(entry.driver)) { + errors.push(`RECONCILED: ledger entry for ${entry.driver}, which is not a driver package.`); + } else if (!classified.has(entry.marker)) { + errors.push(`RECONCILED: ledger entry names ${entry.marker}, which CASE_SETS does not.`); + } + } + + return { drivers, rows, errors }; +} + +function report() { + const { drivers, rows, errors } = audit(); + + const covered = rows.filter((r) => r.state === 'covered').length; + const debt = rows.filter((r) => r.state === 'debt').length; + const exempt = rows.filter((r) => r.state === 'exempt').length; + + const width = Math.max(...drivers.map((d) => d.length), 8); + console.log(`\ndriver conformance matrix (${drivers.length} drivers x ${CASE_SETS.length} case-sets)\n`); + console.log( + ' ' + 'driver'.padEnd(width) + ' ' + CASE_SETS.map((c) => c.marker.replace(/_CASES$/, '')).join(' '), + ); + for (const driver of drivers) { + const cells = CASE_SETS.map((c) => { + const row = rows.find((r) => r.driver === driver && r.marker === c.marker); + const glyph = { covered: 'ok', debt: 'DEBT', exempt: 'exempt', MISSING: 'MISSING' }[row.state]; + return glyph.padEnd(c.marker.replace(/_CASES$/, '').length); + }); + console.log(' ' + driver.padEnd(width) + ' ' + cells.join(' ')); + } + console.log(''); + + if (errors.length) { + for (const e of errors) console.error(` x ${e}`); + console.error(`\ncheck-driver-conformance: ${errors.length} problem(s).\n`); + process.exit(1); + } + + // Print the ledger's reasons, not only its counts. An entry whose + // justification is never surfaced is how a ledger decays into a list nobody + // reads — and these rows are the whole reason this run is green. + for (const entry of LEDGER) { + console.log(` ${entry.kind} ${entry.driver} × ${entry.marker}`); + console.log(` ${entry.why}`); + if (entry.issue) console.log(` tracked: ${entry.issue}`); + } + if (LEDGER.length) console.log(''); + + console.log( + `check-driver-conformance: OK — ${covered} covered cell(s), ${debt} in the DEBT ledger, ` + + `${exempt} exempt.\n`, + ); +} + +// ── Self-test ─────────────────────────────────────────────────────────────── +// +// A guard that cannot fail is not a guard. This drives the three invariants +// against synthetic inputs so a refactor that neuters the detection fails here +// rather than silently passing every future PR. + +function selfTest() { + const failures = []; + const expect = (label, cond) => { + if (!cond) failures.push(label); + }; + + // CONSUMED: an unused import must not count as coverage. + const tmp = join(ROOT, 'node_modules', '.check-driver-conformance-selftest'); + try { + mkdirSync(join(tmp, 'src'), { recursive: true }); + writeFileSync( + join(tmp, 'src', 'a.test.ts'), + "import { PAGINATION_CASES } from '@objectstack/spec/data';\n// never referenced again\n", + ); + expect('unused import must not count as coverage', consumes(tmp, 'PAGINATION_CASES') === null); + + writeFileSync( + join(tmp, 'src', 'a.test.ts'), + "import { PAGINATION_CASES } from '@objectstack/spec/data';\nfor (const c of PAGINATION_CASES) {}\n", + ); + expect('driven import must count as coverage', consumes(tmp, 'PAGINATION_CASES') !== null); + + // A locally re-declared fixture is not the shared standard. + writeFileSync( + join(tmp, 'src', 'a.test.ts'), + 'const PAGINATION_CASES = [];\nfor (const c of PAGINATION_CASES) {}\n', + ); + expect( + 'local re-declaration must not count as coverage', + consumes(tmp, 'PAGINATION_CASES') === null, + ); + } finally { + rmSync(tmp, { recursive: true, force: true }); + } + + // CLASSIFIED: discovery must actually find the case-sets on disk, or the + // reconciliation is vacuous and every future fixture drops out silently. + const found = discoverCaseSets().map((c) => c.marker); + expect('discovers FILTER_LOGIC_CASES on disk', found.includes('FILTER_LOGIC_CASES')); + expect('discovers TEMPORAL_CASES on disk', found.includes('TEMPORAL_CASES')); + expect('discovers PAGINATION_UNORDERED_CASES on disk', found.includes('PAGINATION_UNORDERED_CASES')); + + // Discovery must find the drivers, for the same reason. + const drivers = discoverDrivers(); + expect('discovers driver packages from disk', drivers.length >= 3 && drivers.includes('driver-sql')); + + if (failures.length) { + for (const f of failures) console.error(` x self-test: ${f}`); + console.error(`\ncheck-driver-conformance --self-test: ${failures.length} failure(s).\n`); + process.exit(1); + } + console.log('OK self-test: detects driven / unused / re-declared fixtures, and discovers both axes.'); +} + +if (process.argv.includes('--self-test')) selfTest(); +else report();