fix(tests): assemble the upload batch instead of racing a sleep

The concurrent-dispatch tests went red on the Docker shard with
"expected all 6 printers to be uploaded to concurrently, but the
high-water mark was 4", on a scheduler that was dispatching all six
correctly.

peak == 6 is the claim that the sixth dispatch reaches its upload
before the first one finishes, and the dispatches do not arrive
together: each runs a preamble of database work first. So the
assertion was a race between that spread and a fixed 0.15 s sleep.
Measured here the spread is ~14 ms; on the runner it passed 150 ms.
Padding the sleep would only move the threshold and slow every test
that uses it.

_UploadRecorder(assemble=True) now holds each call until every
upload the pass launched has arrived, reading len(_inflight), which
is filled synchronously at launch and is therefore the batch size.
That states the property the peak assertions are about, directly and
with no time in it, and it ends sooner than the sleep it replaces -
the file runs 7.0s to 5.9s. _BATCH_DEADLINE_SECONDS bounds it so a
dispatch that really has gone serial fails on its peak assertion
instead of hanging to the pytest timeout.

test_check_queue_returns_without_awaiting_the_uploads keeps the
sleep, with the reason recorded next to it: it reads the rows while
the uploads are open, and an assembled batch releases as soon as it
is complete, which would let the dispatches flip those rows
mid-assertion.

The test engine also takes the application's own _set_sqlite_pragmas
listener rather than a copy, so the file is opened the way the
running system opens one - WAL, synchronous = NORMAL, a 15 s busy
timeout. SQLite's defaults spend an fsync per commit and lock the
whole file, which made a dispatch's preamble cost more than the
upload it precedes. That is what turned the previous commit's move
to a file-backed database into a visible failure.

Verified in both directions rather than by a green run: with two
printers' preambles delayed by a second, the old recorder reports
exactly the "high-water mark was 4" the runner saw and the shared
library test reports 3 == 4, while the assembling recorder passes
the same scenario. Without the induced skew, ruff is clean, the full
backend suite is green at 12224 passed, and ten consecutive runs of
the file pinned to one core show no failures.

backend/tests/unit/test_scheduler_concurrent_dispatch.py

@@ -40,13 +40,13 @@ from types import SimpleNamespace
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
-from sqlalchemy import func, select
+from sqlalchemy import event, func, select
 from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
 
 import backend.app.models  # noqa: F401 - populate Base.metadata
 import backend.app.services.archive as archive_module
 import backend.app.services.print_scheduler as scheduler_module
-from backend.app.core.database import Base
+from backend.app.core.database import Base, _set_sqlite_pragmas
 from backend.app.models.archive import PrintArchive
 from backend.app.models.library import LibraryFile
 from backend.app.models.print_queue import PrintQueueItem
@@ -56,6 +56,10 @@ from backend.app.services.print_scheduler import PrintScheduler
 
 UPLOAD_SECONDS = 0.15
 
+# Hard cap on assembling one batch, so a dispatch that has gone serial fails on
+# its peak assertion instead of hanging. See ``_UploadRecorder``.
+_BATCH_DEADLINE_SECONDS = 5.0
+
 
 def _test_engine(tmp_path):
     """An engine that gives every session its own connection.
@@ -76,8 +80,17 @@ def _test_engine(tmp_path):
     A file gets ``AsyncAdaptedQueuePool`` and a connection per session, which is
     what the app runs with in production (``_resolve_pool_kwargs`` in
     ``backend/app/core/database.py``: pool_size 20, max_overflow 200).
+
+    It also takes the app's own connect listener rather than a copy of it, so the
+    file is opened exactly as the running system opens one: WAL instead of a
+    whole-file DELETE journal, ``synchronous = NORMAL`` instead of an fsync per
+    commit, and a 15 s busy timeout. Default SQLite settings would make a
+    dispatch's preamble cost more than the upload it precedes, which is the
+    difference between six uploads overlapping and four.
     """
-    return create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'queue.db'}", echo=False)
+    engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'queue.db'}", echo=False)
+    event.listen(engine.sync_engine, "connect", _set_sqlite_pragmas)
+    return engine
 
 
 @pytest.fixture
