perf(server): resuming the thread list takes one fewer round trip

apps/server/src/rpcInitialItems.memory.test.ts

@@ -16,12 +16,20 @@ it.each(["prefix", "snapshot", "replay"])(
     const { stdout } = await execFile(process.execPath, ["--expose-gc", fixture, root, mode]);
     const result = JSON.parse(stdout) as {
       checkpoints: ReadonlyArray<{ subscriptions: number; retained: number }>;
+      deliveredBatches: ReadonlyArray<ReadonlyArray<ReadonlyArray<string>>>;
     };
     expect(
       result.checkpoints.map(({ subscriptions, retained }) => ({ subscriptions, retained })),
     ).toEqual([
       { subscriptions: 2, retained: 0 },
       { subscriptions: 4, retained: 0 },
     ]);
+    expect(result.deliveredBatches).toEqual(
+      Array.from({ length: 4 }, () => [
+        mode === "prefix"
+          ? ["prefix"]
+          : [mode === "snapshot" ? "snapshot" : "event", "synchronized"],
+      ]),
+    );
   },
 );

apps/server/src/rpcInitialItems.ts

@@ -1,8 +1,12 @@
 import * as Effect from "effect/Effect";
 import * as Stream from "effect/Stream";
 
-export function rpcInitialItems<A>(items: ReadonlyArray<A>): Stream.Stream<A> {
-  let pending: ReadonlyArray<A> | undefined = items;
+export function rpcInitialItems<A, B = never>(
+  items: ReadonlyArray<A>,
+  completion?: B,
+): Stream.Stream<A | B> {
+  let pending: ReadonlyArray<A | B> | undefined =
+    completion === undefined ? items : [...items, completion];
   return Stream.unwrap(
     Effect.sync(() => {
       const initial = pending;

apps/server/src/testkit/RpcInitialMemory.fixture.mjs

@@ -23,6 +23,7 @@ const refs = [];
 const fibers = [];
 const checkpoints = [];
 const deliveredCounts = [];
+const deliveredBatches = [];
 
 function history() {
   const value = JSON.parse(JSON.stringify({ text: NodeCrypto.randomBytes(bytes).toString("hex") }));
@@ -32,6 +33,7 @@ function history() {
 
 function start(ready) {
   const index = deliveredCounts.push(0) - 1;
+  deliveredBatches.push([]);
   const live = Stream.fromEffect(Deferred.succeed(ready, undefined)).pipe(
     Stream.drain,
     Stream.concat(Stream.never),
@@ -80,19 +82,22 @@ function start(ready) {
   return stream.pipe(
     Effect.flatMap((stream) =>
       stream.pipe(
-        Stream.runForEach((item) =>
+        Stream.runForEachArray((items) =>
           Effect.sync(() => {
-            if (mode === "prefix") {
-              NodeAssert.equal(item.text.length, bytes * 2);
-            } else if (delivered === 0) {
-              NodeAssert.equal(item.kind, mode === "replay" ? "event" : "snapshot");
-              const value = mode === "replay" ? item.event.payload : item.projection.messages[0];
-              NodeAssert.equal(value.text.length, bytes * 2);
-            } else {
-              NodeAssert.equal(item.kind, "synchronized");
+            deliveredBatches[index].push(items.map((item) => item.kind ?? "prefix"));
+            for (const item of items) {
+              if (mode === "prefix") {
+                NodeAssert.equal(item.text.length, bytes * 2);
+              } else if (delivered === 0) {
+                NodeAssert.equal(item.kind, mode === "replay" ? "event" : "snapshot");
+                const value = mode === "replay" ? item.event.payload : item.projection.messages[0];
+                NodeAssert.equal(value.text.length, bytes * 2);
+              } else {
+                NodeAssert.equal(item.kind, "synchronized");
+              }
+              delivered++;
+              deliveredCounts[index] = delivered;
             }
-            delivered++;
-            deliveredCounts[index] = delivered;
           }),
         ),
         Effect.forkDetach,
@@ -124,7 +129,7 @@ const result = await Effect.runPromise(
       }
     }
     for (const fiber of fibers) yield* Fiber.interrupt(fiber);
-    return { checkpoints };
+    return { checkpoints, deliveredBatches };
   }),
 );
 console.log(JSON.stringify(result));