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 5b7ef884..b9aed867 100644 --- a/src/SuspenseSubject.ts +++ b/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 extends Subject { private _value: T | undefined; private _hasValue = false; @@ -12,13 +20,19 @@ export class SuspenseSubject extends Subject { private _warmupSubscription: Subscription; private _immutableStatus: ObservableStatus; 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, private _timeoutWindow: number, private _suspenseEnabled: boolean) { + constructor( + innerObservable: Observable, + private _timeoutWindow: number, + _suspenseEnabled: boolean, + private _onEvict: () => void = () => {} + ) { super(); this._firstEmission = new Promise((resolve) => (this._resolveFirstEmission = resolve)); @@ -42,26 +56,53 @@ export class SuspenseSubject extends Subject { 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(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 extends Subject { } 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 extends Subject { 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((resolve) => (this._resolveFirstEmission = resolve)); - this._updateImmutableStatus(); - } - _subscribe(subscriber: Subscriber): 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() { 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(source: Observable, id: string, suspense if (preloadedObservables.has(id)) { return preloadedObservables.get(id) as SuspenseSubject; } else { - const observable = new SuspenseSubject(source, DEFAULT_TIMEOUT, suspenseEnabled); + const observable: SuspenseSubject = 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..2278fb31 100644 --- a/test/useObservable.test.tsx +++ b/test/useObservable.test.tsx @@ -427,4 +427,248 @@ 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(); + const observableId = `sub-release-${suspense}-${Math.random()}`; + + const Obs = () => { + const { data } = useObservable(observableId, subject$, { suspense }); + return
{String(data)}
; + }; + + const { unmount, getByTestId } = render( + loading}> + + + ); + + 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
{this.state.error.message}
; + } + return this.props.children; + } + } + + const Obs = () => { + const source$ = shouldFail ? throwError(() => new Error('permission-denied')) : of('recovered'); + const { data, status } = useObservable(observableId, source$); + return
{status === 'success' ? data : 'loading'}
; + }; + + const first = render( + + + + ); + + 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( + + + + ); + + 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('initial-user-doc'); + const secondStream$ = new BehaviorSubject('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 ?
{this.state.error.message}
: this.props.children; + } + } + + const Obs = () => { + const { data, status } = useObservable(observableId, currentStream$); + return
{status === 'success' ? data : 'loading'}
; + }; + + const first = render( + + + + ); + 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, and remount immediately without waiting 1s. + first.unmount(); + act(() => { + firstStream$.error(new Error('permission-denied-after-unmount')); + }); + + currentStream$ = secondStream$; + + const second = render( + + + + ); + 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('a'); + const observableId = `shared-subscriber-update-${Math.random()}`; + + const Obs = ({ testId }: { testId: string }) => { + const { data } = useObservable(observableId, subject$); + return
{String(data)}
; + }; + + const first = render(); + const second = render(); + 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('a'); + const observableId = `shared-subscriber-release-${Math.random()}`; + + const Obs = ({ testId }: { testId: string }) => { + const { data } = useObservable(observableId, subject$); + return
{String(data)}
; + }; + + const first = render(); + const second = render(); + 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 () => { + const observableId = `grace-window-${Math.random()}`; + let subscribeCount = 0; + const base$ = new BehaviorSubject('cached-val'); + const counted$ = base$.lift({ + call(subscriber, source) { + subscribeCount++; + return source.subscribe(subscriber); + } + }); + + const Obs = () => { + const { data } = useObservable(observableId, counted$); + return
{String(data)}
; + }; + + const first = render(); + 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(); + 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(); + 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(); + 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(); + }); + }); });