fix(vp): close #1780 race — bump slicer-MQTT wait to 5s + retroactive stamp

  Round 2 (166e9f9e) fixed the stash-key mismatch, but @mkoreen's
  2026-06-23 bundle showed BS's MQTT project_file arrived 85 ms past the
  2.0 s wait timeout (FTP done 00:42:02.509, "No slicer options cached"
  00:42:04.509, MQTT 00:42:04.594). Queue item was committed with
  settings defaults; nozzle_mapping never made it onto the wire.

  Three pieces:

  1. _SLICER_OPTIONS_WAIT_TIMEOUT module constant, 2.0 -> 5.0 s. Covers
     wireless / loaded-Pi jitter; one-time +3 s cost only for legacy
     slicers that never send MQTT.

  2. _RECENT_QUEUE_ITEM_TTL fallback: on_print_command retroactively
     UPDATEs slicer-driven fields on a recently-committed queue item
     when the event wait already gave up. Tracked via
     _recent_queue_items dict (30 s TTL, evicted on every queue-add).
     Gated on status='pending' so we never race the dispatcher.
     Multi-plate covered via WHERE id IN (...).

  3. Post-commit last-chance pop. Audit caught a race in (2): MQTT could
     arrive during any await inside _add_to_print_queue (wait_for,
     archive_print, db.flush, db.commit), and on_print_command would
     stash data with no event consumer AND no _recent_queue_items entry
     yet. After populating _recent_queue_items, _add_to_print_queue now
     pops _slicer_print_options[file_path.name] one last time and
     routes any hit through _restamp inline.

backend/app/services/virtual_printer/manager.py

@@ -6,6 +6,7 @@ bound to its dedicated IP address, regardless of mode.
 
 import asyncio
 import logging
+import time
 from collections.abc import Callable
 from datetime import datetime, timezone
 from pathlib import Path
@@ -110,6 +111,23 @@ DEFAULT_VIRTUAL_PRINTER_MODEL = "BL-P001"  # X1C
 # the dict fills, so a long-running VP can't leak unbounded state.
 _SLICER_OPTIONS_CACHE_LIMIT = 128
 
+# How long ``_add_to_print_queue`` waits for the slicer's MQTT
+# ``project_file`` after the FTP upload completes (#1780 round 3).
+# Bambu Studio sends FTP first, then MQTT immediately after — but on
+# wireless / loaded setups the MQTT command can land 2+ s after FTP,
+# which used to time the wait out and silently drop ``nozzle_mapping``
+# + the other slicer-driven flags. The bumped window covers the
+# observed worst case in the field; the late-MQTT fallback in
+# ``on_print_command`` covers the rest.
+_SLICER_OPTIONS_WAIT_TIMEOUT = 5.0
+
+# How long ``on_print_command`` will retroactively stamp slicer fields
+# onto a recently-committed queue item when the MQTT print command
+# arrives after ``_SLICER_OPTIONS_WAIT_TIMEOUT`` expired. Covers
+# extra-late MQTT (slow wireless slicer, NIC drop+retry) and the
+# scheduler tick interval before dispatch picks the item up.
+_RECENT_QUEUE_ITEM_TTL = 30.0
+
 
 def _get_serial_for_model(model: str, serial_suffix: str) -> str:
     """Get serial number for the given model and suffix."""
@@ -200,6 +218,16 @@ class VirtualPrinterInstance:
         self._slicer_print_options: dict[str, dict] = {}
         self._slicer_print_options_events: dict[str, asyncio.Event] = {}
 
+        # Queue items recently committed by `_add_to_print_queue`, keyed by
+        # FTP filename. Used by `on_print_command` to retroactively stamp the
+        # slicer's nozzle_mapping (and the other slicer-driven flags) onto a
+        # queue item when the MQTT `project_file` arrives after the queue-add
+        # wait timed out — the #1780 round-3 race. Value is
+        # (queue_item_ids, monotonic_committed_at); entries older than
+        # `_RECENT_QUEUE_ITEM_TTL` are evicted opportunistically on each
+        # queue-add.
+        self._recent_queue_items: dict[str, tuple[list[int], float]] = {}
+
         # Per-instance services
         self._proxy: SlicerProxyManager | None = None
         self._ftp: VirtualPrinterFTPServer | None = None
