diff --git a/packages/mosaic/src/lease-broker/broker-test-client.spec.ts b/packages/mosaic/src/lease-broker/broker-test-client.spec.ts index cd955d6..16430bf 100644 --- a/packages/mosaic/src/lease-broker/broker-test-client.spec.ts +++ b/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>, + replies: ReadonlyArray | '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=/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()}`); +}