diff --git a/packages/mosaic/package.json b/packages/mosaic/package.json index d1b8b07..3ffda77 100644 --- a/packages/mosaic/package.json +++ b/packages/mosaic/package.json @@ -25,7 +25,7 @@ "lint": "eslint src", "typecheck": "tsc --noEmit", "test": "vitest run --passWithNoTests && pnpm run test:framework-shell", - "test:framework-shell": "python3 src/mutator-gate/runtime_tools_unittest.py && python3 src/mutator-gate/runtime_launch_guard_unittest.py && python3 framework/tools/lease-broker/check-runtime-launches.py --root ../.. && bash framework/tools/codex/test-pr-diff-context.sh" + "test:framework-shell": "python3 src/lease-broker/daemon_deadline_unittest.py && python3 src/mutator-gate/runtime_tools_unittest.py && python3 src/mutator-gate/runtime_launch_guard_unittest.py && python3 framework/tools/lease-broker/check-runtime-launches.py --root ../.. && bash framework/tools/codex/test-pr-diff-context.sh" }, "dependencies": { "@mosaicstack/brain": "workspace:*", 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 16430bf..b4a4278 100644 --- a/packages/mosaic/src/lease-broker/broker-test-client.spec.ts +++ b/packages/mosaic/src/lease-broker/broker-test-client.spec.ts @@ -53,29 +53,39 @@ afterEach(async () => { }); 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')]]); + test('rejects an empty early close without retrying into a false green', async () => { + const broker = await scriptedBroker([[], [Buffer.from('{"ok":true}\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]]); - - const failure = await requestBrokerReply( - broker.socketPath, - { action: 'probe' }, - { earlyCloseRetries: 1 }, - ).catch((error: unknown) => error); + const failure = await requestBrokerReply(broker.socketPath, { action: 'probe' }).catch( + (error: unknown) => error, + ); expect(failure).toBeInstanceOf(BrokerTransportError); expect(failure).toMatchObject({ kind: 'early-close', - attempts: 2, + attempts: 1, + responseLength: 0, + responseHex: '', + }); + expect(failure).toHaveProperty( + 'message', + expect.stringMatching(/closed before newline.*length=0.*hex=/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', }); @@ -83,7 +93,7 @@ describe('newline-framed broker test client', () => { 'message', expect.stringMatching(/closed before newline.*length=6.*bytes=.*ok/i), ); - expect(broker.connections()).toBe(2); + expect(broker.connections()).toBe(1); }); test('rejects a newline-terminated malformed broker reply without retrying', async () => { diff --git a/packages/mosaic/src/lease-broker/daemon_deadline_unittest.py b/packages/mosaic/src/lease-broker/daemon_deadline_unittest.py new file mode 100644 index 0000000..a63f912 --- /dev/null +++ b/packages/mosaic/src/lease-broker/daemon_deadline_unittest.py @@ -0,0 +1,72 @@ +#!/usr/bin/env python3 +"""Regression tests for bounded lease-broker read/handle/send deadlines.""" + +from __future__ import annotations + +import importlib.util +import json +import socket +import threading +import time +import unittest +from pathlib import Path + + +DAEMON_PATH = Path(__file__).parents[2] / "framework/tools/lease-broker/daemon.py" +SPEC = importlib.util.spec_from_file_location("lease_broker_deadline_daemon", DAEMON_PATH) +if SPEC is None or SPEC.loader is None: + raise RuntimeError("unable to load lease broker daemon") +DAEMON = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(DAEMON) + + +def original_connection_budget() -> float: + read_budget = getattr(DAEMON, "READ_DEADLINE_SECONDS", None) + if isinstance(read_budget, (int, float)): + return float(read_budget) + return float(DAEMON.CONNECTION_DEADLINE_SECONDS) + + +class SlowBroker: + def __init__(self, delay: float) -> None: + self.delay = delay + self.calls = 0 + + def handle(self, _peer: tuple[int, int, int], request: dict[str, object]) -> dict[str, object]: + self.calls += 1 + time.sleep(self.delay) + return {"ok": True, "echo": request.get("action")} + + +class BrokerDeadlineTest(unittest.TestCase): + def test_completed_slow_handle_gets_a_complete_framed_reply(self) -> None: + broker = SlowBroker(original_connection_budget() + 0.1) + broker_lock = threading.Lock() + server, client = socket.socketpair(socket.AF_UNIX, socket.SOCK_STREAM) + client.settimeout(original_connection_budget() + 2.0) + worker = threading.Thread( + target=DAEMON.handle_connection, + args=(server, broker, broker_lock), + daemon=True, + ) + worker.start() + client.sendall(b'{"action":"probe"}\n') + client.shutdown(socket.SHUT_WR) + + reply = bytearray() + while True: + chunk = client.recv(4096) + if not chunk: + break + reply.extend(chunk) + worker.join(timeout=2.0) + client.close() + + self.assertFalse(worker.is_alive()) + self.assertEqual(broker.calls, 1) + self.assertTrue(reply.endswith(b"\n"), f"unframed reply: {bytes(reply)!r}") + self.assertEqual(json.loads(reply), {"ok": True, "echo": "probe"}) + + +if __name__ == "__main__": + unittest.main()