fix(receiver): harden prefetch and listener lifecycle

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.

Author: GefMar

docs/guide/cli.md

@@ -162,7 +162,7 @@ The number of signals before a hard kill can be configured with the `--hardkill-
 * `--log-format` is used to set a log format (default `%(asctime)s][%(name)s][%(levelname)-7s][%(processName)s] %(message)s`).
 * `--max-async-tasks` - maximum number of simultaneously running async tasks.
 * `--max-async-tasks-jitter` – Randomly varies the max async task limit between --max-async-tasks and a jittered value, helping prevent simultaneous worker restarts.
-* `--max-prefetch` - number of tasks to be prefetched before execution. (Useful for systems with high message rates, but brokers should support acknowledgements).
+* `--max-prefetch` - maximum number of deliveries buffered in addition to the listener's single in-flight delivery. The default `0` disables additional buffering while still allowing one delivery to move from the broker listener toward execution. The value must be non-negative. This option is useful for systems with high message rates, but brokers should support acknowledgements.
 * `--max-threadpool-threads` - number of threads for sync function execution.
 * `--no-propagate-errors` - if this parameter is enabled, exceptions won't be thrown in generator dependencies.
 * `--receiver` - python path to custom receiver class.

taskiq/api/receiver.py

@@ -45,7 +45,8 @@ async def run_receiver_task(
     :param validate_params: whether to validate params or not.
     :param max_async_tasks: maximum number of simultaneous async tasks.
     :param max_async_tasks_jitter: random jitter to add to max_async_tasks.
-    :param max_prefetch: maximum number of tasks to prefetch.
+    :param max_prefetch: maximum number of deliveries buffered in addition to
+        the listener's single in-flight delivery. Must be non-negative.
     :param propagate_exceptions: whether to propagate exceptions in generators or not.
     :param run_startup: whether to run startup function or not.
     :param ack_time: acknowledge type to use.

taskiq/cli/worker/args.py

@@ -224,7 +224,10 @@ def from_cli(
             type=int,
             dest="max_prefetch",
             default=0,
-            help="Maximum prefetched tasks per worker process. ",
+            help=(
+                "Maximum deliveries buffered beyond the in-flight delivery. "
+                "Must be non-negative. "
+            ),
         )
         parser.add_argument(
             "--no-configure-logging",