fix(ftp): stop a slow upload from being retried on top of itself (#2529)

upload_file_async carried a flat 600s wall-clock deadline and ran the
transfer via asyncio.wait_for(run_in_executor(...)). wait_for cancels the
future, not the executor thread. A 96 MB 3MF to an A1 over WiFi sustains
~75 KB/s and needs ~20 minutes, so the await gave up at ~70 MB, returned
False, and with_ftp_retry started a second STOR of the same file onto the
same printer while the first was still streaming. The reporter filmed two
transfers of one job climbing in parallel at 2% and 72%; the print never
landed and the printer read as having a flaky network.

The deadline is now derived from the file size against a 25 KB/s floor, so a
slow-but-healthy transfer can finish — a link that has actually died is
caught within socket_timeout by the blocking sendall, which is what should be
detecting failure. A deadline expiry now stops the transfer for real: the
worker is signalled, raises UploadCancelled from its progress callback, and
upload_file's existing cancel path breaks the send loop and deletes the
partial file. with_ftp_retry never retries that, and a per-printer lock makes
overlapping uploads impossible however they were triggered.

backend/app/services/bambu_ftp.py

@@ -6,6 +6,7 @@ import socket
 import ssl
 import threading
 import time
+import weakref
 from collections.abc import Awaitable, Callable
 from enum import Enum
 from ftplib import FTP, FTP_TLS  # nosec B402
@@ -17,6 +18,32 @@ logger = logging.getLogger(__name__)
 
 T = TypeVar("T")
 
+# Overall upload deadline (#2529). A flat wall-clock cap punishes big files on
+# slow links rather than catching broken ones: a 96 MB 3MF at the ~75 KB/s an A1
+# sustains over WiFi legitimately needs ~20 minutes, and the old flat 600 s
+# declared it dead at ~70 MB. The deadline is therefore derived from the file
+# size against a deliberately pessimistic floor rate. This is a backstop, not the
+# failure detector — a link that has actually died is caught within
+# ``socket_timeout`` by the blocking ``sendall``, long before this fires.
+_UPLOAD_FLOOR_BYTES_PER_SEC = 25 * 1024
+_UPLOAD_MIN_TIMEOUT = 600.0
+
+# How long to give the worker thread to notice the cancel flag, unwind, and
+# delete its partial file. It checks the flag once per CHUNK_SIZE, so on a link
+# slow enough to have hit the deadline this is one chunk plus the delete.
+_UPLOAD_CANCEL_GRACE = 60.0
+
+
+class UploadCancelled(Exception):
+    """Raised inside the upload worker to abort an in-flight transfer.
+
+    ``upload_file`` treats any exception from its progress callback as "stop
+    now": it breaks out of the send loop, deletes the partial file from the
+    printer, and re-raises. That is the only way to stop a transfer — an
+    executor thread cannot be cancelled from the event loop, so a bare
+    ``asyncio.wait_for`` leaves it streaming (see ``upload_file_async``).
+    """
+
 
 class DeleteResult(Enum):
     """Outcome of an FTP delete attempt.
@@ -980,12 +1007,45 @@ async def download_file_try_paths_async(
     return await loop.run_in_executor(None, _download)
 
 
+def _upload_deadline(local_path: Path) -> float:
+    """Derive an upload deadline from the file size (#2529).
+
+    See ``_UPLOAD_FLOOR_BYTES_PER_SEC``. An unstat-able file falls back to the
+    floor timeout — ``upload_file`` will fail on the open() anyway.
+    """
+    try:
+        size = local_path.stat().st_size
+    except OSError:
+        return _UPLOAD_MIN_TIMEOUT
+    return max(_UPLOAD_MIN_TIMEOUT, size / _UPLOAD_FLOOR_BYTES_PER_SEC)
+
+
+# One upload at a time per printer. Two concurrent STOR commands for the same
+# remote path leave a corrupt file on the SD card, and the printer reads as
+# flaky rather than busy (#2529). Held for the duration of a transfer, so a
+# second dispatch to the same printer queues behind the first instead of racing
+# it. Keyed per event loop: an asyncio.Lock binds to the loop that first awaits
+# it, and the test suite runs each case on a fresh loop.
+_upload_locks: weakref.WeakKeyDictionary[asyncio.AbstractEventLoop, dict[str, asyncio.Lock]] = (
+    weakref.WeakKeyDictionary()
+)
+
+
+def _upload_lock(loop: asyncio.AbstractEventLoop, ip_address: str) -> asyncio.Lock:
+    per_loop = _upload_locks.setdefault(loop, {})
+    lock = per_loop.get(ip_address)
+    if lock is None:
+        lock = asyncio.Lock()
+        per_loop[ip_address] = lock
+    return lock
+
+
 async def upload_file_async(
     ip_address: str,
     access_code: str,
     local_path: Path,
     remote_path: str,
-    timeout: float = 600.0,
+    timeout: float | None = None,
     progress_callback: Callable[[int, int], None] | None = None,
     socket_timeout: float | None = None,
     printer_model: str | None = None,
@@ -1000,19 +1060,31 @@ async def upload_file_async(
         access_code: Printer access code
         local_path: Local file path to upload
         remote_path: Remote path on printer
-        timeout: Overall operation timeout (asyncio)
+        timeout: Overall deadline. ``None`` (the default) derives it from the
+            file size — see ``_upload_deadline``. A caller that passes a number
+            gets exactly that, which is what the tests rely on.
         progress_callback: Optional callback for progress updates
         socket_timeout: FTP socket timeout for slow connections (e.g., A1 printers)
         printer_model: Printer model for A1-specific workarounds
     """
     loop = asyncio.get_event_loop()
     is_a1 = printer_model in BambuFTPClient.A1_MODELS if printer_model else False
+    deadline = _upload_deadline(local_path) if timeout is None else timeout
+
+    # Set when the deadline expires. The worker checks it once per chunk.
+    cancel = threading.Event()
+
+    def _guarded_progress(uploaded: int, total: int) -> None:
+        if cancel.is_set():
+            raise UploadCancelled(f"upload of {remote_path} exceeded its {deadline:.0f}s deadline")
+        if progress_callback:
+            progress_callback(uploaded, total)
 
     def _upload(force_prot_c: bool = False) -> bool:
         mode_str = "prot_c" if force_prot_c else "prot_p"
         logger.info(
             f"FTP connecting to {ip_address} for upload (model={printer_model}, "
-            f"mode={mode_str}, socket_timeout={socket_timeout}s)..."
+            f"mode={mode_str}, socket_timeout={socket_timeout}s, deadline={deadline:.0f}s)..."
         )
         client = BambuFTPClient(
             ip_address, access_code, timeout=socket_timeout, printer_model=printer_model, force_prot_c=force_prot_c
@@ -1020,7 +1092,7 @@ async def upload_file_async(
         if client.connect():
             logger.info("FTP connected to %s", ip_address)
             try:
-                result = client.upload_file(local_path, remote_path, progress_callback)
+                result = client.upload_file(local_path, remote_path, _guarded_progress)
                 if result:
                     # Cache the working mode
                     BambuFTPClient.cache_mode(ip_address, mode_str)
@@ -1030,32 +1102,80 @@ async def upload_file_async(
         logger.warning("FTP connection failed to %s", ip_address)
         return False
 
-    try:
+    async def _attempt(force_prot_c: bool) -> bool:
+        """Run one upload attempt, and make a timeout actually stop the transfer.
+
+        ``asyncio.wait_for`` cancels the *future*, never the executor thread
+        behind it. Before #2529 a slow-but-healthy upload that overran the
+        deadline left that thread streaming: it kept pushing bytes, kept firing
+        the progress callback, and the retry above put a *second* STOR of the
+        same file onto the same printer. The reporter's 96 MB job ran four
+        concurrent transfers and never landed. So on timeout we signal the
+        worker (it raises ``UploadCancelled`` from the progress callback, which
+        breaks the send loop and deletes the partial file) and wait for it to
+        actually go.
+        """
+        fut = loop.run_in_executor(None, lambda: _upload(force_prot_c))
+        try:
+            return await asyncio.wait_for(asyncio.shield(fut), timeout=deadline)
+        except TimeoutError:
+            cancel.set()
+            logger.warning(
+                "FTP upload of %s exceeded its %.0fs deadline — cancelling the transfer",
+                remote_path,
+                deadline,
+            )
+            try:
+                await asyncio.wait_for(asyncio.shield(fut), timeout=_UPLOAD_CANCEL_GRACE)
+            except UploadCancelled:
+                logger.info("FTP upload of %s cancelled; partial file removed from the printer", remote_path)
+            except TimeoutError:
+                # The thread is wedged somewhere that never reaches the callback
+                # (a blocked sendall, say). Nothing more we can do from here —
+                # but consume the eventual result so asyncio doesn't log the
+                # future's exception as unretrieved when it is garbage-collected.
+                logger.error(
+                    "FTP upload thread for %s did not stop within %.0fs of the cancel signal",
+                    remote_path,
+                    _UPLOAD_CANCEL_GRACE,
+                )
+                fut.add_done_callback(_swallow_future_result)
+            except Exception as e:
+                logger.warning("FTP upload of %s errored while cancelling: %s", remote_path, e)
+            # Raise rather than return False: a deadline expiry means the link
+            # sustained less than the floor rate for the whole transfer, and a
+            # retry would only spend another full deadline finding that out
+            # again — with check_queue serialized, four of those block the
+            # entire print queue for hours. ``with_ftp_retry`` never retries it.
+            raise UploadCancelled(
+                f"Upload of {remote_path} to {ip_address} exceeded its {deadline:.0f}s deadline "
+                f"(link sustained less than {_UPLOAD_FLOOR_BYTES_PER_SEC // 1024} KB/s)"
+            ) from None
+
+    async with _upload_lock(loop, ip_address):
         # Check if we have a cached mode for this printer
         cached_mode = BambuFTPClient._mode_cache.get(ip_address)
 
         if cached_mode:
             # Use cached mode
-            force_prot_c = cached_mode == "prot_c"
-            return await asyncio.wait_for(loop.run_in_executor(None, lambda: _upload(force_prot_c)), timeout=timeout)
+            return await _attempt(cached_mode == "prot_c")
 
         # No cached mode - try prot_p first
-        result = await asyncio.wait_for(loop.run_in_executor(None, lambda: _upload(False)), timeout=timeout)
-
-        if result:
+        if await _attempt(False):
             return True
 
         # Upload failed - for A1 models, try prot_c fallback
         if is_a1:
             logger.info("FTP upload failed with prot_p for A1 model, trying prot_c fallback...")
-            result = await asyncio.wait_for(loop.run_in_executor(None, lambda: _upload(True)), timeout=timeout)
-            return result
+            return await _attempt(True)
 
         return False
 
-    except TimeoutError:
-        logger.warning("FTP upload timed out after %ss for %s", timeout, remote_path)
-        return False
+
+def _swallow_future_result(fut: asyncio.Future) -> None:
+    """Retrieve a future's exception so asyncio doesn't log it as unhandled."""
+    if not fut.cancelled():
+        fut.exception()
 
 
 async def list_files_async(
@@ -1213,6 +1333,10 @@ async def with_ftp_retry(
 
     Returns:
         Result of the operation, or None if all attempts fail
+
+    ``UploadCancelled`` is never retried, whatever the caller passes: it means
+    the transfer overran its size-derived deadline, so a retry would spend
+    another full deadline reaching the same conclusion (#2529).
     """
     last_error = None
 
@@ -1227,6 +1351,8 @@ async def with_ftp_retry(
             # Operation returned failure indicator
             if attempt > 0:
                 logger.info("%s attempt %s/%s returned failure", operation_name, attempt + 1, max_retries + 1)
+        except UploadCancelled:
+            raise
         except Exception as e:
             if non_retry_exceptions and isinstance(e, non_retry_exceptions):
                 raise

backend/app/services/print_scheduler.py

@@ -24,6 +24,7 @@ from backend.app.models.smart_plug import SmartPlug
 from backend.app.models.spool_assignment import SpoolAssignment
 from backend.app.models.spoolman_slot_assignment import SpoolmanSlotAssignment
 from backend.app.services.bambu_ftp import (
+    UploadCancelled,
     cache_3mf_download,
     delete_file_async,
     get_ftp_retry_settings,
@@ -2797,6 +2798,10 @@ class PrintScheduler:
 
         progress_bridge = _UploadProgressBridge(toast_uid, item.id)
 
+        # A deadline expiry gets its own message: "check your SD card" is the
+        # wrong advice for a link that was simply too slow to finish (#2529).
+        upload_error: str | None = None
+
         try:
             if ftp_retry_enabled:
                 uploaded = await with_ftp_retry(
@@ -2822,6 +2827,13 @@ class PrintScheduler:
                     printer_model=printer.model,
                     progress_callback=progress_bridge,
                 )
+        except UploadCancelled as e:
+            uploaded = False
+            upload_error = (
+                "Upload was too slow to finish and was cancelled. The printer's connection could not sustain "
+                "the transfer — check its Wi-Fi signal, or move it closer to the access point."
+            )
+            logger.error("Queue item %s: upload deadline exceeded: %s", item.id, e)
         except Exception as e:
             uploaded = False
             logger.error("Queue item %s: FTP error: %s (type: %s)", item.id, e, type(e).__name__)
@@ -2831,7 +2843,7 @@ class PrintScheduler:
             injected_path.unlink(missing_ok=True)
 
         if not uploaded:
-            error_msg = (
+            error_msg = upload_error or (
                 "Failed to upload file to printer. Check if SD card is inserted and properly formatted (FAT32/exFAT). "
                 "See server logs for detailed diagnostics."
             )

backend/tests/unit/services/test_bambu_ftp.py

@@ -13,11 +13,14 @@ Tests against a real mock implicit FTPS server, covering:
 - Failure injection scenarios (regressions for 0.1.8 bugs)
 """
 
+import asyncio
+import threading
 import time
 from pathlib import Path
 
 import pytest
 
+from backend.app.services import bambu_ftp
 from backend.app.services.bambu_ftp import (
     BambuFTPClient,
     FileNotOnPrinterError,
@@ -1422,3 +1425,193 @@ class TestThreeMFCache:
         assert archive_file.exists(), "archive 3mf must not be deleted by cache cleanup"
         assert library_file.exists(), "library 3mf must not be deleted by cache cleanup"
         assert not temp_file.exists(), "temp file should still be cleaned up"
+
+
+@pytest.fixture
+def slow_upload_client(monkeypatch):
+    """Replace BambuFTPClient with a fake whose upload streams slowly.
+
+    Mirrors the real client's contract for the bits that matter here: it fires
+    the progress callback once per chunk and treats a callback exception as
+    "stop now" — break out of the send loop, drop the partial file, re-raise.
+    The returned dict lets a test see what the worker thread actually did,
+    which is the whole point: the #2529 ghost transfer was invisible from the
+    event loop's side.
+    """
+    state = {
+        "attempts": 0,
+        "concurrent": 0,
+        "max_concurrent": 0,
+        "completed": False,
+        "cancelled": False,
+        "deleted": [],
+        "chunks": 20,
+        "chunk_delay": 0.05,
+    }
+    lock = threading.Lock()
+
+    class FakeClient:
+        def __init__(self, *args, **kwargs):
+            pass
+
+        def connect(self):
+            return True
+
+        def upload_file(self, local_path, remote_path, progress_callback=None):
+            with lock:
+                state["attempts"] += 1
+                state["concurrent"] += 1
+                state["max_concurrent"] = max(state["max_concurrent"], state["concurrent"])
+            try:
+                total = state["chunks"]
+                for sent in range(1, total + 1):
+                    time.sleep(state["chunk_delay"])
+                    if progress_callback:
+                        try:
+                            progress_callback(sent, total)
+                        except Exception:
+                            state["cancelled"] = True
+                            state["deleted"].append(remote_path)
+                            raise
+                state["completed"] = True
+                return True
+            finally:
+                with lock:
+                    state["concurrent"] -= 1
+
+        def disconnect(self):
+            pass
+
+    monkeypatch.setattr(bambu_ftp, "BambuFTPClient", FakeClient)
+    monkeypatch.setattr(FakeClient, "_mode_cache", {}, raising=False)
+    monkeypatch.setattr(FakeClient, "A1_MODELS", ("A1", "A1 Mini"), raising=False)
+    monkeypatch.setattr(FakeClient, "cache_mode", staticmethod(lambda ip, mode: None), raising=False)
+    return state
+
+
+# ---------------------------------------------------------------------------
+# TestUploadDeadline (#2529)
+# ---------------------------------------------------------------------------
+class TestUploadDeadline:
+    """The upload deadline must be size-aware, and must actually stop the transfer.
+
+    Regression for #2529: a 96 MB 3MF to an A1 over WiFi sustains ~75 KB/s and
+    needs ~20 minutes. The old flat 600 s wall-clock cap declared it dead at
+    ~70 MB, `asyncio.wait_for` cancelled the *future* but not the executor
+    thread — which kept streaming — and `with_ftp_retry` then started a second
+    STOR of the same file onto the same printer. The reporter's video shows two
+    transfers of the same job climbing in parallel (2% and 72%), and the print
+    never landed.
+    """
+
+    def test_deadline_scales_with_file_size(self, tmp_path):
+        """A big file gets proportionally longer, a small one gets the floor."""
+        small = tmp_path / "small.3mf"
+        small.write_bytes(b"x" * 1024)
+        assert bambu_ftp._upload_deadline(small) == bambu_ftp._UPLOAD_MIN_TIMEOUT
+
+        # The reporter's file. At the 25 KB/s floor rate, 96 MB is ~64 minutes —
+        # far above the 600 s that killed it at 72%.
+        big = tmp_path / "big.3mf"
+        big.write_bytes(b"x" * (96 * 1024 * 1024))
+        deadline = bambu_ftp._upload_deadline(big)
+        assert deadline > bambu_ftp._UPLOAD_MIN_TIMEOUT
+        assert deadline == pytest.approx((96 * 1024 * 1024) / bambu_ftp._UPLOAD_FLOOR_BYTES_PER_SEC)
+
+    def test_deadline_falls_back_to_floor_for_unstatable_file(self, tmp_path):
+        assert bambu_ftp._upload_deadline(tmp_path / "nope.3mf") == bambu_ftp._UPLOAD_MIN_TIMEOUT
+
+    @pytest.mark.asyncio
+    async def test_timeout_stops_the_worker_thread(self, tmp_path, monkeypatch, slow_upload_client):
+        """The transfer stops when the deadline expires, instead of streaming on.
+
+        Mutation check: drop the `cancel.set()` in upload_file_async and the
+        worker runs to completion, which is exactly the ghost transfer #2529
+        reported.
+        """
+        state = slow_upload_client
+        local = tmp_path / "slow.3mf"
+        local.write_bytes(b"x" * 4096)
+
+        with pytest.raises(bambu_ftp.UploadCancelled):
+            await upload_file_async("127.0.0.1", "12345678", local, "/cache/slow.3mf", timeout=0.2, printer_model="X1C")
+
+        # The worker noticed the cancel and unwound — it did not run to the end.
+        await asyncio.sleep(0.5)
+        assert state["cancelled"] is True
+        assert state["completed"] is False
+        # And it cleaned the partial file off the printer on its way out.
+        assert state["deleted"] == ["/cache/slow.3mf"]
+
+    @pytest.mark.asyncio
+    async def test_timeout_is_not_retried(self, tmp_path, monkeypatch, slow_upload_client):
+        """with_ftp_retry must not start a second transfer after a deadline expiry.
+
+        This is the bug the reporter filmed: attempt 2 began while attempt 1 was
+        still sending. One attempt, then a hard failure.
+        """
+        state = slow_upload_client
+        local = tmp_path / "slow.3mf"
+        local.write_bytes(b"x" * 4096)
+
+        with pytest.raises(bambu_ftp.UploadCancelled):
+            await with_ftp_retry(
+                upload_file_async,
+                "127.0.0.1",
+                "12345678",
+                local,
+                "/cache/slow.3mf",
+                timeout=0.2,
+                printer_model="X1C",
+                max_retries=3,
+                retry_delay=0,
+            )
+
+        assert state["attempts"] == 1, "a timed-out upload must not be retried"
+
+    @pytest.mark.asyncio
+    async def test_uploads_to_one_printer_are_serialized(self, tmp_path, monkeypatch, slow_upload_client):
+        """Two dispatches to the same printer queue up; they never overlap.
+
+        Concurrent STORs of the same remote path leave a corrupt file on the SD
+        card and make the printer look like it has a flaky network.
+        """
+        state = slow_upload_client
+        state["chunk_delay"] = 0.05
+        local = tmp_path / "slow.3mf"
+        local.write_bytes(b"x" * 4096)
+
+        async def _dispatch(name: str) -> bool:
+            return await upload_file_async(
+                "127.0.0.1", "12345678", local, f"/cache/{name}.3mf", timeout=30.0, printer_model="X1C"
+            )
+
+        results = await asyncio.gather(_dispatch("a"), _dispatch("b"))
+
+        assert results == [True, True]
+        assert state["attempts"] == 2
+        assert state["max_concurrent"] == 1, "two uploads ran against the same printer at once"
+
+    def test_progress_callback_raising_deletes_the_partial_file(self, ftp_client_factory, ftp_root, tmp_path):
+        """The cancel path in the real client removes what it already wrote.
+
+        This is the mechanism the deadline now hangs off, exercised end to end
+        against the mock FTPS server rather than a fake.
+        """
+        client = ftp_client_factory()
+        assert client.connect() is True
+        try:
+            local = tmp_path / "cancelme.3mf"
+            # Two chunks, so the callback fires while there is a partial file.
+            local.write_bytes(b"x" * (BambuFTPClient.CHUNK_SIZE * 2))
+
+            def _stop_after_first_chunk(uploaded: int, total: int) -> None:
+                raise bambu_ftp.UploadCancelled("stop")
+
+            with pytest.raises(bambu_ftp.UploadCancelled):
+                client.upload_file(local, "/cancelme.3mf", _stop_after_first_chunk)
+        finally:
+            client.disconnect()
+
+        time.sleep(_UPLOAD_FLUSH_DELAY)
+        assert not (Path(ftp_root) / "cancelme.3mf").exists(), "partial file left on the printer"