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.

CHANGELOG.md

@@ -33,6 +33,7 @@ All notable changes to Bambuddy will be documented in this file.
 - **Every FTP session Bambuddy opens now records how it closed (#3009, reported by @grengojbo)** — the report traced a print completion that opened two FTP connections to the printer, deleted one file and then, as far as the log showed, did nothing else until the printer was powered off 21 minutes later, and concluded the connections were being left open. They were not: the post-print SD-card cleanup opens one connection per candidate filename and closes each in a `finally`, which a run against a real FTPS server confirms at the server end for both the delete and the 550 not-here case. The trouble is that nothing in the log could have said so. Neither the clean close nor the hard socket drop logged anything at any level, so a session closed properly and a socket genuinely abandoned produced the same output — none — and the only way to tell them apart was to read the source. Both now log one DEBUG line naming the printer, whether QUIT was acknowledged or the socket had to be dropped without it, why, and how long the session was held. Every connect in a debug log is now paired with a close, so the next person suspecting a leaked FTP connection can settle it from a support bundle rather than by inference. Nothing about the connection handling itself changed, and at default log level nothing new is printed. This does not explain the SD-card read/write error in that report or in #645; it only removes one theory from the list by making it checkable.
 
 ### Fixed
+- **The concurrent-dispatch tests no longer depend on machine speed to prove the scheduler dispatches in parallel** — `tests/unit/test_scheduler_concurrent_dispatch.py` failed on CI and passed locally on a commit that touched nothing but a text file, and it had two independent reasons to. The first: the fixtures built their farm on `sqlite+aiosqlite:///:memory:`, and SQLAlchemy backs an in-memory SQLite with a `StaticPool` — one DBAPI connection handed to every session, with nothing keeping them apart. That was harmless while `check_queue` awaited its uploads inline, because only one session was ever live at a time. Under the refillable upload pool (#2602) the uploads run as concurrent background tasks with a session each, so their transactions interleave on that single connection: a sibling session's `close()` rolls back another's flushed-but-uncommitted UPDATE, which is how six queue items were each logged as `Status set to 'printing'` and four of them read back as `pending`. The fixtures now put the database in the test's own `tmp_path` and open it through the application's own connect listener, so each session gets its own connection and the file is opened the way the running system opens one — WAL, `synchronous = NORMAL`, a 15 s busy timeout — rather than with SQLite's defaults, which spend an fsync per commit and lock the whole file. The second reason was underneath the first and would have outlived it: `peak == 6` means the sixth dispatch reached its upload before the first one finished, and each dispatch runs a preamble of database work first, so the assertion was really a race between that spread and a fixed 0.15 s sleep. The spread is ~14 ms on a workstation and passed 150 ms on a CI runner, which recorded a peak of 4 out of 6 for a scheduler that was dispatching all six correctly. The uploads now hold until every upload the pass launched has arrived — `scheduler._inflight` is filled synchronously at launch, so its length is the batch size — which states the property directly, with no time in it, and ends sooner than the sleep it replaces. Reproduced both ways before and after: with one printer's preamble delayed by a second, the old recorder reports exactly the "high-water mark was 4" CI saw and the new one passes. Nothing in the scheduler changes; the concurrency under test was correct throughout.
 - **An AMS that reports no humidity percentage no longer shows the drop index as one (#3140, reported by @Sawtaytoes)** — Bambu sends two humidity fields that are not the same quantity: `humidity_raw` is relative humidity in percent, and `humidity` is the 1-5 drop index, which runs the other way — a high index means dry where a high percentage means wet. Bambuddy used the index whenever no percentage arrived, so a unit sending only the index rendered as "2%" in the green band while being the second-wettest of the five steps, charted an average of index values as a percentage, and could never cross a humidity alarm or auto-drying threshold, since no index reaches one. Such a unit now reports no humidity at all: the card hides the water-drop indicator, the history chart leaves a gap, and the alarm and auto-drying skip the unit instead of reading it as permanently bone dry. Temperature is recorded and alarmed on as before. No supported printer is known to be affected — the report came from an install running X1Plus, which Bambuddy does not support — and printers that send a percentage keep it, including two readings that previously fell through to the index because they would not parse as a whole number: one with a decimal point, and `38.0`. A unit that sends the index and no percentage now says so once in the log, with its firmware versions requested, so a supported printer that turns out to do this is visible rather than silently blank. Three smaller faults in the same paths went with it: a humidity of exactly 0% was stored as NULL when the firmware sent it as a number, a history window averaging exactly 0 reported no average at all while the minimum and maximum beside it reported 0, and a `humidity_raw` that was not a number at all aborted the whole recording pass for every printer, not just the one that sent it.
 - **Two Bambuddy instances could not have their CA certificates trusted at the same time (#3014, reported by @Steven-Pierce)** — a slicer holding the CA of two Bambuddy installs failed to connect to one of them, with the same generic `code=-1` an install whose CA was never imported gives. Each CA worked on its own; together, one stopped. Every install signed its virtual printers with an authority named exactly `CN=Virtual Printer CA`, and a slicer's trust store is a flat list of certificates that OpenSSL resolves by subject name: it takes the first authority whose name matches the certificate in front of it and fails the chain when that authority turns out not to have signed it, rather than trying the next match. Whichever of the two landed second in the file therefore lost, decided by nothing more than the order they were appended in. A newly generated CA now carries a per-install suffix taken from its own key — `CN=Virtual Printer CA D55808BE` — and publishes that key identifier, which the printer certificate then points back at, so the two are told apart by name and each chain resolves to the authority that really signed it. Nothing has to be re-imported on upgrade: an install that already has a CA keeps it untouched, and the printer certificates signed by it keep exactly the shape they have today. The collision goes away as soon as one of the two CAs is newer than this release — a second install set up after upgrading, or an existing one whose `bbl_ca.crt` and `bbl_ca.key` are deleted from `virtual_printer/certs/` so a fresh CA is generated, which does mean importing that one into the slicer again.
 - **ntfy per-event priorities were set, stored, and never sent (#3139, reported by @Thomansky)** — a per-event priority (#990) never reached ntfy: every notification arrived at the server's default, whatever the dialog showed. The two halves of the feature disagreed about the key. The dialog builds its rows from the provider's event toggles and stores the map under those names — `on_print_complete`, `on_print_failed` — while every sender is called with the bare event name, `print_complete`, so the lookup missed for all 18 events the dialog offers. It looked healthy from both ends: the stored config held exactly the priorities that were set, and the tests were green because they called the ntfy sender directly with the prefixed name, the one spelling the running system never produces. The lookup now accepts either spelling, so existing configurations keep working and nothing has to be migrated. The tests use the bare form, and one of them now runs from a finished print through to the outgoing request, which is the assertion that would have caught this. On upgrade, priorities that have been sitting in the database inertly start being honoured, so an event mapped to **Min** or **Low** will now arrive quieter than it did yesterday — that is the setting taking effect, not alerts going missing. The daily digest is unchanged and still goes out at the server default: it summarises several events at once, and the dialog offers no priority for it.

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)