Keep queued jobs when a printer refuses the upload (#3210)

CHANGELOG.md

@@ -59,6 +59,7 @@ All notable changes to Bambuddy will be documented in this file.
 - **The frontend build no longer warns about `path` and `crypto` being externalized for the STEP previewer (#2976)** — `occt-import-js`, the Emscripten build behind STEP previews, requires both modules, but only inside its `ENVIRONMENT_IS_NODE` branches; in the browser it loads its `.wasm` from the URL the preview worker passes and draws randomness from `crypto.getRandomValues`. Vite still externalized both and printed two warnings on every build. `vite.config.ts` now drops exactly those two warnings for that one package through `build.rolldownOptions.onLog`, so an externalization anywhere else, or of any other module, still shows.
 
 ### Fixed
+- **A printer that refuses uploads no longer empties the print queue (#3210, reported by @Leander-Vh)** — When a printer's file service would not take a file, the queue marked the job failed. That left the printer idle, so the next pass gave it the next job, which failed the same way: one P2S whose file service answered `421 There are too many connections` failed 43 queued jobs in ten minutes. When the file never reached the printer (the connection was refused, timed out or dropped), the job now stays in the queue, and the printer gets no new jobs for five minutes. The wait doubles with each refusal in a row, up to an hour, and the first upload that gets through resets it. Jobs for any printer of that model go to the others, and a job assigned to that printer shows "*printer* is not accepting files — Bambuddy will retry automatically" until it can be sent. One Job Waiting notification goes out per outage, not one per retry, and **Keep bed warm between prints** does not heat a printer that is waiting this way. A rejected access code, a full SD card (including an FTP 452 reply, which used to be read as a network error) or an upload too slow to finish still fails the job, because a retry would fail the same way.
 - **Re-reading a slot's RFID says when the printer refuses, and older firmware gets the command it understands (#3206, reported by @Sawtaytoes)** — The refresh sent `ams_get_rfid` and reported success without listening for the answer, so an X1C on X1Plus (base firmware 01.08.02.00), which answers it with `FAIL` / `ERROR STATE`, showed "Refreshing" while nothing happened, and the K profile was re-applied to a slot that had not been read. Bambuddy now waits for the printer's answer. When `ams_get_rfid` is refused by an X1, P1 or A1 printer that is not printing, it sends `M620 R<tray>`, the command Bambu Studio uses for those models; printers that accept `ams_get_rfid` never see it. If the printer refuses both, the refresh shows an error with the printer's reason and nothing is re-applied. A printer that does not answer at all is still treated as having taken the request, as before.
 - **The `{finish_photo_url}` link in a notification opens when authentication is on** — With authentication on, the link pointed at the archive's photo page, which needs a login that a link tapped in Telegram, CallMeBot or a Home Assistant notification can't carry, so it only ever showed an error. With authentication on the link now points at a copy of the photo under a long random name, which opens that one photo and nothing else and stops working after 3 days. With authentication off the link is unchanged. Photos attached to the message itself were not affected.
 - **The Reorder Alert and Stock Break Alert notifications are now actually sent (#2955, reported and contributed by @ojimpo in #3196)** — Both events could be switched on for a notification provider, but nothing in the backend worked out when a SKU was low: the Forecast panel on the Inventory page did that in the browser, so no alert could ever fire. Bambuddy now runs the same forecast once an hour, from your own inventory or from Spoolman, and sends an alert when a SKU reaches its reorder point or will run out before a reorder could arrive. Each SKU alerts once when it enters a condition and again only after it has cleared; one that gets worse, from reorder to stock break, alerts again. A SKU with its alerts snoozed is skipped, and a provider with both events on gets the stock break message only. What has been sent is stored, so a restart or an update does not send the alerts again. The default templates now name the subtype and colour, so two colours of one filament no longer send the same message; a template you have edited is left as it is.

backend/app/services/bambu_ftp.py

@@ -1275,7 +1275,14 @@ class BambuFTPClient:
             return False
         except (OSError, ftplib.Error) as e:
             logger.error("FTP upload failed for %s: %s (type: %s)", remote_path, e, type(e).__name__)
-            self.last_failure = FtpFailure(FtpFailureKind.NETWORK, str(e), _ftp_reply_code(e))
+            code = _ftp_reply_code(e)
+            # 452 is "insufficient storage" -- a 4xx, so ftplib raises it as
+            # error_temp rather than error_perm, but it is the printer talking
+            # about its card just as 552/553 are. As NETWORK it would read as a
+            # connection problem, and the queue would retry it forever (#3210).
+            is_storage = isinstance(e, ftplib.Error) and code == "452"
+            kind = FtpFailureKind.STORAGE if is_storage else FtpFailureKind.NETWORK
+            self.last_failure = FtpFailure(kind, str(e), code)
             return False
 
     def upload_bytes(self, data: bytes, remote_path: str) -> bool:

backend/app/services/print_scheduler.py

@@ -32,6 +32,7 @@ from backend.app.models.spool_assignment import SpoolAssignment
 from backend.app.models.spoolman_slot_assignment import SpoolmanSlotAssignment
 from backend.app.services import drying_preflight, print_dispatch_context, stock_forecast
 from backend.app.services.bambu_ftp import (
+    FtpFailureKind,
     FtpFailureReport,
     UploadCancelled,
     cache_3mf_download,
@@ -350,6 +351,35 @@ _ACTIVE_PRINT_STATES: frozenset[str] = frozenset({"PREPARE", "SLICING", "RUNNING
 # force-reconnect on the very next attempt — while still bounding the loop.
 DISPATCH_MAX_ATTEMPTS = 3
 
+# Upload failures that mean the file never reached the printer's storage: the
+# file service refused or never answered (#3210). These put the item back in
+# the queue instead of failing it, because nothing about the job is wrong --
+# and failing it left the printer idle, so the next pass handed it the next
+# item, which failed the same way. One P2S whose file service was out of
+# connection slots ate 43 queued jobs in ten minutes that way.
+#
+# Not here: AUTH (a wrong access code needs the user), STORAGE (a full or
+# missing card does too), NOT_FOUND (a Bambuddy-side path problem), UNKNOWN,
+# and an upload that overran its deadline -- each of those would fail the same
+# way on every retry.
+_UPLOAD_REQUEUE_KINDS: frozenset[FtpFailureKind] = frozenset(
+    {FtpFailureKind.HANDSHAKE, FtpFailureKind.COOLOFF, FtpFailureKind.TIMEOUT, FtpFailureKind.NETWORK}
+)
+
+# How long a printer whose upload was put back stays out of dispatch (#3210).
+# The first window matches the FTP client's own handshake cool-off, so the
+# retry lands after the client would talk to the printer again anyway. Each
+# further refusal in a row doubles it, up to the cap: #3210's printer refused
+# for some 40 hours, and every retry costs a preheat cycle where preheat is on,
+# five connection attempts and a page of log. A successful upload resets it.
+UPLOAD_FAILURE_BACKOFF_SECONDS = 300
+UPLOAD_FAILURE_BACKOFF_MAX_SECONDS = 3600
+
+
+def _upload_backoff_seconds(refusals: int) -> int:
+    """Backoff after the *refusals*-th refused upload in a row (1-based)."""
+    return min(UPLOAD_FAILURE_BACKOFF_SECONDS * 2 ** max(refusals - 1, 0), UPLOAD_FAILURE_BACKOFF_MAX_SECONDS)
+
 
 @dataclass(slots=True)
 class _ModelCandidate:
@@ -1053,6 +1083,20 @@ class PrintScheduler:
         # moment it connects.
         self._wake_failures: dict[int, float] = {}
         self._wake_failure_cooloff = 600  # seconds
+        # Printers whose last upload never reached them, mapped to the monotonic
+        # time they may be dispatched to again (#3210). Same expire-on-read shape
+        # as `_wake_failures`. Without it, a printer that cannot take files is
+        # idle on every pass, so it is picked on every pass.
+        self._upload_backoff: dict[int, float] = {}
+        # Refused uploads in a row per printer, which sets the next backoff
+        # window. Cleared by a successful upload to that printer.
+        self._upload_refusals: dict[int, int] = {}
+        # Printers whose current run of refusals has already sent its one
+        # "job waiting" notification. Each retry clears the item's waiting
+        # reason and the next refusal sets it again, which `hold_item` reads as
+        # a new reason; without this, every retry would notify. Cleared with
+        # `_upload_refusals`.
+        self._upload_refusal_notified: set[int] = set()
         # Track which printers are currently auto-drying (printer_id -> start timestamp)
         self._drying_in_progress: dict[int, float] = {}
         # Per-AMS memory of the auto-drying cycles WE armed, keyed by
@@ -1626,6 +1670,34 @@ class PrintScheduler:
             # generalises that workaround instead of repeating it per case.
             dispatching_printers: set[int] = set(busy_printers)
 
+            # Printers whose last upload never reached them (#3210). After the
+            # snapshot above on purpose: such a printer is idle, not about to
+            # print, and auto-drying must keep treating it as idle.
+            now_mono = time.monotonic()
+            # In backoff this pass. Kept apart from busy_printers so keep-warm
+            # can leave them out: there is no print to keep a bed warm for.
+            backoff_printers: set[int] = set()
+            # In backoff, and the "job waiting" notification for this run of
+            # refusals has gone out already. Their holds are written silently.
+            silent_hold_printers: set[int] = set()
+            for backoff_pid, retry_at in list(self._upload_backoff.items()):
+                if now_mono >= retry_at:
+                    del self._upload_backoff[backoff_pid]
+                    continue
+                backoff_printers.add(backoff_pid)
+                if backoff_pid in self._upload_refusal_notified:
+                    silent_hold_printers.add(backoff_pid)
+                else:
+                    self._upload_refusal_notified.add(backoff_pid)
+                mark_busy(
+                    backoff_pid,
+                    f"its file service refused the last upload; retrying in {retry_at - now_mono:.0f}s",
+                )
+                item_hold_reasons.setdefault(
+                    backoff_pid,
+                    f"{printer_label(backoff_pid)} is not accepting files — Bambuddy will retry automatically",
+                )
+
             # Printers held by a Home Assistant sensor interlock (#1148) — an
             # enclosure door left open, say. The fixed-printer branch turns
             # this into a waiting_reason the user can act on; the model-based
@@ -1762,6 +1834,7 @@ class PrintScheduler:
                         await hold_item(
                             item,
                             item_hold_reasons.get(item.printer_id) or f"Busy: {printer_label(item.printer_id)}",
+                            notify=item.printer_id not in silent_hold_printers,
                         )
                         continue
 
@@ -2196,7 +2269,9 @@ class PrintScheduler:
             # auxiliary check wedge the queue. The bed simply stays wherever it
             # was, and the next tick tries again.
             try:
-                await self._apply_keep_warm(db, items, dispatch_ids, busy_printers, require_plate_clear)
+                await self._apply_keep_warm(
+                    db, items, dispatch_ids, busy_printers - backoff_printers, require_plate_clear
+                )
             except Exception as e:
                 logger.warning("Keep-warm pass failed, continuing with dispatch: %s", e, exc_info=True)
 
@@ -2387,6 +2462,59 @@ class PrintScheduler:
                 # dispatchable again on the next tick.
                 await self._clear_dispatch_claim(item_db, item_id)
 
+    async def _requeue_after_upload_refused(
+        self,
+        db: AsyncSession,
+        item: PrintQueueItem,
+        printer: Printer,
+        error_msg: str,
+        toast_uid: int | None,
+    ) -> None:
+        """Put an item back in the queue after its file never reached the printer (#3210).
+
+        The item keeps its printer. Its AMS mapping was computed against that
+        printer's trays, and nothing on the row says whether a mapping was
+        computed or set by the user, so moving it to a sibling could print from
+        the wrong slots. The printer goes into `_upload_backoff` instead, which
+        keeps every other item away from it; this one waits there and is the
+        only thing that knocks again, once per backoff window. The window
+        grows with each refusal in a row; see ``_upload_backoff_seconds``.
+
+        `dispatch_attempts` is not charged: that budget bounds a printer that
+        takes the file and then never starts, which this is not.
+        """
+        refusals = self._upload_refusals.get(printer.id, 0) + 1
+        self._upload_refusals[printer.id] = refusals
+        backoff = _upload_backoff_seconds(refusals)
+        self._upload_backoff[printer.id] = time.monotonic() + backoff
+        # The row is still `pending` -- it only moves to `printing` after a
+        # successful upload -- so there is no status to write back. Writing one
+        # anyway would undo a cancel that landed during the upload.
+        if item.error_message:
+            item.error_message = None
+            await db.commit()
+        logger.warning(
+            "Queue item %s: upload to printer %s (%s) never reached it — %s Kept in the queue; "
+            "refusal %d in a row, so the printer is out of dispatch for %ds.",
+            item.id,
+            printer.id,
+            printer.name,
+            error_msg,
+            refusals,
+            backoff,
+        )
+        try:
+            # Closes the dispatch toast for this attempt. Without it the toast
+            # keeps spinning on an upload that has ended.
+            await ws_manager.send_queue_item_failed(
+                user_id=toast_uid,
+                queue_item_id=item.id,
+                printer_id=item.printer_id,
+                reason="upload_failed",
+            )
+        except Exception:
+            pass  # toast is best-effort
+
     def _rollback_unconfirmed_expected_print(self, item_id: int) -> None:
         """Drop an expectation for a print command that was never sent.
 
@@ -7697,6 +7825,10 @@ class PrintScheduler:
             # way to say so here, so the card got named even for a TLS
             # handshake that never reached the printer's filesystem (#2899).
             error_msg = upload_error or describe_upload_failure(upload_failure.failure)
+            failure = upload_failure.failure
+            if upload_error is None and failure is not None and failure.kind in _UPLOAD_REQUEUE_KINDS:
+                await self._requeue_after_upload_refused(db, item, printer, error_msg, toast_uid)
+                return
             item.status = "failed"
             item.error_message = error_msg
             item.completed_at = datetime.now(timezone.utc)
@@ -7728,6 +7860,11 @@ class PrintScheduler:
             await self._power_off_if_needed(db, item)
             return
 
+        # The printer took a file, so any run of refused uploads is over: the
+        # next refusal starts again at the shortest backoff, and notifies (#3210).
+        self._upload_refusals.pop(printer.id, None)
+        self._upload_refusal_notified.discard(printer.id)
+
         # Parse AMS mapping if stored
         ams_mapping = None
         if item.ams_mapping:

backend/tests/unit/services/test_upload_failure_reason_2899.py

@@ -400,22 +400,10 @@ async def _dispatch_failing_with(dispatch_case, failure: FtpFailure | None):
 
 
 class TestWhatTheQueueEntrySays:
-    async def test_a_handshake_failure_does_not_send_anyone_to_the_sd_card(self, dispatch_case):
-        """The report's own case: a TLS failure, answered with card advice.
-
-        The reporter acted on it and restarted the printer. Nothing in that
-        path reaches the printer's filesystem, and the cool-off that made the
-        next dispatch fail identically lives in Bambuddy's memory, where
-        power-cycling a printer does not reach.
-        """
-        message, reason = await _dispatch_failing_with(
-            dispatch_case, FtpFailure(FtpFailureKind.HANDSHAKE, "WRONG_VERSION_NUMBER")
-        )
-
-        assert "inserted" not in message, message
-        assert "FAT32" not in message, message
-        assert "not with TLS" in message, message
-        assert reason == message
+    # A handshake failure used to be this class's lead case. Since #3210 it no
+    # longer fails the item at all -- the file never reached the printer, so the
+    # item goes back in the queue (test_scheduler_upload_requeue_3210.py). Its
+    # wording is still pinned by TestTheWording above.
 
     async def test_a_553_still_gets_the_card_advice(self, dispatch_case):
         """The advice was written for this case and belongs to it.

backend/tests/unit/test_scheduler_upload_requeue_3210.py

@@ -0,0 +1,518 @@
+"""A failed upload must not eat the queue (#3210).
+
+The reporter queued 20 jobs for "any P2S". One P2S's file service was out of
+connection slots -- it answered port 990 with ``421 There are too many
+connections from your internet address`` -- so every upload to it failed. Each
+failure marked the item ``failed``, which left that printer idle, so the next
+pass handed it the next item. One item died about every nine seconds: 43 jobs
+gone in ten minutes, none printed on that printer.
+
+Two changes, pinned here:
+
+* An upload whose file never reached the printer (handshake refused, cool-off,
+  timeout, dropped connection) puts the item back in the queue. Failures that
+  would repeat on every retry -- a rejected access code, a full card -- still
+  fail it.
+* The printer is out of dispatch for a backoff window, so nothing else is
+  sent to it, and the item that was put back waits there with a reason.
+"""
+
+import asyncio
+import time
+from contextlib import ExitStack, asynccontextmanager
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+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.models.archive import PrintArchive
+from backend.app.models.print_queue import PrintQueueItem
+from backend.app.models.printer import Printer
+from backend.app.services.bambu_ftp import BambuFTPClient, FtpFailure, FtpFailureKind, UploadCancelled
+from backend.app.services.print_scheduler import (
+    UPLOAD_FAILURE_BACKOFF_MAX_SECONDS,
+    UPLOAD_FAILURE_BACKOFF_SECONDS,
+    PrintScheduler,
+)
+
+pytestmark = pytest.mark.unit
+
+REFUSING_IP = "10.0.0.8"
+HEALTHY_IP = "10.0.0.9"
+
+
+@pytest.fixture
+async def farm(tmp_path):
+    """Two printers -- one whose file service refuses everything -- and an item factory."""
+    engine = create_async_engine("sqlite+aiosqlite:///:memory:", echo=False)
+    async with engine.begin() as conn:
+        await conn.run_sync(Base.metadata.create_all)
+    session_maker = async_sessionmaker(engine, expire_on_commit=False)
+
+    base_dir = tmp_path / "farm"
+    (base_dir / "archives").mkdir(parents=True, exist_ok=True)
+
+    async with session_maker() as db:
+        refusing = Printer(
+            name="P2S-8", serial_number="S8", ip_address=REFUSING_IP, access_code="12345678", model="P2S"
+        )
+        healthy = Printer(name="P2S-9", serial_number="S9", ip_address=HEALTHY_IP, access_code="12345678", model="P2S")
+        db.add_all([refusing, healthy])
+        await db.commit()
+        ids = SimpleNamespace(refusing=refusing.id, healthy=healthy.id)
+
+    counter = iter(range(1000))
+
+    async def add_item(*, printer_id: int | None = None, target_model: str | None = None) -> int:
+        n = next(counter)
+        async with session_maker() as db:
+            archive_rel = Path("archives") / f"job-{n}.3mf"
+            (base_dir / archive_rel).write_bytes(b"archive payload")
+            archive = PrintArchive(
+                printer_id=printer_id,
+                filename=f"job-{n}.3mf",
+                file_path=str(archive_rel),
+                file_size=15,
+                status="completed",
+            )
+            db.add(archive)
+            await db.flush()
+            item = PrintQueueItem(
+                printer_id=printer_id,
+                target_model=target_model,
+                archive_id=archive.id,
+                status="pending",
+                position=n,
+            )
+            db.add(item)
+            await db.commit()
+            return item.id
+
+    try:
+        yield SimpleNamespace(session_maker=session_maker, base_dir=base_dir, ids=ids, add_item=add_item)
+    finally:
+        await engine.dispose()
+
+
+def _upload_refused_by(ip: str, failure: FtpFailure):
+    """An upload that fails with *failure* on *ip* and succeeds everywhere else."""
+
+    async def _upload(ip_address, *_args, **kwargs):
+        if ip_address != ip:
+            return True
+        if kwargs.get("failure") is not None:
+            kwargs["failure"].failure = failure
+        return False
+
+    return _upload
+
+
+@asynccontextmanager
+async def _scheduler(
+    ctx,
+    upload,
+    *,
+    busy: set[int] | None = None,
+    scheduler: PrintScheduler | None = None,
+    waiting_notify: AsyncMock | None = None,
+):
+    """A scheduler wired to *ctx*'s database, with every printer idle unless in *busy*.
+
+    Pass the same *waiting_notify* to several passes to count notifications
+    across them.
+    """
+    scheduler = scheduler or PrintScheduler()
+    busy = busy if busy is not None else set()
+    failed_notify = AsyncMock()
+    waiting_notify = waiting_notify 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),
+        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),
+        patch("backend.app.core.database.async_session", ctx.session_maker),
+        patch("backend.app.services.print_scheduler.printer_manager.is_connected", MagicMock(return_value=True)),
+        patch(
+            "backend.app.services.print_scheduler.printer_manager.get_status",
+            MagicMock(return_value=SimpleNamespace(state="IDLE", subtask_id=None, gcode_file=None, raw_data={})),
+        ),
+        patch(
+            "backend.app.services.print_scheduler.printer_manager.is_awaiting_plate_clear",
+            MagicMock(return_value=False),
+        ),
+        patch("backend.app.services.print_scheduler.printer_manager.start_print", MagicMock(return_value=True)),
+        patch("backend.app.services.print_scheduler.printer_manager.set_awaiting_plate_clear", MagicMock()),
+        patch("backend.app.services.print_scheduler.upload_file_async", upload),
+        patch("backend.app.services.print_scheduler.delete_file_async", AsyncMock(return_value=True)),
+        patch(
+            "backend.app.services.print_scheduler.get_ftp_retry_settings",
+            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", _real_spawn),
+        patch("backend.app.services.notification_service.notification_service.on_queue_job_started", AsyncMock()),
+        patch("backend.app.services.notification_service.notification_service.on_queue_job_failed", failed_notify),
+        patch("backend.app.services.notification_service.notification_service.on_queue_job_assigned", AsyncMock()),
+        patch("backend.app.services.notification_service.notification_service.on_queue_job_waiting", waiting_notify),
+        patch("backend.app.services.mqtt_relay.mqtt_relay.on_queue_job_started", AsyncMock()),
+        patch.object(scheduler, "_is_printer_idle", MagicMock(side_effect=lambda pid, *_a, **_k: pid not in busy)),
+        patch.object(scheduler, "_ensure_ams_mapping", AsyncMock(return_value=None)),
+        patch.object(scheduler, "_block_on_filament_deficit", AsyncMock(return_value=False)),
+        patch.object(scheduler, "_propagate_owner_to_printer_manager", AsyncMock()),
+        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)
+        scheduler.failed_notify = failed_notify
+        yield scheduler
+        tasks = [task for (task, _pid) in scheduler._inflight.values()]
+        if tasks:
+            await asyncio.gather(*tasks, return_exceptions=True)
+
+
+async def _item(ctx, item_id: int) -> PrintQueueItem:
+    async with ctx.session_maker() as db:
+        return await db.get(PrintQueueItem, item_id)
+
+
+HANDSHAKE = FtpFailure(FtpFailureKind.HANDSHAKE, "WRONG_VERSION_NUMBER (printer answered in cleartext: 421 ...)")
+
+
+# ---------------------------------------------------------------------------
+# What one failed upload does to its item
+# ---------------------------------------------------------------------------
+class TestOneFailedUpload:
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(
+        "kind",
+        [FtpFailureKind.HANDSHAKE, FtpFailureKind.COOLOFF, FtpFailureKind.TIMEOUT, FtpFailureKind.NETWORK],
+    )
+    async def test_a_file_that_never_arrived_keeps_the_item(self, farm, kind):
+        item_id = await farm.add_item(printer_id=farm.ids.refusing)
+        upload = _upload_refused_by(REFUSING_IP, FtpFailure(kind, "detail"))
+
+        async with _scheduler(farm, upload) as scheduler:
+            await scheduler.check_queue()
+
+        item = await _item(farm, item_id)
+        assert item.status == "pending"
+        assert item.printer_id == farm.ids.refusing
+        assert item.error_message is None
+        assert item.dispatch_attempts == 0, "the start-watchdog budget is for a printer that took the file"
+        scheduler.failed_notify.assert_not_awaited()
+        assert farm.ids.refusing in scheduler._upload_backoff
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(
+        "failure",
+        [
+            FtpFailure(FtpFailureKind.AUTH, "530 Login incorrect.", "530"),
+            FtpFailure(FtpFailureKind.STORAGE, "553 Could not create file.", "553"),
+            FtpFailure(FtpFailureKind.NOT_FOUND, "550 Permission denied.", "550"),
+            FtpFailure(FtpFailureKind.UNKNOWN, "500 what"),
+            None,
+        ],
+        ids=["auth", "storage", "not_found", "unknown", "unreported"],
+    )
+    async def test_a_failure_that_would_repeat_still_fails_the_item(self, farm, failure):
+        """Retrying a wrong access code or a full card every five minutes helps no one."""
+        item_id = await farm.add_item(printer_id=farm.ids.refusing)
+
+        async def upload(*_args, **kwargs):
+            if failure is not None:
+                kwargs["failure"].failure = failure
+            return False
+
+        async with _scheduler(farm, upload) as scheduler:
+            await scheduler.check_queue()
+
+        item = await _item(farm, item_id)
+        assert item.status == "failed"
+        assert item.error_message
+        scheduler.failed_notify.assert_awaited_once()
+        assert farm.ids.refusing not in scheduler._upload_backoff
+
+    @pytest.mark.asyncio
+    async def test_an_upload_that_overran_its_deadline_still_fails(self, farm):
+        """A link too slow to finish would be just as slow next time (#2529)."""
+        item_id = await farm.add_item(printer_id=farm.ids.refusing)
+
+        async def upload(*_args, **kwargs):
+            kwargs["failure"].failure = FtpFailure(FtpFailureKind.TIMEOUT, "deadline")
+            raise UploadCancelled("too slow")
+
+        async with _scheduler(farm, upload) as scheduler:
+            await scheduler.check_queue()
+
+        assert (await _item(farm, item_id)).status == "failed"
+
+    @pytest.mark.asyncio
+    async def test_a_cancel_during_the_upload_is_not_undone(self, farm):
+        """The row is never written back to pending, so a cancel that won stays won."""
+        item_id = await farm.add_item(printer_id=farm.ids.refusing)
+
+        async def upload(*_args, **kwargs):
+            async with farm.session_maker() as other:
+                row = await other.get(PrintQueueItem, item_id)
+                row.status = "cancelled"
+                await other.commit()
+            kwargs["failure"].failure = HANDSHAKE
+            return False
+
+        async with _scheduler(farm, upload) as scheduler:
+            await scheduler.check_queue()
+
+        assert (await _item(farm, item_id)).status == "cancelled"
+
+
+# ---------------------------------------------------------------------------
+# The report: a refusing printer must not drain the queue
+# ---------------------------------------------------------------------------
+class TestTheQueueIsNotDrained:
+    @pytest.mark.asyncio
+    async def test_any_model_items_stop_going_to_the_refusing_printer(self, farm):
+        """#3210's shape: "any P2S" items, one P2S refusing every upload.
+
+        Pass 1 sends one item to each printer; the refusing one keeps its item.
+        Pass 2 has the healthy printer busy printing and the refusing one in
+        backoff, so the rest wait instead of being fed to it one by one.
+        """
+        ids = [await farm.add_item(target_model="P2S") for _ in range(5)]
+        upload = _upload_refused_by(REFUSING_IP, HANDSHAKE)
+        scheduler = PrintScheduler()
+
+        async with _scheduler(farm, upload, scheduler=scheduler):
+            await scheduler.check_queue()
+        async with _scheduler(farm, upload, scheduler=scheduler, busy={farm.ids.healthy}):
+            for _ in range(3):
+                await scheduler.check_queue()
+
+        items = [await _item(farm, i) for i in ids]
+        assert [i.status for i in items].count("failed") == 0, [(i.id, i.status, i.error_message) for i in items]
+        assert [i.status for i in items].count("printing") == 1
+        held = [i for i in items if i.printer_id == farm.ids.refusing]
+        assert len(held) == 1, "exactly one item waits on the refusing printer"
+        assert held[0].status == "pending"
+        waiting = [i for i in items if i.status == "pending" and i.printer_id is None]
+        assert len(waiting) == 3, "the rest stay unassigned, free for any printer that comes free"
+
+    @pytest.mark.asyncio
+    async def test_the_item_kept_on_the_printer_says_why(self, farm):
+        item_id = await farm.add_item(printer_id=farm.ids.refusing)
+        upload = _upload_refused_by(REFUSING_IP, HANDSHAKE)
+        scheduler = PrintScheduler()
+
+        async with _scheduler(farm, upload, scheduler=scheduler):
+            await scheduler.check_queue()
+        async with _scheduler(farm, upload, scheduler=scheduler):
+            await scheduler.check_queue()
+
+        item = await _item(farm, item_id)
+        assert item.status == "pending"
+        assert item.waiting_reason == "P2S-8 is not accepting files — Bambuddy will retry automatically"
+
+    @pytest.mark.asyncio
+    async def test_the_printer_is_tried_again_once_the_backoff_ends(self, farm):
+        item_id = await farm.add_item(printer_id=farm.ids.refusing)
+        scheduler = PrintScheduler()
+
+        async with _scheduler(farm, _upload_refused_by(REFUSING_IP, HANDSHAKE), scheduler=scheduler):
+            await scheduler.check_queue()
+
+        retry_at = scheduler._upload_backoff[farm.ids.refusing]
+        assert retry_at - time.monotonic() == pytest.approx(UPLOAD_FAILURE_BACKOFF_SECONDS, abs=5)
+
+        # The printer recovered and the window has passed.
+        scheduler._upload_backoff[farm.ids.refusing] = time.monotonic() - 1
+        async with _scheduler(farm, AsyncMock(return_value=True), scheduler=scheduler):
+            await scheduler.check_queue()
+
+        assert (await _item(farm, item_id)).status == "printing"
+        assert farm.ids.refusing not in scheduler._upload_backoff
+
+
+# ---------------------------------------------------------------------------
+# A printer that stays broken: retries thin out, and say so once
+# ---------------------------------------------------------------------------
+def _expire_backoff(scheduler: PrintScheduler, printer_id: int) -> None:
+    """Jump past the printer's backoff window, as if the time had passed."""
+    scheduler._upload_backoff[printer_id] = time.monotonic() - 1
+
+
+def _remaining(scheduler: PrintScheduler, printer_id: int) -> float:
+    return scheduler._upload_backoff[printer_id] - time.monotonic()
+
+
+async def _finish(ctx, scheduler: PrintScheduler, item_id: int) -> None:
+    """Mark a dispatched item's print done, so its printer can take the next one."""
+    async with ctx.session_maker() as db:
+        row = await db.get(PrintQueueItem, item_id)
+        assert row.status == "printing"
+        row.status = "completed"
+        await db.commit()
+        scheduler._release_dispatch_hold(row.printer_id)
+
+
+class TestAPrinterThatStaysBroken:
+    @pytest.mark.asyncio
+    async def test_each_refusal_in_a_row_doubles_the_wait_up_to_the_cap(self, farm):
+        """#3210's printer refused for about 40 hours.
+
+        At a flat five minutes that is ~480 retries, each one a preheat cycle
+        where preheat is on, five connection attempts and a page of log.
+        """
+        await farm.add_item(printer_id=farm.ids.refusing)
+        upload = _upload_refused_by(REFUSING_IP, HANDSHAKE)
+        scheduler = PrintScheduler()
+
+        waits = []
+        for _ in range(6):
+            async with _scheduler(farm, upload, scheduler=scheduler):
+                await scheduler.check_queue()
+            waits.append(_remaining(scheduler, farm.ids.refusing))
+            _expire_backoff(scheduler, farm.ids.refusing)
+
+        expected = [300, 600, 1200, 2400, 3600, 3600]
+        assert expected[0] == UPLOAD_FAILURE_BACKOFF_SECONDS
+        assert expected[-1] == UPLOAD_FAILURE_BACKOFF_MAX_SECONDS
+        assert waits == [pytest.approx(w, abs=5) for w in expected]
+
+    @pytest.mark.asyncio
+    async def test_a_successful_upload_resets_the_wait(self, farm):
+        first = await farm.add_item(printer_id=farm.ids.refusing)
+        refused = _upload_refused_by(REFUSING_IP, HANDSHAKE)
+        scheduler = PrintScheduler()
+
+        for _ in range(3):
+            async with _scheduler(farm, refused, scheduler=scheduler):
+                await scheduler.check_queue()
+            _expire_backoff(scheduler, farm.ids.refusing)
+        async with _scheduler(farm, AsyncMock(return_value=True), scheduler=scheduler):
+            await scheduler.check_queue()
+        assert farm.ids.refusing not in scheduler._upload_refusals
+        await _finish(farm, scheduler, first)
+
+        # It breaks again later: back to the first window, not the fourth.
+        await farm.add_item(printer_id=farm.ids.refusing)
+        async with _scheduler(farm, refused, scheduler=scheduler, busy=set()):
+            await scheduler.check_queue()
+        assert _remaining(scheduler, farm.ids.refusing) == pytest.approx(UPLOAD_FAILURE_BACKOFF_SECONDS, abs=5)
+
+    @pytest.mark.asyncio
+    async def test_one_outage_sends_one_waiting_notification(self, farm):
+        """Every retry clears the waiting reason and the next refusal sets it again.
+
+        `hold_item` reads that as a new reason each time, and "Job Waiting" is
+        on by default -- so without a guard a 40-hour outage notifies ~480 times.
+        """
+        await farm.add_item(printer_id=farm.ids.refusing)
+        upload = _upload_refused_by(REFUSING_IP, HANDSHAKE)
+        scheduler = PrintScheduler()
+        waiting = AsyncMock()
+
+        for _ in range(4):
+            # The pass that dispatches and is refused...
+            async with _scheduler(farm, upload, scheduler=scheduler, waiting_notify=waiting):
+                await scheduler.check_queue()
+            # ...and the pass that holds the item while the printer waits.
+            async with _scheduler(farm, upload, scheduler=scheduler, waiting_notify=waiting):
+                await scheduler.check_queue()
+            _expire_backoff(scheduler, farm.ids.refusing)
+
+        assert waiting.await_count == 1
+        assert "not accepting files" in waiting.await_args.kwargs["waiting_reason"]
+
+    @pytest.mark.asyncio
+    async def test_a_new_outage_after_a_recovery_notifies_again(self, farm):
+        first = await farm.add_item(printer_id=farm.ids.refusing)
+        refused = _upload_refused_by(REFUSING_IP, HANDSHAKE)
+        scheduler = PrintScheduler()
+        waiting = AsyncMock()
+
+        async def outage():
+            async with _scheduler(farm, refused, scheduler=scheduler, waiting_notify=waiting):
+                await scheduler.check_queue()
+            async with _scheduler(farm, refused, scheduler=scheduler, waiting_notify=waiting):
+                await scheduler.check_queue()
+            _expire_backoff(scheduler, farm.ids.refusing)
+
+        await outage()
+        async with _scheduler(farm, AsyncMock(return_value=True), scheduler=scheduler, waiting_notify=waiting):
+            await scheduler.check_queue()
+        await _finish(farm, scheduler, first)
+        await farm.add_item(printer_id=farm.ids.refusing)
+        await outage()
+
+        assert waiting.await_count == 2
+
+    @pytest.mark.asyncio
+    async def test_keep_warm_does_not_hold_a_bed_for_a_printer_in_backoff(self, farm):
+        """There is no print coming to keep the bed warm for.
+
+        The printer is in busy_printers and has a pending item, which is
+        exactly what keep-warm looks for -- and each retry would restart its
+        time cap, so the bed could stay hot for the whole outage.
+        """
+        await farm.add_item(printer_id=farm.ids.refusing)
+        upload = _upload_refused_by(REFUSING_IP, HANDSHAKE)
+        scheduler = PrintScheduler()
+
+        async with _scheduler(farm, upload, scheduler=scheduler):
+            await scheduler.check_queue()
+        keep_warm = AsyncMock()
+        async with _scheduler(farm, upload, scheduler=scheduler):
+            with patch.object(scheduler, "_apply_keep_warm", keep_warm):
+                await scheduler.check_queue()
+
+        busy_seen = keep_warm.await_args.args[3]
+        assert farm.ids.refusing not in busy_seen
+
+
+# ---------------------------------------------------------------------------
+# 452 is the card, not the network
+# ---------------------------------------------------------------------------
+class TestA452IsAStorageReply:
+    def _upload_raising(self, error, tmp_path):
+        local = tmp_path / "job.3mf"
+        local.write_bytes(b"x" * 16)
+        client = BambuFTPClient(REFUSING_IP, "12345678")
+        client._ftp = MagicMock()
+        client._ftp.transfercmd.side_effect = error
+        assert client.upload_file(local, "/job.3mf") is False
+        return client.last_failure
+
+    def test_452_is_storage(self, tmp_path):
+        """ftplib raises it as error_temp, but it says the card is full.
+
+        Read as NETWORK it would be retried forever instead of failing with
+        the advice to check the card.
+        """
+        import ftplib  # nosec B402 -- tests construct real ftplib error types
+
+        failure = self._upload_raising(ftplib.error_temp("452 Insufficient storage space."), tmp_path)
+        assert failure.kind is FtpFailureKind.STORAGE
+        assert failure.code == "452"
+
+    def test_other_transient_replies_are_still_network(self, tmp_path):
+        import ftplib  # nosec B402 -- tests construct real ftplib error types
+
+        failure = self._upload_raising(ftplib.error_temp("421 There are too many connections."), tmp_path)
+        assert failure.kind is FtpFailureKind.NETWORK
+
+    def test_a_socket_error_is_never_read_as_a_reply_code(self, tmp_path):
+        failure = self._upload_raising(OSError("452 looks like a code but is not a reply"), tmp_path)
+        assert failure.kind is FtpFailureKind.NETWORK