@@ -342,6 +370,114 @@ class VirtualPrinterInstance:
         event = self._slicer_print_options_events.get(stash_key)
         if event:
             event.set()
+            return
+        # No consumer waiting: `_add_to_print_queue` either already gave up
+        # (wait_for timed out) or hasn't started yet (FTP still uploading).
+        # If a queue item was committed within the last
+        # `_RECENT_QUEUE_ITEM_TTL`, the wait timed out and the row holds
+        # settings defaults instead of the slicer's pick — retroactively
+        # stamp the slicer-driven fields so the dispatcher honours the
+        # user's choice. Covers the #1780 round-3 race where Bambu Studio's
+        # MQTT lands just past the bumped wait ceiling.
+        await self._restamp_recent_queue_item(stash_key, data)
+
+    async def _restamp_recent_queue_item(self, stash_key: str, data: dict) -> None:
+        """Patch slicer-driven fields onto a queue item the MQTT command missed.
+
+        ``_add_to_print_queue`` waits up to ``_SLICER_OPTIONS_WAIT_TIMEOUT``
+        for the slicer's MQTT ``project_file`` before committing the queue
+        item. If the MQTT command arrives after that window — observed in
+        the field at ~2.1 s on H2C / wireless setups (#1780 round 3) — the
+        row was already written with settings defaults. This method runs
+        on the late MQTT path: it looks up the most recent queue items
+        committed for this filename and patches in the slicer's
+        ``nozzle_mapping`` + workflow flags, but only while the items are
+        still ``pending`` (scheduler hasn't dispatched them yet).
+        """
+        if not self._session_factory:
+            return
+        entry = self._recent_queue_items.get(stash_key)
+        if entry is None:
+            return
+        queue_item_ids, committed_at = entry
+        if time.monotonic() - committed_at > _RECENT_QUEUE_ITEM_TTL:
+            self._recent_queue_items.pop(stash_key, None)
+            return
+
+        import json
+
+        # Mirror the field set `_add_to_print_queue` reads off slicer_opts.
+        # MQTT uses `bed_leveling` (single L); the column is `bed_levelling`.
+        # `nozzles_info` is intentionally not stamped — column kept for
+        # legacy rows but never written; see PrintQueueItem.nozzles_info.
+        patch: dict = {}
+        for mqtt_field, column in (
+            ("bed_leveling", "bed_levelling"),
+            ("flow_cali", "flow_cali"),
+            ("vibration_cali", "vibration_cali"),
+            ("layer_inspect", "layer_inspect"),
+            ("timelapse", "timelapse"),
+            ("use_ams", "use_ams"),
+        ):
+            if mqtt_field in data:
+                patch[column] = bool(data[mqtt_field])
+
+        raw = data.get("nozzle_mapping")
+        if raw is not None:
+            if isinstance(raw, str):
+                try:
+                    raw = json.loads(raw)
+                except json.JSONDecodeError:
+                    logger.warning(
+                        "[VP %s] Late MQTT nozzle_mapping is unparseable JSON, dropping: %r",
+                        self.name,
+                        raw,
+                    )
+                    raw = None
+            if raw is not None:
+                patch["nozzle_mapping"] = json.dumps(raw)
+
+        if not patch:
+            self._recent_queue_items.pop(stash_key, None)
+            return
+
+        from sqlalchemy import select, update
+
+        from backend.app.models.print_queue import PrintQueueItem
+
+        try:
+            async with self._session_factory() as db:
+                # Only stamp items still pending; once the scheduler has
+                # picked the row up we can't safely race the dispatcher.
+                result = await db.execute(
+                    select(PrintQueueItem.id).where(
+                        PrintQueueItem.id.in_(queue_item_ids),
+                        PrintQueueItem.status == "pending",
+                    )
+                )
+                eligible_ids = [row[0] for row in result.all()]
+                if not eligible_ids:
+                    self._recent_queue_items.pop(stash_key, None)
+                    return
+                await db.execute(update(PrintQueueItem).where(PrintQueueItem.id.in_(eligible_ids)).values(**patch))
+                await db.commit()
+                logger.info(
+                    "[VP %s] Late slicer MQTT for %s — retroactively stamped %s onto queue item(s) %s",
+                    self.name,
+                    stash_key,
+                    sorted(patch.keys()),
+                    eligible_ids,
+                )
+        except Exception as e:
+            logger.error(
+                "[VP %s] Failed to retroactively stamp queue item(s) %s for %s: %s",
+                self.name,
+                queue_item_ids,
+                stash_key,
+                e,
+            )
+        finally:
+            self._recent_queue_items.pop(stash_key, None)
 
     def _schedule_finish_release(self, filename: str, delay: float = 1.5) -> None:
         """Re-set gcode_state=FINISH on the VP after the project_file ack.
@@ -524,19 +660,21 @@ class VirtualPrinterInstance:
         # queue item can inherit the slicer-side print options the user
         # picked (timelapse, bed_leveling, etc). Slicers send the FTP upload
         # first and the MQTT command immediately after, so the typical lag
-        # is a few hundred ms; 2 s is conservative without making every
-        # VP-queue add visibly slow. Falls back to the global default_*
+        # is a few hundred ms. The window is generous enough to absorb
+        # wireless / loaded-Pi jitter without making every VP-queue add
+        # visibly slow — observed worst case in #1780 round 3 was 2.085 s,
+        # the previous 2.0 s ceiling. Falls back to the global default_*
         # settings if MQTT doesn't arrive in time (legacy behaviour for
         # users on a slicer that doesn't send a print command). #1403.
         # The wait is skipped when there's no MQTT server attached — covers
         # unit tests that invoke `_add_to_print_queue` directly without
-        # going through `on_print_command`, so they don't pay the 2 s tax.
+        # going through `on_print_command`, so they don't pay the wait tax.
         slicer_opts = self._slicer_print_options.pop(file_path.name, None)
         if slicer_opts is None and self._mqtt is not None:
             event = asyncio.Event()
             self._slicer_print_options_events[file_path.name] = event
             try:
-                await asyncio.wait_for(event.wait(), timeout=2.0)
+                await asyncio.wait_for(event.wait(), timeout=_SLICER_OPTIONS_WAIT_TIMEOUT)
                 slicer_opts = self._slicer_print_options.pop(file_path.name, None)
             except asyncio.TimeoutError:
                 slicer_opts = None
@@ -749,6 +887,34 @@ class VirtualPrinterInstance:
                         await db.flush()  # populate queue_item.id before logging
                         queue_item_ids.append(queue_item.id)
                     await db.commit()
+                    # Track the freshly-committed queue items so
+                    # `on_print_command` can retroactively stamp slicer-side
+                    # fields if the MQTT `project_file` lands AFTER the
+                    # `_SLICER_OPTIONS_WAIT_TIMEOUT` window expired — the
+                    # #1780 round-3 race. Eviction of stale entries here
+                    # keeps the dict bounded; the queue path is the only
+                    # writer, so doing it on commit is enough.
+                    now = time.monotonic()
+                    cutoff = now - _RECENT_QUEUE_ITEM_TTL
+                    self._recent_queue_items = {k: v for k, v in self._recent_queue_items.items() if v[1] > cutoff}
+                    self._recent_queue_items[file_path.name] = (list(queue_item_ids), now)
+                    # Last-chance check: MQTT for this filename could have
+                    # arrived during ANY await between the initial pop and
+                    # now — wait_for itself, archive_print, db.flush,
+                    # db.commit. In all those cases `on_print_command`
+                    # stashed its data but neither the event-signal path nor
+                    # the retroactive `_recent_queue_items` path was in
+                    # place to consume it. Pop any late stash and apply
+                    # inline so the late MQTT never leaks past the queue-add.
+                    late_opts = self._slicer_print_options.pop(file_path.name, None)
+                    if late_opts is not None:
+                        logger.info(
+                            "[VP %s] Late slicer MQTT detected for %s during queue-add — "
+                            "applying inline (race vs commit/archive/flush yield)",
+                            self.name,
+                            file_path.name,
+                        )
+                        await self._restamp_recent_queue_item(file_path.name, late_opts)
                     if len(queue_item_ids) == 1:
                         logger.info("[VP %s] Added to queue: %s", self.name, queue_item_ids[0])
                     else:

backend/tests/unit/services/test_virtual_printer.py

@@ -1785,6 +1785,272 @@ class TestVirtualPrinterInstance:
         for item in added_items:
             assert _json.loads(item.nozzle_mapping) == [16, 0]
 
+    @pytest.mark.asyncio
+    async def test_on_print_command_late_mqtt_retroactively_stamps_queue_item(self, tmp_path):
+        """#1780 round 3: Bambu Studio's MQTT project_file can arrive AFTER
+        `_add_to_print_queue` already gave up waiting (observed at 2.085 s
+        on H2C wireless setups). The queue item was committed with settings
+        defaults; the slicer's nozzle_mapping + workflow flags must be
+        patched onto it when MQTT lands, otherwise the H2C firmware falls
+        back to auto-pick.
+        """
+        import json as _json
+
+        from backend.app.services.virtual_printer.manager import VirtualPrinterInstance
+
+        added_items: list = []
+        mock_db = AsyncMock()
+        mock_db.add = MagicMock(
+            side_effect=lambda item: (added_items.append(item), setattr(item, "id", 100 + len(added_items)))[0]
+        )
+
+        async def _flush():
+            # added_items[-1].id was set by `add`; nothing else to do.
+            return None
+
+        mock_db.flush = AsyncMock(side_effect=_flush)
+        mock_db.commit = AsyncMock()
+
+        # First execute() call (the position-max SELECT inside _add_to_print_queue)
+        # returns None; second (the eligible-pending SELECT in
+        # _restamp_recent_queue_item) returns the committed queue id; third
+        # (the UPDATE) is fire-and-forget.
+        position_max_result = MagicMock()
+        position_max_result.scalar = MagicMock(return_value=None)
+        select_pending_result = MagicMock()
+        select_pending_result.all = MagicMock(return_value=[(101,)])
+        update_result = MagicMock()
+        mock_db.execute = AsyncMock(side_effect=[position_max_result, select_pending_result, update_result])
+
+        mock_session_factory = MagicMock()
+        mock_session_ctx = AsyncMock()
+        mock_session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
+        mock_session_ctx.__aexit__ = AsyncMock(return_value=False)
+        mock_session_factory.return_value = mock_session_ctx
+
+        inst = VirtualPrinterInstance(
+            vp_id=99,
+            name="LateMQTT",
+            mode="queue",
+            model="O1C2",
+            access_code="12345678",
+            serial_suffix="391800099",
+            base_dir=tmp_path,
+            session_factory=mock_session_factory,
+        )
+        # MQTT server presence enables the wait_for path; we don't actually
+        # use any methods on it.
+        inst._mqtt = MagicMock()
+
+        file_path = tmp_path / "test.3mf"
+        file_path.write_bytes(b"fake3mf")
+
+        mock_archive = MagicMock()
+        mock_archive.id = 1
+        mock_archive.print_name = "test"
+
+        # 1. _add_to_print_queue runs WITHOUT a prior on_print_command —
+        #    the wait_for times out (settings-default fallback) and the
+        #    queue item is committed.
+        with (
+            patch(
+                "backend.app.api.routes.settings.get_setting",
+                new_callable=AsyncMock,
+                return_value=None,
+            ),
+            patch(
+                "backend.app.services.archive.ArchiveService.archive_print",
+                new_callable=AsyncMock,
+                return_value=mock_archive,
+            ),
+            # Shorten the wait so the test isn't slow.
+            patch(
+                "backend.app.services.virtual_printer.manager._SLICER_OPTIONS_WAIT_TIMEOUT",
+                0.05,
+            ),
+        ):
+            await inst._add_to_print_queue(file_path, "192.168.1.100")
+
+        assert len(added_items) == 1
+        assert added_items[0].nozzle_mapping is None  # MQTT was never received
+        assert file_path.name in inst._recent_queue_items
+
+        # 2. MQTT project_file arrives AFTER the wait expired — must
+        #    retroactively patch the queue item.
+        await inst.on_print_command(
+            file_path.name,
+            {
+                "command": "project_file",
+                "file": file_path.name,
+                "nozzle_mapping": [16, -1, -1, 1],
+                "timelapse": True,
+                "bed_leveling": False,
+            },
+        )
+
+        # The UPDATE call is the third execute. Inspect its values.
+        update_call = mock_db.execute.await_args_list[2]
+        update_stmt = update_call.args[0]
+        compiled = update_stmt.compile(compile_kwargs={"literal_binds": False})
+        params = dict(compiled.params)
+        assert _json.loads(params["nozzle_mapping"]) == [16, -1, -1, 1]
+        assert params["timelapse"] is True
+        assert params["bed_levelling"] is False  # MQTT bed_leveling → column bed_levelling
+        # Recent-queue tracking dict is cleared after the patch.
+        assert file_path.name not in inst._recent_queue_items
+
+    @pytest.mark.asyncio
+    async def test_add_to_print_queue_catches_mqtt_stashed_post_wait_timeout(self, tmp_path):
+        """The actual race-window scenario: wait_for times out, then MQTT
+        arrives and stashes options AFTER the wait but BEFORE the
+        post-commit re-check. The post-commit pop must catch it.
+        """
+        import json as _json
+
+        from backend.app.services.virtual_printer.manager import VirtualPrinterInstance
+
+        added_items: list = []
+        mock_db = AsyncMock()
+        mock_db.add = MagicMock(
+            side_effect=lambda item: (added_items.append(item), setattr(item, "id", 300 + len(added_items)))[0]
+        )
+        mock_db.flush = AsyncMock()
+        mock_db.commit = AsyncMock()
+
+        position_max_result = MagicMock()
+        position_max_result.scalar = MagicMock(return_value=None)
+        select_pending_result = MagicMock()
+        select_pending_result.all = MagicMock(return_value=[(301,)])
+        update_result = MagicMock()
+        mock_db.execute = AsyncMock(side_effect=[position_max_result, select_pending_result, update_result])
+
+        mock_session_factory = MagicMock()
+        mock_session_ctx = AsyncMock()
+        mock_session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
+        mock_session_ctx.__aexit__ = AsyncMock(return_value=False)
+        mock_session_factory.return_value = mock_session_ctx
+
+        inst = VirtualPrinterInstance(
+            vp_id=96,
+            name="RaceCommitYield",
+            mode="queue",
+            model="O1C2",
+            access_code="12345678",
+            serial_suffix="391800096",
+            base_dir=tmp_path,
+            session_factory=mock_session_factory,
+        )
+        inst._mqtt = MagicMock()
+
+        file_path = tmp_path / "test.3mf"
+        file_path.write_bytes(b"fake3mf")
+
+        # Stash MQTT data on the FIRST commit (simulating MQTT arrival
+        # during _add_to_print_queue's commit yield); _restamp also calls
+        # db.commit later, so we one-shot the side effect.
+        commit_calls = {"n": 0}
+
+        async def _delayed_stash(*_args, **_kwargs):
+            commit_calls["n"] += 1
+            if commit_calls["n"] == 1:
+                inst._slicer_print_options[file_path.name] = {
+                    "command": "project_file",
+                    "file": file_path.name,
+                    "nozzle_mapping": [0, 16, -1, -1],
+                    "timelapse": False,
+                }
+            return None
+
+        mock_db.commit = AsyncMock(side_effect=_delayed_stash)
+
+        mock_archive = MagicMock()
+        mock_archive.id = 1
+        mock_archive.print_name = "test"
+
+        with (
+            patch(
+                "backend.app.api.routes.settings.get_setting",
+                new_callable=AsyncMock,
+                return_value=None,
+            ),
+            patch(
+                "backend.app.services.archive.ArchiveService.archive_print",
+                new_callable=AsyncMock,
+                return_value=mock_archive,
+            ),
+            patch(
+                "backend.app.services.virtual_printer.manager._SLICER_OPTIONS_WAIT_TIMEOUT",
+                0.05,
+            ),
+        ):
+            await inst._add_to_print_queue(file_path, "192.168.1.100")
+
+        # Queue item INSERTed with defaults (wait timed out, no slicer_opts).
+        assert len(added_items) == 1
+        # But the post-commit pop caught the late stash and applied the
+        # slicer nozzle_mapping via _restamp's UPDATE.
+        update_call = mock_db.execute.await_args_list[2]
+        update_stmt = update_call.args[0]
+        compiled = update_stmt.compile(compile_kwargs={"literal_binds": False})
+        params = dict(compiled.params)
+        assert _json.loads(params["nozzle_mapping"]) == [0, 16, -1, -1]
+        assert params["timelapse"] is False
+        # _recent_queue_items entry was consumed by the post-commit
+        # _restamp call.
+        assert file_path.name not in inst._recent_queue_items
+        # And the stash is empty.
+        assert file_path.name not in inst._slicer_print_options
+
+    @pytest.mark.asyncio
+    async def test_on_print_command_late_mqtt_skips_already_dispatched_item(self, tmp_path):
+        """Once the scheduler has picked the queue item up (status != pending),
+        the retroactive patch is a no-op — racing the dispatcher would be
+        unsafe.
+        """
+        from backend.app.services.virtual_printer.manager import VirtualPrinterInstance
+
+        mock_db = AsyncMock()
+        # The eligible-pending SELECT returns nothing — item is no longer pending.
+        empty_result = MagicMock()
+        empty_result.all = MagicMock(return_value=[])
+        mock_db.execute = AsyncMock(return_value=empty_result)
+        mock_db.commit = AsyncMock()
+
+        mock_session_factory = MagicMock()
+        mock_session_ctx = AsyncMock()
+        mock_session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
+        mock_session_ctx.__aexit__ = AsyncMock(return_value=False)
+        mock_session_factory.return_value = mock_session_ctx
+
+        inst = VirtualPrinterInstance(
+            vp_id=98,
+            name="LateMQTTDispatched",
+            mode="queue",
+            model="O1C2",
+            access_code="12345678",
+            serial_suffix="391800098",
+            base_dir=tmp_path,
+            session_factory=mock_session_factory,
+        )
+        inst._mqtt = MagicMock()
+        # Pre-seed the recent-queue dict — pretend _add_to_print_queue just
+        # committed item id 42.
+        inst._recent_queue_items["test.3mf"] = ([42], 1_000_000.0)
+        # Drive _restamp via on_print_command on the late-MQTT path.
+        with patch("backend.app.services.virtual_printer.manager.time.monotonic", return_value=1_000_001.0):
+            await inst.on_print_command(
+                "test.3mf",
+                {
+                    "command": "project_file",
+                    "file": "test.3mf",
+                    "nozzle_mapping": [16, -1],
+                },
+            )
+        # No UPDATE was issued — only the eligibility SELECT ran.
+        assert mock_db.execute.await_count == 1
+        mock_db.commit.assert_not_awaited()
+        assert "test.3mf" not in inst._recent_queue_items
+
 
 class TestVirtualPrinterManager:
     """Tests for VirtualPrinterManager orchestrator."""