fix(core): release subscriptions on unmount and evict errored cache entries

src/SuspenseSubject.ts

@@ -1,7 +1,11 @@
-import { empty, Observable, Subject, Subscriber, Subscription } from 'rxjs';
-import { catchError, shareReplay, tap } from 'rxjs/operators';
+import { empty, Observable, ReplaySubject, Subject, Subscriber, Subscription, timer } from 'rxjs';
+import { catchError, share, tap } from 'rxjs/operators';
 import { ObservableStatus } from './useObservable.js';
 
+// Grace period before a zero-subscriber stream detaches its Firebase listener and evicts its cache entry.
+// A 1-second window lets React StrictMode double-mounts and immediate route transitions reuse the active listener.
+const RELEASE_DELAY_MS = 1000;
+
 export class SuspenseSubject<T> extends Subject<T> {
   private _value: T | undefined;
   private _hasValue = false;
@@ -18,7 +22,12 @@ export class SuspenseSubject<T> extends Subject<T> {
   // @ts-expect-error: TODO: double check to see if this is an RXJS thing or if we should listen to TS
   private _resolveFirstEmission: () => void;
 
-  constructor(innerObservable: Observable<T>, private _timeoutWindow: number, private _suspenseEnabled: boolean) {
+  constructor(
+    innerObservable: Observable<T>,
+    private _timeoutWindow: number,
+    _suspenseEnabled: boolean,
+    onEvict: () => void = () => {}
+  ) {
     super();
     this._firstEmission = new Promise<void>((resolve) => (this._resolveFirstEmission = resolve));
 
@@ -42,26 +51,41 @@ export class SuspenseSubject<T> extends Subject<T> {
           this._error = e;
           this._resolveFirstEmission();
           this._updateImmutableStatus();
+          // Evict the errored entry after the current render reads it so a remount or ErrorBoundary retry starts fresh (#742, #485).
+          setTimeout(onEvict, RELEASE_DELAY_MS);
         },
         complete: () => {
           this._isComplete = true;
           this._updateImmutableStatus();
         }
       }),
       catchError(() => empty()),
-      shareReplay(1)
+      // Multicast one underlying Firebase listener across all subscribers:
+      // - `connector`: cache and replay the most recent emission to new subscribers on mount.
+      // - `resetOnError`: tear down the source subscription when the stream errors.
+      // - `resetOnComplete`: keep the final value cached when a one-shot stream (`first()`) completes.
+      // - `resetOnRefCountZero`: unsubscribe from Firebase and evict the cache entry once no subscribers remain for `RELEASE_DELAY_MS`.
+      share({
+        connector: () => new ReplaySubject<T>(1),
+        resetOnError: true,
+        resetOnComplete: false,
+        resetOnRefCountZero: () => timer(RELEASE_DELAY_MS).pipe(tap(onEvict)),
+      })
     );
     // warm up the observable
     this._warmupSubscription = this._innerObservable.subscribe();
 
-    // set a timeout for resetting the cache, subscriptions will cancel the timeout
-    // and reschedule again on unsubscribe
-    if (this._suspenseEnabled) {
-      // Noop if suspense is enabled
-      this._timeoutHandler = setTimeout(() => {}, this._timeoutWindow);
-    } else {
-      this._timeoutHandler = setTimeout(this._reset.bind(this), this._timeoutWindow);
-    }
+    // Release the warm-up subscription if no component subscribes within `_timeoutWindow` after the first emission settles.
+    // Do not release while the first emission is still pending: a suspended component is waiting on `firstEmission`, and
+    // React will not retry the render and call `_subscribe` until that promise resolves (#525).
+    const releaseWarmupWhenSettled = () => {
+      if (this._hasValue || this._error) {
+        this._warmupSubscription.unsubscribe();
+      } else {
+        this._timeoutHandler = setTimeout(releaseWarmupWhenSettled, this._timeoutWindow);
+      }
+    };
+    this._timeoutHandler = setTimeout(releaseWarmupWhenSettled, this._timeoutWindow);
   }
 
   get hasValue(): boolean {
@@ -108,23 +132,13 @@ export class SuspenseSubject<T> extends Subject<T> {
     this._updateImmutableStatus();
   }
 
-  private _reset() {
-    // seems to be undefined in tests?
-    if (this._warmupSubscription) {
-      this._warmupSubscription.unsubscribe();
-    }
-    this._hasValue = false;
-    this._value = undefined;
-    this._error = undefined;
-    this._firstEmission = new Promise<void>((resolve) => (this._resolveFirstEmission = resolve));
-    this._updateImmutableStatus();
-  }
-
   _subscribe(subscriber: Subscriber<T>): Subscription {
     if (this._timeoutHandler) {
       clearTimeout(this._timeoutHandler);
     }
     this._innerSubscriber = this._innerObservable.subscribe(subscriber);
+    // Hand off ownership of the shared source from the warm-up subscription to the real subscriber.
+    this._warmupSubscription.unsubscribe();
     return this._innerSubscriber;
   }
 

src/useObservable.ts

@@ -21,7 +21,7 @@ export function preloadObservable<T>(source: Observable<T>, id: string, suspense
   if (preloadedObservables.has(id)) {
     return preloadedObservables.get(id) as SuspenseSubject<T>;
   } else {
-    const observable = new SuspenseSubject(source, DEFAULT_TIMEOUT, suspenseEnabled);
+    const observable: SuspenseSubject<T> = new SuspenseSubject(source, DEFAULT_TIMEOUT, suspenseEnabled, () => preloadedObservables.get(id) === observable && preloadedObservables.delete(id));
     preloadedObservables.set(id, observable);
     return observable;
   }

test/useObservable.test.tsx

@@ -427,4 +427,109 @@ describe('useObservable', () => {
       expect(html).not.toContain('another-requests-value');
     });
   });
