fix(inmemory): stabilize task ordering and lifecycle

Ensure post_send completes before local execution while preserving
per-invocation ownership and await_inplace semantics.

Drain accepted tasks during shutdown, retain background failures, and
clean up middleware, result backend, and executor resources reliably.

Add concurrency, cancellation, lifecycle, and compatibility coverage.

Closes: #586

Author: GefMar

docs/available-components/brokers.md

@@ -12,6 +12,11 @@ This is a special broker for local development. It uses the same functions to ex
 but all tasks are executed locally in the current thread.
 By default it uses `InMemoryResultBackend` but this can be overridden.
 
+`startup()` and `shutdown()` run both the client and worker event phases because
+the broker performs both roles in one process. They also own middleware and
+result backend lifecycle. Shutdown waits for local background executions before
+closing those resources and the synchronous task executor.
+
 ## ZeroMQBroker
 
 This broker uses [ZMQ](https://zeromq.org/) to communicate between worker and client processes.

docs/guide/testing-taskiq.md

@@ -165,6 +165,10 @@ broker = InMemoryBroker(await_inplace=True)
 
 With this setup all `await function.kiq()` calls will behave similarly to `await function()`, but
 with dependency injection and all taskiq-related functionality.
+The normal middleware boundary is preserved in both execution modes: all
+`post_send` hooks finish before `pre_execute` and the task body begin.
+With `await_inplace=True`, the `kiq()` call also returns only after inline
+execution completes.
 
 2. Alternatively, you can manually await all tasks after invoking the
    target function by using the `wait_all` method.
@@ -182,6 +186,18 @@ async def test_add_one():
     # have been completed and do all the assertions.
 ```
 
+`wait_all()` also surfaces the first pending background callback failure even
+if that callback completed before `wait_all()` was called. The failure is
+consumed after it is raised, so a later `wait_all()` only observes newer work.
+Cancelling a `wait_all()` call cancels only that waiter; accepted executions
+remain tracked and can be drained by a later call.
+Calling `shutdown()` performs the same drain before closing middleware, result
+backend, and executor resources. If shutdown is cancelled while draining, it
+finishes the drain and resource cleanup before propagating the cancellation.
+Both drain methods must be called by the external test or application lifecycle
+owner. Calling `wait_all()` or `shutdown()` from a task or `post_send` hook
+managed by the same broker raises `RuntimeError` instead of waiting on itself.
+
 ## Dependency injection
 
 If you use dependencies in your tasks, you may think that this can become a problem for tests. But it's not.

taskiq/abc/broker.py

@@ -23,7 +23,7 @@
 from taskiq.acks import AckableMessage
 from taskiq.decor import AsyncTaskiqDecoratedTask
 from taskiq.events import TaskiqEvents
-from taskiq.exceptions import TaskBrokerMismatchError
+from taskiq.exceptions import SendTaskError, TaskBrokerMismatchError
 from taskiq.formatters.proxy_formatter import ProxyFormatter
 from taskiq.message import BrokerMessage
 from taskiq.result_backends.dummy import DummyResultBackend
@@ -186,39 +186,90 @@ def add_middlewares(self, *middlewares: "TaskiqMiddleware") -> None:
 
     async def startup(self) -> None:
         """Do something when starting broker."""
-        event = TaskiqEvents.CLIENT_STARTUP
-        if self.is_worker_process:
-            event = TaskiqEvents.WORKER_STARTUP
-
-        for handler in self.event_handlers[event]:
-            await maybe_awaitable(handler(self.state))
+        for event in self._get_startup_events():
+            for handler in self.event_handlers[event]:
+                await maybe_awaitable(handler(self.state))
 
         for middleware in self.middlewares:
             if middleware.__class__.startup != TaskiqMiddleware.startup:
                 await maybe_awaitable(middleware.startup())
 
         await self.result_backend.startup()
 
+    def _get_startup_events(self) -> tuple[TaskiqEvents, ...]:
+        """Return event phases owned by this broker startup."""
+        if self.is_worker_process:
+            return (TaskiqEvents.WORKER_STARTUP,)
+        return (TaskiqEvents.CLIENT_STARTUP,)
+
     async def shutdown(self) -> None:
         """
         Close the broker.
 
         This method is called,
         when broker is closing.
         """
-        event = TaskiqEvents.CLIENT_SHUTDOWN
-        if self.is_worker_process:
-            event = TaskiqEvents.WORKER_SHUTDOWN
+        shutdown_error: BaseException | None = None
 
-        # Call all shutdown events.
-        for handler in self.event_handlers[event]:
-            await maybe_awaitable(handler(self.state))
+        for event in self._get_shutdown_events():
+            for handler in self.event_handlers[event]:
+                try:
+                    await maybe_awaitable(handler(self.state))
+                except BaseException as exc:
+                    shutdown_error = self._remember_shutdown_error(
+                        shutdown_error,
+                        exc,
+                    )
 
         for middleware in self.middlewares:
             if middleware.__class__.shutdown != TaskiqMiddleware.shutdown:
-                await maybe_awaitable(middleware.shutdown())
+                try:
+                    await maybe_awaitable(middleware.shutdown())
+                except BaseException as exc:
+                    shutdown_error = self._remember_shutdown_error(
+                        shutdown_error,
+                        exc,
+                    )
+
+        try:
+            await self.result_backend.shutdown()
+        except BaseException as exc:
+            shutdown_error = self._remember_shutdown_error(shutdown_error, exc)
+
+        if shutdown_error is not None:
+            raise shutdown_error
+
+    def _get_shutdown_events(self) -> tuple[TaskiqEvents, ...]:
+        """Return event phases owned by this broker shutdown."""
+        if self.is_worker_process:
+            return (TaskiqEvents.WORKER_SHUTDOWN,)
+        return (TaskiqEvents.CLIENT_SHUTDOWN,)
+
+    @staticmethod
+    def _remember_shutdown_error(
+        first_error: BaseException | None,
+        current_error: BaseException,
+    ) -> BaseException:
+        """Keep the first shutdown failure while cleanup continues."""
+        if first_error is None:
+            return current_error
+        logger.error(
+            "Additional error while shutting down broker resources.",
+            exc_info=current_error,
+        )
+        return first_error
 
-        await self.result_backend.shutdown()
+    async def _kick_with_post_send(
+        self,
+        message: BrokerMessage,
+        post_send: Callable[[], Awaitable[None]],
+    ) -> None:
+        """Run the package-internal send boundary used by AsyncKicker."""
+        try:
+            await self.kick(message)
+        except Exception as exc:
+            raise SendTaskError from exc
+        await post_send()
 
     @abstractmethod
     async def kick(