Fixed retry middleware. Added logs and parent labels.

Signed-off-by: Pavel Kirilin <[email protected]>

Author: s3rius

taskiq/middlewares/retry_middleware.py

@@ -41,13 +41,21 @@ async def on_error(
         if retry_on_error != "True":
             return
 
-        logger.info(f"Task '{message.task_name}' invocation failed. Retrying.")
-
         # Getting number of previous retries.
         retries = int(message.labels.get("_retries", 0)) + 1
         message.labels["_retries"] = str(retries)
         max_retries = int(message.labels.get("max_retries", self.default_retry_count))
         if retries < max_retries:
+            logger.info(
+                "Task '%s' invocation failed. Retrying.",
+                message.task_name,
+            )
+            message.labels["_parent"] = message.task_id
             message.task_id = self.broker.id_generator()
             broker_message = self.broker.formatter.dumps(message=message)
             await self.broker.kick(broker_message)
+        else:
+            logger.warning(
+                "Task '%s' invocation failed. Maximum retries count is reached.",
+                message.task_name,
+            )