diff --git a/taskiq/receiver/receiver.py b/taskiq/receiver/receiver.py index 871d13d6..d68e6c78 100644 --- a/taskiq/receiver/receiver.py +++ b/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...", diff --git a/tests/receiver/test_receiver.py b/tests/receiver/test_receiver.py index d724b326..ad920855 100644 --- a/tests/receiver/test_receiver.py +++ b/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()