fix(queue): upload to printers in parallel, cap wedge retries, make debug logs survive a farm

The reporter's 19-printer farm started prints "one by one", up to an hour apart.
check_queue awaited each dispatch inline, and a dispatch includes the FTP upload,
so every printer queued behind every other printer's transfer despite being an
independent machine. His logs give the arithmetic: 40978500 bytes in 254.1s,
157 KB/s - a Bambu printer's SD write, not the network, is the bottleneck. Nineteen
of those in series is ~80 minutes, and the next upload started 131 ms after the
previous one finished. The delay is linear in fleet size, which is why it got worse
the more printers he selected.

Dispatch is now collected during the (still sequential) selection loop and run
concurrently afterwards, capped by queue_max_concurrent_uploads - Settings ->
Workflow -> Queue & Dispatch, default 4, 1 restores the old behaviour. Every gate
is untouched; only the transfers overlap. The pass still awaits its uploads before
returning: _start_print flips the row pending -> printing only after the upload,
so an early return would let the next tick re-dispatch the same rows.

FTP work moves to its own thread pool. It was on asyncio's default executor -
min(32, cpu+4), six threads on a 2-core NAS, shared with everything else - which
was survivable only while uploads were serial.

Two problems the same bundle exposed:

A printer that accepts project_file but never starts (#1678) was retried forever:
270s watchdog, revert to pending, re-upload the whole file, repeat. Hence his
"printer who, since the morning, still not launch" - and on a farm each lap also
eats an upload slot the other printers are waiting on. Attempts are now counted on
the queue item; after three it fails with a message pointing at the printer instead
of queueing a fourth re-upload.

The debug bundle we asked him for held 4m49s of history. The push_status dumps fired
on every frame rather than on change - several while their own comment claimed
otherwise - which is 27,727 of the bundle's 29,830 lines and rolls 5 MB in under five
minutes on 19 printers. They now log transitions only. The bundle also read just the
live log while three rotated backups sat next to it, under a byte budget four times
larger than the file it was reading.

Migration verified on SQLite and Postgres: idempotent, backfills legacy NULLs
(dispatch_attempts + 1 is NULL for a NULL row, which would silently disable the cap).

Tests: 6 on concurrent dispatch (overlap, cap honoured, 1 == serial, default applies
with no settings row, a failed printer does not cancel its siblings, no early return),
4 on the retry budget, 6 on the bundle's rotated-log span, 7 on the debug gating.
Each verified to fail against the unfixed code - the first end-to-end log assertion I
wrote passed without the fix and had to be tightened.

backend/app/api/routes/settings.py

@@ -142,6 +142,7 @@ async def _build_settings_response(db: AsyncSession, is_api_key: bool = False) -
             "pipeline_max_copies",
             "preheat_max_wait_seconds",
             "preheat_soak_seconds",
+            "queue_max_concurrent_uploads",
         ]:
             settings_dict[setting.key] = int(setting.value)
         elif setting.key == "default_printer_id":

backend/app/api/routes/support.py

@@ -1119,21 +1119,51 @@ async def _collect_support_info() -> dict:
 
 
 def _get_log_content(max_bytes: int = 10 * 1024 * 1024, sensitive_strings: dict[str, str] | None = None) -> bytes:
-    """Get log file content, limited to max_bytes from the end."""
+    """Get recent log content, limited to max_bytes from the end.
+
+    Spans the rotated files as well as the live one. ``bambuddy.log`` is capped
+    at 5 MB by the RotatingFileHandler, and the bundle used to ship only that
+    file — so on a large fleet with debug logging on, the window we ask a
+    reporter for was far shorter than anyone realised. The 19-printer farm in
+    #2555 emits ~100 lines/s of MQTT frame dumps, which fills 5 MB in under five
+    minutes: the bundle we received to diagnose a *queue* problem barely
+    contained one upload. The three rotated backups were sitting on disk unread.
+
+    Reads oldest -> newest so the result is chronological, then takes the last
+    ``max_bytes``, which is where the budget was all along.
+    """
     log_file = settings.log_dir / "bambuddy.log"
     if not log_file.exists():
         return b"Log file not found"
 
-    file_size = log_file.stat().st_size
-    if file_size <= max_bytes:
-        content = log_file.read_text(encoding="utf-8", errors="replace")
-    else:
-        # Read last max_bytes
-        with open(log_file, "rb") as f:
-            f.seek(file_size - max_bytes)
-            # Skip partial line at start
-            f.readline()
-            content = f.read().decode("utf-8", errors="replace")
+    # RotatingFileHandler names its backups .log.1 (newest) .. .log.N (oldest).
+    # Walk them in reverse so the concatenation reads forwards in time.
+    candidates: list[Path] = []
+    for index in range(settings.log_backup_count, 0, -1):
+        rotated = log_file.with_name(f"{log_file.name}.{index}")
+        if rotated.exists():
+            candidates.append(rotated)
+    candidates.append(log_file)
+
+    chunks: list[str] = []
+    remaining = max_bytes
+    # Fill from the newest backwards so the byte budget is spent on recent
+    # history, then flip back to chronological order for the reader.
+    for path in reversed(candidates):
+        if remaining <= 0:
+            break
+        try:
+            size = path.stat().st_size
+            with open(path, "rb") as f:
+                if size > remaining:
+                    f.seek(size - remaining)
+                    f.readline()  # discard the partial line the seek landed in
+                chunks.append(f.read().decode("utf-8", errors="replace"))
+            remaining -= min(size, remaining)
+        except OSError:
+            logger.debug("Failed to read log file %s for support bundle", path, exc_info=True)
+
+    content = "".join(reversed(chunks))
 
     # Sanitize sensitive data
     content = sanitize_log_content(content, sensitive_strings)
@@ -1278,7 +1308,12 @@ async def generate_support_bundle(
             zf.writestr(f"push-status/printer-{i + 1}.json", snapshot_json)
 
         # Add log file
-        log_content = _get_log_content(sensitive_strings=sensitive_strings)
+        # Off the event loop: this reads up to 10 MB and then runs one full regex
+        # pass per sensitive string over it. Now that the bundle spans the rotated
+        # files it can genuinely reach that ceiling, and the blocking cost scales
+        # with the number of printers (4 redaction patterns each) — i.e. it is
+        # worst on exactly the fleet size this change was written for.
+        log_content = await asyncio.to_thread(_get_log_content, sensitive_strings=sensitive_strings)
         zf.writestr("bambuddy.log", log_content)
 
     zip_buffer.seek(0)

backend/app/core/config.py

@@ -3,6 +3,7 @@ import os
 import re as _re
 from pathlib import Path
 
+from pydantic import Field
 from pydantic_settings import BaseSettings
 
 # Application version - single source of truth
@@ -76,6 +77,13 @@ class Settings(BaseSettings):
     # Logging
     log_level: str = "INFO"  # Override with LOG_LEVEL env var or DEBUG=true
     log_to_file: bool = True  # Set to false to disable file logging
+    # Rotation for bambuddy.log. Read by main.py (which owns the handler) and by
+    # the support bundle (which harvests the backups as well as the live file);
+    # they must agree on the backup count or the bundle silently skips history.
+    # Bounded: RotatingFileHandler treats maxBytes=0 as "never rotate", so a
+    # zero/negative override would grow the log without limit.
+    log_max_bytes: int = Field(default=5 * 1024 * 1024, gt=0)
+    log_backup_count: int = Field(default=3, ge=0)
 
     # API
     api_prefix: str = "/api/v1"

backend/app/core/database.py

@@ -3413,6 +3413,16 @@ async def run_migrations(conn):
     else:
         await _safe_execute(conn, "ALTER TABLE oidc_providers ADD COLUMN is_autologin BOOLEAN DEFAULT false")
 
+    # Migration: Add dispatch_attempts to print_queue (#2555). Counts the times
+    # the start-watchdog reverted the row from 'printing' back to 'pending' so a
+    # printer that never actually starts stops being retried forever. INTEGER
+    # DEFAULT 0 is spelled identically on SQLite and Postgres — no dialect branch.
+    # Verified on both dialects: ADD COLUMN ... DEFAULT 0 backfills existing rows,
+    # so no separate UPDATE is needed (and _safe_execute is DDL-only — see its
+    # docstring). The scheduler reads it as `(item.dispatch_attempts or 0) + 1`
+    # regardless, so even a NULL row could not disable the retry cap.
+    await _safe_execute(conn, "ALTER TABLE print_queue ADD COLUMN dispatch_attempts INTEGER DEFAULT 0")
+
     # Migration: Disambiguate the four ``user_print_*`` notification template
     # names by appending " Email" (#1792). See ``_migrate_rename_user_print_template_names``.
     await _migrate_rename_user_print_template_names(conn)

backend/app/main.py

@@ -288,8 +288,8 @@ if app_settings.log_to_file:
     log_file = app_settings.log_dir / "bambuddy.log"
     file_handler = RotatingFileHandler(
         log_file,
-        maxBytes=5 * 1024 * 1024,  # 5MB
-        backupCount=3,
+        maxBytes=app_settings.log_max_bytes,
+        backupCount=app_settings.log_backup_count,
         encoding="utf-8",
     )
     file_handler.setLevel(log_level)

backend/app/models/print_queue.py

@@ -65,6 +65,14 @@ class PrintQueueItem(Base):
     # Auto-print G-code injection (#422)
     gcode_injection: Mapped[bool] = mapped_column(Boolean, default=False)
 
+    # How many times the start-watchdog has reverted this item from 'printing'
+    # back to 'pending' (#2555). A printer that accepts project_file but never
+    # starts (#1678) used to be retried forever: upload, wait out the watchdog,
+    # revert, upload again — burning a full 3MF transfer per cycle and, with
+    # the queue dispatching serially, dragging every other printer's start time
+    # out with it. The counter bounds that loop; see DISPATCH_MAX_ATTEMPTS.
+    dispatch_attempts: Mapped[int] = mapped_column(Integer, default=0, server_default="0")
+
     # H2C dual-nozzle-rack slicer pick preservation (#1780). BambuStudio's
     # project_file MQTT command for rack-swap-capable models (O1C2 today)
     # carries per-filament physical nozzle position IDs in `nozzle_mapping`,

backend/app/schemas/settings.py

@@ -326,6 +326,16 @@ class AppSettings(BaseModel):
         default=False,
         description="Shortest Job First — scheduler prioritizes shorter print jobs over longer ones",
     )
+    queue_max_concurrent_uploads: int = Field(
+        default=4,
+        ge=1,
+        le=16,
+        description=(
+            "How many printers the queue may upload to at the same time. Printers are independent "
+            "machines, so raising this starts a multi-printer batch proportionally sooner; each "
+            "concurrent upload costs one connection and one thread on the Bambuddy host."
+        ),
+    )
 
     # Preheat / heat-soak before queued prints (#1468). The scheduler stage runs
     # BEFORE FTP upload. Three hardware tiers behave differently:
@@ -553,6 +563,7 @@ class AppSettingsUpdate(BaseModel):
     stagger_interval_minutes: int | None = Field(default=None, ge=1, le=60)
     require_plate_clear: bool | None = None
     queue_shortest_first: bool | None = None
+    queue_max_concurrent_uploads: int | None = Field(default=None, ge=1, le=16)
     preheat_enabled: bool | None = None
     preheat_filament_targets: str | None = None
     preheat_max_wait_seconds: int | None = Field(default=None, ge=60, le=3600)

backend/app/services/bambu_ftp.py

@@ -8,6 +8,7 @@ import threading
 import time
 import weakref
 from collections.abc import Awaitable, Callable
+from concurrent.futures import ThreadPoolExecutor
 from enum import Enum
 from ftplib import FTP, FTP_TLS  # nosec B402
 from io import BytesIO
@@ -18,6 +19,32 @@ logger = logging.getLogger(__name__)
 
 T = TypeVar("T")
 
+# Every FTP call below is blocking ftplib work handed to a thread. They used to
+# run on asyncio's *default* executor, which is sized min(32, cpu_count + 4) —
+# six threads on a 2-core NAS — and is shared with every other ``to_thread`` /
+# ``run_in_executor`` caller in the app. That was survivable only because the
+# scheduler uploaded to exactly one printer at a time. Dispatching to several
+# printers at once (#2555) would park one thread per in-flight upload for
+# minutes at a stretch (a 41 MB 3MF at the ~150 KB/s a Bambu printer sustains
+# takes ~4 min), starving the default pool and stalling unrelated work.
+#
+# A dedicated pool keeps that blast radius inside the FTP layer: the scheduler's
+# own concurrency cap is what limits parallel uploads, and it can never exhaust
+# the executor everything else depends on. Threads are created lazily, so an
+# idle pool costs nothing.
+#
+# Sized well above `queue_max_concurrent_uploads` (max 16), because uploads are
+# not the only traffic here: SD browsing, timelapse/recording listing, cover
+# downloads, deletes and storage checks all run through this pool too, and on a
+# farm they fan out across every printer at once. The pool's work queue is
+# unbounded, so exceeding it does not fail — it queues. But `asyncio.wait_for`
+# starts its clock at submission, not at thread start, so a task that sits in the
+# queue can burn its whole timeout without ever running, and `list_files_async`
+# reports a timeout as an empty listing — a silent "this printer has no files".
+# Keep the headroom.
+_FTP_MAX_WORKERS = 48
+_ftp_executor = ThreadPoolExecutor(max_workers=_FTP_MAX_WORKERS, thread_name_prefix="bambu-ftp")
+
 # Overall upload deadline (#2529). A flat wall-clock cap punishes big files on
 # slow links rather than catching broken ones: a 96 MB 3MF at the ~75 KB/s an A1
 # sustains over WiFi legitimately needs ~20 minutes, and the old flat 600 s
