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

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": {

src/SuspenseSubject.ts

@@ -1,7 +1,15 @@
-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 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;
   private _hasValue = false;
@@ -12,13 +20,19 @@ 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;
   // @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,
+    private _onEvict: () => void = () => {}
+  ) {
     super();
     this._firstEmission = new Promise<void>((resolve) => (this._resolveFirstEmission = resolve));
 
@@ -42,26 +56,53 @@ export class SuspenseSubject<T> extends Subject<T> {
           this._error = e;
           this._resolveFirstEmission();
           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 `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, FIRST_LOAD_ERROR_EVICT_DELAY_MS);
+          }
         },
         complete: () => {
           this._isComplete = true;
           this._updateImmutableStatus();
         }
       }),
-      catchError(() => empty()),
-      shareReplay(1)
+      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 `LISTENER_RELEASE_DELAY_MS` with zero subscribers.
+      share({
+        connector: () => new ReplaySubject<T>(1),
+        resetOnError: true,
+        resetOnComplete: false,
+        resetOnRefCountZero: () => timer(LISTENER_RELEASE_DELAY_MS).pipe(tap(this._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.
+    // 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) {
+          this._onEvict();
+        }
+      } else {
+        this._timeoutHandler = setTimeout(releaseWarmupWhenSettled, this._timeoutWindow);
+      }
+    };
+    this._timeoutHandler = setTimeout(releaseWarmupWhenSettled, this._timeoutWindow);
   }
 
   get hasValue(): boolean {
@@ -72,9 +113,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) {
@@ -108,24 +146,35 @@ 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);
-    return this._innerSubscriber;
+    this._subscriberCount++;
+    const innerSubscription = this._innerObservable.subscribe({
+      next: (v) => subscriber.next(v),
+      error: (e) => subscriber.error(e),
+      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 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(() => {
+      innerSubscription.unsubscribe();
+      this._subscriberCount--;
+      if (this._subscriberCount === 0 && this._error) {
+        this._onEvict();
+      }
+    });
   }
 
   get ourError() {

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;
   }