+
+  describe('Subscription lifecycle & error eviction', () => {
+    it('releases the underlying source subscription after unmount', async () => {
+      for (const suspense of [false, true]) {
+        const subject$ = new Subject<number>();
+        const observableId = `sub-release-${suspense}-${Math.random()}`;
+
+        const Obs = () => {
+          const { data } = useObservable(observableId, subject$, { suspense });
+          return <div data-testid="obs">{String(data)}</div>;
+        };
+
+        const { unmount, getByTestId } = render(
+          <React.Suspense fallback={<div data-testid="fb">loading</div>}>
+            <Obs />
+          </React.Suspense>
+        );
+
+        await act(async () => {
+          subject$.next(42);
+        });
+
+        await waitFor(() => expect(getByTestId('obs')).toHaveTextContent('42'));
+        expect(subject$.observed).toBe(true);
+
+        unmount();
+        await waitFor(() => expect(subject$.observed).toBe(false), { timeout: 2500 });
+      }
+    });
+
+    it('evicts an errored cache entry so a subsequent mount with the same observableId recovers', async () => {
+      const observableId = `error-evict-${Math.random()}`;
+      let shouldFail = true;
+
+      class Boundary extends React.Component<{ children: React.ReactNode }, { error: Error | null }> {
+        state: { error: Error | null } = { error: null };
+        static getDerivedStateFromError(error: Error) {
+          return { error };
+        }
+        render() {
+          if (this.state.error) {
+            return <div data-testid="err">{this.state.error.message}</div>;
+          }
+          return this.props.children;
+        }
+      }
+
+      const Obs = () => {
+        const source$ = shouldFail ? throwError(() => new Error('permission-denied')) : of('recovered');
+        const { data, status } = useObservable(observableId, source$);
+        return <div data-testid="val">{status === 'success' ? data : 'loading'}</div>;
+      };
+
+      const first = render(
+        <Boundary>
+          <Obs />
+        </Boundary>
+      );
+
+      await waitFor(() => expect(first.getByTestId('err')).toHaveTextContent('permission-denied'));
+      first.unmount();
+
+      // Wait for the 1s error eviction window to clear the errored entry from preloadedObservables
+      await new Promise((r) => setTimeout(r, 1100));
+      shouldFail = false;
+
+      const second = render(
+        <Boundary>
+          <Obs />
+        </Boundary>
+      );
+
+      await waitFor(() => expect(second.getByTestId('val')).toHaveTextContent('recovered'));
+      second.unmount();
+    });
+
+    it('evicts a cache entry when an error arrives after unmount during the release grace window (#485)', async () => {
+      const observableId = `error-after-unmount-${Math.random()}`;
+      const firstStream$ = new BehaviorSubject<string>('initial-user-doc');
+      const secondStream$ = new BehaviorSubject<string>('reauthenticated-user-doc');
+      let currentStream$ = firstStream$;
+
+      const Obs = () => {
+        const { data, status } = useObservable(observableId, currentStream$);
+        return <div data-testid="val">{status === 'success' ? data : 'loading'}</div>;
+      };
+
+      const first = render(<Obs />);
+      await waitFor(() => expect(first.getByTestId('val')).toHaveTextContent('initial-user-doc'));
+
+      // Unmount first (e.g. route leaves protected view on sign-out), then emit permission-denied
+      // while the 1s release grace window is still active.
+      first.unmount();
+      act(() => {
+        firstStream$.error(new Error('permission-denied-after-unmount'));
+      });
+
+      await new Promise((r) => setTimeout(r, 1100));
+      currentStream$ = secondStream$;
+
+      const second = render(<Obs />);
+      await waitFor(() => expect(second.getByTestId('val')).toHaveTextContent('reauthenticated-user-doc'));
+      second.unmount();
+    });
+  });
 });