Add ability to limit the consumption of async tasks based on a maximu…

…m number. (#79)

librarian_background/queues.py

@@ -13,7 +13,7 @@
 from pathlib import Path
 from typing import TYPE_CHECKING, Callable
 
-from sqlalchemy import select
+from sqlalchemy import func, select
 
 from hera_librarian.exceptions import LibrarianError
 from hera_librarian.transfer import TransferStatus
@@ -219,6 +219,18 @@ def consume_queue_item(session_maker: Callable[[], "Session"]) -> bool:
             # Nothing to do!
             return False
 
+        # Now, check we don't have too much going on.
+        stmt = (
+            select(func.count(SendQueue.id))
+            .filter_by(consumed=True)
+            .filter_by(completed=False)
+        )
+        in_flight = session.execute(stmt).scalar()
+
+        if in_flight > server_settings.max_async_inflight_transfers:
+            # Too much to do!
+            return False
+
         # Otherwise, we are free to consume this item.
         transfer_list = [
             (Path(x.source_path), Path(x.dest_path)) for x in queue_item.transfers

librarian_server/settings.py

@@ -102,6 +102,9 @@ class ServerSettings(BaseSettings):
 
     # Transfer constraints
     max_async_send_retries: int = 3
+    # The maximum number of in-flight asynchronous transfers to
+    # a specific destination.
+    max_async_inflight_transfers: int = 64
 
     # Slack integration; by default disable this. You will need a slack
     # webhook url, and by default we raise all log_to_database alerts to slack too.