diff --git a/src/state/ObservableScope.test.ts b/src/state/ObservableScope.test.ts index 31728f394..a1a03152c 100644 --- a/src/state/ObservableScope.test.ts +++ b/src/state/ObservableScope.test.ts @@ -237,3 +237,32 @@ describe("Reconcile", () => { expect(setup).toHaveBeenCalledWith(1); }); }); + +describe("behavior", () => { + it("delivers a re-entrant emission to every subscriber, after the current one", () => { + const warn = vi.spyOn(logger, "warn").mockImplementation(() => {}); + const scope = new ObservableScope(); + const source$ = new Subject(); + const behavior$ = scope.behavior(source$, 0); + const seenFirst: number[] = []; + // A subscriber that reacts to the value 1 by synchronously emitting 2 + behavior$.subscribe((v) => { + seenFirst.push(v); + if (v === 1) source$.next(2); + }); + const seenSecond: number[] = []; + behavior$.subscribe((v) => seenSecond.push(v)); + + source$.next(1); + + // Without queueing the second subscriber would be left on 1 + expect(seenFirst).toEqual([0, 1, 2]); + expect(seenSecond).toEqual([0, 1, 2]); + expect(behavior$.value).toBe(2); + expect(warn).toHaveBeenCalledWith( + expect.stringContaining("Behavior re-entered"), + expect.any(String), + ); + scope.end(); + }); +}); diff --git a/src/state/ObservableScope.ts b/src/state/ObservableScope.ts index e3fc644f7..9e0c87717 100644 --- a/src/state/ObservableScope.ts +++ b/src/state/ObservableScope.ts @@ -20,6 +20,8 @@ import { takeUntil, } from "rxjs"; +import { logger } from "matrix-js-sdk/lib/logger"; + import { type Behavior } from "./Behavior"; type MonoTypeOperator = (o: Observable) => Observable; @@ -73,9 +75,34 @@ export class ObservableScope { // they will no longer re-emit their current value upon subscription. We want // to support Observables that complete (for example `of({})`), so we have to // take care to not propagate the completion event. + // If a subscriber synchronously causes this same behavior to emit again, + // rxjs would deliver the nested value to every subscriber and then resume + // delivering the outer (older) value to the remaining subscribers, leaving + // them permanently out of sync. Queue nested emissions instead so that + // every subscriber sees every value, in order, and ends on the latest one. + const pending: T[] = []; + let delivering = false; + let reentryReported = false; setValue$.pipe(this.bind(), distinctUntilChanged()).subscribe({ next(value) { - subject$.next(value); + pending.push(value); + if (delivering) { + if (!reentryReported) { + reentryReported = true; + logger.warn( + "Behavior re-entered while delivering a value; the nested value has been queued", + new Error().stack, + ); + } + return; + } + delivering = true; + try { + while (pending.length > 0) subject$.next(pending.shift()!); + } finally { + delivering = false; + pending.length = 0; + } }, error(err: unknown) { subject$.error(err);