Warn when a behavior is re-entered while delivering a value

rxjs delivers a nested emission to every subscriber and then resumes
delivering the outer, older value to the remaining subscribers, which
leaves them permanently out of sync. Log the first occurrence per
behavior with a stack trace so the re-entrant path can be identified
from a rageshake.
This commit is contained in:
Matthew Hodgson
2026-09-03 17:26:38 +01:00
parent 0b95fd5a65
commit 2ab2197c01
2 changed files with 50 additions and 1 deletions

View File

@@ -237,3 +237,30 @@ describe("Reconcile", () => {
expect(setup).toHaveBeenCalledWith(1); 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<number>();
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();
});
});

View File

@@ -20,6 +20,8 @@ import {
takeUntil, takeUntil,
} from "rxjs"; } from "rxjs";
import { logger } from "matrix-js-sdk/lib/logger";
import { type Behavior } from "./Behavior"; import { type Behavior } from "./Behavior";
type MonoTypeOperator = <T>(o: Observable<T>) => Observable<T>; type MonoTypeOperator = <T>(o: Observable<T>) => Observable<T>;
@@ -73,9 +75,29 @@ export class ObservableScope {
// they will no longer re-emit their current value upon subscription. We want // 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 // to support Observables that complete (for example `of({})`), so we have to
// take care to not propagate the completion event. // 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({ setValue$.pipe(this.bind(), distinctUntilChanged()).subscribe({
next(value) { 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); subject$.next(value);
} finally {
delivering = wasDelivering;
}
}, },
error(err: unknown) { error(err: unknown) {
subject$.error(err); subject$.error(err);