From 36d66432b92cc9154f9404418ee525be2ddfcd7e Mon Sep 17 00:00:00 2001
From: Jeff Huleatt <3759507+jhuleatt@users.noreply.github.com>
Date: Tue, 27 Oct 2026 19:05:13 +0000
Subject: [PATCH 1/5] fix(core): release subscriptions on unmount and evict
 errored cache entries

---
 src/SuspenseSubject.ts      | 62 ++++++++++++++++-----------
 src/useObservable.ts        |  2 +-
 test/useObservable.test.tsx | 84 ++++++++++++++++++++++++++++++++++++-
 3 files changed, 122 insertions(+), 26 deletions(-)

diff --git a/src/SuspenseSubject.ts b/src/SuspenseSubject.ts
index 5b7ef884..6b6f4a1a 100644
--- a/src/SuspenseSubject.ts
+++ b/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,6 +51,8 @@ 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;
@@ -49,19 +60,32 @@ export class SuspenseSubject<T> extends Subject<T> {
         }
       }),
       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;
   }
 
diff --git a/src/useObservable.ts b/src/useObservable.ts
index 237f6e35..53816fcd 100644
--- a/src/useObservable.ts
+++ b/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;
   }
diff --git a/test/useObservable.test.tsx b/test/useObservable.test.tsx
index ada53ab5..5f2018d6 100644
--- a/test/useObservable.test.tsx
+++ b/test/useObservable.test.tsx
@@ -3,7 +3,7 @@ import { act, cleanup, render, renderHook, waitFor } from '@testing-library/reac
 import * as React from 'react';
 import { Writable } from 'node:stream';
 import { renderToString, renderToPipeableStream } from 'react-dom/server';
-import { of, Subject, BehaviorSubject, throwError } from 'rxjs';
+import { Observable, of, Subject, BehaviorSubject, throwError } from 'rxjs';
 import { useObservable, ReactFireOptions } from '../src/index';
 
 describe('useObservable', () => {
@@ -427,4 +427,86 @@ 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;
+      const source$ = new Observable<string>((subscriber) => {
+        if (shouldFail) {
+          subscriber.error(new Error('permission-denied'));
+        } else {
+          subscriber.next('recovered');
+        }
+      });
+
+      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 { 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();
+    });
+  });
 });

