diff --git a/.changeset/microtask-default-batching.md b/.changeset/microtask-default-batching.md new file mode 100644 index 00000000..f9b91da0 --- /dev/null +++ b/.changeset/microtask-default-batching.md @@ -0,0 +1,21 @@ +--- +'@remote-dom/core': minor +--- + +Flush `BatchingRemoteConnection` mutations on a microtask by default + +The default `batch` function used by `BatchingRemoteConnection` now schedules flushes on a microtask (via `queueMicrotask`, falling back to `Promise.resolve().then(...)`) instead of a macrotask via `MessageChannel`/`setTimeout`. + +This ensures that DOM mutations made in the remote environment are delivered to the host before any `await`ed RPC response resolves, avoiding a class of bugs where the host sees an RPC result before the element updates that produced it (for example, a `perform()` callback setting an error on a field, only to have the host run its post-`await` logic before that error mutation arrives). + +If you were relying on the previous macrotask behavior, you can restore it by passing a custom `batch` option, e.g.: + +```ts +new BatchingRemoteConnection(connection, { + batch: (flush) => { + const channel = new MessageChannel(); + channel.port1.onmessage = () => flush(); + channel.port2.postMessage(null); + }, +}); +``` diff --git a/packages/core/README.md b/packages/core/README.md index 6cdf8866..77909eeb 100644 --- a/packages/core/README.md +++ b/packages/core/README.md @@ -474,7 +474,7 @@ root.connect(connection); The `RemoteConnection` object you receive from `RemoteReceiver.connection` is a simple object that immediately communicates all updates to the host environment. When using `RemoteMutationObserver`, documented above, this is not a major issue, since the `MutationObserver` API automatically batches DOM mutations. However, it can be more of a problem when using Remote DOM in a web worker (typically, with the `RemoteRootElement` wrapper), where no such batching is performed. -To improve performance in these cases, you can use the `BatchingRemoteConnection` class, which batches updates from the remote environment that happen in the same JavaScript task. This class is a subclass of `RemoteConnection`, and can be used directly in place of the original connection object: +To improve performance in these cases, you can use the `BatchingRemoteConnection` class, which batches updates from the remote environment. By default, queued mutations are flushed on a microtask (falling back to `Promise.resolve()`), so that mutations are observable on the host before any `await`ed RPC response resolves. You can override this by passing a custom `batch` function — for example, to defer flushes to a macrotask via `MessageChannel` or `setTimeout`. This class is a subclass of `RemoteConnection`, and can be used directly in place of the original connection object: ```ts import { diff --git a/packages/core/source/elements/connection.ts b/packages/core/source/elements/connection.ts index 5d6bc74f..af1d617b 100644 --- a/packages/core/source/elements/connection.ts +++ b/packages/core/source/elements/connection.ts @@ -1,9 +1,12 @@ import type {RemoteConnection, RemoteMutationRecord} from '../types.ts'; /** - * A wrapper around a `RemoteConnection` that batches mutations. By default, this - * all calls are flushed in a queued microtask, but this can be customized by passing - * a custom `batch` option. + * A wrapper around a `RemoteConnection` that batches mutations. By default, + * queued mutations are flushed in a microtask (via `queueMicrotask`, falling + * back to `Promise.resolve().then(...)` in environments where + * `queueMicrotask` is unavailable). This can be customized by passing a + * custom `batch` option — for example, to defer flushes to a macrotask via + * `MessageChannel` or `setTimeout`. */ export class BatchingRemoteConnection { readonly #connection: RemoteConnection; @@ -50,22 +53,16 @@ export class BatchingRemoteConnection { } function createDefaultBatchFunction() { - let channel: MessageChannel; - return (queue: () => void) => { - // In environments without a `MessageChannel`, use a `setTimeout` fallback. - if (typeof MessageChannel !== 'function') { - setTimeout(() => { - queue(); - }, 0); + // Flush on a microtask so that mutations are observable on the host + // before any `await`ed RPC response resolves. In environments without + // `queueMicrotask`, fall back to `Promise.resolve().then(...)`, which + // also schedules on the microtask queue. + if (typeof queueMicrotask === 'function') { + queueMicrotask(queue); return; } - // `MessageChannel` trick that forces the code to run on the next task. - channel ??= new MessageChannel(); - channel.port1.onmessage = () => { - queue(); - }; - channel.port2.postMessage(null); + Promise.resolve().then(queue); }; } diff --git a/packages/core/source/elements/tests/connection.test.ts b/packages/core/source/elements/tests/connection.test.ts index ef2c7224..7cd30058 100644 --- a/packages/core/source/elements/tests/connection.test.ts +++ b/packages/core/source/elements/tests/connection.test.ts @@ -1,11 +1,15 @@ import '../../polyfill/polyfill.ts'; -import {describe, expect, it, vi, type MockedObject} from 'vitest'; +import {afterEach, describe, expect, it, vi, type MockedObject} from 'vitest'; import {BatchingRemoteConnection, RemoteConnection} from '../../elements'; describe('BatchingRemoteConnection', () => { - it('batches mutations', async () => { + afterEach(() => { + vi.restoreAllMocks(); + }); + + it('batches mutations on a microtask', async () => { const connection = createRemoteConnectionSpy(); const batchingConnection = new BatchingRemoteConnection(connection); @@ -14,7 +18,7 @@ describe('BatchingRemoteConnection', () => { expect(connection.mutate).not.toHaveBeenCalled(); - await waitForNextTask(); + await waitForNextMicrotask(); expect(connection.mutate).toHaveBeenCalledTimes(1); expect(connection.mutate).toHaveBeenCalledWith([1, 2, 3, 4, 5, 6]); @@ -23,37 +27,60 @@ describe('BatchingRemoteConnection', () => { expect(connection.mutate).toHaveBeenCalledTimes(1); - await waitForNextTask(); + await waitForNextMicrotask(); expect(connection.mutate).toHaveBeenCalledTimes(2); expect(connection.mutate).toHaveBeenCalledWith([7, 8, 9]); }); - it('batches mutations with setTimeout when there is no MessageChannel', async () => { - vi.useFakeTimers(); - vi.spyOn(globalThis, 'MessageChannel', 'get').mockReturnValue( - undefined as any, - ); - const setTimeoutSpy = vi.spyOn(globalThis, 'setTimeout'); + it('flushes mutations before an awaited promise resolves', async () => { const connection = createRemoteConnectionSpy(); const batchingConnection = new BatchingRemoteConnection(connection); batchingConnection.mutate([1, 2, 3]); - batchingConnection.mutate([4, 5, 6]); - expect(connection.mutate).not.toHaveBeenCalled(); - - vi.runAllTimers(); + // An `await` yields to the microtask queue, so microtask-scheduled + // flushes must have run by the time execution resumes here. + await Promise.resolve(); - expect(setTimeoutSpy).toHaveBeenCalledTimes(1); expect(connection.mutate).toHaveBeenCalledTimes(1); - expect(connection.mutate).toHaveBeenCalledWith([1, 2, 3, 4, 5, 6]); + expect(connection.mutate).toHaveBeenCalledWith([1, 2, 3]); + }); - vi.restoreAllMocks(); - vi.useRealTimers(); + it('uses queueMicrotask to schedule the batch flush', () => { + const queueMicrotaskSpy = vi.spyOn(globalThis, 'queueMicrotask'); + const connection = createRemoteConnectionSpy(); + const batchingConnection = new BatchingRemoteConnection(connection); + + batchingConnection.mutate([1, 2, 3]); + + expect(queueMicrotaskSpy).toHaveBeenCalledTimes(1); + }); + + it('falls back to Promise.resolve().then() when queueMicrotask is unavailable', async () => { + const originalQueueMicrotask = globalThis.queueMicrotask; + // @ts-expect-error — intentionally removing to test fallback. + delete globalThis.queueMicrotask; + + try { + const connection = createRemoteConnectionSpy(); + const batchingConnection = new BatchingRemoteConnection(connection); + + batchingConnection.mutate([1, 2, 3]); + batchingConnection.mutate([4, 5, 6]); + + expect(connection.mutate).not.toHaveBeenCalled(); + + await Promise.resolve(); + + expect(connection.mutate).toHaveBeenCalledTimes(1); + expect(connection.mutate).toHaveBeenCalledWith([1, 2, 3, 4, 5, 6]); + } finally { + globalThis.queueMicrotask = originalQueueMicrotask; + } }); - it('flushes mutations', async () => { + it('flush() drains queued mutations synchronously', () => { const connection = createRemoteConnectionSpy(); const batchingConnection = new BatchingRemoteConnection(connection); @@ -62,14 +89,24 @@ describe('BatchingRemoteConnection', () => { expect(connection.mutate).toHaveBeenCalledOnce(); expect(connection.mutate).toHaveBeenCalledWith([1, 2, 3]); + }); + + it('flush() prevents the pending microtask from double-firing, and a subsequent mutate() schedules a fresh batch', async () => { + const connection = createRemoteConnectionSpy(); + const batchingConnection = new BatchingRemoteConnection(connection); + + batchingConnection.mutate([1, 2, 3]); + batchingConnection.flush(); - await waitForNextTask(); + await waitForNextMicrotask(); - // ensure it wasn't called again + // The scheduled microtask still runs, but flush() drained the queue + // already, so the underlying connection isn't called a second time. expect(connection.mutate).toHaveBeenCalledOnce(); + batchingConnection.mutate([4, 5, 6]); - await waitForNextTask(); + await waitForNextMicrotask(); expect(connection.mutate).toHaveBeenCalledTimes(2); expect(connection.mutate).toHaveBeenCalledWith([4, 5, 6]); @@ -94,14 +131,8 @@ describe('BatchingRemoteConnection', () => { }); }); -async function waitForNextTask() { - const channel = new MessageChannel(); - const promise = new Promise((resolve) => { - channel.port1.onmessage = resolve; - }); - channel.port2.postMessage(null); - - await promise; +async function waitForNextMicrotask() { + await Promise.resolve(); } function createRemoteConnectionSpy(): MockedObject {