diff --git a/backend/app/api/routes/printers.py b/backend/app/api/routes/printers.py index 2fd530d52..102287496 100644 --- a/backend/app/api/routes/printers.py +++ b/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(): diff --git a/backend/app/services/archive.py b/backend/app/services/archive.py index 77e3ca183..cc782efa3 100644 --- a/backend/app/services/archive.py +++ b/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 ```` 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) diff --git a/backend/app/services/bambu_ftp.py b/backend/app/services/bambu_ftp.py index 3650a59d4..a6b3a75ba 100644 --- a/backend/app/services/bambu_ftp.py +++ b/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 - finally: - client.disconnect() + client = BambuFTPClient(ip_address, access_code, timeout=socket_timeout, printer_model=printer_model) + if not client.connect(): + return None - 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 + 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: + done.set() + + 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): diff --git a/backend/app/services/printer_media.py b/backend/app/services/printer_media.py index c52df2f98..f71fe4a26 100644 --- a/backend/app/services/printer_media.py +++ b/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") diff --git a/backend/app/services/usage_tracker.py b/backend/app/services/usage_tracker.py index ba1df7be8..d8b6981f0 100644 --- a/backend/app/services/usage_tracker.py +++ b/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, diff --git a/backend/tests/unit/test_cover_rechecks_3mf_cache_2957.py b/backend/tests/unit/test_cover_rechecks_3mf_cache_2957.py new file mode 100644 index 000000000..d41cad0ef --- /dev/null +++ b/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"] diff --git a/backend/tests/unit/test_donor_3mf_validation_2957.py b/backend/tests/unit/test_donor_3mf_validation_2957.py new file mode 100644 index 000000000..289236a81 --- /dev/null +++ b/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"" for i in range(filaments) + ) + with zipfile.ZipFile(path, "w") as zf: + zf.writestr( + "Metadata/slice_info.config", + "" + f"" + f"{rows}" + "", + ) + 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"" for i in range(filaments) + ) + body += f"{rows}" + with zipfile.ZipFile(path, "w") as zf: + zf.writestr("Metadata/slice_info.config", f"{body}") + 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 diff --git a/backend/tests/unit/test_download_deadline_and_gate_2957.py b/backend/tests/unit/test_download_deadline_and_gate_2957.py new file mode 100644 index 000000000..bcc69d1e8 --- /dev/null +++ b/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}"