Queue re-entrant emissions in scope.behavior

If a subscriber synchronously causes the same behavior to emit again,
rxjs delivers the nested value to every subscriber and then resumes
delivering the outer, older value to the remaining subscribers, which
leaves them permanently out of sync while .value holds the new value.
Queue nested emissions so every subscriber sees every value in order
and ends on the latest one, and warn once per behavior with a stack
trace so the re-entrant paths can be found.

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<number>();
+    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();
+  });
+});

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 = <T>(o: Observable<T>) => Observable<T>;
@@ -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);