fix(db): propagate initial query errors

.changeset/propagate-initial-query-errors.md

@@ -0,0 +1,10 @@
+---
+'@tanstack/db': patch
+'@tanstack/electric-db-collection': patch
+'@tanstack/powersync-db-collection': patch
+'@tanstack/query-db-collection': patch
+'@tanstack/rxdb-db-collection': patch
+'@tanstack/trailbase-db-collection': patch
+---
+
+Propagate initial query sync failures through dependent live queries and readiness promises, including recovery and late subscribers, while preserving a ready cached snapshot on later refetch failures. Let sync adapters pass the original failure to `markError(error)` so readiness promises reject with that cause. Isolate adapter callbacks by sync session, preserve synchronous startup errors, and prevent rejected deduplicated subset requests from creating detached promise rejections.

docs/guides/collection-options-creator.md

@@ -73,10 +73,11 @@ The sync function must return a cleanup function for proper garbage collection:
 
 ```typescript
 const sync: SyncConfig<T>['sync'] = (params) => {
- const { begin, write, commit, markReady, collection } = params
+ const { begin, write, commit, markReady, markError, collection } = params
 
  // 1. Initialize connection to your sync engine
  const connection = initializeConnection(config)
+ const initialSyncAbort = new AbortController()
 
  // 2. Set up real-time subscription FIRST (prevents race conditions)
  const eventBuffer: Array<any> = []
@@ -110,7 +111,7 @@ const sync: SyncConfig<T>['sync'] = (params) => {
  // 3. Perform initial data fetch
  async function initialSync() {
  try {
- const data = await fetchInitialData()
+ const data = await fetchInitialData({ signal: initialSyncAbort.signal })
 
  begin() // Start a transaction
 
@@ -134,20 +135,24 @@ const sync: SyncConfig<T>['sync'] = (params) => {
  commit()
  eventBuffer.splice(0)
  }
-
+
+ // A complete initial snapshot is now available.
+ markReady()
  } catch (error) {
+ if (initialSyncAbort.signal.aborted) return
  console.error('Initial sync failed:', error)
-throw error
-} finally {
- // ALWAYS call markReady, even on error
-markReady()
+// No usable initial snapshot exists.
+ // Only initial startup owns collection readiness. A later refetch
+ // failure must keep the last ready snapshot usable.
+if (collection.status === 'loading') markError(error)
  }
  }
 
  initialSync()
 
  // 4. Return cleanup function
  return () => {
+ initialSyncAbort.abort()
  connection.close()
  // Clean up any timers, intervals, or other resources
  }
@@ -163,7 +168,8 @@ The sync process follows this lifecycle:
 1. **begin()** - Start collecting changes
 2. **write()** - Add changes to the pending transaction (buffered until commit)
 3. **commit()** - Apply all changes atomically to the collection state
-4. **markReady()** - Signal that initial sync is complete
+4. **markReady()** - Signal that a usable initial or recovered snapshot exists
+5. **markError(error?)** - Signal that initial sync failed before producing a usable snapshot; pass the cause so readiness waits reject with it
 
 **Race Condition Prevention:**
 Many sync engines start real-time subscriptions before the initial sync completes. Your implementation MUST deduplicate events that arrive via subscription that represent the same data as the initial sync. Consider:
@@ -900,8 +906,8 @@ const wrappedOnInsert = async (params) => {
 
 ## Best Practices
 
-1. **Always call markReady()** - This signals that the collection has initial data and is ready for use
-2. **Handle errors gracefully** - Call markReady() even on error to avoid blocking the app
+1. **Report initial sync status** - Call `markReady()` after a usable snapshot, or `markError(error)` if initial sync fails
+2. **Recover explicitly** - After an error, call `markReady()` only when a later sync has produced a usable snapshot
 3. **Clean up resources** - Return a cleanup function from sync to prevent memory leaks
 4. **Batch operations** - Use begin/commit to batch multiple changes for better performance
 5. **Race Conditions** - Start listeners before initial fetch and buffer events

docs/guides/error-handling.md

@@ -420,7 +420,7 @@ try {
 
 ### Query Collection Sync Errors
 
-Query collections handle sync errors gracefully and mark the collection as ready even on error to avoid blocking applications:
+Query collections distinguish an initial load failure from a later refetch failure:
 
 ```ts
 import { queryCollectionOptions } from "@tanstack/query-db-collection"