@@ -919,7 +946,7 @@ async def download_file_async(
         done = threading.Event()
         try:
             return await asyncio.wait_for(
-                loop.run_in_executor(None, _download, force_prot_c, completion, done), timeout=timeout
+                loop.run_in_executor(_ftp_executor, _download, force_prot_c, completion, done), timeout=timeout
             )
         except TimeoutError:
             # Slow WiFi links commonly overshoot ftp_timeout by 10–30 s without
@@ -932,6 +959,12 @@ async def download_file_async(
             # floor so artificially small test timeouts still give zombies a
             # realistic window to finish.
             grace = max(min(timeout, 30.0), 0.5)
+            # Deliberately the DEFAULT executor, not `_ftp_executor`: this thread
+            # blocks waiting on `_download`, which is itself an `_ftp_executor`
+            # worker. Parking waiters in the same bounded pool as the workers they
+            # wait for is how you build a deadlock — with enough concurrent
+            # timeouts the waiters would occupy every slot and the downloads they
+            # are waiting for could never be scheduled.
             await loop.run_in_executor(None, done.wait, grace)
             if completion["success"] and local_path.exists() and local_path.stat().st_size > 0:
                 logger.info(
@@ -1004,7 +1037,7 @@ async def download_file_try_paths_async(
         finally:
             client.disconnect()
 
-    return await loop.run_in_executor(None, _download)
+    return await loop.run_in_executor(_ftp_executor, _download)
 
 
 def _upload_deadline(local_path: Path) -> float:
@@ -1115,7 +1148,7 @@ async def upload_file_async(
         breaks the send loop and deletes the partial file) and wait for it to
         actually go.
         """
-        fut = loop.run_in_executor(None, lambda: _upload(force_prot_c))
+        fut = loop.run_in_executor(_ftp_executor, lambda: _upload(force_prot_c))
         try:
             return await asyncio.wait_for(asyncio.shield(fut), timeout=deadline)
         except TimeoutError:
@@ -1204,7 +1237,7 @@ async def list_files_async(
         return []
 
     try:
-        return await asyncio.wait_for(loop.run_in_executor(None, _list), timeout=timeout)
+        return await asyncio.wait_for(loop.run_in_executor(_ftp_executor, _list), timeout=timeout)
     except TimeoutError:
         logger.warning("FTP list_files timed out after %ss for %s", timeout, path)
         return []
@@ -1238,7 +1271,7 @@ async def delete_file_async(
                 client.disconnect()
         return DeleteResult.FAILED
 
-    return await loop.run_in_executor(None, _delete)
+    return await loop.run_in_executor(_ftp_executor, _delete)
 
 
 async def download_file_bytes_async(
@@ -1265,7 +1298,7 @@ async def download_file_bytes_async(
                 client.disconnect()
         return None
 
-    return await loop.run_in_executor(None, _download)
+    return await loop.run_in_executor(_ftp_executor, _download)
 
 
 async def get_storage_info_async(
@@ -1291,7 +1324,7 @@ async def get_storage_info_async(
                 client.disconnect()
         return None
 
-    return await loop.run_in_executor(None, _get_storage)
+    return await loop.run_in_executor(_ftp_executor, _get_storage)
 
 
 async def get_ftp_retry_settings() -> tuple[bool, int, float, float]:

backend/app/services/bambu_mqtt.py

@@ -493,6 +493,8 @@ class BambuMQTTClient:
         self.serial_number = serial_number
         self.access_code = access_code
         self.model = model
+        # Last value logged by _debug_on_change(), keyed by log site. See there.
+        self._debug_last: dict[str, object] = {}
         self.on_state_change = on_state_change
         self.on_print_start = on_print_start
         self.on_print_complete = on_print_complete
@@ -1001,6 +1003,46 @@ class BambuMQTTClient:
                     json.dumps(print_data),
                 )
 
+    def _debug_on_change(self, key: str, value: object, msg: str, *args: object) -> None:
+        """``logger.debug``, but only when ``value`` differs from the last call for ``key``.
+
+        The state dumps in the push_status handler fire whenever their field is
+        *present* in the frame — and a full push_status carries every field, so
+        they fire on every frame regardless of whether anything changed. Several
+        even say "updated" or "changes" in their own comment while doing nothing
+        of the sort.
+
+        On one printer that is ~1.5 lines/s and nobody noticed. On the 19-printer
+        farm in #2555 it is ~100 lines/s, which fills the 5 MB log inside five
+        minutes: the reporter enabled debug logging as asked and the support
+        bundle came back holding under five minutes of history, almost none of it
+        about the queue problem we were chasing. 27,727 of its 29,830 lines were
+        these dumps.
+
+        Deduplicating on the value keeps every transition — which is the only part
+        anyone reads these lines for — and drops the steady-state repetition.
+        ``value`` must capture everything interpolated into ``msg``, or a change
+        will be swallowed; pass a tuple when the message renders several fields.
+        """
+        if not logger.isEnabledFor(logging.DEBUG):
+            # Debug logging is toggled at RUNTIME (POST /support/debug-logging),
+            # and these clients outlive the toggle. Letting INFO-level frames warm
+            # the cache would be self-defeating: the operator turns debug on
+            # precisely to see the printer's current state, and a cache already
+            # holding every steady-state value would suppress that baseline until
+            # something happened to change. On an idle printer the bundle would
+            # come back with none of these lines at all.
+            #
+            # So while debug is off we record nothing and drop whatever we had.
+            # Every enable then starts cold and dumps a full baseline on the next
+            # frame, exactly as it did before this method existed.
+            self._debug_last.clear()
+            return
+        if self._debug_last.get(key) == value:
+            return
+        self._debug_last[key] = value
+        logger.debug(msg, *args)
+
     def _process_message(self, payload: dict):
         """Process incoming MQTT message from printer."""
         # Handle top-level AMS data (comes outside of "print" key)
@@ -1165,9 +1207,14 @@ class BambuMQTTClient:
                 self.state.ams_status_main = (self.state.ams_status >> 8) & 0xFF
 
                 # Log when ams_status changes (for filament change tracking debug)
-                logger.debug(
-                    f"[{self.serial_number}] ams_status: {self.state.ams_status} "
-                    f"(main={self.state.ams_status_main}, sub={self.state.ams_status_sub})"
+                self._debug_on_change(
+                    "ams_status:print",
+                    self.state.ams_status,
+                    "[%s] ams_status: %s (main=%s, sub=%s)",
+                    self.serial_number,
+                    self.state.ams_status,
+                    self.state.ams_status_main,
+                    self.state.ams_status_sub,
                 )
 
             # Check for command responses
@@ -1639,7 +1686,13 @@ class BambuMQTTClient:
             # Log all AMS dict fields to debug tray_now for H2D dual-nozzle
             non_list_fields = {k: v for k, v in ams_data.items() if k != "ams"}
             if non_list_fields:
-                logger.debug("[%s] AMS dict fields: %s", self.serial_number, non_list_fields)
+                self._debug_on_change(
+                    "ams_dict_fields",
+                    non_list_fields,
+                    "[%s] AMS dict fields: %s",
+                    self.serial_number,
+                    non_list_fields,
+                )
 
             # IMPORTANT: Parse ams_status FIRST before tray_now, so we have fresh status
             # when checking if we're in filament change mode for tray_now disambiguation
@@ -1655,9 +1708,14 @@ class BambuMQTTClient:
                 # Compute main and sub status
                 self.state.ams_status_sub = self.state.ams_status & 0xFF
                 self.state.ams_status_main = (self.state.ams_status >> 8) & 0xFF
-                logger.debug(
-                    f"[{self.serial_number}] ams_status: {self.state.ams_status} "
-                    f"(main={self.state.ams_status_main}, sub={self.state.ams_status_sub})"
+                self._debug_on_change(
+                    "ams_status:ams",
+                    self.state.ams_status,
+                    "[%s] ams_status: %s (main=%s, sub=%s)",
+                    self.serial_number,
+                    self.state.ams_status,
+                    self.state.ams_status_main,
+                    self.state.ams_status_sub,
                 )
 
             # Parse tray_now from AMS dict - this is the currently loaded tray global ID
@@ -1878,7 +1936,13 @@ class BambuMQTTClient:
                         )
                     self.state.last_loaded_tray = self.state.tray_now
 
-                logger.debug("[%s] tray_now updated: %s", self.serial_number, self.state.tray_now)
+                self._debug_on_change(
+                    "tray_now",
+                    self.state.tray_now,
+                    "[%s] tray_now updated: %s",
+                    self.serial_number,
+                    self.state.tray_now,
+                )
 
             # NOTE: ams_status is parsed BEFORE tray_now (see above) to ensure correct
             # state when checking filament change mode for H2D disambiguation
@@ -2029,7 +2093,14 @@ class BambuMQTTClient:
         self._apply_ams_version_cache(merged_ams)
         # Update timestamp for RFID refresh detection (frontend can detect "new data arrived")
         self.state.last_ams_update = time.time()
-        logger.debug("[%s] Merged AMS data: %s new units, %s total", self.serial_number, len(ams_list), len(merged_ams))
+        self._debug_on_change(
+            "merged_ams",
+            (len(ams_list), len(merged_ams)),
+            "[%s] Merged AMS data: %s new units, %s total",
+            self.serial_number,
+            len(ams_list),
+            len(merged_ams),
+        )
 
         # Extract ams_extruder_map from each AMS unit's info field
         # BambuStudio DevFilaSystem.cpp parses info as hex string:
@@ -2055,7 +2126,15 @@ class BambuMQTTClient:
                         # 0xE = uninitialized AMS, skip
                         continue
                     ams_extruder_map[str(ams_id)] = extruder_id
-                    logger.debug(f"[{self.serial_number}] AMS {ams_id} info=0x{info} -> extruder {extruder_id}")
+                    self._debug_on_change(
+                        f"ams_info:{ams_id}",
+                        (info, extruder_id),
+                        "[%s] AMS %s info=0x%s -> extruder %s",
+                        self.serial_number,
+                        ams_id,
+                        info,
+                        extruder_id,
+                    )
                 except (ValueError, TypeError):
                     pass  # Skip AMS units with unparseable info bitmask values
         if ams_extruder_map:
@@ -2378,8 +2457,13 @@ class BambuMQTTClient:
                     state_val = ext_data["state"]
                     # Extract bits 12-14 (3 bits) for switch state
                     switch_state = (state_val >> 12) & 0x7
-                    logger.debug(
-                        f"[{self.serial_number}] device.extruder.state={state_val} (switch_state bits 12-14: {switch_state})"
+                    self._debug_on_change(
+                        "extruder_state",
+                        state_val,
+                        "[%s] device.extruder.state=%s (switch_state bits 12-14: %s)",
+                        self.serial_number,
+                        state_val,
+                        switch_state,
                     )
                 # Log 'cur' field if present (might indicate current/active extruder)
                 if "cur" in ext_data:
@@ -2521,7 +2605,13 @@ class BambuMQTTClient:
                         # Valid direct temperature - heater is OFF
                         temps["chamber"] = float(info_temp)
                         temps["chamber_target"] = 0.0  # Direct value means heater off
-                        logger.debug("[%s] info.temp direct: %s°C (heater OFF)", self.serial_number, info_temp)
+                        self._debug_on_change(
+                            "info_temp_direct",
+                            info_temp,
+                            "[%s] info.temp direct: %s°C (heater OFF)",
+                            self.serial_number,
+                            info_temp,
+                        )
             # H2D series: Dual extruder temps are in device.extruder.info array
             # Temperature values are encoded as fixed-point (value / 65536 = °C)
             if "device" in data and isinstance(data["device"], dict):
@@ -2656,7 +2746,13 @@ class BambuMQTTClient:
 
                 # Log ctc_info contents for debugging
                 if ctc_info:
-                    logger.debug("[%s] ctc_info keys: %s", self.serial_number, list(ctc_info.keys()))
+                    self._debug_on_change(
+                        "ctc_info_keys",
+                        tuple(ctc_info.keys()),
+                        "[%s] ctc_info keys: %s",
+                        self.serial_number,
+                        list(ctc_info.keys()),
+                    )
 
                 # FIRST: Parse explicit ctc.info.target if available - this is the authoritative target
                 # (what the slicer shows). This OVERRIDES any previously decoded target.
@@ -2737,14 +2833,31 @@ class BambuMQTTClient:
                     target = self.state.temperatures.get("chamber_target", 0)
 
                 self.state.temperatures["chamber_heating"] = target > 0 and current < target
-                logger.debug(
-                    f"[{self.serial_number}] Chamber heating calculated: target={target}, current={current}, heating={self.state.temperatures['chamber_heating']}, respect_local={respect_local}"
+                self._debug_on_change(
+                    "chamber_heating",
+                    (target, current, self.state.temperatures["chamber_heating"], respect_local),
+                    "[%s] Chamber heating calculated: target=%s, current=%s, heating=%s, respect_local=%s",
+                    self.serial_number,
+                    target,
+                    current,
+                    self.state.temperatures["chamber_heating"],
+                    respect_local,
                 )
 
             # Debug: log chamber value if it was updated
             if "chamber" in temps:
-                logger.debug(
-                    f"[{self.serial_number}] Chamber temp updated to: {self.state.temperatures.get('chamber')}, target: {self.state.temperatures.get('chamber_target')}, heating: {self.state.temperatures.get('chamber_heating')}"
+                self._debug_on_change(
+                    "chamber_temp",
+                    (
+                        self.state.temperatures.get("chamber"),
+                        self.state.temperatures.get("chamber_target"),
+                        self.state.temperatures.get("chamber_heating"),
+                    ),
+                    "[%s] Chamber temp updated to: %s, target: %s, heating: %s",
+                    self.serial_number,
+                    self.state.temperatures.get("chamber"),
+                    self.state.temperatures.get("chamber_target"),
+                    self.state.temperatures.get("chamber_heating"),
                 )
 
             # Calculate nozzle_heating for single nozzle printers (not set by H2D parsing)
@@ -2958,7 +3071,7 @@ class BambuMQTTClient:
         # Parse ipcam/live view status
         if "ipcam" in data:
             ipcam_data = data["ipcam"]
-            logger.debug("[%s] ipcam field: %s", self.serial_number, ipcam_data)
+            self._debug_on_change("ipcam", ipcam_data, "[%s] ipcam field: %s", self.serial_number, ipcam_data)
             if isinstance(ipcam_data, dict):
                 # Check ipcam_record field for live view status
                 self.state.ipcam = ipcam_data.get("ipcam_record") == "enable"
@@ -2980,7 +3093,9 @@ class BambuMQTTClient:
         # Parse WiFi signal strength (dBm)
         if "wifi_signal" in data:
             wifi_signal = data["wifi_signal"]
-            logger.debug("[%s] wifi_signal received: %s", self.serial_number, wifi_signal)
+            self._debug_on_change(
+                "wifi_signal", wifi_signal, "[%s] wifi_signal received: %s", self.serial_number, wifi_signal
+            )
             if isinstance(wifi_signal, (int, float)):
                 self.state.wifi_signal = int(wifi_signal)
             elif isinstance(wifi_signal, str):

backend/app/services/print_scheduler.py

@@ -127,6 +127,15 @@ class _UploadProgressBridge:
 # briefly in SLICING between PREPARE and RUNNING while parsing the g-code.
 _ACTIVE_PRINT_STATES: frozenset[str] = frozenset({"PREPARE", "SLICING", "RUNNING", "PAUSE"})
 
+# How many times the start-watchdog may revert an item to 'pending' before it
+# gives up and fails the row instead (#2555). Each attempt costs a full 3MF
+# re-upload plus the watchdog's wait, so a wedged printer left to retry forever
+# both never recovers and starves the other printers of dispatch slots. Three
+# is chosen to clear the transient causes the watchdog already recovers from —
+# a lost MQTT publish on a half-broken session (#887/#936) is fixed by the
+# force-reconnect on the very next attempt — while still bounding the loop.
+DISPATCH_MAX_ATTEMPTS = 3
+
 # Filament type equivalence groups — types within the same group are
 # interchangeable on the printer side (Bambu Lab firmware treats them as compatible).
 _FILAMENT_TYPE_GROUPS: list[list[str]] = [
@@ -329,6 +338,46 @@ class PrintScheduler:
             # Log skip reasons once per queue check (not per item)
             skip_reasons: dict[str, int] = {}
 
+            # Items selected for dispatch in this pass, one per printer. The
+            # loop below only *decides* — the uploads happen afterwards, in
+            # parallel (#2555). See _dispatch_selected().
+            dispatch_ids: list[int] = []
+
+            # Library rows queued with `cleanup_library_after_dispatch` (the
+            # printer-card "upload and print" flow) are CONSUMED by the dispatch
+            # that prints them: the row is deleted and the 3MF is unlinked from
+            # disk. That was safe only because dispatch was serial. Run two of
+            # them against the same row at once and the second DELETE matches no
+            # row (StaleDataError), and the winner's unlink can pull the file out
+            # from under the loser's in-flight upload.
+            #
+            # Only the cleanup flag mutates the row. An ordinary library print
+            # just reads it, so the common fan-out — one file, many printers,
+            # which is exactly the reporter's workload — still goes out fully in
+            # parallel. Narrow the guard to the mutating case; do not serialise
+            # the case the whole fix exists for.
+            dispatch_libs: set[int] = set()
+            consumed_libs: set[int] = set()
+
+            def _library_row_conflict(candidate: PrintQueueItem) -> bool:
+                """True if dispatching `candidate` now would race another item's cleanup."""
+                lib_id = candidate.library_file_id
+                if lib_id is None:
+                    return False
+                if candidate.cleanup_library_after_dispatch:
+                    # We would delete a row someone else in this pass is reading.
+                    return lib_id in dispatch_libs
+                # Someone else in this pass will delete the row out from under us.
+                return lib_id in consumed_libs
+
+            def _claim_library_row(candidate: PrintQueueItem) -> None:
+                lib_id = candidate.library_file_id
+                if lib_id is None:
+                    return
+                dispatch_libs.add(lib_id)
+                if candidate.cleanup_library_after_dispatch:
+                    consumed_libs.add(lib_id)
+
             for item in items:
                 # Check scheduled time first (scheduled_time is stored in UTC from ISO string)
                 if item.scheduled_time:
@@ -442,8 +491,20 @@ class PrintScheduler:
                     if await self._block_on_filament_deficit(db, item):
                         continue
 
-                    # Start the print
-                    await self._start_print(db, item)
+                    # Hold this item back for the next pass rather than racing
+                    # another dispatch over the same transient library row. The
+                    # printer is still marked busy so a later item does not jump
+                    # its place in this printer's queue.
+                    if _library_row_conflict(item):
+                        skip_reasons["library_row_in_use"] = skip_reasons.get("library_row_in_use", 0) + 1
+                        busy_printers.add(item.printer_id)
+                        continue
+
+                    # Queue the dispatch instead of running it here — see
+                    # _dispatch_selected(). busy_printers still gets the printer
+                    # immediately, so nothing else in this pass can target it.
+                    _claim_library_row(item)
+                    dispatch_ids.append(item.id)
                     busy_printers.add(item.printer_id)
 
                     # SJF starvation guard: mark items that were jumped
@@ -517,6 +578,20 @@ class PrintScheduler:
                             )
 
                     if printer_id:
+                        # Before claiming the printer: hold back rather than race
+                        # another dispatch over the same transient library row.
+                        # Checked here so a held item does not get a printer
+                        # assigned and then sit on it. See _library_row_conflict().
+                        #
+                        # No busy_printers.add() here, unlike the fixed-printer
+                        # branch above: that one protects its printer's own queue
+                        # ordering, but this item was never assigned to `printer_id`
+                        # — the matcher merely offered it. Marking it busy would
+                        # strand an idle printer for the rest of the pass.
+                        if _library_row_conflict(item):
+                            skip_reasons["library_row_in_use"] = skip_reasons.get("library_row_in_use", 0) + 1
+                            continue
+
                         # Check condition (previous print success) before assigning
                         if item.require_previous_success:
                             if not await self._check_previous_success(db, item):
@@ -569,7 +644,8 @@ class PrintScheduler:
                         if await self._block_on_filament_deficit(db, item):
                             continue
 
-                        await self._start_print(db, item)
+                        _claim_library_row(item)
+                        dispatch_ids.append(item.id)
                         busy_printers.add(printer_id)
 
                         # SJF starvation guard: mark model-based items that were jumped
@@ -591,7 +667,10 @@ class PrintScheduler:
                                     other.been_jumped = True
                             await db.commit()
 
-            # Log summary of skip reasons (helps diagnose why queue items aren't starting)
+            # Log the decisions BEFORE dispatching. The dispatch below blocks for
+            # as long as the slowest upload takes (minutes on a big 3MF), and a
+            # skip summary that only lands after the transfers have finished is
+            # useless for working out why an item did not go out.
             if skip_reasons:
                 logger.info("Queue skip summary: %s", skip_reasons)
             if busy_printers:
@@ -609,9 +688,89 @@ class PrintScheduler:
                         awaiting,
                     )
 
+            # Read the concurrency limit BEFORE the commit below, not inside
+            # _dispatch_selected(). A SELECT on this session after the commit
+            # implicitly opens a fresh transaction that nothing then closes, and
+            # it would stay open for the whole dispatch — minutes of "idle in
+            # transaction" on Postgres (pinned MVCC snapshot, vacuum blocked),
+            # and on SQLite a pinned WAL read snapshot that stops the WAL being
+            # checkpointed while every dispatch is writing to it.
+            upload_limit = max(1, await self._get_int_setting(db, "queue_max_concurrent_uploads", default=4))
+
+            # Selection is done; every decision above is recorded on `db`
+            # (model-based printer assignment, computed ams_mapping). Flush it
+            # before the dispatch tasks open their own sessions, or they will
+            # read a row that still says printer_id=None. This also releases the
+            # connection back to the pool for the duration of the dispatch.
+            await db.commit()
+
+            if dispatch_ids:
+                await self._dispatch_selected(dispatch_ids, upload_limit)
+
             # Auto-drying: start drying on idle printers that have no pending queue items
             await self._check_auto_drying(db, items, busy_printers, require_plate_clear=require_plate_clear)
 
+    async def _dispatch_selected(self, item_ids: list[int], limit: int) -> None:
+        """Upload and start every item selected by this queue pass, in parallel.
+
+        Dispatch used to happen inline in the selection loop: ``await
+        _start_print(db, item)`` for each item in turn. Since ``_start_print``
+        performs the FTP upload, that serialized every printer behind every
+        other printer's transfer — even though the printers are entirely
+        independent machines. A Bambu printer's FTP server sustains ~150 KB/s
+        (its own SD write is the bottleneck, not the network), so a 41 MB 3MF
+        takes ~4 minutes. The reporter's 19-printer farm therefore needed ~80
+        minutes before the last printer received its file, and the queue looked
+        like it was starting prints "one by one, very slowly" (#2555).
+
+        Uploads to *different* printers contend for nothing, so they run
+        concurrently here, bounded by ``queue_max_concurrent_uploads``. The
+        bound exists because the printers are independent but the host is not:
+        each in-flight upload holds a thread in the FTP pool, a TLS session and
+        a file handle.
+
+        This is awaited before ``check_queue`` returns, which preserves the
+        invariant the rest of the scheduler is built on: a pass never overlaps
+        with the next one. It matters more than it looks — ``_start_print``
+        flips the row pending -> printing only *after* the upload finishes, so
+        a pass that returned early while uploads were still in flight would let
+        the next pass re-dispatch the very same still-pending rows.
+
+        ``limit`` is read by the caller, on the caller's session, before it
+        commits — reading it here would leave that session idle-in-transaction
+        for the whole dispatch. This function deliberately takes no session.
+        """
+        sem = asyncio.Semaphore(limit)
+
+        async def _one(item_id: int) -> None:
+            # Its own session: these run concurrently, and an AsyncSession is not
+            # safe to share across tasks. It also keeps a slow upload from pinning
+            # the caller's session (and, on SQLite, its transaction) open for the
+            # duration.
+            async with sem, async_session() as item_db:
+                item = await item_db.get(PrintQueueItem, item_id)
+                if not item:
+                    logger.info("Queue item %s vanished before dispatch — skipping", item_id)
+                    return
+                await self._start_print(item_db, item)
+
+        logger.info(
+            "Dispatching %d queue item(s) with up to %d concurrent upload(s): %s",
+            len(item_ids),
+            limit,
+            item_ids,
+        )
+        results = await asyncio.gather(*(_one(i) for i in item_ids), return_exceptions=True)
+
+        # gather() with return_exceptions keeps one printer's failure from
+        # cancelling its siblings' in-flight uploads. _start_print already
+        # handles its own failure modes and marks the item failed; anything
+        # arriving here is unexpected, so log it loudly rather than letting
+        # gather swallow it.
+        for item_id, result in zip(item_ids, results, strict=True):
+            if isinstance(result, BaseException):
+                logger.error("Queue item %s: dispatch raised %s: %s", item_id, type(result).__name__, result)
+
     async def _find_idle_printer_for_model(
         self,
         db: AsyncSession,
@@ -2421,6 +2580,46 @@ class PrintScheduler:
         result = await db.execute(select(Printer).where(Printer.id == printer_id))
         return result.scalar_one_or_none()
 
+    async def _notify_dispatch_gave_up(
+        self,
+        queue_item_id: int,
+        printer_id: int,
+        created_by_id: int | None,
+    ) -> None:
+        """Tell the user the queue item was failed after exhausting its dispatch retries.
+
+        Called from the watchdog, which is a background task with no session of
+        its own — hence the fresh one here. Best-effort throughout: the row is
+        already marked failed and that is the load-bearing part; a notification
+        provider being down must not resurrect the retry loop we just stopped.
+        """
+        try:
+            async with async_session() as db:
+                item = await db.get(PrintQueueItem, queue_item_id)
+                if not item:
+                    return
+                job_name = await self._get_job_name(db, item)
+                printer = await self._get_printer(db, printer_id)
+                await notification_service.on_queue_job_failed(
+                    job_name=job_name,
+                    printer_id=printer_id,
+                    printer_name=printer.name if printer else "Unknown",
+                    reason="Printer accepted the file but never started printing",
+                    db=db,
+                )
+        except Exception as e:
+            logger.warning("Queue item %s: give-up notification failed: %s", queue_item_id, e)
+
+        try:
+            await ws_manager.send_queue_item_failed(
+                user_id=created_by_id,
+                queue_item_id=queue_item_id,
+                printer_id=printer_id,
+                reason="never_started",
+            )
+        except Exception:
+            pass  # toast is best-effort
+
     async def _block_on_filament_deficit(
         self,
         db: AsyncSession,
@@ -3063,12 +3262,24 @@ class PrintScheduler:
                     name=f"watchdog-print-start-{item.id}",
                 )
 
-            # Get estimated time for notification
+            # Get estimated time for notification.
+            #
+            # This used to fall back to `library_file.print_time_seconds`, a column
+            # LibraryFile does not have — the print time it knows about lives in
+            # `file_metadata`. So a library print whose archive carried no parseable
+            # print time (a plain .gcode, or a 3MF the parser could not read) raised
+            # AttributeError right here, *after* the printer had already been sent
+            # the job: the started-notification never fired, and the exception
+            # unwound the whole queue pass, so every other printer still waiting to
+            # be dispatched on that tick silently missed its turn.
+            #
+            # The queue item caches the print time at creation ("Cached from
+            # archive/library"), which is the value this was reaching for.
             estimated_time = None
             if archive and archive.print_time_seconds:
                 estimated_time = archive.print_time_seconds
-            elif library_file and library_file.print_time_seconds:
-                estimated_time = library_file.print_time_seconds
+            elif item.print_time_seconds:
+                estimated_time = item.print_time_seconds
 
             # Send job started notification
             await notification_service.on_queue_job_started(
@@ -3241,8 +3452,10 @@ class PrintScheduler:
         # Drop the in-memory hold so the retry isn't blocked by it.
         scheduler._release_dispatch_hold(printer_id)
 
-        # Three outcomes from the revert attempt, each routed differently:
+        # Four outcomes from the revert attempt, each routed differently:
         #   "reverted":          row flipped from printing -> pending, run recovery
+        #   "gave_up":           same, but the retry budget is spent — row failed
+        #                        rather than pending, so it stops going round again
         #   "already_moved_on":  item.status != 'printing' (completed/cancelled by
         #                        on_print_complete or user). Skip recovery entirely
         #                        — the print clearly landed somewhere even if the
@@ -3250,12 +3463,30 @@ class PrintScheduler:
         #   "revert_failed":     SQLite contention exhausted retries. Still run
         #                        recovery so the MQTT session gets a fresh client_id
         #                        on the half-broken-session path.
+        #
+        # The retry budget (#2555): reverting to 'pending' hands the item straight
+        # back to the next queue pass, which re-uploads the whole 3MF and waits out
+        # the watchdog again. For a printer that is genuinely wedged that loop never
+        # ends — the reporter had one printer "since this morning still not launch"
+        # — and each lap also consumes an upload slot that the other printers in the
+        # farm are waiting on. Retrying is right; retrying forever is not.
         async def _do_revert(db):
             item = await db.get(PrintQueueItem, queue_item_id)
             if not item or item.status != "printing":
                 return "already_moved_on"
-            item.status = "pending"
+            item.dispatch_attempts = (item.dispatch_attempts or 0) + 1
             item.started_at = None
+            if item.dispatch_attempts >= DISPATCH_MAX_ATTEMPTS:
+                item.status = "failed"
+                item.error_message = (
+                    f"The printer accepted the file but never started printing, after "
+                    f"{item.dispatch_attempts} attempts. Check the printer's screen for a "
+                    f"prompt or error, confirm its SD card is readable, and start the job again."
+                )
+                item.completed_at = datetime.now(timezone.utc)
+                await db.commit()
+                return "gave_up"
+            item.status = "pending"
             await db.commit()
             return "reverted"
 
@@ -3279,7 +3510,18 @@ class PrintScheduler:
             return
 
         total_timeout = timeout + (phase_b_timeout if landed_on_subtask else 0.0)
-        if revert_outcome == "reverted":
+        if revert_outcome == "gave_up":
+            logger.error(
+                "Queue item %s: printer %d never started the print after %d dispatch "
+                "attempts (last one waited %.0fs) — marking the item failed instead of "
+                "re-uploading it again (#2555)",
+                queue_item_id,
+                printer_id,
+                DISPATCH_MAX_ATTEMPTS,
+                total_timeout,
+            )
+            await scheduler._notify_dispatch_gave_up(queue_item_id, printer_id, created_by_id)
+        elif revert_outcome == "reverted":
             if landed_on_subtask:
                 logger.warning(
                     "Queue item %s: printer %d accepted project_file (subtask_id "

backend/tests/unit/test_mqtt_debug_on_change.py

@@ -0,0 +1,183 @@
+"""Per-frame debug dumps must log transitions, not every frame (#2555).
+
+The state dumps in the push_status handler fired whenever their field was
+*present* in the frame. A full push_status carries every field, so they fired on
+every frame regardless of whether anything had changed — several while their own
+comment claimed to log "when X changes".
+
+On one printer that is ~1.5 lines/s and nobody noticed. On a 19-printer farm it
+is ~100 lines/s: the reporter turned on debug logging as asked, and the 5 MB log
+rolled over in under five minutes. 27,727 of the 29,830 lines in the support
+bundle were these dumps, and the queue problem we were chasing was nowhere in the
+window.
+"""
+
+import logging
+from unittest.mock import MagicMock, patch
+
+from backend.app.services.bambu_mqtt import BambuMQTTClient
+
+
+def _client() -> BambuMQTTClient:
+    return BambuMQTTClient(ip_address="10.0.0.1", serial_number="SERIAL", access_code="code", model="A1")
+
+
+class TestDebugOnChange:
+    def test_repeated_identical_values_log_once(self):
+        client = _client()
+        with patch("backend.app.services.bambu_mqtt.logger") as log:
+            for _ in range(50):
+                client._debug_on_change("wifi_signal", -52, "[%s] wifi_signal: %s", "SERIAL", -52)
+
+        assert log.debug.call_count == 1, (
+            f"50 identical frames produced {log.debug.call_count} log lines — this is the flood"
+        )
+
+    def test_each_change_is_logged(self):
+        """Suppressing repeats must not suppress transitions — the transitions are
+        the entire reason anyone reads these lines."""
+        client = _client()
+        with patch("backend.app.services.bambu_mqtt.logger") as log:
+            for value in (-52, -52, -60, -60, -52):
+                client._debug_on_change("wifi_signal", value, "[%s] wifi_signal: %s", "SERIAL", value)
+
+        assert log.debug.call_count == 3
+        assert [c.args[-1] for c in log.debug.call_args_list] == [-52, -60, -52]
+
+    def test_keys_are_tracked_independently(self):
+        client = _client()
+        with patch("backend.app.services.bambu_mqtt.logger") as log:
+            client._debug_on_change("tray_now", 1, "tray_now: %s", 1)
+            client._debug_on_change("ams_status", 1, "ams_status: %s", 1)
+            client._debug_on_change("tray_now", 1, "tray_now: %s", 1)  # repeat, suppressed
+
+        assert log.debug.call_count == 2, "same value under a different key must not be swallowed"
+
+    def test_printers_are_tracked_independently(self):
+        """State is per-client. Two printers reporting the same value must each
+        get their own line — a farm is exactly where this matters."""
+        a, b = _client(), _client()
+        with patch("backend.app.services.bambu_mqtt.logger") as log:
+            a._debug_on_change("tray_now", 3, "tray_now: %s", 3)
+            b._debug_on_change("tray_now", 3, "tray_now: %s", 3)
+
+        assert log.debug.call_count == 2
+
+    def test_composite_values_detect_a_change_in_any_field(self):
+        """Messages that render several fields must pass all of them, or a change
+        in the unwatched field is silently dropped."""
+        client = _client()
+        with patch("backend.app.services.bambu_mqtt.logger") as log:
+            client._debug_on_change("chamber", (40.0, 0.0, False), "chamber %s %s %s", 40.0, 0.0, False)
+            client._debug_on_change("chamber", (40.0, 60.0, True), "chamber %s %s %s", 40.0, 60.0, True)
+
+        assert log.debug.call_count == 2, "target/heating changed while current stayed 40.0 — must still log"
+
+    def test_dict_values_compare_by_content(self):
+        """The AMS dict dump is the biggest line by volume; it is a fresh dict every
+        frame, so identity comparison would never suppress anything."""
+        client = _client()
+        with patch("backend.app.services.bambu_mqtt.logger") as log:
+            for _ in range(10):
+                client._debug_on_change("ams", {"tray_now": "0", "bits": "7000000"}, "ams: %s", {})
+            client._debug_on_change("ams", {"tray_now": "1", "bits": "7000000"}, "ams: %s", {})
+
+        assert log.debug.call_count == 2
+
+
+class TestRuntimeDebugToggle:
+    """Debug logging is turned on at RUNTIME (POST /support/debug-logging) and these
+    clients outlive the toggle — which is the whole workflow this change serves:
+    "enable debug logging, reproduce, send the bundle".
+
+    So the cache must not be warmed while running at INFO. If it were, the operator
+    would enable debug, and every steady-state value would already be "seen" — an
+    idle printer's bundle would contain none of these lines at all, which is worse
+    than the flood it replaced.
+    """
+
+    def test_enabling_debug_at_runtime_still_dumps_a_baseline(self):
+        client = _client()
+        mqtt_logger = logging.getLogger("backend.app.services.bambu_mqtt")
+        original = mqtt_logger.level
+        try:
+            # Steady state at INFO: the app has been running for hours.
+            mqtt_logger.setLevel(logging.INFO)
+            with patch("backend.app.services.bambu_mqtt.logger", wraps=mqtt_logger) as log:
+                for _ in range(200):
+                    client._debug_on_change("wifi_signal", -52, "wifi %s", -52)
+                assert log.debug.call_count == 0, "nothing should be emitted at INFO"
+
+            # Operator flips debug on. The value has NOT changed — but they turned
+            # this on to see the printer's state, so the very next frame must dump it.
+            mqtt_logger.setLevel(logging.DEBUG)
+            with patch("backend.app.services.bambu_mqtt.logger", wraps=mqtt_logger) as log:
+                client._debug_on_change("wifi_signal", -52, "wifi %s", -52)
+                assert log.debug.call_count == 1, (
+                    "no baseline after enabling debug — the cache was warmed while at "
+                    "INFO, so the operator sees nothing until the value happens to change"
+                )
+                # ...and it still dedups from there.
+                for _ in range(50):
+                    client._debug_on_change("wifi_signal", -52, "wifi %s", -52)
+                assert log.debug.call_count == 1
+        finally:
+            mqtt_logger.setLevel(original)
+
+    def test_disabling_debug_drops_the_cache(self):
+        """Off -> on must be as cold as a fresh process, not just first-ever-on."""
+        client = _client()
+        mqtt_logger = logging.getLogger("backend.app.services.bambu_mqtt")
+        original = mqtt_logger.level
+        try:
+            mqtt_logger.setLevel(logging.DEBUG)
+            client._debug_on_change("tray_now", 2, "tray_now %s", 2)
+            assert client._debug_last
+
+            mqtt_logger.setLevel(logging.INFO)
+            client._debug_on_change("tray_now", 2, "tray_now %s", 2)
+            assert client._debug_last == {}, "cache must be dropped while debug is off"
+        finally:
+            mqtt_logger.setLevel(original)
+
+
+class TestRealDumpSitesAreGated:
+    """End-to-end: feed the same push_status frame twice and count the lines."""
+
+    def test_identical_push_status_frames_do_not_re_dump_state(self):
+        # Deliberately the client's real PrinterState, not a mock: a MagicMock
+        # state would return the same stub object for every attribute read, so
+        # the values would compare equal and the test would pass even with the
+        # gating removed.
+        client = _client()
+        assert not isinstance(client.state, MagicMock)
+        frame = {
+            "print": {
+                "ams": {
+                    "ams": [],
+                    "ams_exist_bits": "1",
+                    "tray_exist_bits": "f",
+                    "tray_now": "0",
+                },
+                "wifi_signal": "-52dBm",
+                "ipcam": {"ipcam_record": "enable"},
+            }
+        }
+
+        logging.getLogger("backend.app.services.bambu_mqtt").setLevel(logging.DEBUG)
+        with patch("backend.app.services.bambu_mqtt.logger") as log:
+            log.isEnabledFor.return_value = True
+            client._process_message(dict(frame))
+            first = [c.args[0] for c in log.debug.call_args_list]
+            log.debug.reset_mock()
+            client._process_message(dict(frame))
+            second = [c.args[0] for c in log.debug.call_args_list]
+
+        # Frame 1 must still dump — the point is to log transitions, not to go quiet.
+        assert first, "the first frame stopped dumping state entirely — the logs are now useless"
+
+        # Frame 2 is byte-identical, so it must produce NOTHING. Asserting merely
+        # "fewer than frame 1" is not enough: a couple of these sites happen to be
+        # naturally one-shot, so an ungated build still measures 3 < 5 and the
+        # assertion passes while every real dump keeps firing on every frame.
+        assert second == [], f"an identical push_status frame re-dumped {len(second)} line(s): {second}"

backend/tests/unit/test_scheduler_concurrent_dispatch.py

@@ -0,0 +1,468 @@
+"""Concurrent queue dispatch across printers (#2555).
+
+Reported as "prints are sent to the printer one by one, very slowly" on a
+19-printer farm — up to an hour before the last printer started. Not a config
+problem: ``check_queue`` awaited ``_start_print`` inline for each pending item,
+and ``_start_print`` performs the FTP upload, so every printer queued behind
+every other printer's transfer. A Bambu printer's FTP server sustains ~150 KB/s
+(its own SD write is the bottleneck, not the network), so the reporter's 41 MB
+3MF took ~254 s *per printer* — 19 of those in series is ~80 minutes.
+
+Printers are independent machines, so the uploads have no reason to be
+serialized. They now run concurrently, capped by ``queue_max_concurrent_uploads``.
+
+What must stay true:
+
+* Uploads to different printers overlap in time (the actual fix).
+* No more than ``queue_max_concurrent_uploads`` run at once (the host is not
+  infinite: each in-flight upload holds an FTP thread, a TLS session, a handle).
+* Setting it to 1 restores exactly the old serial behaviour.
+* One printer failing must not cancel its siblings' in-flight uploads.
+* A pass still never overlaps with the next one — ``_start_print`` flips the row
+  pending -> printing only *after* the upload completes, so returning early
+  while uploads were in flight would let the next pass re-dispatch the same rows.
+"""
+
+import asyncio
+from contextlib import ExitStack
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+
+import backend.app.models  # noqa: F401 - populate Base.metadata
+import backend.app.services.archive as archive_module
+import backend.app.services.print_scheduler as scheduler_module
+from backend.app.core.database import Base
+from backend.app.models.archive import PrintArchive
+from backend.app.models.library import LibraryFile
+from backend.app.models.print_queue import PrintQueueItem
+from backend.app.models.printer import Printer
+from backend.app.models.settings import Settings
+from backend.app.services.print_scheduler import PrintScheduler
+
+UPLOAD_SECONDS = 0.15
+
+
+@pytest.fixture
+async def farm(tmp_path):
+    """Build a farm of N printers, each with one pending queue item."""
+    engine = create_async_engine("sqlite+aiosqlite:///:memory:", echo=False)
+    async with engine.begin() as conn:
+        await conn.run_sync(Base.metadata.create_all)
+    session_maker = async_sessionmaker(engine, expire_on_commit=False)
+
+    async def make_farm(printer_count: int, *, max_concurrent: int | None = None):
+        base_dir = tmp_path / "farm"
+        (base_dir / "archives").mkdir(parents=True, exist_ok=True)
+
+        async with session_maker() as db:
+            if max_concurrent is not None:
+                db.add(Settings(key="queue_max_concurrent_uploads", value=str(max_concurrent)))
+
+            printer_ids = []
+            for n in range(printer_count):
+                archive_rel = Path("archives") / f"job-{n}.3mf"
+                (base_dir / archive_rel).write_bytes(b"archive payload")
+
+                printer = Printer(
+                    name=f"Printer {n}",
+                    serial_number=f"SERIAL-{n}",
+                    ip_address=f"10.0.0.{n + 1}",
+                    access_code="access-code",
+                    model="A1",
+                )
+                db.add(printer)
+                await db.flush()
+
+                archive = PrintArchive(
+                    printer_id=printer.id,
+                    filename=f"job-{n}.3mf",
+                    file_path=str(archive_rel),
+                    file_size=15,
+                    print_time_seconds=120,
+                    status="completed",
+                )
+                db.add(archive)
+                await db.flush()
+
+                db.add(
+                    PrintQueueItem(
+                        printer_id=printer.id,
+                        archive_id=archive.id,
+                        status="pending",
+                        position=n,
+                    )
+                )
+                printer_ids.append(printer.id)
+            await db.commit()
+
+        return SimpleNamespace(
+            session_maker=session_maker,
+            base_dir=base_dir,
+            printer_ids=printer_ids,
+        )
+
+    try:
+        yield make_farm
+    finally:
+        await engine.dispose()
+
+
+class _UploadRecorder:
+    """Stands in for ``upload_file_async``; records overlap.
+
+    Each call sleeps, so genuinely concurrent uploads have overlapping
+    lifetimes. ``peak`` is the high-water mark of simultaneous in-flight
+    uploads — the number the whole fix turns on.
+    """
+
+    def __init__(self, *, fail_for_ip: str | None = None):
+        self.in_flight = 0
+        self.peak = 0
+        self.order: list[str] = []
+        self.fail_for_ip = fail_for_ip
+
+    async def __call__(self, ip_address, access_code, local_path, remote_path, **kwargs):
+        self.in_flight += 1
+        self.peak = max(self.peak, self.in_flight)
+        self.order.append(ip_address)
+        try:
+            await asyncio.sleep(UPLOAD_SECONDS)
+            if self.fail_for_ip is not None and ip_address == self.fail_for_ip:
+                raise OSError(f"simulated FTP failure for {ip_address}")
+            return True
+        finally:
+            self.in_flight -= 1
+
+
+async def _run_check_queue(ctx, upload, job_started=None):
+    scheduler = PrintScheduler()
+    job_started = job_started or AsyncMock()
+
+    patches = [
+        patch.object(scheduler_module.settings, "base_dir", ctx.base_dir),
+        # The library-file path archives the 3MF before uploading it, and the
+        # archive service resolves its own settings — redirect both or it writes
+        # into the real repo and then fails relative_to(base_dir).
+        patch.object(archive_module.settings, "base_dir", ctx.base_dir),
+        patch.object(archive_module.settings, "archive_dir", ctx.base_dir / "archive"),
+        patch("backend.app.services.print_scheduler.async_session", ctx.session_maker),
+        patch("backend.app.core.database.async_session", ctx.session_maker),
+        patch("backend.app.services.print_scheduler.printer_manager.is_connected", MagicMock(return_value=True)),
+        patch("backend.app.services.print_scheduler.printer_manager.get_status", MagicMock(return_value=None)),
+        patch("backend.app.services.print_scheduler.printer_manager.start_print", MagicMock(return_value=True)),
+        patch("backend.app.services.print_scheduler.printer_manager.set_awaiting_plate_clear", MagicMock()),
+        patch("backend.app.services.print_scheduler.upload_file_async", upload),
+        patch("backend.app.services.print_scheduler.delete_file_async", AsyncMock(return_value=True)),
+        patch(
+            "backend.app.services.print_scheduler.get_ftp_retry_settings",
+            AsyncMock(return_value=(False, 0, 0, 1.0)),
+        ),
+        patch("backend.app.services.print_scheduler.cache_3mf_download", MagicMock()),
+        patch("backend.app.services.print_scheduler.spawn_background_task", MagicMock()),
+        patch("backend.app.services.notification_service.notification_service.on_queue_job_started", job_started),
+        patch("backend.app.services.notification_service.notification_service.on_queue_job_failed", AsyncMock()),
+        patch("backend.app.services.mqtt_relay.mqtt_relay.on_queue_job_started", AsyncMock()),
+        patch.object(scheduler, "_is_printer_idle", MagicMock(return_value=True)),
+        patch.object(scheduler, "_propagate_owner_to_printer_manager", AsyncMock()),
+        patch.object(scheduler, "_power_off_if_needed", AsyncMock()),
+        patch.object(scheduler, "_preheat_and_soak", AsyncMock()),
+        patch.object(scheduler, "_check_auto_drying", AsyncMock()),
+    ]
+
+    with ExitStack() as stack:
+        for patcher in patches:
+            stack.enter_context(patcher)
+        await scheduler.check_queue()
+
+
+async def _statuses(ctx):
+    async with ctx.session_maker() as db:
+        rows = (await db.execute(select(PrintQueueItem).order_by(PrintQueueItem.position))).scalars().all()
+        return [r.status for r in rows]
+
+
+@pytest.mark.asyncio
+async def test_uploads_to_different_printers_overlap(farm):
+    """The headline fix: six printers must not queue behind each other.
+
+    Pre-fix this recorded peak == 1 no matter how many printers were pending.
+    """
+    ctx = await farm(6, max_concurrent=6)
+    upload = _UploadRecorder()
+
+    await _run_check_queue(ctx, upload)
+
+    assert upload.peak == 6, (
+        f"expected all 6 printers to be uploaded to concurrently, but the "
+        f"high-water mark was {upload.peak} — uploads are still serialized"
+    )
+    assert await _statuses(ctx) == ["printing"] * 6
+
+
+@pytest.mark.asyncio
+async def test_concurrency_is_capped_by_the_setting(farm):
+    """Eight pending printers, cap of 3 — never more than 3 uploads at once.
+
+    The cap is the reason this is a setting and not just ``asyncio.gather``:
+    the printers are independent but the Bambuddy host is not.
+    """
+    ctx = await farm(8, max_concurrent=3)
+    upload = _UploadRecorder()
+
+    await _run_check_queue(ctx, upload)
+
+    assert upload.peak == 3, f"cap of 3 not honoured — peak was {upload.peak}"
+    assert len(upload.order) == 8, "every pending item must still be dispatched, just not all at once"
+    assert await _statuses(ctx) == ["printing"] * 8
+
+
+@pytest.mark.asyncio
+async def test_limit_of_one_restores_serial_behaviour(farm):
+    """An escape hatch for weak networks: 1 == the pre-#2555 behaviour."""
+    ctx = await farm(4, max_concurrent=1)
+    upload = _UploadRecorder()
+
+    await _run_check_queue(ctx, upload)
+
+    assert upload.peak == 1
+    assert await _statuses(ctx) == ["printing"] * 4
+
+
+@pytest.mark.asyncio
+async def test_default_concurrency_applies_when_setting_absent(farm):
+    """No Settings row (every existing install) must still dispatch in parallel.
+
+    The whole point is that the reporter's farm gets faster *without* him having
+    to find a new setting first. Default is 4.
+    """
+    ctx = await farm(5, max_concurrent=None)
+    upload = _UploadRecorder()
+
+    await _run_check_queue(ctx, upload)
+
+    assert upload.peak == 4, f"expected the default cap of 4, got {upload.peak}"
+    assert await _statuses(ctx) == ["printing"] * 5
+
+
+@pytest.mark.asyncio
+async def test_one_failing_upload_does_not_cancel_the_others(farm):
+    """A dead printer must not take its siblings' in-flight uploads down with it.
+
+    ``asyncio.gather`` without ``return_exceptions=True`` cancels every sibling
+    task the moment one raises — which would mean a single unreachable printer
+    silently aborts the whole batch mid-transfer.
+    """
+    ctx = await farm(4, max_concurrent=4)
+    upload = _UploadRecorder(fail_for_ip="10.0.0.2")  # printer index 1
+
+    await _run_check_queue(ctx, upload)
+
+    statuses = await _statuses(ctx)
+    assert statuses[1] == "failed", "the unreachable printer's item should be marked failed"
+    assert [s for i, s in enumerate(statuses) if i != 1] == ["printing"] * 3, (
+        "the other three printers must have started despite the failure"
+    )
+
+
+@pytest.mark.asyncio
+async def test_check_queue_awaits_its_dispatches_before_returning(farm):
+    """The pass must not return while uploads are still in flight.
+
+    ``_start_print`` flips the row pending -> printing only *after* the upload
+    finishes. If ``check_queue`` returned early, the next 30-second tick would
+    still see those rows as ``pending`` on an idle-looking printer and dispatch
+    them a second time.
+    """
+    ctx = await farm(3, max_concurrent=3)
+    upload = _UploadRecorder()
+
+    await _run_check_queue(ctx, upload)
+
+    assert upload.in_flight == 0, "check_queue returned with uploads still running"
+    assert await _statuses(ctx) == ["printing"] * 3
+
+
+class TestSharedLibraryRow:
+    """Dispatching in parallel means two items can now reach the same library row
+    at the same time — impossible when dispatch was serial.
+
+    Only the ``cleanup_library_after_dispatch`` flow (printer-card "upload and
+    print") *mutates* that row: it deletes it and unlinks the 3MF from disk once
+    the print is away. Two of those against one row would race — the loser's
+    DELETE matches no row, and the winner's unlink can pull the file out from
+    under the loser's in-flight upload.
+
+    An ordinary library print only reads the row. That distinction is load-bearing:
+    the reporter's own batch was one File Manager file fanned out across his farm
+    (both of the queue items in his log point at library file 116), so a blanket
+    "never share a library row" guard would re-serialize the exact workload this
+    change exists to fix.
+    """
+
+    @staticmethod
+    async def _library_farm(session_maker, tmp_path, printer_count, *, cleanup: bool):
+        """One shared library file, one queue item per printer, all pointing at it."""
+        base_dir = tmp_path / "libfarm"
+        (base_dir / "library").mkdir(parents=True, exist_ok=True)
+        shared = base_dir / "library" / "shared.3mf"
+        shared.write_bytes(b"shared payload")
+
+        async with session_maker() as db:
+            db.add(Settings(key="queue_max_concurrent_uploads", value=str(printer_count)))
+            library_file = LibraryFile(
+                filename="shared.3mf",
+                file_path=str(shared),
+                file_type="3mf",
+                file_size=shared.stat().st_size,
+            )
+            db.add(library_file)
+            await db.flush()
+
+            for n in range(printer_count):
+                printer = Printer(
+                    name=f"Printer {n}",
+                    serial_number=f"LIB-SERIAL-{n}",
+                    ip_address=f"10.1.0.{n + 1}",
+                    access_code="access-code",
+                    model="A1",
+                )
+                db.add(printer)
+                await db.flush()
+                db.add(
+                    PrintQueueItem(
+                        printer_id=printer.id,
+                        library_file_id=library_file.id,
+                        cleanup_library_after_dispatch=cleanup,
+                        status="pending",
+                        position=n,
+                    )
+                )
+            await db.commit()
+
+        return SimpleNamespace(session_maker=session_maker, base_dir=base_dir, printer_ids=None)
+
+    @pytest.mark.asyncio
+    async def test_plain_library_file_still_fans_out_in_parallel(self, tmp_path):
+        """The reporter's actual workload: one File Manager file, four printers.
+
+        Nothing here mutates the library row, so all four must upload at once. If
+        this ever drops to 1 the headline fix is gone.
+        """
+        engine = create_async_engine("sqlite+aiosqlite:///:memory:", echo=False)
+        async with engine.begin() as conn:
+            await conn.run_sync(Base.metadata.create_all)
+        session_maker = async_sessionmaker(engine, expire_on_commit=False)
+        try:
+            ctx = await self._library_farm(session_maker, tmp_path, 4, cleanup=False)
+            upload = _UploadRecorder()
+
+            await _run_check_queue(ctx, upload)
+
+            assert upload.peak == 4, f"a shared library file must not re-serialize the fan-out — peak was {upload.peak}"
+            assert await _statuses(ctx) == ["printing"] * 4
+        finally:
+            await engine.dispose()
+
+    @pytest.mark.asyncio
+    async def test_cleanup_items_never_share_a_row_in_one_pass(self, tmp_path):
+        """The mutating flow must be held to one dispatch per pass.
+
+        Each of these deletes the library row and unlinks the 3MF when it is done.
+        Exactly one may go per pass; the rest stay pending for a later one.
+        """
+        engine = create_async_engine("sqlite+aiosqlite:///:memory:", echo=False)
+        async with engine.begin() as conn:
+            await conn.run_sync(Base.metadata.create_all)
+        session_maker = async_sessionmaker(engine, expire_on_commit=False)
+        try:
+            ctx = await self._library_farm(session_maker, tmp_path, 3, cleanup=True)
+            upload = _UploadRecorder()
+
+            await _run_check_queue(ctx, upload)
+
+            assert upload.peak <= 1, (
+                f"{upload.peak} dispatches raced over one consumable library row — "
+                f"the loser's DELETE finds nothing and its 3MF can be unlinked mid-upload"
+            )
+            statuses = await _statuses(ctx)
+            assert statuses.count("printing") == 1, "exactly one item should have gone out"
+            assert statuses.count("pending") == 2, "the rest must stay queued, not fail"
+        finally:
+            await engine.dispose()
+
+
+@pytest.mark.asyncio
+async def test_library_print_without_a_parseable_print_time_does_not_crash(tmp_path):
+    """Regression: `_start_print` read `library_file.print_time_seconds`, a column
+    LibraryFile does not have.
+
+    It only fired when the archive carried no print time — a plain .gcode, or a 3MF
+    the parser could not read — and it fired *after* the printer had been sent the
+    job. The started-notification was lost, and the AttributeError unwound the whole
+    queue pass, so every other printer still waiting to be dispatched on that tick
+    silently missed its turn. Exactly the "why did only some of them start" shape.
+
+    Two printers here: if the first one's dispatch blows up, the second must still
+    go out.
+    """
+    engine = create_async_engine("sqlite+aiosqlite:///:memory:", echo=False)
+    async with engine.begin() as conn:
+        await conn.run_sync(Base.metadata.create_all)
+    session_maker = async_sessionmaker(engine, expire_on_commit=False)
+    try:
+        base_dir = tmp_path / "nolibtime"
+        (base_dir / "library").mkdir(parents=True, exist_ok=True)
+
+        async with session_maker() as db:
+            db.add(Settings(key="queue_max_concurrent_uploads", value="2"))
+            for n in range(2):
+                src = base_dir / "library" / f"job-{n}.gcode"
+                src.write_bytes(b"G28\n")
+                lib = LibraryFile(
+                    filename=f"job-{n}.gcode",
+                    file_path=str(src),
+                    file_type="gcode",
+                    file_size=src.stat().st_size,
+                )
+                db.add(lib)
+                printer = Printer(
+                    name=f"Printer {n}",
+                    serial_number=f"NT-{n}",
+                    ip_address=f"10.2.0.{n + 1}",
+                    access_code="access-code",
+                    model="A1",
+                )
+                db.add(printer)
+                await db.flush()
+                db.add(
+                    PrintQueueItem(
+                        printer_id=printer.id,
+                        library_file_id=lib.id,
+                        status="pending",
+                        position=n,
+                        print_time_seconds=None,  # nothing cached either — the crashing shape
+                    )
+                )
+            await db.commit()
+
+        ctx = SimpleNamespace(session_maker=session_maker, base_dir=base_dir, printer_ids=None)
+        job_started = AsyncMock()
+
+        await _run_check_queue(ctx, _UploadRecorder(), job_started=job_started)
+
+        assert await _statuses(ctx) == ["printing", "printing"]
+
+        # The status flip happens BEFORE the crash point, so it is not the signal —
+        # both rows read "printing" even with the bug present. The started-notification
+        # is emitted just after it, and is what the AttributeError actually destroyed.
+        assert job_started.await_count == 2, (
+            "the job-started notification was lost — _start_print raised after the "
+            "printer had already been sent the job"
+        )
+    finally:
+        await engine.dispose()

backend/tests/unit/test_scheduler_watchdog.py

@@ -14,12 +14,12 @@ tick.
 """
 
 from types import SimpleNamespace
-from unittest.mock import MagicMock, patch
+from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
 from backend.app.models.print_queue import PrintQueueItem
-from backend.app.services.print_scheduler import PrintScheduler
+from backend.app.services.print_scheduler import DISPATCH_MAX_ATTEMPTS, PrintScheduler
 
 
 @pytest.fixture
@@ -489,3 +489,121 @@ class TestGcodeFileDiscriminator:
             )
 
         client.force_reconnect_stale_session.assert_called_once()
+
+
+class TestWatchdogRetryBudget:
+    """A revert hands the item straight back to the next queue pass, which
+    re-uploads the whole 3MF and waits the watchdog out again. For a printer
+    that is genuinely wedged that loop never terminates — the #2555 reporter had
+    one printer "since this morning still not launch" — and every lap also burns
+    an upload slot the rest of the farm is queueing for. Retrying is right;
+    retrying forever is not.
+    """
+
+    @staticmethod
+    async def _wedge(db_session, *, item_id: int = 1):
+        """Run one watchdog cycle against a printer that accepts but never starts."""
+        get_status = MagicMock(return_value=_status("IDLE", "NEW_SUBTASK", gcode_file="/new.3mf"))
+        get_client = MagicMock(return_value=MagicMock())
+
+        with (
+            patch("backend.app.services.print_scheduler.printer_manager.get_status", get_status),
+            patch("backend.app.services.print_scheduler.printer_manager.get_client", get_client),
+            patch("backend.app.services.print_scheduler.async_session", db_session),
+            patch("backend.app.core.database.async_session", db_session),
+            patch(
+                "backend.app.services.notification_service.notification_service.on_queue_job_failed",
+                AsyncMock(),
+            ) as notify,
+        ):
+            await PrintScheduler._watchdog_print_start(
+                queue_item_id=item_id,
+                printer_id=42,
+                pre_state="IDLE",
+                pre_subtask_id="OLD_SUBTASK",
+                pre_gcode_file="/old.3mf",
+                timeout=0.2,
+                phase_b_timeout=0.2,
+                poll_interval=0.05,
+            )
+        return notify
+
+    @pytest.mark.asyncio
+    async def test_early_wedges_still_revert_for_retry(self, db_session):
+        """Attempts below the budget must keep the existing #1678 behaviour.
+
+        The transient causes are real and the watchdog already recovers from
+        them (a publish lost on a half-broken session is fixed by the forced
+        reconnect on the very next attempt), so the first wedges must not fail
+        the job.
+        """
+        await self._wedge(db_session)
+
+        async with db_session() as db:
+            item = await db.get(PrintQueueItem, 1)
+            assert item.status == "pending", "first wedge must still be retried"
+            assert item.dispatch_attempts == 1
+            assert item.started_at is None
+
+    @pytest.mark.asyncio
+    async def test_attempts_accumulate_across_wedges(self, db_session):
+        """The counter is what bounds the loop, so it must survive the revert."""
+        for expected in (1, 2):
+            # Each pass starts from a fresh dispatch, i.e. the row is 'printing' again.
+            async with db_session() as db:
+                item = await db.get(PrintQueueItem, 1)
+                item.status = "printing"
+                await db.commit()
+
+            await self._wedge(db_session)
+
+            async with db_session() as db:
+                item = await db.get(PrintQueueItem, 1)
+                assert item.dispatch_attempts == expected
+                assert item.status == "pending"
+
+    @pytest.mark.asyncio
+    async def test_gives_up_and_fails_the_item_at_the_budget(self, db_session):
+        """The third wedge fails the row instead of queueing a fourth re-upload."""
+        notify = None
+        for _ in range(DISPATCH_MAX_ATTEMPTS):
+            async with db_session() as db:
+                item = await db.get(PrintQueueItem, 1)
+                item.status = "printing"
+                await db.commit()
+            notify = await self._wedge(db_session)
+
+        async with db_session() as db:
+            item = await db.get(PrintQueueItem, 1)
+            assert item.status == "failed", f"after {DISPATCH_MAX_ATTEMPTS} wedges the item must stop going round again"
+            assert item.dispatch_attempts == DISPATCH_MAX_ATTEMPTS
+            assert item.completed_at is not None
+            # The message has to tell the user where to look — the fault is on
+            # the printer, and no amount of retrying from our side will fix it.
+            assert "never started printing" in item.error_message
+
+        notify.assert_awaited_once()
+
+    @pytest.mark.asyncio
+    async def test_a_successful_start_never_touches_the_counter(self, db_session):
+        """Only the revert path increments. A printer that picks the job up
+        must not accumulate attempts towards a future give-up."""
+        get_status = MagicMock(return_value=_status("RUNNING", "NEW_SUBTASK"))
+        with (
+            patch("backend.app.services.print_scheduler.printer_manager.get_status", get_status),
+            patch("backend.app.services.print_scheduler.async_session", db_session),
+            patch("backend.app.core.database.async_session", db_session),
+        ):
+            await PrintScheduler._watchdog_print_start(
+                queue_item_id=1,
+                printer_id=42,
+                pre_state="IDLE",
+                pre_subtask_id="OLD_SUBTASK",
+                timeout=0.2,
+                poll_interval=0.05,
+            )
+
+        async with db_session() as db:
+            item = await db.get(PrintQueueItem, 1)
+            assert item.status == "printing"
+            assert item.dispatch_attempts == 0

backend/tests/unit/test_support_bundle_rotated_logs.py

@@ -0,0 +1,108 @@
+"""The support bundle must span the rotated logs, not just the live file (#2555).
+
+``bambuddy.log`` is capped at 5 MB by the RotatingFileHandler, and the bundle
+used to ship only that one file — the three rotated backups sat on disk unread.
+
+That cap is invisible on a single printer and brutal on a farm. The 19-printer
+fleet in #2555 emits ~100 lines/s of MQTT frame dumps with debug logging on,
+which fills 5 MB in under five minutes: we asked the reporter for debug logs to
+diagnose a *queue* problem and the bundle came back holding 4m49s of history,
+almost none of it about the queue. The byte budget for the bundle was 10 MB all
+along; it just never looked past the newest file.
+"""
+
+from unittest.mock import patch
+
+from backend.app.api.routes.support import _get_log_content
+
+
+def _write_rotation(log_dir, live: str, backups: list[str]):
+    """Lay out a RotatingFileHandler set: bambuddy.log plus .log.1 .. .log.N.
+
+    ``backups[0]`` becomes ``.log.1``, which the handler defines as the *newest*
+    backup — i.e. the one immediately preceding the live file.
+    """
+    (log_dir / "bambuddy.log").write_text(live, encoding="utf-8")
+    for index, text in enumerate(backups, start=1):
+        (log_dir / f"bambuddy.log.{index}").write_text(text, encoding="utf-8")
+
+
+def test_reads_rotated_backups_as_well_as_the_live_file(tmp_path):
+    _write_rotation(tmp_path, "live line\n", ["newest backup\n", "middle backup\n", "oldest backup\n"])
+
+    with (
+        patch("backend.app.api.routes.support.settings.log_dir", tmp_path),
+        patch("backend.app.api.routes.support.settings.log_backup_count", 3),
+    ):
+        content = _get_log_content().decode()
+
+    for expected in ("oldest backup", "middle backup", "newest backup", "live line"):
+        assert expected in content, f"{expected!r} missing — the bundle dropped rotated history"
+
+
+def test_output_is_chronological_oldest_first(tmp_path):
+    """A log you have to read backwards is not a log. .log.3 is the oldest."""
+    _write_rotation(tmp_path, "D\n", ["C\n", "B\n", "A\n"])
+
+    with (
+        patch("backend.app.api.routes.support.settings.log_dir", tmp_path),
+        patch("backend.app.api.routes.support.settings.log_backup_count", 3),
+    ):
+        content = _get_log_content().decode()
+
+    assert content.split() == ["A", "B", "C", "D"]
+
+
+def test_byte_budget_is_spent_on_the_newest_history(tmp_path):
+    """When the rotation exceeds max_bytes, drop the OLD end, keep the recent.
+
+    Truncating from the wrong end would hand us a bundle full of history that
+    predates the problem being reported.
+    """
+    _write_rotation(tmp_path, "live\n", ["recent\n", "ancient\n"])
+
+    with (
+        patch("backend.app.api.routes.support.settings.log_dir", tmp_path),
+        patch("backend.app.api.routes.support.settings.log_backup_count", 2),
+    ):
+        # Enough for "live\n" + "recent\n" but not for "ancient\n" as well.
+        content = _get_log_content(max_bytes=12).decode()
+
+    assert "live" in content
+    assert "recent" in content
+    assert "ancient" not in content
+
+
+def test_partial_line_at_the_truncation_point_is_discarded(tmp_path):
+    """Seeking into the middle of a line must not emit a mangled fragment."""
+    (tmp_path / "bambuddy.log").write_text("aaaaaaaaaa\nbbbbbbbbbb\n", encoding="utf-8")
+
+    with (
+        patch("backend.app.api.routes.support.settings.log_dir", tmp_path),
+        patch("backend.app.api.routes.support.settings.log_backup_count", 3),
+    ):
+        content = _get_log_content(max_bytes=15).decode()
+
+    assert content == "bbbbbbbbbb\n", "a half-line leaked through the seek"
+
+
+def test_missing_backups_are_skipped_not_fatal(tmp_path):
+    """A fresh install has no .log.N yet; a gap must not abort the bundle."""
+    (tmp_path / "bambuddy.log").write_text("live only\n", encoding="utf-8")
+    (tmp_path / "bambuddy.log.2").write_text("older\n", encoding="utf-8")  # .log.1 absent
+
+    with (
+        patch("backend.app.api.routes.support.settings.log_dir", tmp_path),
+        patch("backend.app.api.routes.support.settings.log_backup_count", 3),
+    ):
+        content = _get_log_content().decode()
+
+    assert content == "older\nlive only\n"
+
+
+def test_absent_log_file_still_reports_cleanly(tmp_path):
+    with (
+        patch("backend.app.api.routes.support.settings.log_dir", tmp_path),
+        patch("backend.app.api.routes.support.settings.log_backup_count", 3),
+    ):
+        assert _get_log_content() == b"Log file not found"

frontend/src/api/client.ts

@@ -1247,6 +1247,10 @@ export interface AppSettings {
   require_plate_clear: boolean;
   // Shortest job first scheduling
   queue_shortest_first: boolean;
+  // How many printers the queue may upload to at once (#2555). 1 restores the
+  // old strictly-serial behaviour, where every printer waited out every other
+  // printer's transfer.
+  queue_max_concurrent_uploads: number;
   // Preheat / heat-soak before queued prints (#1468). Master toggle is the
   // default for new queue items; per-item PrintQueueItem.preheat_override can
   // flip the decision per print. Chamber target derives from the loaded AMS

frontend/src/i18n/locales/de.ts

@@ -2075,6 +2075,10 @@ export default {
     tempFanPresetsChamber: 'Kammertemperatur',
     tempFanPresetsFan: 'Lüftergeschwindigkeit',
     tempFanPresetsReset: 'Auf Standardwerte zurücksetzen',
+    concurrentUploadsTitle: 'Gleichzeitige Uploads',
+    concurrentUploadsDescription: 'Wie viele Drucker die Warteschlange gleichzeitig mit Dateien versorgen darf. Drucker nehmen Dateien nur langsam entgegen (ein großer Druck kann mehrere Minuten dauern), und jeder wartet, bis er an der Reihe ist — bei größeren Farmen verhindert ein höherer Wert also, dass der letzte Drucker eines Stapels erst alle anderen Übertragungen abwarten muss. Verringern Sie ihn, wenn Ihr Netzwerk oder der Bambuddy-Host mit parallelen Übertragungen Probleme hat.',
+    concurrentUploadsLabel: 'Gleichzeitig belieferte Drucker',
+    concurrentUploadsHelp: '1 beliefert immer nur einen Drucker (das bisherige Verhalten). Standard ist 4.',
     staggeredStart: 'Versetzter Start',
     staggeredStartDescription: 'Standard-Gruppengröße und -Intervall beim Staffeln von Mehrdrucker-Batchstarts. Pro Batch im Druck-Dialog überschreibbar.',
     preheatTitle: 'Vorheizen & Heat Soak',

frontend/src/i18n/locales/en.ts

@@ -2094,6 +2094,10 @@ export default {
     tempFanPresetsChamber: 'Chamber temperature',
     tempFanPresetsFan: 'Fan speed',
     tempFanPresetsReset: 'Reset to defaults',
+    concurrentUploadsTitle: 'Concurrent Uploads',
+    concurrentUploadsDescription: 'How many printers the queue may send files to at the same time. Printers receive files slowly (a large print can take several minutes), and each one waits its turn — so on a bigger fleet, raising this is what stops the last printer in a batch from waiting out every transfer before it. Lower it if your network or Bambuddy host struggles with parallel transfers.',
+    concurrentUploadsLabel: 'Printers uploaded to at once',
+    concurrentUploadsHelp: '1 sends to one printer at a time (the old behaviour). Default is 4.',
     staggeredStart: 'Staggered Start',
     staggeredStartDescription: 'Default group size and interval when staggering multi-printer batch starts. Can be overridden per batch in the print modal.',
     preheatTitle: 'Preheat & Heat Soak',

frontend/src/i18n/locales/es.ts

@@ -2078,6 +2078,10 @@ export default {
     tempFanPresetsChamber: 'Temperatura de la cámara',
     tempFanPresetsFan: 'Velocidad del ventilador',
     tempFanPresetsReset: 'Restablecer valores predeterminados',
+    concurrentUploadsTitle: 'Subidas simultáneas',
+    concurrentUploadsDescription: 'Cuántas impresoras pueden recibir archivos de la cola al mismo tiempo. Las impresoras reciben los archivos lentamente (una impresión grande puede tardar varios minutos) y cada una espera su turno, así que en una granja grande subir este valor evita que la última impresora de un lote tenga que esperar a que terminen todas las transferencias anteriores. Bájalo si tu red o el host de Bambuddy tienen problemas con transferencias en paralelo.',
+    concurrentUploadsLabel: 'Impresoras atendidas a la vez',
+    concurrentUploadsHelp: '1 envía a una impresora cada vez (el comportamiento anterior). El valor predeterminado es 4.',
     staggeredStart: 'Inicio escalonado',
     staggeredStartDescription: 'Tamaño de grupo e intervalo predeterminados al escalonar los inicios de lotes en varias impresoras. Se pueden anular por lote en la ventana de impresión.',
     preheatTitle: 'Precalentamiento y Heat Soak',

frontend/src/i18n/locales/fr.ts

@@ -2031,6 +2031,10 @@ export default {
     tempFanPresetsChamber: 'Température de la chambre',
     tempFanPresetsFan: 'Vitesse du ventilateur',
     tempFanPresetsReset: 'Rétablir les valeurs par défaut',
+    concurrentUploadsTitle: 'Envois simultanés',
+    concurrentUploadsDescription: "Nombre d'imprimantes auxquelles la file d'attente peut envoyer des fichiers en même temps. Les imprimantes reçoivent les fichiers lentement (une grosse impression peut prendre plusieurs minutes) et chacune attend son tour : sur un parc important, augmenter cette valeur évite que la dernière imprimante d'un lot doive attendre la fin de tous les transferts précédents. Réduisez-la si votre réseau ou votre hôte Bambuddy supporte mal les transferts parallèles.",
+    concurrentUploadsLabel: 'Imprimantes servies simultanément',
+    concurrentUploadsHelp: "1 n'envoie qu'à une imprimante à la fois (l'ancien comportement). Valeur par défaut : 4.",
     staggeredStart: 'Démarrage échelonné',
     staggeredStartDescription: 'Taille de groupe et intervalle par défaut lors de l\'échelonnement des démarrages de lots multi-imprimantes. Modifiable par lot dans la fenêtre d\'impression.',
     preheatTitle: 'Préchauffage & Heat Soak',

frontend/src/i18n/locales/it.ts

@@ -2031,6 +2031,10 @@ export default {
     tempFanPresetsChamber: 'Temperatura camera',
     tempFanPresetsFan: 'Velocità ventola',
     tempFanPresetsReset: 'Ripristina valori predefiniti',
+    concurrentUploadsTitle: 'Caricamenti simultanei',
+    concurrentUploadsDescription: "Quante stampanti possono ricevere file dalla coda contemporaneamente. Le stampanti ricevono i file lentamente (una stampa grande può richiedere diversi minuti) e ognuna aspetta il proprio turno: su un parco macchine ampio, aumentare questo valore evita che l'ultima stampante di un lotto debba attendere la fine di tutti i trasferimenti precedenti. Riducilo se la rete o l'host Bambuddy fatica con i trasferimenti paralleli.",
+    concurrentUploadsLabel: 'Stampanti servite contemporaneamente',
+    concurrentUploadsHelp: '1 invia a una stampante alla volta (il comportamento precedente). Il valore predefinito è 4.',
     staggeredStart: 'Avvio scaglionato',
     staggeredStartDescription: 'Dimensione gruppo e intervallo predefiniti per scaglionare avvii di batch multi-stampante. Sovrascrivibili per batch nella finestra di stampa.',
     preheatTitle: 'Preriscaldo & Heat Soak',

frontend/src/i18n/locales/ja.ts

@@ -2074,6 +2074,10 @@ export default {
     tempFanPresetsChamber: 'チャンバー温度',
     tempFanPresetsFan: 'ファン速度',
     tempFanPresetsReset: 'デフォルトにリセット',
+    concurrentUploadsTitle: '同時アップロード',
+    concurrentUploadsDescription: 'キューが同時にファイルを送信できるプリンター数です。プリンターのファイル受信は遅く（大きな造形では数分かかることもあります）、各プリンターは順番を待ちます。台数の多い環境では、この値を上げることで、バッチ内の最後のプリンターが先行するすべての転送を待たずに済みます。ネットワークや Bambuddy ホストが並列転送に耐えられない場合は下げてください。',
+    concurrentUploadsLabel: '同時に送信するプリンター数',
+    concurrentUploadsHelp: '1 にすると 1 台ずつ送信します（従来の動作）。既定値は 4 です。',
     staggeredStart: '段階的開始',
     staggeredStartDescription: '複数プリンターのバッチ開始を段階的に行う際のデフォルトのグループサイズと間隔。プリントモーダルでバッチごとに上書き可能。',
     preheatTitle: 'プレヒート & ヒートソーク',

frontend/src/i18n/locales/ko.ts

@@ -1956,6 +1956,10 @@ export default {
     tempFanPresetsChamber: '챔버 온도',
     tempFanPresetsFan: '팬 속도',
     tempFanPresetsReset: '기본값으로 재설정',
+    concurrentUploadsTitle: '동시 업로드',
+    concurrentUploadsDescription: '대기열이 동시에 파일을 보낼 수 있는 프린터 수입니다. 프린터는 파일을 느리게 받으며(큰 출력물은 몇 분이 걸릴 수 있습니다) 각 프린터는 자기 차례를 기다립니다. 따라서 프린터가 많은 환경에서는 이 값을 높이면 배치의 마지막 프린터가 앞선 모든 전송이 끝나기를 기다리지 않아도 됩니다. 네트워크나 Bambuddy 호스트가 병렬 전송을 버거워하면 값을 낮추세요.',
+    concurrentUploadsLabel: '동시에 전송할 프린터 수',
+    concurrentUploadsHelp: '1이면 한 번에 한 대씩 전송합니다(기존 동작). 기본값은 4입니다.',
     staggeredStart: '엇갈린 시작',
     staggeredStartDescription: '다중 프린터 일괄 시작 시 기본 그룹 크기 및 간격. 인쇄 모달에서 배치별로 재정의할 수 있습니다.',
     preheatTitle: '예열 & 히트 소크',

frontend/src/i18n/locales/pt-BR.ts

@@ -2031,6 +2031,10 @@ export default {
     tempFanPresetsChamber: 'Temperatura da câmara',
     tempFanPresetsFan: 'Velocidade do ventilador',
     tempFanPresetsReset: 'Restaurar padrões',
+    concurrentUploadsTitle: 'Envios simultâneos',
+    concurrentUploadsDescription: 'Quantas impressoras a fila pode abastecer com arquivos ao mesmo tempo. As impressoras recebem os arquivos lentamente (uma impressão grande pode levar vários minutos) e cada uma espera sua vez — portanto, em um parque maior, aumentar este valor evita que a última impressora de um lote tenha de esperar todas as transferências anteriores. Reduza-o se a sua rede ou o host do Bambuddy tiver dificuldade com transferências paralelas.',
+    concurrentUploadsLabel: 'Impressoras abastecidas de uma vez',
+    concurrentUploadsHelp: '1 envia para uma impressora por vez (o comportamento anterior). O padrão é 4.',
     staggeredStart: 'Início escalonado',
     staggeredStartDescription: 'Tamanho de grupo e intervalo padrão ao escalonar inícios de lotes multi-impressora. Pode ser sobrescrito por lote no modal de impressão.',
     preheatTitle: 'Pré-aquecimento & Heat Soak',

frontend/src/i18n/locales/tr.ts

@@ -2079,6 +2079,10 @@ export default {
     tempFanPresetsChamber: 'Hazne sıcaklığı',
     tempFanPresetsFan: 'Fan hızı',
     tempFanPresetsReset: 'Varsayılanlara sıfırla',
+    concurrentUploadsTitle: 'Eşzamanlı Yüklemeler',
+    concurrentUploadsDescription: 'Kuyruğun aynı anda kaç yazıcıya dosya gönderebileceği. Yazıcılar dosyaları yavaş alır (büyük bir baskı birkaç dakika sürebilir) ve her biri sırasını bekler; bu nedenle büyük bir filoda bu değeri artırmak, bir gruptaki son yazıcının kendisinden önceki tüm aktarımları beklemesini önler. Ağınız veya Bambuddy sunucunuz paralel aktarımlarda zorlanıyorsa değeri düşürün.',
+    concurrentUploadsLabel: 'Aynı anda dosya gönderilen yazıcı sayısı',
+    concurrentUploadsHelp: "1, aynı anda tek yazıcıya gönderir (eski davranış). Varsayılan 4'tür.",
     staggeredStart: 'Kademeli Başlatma',
     staggeredStartDescription: 'Çoklu yazıcı toplu başlatmaları kademelendirilirken varsayılan grup boyutu ve aralığı. Baskı modalinde yığın başına geçersiz kılınabilir.',
     preheatTitle: 'Ön Isıtma & Heat Soak',

frontend/src/i18n/locales/zh-CN.ts

@@ -2076,6 +2076,10 @@ export default {
     tempFanPresetsChamber: '腔室温度',
     tempFanPresetsFan: '风扇速度',
     tempFanPresetsReset: '恢复默认值',
+    concurrentUploadsTitle: '并发上传',
+    concurrentUploadsDescription: '队列可以同时向多少台打印机发送文件。打印机接收文件很慢（大型打印可能需要几分钟），并且每台都要排队等候——因此在打印机较多时，调高该值可以避免批次中最后一台打印机等完前面所有传输。如果网络或 Bambuddy 主机难以承受并行传输，请调低该值。',
+    concurrentUploadsLabel: '同时上传的打印机数量',
+    concurrentUploadsHelp: '设为 1 时一次只向一台打印机发送（旧行为）。默认值为 4。',
     staggeredStart: '错峰启动',
     staggeredStartDescription: '错峰启动多台打印机批次时的默认组大小和间隔。可在打印对话框中按批次覆盖。',
     preheatTitle: '预热与热保温',

frontend/src/i18n/locales/zh-TW.ts

@@ -2076,6 +2076,10 @@ export default {
     tempFanPresetsChamber: '腔室溫度',
     tempFanPresetsFan: '風扇速度',
     tempFanPresetsReset: '還原預設值',
+    concurrentUploadsTitle: '並行上傳',
+    concurrentUploadsDescription: '佇列可同時向多少台印表機傳送檔案。印表機接收檔案的速度很慢（大型列印可能需要數分鐘），而且每台都得排隊等候——因此在印表機較多時，調高此值可避免批次中最後一台印表機等完前面所有傳輸。若網路或 Bambuddy 主機難以負荷並行傳輸，請調低此值。',
+    concurrentUploadsLabel: '同時上傳的印表機數量',
+    concurrentUploadsHelp: '設為 1 時一次只傳送給一台印表機（舊行為）。預設值為 4。',
     staggeredStart: '錯開啟動',
     staggeredStartDescription: '多台印表機批次啟動時的預設群組大小與間隔。可在列印對話框中逐批覆寫。',
     preheatTitle: '預熱與熱保溫',

frontend/src/pages/SettingsPage.tsx

@@ -1,5 +1,5 @@
 import { useQuery, useMutation, useQueryClient } from '@tanstack/react-query';
-import { Loader2, Plus, Plug, AlertTriangle, RotateCcw, Bell, Download, RefreshCw, ExternalLink, Globe, Droplets, Thermometer, FileText, Edit2, Send, CheckCircle, XCircle, History, Trash2, Zap, TrendingUp, Calendar, DollarSign, Power, PowerOff, Key, Copy, Database, X, Shield, Printer, Cylinder, Wifi, Home, Video, Users, Lock, Unlock, ChevronDown, Save, Mail, Flame, Layers, ListOrdered, Code, Search, Scale, Settings as SettingsIcon, ScanEye, Cog, QrCode, Heart, Briefcase, Workflow } from 'lucide-react';
+import { Loader2, Plus, Plug, AlertTriangle, RotateCcw, Bell, Download, RefreshCw, ExternalLink, Globe, Droplets, Thermometer, FileText, Edit2, Send, CheckCircle, XCircle, History, Trash2, Zap, TrendingUp, Calendar, DollarSign, Power, PowerOff, Key, Copy, Database, X, Shield, Printer, Cylinder, Wifi, Home, Video, Users, Lock, Unlock, ChevronDown, Save, Mail, Flame, Layers, ListOrdered, Code, Search, Scale, Settings as SettingsIcon, ScanEye, Cog, QrCode, Heart, Briefcase, Workflow, UploadCloud } from 'lucide-react';
 import { useTranslation } from 'react-i18next';
 import { useNavigate, useSearchParams } from 'react-router-dom';
 import { api } from '../api/client';
@@ -72,6 +72,7 @@ registerSettingsSearch({ labelKey: 'settings.defaultPrintOptions', labelFallback
 registerSettingsSearch({ labelKey: 'settings.tempFanPresetsTitle', labelFallback: 'Temperature & Fan Presets', tab: 'queue', keywords: 'temperature fan presets nozzle bed chamber quick buttons popover', anchor: 'card-temp-fan-presets' });
 registerSettingsSearch({ labelKey: 'settings.staggeredStart', labelFallback: 'Staggered Start', tab: 'queue', keywords: 'staggered batch delay start queue group', anchor: 'card-staggered' });
 registerSettingsSearch({ labelKey: 'settings.plateClear', labelFallback: 'Plate-Clear Confirmation', tab: 'queue', keywords: 'plate clear confirm auto queue', anchor: 'card-plate' });
+registerSettingsSearch({ labelKey: 'settings.concurrentUploadsTitle', labelFallback: 'Concurrent Uploads', tab: 'queue', keywords: 'concurrent parallel upload transfer ftp queue slow farm simultaneous', anchor: 'card-concurrent-uploads' });
 registerSettingsSearch({ labelKey: 'settings.gcodeInjection', labelFallback: 'G-code Injection', tab: 'queue', keywords: 'gcode injection start end autoprint farmloop swapmod autoclear printflow', anchor: 'card-gcode' });
 registerSettingsSearch({ labelKey: 'settings.slicerCard', labelFallback: 'Slicer', tab: 'queue', keywords: 'slicer orcaslicer bambustudio orca bambu api sidecar url docker preferred', anchor: 'card-slicer' });
 registerSettingsSearch({ labelKey: 'settings.queueDrying', tab: 'queue', keywords: 'drying presets temperature time humidity ams', anchor: 'card-drying' });
@@ -1018,6 +1019,7 @@ export function SettingsPage() {
       (settings.stagger_group_size ?? 2) !== (localSettings.stagger_group_size ?? 2) ||
       (settings.stagger_interval_minutes ?? 5) !== (localSettings.stagger_interval_minutes ?? 5) ||
       (settings.require_plate_clear ?? false) !== (localSettings.require_plate_clear ?? false) ||
+      (settings.queue_max_concurrent_uploads ?? 4) !== (localSettings.queue_max_concurrent_uploads ?? 4) ||
       (settings.preheat_enabled ?? false) !== (localSettings.preheat_enabled ?? false) ||
       (settings.preheat_filament_targets ?? '') !== (localSettings.preheat_filament_targets ?? '') ||
       (settings.preheat_max_wait_seconds ?? 900) !== (localSettings.preheat_max_wait_seconds ?? 900) ||
@@ -1116,6 +1118,7 @@ export function SettingsPage() {
         stagger_group_size: localSettings.stagger_group_size,
         stagger_interval_minutes: localSettings.stagger_interval_minutes,
         require_plate_clear: localSettings.require_plate_clear,
+        queue_max_concurrent_uploads: localSettings.queue_max_concurrent_uploads,
         preheat_enabled: localSettings.preheat_enabled,
         preheat_filament_targets: localSettings.preheat_filament_targets,
         preheat_max_wait_seconds: localSettings.preheat_max_wait_seconds,
@@ -4387,6 +4390,37 @@ export function SettingsPage() {
             </CardContent>
           </Card>
 
+          {/* Concurrent queue uploads (#2555) */}
+          <Card id="card-concurrent-uploads">
+            <CardHeader>
+              <h3 className="text-base font-semibold text-white flex items-center gap-2">
+                <UploadCloud className="w-4 h-4 text-bambu-green" />
+                {t('settings.concurrentUploadsTitle', 'Concurrent Uploads')}
+              </h3>
+            </CardHeader>
+            <CardContent className="space-y-3">
+              <p className="text-xs text-bambu-gray">
+                {t('settings.concurrentUploadsDescription', 'How many printers the queue may send files to at the same time. Printers receive files slowly (a large print can take several minutes), and each one waits its turn — so on a bigger fleet, raising this is what stops the last printer in a batch from waiting out every transfer before it. Lower it if your network or Bambuddy host struggles with parallel transfers.')}
+              </p>
+              <div className="w-full sm:w-1/2">
+                <label className="block text-xs text-bambu-gray mb-1">
+                  {t('settings.concurrentUploadsLabel', 'Printers uploaded to at once')}
+                </label>
+                <input
+                  type="number"
+                  min={1}
+                  max={16}
+                  value={localSettings.queue_max_concurrent_uploads ?? 4}
+                  onChange={(e) => updateSetting('queue_max_concurrent_uploads', Math.max(1, Math.min(16, parseInt(e.target.value) || 1)))}
+                  className="w-full px-3 py-2 bg-bambu-dark border border-bambu-dark-tertiary rounded-lg text-white text-sm focus:outline-none focus:border-bambu-green"
+                />
+                <p className="text-xs text-bambu-gray mt-1">
+                  {t('settings.concurrentUploadsHelp', '1 sends to one printer at a time (the old behaviour). Default is 4.')}
+                </p>
+              </div>
+            </CardContent>
+          </Card>
+
           {/* Preheat & Heat Soak (#1468) */}
           <Card id="card-preheat">
             <CardHeader>

static/index.html

@@ -26,8 +26,8 @@
 
     <!-- Splash screens for iOS -->
     <link rel="apple-touch-startup-image" href="/img/android-chrome-512x512.png" />
-    <script type="module" crossorigin src="/assets/index-Bu5feVhv.js"></script>
-    <link rel="stylesheet" crossorigin href="/assets/index-DASc8Ke0.css">
+    <script type="module" crossorigin src="/assets/index-CevSiltg.js"></script>
+    <link rel="stylesheet" crossorigin href="/assets/index-7t2liT66.css">
   </head>
   <body>
     <div id="root"></div>