From 77c9a82614318148e637059a802a8a250039d344 Mon Sep 17 00:00:00 2001 From: "jason.woltje" Date: Mon, 20 Jul 2026 06:44:37 +0000 Subject: [PATCH] fix(lease-broker): de-flake recovery_runtime b2 broker-socket ConnectionRefused race (#849) (#851) --- .../lease-broker/recovery_runtime_unittest.py | 31 ++++++++++++------- 1 file changed, 20 insertions(+), 11 deletions(-) diff --git a/packages/mosaic/src/lease-broker/recovery_runtime_unittest.py b/packages/mosaic/src/lease-broker/recovery_runtime_unittest.py index 684daece..1b6d2b6d 100644 --- a/packages/mosaic/src/lease-broker/recovery_runtime_unittest.py +++ b/packages/mosaic/src/lease-broker/recovery_runtime_unittest.py @@ -31,17 +31,26 @@ PI_EXTENSION = FRAMEWORK / "runtime/pi/mosaic-extension.ts" def request(socket_path: Path, value: dict[str, object]) -> dict[str, object]: - with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as connection: - connection.settimeout(3.0) - connection.connect(str(socket_path)) - connection.sendall((json.dumps(value, separators=(",", ":")) + "\n").encode()) - connection.shutdown(socket.SHUT_WR) - response = bytearray() - while True: - chunk = connection.recv(4096) - if not chunk: - break - response.extend(chunk) + deadline = time.monotonic() + 5.0 + while True: + with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as connection: + connection.settimeout(3.0) + try: + connection.connect(str(socket_path)) + except ConnectionRefusedError: + if time.monotonic() >= deadline: + raise + time.sleep(0.02) + continue + connection.sendall((json.dumps(value, separators=(",", ":")) + "\n").encode()) + connection.shutdown(socket.SHUT_WR) + response = bytearray() + while True: + chunk = connection.recv(4096) + if not chunk: + break + response.extend(chunk) + break if not response.endswith(b"\n") or response.count(b"\n") != 1: raise AssertionError(f"unframed broker response: {bytes(response)!r}") reply = json.loads(response[:-1])