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.49% +1.20%
==========================================
Files 69 69
Lines 2577 2697 +120
==========================================
+ Hits 2095 2225 +130
+ Misses 482 472 -10 ☔ 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.
There was a problem hiding this comment.
Using ExceptionGroup no longer provides any practical benefit. However, adopting it would change the exception contract and require dropping support for Python 3.10, which is still explicitly supported by the project and is part of the CI matrix.
Therefore, raising the project’s minimum supported Python version is outside the scope of this change.
I’m not in favor of introducing changes that conflict with the project’s current dependency constraints. We can revisit this as part of a technical debt initiative once support for Python 3.10 has been officially dropped.
- remove undocumented prefetch middleware hooks and metrics - preserve listener errors during task group failures - restore capacity when callback handoff fails - strengthen shutdown and cleanup coverage Refs: #528
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.