Migrate from tornado gen.coroutine to native async/await

[gmr]

Summary

Remove the tornado dependency and replace all tornado-specific async patterns with Python's native asyncio throughout the library.

Problem

The library used tornado's @gen.coroutine/yield pattern for async execution, which predates Python 3.5's async def/await syntax. This created a hard dependency on tornado for consumers that don't otherwise need it, and prevented interoperability with pure asyncio codebases.

Solution

• Replace all @gen.coroutine + yield with async def + await
• Replace tornado.locks.Lock/Event with asyncio.Lock/asyncio.Event
• Replace tornado.ioloop.IOLoop with the asyncio event loop
• Replace tornado.concurrent.Future with asyncio.Future
• Replace pika.adapters.tornado_connection with pika.adapters.asyncio_connection
• Replace tornado.iostream.IOStream with stdlib socket for the statsd TCP path
• Replace tornado.testing.AsyncTestCase + gen_test with unittest.IsolatedAsyncioTestCase
• Remove tornado from dependencies entirely

Also fixed a deadlock in the statsd TCP test teardown: wait_closed() was called before the client socket was closed, causing it to hang indefinitely.

Impact

Breaking change for consumers that subclass Consumer or SmartConsumer using @gen.coroutine/yield — these must be updated to async def/await. The public API shape is otherwise identical.

Consumers that were already using async def/await (compatible with tornado since Python 3.5) are unaffected.

All 225 tests pass.

Summary by CodeRabbit

• Chores
  • Removed Tornado from required project dependencies
  • Migrated internal asynchronous operations from Tornado framework to native Python asyncio
  • Updated consumer lifecycle and message connection management to use asyncio primitives
  • Modernized test infrastructure to use Python's standard async test utilities

[coderabbitai]

Warning

Rate limit exceeded

@gmr has exceeded the limit for the number of commits that can be reviewed per hour. Please wait 13 minutes and 0 seconds before requesting another review.

Your organization is not enrolled in usage-based pricing. Contact your admin to enable usage-based pricing to continue reviews beyond the rate limit, or try again in 13 minutes and 0 seconds.

⌛ How to resolve this issue?

After the wait time has elapsed, a review can be triggered using the @coderabbitai review command as a PR comment. Alternatively, push new commits to this PR.

We recommend that you space out your commits to avoid hitting the rate limit.

🚦 How do rate limits work?

CodeRabbit enforces hourly rate limits for each developer per organization.

Our paid plans have higher rate limits than the trial, open-source and free plans. In all cases, we re-allow further reviews after a brief timeout.

Please see our FAQ for further information.

ℹ️ Review info

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro

Run ID: 4f1f1265-cf26-4f09-9cdb-fafaa3a6b7be

📥 Commits

Reviewing files that changed from the base of the PR and between dbb97f9 and 7d2fbfd.

📒 Files selected for processing (1)

• tests/test_statsd.py

📝 Walkthrough

Walkthrough

The codebase undergoes a comprehensive migration from Tornado to native asyncio. Core library modules—consumer, process, mixins, and statsd—convert all async primitives to native async/await syntax, asyncio locks, and asyncio event loops. Test infrastructure is rebased to unittest.IsolatedAsyncioTestCase, and the Tornado dependency is removed from the project configuration.

Changes

