fix(receiver): harden prefetch and listener lifecycle - #645
Conversation
Track delivery and prefetch capacity ownership across shutdown paths. Close pending listener reads, preserve late deliveries, and restore fail-fast transport error propagation. Keep middleware accounting balanced, report discarded deliveries, and retain deferred lifecycle diagnostics when task group failures overlap.
e15d278 to
4321e09
Compare
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## master #645 +/- ##
==========================================
+ Coverage 81.29% 82.71% +1.41%
==========================================
Files 69 69
Lines 2577 2742 +165
==========================================
+ Hits 2095 2268 +173
+ Misses 482 474 -8 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
Reject invalid prefetch values before receiver retries and retain delivery accounting until execution capacity is acquired. Add deterministic coverage for CLI defaults, cancellation races, and listener lifecycle cleanup. Refs #528
| "A Receiver listener lifecycle error was recorded before " | ||
| "the task group failed.", | ||
| exc_info=(type(error), error, error.__traceback__), |
There was a problem hiding this comment.
This one can be simplified.
| "A Receiver listener lifecycle error was recorded before " | |
| "the task group failed.", | |
| exc_info=(type(error), error, error.__traceback__), | |
| "A Receiver listener lifecycle error was recorded before " | |
| "the task group failed.\n%s" % error, | |
| exc_info=True, |
| state.owns_delivery_slot = False | ||
| return late_delivery | ||
|
|
||
| async def _notify_prefetch_hook(self, hook_name: str) -> None: |
There was a problem hiding this comment.
Let's remove this function and whole on_prefetch_queue_add and on_prefetch_queue_remove. They are not part of middelware ABC and should not be used.
I actually missed that thing when opentelemetry support was added. Let's remove this functionality, since it's internal implementation details. I don't think those functions should have existed.
| self._record_listen_error(exc) | ||
| queue.put_nowait(_QueueSignal.DONE) |
There was a problem hiding this comment.
Since we're running in a anyio group, I suppose you can simply do something like this:
| self._record_listen_error(exc) | |
| queue.put_nowait(_QueueSignal.DONE) | |
| queue.put_nowait(_QueueSignal.DONE) | |
| raise |
| finally: | ||
| queue.put_nowait(_QueueSignal.DONE) | ||
| finish_waiter.cancel() | ||
| await asyncio.gather(finish_waiter, return_exceptions=True) |
There was a problem hiding this comment.
| await asyncio.gather(finish_waiter, return_exceptions=True) | |
| with contextlib.supress(asyncio.CanceledError): | |
| await finish_waiter |
Idk, IMO it's just more obvious what is going on with such code.
| await self._settle_delivery_acquire(acquire_task) | ||
| raise | ||
|
|
||
| if finish_waiter in done or finish_event.is_set(): |
There was a problem hiding this comment.
| if finish_waiter in done or finish_event.is_set(): | |
| if finish_waiter.done() or finish_event.is_set(): |
I guess that'd work and will look a bit cleaner. Also, you won't need done.
| ) -> None: | ||
| """Release capacity and instrumentation for abandoned queue entries.""" | ||
| discarded_messages = 0 | ||
| while True: |
There was a problem hiding this comment.
| while True: | |
| while queue: |
You can use this condition to simplify inner body of the loop. Because you won't need to check on QueueEmpty exception.
| ) | ||
| state.owns_delivery_slot = False | ||
| try: | ||
| await self._notify_prefetch_hook("on_prefetch_queue_add") |
| try: | ||
| await maybe_awaitable(hook()) | ||
| except BaseException as exc: | ||
| if first_error is None: |
There was a problem hiding this comment.
This approach ignores all the errors except the first one. I mean, it's fine, because you raise it anyway. But, I'd rather create an exception group.
I think it's safe to bump MPE to 3.11 and start using ExceptionGroup for such case. It'd be a good fit and reasonable upgrade.
Since python 3.10 will be completely discarded in 3 months.
2 participants
Track delivery and prefetch capacity ownership across shutdown paths. Close pending listener reads, preserve late deliveries, and restore fail-fast transport error propagation.
Keep middleware accounting balanced, report discarded deliveries, and retain deferred lifecycle diagnostics when task group failures overlap.
Refs #528
im evaluated PR#630 and retained its valid requirement: the Receiver must not pull a message it cannot execute.
However, that implementation also introduces a broad retry loop that treats transport, middleware, and programming errors alike, while pending-read, iterator, late-delivery, and semaphore ownership remain implicit.
This PR instead fixes the underlying ownership invariant. Each admitted delivery explicitly owns capacity from broker read through callback completion, pending reads are cancelled and awaited, iterators are closed, and failures remain visible to the worker lifecycle owner.
Retry and backoff remain a separate policy rather than being introduced implicitly as part of the prefetch fix.