fix: recover deadlocked Binance receivers

deployment/aliyun/README.md

@@ -16,8 +16,11 @@ symbol coverage, bridge state, gap count, free disk space, and whether the 20GiB
 warning threshold is active. It also reports pending upload count, last upload
 success/error, and an upload warning so OSS failures cannot look fully healthy.
 A silent WebSocket shard fails after
-`STALL_TIMEOUT_SECONDS`; systemd then restarts the service. Low disk space emits a
-warning but does not pause collection. Successfully uploaded segments are deleted
+`STALL_TIMEOUT_SECONDS`. Receiver cleanup is bounded to five seconds so a stuck
+WebSocket close handshake cannot block reconnection. A separate process watchdog
+exits after 180 seconds without any market-data frame, allowing systemd to recover
+even if the asyncio loop deadlocks. Low disk space emits a warning but does not
+pause collection. Successfully uploaded segments are deleted
 from the local spool immediately. Pending segments are retained when OSS upload
 fails so the collector never creates a silent data hole merely to reclaim space.
 The shared pending-diff budget still bounds initialization bursts. Services restart

deployment/aliyun/binance-lob-archiver-spot.env

@@ -8,6 +8,8 @@ SNAPSHOT_LIMIT=100
 SNAPSHOT_REQUESTS_PER_SECOND=15
 SYNC_TIMEOUT_SECONDS=120
 STALL_TIMEOUT_SECONDS=60
+PROCESS_WATCHDOG_SECONDS=180
+TASK_CANCEL_TIMEOUT_SECONDS=5
 MAX_BUFFERED_DIFFS=250000
 MAX_PENDING_DIFFS_TOTAL=250000
 MIN_FREE_GB=20

deployment/aliyun/binance-lob-archiver-usdm.env

@@ -9,6 +9,8 @@ SNAPSHOT_REQUESTS_PER_SECOND=5
 SNAPSHOT_RETRY_ATTEMPTS=6
 SYNC_TIMEOUT_SECONDS=120
 STALL_TIMEOUT_SECONDS=60
+PROCESS_WATCHDOG_SECONDS=180
+TASK_CANCEL_TIMEOUT_SECONDS=5
 MAX_BUFFERED_DIFFS=250000
 MAX_PENDING_DIFFS_TOTAL=250000
 MIN_FREE_GB=20

deployment/aliyun/binance_lob_archiver.py

@@ -66,6 +66,10 @@
 MIN_FREE_GB = int(os.getenv("MIN_FREE_GB", "20"))
 ZSTD_TIMEOUT_SECONDS = int(os.getenv("ZSTD_TIMEOUT_SECONDS", "300"))
 OSS_COPY_TIMEOUT_SECONDS = int(os.getenv("OSS_COPY_TIMEOUT_SECONDS", "300"))
+PROCESS_WATCHDOG_SECONDS = int(os.getenv("PROCESS_WATCHDOG_SECONDS", "180"))
+TASK_CANCEL_TIMEOUT_SECONDS = int(os.getenv("TASK_CANCEL_TIMEOUT_SECONDS", "5"))
+LAST_DATA_AT = time.monotonic()
+PROCESS_WATCHDOG_ARMED = False
 UPLOAD_STATUS: dict[str, str | None] = {
     "last_success_at": None,
     "last_error_at": None,
@@ -94,6 +98,10 @@ def __init__(self, symbol: str, status: int) -> None:
         super().__init__(f"snapshot unavailable symbol={self.symbol} status={status}")
 
 
+class TaskCancellationStuck(RuntimeError):
+    pass
+
+
 class PendingBudget:
     def __init__(self, limit: int) -> None:
         self.limit = limit
@@ -597,6 +605,7 @@ def fetch_snapshot_sync(symbol: str) -> dict:
 
 
 async def receive_url(url: str, queue: asyncio.Queue, stop: asyncio.Event) -> None:
+    global LAST_DATA_AT
     async with connect(
         url, open_timeout=20, ping_interval=20, max_size=8 * 1024 * 1024
     ) as websocket:
@@ -610,6 +619,7 @@ async def receive_url(url: str, queue: asyncio.Queue, stop: asyncio.Event) -> No
                     f"no depth frames for {STALL_TIMEOUT_SECONDS}s on websocket shard"
                 )
             if isinstance(message, str):
+                LAST_DATA_AT = time.monotonic()
                 await queue.put(("diff", time.time_ns(), json.loads(message)))
 
 
@@ -661,6 +671,43 @@ def is_stalled(last_frame_at: float, now: float) -> bool:
     return now - last_frame_at > STALL_TIMEOUT_SECONDS
 
 