Cohort / File(s)	Summary
Dependency & Project Config pyproject.toml	Removed tornado from required dependencies.
Core Consumer & Lifecycle rejected/consumer.py	Converted all lifecycle hooks (initialize, prepare, process, on_finish, on_blocked, on_unblocked, finish, yield_to_ioloop, execute) to native async methods. Shifted initialization from eager __init__ to lazy execution within execute() via _initialized flag. Replaced Tornado future/yield patterns with direct await calls. Removed io_loop property and Tornado-specific exception handling.
Mixin Async Updates rejected/mixins.py	Converted GarbageCollectorMixin.on_finish to async method, updating parent call from synchronous super().on_finish() to await super().on_finish().
Process & RabbitMQ Integration rejected/process.py	Migrated connection from pika.adapters.tornado_connection.TornadoConnection to pika.adapters.asyncio_connection.AsyncioConnection. Replaced Tornado coroutines with native async/await, converted yield self.consumer_lock.acquire() to async with self.consumer_lock, and switched from Tornado IOLoop scheduling to asyncio.ensure_future(). Updated event-loop lifecycle to pure asyncio patterns.
StatsD Transport rejected/statsd.py	Removed Tornado iostream dependency. Replaced IOStream-based TCP transport (_tcp_sock) with raw socket.socket (_tcp_writer). Rewrote TCP connection/reconnect logic, removing callback-based patterns and simplifying to direct socket operations.
Test Harness rejected/testing.py	Rebased AsyncTestCase from Tornado's testing.AsyncTestCase to unittest.IsolatedAsyncioTestCase. Converted process_message to native async method. Updated internal consumer execution to use await and asyncio patterns. Removed Tornado test utilities (gen_test, gen.coroutine).
Consumer Tests tests/test_consumer.py	Removed Tornado testing imports, replaced testing.AsyncTestCase with unittest.IsolatedAsyncioTestCase. Updated test methods to native async/await, changed process mocks to AsyncMock, and replaced initialization assertions with _initialized flag checks.
Process Tests tests/test_process.py	Migrated from Tornado to unittest.IsolatedAsyncioTestCase. Replaced tornado.locks.Lock() with asyncio.Lock(). Converted gen_test-based tests to native async methods using await for invoke_consumer.
StatsD Tests tests/test_statsd.py	Rewrote TCP test server from Tornado's tcpserver.TCPServer to asyncio.Protocol. Converted test class to unittest.IsolatedAsyncioTestCase with asyncSetUp/asyncTearDown. Updated test waiting from Tornado events to asyncio.wait_for(...event.wait()) patterns.
Testing Module Tests tests/test_testing.py	Converted test consumer process methods from Tornado generator coroutines to native async methods. Updated test methods from @gen_test with yield to native async def with await.

Sequence Diagram

sequenceDiagram
    participant Client
    participant Process
    participant Consumer
    participant EventLoop
    participant RabbitMQ

    Note over Process,EventLoop: Asyncio-Based Message Processing Flow (New)
    
    RabbitMQ->>Process: on_message_callback(message)
    Process->>EventLoop: asyncio.ensure_future(invoke_consumer)
    activate EventLoop
    EventLoop->>Process: invoke_consumer()
    activate Process
    Process->>Process: async with consumer_lock
    Process->>Consumer: await execute(message)
    activate Consumer
    Consumer->>Consumer: await prepare()
    Consumer->>Consumer: await process()
    Consumer->>Consumer: await on_finish()
    Consumer-->>Process: return
    deactivate Consumer
    Process-->>EventLoop: complete
    deactivate Process
    deactivate EventLoop
    EventLoop->>Client: task complete

Loading

Estimated code review effort

🎯 4 (Complex) | ⏱️ ~80 minutes

Poem

🐰 With asyncio's grace, we shed Tornado's weight,
Native coroutines now handle the fate,
Sockets dance freely, no IOStream in sight,
The message flows swift through the async night!

🚥 Pre-merge checks | ✅ 2 | ❌ 1

❌ Failed checks (1 warning)

Check name	Status	Explanation	Resolution
Docstring Coverage	⚠️ Warning	Docstring coverage is 35.14% which is insufficient. The required threshold is 80.00%.	Write docstrings for the functions missing them to satisfy the coverage threshold.

✅ Passed checks (2 passed)

Check name	Status	Explanation
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Title check	✅ Passed	The pull request title accurately and concisely describes the main objective—migrating from Tornado's @gen.coroutine patterns to native Python async/await syntax, which is the central theme across all file changes.

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch tornado-to-async

---

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.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[coderabbitai]

