From 1a06da3984182956f8a839c440a27c160f2b151a Mon Sep 17 00:00:00 2001 From: Cabbagec Date: Mon, 27 Jul 2026 02:45:21 +0000 Subject: [PATCH] fix: reconnect after silent control heartbeat loss --- deploy/production/lithium/compose.yaml | 2 +- deploy/production/x1/compose.yaml | 2 +- deploy/production/x2/compose.yaml | 2 +- docs/deployment-and-usage.md | 2 +- pyproject.toml | 2 +- src/archive_clients/daemon.py | 19 ++++++++++- tests/test_daemon.py | 47 ++++++++++++++++++++++++++ 7 files changed, 70 insertions(+), 6 deletions(-) diff --git a/deploy/production/lithium/compose.yaml b/deploy/production/lithium/compose.yaml index e339a36..a3fde7e 100644 --- a/deploy/production/lithium/compose.yaml +++ b/deploy/production/lithium/compose.yaml @@ -2,7 +2,7 @@ name: archive-control-archive services: archive-client: - image: sodium/archive-clients:v0.1.12 + image: sodium/archive-clients:v0.1.13 user: "1000:1000" restart: unless-stopped command: ["--config", "/etc/archive-control/client.toml"] diff --git a/deploy/production/x1/compose.yaml b/deploy/production/x1/compose.yaml index c0458ac..c6a1239 100644 --- a/deploy/production/x1/compose.yaml +++ b/deploy/production/x1/compose.yaml @@ -2,7 +2,7 @@ name: archive-control-cache services: archive-client: - image: sodium/archive-clients:v0.1.12 + image: sodium/archive-clients:v0.1.13 user: "1001:1001" restart: unless-stopped network_mode: host diff --git a/deploy/production/x2/compose.yaml b/deploy/production/x2/compose.yaml index c0458ac..c6a1239 100644 --- a/deploy/production/x2/compose.yaml +++ b/deploy/production/x2/compose.yaml @@ -2,7 +2,7 @@ name: archive-control-cache services: archive-client: - image: sodium/archive-clients:v0.1.12 + image: sodium/archive-clients:v0.1.13 user: "1001:1001" restart: unless-stopped network_mode: host diff --git a/docs/deployment-and-usage.md b/docs/deployment-and-usage.md index 80ace49..bf24a7b 100644 --- a/docs/deployment-and-usage.md +++ b/docs/deployment-and-usage.md @@ -223,7 +223,7 @@ cache/archive routes according to policy. ```yaml services: archive-client: - image: sodium/archive-clients:v0.1.12 + image: sodium/archive-clients:v0.1.13 user: "1001:1001" restart: unless-stopped command: ["archive-client", "--config", "/etc/archive-control/client.toml"] diff --git a/pyproject.toml b/pyproject.toml index f51f35b..f7feb1e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "archive-clients" -version = "0.1.12" +version = "0.1.13" requires-python = ">=3.11" dependencies = ["protobuf==7.35.1", "websockets==16.0"] diff --git a/src/archive_clients/daemon.py b/src/archive_clients/daemon.py index 1edaf54..758f47c 100644 --- a/src/archive_clients/daemon.py +++ b/src/archive_clients/daemon.py @@ -11,6 +11,7 @@ from pathlib import Path, PurePosixPath from typing import Any from websockets.asyncio.client import connect +from websockets.exceptions import ConnectionClosedOK from archive_clients.backup import SQLiteBackupManager from archive_clients.config import ClientConfig @@ -205,7 +206,23 @@ class ArchiveClientDaemon: command_tasks: set[asyncio.Task[None]] = set() try: await self._resume_commands(outbound, command_tasks) - async for frame in websocket: + # A proxy can leave the TCP/WebSocket socket apparently open + # after the control server has discarded its session. The + # server then cannot deliver durable commands and its pending + # outbox remains stranded unless the client independently + # detects the missing application heartbeats and reconnects. + while True: + try: + frame = await asyncio.wait_for( + websocket.recv(), + self.config.connection.offline_timeout, + ) + except ConnectionClosedOK: + return + except asyncio.TimeoutError as exc: + raise RuntimeError( + "control heartbeat timed out" + ) from exc await self._handle(decode(frame), outbound, command_tasks) finally: writer.cancel() diff --git a/tests/test_daemon.py b/tests/test_daemon.py index a56de4d..ed64fbd 100644 --- a/tests/test_daemon.py +++ b/tests/test_daemon.py @@ -22,6 +22,53 @@ from archive_control.v1 import ( class DaemonTransportTests(unittest.IsolatedAsyncioTestCase): + async def test_silent_control_connection_ends_for_reconnect(self): + """A lost server heartbeat must not leave durable commands stranded.""" + + async def control(websocket): + registration = decode(await websocket.recv()) + self.assertEqual(registration.WhichOneof("payload"), "register_request") + response = new_envelope() + response.correlation_id = registration.message_id + response.register_response.status = ( + client_pb2.REGISTRATION_STATUS_ACCEPTED + ) + response.register_response.negotiated_version.major = 1 + await websocket.send(encode(response)) + # Deliberately keep TCP/WebSocket open but send no application + # heartbeats. This models a stale proxy/server-side session. + await websocket.wait_closed() + + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + token = root / "token" + token.write_text("shared-secret", encoding="utf-8") + os.chmod(token, 0o600) + async with serve(control, "127.0.0.1", 0, ping_interval=None) as server: + port = server.sockets[0].getsockname()[1] + service = ServiceConfig( + "http://local", PurePosixPath("/api"), root, + ) + config = ClientConfig( + "cache-1", "Cache 1", "cache", + f"ws://127.0.0.1:{port}", token, + root / "state.db", root / "backups", service, service, + ConnectionConfig( + registration_timeout=1, + offline_timeout=0.05, + reconnect_initial=0.01, + reconnect_max=0.01, + reconnect_jitter=False, + ), + ) + probe = FilesystemProbe(root, True, True, True, True, True) + daemon = ArchiveClientDaemon(config, [probe, probe], []) + await asyncio.to_thread(daemon.store.initialize) + with self.assertRaisesRegex( + RuntimeError, "control heartbeat timed out" + ): + await asyncio.wait_for(daemon._connection(), 1) + async def test_eviction_assignment_and_steps_are_admitted(self): with tempfile.TemporaryDirectory() as directory: root = Path(directory)