Fix error propagation from sync clients to collections

packages/db/src/collection/index.ts

@@ -215,7 +215,7 @@ export class CollectionImpl<
   public utils: Record<string, Fn> = {}
 
   // Managers
-  private _events: CollectionEventsManager
+  public events: CollectionEventsManager
   private _changes: CollectionChangesManager<TOutput, TKey, TSchema, TInput>
   private _lifecycle: CollectionLifecycleManager<TOutput, TKey, TSchema, TInput>
   private _sync: CollectionSyncManager<TOutput, TKey, TSchema, TInput>
@@ -261,7 +261,7 @@ export class CollectionImpl<
     }
 
     this._changes = new CollectionChangesManager()
-    this._events = new CollectionEventsManager()
+    this.events = new CollectionEventsManager()
     this._indexes = new CollectionIndexesManager()
     this._lifecycle = new CollectionLifecycleManager(config, this.id)
     this._mutations = new CollectionMutationsManager(config, this.id)
@@ -272,9 +272,9 @@ export class CollectionImpl<
       collection: this, // Required for passing to CollectionSubscription
       lifecycle: this._lifecycle,
       sync: this._sync,
-      events: this._events,
+      events: this.events,
     })
-    this._events.setDeps({
+    this.events.setDeps({
       collection: this, // Required for adding to emitted events
     })
     this._indexes.setDeps({
@@ -283,7 +283,7 @@ export class CollectionImpl<
     })
     this._lifecycle.setDeps({
       changes: this._changes,
-      events: this._events,
+      events: this.events,
       indexes: this._indexes,
       state: this._state,
       sync: this._sync,
@@ -810,7 +810,7 @@ export class CollectionImpl<
     event: T,
     callback: CollectionEventHandler<T>
   ) {
-    return this._events.on(event, callback)
+    return this.events.on(event, callback)
   }
 
   /**
@@ -820,7 +820,7 @@ export class CollectionImpl<
     event: T,
     callback: CollectionEventHandler<T>
   ) {
-    return this._events.once(event, callback)
+    return this.events.once(event, callback)
   }
 
   /**
@@ -830,7 +830,7 @@ export class CollectionImpl<
     event: T,
     callback: CollectionEventHandler<T>
   ) {
-    this._events.off(event, callback)
+    this.events.off(event, callback)
   }
 
   /**
@@ -840,7 +840,7 @@ export class CollectionImpl<
     event: T,
     timeout?: number
   ) {
-    return this._events.waitFor(event, timeout)
+    return this.events.waitFor(event, timeout)
   }
 
   /**

packages/db/src/collection/lifecycle.ts

@@ -78,7 +78,7 @@ export class CollectionLifecycleManager<
       loading: [`initialCommit`, `ready`, `error`, `cleaned-up`],
       initialCommit: [`ready`, `error`, `cleaned-up`],
       ready: [`cleaned-up`, `error`],
-      error: [`cleaned-up`, `idle`],
+      error: [`cleaned-up`, `idle`, `loading`],
       "cleaned-up": [`loading`, `error`],
     }
 
@@ -144,6 +144,11 @@ export class CollectionLifecycleManager<
    * @private - Should only be called by sync implementations
    */
   public markReady(): void {
+    // If recovering from error, transition to loading first
+    if (this.status === `error`) {
+      this.setStatus(`loading`)
+    }
+
     this.validateStatusTransition(this.status, `ready`)
     // Can transition to ready from loading or initialCommit states
     if (this.status === `loading` || this.status === `initialCommit`) {
@@ -170,6 +175,15 @@ export class CollectionLifecycleManager<
     }
   }
 
+  /**
+   * Mark the collection as being in an error state
+   * This is called by sync implementations when persistent errors occur
+   * @private - Should only be called by sync implementations
+   */
+  public markError(): void {
+    this.setStatus(`error`)
+  }
+
   /**
    * Start the garbage collection timer
    * Called when the collection becomes inactive (no subscribers)

packages/db/src/collection/sync.ts

@@ -159,6 +159,9 @@ export class CollectionSyncManager<
           markReady: () => {
             this.lifecycle.markReady()
           },
+          markError: () => {
+            this.lifecycle.markError()
+          },
           truncate: () => {
             const pendingTransaction =
               this.state.pendingSyncedTransactions[
@@ -221,6 +224,19 @@ export class CollectionSyncManager<
         resolve()
       })
 
+      // Also listen for error status transitions and reject the promise
+      const unsubscribeError = this.collection.events.once(
+        `status:error`,
+        () => {
+          reject(new CollectionIsInErrorStateError())
+        }
+      )
+
+      // Clean up error listener when promise resolves
+      this.lifecycle.onFirstReady(() => {
+        unsubscribeError()
+      })
+
       // Start sync if collection hasn't started yet or was cleaned up
       if (
         this.lifecycle.status === `idle` ||
@@ -229,6 +245,7 @@ export class CollectionSyncManager<
         try {
           this.startSync()
         } catch (error) {
+          unsubscribeError()
           reject(error)
           return
         }

packages/db/src/types.ts

@@ -172,6 +172,7 @@ export interface SyncConfig<
     write: (message: Omit<ChangeMessage<T>, `key`>) => void
     commit: () => void
     markReady: () => void
+    markError: () => void
     truncate: () => void
   }) => void | CleanupFn | SyncConfigRes
 

packages/electric-db-collection/src/electric.ts

@@ -587,6 +587,10 @@ function createElectricSync<T extends Row<unknown>>(
     collectionId?: string
   }
 ): SyncConfig<T> {
+  // Track first error for grace period before setting collection to error status
+  let firstErrorTimestamp: number | null = null
+  let errorGracePeriodTimeout: ReturnType<typeof setTimeout> | null = null
+  const ERROR_GRACE_PERIOD_MS = 10000 // 10 seconds
   const {
     seenTxids,
     seenSnapshots,
@@ -620,7 +624,7 @@ function createElectricSync<T extends Row<unknown>>(
 
   return {
     sync: (params: Parameters<SyncConfig<T>[`sync`]>[0]) => {
-      const { begin, write, commit, markReady, truncate, collection } = params
+      const { begin, write, commit, markReady, markError, truncate } = params
 
       // Abort controller for the stream - wraps the signal if provided
       const abortController = new AbortController()
@@ -655,25 +659,29 @@ function createElectricSync<T extends Row<unknown>>(
         ...shapeOptions,
         signal: abortController.signal,
         onError: (errorParams) => {
-          // Just immediately mark ready if there's an error to avoid blocking
-          // apps waiting for `.preload()` to finish.
           // Note that Electric sends a 409 error on a `must-refetch` message, but the
-          // ShapeStream handled this and it will not reach this handler, therefor
-          // this markReady will not be triggers by a `must-refetch`.
-          markReady()
+          // ShapeStream handles this and it will not reach this handler.
+          // If the error is transitory, ShapeStream will retry and eventually call
+          // markReady() naturally when it receives 'up-to-date'.
+
+          // Track first error for grace period
+          if (firstErrorTimestamp === null) {
+            firstErrorTimestamp = Date.now()
+
+            // After 10 seconds of continuous errors, set collection status to error
+            errorGracePeriodTimeout = setTimeout(() => {
+              markError()
+            }, ERROR_GRACE_PERIOD_MS)
+          }
 
           if (shapeOptions.onError) {
             return shapeOptions.onError(errorParams)
           } else {
-            console.error(
-              `An error occurred while syncing collection: ${collection.id}, \n` +
-                `it has been marked as ready to avoid blocking apps waiting for '.preload()' to finish. \n` +
-                `You can provide an 'onError' handler on the shapeOptions to handle this error, and this message will not be logged.`,
-              errorParams
-            )
+            // If no custom error handler is provided, throw the error
+            // This ensures errors propagate to the app and aligns with
+            // Electric SQL's documented behavior
+            throw errorParams
           }
-
-          return
         },
       })
       let transactionStarted = false
@@ -767,6 +775,15 @@ function createElectricSync<T extends Row<unknown>>(
         }
 
         if (hasUpToDate) {
+          // Clear error tracking on successful sync (recovery from transitory errors)
+          if (firstErrorTimestamp !== null) {
+            firstErrorTimestamp = null
+            if (errorGracePeriodTimeout !== null) {
+              clearTimeout(errorGracePeriodTimeout)
+              errorGracePeriodTimeout = null
+            }
+          }
+
           // Clear the current batch buffer since we're now up-to-date
           currentBatchMessages.setState(() => [])