diff --git a/backend/tests/unit/test_scheduler_concurrent_dispatch.py b/backend/tests/unit/test_scheduler_concurrent_dispatch.py index 4feb37c33..544eef58f 100644 --- a/backend/tests/unit/test_scheduler_concurrent_dispatch.py +++ b/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)