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