fix: complete direct subscribers of preloaded observables on error

src/SuspenseSubject.ts

@@ -14,7 +14,7 @@
   private _hasValue = false;
   private _warmupRelease: Subscription;
   private _firstEmission: Promise<void>;
   private _error: any = undefined;
   private _innerObservable: Observable<T>;
   private _warmupSubscription: Subscription;
   private _immutableStatus: ObservableStatus<T>;
@@ -147,31 +147,31 @@
   }
 
   _subscribe(subscriber: Subscriber<T>): Subscription {
+    // `catchError(() => EMPTY)` converts upstream errors into completions, so a direct subscriber completes on error.
+    return this._attach((value) => subscriber.next(value), () => subscriber.complete(), false);
+  }
+
+  // For `useObservable`. Completing a `Subscriber` auto-unsubscribes it while the hook is still mounted,
+  // so this reports every change through `onStoreChange` and never completes.
+  _subscribeToStore(onStoreChange: () => void): Subscription {
+    return this._attach(onStoreChange, onStoreChange, true);
+  }
+
+  private _attach(next: (value: T) => void, complete: () => void, evictErroredEntryOnUnsubscribe: boolean): Subscription {
     // A mounted subscriber owns the source now; the warm-up release would evict a mounted one-shot entry.
     this._warmupRelease.unsubscribe();
     if (!this._error && !this._isComplete) {
       this._onReregister();
     }
     this._subscriberCount++;
-    const innerSubscription = this._innerObservable.subscribe({
-      next: (value) => subscriber.next(value),
-      complete: () => {
-        if (this._error) {
-          // `catchError(() => EMPTY)` converts upstream errors into completions.
-          // Notify `useObservable` (`onStoreChange`) without calling `subscriber.complete()`,
-          // because `subscriber.complete()` auto-unsubscribes `subscriber` before React unmounts.
-          subscriber.next(this._value);
-        } else {
-          subscriber.complete();
-        }
-      },
-    });
+    const innerSubscription = this._innerObservable.subscribe({ next, complete });
     // Hand off ownership of the shared source from the warm-up subscription to the real subscriber.
     this._warmupSubscription.unsubscribe();
     return new Subscription(() => {
       innerSubscription.unsubscribe();
       this._subscriberCount--;
-      if (this._subscriberCount === 0 && this._error) {
+      // A direct subscriber is unsubscribed when it completes on error, while a suspended render may still need the entry.
+      if (evictErroredEntryOnUnsubscribe && this._subscriberCount === 0 && this._error) {
         this._onEvict();
       }
     });

src/useObservable.ts

@@ -9,10 +9,10 @@
 const DEFAULT_TIMEOUT = 30_000;
 
 // Since we're side-effect free, we need to ensure our observable cache is global
 const preloadedObservables: Map<string, SuspenseSubject<any>> = (globalThis as any as ReactFireGlobals)._reactFirePreloadedObservables || new Map();
 
 if (!(globalThis as any as ReactFireGlobals)._reactFirePreloadedObservables) {
   (globalThis as any as ReactFireGlobals)._reactFirePreloadedObservables = preloadedObservables;
 }
 
 // Starts listening to an Observable.
@@ -92,25 +92,14 @@
   const observable = preloadObservable(source, observableId, suspenseEnabled);
 
   // Suspend if suspense is enabled and no initial data exists
   const hasInitialData = config.hasOwnProperty('initialData') || config.hasOwnProperty('startWithValue');
   const hasData = observable.hasValue || hasInitialData;
   if (suspenseEnabled === true && !hasData) {
     throw observable.firstEmission;
   }
 
   const subscribe = React.useCallback((onStoreChange: () => void) => {
-      const subscription = observable.subscribe({
-        next: () => {
-          onStoreChange();
-        },
-        error: (e) => {
-          onStoreChange();
-          throw e;
-        },
-        complete: () => {
-          onStoreChange();
-        },
-      });
+      const subscription = observable._subscribeToStore(onStoreChange);
 
       return () => {
         subscription.unsubscribe();

test/suspenseSubject-error-branch.test.tsx

@@ -1,15 +1,14 @@
 import '@testing-library/jest-dom/extend-expect';
 import { cleanup, render, waitFor } from '@testing-library/react';
 import * as React from 'react';
-import { Observable, timer, throwError } from 'rxjs';
+import { concat, EmptyError, lastValueFrom, Observable, of, timer, throwError } from 'rxjs';
 import { mergeMap } from 'rxjs/operators';
-import { useObservable } from '../src/index';
+import { preloadObservable, useObservable } from '../src/index';
 
-// Guards the `if (this._error)` branch in SuspenseSubject._subscribe's `complete` handler.
-// `catchError(() => EMPTY)` turns a source error into a completion; that branch turns it back into
-// a notification instead of completing the React subscriber. Completing instead auto-unsubscribes
-// the mounted subscriber, and a mounted hook with `initialData` then resubscribes in a loop.
-describe('SuspenseSubject error-to-completion branch', () => {
+// `catchError(() => EMPTY)` turns a source error into a completion. `_subscribeToStore` reports it to
+// `useObservable` without completing, because completing unsubscribes the still-mounted hook, and its
+// unmount then no longer evicts the errored entry. `_subscribe` completes a direct subscriber.
+describe('SuspenseSubject source errors', () => {
   afterEach(() => cleanup());
 
   class Boundary extends React.Component<{ children: React.ReactNode }, { error: Error | null }> {
@@ -23,7 +22,7 @@
   }
 
   it('surfaces a source error to a mounted initialData hook once, without resubscribing', async () => {
     const consoleError = vi.spyOn(console, 'error').mockImplementation(() => {});
     try {
       let subscriptions = 0;
       const source$ = new Observable<string>((subscriber) => {
@@ -54,4 +53,71 @@
       consoleError.mockRestore();
     }
   }, 10000);
+
+  // A direct subscriber of a `preload*` result completes, as in 4.3.0.
+  it('completes a direct subscriber of a preloaded observable when the source errors', async () => {
+    const source$ = timer(50).pipe(mergeMap(() => throwError(() => new Error('permission-denied'))));
+    const preloaded$ = preloadObservable(source$, `error-branch-direct-${Math.random()}`);
+
+    const events: string[] = [];
+    const observer = {
+      next: (value: unknown) => events.push(`next:${String(value)}`),
+      error: () => events.push('error'),
+      complete: () => events.push('complete'),
+    };
+
+    preloaded$.subscribe(observer);
+    await waitFor(() => expect(events).toEqual(['complete']));
+
+    // A subscriber that arrives after the error completes too.
+    events.length = 0;
+    preloaded$.subscribe(observer);
+    expect(events).toEqual(['complete']);
+    await expect(lastValueFrom(preloaded$)).rejects.toBeInstanceOf(EmptyError);
+  });
+
+  it('delivers a value to a direct subscriber before completing it on a later error', async () => {
+    const source$ = concat(of('first'), timer(50).pipe(mergeMap(() => throwError(() => new Error('permission-denied')))));
+    const events: string[] = [];
+    preloadObservable(source$, `error-branch-value-${Math.random()}`).subscribe({
+      next: (value) => events.push(`next:${String(value)}`),
+      error: () => events.push('error'),
+      complete: () => events.push('complete'),
+    });
+
+    await waitFor(() => expect(events).toEqual(['next:first', 'complete']));
+  });
+
+  it('keeps the errored entry for a suspended reader when a direct subscriber completes', async () => {
+    const consoleError = vi.spyOn(console, 'error').mockImplementation(() => {});
+    try {
+      let subscriptions = 0;
+      const source$ = new Observable<string>((subscriber) => {
+        subscriptions++;
+        return timer(50)
+          .pipe(mergeMap(() => throwError(() => new Error('permission-denied'))))
+          .subscribe(subscriber);
+      });
+      const observableId = `error-branch-render-${Math.random()}`;
+
+      const Reader = () => {
+        void lastValueFrom(preloadObservable(source$, observableId)).catch(() => undefined);
+        const { data } = useObservable(observableId, source$, { suspense: true });
+        return <div data-testid="val">{data}</div>;
+      };
+
+      const { getByTestId } = render(
+        <Boundary>
+          <React.Suspense fallback={<div data-testid="fallback" />}>
+            <Reader />
+          </React.Suspense>
+        </Boundary>
+      );
+
+      await waitFor(() => expect(getByTestId('err')).toHaveTextContent('permission-denied'));
+      expect(subscriptions).toBe(1);
+    } finally {
+      consoleError.mockRestore();
+    }
+  });
 });