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

import { mkdtemp, rm } from 'node:fs/promises';
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 { BrokerTransportError, readBrokerReply, requestBrokerReply } from './broker-test-client.js';

const roots: string[] = [];
const servers: Server[] = [];
const sockets: Socket[] = [];

async function scriptedBroker(
  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({ allowHalfOpen: true }, (socket) => {
    sockets.push(socket);
    const chunks = replies[connections] ?? replies.at(-1) ?? [];
    connections += 1;
    socket.once('end', () => {
      if (chunks === 'hang') return;
      void (async () => {
        for (const chunk of chunks) {
          socket.write(chunk);
          await new Promise<void>((resolve) => setTimeout(resolve, 5));
        }
        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 () => {
  for (const socket of sockets.splice(0)) socket.destroy();
  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('rejects an empty early close without retrying into a false green', async () => {
    const broker = await scriptedBroker([[], [Buffer.from('{"ok":true}\n')]]);

    const failure = await requestBrokerReply(broker.socketPath, { action: 'probe' }).catch(
      (error: unknown) => error,
    );

    expect(failure).toBeInstanceOf(BrokerTransportError);
    expect(failure).toMatchObject({
      kind: 'early-close',
      attempts: 1,
      responseLength: 0,
      responseHex: '',
    });
    expect(failure).toHaveProperty(
      'message',
      expect.stringMatching(/closed before newline.*length=0.*hex=<empty>/i),
    );
    expect(broker.connections()).toBe(1);
  });

  test('rejects repeated truncated early closes with response bytes and length', async () => {
    const truncated = Buffer.from('{"ok":');
    const broker = await scriptedBroker([[truncated], [Buffer.from('{"ok":true}\n')]]);

    const failure = await requestBrokerReply(broker.socketPath, { action: 'probe' }).catch(
      (error: unknown) => error,
    );

    expect(failure).toBeInstanceOf(BrokerTransportError);
    expect(failure).toMatchObject({
      kind: 'early-close',
      attempts: 1,
      responseLength: 6,
      responseHex: '7b226f6b223a',
    });
    expect(failure).toHaveProperty(
      'message',
      expect.stringMatching(/closed before newline.*length=6.*bytes=.*ok/i),
    );
    expect(broker.connections()).toBe(1);
  });

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

  test('rejects trailing bytes delivered after a complete frame in a later data event', async () => {
    const broker = await scriptedBroker([[Buffer.from('{"ok":true}\n'), Buffer.from('extra')]]);

    const failure = await requestBrokerReply(broker.socketPath, { action: 'probe' }).catch(
      (error: unknown) => error,
    );

    expect(failure).toBeInstanceOf(BrokerTransportError);
    expect(failure).toMatchObject({ kind: 'malformed-reply', responseLength: 17 });
    expect(failure).toHaveProperty(
      'message',
      expect.stringMatching(/bytes after the newline terminator/i),
    );
  });

  test('redacts security tokens from typed transport diagnostics', async () => {
    const token = 'a'.repeat(64);
    const reply = Buffer.from(`{"ok":true,"promotion_token":"${token}`);
    const broker = await scriptedBroker([[reply]]);

    const failure = await requestBrokerReply(broker.socketPath, { action: 'probe' }).catch(
      (error: unknown) => error,
    );

    expect(failure).toBeInstanceOf(BrokerTransportError);
    expect(failure).toMatchObject({
      kind: 'early-close',
      responseLength: reply.length,
      responsePreview: '<redacted-sensitive-reply>',
    });
    expect(String((failure as Error).message)).not.toContain(token);
    expect(JSON.stringify(failure)).not.toContain(token);
  });

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