fix: recover deadlocked Binance receivers - #6
Conversation
📝 WalkthroughWalkthroughThe Aliyun Binance archiver now includes configurable process watchdog and bounded task-cancellation controls. It records market-data activity, exits after prolonged stalls, detects stuck shutdown tasks, and documents the recovery behavior. ChangesAliyun process recovery
Estimated code review effort: 3 (Moderate) | ~20 minutes Sequence Diagram(s)sequenceDiagram
participant WebSocketReceiveLoop
participant LAST_DATA_AT
participant run_process_watchdog
participant systemd
WebSocketReceiveLoop->>LAST_DATA_AT: record depth-frame timestamp
run_process_watchdog->>LAST_DATA_AT: check elapsed time
run_process_watchdog->>systemd: exit process after watchdog deadline
sequenceDiagram
participant run_session
participant cancel_tasks_bounded
participant asyncioTasks
participant collect
run_session->>cancel_tasks_bounded: cancel session tasks
cancel_tasks_bounded->>asyncioTasks: await cancellation with timeout
cancel_tasks_bounded-->>collect: raise TaskCancellationStuck
collect->>collect: write fatal health record and re-raise
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: ca27c24c94
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| await asyncio.gather( | ||
| *tasks, *resync_tasks.values(), return_exceptions=True | ||
| ) | ||
| await cancel_tasks_bounded(tuple([*tasks, *resync_tasks.values()])) |
There was a problem hiding this comment.
When cancel_tasks_bounded times out, it raises here before the existing queue-drain loop runs. In the new stuck-close fatal path, any diff/snapshot frames already read from the websocket and sitting in queue are discarded on process restart instead of being written as archived-only/replayed-unsafe as the following drain logic intends; preserve the timeout error but flush the queued frames before re-raising it.
Useful? React with 👍 / 👎.
| SPOOL_DIR.mkdir(parents=True, exist_ok=True) | ||
| recover_parts() | ||
| LAST_DATA_AT = time.monotonic() | ||
| PROCESS_WATCHDOG_ARMED = True |
There was a problem hiding this comment.
Arming the process watchdog for the entire collector also leaves it active while ArchiveRuntime.rotate() is compressing a segment, even though finalize_segment() is allowed to spend up to ZSTD_TIMEOUT_SECONDS (300s) in zstd. On the full-market services, if a scheduled/shutdown rotation or compression stall applies backpressure long enough that receivers stop reading for more than PROCESS_WATCHDOG_SECONDS (180s), the watchdog takes the os._exit path and drops the in-memory queue instead of letting the normal .part recovery path preserve the segment; disarm or heartbeat the watchdog around intentional long rotations.
Useful? React with 👍 / 👎.
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
🧹 Nitpick comments (1)
deployment/aliyun/binance_lob_archiver.py (1)
1031-1031: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueOptional: simplify tuple construction.
tuple([*tasks, *resync_tasks.values()])allocates an intermediate list. A direct tuple literal avoids that.♻️ Proposed refactor
- await cancel_tasks_bounded(tuple([*tasks, *resync_tasks.values()])) + await cancel_tasks_bounded((*tasks, *resync_tasks.values()))🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@deployment/aliyun/binance_lob_archiver.py` at line 1031, In the task cancellation call, replace the intermediate-list tuple construction around cancel_tasks_bounded with a direct tuple construction combining tasks and resync_tasks.values(), preserving the same elements and ordering.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Nitpick comments:
In `@deployment/aliyun/binance_lob_archiver.py`:
- Line 1031: In the task cancellation call, replace the intermediate-list tuple
construction around cancel_tasks_bounded with a direct tuple construction
combining tasks and resync_tasks.values(), preserving the same elements and
ordering.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 12db39ec-5639-4854-a012-dea90b9defae
📒 Files selected for processing (5)
deployment/aliyun/README.mddeployment/aliyun/binance-lob-archiver-spot.envdeployment/aliyun/binance-lob-archiver-usdm.envdeployment/aliyun/binance_lob_archiver.pydeployment/aliyun/test_binance_lob_archiver.py
1 participant
Summary
systemd activestate.Root cause evidence
Both services had no network socket and no file growth after approximately 22:17, while systemd remained active. The Python main threads were idle in
epoll_waitwith no scheduled timeout, consistent withrun_sessionhanging while gathering cancelled receiver tasks.Verification
git diff --checkpass.status=synced,sequence_gaps=0.Test plan
python3 -m unittest deployment/aliyun/test_binance_lob_archiver.pypython3 deployment/aliyun/binance_lob_archiver.py --self-testpython3 -m py_compile deployment/aliyun/binance_lob_archiver.pygit diff --checkSummary by CodeRabbit
Bug Fixes
Documentation