Refuse a same-named 3MF that contradicts the running print (issue #2957)

    When a print's own 3MF cannot be fetched the usage tracker borrows one from the
    library or a previous archive, matching on the filename stem. That is far weaker
    evidence than it looks: Bambu Studio writes the printer-side filename from the
    project's Title metadata, so every plate of a project reaches the printer under
    one name however the file was renamed on disk.

    The reporter's single-filament job was handed a previous archive's three-filament
    plate. Three spools were debited for material that was never extruded, and
    nothing on the archive said the numbers were someone else's.

    A candidate is now rejected when it positively contradicts the print - a
    different plate, or a filament count the slicer's ams_mapping disagrees with -
    and the accepted one is logged with both expectations. Only on a contradiction:
    the plate needs firmware that echoes it and the count needs a print command
    Bambuddy saw, and refusing everything uncorroborated would retire the fallback
    recovery this same issue asked for.

    The count is scoped to one plate or not compared at all. Unscoped, the filament
    reader collects every <filament> in the file, and that sum against one plate's
    count would reject every multi-plate library upload on exactly the firmwares
    that cannot tell us the plate.

    -----

    Give a download the time the file needs, and one printer at a time (issue #2957)

    ftp_timeout is handed to every download as both the socket inactivity timeout
    and the whole-transfer deadline, which makes its 30s default a cap on how big a
    file a printer may serve. The reporter measured one 5.4 MB 3MF at 45s off a worn
    P1S SD card and 25s off a new one, and a 15.15 MB 3MF at 105s. None of those
    links were broken - they were slow, which is what the inactivity timeout exists
    to tell apart.

    download_to_file already asks for SIZE. It now reports it, and the total
    deadline follows the file at the 25 KB/s floor _upload_deadline has used since
    do, so #2572's cap on the executor queue wait is untouched. Capped at 300s for a
    reason that is not about FTP: on_print_start holds a pooled DB connection across
    its whole 3MF hunt.

    Downloads also take turns per printer now. He watched Bambu Studio lose its own
    connection while Bambuddy pulled a 12 MB 3MF, and a later log caught two
    Bambuddy downloads of the same file overlapping at print start. The gate is
    soft - whoever cannot have it within 30s goes anyway, because a print losing its
    3MF to queueing is worse than the contention, and a soft gate cannot deadlock.

    It is also meaningful for the first time. The 90s cap on a multi-path lookup
    returned while its worker kept walking the remaining paths, still on the
    printer's socket; that walk is now cancelled and waited out before the printer
    is handed on.

    -----

    Look in the shared 3MF cache again before each cover retry (issue #2957)

    The cover endpoint and the print-start archive flow share a cache so whichever
    fetches the 3MF first hands it to the other (#972). The cover consulted it once
    on the way in, then retried for up to two and a half minutes without looking
    again.

    In the reporter's log the archive flow published the file 42 seconds into that
    sequence and the cover's third attempt still pulled its own 5,250,969-byte copy,
    off a printer that was mid-print on the same SD card.

    A file picked up that way is left alone rather than re-registered under this
    endpoint's own name or deleted on the way out. It is the archive flow's.

backend/app/api/routes/printers.py

@@ -1320,16 +1320,23 @@ async def _produce_cover_image(
     # trip through subtask_name (#2856).
     downloaded = False
     using_cached = False
-    for candidate_name in (*possible_filenames, storage.probe_filename):
-        if not candidate_name:
-            continue
-        cached = get_cached_3mf(printer_id, candidate_name)
-        if cached:
-            logger.info("Cover using cached 3MF from %s (avoided duplicate FTP)", cached)
-            temp_path = cached
-            downloaded = True
-            using_cached = True
-            break
+
+    def _cached_source() -> Path | None:
+        """The 3MF another flow has already published for this print, if any."""
+        for candidate_name in (*possible_filenames, storage.probe_filename):
+            if not candidate_name:
+                continue
+            cached = get_cached_3mf(printer_id, candidate_name)
+            if cached:
+                return cached
+        return None
+
+    cached = _cached_source()
+    if cached:
+        logger.info("Cover using cached 3MF from %s (avoided duplicate FTP)", cached)
+        temp_path = cached
+        downloaded = True
+        using_cached = True
 
     if not downloaded:
         # Same idea, one step further back: that in-memory cache dies with the
@@ -1383,6 +1390,26 @@ async def _produce_cover_image(
         last_error = None
 
         for attempt in range(max_retries + 1):
+            if attempt:
+                # Look again before spending another transfer. The entry check
+                # above only settles the race when the two flows do not
+                # overlap, and on a P1S at print start they overlap for
+                # minutes: a reported run had the archive flow publish the file
+                # 42 seconds into this endpoint's 2.5-minute retry sequence,
+                # and the third attempt still pulled its own 5 MB copy of it
+                # over the same socket the printer was serving the print from
+                # (#2957).
+                cached = _cached_source()
+                if cached:
+                    logger.info(
+                        "Cover picked up the 3MF another flow finished downloading (%s) — skipping retry %s",
+                        cached,
+                        attempt + 1,
+                    )
+                    temp_path = cached
+                    downloaded = True
+                    using_cached = True
+                    break
             if ftps_handshake_blocked(printer.ip_address):
                 # Nothing to retry: the printer is not completing a TLS
                 # handshake on port 990, so no path and no attempt reaches it
@@ -1432,8 +1459,11 @@ async def _produce_cover_image(
                 f"Could not download 3MF file for '{subtask_name}' from printer {printer.ip_address}. Tried: {possible_filenames}",
             )
 
-        # Share the fresh download with the archive flow.
-        cache_3mf_download(printer_id, temp_filename, temp_path)
+        # Share the fresh download with the archive flow — unless the file is
+        # already theirs, in which case re-registering it under this endpoint's
+        # own name would only add a second key pointing at the same bytes.
+        if not using_cached:
+            cache_3mf_download(printer_id, temp_filename, temp_path)
 
     # Verify file actually exists and has content
     if not temp_path.exists():

backend/app/services/archive.py

@@ -92,6 +92,25 @@ def _read_plate_index(plate) -> int | None:
     return None
 
 
+def plate_indexes_in_3mf(file_path: Path) -> list[int | None]:
+    """Return one entry per ``<plate>`` a Bambu 3MF declares, in file order.
+
+    Reads only ``Metadata/slice_info.config``. An entry is None when that plate
+    carries no readable index, and the list is empty for a file that could not
+    be read at all — callers must not confuse either with "this file has plates
+    and yours is not among them". An unreadable 3MF is a parse this code does
+    not understand, not evidence about which plate it holds (#2957).
+    """
+    try:
+        with zipfile.ZipFile(file_path, "r") as zf:
+            if "Metadata/slice_info.config" not in zf.namelist():
+                return []
+            root = ET.fromstring(zf.read("Metadata/slice_info.config").decode())
+            return [_read_plate_index(plate) for plate in root.findall(".//plate")]
+    except Exception:
+        return []
+
+
 def peek_plate_index_in_3mf(file_path: Path) -> int | None:
     """Return the plate index a single-plate Bambu 3MF represents, or None.
 
@@ -105,18 +124,8 @@ def peek_plate_index_in_3mf(file_path: Path) -> int | None:
     plate 1 out of such a file, declaring a mismatch against the plate that
     is really running, and discarding a perfectly good 3MF (#2522).
     """
-    try:
-        with zipfile.ZipFile(file_path, "r") as zf:
-            if "Metadata/slice_info.config" not in zf.namelist():
-                return None
-            content = zf.read("Metadata/slice_info.config").decode()
-            root = ET.fromstring(content)
-            plates = root.findall(".//plate")
-            if len(plates) != 1:
-                return None
-            return _read_plate_index(plates[0])
-    except Exception:
-        return None
+    plates = plate_indexes_in_3mf(file_path)
+    return plates[0] if len(plates) == 1 else None
 
 
 _PLATE_SUFFIX_RE = re.compile(r"^(.*?)(\s*-\s*Plate\s+|_plate_)(\d+)$", re.IGNORECASE)

backend/app/services/bambu_ftp.py

@@ -8,8 +8,9 @@ import ssl
 import threading
 import time
 import weakref
-from collections.abc import Awaitable, Callable
+from collections.abc import AsyncIterator, Awaitable, Callable
 from concurrent.futures import ThreadPoolExecutor
+from contextlib import asynccontextmanager
 from dataclasses import dataclass
 from enum import Enum
 from ftplib import FTP, FTP_TLS  # nosec B402
@@ -57,6 +58,147 @@ _ftp_executor = ThreadPoolExecutor(max_workers=_FTP_MAX_WORKERS, thread_name_pre
 _UPLOAD_FLOOR_BYTES_PER_SEC = 25 * 1024
 _UPLOAD_MIN_TIMEOUT = 600.0
 
+# The same idea for the other direction (#2957). ``ftp_timeout`` is handed to
+# every download as BOTH the socket inactivity timeout and the whole-transfer
+# deadline, so its 30 s default is a cap on how big a file the printer is
+# allowed to serve. A reporter measured the same 5.4 MB 3MF at 45 s off a worn
+# P1S SD card and 25 s off a new one, and a 15.15 MB 3MF at 105 s; a 7.8 MB
+# archive in his older logs survived only because it finished inside the retry
+# grace. None of those transfers were unhealthy -- they were slow, which is what
+# the inactivity timeout is for and what a total deadline cannot tell apart.
+#
+# So the total deadline follows the file instead, at the same pessimistic floor
+# rate the upload path uses. The extension is granted only once the printer has
+# answered SIZE, which it can only do from a running worker: the queue wait that
+# #2572's cap exists to bound is not lengthened by any of this.
+_DOWNLOAD_FLOOR_BYTES_PER_SEC = 25 * 1024
+
+# ...but not without a ceiling, and for a reason that has nothing to do with FTP:
+# ``on_print_start`` runs its whole 3MF hunt inside one ``async_session``, so
+# every second a download is allowed is a second a pooled DB connection is held
+# (the same coupling behind #2572's cap). 300 s is ~7.5 MB at the floor rate and
+# covers the reporter's 15.15 MB / 105 s measurement three times over, because
+# the floor is pessimistic by design and a real link is not that slow.
+_DOWNLOAD_MAX_TIMEOUT = 300.0
+
+
+def _download_extension(size: int | None, base_timeout: float) -> float:
+    """Extra seconds to allow a download the printer says is this big.
+
+    Zero when the size is unknown -- no SIZE reply means no transfer got under
+    way, so the base deadline stands and a printer that is not answering still
+    fails on schedule. Zero, too, once the base deadline is already the more
+    generous of the two: this only ever lengthens a deadline.
+    """
+    if not size or size <= 0:
+        return 0.0
+    return max(0.0, min(size / _DOWNLOAD_FLOOR_BYTES_PER_SEC, _DOWNLOAD_MAX_TIMEOUT) - base_timeout)
+
+
+# How long a download will wait for another one on the same printer to finish
+# before going ahead alongside it (#2957). A P1S at print start is already
+# serving the print off the same SD card and talking MQTT to the slicer, and a
+# reporter watched Bambu Studio itself lose its connection while Bambuddy pulled
+# a 12 MB 3MF -- with a second Bambuddy transfer for the same file running at
+# the same time. 30 s covers the transfer sizes that actually overlap at print
+# start -- the reporter's 5.4 MB 3MF took 25 s off a healthy SD card -- and the
+# wait is deliberately no longer, because the gate is contention relief and not
+# a correctness control. Whoever cannot have it goes anyway, exactly as every
+# download did before this existed: a print must never lose its 3MF to queueing,
+# and the caller's own deadline stays untouched either way.
+_DOWNLOAD_GATE_WAIT_SECONDS = 30.0
+
+# How long to wait for a cancelled path-walk worker to unwind before releasing
+# the printer to the next download. It checks the flag once per 8 KiB chunk, so
+# this is one chunk on a link slow enough to have blown the deadline.
+_DOWNLOAD_UNWIND_SECONDS = 30.0
+
+
+def _discard_worker_outcome(worker: asyncio.Future) -> None:
+    """Read a shielded worker's result so asyncio does not complain about it.
+
+    ``asyncio.wait_for`` cancels the shield, not the executor thread behind it,
+    so the worker future outlives the call and nobody is left to look at what it
+    raised. An unretrieved exception surfaces later as a loop-level
+    ``Future exception was never retrieved`` ERROR with a traceback, logged
+    after the caller has already reported the real failure -- the same class of
+    noise as #2968, and measurably reproducible with a transport error that
+    lands just after the cap. The value is genuinely unwanted here; only the
+    fact that something read it matters.
+    """
+
+    def _read(fut: asyncio.Future) -> None:
+        if fut.cancelled():
+            return
+        exc = fut.exception()
+        if exc is not None:
+            logger.debug("FTP path-walk worker failed after its caller gave up: %s", exc)
+
+    if worker.done():
+        _read(worker)
+    else:
+        worker.add_done_callback(_read)
+
+
+# One heavy download at a time per printer. Keyed per event loop for the same
+# reason ``_upload_locks`` is: an asyncio.Lock binds to the loop that first
+# awaits it, and the test suite runs each case on a fresh loop.
+_download_locks: weakref.WeakKeyDictionary[asyncio.AbstractEventLoop, dict[str, asyncio.Lock]] = (
+    weakref.WeakKeyDictionary()
+)
+
+
+def _download_lock(loop: asyncio.AbstractEventLoop, ip_address: str) -> asyncio.Lock:
+    per_loop = _download_locks.setdefault(loop, {})
+    lock = per_loop.get(ip_address)
+    if lock is None:
+        lock = asyncio.Lock()
+        per_loop[ip_address] = lock
+    return lock
+
+
+@asynccontextmanager
+async def _serialized_download(ip_address: str, what: str, *, enabled: bool = True) -> AsyncIterator[bool]:
+    """Hold this printer's download gate for the block, or go without it.
+
+    Yields whether the gate was actually held, which is what the tests assert on
+    -- from the outside a serialized download and a concurrent one differ only
+    in timing. The wait is its own budget rather than a slice of the caller's
+    transfer deadline: a print-start download that queued behind a thumbnail
+    would otherwise fail on a timer and leave the print with no archive, which
+    is a worse outcome than the contention this is here to relieve.
+
+    ``enabled=False`` skips the gate entirely, for the callers that documented
+    themselves as lock-free before this existed -- see ``serialize`` on
+    :func:`download_file_async`.
+    """
+    if not enabled:
+        yield False
+        return
+    loop = asyncio.get_event_loop()
+    lock = _download_lock(loop, ip_address)
+    started = loop.time()
+    held = False
+    try:
+        await asyncio.wait_for(lock.acquire(), timeout=_DOWNLOAD_GATE_WAIT_SECONDS)
+        held = True
+        waited = loop.time() - started
+        if waited > 1.0:
+            logger.info("Waited %.1fs for printer %s to finish its other download before %s", waited, ip_address, what)
+    except TimeoutError:
+        logger.warning(
+            "Printer %s is still busy with another download after %ss — starting %s alongside it",
+            ip_address,
+            _DOWNLOAD_GATE_WAIT_SECONDS,
+            what,
+        )
+    try:
+        yield held
+    finally:
+        if held:
+            lock.release()
+
+
 # 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.
@@ -86,6 +228,20 @@ class DownloadInsufficientSpace(Exception):
     """Raised before an FTP callback consumes the application's disk reserve."""
 
 
+class DownloadDeadlineExceeded(Exception):
+    """A transfer overran the deadline derived from the size the printer reported.
+
+    Never retried, for the reason ``UploadCancelled`` is not (#2529): the
+    deadline was already stretched to fit the file at a floor rate no working
+    link falls below, so another attempt would spend another full deadline
+    reaching the same conclusion -- and ``on_print_start`` spends it holding a
+    pooled database connection. Raised instead of returning False so
+    ``with_ftp_retry`` can tell this apart from an ordinary failed attempt
+    (#2957); callers that do not retry see it through their existing handlers,
+    which is why the archive flow advances to its next candidate path.
+    """
+
+
 @dataclass(frozen=True)
 class FileListResult:
     """A directory listing that distinguishes empty from unreachable."""
@@ -735,8 +891,15 @@ class BambuFTPClient:
         max_bytes: int | None = None,
         cancel_event: threading.Event | None = None,
         min_free_bytes: int | None = None,
+        size_callback: Callable[[int], None] | None = None,
     ) -> bool:
-        """Download a file with cooperative cancellation and byte bounds."""
+        """Download a file with cooperative cancellation and byte bounds.
+
+        ``size_callback`` is handed the size the printer reported for this file,
+        once, before the transfer starts. The async wrappers use it to grow a
+        whole-transfer deadline that was set before anyone knew how big the file
+        was (#2957); it must not raise.
+        """
         if not self._ftp:
             logger.warning("download_to_file called but FTP not connected")
             return False
@@ -757,6 +920,8 @@ class BambuFTPClient:
             if min_free_bytes is not None and authoritative_size is not None:
                 if shutil.disk_usage(local_path.parent).free < min_free_bytes + authoritative_size:
                     raise DownloadInsufficientSpace(remote_path)
+            if size_callback is not None and authoritative_size is not None and authoritative_size > 0:
+                size_callback(authoritative_size)
             with open(local_path, "wb") as f:
                 written = 0
                 # retrbinary hands over 8 KiB at a time, so checking the volume
@@ -1399,6 +1564,7 @@ async def download_file_async(
     max_bytes: int | None = None,
     cancel_event: threading.Event | None = None,
     min_free_bytes: int | None = None,
+    serialize: bool = True,
 ) -> bool:
     """Async wrapper for downloading a file with timeout.
 
@@ -1420,6 +1586,12 @@ async def download_file_async(
         timeout: Overall operation timeout (asyncio)
         socket_timeout: FTP socket timeout for slow connections (e.g., A1 printers)
         printer_model: Printer model for A1-specific workarounds
+        serialize: take this printer's download gate for the transfer (#2957).
+            Pass False from a path that must neither queue behind another
+            download nor make one queue behind it -- the printer file browser
+            is both, and says so: a preview must not wait out somebody else's
+            ten-gigabyte selection, and that selection must not hold the printer
+            for the twenty minutes it legitimately takes.
     """
     loop = asyncio.get_event_loop()
     is_a1 = printer_model in BambuFTPClient.A1_MODELS if printer_model else False
@@ -1468,6 +1640,7 @@ async def download_file_async(
                         max_bytes=max_bytes,
                         cancel_event=combined_cancel,
                         min_free_bytes=min_free_bytes,
+                        size_callback=lambda n: completion.__setitem__("size", n),
                     )
                     if result:
                         BambuFTPClient.cache_mode(ip_address, mode_str)
@@ -1484,8 +1657,32 @@ async def download_file_async(
         done = threading.Event()
         attempt_cancel = threading.Event()
         worker = loop.run_in_executor(_ftp_executor, _download, force_prot_c, completion, done, attempt_cancel)
+        # What this attempt was actually allowed, for the log lines below: the
+        # size-derived extension moves it after the fact.
+        allowed = timeout
+        extended = False
         try:
-            return await asyncio.wait_for(asyncio.shield(worker), timeout=timeout)
+            try:
+                return await asyncio.wait_for(asyncio.shield(worker), timeout=timeout)
+            except TimeoutError:
+                # The deadline was set before anyone knew the file's size. Now
+                # the printer has told us, so give a transfer that is genuinely
+                # under way the time that size needs (#2957). Re-raises into the
+                # handler below when the size is unknown or already covered.
+                extension = _download_extension(completion.get("size"), timeout)
+                if extension <= 0:
+                    raise
+                logger.info(
+                    "FTP download of %s passed its %ss deadline but the printer reports %s bytes — "
+                    "allowing %.0fs more rather than declaring a slow transfer dead (#2957)",
+                    remote_path,
+                    timeout,
+                    completion.get("size"),
+                    extension,
+                )
+                allowed = timeout + extension
+                extended = True
+                return await asyncio.wait_for(asyncio.shield(worker), timeout=extension)
         except asyncio.CancelledError:
             # Cancelling an asyncio Future cannot stop its executor thread. Set
             # the callback-visible flag and do not let the caller unlink the
@@ -1526,15 +1723,26 @@ async def download_file_async(
             if completion["success"] and local_path.exists() and local_path.stat().st_size > 0:
                 logger.info(
                     "FTP download wait_for timed out after %ss for %s, but thread completed within %ss grace (%s bytes) — salvaging",
-                    timeout,
+                    allowed,
                     remote_path,
                     grace,
                     local_path.stat().st_size,
                 )
                 return True
+            if extended:
+                # The transfer had already been given the time its own reported
+                # size needs. Retrying spends that again to learn the same
+                # thing, so say so rather than reporting an ordinary miss.
+                logger.warning(
+                    "FTP download of %s did not finish inside the %ss its size bought it (plus %ss grace)",
+                    remote_path,
+                    allowed,
+                    grace,
+                )
+                raise DownloadDeadlineExceeded(remote_path)
             logger.warning(
                 "FTP download timed out after %ss (plus %ss grace) for %s",
-                timeout,
+                allowed,
                 grace,
                 remote_path,
             )
@@ -1543,20 +1751,24 @@ async def download_file_async(
     # Check if we have a cached mode for this printer
     cached_mode = BambuFTPClient._mode_cache.get(ip_address)
 
-    if cached_mode:
-        force_prot_c = cached_mode == "prot_c"
-        return await _run(force_prot_c)
+    # The gate spans the prot_c fallback too: those are two attempts at one
+    # transfer, and letting go between them would hand the printer to a waiter
+    # mid-download (#2957).
+    async with _serialized_download(ip_address, f"a download of {remote_path}", enabled=serialize):
+        if cached_mode:
+            force_prot_c = cached_mode == "prot_c"
+            return await _run(force_prot_c)
 
-    # No cached mode - try prot_p first
-    if await _run(False):
-        return True
+        # No cached mode - try prot_p first
+        if await _run(False):
+            return True
 
-    # Download failed - for A1 models, try prot_c fallback
-    if is_a1:
-        logger.info("FTP download failed with prot_p for A1 model, trying prot_c fallback...")
-        return await _run(True)
+        # Download failed - for A1 models, try prot_c fallback
+        if is_a1:
+            logger.info("FTP download failed with prot_p for A1 model, trying prot_c fallback...")
+            return await _run(True)
 
-    return False
+        return False
 
 
 async def download_file_try_paths_async(
@@ -1587,34 +1799,67 @@ async def download_file_try_paths_async(
             connects, that queue wait is otherwise unbounded — and any caller
             holding a DB connection while awaiting this would pin it until the
             pool is exhausted (#2572). The cap converts that into a bounded
-            wait; the orphaned worker finishes and its result is discarded.
+            wait; the worker is then cancelled and waited out rather than
+            orphaned (#2957), so a DB-holding caller's worst case is the gate
+            wait plus this cap plus one unwind -- still bounded, and the
+            orphaned worker no longer keeps the printer's socket after it.
     """
     loop = asyncio.get_event_loop()
+    # An executor thread cannot be cancelled, so the cap alone used to leave a
+    # worker walking the remaining paths -- still holding the printer's FTP
+    # socket -- long after this coroutine had given up on it. A reporter's log
+    # shows one of those still going as the archive flow's own download landed,
+    # two Bambuddy transfers deep into a P1S that was mid-print (#2957). The
+    # flag stops it at the next chunk instead.
+    cancel = threading.Event()
+    done = threading.Event()
 
     def _download():
-        client = BambuFTPClient(ip_address, access_code, timeout=socket_timeout, printer_model=printer_model)
-        if not client.connect():
-            return None
-
         try:
-            # FileNotOnPrinterError signals "try the next path", not "give up" —
-            # this function's whole purpose is to walk a list of candidates
-            # over one connection. Only a real transport error should bubble.
-            for remote_path in remote_paths:
-                try:
-                    if client.download_to_file(remote_path, local_path):
-                        return remote_path
-                except FileNotOnPrinterError:
-                    continue
-            return None
+            client = BambuFTPClient(ip_address, access_code, timeout=socket_timeout, printer_model=printer_model)
+            if not client.connect():
+                return None
+
+            try:
+                # FileNotOnPrinterError signals "try the next path", not "give up" —
+                # this function's whole purpose is to walk a list of candidates
+                # over one connection. Only a real transport error should bubble.
+                for remote_path in remote_paths:
+                    if cancel.is_set():
+                        return None
+                    try:
+                        if client.download_to_file(remote_path, local_path, cancel_event=cancel):
+                            return remote_path
+                    except FileNotOnPrinterError:
+                        continue
+                    except DownloadCancelled:
+                        return None
+                return None
+            finally:
+                client.disconnect()
         finally:
-            client.disconnect()
+            done.set()
 
-    try:
-        return await asyncio.wait_for(loop.run_in_executor(_ftp_executor, _download), timeout=timeout)
-    except TimeoutError:
-        logger.warning("FTP download_try_paths exceeded its %ss cap for %s (#2572)", timeout, ip_address)
-        return None
+    async with _serialized_download(ip_address, f"a {len(remote_paths)}-path lookup"):
+        worker = loop.run_in_executor(_ftp_executor, _download)
+        try:
+            return await asyncio.wait_for(asyncio.shield(worker), timeout=timeout)
+        except asyncio.CancelledError:
+            # The caller is going away and should not be made to wait, but the
+            # worker must not keep the printer to itself either.
+            cancel.set()
+            _discard_worker_outcome(worker)
+            raise
+        except TimeoutError:
+            logger.warning("FTP download_try_paths exceeded its %ss cap for %s (#2572)", timeout, ip_address)
+            cancel.set()
+            # Do not hand the printer to the next download while this worker is
+            # still on its socket. The DEFAULT executor, never ``_ftp_executor``:
+            # parking a waiter in the same bounded pool as the worker it waits
+            # for is how a deadlock gets built.
+            await loop.run_in_executor(None, done.wait, _DOWNLOAD_UNWIND_SECONDS)
+            _discard_worker_outcome(worker)
+            return None
 
 
 def _upload_deadline(local_path: Path) -> float:
@@ -2207,6 +2452,8 @@ async def with_ftp_retry(
     ``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).
+    ``DownloadDeadlineExceeded`` is the same thing in the other direction
+    (#2957) and is treated the same way.
     """
     last_error = None
     attempts_made = 0
@@ -2223,7 +2470,7 @@ 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:
+        except (UploadCancelled, DownloadDeadlineExceeded):
             raise
         except Exception as e:
             if non_retry_exceptions and isinstance(e, non_retry_exceptions):

backend/app/services/printer_media.py

@@ -383,6 +383,12 @@ async def build_printer_files_zip(
                         max_bytes=MAX_PRINTER_ZIP_BYTES - total_bytes,
                         cancel_event=cancel_signal,
                         min_free_bytes=PRINTER_ZIP_FREE_SPACE_RESERVE,
+                        # Outside the per-printer download gate (#2957), in both
+                        # directions. A selection of ~250 MB /ipcam chunks holds
+                        # the printer for as long as it legitimately takes, and
+                        # nothing else should be made to wait that out; equally,
+                        # each file here must not stall behind a thumbnail.
+                        serialize=False,
                     )
                     if not downloaded:
                         failed_paths.append(remote_path)
@@ -509,6 +515,9 @@ async def build_printer_file(
             max_bytes=MAX_PRINTER_ZIP_BYTES,
             cancel_event=cancel_signal,
             min_free_bytes=PRINTER_ZIP_FREE_SPACE_RESERVE,
+            # The lock-free promise in this function's docstring, kept: a preview
+            # must not queue behind somebody else's selection (#2957).
+            serialize=False,
         )
         if not downloaded:
             raise FileNotFoundError("The selected printer file could not be downloaded")

backend/app/services/usage_tracker.py

@@ -1067,11 +1067,76 @@ def _stem_matches(column, stem: str):
     return column.ilike(f"{escaped}.%", escape="\\") | column.ilike(f"%/{escaped}.%", escape="\\")
 
 
+def _expected_plate_for_print(plate_id: int | None, gcode_file: str | None) -> int | None:
+    """The plate a running print is on, from whatever was recorded about it.
+
+    ``plate_id`` is the reliable source, and the archives that need a donor 3MF
+    have none: the no-3MF fallback row is created before any 3MF is read, so
+    the column is never filled. The gcode path the printer echoed is the other
+    source, exact on the firmwares that echo ``Metadata/plate_N.gcode``. Some
+    P1S builds echo only the 3MF filename, and then the plate is simply not
+    knowable at print start (#2957).
+    """
+    from backend.app.services.printer_manager import parse_plate_id
+
+    if plate_id is not None:
+        return plate_id
+    return parse_plate_id(gcode_file)
+
+
+def _donor_3mf_conflicts(candidate, expected_plate: int | None) -> str | None:
+    """Why *candidate* cannot be this print's 3MF, or None if nothing rules it out.
+
+    A same-name 3MF is not the same print. Bambu Studio writes the printer-side
+    filename from the project's ``Title`` metadata, so every plate of a project
+    arrives under one name however the user renamed the file on disk, and a
+    donor chosen on the name alone hands one plate's slicer estimates to another
+    plate's print. A reporter's single-filament job was charged against three
+    spools that way, and nothing about the deduction said it was a guess
+    (#2957).
+
+    The plate is the one thing that can settle this. It is the same comparison
+    #1204 already makes against a freshly downloaded 3MF, so a single-plate
+    export is known to carry its original index rather than a renumbered 1.
+
+    Filament *count* deliberately is not checked, however tempting: the slicer's
+    ``ams_mapping`` is indexed by the project's filament slot -- see
+    ``slot_to_tray[slot_id - 1]`` below -- not by the plate's, so a real
+    single-filament print reports ``[0, -1, -1, -1]`` and its length says
+    nothing about how many filaments the plate uses.
+    """
+    from backend.app.services.archive import plate_indexes_in_3mf
+
+    if expected_plate is None:
+        return None
+
+    plates = plate_indexes_in_3mf(candidate)
+    if not plates or any(plate is None for plate in plates):
+        # Nothing was read, or not all of it was, and neither is evidence about
+        # the plate. Refusing here would drop the fallback for every 3MF variant
+        # this parser does not understand; downstream reports that honestly as
+        # "no filament usage data".
+        return None
+    if len(plates) == 1 and plates[0] != expected_plate:
+        return f"it holds plate {plates[0]}, this print is plate {expected_plate}"
+    if expected_plate not in plates:
+        # An all-plates export is a good donor precisely when it carries the
+        # plate that is running. Without this the plate is looked for
+        # downstream, found missing, and the whole file's filaments are summed
+        # onto one plate's print.
+        return f"it has no plate {expected_plate}"
+    return None
+
+
 async def _resolve_3mf_fallback(archive, db: AsyncSession, base_dir):
     """Try to find a 3MF file from library or a previous archive when the current archive has none.
 
     This handles fallback archives (FTP download failed) where the 3MF may already exist
     locally from a library upload or a previous successful print of the same file.
+
+    A name match alone does not make a candidate this print's file, so every
+    candidate is put through :func:`_donor_3mf_conflicts` before it is handed
+    back (#2957).
     """
     from pathlib import Path
 
@@ -1084,6 +1149,24 @@ async def _resolve_3mf_fallback(archive, db: AsyncSession, base_dir):
     if not search_base:
         return None
 
+    print_data = (getattr(archive, "extra_data", None) or {}).get("_print_data") or {}
+    expected_plate = _expected_plate_for_print(
+        getattr(archive, "plate_id", None),
+        archive.filename or print_data.get("filename"),
+    )
+    if expected_plate is None:
+        # Worth saying out loud. On the firmwares that echo only the 3MF
+        # filename there is nothing to check a donor against, so whatever is
+        # accepted below is accepted on its name alone -- which is how the
+        # reporter's spools were debited for another plate's filament. The
+        # deduction being silent was half the bug (#2957).
+        logger.warning(
+            "[UsageTracker] 3MF fallback: archive %s does not know its plate (%r), so a same-named "
+            "3MF can only be matched on its name",
+            archive.id,
+            archive.filename,
+        )
+
     # 1. Try library files matching the name (match base name at file boundary)
     try:
         lib_result = await db.execute(
@@ -1097,7 +1180,21 @@ async def _resolve_3mf_fallback(archive, db: AsyncSession, base_dir):
             lib_path = Path(lib_file.file_path)
             candidate = lib_path if lib_path.is_absolute() else base_dir / lib_file.file_path
             if candidate.exists() and candidate.suffix == ".3mf":
-                logger.info("[UsageTracker] 3MF fallback: found library file %s for archive %s", candidate, archive.id)
+                conflict = _donor_3mf_conflicts(candidate, expected_plate)
+                if conflict:
+                    logger.warning(
+                        "[UsageTracker] 3MF fallback: not using library file %s for archive %s — %s",
+                        candidate,
+                        archive.id,
+                        conflict,
+                    )
+                    continue
+                logger.info(
+                    "[UsageTracker] 3MF fallback: found library file %s for archive %s (expected plate=%s)",
+                    candidate,
+                    archive.id,
+                    expected_plate,
+                )
                 return candidate
     except Exception as e:
         logger.debug("[UsageTracker] 3MF fallback: library lookup failed: %s", e)
@@ -1117,10 +1214,20 @@ async def _resolve_3mf_fallback(archive, db: AsyncSession, base_dir):
         for prev_archive in prev_result.scalars().all():
             candidate = base_dir / prev_archive.file_path
             if candidate.exists() and candidate.suffix == ".3mf":
+                conflict = _donor_3mf_conflicts(candidate, expected_plate)
+                if conflict:
+                    logger.warning(
+                        "[UsageTracker] 3MF fallback: not using archive %s's file for archive %s — %s",
+                        prev_archive.id,
+                        archive.id,
+                        conflict,
+                    )
+                    continue
                 logger.info(
-                    "[UsageTracker] 3MF fallback: found previous archive %s file for archive %s",
+                    "[UsageTracker] 3MF fallback: found previous archive %s file for archive %s (expected plate=%s)",
                     prev_archive.id,
                     archive.id,
+                    expected_plate,
                 )
                 return candidate
     except Exception as e:
@@ -1142,7 +1249,9 @@ async def _find_3mf_by_filename(
     need the 3MF slicer data for filament usage tracking.
 
     ``print_name`` is the model name to fall back to when ``filename`` is the
-    printer's plate path, which names no model at all.
+    printer's plate path, which names no model at all -- and when it is that
+    plate path, it is also what keeps a same-named file for a different plate
+    from being adopted (#2957); see :func:`_donor_3mf_conflicts`.
     """
     from pathlib import Path
 
@@ -1153,6 +1262,8 @@ async def _find_3mf_by_filename(
     if not search_base:
         return None
 
+    expected_plate = _expected_plate_for_print(None, filename)
+
     # 1. Try library files matching the name
     try:
         lib_result = await db.execute(
@@ -1166,6 +1277,15 @@ async def _find_3mf_by_filename(
             lib_path = Path(lib_file.file_path)
             candidate = lib_path if lib_path.is_absolute() else base_dir / lib_file.file_path
             if candidate.exists() and candidate.suffix == ".3mf":
+                conflict = _donor_3mf_conflicts(candidate, expected_plate)
+                if conflict:
+                    logger.warning(
+                        "[UsageTracker] 3MF (no-archive): not using library file %s for '%s' — %s",
+                        candidate,
+                        filename,
+                        conflict,
+                    )
+                    continue
                 logger.info("[UsageTracker] 3MF (no-archive): found library file %s for '%s'", candidate, filename)
                 return candidate
     except Exception as e:
@@ -1185,6 +1305,15 @@ async def _find_3mf_by_filename(
         for prev_archive in prev_result.scalars().all():
             candidate = base_dir / prev_archive.file_path
             if candidate.exists() and candidate.suffix == ".3mf":
+                conflict = _donor_3mf_conflicts(candidate, expected_plate)
+                if conflict:
+                    logger.warning(
+                        "[UsageTracker] 3MF (no-archive): not using archive %s's file for '%s' — %s",
+                        prev_archive.id,
+                        filename,
+                        conflict,
+                    )
+                    continue
                 logger.info(
                     "[UsageTracker] 3MF (no-archive): found previous archive %s file for '%s'",
                     prev_archive.id,

backend/tests/unit/test_cover_rechecks_3mf_cache_2957.py

@@ -0,0 +1,168 @@
+"""The cover endpoint stops re-fetching a 3MF another flow already has (#2957).
+
+Both the cover endpoint and the print-start archive flow want the running
+print's 3MF, and #972 gave them a shared cache so whichever gets it first hands
+it to the other. The cover endpoint looked in that cache exactly once, on the
+way in, and then fell into a retry loop that never looked again.
+
+On a P1S the two flows overlap for minutes. The reporter's log has the cover
+request starting at 13:31:09, its first attempt burning the whole 90-second
+path-walk cap, the archive flow publishing the file to the cache at 13:32:47 --
+and the cover's third attempt pulling its own 5,250,969-byte copy of that same
+file at 13:33:29, off a printer that was mid-print on the same SD card.
+
+These tests pin the re-check: the file is picked up between attempts, the
+retries still happen when there is genuinely nothing to pick up, and a file that
+came from the cache is neither re-registered under this endpoint's own name nor
+deleted on the way out -- it belongs to the archive flow.
+"""
+
+from __future__ import annotations
+
+import zipfile
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+import pytest
+from fastapi import HTTPException
+
+import backend.app.api.routes.printers as printers_mod
+from backend.app.api.routes.printers import _produce_cover_image
+
+pytestmark = pytest.mark.asyncio
+
+SUBTASK = "bambu_lab_spool"
+COVER_BYTES = b"\x89PNG\r\n\x1a\nplate-1-thumbnail"
+
+
+def _write_3mf(path: Path) -> Path:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    with zipfile.ZipFile(path, "w") as zf:
+        zf.writestr("Metadata/plate_1.png", COVER_BYTES)
+    return path
+
+
+@pytest.fixture(autouse=True)
+def _clear_cover_state():
+    printers_mod._cover_cache.clear()
+    printers_mod._cover_404_cache.clear()
+    printers_mod._cover_inflight.clear()
+    yield
+    printers_mod._cover_cache.clear()
+    printers_mod._cover_404_cache.clear()
+    printers_mod._cover_inflight.clear()
+
+
+class _Harness:
+    """The cover endpoint with its FTP, storage verdict and cache faked out."""
+
+    def __init__(self, tmp_path: Path):
+        self.tmp_path = tmp_path
+        self.downloads = 0
+        self.cache: dict[str, Path] = {}
+        self.registered: list[tuple[int, str, Path]] = []
+        self.on_download = None
+        self.serves_the_file = False
+        self.printer = SimpleNamespace(id=1, ip_address="172.25.12.149", access_code="x", model="P1S", name="P1S")
+
+    def _get_cached(self, printer_id, name):
+        return self.cache.get("path")
+
+    async def _download(self, ip_address, access_code, remote_paths, local_path, **kwargs):
+        self.downloads += 1
+        if self.on_download is not None:
+            self.on_download(self)
+        if self.serves_the_file:
+            _write_3mf(local_path)
+            return remote_paths[0]
+        return None
+
+    async def run(self, **kwargs):
+        async def _no_recovery(printer_id, name, path):
+            return False
+
+        with (
+            patch.object(printers_mod.settings, "archive_dir", self.tmp_path / "archive"),
+            patch.object(printers_mod.printer_manager, "get_status", MagicMock(return_value=SimpleNamespace())),
+            patch.object(
+                printers_mod,
+                "print_file_reachable_over_ftp",
+                MagicMock(return_value=SimpleNamespace(reachable=True, probe_filename=None, reason="")),
+            ),
+            patch.object(printers_mod, "get_cached_3mf", self._get_cached),
+            patch.object(
+                printers_mod,
+                "cache_3mf_download",
+                lambda pid, name, path: self.registered.append((pid, name, path)),
+            ),
+            patch.object(printers_mod, "download_file_try_paths_async", self._download),
+            patch("backend.app.main.try_recover_fallback_archive", _no_recovery),
+            patch.object(printers_mod.asyncio, "sleep", lambda *_: _noop()),
+        ):
+            return await _produce_cover_image(
+                self.printer, 1, SUBTASK, None, "default", None, (SUBTASK, "default"), **kwargs
+            )
+
+
+async def _noop():
+    return None
+
+
+class TestItLooksAgainBetweenAttempts:
+    async def test_a_file_published_mid_retry_is_picked_up(self, tmp_path):
+        """The reported sequence: the archive flow finishes while this endpoint
+        is between retries, and the retry must not spend a second transfer."""
+        harness = _Harness(tmp_path)
+        source = _write_3mf(tmp_path / "archive" / "temp" / f"{SUBTASK}.gcode.3mf")
+
+        def publish(h):
+            h.cache["path"] = source  # the archive flow's download lands
+
+        harness.on_download = publish
+
+        assert await harness.run() == COVER_BYTES
+        assert harness.downloads == 1, "the cover re-downloaded a 3MF the cache already held"
+
+    async def test_the_cached_file_is_left_to_its_owner(self, tmp_path):
+        """It is the archive flow's temp file. Re-registering it under this
+        endpoint's own key would point the cache at bytes it does not own, and
+        deleting it would force the archive flow to fetch it again."""
+        harness = _Harness(tmp_path)
+        source = _write_3mf(tmp_path / "archive" / "temp" / f"{SUBTASK}.gcode.3mf")
+        harness.on_download = lambda h: h.cache.__setitem__("path", source)
+
+        await harness.run()
+
+        assert harness.registered == []
+        assert source.exists()
+
+
+class TestWhatItMustNotChange:
+    async def test_retries_still_run_when_there_is_nothing_to_pick_up(self, tmp_path):
+        """max_retries + 1 attempts, exactly as before -- the re-check must not
+        become an early exit for a printer that simply has not answered yet."""
+        harness = _Harness(tmp_path)
+
+        with pytest.raises(HTTPException) as exc:
+            await harness.run()
+
+        assert exc.value.status_code == 404
+        assert harness.downloads == 3
+
+    async def test_a_hit_on_the_way_in_still_skips_ftp_entirely(self, tmp_path):
+        harness = _Harness(tmp_path)
+        harness.cache["path"] = _write_3mf(tmp_path / "archive" / "temp" / f"{SUBTASK}.gcode.3mf")
+
+        assert await harness.run() == COVER_BYTES
+        assert harness.downloads == 0
+
+    async def test_its_own_download_is_still_shared(self, tmp_path):
+        """The other half of #972: a cover that really did fetch the bytes must
+        still publish them, or the archive flow refetches the same file."""
+        harness = _Harness(tmp_path)
+        harness.serves_the_file = True
+
+        assert await harness.run() == COVER_BYTES
+        assert harness.downloads == 1
+        assert [name for _, name, _ in harness.registered] == [f"{SUBTASK}.gcode.3mf"]

backend/tests/unit/test_donor_3mf_validation_2957.py

@@ -0,0 +1,227 @@
+"""A same-named 3MF is not automatically this print's 3MF (#2957).
+
+When a print's own 3MF could not be fetched, the usage tracker looks for one in
+the library or in a previous archive and matches on the filename stem. Bambu
+Studio writes the printer-side filename from the project's ``Title`` metadata,
+so every plate of a project reaches the printer under one name however the user
+renamed the file on disk -- filename equality says almost nothing.
+
+The reporter's archive 94 was handed archive 92's file that way. The real print
+used one filament; the donor plate declared three, and three spools were debited
+for material they never extruded. Nothing in the archive said the numbers were
+someone else's.
+
+The plate is the only sound discriminator available here, and these tests pin
+both halves of it: a donor holding a different plate is refused, and an
+all-plates export is refused unless it actually carries the plate that is
+running. The tempting second check -- comparing the donor's filament count
+against the slicer's ``ams_mapping`` -- is deliberately absent and has a test of
+its own saying why: that field is indexed by the *project's* filament slots, so
+a genuine single-filament print reports ``[0, -1, -1, -1]``.
+
+Where the plate cannot be known at all, which is the reporter's own firmware,
+the donor is still accepted on its name and a warning says so. That is the
+honest limit of what the data supports.
+"""
+
+from __future__ import annotations
+
+import zipfile
+from pathlib import Path
+
+import pytest
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from backend.app.models.archive import PrintArchive
+from backend.app.models.library import LibraryFile
+from backend.app.services.usage_tracker import (
+    _donor_3mf_conflicts,
+    _expected_plate_for_print,
+    _resolve_3mf_fallback,
+)
+
+
+def _write_3mf(path: Path, *, plate: int, filaments: int) -> Path:
+    """A single-plate 3MF declaring *filaments* filaments on plate *plate*."""
+    path.parent.mkdir(parents=True, exist_ok=True)
+    rows = "".join(
+        f"<filament id='{i + 1}' type='PLA' color='#00AE42' used_g='10' used_m='3.4'/>" for i in range(filaments)
+    )
+    with zipfile.ZipFile(path, "w") as zf:
+        zf.writestr(
+            "Metadata/slice_info.config",
+            "<?xml version='1.0' encoding='UTF-8'?><config><plate>"
+            f"<metadata key='index' value='{plate}'/>"
+            f"<metadata key='prediction' value='3600'/>{rows}"
+            "</plate></config>",
+        )
+    return path
+
+
+def _write_multiplate_3mf(path: Path, plates: dict[int, int]) -> Path:
+    """An all-plates export: ``{plate index: filament count}``."""
+    path.parent.mkdir(parents=True, exist_ok=True)
+    body = ""
+    for plate, filaments in plates.items():
+        rows = "".join(
+            f"<filament id='{i + 1}' type='PLA' color='#00AE42' used_g='10' used_m='3.4'/>" for i in range(filaments)
+        )
+        body += f"<plate><metadata key='index' value='{plate}'/><metadata key='prediction' value='60'/>{rows}</plate>"
+    with zipfile.ZipFile(path, "w") as zf:
+        zf.writestr("Metadata/slice_info.config", f"<?xml version='1.0' encoding='UTF-8'?><config>{body}</config>")
+    return path
+
+
+class TestWhatRulesADonorOut:
+    def test_a_donor_holding_a_different_plate(self, tmp_path):
+        donor = _write_3mf(tmp_path / "donor.3mf", plate=2, filaments=1)
+
+        conflict = _donor_3mf_conflicts(donor, expected_plate=1)
+
+        assert conflict is not None
+        assert "plate 2" in conflict
+
+    def test_an_all_plates_export_without_the_running_plate(self, tmp_path):
+        """Left alone this is the silent one: the plate is looked for
+        downstream, found missing, and every filament in the file is summed onto
+        a single plate's print."""
+        donor = _write_multiplate_3mf(tmp_path / "donor.3mf", {1: 1, 2: 3})
+
+        assert _donor_3mf_conflicts(donor, expected_plate=5) is not None
+
+    def test_an_unreadable_donor_is_not_rejected_on_that_alone(self, tmp_path):
+        """Refusing a file we merely could not parse would take the fallback
+        away from every 3MF variant this parser does not understand -- and an
+        unreadable file is not evidence about which plate it holds. "No plates
+        found" must not be read as "not your plate". The parse failure surfaces
+        downstream as "no filament usage data" instead.
+        """
+        donor = tmp_path / "broken.3mf"
+        donor.write_bytes(b"PK\x03\x04not-really-a-3mf")
+
+        assert _donor_3mf_conflicts(donor, expected_plate=None) is None
+        assert _donor_3mf_conflicts(donor, expected_plate=2) is None
+
+
+class TestWhatMustStillBeAccepted:
+    def test_the_matching_plate(self, tmp_path):
+        donor = _write_3mf(tmp_path / "donor.3mf", plate=2, filaments=2)
+
+        assert _donor_3mf_conflicts(donor, expected_plate=2) is None
+
+    def test_an_all_plates_export_that_carries_the_running_plate(self, tmp_path):
+        """``peek_plate_index_in_3mf`` returns None for a multi-plate file --
+        "which plate is this" has no answer (#2522) -- so the file is judged on
+        whether it holds the plate instead."""
+        donor = _write_multiplate_3mf(tmp_path / "donor.3mf", {1: 3, 2: 1})
+
+        assert _donor_3mf_conflicts(donor, expected_plate=2) is None
+
+    def test_nothing_known_accepts_anything(self, tmp_path):
+        """The reporter's firmware echoes only the 3MF filename and the print
+        was not one Bambuddy dispatched, so the plate is unknowable. Accepting
+        is the pre-existing behaviour and stays -- refusing here would retire
+        the fallback recovery this same issue asked for -- but it is logged."""
+        donor = _write_3mf(tmp_path / "donor.3mf", plate=7, filaments=4)
+
+        assert _donor_3mf_conflicts(donor, expected_plate=None) is None
+
+
+class TestTheCheckThatIsDeliberatelyNotMade:
+    def test_the_filament_count_is_not_compared(self, tmp_path):
+        """A donor whose plate matches is accepted however many filaments it
+        declares, and this is the load-bearing reason why.
+
+        ``ams_mapping`` is indexed by the *project's* filament slots, not the
+        plate's -- ``slot_to_tray[slot_id - 1]`` in the same module -- so a
+        genuine single-filament print publishes ``[0, -1, -1, -1]``. Comparing
+        its length against a plate's filament count would reject correct donors
+        far more often than wrong ones, on every multi-filament project.
+        """
+        donor = _write_3mf(tmp_path / "donor.3mf", plate=2, filaments=3)
+
+        assert _donor_3mf_conflicts(donor, expected_plate=2) is None
+
+
+class TestWhereTheExpectationsComeFrom:
+    def test_the_plate_column_wins_when_it_is_set(self):
+        assert _expected_plate_for_print(3, "Metadata/plate_1.gcode") == 3
+
+    def test_otherwise_the_gcode_path_the_printer_echoed(self):
+        assert _expected_plate_for_print(None, "Metadata/plate_2.gcode") == 2
+
+    def test_a_p1s_that_echoes_only_the_3mf_name_knows_no_plate(self):
+        """Verbatim from the report: ``PRINT START detected - file:
+        Desktop_Goose.gcode.3mf``. There is no plate in that."""
+        assert _expected_plate_for_print(None, "Desktop_Goose.gcode.3mf") is None
+
+
+@pytest.mark.asyncio
+class TestTheLookupItself:
+    async def _seed(self, engine, tmp_path, *, donor_plate: int):
+        maker = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
+        donor_rel = "archive/1/donor.gcode.3mf"
+        _write_3mf(tmp_path / donor_rel, plate=donor_plate, filaments=1)
+        async with maker() as db:
+            # A successfully archived print keeps the 3MF's name; the fallback
+            # row keeps whatever the printer echoed, which here is the plate
+            # path that tells us which plate is running.
+            donor = PrintArchive(
+                printer_id=1,
+                filename="Trent.gcode.3mf",
+                file_path=donor_rel,
+                file_size=1,
+                print_name="Trent",
+                status="completed",
+            )
+            fallback = PrintArchive(
+                printer_id=1,
+                filename="Metadata/plate_2.gcode",
+                file_path="",
+                file_size=0,
+                print_name="Trent",
+                status="printing",
+                extra_data={"no_3mf_available": True},
+            )
+            db.add_all([donor, fallback])
+            await db.commit()
+            await db.refresh(donor)
+            await db.refresh(fallback)
+            return maker, donor.id, fallback.id
+
+    async def test_a_wrong_donor_is_refused(self, test_engine, tmp_path):
+        maker, _, fallback_id = await self._seed(test_engine, tmp_path, donor_plate=1)
+        async with maker() as db:
+            archive = await db.get(PrintArchive, fallback_id)
+            assert await _resolve_3mf_fallback(archive, db, tmp_path) is None
+
+    async def test_a_matching_donor_is_still_used(self, test_engine, tmp_path):
+        maker, _, fallback_id = await self._seed(test_engine, tmp_path, donor_plate=2)
+        async with maker() as db:
+            archive = await db.get(PrintArchive, fallback_id)
+            resolved = await _resolve_3mf_fallback(archive, db, tmp_path)
+            assert resolved is not None and resolved.name == "donor.gcode.3mf"
+
+    async def test_the_library_branch_is_guarded_too(self, test_engine, tmp_path):
+        """A library upload can be the wrong plate for exactly the same reason a
+        previous archive can, and it is consulted first."""
+        maker = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False)
+        _write_3mf(tmp_path / "library/Trent.3mf", plate=1, filaments=3)
+        async with maker() as db:
+            db.add(LibraryFile(filename="Trent.3mf", file_path="library/Trent.3mf", file_type="3mf", file_size=1))
+            # The printer echoes the plate path, so the plate is knowable; the
+            # search stem falls back to the print name, which is what finds the
+            # library upload in the first place.
+            archive = PrintArchive(
+                printer_id=1,
+                filename="Metadata/plate_2.gcode",
+                file_path="",
+                file_size=0,
+                print_name="Trent",
+                status="printing",
+            )
+            db.add(archive)
+            await db.commit()
+            await db.refresh(archive)
+
+            assert await _resolve_3mf_fallback(archive, db, tmp_path) is None

backend/tests/unit/test_download_deadline_and_gate_2957.py

@@ -0,0 +1,340 @@
+"""Downloads get a deadline that fits the file, and take turns on a printer (#2957).
+
+Two things about ``ftp_timeout``. It is passed as *both* the socket inactivity
+timeout and the whole-transfer deadline, so its 30 s default is really a cap on
+how big a file a printer is allowed to serve: the reporter measured the same
+5.4 MB 3MF at 45 s off a worn P1S SD card and 25 s off a new one, and a 15.15 MB
+3MF at 105 s. None of those transfers were unhealthy. And it bounded nothing
+about concurrency -- he watched Bambu Studio lose its own connection to the
+printer while two Bambuddy downloads for the same file ran against it at once.
+
+So the total deadline now follows the size the printer reports, and a printer
+serves one Bambuddy download at a time. Both are deliberately soft: the
+extension is granted only once SIZE has been answered (so a dead printer still
+fails on schedule, and the queue wait #2572 capped is untouched), and a download
+that cannot have the gate goes anyway rather than letting a print lose its 3MF
+to queueing.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import gc
+import threading
+import time
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+
+import backend.app.services.bambu_ftp as ftp_mod
+from backend.app.services.bambu_ftp import (
+    _DOWNLOAD_FLOOR_BYTES_PER_SEC,
+    _DOWNLOAD_MAX_TIMEOUT,
+    _download_extension,
+    _serialized_download,
+    download_file_async,
+    download_file_try_paths_async,
+)
+
+
+class _FakeClient:
+    """Enough of ``BambuFTPClient`` for the async wrappers to drive it."""
+
+    _mode_cache: dict[str, str] = {}
+    A1_MODELS = ()
+
+    def __init__(self, *a, **kw):
+        pass
+
+    @classmethod
+    def cache_mode(cls, ip_address, mode):
+        pass
+
+    def connect(self):
+        return True
+
+    def disconnect(self):
+        pass
+
+
+class TestTheDeadlineFollowsTheFile:
+    def test_a_15mb_3mf_gets_far_more_than_the_30s_default(self):
+        """The reporter's file. 105 s measured, 30 s allowed."""
+        assert _download_extension(15_150_000, 30.0) > 105.0
+
+    def test_an_unknown_size_extends_nothing(self):
+        """No SIZE reply means no transfer got under way. A printer that is not
+        answering must still fail on the base deadline."""
+        assert _download_extension(None, 30.0) == 0.0
+        assert _download_extension(0, 30.0) == 0.0
+
+    def test_a_small_file_that_already_fits_extends_nothing(self):
+        assert _download_extension(64 * 1024, 30.0) == 0.0
+
+    def test_it_is_capped(self):
+        """``on_print_start`` holds a pooled DB connection across the whole 3MF
+        hunt, so an unbounded deadline is a connection leak with extra steps."""
+        assert _download_extension(10 * 1024 * 1024 * 1024, 30.0) == _DOWNLOAD_MAX_TIMEOUT - 30.0
+
+    def test_the_floor_is_pessimistic_not_the_measured_rate(self):
+        """25 KB/s. The reporter's P1S managed ~145 KB/s on its bad day, so the
+        allowance is several times what a real slow link needs."""
+        assert _DOWNLOAD_FLOOR_BYTES_PER_SEC == 25 * 1024
+
+
+@pytest.mark.asyncio
+class TestASlowTransferSurvivesItsDeadline:
+    async def test_a_transfer_that_reports_its_size_is_given_the_time(self, tmp_path):
+        """The whole point, at 1/1000 scale: a deadline the transfer blows past,
+        and a printer that answered SIZE. 1 MB at the 25 KB/s floor buys ~39 s,
+        so a transfer that takes 0.4 s finishes instead of being declared dead
+        at 0.05 s. Nothing is patched here but the socket."""
+        payload = b"x" * 4096
+
+        class _Client(_FakeClient):
+            def download_to_file(self, remote_path, local_path, *, size_callback=None, cancel_event=None, **kwargs):
+                size_callback(1_000_000)
+                # Honouring the cancel flag is what makes this a real test: the
+                # expired-deadline path sets it, and a transfer that ignored it
+                # would be salvaged by the #1014 grace and prove nothing.
+                for _ in range(40):
+                    if cancel_event is not None and cancel_event.is_set():
+                        raise ftp_mod.DownloadCancelled(remote_path)
+                    time.sleep(0.01)
+                local_path.write_bytes(payload)
+                return True
+
+        with patch.object(ftp_mod, "BambuFTPClient", _Client):
+            ok = await download_file_async("10.0.0.1", "x", "/f.3mf", tmp_path / "f.3mf", timeout=0.05)
+
+        assert ok is True
+        assert (tmp_path / "f.3mf").read_bytes() == payload
+
+    async def test_a_transfer_that_blows_even_the_size_deadline_is_not_retried(self, tmp_path):
+        """Otherwise the retry loop spends the whole stretched deadline again to
+        reach the same conclusion -- four times, holding a pooled database
+        connection, because ``on_print_start`` never lets go of one. Same reason
+        ``UploadCancelled`` has been non-retryable since #2529."""
+        attempts = {"n": 0}
+
+        class _Client(_FakeClient):
+            def download_to_file(self, remote_path, local_path, *, size_callback=None, cancel_event=None, **kwargs):
+                attempts["n"] += 1
+                size_callback(1_000_000)
+                for _ in range(200):
+                    if cancel_event is not None and cancel_event.is_set():
+                        raise ftp_mod.DownloadCancelled(remote_path)
+                    time.sleep(0.01)
+                return False
+
+        with (
+            patch.object(ftp_mod, "BambuFTPClient", _Client),
+            patch.object(ftp_mod, "_download_extension", lambda size, base: 0.2 if size else 0.0),
+            pytest.raises(ftp_mod.DownloadDeadlineExceeded),
+        ):
+            await ftp_mod.with_ftp_retry(
+                download_file_async,
+                "10.0.0.11",
+                "x",
+                "/f.3mf",
+                tmp_path / "f.3mf",
+                timeout=0.05,
+                max_retries=3,
+                retry_delay=0,
+            )
+
+        assert attempts["n"] == 1, "a transfer that already had its full size-derived deadline was retried"
+
+    async def test_an_ordinary_timeout_is_still_an_ordinary_retryable_miss(self, tmp_path):
+        """No SIZE, no extension, no new exception -- the pre-existing contract."""
+
+        class _Client(_FakeClient):
+            def download_to_file(self, remote_path, local_path, **kwargs):
+                time.sleep(0.6)
+                return False
+
+        with patch.object(ftp_mod, "BambuFTPClient", _Client):
+            assert await download_file_async("10.0.0.12", "x", "/f.3mf", tmp_path / "f.3mf", timeout=0.05) is False
+
+    async def test_a_printer_that_never_answers_size_still_fails_on_time(self, tmp_path):
+        class _Client(_FakeClient):
+            def download_to_file(self, remote_path, local_path, **kwargs):
+                time.sleep(1.5)
+                return False
+
+        started = time.monotonic()
+        with patch.object(ftp_mod, "BambuFTPClient", _Client):
+            ok = await download_file_async("10.0.0.2", "x", "/f.3mf", tmp_path / "f.3mf", timeout=0.2)
+        elapsed = time.monotonic() - started
+
+        assert ok is False
+        assert elapsed < 5.0, "an unknown size must not buy a transfer any extra time"
+
+
+@pytest.mark.asyncio
+class TestOnlyOneDownloadPerPrinter:
+    async def test_the_second_download_waits_for_the_first(self):
+        order: list[str] = []
+
+        async def _hold(tag: str, seconds: float):
+            async with _serialized_download("10.0.0.3", tag) as held:
+                order.append(f"{tag}:in:{held}")
+                await asyncio.sleep(seconds)
+                order.append(f"{tag}:out")
+
+        await asyncio.gather(_hold("a", 0.15), _hold("b", 0.01))
+
+        assert order == ["a:in:True", "a:out", "b:in:True", "b:out"]
+
+    async def test_a_waiter_that_gives_up_goes_anyway(self):
+        """The gate is contention relief, not a correctness control. A print
+        that lost its 3MF because a thumbnail held the printer would be a worse
+        bug than the contention."""
+        with patch.object(ftp_mod, "_DOWNLOAD_GATE_WAIT_SECONDS", 0.05):
+
+            async def _holder():
+                async with _serialized_download("10.0.0.4", "holder"):
+                    await asyncio.sleep(0.3)
+
+            async def _waiter():
+                async with _serialized_download("10.0.0.4", "waiter") as held:
+                    return held
+
+            holder = asyncio.create_task(_holder())
+            await asyncio.sleep(0.01)
+            went_anyway = await _waiter()
+            await holder
+
+        assert went_anyway is False
+
+    async def test_the_gate_is_released_when_the_body_raises(self):
+        with pytest.raises(RuntimeError):
+            async with _serialized_download("10.0.0.5", "boom"):
+                raise RuntimeError("boom")
+
+        async with _serialized_download("10.0.0.5", "after") as held:
+            assert held is True
+
+    async def test_a_real_download_takes_the_gate(self, tmp_path):
+        """Not just the helper: the two entry points every download goes
+        through have to be the ones holding it."""
+        concurrent = {"max": 0, "now": 0}
+        lock = threading.Lock()
+
+        class _Client(_FakeClient):
+            def download_to_file(self, remote_path, local_path: Path, **kwargs):
+                with lock:
+                    concurrent["now"] += 1
+                    concurrent["max"] = max(concurrent["max"], concurrent["now"])
+                time.sleep(0.1)
+                with lock:
+                    concurrent["now"] -= 1
+                local_path.write_bytes(b"data")
+                return True
+
+        with patch.object(ftp_mod, "BambuFTPClient", _Client):
+            await asyncio.gather(
+                download_file_async("10.0.0.6", "x", "/a.3mf", tmp_path / "a.3mf", timeout=30),
+                download_file_try_paths_async("10.0.0.6", "x", ["/b.3mf"], tmp_path / "b.3mf", timeout=30),
+            )
+
+        assert concurrent["max"] == 1, "two downloads ran against one printer at the same time"
+
+    async def test_the_file_browser_stays_outside_the_gate(self, tmp_path):
+        """``printer_media`` documented itself lock-free before this gate
+        existed, in both directions: a 3MF preview must not wait out somebody
+        else's ten-gigabyte selection, and that selection must not hold the
+        printer for the twenty minutes it legitimately takes."""
+        overlapped = asyncio.Event()
+
+        class _Client(_FakeClient):
+            def download_to_file(self, remote_path, local_path: Path, **kwargs):
+                time.sleep(0.15)
+                local_path.write_bytes(b"data")
+                return True
+
+        async def _holder():
+            async with _serialized_download("10.0.0.10", "holder"):
+                overlapped.set()
+                await asyncio.sleep(0.3)
+
+        with patch.object(ftp_mod, "BambuFTPClient", _Client):
+            holder = asyncio.create_task(_holder())
+            await overlapped.wait()
+            started = time.monotonic()
+            ok = await download_file_async(
+                "10.0.0.10", "x", "/big.mp4", tmp_path / "big.mp4", timeout=30, serialize=False
+            )
+            elapsed = time.monotonic() - started
+            await holder
+
+        assert ok is True
+        assert elapsed < 1.0, "an opted-out download queued behind the gate anyway"
+
+    async def test_different_printers_do_not_queue_behind_each_other(self):
+        started = asyncio.Event()
+
+        async def _slow():
+            async with _serialized_download("10.0.0.7", "slow"):
+                started.set()
+                await asyncio.sleep(0.3)
+
+        task = asyncio.create_task(_slow())
+        await started.wait()
+        async with _serialized_download("10.0.0.8", "other") as held:
+            assert held is True
+        await task
+
+
+@pytest.mark.asyncio
+class TestTheCapNoLongerLeavesAWorkerOnTheSocket:
+    async def test_a_capped_path_walk_stops_its_worker(self, tmp_path):
+        """``asyncio.wait_for`` cannot cancel an executor thread, so the cap used
+        to return while the worker kept walking the remaining paths -- still
+        holding the printer's FTP socket. The reporter's log has one of those
+        still going as the archive flow's own download landed."""
+        cancelled = threading.Event()
+        walked: list[str] = []
+
+        class _Client(_FakeClient):
+            def download_to_file(self, remote_path, local_path, *, cancel_event=None, **kwargs):
+                walked.append(remote_path)
+                for _ in range(60):
+                    if cancel_event is not None and cancel_event.is_set():
+                        cancelled.set()
+                        raise ftp_mod.DownloadCancelled(remote_path)
+                    time.sleep(0.01)
+                return False
+
+        with patch.object(ftp_mod, "BambuFTPClient", _Client):
+            hit = await download_file_try_paths_async(
+                "10.0.0.9", "x", ["/1.3mf", "/2.3mf", "/3.3mf"], tmp_path / "f.3mf", timeout=0.1
+            )
+
+        assert hit is None
+        assert cancelled.is_set(), "the capped worker was left running on the printer's socket"
+        assert walked == ["/1.3mf"], "the worker kept walking paths after its caller had given up"
+
+    async def test_a_late_transport_error_is_not_logged_as_loop_noise(self, tmp_path):
+        """Shielding the worker so the cap can wait it out means nobody is left
+        to read what it raised, and asyncio reports that as a bare
+        ``Future exception was never retrieved`` ERROR with a traceback -- after
+        the caller has already logged the real failure. Precisely the class of
+        noise #2968 was about, so it must not come back in through this door."""
+        loop_errors: list[str] = []
+        asyncio.get_running_loop().set_exception_handler(lambda _loop, ctx: loop_errors.append(ctx.get("message", "")))
+
+        class _Client(_FakeClient):
+            def download_to_file(self, remote_path, local_path, **kwargs):
+                time.sleep(0.3)
+                raise OSError("late transport failure")
+
+        with patch.object(ftp_mod, "BambuFTPClient", _Client):
+            assert await download_file_try_paths_async("10.0.0.13", "x", ["/a"], tmp_path / "a", timeout=0.05) is None
+
+        await asyncio.sleep(0.6)
+        gc.collect()
+        await asyncio.sleep(0)
+
+        assert loop_errors == [], f"the shielded worker leaked its failure into the log: {loop_errors}"