diff --git a/.changeset/add-cached-cursor-pagination.md b/.changeset/add-cached-cursor-pagination.md new file mode 100644 index 0000000000..0b300bd6eb --- /dev/null +++ b/.changeset/add-cached-cursor-pagination.md @@ -0,0 +1,7 @@ +--- +'@tanstack/query-db-collection': patch +--- + +Add `createCursorPager` to fulfill offset/limit requests from endpoints with opaque continuation tokens. Reuse fresh backend pages through the existing QueryClient, with Query-managed expiry, invalidation and garbage collection, while retaining the existing UI peek-ahead behavior. + +Keep manual raw-row writes from overwriting other cache formats or marking them fresh. Preserve wrapped-response writes and seeding of empty row caches. diff --git a/docs/collections/query-collection.md b/docs/collections/query-collection.md index 25b5f565de..543bddeeb9 100644 --- a/docs/collections/query-collection.md +++ b/docs/collections/query-collection.md @@ -831,16 +831,155 @@ const { data, fetchNextPage, hasNextPage } = useLiveInfiniteQuery( ``` Reject failed requests instead of returning partial rows as success. An -unlimited request must drain until the endpoint reports exhaustion. If the -endpoint uses opaque cursors instead of page numbers, keep that cursor handling -inside `queryFn` or its adapter; honoring a new offset may require starting at -the beginning again. The hook does not maintain remote cursor history. +unlimited request must drain until the endpoint reports exhaustion. For an +endpoint with opaque cursors, use `createCursorPager` as shown below. The hook +does not maintain remote cursor history on its own. Manually appending rows with `writeUpsert` is a separate, lower-level loading strategy. It does not make an eager `queryFn` incremental: a later successful refetch still replaces its complete state and can remove appended rows. `staleTime: Infinity` does not prevent explicit refetch or invalidation. +#### Endpoints with opaque cursors + +Query Collections ask for arbitrary windows, such as rows 40–59. An +opaque-cursor API cannot jump to row 40. It can only fetch the first page, then +follow the continuation token returned with each page. + +`createCursorPager` bridges those two protocols. It stores complete backend +pages in your existing QueryClient and follows their cursors until it has enough +rows for the requested `offset` and `limit`. A later request can reuse that +fresh run of pages and fetch only the missing suffix. The hook still requests +an extra row to decide `hasNextPage`; the pager just fulfills the resulting +window. + +```typescript +import { createCollection } from '@tanstack/db' +import { createCursorPager, queryCollectionOptions } from '@tanstack/query-db-collection' + +type Post = { id: number; createdAt: number; title: string } + +const postsCollection = createCollection( + queryCollectionOptions({ + queryClient, + queryKey: ['posts', 'rows'], + syncMode: 'on-demand', + getKey: (post: Post) => post.id, + queryFn: async (ctx) => { + const { where, orderBy, offset = 0, limit } = ctx.meta?.loadSubsetOptions ?? {} + + // Translate every filter and sort, or reject unsupported expressions. + // This stable, serializable value identifies one ordered result sequence. + const request = api.translatePostQuery({ where, orderBy }) + + const pager = createCursorPager({ + queryClient, + queryKey: ['posts', 'cursor-pages', request], + staleTime: 60_000, + gcTime: 5 * 60_000, + fetchPage: async (cursor, signal) => { + const response = await api.listPosts({ ...request, cursor, signal }) + + // Only null means that the source is exhausted. + return { rows: response.items, nextCursor: response.nextCursor ?? null } + }, + }) + + return pager.read({ offset, limit }, ctx.signal) + }, + }), +) +``` + +##### Give each result sequence its own key + +The pager's query key identifies one ordered backend result, not one requested +window. Leave `offset` and `limit` out of the key so wider reads can extend the +same cached prefix. Include everything that changes the result—such as the +tenant, source, filters, and order. + +Keep row data and cursor pages under sibling keys: + +```text +['posts', 'rows'] QueryCollection row array +['posts', 'cursor-pages', request] backend pages and cursors +``` + +They cannot share a key because the two cache entries have different shapes. +Do not put cursor pages beneath the row key either: manual collection writes +target entries under the row prefix. A shared resource prefix such as +`['posts']` still lets you cancel or invalidate both kinds of data together. + +The example creates a small pager object for each `queryFn` call. Those objects +still share cached pages and in-flight fetches through QueryClient. Each pager +serializes its own reads. Retain one pager per result sequence only if calls +must also share that queue or you need to call `pager.reset()` later. + +##### What happens when the collection asks for rows + +For fresh data, the pager uses every cached page it can and fetches only the +missing suffix. `nextCursor: null` is the sole exhaustion signal: a short or +empty page with another cursor does not end the sequence. An omitted `limit` +drains the source; a zero limit performs no fetch. Repeated continuation tokens +throw instead of looping forever. + +Query owns freshness, garbage collection, retries, and invalidation. Pass +`staleTime` and `gcTime` to the pager or let it inherit QueryClient defaults. +Configure retries through QueryClient defaults for the page-key prefix; +imperative Query fetches do not retry by default. The pager keeps every page in +its original form, so Query defaults for `maxPages` and `select` do not apply. + +When cached pages are stale, the next read refetches every page loaded so far, +starting with the first. It does this even for a shallow window because old +continuation tokens belong to the old sequence. Staleness does not start a +background timer. The page query has no lasting observer, so its `gcTime` may +expire while the collection's published rows remain visible. + +A failed read rejects rather than returning a partial window. Query keeps the +last successful pages, so a later read can retry. Treat their rows as immutable, +just like other Query data. `read()` returns references from the cached pages, +not detached copies, and does not promise the same array or object identity +across calls. + +##### Refresh or cancel the sequence + +After a mutation—or whenever you need a forced refresh—cancel the shared +resource prefix before invalidating it: + +```typescript +await queryClient.cancelQueries({ queryKey: ['posts'] }) +await queryClient.invalidateQueries({ queryKey: ['posts'] }) +``` + +Cancelling first stops an old page append from completing after invalidation and +marking the old sequence fresh again. It rejects readers waiting on that fetch +with an `AbortError` and prevents a late result from entering the cache. The +backend must honor the signal if you also want to stop its network work. + +Aborting the signal passed to `read()` rejects only that reader; it does not +cancel a page fetch shared with another reader. A cancelling Query refetch moves +waiting readers to the replacement fetch. `collection.utils.refetch()` refreshes +the row queries but may reuse fresh cursor pages. `pager.reset()` removes one +pager's exact page key and cancels its queued reads, but does not refresh the +collection's row query. + +##### Backend contract + +The helper handles cursor traversal only. Your adapter must: + +- translate every requested `where` and `orderBy`, or reject unsupported + expressions; +- apply filters and ordering before pagination; +- use a deterministic total order, including a unique tie-breaker; +- keep that ordered result consistent while its cursor sequence is read; and +- return `nextCursor: null` only when the result is authoritatively exhausted. + +`LoadSubsetOptions.cursor` contains query-expression hints, not the backend's +opaque token. `createCursorPager` therefore addresses the cursor sequence by +offset and deliberately does not consume those hints. No client-side pager can +make a stable snapshot from an endpoint whose ordering changes between pages; +the endpoint must define its own consistency and cursor-expiry rules. + ## Important Behaviors ### Full State Sync diff --git a/docs/contributing/oracle-coverage.md b/docs/contributing/oracle-coverage.md index 1b2aa2f157..28376a378d 100644 --- a/docs/contributing/oracle-coverage.md +++ b/docs/contributing/oracle-coverage.md @@ -35,6 +35,7 @@ comment and the current API/architecture contract before extending its model. | Drafts and native values | [proxy](../../packages/db/tests/proxy.test.ts), [detachment](../../packages/db/tests/proxy-detachment-contract.test.ts), [iteration](../../packages/db/tests/proxy-iteration-contract.test.ts) | Native-operation controls, exact patches and actual stored rows; alias/cycle/adversarial-key histories. General native-mutator and symbol-write support is not established by a plain-object oracle. | | Query DB and observer | [ownership](../../packages/query-db-collection/tests/ownership-lifecycle.oracle.test.ts), [load lifecycle](../../packages/query-db-collection/tests/load-subset-lifecycle-oracle.test.ts), [observer histories](../../packages/db/tests/live-query-observer-history.property.test.ts) | Real QueryClient boundary and a per-listener eligibility ledger, not a duplicate dispatch queue. Check reentry, peer survival, FIFO and disposal independently of final rows. | | Ordered acquisition | [pagination](../../packages/db/tests/query/pagination-oracle.property.test.ts), [ordered work](../../packages/db/tests/query/ordered-work-oracle.property.test.ts), [ordered lifecycle](../../packages/db/tests/query/ordered-lifecycle-oracle.property.test.ts) | Complete finite provider results, pending windows, ties/nulls, ownership and documented repair timing. Request completion is not proof of unrequested source extent. | +| Opaque backend pagination | [window oracle](../../packages/query-db-collection/tests/cursor-pagination.oracle.test.ts), [cache histories](../../packages/query-db-collection/tests/cursor-pagination.cache-oracle.test.ts), [cache publication](../../packages/query-db-collection/tests/cursor-pagination.publication-oracle.test.ts), [browser acquisition boundaries](../../packages/query-db-collection/tests/cursor-pagination.boundary-oracle.test.ts), [QueryCollection integration](../../packages/query-db-collection/tests/cursor-pagination.integration.test.ts) | Full filter/sort/slice reference, opaque token transport, actual Query cache expiry/invalidation/GC, forced refresh during growth, protocol failure publication/recovery, bounded slice work, nested cancellation/replacement, reader abort, browser retry defaults, manual-write cache isolation, and production window publications. Stable backend sequences; not snapshot guarantees for changing endpoints. Peek-ahead remains enabled. | | Electric and TrailBase | [Electric histories](../../packages/electric-db-collection/tests/electric-oracle.property.test.ts), [PostgreSQL semantics](../../packages/electric-db-collection/e2e/sql-predicate-semantics.e2e.test.ts), [TrailBase contract](../../packages/trailbase-db-collection/tests/ORACLE.md) | Installed SDK delivery/framing, independent predicates, exact subscription arguments and late errors. SDK fixtures and a real service test earn different credit. | | PowerSync | [tests](../../packages/powersync-db-collection/tests) | Applied receipt positions crossed with held peers, native SQLite/SDK and cleanup evidence. A timeout mutant proves a progress failure, not every value assertion. | | SQLite persistence and native hosts | [persisted histories](../../packages/db-sqlite-persistence-core/tests/persisted.test.ts), [driver contracts](../../packages/db-sqlite-persistence-core/tests/contracts/sqlite-driver-contract.ts), [113-law manifest](../../packages/db-collection-e2e/src/fixtures/persisted-conformance-manifest.ts) | Cache/remote rejection/peer/reopen histories and exact driver results. The manifest excludes progressive and move suites; registration and shim runs are not device execution. | diff --git a/packages/db/tests/oracle-config.ts b/packages/db/tests/oracle-config.ts index 582d21686d..d9b4d3749b 100644 --- a/packages/db/tests/oracle-config.ts +++ b/packages/db/tests/oracle-config.ts @@ -3,6 +3,21 @@ import { oracleReplayReporter } from './oracle-replay-witness.js' type OracleEnvironment = Record const staticOracleProperties = [ + `cursor-pagination.no-peek`, + `cursor-pagination.history`, + `cursor-pagination.cache`, + `cursor-pagination.nested-cancellation`, + `cursor-pagination.reader-abort`, + `cursor-pagination.manual-write`, + `cursor-pagination.backend-ownership`, + `cursor-pagination.refresh-publication`, + `cursor-pagination.protocol-publication`, + `cursor-pagination.slice-work`, + `cursor-pagination.defaults`, + `cursor-pagination.cancellation`, + `cursor-pagination.partition`, + `cursor-pagination.reset`, + `cursor-pagination.failure`, `oracle-replay.calibration`, `trailbase.lifecycle`, `electric.bound-descriptor-history`, diff --git a/packages/query-db-collection/package.json b/packages/query-db-collection/package.json index 29264d93a8..e451088280 100644 --- a/packages/query-db-collection/package.json +++ b/packages/query-db-collection/package.json @@ -21,7 +21,7 @@ "dev": "vite build --watch", "lint": "eslint . --fix", "test": "vitest run", - "test:oracles": "vitest run tests/includes-work-counter-oracle.test.ts tests/load-subset-lifecycle-oracle.test.ts tests/ownership-lifecycle.oracle.test.ts", + "test:oracles": "vitest run tests/includes-work-counter-oracle.test.ts tests/load-subset-lifecycle-oracle.test.ts tests/ownership-lifecycle.oracle.test.ts tests/cursor-pagination.oracle.test.ts tests/cursor-pagination.cache-oracle.test.ts tests/cursor-pagination.publication-oracle.test.ts tests/cursor-pagination.boundary-oracle.test.ts tests/cursor-pagination.integration.test.ts", "test:e2e": "vitest run --config vitest.e2e.config.ts" }, "type": "module", diff --git a/packages/query-db-collection/src/cursor-pagination.ts b/packages/query-db-collection/src/cursor-pagination.ts new file mode 100644 index 0000000000..7f9fbfc78a --- /dev/null +++ b/packages/query-db-collection/src/cursor-pagination.ts @@ -0,0 +1,241 @@ +import { InfiniteQueryObserver, isCancelledError } from '@tanstack/query-core' +import type { InfiniteData, QueryClient, QueryKey } from '@tanstack/query-core' + +/** One backend page. Only null means the ordered result is exhausted. */ +export interface CursorPage { + rows: ReadonlyArray + nextCursor: string | null +} + +export interface CursorPagerOptions { + queryClient: QueryClient + /** Dedicated infinite-query key: include the source, filters and order, not the window. */ + queryKey: QueryKey + fetchPage: ( + cursor: string | undefined, + signal: AbortSignal, + ) => Promise> + /** Query's freshness interval. Defaults to the QueryClient's setting. */ + staleTime?: number + /** Query's inactive cache lifetime. Defaults to the QueryClient's setting. */ + gcTime?: number +} + +export interface CursorPager { + read: ( + window: { offset?: number; limit?: number }, + signal?: AbortSignal, + ) => Promise> + /** Remove this key's cached pages and invalidate this pager's queued reads. */ + reset: () => void +} + +/** + * Fulfill offset/limit windows using opaque backend cursors. Query owns the + * pages, freshness, invalidation and garbage collection. Use one dedicated key + * per filtered, totally ordered source. Loading more reuses fresh pages; refreshing + * stale data rebuilds the loaded sequence from its first page. + * + * The key must not also be used for ordinary QueryCollection row arrays. + * Use sibling row/page prefixes under one resource prefix, not pages beneath + * the row prefix (manual writes update that whole prefix). For a forced refresh, + * cancel the resource prefix before invalidating it. A positive staleTime avoids + * refreshing on every read. + * + * Reads on one pager serialize; separate pagers share Query's cache and fetches. + * Aborting a reader discards its answer, not shared cached + * pages. Cancel the query through QueryClient to cancel its transport. Neither + * cancellation nor a TTL can repair a backend's inconsistent cursor sequence. + */ +export function createCursorPager({ + queryClient, + queryKey, + fetchPage, + staleTime, + gcTime, +}: CursorPagerOptions): CursorPager { + let generation = 0 + let tail = Promise.resolve() + const sequences = new WeakMap>() + type Pages = InfiniteData, string | undefined> + const nextCursor = (page: CursorPage) => { + if ( + !Array.isArray(page.rows) || + (page.nextCursor !== null && typeof page.nextCursor !== `string`) + ) + throw new TypeError( + `Invalid cursor page: expected rows and a string or null nextCursor`, + ) + return page.nextCursor + } + const options = { + queryKey, + // Offsets address the complete sequence, never a selected or evicted prefix. + maxPages: 0, + select: undefined, + enabled: true, + ...(staleTime === undefined ? {} : { staleTime }), + ...(gcTime === undefined ? {} : { gcTime }), + initialPageParam: undefined as string | undefined, + queryFn: async ({ + pageParam, + signal, + }: { + pageParam: string | undefined + signal: AbortSignal + }) => { + let params = sequences.get(signal) + if (pageParam === undefined || !params) { + // Refresh starts a new sequence; growth extends the validated cache. + // Query gives every page/retry in an acquisition the same signal. + params = new Set( + pageParam === undefined + ? [] + : queryClient.getQueryData< + InfiniteData, string | undefined> + >(queryKey)?.pageParams, + ) + sequences.set(signal, params) + } + params.add(pageParam) + const page = await fetchPage(pageParam, signal) + // Validate even the final response, before Query publishes success or + // resolves shared waiters. getNextPageParam runs too late for that. + const next = nextCursor(page) + if (next !== null && params.has(next)) { + throw new Error(`Backend repeated a continuation cursor`) + } + return page + }, + getNextPageParam: nextCursor, + } + + const abortError = () => + new DOMException(`Cursor acquisition was canceled`, `AbortError`) + const abortable = ( + request: Promise, + signal?: AbortSignal, + ) => { + if (!signal) return request + return new Promise((resolve, reject) => { + const abort = () => reject(signal.reason ?? abortError()) + const cleanup = () => signal.removeEventListener(`abort`, abort) + request.then( + (value) => { + cleanup() + resolve(value) + }, + (error) => { + cleanup() + reject(error) + }, + ) + if (signal.aborted) abort() + else signal.addEventListener(`abort`, abort, { once: true }) + }) + } + + return { + read({ offset = 0, limit }, signal) { + const requestedGeneration = generation + const checkCurrent = () => { + if (signal?.aborted) throw signal.reason ?? abortError() + if (requestedGeneration !== generation) { + throw new DOMException(`Cursor sequence was reset`, `AbortError`) + } + } + const run = async () => { + checkCurrent() + const end = limit === undefined ? Infinity : offset + limit + if ( + !Number.isSafeInteger(offset) || + offset < 0 || + (limit !== undefined && + (!Number.isSafeInteger(limit) || + limit < 0 || + !Number.isSafeInteger(end))) + ) + throw new RangeError(`Expected a nonnegative finite integer window`) + if (limit === 0) return [] + try { + const defaults = queryClient.defaultQueryOptions(options) + // Match imperative fetchQuery defaults in browsers as well as Node. + defaults.retry ??= false + const observer = new InfiniteQueryObserver< + CursorPage, + Error, + Pages, + QueryKey, + string | undefined + >(queryClient, { ...options, retry: defaults.retry }) + const query = observer.getCurrentQuery() + const acquire = async (next: boolean): Promise => { + let request = ( + next + ? observer.fetchNextPage({ + throwOnError: true, + cancelRefetch: false, + }) + : observer.refetch({ throwOnError: true, cancelRefetch: false }) + ).then((result) => result.data!) + let acquisition = query.promise! + for (;;) { + try { + const [data] = await Promise.all([request, acquisition]) + return data + } catch (error) { + if (!isCancelledError(error)) throw error + if (!error.silent || query.promise === acquisition) + throw abortError() + // A cancelling refetch supersedes the old transport. Follow + // its replacement, but never leak Query's control error into + // an enclosing row query as though that query was canceled. + acquisition = query.promise! + request = acquisition + } + } + } + const current = observer.getCurrentResult() + let data = + current.data && !current.isStale + ? current.data + : await acquire(false) + checkCurrent() + if (!data.pages.length) + throw new TypeError(`Invalid cursor page: empty cached sequence`) + while ( + data.pages.reduce((count, page) => count + page.rows.length, 0) < + end && + data.pages[data.pages.length - 1]?.nextCursor !== null + ) { + data = await acquire(true) + checkCurrent() + } + const rows: Array = [] + let start = 0 + for (const page of data.pages) { + if (start >= end) break + const stop = Math.min(page.rows.length, end - start) + for (let index = Math.max(0, offset - start); index < stop; index++) + rows.push(page.rows[index]!) + start += page.rows.length + } + return rows + } catch (error) { + checkCurrent() + throw error + } + } + const result = tail.then(() => abortable(run(), signal)) + tail = result.then( + () => {}, + () => {}, + ) + return abortable(result, signal) + }, + reset() { + generation++ + queryClient.removeQueries({ queryKey, exact: true }) + }, + } +} diff --git a/packages/query-db-collection/src/index.ts b/packages/query-db-collection/src/index.ts index 14a5fe7022..565d141b34 100644 --- a/packages/query-db-collection/src/index.ts +++ b/packages/query-db-collection/src/index.ts @@ -11,6 +11,13 @@ export { export * from './errors' +export { createCursorPager } from './cursor-pagination' +export type { + CursorPage, + CursorPager, + CursorPagerOptions, +} from './cursor-pagination' + // Re-export expression helpers from @tanstack/db export { parseWhereExpression, diff --git a/packages/query-db-collection/src/query.ts b/packages/query-db-collection/src/query.ts index cc797f59bb..a192f0b621 100644 --- a/packages/query-db-collection/src/query.ts +++ b/packages/query-db-collection/src/query.ts @@ -2248,8 +2248,11 @@ export function queryCollectionOptions( return oldData }) } else { - // No select - cache contains raw array, just set it directly - queryClient.setQueryData(key, items) + // Raw row writes must not overwrite a different cache format. Avoid even + // a no-op setQueryData: it marks unrelated data fresh and clears invalidation. + const previous = queryClient.getQueryData(key) + if (previous === undefined || Array.isArray(previous)) + queryClient.setQueryData(key, items) } } diff --git a/packages/query-db-collection/tests/cursor-pagination.boundary-oracle.test.ts b/packages/query-db-collection/tests/cursor-pagination.boundary-oracle.test.ts new file mode 100644 index 0000000000..d1b77a76ee --- /dev/null +++ b/packages/query-db-collection/tests/cursor-pagination.boundary-oracle.test.ts @@ -0,0 +1,322 @@ +// @vitest-environment jsdom +import { QueryClient, hashKey } from '@tanstack/query-core' +import { createCollection } from '@tanstack/db' +import fc from 'fast-check' +import { describe, expect, it, vi } from 'vitest' +import { createDeferred } from '../../db/src/deferred.js' +import { oraclePropertyOptions } from '../../db/tests/oracle-config.js' +import { createCursorPager, queryCollectionOptions } from '../src/index.js' + +const tick = () => new Promise((resolve) => setTimeout(resolve, 0)) +const makeClient = () => + new QueryClient({ + defaultOptions: { + queries: { + retry: false, + staleTime: Infinity, + gcTime: Infinity, + }, + }, + }) +const rowsFor = (size: number) => + Array.from({ length: size * 2 }, (_, id) => ({ id })) + +describe(`cursor acquisition boundaries`, () => { + it.each([false, true])( + `nested row query settles when pages cancel; replacement=%s`, + async (replace) => { + await fc.assert( + fc.asyncProperty(fc.integer({ min: 1, max: 4 }), async (size) => { + const client = makeClient() + const rows = rowsFor(size) + const key = [`resource`, `pages`] + const entered = createDeferred() + const release = createDeferred() + let hold = true + let version = 0 + const logged = vi.spyOn(console, `error`).mockImplementation(() => {}) + const options = { + queryClient: client, + queryKey: key, + fetchPage: async (cursor: string | undefined) => { + const page = { + rows: rows + .slice(cursor ? size : 0, cursor ? rows.length : size) + .map((row) => ({ ...row, version })), + nextCursor: cursor ? null : `next`, + } + if (cursor && hold) { + entered.resolve() + await release.promise + } + return page + }, + } + await createCursorPager(options).read({ limit: size }) + const collection = createCollection( + queryCollectionOptions({ + queryClient: client, + queryKey: [`resource`, `rows`], + getKey: (row: { id: number; version: number }) => row.id, + queryFn: () => createCursorPager(options).read({}), + }), + ) + const pending = collection.preload().then( + () => ({ ok: true }), + (error: unknown) => ({ ok: false, error }), + ) + try { + await entered.promise + if (replace) { + hold = false + version = 1 + await client.refetchQueries({ queryKey: key, exact: true }) + } else await client.cancelQueries({ queryKey: key, exact: true }) + await tick() + expect( + client.getQueryState([`resource`, `rows`])?.fetchStatus, + ).toBe(`idle`) + const outcome = await pending + expect(outcome.ok).toBe(replace) + if (replace) + expect( + [...collection.values()].map(({ id, version: rowVersion }) => ({ + id, + version: rowVersion, + })), + ).toEqual(rows.map((row) => ({ ...row, version: 1 }))) + else { + expect(client.getQueryState([`resource`, `rows`])?.status).toBe( + `error`, + ) + if (!outcome.ok && `error` in outcome) + expect((outcome.error as Error).name).toBe(`AbortError`) + } + expect(logged).toHaveBeenCalledTimes(replace ? 0 : 1) + } finally { + logged.mockRestore() + release.resolve() + await collection.cleanup() + client.clear() + } + }), + oraclePropertyOptions(30, `cursor-pagination.nested-cancellation`), + ) + }, + ) + + it(`reader abort releases its queue without canceling shared transport`, async () => { + await fc.assert( + fc.asyncProperty(fc.integer({ min: 1, max: 4 }), async (size) => { + const client = makeClient(), + rows = rowsFor(size), + entered = createDeferred(), + release = createDeferred() + const pager = createCursorPager({ + queryClient: client, + queryKey: [`abort`], + fetchPage: async (cursor, signal) => { + if (cursor) { + entered.resolve(signal) + await release.promise + } + return { + rows: rows.slice(cursor ? size : 0, cursor ? rows.length : size), + nextCursor: cursor ? null : `next`, + } + }, + }) + await pager.read({ limit: size }) + const abort = new AbortController(), + reason = new Error(`reader left`) + let state: unknown = `pending` + const pending = pager.read({}, abort.signal).then( + () => { + state = `success` + }, + (error) => { + state = error + }, + ) + try { + const transport = await entered.promise + abort.abort(reason) + await tick() + expect(state).toBe(reason) + expect(transport.aborted).toBe(false) + expect(await pager.read({ limit: size })).toEqual(rows.slice(0, size)) + } finally { + release.resolve() + await pending + client.clear() + } + }), + oraclePropertyOptions(30, `cursor-pagination.reader-abort`), + ) + }) + + it(`a fresh hit does not inherit another acquisition's same-turn cancellation`, async () => { + const client = makeClient(), + entered = createDeferred(), + release = createDeferred() + const key = [`hit`], + rows = rowsFor(1) + const options = { + queryClient: client, + queryKey: key, + fetchPage: async (cursor: string | undefined) => { + if (cursor) { + entered.resolve() + await release.promise + } + return { + rows: [rows[cursor ? 1 : 0]!], + nextCursor: cursor ? null : `next`, + } + }, + } + await createCursorPager(options).read({ limit: 1 }) + const pending = createCursorPager(options) + .read({}) + .catch(() => {}) + try { + await entered.promise + const hit = createCursorPager(options).read({ limit: 1 }) + void client.cancelQueries( + { queryKey: key, exact: true }, + { revert: false }, + ) + expect(await hit).toEqual(rows.slice(0, 1)) + } finally { + release.resolve() + await pending + client.clear() + } + }) + + it.each([undefined, false, 1] as const)( + `browser retry policy is the same across acquisition phases: %s`, + async (retry) => { + for (const phase of [`initial`, `growth`] as const) { + const client = new QueryClient({ + defaultOptions: { + queries: { gcTime: Infinity, staleTime: Infinity, retryDelay: 0 }, + }, + }) + if (retry !== undefined) client.setQueryDefaults([`retry`], { retry }) + let calls = 0 + const error = new Error(`page failed`) + const pager = createCursorPager({ + queryClient: client, + queryKey: [`retry`], + fetchPage: (cursor) => { + if (phase === `initial` || cursor !== undefined) { + calls++ + return Promise.reject(error) + } + return Promise.resolve({ rows: [{ id: 0 }], nextCursor: `next` }) + }, + }) + try { + await expect(pager.read({})).rejects.toBe(error) + expect(calls).toBe(retry === 1 ? 2 : 1) + } finally { + client.clear() + } + } + }, + ) + + it(`supports an AbortSignal without throwIfAborted`, async () => { + const client = makeClient(), + abort = new AbortController(), + rows = rowsFor(1) + Object.defineProperty(abort.signal, `throwIfAborted`, { value: undefined }) + const pager = createCursorPager({ + queryClient: client, + queryKey: [`compat`], + fetchPage: () => Promise.resolve({ rows, nextCursor: null }), + }) + try { + expect(await pager.read({}, abort.signal)).toEqual(rows) + abort.abort() + await expect(pager.read({}, abort.signal)).rejects.toMatchObject({ + name: `AbortError`, + }) + } finally { + client.clear() + } + }) + + it.each([`response`, `cache`] as const)( + `invalid continuation rejects without spinning from %s`, + async (origin) => { + const client = makeClient(), + key = [`invalid`] + const invalid = { rows: rowsFor(1), nextCursor: undefined } + const pager = createCursorPager({ + queryClient: client, + queryKey: key, + fetchPage: () => + Promise.resolve( + invalid as unknown as { + rows: Array<{ id: number }> + nextCursor: null + }, + ), + }) + if (origin === `cache`) + client.setQueryData(key, { pages: [invalid], pageParams: [undefined] }) + // A semantic work budget turns a microtask loop into a finite RED witness. + let attempts = 0 + if (origin === `cache`) { + const query = client + .getQueryCache() + .find({ queryKey: key, exact: true })! + const run = query.fetch.bind(query) + vi.spyOn(query, `fetch`).mockImplementation((...input) => { + if (++attempts > 3) throw new Error(`loop budget exceeded`) + return run(...input) + }) + } + try { + await expect(pager.read({ limit: 10 })).rejects.toThrow( + `Invalid cursor page`, + ) + expect(attempts).toBeLessThanOrEqual(1) + } finally { + vi.restoreAllMocks() + client.clear() + } + }, + ) + + it(`fresh reads do not hash unrelated cached queries`, async () => { + const client = makeClient() + let hashes = 0 + client.setDefaultOptions({ + queries: { + ...client.getDefaultOptions().queries, + queryKeyHashFn: (key) => { + hashes++ + return hashKey(key) + }, + }, + }) + for (let index = 0; index < 100; index++) + client.setQueryData([`unrelated`, index], index) + const pager = createCursorPager({ + queryClient: client, + queryKey: [`work`], + fetchPage: () => Promise.resolve({ rows: rowsFor(1), nextCursor: null }), + }) + try { + await pager.read({}) + hashes = 0 + await pager.read({ limit: 1 }) + expect(hashes).toBeLessThan(10) + } finally { + client.clear() + } + }) +}) diff --git a/packages/query-db-collection/tests/cursor-pagination.cache-oracle.test.ts b/packages/query-db-collection/tests/cursor-pagination.cache-oracle.test.ts new file mode 100644 index 0000000000..e9e9b26826 --- /dev/null +++ b/packages/query-db-collection/tests/cursor-pagination.cache-oracle.test.ts @@ -0,0 +1,532 @@ +import { QueryClient } from '@tanstack/query-core' +import fc from 'fast-check' +import { describe, expect, it, vi } from 'vitest' +import { createDeferred } from '../../db/src/deferred.js' +import { oraclePropertyOptions } from '../../db/tests/oracle-config.js' +import { createCursorPager } from '../src/index.js' +import { createBackend } from './cursor-pagination/backend.js' +import { expectedRows } from './cursor-pagination/model.js' +import type { CursorPager } from '../src/index.js' +import type { Row } from './cursor-pagination/model.js' + +const scope = { group: undefined, descending: false } +const makeRows = (count: number, version = 0): Array => + Array.from({ length: count }, (_, id) => ({ + id, + rank: Math.floor(id / 3), + group: version, + })) + +async function checkWindow( + pager: CursorPager, + source: Array, + width: number, +) { + expect( + await pager.read({ limit: width }), + `window matches the permitted source snapshot`, + ).toEqual(expectedRows(source, scope, { offset: 0, limit: width })) +} + +/** Row truth is still a full relation. A fake clock and real QueryClient drive + * freshness; the model keeps only the permitted source snapshot and a deadline, + * not Query's page cache, retryer, observer or garbage-collection state. */ +describe(`cursor cache lifecycle`, () => { + it.each([ + [`global`, `maxPages`], + [`global`, `select`], + [`global`, `both`], + [`key`, `maxPages`], + [`key`, `select`], + [`key`, `both`], + ] as const)( + `preserves full windows with %s %s defaults`, + async (level, mode) => { + await fc.assert( + fc.asyncProperty( + fc.integer({ min: 1, max: 4 }), + fc.integer({ min: 1, max: 3 }), + async (size, maxPages) => { + const client = new QueryClient({ + defaultOptions: { + queries: { + retry: false, + gcTime: Infinity, + staleTime: Infinity, + }, + }, + }) + const defaults = { + ...(mode !== `select` ? { maxPages } : {}), + ...(mode !== `maxPages` + ? { select: (data: unknown) => ({ selected: data }) } + : {}), + } + const key = [`defaults`, level, mode] + if (level === `global`) { + client.setDefaultOptions({ + queries: { ...client.getDefaultOptions().queries, ...defaults }, + }) + } else client.setQueryDefaults([`defaults`], defaults) + let rows = makeRows(size * (maxPages + 2)) + let backend = createBackend(rows, scope, size) + const pager = createCursorPager({ + queryClient: client, + queryKey: key, + fetchPage: (cursor, signal) => { + if (cursor === undefined) + backend = createBackend(rows, scope, size) + return backend.fetchPage(cursor, signal) + }, + }) + try { + await checkWindow(pager, rows, size) + await checkWindow(pager, rows, rows.length) + // Growing must not move the origin of later offset reads. + for (const offset of [0, size, rows.length - 1]) { + const window = { offset, limit: size } + expect(await pager.read(window)).toEqual( + expectedRows(rows, scope, window), + ) + } + rows = makeRows(rows.length, 1) + await client.invalidateQueries({ queryKey: key }) + await checkWindow(pager, rows, rows.length) + } finally { + client.clear() + } + }, + ), + oraclePropertyOptions(50, `cursor-pagination.defaults`), + ) + }, + ) + + it.each([`initial`, `growth`, `refresh`] as const)( + `cancellation stops a held %s acquisition and permits later recovery`, + async (phase) => { + await fc.assert( + fc.asyncProperty( + fc.integer({ min: 1, max: 4 }), + fc.integer({ min: 1, max: 3 }), + async (size, depth) => { + const client = new QueryClient({ + defaultOptions: { + queries: { + staleTime: Infinity, + gcTime: Infinity, + retry: false, + }, + }, + }) + const key = [`cancel-history`] + const hold = createDeferred() + const started = createDeferred() + const delivered = createDeferred() + let holdAt = Infinity + let calls = 0 + let rows = makeRows(size * (depth + 2)) + let backend = createBackend(rows, scope, size) + const options = { + queryClient: client, + queryKey: key, + fetchPage: async ( + cursor: string | undefined, + signal: AbortSignal, + ) => { + if (cursor === undefined) + backend = createBackend(rows, scope, size) + const page = await backend.fetchPage(cursor, signal) + if (++calls === holdAt) { + started.resolve(signal) + // Hold real response delivery; deliberately ignore abort to + // prove Query also fences a transport's late completion. + await hold.promise + delivered.resolve() + } + return page + }, + } + const pager = createCursorPager(options) + try { + if (phase !== `initial`) { + await checkWindow(pager, rows, size * (depth + 1)) + } + const previous = client.getQueryData(key) + if (phase === `refresh`) { + rows = makeRows(rows.length, 1) + await client.invalidateQueries({ queryKey: key }) + } + holdAt = calls + (phase === `refresh` ? depth : 1) + const window = { limit: rows.length } + const pending = pager.read(window).then( + (value) => ({ status: `fulfilled` as const, value }), + (reason: unknown) => ({ status: `rejected` as const, reason }), + ) + const signal = await started.promise + if (phase === `growth`) { + // A reader whose whole window is cached need not wait for a + // peer's deeper acquisition, even though they share a key. + await checkWindow(createCursorPager(options), rows, size) + } + const query = client + .getQueryCache() + .find({ queryKey: key, exact: true })! + const joined = createDeferred() + const fetch = query.fetch.bind(query) + const joinWitness = vi + .spyOn(query, `fetch`) + .mockImplementation((...args) => { + const result = fetch(...args) + joined.resolve() + return result + }) + const peer = createCursorPager(options) + .read(window) + .then( + (value) => ({ status: `fulfilled` as const, value }), + (reason: unknown) => ({ + status: `rejected` as const, + reason, + }), + ) + // Observe the real fetch join instead of guessing how many + // microtasks the peer takes to reach the acquisition boundary. + await joined.promise + joinWitness.mockRestore() + await client.cancelQueries({ queryKey: key, exact: true }) + expect(signal.aborted).toBe(true) + for (const result of await Promise.all([pending, peer])) { + expect(result.status).toBe(`rejected`) + if (result.status === `rejected`) + expect(result.reason).toMatchObject({ name: `AbortError` }) + } + expect( + calls, + `cancellation must not start replacement transport`, + ).toBe(holdAt) + expect(client.getQueryData(key)).toBe(previous) + hold.resolve() + await delivered.promise + await new Promise((resolve) => setTimeout(resolve, 0)) + expect(client.getQueryData(key)).toBe(previous) + await checkWindow(pager, rows, rows.length) + } finally { + hold.resolve() + client.clear() + } + }, + ), + oraclePropertyOptions(50, `cursor-pagination.cancellation`), + ) + }, + ) + + it(`the cache oracle rejects reuse after invalidation and accepts a real refresh`, async () => { + for (const ignoreInvalidation of [false, true]) { + const client = new QueryClient({ + defaultOptions: { + queries: { staleTime: Infinity, gcTime: Infinity, retry: false }, + }, + }) + let rows = makeRows(2) + const key = [`sensitivity`] + const pager = createCursorPager({ + queryClient: client, + queryKey: key, + fetchPage: () => Promise.resolve({ rows, nextCursor: null }), + }) + try { + await checkWindow(pager, rows, 2) + rows = makeRows(2, 1) + await client.invalidateQueries({ queryKey: key }) + if (ignoreInvalidation) { + // Deliberately bypass Query's stale check, but still execute the + // production pager and the same value checker as the history law. + client + .getQueryCache() + .find({ queryKey: key, exact: true })! + .setState({ isInvalidated: false }) + await expect(checkWindow(pager, rows, 2)).rejects.toThrow( + `window matches`, + ) + } else await checkWindow(pager, rows, 2) + } finally { + vi.restoreAllMocks() + client.clear() + } + } + }) + + it(`QueryClient cancellation rejects held work without installing its late page`, async () => { + const client = new QueryClient({ + defaultOptions: { + queries: { staleTime: Infinity, gcTime: Infinity, retry: false }, + }, + }) + const key = [`cancel`] + const hold = createDeferred() + const delivered = createDeferred() + let deliveredSignal: AbortSignal | undefined + let holdResponse = true + const rows = makeRows(2) + const pager = createCursorPager({ + queryClient: client, + queryKey: key, + fetchPage: async (_cursor, signal) => { + deliveredSignal = signal + if (holdResponse) { + await hold.promise + delivered.resolve() + } + return { rows, nextCursor: null } + }, + }) + const pending = pager.read({}).then( + () => `success`, + () => `rejected`, + ) + try { + await vi.waitFor(() => expect(deliveredSignal).toBeDefined()) + await client.cancelQueries({ queryKey: key, exact: true }) + expect(await pending).toBe(`rejected`) + expect(deliveredSignal?.aborted).toBe(true) + hold.resolve() + await delivered.promise + await new Promise((resolve) => setTimeout(resolve, 0)) + expect(client.getQueryData(key)).toBeUndefined() + holdResponse = false + await checkWindow(pager, rows, 2) + } finally { + hold.resolve() + await pending + client.clear() + } + }) + + it(`reuses pages until expiry or invalidation across generated histories`, async () => { + vi.useFakeTimers({ toFake: [`Date`] }) + try { + await fc.assert( + fc.asyncProperty( + fc.integer({ min: 1, max: 12 }), + fc.integer({ min: 1, max: 5 }), + fc.array( + fc.constantFrom( + `read`, + `grow`, + `age`, + `invalidate`, + `change`, + `remove`, + ), + { minLength: 1, maxLength: 25 }, + ), + async (count, size, actions) => { + let now = 1000 + vi.setSystemTime(now) + const client = new QueryClient({ + defaultOptions: { queries: { retry: false, gcTime: Infinity } }, + }) + let source = makeRows(count) + let expectedSource = source + let deadline = 0 + let width = 1 + let version = 0 + let cached = false + let calls = 0 + let starts = 0 + let backend = createBackend(source, scope, size) + const key = [`cache-law`] + const pager = createCursorPager({ + queryClient: client, + queryKey: key, + staleTime: 100, + fetchPage: (cursor, signal) => { + calls++ + if (cursor === undefined) { + starts++ + backend = createBackend(source, scope, size) + } + return backend.fetchPage(cursor, signal) + }, + }) + try { + for (const action of [`read`, ...actions] as const) { + if (action === `change`) { + source = makeRows(count, ++version) + continue + } + if (action === `age`) { + now += 100 + vi.setSystemTime(now) + continue + } + if (action === `invalidate`) { + await client.invalidateQueries({ queryKey: key }) + deadline = 0 + continue + } + if (action === `remove`) { + client.removeQueries({ queryKey: key }) + cached = false + continue + } + if (action === `grow`) width += size + const refresh = !cached || now >= deadline + if (refresh) expectedSource = source + const beforeCalls = calls + const beforeStarts = starts + const window = { offset: 0, limit: width } + await checkWindow(pager, expectedSource, width) + expect(starts - beforeStarts).toBe(refresh ? 1 : 0) + // Query dates the cache from its latest successful acquisition, + // including fetchNextPage, rather than aging each page separately. + if (calls !== beforeCalls) deadline = now + 100 + cached = true + const settledCalls = calls + expect(await pager.read(window)).toEqual( + expectedRows(expectedSource, scope, window), + ) + expect(calls).toBe(settledCalls) + } + } finally { + client.clear() + } + }, + ), + oraclePropertyOptions(100, `cursor-pagination.cache`), + ) + } finally { + vi.useRealTimers() + } + }) + + it.each([0, 25])( + `collects inactive pages independently of freshness with gcTime %i`, + async (gcTime) => { + vi.useFakeTimers() + const client = new QueryClient({ + defaultOptions: { queries: { retry: false } }, + }) + const backend = createBackend(makeRows(6), scope, 2) + const key = [`collect`] + const pager = createCursorPager({ + queryClient: client, + queryKey: key, + staleTime: Infinity, + gcTime, + fetchPage: backend.fetchPage, + }) + try { + await pager.read({ limit: 3 }) + expect(backend.calls).toHaveLength(2) + expect(client.getQueryData(key)).toBeDefined() + await vi.advanceTimersByTimeAsync(gcTime + 1) + expect(client.getQueryData(key)).toBeUndefined() + expect(await pager.read({ limit: 3 })).toEqual( + expectedRows(makeRows(6), scope, { offset: 0, limit: 3 }), + ) + expect(backend.calls).toHaveLength(4) + } finally { + client.clear() + vi.useRealTimers() + } + }, + ) + + it.each([`expiry`, `invalidate`] as const)( + `keeps cached data while a %s refresh is held or fails`, + async (cause) => { + vi.useFakeTimers({ toFake: [`Date`] }) + vi.setSystemTime(1000) + const client = new QueryClient({ + defaultOptions: { queries: { retry: false, gcTime: Infinity } }, + }) + const key = [`refresh`] + let source = makeRows(6) + let backend = createBackend(source, scope, 2) + let hold: ReturnType> | undefined + const pager = createCursorPager({ + queryClient: client, + queryKey: key, + staleTime: 100, + fetchPage: async (cursor, signal) => { + if (cursor === undefined) backend = createBackend(source, scope, 2) + const page = await backend.fetchPage(cursor, signal) + if (hold) await hold.promise + return page + }, + }) + try { + await pager.read({ limit: 4 }) + const old = client.getQueryData(key) + source = makeRows(6, 1) + if (cause === `expiry`) vi.setSystemTime(1100) + else await client.invalidateQueries({ queryKey: key }) + hold = createDeferred() + const error = new Error(`refresh failed`) + const pending = pager + .read({ limit: 4 }) + .catch((reason: unknown) => reason) + await vi.waitFor(() => expect(client.isFetching()).toBe(1)) + expect(client.getQueryData(key)).toBe(old) + hold.reject(error) + expect(await pending).toBe(error) + expect(client.getQueryData(key)).toBe(old) + hold = undefined + expect(await pager.read({ limit: 4 })).toEqual( + expectedRows(source, scope, { offset: 0, limit: 4 }), + ) + } finally { + hold?.resolve() + client.clear() + vi.useRealTimers() + } + }, + ) + + it(`shares acquisitions across readers without letting one abort cancel its peer`, async () => { + const client = new QueryClient({ + defaultOptions: { + queries: { retry: false, gcTime: Infinity, staleTime: Infinity }, + }, + }) + const backend = createBackend(makeRows(6), scope, 2) + const hold = createDeferred() + const options = { + queryClient: client, + queryKey: [`shared`], + fetchPage: async (cursor: string | undefined, signal: AbortSignal) => { + const page = await backend.fetchPage(cursor, signal) + await hold.promise + return page + }, + } + const a = createCursorPager(options) + const b = createCursorPager(options) + const abort = new AbortController() + const error = new Error(`reader left`) + const pending = Promise.allSettled([ + a.read({ limit: 2 }, abort.signal), + b.read({ limit: 4 }), + ]) + try { + await vi.waitFor(() => expect(backend.calls).toHaveLength(1)) + abort.abort(error) + hold.resolve() + expect(await pending).toEqual([ + { status: `rejected`, reason: error }, + { + status: `fulfilled`, + value: expectedRows(makeRows(6), scope, { offset: 0, limit: 4 }), + }, + ]) + expect(backend.calls).toHaveLength(2) + } finally { + hold.resolve() + await pending + client.clear() + } + }) +}) diff --git a/packages/query-db-collection/tests/cursor-pagination.integration.test.ts b/packages/query-db-collection/tests/cursor-pagination.integration.test.ts new file mode 100644 index 0000000000..013d5c44e0 --- /dev/null +++ b/packages/query-db-collection/tests/cursor-pagination.integration.test.ts @@ -0,0 +1,455 @@ +import { QueryClient } from '@tanstack/query-core' +import { + BasicIndex, + createCollection, + createLiveQueryCollection, +} from '@tanstack/db' +import { describe, expect, it, vi } from 'vitest' +import fc from 'fast-check' +import { oraclePropertyOptions } from '../../db/tests/oracle-config.js' +import { createDeferred } from '../../db/src/deferred.js' +import { createLiveQueryWindowController } from '../../db/src/live-query-window-controller.js' +import { createCursorPager, queryCollectionOptions } from '../src/index.js' +import { createBackend } from './cursor-pagination/backend.js' +import { expectedWindow } from './cursor-pagination/model.js' +import type { Row, Scope } from './cursor-pagination/model.js' + +/** Real QueryClient -> QueryCollection -> graph -> shared window controller. + * The fixture endpoint supports prefix and rank-equality tie requests. Other + * predicates and cursor expressions reject rather than silently dropping IR. + * Each tie filter has its own opaque backend sequence, just like the prefix. + */ +function createFixture( + rows: Array, + scope: Scope, + backendSize: number, + holdFirst = false, +) { + const backend = { calls: [] as Array } + const gate = createDeferred() + let received = 0 + const client = new QueryClient({ + defaultOptions: { + queries: { retry: false, staleTime: Infinity, gcTime: Infinity }, + }, + }) + const makePager = (rank?: number) => { + const makeBackend = () => + createBackend( + rank === undefined ? rows : rows.filter((row) => row.rank === rank), + scope, + backendSize, + ) + let transport = makeBackend() + return createCursorPager({ + queryClient: client, + queryKey: [`cursor-experiment`, scope, `opaque-pages`, rank ?? null], + fetchPage: async (cursor, signal) => { + if (cursor === undefined) transport = makeBackend() + if (rank === undefined) backend.calls.push(cursor) + const page = await transport.fetchPage(cursor, signal) + if (++received === 1 && holdFirst) await gate.promise + return page + }, + }) + } + const pager = makePager() + const ties = new Map>() + const requests: Array<{ offset: number; limit: number | undefined }> = [] + const direction = scope.descending ? `desc` : `asc` + const source = createCollection( + queryCollectionOptions({ + queryClient: client, + queryKey: [`cursor-experiment`, scope, `rows`], + syncMode: `on-demand`, + autoIndex: `eager`, + defaultIndexType: BasicIndex, + getKey: (row) => row.id, + queryFn: async (ctx) => { + const options = ctx.meta?.loadSubsetOptions + let reader = pager + const where = options?.where + if (where !== undefined) { + if ( + where.type !== `func` || + where.name !== `eq` || + where.args.length !== 2 || + where.args[0]?.type !== `ref` || + where.args[0].path.length !== 1 || + where.args[0].path[0] !== `rank` || + where.args[1]?.type !== `val` || + typeof where.args[1].value !== `number` + ) { + throw new Error(`Unsupported fixture predicate`) + } + const rank = where.args[1].value + let tie = ties.get(rank) + if (!tie) { + tie = makePager(rank) + ties.set(rank, tie) + } + reader = tie + } + expect( + options?.cursor, + `fixture uses offset requests, not IR cursors`, + ).toBeUndefined() + if (options?.orderBy) { + expect(options.orderBy).toMatchObject([ + { expression: { path: [`rank`] }, compareOptions: { direction } }, + { expression: { path: [`id`] }, compareOptions: { direction } }, + ]) + } + const window = { offset: options?.offset ?? 0, limit: options?.limit } + requests.push(window) + return reader.read(window, ctx.signal) + }, + }), + ) + const live = createLiveQueryCollection((q) => + q + .from({ row: source }) + .orderBy(({ row }) => row.rank, direction) + .orderBy(({ row }) => row.id, direction) + .limit(1), + ) + return { + backend, + gate, + requests, + live, + source, + refresh: () => + client.invalidateQueries( + { queryKey: [`cursor-experiment`, scope] }, + { throwOnError: true }, + ), + received: () => received, + async cleanup() { + gate.resolve() + await live.cleanup() + await source.cleanup() + client.clear() + }, + } +} + +const matrix = [0, 1, 9, 20].flatMap((count) => + [2, 5].flatMap((backendSize) => + [3, 7].flatMap((pageSize) => + [false, true].map((descending) => ({ + count, + backendSize, + pageSize, + descending, + })), + ), + ), +) + +// This oracle checks all user row fields and their order. Virtual Collection +// fields ($key/$origin/etc.) are outside the cursor adapter's responsibility. +const observeRows = (rows: ReadonlyArray) => + rows.map(({ id, rank, group }) => ({ id, rank, group })) + +describe(`cursor adapter through production pagination`, () => { + it.each([ + { wrapped: false, nested: false }, + { wrapped: false, nested: true }, + { wrapped: true, nested: false }, + ])( + `manual writes preserve cache shapes: $wrapped/$nested`, + async ({ wrapped, nested }) => { + await fc.assert( + fc.asyncProperty( + fc.array( + fc.record({ + kind: fc.constantFrom(`insert`, `update`, `delete`), + position: fc.nat({ max: 10 }), + }), + { minLength: 1, maxLength: 8 }, + ), + fc.integer({ min: 2, max: 6 }), + async (operations, count) => { + const client = new QueryClient({ + defaultOptions: { + queries: { + retry: false, + staleTime: Infinity, + gcTime: Infinity, + }, + }, + }) + const rows = Array.from({ length: count }, (_, id) => ({ + id, + rank: id, + group: 0, + })) + const pageKey = [`manual`, `pages`] + const rowKey = nested ? [`manual`] : [`manual`, `rows`] + const pager = createCursorPager({ + queryClient: client, + queryKey: pageKey, + fetchPage: () => Promise.resolve({ rows, nextCursor: null }), + }) + const common = { + queryClient: client, + queryKey: rowKey, + getKey: (row: Row) => row.id, + } + const source = createCollection( + wrapped + ? queryCollectionOptions({ + ...common, + queryFn: async () => ({ + items: await pager.read({}), + label: `keep`, + }), + select: (response) => response.items, + }) + : queryCollectionOptions({ + ...common, + queryFn: () => pager.read({}), + }), + ) + try { + await source.preload() + // Raw row caches with no data must still be seeded by manual writes. + const emptyKey = [...rowKey, `unloaded`] + if (!wrapped) + client.getQueryCache().build(client, { queryKey: emptyKey }) + const expected = rows.map((row) => ({ ...row })) + let nextId = count + for (const [step, operation] of operations.entries()) { + await client.invalidateQueries({ + queryKey: pageKey, + exact: true, + refetchType: `none`, + }) + const previous = client.getQueryState(pageKey) + const position = + operation.position % Math.max(1, expected.length) + if (operation.kind === `insert` || !expected.length) { + const row = { id: nextId, rank: nextId++, group: 0 } + source.utils.writeInsert({ ...row }) + expected.push(row) + } else if (operation.kind === `update`) { + source.utils.writeUpdate({ + id: expected[position]!.id, + group: step + 1, + }) + expected[position]!.group = step + 1 + } else { + source.utils.writeDelete(expected[position]!.id) + expected.splice(position, 1) + } + expect(observeRows([...source.values()])).toEqual(expected) + expect(client.getQueryData(rowKey)).toEqual( + wrapped ? { items: expected, label: `keep` } : expected, + ) + if (!wrapped) + expect(client.getQueryData(emptyKey)).toEqual(expected) + // Preserving the object alone is insufficient: a no-op setQueryData + // also clears invalidation and marks an unrelated sequence fresh. + expect( + client.getQueryState(pageKey), + `manual writes must not touch page state`, + ).toBe(previous) + expect(await pager.read({})).toEqual(rows) + } + } finally { + await source.cleanup() + client.clear() + } + }, + ), + oraclePropertyOptions(50, `cursor-pagination.manual-write`), + ) + }, + ) + it.each( + [false, true].flatMap((descending) => + [2, 5].map((backendSize) => ({ descending, backendSize })), + ), + )( + `refreshes changed backend rows before growing the window $descending/$backendSize`, + async ({ descending, backendSize }) => { + const rows = Array.from({ length: 3 }, (_, id) => ({ + id, + rank: 0, + group: 0, + })) + const scope = { group: undefined, descending } + const fixture = createFixture(rows, scope, backendSize) + const controller = createLiveQueryWindowController(fixture.live, { + pageSize: 3, + }) + const stop = controller.subscribe(() => {}) + try { + await controller.preload() + expect(observeRows(controller.getSnapshot().data)).toEqual( + expectedWindow(rows, scope, 3).rows, + ) + expect(controller.getSnapshot().hasNextPage).toBe(false) + for (const count of [9, 2, 0, 7]) { + const previous = observeRows(controller.getSnapshot().data) + const calls = fixture.backend.calls.length + rows.splice( + 0, + rows.length, + ...Array.from({ length: count }, (_, id) => ({ + id, + rank: Math.floor(id / 6), + group: count, + })), + ) + // Refreshing only the outer row query still allows its fresh page + // cache. Prefix invalidation is the explicit two-cache refresh path. + await fixture.source.utils.refetch({ throwOnError: true }) + expect(observeRows(controller.getSnapshot().data)).toEqual(previous) + expect(fixture.backend.calls).toHaveLength(calls) + await fixture.refresh() + const expected = expectedWindow(rows, scope, 3) + await vi.waitFor(() => { + expect(observeRows(controller.getSnapshot().data)).toEqual( + expected.rows, + ) + expect(controller.getSnapshot().hasNextPage).toBe( + expected.hasNextPage, + ) + }) + } + await controller.fetchNextPage() + expect(observeRows(controller.getSnapshot().data)).toEqual( + expectedWindow(rows, scope, 6).rows, + ) + expect(controller.getSnapshot().hasNextPage).toBe(true) + } finally { + stop() + controller.dispose() + await fixture.cleanup() + } + }, + ) + + it.each(matrix)( + `publishes exact pages and exhaustion $count/$backendSize/$pageSize/$descending`, + async ({ count, backendSize, pageSize, descending }) => { + const rows = Array.from({ length: count }, (_, id) => ({ + id, + rank: Math.floor(id / 6), + group: id % 2, + })) + const scope = { group: undefined, descending } + const fixture = createFixture(rows, scope, backendSize) + const controller = createLiveQueryWindowController(fixture.live, { + pageSize, + }) + const publications: Array<{ + rows: Array + pages: number + more: boolean + ready: boolean + }> = [] + const unsubscribe = controller.subscribe(() => { + const snapshot = controller.getSnapshot() + publications.push({ + rows: observeRows(snapshot.data), + pages: snapshot.pageParams.length, + more: snapshot.hasNextPage, + ready: snapshot.isReady, + }) + }) + try { + await controller.preload() + for (let page = 1; page <= Math.ceil(count / pageSize) + 1; page++) { + const expected = expectedWindow(rows, scope, page * pageSize) + const snapshot = controller.getSnapshot() + expect(observeRows(snapshot.data)).toEqual(expected.rows) + expect(snapshot.hasNextPage).toBe(expected.hasNextPage) + expect(observeRows(snapshot.pages.flat())).toEqual(expected.rows) + if (!expected.hasNextPage) break + await controller.fetchNextPage() + } + expect(publications.length).toBeGreaterThan(0) + expect(publications.some((published) => published.ready)).toBe(true) + for (const published of publications) { + if (published.ready) { + const expected = expectedWindow( + rows, + scope, + published.pages * pageSize, + ) + expect(published.rows).toEqual(expected.rows) + expect(published.more).toBe(expected.hasNextPage) + } + } + expect(fixture.requests.length).toBeGreaterThan(0) + expect(fixture.requests[0]?.limit).toBe(pageSize + 1) + expect(fixture.backend.calls).toHaveLength( + Math.max(1, Math.ceil(count / backendSize)), + ) + } finally { + unsubscribe() + controller.dispose() + await fixture.cleanup() + } + }, + ) + + it(`holds real response delivery and preserves a shallower peer after deep exhaustion`, async () => { + const rows = Array.from({ length: 9 }, (_, id) => ({ + id, + rank: Math.floor(id / 6), + group: 0, + })) + const scope = { group: undefined, descending: false } + const fixture = createFixture(rows, scope, 2, true) + const shallow = createLiveQueryWindowController(fixture.live, { + pageSize: 3, + }) + const deep = createLiveQueryWindowController(fixture.live, { pageSize: 10 }) + const publications: Array> = [] + const stopShallow = shallow.subscribe(() => + publications.push(shallow.getSnapshot().data.map((row) => row.id)), + ) + const stopDeep = deep.subscribe(() => {}) + let settled = 0 + const pending = Promise.allSettled( + [shallow.preload(), deep.preload()].map((promise) => + promise.then(() => { + settled++ + }), + ), + ) + try { + await vi.waitFor(() => expect(fixture.received()).toBe(1)) + expect(settled).toBe(0) + expect(shallow.getSnapshot().data).toEqual([]) + expect(publications.every((batch) => batch.length === 0)).toBe(true) + fixture.gate.resolve() + expect((await pending).map((result) => result.status)).toEqual([ + `fulfilled`, + `fulfilled`, + ]) + expect(shallow.getSnapshot().hasNextPage).toBe(true) + expect(deep.getSnapshot().hasNextPage).toBe(false) + stopDeep() + deep.dispose() + await shallow.preload() + expect(observeRows(shallow.getSnapshot().data)).toEqual(rows.slice(0, 3)) + expect(shallow.getSnapshot().hasNextPage).toBe(true) + await shallow.fetchNextPage() + expect(observeRows(shallow.getSnapshot().data)).toEqual(rows.slice(0, 6)) + expect(fixture.backend.calls).toHaveLength(5) + } finally { + fixture.gate.resolve() + await pending + stopShallow() + stopDeep() + shallow.dispose() + deep.dispose() + await fixture.cleanup() + } + }) +}) diff --git a/packages/query-db-collection/tests/cursor-pagination.no-peek.integration.test.ts b/packages/query-db-collection/tests/cursor-pagination.no-peek.integration.test.ts new file mode 100644 index 0000000000..4126a5a0f3 --- /dev/null +++ b/packages/query-db-collection/tests/cursor-pagination.no-peek.integration.test.ts @@ -0,0 +1,215 @@ +import { + BasicIndex, + createCollection, + createLiveQueryCollection, +} from '@tanstack/db' +import { describe, expect, it } from 'vitest' +import { createDeferred } from '../../db/src/deferred.js' +import { createLiveQueryWindowController } from '../../db/src/live-query-window-controller.js' +import { expectedWindow } from './cursor-pagination/model.js' +import { createNoPeekSession } from './cursor-pagination/no-peek.js' +import { createFactTransport } from './cursor-pagination/no-peek-transport.js' +import type { Row } from './cursor-pagination/model.js' +import type { PrefixPublication } from './cursor-pagination/no-peek.js' + +const scope = { group: undefined, descending: false } + +/** Narrow bridge for one immutable source ordered by unique id. Facts are kept + * outside production and admitted only after the real setWindow/preload gate. + * This intentionally does not claim generic compiler eligibility or automatic + * metadata-only notification support. */ +function createGraphFixture( + count: number, + pageSize: number, + hold = false, + filtered = false, +) { + const rows: Array = Array.from({ length: count }, (_, id) => ({ + id, + rank: id, + group: 0, + })) + const transport = createFactTransport(rows, pageSize, scope) + const received = createDeferred() + const release = createDeferred() + let fact: PrefixPublication | undefined + const loaded = new Set() + const source = createCollection({ + getKey: (row) => row.id, + syncMode: `on-demand`, + autoIndex: `eager`, + defaultIndexType: BasicIndex, + sync: { + sync: ({ begin, write, commit, markReady }) => { + markReady() + return { + loadSubset: async (options) => { + let batch: Array + if (options.where) { + const where = options.where + if ( + where.type !== `func` || + where.name !== `eq` || + where.args[0]?.type !== `ref` || + where.args[0].path[0] !== `id` || + where.args[1]?.type !== `val` + ) { + throw new Error(`Unsupported probe filter`) + } + const id = where.args[1].value + batch = rows.filter((row) => row.id === id) + } else { + // This endpoint uses offset as permitted by the source contract; + // cursor hints are not substituted for the offset window. + const offset = options.offset ?? 0 + const packet = await transport.read( + offset + (options.limit ?? rows.length + 1), + ) + if (offset === 0) fact = packet + batch = packet.rows.slice(offset) + received.resolve() + if (hold) await release.promise + } + begin() + for (const row of batch) { + if (!loaded.has(row.id)) { + write({ type: `insert`, value: row }) + loaded.add(row.id) + } + } + await commit() + }, + } + }, + }, + }) + const live = createLiveQueryCollection((q) => { + const query = q + .from({ row: source }) + .orderBy(({ row }) => row.id) + .limit(1) + return filtered ? query.fn.where(({ row }) => row.id < 3) : query + }) + const windows: Array = [] + const acquire = async (requested: number): Promise => { + windows.push(requested) + const settled = live.utils.setWindow({ offset: 0, limit: requested }) + if (settled !== true) await settled + await live.preload() + const stamp = fact?.stamp ?? {} + return { + rows: live.toArray.map(({ id, rank, group }) => ({ id, rank, group })), + requested, + stamp, + transparent: !filtered, + fact: fact?.fact, + } + } + return { + rows, + live, + transport, + acquire, + windows, + received, + release, + fact: () => fact, + async cleanup() { + release.resolve() + await live.cleanup() + await source.cleanup() + }, + } +} + +describe(`no-peek bridge at real graph publication`, () => { + it.each([0, 3, 4, 9])( + `matches peek-ahead output with %s remote rows`, + async (count) => { + const candidate = createGraphFixture(count, 3) + const baseline = createGraphFixture(count, 3) + const session = createNoPeekSession(candidate.acquire) + const controller = createLiveQueryWindowController(baseline.live, { + pageSize: 3, + }) + const unsubscribe = controller.subscribe(() => {}) + try { + session.request(`a`, 3) + await session.refresh() + await controller.preload() + expect(session.get(`a`)).toEqual( + expectedWindow(candidate.rows, scope, 3), + ) + expect(controller.getSnapshot().hasNextPage).toBe( + session.get(`a`)?.hasNextPage, + ) + expect(controller.getSnapshot().data.map(({ id }) => id)).toEqual( + session.get(`a`)?.rows.map(({ id }) => id), + ) + expect(candidate.transport.requests[0]).toBe(3) + expect(baseline.transport.requests[0]).toBe(4) + expect(candidate.transport.backend.calls).toHaveLength(1) + expect(baseline.transport.backend.calls).toHaveLength(count > 3 ? 2 : 1) + } finally { + unsubscribe() + controller.dispose() + await candidate.cleanup() + await baseline.cleanup() + } + }, + ) + + it(`holds continuation behind actual source and graph publication`, async () => { + const fixture = createGraphFixture(9, 3, true) + const session = createNoPeekSession(fixture.acquire) + session.request(`a`, 3) + let notifications = 0 + session.subscribe(() => notifications++) + const pending = session.refresh() + const observed = pending.then( + () => undefined, + (error: unknown) => error, + ) + try { + await fixture.received.promise + expect(fixture.fact()?.fact?.hasMore).toBe(true) + expect(fixture.live.toArray).toEqual([]) + expect(session.get(`a`)).toBeUndefined() + expect(notifications).toBe(0) + fixture.release.resolve() + expect(await observed).toBeUndefined() + expect(session.get(`a`)).toEqual(expectedWindow(fixture.rows, scope, 3)) + expect(notifications).toBe(1) + } finally { + fixture.release.resolve() + await observed + await fixture.cleanup() + } + }) + + it(`falls back for a local filter even though the order is unchanged`, async () => { + const fixture = createGraphFixture(9, 3, false, true) + const session = createNoPeekSession(fixture.acquire, false) + session.request(`a`, 3) + try { + const rawPrefix = await createFactTransport(fixture.rows, 3, scope).read( + 3, + ) + expect(rawPrefix.fact?.hasMore).toBe(true) + await session.refresh() + expect(session.get(`a`)).toEqual( + expectedWindow( + fixture.rows.filter((row) => row.id < 3), + scope, + 3, + ), + ) + expect(fixture.windows).toEqual([4]) + expect(session.get(`a`)?.hasNextPage).toBe(false) + // An opaque local filter requires the existing full-source loading path. + expect(fixture.transport.requests[0]).toBe(10) + } finally { + await fixture.cleanup() + } + }) +}) diff --git a/packages/query-db-collection/tests/cursor-pagination.no-peek.test.ts b/packages/query-db-collection/tests/cursor-pagination.no-peek.test.ts new file mode 100644 index 0000000000..19154ca889 --- /dev/null +++ b/packages/query-db-collection/tests/cursor-pagination.no-peek.test.ts @@ -0,0 +1,239 @@ +import { describe, expect, it } from 'vitest' +import fc from 'fast-check' +import { createDeferred } from '../../db/src/deferred.js' +import { oraclePropertyOptions } from '../../db/tests/oracle-config.js' +import { expectedWindow } from './cursor-pagination/model.js' +import { createNoPeekSession } from './cursor-pagination/no-peek.js' +import { createFactTransport } from './cursor-pagination/no-peek-transport.js' +import type { Row, Scope } from './cursor-pagination/model.js' + +const scope: Scope = { group: undefined, descending: false } +const rowsOf = (count: number): Array => + Array.from({ length: count }, (_, id) => ({ + id, + rank: Math.floor(id / 3), + group: id % 2, + })) + +/** Transport sees only opaque pages. In these protocol tests its completion is + * also the synthetic publication boundary; the production probe is separate. */ +function createTransport(rows: Array, size: number, selected = scope) { + return createFactTransport(rows, size, selected) +} + +describe(`experimental publication-bound no-peek pagination`, () => { + it(`keeps shallow continuation after a deeper consumer reaches the end`, async () => { + const rows = rowsOf(9) + const transport = createTransport(rows, 3) + const session = createNoPeekSession(transport.read) + session.request(`shallow`, 3) + session.request(`deep`, 10) + await session.refresh() + expect(session.get(`shallow`)).toEqual(expectedWindow(rows, scope, 3)) + expect(session.get(`deep`)).toEqual(expectedWindow(rows, scope, 10)) + session.release(`deep`) + await session.refresh() + expect(session.get(`shallow`)).toEqual(expectedWindow(rows, scope, 3)) + }) + + it.each([ + `missing`, + `foreign-stamp`, + `wrong-boundary`, + `transformed`, + ] as const)( + `acquires and retains a peek witness for %s facts`, + async (fault) => { + const rows = rowsOf(9) + const transport = createTransport(rows, 3) + const session = createNoPeekSession(async (limit) => { + const packet = await transport.read(limit) + if (fault === `missing`) delete packet.fact + if (fault === `foreign-stamp`) packet.fact!.stamp = {} + if (fault === `wrong-boundary`) packet.fact!.end++ + if (fault === `transformed`) packet.transparent = false + return packet + }) + session.request(`a`, 3) + await session.refresh() + expect(session.get(`a`)).toEqual(expectedWindow(rows, scope, 3)) + expect(transport.requests).toEqual([3, 4]) + session.request(`b`, 10) + await session.refresh() + session.release(`b`) + await session.refresh() + expect(transport.requests.at(-1)).toBe(4) + expect(session.get(`a`)).toEqual(expectedWindow(rows, scope, 3)) + }, + ) + + it(`agrees with full-relation truth across scopes, peers, release and repartition`, async () => { + await fc.assert( + fc.asyncProperty( + fc.integer({ min: 0, max: 35 }), + fc.integer({ min: 1, max: 9 }), + fc.record({ + descending: fc.boolean(), + group: fc.option(fc.integer({ min: 0, max: 1 }), { nil: undefined }), + }), + fc.array( + fc.record({ + id: fc.constantFrom(`a`, `b`, `c`), + count: fc.integer({ min: 1, max: 40 }), + release: fc.boolean(), + metadata: fc.boolean(), + }), + { minLength: 1, maxLength: 20 }, + ), + async (count, size, selected, history) => { + const rows = rowsOf(count) + const transport = createTransport(rows, size, selected) + let metadata = true + const session = createNoPeekSession(async (limit) => { + const packet = await transport.read(limit) + if (!metadata) delete packet.fact + return packet + }) + const windows = new Map() + session.subscribe(() => { + for (const [id, n] of windows) + expect(session.get(id)).toEqual(expectedWindow(rows, selected, n)) + }) + for (const action of history) { + metadata = action.metadata + if (action.release) { + session.release(action.id) + windows.delete(action.id) + } else { + session.request(action.id, action.count) + windows.set(action.id, action.count) + } + await session.refresh() + for (const [id, n] of windows) + expect(session.get(id)).toEqual(expectedWindow(rows, selected, n)) + } + }, + ), + oraclePropertyOptions(150, `cursor-pagination.no-peek`), + ) + }) + + it(`does not expose response facts before complete publication, or after reset`, async () => { + const transport = createTransport(rowsOf(9), 3) + const applied = createDeferred() + const received = createDeferred() + let held = false + const session = createNoPeekSession(async (limit) => { + const packet = await transport.read(limit) + if (held) { + received.resolve() + await applied.promise + } + return packet + }) + session.request(`a`, 3) + await session.refresh() + const old = session.get(`a`) + held = true + session.request(`a`, 10) + const pending = session.refresh() + const observed = pending.then( + () => undefined, + (error: unknown) => error, + ) + await received.promise + expect(session.get(`a`)).toBe(old) + session.reset() + applied.resolve() + expect(await observed).toMatchObject({ name: `AbortError` }) + expect(session.get(`a`)).toBeUndefined() + held = false + await session.refresh() + expect(session.get(`a`)).toEqual(expectedWindow(rowsOf(9), scope, 10)) + }) + + it(`notifies when only the published continuation changes`, async () => { + let transport = createTransport(rowsOf(3), 3) + const session = createNoPeekSession((limit) => transport.read(limit)) + session.request(`a`, 3) + const snapshots: Array = [] + session.subscribe(() => snapshots.push(session.get(`a`)?.hasNextPage)) + await session.refresh() + transport = createTransport(rowsOf(4), 3) + await session.refresh() + expect(snapshots).toEqual([false, true]) + expect(session.get(`a`)?.rows).toEqual(rowsOf(3)) + }) + + it.each([`acquisition`, `fallback`] as const)( + `keeps the last snapshot and queued peer usable after %s failure`, + async (boundary) => { + const transport = createTransport(rowsOf(9), 3) + const failure = new Error(`Failed publication`) + let fail = false + const session = createNoPeekSession(async (limit) => { + if (fail && (boundary === `acquisition` || limit === 7)) { + fail = false + throw failure + } + const packet = await transport.read(limit) + if (boundary === `fallback` && limit >= 6) delete packet.fact + return packet + }) + session.request(`a`, 3) + await session.refresh() + const old = session.get(`a`) + session.request(`a`, 6) + fail = true + let publications = 0 + session.subscribe(() => publications++) + await expect(session.refresh()).rejects.toBe(failure) + expect(session.get(`a`)).toBe(old) + expect(publications).toBe(0) + session.request(`peer`, 9) + const results = await Promise.allSettled([ + session.refresh(), + session.refresh(), + ]) + expect(results.map((result) => result.status)).toEqual([ + `fulfilled`, + `fulfilled`, + ]) + expect(session.get(`a`)).toEqual(expectedWindow(rowsOf(9), scope, 6)) + expect(session.get(`peer`)).toEqual(expectedWindow(rowsOf(9), scope, 9)) + }, + ) + + it(`restores peek when metadata is withdrawn without changing the rows`, async () => { + const transport = createTransport(rowsOf(9), 3) + let metadata = true + const session = createNoPeekSession(async (limit) => { + const packet = await transport.read(limit) + if (!metadata) delete packet.fact + return packet + }) + session.request(`a`, 3) + await session.refresh() + const before = session.get(`a`) + metadata = false + await session.refresh() + expect(transport.requests).toEqual([3, 3, 4]) + expect(session.get(`a`)).toEqual(before) + await session.refresh() + expect(transport.requests.at(-1)).toBe(4) + }) + + it(`saves a backend request only when peek crosses a page boundary`, async () => { + for (const size of [3, 4, 50]) { + const transport = createTransport(rowsOf(10), size) + const session = createNoPeekSession(transport.read) + session.request(`a`, 3) + await session.refresh() + expect(transport.requests).toEqual([3]) + expect(transport.backend.calls).toHaveLength(1) + const peek = createTransport(rowsOf(10), size) + await peek.read(4) + expect(peek.backend.calls).toHaveLength(size === 3 ? 2 : 1) + } + }) +}) diff --git a/packages/query-db-collection/tests/cursor-pagination.oracle.test.ts b/packages/query-db-collection/tests/cursor-pagination.oracle.test.ts new file mode 100644 index 0000000000..8db5a81566 --- /dev/null +++ b/packages/query-db-collection/tests/cursor-pagination.oracle.test.ts @@ -0,0 +1,387 @@ +import fc from 'fast-check' +import { describe, expect, it, vi } from 'vitest' +import { createDeferred } from '../../db/src/deferred.js' +import { oraclePropertyOptions } from '../../db/tests/oracle-config.js' +import { createBackend } from './cursor-pagination/backend.js' +import { expectedRows } from './cursor-pagination/model.js' +import { createCursorPager } from './cursor-pagination/pager.js' +import type { Row, Scope, Window } from './cursor-pagination/model.js' + +const rowsArbitrary = fc.uniqueArray( + fc.record({ + id: fc.integer({ min: 0, max: 60 }), + rank: fc.integer({ min: -3, max: 3 }), + group: fc.integer({ min: 0, max: 2 }), + }), + { selector: (row) => row.id, maxLength: 30 }, +) +const scopeArbitrary = fc.record({ + group: fc.option(fc.integer({ min: 0, max: 2 }), { nil: undefined }), + descending: fc.boolean(), +}) +const windowArbitrary = fc.record({ + offset: fc.integer({ min: 0, max: 35 }), + limit: fc.option(fc.integer({ min: 0, max: 35 }), { nil: undefined }), +}) +const fixtureRows: Array = Array.from({ length: 9 }, (_, id) => ({ + id, + rank: Math.floor(id / 6), + group: id % 2, +})) +const allAscending: Scope = { group: undefined, descending: false } + +it(`backend sequences reject foreign tokens while retaining their own continuations`, async () => { + await fc.assert( + fc.asyncProperty( + fc.integer({ min: 1, max: 5 }), + fc.boolean(), + async (size, descending) => { + const scope = { group: undefined, descending } + const rows = Array.from({ length: size * 3 }, (_, id) => ({ + id, + rank: id, + group: 0, + })) + const a = createBackend(rows, scope, size) + const b = createBackend(rows, scope, size) + const firstA = await a.fetchPage(undefined), + firstB = await b.fetchPage(undefined) + expect(firstA.nextCursor).not.toBe(firstB.nextCursor) + await expect( + Promise.resolve().then(() => b.fetchPage(firstA.nextCursor!)), + ).rejects.toThrow(`Foreign`) + const own = await b.fetchPage(firstB.nextCursor!) + expect(own.rows).toEqual( + expectedRows(rows, scope, { offset: size, limit: size }), + ) + }, + ), + oraclePropertyOptions(50, `cursor-pagination.backend-ownership`), + ) +}) + +async function checkRead( + source: Array, + scope: Scope, + window: Window, + read: (window: Window) => Promise>, +) { + const expected = expectedRows(source, scope, window) + expect(await read(window), `ordered slice ${JSON.stringify(window)}`).toEqual( + expected, + ) +} + +describe(`cursor pagination reference`, () => { + it(`preserves ties, filtering and reverse total order`, () => { + expect( + expectedRows( + fixtureRows, + { group: 1, descending: true }, + { offset: 1, limit: 3 }, + ).map((row) => row.id), + ).toEqual([5, 3, 1]) + expect( + expectedRows(fixtureRows, allAscending, { offset: 0, limit: 0 }), + ).toEqual([]) + }) + + it(`rejects capped-page, wrong-order and duplicate-row answers`, async () => { + const window = { offset: 0, limit: 8 } + const expected = expectedRows(fixtureRows, allAscending, window) + for (const wrong of [ + expected.slice(0, 2), + [...expected].reverse(), + [...expected.slice(0, 7), expected[0]!], + ]) { + await expect( + checkRead(fixtureRows, allAscending, window, () => + Promise.resolve(wrong), + ), + ).rejects.toThrow(`ordered slice`) + } + await checkRead(fixtureRows, allAscending, window, () => + Promise.resolve(expected), + ) + }) +}) + +describe(`opaque cursor adapter`, () => { + it.each([ + { offset: -1, limit: 1 }, + { offset: 0.5, limit: 1 }, + { offset: Infinity, limit: 1 }, + { offset: NaN, limit: 1 }, + { offset: 0, limit: -1 }, + { offset: 0, limit: 0.5 }, + { offset: 0, limit: Infinity }, + { offset: 0, limit: NaN }, + { offset: Number.MAX_SAFE_INTEGER, limit: 1 }, + ])( + `rejects invalid windows before transport: $offset/$limit`, + async (window) => { + const backend = createBackend(fixtureRows, allAscending, 2) + const pager = createCursorPager(backend.fetchPage) + await expect(pager.read(window)).rejects.toThrow(RangeError) + expect(backend.calls).toEqual([]) + await checkRead( + fixtureRows, + allAscending, + { offset: 0, limit: 2 }, + pager.read, + ) + }, + ) + + it(`captures each requested window before queued work starts`, async () => { + const backend = createBackend(fixtureRows, allAscending, 2) + const pager = createCursorPager(backend.fetchPage) + const window = { offset: 1, limit: 3 } + const expected = expectedRows(fixtureRows, allAscending, window) + const pending = pager.read(window) + window.offset = 7 + window.limit = 1 + expect(await pending).toEqual(expected) + }) + + it(`accepts large backend pages without spreading rows into arguments`, async () => { + const rows = Array.from({ length: 150_000 }, (_, id) => id) + const pager = createCursorPager(() => + Promise.resolve({ rows, nextCursor: null }), + ) + expect(await pager.read({ offset: 149_997, limit: undefined })).toEqual( + rows.slice(149_997), + ) + expect(await pager.read({ offset: 0, limit: 2 })).toEqual(rows.slice(0, 2)) + }) + + it.each([1, 2, 5, 50])( + `fulfills windows across backend pages of %i`, + async (pageSize) => { + const backend = createBackend(fixtureRows, allAscending, pageSize) + const pager = createCursorPager(backend.fetchPage) + for (const window of [ + { offset: 0, limit: 0 }, + { offset: 0, limit: 3 }, + { offset: 2, limit: 6 }, + { offset: 0, limit: undefined }, + { offset: 30, limit: 2 }, + ]) { + await checkRead(fixtureRows, allAscending, window, pager.read) + } + expect(backend.calls).toHaveLength( + Math.ceil(fixtureRows.length / pageSize), + ) + }, + ) + + it(`matches whole-relation slices across random window histories`, async () => { + await fc.assert( + fc.asyncProperty( + rowsArbitrary, + scopeArbitrary, + fc.integer({ min: 1, max: 8 }), + fc.array(windowArbitrary, { minLength: 1, maxLength: 20 }), + async (rows, scope, pageSize, windows) => { + const backend = createBackend(rows, scope, pageSize) + const pager = createCursorPager(backend.fetchPage) + for (const window of windows) + await checkRead(rows, scope, window, pager.read) + const calls = backend.calls.length + for (const window of [...windows].reverse()) + await checkRead(rows, scope, window, pager.read) + expect(backend.calls).toHaveLength(calls) + }, + ), + oraclePropertyOptions(100, `cursor-pagination.history`), + ) + }) + + it(`keeps answers invariant under backend page repartition`, async () => { + await fc.assert( + fc.asyncProperty( + rowsArbitrary, + scopeArbitrary, + windowArbitrary, + async (rows, scope, window) => { + for (const size of [1, 4, 50]) { + const backend = createBackend(rows, scope, size, true) + await checkRead( + rows, + scope, + window, + createCursorPager(backend.fetchPage).read, + ) + } + }, + ), + oraclePropertyOptions(100, `cursor-pagination.partition`), + ) + }) + + it.each([`abort`, `reject`, `reset`] as const)( + `keeps a healthy peer usable after held response %s`, + async (action) => { + const backend = createBackend(fixtureRows, allAscending, 2) + const gate = createDeferred() + const abort = new AbortController() + const failure = new Error(`backend failed`) + let calls = 0 + const pager = createCursorPager(async (cursor, signal) => { + const page = await backend.fetchPage(cursor, signal) + if (++calls === 1) await gate.promise + return page + }) + const outcomes: Array = [] + const first = pager.read({ offset: 0, limit: 2 }, abort.signal).then( + () => { + outcomes.push(`first-success`) + return undefined + }, + (error: unknown) => { + outcomes.push(`first-error`) + return error + }, + ) + await vi.waitFor(() => expect(calls).toBe(1)) + expect(outcomes).toEqual([]) + if (action === `abort`) abort.abort() + if (action === `reset`) pager.reset() + const peer = pager.read({ offset: 0, limit: 6 }) + const peerObserved = peer.then((rows) => { + outcomes.push(`peer-success`) + return rows + }) + if (action === `reject`) gate.reject(failure) + else gate.resolve() + const error = await first + if (action === `reject`) expect(error).toBe(failure) + else expect(error).toMatchObject({ name: `AbortError` }) + await expect(peerObserved).resolves.toEqual( + expectedRows(fixtureRows, allAscending, { offset: 0, limit: 6 }), + ) + expect(outcomes).toEqual([`first-error`, `peer-success`]) + // Cancelling a reader does not cancel Query's shared acquisition. A + // successful page remains reusable; failed/reset acquisitions restart. + if (action === `abort`) { + expect(backend.calls[1]).toEqual(expect.any(String)) + expect(backend.calls).toHaveLength(3) + } else { + expect(backend.calls.slice(0, 2)).toEqual([undefined, undefined]) + } + }, + ) + + it(`serializes unequal peer windows without refetching cached pages`, async () => { + const backend = createBackend(fixtureRows, allAscending, 2) + const pager = createCursorPager(backend.fetchPage) + const windows = [ + { offset: 0, limit: 3 }, + { offset: 2, limit: 6 }, + { offset: 0, limit: 2 }, + ] + const results = await Promise.all( + windows.map((window) => pager.read(window)), + ) + expect(results).toEqual( + windows.map((window) => expectedRows(fixtureRows, allAscending, window)), + ) + expect(backend.calls).toHaveLength(4) + await checkRead(fixtureRows, allAscending, windows[0]!, pager.read) + expect(backend.calls).toHaveLength(4) + }) + + it(`resets source generations and isolates filter/order scopes`, async () => { + await fc.assert( + fc.asyncProperty( + rowsArbitrary, + scopeArbitrary, + scopeArbitrary, + async (rows, firstScope, secondScope) => { + let backend = createBackend(rows, firstScope, 3) + const pager = createCursorPager((cursor, signal) => + backend.fetchPage(cursor, signal), + ) + const all = { offset: 0, limit: undefined } + await checkRead(rows, firstScope, all, pager.read) + const changed = rows.map((row) => ({ + ...row, + rank: -row.rank, + id: row.id + 100, + })) + backend = createBackend(changed, secondScope, 2) + pager.reset() + await checkRead(changed, secondScope, all, pager.read) + const independent = createCursorPager( + createBackend(rows, firstScope, 4).fetchPage, + ) + await checkRead(rows, firstScope, all, independent.read) + await checkRead(changed, secondScope, all, pager.read) + }, + ), + oraclePropertyOptions(50, `cursor-pagination.reset`), + ) + }) + + it(`rejects a repeated opaque cursor without looping`, async () => { + let calls = 0 + const pager = createCursorPager(() => { + calls++ + return Promise.resolve({ rows: [], nextCursor: `same` }) + }) + await expect(pager.read({ offset: 0, limit: 1 })).rejects.toThrow( + `repeated`, + ) + expect(calls).toBe(2) + }) + + it(`recovers after faults at generated intermediate page boundaries`, async () => { + await fc.assert( + fc.asyncProperty( + fc.array(fc.integer({ min: -2, max: 2 }), { + minLength: 12, + maxLength: 24, + }), + fc.integer({ min: 1, max: 4 }), + fc.integer({ min: 0, max: 2 }), + fc.boolean(), + fc.constantFrom(`reject`, `abort`), + async (ranks, size, faultPage, descending, fault) => { + const rows = ranks.map((rank, id) => ({ id, rank, group: id % 2 })) + const scope = { group: undefined, descending } + const backend = createBackend(rows, scope, size) + const abort = new AbortController() + const error = new Error(`page failure`) + let calls = 0 + const pager = createCursorPager(async (cursor, signal) => { + const page = await backend.fetchPage(cursor, signal) + if (calls++ === faultPage) { + if (fault === `reject`) throw error + abort.abort(error) + } + return page + }) + const all = { offset: 0, limit: undefined } + const failed = pager.read(all, abort.signal) + // Attach observers before the healthy peer can expose a failure. + const outcomes = Promise.allSettled([failed, pager.read(all)]) + const [first, peer] = await outcomes + expect(first).toEqual({ status: `rejected`, reason: error }) + expect(peer).toEqual({ + status: `fulfilled`, + value: expectedRows(rows, scope, all), + }) + expect(backend.calls).toHaveLength( + // Query marks failed acquisitions stale and rebuilds their prefix. + Math.ceil(rows.length / size) + + (fault === `reject` ? faultPage + 1 : 0), + ) + const before = backend.calls.length + await checkRead(rows, scope, { offset: 1, limit: 3 }, pager.read) + expect(backend.calls).toHaveLength(before) + }, + ), + oraclePropertyOptions(100, `cursor-pagination.failure`), + ) + }) +}) diff --git a/packages/query-db-collection/tests/cursor-pagination.publication-oracle.test.ts b/packages/query-db-collection/tests/cursor-pagination.publication-oracle.test.ts new file mode 100644 index 0000000000..6f42d992ba --- /dev/null +++ b/packages/query-db-collection/tests/cursor-pagination.publication-oracle.test.ts @@ -0,0 +1,324 @@ +import { QueryClient, QueryObserver } from '@tanstack/query-core' +import fc from 'fast-check' +import { describe, expect, it, vi } from 'vitest' +import { createDeferred } from '../../db/src/deferred.js' +import { oraclePropertyOptions } from '../../db/tests/oracle-config.js' +import { createCursorPager } from '../src/index.js' +import { createBackend } from './cursor-pagination/backend.js' +import { expectedRows } from './cursor-pagination/model.js' +import type { Row } from './cursor-pagination/model.js' + +const scope = { group: undefined, descending: false } +const rowsFor = (count: number, version = 0): Array => + Array.from({ length: count }, (_, id) => ({ id, rank: id, group: version })) +const createClient = () => + new QueryClient({ + defaultOptions: { + queries: { retry: false, staleTime: Infinity, gcTime: Infinity }, + }, + }) + +// The reference remains a whole relation. These laws add publication and +// next-use observations, not a model of Query's retryer or page cache. +describe(`cursor cache publication`, () => { + it.each( + [true, false].flatMap((cancel) => + [false, true].map((sharedPager) => ({ cancel, sharedPager })), + ), + )( + `force refresh requires cancellation of held growth: $cancel/$sharedPager`, + async ({ cancel, sharedPager }) => { + await fc.assert( + fc.asyncProperty( + fc.integer({ min: 1, max: 4 }), + fc.integer({ min: 1, max: 3 }), + async (size, depth) => { + const client = createClient() + const key = [`posts`, `cursor-pages`] + const entered = createDeferred() + const release = createDeferred() + let source = rowsFor(size * (depth + 1)) + // Each token retains its original immutable backend, even when a new + // sequence starts. The fixture must not rescue stale cursors. + const routes = new Map>() + let serial = 0 + let starts = 0 + let calls = 0 + let holdAt = Infinity + const options = { + queryClient: client, + queryKey: key, + fetchPage: async ( + cursor: string | undefined, + signal: AbortSignal, + ) => { + const backend = + cursor === undefined + ? createBackend(source, scope, size) + : routes.get(cursor)! + if (cursor === undefined) starts++ + const backendCursor = cursor?.slice(cursor.indexOf(`:`) + 1) + const page = await backend.fetchPage(backendCursor, signal) + const nextCursor = + page.nextCursor === null + ? null + : `${++serial}:${page.nextCursor}` + if (nextCursor !== null) routes.set(nextCursor, backend) + if (++calls === holdAt) { + entered.resolve() + await release.promise // Deliberately deliver even after abort. + } + return { rows: page.rows, nextCursor } + }, + } + const retained = createCursorPager(options) + const reader = () => + sharedPager ? retained : createCursorPager(options) + let refreshEntered: + | ReturnType> + | undefined + const outer = new QueryObserver(client, { + queryKey: [`posts`, `rows`], + queryFn: () => { + refreshEntered?.resolve() + return reader().read({ limit: size * depth }) + }, + }) + const unsubscribe = outer.subscribe(() => {}) + try { + await outer.refetch({ cancelRefetch: false, throwOnError: true }) + holdAt = calls + 1 + const growth = reader() + .read({}) + .catch((error: unknown) => error) + await entered.promise + source = rowsFor(source.length, 1) + const query = client + .getQueryCache() + .find({ queryKey: key, exact: true })! + const joined = createDeferred() + const fetch = query.fetch.bind(query) + const witness = vi + .spyOn(query, `fetch`) + .mockImplementation((...args) => { + const result = fetch(...args) + joined.resolve() + return result + }) + // Follow the guide's force-refresh recipe through a real active + // outer query. Observe its acquisition before releasing old data. + if (cancel) await client.cancelQueries({ queryKey: [`posts`] }) + refreshEntered = createDeferred() + const refresh = client.invalidateQueries({ queryKey: [`posts`] }) + await refreshEntered.promise + if (sharedPager && !cancel) { + // This read is behind the old growth in the same queue. That + // success clears invalidation before the queued read can see it. + expect(witness).not.toHaveBeenCalled() + } else await joined.promise + witness.mockRestore() + release.resolve() + await refresh + const oldResult = await growth + const expected = expectedRows(source, scope, { + offset: 0, + limit: size * depth, + }) + const check = () => + expect( + outer.getCurrentResult().data, + `refresh publishes the new snapshot`, + ).toEqual(expected) + if (cancel) { + check() + expect(oldResult).toMatchObject({ name: `AbortError` }) + expect(starts).toBe(2) + expect( + await createCursorPager(options).read({ + limit: size * depth, + }), + ).toEqual(expected) + } else { + // Fault control: omitting the documented cancellation reaches + // the same checker with a valid but obsolete backend sequence. + expect(check).toThrow(`refresh publishes`) + expect(outer.getCurrentResult().data).toEqual( + expectedRows(rowsFor(source.length), scope, { + offset: 0, + limit: size * depth, + }), + ) + expect(starts).toBe(1) + } + } finally { + release.resolve() + unsubscribe() + client.clear() + } + }, + ), + oraclePropertyOptions(50, `cursor-pagination.refresh-publication`), + ) + }, + ) + + it.each([`growth`, `refresh`] as const)( + `%s rejects malformed final continuations without poisoning the cache`, + async (phase) => { + await fc.assert( + fc.asyncProperty( + fc.integer({ min: 1, max: 4 }), + fc.integer({ min: 1, max: 4 }), + fc.nat({ max: 3 }), + fc.boolean(), + async (size, depth, repeat, retry) => { + const client = createClient() + client.setDefaultOptions({ + queries: { + ...client.getDefaultOptions().queries, + retry: retry ? 1 : false, + retryDelay: 0, + }, + }) + const key = [`protocol`] + let source = rowsFor(size * (depth + 1)) + let backend = createBackend(source, scope, size) + let params: Array = [] + let malformed = false + let calls = 0 + const entered = createDeferred() + const release = createDeferred() + const options = { + queryClient: client, + queryKey: key, + fetchPage: async ( + cursor: string | undefined, + signal: AbortSignal, + ) => { + calls++ + if (cursor === undefined) { + backend = createBackend(source, scope, size) + params = [] + } else params.push(cursor) + const page = await backend.fetchPage(cursor, signal) + if (malformed && page.nextCursor === null) { + entered.resolve() + await release.promise + } + return malformed && page.nextCursor === null + ? { ...page, nextCursor: params[repeat % params.length]! } + : page + }, + } + const pager = createCursorPager(options) + try { + await pager.read({ + limit: phase === `growth` ? size * depth : source.length, + }) + if (phase === `refresh`) { + source = rowsFor(source.length, 1) + await client.invalidateQueries({ queryKey: key }) + } + const previous = client.getQueryData(key) + const published: Array = [] + const unsubscribe = client.getQueryCache().subscribe((event) => { + if (event.type === `updated` && event.action.type === `success`) + published.push(event.query.state.data) + }) + malformed = true + try { + const pending = pager.read({}).catch((error: unknown) => error) + await entered.promise + const query = client + .getQueryCache() + .find({ queryKey: key, exact: true })! + const joined = createDeferred() + const fetch = query.fetch.bind(query) + const witness = vi + .spyOn(query, `fetch`) + .mockImplementation((...args) => { + const result = fetch(...args) + joined.resolve() + return result + }) + const peer = createCursorPager(options) + .read({}) + .catch((error: unknown) => error) + await joined.promise + witness.mockRestore() + release.resolve() + for (const result of await Promise.all([pending, peer])) { + expect(result).toBeInstanceOf(Error) + expect(String(result)).toContain(`repeated`) + } + expect( + published, + `invalid acquisition must not publish success`, + ).toEqual([]) + expect(client.getQueryData(key)).toBe(previous) + } finally { + unsubscribe() + } + malformed = false + const before = calls + expect(await pager.read({})).toEqual( + expectedRows(source, scope, { offset: 0, limit: undefined }), + ) + expect(calls).toBeGreaterThan(before) + const settled = calls + expect(await pager.read({})).toEqual( + expectedRows(source, scope, { offset: 0, limit: undefined }), + ) + expect(calls).toBe(settled) + } finally { + release.resolve() + client.clear() + } + }, + ), + oraclePropertyOptions(50, `cursor-pagination.protocol-publication`), + ) + }, + ) + + it(`cached slices visit only requested rows`, async () => { + await fc.assert( + fc.asyncProperty( + fc.integer({ min: 1, max: 100 }), + fc.nat({ max: 120 }), + fc.option(fc.nat({ max: 120 }), { nil: undefined }), + async (count, offset, limit) => { + const client = createClient() + const rows = rowsFor(count) + let visits = 0 + const observed = rows.map((row) => row) + for (let index = 0; index < count; index++) { + Object.defineProperty(observed, index, { + enumerable: true, + get: () => { + visits++ + return rows[index] + }, + }) + } + const pager = createCursorPager({ + queryClient: client, + queryKey: [`slice`], + fetchPage: () => + Promise.resolve({ rows: observed, nextCursor: null }), + }) + try { + await pager.read({}) + visits = 0 + const expected = expectedRows(rows, scope, { offset, limit }) + expect(await pager.read({ offset, limit })).toEqual(expected) + expect(visits).toBe(expected.length) + } finally { + client.clear() + } + }, + ), + oraclePropertyOptions(100, `cursor-pagination.slice-work`), + ) + }) +}) diff --git a/packages/query-db-collection/tests/cursor-pagination/LOSS-AUDIT.md b/packages/query-db-collection/tests/cursor-pagination/LOSS-AUDIT.md new file mode 100644 index 0000000000..169f6f7488 --- /dev/null +++ b/packages/query-db-collection/tests/cursor-pagination/LOSS-AUDIT.md @@ -0,0 +1,198 @@ +# Cursor implementation loss audit + +2026-09-15. Base `3c54e89ae`, uncommitted cursor helper and tests. + +The frozen candidate was the 143-line Query-owned `createCursorPager`, its +documentation and 87 passing tests. Two fresh, source-isolated agents scanned +the issue and historical plans separately. A third fresh agent was unavailable; +the implementing agent checked the oracle guide. That last pass is correlated, +not an independent sign-off. These are loss traces, not a completeness proof. + +## Issue-only recovered material + +Source: [issue #863](https://github.com/TanStack/db/issues/863), including the +three comments below. Exploratory proposals are not established requirements. + +| Recovered item | Where the frozen candidate lost it | Disposition under the chosen scope | +| ---------------------------------------------------------------------------------------------------------------------------- | ---------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| Automatic scope identity from collection/filter/order, excluding windows | Replaced by caller-supplied Query keys | Explicit responsibility shift. Applications translate arbitrary endpoint parameters; the helper cannot derive that translation. Keys must describe source, tenant, filter and order. No automatic IR/session registry. | +| Existing subscription as the session owner | QueryClient/key ownership replaces it | Explicit alternative not taken. Cached pages outlive an individual subscription and reuse Query's existing cache/GC rules. No additional subscription-owned state. | +| PK tie-breaking might remove the original cost difference; opaque cursors still have separate value | Compressed away with the metadata proposal | Preserve as Sam's argument, not our measurement. The experiment's one-call saving appears only when N+1 crosses a physical page boundary. No general bandwidth/latency savings claim. | +| Declarative `queryCollectionOptions` pagination config, initial numeric page parameter, next-page callback, simplified hooks | Reduced to a helper composed inside queryFn | Explicitly not implemented. The helper starts at undefined and follows string tokens ending at null. Numeric page APIs keep the existing documented adapter pattern. No new hook configuration. | +| Enforce a PK tie-breaker and detect order changes | Became a caller/backend precondition | Explicitly not implemented. Endpoint order must be total; changed scope needs a different key. The helper cannot inspect an opaque endpoint's sorting. | +| Generic facility for all collection types, outside the query engine | Narrowed to Query-backed pagination | Keep the no-engine-change boundary. No cross-adapter metadata extension point in this PR. | +| Content-agnostic context, with interpretation callback either on the hook or collection | Both alternatives collapsed into “no metadata” | Both remain excluded. This also excludes total counts, previous-page metadata, sync progress, stream/snapshot positions and Electric transaction IDs. | +| Initial remote offset jump, then cursor continuation | Local offset slicing can sound equivalent | Explicitly excluded. Offset means traversing enough cursor pages and slicing; it is not a direct backend jump. | +| Metadata-driven no-peek with fallback | Explicitly rejected earlier | Still excluded. The maintainer chose to close #863 with partial implementation: cursor loading ships; metadata/no-peek is intentionally not pursued. | + +Comment provenance: [Sam's identity/cost/architecture alternatives](https://github.com/TanStack/db/issues/863#issuecomment-3556836514), +[Kevin's two context API placements](https://github.com/TanStack/db/issues/863#issuecomment-3557243731), +[Enyel's declarative pagination proposal](https://github.com/TanStack/db/issues/863#issuecomment-3559079761). + +## Plan-only recovered material + +Sources: [contract and plan](./README.md), [no-peek results](./NO-PEEK-RESULTS.md), +and the user's choices to ship one PR, retain caching, and reuse Query semantics. + +| Recovered item | Loss mechanism | Disposition | +| ---------------------------------------------------------------------------------------- | ------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| Immutable cached rows; no cross-read identity promise | Summary omitted caller obligations | Added to public guide. Read returns a sliced array, not detached row objects. | +| Reset does not refresh collection rows | Two-cache boundary compressed | Public guide explicitly distinguishes reset, outer refetch and shared-prefix invalidation. Integration now checks that outer refetch alone can reuse fresh pages before invalidation refreshes both. | +| Endpoint owns cursor validity/invalidation rules | Responsibility became only a stable-source precondition | Kept explicit in the public guide. No TTL claims to repair an inconsistent backend sequence. | +| Prefix transport counts exclude separate tie-group requests; virtual row fields excluded | Verification summary compressed scope | Preserve these limits in README and final receipts. Pager work laws cover all calls within their single scope. | +| Historical cancellation fault receipts use the old reader-owned acquisition contract | Temporal flattening | Keep receipts labeled historical. Current Query-owned law retains successful pages after a reader abort; QueryClient cancellation separately discards late acquisition results. | +| Specific no-peek benefit and safety laws | Entire feature excluded | Keep the experiment and result file: boundary-only request saving; retained-tail truth; publication/prefix association; sticky per-consumer fallback; fact-only notification. These are not newly owed shipping features. | +| Generic eligibility and metadata wiring were unproven | Could be mistaken for completed features later removed | Preserve their unproven status. No claim that static scoped tests establish mutable multi-source metadata correctness. | +| Fresh-per-call refetching and oldest-page TTL | Replaced by Query-native semantics | Deliberate change. New helper instances share pages by QueryClient/key; successful page growth updates Query's freshness timestamp. No independent clock/TTL implementation. | +| One PR | Could be confused with earlier “independent” adapter recommendation | Preserved: helper, tests and docs belong in one PR. | + +## Oracle-guide pass and recovered tests + +Source: [Writing reliable oracle tests](../../../../docs/contributing/oracle-tests.md). +This pass shares the implementer's context; it is weaker than the two isolated +readings above. It compares promises and test evidence, not every possible state. + +| Guide obligation | Evidence or recovered gap | Action / limit | +| -------------------------------------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| Reference substantially simpler than production | Full filter/sort/slice relation; cache model adds only permitted source snapshot and deadline | Retained. No copied Query observer/retry/cache model. | +| Real production path, fixture must not perform the tested work | Opaque backend only serves pages; actual helper follows tokens. Integration runs QueryClient → QueryCollection → graph → controller | Retained. Numeric rank/id fixture rejects unsupported predicates; not a generic IR interpreter or native framework run. | +| Calibrate checker sensitivity, not merely green execution | Existing capped/reversed/duplicate controls; new cache law lacked a faulty-cache control | Added a deliberate `ensureInfiniteQueryData` substitution that ignores invalidation. The same window checker rejects stale rows; normal refresh passes. This is an assertion failure, not timeout/setup failure. | +| Control the claimed cancellation boundary | Reader abort/reset existed; QueryClient transport cancellation newly documented without a witness | Added held transport cancellation, aborted-signal witness, rejected read, no late cache installation, and successful subsequent read. | +| Distinguish outer request success from fresh backend data | Guide explained prefix invalidation, but test only exercised that happy path | Extended all four mutable-refetch cells: outer refetch first reuses cached rows with zero extra prefix requests; shared-prefix invalidation then exposes changed backend rows. | +| Preserve valid histories when replacing machinery | Offset/backward/overlap/zero/unlimited, partition, ties, failures, reset, peers remain | No prior test file deleted. Abort/retry work expectations changed explicitly to Query ownership; exact rows/error/peer outcomes remain checked. | +| Async observations and public scope | Held initial delivery; Query cache held/failed refresh; every ready static window publication; settled mutable-refetch rows/hasNextPage | Does not establish every intermediate mutable-refresh publication or all framework schedules. Those broader contracts stay with existing collection/publication owners. | +| Replay, shrinking and executable registration | Stable property names and shared seed/path parser; fixed/random lanes; new files in package oracle script | Fixed-seed 100× and default runs executed. No claim that one seed schedules real network/OS work, or that every counterexample has a separate replay-process receipt. | +| Domain boundaries and lifecycle work | Finite stable backend sequences, controlled faults; real expiry/invalidation/GC; cancellation and reset | No snapshot guarantee for changing HTTP endpoints, persistent cursor cache guarantee, per-page eviction or generic ordering validation. | + +## Evidence after recovery + +Before audit recovery: package 341 tests passed, types/lint/build passed; +100× seed 863 passed 87 tests and 60,000 generated histories. Recovery adds +two tests and strengthens four existing integration cells; final receipts are +recorded in README after rerunning the gates. + +Browser ESM, esbuild minification, target ES2020, same checkout/settings: +`queryCollectionOptions` alone: 46,590 bytes / 14,985 gzip. +Adding `createCursorPager`: 50,667 bytes / 16,475 gzip (+4,077 / +1,490). +This is an opt-in diagnostic import comparison, not every application's bundle +or an all-core comparison. The helper adds no core/hook source changes. + +## Subsequent prep-pr corrections + +The earlier cancellation witness covered only an initial request, where Query +has no cached result to restore. Review found that growth/refresh cancellation +could return the old cache as success and trigger another transport. It also +found that inherited `maxPages` evicted the prefix addressed by offset reads and +inherited `select` changed the observer's page format. The earlier cache model +was small enough; its configuration and scheduling domains were too narrow. + +The expanded oracle crosses global/key defaults with selection/page eviction +and generated growth, backward windows and invalidation. Cancellation now crosses +initial/growth/refresh, page size/depth, shared waiters, late response delivery +and subsequent recovery. The real peer fetch join is observed before cancellation; +a shallow cache hit is checked while a deeper peer remains held. + +Before the production fixes, eight of the expanded suite's 17 cells failed; +afterward all pass. Seed 863 minimized failing size/depth inputs to `[1, 1]`. +The fixes disable incompatible internal-page defaults and preserve Query's raw +acquisition rejection without blocking fresh cache hits. No reference-model +branch was added to imitate Query internals. Final full-package/stress/build +receipts and updated size measurements are in README; the measurements above +describe the earlier audit candidate, not the final helper. + +The audit can itself turn old suggestions into obligations or over-rescue +interesting detail. The tables separate source observations from the later +scope dispositions. They do not infer consensus among issue commenters or +certify that the implementation has no undiscovered bugs. + +## Cache publication review + +The next external review exposed two further false-green boundaries. The cache +history awaited every read, so it never invalidated during growth. The repeated +cursor test observed rejection, but not whether invalid data had already been +published or whether a later read could recover. + +The publication oracle keeps the same full-relation reference. It adds: + +- Immutable cursor sequences with held suffix delivery, a real active outer + QueryObserver, and the documented cancel-then-invalidate procedure. Omitting + cancellation is a fault control rejected by the same row checker. +- Growth and refresh with a malformed final continuation, generated page sizes, + depths and backward token targets, shared waiters, and Query retry enabled or + disabled. Assertions cover rejection, no successful cache publication, retained + last-good data, repair, and a subsequent fresh cache hit. +- Generated cached windows checked against full filter/sort/slice truth, with + row-access counts requiring work proportional to the returned slice rather + than all cached rows. + +Before fixes, all four initial cells failed with seed 863. Growth/refresh +protocol failures shrank to size/depth/target `[1, 1, 0]`, where an invalid +acquisition emitted a successful cache publication. The slice failure shrank +to `[1, 1, undefined]`: a beyond-end empty read still accessed the cached row. +The held-growth failure returned version 0 after a required version-1 refresh. + +Refresh is a documented procedure correction, not a new invalidation engine. +The driver now cancels before invalidating; the old procedure remains a negative +control. Protocol validation moved into response acquisition, before Query can +publish or resolve shared waiters. Its weakly held token set lasts only as long +as the acquisition signal; cached page parameters seed later growth. Slice +collection adds no persistent state. No production reference model, core query +change, metadata API or no-peek feature was added. + +## Nested acquisition and fixture review + +The next review tested boundaries the standalone cache oracle did not own: +an actual outer QueryCollection, browser retry defaults, aborting a queued reader, +and manual row writes. Nine boundary checks failed before fixes; a separate +generated manual-write law reproduced the documented prefix collision. + +- The browser boundary oracle uses real QueryCollection preload and Query state. + Page cancellation must settle the outer row query with an ordinary AbortError; + silently replaced acquisitions must deliver the replacement's new values. + It also checks prompt reader abort without transport cancellation, fresh reads + beside same-turn cancellation, global/default retry consistency, legacy signals, + invalid response/cache continuations, and unrelated-query hash work. +- Actual acquisitions and cache hits now follow separate paths. One Query observer + identifies its current query directly; the cache no longer gets scanned per page. + Readers follow silent replacements and translate explicit acquisition cancellation + out of Query's internal control-error type. Aborted readers release the queue + without canceling the shared transport; generation/reset isolation stays intact. +- The documented row/page keys are siblings beneath a resource prefix. The manual + write law crosses insert/update/delete and sizes, retaining exact public row + values and unchanged page-cache records. This fixes the recipe, not QueryCollection's + intentional prefix-wide manual-write behavior. +- CodeRabbit's late-delivery observations now await a transport marker and drain + the promise work before checking cache fences. Its backend fixture finding is + covered by unique sequence namespaces and a generated foreign-token rejection + law. Fixed `next` tokens in the boundary suite still exercise lawful reuse across + sequential acquisitions; fixture isolation is not a new backend API requirement. + +The reference relation remains unchanged. Undefined continuations are not valid +under the null-only protocol: the helper now gives a clear error, rejects malformed +cached continuations instead of looping, and the example normalizes an endpoint's +omitted terminal cursor. Full-prefix stale refresh and independent cache GC are +retained Query semantics. No-peek experiments remain valuable tests, not shipping +features. The maintainer's patch-release and partial-closeout decisions stand. + +## Follow-up: defensive cache writes and queued refresh + +The manual-write oracle now varies raw versus selected responses, sibling versus +nested page keys for raw responses, and generated insert/update/delete histories. +It checks exact collection rows, row-cache contents, selected wrapper metadata, +seeding an empty raw cache, and the full untouched page-query state. The reference +is still an independent array; no Query cache state machine was added. + +The nested raw-cache cell failed before the guard, shrinking to one insert and +two initial rows (seed -441317402, path 0:0:0:0). A deliberate guard mutant which +called setQueryData with the unchanged page object also failed: that call clears +invalidation and changes freshness despite preserving the rows. The final guard +does not call setQueryData for an existing non-array record in the raw write path. +Selected response writes are unchanged. Mixed selected/page caches under one +prefix remain unsupported; the guide still prescribes sibling prefixes. + +The existing forced-refresh oracle now crosses new-per-call and retained pagers. +With a retained pager, the old growth can finish before a queued read observes +invalidation. Cancel-then-invalidate passes both paths. Invalidation alone remains +a negative control in both; the scratch cancelRefetch-on-invalidated candidate +repairs only the new-pager path. Automatic invalidation repair would require a +separate contract decision, not merely that one-line option change. diff --git a/packages/query-db-collection/tests/cursor-pagination/NO-PEEK-RESULTS.md b/packages/query-db-collection/tests/cursor-pagination/NO-PEEK-RESULTS.md new file mode 100644 index 0000000000..48de6d7141 --- /dev/null +++ b/packages/query-db-collection/tests/cursor-pagination/NO-PEEK-RESULTS.md @@ -0,0 +1,127 @@ +# No-peek experiment: result and limits + +The narrow design works in the tested domain. An authoritative continuation +fact can replace the extra output row when the backend prefix maps directly to +the published query prefix. It is not safe as a collection-wide boolean. + +This is test-only experimental code, validated on merge head `3c54e89ae` with +rebuilt core and db-ivm packages. No production changes or public API are added. + +## Measured benefit + +For a three-row UI window and three-row backend pages over nine rows: + +| Path | Initial output demand | Backend page calls | Displayed rows | Has next | +| -------------------------------------------- | --------------------: | -----------------: | -------------- | -------- | +| Current production window controller | 4 | 2 | 0, 1, 2 | true | +| Experimental session and narrow graph bridge | 3 | 1 | 0, 1, 2 | true | + +Both paths use the same opaque-page adapter logic and real collection graph. +These are request counts, not latency or bundle-size benchmarks. When the +backend page already contains the extra row (page sizes four or fifty in the +protocol comparison), both paths make one call. At authoritative exhaustion +after exactly three rows, both make one call too: the cursor adapter already +knows it has reached the end, even if the loader requests a further slice. + +## The independent oracle + +The model remains the full relation filtered, sorted and sliced, with next-page +truth determined by its total length. It has no cursors, stamps, lease state or +fallback state. The transport fixture derives its continuation only from opaque +backend pages and its retained tail, not from that model. + +Generated histories vary source size, filter, direction, backend page size, +consumer count/window size, peer release and metadata omission. Every session +publication is checked against each consumer's reference window. Fixed probes +add reset during held delivery, missing/mismatched facts, fact-only changes, +metadata withdrawal, acquisition/fallback failure and healthy peer recovery. + +## What the failures taught us + +The first candidate used the latest response's `hasMore` for all consumers and +treated missing metadata as false. It failed seven of nine tests. Seed `863` +shrunk one failure to two source rows, backend page size one, a one-row window, +and missing metadata. It incorrectly announced the end. + +The tested repair has four rules: + +1. A real output row beyond a consumer's window proves continuation, regardless + of a deeper consumer's terminal fact. +2. Otherwise, accept only a fact for the same complete publication and matching + prefix boundary, with a transparent source-to-output mapping. +3. Missing or inapplicable facts acquire `N+1`. Keep this requirement on that + consumer's lease, so a peer's departure cannot silently remove its witness. +4. A new continuation fact changes the session snapshot even if rows are equal. + Do not publish response-time metadata while the corresponding output is held. + +Reset rejects old-generation completions. Failed requests preserve the previous +snapshot and do not poison later reads. These are rules of the test candidate; +they are not new guarantees supplied by the current production API. + +## Real production boundary tested + +The integration probe runs an on-demand source through a real live query. The +test-only bridge admits its fact after `setWindow` and `preload` complete, and +reads actual public rows. It compares the candidate to the unchanged production +window controller for source sizes zero, three, four and nine. A held transport +response proves no candidate snapshot is published merely because the fact has +arrived. The bridge is deliberately restricted to an immutable source with a +unique-id order. It supports the loader's offset and id-equality refinements. + +An opaque local filter preserves order but changes membership: the raw source +prefix says more rows exist while the complete filtered query has no next page. +The ineligible candidate correctly requests `N+1` output rows. The existing +compiler chooses full-source loading for this filter; the experiment does not +replace that path or claim it is cheap. + +## What remains unproven / unimplemented + +- Generic eligibility detection. Matching order alone is insufficient. The + backend must also describe the same result membership and row cardinality. + Joins, aggregates, local filters and optimistic changes require rejection or + an independently justified transfer rule. Simple one-to-one projections need + not be excluded merely because field names change. +- General fact transport through QueryCollection, graph refinement and shared + window publication. The bridge here is test-only and manually scoped; it is + not production metadata plumbing and does not cover mutable multi-source + graphs or all response/application interleavings. +- Automatic metadata-only notification and snapshot-cache invalidation in the + existing framework controller. The candidate demonstrates the required law; + the current controller still derives continuation solely from output rows. +- Automatic invalidation on local writes/refetch and ownership/cleanup wiring. + The protocol tests explicitly drive refresh and reset. They do not prove that + all production events trigger those actions. +- Incorrect backend claims or incomplete load fulfillment. A successful public + prefix must still obey the existing request-completion contract. A `hasMore` + flag is not a remedy for arbitrary adapters silently returning partial data. + +The prototype's publication stamp is a test identity witness, not a proposal +for a new global token registry. Production should reuse existing publication +and ownership boundaries where possible. Its small controller is not a full +replacement for the production controller's lifecycle and error handling. + +## Verification + +- First red: seven assertion failures, two passes; no setup failure or timeout. +- Final fixed-seed campaign: all 66 tests across both experiments passed. + The new no-peek property ran 15,000 histories at multiplier 100, seed `863`; + the preceding cursor experiment ran another 35,000 generated cases. +- A separate default random-seed run covers the no-peek files. +- Package TypeScript and lint on all experiment TypeScript passed. + +From `packages/query-db-collection`: + +```sh +TANSTACK_DB_ORACLE_RUNS_MULTIPLIER=100 TANSTACK_DB_ORACLE_SEED=863 \ + ../../node_modules/.bin/vitest run \ + tests/cursor-pagination.no-peek.test.ts \ + tests/cursor-pagination.no-peek.integration.test.ts \ + tests/cursor-pagination.oracle.test.ts \ + tests/cursor-pagination.integration.test.ts \ + --typecheck.enabled=false --maxWorkers=1 +``` + +Recommendation: retain the cursor adapter independently. If proceeding with +no-peek, start with explicitly eligible single-source queries, publication-bound +facts, and per-consumer fallback. Do not implement the original global setter +shortcut, and do not build a general metadata/coverage platform for this benefit. diff --git a/packages/query-db-collection/tests/cursor-pagination/README.md b/packages/query-db-collection/tests/cursor-pagination/README.md new file mode 100644 index 0000000000..c322f6cf38 --- /dev/null +++ b/packages/query-db-collection/tests/cursor-pagination/README.md @@ -0,0 +1,256 @@ +# Cursor adapter: contract, experiments and implementation + +Work arising from https://github.com/TanStack/db/issues/863. The independent +test-first experiment now exercises the public `createCursorPager` helper in +`src/cursor-pagination.ts`. Keep the existing output-row peek-ahead contract. + +The separate [no-peek experiment results](./NO-PEEK-RESULTS.md) evaluate an +opt-in publication-bound continuation fact. That candidate does not change the +cursor experiment or the existing production controller. + +## Contract before implementation + +- A pager belongs to one source/filter/total-order sequence. The adapter creates + separate pagers for separate scopes; it never shares a single latest cursor. +- A successful read returns exactly the requested offset/limit slice, or fewer + rows only at authoritative exhaustion. Undefined limit drains the source. +- Backend cursors are opaque and point after the backend response, not after + the UI slice. Retain the unreturned tail before advancing that cursor. +- Zero limit performs no transport work. Short and empty nonterminal pages do + not establish exhaustion. A repeated cursor fails clearly rather than loops. +- Reads of one pager serialize. Reader cancellation rejects that reader but + does not cancel Query's shared acquisition or discard its valid page. Use + QueryClient cancellation to stop transport; reset rejects old reads. Failed + acquisitions leave the previous cached result available and can cause Query + to rebuild the prefix when retried. These are Query's semantics, replacing + the prototype's rule that an aborted reader must discard its response. +- Query owns cache expiry, garbage collection and invalidation. Use a distinct + infinite-query key for each source/filter/order, beside the row prefix under + a shared resource prefix. Never put pages inside the row collection's prefix: + manual row writes target queries beneath it. Raw-array writes defensively skip + other cache formats; selected response formats still require separate prefixes. + For forced refresh, cancel the shared prefix before invalidating both caches; + invalidation alone can join an old in-flight append. Collection refetch alone may reuse + fresh pages. Reset removes pages, but does not refresh collection rows. +- The internal page format and full prefix are fixed: inherited `select` and + `maxPages` settings cannot change them. Explicit acquisition cancellation rejects its + waiting reads with AbortError without starting new work; a silent cancelling + refetch moves waiters to the replacement. Aborted readers release their own + queue position without cancelling transport. Cached reads + need not wait for a deeper peer acquisition. +- Rows are immutable request/cache values. Reading may return a fresh array; + cross-read object identity is not promised. + +The backend supplies a stable ordered sequence within a pagination interval. This is +not a promise of snapshot consistency from arbitrary changing HTTP endpoints. +An adapter must define its real endpoint's cursor validity/invalidation rules. + +## Oracle and responsibility boundary + +`model.ts` filters/sorts the complete independent dataset and slices it. It has +no cursor or acquisition state. Generated histories vary windows, ties, backend +page size, source scope, failure/retry and reset. Fixed products retain empty, +zero, unlimited, peer and backend/UI boundary witnesses. Fault controls must +fail value/protocol assertions, not merely fail setup or time out. + +The fake backend alone understands its token table. The production pager +must obtain rows by traversing those tokens; the backend does not remove rows +already delivered or infer missing progress for the pager. + +Layers earn separate credit: model calibration; pager against opaque transport; +real QueryClient/QueryCollection/live-window integration. The last layer must +observe production publications, not manufacture them in the fixture. + +## Work queue + +- [x] Calibrate independent reference and demonstrate a one-page reader fails. +- [x] Implement test-only pager; retain red/green evidence. +- [x] Generated windows, page repartition equivalence, scope/reset, work checks. +- [x] Held response, abort, failure/retry, healthy peer, late reset checks. +- [x] Real QueryCollection and window-controller integration; unchanged peek-ahead. +- [x] Focused tests, types, lint, stress; record exact commands and limits. + +## Chosen implementation and explicit exclusions + +- Use Query's infinite-query page cache, not a parallel TTL/registry. Loading + more fetches the missing suffix while pages are fresh. Expiry or invalidation + rebuilds the loaded sequence from the beginning. Query measures freshness + from the latest successful acquisition, including page growth. This differs + from the earlier suggestion to date a session from its oldest page. +- The fresh-pager-per-queryFn prototype was rejected: it would repeat fetching + on ordinary page growth. Creating a helper per call is now safe **only + because its stable QueryClient/key retains pages outside that helper**. +- Page queries have no lasting observer. Their inactive `gcTime` controls + retention even when the outer collection is visible. Memory can still grow + with fetched data inside that interval; no per-page size/eviction policy. +- Keep N+1. No metadata setter, no-peek optimization, metadata-only notification, + eligibility planner, new D2 graph, total-count API, previous-page API, cursor + persistence promise, backend snapshot guarantee or offset/cursor jump hybrid. +- Query-native key sharing reuses acquisitions. We do not add cross-owner + cancellation leases or replace Query's own concurrent-fetch semantics. +- The hook and core are unchanged. The maintainer chose to close #863 with this + partial implementation. Cached opaque-cursor loading ships; metadata/no-peek + is intentionally excluded, not unfinished work required to close the issue. + +## Implementation verification + +- Before promotion, request capture returned the mutated caller window and a + 150,000-row page overflowed spread arguments: two assertion/runtime failures. + The helper captures numeric fields on invocation and avoids argument spread. +- A long-lived cache without invalidation failed all four real-refetch cells, + retaining old rows. The Query-owned cache with shared-prefix invalidation + passes backend growth, shrink, empty and regrowth plus later page expansion. +- The frozen candidate passed 87 tests across five files with multiplier 100, + seed 863: 35,000 cursor histories, 10,000 cache histories and 15,000 no-peek + experiment histories. The latter are retained experiments, not shipped + metadata behavior. A loss audit and final package gates follow. + +The cache oracle uses a full source snapshot plus a freshness deadline, not a +copy of Query's cache/observer/retry state. The integration test uses the actual +QueryClient, QueryCollection, graph and window controller. The supporting +`pager.ts` only configures an isolated QueryClient for stable-snapshot laws. + +## Final implementation gates and loss audit + +[Loss audit](./LOSS-AUDIT.md): two source-isolated readers (issue and plans), +plus an explicitly correlated guide audit. Recovered documentation boundaries, +added invalidation fault calibration and QueryClient transport cancellation, +and strengthened all four real-refetch cells. No production changes were +needed after that audit. No prior experiment test was deleted. + +The subsequent prep-pr review found three real gaps. The expanded cache oracle +failed eight cells before production changed (seed 863, minimized size/depth +`[1, 1]`): global/key `maxPages`, `select`, their combination, and cancellation +with an existing prefix during growth/refresh. Initial cancellation stayed green. + +The fixes pin the internal page format and preserve acquisition rejection even +when Query's fetch API returns reverted cache data. Generated cancellation +histories hold actual response delivery at varied page depths, observe a real +peer fetch join, assert both waiters reject without replacement transport, fence +late completion, and retry. A fresh cached shallow read still completes while a +deeper acquisition is held. The reference remains full filter/sort/slice; no +Query state machine was added to it. The reviewer independently rechecked the +three fixes after implementation. + +The cache-publication review then exposed invalidation during held growth and +protocol errors after cache publication. Its oracle was red in four cells before +the fixes. The supported refresh procedure now cancels before invalidation; +response validation runs before Query can publish or resolve shared waiters. +The new five-cell suite retains an invalidation-only fault control and covers +shared readers, retries, malformed final tokens, recovery and bounded slice work. +See LOSS-AUDIT.md for the distinct generator and observation gaps. + +Verification for the acquisition-boundary review fixes on base head `3f43deeb6`: + +- Full Query DB package: **370 tests in 14 files**, default random-seed lane, + exit 0, 7.82 seconds. +- Final stress: **116 tests in seven files**, multiplier 100, seed 863, exit 0, + 73.48 seconds. 154,000 generated histories: the previous 135,000 plus + 6,000 nested cancellation/replacement, 3,000 reader abort, 5,000 manual-write + and 5,000 backend-token isolation histories. 98 tests concern the + shipping cursor helper; 18 retain the excluded experiment. +- The first stress attempt hit the ordinary five-second test timeout in the + retry-heavy refresh property, with no assertion mismatch. The final stress + command uses a 60-second timeout; the default suite limit is unchanged. +- Package TypeScript, targeted ESLint, formatting checks and Vite build pass. +- Browser ESM diagnostic import, esbuild minification, target ES2020: + `queryCollectionOptions` alone 46,590 bytes / 14,985 gzip; with the helper + 52,014 / 16,999 (+5,424 / +2,014). These boundary fixes add 334 gzip + bytes to published head `3f43deeb6`. This is an opt-in import comparison, + not a universal application bundle measurement. + +Final stress command, from `packages/query-db-collection`: + +```sh +TANSTACK_DB_ORACLE_RUNS_MULTIPLIER=100 TANSTACK_DB_ORACLE_SEED=863 \ + ../../node_modules/.bin/vitest run \ + tests/cursor-pagination.oracle.test.ts \ + tests/cursor-pagination.cache-oracle.test.ts \ + tests/cursor-pagination.publication-oracle.test.ts \ + tests/cursor-pagination.boundary-oracle.test.ts \ + tests/cursor-pagination.integration.test.ts \ + tests/cursor-pagination.no-peek.test.ts \ + tests/cursor-pagination.no-peek.integration.test.ts \ + --typecheck.enabled=false --maxWorkers=1 --testTimeout=60000 +``` + +Full-package command: `../../node_modules/.bin/vitest run --typecheck.enabled=false --maxWorkers=2`. +Types are checked separately with `tsc --noEmit -p packages/query-db-collection/tsconfig.json` +from the repository root. The installed local binaries avoid pnpm's unrelated +attempt to replace this checkout's existing node_modules. + +## Cache-guard follow-up verification + +Against base head `37dc8057c`, the manual-write oracle now crosses raw/selected +responses and raw nested/sibling page keys over generated operation histories. +It checks cache state as well as data, including empty raw-cache seeding. A +no-op cache-write mutant fails because it clears the page query's invalidation. +The refresh oracle also crosses new-per-call and retained pagers, including reads +queued behind growth. See LOSS-AUDIT.md for RED/GREEN evidence and scope. + +- Full package: **374 tests / 14 files**, exit 0, 8.98 seconds. +- Stress: **120 tests / seven files**, multiplier 100, seed 863, exit 0, + 83.62 seconds. **174,000 generated histories**: 154,000 from the preceding + boundary suite plus 10,000 additional manual-write and 10,000 retained-pager + refresh histories. The stress command above is unchanged. +- Package types, Vite build and formatting pass. ESLint reports no errors and + two pre-existing `no-shadow` warnings in unchanged parts of `query.ts`. +- Same ES2020 browser diagnostic imports: `queryCollectionOptions` alone + **46,655 minified / 15,000 gzip**; with `createCursorPager`, **52,079 / 17,017**. + The guard adds 65 minified bytes and 15/18 gzip bytes respectively over the + base head. No full-prefix refresh or automatic invalidation policy changed. + +## Historical experiment verification receipts + +Validated after merging `origin/main` at `3b991173f` into merge head `3c54e89ae`. +The core and db-ivm packages were rebuilt from that checkout for the integration +tests. The experiment remains test-only; these results do not establish the +safety of removing peek-ahead or adding a public metadata API. + +- First red: a one-backend-page reader failed four tests. Seed `1729857442` + shrank the generated failure to two rows, backend page size one, and an + unlimited read: the reader returned only the first row. +- Negative control: removing the post-response cancellation/generation check + failed both held-response abort and reset cases. The rejection control still + passed. Restoring the check restored green; these were assertion failures, + not timeouts. +- Final green: 48 tests, including 32 backend/UI page-size integration cells + and a held-response shared-consumer case. Four generated properties ran + 35,000 cases in total at multiplier 100 with seed `863`. A separate normal + random-seed run also passed. +- Package TypeScript check and lint on all changed TypeScript files passed. + +From `packages/query-db-collection`, using the installed local binaries: + +```sh +TANSTACK_DB_ORACLE_RUNS_MULTIPLIER=100 TANSTACK_DB_ORACLE_SEED=863 \ + ../../node_modules/.bin/vitest run \ + tests/cursor-pagination.oracle.test.ts \ + tests/cursor-pagination.integration.test.ts \ + --typecheck.enabled=false --maxWorkers=1 +``` + +From the repository root: + +```sh +./node_modules/.bin/tsc --noEmit -p packages/query-db-collection/tsconfig.json +./node_modules/.bin/eslint packages/query-db-collection/tests/cursor-pagination*.test.ts \ + packages/query-db-collection/tests/cursor-pagination/*.ts packages/db/tests/oracle-config.ts +``` + +## Integration limits + +The integration fixture handles its specific rank/id ordering and rank-equality +tie requests, with an independent cursor sequence for each filter. It rejects +unsupported predicates and cursor hints; it is not a generic IR interpreter. +Transport-count assertions there cover the primary prefix sequence, not the +separate tie-group sequences. The pager-level oracle checks total backend calls +within its single scope. + +The real framework window controller runs without React. Assertions compare +public user fields and window state after every ready publication; virtual row +metadata is outside this experiment's contract. + +The tests do not establish snapshot consistency for changing endpoints, actual +React/Vue/Svelte rendering, every IR expression or persistence of page caches. +The no-peek experiment uses a manually scoped bridge and remains test-only. diff --git a/packages/query-db-collection/tests/cursor-pagination/backend.ts b/packages/query-db-collection/tests/cursor-pagination/backend.ts new file mode 100644 index 0000000000..18c41b3d5d --- /dev/null +++ b/packages/query-db-collection/tests/cursor-pagination/backend.ts @@ -0,0 +1,55 @@ +import type { Row, Scope } from './model.js' + +export type Page = { rows: Array; nextCursor: string | null } +export type FetchPage = ( + cursor: string | undefined, + signal?: AbortSignal, +) => Promise> + +let nextSequence = 0 + +/** Backend token table is private to the fixture; the pager cannot decode it. */ +export function createBackend( + source: ReadonlyArray, + scope: Scope, + pageSize: number, + emptyFirst = false, +) { + const sequence = ++nextSequence + const rows: Array = [] + // Separate formulation from the model's filter/sort/slice. The backend's + // trusted job is to expose one stable relation in opaque, finite pages. + for (const row of source) { + if (scope.group !== undefined && row.group !== scope.group) continue + const position = rows.findIndex((other) => { + const before = + row.rank < other.rank || (row.rank === other.rank && row.id < other.id) + return scope.descending ? !before : before + }) + rows.splice(position < 0 ? rows.length : position, 0, { ...row }) + } + const tokens = new Map() + let serial = 0 + const calls: Array = [] + const token = (offset: number): string => { + const value = `opaque-${sequence}-${++serial}-${Math.imul(serial, 2654435761) >>> 0}` + tokens.set(value, offset) + return value + } + const fetchPage: FetchPage = (cursor, signal) => { + signal?.throwIfAborted() + calls.push(cursor) + if (emptyFirst && cursor === undefined) { + return Promise.resolve({ rows: [], nextCursor: token(0) }) + } + const offset = cursor === undefined ? 0 : tokens.get(cursor) + if (offset === undefined) throw new Error(`Foreign or invented cursor`) + const page = rows.slice(offset, offset + pageSize) + const end = offset + page.length + return Promise.resolve({ + rows: page.map((row) => ({ ...row })), + nextCursor: end < rows.length ? token(end) : null, + }) + } + return { fetchPage, calls } +} diff --git a/packages/query-db-collection/tests/cursor-pagination/model.ts b/packages/query-db-collection/tests/cursor-pagination/model.ts new file mode 100644 index 0000000000..7c33c0bdfc --- /dev/null +++ b/packages/query-db-collection/tests/cursor-pagination/model.ts @@ -0,0 +1,33 @@ +export type Row = { id: number; rank: number; group: number } +export type Scope = { group: number | undefined; descending: boolean } +export type Window = { offset: number; limit: number | undefined } + +/** + * Reference authority: an exact ordered request is a slice of the whole + * filtered relation. Backend pages, cursors, caches and promises do not enter + * this model. Numeric rank then ID is the declared total order in this domain. + */ +export function expectedRows( + source: ReadonlyArray, + scope: Scope, + window: Window, +): Array { + const direction = scope.descending ? -1 : 1 + return source + .filter((row) => scope.group === undefined || row.group === scope.group) + .sort((a, b) => direction * (a.rank - b.rank || a.id - b.id)) + .slice( + window.offset, + window.limit === undefined ? undefined : window.offset + window.limit, + ) + .map((row) => ({ ...row })) +} + +export function expectedWindow( + source: ReadonlyArray, + scope: Scope, + n: number, +) { + const all = expectedRows(source, scope, { offset: 0, limit: undefined }) + return { rows: all.slice(0, n), hasNextPage: all.length > n } +} diff --git a/packages/query-db-collection/tests/cursor-pagination/no-peek-transport.ts b/packages/query-db-collection/tests/cursor-pagination/no-peek-transport.ts new file mode 100644 index 0000000000..9639c82168 --- /dev/null +++ b/packages/query-db-collection/tests/cursor-pagination/no-peek-transport.ts @@ -0,0 +1,40 @@ +import { createBackend } from './backend.js' +import { createCursorPager } from './pager.js' +import type { Row, Scope } from './model.js' +import type { PrefixPublication } from './no-peek.js' + +/** Endpoint fixture derives continuation from opaque pages and retained tails, + * never from the full-relation oracle or the window controller. */ +export function createFactTransport( + rows: Array, + size: number, + scope: Scope, +) { + const backend = createBackend(rows, scope, size) + let fetched = 0 + let remaining = true + const pager = createCursorPager(async (cursor, signal) => { + const page = await backend.fetchPage(cursor, signal) + fetched += page.rows.length + remaining = page.nextCursor !== null + return page + }) + const requests: Array = [] + const read = async (limit: number): Promise => { + requests.push(limit) + const result = await pager.read({ offset: 0, limit }) + const stamp = {} + return { + rows: result, + requested: limit, + stamp, + transparent: true, + fact: { + stamp, + end: result.length, + hasMore: fetched > result.length || remaining, + }, + } + } + return { read, requests, backend } +} diff --git a/packages/query-db-collection/tests/cursor-pagination/no-peek.ts b/packages/query-db-collection/tests/cursor-pagination/no-peek.ts new file mode 100644 index 0000000000..905b49abad --- /dev/null +++ b/packages/query-db-collection/tests/cursor-pagination/no-peek.ts @@ -0,0 +1,119 @@ +import type { Row } from './model.js' + +/** Proposed test-only boundary: this packet describes a complete public prefix, + * not a transport response or merely applied source rows. */ +export type PrefixPublication = { + rows: Array + requested: number + stamp: object + transparent: boolean + fact?: { stamp: object; end: number; hasMore: boolean } +} + +export type WindowSnapshot = { rows: Array; hasNextPage: boolean } + +function continuation( + packet: PrefixPublication, + count: number, +): boolean | undefined { + // A deeper peer's terminal fact cannot erase this consumer's actual row N+1. + if (packet.rows.length > count) return true + if (packet.requested > count) return false + const fact = packet.fact + if ( + packet.transparent && + fact?.stamp === packet.stamp && + fact.end === packet.rows.length && + (fact.end === count || !fact.hasMore) + ) + return fact.hasMore + return undefined +} + +/** Test-only coordinator, not a replacement for the production window controller. + * Acquire must settle at public-window completion. No source-lifecycle machinery + * is duplicated here; the missing production bridge remains explicit. */ +export function createNoPeekSession( + acquire: (limit: number) => Promise, + eligible = true, +) { + const leases = new Map() + const snapshots = new Map() + const listeners = new Set<() => void>() + let generation = 0 + let tail = Promise.resolve() + const desired = () => + Math.max( + ...[...leases.values()].map(({ count, peek }) => count + Number(peek)), + ) + return { + request(id: string, limit: number) { + if ( + !Number.isSafeInteger(limit) || + limit <= 0 || + limit >= Number.MAX_SAFE_INTEGER + ) { + throw new RangeError(`Expected a positive safe window`) + } + leases.set(id, { count: limit, peek: leases.get(id)?.peek ?? !eligible }) + }, + release(id: string) { + leases.delete(id) + snapshots.delete(id) + }, + reset() { + generation++ + for (const lease of leases.values()) lease.peek = !eligible + snapshots.clear() + }, + subscribe(listener: () => void) { + listeners.add(listener) + return () => listeners.delete(listener) + }, + get(id: string) { + return snapshots.get(id) + }, + refresh() { + const requestedGeneration = generation + const checkCurrent = () => { + if (requestedGeneration !== generation) + throw new DOMException(`Publication was reset`, `AbortError`) + } + const result = tail.then(async () => { + checkCurrent() + while (leases.size > 0) { + const limit = desired() + const publication = await acquire(limit) + checkCurrent() + if ( + publication.requested !== limit || + publication.rows.length > limit + ) { + throw new Error(`Publication does not match acquired prefix`) + } + const next = new Map() + for (const [id, lease] of leases) { + const more = continuation(publication, lease.count) + if (more === undefined) lease.peek = true + else + next.set(id, { + rows: publication.rows.slice(0, lease.count), + hasNextPage: more, + }) + } + // Keep fallback on each consumer's lease, including after deeper peers + // release. A snapshot cache must also be invalidated by fact-only changes. + if (next.size !== leases.size) continue + for (const [id, snapshot] of next) snapshots.set(id, snapshot) + for (const listener of listeners) listener() + return + } + }) + tail = result.then( + () => {}, + () => {}, + ) + return result + }, + } +} diff --git a/packages/query-db-collection/tests/cursor-pagination/pager.ts b/packages/query-db-collection/tests/cursor-pagination/pager.ts new file mode 100644 index 0000000000..6b5d4b57f2 --- /dev/null +++ b/packages/query-db-collection/tests/cursor-pagination/pager.ts @@ -0,0 +1,16 @@ +import { QueryClient } from '@tanstack/query-core' +import { createCursorPager as createProductionPager } from '../../src/index.js' +import type { FetchPage } from './backend.js' + +/** Stable-snapshot tests use isolated caches without timers or expiry. */ +export function createCursorPager(fetchPage: FetchPage) { + return createProductionPager({ + queryClient: new QueryClient({ + defaultOptions: { queries: { retry: false } }, + }), + queryKey: [`cursor-pages`], + staleTime: Infinity, + gcTime: Infinity, + fetchPage, + }) +}