Actionable comments posted: 4

Caution

Some comments are outside the diff and can’t be posted inline due to platform limitations.

⚠️ Outside diff range comments (1)

rejected/statsd.py (1)

146-172: ⚠️ Potential issue | 🟠 Major

_tcp_on_closed() is orphan code—TCP reconnection after disconnect is not implemented.

The _tcp_on_closed() method is defined but never invoked anywhere in the codebase. In the previous Tornado implementation, this would have been wired via IOStream.set_close_callback(). After migration to raw sockets, this callback integration is missing, so TCP reconnection on disconnect will not occur.

When _tcp_socket() fails and returns None (line 169), _tcp_on_closed() assigns it directly to _tcp_writer without checking (line 151). However, this is handled safely in _send() by the guard clause on line 119 (if self._tcp_writer:), which falls back to UDP when _tcp_writer is None. Still, once a TCP connection fails, the system silently switches to UDP instead of attempting to reconnect.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@rejected/statsd.py` around lines 146 - 172, _tcp_on_closed is never invoked
after migrating away from Tornado, so TCP reconnection never happens; update the
TCP path so socket-close detection triggers _tcp_on_closed and make
_tcp_on_closed perform retry/backoff reconnects instead of simply assigning the
return of _tcp_socket to _tcp_writer. Specifically, wire a reader/monitor or
non-blocking recv/selector callback (or spawn a short-lived thread) that detects
socket shutdown and calls _tcp_on_closed, modify _tcp_on_closed to cancel/clean
the old socket, schedule repeated attempts to call _tcp_socket with exponential
backoff until it returns a socket (setting self._tcp_writer only on success),
and change _tcp_socket so connection failures do not call the global shutdown
callback immediately but instead return None (or raise a controlled exception)
so the reconnection logic can handle retries while _send continues to fall back
to UDP.

🧹 Nitpick comments (3)

rejected/statsd.py (2)

119-127: TCP error handling invokes _failure_callback() but does not clean up _tcp_writer.

When an OSError occurs during _send() on the TCP path, _failure_callback() is called but _tcp_writer is not cleared. This leaves a potentially broken socket reference. Subsequent sends would continue to fail and repeatedly call _failure_callback().

Consider clearing _tcp_writer when an error occurs:

♻️ Proposed fix

         except OSError as error:  # pragma: nocover
             if self._connected:
                 LOGGER.exception('Error sending statsd metric: %s', error)
                 self._connected = False
+                if self._tcp_writer:
+                    try:
+                        self._tcp_writer.close()
+                    except OSError:
+                        pass
+                    self._tcp_writer = None
                 self._failure_callback()

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@rejected/statsd.py` around lines 119 - 127, The OSError handler in _send()
clears _connected and calls _failure_callback() but leaves the broken
_tcp_writer reference; update the except block to also clean up the TCP writer
(e.g., close/terminate and set self._tcp_writer = None) when self._tcp_writer
was being used, so future sends don't reuse a bad socket; ensure the logic
touches the same symbols (_tcp_writer, _connected, _failure_callback, _send) and
still calls _failure_callback() after safely disposing the writer.

---

159-172: Blocking socket.connect() will stall the event loop if reconnection is restored.

If the _tcp_on_closed() reconnection logic is fixed to be invoked (e.g., via error handling), the synchronous socket.connect() call will block the asyncio event loop. Consider using loop.sock_connect() for non-blocking TCP connections, or document that TCP statsd is best-effort and connection failures trigger shutdown.

