Uh oh!
There was an error while loading. Please reload this page.
fix: atexit final flush + dispose-during-connect (closes #6) - #28
Merged
Conversation
The old _LocalTransport.dispose() queued a None sentinel and joined with a 2s timeout. If the loop was inside websockets.connect()'s blocking upgrade handshake, the sentinel sat in the queue until the OS TCP timeout (~75 s on Linux), so the join timed out and the daemon thread plus its open socket FD leaked. Schedule loop.stop() via call_soon_threadsafe instead — this interrupts the connect coroutine and lets run_until_complete return promptly. Bound the join with shutdown_flush_timeout_ms (wired through Transport) and warn (not block) if the thread is still alive after that. Regression test stands up a TCP server that accepts but never completes the WebSocket upgrade, then dispose()s mid-handshake and asserts the join returns in under 3 s. Refs #6
Short-lived processes (cron jobs, Lambda invocations, one-shot CLI scripts, SIGTERM'd containers) dropped their last aggregator bucket because the flush timer is a daemon thread and dies on exit. The user either had to remember to call handle.dispose() manually or accept silent data loss for the final window. Register handle.dispose() via atexit when init() runs, gated by the new RecostConfig.auto_shutdown_handlers (default True). The callback is idempotent and unregisters itself on explicit dispose so a process that cycles init/dispose does not accumulate dead atexit hooks. Regression tests: two in-process checks (registered when enabled, NOT registered when opted out) plus a subprocess end-to-end test that verifies a normal sys.exit(0) triggers the final flush. Refs #6
📝 WalkthroughWalkthroughThis PR addresses process lifecycle gaps by implementing graceful shutdown through atexit registration and fixing resource leaks during dispose. It adds a configurable shutdown timeout to prevent FD accumulation in long-lived processes, registers automatic final flushes on normal exit for short-lived processes, and provides comprehensive test coverage including subprocess and blocking-connect scenarios. ChangesProcess Lifecycle and Transport Shutdown
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~25 minutes Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ 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 |
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 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.
Inline comments:
In `@tests/test_transport.py`:
- Around line 417-437: The test helper _start_blackhole_server currently
accumulates accepted client sockets in the local accepted list and never closes
them, leaking file descriptors; modify the implementation so accepted sockets
are closed during teardown by either (a) attaching the accepted list to the
returned srv object (e.g. srv._accepted = accepted) or returning a tuple (srv,
accepted) and then updating tests to iterate over accepted and call .close()
before/after closing srv, and also ensure accept_loop closes any connections
when detecting srv.fileno() == -1 (or on shutdown) to avoid leaving sockets
open; update references to accept_loop, accepted, and _start_blackhole_server
accordingly.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: 574fcaa6-ac40-41fe-b206-7f9771016e62
📒 Files selected for processing (5)
recost/_init.pyrecost/_transport.pyrecost/_types.pytests/test_init.pytests/test_transport.py
Comment on lines
+417
to
+437
| def _start_blackhole_server(port: int) -> socket.socket: | ||
| """Bind+listen on `port` and accept connections but never respond | ||
| to the HTTP upgrade. websockets.connect() will TCP-connect, send its | ||
| upgrade request, and then block waiting for an HTTP response.""" | ||
| srv = socket.socket() | ||
| srv.bind(("127.0.0.1", port)) | ||
| srv.listen(8) | ||
| accepted: list[socket.socket] = [] | ||
| def accept_loop() -> None: | ||
| srv.settimeout(0.5) | ||
| while True: | ||
| try: | ||
| conn, _ = srv.accept() | ||
| accepted.append(conn) | ||
| except (OSError, socket.timeout): | ||
| if srv.fileno() == -1: | ||
| return | ||
| threading.Thread(target=accept_loop, daemon=True).start() | ||
| return srv |
There was a problem hiding this comment.
Close accepted blackhole sockets during teardown.
At Line 431, accepted client sockets are retained but never closed; Line 462 only closes the listening socket. This can leak FDs across tests.
Proposed fix
- def _start_blackhole_server(port: int) -> socket.socket:+ def _start_blackhole_server(port: int) -> tuple[socket.socket, list[socket.socket]]:
@@
- return srv+ return srv, accepted
@@
- server = self._start_blackhole_server(port)+ server, accepted = self._start_blackhole_server(port)
@@
finally:
+ for conn in accepted:+ try:+ conn.close()+ except OSError:+ pass
server.close()Also applies to: 461-463
🤖 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 `@tests/test_transport.py` around lines 417 - 437, The test helper
_start_blackhole_server currently accumulates accepted client sockets in the
local accepted list and never closes them, leaking file descriptors; modify the
implementation so accepted sockets are closed during teardown by either (a)
attaching the accepted list to the returned srv object (e.g. srv._accepted =
accepted) or returning a tuple (srv, accepted) and then updating tests to
iterate over accepted and call .close() before/after closing srv, and also
ensure accept_loop closes any connections when detecting srv.fileno() == -1 (or
on shutdown) to avoid leaving sockets open; update references to accept_loop,
accepted, and _start_blackhole_server accordingly.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for freeto join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Closes#6.
Two surgical fixes to process lifecycle:
init()registershandle.dispose()viaatexitso short-lived processes (cron, Lambda, one-shot CLI, SIGTERMin a container) flush their last bucket instead of dropping it when
the daemon flush thread is killed. Opt-out via
RecostConfig.auto_shutdown_handlers=False._LocalTransport.dispose()now schedulesloop.stop()viacall_soon_threadsafeinstead of waiting for aqueue sentinel. If the loop is blocked inside
websockets.connect()'supgrade handshake, the sentinel-in-queue dispose used to leave the
daemon thread + socket FD pinned until the OS TCP timeout (~75 s).
Tests
+ "" +tests/test_transport.py::TestDisposeDuringConnect+ "" +— stands up aTCP server that accepts the connection but never responds to the
WebSocket upgrade, then calls
+ "" +dispose()+ "" +mid-handshake. Post-fixdispose returns in well under one second (threshold 1.5 s).
+ "" +tests/test_init.py::TestAtexitFlush+ "" +— three tests: atexit ISregistered when enabled, atexit is NOT registered when opted out,
and an end-to-end subprocess test that verifies a real
+ "" +sys.exit(0)+ "" +triggers the final flush.
Notes
+ "" +RecostConfig.auto_shutdown_handlers+ "" +defaults to
+ "" +True+ "" +— existing callers automatically get the betterbehavior with no migration.
+ "" +dispose()+ "" +solong-lived processes that cycle
+ "" +init+ "" +/+ "" +dispose+ "" +don't accumulatedead callbacks.
_handle, install/uninstall, init-vs-dispose #4) —+ "" +_init_lock+ "" +already serializes+ "" +init+ "" +/+ "" +dispose+ "" +, so atexit re-entering+ "" +dispose()+ "" +from the mainthread is safe against user-driven dispose.
+ "" +loop.stop()+ "" +on the loop thread nowcauses
+ "" +run_until_complete+ "" +to raise+ "" +RuntimeError: Event loop stopped before Future completed+ "" +—caught with a narrow message match so genuine coroutine bugs still
surface.
🤖 Generated with Claude Code
Summary by CodeRabbit
New Features
Bug Fixes