fix(queue): make upload concurrency a refillable pool, not a per-batch cap (#2602)

check_queue awaited asyncio.gather() over the whole selected batch before
returning, so the scheduler run loop was blocked until the slowest FTP
upload in the batch finished. On a large farm a 513s upload left 15 of 16
configured upload slots idle for 8.5 minutes while other printers came
free — the setting behaved as a per-batch cap, not a worker pool.

Launch uploads as independent background tasks tracked in a _inflight pool.
Each tick excludes in-flight item rows and their printers from selection,
launches at most limit - len(_inflight) new uploads, and returns
immediately, so a freed slot refills on the next fast tick. The no-double-
dispatch invariant the batch-await provided (rows stay pending until upload
completes) is now carried by the in-flight exclusion; the pending->printing
CAS, busy-printer guard (#2598), per-printer hold, auto-drying exclusion,
and per-item failure isolation are all preserved per task.

Rewrites the concurrent-dispatch tests around pool/reservation/refill
semantics and adds coverage for slot refill, in-flight exclusion, and the
non-blocking return.

backend/tests/unit/test_scheduler_concurrent_dispatch.py

@@ -1,36 +1,46 @@
-"""Concurrent queue dispatch across printers (#2555).
+"""Concurrent queue dispatch as a refillable upload pool (#2555, #2602).
 
-Reported as "prints are sent to the printer one by one, very slowly" on a
-19-printer farm — up to an hour before the last printer started. Not a config
-problem: ``check_queue`` awaited ``_start_print`` inline for each pending item,
-and ``_start_print`` performs the FTP upload, so every printer queued behind
-every other printer's transfer. A Bambu printer's FTP server sustains ~150 KB/s
-(its own SD write is the bottleneck, not the network), so the reporter's 41 MB
-3MF took ~254 s *per printer* — 19 of those in series is ~80 minutes.
+Reported first (#2555) as "prints are sent to the printer one by one, very
+slowly" on a 19-printer farm: ``check_queue`` awaited ``_start_print`` inline per
+item, and ``_start_print`` performs the FTP upload, so every printer queued
+behind every other printer's transfer. #2555 moved the uploads to a parallel
+``asyncio.gather()`` — but that gather was *awaited before check_queue returned*,
+so the run loop stayed blocked until the slowest upload in the batch finished. On
+a 93-printer farm (#2602) a 513 s upload left 15 of 16 configured slots idle for
+8.5 minutes while other printers came free.
 
-Printers are independent machines, so the uploads have no reason to be
-serialized. They now run concurrently, capped by ``queue_max_concurrent_uploads``.
+The uploads now run as independent background tasks tracked in
+``scheduler._inflight``; each tick launches at most ``limit - len(_inflight)`` new
+ones and returns immediately, so a freed slot refills on the next fast tick.
 
 What must stay true:
 
-* Uploads to different printers overlap in time (the actual fix).
-* No more than ``queue_max_concurrent_uploads`` run at once (the host is not
-  infinite: each in-flight upload holds an FTP thread, a TLS session, a handle).
-* Setting it to 1 restores exactly the old serial behaviour.
-* One printer failing must not cancel its siblings' in-flight uploads.
-* A pass still never overlaps with the next one — ``_start_print`` flips the row
-  pending -> printing only *after* the upload completes, so returning early
-  while uploads were in flight would let the next pass re-dispatch the same rows.
+* Uploads to different printers overlap in time (the #2555 fix).
+* No more than ``queue_max_concurrent_uploads`` run at once — as a *pool*, across
+  ticks, not just within one batch (#2602).
+* A freed slot is refilled by a later tick (#2602).
+* An item whose upload is in flight — and its printer — are excluded from the
+  next pass, so a still-`pending` row is never dispatched twice (#2602).
+* check_queue returns *without* waiting for the uploads (#2602), reporting a
+  productive/in-flight pass so ``run()`` re-checks on the fast interval.
+* Setting the cap to 1 restores serial behaviour; one printer failing must not
+  cancel its siblings' in-flight uploads.
+
+Test model: the scheduler now launches uploads via ``spawn_background_task``, so
+the harness swaps in a real task-spawning shim and drains ``_inflight`` explicitly
+(inside the patched context, so the upload/session patches are still active while
+the pool workers run). ``_run_to_completion`` loops check_queue + drain to model
+the run loop draining a queue that exceeds the cap.
 """
 
 import asyncio
-from contextlib import ExitStack
+from contextlib import ExitStack, asynccontextmanager
 from pathlib import Path
 from types import SimpleNamespace
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
-from sqlalchemy import select
+from sqlalchemy import func, select
 from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
 
 import backend.app.models  # noqa: F401 - populate Base.metadata
@@ -117,7 +127,7 @@ class _UploadRecorder:
 
     Each call sleeps, so genuinely concurrent uploads have overlapping
     lifetimes. ``peak`` is the high-water mark of simultaneous in-flight
-    uploads — the number the whole fix turns on.
+    uploads — the number the pool cap turns on.
     """
 
     def __init__(self, *, fail_for_ip: str | None = None):
@@ -139,15 +149,25 @@ class _UploadRecorder:
             self.in_flight -= 1
 
 
-async def _run_check_queue(ctx, upload, job_started=None):
+@asynccontextmanager
+async def _scheduler_ctx(ctx, upload, job_started=None):
+    """Yield a scheduler with all I/O patched, and a real task-spawning shim.
+
+    The scheduler launches uploads through ``spawn_background_task`` (#2602), so
+    the harness gives it a real ``create_task`` shim rather than the no-op mock
+    used before — otherwise the pool workers never run and rows stay ``pending``.
+    The watchdog (also spawned per dispatch) is stubbed so it doesn't poll for
+    the whole test. Drain ``_inflight`` *inside* this context so the workers run
+    while the upload/session patches are still active.
+    """
     scheduler = PrintScheduler()
     job_started = job_started or AsyncMock()
 
+    def _real_spawn(coro, *, name=None):
+        return asyncio.create_task(coro, name=name)
+
     patches = [
         patch.object(scheduler_module.settings, "base_dir", ctx.base_dir),
-        # The library-file path archives the 3MF before uploading it, and the
-        # archive service resolves its own settings — redirect both or it writes
-        # into the real repo and then fails relative_to(base_dir).
         patch.object(archive_module.settings, "base_dir", ctx.base_dir),
         patch.object(archive_module.settings, "archive_dir", ctx.base_dir / "archive"),
         patch("backend.app.services.print_scheduler.async_session", ctx.session_maker),
@@ -163,7 +183,7 @@ async def _run_check_queue(ctx, upload, job_started=None):
             AsyncMock(return_value=(False, 0, 0, 1.0)),
         ),
         patch("backend.app.services.print_scheduler.cache_3mf_download", MagicMock()),
-        patch("backend.app.services.print_scheduler.spawn_background_task", MagicMock()),
+        patch("backend.app.services.print_scheduler.spawn_background_task", _real_spawn),
         patch("backend.app.services.notification_service.notification_service.on_queue_job_started", job_started),
         patch("backend.app.services.notification_service.notification_service.on_queue_job_failed", AsyncMock()),
         patch("backend.app.services.mqtt_relay.mqtt_relay.on_queue_job_started", AsyncMock()),
@@ -172,12 +192,51 @@ async def _run_check_queue(ctx, upload, job_started=None):
         patch.object(scheduler, "_power_off_if_needed", AsyncMock()),
         patch.object(scheduler, "_preheat_and_soak", AsyncMock()),
         patch.object(scheduler, "_check_auto_drying", AsyncMock()),
+        patch.object(scheduler, "_watchdog_print_start", AsyncMock()),
     ]
 
     with ExitStack() as stack:
         for patcher in patches:
             stack.enter_context(patcher)
-        return await scheduler.check_queue()
+        yield scheduler
+
+
+async def _drain(scheduler):
+    """Run the currently in-flight pool workers to completion."""
+    tasks = [task for (task, _pid) in scheduler._inflight.values()]
+    if tasks:
+        await asyncio.gather(*tasks, return_exceptions=True)
+
+
+async def _run_check_queue(ctx, upload, job_started=None, *, drain=True):
+    """Run one check_queue pass; by default also drain the launched uploads.
+
+    Returns the check_queue result (True if the pass was productive / has uploads
+    still in flight).
+    """
+    async with _scheduler_ctx(ctx, upload, job_started) as scheduler:
+        result = await scheduler.check_queue()
+        if drain:
+            await _drain(scheduler)
+        return result
+
+
+async def _run_to_completion(ctx, upload, job_started=None, *, max_ticks: int = 50) -> int:
+    """Model the run loop: check_queue + drain until the queue is empty.
+
+    Draining fully between ticks makes each tick a fresh batch of at most the cap,
+    which is enough to prove the cap holds across the whole drain and every item
+    eventually goes out. Returns the number of ticks it took.
+    """
+    ticks = 0
+    async with _scheduler_ctx(ctx, upload, job_started) as scheduler:
+        while ticks < max_ticks:
+            await scheduler.check_queue()
+            await _drain(scheduler)
+            ticks += 1
+            if await _pending_count(ctx) == 0 and not scheduler._inflight:
+                break
+    return ticks
 
 
 async def _statuses(ctx):
@@ -186,9 +245,16 @@ async def _statuses(ctx):
         return [r.status for r in rows]
 
 
+async def _pending_count(ctx) -> int:
+    async with ctx.session_maker() as db:
+        return await db.scalar(
+            select(func.count()).select_from(PrintQueueItem).where(PrintQueueItem.status == "pending")
+        )
+
+
 @pytest.mark.asyncio
 async def test_uploads_to_different_printers_overlap(farm):
-    """The headline fix: six printers must not queue behind each other.
+    """The #2555 headline: six printers must not queue behind each other.
 
     Pre-fix this recorded peak == 1 no matter how many printers were pending.
     """
@@ -205,29 +271,119 @@ async def test_uploads_to_different_printers_overlap(farm):
 
 
 @pytest.mark.asyncio
-async def test_concurrency_is_capped_by_the_setting(farm):
+async def test_pool_cap_holds_across_refills(farm):
     """Eight pending printers, cap of 3 — never more than 3 uploads at once.
 
-    The cap is the reason this is a setting and not just ``asyncio.gather``:
-    the printers are independent but the Bambuddy host is not.
+    Under the pool model (#2602) one tick launches at most 3; the queue drains
+    over several ticks. The cap must hold across the *whole* drain, and every
+    item must still go out.
     """
     ctx = await farm(8, max_concurrent=3)
     upload = _UploadRecorder()
 
-    await _run_check_queue(ctx, upload)
+    ticks = await _run_to_completion(ctx, upload)
 
-    assert upload.peak == 3, f"cap of 3 not honoured — peak was {upload.peak}"
+    assert upload.peak == 3, f"cap of 3 not honoured across the drain — peak was {upload.peak}"
     assert len(upload.order) == 8, "every pending item must still be dispatched, just not all at once"
     assert await _statuses(ctx) == ["printing"] * 8
+    assert ticks >= 3, "8 items at a cap of 3 must take at least 3 ticks to drain"
+
+
+@pytest.mark.asyncio
+async def test_freed_slot_is_refilled_on_the_next_tick(farm):
+    """The #2602 fix: a busy pool doesn't block, and a freed slot refills.
+
+    Cap of 1, two printers. Tick 1 launches printer A. A second tick while A is
+    still in flight must launch nothing (pool full) rather than block. Once A
+    finishes, the next tick fills the freed slot with printer B.
+    """
+    ctx = await farm(2, max_concurrent=1)
+    upload = _UploadRecorder()
+
+    async with _scheduler_ctx(ctx, upload) as scheduler:
+        # Tick 1: one slot, one launch. Don't drain — A is now "in flight".
+        assert await scheduler.check_queue() is True
+        assert len(scheduler._inflight) == 1
+
+        # Tick 2 while A is in flight: pool full → no new launch, no blocking.
+        assert await scheduler.check_queue() is True
+        assert len(scheduler._inflight) == 1, "a full pool must not launch a second upload"
+
+        # A completes, freeing the slot.
+        await _drain(scheduler)
+        assert not scheduler._inflight
+
+        # Tick 3: the freed slot is refilled with the second printer.
+        assert await scheduler.check_queue() is True
+        assert len(scheduler._inflight) == 1
+        await _drain(scheduler)
+
+    assert await _statuses(ctx) == ["printing", "printing"]
+    assert upload.peak == 1, "cap of 1 must never overlap two uploads"
+
+
+@pytest.mark.asyncio
+async def test_inflight_item_and_printer_are_excluded_from_reselection(farm):
+    """A still-`pending` in-flight row must not be dispatched a second time (#2602).
+
+    The row flips pending -> printing only after its upload completes, so the
+    reservation that stops a fast tick re-dispatching it is the in-flight
+    exclusion, not the DB status.
+    """
+    ctx = await farm(1, max_concurrent=4)
+    upload = _UploadRecorder()
+
+    async with _scheduler_ctx(ctx, upload) as scheduler:
+        await scheduler.check_queue()
+        inflight_before = set(scheduler._inflight)
+        assert len(inflight_before) == 1
+
+        # Second tick while the upload is in flight (row still pending): the item
+        # and its printer must be excluded — no new task, pool unchanged.
+        await scheduler.check_queue()
+        assert set(scheduler._inflight) == inflight_before, "an in-flight item was re-selected"
+
+        await _drain(scheduler)
+
+    assert await _statuses(ctx) == ["printing"]
+    assert len(upload.order) == 1, "the item must be uploaded exactly once, not twice"
+
+
+@pytest.mark.asyncio
+async def test_inflight_printer_is_kept_out_of_auto_drying(farm):
+    """A printer with an upload in flight must not be auto-dried in the gap (#2602).
+
+    Once check_queue returns while the upload runs, the only pending row is the
+    in-flight one — so the pass takes the "no dispatchable items" path. That path
+    must still exclude the in-flight printer from auto-drying, because its print
+    is imminent (the row flips to printing the moment the upload finishes).
+    """
+    ctx = await farm(1, max_concurrent=4)
+    printer_id = ctx.printer_ids[0]
+    upload = _UploadRecorder()
+
+    async with _scheduler_ctx(ctx, upload) as scheduler:
+        await scheduler.check_queue()  # launch the only item; now in flight
+        scheduler._check_auto_drying.reset_mock()
+
+        # Second tick: the sole pending row is in flight, so this hits the
+        # empty-items path. It must report the in-flight printer as busy.
+        result = await scheduler.check_queue()
+        assert result is True, "in-flight uploads keep the loop on the fast interval"
+        assert scheduler._check_auto_drying.await_count == 1
+        busy_arg = scheduler._check_auto_drying.await_args.args[2]
+        assert printer_id in busy_arg, "the in-flight printer must be excluded from auto-drying"
+
+        await _drain(scheduler)
 
 
 @pytest.mark.asyncio
 async def test_limit_of_one_restores_serial_behaviour(farm):
-    """An escape hatch for weak networks: 1 == the pre-#2555 behaviour."""
+    """An escape hatch for weak networks: 1 == one upload at a time."""
     ctx = await farm(4, max_concurrent=1)
     upload = _UploadRecorder()
 
-    await _run_check_queue(ctx, upload)
+    await _run_to_completion(ctx, upload)
 
     assert upload.peak == 1
     assert await _statuses(ctx) == ["printing"] * 4
@@ -237,13 +393,12 @@ async def test_limit_of_one_restores_serial_behaviour(farm):
 async def test_default_concurrency_applies_when_setting_absent(farm):
     """No Settings row (every existing install) must still dispatch in parallel.
 
-    The whole point is that the reporter's farm gets faster *without* him having
-    to find a new setting first. Default is 4.
+    Default cap is 4.
     """
     ctx = await farm(5, max_concurrent=None)
     upload = _UploadRecorder()
 
-    await _run_check_queue(ctx, upload)
+    await _run_to_completion(ctx, upload)
 
     assert upload.peak == 4, f"expected the default cap of 4, got {upload.peak}"
     assert await _statuses(ctx) == ["printing"] * 5
@@ -253,9 +408,8 @@ async def test_default_concurrency_applies_when_setting_absent(farm):
 async def test_one_failing_upload_does_not_cancel_the_others(farm):
     """A dead printer must not take its siblings' in-flight uploads down with it.
 
-    ``asyncio.gather`` without ``return_exceptions=True`` cancels every sibling
-    task the moment one raises — which would mean a single unreachable printer
-    silently aborts the whole batch mid-transfer.
+    Each upload is an independent task, so one raising cannot cancel the others;
+    _start_print marks that one item failed and the rest proceed.
     """
     ctx = await farm(4, max_concurrent=4)
     upload = _UploadRecorder(fail_for_ip="10.0.0.2")  # printer index 1
@@ -271,11 +425,7 @@ async def test_one_failing_upload_does_not_cancel_the_others(farm):
 
 @pytest.mark.asyncio
 async def test_check_queue_reports_it_dispatched(farm):
-    """A productive pass returns True so ``run()`` re-checks quickly (#2555).
-
-    The fast re-tick is what stops a draining batch from stalling 30 s behind
-    the idle sleep every time a printer frees up.
-    """
+    """A productive pass returns True so ``run()`` re-checks quickly (#2555)."""
     ctx = await farm(3, max_concurrent=3)
 
     dispatched = await _run_check_queue(ctx, _UploadRecorder())
@@ -294,38 +444,41 @@ async def test_check_queue_reports_nothing_dispatched_when_empty(farm):
 
 
 @pytest.mark.asyncio
-async def test_check_queue_awaits_its_dispatches_before_returning(farm):
-    """The pass must not return while uploads are still in flight.
+async def test_check_queue_returns_without_awaiting_the_uploads(farm):
+    """The pass must return *before* the uploads finish (#2602).
 
-    ``_start_print`` flips the row pending -> printing only *after* the upload
-    finishes. If ``check_queue`` returned early, the next 30-second tick would
-    still see those rows as ``pending`` on an idle-looking printer and dispatch
-    them a second time.
+    This is the inversion of the old contract: check_queue no longer blocks on
+    the batch. It launches the uploads as tracked background tasks, leaves the
+    rows ``pending`` (they flip to ``printing`` only when each upload completes),
+    and returns True so the run loop keeps ticking fast while they drain.
     """
     ctx = await farm(3, max_concurrent=3)
     upload = _UploadRecorder()
 
-    await _run_check_queue(ctx, upload)
+    async with _scheduler_ctx(ctx, upload) as scheduler:
+        result = await scheduler.check_queue()
+
+        # Uploads are tracked but have not been awaited: rows are still pending.
+        assert result is True
+        assert len(scheduler._inflight) == 3
+        assert await _statuses(ctx) == ["pending"] * 3
+
+        await _drain(scheduler)
 
-    assert upload.in_flight == 0, "check_queue returned with uploads still running"
     assert await _statuses(ctx) == ["printing"] * 3
+    assert upload.peak == 3
 
 
 class TestSharedLibraryRow:
-    """Dispatching in parallel means two items can now reach the same library row
-    at the same time — impossible when dispatch was serial.
+    """Dispatching in parallel means two items can reach the same library row at
+    the same time — impossible when dispatch was serial.
 
     Only the ``cleanup_library_after_dispatch`` flow (printer-card "upload and
-    print") *mutates* that row: it deletes it and unlinks the 3MF from disk once
-    the print is away. Two of those against one row would race — the loser's
-    DELETE matches no row, and the winner's unlink can pull the file out from
-    under the loser's in-flight upload.
-
-    An ordinary library print only reads the row. That distinction is load-bearing:
-    the reporter's own batch was one File Manager file fanned out across his farm
-    (both of the queue items in his log point at library file 116), so a blanket
-    "never share a library row" guard would re-serialize the exact workload this
-    change exists to fix.
+    print") *mutates* that row: it deletes it and unlinks the 3MF once the print
+    is away. Two of those against one row would race. An ordinary library print
+    only reads the row, and the reporter's own batch was one File Manager file
+    fanned out across his farm, so a blanket "never share a library row" guard
+    would re-serialize the exact workload this exists to fix.
     """
 
     @staticmethod
@@ -374,8 +527,7 @@ class TestSharedLibraryRow:
     async def test_plain_library_file_still_fans_out_in_parallel(self, tmp_path):
         """The reporter's actual workload: one File Manager file, four printers.
 
-        Nothing here mutates the library row, so all four must upload at once. If
-        this ever drops to 1 the headline fix is gone.
+        Nothing here mutates the library row, so all four must upload at once.
         """
         engine = create_async_engine("sqlite+aiosqlite:///:memory:", echo=False)
         async with engine.begin() as conn:
@@ -396,7 +548,7 @@ class TestSharedLibraryRow:
     async def test_cleanup_items_never_share_a_row_in_one_pass(self, tmp_path):
         """The mutating flow must be held to one dispatch per pass.
 
-        Each of these deletes the library row and unlinks the 3MF when it is done.
+        Each of these deletes the library row and unlinks the 3MF when done.
         Exactly one may go per pass; the rest stay pending for a later one.
         """
         engine = create_async_engine("sqlite+aiosqlite:///:memory:", echo=False)
@@ -427,12 +579,9 @@ async def test_library_print_without_a_parseable_print_time_does_not_crash(tmp_p
 
     It only fired when the archive carried no print time — a plain .gcode, or a 3MF
     the parser could not read — and it fired *after* the printer had been sent the
-    job. The started-notification was lost, and the AttributeError unwound the whole
-    queue pass, so every other printer still waiting to be dispatched on that tick
-    silently missed its turn. Exactly the "why did only some of them start" shape.
-
-    Two printers here: if the first one's dispatch blows up, the second must still
-    go out.
+    job. The started-notification was lost and the AttributeError unwound the
+    dispatch. Two printers here: if the first one's dispatch blows up, the second
+    must still go out.
     """
     engine = create_async_engine("sqlite+aiosqlite:///:memory:", echo=False)
     async with engine.begin() as conn:
@@ -480,10 +629,6 @@ async def test_library_print_without_a_parseable_print_time_does_not_crash(tmp_p
         await _run_check_queue(ctx, _UploadRecorder(), job_started=job_started)
 
         assert await _statuses(ctx) == ["printing", "printing"]
-
-        # The status flip happens BEFORE the crash point, so it is not the signal —
-        # both rows read "printing" even with the bug present. The started-notification
-        # is emitted just after it, and is what the AttributeError actually destroyed.
         assert job_started.await_count == 2, (
             "the job-started notification was lost — _start_print raised after the "
             "printer had already been sent the job"