From abf68088803e972b31ba48c5e0430e73d1db4927 Mon Sep 17 00:00:00 2001 From: proerror Date: Mon, 13 Jul 2026 22:33:34 +0800 Subject: [PATCH 1/2] fix: recover deadlocked Binance receivers --- deployment/aliyun/README.md | 7 ++- .../aliyun/binance-lob-archiver-spot.env | 2 + .../aliyun/binance-lob-archiver-usdm.env | 2 + deployment/aliyun/binance_lob_archiver.py | 49 +++++++++++++++++-- .../aliyun/test_binance_lob_archiver.py | 26 ++++++++++ 5 files changed, 79 insertions(+), 7 deletions(-) diff --git a/deployment/aliyun/README.md b/deployment/aliyun/README.md index 8ce8589ea..4ab9de521 100644 --- a/deployment/aliyun/README.md +++ b/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 diff --git a/deployment/aliyun/binance-lob-archiver-spot.env b/deployment/aliyun/binance-lob-archiver-spot.env index d7e663002..e27296307 100644 --- a/deployment/aliyun/binance-lob-archiver-spot.env +++ b/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 diff --git a/deployment/aliyun/binance-lob-archiver-usdm.env b/deployment/aliyun/binance-lob-archiver-usdm.env index e6ed24d71..b50a843f6 100644 --- a/deployment/aliyun/binance-lob-archiver-usdm.env +++ b/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 diff --git a/deployment/aliyun/binance_lob_archiver.py b/deployment/aliyun/binance_lob_archiver.py index bcb8c795c..d71a77445 100644 --- a/deployment/aliyun/binance_lob_archiver.py +++ b/deployment/aliyun/binance_lob_archiver.py @@ -66,6 +66,9 @@ 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() UPLOAD_STATUS: dict[str, str | None] = { "last_success_at": None, "last_error_at": None, @@ -597,6 +600,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 +614,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 +666,39 @@ 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) + 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, ...]) -> int: + if not tasks: + return 0 + 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), + ) + return len(pending) + + def bridge_timed_out( previously_synced: bool, sync_deadline: float | None, now: float ) -> bool: @@ -981,11 +1019,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(): @@ -1152,6 +1186,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)) diff --git a/deployment/aliyun/test_binance_lob_archiver.py b/deployment/aliyun/test_binance_lob_archiver.py index fda5bf794..2c9568842 100644 --- a/deployment/aliyun/test_binance_lob_archiver.py +++ b/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"), + ): + pending = await ARCHIVER.cancel_tasks_bounded((task,)) + + self.assertEqual(pending, 1) + 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)) From ca27c24c9476f3435c8880ad060e98e21aee3eaa Mon Sep 17 00:00:00 2001 From: proerror Date: Mon, 13 Jul 2026 22:52:03 +0800 Subject: [PATCH 2/2] fix: make receiver stalls process-fatal --- deployment/aliyun/binance_lob_archiver.py | 27 ++++++++++++++++--- .../aliyun/test_binance_lob_archiver.py | 4 +-- 2 files changed, 26 insertions(+), 5 deletions(-) diff --git a/deployment/aliyun/binance_lob_archiver.py b/deployment/aliyun/binance_lob_archiver.py index d71a77445..3a61d6631 100644 --- a/deployment/aliyun/binance_lob_archiver.py +++ b/deployment/aliyun/binance_lob_archiver.py @@ -69,6 +69,7 @@ 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, @@ -97,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 @@ -674,6 +679,8 @@ 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( @@ -683,9 +690,9 @@ def run_process_watchdog() -> None: os._exit(75) -async def cancel_tasks_bounded(tasks: tuple[asyncio.Task, ...]) -> int: +async def cancel_tasks_bounded(tasks: tuple[asyncio.Task, ...]) -> None: if not tasks: - return 0 + return for task in tasks: task.cancel() _, pending = await asyncio.wait( @@ -696,7 +703,9 @@ async def cancel_tasks_bounded(tasks: tuple[asyncio.Task, ...]) -> int: "task cancellation timed out pending=%s", len(pending), ) - return len(pending) + raise TaskCancellationStuck( + f"task cancellation timed out pending={len(pending)}" + ) def bridge_timed_out( @@ -1064,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) @@ -1115,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) diff --git a/deployment/aliyun/test_binance_lob_archiver.py b/deployment/aliyun/test_binance_lob_archiver.py index 2c9568842..58ecacab2 100644 --- a/deployment/aliyun/test_binance_lob_archiver.py +++ b/deployment/aliyun/test_binance_lob_archiver.py @@ -348,9 +348,9 @@ async def stubborn(): patch.object(ARCHIVER, "TASK_CANCEL_TIMEOUT_SECONDS", 0.01), self.assertLogs("binance-lob-archiver", level="ERROR"), ): - pending = await ARCHIVER.cancel_tasks_bounded((task,)) + with self.assertRaises(ARCHIVER.TaskCancellationStuck): + await ARCHIVER.cancel_tasks_bounded((task,)) - self.assertEqual(pending, 1) release.set() await task