Fix message-processing lifecycle, shutdown, and reconnect bugs#100
Conversation
Fixes a group of verified bugs across the process/consumer/connection message lifecycle. Each is covered by a new or updated unit test. Closes #68 - Consumer._process_standard now calls finish() after process() Closes #69 - _republish preserves original AMQP properties, not just headers Closes #70 - invoke_consumer pops the in-flight entry on DecodeError Closes #71 - stop() sets STOP_REQUESTED before shutting down connections so the wait-for-in-flight path is no longer dead code Closes #72 - pending shutdown tasks (consumer.shutdown, codec.close) are run to completion after the loop stops instead of being abandoned Closes #73 - _in_flight is keyed by (connection_name, delivery_tag) to avoid cross-connection tag collisions Closes #74 - Connection.shutdown() is a no-op when closed, guards connection.close(), no longer deletes self.connection; reconnect assigns the new AsyncioConnection Closes #75 - a terminal channel error while processing now reconnects rather than leaving a zombie process Closes #78 - the legacy SSL path sets check_hostname before verify_mode Closes #80 - a non-numeric X-Processing-Exceptions header is treated as 0 instead of raising ValueError into an infinite requeue loop Closes #81 - correlation id is carried per-message (contextvar + ctx), Consumer runs _pre_execute inside the lock, and initialize() is guarded Closes #82 - the KeyboardInterrupt handler no longer double-nacks Closes #84 - a reconnecting connection re-issues its own basic_consume Closes #85 - reject() is a metrics-only no-op under no_ack instead of raising Closes #86 - returned messages (delivery_tag=None) skip broker ack/nack Closes #87 - shutdown_connections also shuts down CONNECTING connections Closes #88 - SIGABRT/SIGPROF handlers only schedule work via call_soon_threadsafe, keeping pika interaction on the loop Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (4)
🚧 Files skipped from review as they are similar to previous changes (4)
📝 WalkthroughWalkthroughThis PR updates connection shutdown and TLS handling, makes consumer initialization and correlation tracking concurrency-safe, preserves AMQP properties on republish, hardens processing-exception parsing, and revises process in-flight tracking, reconnect, shutdown, and signal behavior with new tests. ChangesConnection Shutdown & TLS Hardening
Consumer Concurrency & Lifecycle Fixes
Process In-Flight Tracking & Shutdown Reliability
Estimated code review effort: 4 (Complex) | ~75 minutes Sequence Diagram(s)sequenceDiagram
participant MessageA
participant MessageB
participant Consumer
participant InitLock
participant LogAdapter
MessageA->>Consumer: execute()
MessageB->>Consumer: execute()
Consumer->>InitLock: _ensure_initialized()
InitLock-->>Consumer: initialize() runs once
Consumer->>LogAdapter: set correlation_id context
LogAdapter-->>Consumer: task-local correlation_id
sequenceDiagram
participant Broker
participant Process
participant Connection
participant IOLoop
Broker--xConnection: failure / close
Connection->>Process: on_connection_failure()
Process->>Connection: connect() when processing
Connection-->>Process: on_connection_ready()
Process->>Connection: consume()
Process->>IOLoop: call_soon_threadsafe(stop or _report_stats)
Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
Comment |
|
@coderabbitai review |
✅ Action performedReview finished.
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (2)
rejected/connection.py (1)
87-93: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winSibling close paths remain unguarded.
The new
elsebranch guardsself.connection.close()againstAttributeError/ConnectionWrongStateError, but theis_active/self.channel→basic_cancel(...)branch (Lines 79-86) and theelif self.channel: self.channel.close()branch (Line 87-88) have no equivalent guard. A channel already mid-close or in a wrong state could raise the same class of pika exceptions this fix is meant to prevent.♻️ Suggested guard for the channel-close branch
elif self.channel: - self.channel.close() + try: + self.channel.close() + except (AttributeError, pika.exceptions.ChannelWrongStateError): + pass🤖 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 `@rejected/connection.py` around lines 87 - 93, The close logic in the connection teardown path still leaves the channel-based branches unprotected, so `ConnectionWrongStateError` or `AttributeError` can escape from `basic_cancel(...)` or `self.channel.close()`. Update the close flow in `Connection.close` to apply the same exception guard used for `self.connection.close()` to the `is_active`/`self.channel.basic_cancel(...)` path and the `elif self.channel: self.channel.close()` path, using the existing `self.channel` and `self.connection` branches as the fix points.tests/test_process.py (1)
428-435: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winTest covers only state transition, not the actual deferred-shutdown drain.
test_stop_while_processing_defers_shutdownverifiesstop()setsSTATE_STOP_REQUESTEDand skips immediateshutdown_connections, but it does not assert that once_in_flightdrains,on_processed()actually invokesshutdown_connections. Sinceon_processedgates that onis_waiting_to_shutdownwhileis_processingalso matchesSTATE_STOP_REQUESTED, adding a follow-through test (drain last in-flight message, assert shutdown) would close the gap for bug#71.🤖 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_process.py` around lines 428 - 435, The current test only checks that stop() switches Process to STATE_STOP_REQUESTED and skips immediate shutdown, but it misses the deferred drain path. Extend test_stop_while_processing_defers_shutdown in test_process.py to follow through by simulating the last in-flight message finishing and calling on_processed(), then assert that shutdown_connections is invoked once _in_flight reaches zero. Use the existing Process methods stop(), on_processed(), and the is_waiting_to_shutdown/STATE_STOP_REQUESTED flow to verify the deferred shutdown behavior end to end.
🤖 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 `@rejected/consumer.py`:
- Around line 166-173: The correlation-id handling in FunctionalConsumer is
still using shared instance state and an inconsistent missing-value check.
Update the correlation_id property and the execute/_pre_execute path so
per-message reads use task-local state (or otherwise avoid relying on
self._correlation_id across concurrent process(ctx) calls), keeping the public
property aligned with the contextvar-based logging behavior. Also make the
msg.correlation_id write-back condition match the cid selection logic in the
same branch, so empty-string correlation IDs are treated consistently with None
and the message is republished with the same id that was logged.
---
Nitpick comments:
In `@rejected/connection.py`:
- Around line 87-93: The close logic in the connection teardown path still
leaves the channel-based branches unprotected, so `ConnectionWrongStateError` or
`AttributeError` can escape from `basic_cancel(...)` or `self.channel.close()`.
Update the close flow in `Connection.close` to apply the same exception guard
used for `self.connection.close()` to the
`is_active`/`self.channel.basic_cancel(...)` path and the `elif self.channel:
self.channel.close()` path, using the existing `self.channel` and
`self.connection` branches as the fix points.
In `@tests/test_process.py`:
- Around line 428-435: The current test only checks that stop() switches Process
to STATE_STOP_REQUESTED and skips immediate shutdown, but it misses the deferred
drain path. Extend test_stop_while_processing_defers_shutdown in test_process.py
to follow through by simulating the last in-flight message finishing and calling
on_processed(), then assert that shutdown_connections is invoked once _in_flight
reaches zero. Use the existing Process methods stop(), on_processed(), and the
is_waiting_to_shutdown/STATE_STOP_REQUESTED flow to verify the deferred shutdown
behavior end to end.
🪄 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: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: eebc78d2-3106-48b3-9437-abdc55a5464e
📒 Files selected for processing (7)
rejected/connection.pyrejected/consumer.pyrejected/log.pyrejected/process.pytests/test_connection.pytests/test_consumer.pytests/test_process.py
- consumer.py: make correlation_id write-back use `not msg.correlation_id` so an empty-string id is treated the same as None and the message is republished with the id that was logged. - consumer.py: have the public `correlation_id` property prefer the task-local contextvar, so concurrent FunctionalConsumer messages don't read another in-flight message's id via shared instance state. - connection.py: guard the channel.close() teardown branch against AttributeError/ChannelWrongStateError, matching the connection.close() branch (the basic_cancel branch is intentionally left unguarded). - tests: extend correlation isolation test to assert the property is per-message, and add a deferred-shutdown drain follow-through test for issue #71. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
|
Addressed the two nitpicks in 4d1a102:
|
1 participant
Summary
Fixes 17 verified bugs in the message-processing lifecycle across
process.py,consumer.py,connection.py, andlog.py. These cover the ack/nack path, in-flight tracking, connection reconnect/shutdown, signal handling, and consumer lifecycle hooks.Problem / Solution
Each fix is surgical and covered by a reproducing test.
Consumer.finish()was documented in the lifecycle but never called → addedawait self.finish()in_process_standard._republishdropped all AMQP properties except headers, breaking error/drop-exchange retry pipelines → now copies content type/encoding, type, correlation_id, priority, expiration.DecodeErrorleaked the message in_in_flight, wedging state/shutdown → pop the tag in the error branch.Process.stop()was dead code → setSTATE_STOP_REQUESTEDbeforeshutdown_connections.consumer.shutdown()/codec.close()ran → pending tasks now run to completion afterrun_forever._in_flightkeyed by delivery tag alone collided across connections → now keyed by(connection.name, delivery_tag).Connection.shutdown()crashed on an already-closed connection → no-op when closed, guardedclose(), stopped deletingself.connection.Process.on_connection_failurereconnects when processing.check_hostname = Trueset beforeverify_mode.X-Processing-Exceptionsheader caused an infinite requeue loop → conversions wrapped, default to 0.contextvar;_pre_executeruns inside the lock;initialize()guarded.KeyboardInterrupthandler double-nacked the same delivery tag → removed the redundantreject().basic_consumenever re-issued after reconnect in multi-connection consumers → re-issues for the specific connection.ack: falseconsumers raisedRuntimeErroron any non-ACK result → rejection branches skip the broker nack whenno_ack.Basic.Return,delivery_tag=None) hit the ack/nack path → short-circuited.shutdown_connectionsskippedSTATE_CONNECTINGconnections, hanging shutdown → now shuts them down too.call_soon_threadsafeonto the loop.Testing
Full suite green (231 tests) + ruff clean. New/updated tests in
tests/test_process.py,tests/test_consumer.py, and newtests/test_connection.py.Closes #68
Closes #69
Closes #70
Closes #71
Closes #72
Closes #73
Closes #74
Closes #75
Closes #78
Closes #80
Closes #81
Closes #82
Closes #84
Closes #85
Closes #86
Closes #87
Closes #88
Summary by CodeRabbit