@@ -447,9 +447,11 @@ const todoCollection = createCollection(
 
 When sync errors occur:
 - Error is logged to console: `[QueryCollection] Error observing query...`
-- Collection is marked as ready to prevent blocking the application
-- Cached data remains available
+- An initial failure marks the collection as `error` because no usable snapshot exists
+- Readiness waits such as `preload()` and `toArrayWhenReady()` reject with the cause passed to `markError(error)` while the collection is in that initial error state
+- A later refetch failure keeps the collection `ready` and preserves its cached data
 - Error tracking counters are updated (`lastError`, `errorCount`)
+- A later successful refetch recovers an initial `error` collection to `ready`; a new readiness wait then resolves normally
 
 ### Sync Write Errors
 

packages/db/skills/db-core/custom-adapter/SKILL.md

@@ -2,7 +2,7 @@
 name: db-core/custom-adapter
 description: >
  Building custom collection adapters for new backends. SyncConfig interface:
- sync function receiving begin, write, commit, markReady, truncate, metadata
+ sync function receiving begin, write, commit, markReady, markError, truncate, metadata
  primitives and returning cleanup, loadSubset, and optional unloadSubset
  handlers.
  ChangeMessage format (insert, update, delete). On-demand LoadSubsetOptions
@@ -48,9 +48,10 @@ function myBackendCollectionOptions<T extends object>(config: {
  return {
  getKey: config.getKey,
  sync: {
- sync: ({ begin, write, commit, markReady }) => {
+ sync: ({ begin, write, commit, markReady, markError, collection }) => {
  let isInitialSyncComplete = false
  const bufferedEvents: Array<BackendEvent<T>> = []
+ const initialSyncAbort = new AbortController()
 
  // 1. Subscribe to real-time events FIRST
  const unsubscribe = myWebSocket.subscribe(config.endpoint, (event) => {
@@ -64,28 +65,37 @@ function myBackendCollectionOptions<T extends object>(config: {
  })
 
  // 2. Fetch initial data
- fetch(config.endpoint).then(async (res) => {
- const items = await res.json()
- begin()
- for (const item of items) {
- write({ type: 'insert', value: item })
- }
- commit()
-
- // 3. Process buffered events
- isInitialSyncComplete = true
- for (const event of bufferedEvents) {
+ void fetch(config.endpoint, { signal: initialSyncAbort.signal })
+ .then(async (res) => {
+ const items = await res.json()
  begin()
- write({ type: event.type, key: event.id, value: event.data })
+ for (const item of items) {
+ write({ type: 'insert', value: item })
+ }
  commit()
- }
 
- // 4. Signal readiness
- markReady()
- })
+ // 3. Process buffered events
+ isInitialSyncComplete = true
+ for (const event of bufferedEvents) {
+ begin()
+ write({ type: event.type, key: event.id, value: event.data })
+ commit()
+ }
+
+ // 4. Signal that a usable snapshot exists
+ markReady()
+ })
+ .catch((error) => {
+ if (initialSyncAbort.signal.aborted) return
+ console.error('Initial sync failed:', error)
+ // Only initial startup owns collection readiness. A later refetch
+ // failure must keep the last ready snapshot usable.
+ if (collection.status === 'loading') markError(error)
+ })
 
  // 5. Return cleanup function
  return () => {
+ initialSyncAbort.abort()
  unsubscribe()
  }
  },
@@ -210,7 +220,7 @@ Without persistence the metadata is in-memory only and does not survive
 reloads. With persistence, it is durable across sessions.
 
 ```ts
-sync: ({ begin, write, commit, markReady, metadata }) => {
+sync: ({ begin, write, commit, markReady, markError, metadata }) => {
  if (!metadata) throw new Error('Sync metadata API is unavailable')
 
  // Row metadata: store per-row state (e.g. server version, ETag)
@@ -254,6 +264,7 @@ sync: ({ begin, write, commit, markReady, metadata }) => {
  })
 
  stream.on('ready', () => markReady())
+ stream.on('initial-error', (error) => markError(error))
  return () => stream.close()
 }
 ```
@@ -322,6 +333,12 @@ sync: ({ begin, write, commit, markReady }) => {
 
 `markReady()` transitions the collection to "ready" status. Without it, live queries never resolve and `useLiveSuspenseQuery` hangs forever in Suspense.
 
+If initial sync fails before it produces a usable snapshot, call
+`markError(error)` instead. This rejects readiness waits with the supplied cause
+and moves dependent live queries to the error state. Calling `markError()`
+without a cause remains supported and rejects with a generic collection-state
+error. A later successful sync can call `markReady()` to recover.
+
 Source: docs/guides/collection-options-creator.md
 
 ### HIGH Race condition: subscribing after initial fetch

packages/db/src/collection/index.ts

@@ -464,7 +464,7 @@ export class CollectionImpl<
  * // Safe to access collection.state now
  * })
  */
- public onFirstReady(callback: () => void): void {
+ public onFirstReady(callback: () => void): () => void {
  return this._lifecycle.onFirstReady(callback)
  }
 

packages/db/src/collection/lifecycle.ts

@@ -36,6 +36,7 @@ export class CollectionLifecycleManager<
  public hasReceivedFirstCommit = false
  public onFirstReadyCallbacks: Array<() => void> = []
  private idleCallbackId: number | null = null
+ private syncError: unknown
 
  /**
  * Creates a new CollectionLifecycleManager instance
@@ -77,7 +78,7 @@ export class CollectionLifecycleManager<
  idle: [`loading`, `error`, `cleaned-up`],
  loading: [`ready`, `error`, `cleaned-up`],
  ready: [`cleaned-up`, `error`],
- error: [`cleaned-up`, `idle`],
+ error: [`ready`, `cleaned-up`, `idle`],
  'cleaned-up': [`loading`, `error`],
  }
 
@@ -133,8 +134,9 @@ export class CollectionLifecycleManager<
  */
  public markReady(): void {
  this.validateStatusTransition(this.status, `ready`)
- // Can transition to ready from loading state
- if (this.status === `loading`) {
+ // A successful initial sync or recovery establishes a ready snapshot.
+ if (this.status === `loading` || this.status === `error`) {
+ this.syncError = undefined
  this.setStatus(`ready`, true)
 
  // Call any registered first ready callbacks (only on first time becoming ready)
@@ -158,6 +160,18 @@ export class CollectionLifecycleManager<
  }
  }
 
+ /** Mark an asynchronous sync failure after sync has started. */
+ public markError(error?: unknown): void {
+ this.validateStatusTransition(this.status, `error`)
+ this.syncError = error
+ this.setStatus(`error`)
+ }
+
+ /** Return the cause supplied by the current sync session, if any. */
+ public getSyncError(): unknown {
+ return this.syncError
+ }
+
  /**
  * Start the garbage collection timer
  * Called when the collection becomes inactive (no subscribers)
@@ -243,6 +257,7 @@ export class CollectionLifecycleManager<
  CleanupQueue.getInstance().cancel(this)
 
  this.hasBeenReady = false
+ this.syncError = undefined
 
  // Call any pending onFirstReady callbacks before clearing them.
  // This ensures preload() promises resolve during cleanup instead of hanging.
@@ -282,14 +297,20 @@ export class CollectionLifecycleManager<
  * Useful for preloading collections
  * @param callback Function to call when the collection first becomes ready
  */
- public onFirstReady(callback: () => void): void {
+ public onFirstReady(callback: () => void): () => void {
  // If already ready, call immediately
  if (this.hasBeenReady) {
  callback()
- return
+ return () => {}
  }
 
  this.onFirstReadyCallbacks.push(callback)
+ return () => {
+ const index = this.onFirstReadyCallbacks.indexOf(callback)
+ if (index !== -1) {
+ this.onFirstReadyCallbacks.splice(index, 1)
+ }
+ }
  }
 
  public cleanup(): void {