test(#838): require typed broker transport failures

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

@@ -1,26 +1,29 @@
 import { mkdtemp, rm } from 'node:fs/promises';
-import { createServer, type Server } from 'node:net';
+import { createServer, type Server, type Socket } 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';
+import { BrokerTransportError, readBrokerReply, requestBrokerReply } from './broker-test-client.js';
 
 const roots: string[] = [];
 const servers: Server[] = [];
+const sockets: Socket[] = [];
 
 async function scriptedBroker(
-  replies: ReadonlyArray<ReadonlyArray<Buffer>>,
+  replies: ReadonlyArray<ReadonlyArray<Buffer> | 'hang'>,
 ): 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 server = createServer({ allowHalfOpen: true }, (socket) => {
+    sockets.push(socket);
     const chunks = replies[connections] ?? replies.at(-1) ?? [];
     connections += 1;
     socket.once('end', () => {
+      if (chunks === 'hang') return;
       for (const chunk of chunks) socket.write(chunk);
       socket.end();
     });
@@ -35,6 +38,7 @@ async function scriptedBroker(
 }
 
 afterEach(async () => {
+  for (const socket of sockets.splice(0)) socket.destroy();
   await Promise.all(
     servers
       .splice(0)
@@ -62,9 +66,23 @@ describe('newline-framed broker test client', () => {
     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);
+    const failure = await requestBrokerReply(
+      broker.socketPath,
+      { action: 'probe' },
+      { earlyCloseRetries: 1 },
+    ).catch((error: unknown) => error);
+
+    expect(failure).toBeInstanceOf(BrokerTransportError);
+    expect(failure).toMatchObject({
+      kind: 'early-close',
+      attempts: 2,
+      responseLength: 6,
+      responseHex: '7b226f6b223a',
+    });
+    expect(failure).toHaveProperty(
+      'message',
+      expect.stringMatching(/closed before newline.*length=6.*bytes=.*ok/i),
+    );
     expect(broker.connections()).toBe(2);
   });
 
@@ -76,4 +94,39 @@ describe('newline-framed broker test client', () => {
     );
     expect(broker.connections()).toBe(1);
   });
+
+  test.each([
+    ['trailing bytes', Buffer.from('{"ok":true}\nextra'), /bytes after the newline/i],
+    ['non-object JSON', Buffer.from('[]\n'), /JSON value is not an object/i],
+    ['oversized frame', Buffer.alloc(64 * 1024 + 1, 0x78), /exceeds 65536 bytes/i],
+  ])('rejects %s with deterministic framing context', async (_label, reply, message) => {
+    const broker = await scriptedBroker([[reply as Buffer]]);
+
+    await expect(requestBrokerReply(broker.socketPath, { action: 'probe' })).rejects.toThrow(
+      message as RegExp,
+    );
+    expect(broker.connections()).toBe(1);
+  });
+
+  test('reports timeout, connection, and writer failures as promise rejections', async () => {
+    const hanging = await scriptedBroker(['hang']);
+    await expect(
+      requestBrokerReply(hanging.socketPath, { action: 'probe' }, { timeoutMs: 10 }),
+    ).rejects.toThrow(/timed out before newline.*length=0.*hex=<empty>/i);
+
+    await expect(
+      requestBrokerReply(join(rootForMissingSocket(), 'missing.sock'), {}),
+    ).rejects.toThrow(/socket error before newline.*length=0/i);
+
+    const writerFailure = await scriptedBroker(['hang']);
+    await expect(
+      readBrokerReply(writerFailure.socketPath, () => {
+        throw new Error('writer failed');
+      }),
+    ).rejects.toThrow('writer failed');
+  });
 });
+
+function rootForMissingSocket(): string {
+  return join(tmpdir(), `mosaic-missing-broker-${process.pid}-${Date.now()}`);
+}