From 5f110920ca36373ccd54d0ecaca46be27ad28e96 Mon Sep 17 00:00:00 2001 From: Kyle Mathews Date: Wed, 16 Sep 2026 15:47:42 -0600 Subject: [PATCH 01/12] fix: publish Query Collection mutation results --- packages/db/src/collection/index.ts | 28 +- packages/db/src/collection/mutations.ts | 19 +- .../db/tests/collection-lifecycle.test.ts | 344 +++++++++++++- packages/query-db-collection/src/query.ts | 128 ++++- .../tests/ownership-lifecycle.oracle.test.ts | 437 +++++++++++++++++- 5 files changed, 923 insertions(+), 33 deletions(-) diff --git a/packages/db/src/collection/index.ts b/packages/db/src/collection/index.ts index 68c4a968c1..918cc51184 100644 --- a/packages/db/src/collection/index.ts +++ b/packages/db/src/collection/index.ts @@ -59,17 +59,22 @@ type CollectionSyncConfigWithFactory = TSync & { readonly [collectionSyncConfigFactory]: ( this: TSync, utilities: object, + startSync: () => void, ) => TSync } -/** @internal Lets adapters bind a sync config to each collection instance. */ +/** + * @internal Lets adapters bind a sync config to each collection instance. + * The factory may retain `startSync` for later use, but must not call it during + * materialization before Collection construction has completed. + */ export function withCollectionSyncConfigFactory( sync: TSync, - factory: (source: TSync, utilities: object) => TSync, + factory: (source: TSync, utilities: object, startSync: () => void) => TSync, ): CollectionSyncConfigWithFactory { Object.defineProperty(sync, collectionSyncConfigFactory, { - value(this: TSync, utilities: object) { - return factory(this, utilities) + value(this: TSync, utilities: object, startSync: () => void) { + return factory(this, utilities, startSync) }, // Preserve the hook when callers wrap a sync config with object spread. enumerable: true, @@ -92,7 +97,11 @@ export function withCollectionSyncConfigCleanup( function materializeCollectionSyncConfig< TSync extends object, TUtils extends object, ->(sync: TSync, utilities: TUtils): { sync: TSync; utilities: TUtils } { +>( + sync: TSync, + utilities: TUtils, + startSync: () => void, +): { sync: TSync; utilities: TUtils } { const factory = ( sync as unknown as Partial> )[collectionSyncConfigFactory] @@ -104,7 +113,10 @@ function materializeCollectionSyncConfig< Object.getPrototypeOf(utilities), Object.getOwnPropertyDescriptors(utilities), ) as TUtils - return { sync: factory.call(sync, ownedUtilities), utilities: ownedUtilities } + return { + sync: factory.call(sync, ownedUtilities, startSync), + utilities: ownedUtilities, + } } function cleanupCollectionSyncConfig(sync: object): void { @@ -396,7 +408,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 ?? {}, () => + 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 f230a51a98..c5204619aa 100644 --- a/packages/db/src/collection/mutations.ts +++ b/packages/db/src/collection/mutations.ts @@ -76,6 +76,10 @@ export class CollectionMutationsManager< : getActiveTransaction() } + private startSyncForMutation(): void { + if (this.lifecycle.status === `idle`) this.collection._sync.startSync() + } + private createTransaction(config: TransactionConfig) { return this.transactionScope ? this.transactionScope.createTransaction(config) @@ -206,9 +210,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 +245,13 @@ export class CollectionMutationsManager< mutations.push(mutation) }) + this.startSyncForMutation() + for (const mutation of mutations) { + if (this.state.has(mutation.key)) { + throw new DuplicateKeyError(mutation.key) + } + } + // If an ambient transaction exists, use it if (ambientTransaction) { ambientTransaction.applyMutations(mutations) @@ -316,6 +327,8 @@ export class CollectionMutationsManager< throw new NoKeysPassedToUpdateError() } + this.startSyncForMutation() + const callback = typeof configOrCallback === `function` ? configOrCallback : maybeCallback! const config = @@ -497,6 +510,8 @@ export class CollectionMutationsManager< } const keysArray = Array.isArray(keys) ? keys : [keys] + this.startSyncForMutation() + const mutations: Array< PendingMutation< TOutput, diff --git a/packages/db/tests/collection-lifecycle.test.ts b/packages/db/tests/collection-lifecycle.test.ts index 7779c08e75..bb0b0a88d2 100644 --- a/packages/db/tests/collection-lifecycle.test.ts +++ b/packages/db/tests/collection-lifecycle.test.ts @@ -1,7 +1,19 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { z } from 'zod' import { createCollection } from '../src/collection/index.js' import { CleanupQueue } from '../src/collection/cleanup-queue.js' -import { InvalidCollectionStatusTransitionError } from '../src/errors.js' +import { + DuplicateKeyError, + InvalidCollectionStatusTransitionError, + InvalidKeyError, + MissingDeleteHandlerError, + MissingInsertHandlerError, + MissingUpdateHandlerError, + NoKeysPassedToDeleteError, + NoKeysPassedToUpdateError, + SchemaValidationError, + UndefinedKeyError, +} from '../src/errors.js' import { getActivePublicationContext, transactionScopedScheduler, @@ -24,6 +36,336 @@ function getChangesManager(collection: object): { } describe(`Collection Lifecycle Management`, () => { + it(`does not start idle sync when an insert has no handler`, async () => { + type Row = { id: string; value: string } + let syncStarts = 0 + const collection = createCollection({ + id: `rejected-mutation-missing-handler`, + getKey: (row) => row.id, + startSync: false, + sync: { + sync: ({ markReady }) => { + syncStarts++ + markReady() + }, + }, + }) + + try { + expect(() => + collection.insert({ id: `rejected`, value: `rejected` }), + ).toThrow(MissingInsertHandlerError) + expect(syncStarts).toBe(0) + expect(collection.status).toBe(`idle`) + } finally { + await collection.cleanup() + } + }) + + it(`does not start idle sync when insert schema validation rejects`, async () => { + let syncStarts = 0 + const collection = createCollection({ + id: `rejected-mutation-schema`, + getKey: (row) => row.id, + schema: z.object({ id: z.string(), value: z.string().min(1) }), + startSync: false, + sync: { + sync: ({ markReady }) => { + syncStarts++ + markReady() + }, + }, + onInsert: async () => {}, + }) + + try { + expect(() => collection.insert({ id: `rejected`, value: `` })).toThrow( + SchemaValidationError, + ) + expect(syncStarts).toBe(0) + expect(collection.status).toBe(`idle`) + } finally { + await collection.cleanup() + } + }) + + it(`does not start idle sync when insert key validation rejects`, async () => { + type Row = { id: string; value: string } + let syncStarts = 0 + const collection = createCollection({ + id: `rejected-mutation-invalid-key`, + getKey: () => true as never, + startSync: false, + sync: { + sync: ({ markReady }) => { + syncStarts++ + markReady() + }, + }, + onInsert: async () => {}, + }) + + try { + expect(() => + collection.insert({ id: `rejected`, value: `rejected` }), + ).toThrow(InvalidKeyError) + expect(syncStarts).toBe(0) + expect(collection.status).toBe(`idle`) + } finally { + await collection.cleanup() + } + }) + + it(`does not start idle sync when an insert key is missing`, async () => { + type Row = { id: string; value: string } + let syncStarts = 0 + const collection = createCollection({ + id: `rejected-mutation-missing-key`, + getKey: () => undefined as never, + startSync: false, + sync: { + sync: ({ markReady }) => { + syncStarts++ + markReady() + }, + }, + onInsert: async () => {}, + }) + + try { + expect(() => + collection.insert({ id: `rejected`, value: `rejected` }), + ).toThrow(UndefinedKeyError) + expect(syncStarts).toBe(0) + expect(collection.status).toBe(`idle`) + } finally { + await collection.cleanup() + } + }) + + it(`starts idle sync exactly once when a mutation is accepted`, async () => { + type Row = { id: string; value: string } + let syncStarts = 0 + const collection = createCollection({ + id: `mutation-starts-idle-sync`, + getKey: (row) => row.id, + startSync: false, + sync: { + sync: ({ markReady }) => { + syncStarts++ + markReady() + }, + }, + onInsert: async () => {}, + }) + + try { + 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(collection.toArray.map(({ id }) => id).sort()).toEqual([ + `first`, + `second`, + ]) + } finally { + await collection.cleanup() + } + }) + + it.each([`update`, `delete`] as const)( + `does not start idle sync when %s has no handler`, + async (operation) => { + type Row = { id: string; value: string } + let syncStarts = 0 + const collection = createCollection({ + id: `rejected-${operation}-missing-handler`, + getKey: (row) => row.id, + startSync: false, + sync: { + sync: ({ markReady }) => { + syncStarts++ + markReady() + }, + }, + }) + + try { + const mutate = () => + operation === `update` + ? collection.update(`target`, (draft) => { + draft.value = `updated` + }) + : collection.delete(`target`) + + expect(mutate).toThrow( + operation === `update` + ? MissingUpdateHandlerError + : MissingDeleteHandlerError, + ) + expect(syncStarts).toBe(0) + expect(collection.status).toBe(`idle`) + } finally { + await collection.cleanup() + } + }, + ) + + it.each([`update`, `delete`] as const)( + `does not start idle sync when %s receives no keys`, + async (operation) => { + type Row = { id: string; value: string } + let syncStarts = 0 + const collection = createCollection({ + id: `rejected-${operation}-empty-keys`, + getKey: (row) => row.id, + startSync: false, + sync: { + sync: ({ markReady }) => { + syncStarts++ + markReady() + }, + }, + onUpdate: async () => {}, + onDelete: async () => {}, + }) + + try { + const mutate = () => + operation === `update` + ? collection.update([], () => {}) + : collection.delete([]) + + expect(mutate).toThrow( + operation === `update` + ? NoKeysPassedToUpdateError + : NoKeysPassedToDeleteError, + ) + expect(syncStarts).toBe(0) + expect(collection.status).toBe(`idle`) + } finally { + await collection.cleanup() + } + }, + ) + + it.each([`update`, `delete`] as const)( + `starts idle sync before %s checks collection state`, + async (operation) => { + type Row = { id: string; value: string } + const target = { id: `target`, value: `original` } + let syncStarts = 0 + const collection = createCollection({ + id: `accepted-${operation}-hydrates-target`, + getKey: (row) => row.id, + startSync: false, + sync: { + sync: ({ begin, write, commit, markReady }) => { + syncStarts++ + begin() + write({ type: `insert`, value: target }) + commit() + markReady() + }, + }, + onUpdate: async () => {}, + onDelete: async () => {}, + }) + + try { + 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.map(({ key, type }) => ({ key, type })), + ).toEqual([{ key: `target`, type: operation }]) + await transaction.isPersisted.promise + } finally { + await collection.cleanup() + } + }, + ) + + it(`does not start idle sync when an insert batch has duplicate keys`, async () => { + type Row = { id: string; value: string } + let syncStarts = 0 + const collection = createCollection({ + id: `rejected-insert-batch-duplicate`, + getKey: (row) => row.id, + startSync: false, + sync: { + sync: ({ markReady }) => { + syncStarts++ + markReady() + }, + }, + onInsert: async () => {}, + }) + + try { + expect(() => + collection.insert([ + { id: `duplicate`, value: `first` }, + { id: `duplicate`, value: `second` }, + ]), + ).toThrow(DuplicateKeyError) + expect(syncStarts).toBe(0) + expect(collection.status).toBe(`idle`) + } finally { + await collection.cleanup() + } + }) + + it(`checks hydrated collection keys before admitting an idle insert`, async () => { + type Row = { id: string; value: string } + const target = { id: `target`, value: `synced` } + const onInsert = vi.fn(async () => {}) + let syncStarts = 0 + const collection = createCollection({ + id: `idle-insert-hydrated-duplicate`, + getKey: (row) => row.id, + startSync: false, + sync: { + sync: ({ begin, write, commit, markReady }) => { + syncStarts++ + begin() + write({ type: `insert`, value: target }) + commit() + markReady() + }, + }, + onInsert, + }) + + try { + expect(() => collection.insert({ id: `target`, value: `local` })).toThrow( + DuplicateKeyError, + ) + expect(syncStarts).toBe(1) + expect(collection.status).toBe(`ready`) + expect(collection.toArray.map(({ id, value }) => ({ id, value }))).toEqual( + [target], + ) + expect(onInsert).not.toHaveBeenCalled() + } finally { + await collection.cleanup() + } + }) + it.each( ([`same`, `missing`, `changed`, `empty`] as const).flatMap((shape) => ([`atomic`, `split`] as const).map((delivery) => ({ shape, delivery })), diff --git a/packages/query-db-collection/src/query.ts b/packages/query-db-collection/src/query.ts index 3c2e45933a..20a88c0c73 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, @@ -917,6 +918,7 @@ export function queryCollectionOptions( // Eager startup holds one reference until cleanup. Cache removal detaches // observation, not that ownership or its rows. let ensureEagerSubscription = () => {} + let replaceMutationResultApplications = () => {} const addRowOwner = (rowKey: string | number, hashedQueryKey: string) => { const owners = rowToQueries.get(rowKey) || new Set() @@ -995,7 +997,8 @@ export function queryCollectionOptions( const retainedQueriesPendingRevalidation = new Set() const pendingResultApplications = new Map>() const failedResultApplications = new Map() - const resultApplicationTokens = new Map() + const retainedResultApplicationTokens = new Map() + const activeResultApplicationTokens = new Map() const resultApplicationControllers = new Map>() const effectivePersistedGcTimes = new Map() const persistedRetentionTimers = new Map< @@ -1007,7 +1010,8 @@ export function queryCollectionOptions( const invalidatePendingResultApplication = (hashedQueryKey: string) => { pendingResultApplications.delete(hashedQueryKey) failedResultApplications.delete(hashedQueryKey) - resultApplicationTokens.delete(hashedQueryKey) + retainedResultApplicationTokens.delete(hashedQueryKey) + activeResultApplicationTokens.delete(hashedQueryKey) resultApplicationControllers .get(hashedQueryKey) ?.forEach((controller) => controller.abort()) @@ -1589,6 +1593,7 @@ export function queryCollectionOptions( const applySuccessfulResult = async ( queryKey: QueryKey, result: QueryObserverResult, + applicationToken: object, persistedBaseline?: Map< string | number, { @@ -1652,6 +1657,14 @@ export function queryCollectionOptions( const restoreOwnershipTracking = () => { if (!state.observers.has(hashedQueryKey)) return + const currentApplicationToken = + activeResultApplicationTokens.get(hashedQueryKey) + if ( + currentApplicationToken !== undefined && + currentApplicationToken !== applicationToken + ) { + return + } if (previousOwnedRows === undefined) { queryToRows.delete(hashedQueryKey) @@ -1764,11 +1777,17 @@ export function queryCollectionOptions( await loadPersistedBaselineForQuery(hashedQueryKey) if ( collection.status === `cleaned-up` || - resultApplicationTokens.get(hashedQueryKey) !== applicationToken + retainedResultApplicationTokens.get(hashedQueryKey) !== applicationToken ) { return } - await applySuccessfulResult(queryKey, result, persistedBaseline, signal) + await applySuccessfulResult( + queryKey, + result, + applicationToken, + persistedBaseline, + signal, + ) } const trackResultApplication = ( @@ -1806,15 +1825,20 @@ export function queryCollectionOptions( const enqueueResultApplication = ( hashedQueryKey: string, - apply: (signal: AbortSignal) => Promise, + apply: (signal: AbortSignal, applicationToken: object) => Promise, + observedApplicationToken?: object, ): void => { const controller = new AbortController() + const applicationToken = observedApplicationToken ?? {} 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 run = () => { + activeResultApplicationTokens.set(hashedQueryKey, applicationToken) + return apply(controller.signal, applicationToken) + } const application = previousApplication ? previousApplication.then(run, run) : run() @@ -1913,18 +1937,32 @@ export function queryCollectionOptions( } const applicationToken = {} - resultApplicationTokens.set(hashedQueryKey, applicationToken) - enqueueResultApplication(hashedQueryKey, (signal) => - reconcileSuccessfulResult( - queryKey, - result, - applicationToken, - signal, - ), + retainedResultApplicationTokens.set( + hashedQueryKey, + applicationToken, + ) + enqueueResultApplication( + hashedQueryKey, + (signal, token) => + reconcileSuccessfulResult( + queryKey, + result, + token, + signal, + ), + applicationToken, ) } else { - enqueueResultApplication(hashedQueryKey, (signal) => - applySuccessfulResult(queryKey, result, undefined, signal), + enqueueResultApplication( + hashedQueryKey, + (signal, token) => + applySuccessfulResult( + queryKey, + result, + token, + undefined, + signal, + ), ) } } else { @@ -1960,6 +1998,26 @@ export function queryCollectionOptions( return handleQueryResult } + replaceMutationResultApplications = () => { + state.observers.forEach((observer, hashedQueryKey) => { + const result = observer.getCurrentResult() + if ( + !result.isSuccess || + result.isFetching || + !hasPostWriteAuthority(hashedQueryKey, observer.getCurrentQuery()) + ) { + return + } + + // Earlier snapshots can be waiting behind this persisting mutation. + // Replace them with the completed refetch so its sync transaction is + // staged before the mutation handler returns; awaiting it here would + // deadlock on the mutation that owns the publication barrier. + invalidatePendingResultApplication(hashedQueryKey) + makeQueryResultHandler(hashToQueryKey.get(hashedQueryKey)!)(result) + }) + } + const isSubscribed = (hashedQueryKey: string) => { return unsubscribes.has(hashedQueryKey) } @@ -2261,6 +2319,7 @@ export function queryCollectionOptions( const cleanup = () => { pendingStartupLoads.clear() ensureEagerSubscription = () => {} + replaceMutationResultApplications = () => {} unsubscribeFromCollectionEvents() unsubscribeFromQueries() persistedRetentionTimers.forEach((timer) => { @@ -2398,6 +2457,11 @@ export function queryCollectionOptions( return Promise.all(refetchPromises) } + const refetchAfterMutation = async () => { + await refetch() + replaceMutationResultApplications() + } + /** * Updates a single query key in the cache with new items, handling both direct arrays * and wrapped response formats (when `select` is used). @@ -2729,7 +2793,7 @@ export function queryCollectionOptions( (handlerResult as { refetch?: boolean }).refetch !== false if (shouldRefetch) { - await refetch() + await refetchAfterMutation() } return handlerResult @@ -2743,7 +2807,7 @@ export function queryCollectionOptions( (handlerResult as { refetch?: boolean }).refetch !== false if (shouldRefetch) { - await refetch() + await refetchAfterMutation() } return handlerResult @@ -2757,7 +2821,7 @@ export function queryCollectionOptions( (handlerResult as { refetch?: boolean }).refetch !== false if (shouldRefetch) { - await refetch() + await refetchAfterMutation() } return handlerResult @@ -2767,11 +2831,35 @@ 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, startSync) => { + const boundUtilities = utilities as Record< + string, + (...args: Array) => any + > + for (const name of [ + `writeInsert`, + `writeUpdate`, + `writeDelete`, + `writeUpsert`, + `writeBatch`, + ]) { + const write = boundUtilities[name]! + boundUtilities[name] = (...args) => { + startSync() + 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 e2f325c736..03c3da263e 100644 --- a/packages/query-db-collection/tests/ownership-lifecycle.oracle.test.ts +++ b/packages/query-db-collection/tests/ownership-lifecycle.oracle.test.ts @@ -14,6 +14,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 +32,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 +58,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 +133,7 @@ function createOwnershipFixture({ syncMode = `on-demand`, metadataRecorder, setupMetadata, + scanPersisted, customHash, staleTime, }: OwnershipFixtureOptions): OwnershipFixture { @@ -137,7 +155,7 @@ function createOwnershipFixture({ const originalSync = baseOptions.sync let pendingSetup = setupMetadata const collection = createCollection( - metadataRecorder || setupMetadata + metadataRecorder || setupMetadata || scanPersisted ? { ...baseOptions, sync: { @@ -148,15 +166,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 +200,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, @@ -465,6 +520,335 @@ 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.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(`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(`keeps cached rows until the final exact acquisition is released`, async () => { const { collection, queryFn } = createOwnershipFixture({ id: `shared-acquisition`, @@ -959,6 +1343,53 @@ 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() From eff30c02f95d93071472d06020a5ef6dc5fde5f6 Mon Sep 17 00:00:00 2001 From: Kyle Mathews Date: Wed, 16 Sep 2026 15:48:02 -0600 Subject: [PATCH 02/12] chore: add Query Collection lifecycle changeset --- .changeset/fix-query-collection-lifecycle.md | 6 ++++++ 1 file changed, 6 insertions(+) create mode 100644 .changeset/fix-query-collection-lifecycle.md diff --git a/.changeset/fix-query-collection-lifecycle.md b/.changeset/fix-query-collection-lifecycle.md new file mode 100644 index 0000000000..0bea5f8af5 --- /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 mutation validation succeeds, and publish authoritative Query Collection refetch results without stale intermediate snapshots. From e651f6c9e6b3431068934a6d6f231308e86a648e Mon Sep 17 00:00:00 2001 From: "autofix-ci[bot]" <114827586+autofix-ci[bot]@users.noreply.github.com> Date: Wed, 16 Sep 2026 21:53:55 +0000 Subject: [PATCH 03/12] ci: apply automated fixes --- .../db/tests/collection-lifecycle.test.ts | 6 +++--- packages/query-db-collection/src/query.ts | 19 +++---------------- .../tests/ownership-lifecycle.oracle.test.ts | 11 +++++------ 3 files changed, 11 insertions(+), 25 deletions(-) diff --git a/packages/db/tests/collection-lifecycle.test.ts b/packages/db/tests/collection-lifecycle.test.ts index bb0b0a88d2..d74e29662c 100644 --- a/packages/db/tests/collection-lifecycle.test.ts +++ b/packages/db/tests/collection-lifecycle.test.ts @@ -357,9 +357,9 @@ describe(`Collection Lifecycle Management`, () => { ) expect(syncStarts).toBe(1) expect(collection.status).toBe(`ready`) - expect(collection.toArray.map(({ id, value }) => ({ id, value }))).toEqual( - [target], - ) + expect( + collection.toArray.map(({ id, value }) => ({ id, value })), + ).toEqual([target]) expect(onInsert).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 20a88c0c73..550719a660 100644 --- a/packages/query-db-collection/src/query.ts +++ b/packages/query-db-collection/src/query.ts @@ -1944,25 +1944,12 @@ export function queryCollectionOptions( enqueueResultApplication( hashedQueryKey, (signal, token) => - reconcileSuccessfulResult( - queryKey, - result, - token, - signal, - ), + reconcileSuccessfulResult(queryKey, result, token, signal), applicationToken, ) } else { - enqueueResultApplication( - hashedQueryKey, - (signal, token) => - applySuccessfulResult( - queryKey, - result, - token, - undefined, - signal, - ), + enqueueResultApplication(hashedQueryKey, (signal, token) => + applySuccessfulResult(queryKey, result, token, undefined, signal), ) } } else { 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 03c3da263e..62e505fb58 100644 --- a/packages/query-db-collection/tests/ownership-lifecycle.oracle.test.ts +++ b/packages/query-db-collection/tests/ownership-lifecycle.oracle.test.ts @@ -720,9 +720,7 @@ describe(`query collection ownership lifecycle`, () => { name: `3-third`, } const allVisibleIds = [...visibleIds, thirdItem.id] - expectedPublication = ordered - ? allVisibleIds - : [...allVisibleIds].sort() + expectedPublication = ordered ? allVisibleIds : [...allVisibleIds].sort() const third = collection.insert(thirdItem) expect(persistenceGates).toHaveLength(3) expectMutationLifecycleSnapshot( @@ -1349,9 +1347,10 @@ describe(`query collection ownership lifecycle`, () => { 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 firstScan = + createDeferred< + Array<{ key: string | number; value: Item; metadata?: unknown }> + >() const scanPersisted = vi .fn() .mockReturnValueOnce(firstScan.promise) From 1abb3ea5f2d63d2026e0ce5993a7abf338cf79cd Mon Sep 17 00:00:00 2001 From: Kyle Mathews Date: Wed, 16 Sep 2026 17:39:48 -0600 Subject: [PATCH 04/12] fix(query-db-collection): supersede stale result applications --- packages/db/src/collection/index.ts | 30 +- packages/db/src/collection/mutations.ts | 23 +- .../db/tests/collection-lifecycle.test.ts | 62 +++ packages/query-db-collection/src/query.ts | 144 +++--- .../tests/ownership-lifecycle.oracle.test.ts | 467 ++++++++++++++++++ 5 files changed, 610 insertions(+), 116 deletions(-) diff --git a/packages/db/src/collection/index.ts b/packages/db/src/collection/index.ts index 918cc51184..caf93721de 100644 --- a/packages/db/src/collection/index.ts +++ b/packages/db/src/collection/index.ts @@ -57,25 +57,23 @@ const collectionSyncConfigCleanup: unique symbol = Symbol.for( type CollectionSyncConfigWithFactory = TSync & { readonly [collectionSyncConfigFactory]: ( - this: TSync, + source: TSync, utilities: object, - startSync: () => void, + startSyncIfIdle: () => void, ) => TSync } -/** - * @internal Lets adapters bind a sync config to each collection instance. - * The factory may retain `startSync` for later use, but must not call it during - * materialization before Collection construction has completed. - */ +/** @internal The factory must defer `startSyncIfIdle` until construction ends. */ export function withCollectionSyncConfigFactory( sync: TSync, - factory: (source: TSync, utilities: object, startSync: () => void) => TSync, + factory: ( + source: TSync, + utilities: object, + startSyncIfIdle: () => void, + ) => TSync, ): CollectionSyncConfigWithFactory { Object.defineProperty(sync, collectionSyncConfigFactory, { - value(this: TSync, utilities: object, startSync: () => void) { - return factory(this, utilities, startSync) - }, + value: factory, // Preserve the hook when callers wrap a sync config with object spread. enumerable: true, }) @@ -100,7 +98,7 @@ function materializeCollectionSyncConfig< >( sync: TSync, utilities: TUtils, - startSync: () => void, + startSyncIfIdle: () => void, ): { sync: TSync; utilities: TUtils } { const factory = ( sync as unknown as Partial> @@ -114,7 +112,7 @@ function materializeCollectionSyncConfig< Object.getOwnPropertyDescriptors(utilities), ) as TUtils return { - sync: factory.call(sync, ownedUtilities, startSync), + sync: factory(sync, ownedUtilities, startSyncIfIdle), utilities: ownedUtilities, } } @@ -408,9 +406,9 @@ export class CollectionImpl< // Set default values for optional config properties const { sync: collectionSync, utilities: collectionUtils } = - materializeCollectionSyncConfig(config.sync, config.utils ?? {}, () => - this._sync.startSync(), - ) + 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 c5204619aa..a678bcd436 100644 --- a/packages/db/src/collection/mutations.ts +++ b/packages/db/src/collection/mutations.ts @@ -76,10 +76,6 @@ export class CollectionMutationsManager< : getActiveTransaction() } - private startSyncForMutation(): void { - if (this.lifecycle.status === `idle`) this.collection._sync.startSync() - } - private createTransaction(config: TransactionConfig) { return this.transactionScope ? this.transactionScope.createTransaction(config) @@ -245,12 +241,9 @@ export class CollectionMutationsManager< mutations.push(mutation) }) - this.startSyncForMutation() - for (const mutation of mutations) { - if (this.state.has(mutation.key)) { - throw new DuplicateKeyError(mutation.key) - } - } + this.collection._sync.startSync() + const duplicate = mutations.find(({ key }) => this.state.has(key)) + if (duplicate) throw new DuplicateKeyError(duplicate.key) // If an ambient transaction exists, use it if (ambientTransaction) { @@ -327,13 +320,14 @@ export class CollectionMutationsManager< throw new NoKeysPassedToUpdateError() } - this.startSyncForMutation() - 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) @@ -510,8 +504,7 @@ export class CollectionMutationsManager< } const keysArray = Array.isArray(keys) ? keys : [keys] - this.startSyncForMutation() - + this.collection._sync.startSync() const mutations: Array< PendingMutation< TOutput, diff --git a/packages/db/tests/collection-lifecycle.test.ts b/packages/db/tests/collection-lifecycle.test.ts index d74e29662c..e9a485ce82 100644 --- a/packages/db/tests/collection-lifecycle.test.ts +++ b/packages/db/tests/collection-lifecycle.test.ts @@ -180,6 +180,35 @@ describe(`Collection Lifecycle Management`, () => { } }) + it(`surfaces idle sync startup errors before applying a mutation`, async () => { + type Row = { id: string; value: string } + const startupError = new Error(`sync startup failed`) + let syncStarts = 0 + const collection = createCollection({ + id: `mutation-sync-startup-error`, + getKey: (row) => row.id, + startSync: false, + sync: { + sync: () => { + syncStarts++ + throw startupError + }, + }, + onInsert: async () => {}, + }) + + try { + expect(() => + collection.insert({ id: `rejected`, value: `rejected` }), + ).toThrow(startupError) + expect(syncStarts).toBe(1) + expect(collection.status).toBe(`error`) + expect(collection.size).toBe(0) + } finally { + await collection.cleanup() + } + }) + it.each([`update`, `delete`] as const)( `does not start idle sync when %s has no handler`, async (operation) => { @@ -256,6 +285,39 @@ describe(`Collection Lifecycle Management`, () => { }, ) + it(`does not start idle sync when update receives no callback`, async () => { + type Row = { id: string; value: string } + let syncStarts = 0 + const collection = createCollection({ + id: `rejected-update-missing-callback`, + getKey: (row) => row.id, + startSync: false, + sync: { + sync: ({ begin, write, commit, markReady }) => { + syncStarts++ + begin() + write({ type: `insert`, value: { id: `target`, value: `original` } }) + commit() + markReady() + }, + }, + onUpdate: async () => {}, + }) + + try { + expect(() => + (collection.update as (key: string, config: object) => unknown)( + `target`, + {}, + ), + ).toThrow(TypeError) + expect(syncStarts).toBe(0) + expect(collection.status).toBe(`idle`) + } finally { + await collection.cleanup() + } + }) + it.each([`update`, `delete`] as const)( `starts idle sync before %s checks collection state`, async (operation) => { diff --git a/packages/query-db-collection/src/query.ts b/packages/query-db-collection/src/query.ts index 550719a660..7c4233ceb2 100644 --- a/packages/query-db-collection/src/query.ts +++ b/packages/query-db-collection/src/query.ts @@ -918,7 +918,6 @@ export function queryCollectionOptions( // Eager startup holds one reference until cleanup. Cache removal detaches // observation, not that ownership or its rows. let ensureEagerSubscription = () => {} - let replaceMutationResultApplications = () => {} const addRowOwner = (rowKey: string | number, hashedQueryKey: string) => { const owners = rowToQueries.get(rowKey) || new Set() @@ -997,9 +996,13 @@ export function queryCollectionOptions( const retainedQueriesPendingRevalidation = new Set() const pendingResultApplications = new Map>() const failedResultApplications = new Map() - const retainedResultApplicationTokens = new Map() - const activeResultApplicationTokens = new Map() - const resultApplicationControllers = new Map>() + type ResultApplicationController = AbortController & { + restoreOwnershipTracking?: () => void + } + const resultApplicationControllers = new Map< + string, + ResultApplicationController + >() const effectivePersistedGcTimes = new Map() const persistedRetentionTimers = new Map< string, @@ -1008,21 +1011,37 @@ export function queryCollectionOptions( let persistedRetentionMaintenance = Promise.resolve() const invalidatePendingResultApplication = (hashedQueryKey: string) => { + const controller = resultApplicationControllers.get(hashedQueryKey) + controller?.restoreOwnershipTracking?.() pendingResultApplications.delete(hashedQueryKey) failedResultApplications.delete(hashedQueryKey) - retainedResultApplicationTokens.delete(hashedQueryKey) - activeResultApplicationTokens.delete(hashedQueryKey) - resultApplicationControllers - .get(hashedQueryKey) - ?.forEach((controller) => controller.abort()) resultApplicationControllers.delete(hashedQueryKey) + controller?.abort() + } + + const waitForCurrentResultApplication = async ( + hashedQueryKey: string, + ): Promise => { + while (true) { + const application = pendingResultApplications.get(hashedQueryKey) + if (!application) return + try { + await application + } catch (error) { + if (pendingResultApplications.get(hashedQueryKey) === application) { + 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)) @@ -1593,7 +1612,7 @@ export function queryCollectionOptions( const applySuccessfulResult = async ( queryKey: QueryKey, result: QueryObserverResult, - applicationToken: object, + applicationToken: ResultApplicationController, persistedBaseline?: Map< string | number, { @@ -1657,11 +1676,8 @@ export function queryCollectionOptions( const restoreOwnershipTracking = () => { if (!state.observers.has(hashedQueryKey)) return - const currentApplicationToken = - activeResultApplicationTokens.get(hashedQueryKey) if ( - currentApplicationToken !== undefined && - currentApplicationToken !== applicationToken + resultApplicationControllers.get(hashedQueryKey) !== applicationToken ) { return } @@ -1679,6 +1695,7 @@ export function queryCollectionOptions( } }) } + applicationToken.restoreOwnershipTracking = restoreOwnershipTracking try { // From this point onward the result, including an empty result, is the @@ -1735,6 +1752,7 @@ export function queryCollectionOptions( } }) + applicationToken.restoreOwnershipTracking = undefined const applied = commit(signal) transactionActive = false retainedQueriesPendingRevalidation.delete(hashedQueryKey) @@ -1743,15 +1761,18 @@ export function queryCollectionOptions( // Readiness is publication: do not expose it until the establishing // transaction's rows and events are visible. if (applied !== true) { + applicationToken.restoreOwnershipTracking = restoreOwnershipTracking await applied } if (signal?.aborted) { restoreOwnershipTracking() return } + applicationToken.restoreOwnershipTracking = undefined markReady() } catch (error) { restoreOwnershipTracking() + applicationToken.restoreOwnershipTracking = undefined if (transactionActive) { const cancellation = new AbortController() @@ -1769,7 +1790,7 @@ export function queryCollectionOptions( const reconcileSuccessfulResult = async ( queryKey: QueryKey, result: QueryObserverResult, - applicationToken: object, + applicationToken: ResultApplicationController, signal: AbortSignal, ) => { const hashedQueryKey = hashKey(queryKey) @@ -1777,7 +1798,7 @@ export function queryCollectionOptions( await loadPersistedBaselineForQuery(hashedQueryKey) if ( collection.status === `cleaned-up` || - retainedResultApplicationTokens.get(hashedQueryKey) !== applicationToken + resultApplicationControllers.get(hashedQueryKey) !== applicationToken ) { return } @@ -1825,31 +1846,24 @@ export function queryCollectionOptions( const enqueueResultApplication = ( hashedQueryKey: string, - apply: (signal: AbortSignal, applicationToken: object) => Promise, - observedApplicationToken?: object, + apply: ( + signal: AbortSignal, + applicationToken: ResultApplicationController, + ) => Promise, ): void => { - const controller = new AbortController() - const applicationToken = observedApplicationToken ?? {} - const controllers = - resultApplicationControllers.get(hashedQueryKey) ?? new Set() - controllers.add(controller) - resultApplicationControllers.set(hashedQueryKey, controllers) - const previousApplication = pendingResultApplications.get(hashedQueryKey) - const run = () => { - activeResultApplicationTokens.set(hashedQueryKey, applicationToken) - return apply(controller.signal, applicationToken) - } - 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 @@ -1935,17 +1949,10 @@ export function queryCollectionOptions( } return } + if (result.isFetching) return - const applicationToken = {} - retainedResultApplicationTokens.set( - hashedQueryKey, - applicationToken, - ) - enqueueResultApplication( - hashedQueryKey, - (signal, token) => - reconcileSuccessfulResult(queryKey, result, token, signal), - applicationToken, + enqueueResultApplication(hashedQueryKey, (signal, token) => + reconcileSuccessfulResult(queryKey, result, token, signal), ) } else { enqueueResultApplication(hashedQueryKey, (signal, token) => @@ -1985,26 +1992,6 @@ export function queryCollectionOptions( return handleQueryResult } - replaceMutationResultApplications = () => { - state.observers.forEach((observer, hashedQueryKey) => { - const result = observer.getCurrentResult() - if ( - !result.isSuccess || - result.isFetching || - !hasPostWriteAuthority(hashedQueryKey, observer.getCurrentQuery()) - ) { - return - } - - // Earlier snapshots can be waiting behind this persisting mutation. - // Replace them with the completed refetch so its sync transaction is - // staged before the mutation handler returns; awaiting it here would - // deadlock on the mutation that owns the publication barrier. - invalidatePendingResultApplication(hashedQueryKey) - makeQueryResultHandler(hashToQueryKey.get(hashedQueryKey)!)(result) - }) - } - const isSubscribed = (hashedQueryKey: string) => { return unsubscribes.has(hashedQueryKey) } @@ -2197,7 +2184,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. @@ -2306,7 +2292,6 @@ export function queryCollectionOptions( const cleanup = () => { pendingStartupLoads.clear() ensureEagerSubscription = () => {} - replaceMutationResultApplications = () => {} unsubscribeFromCollectionEvents() unsubscribeFromQueries() persistedRetentionTimers.forEach((timer) => { @@ -2444,11 +2429,6 @@ export function queryCollectionOptions( return Promise.all(refetchPromises) } - const refetchAfterMutation = async () => { - await refetch() - replaceMutationResultApplications() - } - /** * Updates a single query key in the cache with new items, handling both direct arrays * and wrapped response formats (when `select` is used). @@ -2780,7 +2760,7 @@ export function queryCollectionOptions( (handlerResult as { refetch?: boolean }).refetch !== false if (shouldRefetch) { - await refetchAfterMutation() + await refetch() } return handlerResult @@ -2794,7 +2774,7 @@ export function queryCollectionOptions( (handlerResult as { refetch?: boolean }).refetch !== false if (shouldRefetch) { - await refetchAfterMutation() + await refetch() } return handlerResult @@ -2808,7 +2788,7 @@ export function queryCollectionOptions( (handlerResult as { refetch?: boolean }).refetch !== false if (shouldRefetch) { - await refetchAfterMutation() + await refetch() } return handlerResult @@ -2820,21 +2800,15 @@ export function queryCollectionOptions( const sync = withCollectionSyncConfigFactory( { sync: enhancedInternalSync }, - (source, utilities, startSync) => { + (source, utilities, startSyncIfIdle) => { const boundUtilities = utilities as Record< string, (...args: Array) => any > - for (const name of [ - `writeInsert`, - `writeUpdate`, - `writeDelete`, - `writeUpsert`, - `writeBatch`, - ]) { + for (const name of Object.keys(writeUtils)) { const write = boundUtilities[name]! boundUtilities[name] = (...args) => { - startSync() + startSyncIfIdle() return write(...args) } } 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 62e505fb58..34771f4299 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' @@ -592,6 +593,85 @@ describe(`query collection ownership lifecycle`, () => { 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] }, @@ -797,6 +877,243 @@ describe(`query collection ownership lifecycle`, () => { ).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(`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 + const subscription = collection.subscribeChanges(() => { + if (!injected && collection.has(outer.id)) { + injected = true + queryClient.setQueryData(queryKey, [invalid]) + } + }) + cleanups.push(async () => subscription.unsubscribe()) + + queryClient.setQueryData(queryKey, [outer]) + await vi.waitFor(() => expect(collection.utils.errorCount).toBe(1)) + 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 @@ -847,6 +1164,156 @@ describe(`query collection ownership lifecycle`, () => { }) }) + 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`, From 95c3ae71bae8f3150f615c5fc22a601140d23b39 Mon Sep 17 00:00:00 2001 From: Kyle Mathews Date: Wed, 16 Sep 2026 18:05:13 -0600 Subject: [PATCH 05/12] fix: address Query Collection lifecycle review --- packages/db/src/collection/mutations.ts | 4 ++- .../db/tests/collection-lifecycle.test.ts | 35 +++++++++++++++++++ packages/query-db-collection/src/query.ts | 6 +++- .../tests/ownership-lifecycle.oracle.test.ts | 6 ++++ 4 files changed, 49 insertions(+), 2 deletions(-) diff --git a/packages/db/src/collection/mutations.ts b/packages/db/src/collection/mutations.ts index a678bcd436..eb664efb85 100644 --- a/packages/db/src/collection/mutations.ts +++ b/packages/db/src/collection/mutations.ts @@ -241,8 +241,10 @@ export class CollectionMutationsManager< mutations.push(mutation) }) + let duplicate = mutations.find(({ key }) => this.state.has(key)) + if (duplicate) throw new DuplicateKeyError(duplicate.key) this.collection._sync.startSync() - const duplicate = mutations.find(({ key }) => this.state.has(key)) + duplicate = mutations.find(({ key }) => this.state.has(key)) if (duplicate) throw new DuplicateKeyError(duplicate.key) // If an ambient transaction exists, use it diff --git a/packages/db/tests/collection-lifecycle.test.ts b/packages/db/tests/collection-lifecycle.test.ts index e9a485ce82..8344185110 100644 --- a/packages/db/tests/collection-lifecycle.test.ts +++ b/packages/db/tests/collection-lifecycle.test.ts @@ -1,6 +1,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' import { z } from 'zod' import { createCollection } from '../src/collection/index.js' +import { DbClient, collectionOptions } from '../src/client.js' import { CleanupQueue } from '../src/collection/cleanup-queue.js' import { DuplicateKeyError, @@ -392,6 +393,40 @@ describe(`Collection Lifecycle Management`, () => { } }) + it(`does not start idle sync when an insert duplicates initial data`, async () => { + type Row = { id: string; value: string } + const existing = { id: `existing`, value: `initial` } + const startupError = new Error(`sync startup failed`) + let syncStarts = 0 + const descriptor = collectionOptions({ + id: `rejected-insert-initial-duplicate`, + getKey: (row: Row) => row.id, + startSync: false, + sync: { + sync: () => { + syncStarts++ + throw startupError + }, + }, + onInsert: async () => {}, + }) + const collection = new DbClient().collection(descriptor, { + initialData: [existing], + }) + + try { + expect(collection.status).toBe(`idle`) + expect(() => + collection.insert({ id: existing.id, value: `duplicate` }), + ).toThrow(DuplicateKeyError) + expect(syncStarts).toBe(0) + expect(collection.status).toBe(`idle`) + expect(collection.get(existing.id)).toMatchObject(existing) + } finally { + await collection.cleanup() + } + }) + it(`checks hydrated collection keys before admitting an idle insert`, async () => { type Row = { id: string; value: string } const target = { id: `target`, value: `synced` } diff --git a/packages/query-db-collection/src/query.ts b/packages/query-db-collection/src/query.ts index 7c4233ceb2..06370d24a1 100644 --- a/packages/query-db-collection/src/query.ts +++ b/packages/query-db-collection/src/query.ts @@ -1028,7 +1028,11 @@ export function queryCollectionOptions( try { await application } catch (error) { - if (pendingResultApplications.get(hashedQueryKey) === application) { + if ( + pendingResultApplications.get(hashedQueryKey) === application || + (failedResultApplications.has(hashedQueryKey) && + failedResultApplications.get(hashedQueryKey) === error) + ) { throw error } } 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 34771f4299..504463e6e3 100644 --- a/packages/query-db-collection/tests/ownership-lifecycle.oracle.test.ts +++ b/packages/query-db-collection/tests/ownership-lifecycle.oracle.test.ts @@ -1099,16 +1099,22 @@ describe(`query collection ownership lifecycle`, () => { 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, ) From a5b54a90fa7a8d88ebdc4e1c94925df3b4f7a8b7 Mon Sep 17 00:00:00 2001 From: Kyle Mathews Date: Thu, 17 Sep 2026 07:07:33 -0600 Subject: [PATCH 06/12] docs(db): clarify duplicate validation around sync --- packages/db/src/collection/mutations.ts | 2 ++ 1 file changed, 2 insertions(+) diff --git a/packages/db/src/collection/mutations.ts b/packages/db/src/collection/mutations.ts index eb664efb85..b50ec1365d 100644 --- a/packages/db/src/collection/mutations.ts +++ b/packages/db/src/collection/mutations.ts @@ -241,6 +241,8 @@ 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 }) => this.state.has(key)) if (duplicate) throw new DuplicateKeyError(duplicate.key) this.collection._sync.startSync() From 0ba1786901c9067d7f38c3b1d798b366910d6b19 Mon Sep 17 00:00:00 2001 From: Kyle Mathews Date: Thu, 17 Sep 2026 08:30:32 -0600 Subject: [PATCH 07/12] test(db): register mutation startup oracle --- docs/contributing/oracle-coverage.md | 2 +- packages/db/package.json | 2 +- .../db/tests/collection-lifecycle.test.ts | 441 +-------------- ...collection-mutation-startup-oracle.test.ts | 504 ++++++++++++++++++ 4 files changed, 507 insertions(+), 442 deletions(-) create mode 100644 packages/db/tests/collection-mutation-startup-oracle.test.ts diff --git a/docs/contributing/oracle-coverage.md b/docs/contributing/oracle-coverage.md index 528cabbe63..1f296b5a36 100644 --- a/docs/contributing/oracle-coverage.md +++ b/docs/contributing/oracle-coverage.md @@ -30,7 +30,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 18857f854b..07923c2fe5 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/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/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/tests/collection-lifecycle.test.ts b/packages/db/tests/collection-lifecycle.test.ts index 8344185110..7779c08e75 100644 --- a/packages/db/tests/collection-lifecycle.test.ts +++ b/packages/db/tests/collection-lifecycle.test.ts @@ -1,20 +1,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' -import { z } from 'zod' import { createCollection } from '../src/collection/index.js' -import { DbClient, collectionOptions } from '../src/client.js' import { CleanupQueue } from '../src/collection/cleanup-queue.js' -import { - DuplicateKeyError, - InvalidCollectionStatusTransitionError, - InvalidKeyError, - MissingDeleteHandlerError, - MissingInsertHandlerError, - MissingUpdateHandlerError, - NoKeysPassedToDeleteError, - NoKeysPassedToUpdateError, - SchemaValidationError, - UndefinedKeyError, -} from '../src/errors.js' +import { InvalidCollectionStatusTransitionError } from '../src/errors.js' import { getActivePublicationContext, transactionScopedScheduler, @@ -37,432 +24,6 @@ function getChangesManager(collection: object): { } describe(`Collection Lifecycle Management`, () => { - it(`does not start idle sync when an insert has no handler`, async () => { - type Row = { id: string; value: string } - let syncStarts = 0 - const collection = createCollection({ - id: `rejected-mutation-missing-handler`, - getKey: (row) => row.id, - startSync: false, - sync: { - sync: ({ markReady }) => { - syncStarts++ - markReady() - }, - }, - }) - - try { - expect(() => - collection.insert({ id: `rejected`, value: `rejected` }), - ).toThrow(MissingInsertHandlerError) - expect(syncStarts).toBe(0) - expect(collection.status).toBe(`idle`) - } finally { - await collection.cleanup() - } - }) - - it(`does not start idle sync when insert schema validation rejects`, async () => { - let syncStarts = 0 - const collection = createCollection({ - id: `rejected-mutation-schema`, - getKey: (row) => row.id, - schema: z.object({ id: z.string(), value: z.string().min(1) }), - startSync: false, - sync: { - sync: ({ markReady }) => { - syncStarts++ - markReady() - }, - }, - onInsert: async () => {}, - }) - - try { - expect(() => collection.insert({ id: `rejected`, value: `` })).toThrow( - SchemaValidationError, - ) - expect(syncStarts).toBe(0) - expect(collection.status).toBe(`idle`) - } finally { - await collection.cleanup() - } - }) - - it(`does not start idle sync when insert key validation rejects`, async () => { - type Row = { id: string; value: string } - let syncStarts = 0 - const collection = createCollection({ - id: `rejected-mutation-invalid-key`, - getKey: () => true as never, - startSync: false, - sync: { - sync: ({ markReady }) => { - syncStarts++ - markReady() - }, - }, - onInsert: async () => {}, - }) - - try { - expect(() => - collection.insert({ id: `rejected`, value: `rejected` }), - ).toThrow(InvalidKeyError) - expect(syncStarts).toBe(0) - expect(collection.status).toBe(`idle`) - } finally { - await collection.cleanup() - } - }) - - it(`does not start idle sync when an insert key is missing`, async () => { - type Row = { id: string; value: string } - let syncStarts = 0 - const collection = createCollection({ - id: `rejected-mutation-missing-key`, - getKey: () => undefined as never, - startSync: false, - sync: { - sync: ({ markReady }) => { - syncStarts++ - markReady() - }, - }, - onInsert: async () => {}, - }) - - try { - expect(() => - collection.insert({ id: `rejected`, value: `rejected` }), - ).toThrow(UndefinedKeyError) - expect(syncStarts).toBe(0) - expect(collection.status).toBe(`idle`) - } finally { - await collection.cleanup() - } - }) - - it(`starts idle sync exactly once when a mutation is accepted`, async () => { - type Row = { id: string; value: string } - let syncStarts = 0 - const collection = createCollection({ - id: `mutation-starts-idle-sync`, - getKey: (row) => row.id, - startSync: false, - sync: { - sync: ({ markReady }) => { - syncStarts++ - markReady() - }, - }, - onInsert: async () => {}, - }) - - try { - 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(collection.toArray.map(({ id }) => id).sort()).toEqual([ - `first`, - `second`, - ]) - } finally { - await collection.cleanup() - } - }) - - it(`surfaces idle sync startup errors before applying a mutation`, async () => { - type Row = { id: string; value: string } - const startupError = new Error(`sync startup failed`) - let syncStarts = 0 - const collection = createCollection({ - id: `mutation-sync-startup-error`, - getKey: (row) => row.id, - startSync: false, - sync: { - sync: () => { - syncStarts++ - throw startupError - }, - }, - onInsert: async () => {}, - }) - - try { - expect(() => - collection.insert({ id: `rejected`, value: `rejected` }), - ).toThrow(startupError) - expect(syncStarts).toBe(1) - expect(collection.status).toBe(`error`) - expect(collection.size).toBe(0) - } finally { - await collection.cleanup() - } - }) - - it.each([`update`, `delete`] as const)( - `does not start idle sync when %s has no handler`, - async (operation) => { - type Row = { id: string; value: string } - let syncStarts = 0 - const collection = createCollection({ - id: `rejected-${operation}-missing-handler`, - getKey: (row) => row.id, - startSync: false, - sync: { - sync: ({ markReady }) => { - syncStarts++ - markReady() - }, - }, - }) - - try { - const mutate = () => - operation === `update` - ? collection.update(`target`, (draft) => { - draft.value = `updated` - }) - : collection.delete(`target`) - - expect(mutate).toThrow( - operation === `update` - ? MissingUpdateHandlerError - : MissingDeleteHandlerError, - ) - expect(syncStarts).toBe(0) - expect(collection.status).toBe(`idle`) - } finally { - await collection.cleanup() - } - }, - ) - - it.each([`update`, `delete`] as const)( - `does not start idle sync when %s receives no keys`, - async (operation) => { - type Row = { id: string; value: string } - let syncStarts = 0 - const collection = createCollection({ - id: `rejected-${operation}-empty-keys`, - getKey: (row) => row.id, - startSync: false, - sync: { - sync: ({ markReady }) => { - syncStarts++ - markReady() - }, - }, - onUpdate: async () => {}, - onDelete: async () => {}, - }) - - try { - const mutate = () => - operation === `update` - ? collection.update([], () => {}) - : collection.delete([]) - - expect(mutate).toThrow( - operation === `update` - ? NoKeysPassedToUpdateError - : NoKeysPassedToDeleteError, - ) - expect(syncStarts).toBe(0) - expect(collection.status).toBe(`idle`) - } finally { - await collection.cleanup() - } - }, - ) - - it(`does not start idle sync when update receives no callback`, async () => { - type Row = { id: string; value: string } - let syncStarts = 0 - const collection = createCollection({ - id: `rejected-update-missing-callback`, - getKey: (row) => row.id, - startSync: false, - sync: { - sync: ({ begin, write, commit, markReady }) => { - syncStarts++ - begin() - write({ type: `insert`, value: { id: `target`, value: `original` } }) - commit() - markReady() - }, - }, - onUpdate: async () => {}, - }) - - try { - expect(() => - (collection.update as (key: string, config: object) => unknown)( - `target`, - {}, - ), - ).toThrow(TypeError) - expect(syncStarts).toBe(0) - expect(collection.status).toBe(`idle`) - } finally { - await collection.cleanup() - } - }) - - it.each([`update`, `delete`] as const)( - `starts idle sync before %s checks collection state`, - async (operation) => { - type Row = { id: string; value: string } - const target = { id: `target`, value: `original` } - let syncStarts = 0 - const collection = createCollection({ - id: `accepted-${operation}-hydrates-target`, - getKey: (row) => row.id, - startSync: false, - sync: { - sync: ({ begin, write, commit, markReady }) => { - syncStarts++ - begin() - write({ type: `insert`, value: target }) - commit() - markReady() - }, - }, - onUpdate: async () => {}, - onDelete: async () => {}, - }) - - try { - 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.map(({ key, type }) => ({ key, type })), - ).toEqual([{ key: `target`, type: operation }]) - await transaction.isPersisted.promise - } finally { - await collection.cleanup() - } - }, - ) - - it(`does not start idle sync when an insert batch has duplicate keys`, async () => { - type Row = { id: string; value: string } - let syncStarts = 0 - const collection = createCollection({ - id: `rejected-insert-batch-duplicate`, - getKey: (row) => row.id, - startSync: false, - sync: { - sync: ({ markReady }) => { - syncStarts++ - markReady() - }, - }, - onInsert: async () => {}, - }) - - try { - expect(() => - collection.insert([ - { id: `duplicate`, value: `first` }, - { id: `duplicate`, value: `second` }, - ]), - ).toThrow(DuplicateKeyError) - expect(syncStarts).toBe(0) - expect(collection.status).toBe(`idle`) - } finally { - await collection.cleanup() - } - }) - - it(`does not start idle sync when an insert duplicates initial data`, async () => { - type Row = { id: string; value: string } - const existing = { id: `existing`, value: `initial` } - const startupError = new Error(`sync startup failed`) - let syncStarts = 0 - const descriptor = collectionOptions({ - id: `rejected-insert-initial-duplicate`, - getKey: (row: Row) => row.id, - startSync: false, - sync: { - sync: () => { - syncStarts++ - throw startupError - }, - }, - onInsert: async () => {}, - }) - const collection = new DbClient().collection(descriptor, { - initialData: [existing], - }) - - try { - expect(collection.status).toBe(`idle`) - expect(() => - collection.insert({ id: existing.id, value: `duplicate` }), - ).toThrow(DuplicateKeyError) - expect(syncStarts).toBe(0) - expect(collection.status).toBe(`idle`) - expect(collection.get(existing.id)).toMatchObject(existing) - } finally { - await collection.cleanup() - } - }) - - it(`checks hydrated collection keys before admitting an idle insert`, async () => { - type Row = { id: string; value: string } - const target = { id: `target`, value: `synced` } - const onInsert = vi.fn(async () => {}) - let syncStarts = 0 - const collection = createCollection({ - id: `idle-insert-hydrated-duplicate`, - getKey: (row) => row.id, - startSync: false, - sync: { - sync: ({ begin, write, commit, markReady }) => { - syncStarts++ - begin() - write({ type: `insert`, value: target }) - commit() - markReady() - }, - }, - onInsert, - }) - - try { - expect(() => collection.insert({ id: `target`, value: `local` })).toThrow( - DuplicateKeyError, - ) - expect(syncStarts).toBe(1) - expect(collection.status).toBe(`ready`) - expect( - collection.toArray.map(({ id, value }) => ({ id, value })), - ).toEqual([target]) - expect(onInsert).not.toHaveBeenCalled() - } finally { - await collection.cleanup() - } - }) - it.each( ([`same`, `missing`, `changed`, `empty`] as const).flatMap((shape) => ([`atomic`, `split`] as const).map((delivery) => ({ shape, delivery })), 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 0000000000..08224bc722 --- /dev/null +++ b/packages/db/tests/collection-mutation-startup-oracle.test.ts @@ -0,0 +1,504 @@ +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 { + const first = collection.insert({ id: `first`, value: `first` }) + const second = collection.insert({ id: `second`, value: `second` }) + await Promise.all([first.isPersisted.promise, second.isPersisted.promise]) + expect(syncStarts).toBe(1) + 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() + } + }, + ) +}) From a00e73ac0083ad224b6b7a28fd6ea8b144d258a7 Mon Sep 17 00:00:00 2001 From: Kyle Mathews Date: Thu, 17 Sep 2026 08:53:49 -0600 Subject: [PATCH 08/12] test(db): preserve mutation startup checkpoints --- .changeset/fix-query-collection-lifecycle.md | 2 +- packages/db/tests/collection-mutation-startup-oracle.test.ts | 5 +++++ 2 files changed, 6 insertions(+), 1 deletion(-) diff --git a/.changeset/fix-query-collection-lifecycle.md b/.changeset/fix-query-collection-lifecycle.md index 0bea5f8af5..e7c5cd4ab3 100644 --- a/.changeset/fix-query-collection-lifecycle.md +++ b/.changeset/fix-query-collection-lifecycle.md @@ -3,4 +3,4 @@ '@tanstack/query-db-collection': patch --- -Start idle collections only after mutation validation succeeds, and publish authoritative Query Collection refetch results without stale intermediate snapshots. +Start idle collections only after locally decidable mutation validation succeeds, and publish authoritative Query Collection refetch results without stale intermediate snapshots. diff --git a/packages/db/tests/collection-mutation-startup-oracle.test.ts b/packages/db/tests/collection-mutation-startup-oracle.test.ts index 08224bc722..59541a35b0 100644 --- a/packages/db/tests/collection-mutation-startup-oracle.test.ts +++ b/packages/db/tests/collection-mutation-startup-oracle.test.ts @@ -435,10 +435,15 @@ describe(`Collection mutation startup oracle`, () => { }) 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`, From fe9e48a98bd61021f57e983583581aad684dee07 Mon Sep 17 00:00:00 2001 From: Kyle Mathews Date: Thu, 17 Sep 2026 11:02:27 -0600 Subject: [PATCH 09/12] fix(query-db): preserve ownership after publication --- packages/query-db-collection/src/query.ts | 27 +- .../tests/ownership-lifecycle.oracle.test.ts | 290 +++++++++++++++++- 2 files changed, 294 insertions(+), 23 deletions(-) diff --git a/packages/query-db-collection/src/query.ts b/packages/query-db-collection/src/query.ts index 06370d24a1..3a39c0de85 100644 --- a/packages/query-db-collection/src/query.ts +++ b/packages/query-db-collection/src/query.ts @@ -997,7 +997,8 @@ export function queryCollectionOptions( const pendingResultApplications = new Map>() const failedResultApplications = new Map() type ResultApplicationController = AbortController & { - restoreOwnershipTracking?: () => void + tx?: { applicationStarted: boolean } + rollback?: () => void } const resultApplicationControllers = new Map< string, @@ -1012,7 +1013,8 @@ export function queryCollectionOptions( const invalidatePendingResultApplication = (hashedQueryKey: string) => { const controller = resultApplicationControllers.get(hashedQueryKey) - controller?.restoreOwnershipTracking?.() + // Core flips this at its no-cancel point before publication can reenter. + if (!controller?.tx?.applicationStarted) controller?.rollback?.() pendingResultApplications.delete(hashedQueryKey) failedResultApplications.delete(hashedQueryKey) resultApplicationControllers.delete(hashedQueryKey) @@ -1022,7 +1024,7 @@ export function queryCollectionOptions( const waitForCurrentResultApplication = async ( hashedQueryKey: string, ): Promise => { - while (true) { + for (;;) { const application = pendingResultApplications.get(hashedQueryKey) if (!application) return try { @@ -1699,7 +1701,7 @@ export function queryCollectionOptions( } }) } - applicationToken.restoreOwnershipTracking = restoreOwnershipTracking + applicationToken.rollback = restoreOwnershipTracking try { // From this point onward the result, including an empty result, is the @@ -1756,7 +1758,7 @@ export function queryCollectionOptions( } }) - applicationToken.restoreOwnershipTracking = undefined + applicationToken.tx = collection._state.pendingSyncedTransactions.at(-1) const applied = commit(signal) transactionActive = false retainedQueriesPendingRevalidation.delete(hashedQueryKey) @@ -1764,19 +1766,12 @@ export function queryCollectionOptions( // Readiness is publication: do not expose it until the establishing // transaction's rows and events are visible. - if (applied !== true) { - applicationToken.restoreOwnershipTracking = restoreOwnershipTracking - await applied - } - if (signal?.aborted) { + if (applied !== true) await applied + if (!signal?.aborted) markReady() + } catch (error) { + if (!applicationToken.tx?.applicationStarted) { restoreOwnershipTracking() - return } - applicationToken.restoreOwnershipTracking = undefined - markReady() - } catch (error) { - restoreOwnershipTracking() - applicationToken.restoreOwnershipTracking = undefined if (transactionActive) { const cancellation = new AbortController() 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 504463e6e3..4953857f6c 100644 --- a/packages/query-db-collection/tests/ownership-lifecycle.oracle.test.ts +++ b/packages/query-db-collection/tests/ownership-lifecycle.oracle.test.ts @@ -293,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(), @@ -335,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 } @@ -429,7 +426,7 @@ function createPersistedOwnershipFixture( queryClient.clear() } }) - return { collection, queryFn } + return { collection, queryClient, queryFn } } function storedItems( @@ -445,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, @@ -1025,6 +1022,285 @@ describe(`query collection ownership lifecycle`, () => { }) }) + 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(`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 From 10bd80fe61728eff513de8ac3e8eaa2dc2c69307 Mon Sep 17 00:00:00 2001 From: Kyle Mathews Date: Thu, 17 Sep 2026 11:09:41 -0600 Subject: [PATCH 10/12] test(query-db): preserve failed replacement readiness --- .../tests/ownership-lifecycle.oracle.test.ts | 50 +++++++++++++++++++ 1 file changed, 50 insertions(+) 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 4953857f6c..8546c6e555 100644 --- a/packages/query-db-collection/tests/ownership-lifecycle.oracle.test.ts +++ b/packages/query-db-collection/tests/ownership-lifecycle.oracle.test.ts @@ -1157,6 +1157,56 @@ describe(`query collection ownership lifecycle`, () => { ).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(`retires a publishing result when its final subset unloads`, async () => { const id = `reentrant-result-unload` const queryKey = [id] as const From e5ef635b2987055c85985dd88410b5debab37c67 Mon Sep 17 00:00:00 2001 From: Kyle Mathews Date: Thu, 17 Sep 2026 11:11:53 -0600 Subject: [PATCH 11/12] test(query-db): protect prepublication rollback --- .../tests/ownership-lifecycle.oracle.test.ts | 65 +++++++++++++++++++ 1 file changed, 65 insertions(+) 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 8546c6e555..3a5958f79e 100644 --- a/packages/query-db-collection/tests/ownership-lifecycle.oracle.test.ts +++ b/packages/query-db-collection/tests/ownership-lifecycle.oracle.test.ts @@ -1207,6 +1207,71 @@ describe(`query collection ownership lifecycle`, () => { 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 From 5ca8f919fc79c0acaeab87031aa0c9f317684d5d Mon Sep 17 00:00:00 2001 From: Kyle Mathews Date: Thu, 17 Sep 2026 11:24:31 -0600 Subject: [PATCH 12/12] refactor: simplify lifecycle phase tracking --- packages/db/src/collection/mutations.ts | 4 ++-- packages/query-db-collection/src/query.ts | 13 ++++++------- 2 files changed, 8 insertions(+), 9 deletions(-) diff --git a/packages/db/src/collection/mutations.ts b/packages/db/src/collection/mutations.ts index b50ec1365d..056887bc3b 100644 --- a/packages/db/src/collection/mutations.ts +++ b/packages/db/src/collection/mutations.ts @@ -243,10 +243,10 @@ export class CollectionMutationsManager< // Reject duplicates already visible before explicitly starting sync; startup may // synchronously reveal additional keys, so check again afterward. - let duplicate = mutations.find(({ key }) => this.state.has(key)) + let duplicate = mutations.find(({ key }) => state.has(key)) if (duplicate) throw new DuplicateKeyError(duplicate.key) this.collection._sync.startSync() - duplicate = mutations.find(({ key }) => this.state.has(key)) + duplicate = mutations.find(({ key }) => state.has(key)) if (duplicate) throw new DuplicateKeyError(duplicate.key) // If an ambient transaction exists, use it diff --git a/packages/query-db-collection/src/query.ts b/packages/query-db-collection/src/query.ts index 3a39c0de85..a9b83ee8f9 100644 --- a/packages/query-db-collection/src/query.ts +++ b/packages/query-db-collection/src/query.ts @@ -997,7 +997,6 @@ export function queryCollectionOptions( const pendingResultApplications = new Map>() const failedResultApplications = new Map() type ResultApplicationController = AbortController & { - tx?: { applicationStarted: boolean } rollback?: () => void } const resultApplicationControllers = new Map< @@ -1013,8 +1012,7 @@ export function queryCollectionOptions( const invalidatePendingResultApplication = (hashedQueryKey: string) => { const controller = resultApplicationControllers.get(hashedQueryKey) - // Core flips this at its no-cancel point before publication can reenter. - if (!controller?.tx?.applicationStarted) controller?.rollback?.() + controller?.rollback?.() pendingResultApplications.delete(hashedQueryKey) failedResultApplications.delete(hashedQueryKey) resultApplicationControllers.delete(hashedQueryKey) @@ -1679,8 +1677,11 @@ 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 @@ -1758,7 +1759,7 @@ export function queryCollectionOptions( } }) - applicationToken.tx = collection._state.pendingSyncedTransactions.at(-1) + resultTransaction = collection._state.pendingSyncedTransactions.at(-1) const applied = commit(signal) transactionActive = false retainedQueriesPendingRevalidation.delete(hashedQueryKey) @@ -1769,9 +1770,7 @@ export function queryCollectionOptions( if (applied !== true) await applied if (!signal?.aborted) markReady() } catch (error) { - if (!applicationToken.tx?.applicationStarted) { - restoreOwnershipTracking() - } + restoreOwnershipTracking() if (transactionActive) { const cancellation = new AbortController()