[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.
This commit is contained in:
Timo K.
2026-10-13 16:13:47 +02:00
parent 7c3f3d3ac7
commit d84dcc75de
2 changed files with 37 additions and 11 deletions
+21
View File
@@ -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<number>(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();
+11 -6
View File
@@ -165,13 +165,15 @@ export class ObservableScope {
callback: (value: T) => Promise<(() => Promise<void>) | void>,
): void {
let latestValue: T | typeof nothing = nothing;
let reconcilePromise: Promise<void> | undefined = undefined;
let running = false;
let cleanUp: (() => Promise<void>) | 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<void> => {
running = true;
try {
while (latestValue !== prevVal) {
await cleanUp?.(); // Call the previous value's clean-up handler
prevVal = latestValue;
@@ -179,6 +181,12 @@ export class ObservableScope {
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;
}
};
value$
@@ -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();
});
}