move span closuse to post_save,

middleware post_send, post_execute, post_save, on_error are now called in reverse order

Author: soapun

taskiq/kicker.py

@@ -168,7 +168,7 @@ async def kiq(
         except Exception as exc:
             raise SendTaskError from exc
 
-        for middleware in self.broker.middlewares:
+        for middleware in reversed(self.broker.middlewares):
             if middleware.__class__.post_send != TaskiqMiddleware.post_send:
                 await maybe_awaitable(middleware.post_send(message))
 

taskiq/middlewares/opentelemetry_middleware.py

@@ -222,7 +222,7 @@ def pre_execute(self, message: TaskiqMessage) -> TaskiqMessage:
         attach_context(message, span, activation, token)
         return message
 
-    def post_execute(  # pylint: disable=R6301
+    def post_save(  # pylint: disable=R6301
         self,
         message: TaskiqMessage,
         result: TaskiqResult[T],

taskiq/receiver/receiver.py

@@ -139,15 +139,15 @@ async def callback(  # noqa: C901, PLR0912
         ):
             await maybe_awaitable(message.ack())
 
-        for middleware in self.broker.middlewares:
+        for middleware in reversed(self.broker.middlewares):
             if middleware.__class__.post_execute != TaskiqMiddleware.post_execute:
                 await maybe_awaitable(middleware.post_execute(taskiq_msg, result))
 
         try:
             if not isinstance(result.error, NoResultError):
                 await self.broker.result_backend.set_result(taskiq_msg.task_id, result)
 
-                for middleware in self.broker.middlewares:
+                for middleware in reversed(self.broker.middlewares):
                     if middleware.__class__.post_save != TaskiqMiddleware.post_save:
                         await maybe_awaitable(middleware.post_save(taskiq_msg, result))
 
@@ -289,7 +289,7 @@ async def run_task(  # noqa: C901, PLR0912, PLR0915
         )
         # If exception is found we execute middlewares.
         if found_exception is not None:
-            for middleware in self.broker.middlewares:
+            for middleware in reversed(self.broker.middlewares):
                 if middleware.__class__.on_error != TaskiqMiddleware.on_error:
                     await maybe_awaitable(
                         middleware.on_error(