Improve handling of fail queues (#6781)

scripts/reindex.py

@@ -143,6 +143,7 @@ def main(argv: list[str]):
     reject(args.deindex and (args.delete or args.create),
            '--deindex is incompatible with --create and --delete.')
 
+    azul.check_fail_queues_state()
     deindex = args.deindex or (args.delete and not every_source)
     delete = args.delete and every_source
 
@@ -184,6 +185,7 @@ def main(argv: list[str]):
                             args.prefix, args.catalogs)
             else:
                 azul.wait_for_indexer()
+                azul.check_fail_queues_state()
 
 
 if __name__ == '__main__':

src/azul/azulclient.py

@@ -600,6 +600,16 @@ def message(source: SourceRef):
         messages = map(message, sources)
         self.queue_mirror_messages(messages)
 
+    def check_fail_queues_state(self):
+        queue_states = set()
+        for fail_queue in config.fail_queue_names:
+            queue_state = self.is_queue_empty(fail_queue)
+            queue_states.add(queue_state)
+            log.info('Queue %r is empty: %r', fail_queue, queue_state)
+        assert len(queue_states) > 0, R('No fail queues found')
+        assert all(queue_states), R(
+            'Non empty fail queues; run `manage_queue.py dump --delete` to empty them.')
+
 
 class AzulClientError(RuntimeError):
     pass

test/integration_test.py

@@ -429,6 +429,7 @@ class Catalog:
 
         def _wait_for_indexer():
             self.azul_client.wait_for_indexer()
+            self.azul_client.check_fail_queues_state()
 
         flags = config.it_flags
         index, delete = ['no_' + flag not in flags for flag in ['index', 'delete']]
@@ -438,6 +439,7 @@ def _wait_for_indexer():
         else:
             log.warning('Will skip indexing due to overriding IT flag.')
 
+        self.azul_client.check_fail_queues_state()
         catalogs: list[Catalog] = []
         for catalog in config.integration_test_catalogs:
             if index:
@@ -495,6 +497,7 @@ def _wait_for_indexer():
             self._test_mirroring()
 
     def _reset_indexer(self):
+        self.azul_client.check_fail_queues_state()
         # While it's OK to erase the integration test catalog, the queues are
         # shared by all catalogs and we can't afford to trash them in a stable
         # deployment like production.