Report re-entrancy depth and tag torn behaviors

A single re-entrant emission can only strand a contiguous run of a
behavior's subscribers, so a tile that is both 'speaking' and 'waiting
for media' (a middle subscriber stranded) implies a nested re-entry.
Report the nesting depth and log each new depth once instead of only the
first re-entry, and tag splitBehavior-derived behaviors with their field
name so the warning identifies which behavior tore.
This commit is contained in:
Matthew Hodgson
2026-09-03 18:08:53 +01:00
parent 2ab2197c01
commit b1393998a7
2 changed files with 38 additions and 21 deletions

View File

@@ -239,27 +239,33 @@ describe("Reconcile", () => {
}); });
describe("behavior", () => { describe("behavior", () => {
it("warns when a subscriber re-enters the behavior synchronously", () => { it("warns with the tag and nesting depth, logging each new depth once", () => {
const warn = vi.spyOn(logger, "warn").mockImplementation(() => {}); const warn = vi.spyOn(logger, "warn").mockImplementation(() => {});
const scope = new ObservableScope(); const scope = new ObservableScope();
const source$ = new Subject<number>(); const source$ = new Subject<number>();
const behavior$ = scope.behavior(source$, 0); const behavior$ = scope.behavior(source$, 0, "tagged");
// A subscriber that reacts to the value 1 by synchronously emitting 2 // Re-enter once on value 1 (depth 2), then a nested re-entry on value 2
// (depth 3), which is the shape needed to strand a middle subscriber.
behavior$.subscribe((v) => { behavior$.subscribe((v) => {
if (v === 1) source$.next(2); if (v === 1) source$.next(2);
}); });
behavior$.subscribe((v) => {
if (v === 2) source$.next(3);
});
const seen: number[] = []; const seen: number[] = [];
behavior$.subscribe((v) => seen.push(v)); behavior$.subscribe((v) => seen.push(v));
source$.next(1); source$.next(1);
expect(warn).toHaveBeenCalledWith( const messages = warn.mock.calls.map((c) => c[0] as string);
expect.stringContaining("Behavior re-entered"), expect(messages).toEqual([
expect.any(String), expect.stringContaining("Behavior (tagged) re-entered at depth 2"),
); expect.stringContaining("Behavior (tagged) re-entered at depth 3"),
// 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. warn.mock.calls.forEach((c) => expect(c[1]).toEqual(expect.any(String)));
expect(behavior$.value).toBe(2); // The behavior settles on the newest value while the last subscriber is
// left stranded on the oldest.
expect(behavior$.value).toBe(3);
expect(seen.at(-1)).toBe(1); expect(seen.at(-1)).toBe(1);
scope.end(); scope.end();
}); });

View File

@@ -68,6 +68,9 @@ export class ObservableScope {
public behavior<T>( public behavior<T>(
setValue$: Observable<T>, setValue$: Observable<T>,
initialValue: T | typeof nothing = nothing, initialValue: T | typeof nothing = nothing,
// An optional label used only in the re-entrancy warning below, so that a
// torn behavior can be identified from a rageshake.
tag?: string,
): Behavior<T> { ): Behavior<T> {
const subject$ = new BehaviorSubject(initialValue); const subject$ = new BehaviorSubject(initialValue);
// Push values from the Observable into the BehaviorSubject. // Push values from the Observable into the BehaviorSubject.
@@ -78,25 +81,29 @@ export class ObservableScope {
// If a subscriber synchronously causes this same behavior to emit again, // If a subscriber synchronously causes this same behavior to emit again,
// rxjs delivers the nested value to every subscriber first and then // rxjs delivers the nested value to every subscriber first and then
// resumes delivering the outer (older) value to the remaining subscribers, // resumes delivering the outer (older) value to the remaining subscribers,
// leaving them permanently out of sync with the others. Log the first // leaving them permanently out of sync with the others. A single such
// occurrence with a stack trace so that the re-entrant path can be found. // re-entry can only strand a contiguous run of subscribers; stranding a
let delivering = false; // subscriber in the middle of the list needs a nested (deeper) re-entry, so
let reentryReported = false; // we report the depth and log each new depth (with a stack trace) rather
// than only the first occurrence, to make a nested re-entry visible.
let depth = 0;
let maxReportedDepth = 1;
setValue$.pipe(this.bind(), distinctUntilChanged()).subscribe({ setValue$.pipe(this.bind(), distinctUntilChanged()).subscribe({
next(value) { next(value) {
if (delivering && !reentryReported) { if (depth > 0 && depth + 1 > maxReportedDepth) {
reentryReported = true; maxReportedDepth = depth + 1;
logger.warn( logger.warn(
"Behavior re-entered while delivering a value; later subscribers will be left with a stale value", `Behavior${tag ? ` (${tag})` : ""} re-entered at depth ${
depth + 1
} while delivering a value; later subscribers will be left with a stale value`,
new Error().stack, new Error().stack,
); );
} }
const wasDelivering = delivering; depth++;
delivering = true;
try { try {
subject$.next(value); subject$.next(value);
} finally { } finally {
delivering = wasDelivering; depth--;
} }
}, },
error(err: unknown) { error(err: unknown) {
@@ -193,7 +200,11 @@ export class ObservableScope {
return Object.fromEntries( return Object.fromEntries(
Object.keys(input$.value).map((key) => [ Object.keys(input$.value).map((key) => [
`${key}$`, `${key}$`,
this.behavior(input$.pipe(map((input) => input[key as keyof T]))), this.behavior(
input$.pipe(map((input) => input[key as keyof T])),
nothing,
key,
),
]), ]),
) as SplitBehavior<T>; ) as SplitBehavior<T>;
} }