fix(queue): re-check the queue quickly after a dispatch instead of always waiting 30s (issue #2555)

The scheduler slept a fixed 30s after every pass, so each printer that
freed up during a batch waited up to a full interval before its next job
was dispatched — on a farm, that idle gap stacked into the "several long
minutes" reporters saw between requesting prints and them starting (#2555).

check_queue() now reports whether it dispatched anything; run() loops again
after 3s on a productive pass and falls back to 30s otherwise. Fast ticks
only continue while the queue is actively draining, so this can't tight-loop:
a pass that dispatches nothing (all pending items behind busy printers, or a
wedged head-of-line job holding its printer) reverts to the normal interval.

backend/app/services/print_scheduler.py

@@ -217,6 +217,17 @@ class PrintScheduler:
     def __init__(self):
         self._running = False
         self._check_interval = 30  # seconds
+        # After a pass that actually dispatched something, loop again almost
+        # immediately instead of sleeping the full interval (#2555). A dispatch
+        # changes printer state — a batch launch fans out over several passes as
+        # printers free up, a wedged head-of-line job reverts to pending, an
+        # upload slot opens — and the next batch of ready work should not have to
+        # wait 30 s behind an idle sleep. When a pass dispatches nothing (all
+        # pending items are behind printers that are genuinely busy printing),
+        # there is nothing to react to, so we fall back to the normal interval;
+        # that also means this can never tight-loop, since fast ticks only
+        # continue while dispatches keep happening and the queue is draining.
+        self._fast_check_interval = 3  # seconds
         self._power_on_wait_time = 180  # seconds to wait for printer after power on (3 min)
         self._power_on_check_interval = 10  # seconds between connection checks
         # Track which printers are currently auto-drying (printer_id -> start timestamp)
@@ -246,20 +257,27 @@ class PrintScheduler:
         logger.info("Print scheduler started")
 
         while self._running:
+            dispatched = False
             try:
-                await self.check_queue()
+                dispatched = await self.check_queue()
             except Exception as e:
                 logger.error("Scheduler error: %s", e)
 
-            await asyncio.sleep(self._check_interval)
+            # Re-check quickly after a productive pass so a draining batch does
+            # not stall behind the idle interval; otherwise sleep normally (#2555).
+            await asyncio.sleep(self._fast_check_interval if dispatched else self._check_interval)
 
     def stop(self):
         """Stop the scheduler."""
         self._running = False
         logger.info("Print scheduler stopped")
 
-    async def check_queue(self):
-        """Check for prints ready to start."""
+    async def check_queue(self) -> bool:
+        """Check for prints ready to start.
+
+        Returns True if this pass dispatched at least one item, so the caller
+        can loop again quickly instead of sleeping the full interval (#2555).
+        """
         async with async_session() as db:
             # Check if shortest-job-first scheduling is enabled
             sjf_enabled = await self._get_bool_setting(db, "queue_shortest_first")
@@ -300,7 +318,7 @@ class PrintScheduler:
             if not items:
                 # No pending items — still check auto-drying on idle printers
                 await self._check_auto_drying(db, [], set(), require_plate_clear=require_plate_clear)
-                return
+                return False
 
             logger.info(
                 "Queue check: found %d pending items: %s",
@@ -710,6 +728,8 @@ class PrintScheduler:
             # Auto-drying: start drying on idle printers that have no pending queue items
             await self._check_auto_drying(db, items, busy_printers, require_plate_clear=require_plate_clear)
 
+            return bool(dispatch_ids)
+
     async def _dispatch_selected(self, item_ids: list[int], limit: int) -> None:
         """Upload and start every item selected by this queue pass, in parallel.
 

backend/tests/unit/test_scheduler_concurrent_dispatch.py

@@ -177,7 +177,7 @@ async def _run_check_queue(ctx, upload, job_started=None):
     with ExitStack() as stack:
         for patcher in patches:
             stack.enter_context(patcher)
-        await scheduler.check_queue()
+        return await scheduler.check_queue()
 
 
 async def _statuses(ctx):
@@ -269,6 +269,30 @@ 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.
+    """
+    ctx = await farm(3, max_concurrent=3)
+
+    dispatched = await _run_check_queue(ctx, _UploadRecorder())
+
+    assert dispatched is True, "check_queue dispatched 3 items but did not report it"
+
+
+@pytest.mark.asyncio
+async def test_check_queue_reports_nothing_dispatched_when_empty(farm):
+    """An empty queue returns False so ``run()`` falls back to the idle interval."""
+    ctx = await farm(0, max_concurrent=3)
+
+    dispatched = await _run_check_queue(ctx, _UploadRecorder())
+
+    assert dispatched is False, "an empty pass must not trigger a fast re-tick"
+
+
 @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.