Given that _failure_callback() triggers process shutdown (per context snippet in rejected/process.py:1129-1133), the current behavior may be intentional—failure means shutdown, no reconnection. If so, _tcp_on_closed() can be removed as dead code.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@rejected/statsd.py` around lines 159 - 172, The current synchronous
socket.connect(self._address) in the TCP connection path blocks the asyncio
event loop; change the implementation in the statsd TCP connect routine
(referenced by self._address, self._failure_callback and _tcp_on_closed) to
perform a non-blocking connect via the event loop (use loop =
asyncio.get_running_loop(), set the socket non-blocking and await
loop.sock_connect(sock, self._address)), catch exceptions and call
self._failure_callback() the same way, set self._connected and return the socket
after the await; alternatively, if TCP failures are intended to be terminal, add
a clear comment and remove the now-dead _tcp_on_closed reconnection logic
instead of leaving blocking connect in place.

rejected/consumer.py (1)

230-234: Minor docstring inconsistency: still mentions Future semantics.

The docstring mentions "If this method returns a coroutine or asyncio.Future" but now prepare() is always an async def and must be awaited. Consider simplifying the docstring since the method is now natively async.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@rejected/consumer.py` around lines 230 - 234, The docstring for prepare()
still references "If this method returns a coroutine or asyncio.Future" which is
outdated because prepare() is now always defined as async; update the prepare()
docstring to remove the Future/coroutine conditional and instead state that
prepare() is an async def method that must be awaited and runs before
consumption starts (keep the "Asynchronous support" note but simplify wording to
reflect native async semantics). Make the change in the prepare() method's
docstring so the language clarifies it's natively asynchronous and requires
await.

🤖 Prompt for all review comments with AI agents

Verify each finding against the current code and only fix it if needed.

Inline comments:
In `@tests/test_statsd.py`:
- Around line 225-226: The test currently bypasses the client's reconnect logic
by directly assigning self.statsd._tcp_writer = self.statsd._tcp_socket() before
calling set_gauge('bar', 10); instead, simulate a broken connection (e.g.,
replace _tcp_writer with a writer that will fail on write or close it) and let
set_gauge('bar', 10) trigger the client's reconnect-on-send path, or monkeypatch
_tcp_socket() to provide a new socket when the client attempts to reconnect; do
not directly reassign _tcp_writer in the test so the client's own
broken-connection detection and reconnect logic (methods involving _tcp_writer
and _tcp_socket()) are exercised.
- Around line 152-164: The test currently reserves an ephemeral port by creating
and closing a temporary socket (sock.bind(('127.0.0.1', 0))) before calling
loop.create_server, which opens a race; instead let create_server bind to port 0
directly and then read the assigned port from the server socket(s). Replace the
temporary-socket block with calling loop.create_server(lambda:
self._server_protocol, '127.0.0.1', 0), await it to set self._server, and then
set self.port = self._server.sockets[0].getsockname()[1] so the test uses the
actual bound port; keep using StatsdServer, self._server_protocol and
self._server as-is.
- Around line 171-172: Remove the nondeterministic asyncio.sleep calls and
replace them with deterministic event waits: delete the short await
asyncio.sleep(0.01) after creating the statsd Client because _tcp_socket() is
synchronous and the connection is already established; for the later 2-second
sleep (after the server receives the 'reconnect' packet and reconnect_receive is
set), replace it with an explicit await on the concrete synchronization
primitive (e.g., await asyncio.wait_for(self._server_protocol.event.wait(),
timeout=...) or await asyncio.wait_for(reconnect_receive.wait(), timeout=...))
so the test waits for the actual state change instead of wall-clock time.

---

Outside diff comments:
In `@rejected/statsd.py`:
- Around line 146-172: _tcp_on_closed is never invoked after migrating away from
Tornado, so TCP reconnection never happens; update the TCP path so socket-close
detection triggers _tcp_on_closed and make _tcp_on_closed perform retry/backoff
reconnects instead of simply assigning the return of _tcp_socket to _tcp_writer.
Specifically, wire a reader/monitor or non-blocking recv/selector callback (or
spawn a short-lived thread) that detects socket shutdown and calls
_tcp_on_closed, modify _tcp_on_closed to cancel/clean the old socket, schedule
repeated attempts to call _tcp_socket with exponential backoff until it returns
a socket (setting self._tcp_writer only on success), and change _tcp_socket so
connection failures do not call the global shutdown callback immediately but
instead return None (or raise a controlled exception) so the reconnection logic
can handle retries while _send continues to fall back to UDP.

