fix: finish loadSubset review and recovery follow-ups

.changeset/finish-review-recovery.md

@@ -0,0 +1,11 @@
+---
+'@tanstack/db': patch
+'@tanstack/db-sqlite-persistence-core': patch
+'@tanstack/powersync-db-collection': patch
+---
+
+Preserve native values, arbitrary class references, and draft cycles during mutation detachment; keep transaction persistence receipts settled after publication errors and avoid restoring an acknowledged direct insert over its server row. Keep a delete/reinsert visible when the old synced row has not yet been replaced.
+
+Retire replaced ordered prefixes without interrupting successful-load bookkeeping if release throws. Retry automatic ordered repair at most twice while retaining stale results and exposing the error; cleanup cancels retries and explicit window retry remains available.
+
+Keep persisted acquisitions independent, avoid retaining one-shot refreshes as permanent demand, and reject upstream load failures without discarding cached rows. Restore PowerSync readiness only after the recovered baseline also removes rows deleted or moved outside active filters during the tracking outage.

docs/guides/error-handling.md

@@ -155,6 +155,19 @@ try {
 Effects report subset failures through `onSourceError` and dispose because
 their incremental result can no longer be kept complete.
 
+When a source change invalidates an ordered window, automatic full-source
+repair keeps the last complete snapshot visible. A failed repair exposes
+`utils.lastSubsetError` and retries at most twice, after 250 ms and 500 ms.
+Exhausting those retries does not put an already-ready query into a terminal
+error state or clear its rows. The app can show the error and explicitly retry
+with `setWindow()`. Cleanup or truncate cancels the old repair timer. Failed
+imperative window moves and initial loads do not use this background retry.
+
+For SQLite-persisted on-demand collections, a failed upstream `loadSubset`
+rejects even when hydration succeeded. Cached rows remain readable; their
+availability does not mean the remote request succeeded. Background coordinator
+retry, where supported, does not change the failed caller's outcome.
+
 When a must-refetch truncate cannot reload every active subset, a subscription
 keeps its last successful snapshot and reports the subset error. It discards
 the incomplete replay batch and keeps later source changes private because they
@@ -323,6 +336,18 @@ try {
 }
 ```
 
+Explicit cancellation is different from a mutation failure. If you call
+`tx.rollback()` while `mutationFn` is pending, the rollback settles
+`tx.isPersisted.promise` as rejected. A later result or rejection from that
+mutation function is ignored: the outstanding `commit()` call resolves and
+`tx.error` is not populated by that late rejection. Observe `isPersisted.promise`
+when you need the transaction's outcome, including explicit cancellation.
+
+After the mutation function succeeds, a publication listener can still throw
+while the completed transaction updates its collections. In that case
+`commit()` rejects with the listener error, but `isPersisted.promise` resolves
+and the transaction remains completed. This is not a persistence failure.
+
 ## Collection Operation Errors
 
 ### Invalid Collection State

docs/guides/mutations.md

@@ -364,6 +364,13 @@ need to keep a new caller-owned object unchanged during the callback, insert
 your own copy. A thrown callback does not roll back edits to that caller-owned
 object; it leaves existing collection data unchanged.
 
+Arbitrary class instances are an exception: newly assigned instances stay by
+reference so their methods, prototypes, and private fields remain intact.
+Later changes to such an instance can therefore affect stored data without a
+new update or notification. Treat those instances as immutable, or convert them
+to plain data before assignment when you need isolation. Supported native values
+such as `URL`, `Date`, `RegExp`, and typed arrays are copied instead.
+
 ### Delete
 
 Remove items from a collection:

packages/db-sqlite-persistence-core/src/persisted.ts

@@ -352,6 +352,7 @@ export interface PersistedCollectionUtils extends UtilsRecord {
     mutations: Array<PendingMutation<Record<string, unknown>>>
   }) => Promise<void> | void
   getLeadershipState?: () => PersistedCollectionLeadershipState
+  /** Hydrate once without acquiring a new ongoing subset lease. */
   forceReloadSubset?: (options: LoadSubsetOptions) => Promise<void> | void
 }
 
@@ -707,18 +708,6 @@ function stableSerialize(value: unknown): string {
   return JSON.stringify(toStableSerializable(value) ?? null)
 }
 
-function normalizeSubsetOptionsForKey(
-  options: LoadSubsetOptions,
-): Record<string, unknown> {
-  return {
-    where: toStableSerializable(options.where),
-    orderBy: toStableSerializable(options.orderBy),
-    limit: options.limit,
-    cursor: toStableSerializable(options.cursor),
-    offset: options.offset,
-  }
-}
-
 function normalizeSyncFnResult(result: void | (() => void) | SyncConfigRes) {
   if (typeof result === `function`) {
     return { cleanup: result } satisfies SyncConfigRes
@@ -802,7 +791,7 @@ class PersistedCollectionRuntime<
     BufferedSyncTransaction<T, TKey>
   > = []
   private readonly queuedTxCommitted: Array<TxCommitted> = []
-  private readonly subscriptionIds = new WeakMap<object, string>()
+  private readonly requestIds = new WeakMap<LoadSubsetOptions, string>()
 
   private collection: Collection<T, TKey, PersistedCollectionUtils> | null =
     null
@@ -824,7 +813,7 @@ class PersistedCollectionRuntime<
   private indexAddedUnsubscribe: (() => void) | null = null
   private indexRemovedUnsubscribe: (() => void) | null = null
   private remoteEnsureRetryTimer: ReturnType<typeof setTimeout> | null = null
-  private nextSubscriptionId = 0
+  private nextRequestId = 0
 
   private latestTerm = 0
   private latestSeq = 0
@@ -938,7 +927,8 @@ class PersistedCollectionRuntime<
     await this.bootstrapPersistedIndexes(indexBootstrapSnapshot)
 
     if (this.syncMode !== `on-demand`) {
-      this.activeSubsets.set(this.getSubsetKey({}), {})
+      const initialSubset = {}
+      this.activeSubsets.set(this.getSubsetKey(initialSubset), initialSubset)
       const appliedCursor = this.appliedReceiptSequence
       await this.applyMutex.run(() =>
         this.hydrateSubsetUnsafe({}, { requestRemoteEnsure: false }),
@@ -1044,6 +1034,8 @@ class PersistedCollectionRuntime<
         }
         console.warn(`Failed to trigger remote subset load:`, error)
         this.queueRemoteSubsetEnsure(options)
+        // Hydration remains readable, but it does not satisfy remote demand.
+        throw error
       }
     }
   }
@@ -1058,7 +1050,7 @@ class PersistedCollectionRuntime<
   }
 
   async forceReloadSubset(options: LoadSubsetOptions): Promise<void> {
-    this.activeSubsets.set(this.getSubsetKey(options), options)
+    // A one-shot refresh does not acquire an enduring subscription lease.
     await this.applyMutex.run(() =>
       this.hydrateSubsetUnsafe(options, { requestRemoteEnsure: false }),
     )
@@ -1808,20 +1800,14 @@ class PersistedCollectionRuntime<
   }
 
   private getSubsetKey(options: LoadSubsetOptions): string {
-    const subscription = options.subscription as object | undefined
-    if (subscription && typeof subscription === `object`) {
-      const existingId = this.subscriptionIds.get(subscription)
-      if (existingId) {
-        return existingId
-      }
-
-      this.nextSubscriptionId++
-      const id = `sub:${this.nextSubscriptionId}`
-      this.subscriptionIds.set(subscription, id)
-      return id
+    // A subscription can own several independent acquisitions, including
+    // identical requests. Only releasing this options object ends its lease.
+    let id = this.requestIds.get(options)
+    if (id === undefined) {
+      id = `request:${++this.nextRequestId}`
+      this.requestIds.set(options, id)
     }
-
-    return `opts:${stableSerialize(normalizeSubsetOptionsForKey(options))}`
+    return id
   }
 
   private queueRemoteSubsetEnsure(options: LoadSubsetOptions): void {
@@ -2605,6 +2591,8 @@ function createWrappedSyncConfig<
             if (!resolvedSourceResult.loadSubset) return true
             acquisition.forwarded = true
             try {
+              // Returning a promise transfers its lease even if it rejects.
+              // Only a synchronous throw leaves no upstream lease to release.
               return resolvedSourceResult.loadSubset(loadOptions)
             } catch (error) {
               acquisition.forwarded = false

packages/db-sqlite-persistence-core/tests/persisted.test.ts

@@ -1744,6 +1744,201 @@ describe(`persistedCollectionOptions`, () => {
     expect(collection.get(`2`)).toBeUndefined()
   })
 
+  it.each(
+    [false, true].flatMap((sharedSubscription) =>
+      [false, true].map((identical) => ({ sharedSubscription, identical })),
+    ),
+  )(
+    `keeps sibling requests owned after one release: %j`,
+    async ({ sharedSubscription, identical }) => {
+      const adapter = createRecordingAdapter([{ id: `1`, title: `Before` }])
+      const coordinator = createCoordinatorHarness()
+      const collection = createCollection(
+        persistedCollectionOptions<Todo, string>({
+          id: `sync-present`,
+          getKey: (row) => row.id,
+          syncMode: `on-demand`,
+          sync: {
+            sync: ({ markReady }) => {
+              markReady()
+              return { loadSubset: () => true }
+            },
+          },
+          persistence: { adapter, coordinator },
+        }),
+      )
+      collection.startSyncImmediate()
+      const subscription = collection.subscribeChanges(() => {})
+      const owner = sharedSubscription ? { subscription } : {}
+      const page: LoadSubsetOptions = {
+        ...owner,
+        ...(identical ? {} : { limit: 1 }),
+      }
+      const all: LoadSubsetOptions = { ...owner }
+      try {
+        await collection._sync.loadSubset(page)
+        await collection._sync.loadSubset(all)
+        collection._sync.unloadSubset(page)
+        coordinator.emit({
+          type: `tx:committed`,
+          term: 1,
+          seq: 1,
+          txId: `sibling-update`,
+          latestRowVersion: 1,
+          requiresFullReload: false,
+          changedRows: [{ key: `1`, value: { id: `1`, title: `After` } }],
+          deletedKeys: [],
+        })
+        await flushAsyncWork()
+        expect(stripVirtualProps(collection.get(`1`))).toEqual({
+          id: `1`,
+          title: `After`,
+        })
+      } finally {
+        subscription.unsubscribe()
+        await collection.cleanup()
+      }
+    },
+  )
+
+  it(`does not retain refresh history as permanent subset demand`, async () => {
+    const adapter = createRecordingAdapter([{ id: `1`, title: `Before` }])
+    const coordinator = createCoordinatorHarness()
+    const collection = createCollection(
+      persistedCollectionOptions<Todo, string>({
+        id: `sync-present`,
+        getKey: (row) => row.id,
+        syncMode: `on-demand`,
+        persistence: { adapter, coordinator },
+      }),
+    )
+    collection.startSyncImmediate()
+    try {
+      await collection._sync.loadSubset({ limit: 1 })
+      for (let i = 0; i < 20; i++)
+        await collection.utils.forceReloadSubset!({ limit: 1 })
+      const before = adapter.loadSubsetCalls.length
+      coordinator.emit({
+        type: `tx:committed`,
+        term: 1,
+        seq: 1,
+        txId: `refresh-invalidation`,
+        latestRowVersion: 1,
+        requiresFullReload: true,
+      })
+      await flushAsyncWork()
+      await flushAsyncWork()
+      expect(adapter.loadSubsetCalls.length - before).toBe(1)
+    } finally {
+      await collection.cleanup()
+    }
+  })
+
+  it(`reports a non-abort upstream failure rather than treating hydration as remote success`, async () => {
+    const failure = new Error(`remote acquisition failed`)
+    const warn = vi.spyOn(console, `warn`).mockImplementation(() => {})
+    const collection = createCollection(
+      persistedCollectionOptions<Todo, string>({
+        id: `remote-acquisition-failure`,
+        getKey: (row) => row.id,
+        syncMode: `on-demand`,
+        sync: {
+          sync: ({ markReady }) => {
+            markReady()
+            return {
+              loadSubset: () => Promise.reject(failure),
+            }
+          },
+        },
+        persistence: {
+          adapter: createRecordingAdapter([
+            { id: `cached`, title: `Last known row` },
+          ]),
+        },
+      }),
+    )
+    collection.startSyncImmediate()
+    try {
+      await expect(
+        Promise.resolve(collection._sync.loadSubset({})),
+      ).rejects.toBe(failure)
+      expect(stripVirtualProps(collection.get(`cached`))).toEqual({
+        id: `cached`,
+        title: `Last known row`,
+      })
+    } finally {
+      warn.mockRestore()
+      await collection.cleanup()
+    }
+  })
+
+  it.each([`throw`, `reject`] as const)(
+    `releases only transferred upstream ownership after a load %s`,
+    async (mode) => {
+      const failure = new Error(`failed upstream load`)
+      const warn = vi.spyOn(console, `warn`).mockImplementation(() => {})
+      const peer: LoadSubsetOptions = { limit: 1 }
+      const failed: LoadSubsetOptions = { limit: 1 }
+      const leases = new Set<LoadSubsetOptions>()
+      let publish!: (title: string) => Promise<void>
+      const unload = vi.fn((options: LoadSubsetOptions) => {
+        leases.delete(options)
+      })
+      const collection = createCollection(
+        persistedCollectionOptions<Todo, string>({
+          id: `failed-load-ownership-${mode}`,
+          getKey: (row) => row.id,
+          syncMode: `on-demand`,
+          sync: {
+            sync: ({ begin, write, commit, markReady }) => {
+              publish = async (title) => {
+                if (!leases.has(peer)) return
+                begin()
+                write({
+                  type: collection.has(`live`) ? `update` : `insert`,
+                  value: { id: `live`, title },
+                })
+                await commit()
+              }
+              markReady()
+              return {
+                loadSubset: (options) => {
+                  if (options === failed && mode === `throw`) throw failure
+                  // Returning a promise transfers the ongoing lease, even if
+                  // fetching its initial snapshot subsequently fails.
+                  leases.add(options)
+                  return options === failed ? Promise.reject(failure) : true
+                },
+                unloadSubset: unload,
+              }
+            },
+          },
+          persistence: { adapter: createRecordingAdapter() },
+        }),
+      )
+      collection.startSyncImmediate()
+      try {
+        await collection._sync.loadSubset(peer)
+        await expect(collection._sync.loadSubset(failed)).rejects.toBe(failure)
+        expect(leases.has(failed)).toBe(mode === `reject`)
+        collection._sync.unloadSubset(failed)
+        expect(unload.mock.calls.map(([options]) => options)).toEqual(
+          mode === `reject` ? [failed] : [],
+        )
+        expect(leases).toEqual(new Set([peer]))
+        await publish(`Peer still live`)
+        expect(collection.get(`live`)?.title).toBe(`Peer still live`)
+        collection._sync.unloadSubset(peer)
+        expect(leases.size).toBe(0)
+        await publish(`Must not arrive`)
+        expect(collection.get(`live`)?.title).toBe(`Peer still live`)
+      } finally {
+        warn.mockRestore()
+        await collection.cleanup()
+      }
+    },
+  )
+
   it(`does not release or acquire an upstream lease cancelled during hydration`, async () => {
     const adapter = createRecordingAdapter()
     const hydrate = adapter.loadSubset
@@ -1861,7 +2056,7 @@ describe(`persistedCollectionOptions`, () => {
         const callsBeforeRetry = ensure.mock.calls.length
         await vi.advanceTimersByTimeAsync(200)
         if (action === `offline`) {
-          expect(result).toBe(`ready`)
+          expect(result).toBe(failure)
           expect(ensure.mock.calls.length).toBeGreaterThan(callsBeforeRetry)
         } else {
           if (action === `abort`) expect(result).toBe(failure)

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/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/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-lifecycle-oracle.property.test.ts tests/query/ordered-work-oracle.property.test.ts tests/query/pagination-oracle.property.test.ts"
+    "test:oracles": "vitest --run tests/collection-cleanup-restart-oracle.test.ts tests/effect-disposal-oracle.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/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"
   },
   "type": "module",
   "main": "dist/cjs/index.cjs",