fix(receiver): ack messages that cannot be parsed or resolved

When a message fails formatter parsing or names a task the broker cannot
find, the receiver logged a warning and returned without acknowledging.
For ackable brokers (e.g. RabbitMQ) the message stayed unacked and
pending forever: it was already prefetched out of the queue, so purging
the queue did not clear it and it only reappeared when the consumer was
terminated.

Both early-return paths now ack the message when it is ackable, since a
message that cannot be parsed or whose task does not exist will never
become processable by a redelivery.

Adds regression tests for both paths.

Closes #569

Author: Sanjays2402

taskiq/receiver/receiver.py

@@ -130,6 +130,8 @@ async def callback(  # noqa: C901, PLR0912
                 exc,
                 exc_info=True,
             )
+            if ack_controller.is_ackable:
+                await ack_controller.ack()
             return
         logger.debug(f"Received message: {taskiq_msg}")
         task = self.broker.find_task(taskiq_msg.task_name)
@@ -138,6 +140,8 @@ async def callback(  # noqa: C901, PLR0912
                 'task "%s" is not found. Maybe you forgot to import it?',
                 taskiq_msg.task_name,
             )
+            if ack_controller.is_ackable:
+                await ack_controller.ack()
             return
         logger.debug(
             "Function for task %s is resolved. Executing...",

tests/receiver/test_receiver.py

@@ -334,6 +334,56 @@ def ack_callback() -> None:
     assert acked
 
 
+async def test_callback_acks_unknown_task() -> None:
+    """Test that a message for an unknown task is acked, not left pending."""
+    broker = InMemoryBroker()
+    acked = False
+
+    def ack_callback() -> None:
+        nonlocal acked
+        acked = True
+
+    receiver = get_receiver(broker)
+
+    broker_message = broker.formatter.dumps(
+        TaskiqMessage(
+            task_id="task_id",
+            task_name="unknown_task_name",
+            labels={},
+            args=[],
+            kwargs={},
+        ),
+    )
+
+    await receiver.callback(
+        AckableMessage(
+            data=broker_message.message,
+            ack=ack_callback,
+        ),
+    )
+    assert acked
+
+
+async def test_callback_acks_unparsable_message() -> None:
+    """Test that an unparsable message is acked, not left pending."""
+    broker = InMemoryBroker()
+    acked = False
+
+    def ack_callback() -> None:
+        nonlocal acked
+        acked = True
+
+    receiver = get_receiver(broker)
+
+    await receiver.callback(
+        AckableMessage(
+            data=b"not a valid taskiq message",
+            ack=ack_callback,
+        ),
+    )
+    assert acked
+
+
 async def test_callback_success_ackable_async() -> None:
     """Test that acks work with async functions."""
     broker = InMemoryBroker()