From d84dcc75de55f743d0e674b9aba2b5d091c6054e Mon Sep 17 00:00:00 2001 From: "Timo K." Date: Tue, 13 Oct 2026 16:13:47 +0200 Subject: [PATCH] [SDK] Keep a value that reaches reconcile as its loop winds down The loop that drives `scope.reconcile` was marked as running until the promise of its last run had settled, which is a few microtasks after the loop itself had checked for new values. A value arriving in that window found the loop "running", was left to it, and was never processed: a connection appearing at that moment would never get its publisher. The flag is now cleared the moment the loop ends, so such a value starts a new loop. The regression test drives the window microtask by microtask. --- sdk/src/reactive/ObservableScope.test.ts | 21 ++++++++++++++++++ sdk/src/reactive/ObservableScope.ts | 27 ++++++++++++++---------- 2 files changed, 37 insertions(+), 11 deletions(-) diff --git a/sdk/src/reactive/ObservableScope.test.ts b/sdk/src/reactive/ObservableScope.test.ts index 25bb14324..e55348532 100644 --- a/sdk/src/reactive/ObservableScope.test.ts +++ b/sdk/src/reactive/ObservableScope.test.ts @@ -192,6 +192,27 @@ describe("Reconcile", () => { expect(setup).not.toHaveBeenCalledWith(3); }); + // The loop ends a few microtasks before its promise settles; a value + // arriving in that window used to be dropped + it.each([0, 1, 2, 3, 4, 5, 6])( + "picks up a value that arrives %i microtasks after the previous loop ended", + async (ticks) => { + const scope = testScope(); + const behavior$ = new BehaviorSubject(0); + const setup = vi.fn(); + scope.reconcile(behavior$, async (value) => { + setup(value); + return Promise.resolve(); + }); + + for (let i = 0; i < ticks; i++) await Promise.resolve(); + behavior$.next(1); + await vi.advanceTimersByTimeAsync(10); + + expect(setup).toHaveBeenCalledWith(1); + }, + ); + it("should wait for setup to complete before starting cleanup", async () => { vi.useFakeTimers(); const scope = testScope(); diff --git a/sdk/src/reactive/ObservableScope.ts b/sdk/src/reactive/ObservableScope.ts index 4f657b96a..beb45e958 100644 --- a/sdk/src/reactive/ObservableScope.ts +++ b/sdk/src/reactive/ObservableScope.ts @@ -165,19 +165,27 @@ export class ObservableScope { callback: (value: T) => Promise<(() => Promise) | void>, ): void { let latestValue: T | typeof nothing = nothing; - let reconcilePromise: Promise | undefined = undefined; + let running = false; let cleanUp: (() => Promise) | void = undefined; let prevVal: T | typeof nothing = nothing; // While this loop runs it will process the latest from `value$` until it caught up with the updates. // It might skip updates from `value$` and only process the newest value after callback has resolved. const reconcileLoop = async (): Promise => { - while (latestValue !== prevVal) { - await cleanUp?.(); // Call the previous value's clean-up handler - prevVal = latestValue; + running = true; + try { + while (latestValue !== prevVal) { + await cleanUp?.(); // Call the previous value's clean-up handler + prevVal = latestValue; - if (latestValue !== nothing) cleanUp = await callback(latestValue); // Sync current value... - // `latestValue` might have gotten updated during the `await callback`. That is why we loop here + if (latestValue !== nothing) cleanUp = await callback(latestValue); // Sync current value... + // `latestValue` might have gotten updated during the `await callback`. That is why we loop here + } + } finally { + // Cleared the moment the loop is done, not when its promise settles: + // a value arriving in between would otherwise find the loop still + // marked as running and never be processed + running = false; } }; @@ -192,11 +200,8 @@ export class ObservableScope { latestValue = value; // There's already an instance of the below 'reconcileLoop' loop running // concurrently. So lets let the loop handle it. NEVER instanciate two `reconcileLoop`s. - if (reconcilePromise) return; - - reconcilePromise = reconcileLoop().finally(() => { - reconcilePromise = undefined; - }); + if (running) return; + void reconcileLoop(); }); }