diff --git a/src/state/ObservableScope.test.ts b/src/state/ObservableScope.test.ts index 31728f394..353ecaed5 100644 --- a/src/state/ObservableScope.test.ts +++ b/src/state/ObservableScope.test.ts @@ -237,3 +237,30 @@ describe("Reconcile", () => { expect(setup).toHaveBeenCalledWith(1); }); }); + +describe("behavior", () => { + it("warns when a subscriber re-enters the behavior synchronously", () => { + const warn = vi.spyOn(logger, "warn").mockImplementation(() => {}); + const scope = new ObservableScope(); + const source$ = new Subject(); + const behavior$ = scope.behavior(source$, 0); + // A subscriber that reacts to the value 1 by synchronously emitting 2 + behavior$.subscribe((v) => { + if (v === 1) source$.next(2); + }); + const seen: number[] = []; + behavior$.subscribe((v) => seen.push(v)); + + source$.next(1); + + expect(warn).toHaveBeenCalledWith( + expect.stringContaining("Behavior re-entered"), + expect.any(String), + ); + // Documents the hazard the warning is about: the later subscriber ends up + // with the stale value 1 even though the behavior's value is 2. + expect(behavior$.value).toBe(2); + expect(seen.at(-1)).toBe(1); + scope.end(); + }); +}); diff --git a/src/state/ObservableScope.ts b/src/state/ObservableScope.ts index e3fc644f7..0a64e2a42 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,29 @@ 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 delivers the nested value to every subscriber first and then + // resumes delivering the outer (older) value to the remaining subscribers, + // leaving them permanently out of sync with the others. Log the first + // occurrence with a stack trace so that the re-entrant path can be found. + let delivering = false; + let reentryReported = false; setValue$.pipe(this.bind(), distinctUntilChanged()).subscribe({ next(value) { - subject$.next(value); + if (delivering && !reentryReported) { + reentryReported = true; + logger.warn( + "Behavior re-entered while delivering a value; later subscribers will be left with a stale value", + new Error().stack, + ); + } + const wasDelivering = delivering; + delivering = true; + try { + subject$.next(value); + } finally { + delivering = wasDelivering; + } }, error(err: unknown) { subject$.error(err);