Flush BatchingRemoteConnection on a microtask by default

.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);
+  },
+});
+```

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 {

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

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<RemoteConnection> {