@@ -148,23 +161,70 @@ async def farm(tmp_path):
 class _UploadRecorder:
     """Stands in for ``upload_file_async``; records overlap.
 
-    Each call sleeps, so genuinely concurrent uploads have overlapping
-    lifetimes. ``peak`` is the high-water mark of simultaneous in-flight
-    uploads — the number the pool cap turns on.
+    Each call stays open for a while, so genuinely concurrent uploads have
+    overlapping lifetimes. ``peak`` is the high-water mark of simultaneous
+    in-flight uploads — the number the pool cap turns on.
+
+    How long a call stays open is the whole difficulty. ``peak == 6`` is really
+    the claim that the sixth dispatch reaches its upload before the first one
+    finishes, and the dispatches do not arrive together: each runs a preamble of
+    database work first. Measured here that spread is ~14 ms; on a CI runner it
+    passed 150 ms, and a fixed 0.15 s sleep then recorded a peak of 4 out of 6
+    for a scheduler that was dispatching all six correctly. Padding the sleep
+    only moves the threshold and slows every test that uses it.
+
+    With ``assemble=True`` the call instead holds until every upload the pass
+    launched has arrived — ``scheduler._inflight`` is populated synchronously at
+    launch (see ``_launch_uploads``), so its length is the batch size and
+    ``in_flight`` catching up to it means the batch is assembled. That is the
+    property the peak assertions are about, stated directly and with no time in
+    it, so machine speed cannot change the answer. It also ends sooner than the
+    sleep it replaces. A batch that never assembles is the failure being looked
+    for, and ``_BATCH_DEADLINE_SECONDS`` bounds it: a serialized dispatch fails
+    on the peak assertion rather than hanging until the pytest timeout.
+
+    Without ``assemble`` — for the tests whose assertion is an upper bound, where
+    nothing has to assemble — the call just sleeps.
     """
 
-    def __init__(self, *, fail_for_ip: str | None = None):
+    def __init__(self, *, fail_for_ip: str | None = None, assemble: bool = False):
         self.in_flight = 0
         self.peak = 0
         self.order: list[str] = []
         self.fail_for_ip = fail_for_ip
+        self.assemble = assemble
+        self.scheduler = None  # bound by _scheduler_ctx when assemble is on
+        self._deadline = 0.0
+        self._assembled = False
+
+    async def _await_batch(self):
+        """Hold until every upload this pass launched has reached this point.
+
+        ``_assembled`` latches, and has to: the moment the batch is complete its
+        members start leaving and ``_inflight`` starts emptying, so a member
+        still comparing the two counts would find them equal for the wrong
+        reason. Latching releases the batch as one. It is cleared by the next
+        batch's first arrival, since ``in_flight`` returns to 0 between ticks.
+        """
+        loop = asyncio.get_running_loop()
+        if self.in_flight == 1:
+            self._deadline = loop.time() + _BATCH_DEADLINE_SECONDS
+            self._assembled = False
+        while not self._assembled:
+            if self.in_flight >= len(self.scheduler._inflight) or loop.time() >= self._deadline:
+                self._assembled = True
+                return
+            await asyncio.sleep(0.005)
 
     async def __call__(self, ip_address, access_code, local_path, remote_path, **kwargs):
         self.in_flight += 1
         self.peak = max(self.peak, self.in_flight)
         self.order.append(ip_address)
         try:
-            await asyncio.sleep(UPLOAD_SECONDS)
+            if self.assemble:
+                await self._await_batch()
+            else:
+                await asyncio.sleep(UPLOAD_SECONDS)
             if self.fail_for_ip is not None and ip_address == self.fail_for_ip:
                 raise OSError(f"simulated FTP failure for {ip_address}")
             return True
@@ -184,6 +244,7 @@ async def _scheduler_ctx(ctx, upload, job_started=None):
     while the upload/session patches are still active.
     """
     scheduler = PrintScheduler()
+    upload.scheduler = scheduler
     job_started = job_started or AsyncMock()
 
     def _real_spawn(coro, *, name=None):
@@ -282,7 +343,7 @@ async def test_uploads_to_different_printers_overlap(farm):
     Pre-fix this recorded peak == 1 no matter how many printers were pending.
     """
     ctx = await farm(6, max_concurrent=6)
-    upload = _UploadRecorder()
+    upload = _UploadRecorder(assemble=True)
 
     await _run_check_queue(ctx, upload)
 
@@ -302,7 +363,7 @@ async def test_pool_cap_holds_across_refills(farm):
     item must still go out.
     """
     ctx = await farm(8, max_concurrent=3)
-    upload = _UploadRecorder()
+    upload = _UploadRecorder(assemble=True)
 
     ticks = await _run_to_completion(ctx, upload)
 
@@ -419,7 +480,7 @@ async def test_default_concurrency_applies_when_setting_absent(farm):
     Default cap is 4.
     """
     ctx = await farm(5, max_concurrent=None)
-    upload = _UploadRecorder()
+    upload = _UploadRecorder(assemble=True)
 
     await _run_to_completion(ctx, upload)
 
@@ -476,6 +537,9 @@ async def test_check_queue_returns_without_awaiting_the_uploads(farm):
     and returns True so the run loop keeps ticking fast while they drain.
     """
     ctx = await farm(3, max_concurrent=3)
+    # Not assembling: this test reads the rows while the uploads are open, and
+    # an assembled batch releases as soon as it is complete, which would let the
+    # dispatches flip those rows to 'printing' mid-assertion.
     upload = _UploadRecorder()
 
     async with _scheduler_ctx(ctx, upload) as scheduler:
@@ -558,7 +622,7 @@ class TestSharedLibraryRow:
         session_maker = async_sessionmaker(engine, expire_on_commit=False)
         try:
             ctx = await self._library_farm(session_maker, tmp_path, 4, cleanup=False)
-            upload = _UploadRecorder()
+            upload = _UploadRecorder(assemble=True)
 
             await _run_check_queue(ctx, upload)