+def process_watchdog_expired(last_data_at: float, now: float) -> bool:
+    return now - last_data_at > PROCESS_WATCHDOG_SECONDS
+
+
+def run_process_watchdog() -> None:
+    interval = max(1.0, min(10.0, PROCESS_WATCHDOG_SECONDS / 4))
+    while True:
+        time.sleep(interval)
+        if not PROCESS_WATCHDOG_ARMED:
+            continue
+        now = time.monotonic()
+        if process_watchdog_expired(LAST_DATA_AT, now):
+            LOG.critical(
+                "process watchdog exiting after %.1fs without market data",
+                now - LAST_DATA_AT,
+            )
+            os._exit(75)
+
+
+async def cancel_tasks_bounded(tasks: tuple[asyncio.Task, ...]) -> None:
+    if not tasks:
+        return
+    for task in tasks:
+        task.cancel()
+    _, pending = await asyncio.wait(
+        tasks, timeout=TASK_CANCEL_TIMEOUT_SECONDS
+    )
+    if pending:
+        LOG.error(
+            "task cancellation timed out pending=%s",
+            len(pending),
+        )
+        raise TaskCancellationStuck(
+            f"task cancellation timed out pending={len(pending)}"
+        )
+
+
 def bridge_timed_out(
     previously_synced: bool, sync_deadline: float | None, now: float
 ) -> bool:
@@ -981,11 +1028,7 @@ async def run_session(
         failure = error
         raise
     finally:
-        for task in [*tasks, *resync_tasks.values()]:
-            task.cancel()
-        await asyncio.gather(
-            *tasks, *resync_tasks.values(), return_exceptions=True
-        )
+        await cancel_tasks_bounded(tuple([*tasks, *resync_tasks.values()]))
         archive_only = isinstance(failure, SequenceGap)
         drain_gap: SequenceGap | None = None
         while not queue.empty():
@@ -1030,8 +1073,11 @@ async def run_session(
 
 async def collect(stop: asyncio.Event) -> None:
     global SYMBOLS, SECURITY_TOKEN_SYMBOLS, EXCLUDED_SYMBOLS
+    global LAST_DATA_AT, PROCESS_WATCHDOG_ARMED
     SPOOL_DIR.mkdir(parents=True, exist_ok=True)
     recover_parts()
+    LAST_DATA_AT = time.monotonic()
+    PROCESS_WATCHDOG_ARMED = True
     runtime = ArchiveRuntime()
     backoff = 1
     initial_budget = PendingBudget(MAX_PENDING_DIFFS_TOTAL)
@@ -1081,6 +1127,15 @@ async def collect(stop: asyncio.Event) -> None:
                     )
                 await asyncio.sleep(backoff)
                 backoff = min(backoff * 2, 30)
+            except TaskCancellationStuck:
+                write_health(
+                    states, session_id, "fatal_task_stall", runtime.total_gaps
+                )
+                LOG.critical(
+                    "receiver cancellation stuck; exiting for systemd restart",
+                    exc_info=True,
+                )
+                raise
             except Exception:
                 write_health(states, session_id, "reconnecting", runtime.total_gaps)
                 LOG.exception("websocket session failed; reconnecting in %ss", backoff)
@@ -1152,6 +1207,11 @@ def main() -> None:
     if STARTUP_DELAY_SECONDS:
         LOG.info("startup delay=%ss", STARTUP_DELAY_SECONDS)
         time.sleep(STARTUP_DELAY_SECONDS)
+    threading.Thread(
+        target=run_process_watchdog,
+        name=f"binance-data-watchdog-{MARKET}",
+        daemon=True,
+    ).start()
     loop.run_until_complete(collect(stop))
 
 

deployment/aliyun/test_binance_lob_archiver.py

@@ -328,6 +328,32 @@ async def test_stall_watchdog_trips_only_after_timeout(self):
             self.assertFalse(ARCHIVER.is_stalled(100, 160))
             self.assertTrue(ARCHIVER.is_stalled(100, 160.1))
 
+    async def test_process_watchdog_has_independent_deadline(self):
+        with patch.object(ARCHIVER, "PROCESS_WATCHDOG_SECONDS", 180):
+            self.assertFalse(ARCHIVER.process_watchdog_expired(100, 280))
+            self.assertTrue(ARCHIVER.process_watchdog_expired(100, 280.1))
+
+    async def test_task_cancellation_is_bounded(self):
+        release = ARCHIVER.asyncio.Event()
+
+        async def stubborn():
+            try:
+                await release.wait()
+            except ARCHIVER.asyncio.CancelledError:
+                await release.wait()
+
+        task = ARCHIVER.asyncio.create_task(stubborn())
+        await ARCHIVER.asyncio.sleep(0)
+        with (
+            patch.object(ARCHIVER, "TASK_CANCEL_TIMEOUT_SECONDS", 0.01),
+            self.assertLogs("binance-lob-archiver", level="ERROR"),
+        ):
+            with self.assertRaises(ARCHIVER.TaskCancellationStuck):
+                await ARCHIVER.cancel_tasks_bounded((task,))
+
+        release.set()
+        await task
+
     async def test_resync_does_not_reuse_expired_initial_deadline(self):
         self.assertTrue(ARCHIVER.bridge_timed_out(False, 100, 101))
         self.assertFalse(ARCHIVER.bridge_timed_out(True, 100, 101))