diff --git a/CHANGELOG.md b/CHANGELOG.md index 70b241620..a8ddbcaeb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,7 @@ All notable changes to Bambuddy will be documented in this file. - **Orca Cloud profile sync now connects by approving a code instead of the copy-paste sign-in** — Connecting Bambuddy to Orca Cloud used to mean opening an OAuth sign-in in a new tab, watching it redirect to a `localhost` URL that fails to load, then copying that dead URL out of the address bar and pasting it back into Bambuddy. That dance existed only because Orca's auth backend (Supabase) accepts no redirect target other than `localhost`, and the deliberately-broken redirect page confused nearly everyone who reached it. OrcaSlicer has since shipped a first-class external-app pairing API (the OAuth 2.0 Device Authorization Grant, RFC 8628), so the flow is now: click **Connect**, approve a short code on your Orca Cloud settings page, and Bambuddy pairs itself — no redirect, no paste, no client secret, and it behaves identically from a LAN IP, `localhost`, or behind a reverse proxy. Bambuddy requests **read-only** access (it only lists and views your Orca Cloud profiles), keeps the pairing alive with the API's rotating refresh tokens (validated end-to-end against Orca's staging and production servers), and stores nothing beyond the issued token pair. The profile list and detail views are unchanged, so nothing downstream of the connect step looks different. The old paste-based sign-in and the email/password fallback are removed. Points at production Orca Cloud by default; `ORCA_CLOUD_API_BASE` overrides the endpoint for testing. ### Fixed +- **`queue_max_concurrent_uploads` behaved as a per-batch cap instead of a refillable pool (#2602, reporter @Jostxxl)** — On a large farm, unused upload slots sat idle whenever any upload from the current batch was still running. **Root cause.** `check_queue()` awaited `_dispatch_selected()`, which awaited `asyncio.gather()` over the whole selected batch before returning — so the scheduler's run loop was blocked until the *slowest* FTP transfer in the batch finished. A 96 MB 3MF that took 513 s to upload left 15 of 16 configured slots unused for 8.5 minutes on a 93-printer farm, even as other printers came free; jobs that became eligible during the long upload couldn't be dispatched. The batch-await was load-bearing for one reason: `_start_print` flips a row `pending → printing` only *after* its upload completes, so returning early would have let the next pass re-dispatch the still-`pending` in-flight rows. **Fix.** Uploads now run 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 (3 s) tick instead of waiting out the whole batch, and the configured limit finally behaves as a continuously-refillable worker pool. The no-double-dispatch invariant the batch-await used to provide is now carried by the in-flight exclusion; the `pending → printing` CAS, the busy-printer guard (#2598), the per-printer dispatch hold, auto-drying exclusion (in-flight printers stay out, including on the no-pending-items path), and per-item failure isolation are all preserved and run per task. Investigated with the reporter's large-farm hotfix and reproduction; covered by rewritten pool tests (cap holds across refills, freed slot refills, in-flight item/printer excluded from re-selection, check_queue returns without awaiting uploads). - **Configuring a built-in/generic filament on an AMS slot reverted to the old profile a moment later (#2604, reporter @Jostxxl)** — Selecting a built-in preset (e.g. Generic ABS) through **Printer → AMS slot → Configure** briefly showed the new material on the printer, then the slot snapped back to whatever was there before (e.g. an old Generic PETG). **Root cause.** The Configure AMS Slot modal sends built-in, local, and Orca-generic presets with a `GF*` `tray_info_idx` but an **empty** `setting_id` (those presets carry no Bambu Cloud setting id of their own), and the `configure_ams_slot` route forwarded that empty value straight to `ams_filament_setting`. The firmware treats a slot that has a filament id but no setting id as half-configured: it accepts the update, then reverts to its previously stored profile. The inventory/assignment path already guards against this by deriving the setting id from the filament id, but the manual Configure path didn't, leaving two inconsistent code paths. **Fix.** `configure_ams_slot` now back-fills `setting_id` from the resolved `tray_info_idx` via `filament_id_to_setting_id` whenever the client sent none (e.g. `GFB99` → `GFSB99`), mirroring the inventory path. Doing it server-side also protects API callers and any future frontend. `P*` user presets and already-`GFS*` values are left untouched, and an explicitly-supplied `setting_id` (including the `PFUS*` pair) still passes through unchanged. Covered by tests for the built-in empty-`setting_id` case and the material-only generic-fallback case both publishing a derived `GFS*` id. - **The HT-A (AMS-HT) spool vanished a few seconds after every power-on (#2594, reporter @GuillaumeHouba)** — On an H2C, the spool in the HT-A high-temp AMS on the left nozzle showed correctly with its RFID assignment on power-on, then disappeared seconds later; the regular AMS spools stayed. **Root cause.** The AMS merge in `_handle_ams_data` clears a tray when it receives a partial `{id, state}` update whose `state != 11` — the rule that lets 4-slot AMS units (e.g. H2D) report an emptied slot with just `{id, state}` and no `tray_type` (#784), where `11` = loaded. But an **AMS-HT** (single-tray high-temp dry box, unit id ≥ 128) reports its *loaded* tray as `state=9`, not 11 — it doesn't feed filament into a shared buffer the way a 4-slot AMS does. So the partial `{id:0, state:9}` the printer sends for the HT tray on power-on was misread as "slot emptied," and Bambuddy wiped the tray's `tray_type` / RFID / Spoolman assignment. The support log showed it plainly: every "state=9 (not loaded) — clearing stale tray data" was on AMS 128, never on the regular AMS unit 0 (which correctly reports 11). **Fix.** The `state != 11 → empty` heuristic is now skipped for AMS-HT units (id ≥ 128); their differing single-tray state semantics mean a partial state update must not clear a present spool. A genuine HT spool removal still clears through the explicit `tray_type == ""` update and the `tray_exist_bits` cleanup, both unchanged, and regular AMS behavior (id < 128) is untouched. Covered by tests for the HT tray surviving a `state=9` partial, the HT still clearing on an explicit empty, and the existing regular-AMS `state=9`/`10`/`11` cases. - **A start-print dispatched to an already-busy printer could cancel the running job (#2598, reporter @khaosdoctor)** — On an A1 mini across a night of prints, jobs were cancelled with no apparent cause; debug logs showed Bambuddy sending `project_file` twice ~3 minutes apart with no completion between, and the printer answering `0500_4004` ("Device is busy and cannot start a new task") — which on that model cancels the RUNNING print. **Root cause.** `start_print()` in the MQTT client guarded only on connection state (`self._client and self.state.connected`) — it published `project_file` with no check on the printer's `gcode_state`. The scheduler *does* gate dispatch on an idle check, but that check treats `FINISH` as idle, and a printer can keep reporting `FINISH` for tens of seconds *after* it accepted a `project_file`; combined with a dispatch watchdog that reverts a queue item and releases its dispatch hold when it doesn't observe the active-state transition in time (#2555), a re-selected item could reach the FTP upload while the printer had actually started, so the start command landed on a live print. **Fix (defense-in-depth).** (a) `start_print()` now refuses to publish `project_file` when the printer is in an active state (`PREPARE` / `SLICING` / `RUNNING` / `PAUSE`) and returns without sending — a single guard at the one publish choke point that every dispatch path (queue, manual, webhook, Virtual-Printer forward) funnels through. `IDLE` / `FINISH` / `FAILED` remain valid start targets. (b) The scheduler re-checks the live printer state right before the FTP upload and *defers* a busy printer (leaves the item pending for a later tick) instead of uploading and dispatching — no wasted transfer, no collision. (c) If the printer goes busy during the upload window and the start command is refused, the scheduler reverts the item to pending (a deferral) rather than marking it failed — a busy printer is not a failure. Covered by tests for the client-level guard (refused while busy, published while idle, guard precedes the connection check) and the scheduler deferring both before the upload and after a busy-refused start. Note: a transport-level MQTT QoS-1 replay on reconnect would bypass the client guard, but the dispatch/watchdog reconnect path already hard-resets the client with a fresh session so it has no inflight `project_file` to replay. diff --git a/backend/app/services/print_scheduler.py b/backend/app/services/print_scheduler.py index 95fbc54da..641a7de63 100644 --- a/backend/app/services/print_scheduler.py +++ b/backend/app/services/print_scheduler.py @@ -269,6 +269,17 @@ class PrintScheduler: # Matches the watchdog timeout (90 s) plus a safety margin so the # watchdog runs first on the unhappy path. self._dispatch_max_hold = 180.0 + # Refillable upload pool (#2602). Items whose FTP upload was launched by + # an earlier pass and is still running. `_start_print` flips the row + # pending -> printing only *after* the upload completes, so until then + # the row stays `pending`: each tick, check_queue excludes these + # item_ids from re-selection and their printers from new dispatch / + # auto-drying, and launches only `limit - len(_inflight)` new uploads so + # freed slots refill on the next fast tick. check_queue is the sole, + # sequential caller and the prune done-callbacks run in the same + # event-loop thread, so this dict needs no lock. + # item_id -> (task, printer_id) + self._inflight: dict[int, tuple[asyncio.Task, int | None]] = {} async def run(self): """Main loop - check queue every interval.""" @@ -336,6 +347,14 @@ class PrintScheduler: ) items = list(result.scalars().all()) + # Drop rows whose upload is still in flight from an earlier pass + # (#2602). They stay `pending` until the upload finishes, so without + # this a fast tick would re-select and re-dispatch the same row. + # Belt-and-suspenders with the printer exclusion below. + if self._inflight: + inflight_ids = set(self._inflight) + items = [it for it in items if it.id not in inflight_ids] + # Read plate-clear setting once per queue check. Default MUST be # False to match the schema (SettingsSchema.require_plate_clear # defaults False) and the frontend (toggle + card badge both treat a @@ -346,9 +365,15 @@ class PrintScheduler: require_plate_clear = await self._get_bool_setting(db, "require_plate_clear", default=False) 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 False + # No dispatchable pending items — still check auto-drying on idle + # printers, but keep any printer with an upload still in flight + # from an earlier pass out of it (#2602): its print is imminent, + # so it must not be auto-dried in the gap before the row flips to + # printing. Report the pass as productive while uploads run so the + # loop stays on the fast interval. + inflight_printers = {pid for (_task, pid) in self._inflight.values() if pid is not None} + await self._check_auto_drying(db, [], inflight_printers, require_plate_clear=require_plate_clear) + return bool(self._inflight) logger.info( "Queue check: found %d pending items: %s", @@ -383,6 +408,15 @@ class PrintScheduler: if self._printer_in_dispatch_hold(held_printer_id): busy_printers.add(held_printer_id) + # Exclude printers whose upload is still in flight from an earlier + # pass (#2602). The row is `pending` until the upload finishes and + # the printing-state seed / dispatch hold above only arm once the + # upload completes, so this is what holds the printer (and, via + # busy_printers, its auto-drying) out of the pass during the upload. + for _task, inflight_pid in self._inflight.values(): + if inflight_pid is not None: + busy_printers.add(inflight_pid) + # Log skip reasons once per queue check (not per item) skip_reasons: dict[str, int] = {} @@ -763,73 +797,94 @@ class PrintScheduler: await db.commit() if dispatch_ids: - await self._dispatch_selected(dispatch_ids, upload_limit) + item_printers = {it.id: it.printer_id for it in items} + self._launch_uploads(dispatch_ids, item_printers, upload_limit) # 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) + # Keep the loop on the fast interval while any upload is in flight so + # a slot freed mid-tick refills within seconds rather than after the + # 30 s idle sleep (#2602). Selecting anything this pass (launched or + # deferred because the pool was full) also counts as productive. + return bool(dispatch_ids) or bool(self._inflight) - async def _dispatch_selected(self, item_ids: list[int], limit: int) -> None: - """Upload and start every item selected by this queue pass, in parallel. + def _launch_uploads(self, item_ids: list[int], item_printers: dict[int, int | None], limit: int) -> None: + """Launch selected uploads as a refillable pool, capped at ``limit`` (#2602). Dispatch used to happen inline in the selection loop: ``await - _start_print(db, item)`` for each item in turn. Since ``_start_print`` + _start_print(db, item)`` per item in turn. Since ``_start_print`` performs the FTP upload, that serialized every printer behind every - other printer's transfer — even though the printers are entirely - independent machines. A Bambu printer's FTP server sustains ~150 KB/s - (its own SD write is the bottleneck, not the network), so a 41 MB 3MF - takes ~4 minutes. The reporter's 19-printer farm therefore needed ~80 - minutes before the last printer received its file, and the queue looked - like it was starting prints "one by one, very slowly" (#2555). + other printer's transfer even though the printers are independent + machines; #2555 moved it 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 — a + 513 s upload left 15 of 16 configured slots idle for 8.5 minutes on a + 93-printer farm even as other printers came free (#2602). - Uploads to *different* printers contend for nothing, so they run - concurrently here, bounded by ``queue_max_concurrent_uploads``. The - bound exists because the printers are independent but the host is not: - each in-flight upload holds a thread in the FTP pool, a TLS session and - a file handle. + Each upload now runs as an independent background task tracked in + ``self._inflight``. check_queue excludes in-flight item_ids (still + `pending` until their upload completes) and their printers from the + next pass's selection, and this method launches at most + ``limit - len(self._inflight)`` new uploads, so a freed slot refills on + the next fast tick instead of waiting out the whole batch. The bound + exists because the printers are independent but the host is not: each + in-flight upload holds a thread in the FTP pool, a TLS session and a + file handle. - This is awaited before ``check_queue`` returns, which preserves the - invariant the rest of the scheduler is built on: a pass never overlaps - with the next one. It matters more than it looks — ``_start_print`` - flips the row pending -> printing only *after* the upload finishes, so - a pass that returned early while uploads were still in flight would let - the next pass re-dispatch the very same still-pending rows. + The no-overlapping-dispatch invariant the batch-await used to provide + is now carried by the in-flight exclusion in check_queue. Everything + else — the pending->printing CAS, the busy-printer guard (#2598), the + per-printer hold, and each item's independent failure handling — still + lives in ``_start_print`` and runs per task exactly as before. - ``limit`` is read by the caller, on the caller's session, before it - commits — reading it here would leave that session idle-in-transaction - for the whole dispatch. This function deliberately takes no session. + Synchronous on purpose: it registers every launched task into + ``self._inflight`` before returning, so the next (sequential) tick sees + an accurate in-flight count with no interleaving await. """ - sem = asyncio.Semaphore(limit) - - async def _one(item_id: int) -> None: - # Its own session: these run concurrently, and an AsyncSession is not - # safe to share across tasks. It also keeps a slow upload from pinning - # the caller's session (and, on SQLite, its transaction) open for the - # duration. - async with sem, async_session() as item_db: - item = await item_db.get(PrintQueueItem, item_id) - if not item: - logger.info("Queue item %s vanished before dispatch — skipping", item_id) - return - await self._start_print(item_db, item) + free = limit - len(self._inflight) + if free <= 0: + logger.info( + "Upload pool full (%d/%d in flight) — deferring %d item(s) to a later tick: %s", + len(self._inflight), + limit, + len(item_ids), + item_ids, + ) + return + to_launch = item_ids[:free] + deferred = item_ids[free:] logger.info( - "Dispatching %d queue item(s) with up to %d concurrent upload(s): %s", - len(item_ids), + "Launching %d upload(s) (pool %d/%d in flight)%s", + len(to_launch), + len(self._inflight), limit, - item_ids, + f" — deferring {deferred} to a later tick" if deferred else "", ) - results = await asyncio.gather(*(_one(i) for i in item_ids), return_exceptions=True) - # gather() with return_exceptions keeps one printer's failure from - # cancelling its siblings' in-flight uploads. _start_print already - # handles its own failure modes and marks the item failed; anything - # arriving here is unexpected, so log it loudly rather than letting - # gather swallow it. - for item_id, result in zip(item_ids, results, strict=True): - if isinstance(result, BaseException): - logger.error("Queue item %s: dispatch raised %s: %s", item_id, type(result).__name__, result) + for item_id in to_launch: + task = spawn_background_task(self._dispatch_one(item_id), name=f"queue-upload-{item_id}") + self._inflight[item_id] = (task, item_printers.get(item_id)) + # Prune on completion so the freed slot is refillable next tick. + # spawn_background_task already logs any uncaught exception; this + # only reclaims the pool slot (fires on success, failure, or cancel). + task.add_done_callback(lambda _t, iid=item_id: self._inflight.pop(iid, None)) + + async def _dispatch_one(self, item_id: int) -> None: + """Upload + start one queue item in its own session (pool worker, #2602). + + Its own session: pool workers run concurrently and an AsyncSession is + not safe to share across tasks; it also keeps a slow upload from pinning + the scheduler's session (and, on SQLite, its transaction) open for the + transfer's duration. + """ + async with async_session() as item_db: + item = await item_db.get(PrintQueueItem, item_id) + if not item: + logger.info("Queue item %s vanished before dispatch — skipping", item_id) + return + await self._start_print(item_db, item) async def _find_idle_printer_for_model( self, diff --git a/backend/tests/unit/test_scheduler_concurrent_dispatch.py b/backend/tests/unit/test_scheduler_concurrent_dispatch.py index cf6894293..6066873c5 100644 --- a/backend/tests/unit/test_scheduler_concurrent_dispatch.py +++ b/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"