From 8ab70ea4cf6f9e07a04d104f94f0e0aa2be00b64 Mon Sep 17 00:00:00 2001 From: Kyle Mathews Date: Wed, 12 Aug 2026 13:48:39 -0600 Subject: [PATCH] test(db): add controlled includes oracle drivers --- packages/db/tests/expected-failure.test.ts | 66 ++ packages/db/tests/expected-failure.ts | 29 + .../query/includes-oracle.property.test.ts | 547 +++++++++----- .../query/includes-temporal-oracle.test.ts | 666 ++++++++++++++++++ packages/db/tests/trace-runner.test.ts | 58 +- packages/db/tests/trace-runner.ts | 27 +- 6 files changed, 1192 insertions(+), 201 deletions(-) create mode 100644 packages/db/tests/expected-failure.test.ts create mode 100644 packages/db/tests/expected-failure.ts create mode 100644 packages/db/tests/query/includes-temporal-oracle.test.ts diff --git a/packages/db/tests/expected-failure.test.ts b/packages/db/tests/expected-failure.test.ts new file mode 100644 index 000000000..4bb2f3cfb --- /dev/null +++ b/packages/db/tests/expected-failure.test.ts @@ -0,0 +1,66 @@ +import { describe, expect, it } from 'vitest' +import { expectAssertionFailure } from './expected-failure.js' +import { TraceAssertionError } from './trace-runner.js' + +describe(`expected failure guard`, () => { + it(`accepts an assertion mismatch at the expected checkpoint`, async () => { + const guarded = expectAssertionFailure( + () => { + try { + expect(`observed`).toBe(`expected`) + return Promise.resolve() + } catch (error) { + return Promise.reject(new TraceAssertionError(2, error)) + } + }, + { checkpoint: 2 }, + ) + + await guarded() + }) + + it(`rejects an assertion mismatch from the wrong checkpoint`, async () => { + const guarded = expectAssertionFailure( + () => + Promise.reject( + new TraceAssertionError(0, new Error(`startup mismatch`)), + ), + { checkpoint: 2 }, + ) + + await expect(guarded()).rejects.toBeInstanceOf(Error) + }) + + it(`rejects a runtime error from the expected checkpoint`, async () => { + const guarded = expectAssertionFailure( + () => + Promise.reject( + new TraceAssertionError(2, new TypeError(`projection failed`)), + ), + { checkpoint: 2 }, + ) + + await expect(guarded()).rejects.toBeInstanceOf(Error) + }) + + it(`accepts an assertion mismatch with the expected message`, async () => { + const guarded = expectAssertionFailure( + () => + Promise.resolve().then(() => { + expect([`actual`]).toEqual([`expected`]) + }), + { message: /expected/ }, + ) + + await guarded() + }) + + it(`rejects runtime errors that happen to have the expected message`, async () => { + const runtimeError = new TypeError(`expected value is missing`) + const guarded = expectAssertionFailure(() => Promise.reject(runtimeError), { + message: /expected/, + }) + + await expect(guarded()).rejects.toBeInstanceOf(Error) + }) +}) diff --git a/packages/db/tests/expected-failure.ts b/packages/db/tests/expected-failure.ts new file mode 100644 index 000000000..8cf70dc22 --- /dev/null +++ b/packages/db/tests/expected-failure.ts @@ -0,0 +1,29 @@ +import { expect } from 'vitest' + +type ExpectedAssertionFailure = + | { checkpoint: number } + | { message: string | RegExp } + +export function expectAssertionFailure>( + assertion: (...args: TArgs) => Promise, + expected: ExpectedAssertionFailure, +): (...args: TArgs) => Promise { + return async (...args) => { + if (`checkpoint` in expected) { + await expect(assertion(...args)).rejects.toMatchObject({ + name: `TraceAssertionError`, + checkpoint: expected.checkpoint, + cause: { name: `AssertionError` }, + }) + return + } + + await expect(assertion(...args)).rejects.toMatchObject({ + name: `AssertionError`, + message: + typeof expected.message === `string` + ? expected.message + : expect.stringMatching(expected.message), + }) + } +} diff --git a/packages/db/tests/query/includes-oracle.property.test.ts b/packages/db/tests/query/includes-oracle.property.test.ts index 6c6bd3777..b0deb00c8 100644 --- a/packages/db/tests/query/includes-oracle.property.test.ts +++ b/packages/db/tests/query/includes-oracle.property.test.ts @@ -14,6 +14,7 @@ import { mockSyncCollectionOptions, withExpectedRejection, } from '../utils.js' +import { expectAssertionFailure } from '../expected-failure.js' import { runTrace } from '../trace-runner.js' import type { TraceCheckpoint, @@ -113,20 +114,31 @@ function levelArbitrary( } function actionArbitrary(depth: IncludeDepth): fc.Arbitrary { - return fc.record({ - type: fc.constantFrom( - `put`, - `delete`, - `optimisticConfirm`, - `optimisticRollback`, - ), - level: levelArbitrary(depth), - id: fc.integer({ min: 0, max: 5 }), - parentGroup: fc.integer({ min: 0, max: 2 }), - group: fc.integer({ min: 0, max: 2 }), - value: fc.integer({ min: -3, max: 3 }), - position: fc.integer({ min: -2, max: 2 }), - }) + return levelArbitrary(depth).chain((level) => + fc.record({ + // Root delete/reinsert has a deterministic expected-failure trace below. + // Keep the green fuzz corpus from rediscovering the same defect class. + type: + level === 0 + ? fc.constantFrom( + `put` as const, + `optimisticConfirm` as const, + `optimisticRollback` as const, + ) + : fc.constantFrom( + `put` as const, + `delete` as const, + `optimisticConfirm` as const, + `optimisticRollback` as const, + ), + level: fc.constant(level), + id: fc.integer({ min: 0, max: 5 }), + parentGroup: fc.integer({ min: 0, max: 2 }), + group: fc.integer({ min: 0, max: 2 }), + value: fc.integer({ min: -3, max: 3 }), + position: fc.integer({ min: -2, max: 2 }), + }), + ) } function ensureActionsTargetRows( @@ -611,16 +623,6 @@ async function settleOptimisticAction( }) } -function expectAssertionFailure>( - assertion: (...args: TArgs) => Promise, -): (...args: TArgs) => Promise { - return async (...args) => { - await expect(assertion(...args)).rejects.toMatchObject({ - name: `AssertionError`, - }) - } -} - async function applyAction( action: HistoryAction, sources: Sources, @@ -999,6 +1001,36 @@ function createConnectedBatchPrefix( return steps } +function createConnectedBatchBranches( + depth: IncludeDepth, +): Array { + const branchRoots = [100, 200] + const steps: Array = [ + { + level: 0, + changes: branchRoots.map((id) => ({ + type: `insert`, + value: batchRoot(id, id, id, 0), + })), + }, + ] + + for (let level = 1; level <= depth; level++) { + steps.push({ + level: level as IncludeDepth, + changes: branchRoots.map((rootId) => ({ + type: `insert`, + value: { + ...batchChild(rootId + level, rootId + level - 1, rootId + level, 0), + group: rootId + level, + }, + })), + }) + } + + return steps +} + function normalizeFullRowBatchInputs( depth: IncludeDepth, inputs: Array, @@ -1076,25 +1108,75 @@ function normalizeFullRowBatchInputs( return { depth, steps } } -function fullRowBatchScenarioArbitrary( - allowChildRelationshipUpdates: boolean, - levels: `all` | `children` = `all`, - maxBatchSize = 3, +function fullRowBatchScenarioAtDepthArbitrary( + depth: IncludeDepth, ): fc.Arbitrary { - return depthArbitrary.chain((depth) => - fc - .array(fullRowBatchInputArbitrary(depth, levels, maxBatchSize), { - minLength: 1, - maxLength: 10, - }) - .map((inputs) => - normalizeFullRowBatchInputs( - depth, - inputs, - allowChildRelationshipUpdates, - ), - ), - ) + return fc + .array(fullRowBatchInputArbitrary(depth, `all`), { + minLength: 1, + maxLength: 10, + }) + .map((inputs) => { + const noise = normalizeFullRowBatchInputs( + depth, + inputs, + false, + ).steps.slice(depth + 1) + const changes: Array> = [100, 200].map((rootId) => ({ + type: `update`, + value: { + ...batchChild( + rootId + depth, + rootId + depth - 1, + rootId + depth + 1, + 0, + ), + group: rootId + depth, + }, + })) + + return { + depth, + steps: [ + ...createConnectedBatchBranches(depth), + ...noise, + { level: depth, changes }, + ], + } + }) +} + +type VisibleRelationshipTransition = `reparent` | `rekey` + +function visibleRelationshipScenarioArbitrary( + depth: IncludeDepth, + transition: VisibleRelationshipTransition, +): fc.Arbitrary { + return fc + .array(fullRowBatchInputArbitrary(depth, `children`, 1), { + minLength: 1, + maxLength: 10, + }) + .map((inputs) => { + const noise = normalizeFullRowBatchInputs( + depth, + inputs, + true, + ).steps.slice(depth + 1) + const value: ChildRow = { + ...batchChild(101, transition === `reparent` ? 200 : 100, 101, 0), + group: transition === `rekey` ? 150 : 101, + } + + return { + depth, + steps: [ + ...createConnectedBatchBranches(depth), + ...noise, + { level: 1, changes: [{ type: `update`, value }] }, + ], + } + }) } async function expectFullRowBatchScenarioMatches({ @@ -1108,6 +1190,24 @@ async function expectFullRowBatchScenarioMatches({ }) } +function recomputeFullRowBatchScenario( + { depth, steps }: FullRowBatchScenario, + stepCount: number, +): Array { + const roots = new Map() + const levels = Array.from({ length: 4 }, () => new Map()) + + for (const step of steps.slice(0, stepCount)) { + if (step.level === 0) { + updateModel(roots, step.changes) + } else { + updateModel(levels[step.level - 1]!, step.changes) + } + } + + return recompute(roots, levels, depth) +} + type FlatMaterialization = `array` | `concat` function createFlatMaterializationQuery( @@ -1554,15 +1654,38 @@ const intraBatchChildHandOffScenario: FullRowBatchScenario = { } describe(`includes recompute oracle`, () => { + fcTest(`covers a visible relationship transition at every depth`, () => { + const scenarios = ([1, 2, 3, 4] as const).map( + (depth) => + fc.sample(visibleRelationshipScenarioArbitrary(depth, `reparent`), { + numRuns: 1, + seed: 1721 + depth, + })[0]!, + ) + + for (const scenario of scenarios) { + const beforeTransition = recomputeFullRowBatchScenario( + scenario, + scenario.steps.length - 1, + ) + expect( + recomputeFullRowBatchScenario(scenario, scenario.steps.length), + ).not.toEqual(beforeTransition) + } + }) + for (const materialization of [`array`, `concat`] as const) { fcTest( `discovered trace: ${materialization} follows an intra-batch child hand-off`, - expectAssertionFailure(async () => { - await expectFlatMaterializationScenarioMatches( - materialization, - intraBatchChildHandOffScenario, - ) - }), + expectAssertionFailure( + async () => { + await expectFlatMaterializationScenarioMatches( + materialization, + intraBatchChildHandOffScenario, + ) + }, + { checkpoint: 3 }, + ), ) } @@ -1579,37 +1702,67 @@ describe(`includes recompute oracle`, () => { expectFlatMaterializationScenarioMatches(kind, scenario), ) - fcTest.prop([fullRowBatchScenarioArbitrary(false)], { - numRuns: 40, - seed: 1719, - })( - `matches recomputation for generated full-row batches at every depth`, - expectFullRowBatchScenarioMatches, - ) + for (const depth of [1, 2, 3, 4] as const) { + fcTest.prop([fullRowBatchScenarioAtDepthArbitrary(depth)], { + numRuns: 10, + seed: 1719 + depth, + })( + `matches recomputation for visible multi-row batches at depth ${depth}`, + expectFullRowBatchScenarioMatches, + ) - fcTest.prop([fullRowBatchScenarioArbitrary(true, `children`, 1)], { - numRuns: 40, - seed: 1721, - })( - `matches recomputation for single-row child reparenting and rekeying`, - expectFullRowBatchScenarioMatches, - ) + const transitions: Array = + depth === 1 ? [`reparent`] : [`reparent`, `rekey`] + for (const transition of transitions) { + fcTest.prop([visibleRelationshipScenarioArbitrary(depth, transition)], { + numRuns: 6, + seed: 1721 + depth, + })( + transition === `rekey` && depth >= 3 + ? `discovered trace: a visible rekey at depth ${depth}` + : `matches recomputation for a visible ${transition} at depth ${depth}`, + async (scenario) => { + const beforeTransition = recomputeFullRowBatchScenario( + scenario, + scenario.steps.length - 1, + ) + const result = recomputeFullRowBatchScenario( + scenario, + scenario.steps.length, + ) + + expect(result).not.toEqual(beforeTransition) + if (transition === `rekey` && depth >= 3) { + await expectAssertionFailure( + () => expectFullRowBatchScenarioMatches(scenario), + { checkpoint: scenario.steps.length }, + )() + } else { + await expectFullRowBatchScenarioMatches(scenario) + } + }, + ) + } + } fcTest( `discovered seed: nested scalar materialization follows a reference update`, - expectAssertionFailure(async () => { - await runTrace({ - steps: [ - { type: `insert`, insert: `root-1` }, - { type: `insert`, insert: `middle-1` }, - { type: `insert`, insert: `shared-1` }, - { type: `insert`, insert: `leaf-1` }, - { type: `redirectMiddle`, id: 1, sharedId: 2 }, - ], - driver: createMaterializeTraceDriver(false), - projection: materializeProjection, - }) - }), + expectAssertionFailure( + async () => { + await runTrace({ + steps: [ + { type: `insert`, insert: `root-1` }, + { type: `insert`, insert: `middle-1` }, + { type: `insert`, insert: `shared-1` }, + { type: `insert`, insert: `leaf-1` }, + { type: `redirectMiddle`, id: 1, sharedId: 2 }, + ], + driver: createMaterializeTraceDriver(false), + projection: materializeProjection, + }) + }, + { checkpoint: 5 }, + ), ) fcTest(`matches recomputation for full-row sync batches`, async () => { @@ -1622,8 +1775,9 @@ describe(`includes recompute oracle`, () => { fcTest( `discovered seed: a reinserted parent drops its old shared route`, - expectAssertionFailure(() => - expectFullRowBatchScenarioMatches(fullRowSharedRoutingSeed), + expectAssertionFailure( + () => expectFullRowBatchScenarioMatches(fullRowSharedRoutingSeed), + { checkpoint: 4 }, ), ) @@ -1903,7 +2057,7 @@ describe(`includes recompute oracle`, () => { seed: 1685, })( `known seed: shared scalar materialization preserves the deepest row`, - expectAssertionFailure(expectMaterializeScenarioMatches), + expectAssertionFailure(expectMaterializeScenarioMatches, { checkpoint: 6 }), ) fcTest.prop([fc.constant(`correlation-key-update`)], { @@ -1911,130 +2065,139 @@ describe(`includes recompute oracle`, () => { seed: 1658, })( `discovered seed: parent correlation-key update rematerializes children`, - expectAssertionFailure(async () => { - const roots = createControlledCollection( - `correlation-seed-roots`, - ) - const children = createControlledCollection( - `correlation-seed-children`, - ) - const live = createLiveQueryCollection((q) => - q.from({ root: roots.collection }).select(({ root }) => ({ - id: root.id, - group: root.group, - children: toArray( - q - .from({ child: children.collection }) - .where(({ child }) => eq(child.parentGroup, root.group)) - .select(({ child }) => ({ id: child.id })), - ), - })), - ) - - try { - await live.preload() - children.write(`insert`, { - id: 1, - parentGroup: 0, - group: 0, - value: 0, - position: 0, - }) - children.write(`insert`, { - id: 2, - parentGroup: 1, - group: 0, - value: 0, - position: 0, - }) - roots.write(`insert`, { id: 1, group: 1, value: 0, position: 0 }) - roots.write(`update`, { id: 1, group: 0, value: 0, position: 0 }) - - expect(stripVirtualProperties(live.toArray)).toEqual([ - { id: 1, group: 0, children: [{ id: 1 }] }, - ]) - } finally { - await live.cleanup() - await Promise.all([ - roots.collection.cleanup(), - children.collection.cleanup(), - ]) - } - }), - ) - - fcTest.prop([fc.constant(`#1454`)], { numRuns: 1, seed: 1454 })( - `known seed: alpha-renaming a duplicate sibling alias preserves results`, - expectAssertionFailure(async () => { - const roots = createControlledCollection(`alias-seed-roots`, [ - { id: 1, group: 1, value: 0, position: 0 }, - ]) - const issues = createControlledCollection(`alias-seed-issues`, [ - { - id: 10, - parentGroup: 1, - group: 10, - value: 10, - position: 0, - }, - ]) - const tags = createControlledCollection(`alias-seed-tags`, [ - { - id: 20, - parentGroup: 1, - group: 20, - value: 20, - position: 0, - }, - ]) - - try { - const uniqueAliases = await queryOnce((q) => - q.from({ root: roots.collection }).select(({ root }) => ({ - id: root.id, - issues: toArray( - q - .from({ issue: issues.collection }) - .where(({ issue }) => eq(issue.parentGroup, root.group)) - .select(({ issue }) => ({ id: issue.id })), - ), - tags: toArray( - q - .from({ tag: tags.collection }) - .where(({ tag }) => eq(tag.parentGroup, root.group)) - .select(({ tag }) => ({ id: tag.id })), - ), - })), + expectAssertionFailure( + async () => { + const roots = createControlledCollection( + `correlation-seed-roots`, + ) + const children = createControlledCollection( + `correlation-seed-children`, ) - const duplicateAliases = await queryOnce((q) => + const live = createLiveQueryCollection((q) => q.from({ root: roots.collection }).select(({ root }) => ({ id: root.id, - issues: toArray( - q - .from({ item: issues.collection }) - .where(({ item }) => eq(item.parentGroup, root.group)) - .select(({ item }) => ({ id: item.id })), - ), - tags: toArray( + group: root.group, + children: toArray( q - .from({ item: tags.collection }) - .where(({ item }) => eq(item.parentGroup, root.group)) - .select(({ item }) => ({ id: item.id })), + .from({ child: children.collection }) + .where(({ child }) => eq(child.parentGroup, root.group)) + .select(({ child }) => ({ id: child.id })), ), })), ) - expect(stripVirtualProperties(duplicateAliases)).toEqual( - stripVirtualProperties(uniqueAliases), + try { + await live.preload() + children.write(`insert`, { + id: 1, + parentGroup: 0, + group: 0, + value: 0, + position: 0, + }) + children.write(`insert`, { + id: 2, + parentGroup: 1, + group: 0, + value: 0, + position: 0, + }) + roots.write(`insert`, { id: 1, group: 1, value: 0, position: 0 }) + roots.write(`update`, { id: 1, group: 0, value: 0, position: 0 }) + + expect(stripVirtualProperties(live.toArray)).toEqual([ + { id: 1, group: 0, children: [{ id: 1 }] }, + ]) + } finally { + await live.cleanup() + await Promise.all([ + roots.collection.cleanup(), + children.collection.cleanup(), + ]) + } + }, + { message: /children/ }, + ), + ) + + fcTest.prop([fc.constant(`#1454`)], { numRuns: 1, seed: 1454 })( + `known seed: alpha-renaming a duplicate sibling alias preserves results`, + expectAssertionFailure( + async () => { + const roots = createControlledCollection(`alias-seed-roots`, [ + { id: 1, group: 1, value: 0, position: 0 }, + ]) + const issues = createControlledCollection( + `alias-seed-issues`, + [ + { + id: 10, + parentGroup: 1, + group: 10, + value: 10, + position: 0, + }, + ], ) - } finally { - await Promise.all([ - roots.collection.cleanup(), - issues.collection.cleanup(), - tags.collection.cleanup(), + const tags = createControlledCollection(`alias-seed-tags`, [ + { + id: 20, + parentGroup: 1, + group: 20, + value: 20, + position: 0, + }, ]) - } - }), + + try { + const uniqueAliases = await queryOnce((q) => + q.from({ root: roots.collection }).select(({ root }) => ({ + id: root.id, + issues: toArray( + q + .from({ issue: issues.collection }) + .where(({ issue }) => eq(issue.parentGroup, root.group)) + .select(({ issue }) => ({ id: issue.id })), + ), + tags: toArray( + q + .from({ tag: tags.collection }) + .where(({ tag }) => eq(tag.parentGroup, root.group)) + .select(({ tag }) => ({ id: tag.id })), + ), + })), + ) + const duplicateAliases = await queryOnce((q) => + q.from({ root: roots.collection }).select(({ root }) => ({ + id: root.id, + issues: toArray( + q + .from({ item: issues.collection }) + .where(({ item }) => eq(item.parentGroup, root.group)) + .select(({ item }) => ({ id: item.id })), + ), + tags: toArray( + q + .from({ item: tags.collection }) + .where(({ item }) => eq(item.parentGroup, root.group)) + .select(({ item }) => ({ id: item.id })), + ), + })), + ) + + expect(stripVirtualProperties(duplicateAliases)).toEqual( + stripVirtualProperties(uniqueAliases), + ) + } finally { + await Promise.all([ + roots.collection.cleanup(), + issues.collection.cleanup(), + tags.collection.cleanup(), + ]) + } + }, + { message: /deeply equal/ }, + ), ) fcTest.prop([fc.constant(`#1444`)], { numRuns: 1, seed: 1444 })( diff --git a/packages/db/tests/query/includes-temporal-oracle.test.ts b/packages/db/tests/query/includes-temporal-oracle.test.ts new file mode 100644 index 000000000..b3cd0a3bf --- /dev/null +++ b/packages/db/tests/query/includes-temporal-oracle.test.ts @@ -0,0 +1,666 @@ +import { describe, expect, it } from 'vitest' +import { createCollection } from '../../src/collection/index.js' +import { createDeferred } from '../../src/deferred.js' +import { extractSimpleComparisons } from '../../src/query/expression-helpers.js' +import { + createLiveQueryCollection, + eq, + toArray, +} from '../../src/query/index.js' +import { expectAssertionFailure } from '../expected-failure.js' +import { runTrace } from '../trace-runner.js' +import type { Collection } from '../../src/collection/index.js' +import type { Deferred } from '../../src/deferred.js' +import type { LoadSubsetOptions } from '../../src/types.js' +import type { TraceDriver, TraceProjection } from '../trace-runner.js' + +type Post = { + id: number + authorId: string + title: string +} + +type Comment = { + id: number + postId: number + body: string +} + +type User = { + id: number + name: string +} + +type ProgressivePost = { + id: number + userId: number + title: string +} + +let collectionId = 0 + +function nextCollectionId(prefix: string): string { + collectionId += 1 + return `${prefix}-${collectionId}` +} + +type PreloadState = { + preloadFailure?: { error: unknown } + preloadOutcome?: Promise + preloadSettled: boolean +} + +function startPreload( + live: ReturnType, + state: PreloadState, +): Promise { + const preload = live.preload() + state.preloadOutcome = preload.then( + () => { + state.preloadSettled = true + }, + (error) => { + state.preloadFailure = { error } + state.preloadSettled = true + }, + ) + return preload +} + +async function finishPreload(state: PreloadState): Promise { + await state.preloadOutcome + if (state.preloadFailure) throw state.preloadFailure.error +} + +function correlationKeys( + loads: ReadonlyArray, + field: string, +): Array { + return [ + ...new Set( + loads.flatMap((load) => + extractSimpleComparisons(load.where).flatMap((filter) => { + if (filter.field[0] !== field) return [] + if (filter.operator === `eq` && typeof filter.value === `number`) { + return [filter.value] + } + if (filter.operator !== `in` || !Array.isArray(filter.value)) { + return [] + } + return filter.value.filter( + (value): value is number => typeof value === `number`, + ) + }), + ), + ), + ].sort((left, right) => left - right) +} + +function createColdPosts(initial: ReadonlyArray): { + collection: Collection + loaded: Deferred +} { + const loaded = createDeferred() + const collection = createCollection({ + id: nextCollectionId(`temporal-posts`), + getKey: (post) => post.id, + syncMode: `on-demand`, + sync: { + sync: ({ begin, write, commit, markReady }) => ({ + loadSubset: () => { + begin() + for (const post of initial) { + write({ type: `insert`, value: post }) + } + commit() + markReady() + loaded.resolve() + return Promise.resolve() + }, + }), + }, + }) + return { collection, loaded } +} + +function createColdComments(): { + collection: Collection + loads: Array +} { + const loads: Array = [] + const comments: Array = [ + { id: 100, postId: 1, body: `one` }, + { id: 200, postId: 2, body: `two` }, + ] + const collection = createCollection({ + id: nextCollectionId(`temporal-comments`), + getKey: (comment) => comment.id, + syncMode: `on-demand`, + sync: { + sync: ({ begin, write, commit, markReady }) => ({ + loadSubset: (options) => { + loads.push(options) + const requested = new Set(correlationKeys([options], `postId`)) + begin() + for (const comment of comments) { + if (requested.has(comment.postId)) { + write({ type: `insert`, value: comment }) + } + } + commit() + markReady() + return Promise.resolve() + }, + }), + }, + }) + return { collection, loads } +} + +type ReadinessObservation = { + ready: boolean + preloadSettled: boolean + rowCount: number + childLoadCount: number + loadedPostIds: Array +} + +type ReadinessContext = { + posts: Collection + comments: Collection + live: ReturnType + loads: Array + preload: PreloadState + parentLoaded: Deferred + expected: ReadinessObservation +} + +function createReadinessDriver( + initialPosts: ReadonlyArray, +): TraceDriver { + return { + setup: () => { + const { collection: postCollection, loaded: parentLoaded } = + createColdPosts(initialPosts) + const { collection: comments, loads } = createColdComments() + const live = createLiveQueryCollection((q) => + q + .from({ post: postCollection }) + .where(({ post }) => eq(post.authorId, `selected`)) + .select(({ post }) => ({ + id: post.id, + comments: toArray( + q + .from({ comment: comments }) + .where(({ comment }) => eq(comment.postId, post.id)), + ), + })), + ) + + return { + posts: postCollection, + comments, + live, + loads, + preload: { preloadSettled: false }, + parentLoaded, + expected: { + ready: true, + preloadSettled: true, + rowCount: initialPosts.length, + childLoadCount: initialPosts.length === 0 ? 0 : 1, + loadedPostIds: initialPosts.map(({ id }) => id), + }, + } + }, + start: async (context) => { + const preload = startPreload(context.live, context.preload) + await context.parentLoaded.promise + if (initialPosts.length > 0) await preload + }, + apply: () => undefined, + cleanup: async ({ posts, comments, live, preload }) => { + await live.cleanup() + await finishPreload(preload) + await Promise.all([posts.cleanup(), comments.cleanup()]) + }, + } +} + +const readinessProjection: TraceProjection< + ReadinessContext, + ReadinessObservation +> = { + observe: ({ live, loads, preload }) => ({ + ready: live.isReady(), + preloadSettled: preload.preloadSettled, + rowCount: live.size, + childLoadCount: loads.length, + loadedPostIds: correlationKeys(loads, `postId`), + }), + recompute: ({ expected }) => expected, + assertEqual: (observed, expected) => { + expect(observed).toEqual(expected) + return undefined + }, +} + +async function expectReadinessMatches( + posts: ReadonlyArray, +): Promise { + await runTrace({ + steps: [], + driver: createReadinessDriver(posts), + projection: readinessProjection, + }) +} + +type DemandCancellationObservation = { + ready: boolean + rowCount: number + childLoadStarted: boolean + childLoadPending: boolean +} + +type DemandCancellationContext = { + posts: Collection + comments: Collection + live: ReturnType + removePost: () => void + childLoad: ReturnType> + childLoadStarted: Deferred + preload: PreloadState + expected: DemandCancellationObservation +} + +function createRemovablePost(): { + collection: Collection + remove: () => void +} { + const post: Post = { + id: 1, + authorId: `selected`, + title: `selected`, + } + let remove: () => void = () => { + throw new Error(`Post collection has not started`) + } + const collection = createCollection({ + id: nextCollectionId(`temporal-removable-post`), + getKey: (row) => row.id, + sync: { + sync: ({ begin, write, commit, markReady }) => { + begin() + write({ type: `insert`, value: post }) + commit() + markReady() + remove = () => { + begin() + write({ type: `delete`, value: post }) + commit() + } + }, + }, + }) + return { collection, remove: () => remove() } +} + +function createDemandCancellationDriver(): TraceDriver< + `remove-parent`, + DemandCancellationContext +> { + return { + setup: () => { + const { collection: posts, remove } = createRemovablePost() + const childLoad = createDeferred() + const childLoadStarted = createDeferred() + const comments = createCollection({ + id: nextCollectionId(`temporal-pending-comments`), + getKey: (comment) => comment.id, + syncMode: `on-demand`, + sync: { + sync: () => ({ + loadSubset: () => { + childLoadStarted.resolve() + return childLoad.promise + }, + }), + }, + }) + const live = createLiveQueryCollection((q) => + q.from({ post: posts }).select(({ post }) => ({ + id: post.id, + comments: toArray( + q + .from({ comment: comments }) + .where(({ comment }) => eq(comment.postId, post.id)), + ), + })), + ) + return { + posts, + comments, + live, + removePost: remove, + childLoad, + childLoadStarted, + preload: { preloadSettled: false }, + expected: { + ready: false, + rowCount: 1, + childLoadStarted: true, + childLoadPending: true, + }, + } + }, + start: async (context) => { + startPreload(context.live, context.preload) + await context.childLoadStarted.promise + }, + apply: (_step, context) => { + context.removePost() + context.expected = { + ready: true, + rowCount: 0, + childLoadStarted: true, + childLoadPending: true, + } + }, + cleanup: async ({ posts, comments, live, childLoad, preload }) => { + childLoad.resolve() + await live.cleanup() + await finishPreload(preload) + await Promise.all([posts.cleanup(), comments.cleanup()]) + }, + } +} + +const demandCancellationProjection: TraceProjection< + DemandCancellationContext, + DemandCancellationObservation +> = { + observe: ({ live, childLoadStarted, childLoad }) => ({ + ready: live.isReady(), + rowCount: live.size, + childLoadStarted: !childLoadStarted.isPending(), + childLoadPending: childLoad.isPending(), + }), + recompute: ({ expected }) => expected, + assertEqual: (observed, expected) => { + expect(observed).toEqual(expected) + return undefined + }, +} + +async function expectObsoleteDemandDoesNotBlockReadiness(): Promise { + await runTrace({ + steps: [`remove-parent`], + driver: createDemandCancellationDriver(), + projection: demandCancellationProjection, + }) +} + +type FastPathEvent = { + phase: `fast` | `late` + keys: Array +} + +type ProgressiveObservation = { + events: Array + ready: boolean + preloadSettled: boolean +} + +type ProgressiveStep = `release-parent` + +type ProgressiveContext = { + users: Collection | undefined + posts: Collection + live: ReturnType + events: Array + closeWindow: () => void + releaseParent: (() => void) | undefined + startReached: Deferred + parentDelivery: Promise | undefined + preload: PreloadState + expected: ProgressiveObservation +} + +function createProgressivePosts(): { + collection: Collection + events: Array + closeWindow: () => void + syncStarted: Deferred +} { + let windowOpen = true + const events: Array = [] + const syncStarted = createDeferred() + const collection = createCollection({ + id: nextCollectionId(`temporal-progressive-posts`), + getKey: (post) => post.id, + syncMode: `on-demand`, + sync: { + sync: ({ begin, commit, markReady }) => { + syncStarted.resolve() + begin() + commit() + markReady() + return { + loadSubset: (options) => { + events.push({ + phase: windowOpen ? `fast` : `late`, + keys: correlationKeys([options], `userId`), + }) + return Promise.resolve() + }, + } + }, + }, + }) + return { + collection, + events, + syncStarted, + closeWindow: () => { + windowOpen = false + }, + } +} + +function createGatedUsers(): { + collection: Collection + release: () => void + started: Deferred + delivery: Promise +} { + const gate = createDeferred() + const started = createDeferred() + const delivery = createDeferred() + const collection = createCollection({ + id: nextCollectionId(`temporal-users`), + getKey: (user) => user.id, + sync: { + sync: ({ begin, write, commit, markReady }) => { + started.resolve() + gate.promise.then( + () => { + begin() + write({ type: `insert`, value: { id: 2, name: `selected` } }) + commit() + markReady() + delivery.resolve() + }, + (error) => delivery.reject(error), + ) + }, + }, + }) + return { + collection, + release: () => gate.resolve(), + started, + delivery: delivery.promise, + } +} + +function createProgressiveDriver( + mode: `direct` | `nested`, +): TraceDriver { + return { + setup: () => { + const { + collection: posts, + events, + closeWindow, + syncStarted, + } = createProgressivePosts() + + if (mode === `direct`) { + const live = createLiveQueryCollection((q) => + q.from({ post: posts }).where(({ post }) => eq(post.userId, 2)), + ) + return { + users: undefined, + posts, + live, + events, + closeWindow, + releaseParent: undefined, + startReached: syncStarted, + parentDelivery: undefined, + preload: { preloadSettled: false }, + expected: { + events: [{ phase: `fast`, keys: [2] }], + ready: true, + preloadSettled: true, + }, + } + } + + const { + collection: users, + release, + started, + delivery, + } = createGatedUsers() + const live = createLiveQueryCollection((q) => + q + .from({ user: users }) + .where(({ user }) => eq(user.id, 2)) + .select(({ user }) => ({ + id: user.id, + posts: toArray( + q + .from({ post: posts }) + .where(({ post }) => eq(post.userId, user.id)), + ), + })), + ) + return { + users, + posts, + live, + events, + closeWindow, + releaseParent: release, + startReached: started, + parentDelivery: delivery, + preload: { preloadSettled: false }, + expected: { + events: [{ phase: `fast`, keys: [2] }], + ready: false, + preloadSettled: false, + }, + } + }, + start: async (context) => { + const preload = startPreload(context.live, context.preload) + await context.startReached.promise + context.closeWindow() + if (mode === `direct`) await preload + }, + apply: async (_step, context) => { + context.releaseParent?.() + context.expected = { + events: [{ phase: `fast`, keys: [2] }], + ready: true, + preloadSettled: true, + } + await finishPreload(context.preload) + }, + cleanup: async ({ + users, + posts, + live, + releaseParent, + parentDelivery, + preload, + }) => { + releaseParent?.() + await parentDelivery + await live.cleanup() + await finishPreload(preload) + await Promise.all([users?.cleanup(), posts.cleanup()]) + }, + } +} + +const progressiveProjection: TraceProjection< + ProgressiveContext, + ProgressiveObservation +> = { + observe: ({ events, live, preload }) => ({ + events: [...events], + ready: live.isReady(), + preloadSettled: preload.preloadSettled, + }), + recompute: ({ expected }) => expected, + assertEqual: (observed, expected) => { + expect(observed).toEqual(expected) + return undefined + }, +} + +async function expectProgressiveTraceMatches( + mode: `direct` | `nested`, +): Promise { + await runTrace({ + steps: mode === `nested` ? [`release-parent`] : [], + driver: createProgressiveDriver(mode), + projection: progressiveProjection, + }) +} + +describe(`includes temporal oracle`, () => { + it( + `discovered trace: an empty outer does not wait for an undemanded child`, + expectAssertionFailure(() => expectReadinessMatches([]), { + checkpoint: 0, + }), + ) + + it(`loads a demanded child before becoming ready`, async () => { + await expectReadinessMatches([ + { id: 1, authorId: `selected`, title: `one` }, + { id: 2, authorId: `selected`, title: `two` }, + ]) + }) + + it( + `discovered trace: obsolete child demand does not block readiness`, + expectAssertionFailure(expectObsoleteDemandDoesNotBlockReadiness, { + checkpoint: 1, + }), + ) + + it(`loads a direct progressive subset inside the fast-path window`, async () => { + await expectProgressiveTraceMatches(`direct`) + }) + + it( + `discovered trace: a nested progressive subset loads inside the fast-path window`, + expectAssertionFailure(() => expectProgressiveTraceMatches(`nested`), { + checkpoint: 0, + }), + ) +}) diff --git a/packages/db/tests/trace-runner.test.ts b/packages/db/tests/trace-runner.test.ts index 75f0eaab6..ea90cffff 100644 --- a/packages/db/tests/trace-runner.test.ts +++ b/packages/db/tests/trace-runner.test.ts @@ -92,7 +92,6 @@ describe(`runTrace`, () => { }) it(`preserves the trace failure when cleanup also fails`, async () => { - const traceError = new Error(`trace failed`) const cleanupError = new Error(`cleanup failed`) const run = runTrace({ @@ -107,18 +106,67 @@ describe(`runTrace`, () => { projection: { observe: (context) => context.observed, recompute: (context) => context.expected, - assertEqual: () => { - throw traceError + assertEqual: (observed, expected) => { + expect(observed).toBe(expected) }, }, }) - await expect(run).rejects.toBe(traceError) + const traceFailure = await run.catch((error: unknown) => error) + expect(traceFailure).toMatchObject({ + name: `TraceAssertionError`, + checkpoint: 0, + cause: { name: `AssertionError` }, + }) expect( - (traceError as Error & { suppressed?: Array }).suppressed, + (traceFailure as Error & { suppressed?: Array }).suppressed, ).toEqual([cleanupError]) }) + it(`does not wrap observation errors as assertion failures`, async () => { + const observationError = new Error(`observation failed`) + + await expect( + runTrace({ + steps: [], + driver: { + setup: () => undefined, + apply: () => undefined, + cleanup: () => undefined, + }, + projection: { + observe: () => { + throw observationError + }, + recompute: () => undefined, + assertEqual: () => undefined, + }, + }), + ).rejects.toBe(observationError) + }) + + it(`does not wrap runtime errors from assertion callbacks`, async () => { + const runtimeError = new TypeError(`assertion callback failed`) + + await expect( + runTrace({ + steps: [], + driver: { + setup: () => undefined, + apply: () => undefined, + cleanup: () => undefined, + }, + projection: { + observe: () => undefined, + recompute: () => undefined, + assertEqual: () => { + throw runtimeError + }, + }, + }), + ).rejects.toBe(runtimeError) + }) + it(`throws cleanup failures when the trace succeeds`, async () => { const cleanupError = new Error(`cleanup failed`) diff --git a/packages/db/tests/trace-runner.ts b/packages/db/tests/trace-runner.ts index 0028bb20f..3c8868d52 100644 --- a/packages/db/tests/trace-runner.ts +++ b/packages/db/tests/trace-runner.ts @@ -4,6 +4,16 @@ type ErrorWithSuppressed = Error & { suppressed?: Array } +export class TraceAssertionError extends Error { + readonly checkpoint: number + + constructor(checkpoint: number, cause: unknown) { + super(`Trace assertion failed at checkpoint ${checkpoint}`, { cause }) + this.name = `TraceAssertionError` + this.checkpoint = checkpoint + } +} + export type TraceCheckpoint = () => undefined export type TraceDriver = { @@ -64,11 +74,20 @@ export async function runTrace({ }: RunTraceOptions): Promise { const setupResult = driver.setup() const context = isPromiseLike(setupResult) ? await setupResult : setupResult + let checkpointIndex = 0 const checkpoint: TraceCheckpoint = () => { - projection.assertEqual( - projection.observe(context), - projection.recompute(context), - ) + const currentCheckpoint = checkpointIndex + checkpointIndex += 1 + const observed = projection.observe(context) + const expected = projection.recompute(context) + try { + projection.assertEqual(observed, expected) + } catch (error) { + if (!(error instanceof Error) || error.name !== `AssertionError`) { + throw error + } + throw new TraceAssertionError(currentCheckpoint, error) + } return undefined }