test(#838): define broker reply framing behavior

packages/mosaic/src/lease-broker/broker-test-client.spec.ts

@@ -0,0 +1,79 @@
+import { mkdtemp, rm } from 'node:fs/promises';
+import { createServer, type Server } from 'node:net';
+import { tmpdir } from 'node:os';
+import { join } from 'node:path';
+
+import { afterEach, describe, expect, test } from 'vitest';
+
+import { requestBrokerReply } from './broker-test-client.js';
+
+const roots: string[] = [];
+const servers: Server[] = [];
+
+async function scriptedBroker(
+  replies: ReadonlyArray<ReadonlyArray<Buffer>>,
+): Promise<{ socketPath: string; connections: () => number }> {
+  const root = await mkdtemp(join(tmpdir(), 'mosaic-broker-client-'));
+  roots.push(root);
+  const socketPath = join(root, 'broker.sock');
+  let connections = 0;
+  const server = createServer((socket) => {
+    const chunks = replies[connections] ?? replies.at(-1) ?? [];
+    connections += 1;
+    socket.once('end', () => {
+      for (const chunk of chunks) socket.write(chunk);
+      socket.end();
+    });
+    socket.resume();
+  });
+  servers.push(server);
+  await new Promise<void>((resolve, reject) => {
+    server.once('error', reject);
+    server.listen(socketPath, resolve);
+  });
+  return { socketPath, connections: () => connections };
+}
+
+afterEach(async () => {
+  await Promise.all(
+    servers
+      .splice(0)
+      .map(
+        (server) =>
+          new Promise<void>((resolve, reject) =>
+            server.close((error) => (error ? reject(error) : resolve())),
+          ),
+      ),
+  );
+  await Promise.all(roots.splice(0).map((root) => rm(root, { recursive: true, force: true })));
+});
+
+describe('newline-framed broker test client', () => {
+  test('retries an empty early close and parses the complete newline-terminated retry', async () => {
+    const broker = await scriptedBroker([[], [Buffer.from('{"ok":true'), Buffer.from('}\n')]]);
+
+    await expect(requestBrokerReply(broker.socketPath, { action: 'probe' })).resolves.toEqual({
+      ok: true,
+    });
+    expect(broker.connections()).toBe(2);
+  });
+
+  test('rejects repeated truncated early closes with response bytes and length', async () => {
+    const truncated = Buffer.from('{"ok":');
+    const broker = await scriptedBroker([[truncated], [truncated]]);
+
+    await expect(
+      requestBrokerReply(broker.socketPath, { action: 'probe' }, { earlyCloseRetries: 1 }),
+    ).rejects.toThrow(/closed before newline.*attempts=2.*length=6.*bytes="\{\\"ok\\":"]/i);
+    expect(broker.connections()).toBe(2);
+  });
+
+  test('rejects a newline-terminated malformed broker reply without retrying', async () => {
+    const broker = await scriptedBroker([[Buffer.from('{bad}\n')]]);
+
+    await expect(requestBrokerReply(broker.socketPath, { action: 'probe' })).rejects.toThrow(
+      /malformed broker reply.*length=6.*bytes="\{bad\}\\n"/i,
+    );
+    expect(broker.connections()).toBe(1);
+  });
+});