fix(jobs): ack MqWatcher messages so prefetched buffer is not logged as lost on shutdown

MqWatcher.consumes signals from RabbitMQ purely to wake up a DB-driven
processor; data lives in pending_txs / pending_messages, not in the
message body. Using no_ack=True meant aio_pika logged a warning per
prefetched message still in the iterator buffer at shutdown
("Message ... lost for consumer with no_ack ..."), which was alarming
on Ctrl-C even though nothing was actually lost.

Switch to no_ack=False with manual ack. Acked messages are gone; any
prefetched-but-not-yet-iterated messages are nack'd with requeue=True
by aio_pika's iterator close path, so they reappear on next startup
and trigger the watcher.

Author: odesenfans

src/aleph/jobs/job_utils.py

@@ -157,8 +157,12 @@ def __init__(self, mq_queue: aio_pika.abc.AbstractQueue):
         self._event = asyncio.Event()
 
     async def _check_for_message(self):
-        async with self.mq_queue.iterator(no_ack=True) as queue_iter:
-            async for _ in queue_iter:
+        # Manual ack (no_ack=False) so that prefetched messages still in the
+        # iterator buffer on shutdown are nack'd with requeue=True by aio_pika,
+        # instead of being logged as "lost for consumer with no_ack".
+        async with self.mq_queue.iterator(no_ack=False) as queue_iter:
+            async for message in queue_iter:
+                await message.ack()
                 self._event.set()
 
     async def __aenter__(self):