From b092103ca85d0a342d51af1be40e255bb604c63e Mon Sep 17 00:00:00 2001
From: Jeff Huleatt <3759507+jhuleatt@users.noreply.github.com>
Date: Tue, 27 Oct 2026 19:59:32 +0000
Subject: [PATCH 2/5] test(core): add error-after-unmount test (#485) and avoid
 top-level import conflict

---
 test/useObservable.test.tsx | 39 +++++++++++++++++++++++++++++--------
 1 file changed, 31 insertions(+), 8 deletions(-)

diff --git a/test/useObservable.test.tsx b/test/useObservable.test.tsx
index 5f2018d6..298ebad8 100644
--- a/test/useObservable.test.tsx
+++ b/test/useObservable.test.tsx
@@ -3,7 +3,7 @@ import { act, cleanup, render, renderHook, waitFor } from '@testing-library/reac
 import * as React from 'react';
 import { Writable } from 'node:stream';
 import { renderToString, renderToPipeableStream } from 'react-dom/server';
-import { Observable, of, Subject, BehaviorSubject, throwError } from 'rxjs';
+import { of, Subject, BehaviorSubject, throwError } from 'rxjs';
 import { useObservable, ReactFireOptions } from '../src/index';
 
 describe('useObservable', () => {
@@ -460,13 +460,6 @@ describe('useObservable', () => {
     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;
-      const source$ = new Observable<string>((subscriber) => {
-        if (shouldFail) {
-          subscriber.error(new Error('permission-denied'));
-        } else {
-          subscriber.next('recovered');
-        }
-      });
 
       class Boundary extends React.Component<{ children: React.ReactNode }, { error: Error | null }> {
         state: { error: Error | null } = { error: null };
@@ -482,6 +475,7 @@ describe('useObservable', () => {
       }
 
       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>;
       };
@@ -508,5 +502,34 @@ describe('useObservable', () => {
       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();
+    });
   });
 });

From ae14caaab5e5e1446ab886914da7d67c386917c6 Mon Sep 17 00:00:00 2001
From: Jeff Huleatt <3759507+jhuleatt@users.noreply.github.com>
Date: Wed, 28 Oct 2026 14:41:24 +0000
Subject: [PATCH 3/5] fix(core): require rxjs ^7.1.0, evict unconsumed
 completed warmups, and add lifecycle tests

---
 package-lock.json           |  2 +-
 package.json                |  2 +-
 src/SuspenseSubject.ts      | 27 ++++++------
 test/useObservable.test.tsx | 83 +++++++++++++++++++++++++++++++++++++
 4 files changed, 100 insertions(+), 14 deletions(-)

diff --git a/package-lock.json b/package-lock.json
index 1e974b3b..87698f63 100644
--- a/package-lock.json
+++ b/package-lock.json
@@ -10,7 +10,7 @@
       "license": "MIT",
       "dependencies": {
         "rxfire": "^6.1.0",
-        "rxjs": "^6.6.3 || ^7.0.1",
+        "rxjs": "^7.1.0",
         "use-sync-external-store": "^1.2.0"
       },
       "devDependencies": {
diff --git a/package.json b/package.json
index 6b6b707f..82e7ed22 100644
--- a/package.json
+++ b/package.json
@@ -108,7 +108,7 @@
   },
   "dependencies": {
     "rxfire": "^6.1.0",
-    "rxjs": "^6.6.3 || ^7.0.1",
+    "rxjs": "^7.1.0",
     "use-sync-external-store": "^1.2.0"
   },
   "allowScripts": {
diff --git a/src/SuspenseSubject.ts b/src/SuspenseSubject.ts
index 6b6f4a1a..9d7c2cc5 100644
--- a/src/SuspenseSubject.ts
+++ b/src/SuspenseSubject.ts
@@ -1,4 +1,4 @@
-import { empty, Observable, ReplaySubject, Subject, Subscriber, Subscription, timer } from 'rxjs';
+import { EMPTY, Observable, ReplaySubject, Subject, Subscriber, Subscription, timer } from 'rxjs';
 import { catchError, share, tap } from 'rxjs/operators';
 import { ObservableStatus } from './useObservable.js';
 
@@ -52,6 +52,7 @@ export class SuspenseSubject<T> extends Subject<T> {
           this._resolveFirstEmission();
           this._updateImmutableStatus();
           // Evict the errored entry after the current render reads it so a remount or ErrorBoundary retry starts fresh (#742, #485).
+          // `catchError(() => EMPTY)` below converts errors into completions before `share` sees them, so this timer alone evicts errored entries.
           setTimeout(onEvict, RELEASE_DELAY_MS);
         },
         complete: () => {
@@ -59,12 +60,14 @@ export class SuspenseSubject<T> extends Subject<T> {
           this._updateImmutableStatus();
         }
       }),
-      catchError(() => empty()),
-      // 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`.
+      catchError(() => EMPTY),
+      // Multicast one Firebase listener across all subscribers:
+      // - `connector`: replay the latest emission to new subscribers on mount.
+      // - `resetOnError`: tear down the source subscription on error (upstream `catchError` converts errors to completions first).
+      // - `resetOnComplete`: keep the final value cached when a one-shot stream (`first()` in `*Once` hooks) completes,
+      //   because RxJS auto-closes the subscriber on completion while the component remains mounted; evicting on completion
+      //   would cause mounted `*Once` components to re-fetch on subsequent renders.
+      // - `resetOnRefCountZero`: detach the Firebase listener and evict the cache entry after `RELEASE_DELAY_MS` with zero subscribers.
       share({
         connector: () => new ReplaySubject<T>(1),
         resetOnError: true,
@@ -76,11 +79,14 @@ export class SuspenseSubject<T> extends Subject<T> {
     this._warmupSubscription = this._innerObservable.subscribe();
 
     // 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).
+    // Wait until the first emission settles so a suspended component can resolve `firstEmission`, re-render, and call `_subscribe` (#525).
+    // If the stream already completed during warm-up (`first()`), `share` skips `resetOnRefCountZero`, so call `onEvict` directly (#790).
     const releaseWarmupWhenSettled = () => {
       if (this._hasValue || this._error) {
         this._warmupSubscription.unsubscribe();
+        if (this._isComplete) {
+          onEvict();
+        }
       } else {
         this._timeoutHandler = setTimeout(releaseWarmupWhenSettled, this._timeoutWindow);
       }
@@ -96,9 +102,6 @@ export class SuspenseSubject<T> extends Subject<T> {
   }
 
   get value(): T {
-    // TODO figure out how to reset the cache here, if I _reset() here before throwing
-    // it doesn't seem to work.
-    // As it is now, this will burn the cache entry until the timeout fires.
     if (this._error) {
       throw this._error;
     } else if (!this.hasValue) {
diff --git a/test/useObservable.test.tsx b/test/useObservable.test.tsx
index 298ebad8..7fbcea5b 100644
--- a/test/useObservable.test.tsx
+++ b/test/useObservable.test.tsx
@@ -531,5 +531,88 @@ describe('useObservable', () => {
       await waitFor(() => expect(second.getByTestId('val')).toHaveTextContent('reauthenticated-user-doc'));
       second.unmount();
     });
+
+    it('reuses the active subscription when remounting within the 1s release grace window', async () => {
+      const observableId = `grace-window-${Math.random()}`;
+      let subscribeCount = 0;
+      const base$ = new BehaviorSubject<string>('cached-val');
+      const counted$ = base$.lift({
+        call(subscriber, source) {
+          subscribeCount++;
+          return source.subscribe(subscriber);
+        }
+      });
+
+      const Obs = () => {
+        const { data } = useObservable(observableId, counted$);
+        return <div data-testid="val">{String(data)}</div>;
+      };
+
+      const first = render(<Obs />);
+      await waitFor(() => expect(first.getByTestId('val')).toHaveTextContent('cached-val'));
+      expect(subscribeCount).toBe(1);
+      first.unmount();
+
+      // Remount after 200ms (inside the 1000ms grace window); must reuse the active subscription.
+      await new Promise((r) => setTimeout(r, 200));
+      const second = render(<Obs />);
+      await waitFor(() => expect(second.getByTestId('val')).toHaveTextContent('cached-val'));
+      expect(subscribeCount).toBe(1);
+      second.unmount();
+
+      // After the 1000ms grace window expires, the next mount subscribes fresh.
+      await new Promise((r) => setTimeout(r, 1100));
+      const third = render(<Obs />);
+      await waitFor(() => expect(third.getByTestId('val')).toHaveTextContent('cached-val'));
+      expect(subscribeCount).toBe(2);
+      third.unmount();
+    });
+
+    it('releases an unconsumed warm-up subscription only after the first emission settles (#525, #790)', async () => {
+      const { SuspenseSubject } = await import('../src/SuspenseSubject');
+      const source$ = new Subject<string>();
+      let evicted = false;
+      new SuspenseSubject(source$, 50, true, () => {
+        evicted = true;
+      });
+
+      // After the first 50ms window, the first emission is still pending, so warm-up stays subscribed (#525).
+      await new Promise((r) => setTimeout(r, 80));
+      expect(source$.observed).toBe(true);
+
+      // Settle and complete the stream without any subscriber mounting (e.g. SSR or unused preload).
+      source$.next('settled');
+      source$.complete();
+
+      // On the next 50ms check, releaseWarmupWhenSettled unsubscribes warm-up and evicts the completed entry (#790).
+      await waitFor(() => expect(evicted).toBe(true), { timeout: 500 });
+      expect(source$.observed).toBe(false);
+    });
+
+    it('preserves a newer cache entry when a late subscriber on an evicted subject unsubscribes', async () => {
+      const { preloadObservable } = await import('../src/useObservable');
+      const observableId = `identity-guard-${Math.random()}`;
+
+      const firstSubject = preloadObservable(new BehaviorSubject('first'), observableId);
+      const initialSub = firstSubject.subscribe();
+      initialSub.unsubscribe();
+
+      // Wait for firstSubject to be evicted from preloadedObservables.
+      await new Promise((r) => setTimeout(r, 1100));
+
+      // Register a new subject under the same observableId and keep a subscriber active.
+      const secondSubject = preloadObservable(new BehaviorSubject('second'), observableId);
+      expect(secondSubject).not.toBe(firstSubject);
+      const activeSub = secondSubject.subscribe();
+
+      // A late caller holding the evicted firstSubject subscribes and unsubscribes, re-arming firstSubject's eviction timer.
+      const lateSub = firstSubject.subscribe();
+      lateSub.unsubscribe();
+      await new Promise((r) => setTimeout(r, 1100));
+
+      // The identity guard in onEvict must keep secondSubject in the cache.
+      expect(preloadObservable(new BehaviorSubject('third'), observableId)).toBe(secondSubject);
+      activeSub.unsubscribe();
+    });
   });
 });

From 8f99728d12dcd637235c8023ea3dd454ed2406d9 Mon Sep 17 00:00:00 2001
From: Jeff Huleatt <3759507+jhuleatt@users.noreply.github.com>
Date: Wed, 28 Oct 2026 15:53:23 +0000
Subject: [PATCH 4/5] fix(core): evict immediately when an unmounted listener
 errors (#485)

---
 src/SuspenseSubject.ts      | 40 ++++++++++++++++++++++++++++---------
 test/useObservable.test.tsx |  3 +--
 2 files changed, 32 insertions(+), 11 deletions(-)

diff --git a/src/SuspenseSubject.ts b/src/SuspenseSubject.ts
index 9d7c2cc5..654d9bdf 100644
--- a/src/SuspenseSubject.ts
+++ b/src/SuspenseSubject.ts
@@ -16,6 +16,7 @@ export class SuspenseSubject<T> extends Subject<T> {
   private _warmupSubscription: Subscription;
   private _immutableStatus: ObservableStatus<T>;
   private _isComplete = false;
+  private _subscriberCount = 0;
 
   // @ts-expect-error: TODO: double check to see if this is an RXJS thing or if we should listen to TS
   private _innerSubscriber: Subscription;
@@ -26,7 +27,7 @@ export class SuspenseSubject<T> extends Subject<T> {
     innerObservable: Observable<T>,
     private _timeoutWindow: number,
     _suspenseEnabled: boolean,
-    onEvict: () => void = () => {}
+    private _onEvict: () => void = () => {}
   ) {
     super();
     this._firstEmission = new Promise<void>((resolve) => (this._resolveFirstEmission = resolve));
@@ -51,9 +52,15 @@ 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).
-          // `catchError(() => EMPTY)` below converts errors into completions before `share` sees them, so this timer alone evicts errored entries.
-          setTimeout(onEvict, RELEASE_DELAY_MS);
+          // If all subscribers have already unmounted (such as a listener rejected after sign-out during the release grace window),
+          // evict immediately so a fast re-sign-in never reads the stale error (#485).
+          // Otherwise, keep the entry for `RELEASE_DELAY_MS` so a suspended or active render can read and throw it (#742).
+          // `catchError(() => EMPTY)` below converts errors into completions before `share` sees them, so `tap.error` and unmount handle eviction.
+          if (this._subscriberCount === 0 && this._warmupSubscription?.closed) {
+            this._onEvict();
+          } else {
+            setTimeout(this._onEvict, RELEASE_DELAY_MS);
+          }
         },
         complete: () => {
           this._isComplete = true;
@@ -72,7 +79,7 @@ export class SuspenseSubject<T> extends Subject<T> {
         connector: () => new ReplaySubject<T>(1),
         resetOnError: true,
         resetOnComplete: false,
-        resetOnRefCountZero: () => timer(RELEASE_DELAY_MS).pipe(tap(onEvict)),
+        resetOnRefCountZero: () => timer(RELEASE_DELAY_MS).pipe(tap(this._onEvict)),
       })
     );
     // warm up the observable
@@ -80,12 +87,12 @@ export class SuspenseSubject<T> extends Subject<T> {
 
     // Release the warm-up subscription if no component subscribes within `_timeoutWindow` after the first emission settles.
     // Wait until the first emission settles so a suspended component can resolve `firstEmission`, re-render, and call `_subscribe` (#525).
-    // If the stream already completed during warm-up (`first()`), `share` skips `resetOnRefCountZero`, so call `onEvict` directly (#790).
+    // If the stream already completed during warm-up (`first()`), `share` skips `resetOnRefCountZero`, so call `_onEvict` directly (#790).
     const releaseWarmupWhenSettled = () => {
       if (this._hasValue || this._error) {
         this._warmupSubscription.unsubscribe();
         if (this._isComplete) {
-          onEvict();
+          this._onEvict();
         }
       } else {
         this._timeoutHandler = setTimeout(releaseWarmupWhenSettled, this._timeoutWindow);
@@ -139,10 +146,25 @@ export class SuspenseSubject<T> extends Subject<T> {
     if (this._timeoutHandler) {
       clearTimeout(this._timeoutHandler);
     }
-    this._innerSubscriber = this._innerObservable.subscribe(subscriber);
+    this._subscriberCount++;
+    this._innerSubscriber = this._innerObservable.subscribe({
+      next: (v) => subscriber.next(v),
+      error: (e) => subscriber.error(e),
+      complete: () => {
+        // Notify `useObservable` (`onStoreChange`) without calling `subscriber.complete()`,
+        // because `subscriber.complete()` auto-unsubscribes `subscriber` before React unmounts.
+        subscriber.next(this._value as T);
+      },
+    });
     // Hand off ownership of the shared source from the warm-up subscription to the real subscriber.
     this._warmupSubscription.unsubscribe();
-    return this._innerSubscriber;
+    return new Subscription(() => {
+      this._innerSubscriber.unsubscribe();
+      this._subscriberCount--;
+      if (this._subscriberCount === 0 && this._error) {
+        this._onEvict();
+      }
+    });
   }
 
   get ourError() {
diff --git a/test/useObservable.test.tsx b/test/useObservable.test.tsx
index 7fbcea5b..b0b35ea0 100644
--- a/test/useObservable.test.tsx
+++ b/test/useObservable.test.tsx
@@ -518,13 +518,12 @@ describe('useObservable', () => {
       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.
+      // while the 1s release grace window is still active, and remount immediately without waiting 1s.
       first.unmount();
       act(() => {
         firstStream$.error(new Error('permission-denied-after-unmount'));
       });
 
-      await new Promise((r) => setTimeout(r, 1100));
       currentStream$ = secondStream$;
 
       const second = render(<Obs />);

From d9d2a9736572a3fdf49434545ffbdd7b77639c90 Mon Sep 17 00:00:00 2001
From: Jeff Huleatt <3759507+jhuleatt@users.noreply.github.com>
Date: Wed, 28 Oct 2026 16:15:46 +0000
Subject: [PATCH 5/5] fix(core): scope inner subscription teardown per
 subscriber in SuspenseSubject

---
 src/SuspenseSubject.ts      | 30 ++++++++++++------
 test/useObservable.test.tsx | 63 +++++++++++++++++++++++++++++++++++--
 2 files changed, 80 insertions(+), 13 deletions(-)

diff --git a/src/SuspenseSubject.ts b/src/SuspenseSubject.ts
index 654d9bdf..b9aed867 100644
--- a/src/SuspenseSubject.ts
+++ b/src/SuspenseSubject.ts
@@ -4,7 +4,11 @@ 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;
+const LISTENER_RELEASE_DELAY_MS = 1000;
+
+// Fallback retention window for an error that arrives during an initial Suspense load before a component subscribes,
+// giving React time to retry the suspended render and throw the error to an Error Boundary.
+const FIRST_LOAD_ERROR_EVICT_DELAY_MS = 1000;
 
 export class SuspenseSubject<T> extends Subject<T> {
   private _value: T | undefined;
@@ -54,12 +58,12 @@ export class SuspenseSubject<T> extends Subject<T> {
           this._updateImmutableStatus();
           // If all subscribers have already unmounted (such as a listener rejected after sign-out during the release grace window),
           // evict immediately so a fast re-sign-in never reads the stale error (#485).
-          // Otherwise, keep the entry for `RELEASE_DELAY_MS` so a suspended or active render can read and throw it (#742).
+          // Otherwise, keep the entry for `FIRST_LOAD_ERROR_EVICT_DELAY_MS` so a suspended first load or active render can read and throw it (#742).
           // `catchError(() => EMPTY)` below converts errors into completions before `share` sees them, so `tap.error` and unmount handle eviction.
           if (this._subscriberCount === 0 && this._warmupSubscription?.closed) {
             this._onEvict();
           } else {
-            setTimeout(this._onEvict, RELEASE_DELAY_MS);
+            setTimeout(this._onEvict, FIRST_LOAD_ERROR_EVICT_DELAY_MS);
           }
         },
         complete: () => {
@@ -74,12 +78,12 @@ export class SuspenseSubject<T> extends Subject<T> {
       // - `resetOnComplete`: keep the final value cached when a one-shot stream (`first()` in `*Once` hooks) completes,
       //   because RxJS auto-closes the subscriber on completion while the component remains mounted; evicting on completion
       //   would cause mounted `*Once` components to re-fetch on subsequent renders.
-      // - `resetOnRefCountZero`: detach the Firebase listener and evict the cache entry after `RELEASE_DELAY_MS` with zero subscribers.
+      // - `resetOnRefCountZero`: detach the Firebase listener and evict the cache entry after `LISTENER_RELEASE_DELAY_MS` with zero subscribers.
       share({
         connector: () => new ReplaySubject<T>(1),
         resetOnError: true,
         resetOnComplete: false,
-        resetOnRefCountZero: () => timer(RELEASE_DELAY_MS).pipe(tap(this._onEvict)),
+        resetOnRefCountZero: () => timer(LISTENER_RELEASE_DELAY_MS).pipe(tap(this._onEvict)),
       })
     );
     // warm up the observable
@@ -147,19 +151,25 @@ export class SuspenseSubject<T> extends Subject<T> {
       clearTimeout(this._timeoutHandler);
     }
     this._subscriberCount++;
-    this._innerSubscriber = this._innerObservable.subscribe({
+    const innerSubscription = this._innerObservable.subscribe({
       next: (v) => subscriber.next(v),
       error: (e) => subscriber.error(e),
       complete: () => {
-        // Notify `useObservable` (`onStoreChange`) without calling `subscriber.complete()`,
-        // because `subscriber.complete()` auto-unsubscribes `subscriber` before React unmounts.
-        subscriber.next(this._value as T);
+        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 as T);
+        } else {
+          subscriber.complete();
+        }
       },
     });
     // Hand off ownership of the shared source from the warm-up subscription to the real subscriber.
     this._warmupSubscription.unsubscribe();
+    this._innerSubscriber = innerSubscription;
     return new Subscription(() => {
-      this._innerSubscriber.unsubscribe();
+      innerSubscription.unsubscribe();
       this._subscriberCount--;
       if (this._subscriberCount === 0 && this._error) {
         this._onEvict();
diff --git a/test/useObservable.test.tsx b/test/useObservable.test.tsx
index b0b35ea0..2278fb31 100644
--- a/test/useObservable.test.tsx
+++ b/test/useObservable.test.tsx
@@ -509,12 +509,26 @@ describe('useObservable', () => {
       const secondStream$ = new BehaviorSubject<string>('reauthenticated-user-doc');
       let currentStream$ = firstStream$;
 
+      class Boundary extends React.Component<{ children: React.ReactNode }, { error: Error | null }> {
+        state: { error: Error | null } = { error: null };
+        static getDerivedStateFromError(error: Error) {
+          return { error };
+        }
+        render() {
+          return this.state.error ? <div data-testid="err">{this.state.error.message}</div> : this.props.children;
+        }
+      }
+
       const Obs = () => {
         const { data, status } = useObservable(observableId, currentStream$);
         return <div data-testid="val">{status === 'success' ? data : 'loading'}</div>;
       };
 
-      const first = render(<Obs />);
+      const first = render(
+        <Boundary>
+          <Obs />
+        </Boundary>
+      );
       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
@@ -526,9 +540,52 @@ describe('useObservable', () => {
 
       currentStream$ = secondStream$;
 
-      const second = render(<Obs />);
-      await waitFor(() => expect(second.getByTestId('val')).toHaveTextContent('reauthenticated-user-doc'));
+      const second = render(
+        <Boundary>
+          <Obs />
+        </Boundary>
+      );
+      await waitFor(() => expect(second.queryByTestId('val')).toHaveTextContent('reauthenticated-user-doc'));
+      second.unmount();
+    });
+
+    it('keeps updating the remaining subscriber when another subscriber of the same observableId unmounts, and releases the source after both unmount', async () => {
+      const subject$ = new BehaviorSubject<string>('a');
+      const observableId = `shared-subscriber-update-${Math.random()}`;
+
+      const Obs = ({ testId }: { testId: string }) => {
+        const { data } = useObservable(observableId, subject$);
+        return <div data-testid={testId}>{String(data)}</div>;
+      };
+
+      const first = render(<Obs testId="one" />);
+      const second = render(<Obs testId="two" />);
+      await waitFor(() => expect(second.getByTestId('two')).toHaveTextContent('a'));
+
+      first.unmount();
+      act(() => subject$.next('b'));
+      await waitFor(() => expect(second.getByTestId('two')).toHaveTextContent('b'), { timeout: 1000 });
+
+      second.unmount();
+      await waitFor(() => expect(subject$.observed).toBe(false), { timeout: 2500 });
+    });
+
+    it('detaches the source after two subscribers of the same observableId unmount', async () => {
+      const subject$ = new BehaviorSubject<string>('a');
+      const observableId = `shared-subscriber-release-${Math.random()}`;
+
+      const Obs = ({ testId }: { testId: string }) => {
+        const { data } = useObservable(observableId, subject$);
+        return <div data-testid={testId}>{String(data)}</div>;
+      };
+
+      const first = render(<Obs testId="one" />);
+      const second = render(<Obs testId="two" />);
+      await waitFor(() => expect(second.getByTestId('two')).toHaveTextContent('a'));
+
+      first.unmount();
       second.unmount();
+      await waitFor(() => expect(subject$.observed).toBe(false), { timeout: 3000 });
     });
 
     it('reuses the active subscription when remounting within the 1s release grace window', async () => {