---

Nitpick comments:
In `@rejected/consumer.py`:
- Around line 230-234: The docstring for prepare() still references "If this
method returns a coroutine or asyncio.Future" which is outdated because
prepare() is now always defined as async; update the prepare() docstring to
remove the Future/coroutine conditional and instead state that prepare() is an
async def method that must be awaited and runs before consumption starts (keep
the "Asynchronous support" note but simplify wording to reflect native async
semantics). Make the change in the prepare() method's docstring so the language
clarifies it's natively asynchronous and requires await.

In `@rejected/statsd.py`:
- Around line 119-127: The OSError handler in _send() clears _connected and
calls _failure_callback() but leaves the broken _tcp_writer reference; update
the except block to also clean up the TCP writer (e.g., close/terminate and set
self._tcp_writer = None) when self._tcp_writer was being used, so future sends
don't reuse a bad socket; ensure the logic touches the same symbols
(_tcp_writer, _connected, _failure_callback, _send) and still calls
_failure_callback() after safely disposing the writer.
- Around line 159-172: The current synchronous socket.connect(self._address) in
the TCP connection path blocks the asyncio event loop; change the implementation
in the statsd TCP connect routine (referenced by self._address,
self._failure_callback and _tcp_on_closed) to perform a non-blocking connect via
the event loop (use loop = asyncio.get_running_loop(), set the socket
non-blocking and await loop.sock_connect(sock, self._address)), catch exceptions
and call self._failure_callback() the same way, set self._connected and return
the socket after the await; alternatively, if TCP failures are intended to be
terminal, add a clear comment and remove the now-dead _tcp_on_closed
reconnection logic instead of leaving blocking connect in place.

🪄 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

Run ID: e533143b-b7db-470c-8b14-08c0c3109b40

📥 Commits

Reviewing files that changed from the base of the PR and between c65bc0a and dbb97f9.

⛔ Files ignored due to path filters (1)

• uv.lock is excluded by !**/*.lock

📒 Files selected for processing (10)

• pyproject.toml
• rejected/consumer.py
• rejected/mixins.py
• rejected/process.py
• rejected/statsd.py
• rejected/testing.py
• tests/test_consumer.py
• tests/test_process.py
• tests/test_statsd.py
• tests/test_testing.py

💤 Files with no reviewable changes (1)

• pyproject.toml

[gmr]

🤖 This comment was posted by Claude on behalf of @gmr

Response to CodeRabbit's Review

Re: tests/test_statsd.py:164 — free-port probe race condition

Fixed in 7d2fbfd. The server now binds directly to port 0 and the assigned port is read from self._server.sockets[0].getsockname()[1], eliminating the TOCTOU window.

Re: tests/test_statsd.py:172 — unnecessary asyncio.sleep(0.01)

Also fixed in 7d2fbfd. Since _tcp_socket() is a synchronous blocking socket.connect(), the connection is fully established when the constructor returns. The sleep was unnecessary and has been removed.

Re: tests/test_statsd.py:226 — reconnect logic short-circuit

Respectfully disagreeing here. The statsd Client._send() method catches OSError and calls _failure_callback() — there is no automatic reconnect-on-send path in the client. The _tcp_on_closed() method exists but is never invoked from _send(). The test accurately reflects the actual behavior: the client detects send failures and signals via the callback, after which the caller is responsible for reconnecting. Testing a reconnect path that doesn't exist would be misleading.

Re: rejected/process.py:36 — AsyncSentryClient / Tornado dependency

This import block is pre-existing code unchanged by this PR. The PR's scope is migrating gen.coroutine to native async/await. The Sentry/raven dependency question (whether to use raven.Client vs AsyncSentryClient vs sentry-sdk) is a separate concern best addressed in a follow-up PR with proper testing of the Sentry integration.