diff --git a/.changeset/fix-query-collection-lifecycle.md b/.changeset/fix-query-collection-lifecycle.md new file mode 100644 index 000000000..e7c5cd4ab --- /dev/null +++ b/.changeset/fix-query-collection-lifecycle.md @@ -0,0 +1,6 @@ +--- +'@tanstack/db': patch +'@tanstack/query-db-collection': patch +--- + +Start idle collections only after locally decidable mutation validation succeeds, and publish authoritative Query Collection refetch results without stale intermediate snapshots. diff --git a/docs/contributing/oracle-coverage.md b/docs/contributing/oracle-coverage.md index 13a46cd6f..cd17a6c50 100644 --- a/docs/contributing/oracle-coverage.md +++ b/docs/contributing/oracle-coverage.md @@ -45,7 +45,7 @@ comment and the current API/architecture contract before extending its model. | --- | --- | --- | | Ordered relations and BTree | [top-K relation oracle](../../packages/db-ivm/tests/operators/topk-relation-oracle.test.ts), [BTree/Map](../../packages/db/tests/btree-map-oracle.test.ts), [incrementalization laws](../../packages/db-ivm/tests/incrementalization-law.property.test.ts) | Independent ordered relations and cumulative signed output. Top-K consolidation compares same-key values without hashing, including cyclic replacements and fresh transient cancellation. Other hash-based operators retain hashing's declared domain. Algebra does not specify client readiness. | | Includes and publication | [cross-formulation](../../packages/db/tests/query/includes-cross-formulation-oracle.property.test.ts), [temporal](../../packages/db/tests/query/includes-temporal-oracle.test.ts), [Collection includes](../../packages/db/tests/query/includes-collection-oracle.property.test.ts), [architecture and complete suite map](../../packages/db/src/query/live/ARCHITECTURE.md#executable-contracts) | Per-parent/flat-join/partition relations, callback-time rows, nested values, and route histories. Observe raw promised order; fresh queries do not establish continuous publication safety. | -| Collection lifecycle | [history](../../packages/db/tests/collection-subscription-lifecycle-history.property.test.ts), [publication](../../packages/db/tests/collection-subscription-lifecycle-publication.property.test.ts), [replay](../../packages/db/tests/collection-subscription-replay-oracle.property.test.ts), [effect disposal](../../packages/db/tests/effect-disposal-oracle.test.ts) | Ownership and phase histories, exact caller/error/publication evidence, late completion and restart. Effect self-dependent disposal remains a separate contract question. | +| Collection lifecycle | [mutation startup](../../packages/db/tests/collection-mutation-startup-oracle.test.ts), [history](../../packages/db/tests/collection-subscription-lifecycle-history.property.test.ts), [publication](../../packages/db/tests/collection-subscription-lifecycle-publication.property.test.ts), [replay](../../packages/db/tests/collection-subscription-replay-oracle.property.test.ts), [effect disposal](../../packages/db/tests/effect-disposal-oracle.test.ts) | Core Collection `insert`/`update`/`delete` admission while `startSync:false` is idle; ownership and phase histories; exact caller/error/publication evidence; late completion and restart. Query write utilities and effect self-dependent disposal remain separate contracts. | | Optimistic state | [history model](../../packages/db/tests/optimistic-history-oracle.ts), [generated histories](../../packages/db/tests/optimistic-transaction-oracle.property.test.ts), [outcomes](../../packages/db/tests/optimistic-history-outcomes.test.ts), [publication](../../packages/db/tests/optimistic-history-publication.test.ts) | Independent whole-row snapshots, rollback dependencies, metadata and prior-value events. Never rebase a pending snapshot merely to simplify the 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. | diff --git a/packages/db/package.json b/packages/db/package.json index af34a1b2e..400431147 100644 --- a/packages/db/package.json +++ b/packages/db/package.json @@ -22,7 +22,7 @@ "lint": "eslint . --fix", "test": "vitest --run", "test:facade-retention": "node --expose-gc --import tsx tests/facade-retention.probe.ts", - "test:oracles": "vitest --run tests/collection-cleanup-restart-oracle.test.ts tests/effect-disposal-oracle.test.ts tests/optimistic-transaction-oracle.property.test.ts tests/optimistic-settlement-boundaries.test.ts tests/optimistic-history-publication.test.ts tests/optimistic-history-outcomes.test.ts tests/collection-metadata-publication-oracle.property.test.ts tests/collection-state-retention-oracle.property.test.ts tests/collection-subscription-lifecycle-history.property.test.ts tests/collection-subscription-lifecycle-oracle.test.ts tests/collection-subscription-lifecycle-publication.property.test.ts tests/collection-subscription-replay-oracle.property.test.ts tests/d2-source-reconciliation-oracle.property.test.ts tests/live-query-observer-history.property.test.ts tests/query/cold-join-reconciliation-oracle.test.ts tests/query/includes-collection-oracle.property.test.ts tests/query/includes-functional-projection-oracle.test.ts tests/query/includes-functional-input-boundary.test.ts tests/query/includes-context-transport-oracle.test.ts tests/query/includes-cross-formulation-oracle.property.test.ts tests/query/includes-optimistic-oracle.property.test.ts tests/query/includes-oracle.property.test.ts tests/query/includes-publication-oracle.test.ts tests/query/includes-query-shape-oracle.test.ts tests/query/includes-temporal-oracle.test.ts tests/query/includes-work-counter-oracle.test.ts tests/query/load-subset-oracle.property.test.ts tests/query/load-subset-replay-refinement-oracle.test.ts tests/query/load-subset-source-readiness-refinement-oracle.test.ts tests/query/load-subset-transaction-refinement-oracle.test.ts tests/query/ordered-source-loader-state.test.ts tests/query/ordered-demand-retirement.test.ts tests/query/ordered-default-work.test.ts tests/query/ordered-lifecycle-oracle.property.test.ts tests/query/ordered-work-oracle.property.test.ts tests/query/pagination-oracle.property.test.ts tests/query/includes-space-oracle.test.ts", + "test:oracles": "vitest --run tests/collection-mutation-startup-oracle.test.ts tests/collection-cleanup-restart-oracle.test.ts tests/effect-disposal-oracle.test.ts tests/optimistic-transaction-oracle.property.test.ts tests/optimistic-settlement-boundaries.test.ts tests/optimistic-history-publication.test.ts tests/optimistic-history-outcomes.test.ts tests/collection-metadata-publication-oracle.property.test.ts tests/collection-state-retention-oracle.property.test.ts tests/collection-subscription-lifecycle-history.property.test.ts tests/collection-subscription-lifecycle-oracle.test.ts tests/collection-subscription-lifecycle-publication.property.test.ts tests/collection-subscription-replay-oracle.property.test.ts tests/d2-source-reconciliation-oracle.property.test.ts tests/live-query-observer-history.property.test.ts tests/query/cold-join-reconciliation-oracle.test.ts tests/query/includes-collection-oracle.property.test.ts tests/query/includes-functional-projection-oracle.test.ts tests/query/includes-functional-input-boundary.test.ts tests/query/includes-context-transport-oracle.test.ts tests/query/includes-cross-formulation-oracle.property.test.ts tests/query/includes-optimistic-oracle.property.test.ts tests/query/includes-oracle.property.test.ts tests/query/includes-publication-oracle.test.ts tests/query/includes-query-shape-oracle.test.ts tests/query/includes-temporal-oracle.test.ts tests/query/includes-work-counter-oracle.test.ts tests/query/load-subset-oracle.property.test.ts tests/query/load-subset-replay-refinement-oracle.test.ts tests/query/load-subset-source-readiness-refinement-oracle.test.ts tests/query/load-subset-transaction-refinement-oracle.test.ts tests/query/ordered-source-loader-state.test.ts tests/query/ordered-demand-retirement.test.ts tests/query/ordered-default-work.test.ts tests/query/ordered-lifecycle-oracle.property.test.ts tests/query/ordered-work-oracle.property.test.ts tests/query/pagination-oracle.property.test.ts tests/query/includes-space-oracle.test.ts", "bench:nested-includes": "vitest bench tests/query/includes-performance.bench.ts --run" }, "type": "module", diff --git a/packages/db/src/collection/index.ts b/packages/db/src/collection/index.ts index 68c4a968c..caf93721d 100644 --- a/packages/db/src/collection/index.ts +++ b/packages/db/src/collection/index.ts @@ -57,20 +57,23 @@ const collectionSyncConfigCleanup: unique symbol = Symbol.for( type CollectionSyncConfigWithFactory = TSync & { readonly [collectionSyncConfigFactory]: ( - this: TSync, + source: TSync, utilities: object, + startSyncIfIdle: () => void, ) => TSync } -/** @internal Lets adapters bind a sync config to each collection instance. */ +/** @internal The factory must defer `startSyncIfIdle` until construction ends. */ export function withCollectionSyncConfigFactory( sync: TSync, - factory: (source: TSync, utilities: object) => TSync, + factory: ( + source: TSync, + utilities: object, + startSyncIfIdle: () => void, + ) => TSync, ): CollectionSyncConfigWithFactory { Object.defineProperty(sync, collectionSyncConfigFactory, { - value(this: TSync, utilities: object) { - return factory(this, utilities) - }, + value: factory, // Preserve the hook when callers wrap a sync config with object spread. enumerable: true, }) @@ -92,7 +95,11 @@ export function withCollectionSyncConfigCleanup( function materializeCollectionSyncConfig< TSync extends object, TUtils extends object, ->(sync: TSync, utilities: TUtils): { sync: TSync; utilities: TUtils } { +>( + sync: TSync, + utilities: TUtils, + startSyncIfIdle: () => void, +): { sync: TSync; utilities: TUtils } { const factory = ( sync as unknown as Partial> )[collectionSyncConfigFactory] @@ -104,7 +111,10 @@ function materializeCollectionSyncConfig< Object.getPrototypeOf(utilities), Object.getOwnPropertyDescriptors(utilities), ) as TUtils - return { sync: factory.call(sync, ownedUtilities), utilities: ownedUtilities } + return { + sync: factory(sync, ownedUtilities, startSyncIfIdle), + utilities: ownedUtilities, + } } function cleanupCollectionSyncConfig(sync: object): void { @@ -396,7 +406,9 @@ export class CollectionImpl< // Set default values for optional config properties const { sync: collectionSync, utilities: collectionUtils } = - materializeCollectionSyncConfig(config.sync, config.utils ?? {}) + materializeCollectionSyncConfig(config.sync, config.utils ?? {}, () => { + if (this._lifecycle.status === `idle`) this._sync.startSync() + }) this.config = { ...config, sync: collectionSync, diff --git a/packages/db/src/collection/mutations.ts b/packages/db/src/collection/mutations.ts index f230a51a9..056887bc3 100644 --- a/packages/db/src/collection/mutations.ts +++ b/packages/db/src/collection/mutations.ts @@ -206,9 +206,9 @@ export class CollectionMutationsManager< // Validate the data against the schema if one exists const validatedData = this.validateData(item, `insert`) - // Check if an item with this ID already exists in the collection or in the current batch + // Reject duplicate keys within this batch before starting sync. const key = this.config.getKey(validatedData) - if (this.state.has(key) || keysInCurrentBatch.has(key)) { + if (keysInCurrentBatch.has(key)) { throw new DuplicateKeyError(key) } keysInCurrentBatch.add(key) @@ -241,6 +241,14 @@ export class CollectionMutationsManager< mutations.push(mutation) }) + // Reject duplicates already visible before explicitly starting sync; startup may + // synchronously reveal additional keys, so check again afterward. + let duplicate = mutations.find(({ key }) => state.has(key)) + if (duplicate) throw new DuplicateKeyError(duplicate.key) + this.collection._sync.startSync() + duplicate = mutations.find(({ key }) => state.has(key)) + if (duplicate) throw new DuplicateKeyError(duplicate.key) + // If an ambient transaction exists, use it if (ambientTransaction) { ambientTransaction.applyMutations(mutations) @@ -317,10 +325,13 @@ export class CollectionMutationsManager< } const callback = - typeof configOrCallback === `function` ? configOrCallback : maybeCallback! + typeof configOrCallback === `function` ? configOrCallback : maybeCallback + if (typeof callback !== `function`) throw new TypeError() const config = typeof configOrCallback === `function` ? {} : configOrCallback + this.collection._sync.startSync() + // Get the current objects or empty objects if they don't exist const currentObjects = keysArray.map((key) => { const item = this.state.get(key) @@ -497,6 +508,7 @@ export class CollectionMutationsManager< } const keysArray = Array.isArray(keys) ? keys : [keys] + this.collection._sync.startSync() const mutations: Array< PendingMutation< TOutput, diff --git a/packages/db/tests/collection-mutation-startup-oracle.test.ts b/packages/db/tests/collection-mutation-startup-oracle.test.ts new file mode 100644 index 000000000..59541a35b --- /dev/null +++ b/packages/db/tests/collection-mutation-startup-oracle.test.ts @@ -0,0 +1,509 @@ +import { describe, expect, it, vi } from 'vitest' +import { z } from 'zod' +import { DbClient, collectionOptions } from '../src/client.js' +import { createCollection } from '../src/collection/index.js' +import { + DuplicateKeyError, + InvalidKeyError, + MissingDeleteHandlerError, + MissingInsertHandlerError, + MissingUpdateHandlerError, + NoKeysPassedToDeleteError, + NoKeysPassedToUpdateError, + SchemaValidationError, + UndefinedKeyError, +} from '../src/errors.js' +import type { Collection } from '../src/collection/index.js' +import type { SyncConfig } from '../src/types.js' + +/** + * Oracle review card + * Owner: core Collection insert/update/delete admission while startSync:false is idle. + * Sources: #918's regular-mutation path, #929's batch-key guard, and CodeRabbit's + * #1840 pre-start/post-start duplicate review. + * Model: finite rejection, synchronous hydration, duplicate-visibility, accepted, + * and startup-failure cells; rejected cells compare ready and throwing adapters. + * Path: public mutations through their production validation and sync entry points. + * Observations: exact error class/identity, starts, handler calls, status, rows, + * mutation type/key, and persistence. + * Mutants: eager/omitted/repeated/late startup, removed or over-broad duplicate + * checks, batch-key loss, wrong handler dispatch, and application before failure. + * Limits: no Query write utilities, deferred startup, ambient transactions, + * cleanup, asynchronous providers, reconciliation, publication, or settlement. + */ +type Row = { id: string; value: string } +type IdleCollection = Collection +type RejectionFixture = { + collection: IdleCollection + mutate: () => unknown + handlerCalls: () => number + assertError: (error: unknown) => void +} +type RejectionCase = { + name: string + create: (sync: SyncConfig) => RejectionFixture +} + +const noop = () => Promise.resolve() + +function idleConfig(sync: SyncConfig) { + return { + getKey: (row: Row) => row.id, + startSync: false, + sync, + } +} + +const rejectionCases: Array = [ + { + name: `insert without a handler`, + create: (sync) => { + const collection = createCollection(idleConfig(sync)) + return { + collection, + mutate: () => collection.insert({ id: `target`, value: `local` }), + handlerCalls: () => 0, + assertError: (error) => + expect(error).toBeInstanceOf(MissingInsertHandlerError), + } + }, + }, + { + name: `insert rejected by its schema`, + create: (sync) => { + const onInsert = vi.fn(noop) + const collection = createCollection({ + ...idleConfig(sync), + schema: z.object({ id: z.string(), value: z.string().min(1) }), + onInsert, + }) + return { + collection, + mutate: () => collection.insert({ id: `target`, value: `` }), + handlerCalls: () => onInsert.mock.calls.length, + assertError: (error) => + expect(error).toBeInstanceOf(SchemaValidationError), + } + }, + }, + { + name: `insert with a non-key value`, + create: (sync) => { + const onInsert = vi.fn(noop) + const collection = createCollection({ + ...idleConfig(sync), + getKey: () => true as never, + onInsert, + }) + return { + collection, + mutate: () => collection.insert({ id: `target`, value: `local` }), + handlerCalls: () => onInsert.mock.calls.length, + assertError: (error) => expect(error).toBeInstanceOf(InvalidKeyError), + } + }, + }, + { + name: `insert with an undefined key`, + create: (sync) => { + const onInsert = vi.fn(noop) + const collection = createCollection({ + ...idleConfig(sync), + getKey: () => undefined as never, + onInsert, + }) + return { + collection, + mutate: () => collection.insert({ id: `target`, value: `local` }), + handlerCalls: () => onInsert.mock.calls.length, + assertError: (error) => expect(error).toBeInstanceOf(UndefinedKeyError), + } + }, + }, + { + name: `insert batch with duplicate keys`, + create: (sync) => { + const onInsert = vi.fn(noop) + const collection = createCollection({ + ...idleConfig(sync), + onInsert, + }) + return { + collection, + mutate: () => + collection.insert([ + { id: `target`, value: `first` }, + { id: `target`, value: `second` }, + ]), + handlerCalls: () => onInsert.mock.calls.length, + assertError: (error) => expect(error).toBeInstanceOf(DuplicateKeyError), + } + }, + }, + { + name: `update without a handler`, + create: (sync) => { + const collection = createCollection(idleConfig(sync)) + return { + collection, + mutate: () => collection.update(`target`, () => {}), + handlerCalls: () => 0, + assertError: (error) => + expect(error).toBeInstanceOf(MissingUpdateHandlerError), + } + }, + }, + { + name: `update without keys`, + create: (sync) => { + const onUpdate = vi.fn(noop) + const collection = createCollection({ + ...idleConfig(sync), + onUpdate, + }) + return { + collection, + mutate: () => collection.update([], () => {}), + handlerCalls: () => onUpdate.mock.calls.length, + assertError: (error) => + expect(error).toBeInstanceOf(NoKeysPassedToUpdateError), + } + }, + }, + { + name: `update without a callback`, + create: (sync) => { + const onUpdate = vi.fn(noop) + const collection = createCollection({ + ...idleConfig(sync), + onUpdate, + }) + return { + collection, + mutate: () => + (collection.update as (key: string, config: object) => unknown)( + `target`, + {}, + ), + handlerCalls: () => onUpdate.mock.calls.length, + assertError: (error) => expect(error).toBeInstanceOf(TypeError), + } + }, + }, + { + name: `delete without a handler`, + create: (sync) => { + const collection = createCollection(idleConfig(sync)) + return { + collection, + mutate: () => collection.delete(`target`), + handlerCalls: () => 0, + assertError: (error) => + expect(error).toBeInstanceOf(MissingDeleteHandlerError), + } + }, + }, + { + name: `delete without keys`, + create: (sync) => { + const onDelete = vi.fn(noop) + const collection = createCollection({ + ...idleConfig(sync), + onDelete, + }) + return { + collection, + mutate: () => collection.delete([]), + handlerCalls: () => onDelete.mock.calls.length, + assertError: (error) => + expect(error).toBeInstanceOf(NoKeysPassedToDeleteError), + } + }, + }, +] + +function captureError(run: () => unknown): unknown { + try { + run() + } catch (error) { + return error + } + throw new Error(`expected mutation to reject synchronously`) +} + +describe(`Collection mutation startup oracle`, () => { + it.each(rejectionCases)( + `keeps local rejection inert before startup: $name`, + async ({ create }) => { + let readyStarts = 0 + let throwingStarts = 0 + const startupError = new Error(`startup must remain unreachable`) + const fixtures = [ + create({ + sync: ({ markReady }) => { + readyStarts++ + markReady() + }, + }), + create({ + sync: () => { + throwingStarts++ + throw startupError + }, + }), + ] + + try { + const errors = fixtures.map(({ mutate }) => captureError(mutate)) + errors.forEach(fixtures[0]!.assertError) + expect([readyStarts, throwingStarts]).toEqual([0, 0]) + for (const fixture of fixtures) { + expect(fixture.collection.status).toBe(`idle`) + expect(fixture.collection.toArray).toEqual([]) + expect(fixture.handlerCalls()).toBe(0) + } + } finally { + await Promise.all( + fixtures.map(({ collection }) => collection.cleanup()), + ) + } + }, + ) + + it.each([`update`, `delete`] as const)( + `hydrates an idle target before valid %s state lookup`, + async (operation) => { + const target = { id: `target`, value: `original` } + let syncStarts = 0 + const onUpdate = vi.fn(noop) + const onDelete = vi.fn(noop) + const collection = createCollection({ + getKey: (row) => row.id, + startSync: false, + sync: { + sync: ({ begin, write, commit, markReady }) => { + syncStarts++ + begin() + write({ type: `insert`, value: target }) + commit() + markReady() + }, + }, + onUpdate, + onDelete, + }) + + try { + expect(syncStarts).toBe(0) + expect(collection.status).toBe(`idle`) + const transaction = + operation === `update` + ? collection.update(`target`, (draft) => { + draft.value = `updated` + }) + : collection.delete(`target`) + expect(syncStarts).toBe(1) + expect(collection.status).toBe(`ready`) + expect(transaction.mutations).toMatchObject([ + { key: `target`, type: operation }, + ]) + await transaction.isPersisted.promise + expect(onUpdate).toHaveBeenCalledTimes(operation === `update` ? 1 : 0) + expect(onDelete).toHaveBeenCalledTimes(operation === `delete` ? 1 : 0) + } finally { + await collection.cleanup() + } + }, + ) + + it(`rejects a duplicate visible before startup`, async () => { + const initial = { id: `initial`, value: `initial` } + let initialStarts = 0 + const collection = new DbClient().collection( + collectionOptions({ + id: `mutation-startup-oracle-initial-data`, + getKey: (row: Row) => row.id, + startSync: false, + sync: { + sync: () => { + initialStarts++ + throw new Error(`initial duplicate must reject before startup`) + }, + }, + onInsert: noop, + }), + { initialData: [initial] }, + ) + + try { + expect(() => collection.insert({ ...initial })).toThrow(DuplicateKeyError) + expect(initialStarts).toBe(0) + expect(collection.status).toBe(`idle`) + expect( + collection.toArray.map(({ id, value }) => ({ id, value })), + ).toEqual([initial]) + } finally { + await collection.cleanup() + } + }) + + it(`rejects a duplicate revealed during startup`, async () => { + const hydrated = { id: `hydrated`, value: `remote` } + const onInsert = vi.fn(noop) + let syncStarts = 0 + const collection = createCollection({ + getKey: (row) => row.id, + startSync: false, + sync: { + sync: ({ begin, write, commit, markReady }) => { + syncStarts++ + begin() + write({ type: `insert`, value: hydrated }) + commit() + markReady() + }, + }, + onInsert, + }) + + try { + expect(() => + collection.insert({ id: hydrated.id, value: `local` }), + ).toThrow(DuplicateKeyError) + expect(syncStarts).toBe(1) + expect(collection.status).toBe(`ready`) + expect( + collection.toArray.map(({ id, value }) => ({ id, value })), + ).toEqual([hydrated]) + expect(onInsert).not.toHaveBeenCalled() + } finally { + await collection.cleanup() + } + }) + + it(`admits a distinct insert after startup hydrates another key`, async () => { + const remote = { id: `remote`, value: `remote` } + const local = { id: `local`, value: `local` } + const onInsert = vi.fn(noop) + let syncStarts = 0 + const collection = createCollection({ + getKey: (row) => row.id, + startSync: false, + sync: { + sync: ({ begin, write, commit, markReady }) => { + syncStarts++ + begin() + write({ type: `insert`, value: remote }) + commit() + markReady() + }, + }, + onInsert, + }) + + try { + const transaction = collection.insert(local) + await transaction.isPersisted.promise + expect(syncStarts).toBe(1) + expect(onInsert).toHaveBeenCalledTimes(1) + expect( + collection.toArray + .map(({ id, value }) => ({ id, value })) + .sort((left, right) => left.id.localeCompare(right.id)), + ).toEqual([local, remote]) + } finally { + await collection.cleanup() + } + }) + + it(`starts once across accepted idle inserts and persists both`, async () => { + let syncStarts = 0 + let handlerCalls = 0 + const collection = createCollection({ + getKey: (row) => row.id, + startSync: false, + sync: { + sync: ({ markReady }) => { + syncStarts++ + markReady() + }, + }, + onInsert: () => { + handlerCalls++ + return Promise.resolve() + }, + }) + + try { + expect(syncStarts).toBe(0) + expect(collection.status).toBe(`idle`) + const first = collection.insert({ id: `first`, value: `first` }) + expect(syncStarts).toBe(1) + expect(collection.status).toBe(`ready`) + const second = collection.insert({ id: `second`, value: `second` }) + await Promise.all([first.isPersisted.promise, second.isPersisted.promise]) + expect(syncStarts).toBe(1) + expect(collection.status).toBe(`ready`) + expect(handlerCalls).toBe(2) + expect(collection.toArray.map(({ id }) => id).sort()).toEqual([ + `first`, + `second`, + ]) + } finally { + await collection.cleanup() + } + }) + + it.each([`insert`, `update`, `delete`] as const)( + `propagates %s startup failure before mutation application`, + async (operation) => { + const startupError = new Error(`${operation} startup failed`) + const target = { id: `target`, value: `original` } + const initialData = operation === `insert` ? [] : [target] + let syncStarts = 0 + const onInsert = vi.fn(noop) + const onUpdate = vi.fn(noop) + const onDelete = vi.fn(noop) + const collection = new DbClient().collection( + collectionOptions({ + id: `mutation-startup-oracle-failure-${operation}`, + getKey: (row: Row) => row.id, + startSync: false, + sync: { + sync: () => { + syncStarts++ + throw startupError + }, + }, + onInsert, + onUpdate, + onDelete, + }), + { initialData }, + ) + + try { + const mutate = () => { + if (operation === `insert`) + return collection.insert({ id: `target`, value: `local` }) + if (operation === `update`) + return collection.update(`target`, (draft) => { + draft.value = `updated` + }) + return collection.delete(`target`) + } + expect(captureError(mutate)).toBe(startupError) + expect(syncStarts).toBe(1) + expect(collection.status).toBe(`error`) + expect( + collection.toArray.map(({ id, value }) => ({ id, value })), + ).toEqual(initialData) + expect(onInsert).not.toHaveBeenCalled() + expect(onUpdate).not.toHaveBeenCalled() + expect(onDelete).not.toHaveBeenCalled() + } finally { + await collection.cleanup() + } + }, + ) +}) diff --git a/packages/query-db-collection/src/query.ts b/packages/query-db-collection/src/query.ts index 3c2e45933..a9b83ee8f 100644 --- a/packages/query-db-collection/src/query.ts +++ b/packages/query-db-collection/src/query.ts @@ -4,6 +4,7 @@ import { deepEquals, getLoadSubsetDemandKey, withCollectionConfigFactory, + withCollectionSyncConfigFactory, } from '@tanstack/db' import { GetKeyRequiredError, @@ -995,8 +996,13 @@ export function queryCollectionOptions( const retainedQueriesPendingRevalidation = new Set() const pendingResultApplications = new Map>() const failedResultApplications = new Map() - const resultApplicationTokens = new Map() - const resultApplicationControllers = new Map>() + type ResultApplicationController = AbortController & { + rollback?: () => void + } + const resultApplicationControllers = new Map< + string, + ResultApplicationController + >() const effectivePersistedGcTimes = new Map() const persistedRetentionTimers = new Map< string, @@ -1005,20 +1011,41 @@ export function queryCollectionOptions( let persistedRetentionMaintenance = Promise.resolve() const invalidatePendingResultApplication = (hashedQueryKey: string) => { + const controller = resultApplicationControllers.get(hashedQueryKey) + controller?.rollback?.() pendingResultApplications.delete(hashedQueryKey) failedResultApplications.delete(hashedQueryKey) - resultApplicationTokens.delete(hashedQueryKey) - resultApplicationControllers - .get(hashedQueryKey) - ?.forEach((controller) => controller.abort()) resultApplicationControllers.delete(hashedQueryKey) + controller?.abort() + } + + const waitForCurrentResultApplication = async ( + hashedQueryKey: string, + ): Promise => { + for (;;) { + const application = pendingResultApplications.get(hashedQueryKey) + if (!application) return + try { + await application + } catch (error) { + if ( + pendingResultApplications.get(hashedQueryKey) === application || + (failedResultApplications.has(hashedQueryKey) && + failedResultApplications.get(hashedQueryKey) === error) + ) { + throw error + } + } + } } const getResultApplicationSettlement = ( hashedQueryKey: string, ): true | Promise => { const pending = pendingResultApplications.get(hashedQueryKey) - if (pending) return pending + if (pending) { + return waitForCurrentResultApplication(hashedQueryKey) + } if (failedResultApplications.has(hashedQueryKey)) { return Promise.reject(failedResultApplications.get(hashedQueryKey)) @@ -1589,6 +1616,7 @@ export function queryCollectionOptions( const applySuccessfulResult = async ( queryKey: QueryKey, result: QueryObserverResult, + applicationToken: ResultApplicationController, persistedBaseline?: Map< string | number, { @@ -1649,9 +1677,17 @@ export function queryCollectionOptions( previousOwnersByRow.set(key, owners ? new Set(owners) : undefined) }) let transactionActive = false + let resultTransaction: { applicationStarted: boolean } | undefined const restoreOwnershipTracking = () => { + // Core flips this at its no-cancel point before publication can reenter. + if (resultTransaction?.applicationStarted) return if (!state.observers.has(hashedQueryKey)) return + if ( + resultApplicationControllers.get(hashedQueryKey) !== applicationToken + ) { + return + } if (previousOwnedRows === undefined) { queryToRows.delete(hashedQueryKey) @@ -1666,6 +1702,7 @@ export function queryCollectionOptions( } }) } + applicationToken.rollback = restoreOwnershipTracking try { // From this point onward the result, including an empty result, is the @@ -1722,6 +1759,7 @@ export function queryCollectionOptions( } }) + resultTransaction = collection._state.pendingSyncedTransactions.at(-1) const applied = commit(signal) transactionActive = false retainedQueriesPendingRevalidation.delete(hashedQueryKey) @@ -1729,14 +1767,8 @@ export function queryCollectionOptions( // Readiness is publication: do not expose it until the establishing // transaction's rows and events are visible. - if (applied !== true) { - await applied - } - if (signal?.aborted) { - restoreOwnershipTracking() - return - } - markReady() + if (applied !== true) await applied + if (!signal?.aborted) markReady() } catch (error) { restoreOwnershipTracking() @@ -1756,7 +1788,7 @@ export function queryCollectionOptions( const reconcileSuccessfulResult = async ( queryKey: QueryKey, result: QueryObserverResult, - applicationToken: object, + applicationToken: ResultApplicationController, signal: AbortSignal, ) => { const hashedQueryKey = hashKey(queryKey) @@ -1764,11 +1796,17 @@ export function queryCollectionOptions( await loadPersistedBaselineForQuery(hashedQueryKey) if ( collection.status === `cleaned-up` || - resultApplicationTokens.get(hashedQueryKey) !== applicationToken + resultApplicationControllers.get(hashedQueryKey) !== applicationToken ) { return } - await applySuccessfulResult(queryKey, result, persistedBaseline, signal) + await applySuccessfulResult( + queryKey, + result, + applicationToken, + persistedBaseline, + signal, + ) } const trackResultApplication = ( @@ -1806,26 +1844,24 @@ export function queryCollectionOptions( const enqueueResultApplication = ( hashedQueryKey: string, - apply: (signal: AbortSignal) => Promise, + apply: ( + signal: AbortSignal, + applicationToken: ResultApplicationController, + ) => Promise, ): void => { - const controller = new AbortController() - const controllers = - resultApplicationControllers.get(hashedQueryKey) ?? new Set() - controllers.add(controller) - resultApplicationControllers.set(hashedQueryKey, controllers) - const previousApplication = pendingResultApplications.get(hashedQueryKey) - const run = () => apply(controller.signal) - const application = previousApplication - ? previousApplication.then(run, run) - : run() + invalidatePendingResultApplication(hashedQueryKey) + const controller: ResultApplicationController = new AbortController() + resultApplicationControllers.set(hashedQueryKey, controller) + const application = apply(controller.signal, controller) const cleanupController = () => { - controllers.delete(controller) - if (controllers.size === 0) { + if (resultApplicationControllers.get(hashedQueryKey) === controller) { resultApplicationControllers.delete(hashedQueryKey) } } void application.then(cleanupController, cleanupController) - trackResultApplication(hashedQueryKey, application) + if (resultApplicationControllers.get(hashedQueryKey) === controller) { + trackResultApplication(hashedQueryKey, application) + } } // eslint-disable-next-line no-shadow @@ -1911,20 +1947,14 @@ export function queryCollectionOptions( } return } + if (result.isFetching) return - const applicationToken = {} - resultApplicationTokens.set(hashedQueryKey, applicationToken) - enqueueResultApplication(hashedQueryKey, (signal) => - reconcileSuccessfulResult( - queryKey, - result, - applicationToken, - signal, - ), + enqueueResultApplication(hashedQueryKey, (signal, token) => + reconcileSuccessfulResult(queryKey, result, token, signal), ) } else { - enqueueResultApplication(hashedQueryKey, (signal) => - applySuccessfulResult(queryKey, result, undefined, signal), + enqueueResultApplication(hashedQueryKey, (signal, token) => + applySuccessfulResult(queryKey, result, token, undefined, signal), ) } } else { @@ -2152,7 +2182,6 @@ export function queryCollectionOptions( } const hasListeners = observer?.hasListeners() ?? false - if (hasListeners) { // During invalidateQueries, TanStack Query keeps internal listeners alive. // Leave refcount at 0 but keep observer so it can resubscribe. @@ -2767,11 +2796,29 @@ export function queryCollectionOptions( // Create utils instance with state and dependencies passed explicitly const utils: any = new QueryCollectionUtilsImpl(state, refetch, writeUtils) + const sync = withCollectionSyncConfigFactory( + { sync: enhancedInternalSync }, + (source, utilities, startSyncIfIdle) => { + const boundUtilities = utilities as Record< + string, + (...args: Array) => any + > + for (const name of Object.keys(writeUtils)) { + const write = boundUtilities[name]! + boundUtilities[name] = (...args) => { + startSyncIfIdle() + return write(...args) + } + } + return source + }, + ) + const options = { ...baseCollectionConfig, getKey, syncMode, - sync: { sync: enhancedInternalSync }, + sync, onInsert: wrappedOnInsert, onUpdate: wrappedOnUpdate, onDelete: wrappedOnDelete, diff --git a/packages/query-db-collection/tests/ownership-lifecycle.oracle.test.ts b/packages/query-db-collection/tests/ownership-lifecycle.oracle.test.ts index e2f325c73..3a5958f79 100644 --- a/packages/query-db-collection/tests/ownership-lifecycle.oracle.test.ts +++ b/packages/query-db-collection/tests/ownership-lifecycle.oracle.test.ts @@ -1,6 +1,7 @@ import { QueryClient, QueryObserver, + focusManager, hashKey, isCancelledError, } from '@tanstack/query-core' @@ -14,6 +15,7 @@ import { import { afterEach, describe, expect, it, vi } from 'vitest' import { createDeferred } from '../../db/src/deferred.js' import { persistedCollectionOptions } from '../../db-sqlite-persistence-core/src/index.js' +import { SyncNotInitializedError } from '../src/errors.js' import { queryCollectionOptions } from '../src/query.js' import type { Collection, @@ -31,6 +33,19 @@ type Item = { name: string } +type MutationLifecycleSnapshot = { + server: Array + cache: Array + synced: Array + source: Array + derived: Array +} + +type MutationPublicationSnapshot = Pick< + MutationLifecycleSnapshot, + `source` | `derived` +> & { expected: Array } + type MetadataRecorder = { rows: Map writes: Array<{ type: `set` | `delete`; key: string | number }> @@ -44,6 +59,9 @@ type OwnershipFixtureOptions = { staleTime?: number metadataRecorder?: MetadataRecorder setupMetadata?: (metadata: SyncMetadataApi) => void + scanPersisted?: () => Promise< + Array<{ key: string | number; value: Item; metadata?: unknown }> + > } type OwnershipFixture = { @@ -116,6 +134,7 @@ function createOwnershipFixture({ syncMode = `on-demand`, metadataRecorder, setupMetadata, + scanPersisted, customHash, staleTime, }: OwnershipFixtureOptions): OwnershipFixture { @@ -137,7 +156,7 @@ function createOwnershipFixture({ const originalSync = baseOptions.sync let pendingSetup = setupMetadata const collection = createCollection( - metadataRecorder || setupMetadata + metadataRecorder || setupMetadata || scanPersisted ? { ...baseOptions, sync: { @@ -148,15 +167,21 @@ function createOwnershipFixture({ const observedMetadata = metadataRecorder ? recordMetadata(params.metadata, metadataRecorder) : params.metadata + const metadataWithPersistedScan = scanPersisted + ? ({ + ...observedMetadata, + row: { ...observedMetadata.row, scanPersisted }, + } as SyncMetadataApi) + : observedMetadata if (pendingSetup) { params.begin() - pendingSetup(observedMetadata) + pendingSetup(metadataWithPersistedScan) params.commit() pendingSetup = undefined } return originalSync.sync({ ...params, - metadata: observedMetadata, + metadata: metadataWithPersistedScan, }) }, }, @@ -176,6 +201,37 @@ function rows(collection: { return Array.from(collection.keys()).map(String).sort() } +function itemIds(items: Iterable): Array { + return Array.from(items, ({ id }) => id).sort() +} + +function expectMutationLifecycleSnapshot( + snapshot: MutationLifecycleSnapshot, + authoritative: ReadonlyArray, + visible: ReadonlyArray, + ordered: boolean, +): void { + expect(snapshot).toEqual({ + server: [...authoritative].sort(), + cache: [...authoritative].sort(), + synced: [...authoritative].sort(), + source: [...visible].sort(), + derived: ordered ? [...visible] : [...visible].sort(), + }) +} + +function expectMutationPublicationIntegrity( + publications: ReadonlyArray, +): void { + for (const publication of publications) { + expect(publication).toEqual({ + source: publication.expected, + derived: publication.expected, + expected: publication.expected, + }) + } +} + function persistedOwners( metadata: ReadonlyMap, rowId: string, @@ -237,10 +293,7 @@ function selectOwnershipRows( .map((item) => structuredClone(item)) } -function createOwnershipStorage( - seed?: StoredOwnership, - gateFirstCommit = false, -) { +function createOwnershipStorage(seed?: StoredOwnership, gatedCommit?: number) { const state: StoredOwnership = structuredClone( seed ?? { rows: new Map(), @@ -279,7 +332,7 @@ function createOwnershipStorage( applyCommittedTx: async (_id, transaction) => { const tx = structuredClone(transaction) commitCount++ - if (gateFirstCommit && commitCount === 1) { + if (commitCount === gatedCommit) { entered.resolve() await released.promise } @@ -373,7 +426,7 @@ function createPersistedOwnershipFixture( queryClient.clear() } }) - return { collection, queryFn } + return { collection, queryClient, queryFn } } function storedItems( @@ -389,7 +442,7 @@ type ColdOwnershipObservation = { stored: Array; visible: Array } async function observeColdOwnerRevalidation( dropMetadata = false, ): Promise> { - const hotStorage = createOwnershipStorage(undefined, true) + const hotStorage = createOwnershipStorage(undefined, 1) const hot = createPersistedOwnershipFixture( `cold-owner-revalidation`, hotStorage, @@ -465,6 +518,1199 @@ describe(`query collection ownership lifecycle`, () => { await Promise.all(cleanups.splice(0).map((cleanup) => cleanup())) }) + it(`starts an idle collection only when a direct write is invoked`, () => { + const queryClient = createQueryClient() + const queryFn = vi.fn((): Promise> => Promise.resolve([])) + const collection = createCollection( + queryCollectionOptions({ + id: `idle-direct-write-startup`, + queryClient, + queryKey: [`idle-direct-write-startup`], + queryFn, + getKey: (item) => item.id, + syncMode: `on-demand`, + startSync: false, + }), + ) + cleanups.push(async () => { + await collection.cleanup() + queryClient.clear() + }) + + expect(collection.status).toBe(`idle`) + expect(collection.utils.isError).toBe(false) + expect(collection.status).toBe(`idle`) + + collection.utils.writeUpsert({ + id: `first`, + category: `direct`, + name: `First`, + }) + + expect(collection.status).toBe(`ready`) + expect(rows(collection)).toEqual([`first`]) + expect(queryFn).not.toHaveBeenCalled() + }) + + it(`fails fast when a direct write is attempted during deferred startup`, () => { + const queryClient = createQueryClient() + const queryFn = vi.fn((): Promise> => Promise.resolve([])) + const collection = createCollection( + queryCollectionOptions({ + id: `deferred-direct-write`, + queryClient, + queryKey: [`deferred-direct-write`], + queryFn, + getKey: (item) => item.id, + syncMode: `on-demand`, + startSync: false, + }), + ) + cleanups.push(async () => { + await collection.cleanup() + queryClient.clear() + }) + + expect(collection._deferSyncStart()).toBe(true) + for (const write of [ + () => collection.utils.writeInsert(shared), + () => collection.utils.writeUpdate(shared), + () => collection.utils.writeDelete(shared.id), + () => collection.utils.writeUpsert(shared), + () => collection.utils.writeBatch(() => {}), + ]) { + expect(write).toThrow(SyncNotInitializedError) + } + expect(collection.status).toBe(`idle`) + expect(rows(collection)).toEqual([]) + + collection._resumeSyncStart() + expect(collection.status).toBe(`ready`) + expect(rows(collection)).toEqual([]) + expect(queryFn).not.toHaveBeenCalled() + }) + + it(`does not restart a cleaned-up idle collection for a direct write`, async () => { + const id = `cleaned-idle-direct-write` + const queryClient = createQueryClient() + const queryFn = vi.fn((): Promise> => Promise.resolve([])) + const collection = createCollection( + queryCollectionOptions({ + id, + queryClient, + queryKey: [id], + queryFn, + getKey: (item) => item.id, + syncMode: `on-demand`, + startSync: false, + }), + ) + cleanups.push(async () => { + await collection.cleanup() + queryClient.clear() + }) + + await collection.cleanup() + expect(() => collection.utils.writeUpsert(shared)).toThrow( + SyncNotInitializedError, + ) + expect(collection.status).toBe(`cleaned-up`) + expect(rows(collection)).toEqual([]) + expect(queryFn).not.toHaveBeenCalled() + }) + + it(`does not restart a cleaned-up collection for a late mutation refetch`, async () => { + const id = `late-mutation-after-cleanup` + const queryKey = [id] as const + const inserted = { id: `late`, category: `mutation`, name: `Late` } + const serverRows: Array = [] + const handlerEntered = createDeferred() + const releaseHandler = createDeferred() + const queryClient = createQueryClient() + const queryFn = vi.fn(() => Promise.resolve(structuredClone(serverRows))) + let writeLate = (_item: Item) => {} + const collection = createCollection( + queryCollectionOptions({ + id, + queryClient, + queryKey, + queryFn, + getKey: (item) => item.id, + startSync: true, + onInsert: async ({ transaction }) => { + handlerEntered.resolve() + await releaseHandler.promise + transaction.mutations.forEach(({ modified }) => + writeLate(structuredClone(modified)), + ) + return { refetch: false } + }, + }), + ) + writeLate = (item) => collection.utils.writeUpsert(item) + cleanups.push(async () => { + releaseHandler.resolve() + await collection.cleanup() + queryClient.clear() + }) + + await collection.stateWhenReady() + expect(queryFn).toHaveBeenCalledOnce() + + const mutation = collection.insert(inserted) + await handlerEntered.promise + await collection.cleanup() + releaseHandler.resolve() + await mutation.isPersisted.promise + + expect(collection.status).toBe(`cleaned-up`) + expect(queryFn).toHaveBeenCalledOnce() + expect(itemIds(collection._state.syncedData.values())).toEqual([]) + expect(queryClient.getQueryData(queryKey)).toEqual([]) + }) + + it.each([ + { ordered: false, settlementOrder: [0, 1] }, + { ordered: false, settlementOrder: [1, 0] }, + { ordered: true, settlementOrder: [0, 1] }, + { ordered: true, settlementOrder: [1, 0] }, + ] as const)( + `publishes mutation refetches through source and downstream view: %j`, + async ({ ordered, settlementOrder }) => { + const id = `mutation-publication-${ordered ? `ordered` : `unordered`}-${settlementOrder.join(``)}` + const queryClient = createQueryClient() + const serverRows: Array = [] + const persistenceGates: Array>> = + [] + const queryFn = vi.fn(() => + Promise.resolve(serverRows.map((item) => structuredClone(item))), + ) + const collection = createCollection( + queryCollectionOptions({ + id, + queryClient, + queryKey: [id], + queryFn, + getKey: (item) => item.id, + startSync: true, + onInsert: async ({ transaction }) => { + const gate = createDeferred() + persistenceGates.push(gate) + await gate.promise + serverRows.push( + ...transaction.mutations.map(({ modified }) => + structuredClone(modified), + ), + ) + }, + }), + ) + const derived = createLiveQueryCollection((query) => { + const source = query.from({ item: collection }) + const result = ordered + ? source.orderBy(({ item }) => item.name, `asc`) + : source + return result.select(({ item }) => ({ ...item })) + }) + const publications: Array = [] + let expectedPublication: Array = [] + const subscription = derived.subscribeChanges(() => { + publications.push({ + source: ordered + ? [...collection.toArray] + .sort((left, right) => left.name.localeCompare(right.name)) + .map(({ id: rowId }) => rowId) + : rows(collection), + derived: ordered + ? derived.toArray.map(({ id: rowId }) => rowId) + : rows(derived), + expected: expectedPublication, + }) + }) + cleanups.push(async () => { + persistenceGates.forEach((gate) => gate.resolve()) + subscription.unsubscribe() + await derived.cleanup() + await collection.cleanup() + queryClient.clear() + }) + + const firstItem = { + id: `z-first`, + category: `mutation`, + name: `1-first`, + } + const secondItem = { + id: `a-second`, + category: `mutation`, + name: `2-second`, + } + await derived.preload() + expectedPublication = [firstItem.id] + const first = collection.insert(firstItem) + const visibleIds = [firstItem.id, secondItem.id] + expectedPublication = ordered ? visibleIds : [...visibleIds].sort() + const second = collection.insert(secondItem) + const transactions = [first, second] + const items = [firstItem, secondItem] + const queryKey = [id] as const + const capture = (): MutationLifecycleSnapshot => ({ + server: itemIds(serverRows), + cache: itemIds(queryClient.getQueryData>(queryKey) ?? []), + synced: itemIds(collection._state.syncedData.values()), + source: itemIds(collection.toArray), + derived: ordered + ? derived.toArray.map(({ id: rowId }) => rowId) + : itemIds(derived.toArray), + }) + + expect(persistenceGates).toHaveLength(2) + expect(queryFn).toHaveBeenCalledOnce() + expectMutationLifecycleSnapshot(capture(), [], visibleIds, ordered) + + const settled = new Set() + for (const transactionIndex of settlementOrder) { + persistenceGates[transactionIndex]!.resolve() + await transactions[transactionIndex]!.isPersisted.promise + settled.add(transactionIndex) + + const authoritativeIds = [...settled].map((index) => items[index]!.id) + expect(queryFn).toHaveBeenCalledTimes(1 + settled.size) + const expectedSynced = + settled.size === transactions.length ? authoritativeIds : [] + expect(capture()).toEqual({ + server: [...authoritativeIds].sort(), + cache: [...authoritativeIds].sort(), + synced: [...expectedSynced].sort(), + source: [...visibleIds].sort(), + derived: ordered ? visibleIds : [...visibleIds].sort(), + }) + expectMutationPublicationIntegrity(publications) + expect(publications.at(-1)?.derived).toEqual( + ordered ? visibleIds : [...visibleIds].sort(), + ) + } + + const thirdItem = { + id: `m-third`, + category: `mutation`, + name: `3-third`, + } + const allVisibleIds = [...visibleIds, thirdItem.id] + expectedPublication = ordered ? allVisibleIds : [...allVisibleIds].sort() + const third = collection.insert(thirdItem) + expect(persistenceGates).toHaveLength(3) + expectMutationLifecycleSnapshot( + capture(), + visibleIds, + allVisibleIds, + ordered, + ) + + persistenceGates[2]!.resolve() + await third.isPersisted.promise + + expect(queryFn).toHaveBeenCalledTimes(4) + expectMutationLifecycleSnapshot( + capture(), + allVisibleIds, + allVisibleIds, + ordered, + ) + expectMutationPublicationIntegrity(publications) + expect(publications.at(-1)?.derived).toEqual( + ordered ? allVisibleIds : [...allVisibleIds].sort(), + ) + }, + ) + + it(`rejects incomplete and misordered mutation publication receipts`, () => { + const visible = [`z-first`, `a-second`, `m-third`] + const valid: MutationLifecycleSnapshot = { + server: [...visible].sort(), + cache: [...visible].sort(), + synced: [...visible].sort(), + source: [...visible].sort(), + derived: visible, + } + + expectMutationLifecycleSnapshot(valid, visible, visible, true) + for (const mutant of [ + { ...valid, synced: [] }, + { ...valid, source: [] }, + { ...valid, derived: visible.slice(0, 2) }, + { ...valid, derived: [...visible].reverse() }, + { ...valid, source: [...valid.source, `stale`] }, + ]) { + expect(() => + expectMutationLifecycleSnapshot(mutant, visible, visible, true), + ).toThrow() + } + }) + + it(`rejects a torn intermediate publication before a complete final snapshot`, () => { + const first = [`z-first`] + const firstAndSecond = [`z-first`, `a-second`] + const allVisible = [`z-first`, `a-second`, `m-third`] + const legitimateSequence: Array = [ + { source: first, derived: first, expected: first }, + { + source: firstAndSecond, + derived: firstAndSecond, + expected: firstAndSecond, + }, + { source: allVisible, derived: allVisible, expected: allVisible }, + ] + + expect(() => + expectMutationPublicationIntegrity(legitimateSequence), + ).not.toThrow() + expect(() => + expectMutationPublicationIntegrity([ + ...legitimateSequence.slice(0, 2), + { source: [], derived: [], expected: allVisible }, + legitimateSequence[2]!, + ]), + ).toThrow() + }) + + it(`supersedes a stale focus result before publishing a newer mutation snapshot`, async () => { + const id = `focus-result-supersession` + const queryKey = [id] as const + const initial = { id: `a`, category: `mutation`, name: `A` } + const retired = { id: `b`, category: `mutation`, name: `B` } + const inserted = { id: `c`, category: `mutation`, name: `C` } + const serverRows = [initial, retired] + const firstFocus = createDeferred>() + const secondFocus = createDeferred>() + const handlerEntered = createDeferred() + const releaseHandler = createDeferred() + const queryClient = createQueryClient(false, 0) + let queryCalls = 0 + const queryFn = vi.fn(() => { + queryCalls++ + if (queryCalls === 2) return firstFocus.promise + if (queryCalls === 3) return secondFocus.promise + return Promise.resolve(structuredClone(serverRows)) + }) + const collection = createCollection( + queryCollectionOptions({ + id, + queryClient, + queryKey, + queryFn, + getKey: (item) => item.id, + syncMode: `on-demand`, + startSync: true, + staleTime: 0, + refetchOnWindowFocus: true, + onInsert: async ({ transaction }) => { + serverRows.splice( + 0, + serverRows.length, + initial, + ...transaction.mutations.map(({ modified }) => + structuredClone(modified), + ), + ) + handlerEntered.resolve() + await releaseHandler.promise + return { refetch: false } + }, + }), + ) + const derived = createLiveQueryCollection((query) => + query.from({ item: collection }).select(({ item }) => ({ + id: item.id, + category: item.category, + name: item.name, + })), + ) + const sourcePublications: Array> = [] + const derivedPublications: Array = [] + const sourceSubscription = collection.subscribeChanges(() => { + sourcePublications.push(itemIds(collection.toArray)) + }) + const derivedSubscription = derived.subscribeChanges(() => { + const source = itemIds(collection.toArray) + derivedPublications.push({ + source, + derived: itemIds(derived.toArray), + expected: source, + }) + }) + cleanups.push(async () => { + releaseHandler.resolve() + sourceSubscription.unsubscribe() + derivedSubscription.unsubscribe() + await derived.cleanup() + await collection.cleanup() + queryClient.clear() + focusManager.setFocused(undefined) + }) + + await derived.preload() + expect(queryFn).toHaveBeenCalledOnce() + sourcePublications.length = 0 + derivedPublications.length = 0 + + focusManager.setFocused(false) + focusManager.setFocused(true) + await vi.waitFor(() => expect(queryFn).toHaveBeenCalledTimes(2)) + + const mutation = collection.insert(inserted) + await handlerEntered.promise + firstFocus.resolve([structuredClone(initial)]) + await vi.waitFor(() => + expect(queryClient.getQueryState(queryKey)?.fetchStatus).toBe(`idle`), + ) + + let waiterOutcome: `pending` | `resolved` | `rejected` = `pending` + const waiter = Promise.resolve(collection._sync.loadSubset({})).then( + () => { + waiterOutcome = `resolved` + }, + (error) => { + waiterOutcome = `rejected` + throw error + }, + ) + + focusManager.setFocused(false) + focusManager.setFocused(true) + await vi.waitFor(() => expect(queryFn).toHaveBeenCalledTimes(3)) + secondFocus.resolve(structuredClone(serverRows)) + await vi.waitFor(() => + expect(queryClient.getQueryState(queryKey)?.fetchStatus).toBe(`idle`), + ) + + releaseHandler.resolve() + await mutation.isPersisted.promise + await waiter + await vi.waitFor(() => + expect(itemIds(derived.toArray)).toEqual([initial.id, inserted.id]), + ) + + expect(waiterOutcome).toBe(`resolved`) + expect(sourcePublications).not.toContainEqual([initial.id]) + for (const publication of sourcePublications) { + expect([ + [initial.id, inserted.id], + [initial.id, retired.id, inserted.id].sort(), + ]).toContainEqual(publication) + } + expectMutationPublicationIntegrity(derivedPublications) + expect({ + cache: itemIds(queryClient.getQueryData>(queryKey) ?? []), + synced: itemIds(collection._state.syncedData.values()), + source: itemIds(collection.toArray), + derived: itemIds(derived.toArray), + }).toEqual({ + cache: [initial.id, inserted.id], + synced: [initial.id, inserted.id], + source: [initial.id, inserted.id], + derived: [initial.id, inserted.id], + }) + + queryClient.setQueryData(queryKey, []) + await vi.waitFor(() => { + expect({ + synced: itemIds(collection._state.syncedData.values()), + source: itemIds(collection.toArray), + derived: itemIds(derived.toArray), + }).toEqual({ synced: [], source: [], derived: [] }) + }) + }) + + it(`removes rows superseded while an older result is publishing`, async () => { + const id = `reentrant-committed-result-supersession` + const queryKey = [id] as const + const initial = { id: `a`, category: `result`, name: `A` } + const sibling = { id: `b`, category: `result`, name: `B` } + const updated = { ...initial, name: `A from server` } + const orphaned = { id: `c`, category: `result`, name: `C` } + const handlerEntered = createDeferred() + const releaseHandler = createDeferred() + const queryClient = createQueryClient() + const queryFn = vi.fn(() => Promise.resolve([initial, sibling])) + const onUpdate = vi.fn(async () => { + handlerEntered.resolve() + await releaseHandler.promise + return { refetch: false } + }) + const collection = createCollection( + queryCollectionOptions({ + id, + queryClient, + queryKey, + queryFn, + getKey: (item) => item.id, + startSync: true, + onUpdate, + }), + ) + cleanups.push(async () => { + releaseHandler.resolve() + await collection.cleanup() + queryClient.clear() + }) + + await collection.stateWhenReady() + const mutation = collection.update(initial.id, (draft) => { + draft.name = `A optimistic` + }) + await handlerEntered.promise + + const publications: Array> = [] + let superseded = false + const subscription = collection.subscribeChanges(() => { + const ids = itemIds(collection.toArray) + publications.push(ids) + if (!superseded && ids.includes(orphaned.id)) { + superseded = true + queryClient.setQueryData(queryKey, [updated, sibling]) + } + }) + cleanups.push(() => Promise.resolve(subscription.unsubscribe())) + + queryClient.setQueryData(queryKey, [updated, sibling, orphaned]) + expect(itemIds(collection._state.syncedData.values())).toEqual([ + initial.id, + sibling.id, + ]) + + releaseHandler.resolve() + await mutation.isPersisted.promise + await collection._sync.loadSubset({}) + + expect(queryFn).toHaveBeenCalledOnce() + expect(onUpdate).toHaveBeenCalledOnce() + expect(superseded).toBe(true) + expect(publications).toContainEqual([initial.id, sibling.id, orphaned.id]) + expect({ + cache: itemIds(queryClient.getQueryData>(queryKey) ?? []), + synced: itemIds(collection._state.syncedData.values()), + source: itemIds(collection.toArray), + }).toEqual({ + cache: [initial.id, sibling.id], + synced: [initial.id, sibling.id], + source: [initial.id, sibling.id], + }) + expect( + persistedOwners(collection._state.syncedMetadata, orphaned.id), + ).toEqual([]) + }) + + it(`retains ownership after a publication listener throws`, async () => { + const id = `failed-result-publication` + const queryKey = [id] as const + const initial = { id: `a`, category: `result`, name: `A` } + const sibling = { id: `b`, category: `result`, name: `B` } + const transient = { id: `c`, category: `result`, name: `C` } + const publicationError = new Error(`Publication failed`) + const queryClient = createQueryClient() + const queryFn = vi.fn(() => Promise.resolve([initial, sibling])) + const consoleError = vi.spyOn(console, `error`).mockImplementation(() => {}) + const collection = createCollection( + queryCollectionOptions({ + id, + queryClient, + queryKey, + queryFn, + getKey: (item) => item.id, + startSync: true, + }), + ) + cleanups.push(async () => { + consoleError.mockRestore() + await collection.cleanup() + queryClient.clear() + }) + + await collection.stateWhenReady() + let threw = false + const subscription = collection.subscribeChanges(() => { + if (!threw && collection.has(transient.id)) { + threw = true + throw publicationError + } + }) + cleanups.push(() => Promise.resolve(subscription.unsubscribe())) + + queryClient.setQueryData(queryKey, [initial, sibling, transient]) + await vi.waitFor(() => expect(threw).toBe(true)) + queryClient.setQueryData(queryKey, [initial, sibling]) + await collection._sync.loadSubset({}) + + expect(queryFn).toHaveBeenCalledOnce() + expect({ + cache: itemIds(queryClient.getQueryData>(queryKey) ?? []), + synced: itemIds(collection._state.syncedData.values()), + source: itemIds(collection.toArray), + }).toEqual({ + cache: [initial.id, sibling.id], + synced: [initial.id, sibling.id], + source: [initial.id, sibling.id], + }) + expect( + persistedOwners(collection._state.syncedMetadata, transient.id), + ).toEqual([]) + }) + + it(`reports a reentrant initial replacement failure before readiness`, async () => { + const id = `failed-initial-result-replacement` + const queryKey = [id] as const + const initial = { id: `a`, category: `result`, name: `A` } + const invalid = { id: `invalid`, category: `result`, name: `Invalid` } + const applicationError = new Error(`Replacement application failed`) + const firstResult = createDeferred>() + const queryClient = createQueryClient() + const queryFn = vi.fn(() => firstResult.promise) + const consoleError = vi.spyOn(console, `error`).mockImplementation(() => {}) + const collection = createCollection( + queryCollectionOptions({ + id, + queryClient, + queryKey, + queryFn, + getKey: (item) => { + if (item.id === invalid.id) throw applicationError + return item.id + }, + startSync: false, + }), + ) + cleanups.push(async () => { + firstResult.resolve([initial]) + consoleError.mockRestore() + await collection.cleanup() + queryClient.clear() + }) + + collection.startSyncImmediate() + expect(collection.status).toBe(`loading`) + let replaced = false + const subscription = collection.subscribeChanges(() => { + if (!replaced && collection.has(initial.id)) { + replaced = true + queryClient.setQueryData(queryKey, [invalid]) + } + }) + cleanups.push(() => Promise.resolve(subscription.unsubscribe())) + + firstResult.resolve([initial]) + await vi.waitFor(() => expect(collection.status).toBe(`error`)) + + expect(queryFn).toHaveBeenCalledOnce() + expect(replaced).toBe(true) + expect(collection.utils.lastError).toBe(applicationError) + expect(itemIds(collection.toArray)).toEqual([initial.id]) + }) + + it(`releases provisional ownership when commit fails before publication`, async () => { + const id = `failed-prepublication-commit` + const detailSubset = { where: eq(`category`, `detail`) } + const listSubset = { where: eq(`category`, `list`) } + const detailKey = [id, getLoadSubsetDemandKey(detailSubset)] as const + const detail = { id: `a`, category: `detail`, name: `A` } + const list = { id: `c`, category: `list`, name: `C` } + const commitError = new Error(`Commit failed before publication`) + const queryClient = createQueryClient() + const queryFn = vi + .fn<() => Promise>>() + .mockResolvedValueOnce([detail]) + .mockResolvedValueOnce([list]) + const consoleError = vi.spyOn(console, `error`).mockImplementation(() => {}) + const baseOptions = queryCollectionOptions({ + id, + queryClient, + queryKey: [id], + queryFn, + getKey: (item) => item.id, + syncMode: `on-demand`, + startSync: true, + }) + const originalSync = baseOptions.sync + let failNextCommit = false + const collection = createCollection({ + ...baseOptions, + sync: { + sync: (params: Parameters[0]) => + originalSync.sync({ + ...params, + commit: (signal) => { + if (failNextCommit) { + failNextCommit = false + throw commitError + } + return params.commit(signal) + }, + }), + }, + }) + cleanups.push(async () => { + collection._sync.unloadSubset(detailSubset) + collection._sync.unloadSubset(listSubset) + consoleError.mockRestore() + await collection.cleanup() + queryClient.clear() + }) + + await collection._sync.loadSubset(detailSubset) + expect(itemIds(collection.toArray)).toEqual([detail.id]) + + failNextCommit = true + queryClient.setQueryData(detailKey, [detail, list]) + await vi.waitFor(() => expect(collection.utils.lastError).toBe(commitError)) + expect(itemIds(collection.toArray)).toEqual([detail.id]) + + await collection._sync.loadSubset(listSubset) + expect(itemIds(collection.toArray)).toEqual([detail.id, list.id]) + collection._sync.unloadSubset(listSubset) + await vi.waitFor(() => + expect(itemIds(collection.toArray)).toEqual([detail.id]), + ) + }) + + it(`retires a publishing result when its final subset unloads`, async () => { + const id = `reentrant-result-unload` + const queryKey = [id] as const + const initial = { id: `a`, category: `result`, name: `A` } + const sibling = { id: `b`, category: `result`, name: `B` } + const updated = { ...initial, name: `A from server` } + const orphaned = { id: `c`, category: `result`, name: `C` } + const handlerEntered = createDeferred() + const releaseHandler = createDeferred() + const queryClient = createQueryClient() + const queryFn = vi.fn(() => Promise.resolve([initial, sibling])) + const onUpdate = vi.fn(async () => { + handlerEntered.resolve() + await releaseHandler.promise + return { refetch: false } + }) + const collection = createCollection( + queryCollectionOptions({ + id, + queryClient, + queryKey, + queryFn, + getKey: (item) => item.id, + syncMode: `on-demand`, + startSync: true, + onUpdate, + }), + ) + cleanups.push(async () => { + releaseHandler.resolve() + await collection.cleanup() + queryClient.clear() + }) + + await collection._sync.loadSubset({}) + let unloaded = false + const subscription = collection.subscribeChanges(() => { + if (!unloaded && collection.has(orphaned.id)) { + unloaded = true + collection._sync.unloadSubset({}) + } + }) + cleanups.push(() => Promise.resolve(subscription.unsubscribe())) + + const mutation = collection.update(initial.id, (draft) => { + draft.name = `A optimistic` + }) + await handlerEntered.promise + queryClient.setQueryData(queryKey, [updated, sibling, orphaned]) + expect(itemIds(collection._state.syncedData.values())).toEqual([ + initial.id, + sibling.id, + ]) + + releaseHandler.resolve() + await mutation.isPersisted.promise + await vi.waitFor(() => expect(unloaded).toBe(true)) + await vi.waitFor(() => { + expect({ + synced: itemIds(collection._state.syncedData.values()), + source: itemIds(collection.toArray), + }).toEqual({ synced: [], source: [] }) + }) + + expect(queryFn).toHaveBeenCalledOnce() + expect(onUpdate).toHaveBeenCalledOnce() + expect( + persistedOwners(collection._state.syncedMetadata, orphaned.id), + ).toEqual([]) + }) + + it(`retains post-publication ownership through durable persistence`, async () => { + const id = `persisted-post-application-unload` + const queryKey = [id] as const + const initial = { id: `a`, category: `result`, name: `A` } + const sibling = { id: `b`, category: `result`, name: `B` } + const transient = { id: `c`, category: `result`, name: `C` } + const storage = createOwnershipStorage(undefined, 2) + const { collection, queryClient, queryFn } = + createPersistedOwnershipFixture(id, storage, [initial, sibling]) + const createDerived = () => + createLiveQueryCollection((query) => + query.from({ item: collection }).select(({ item }) => ({ ...item })), + ) + const firstDerived = createDerived() + let secondDerived: ReturnType | undefined + + try { + await firstDerived.preload() + expect(queryFn).toHaveBeenCalledOnce() + expect(itemIds(storage.snapshot().rows.values())).toEqual([ + initial.id, + sibling.id, + ]) + + queryClient.setQueryData(queryKey, [initial, sibling, transient]) + await storage.entered + expect({ + synced: itemIds(collection._state.syncedData.values()), + derived: itemIds(firstDerived.toArray), + stored: itemIds(storage.snapshot().rows.values()), + }).toEqual({ + synced: [initial.id, sibling.id, transient.id], + derived: [initial.id, sibling.id, transient.id], + stored: [initial.id, sibling.id], + }) + + await firstDerived.cleanup() + storage.release() + await vi.waitFor(() => + expect(itemIds(storage.snapshot().rows.values())).toEqual([ + initial.id, + sibling.id, + transient.id, + ]), + ) + + queryClient.removeQueries({ queryKey, exact: true }) + secondDerived = createDerived() + await secondDerived.preload() + await vi.waitFor(() => expect(queryFn).toHaveBeenCalledTimes(2)) + await vi.waitFor(() => { + expect({ + synced: itemIds(collection._state.syncedData.values()), + source: itemIds(collection.toArray), + derived: itemIds(secondDerived?.toArray ?? []), + stored: itemIds(storage.snapshot().rows.values()), + }).toEqual({ + synced: [initial.id, sibling.id], + source: [initial.id, sibling.id], + derived: [initial.id, sibling.id], + stored: [initial.id, sibling.id], + }) + }) + expect( + persistedOwners(storage.snapshot().rowMetadata, transient.id), + ).toEqual([]) + } finally { + storage.release() + await secondDerived?.cleanup() + if (firstDerived.status !== `cleaned-up`) await firstDerived.cleanup() + } + }) + + it(`keeps only the newest cache result when publication reenters application`, async () => { + const id = `reentrant-result-application` + const queryKey = [id] as const + const initial = { id: `a`, category: `result`, name: `A` } + const outer = { id: `b`, category: `result`, name: `B` } + const inner = { id: `c`, category: `result`, name: `C` } + const queryClient = createQueryClient() + const collection = createCollection( + queryCollectionOptions({ + id, + queryClient, + queryKey, + queryFn: () => Promise.resolve([initial]), + getKey: (item) => item.id, + startSync: true, + }), + ) + cleanups.push(async () => { + await collection.cleanup() + queryClient.clear() + }) + + await collection.stateWhenReady() + const publications: Array> = [] + let injected = false + const subscription = collection.subscribeChanges(() => { + const ids = itemIds(collection.toArray) + publications.push(ids) + if (!injected && ids.includes(outer.id)) { + injected = true + queryClient.setQueryData(queryKey, [inner]) + } + }) + cleanups.push(async () => subscription.unsubscribe()) + + queryClient.setQueryData(queryKey, [outer]) + await vi.waitFor(() => expect(itemIds(collection.toArray)).toEqual([`c`])) + + expect(publications).toEqual([[`b`], [`c`]]) + queryClient.setQueryData(queryKey, []) + await vi.waitFor(() => expect(collection.toArray).toEqual([])) + }) + + it(`retains a reentrant newer application failure for subset settlement`, async () => { + const id = `reentrant-result-failure` + const queryKey = [id] as const + const initial = { id: `a`, category: `result`, name: `A` } + const outer = { id: `b`, category: `result`, name: `B` } + const invalid = { id: `invalid`, category: `result`, name: `Invalid` } + const applicationError = new Error(`newest result application failed`) + const queryClient = createQueryClient() + const consoleError = vi.spyOn(console, `error`).mockImplementation(() => {}) + const collection = createCollection( + queryCollectionOptions({ + id, + queryClient, + queryKey, + queryFn: () => Promise.resolve([initial]), + getKey: (item) => { + if (item.id === invalid.id) throw applicationError + return item.id + }, + syncMode: `on-demand`, + startSync: true, + }), + ) + cleanups.push(async () => { + consoleError.mockRestore() + await collection.cleanup() + queryClient.clear() + }) + + await collection._sync.loadSubset({}) + let injected = false + let pendingWaiter: Promise | undefined + const subscription = collection.subscribeChanges(() => { + if (!injected && collection.has(outer.id)) { + injected = true + queryClient.setQueryData(queryKey, [invalid]) + pendingWaiter = expect( + Promise.resolve(collection._sync.loadSubset({})), + ).rejects.toBe(applicationError) + void pendingWaiter.catch(() => {}) + } + }) + cleanups.push(async () => subscription.unsubscribe()) + + queryClient.setQueryData(queryKey, [outer]) + await vi.waitFor(() => expect(collection.utils.errorCount).toBe(1)) + await pendingWaiter + await expect(Promise.resolve(collection._sync.loadSubset({}))).rejects.toBe( + applicationError, + ) + }) + + it(`retires mutation ownership when a later cache result is empty`, async () => { + const id = `mutation-ownership-replacement` + const queryKey = [id] as const + const initial = { id: `a`, category: `mutation`, name: `A` } + const inserted = { id: `b`, category: `mutation`, name: `B` } + const serverRows = [initial] + const queryClient = createQueryClient() + const queryFn = vi.fn(() => Promise.resolve(structuredClone(serverRows))) + const collection = createCollection( + queryCollectionOptions({ + id, + queryClient, + queryKey, + queryFn, + getKey: (item) => item.id, + startSync: true, + onInsert: ({ transaction }) => { + serverRows.push( + ...transaction.mutations.map(({ modified }) => + structuredClone(modified), + ), + ) + return Promise.resolve() + }, + }), + ) + cleanups.push(async () => { + await collection.cleanup() + queryClient.clear() + }) + + await collection.stateWhenReady() + const mutation = collection.insert(inserted) + await mutation.isPersisted.promise + expect({ + cache: itemIds(queryClient.getQueryData>(queryKey) ?? []), + synced: itemIds(collection._state.syncedData.values()), + source: itemIds(collection.toArray), + }).toEqual({ cache: [`a`, `b`], synced: [`a`, `b`], source: [`a`, `b`] }) + + queryClient.setQueryData(queryKey, []) + await vi.waitFor(() => { + expect({ + cache: itemIds(queryClient.getQueryData>(queryKey) ?? []), + synced: itemIds(collection._state.syncedData.values()), + source: itemIds(collection.toArray), + }).toEqual({ cache: [], synced: [], source: [] }) + }) + }) + + it(`publishes an authoritative delete after its mutation refetch`, async () => { + const id = `mutation-delete-publication` + const queryKey = [id] as const + const initial = { id: `a`, category: `mutation`, name: `A` } + const serverRows = [initial] + const queryClient = createQueryClient() + const preexistingRefetchResult = createDeferred>() + const mutationRefetchResult = createDeferred>() + const persistenceGate = createDeferred() + const metadata: MetadataRecorder = { rows: new Map(), writes: [] } + const commitRequests: Array<{ + writes: Array + receipt: `immediate` | `pending` + outcome?: `applied` | `aborted` | `rejected` + }> = [] + const queryFn = vi + .fn<() => Promise>>() + .mockResolvedValueOnce(structuredClone(serverRows)) + .mockReturnValueOnce(preexistingRefetchResult.promise) + .mockReturnValueOnce(mutationRefetchResult.promise) + const baseOptions = queryCollectionOptions({ + id, + queryClient, + queryKey, + queryFn, + getKey: (item) => item.id, + startSync: true, + onDelete: async ({ transaction }) => { + await persistenceGate.promise + const deleted = new Set( + transaction.mutations.map(({ original }) => original.id), + ) + serverRows.splice( + 0, + serverRows.length, + ...serverRows.filter(({ id: rowId }) => !deleted.has(rowId)), + ) + }, + }) + const originalSync = baseOptions.sync + const collection = createCollection({ + ...baseOptions, + sync: { + sync: (params: Parameters[0]) => { + let writes: Array = [] + return originalSync.sync({ + ...params, + metadata: recordMetadata(params.metadata!, metadata), + begin: () => { + writes = [] + params.begin() + }, + write: (change) => { + writes.push(change.type) + params.write(change) + }, + commit: (signal) => { + const result = params.commit(signal) + const request: (typeof commitRequests)[number] = { + writes: [...writes], + receipt: result === true ? `immediate` : `pending`, + } + commitRequests.push(request) + if (result === true) { + request.outcome = `applied` + } else { + void result.then( + () => { + request.outcome = `applied` + }, + (error) => { + request.outcome = isCancelledError(error) + ? `aborted` + : `rejected` + }, + ) + } + return result + }, + }) + }, + }, + }) + const derived = createLiveQueryCollection((query) => + query.from({ item: collection }).select(({ item }) => ({ + id: item.id, + category: item.category, + name: item.name, + })), + ) + const capture = (): MutationLifecycleSnapshot => ({ + server: itemIds(serverRows), + cache: itemIds(queryClient.getQueryData>(queryKey) ?? []), + synced: itemIds(collection._state.syncedData.values()), + source: itemIds(collection.toArray), + derived: itemIds(derived.toArray), + }) + const publications: Array = [] + const subscription = derived.subscribeChanges(() => { + publications.push(capture()) + }) + cleanups.push(async () => { + persistenceGate.resolve() + preexistingRefetchResult.resolve([]) + mutationRefetchResult.resolve([]) + subscription.unsubscribe() + await derived.cleanup() + await collection.cleanup() + queryClient.clear() + }) + + await derived.preload() + expectMutationLifecycleSnapshot( + capture(), + [initial.id], + [initial.id], + false, + ) + + const preexistingRefetch = collection.utils.refetch({ throwOnError: true }) + void preexistingRefetch.catch(() => undefined) + await vi.waitFor(() => expect(queryFn).toHaveBeenCalledTimes(2)) + + const mutation = collection.delete(initial.id) + expect(itemIds(collection.toArray)).toEqual([]) + expect(itemIds(derived.toArray)).toEqual([]) + + persistenceGate.resolve() + await vi.waitFor(() => expect(queryFn).toHaveBeenCalledTimes(3)) + mutationRefetchResult.resolve([]) + await mutation.isPersisted.promise + + expect(queryFn).toHaveBeenCalledTimes(3) + expect(commitRequests.some(({ writes }) => writes.includes(`delete`))).toBe( + true, + ) + expect(commitRequests.some(({ receipt }) => receipt === `pending`)).toBe( + true, + ) + expect(metadata.writes).toContainEqual({ type: `delete`, key: initial.id }) + await vi.waitFor(() => + expect(commitRequests.every(({ outcome }) => outcome !== undefined)).toBe( + true, + ), + ) + expectMutationLifecycleSnapshot(capture(), [], [], false) + expect(publications.at(-1)?.source).toEqual([]) + expect(publications.at(-1)?.derived).toEqual([]) + }) + it(`keeps cached rows until the final exact acquisition is released`, async () => { const { collection, queryFn } = createOwnershipFixture({ id: `shared-acquisition`, @@ -959,6 +2205,54 @@ describe(`query collection ownership lifecycle`, () => { expectColdOwnerRevalidation(await observeColdOwnerRevalidation()) }) + it(`does not publish a superseded result after its persisted scan resolves`, async () => { + const id = `superseded-retained-scan` + const queryKey = [id] as const + const queryHash = hashKey(queryKey) + const stale = { id: `stale`, category: `retained`, name: `Stale` } + const fresh = { id: `fresh`, category: `retained`, name: `Fresh` } + const firstScan = + createDeferred< + Array<{ key: string | number; value: Item; metadata?: unknown }> + >() + const scanPersisted = vi + .fn() + .mockReturnValueOnce(firstScan.promise) + .mockResolvedValue([]) + const { collection, queryFn } = createOwnershipFixture({ + id, + results: [[stale], [fresh]], + syncMode: `eager`, + scanPersisted, + setupMetadata: (metadata) => { + metadata.collection.set(`queryCollection:gc:${queryHash}`, { + queryHash, + mode: `until-revalidated`, + }) + }, + }) + const publications: Array> = [] + const subscription = collection.subscribeChanges(() => { + publications.push(itemIds(collection.toArray)) + }) + cleanups.push(() => { + subscription.unsubscribe() + return Promise.resolve() + }) + + await vi.waitFor(() => expect(scanPersisted).toHaveBeenCalledOnce()) + expect(queryFn).toHaveBeenCalledOnce() + const refetch = collection.utils.refetch({ throwOnError: true }) + await vi.waitFor(() => expect(queryFn).toHaveBeenCalledTimes(2)) + await refetch + firstScan.resolve([]) + + await vi.waitFor(() => { + expect(itemIds(collection.toArray)).toEqual([fresh.id]) + }) + expect(publications).not.toContainEqual([stale.id]) + }) + it(`rejects emitted ownership that is absent from cold storage`, async () => { const observations = await observeColdOwnerRevalidation(true) expect(() => expectColdOwnerRevalidation(observations)).toThrow()