From 565514f467cb80837f7197065221cd37354c2a78 Mon Sep 17 00:00:00 2001 From: Tyler Dixon Date: Wed, 11 Nov 2026 11:58:39 -0800 Subject: [PATCH 1/3] fix: complete direct subscribers of preloaded observables on error Since #839, a direct subscriber of a preloadFirestoreDoc or preloadObservable result received next(undefined) on a source error and never completed, so firstValueFrom resolved undefined and lastValueFrom never settled. 4.3.0 completed. useObservable now subscribes through _subscribeToStore, which notifies without completing, and every other subscriber completes on error again. --- src/SuspenseSubject.ts | 25 ++++++++++----------- src/useObservable.ts | 13 +---------- test/suspenseSubject-error-branch.test.tsx | 26 ++++++++++++++++++++-- 3 files changed, 37 insertions(+), 27 deletions(-) diff --git a/src/SuspenseSubject.ts b/src/SuspenseSubject.ts index 4514a586..c04210ae 100644 --- a/src/SuspenseSubject.ts +++ b/src/SuspenseSubject.ts @@ -147,25 +147,24 @@ export class SuspenseSubject extends Subject { } _subscribe(subscriber: Subscriber): Subscription { + // `catchError(() => EMPTY)` converts upstream errors into completions, so a direct subscriber completes on error. + return this._attach((value) => subscriber.next(value), () => subscriber.complete()); + } + + // For `useObservable`. Completing a `Subscriber` auto-unsubscribes it, which on error would evict the entry + // 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); + } + + private _attach(next: (value: T) => void, complete: () => void): 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(() => { diff --git a/src/useObservable.ts b/src/useObservable.ts index 42575419..296ddb76 100644 --- a/src/useObservable.ts +++ b/src/useObservable.ts @@ -99,18 +99,7 @@ export function useObservable(observableId: string, source: Observa } 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(); diff --git a/test/suspenseSubject-error-branch.test.tsx b/test/suspenseSubject-error-branch.test.tsx index 1119c44f..ee141e91 100644 --- a/test/suspenseSubject-error-branch.test.tsx +++ b/test/suspenseSubject-error-branch.test.tsx @@ -1,9 +1,9 @@ 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 { EmptyError, lastValueFrom, Observable, 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 @@ -54,4 +54,26 @@ describe('SuspenseSubject error-to-completion branch', () => { consoleError.mockRestore(); } }, 10000); + + // That branch is for `useObservable` only. 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); + }); }); From 4738b10d9b94a5fc74c8ce89cc29fba0dc0958de Mon Sep 17 00:00:00 2001 From: Armando Navarro Date: Wed, 11 Nov 2026 14:36:36 -0800 Subject: [PATCH 2/3] fix: keep an errored entry when a direct subscriber completes Completing a direct subscriber on a source error unsubscribes it, and its teardown evicted the errored entry. A Suspense render suspended on the same id then retried onto a new entry and read the source again, without end when the subscription was made in that component's render body. 4.3.0 never evicted here. Only the useObservable subscription now evicts an errored entry on unsubscribe. Adds tests for the render-body case and for a value followed by an error, and updates the test file's comments. --- src/SuspenseSubject.ts | 13 ++--- test/suspenseSubject-error-branch.test.tsx | 58 +++++++++++++++++++--- 2 files changed, 58 insertions(+), 13 deletions(-) diff --git a/src/SuspenseSubject.ts b/src/SuspenseSubject.ts index c04210ae..4bb74133 100644 --- a/src/SuspenseSubject.ts +++ b/src/SuspenseSubject.ts @@ -148,16 +148,16 @@ export class SuspenseSubject extends Subject { _subscribe(subscriber: Subscriber): Subscription { // `catchError(() => EMPTY)` converts upstream errors into completions, so a direct subscriber completes on error. - return this._attach((value) => subscriber.next(value), () => subscriber.complete()); + return this._attach((value) => subscriber.next(value), () => subscriber.complete(), false); } - // For `useObservable`. Completing a `Subscriber` auto-unsubscribes it, which on error would evict the entry - // while the hook is still mounted, so this reports every change through `onStoreChange` and never completes. + // 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); + return this._attach(onStoreChange, onStoreChange, true); } - private _attach(next: (value: T) => void, complete: () => void): Subscription { + 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) { @@ -170,7 +170,8 @@ export class SuspenseSubject extends Subject { 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(); } }); diff --git a/test/suspenseSubject-error-branch.test.tsx b/test/suspenseSubject-error-branch.test.tsx index ee141e91..231858aa 100644 --- a/test/suspenseSubject-error-branch.test.tsx +++ b/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 { EmptyError, lastValueFrom, Observable, timer, throwError } from 'rxjs'; +import { concat, EmptyError, lastValueFrom, Observable, of, timer, throwError } from 'rxjs'; import { mergeMap } from 'rxjs/operators'; 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 auto-unsubscribes the mounted subscriber and a +// mounted hook with `initialData` then resubscribes in a loop. `_subscribe` completes a direct subscriber. +describe('SuspenseSubject source errors', () => { afterEach(() => cleanup()); class Boundary extends React.Component<{ children: React.ReactNode }, { error: Error | null }> { @@ -55,7 +54,7 @@ describe('SuspenseSubject error-to-completion branch', () => { } }, 10000); - // That branch is for `useObservable` only. A direct subscriber of a `preload*` result completes, as in 4.3.0. + // 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()}`); @@ -76,4 +75,49 @@ describe('SuspenseSubject error-to-completion branch', () => { 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((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
{data}
; + }; + + const { getByTestId } = render( + + }> + + + + ); + + await waitFor(() => expect(getByTestId('err')).toHaveTextContent('permission-denied')); + expect(subscriptions).toBe(1); + } finally { + consoleError.mockRestore(); + } + }); }); From 49b6afe178bfb4a97eeb81b7547e2e1c0dfaba76 Mon Sep 17 00:00:00 2001 From: Tyler Dixon Date: Wed, 11 Nov 2026 15:28:39 -0800 Subject: [PATCH 3/3] test: correct the error-branch header comment --- test/suspenseSubject-error-branch.test.tsx | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/test/suspenseSubject-error-branch.test.tsx b/test/suspenseSubject-error-branch.test.tsx index 231858aa..26365e2c 100644 --- a/test/suspenseSubject-error-branch.test.tsx +++ b/test/suspenseSubject-error-branch.test.tsx @@ -6,8 +6,8 @@ import { mergeMap } from 'rxjs/operators'; import { preloadObservable, useObservable } from '../src/index'; // `catchError(() => EMPTY)` turns a source error into a completion. `_subscribeToStore` reports it to -// `useObservable` without completing, because completing auto-unsubscribes the mounted subscriber and a -// mounted hook with `initialData` then resubscribes in a loop. `_subscribe` completes a direct subscriber. +// `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());