fix(queue): a job edited during a scheduler round waits for the next one instead of being dispatched or mapped from the stale copy; the claim requires the selected printer; bulk edit rechecks the claim; no transaction held through a power-on wait

CHANGELOG.md

@@ -164,6 +164,7 @@ All notable changes to Bambuddy will be documented in this file.
 - **The frontend build no longer warns about `path` and `crypto` being externalized for the STEP previewer (#2976)** — `occt-import-js`, the Emscripten build behind STEP previews, requires both modules, but only inside its `ENVIRONMENT_IS_NODE` branches; in the browser it loads its `.wasm` from the URL the preview worker passes and draws randomness from `crypto.getRandomValues`. Vite still externalized both and printed two warnings on every build. `vite.config.ts` now drops exactly those two warnings for that one package through `build.rolldownOptions.onLog`, so an externalization anywhere else, or of any other module, still shows.
 
 ### Fixed
+- **Queue edits made while the queue is being worked through are kept** — The scheduler works on the queue as it read it at the start of each round, and a round can take minutes while a smart plug powers a printer on. A job edited in the meantime lost the edit: an AMS mapping set by hand was overwritten, a job moved to another printer (or an "Any model" job pinned to one) could still be sent to the printer the round had picked, and a new plate could be mapped from the old plate's filaments. A job edited during a round now waits for the next one, which reads the edit. A bulk edit likewise leaves a job that started dispatching while the edit was checked, instead of changing its printer or cost centre under the upload. While waiting for a printer to power on, the scheduler no longer keeps a database transaction open.
 - **A print's filament and cost are charged with its own plate and AMS mapping** — Bambuddy remembers per archive which plate and AMS slots a dispatch uses, for the usage and cost charged when the print ends. Three cases used the wrong ones:
   - **A whole file queued after one plate of it** was sent but never started (refused, or the printer stayed idle) was charged as that plate, with its slots.
   - **A plate that failed to send** wiped the plate number and slots of another plate of the same archive still printing, whose end then charged the whole file.

backend/app/api/routes/print_queue.py

@@ -1336,17 +1336,12 @@ async def bulk_update_queue_items(
             continue
 
         item_update_data = update_data.copy()
-        if "printer_id" in item_update_data and item_update_data["printer_id"] != item.printer_id:
-            # A job with printer alternatives picks its own printer (#671),
-            # as the single edit refuses to override.
-            if item.variants:
-                skipped_count += 1
-                continue
-            # Now bound to this printer: not to a model, and nothing worked
-            # out for the old printer carries over.
-            item.target_model = None
-            item.target_location = None
-            drop_printer_resolved(item)
+        rebinds = "printer_id" in item_update_data and item_update_data["printer_id"] != item.printer_id
+        # A job with printer alternatives picks its own printer (#671), as the
+        # single edit refuses to override.
+        if rebinds and item.variants:
+            skipped_count += 1
+            continue
         if validates_billing_fields:
             trusted_estimated_cost = await _trusted_item_estimated_cost(
                 db,
@@ -1364,6 +1359,25 @@ async def bulk_update_queue_items(
                 exclude_queue_item_id=item.id,
             )
 
+        # Re-check right before changing it, as the single edit does (#2615):
+        # the validations above are awaited, and a scheduler worker may have
+        # claimed the row (or it stopped being pending) meanwhile. Nothing on
+        # this row has been changed yet, so skipping it leaves it as it is.
+        current = (
+            await db.execute(
+                select(PrintQueueItem.status, PrintQueueItem.dispatching_at).where(PrintQueueItem.id == item.id)
+            )
+        ).one_or_none()
+        if current is None or current.status != "pending" or current.dispatching_at is not None:
+            skipped_count += 1
+            continue
+
+        if rebinds:
+            # Now bound to this printer: not to a model, and nothing worked
+            # out for the old printer carries over.
+            item.target_model = None
+            item.target_location = None
+            drop_printer_resolved(item)
         for field, value in item_update_data.items():
             setattr(item, field, value)
         updated_count += 1

backend/app/services/print_scheduler.py

@@ -733,6 +733,41 @@ def _int_or(value, default: int) -> int:
         return default
 
 
+# What a queue job's dispatch is worked out from, and an edit can change.
+_DISPATCH_INPUTS = (
+    "printer_id",
+    "status",
+    "ams_mapping",
+    "plate_id",
+    "filament_overrides",
+    "skip_filament_check",
+    "target_model",
+    "target_location",
+)
+
+
+async def _edited_since_loaded(db: AsyncSession, item: PrintQueueItem) -> bool:
+    """Whether ``item`` was changed or removed since ``check_queue`` loaded it.
+
+    The pass works on the jobs as loaded at its start, and it may have waited
+    since: a plug powering a printer on takes minutes. A job edited meanwhile
+    (moved to another printer, given a mapping or another plate, cancelled)
+    is left to the next pass, which reads it fresh: deciding from the stale
+    copy would overwrite the edit, or send the job where it no longer goes.
+    The query flushes the pass's own pending changes first, so those don't
+    count as edits.
+    """
+    row = (
+        await db.execute(
+            select(*(getattr(PrintQueueItem, name) for name in _DISPATCH_INPUTS)).where(PrintQueueItem.id == item.id)
+        )
+    ).one_or_none()
+    if row is not None and all(getattr(item, name) == value for name, value in zip(_DISPATCH_INPUTS, row, strict=True)):
+        return False
+    logger.info("Queue item %s changed during this pass; leaving it to the next", item.id)
+    return True
+
+
 def _global_tray_id(ams_id: int, tray_id: int) -> int:
     """Bambu's flat tray addressing: ``ams_id * 4 + tray_id`` for a four-slot
     unit, and the bare unit id for an AMS-HT (ids from 128, one tray each).
@@ -2131,6 +2166,9 @@ class PrintScheduler:
                     # (all -1). A stored all-[-1] mapping is a bug artifact — a
                     # frontend status-load race can persist [-1] (#2589) — and
                     # must be recomputed from live trays rather than trusted.
+                    if await _edited_since_loaded(db, item):
+                        continue
+
                     unmappable = await self._ensure_ams_mapping(db, item.printer_id, item)
                     if unmappable:
                         await self._fail_unmappable_item(db, item, item.printer_id, unmappable)
@@ -2396,6 +2434,9 @@ class PrintScheduler:
                             skip_reasons["library_row_in_use"] = skip_reasons.get("library_row_in_use", 0) + 1
                             continue
 
+                        if await _edited_since_loaded(db, item):
+                            continue
+
                         # Assign printer and start - clear waiting reason
                         item.printer_id = printer_id
                         item.waiting_reason = None
@@ -2655,7 +2696,7 @@ class PrintScheduler:
             # routes reject reassignment (409), so printer_id can't change out from
             # under the in-flight upload and split the queue row from the
             # archive/expected-print/physical command.
-            if not await self._claim_for_dispatch(item_db, item_id):
+            if not await self._claim_for_dispatch(item_db, item_id, selected_printer_id):
                 logger.info(
                     "Queue item %s not claimable for dispatch (cancelled, removed, or already claimed) — skipping",
                     item_id,
@@ -2959,19 +3000,24 @@ class PrintScheduler:
             sorted(pin),
         )
 
-    async def _claim_for_dispatch(self, db: AsyncSession, item_id: int) -> bool:
+    async def _claim_for_dispatch(self, db: AsyncSession, item_id: int, printer_id: int | None = None) -> bool:
         """Atomically stamp ``dispatching_at`` on a still-pending, unclaimed row.
 
         Returns True if this call won the claim, False if the row was already
         claimed, no longer pending (cancelled mid-tick), or removed. The CAS is
-        the load-bearing guard against reassign-during-dispatch (#2615)."""
-        res = await db.execute(
+        the load-bearing guard against reassign-during-dispatch (#2615).
+        ``printer_id`` is the printer the item was selected for: a row moved to
+        another printer since is not claimed, and the next pass selects it for
+        that one."""
+        query = (
             update(PrintQueueItem)
             .where(PrintQueueItem.id == item_id)
             .where(PrintQueueItem.status == "pending")
             .where(PrintQueueItem.dispatching_at.is_(None))
-            .values(dispatching_at=datetime.now(timezone.utc))
         )
+        if printer_id is not None:
+            query = query.where(PrintQueueItem.printer_id == printer_id)
+        res = await db.execute(query.values(dispatching_at=datetime.now(timezone.utc)))
         await db.commit()
         return res.rowcount > 0
 
@@ -7630,6 +7676,11 @@ class PrintScheduler:
         if not printer:
             logger.error("Printer %s not found in database", printer_id)
             return False
+        # End the pass's transaction before waiting up to minutes for the boot:
+        # held open, it pins a pooled connection idle in transaction
+        # (PostgreSQL) and a stale read snapshot (SQLite). Nothing below reads
+        # through it.
+        await db.commit()
 
         # Wait for printer to boot (give it some time before trying to connect)
         logger.info("Waiting 30s for printer %s to boot...", printer_id)

backend/tests/integration/test_bulk_edit_dispatch_claim.py

@@ -0,0 +1,58 @@
+"""Bulk edit leaves a row a dispatch claimed during the edit (DM/BULK).
+
+The bulk edit checked the claim on the rows it loaded, then awaited cost and
+budget checks per row before changing it. A row the scheduler claimed for
+dispatch in between was changed anyway, so its printer or cost centre no
+longer matched the upload already under way.
+"""
+
+from datetime import datetime, timezone
+from unittest.mock import AsyncMock, patch
+
+import pytest
+from sqlalchemy import update
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from backend.app.models.finance import CostCenter
+from backend.app.models.print_queue import PrintQueueItem
+
+pytestmark = [pytest.mark.asyncio, pytest.mark.integration]
+
+
+async def test_a_row_claimed_during_the_edit_is_skipped(
+    async_client, db_session, test_engine, printer_factory, archive_factory
+):
+    printer = await printer_factory()
+    other_printer = await printer_factory(name="Other")
+    archive = await archive_factory(printer.id)
+    center = CostCenter(name="Lab", is_active=True, is_private=False)
+    db_session.add(center)
+    claimed = PrintQueueItem(printer_id=printer.id, archive_id=archive.id, status="pending", target_model="X1C")
+    free = PrintQueueItem(printer_id=printer.id, archive_id=archive.id, status="pending")
+    db_session.add_all([claimed, free])
+    await db_session.commit()
+    other = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False)
+
+    async def dispatch_claims(*_args, exclude_queue_item_id=None, **_kwargs):
+        if exclude_queue_item_id == claimed.id:
+            async with other() as db:
+                await db.execute(
+                    update(PrintQueueItem)
+                    .where(PrintQueueItem.id == claimed.id)
+                    .values(dispatching_at=datetime.now(timezone.utc))
+                )
+                await db.commit()
+
+    with patch("backend.app.api.routes.print_queue.validate_print_budget", AsyncMock(side_effect=dispatch_claims)):
+        response = await async_client.patch(
+            "/api/v1/queue/bulk",
+            json={"item_ids": [claimed.id, free.id], "cost_center_id": center.id, "printer_id": other_printer.id},
+        )
+
+    assert response.status_code == 200, response.text
+    assert (response.json()["updated_count"], response.json()["skipped_count"]) == (1, 1)
+    async with other() as db:
+        claimed_row = await db.get(PrintQueueItem, claimed.id)
+        free_row = await db.get(PrintQueueItem, free.id)
+    assert (claimed_row.printer_id, claimed_row.cost_center_id, claimed_row.target_model) == (printer.id, None, "X1C")
+    assert (free_row.printer_id, free_row.cost_center_id) == (other_printer.id, center.id)

backend/tests/unit/test_scheduler_snapshot_races.py

@@ -0,0 +1,196 @@
+"""An edit made while the scheduler's pass runs is not overwritten (Q2).
+
+``check_queue`` loads the pending items once and works on that copy. A queue
+item edited in between (a mapping set by hand, a move to another printer) was
+overwritten when the pass computed and saved its AMS mapping, or dispatched
+to the printer it was on before. The edit lands where the pass waits on
+something; the previous-print check is such a place.
+"""
+
+from unittest.mock import AsyncMock, patch
+
+import pytest
+from sqlalchemy import update
+
+from backend.app.models.print_queue import PrintQueueItem
+from backend.app.models.printer import Printer
+from backend.tests.unit.test_scheduler_busy_reasons_3018 import _scheduler, one_printer  # noqa: F401
+
+
+async def _pinned_item(ctx, ams_mapping: str | None = None) -> int:
+    item_id = await ctx.add_item()
+    async with ctx.session_maker() as db:
+        await db.execute(
+            update(PrintQueueItem)
+            .where(PrintQueueItem.id == item_id)
+            .values(require_previous_success=True, ams_mapping=ams_mapping)
+        )
+        await db.commit()
+    return item_id
+
+
+def _edit_during_the_pass(ctx, scheduler, **values):
+    async def edit(_db, item, *_args, **_kwargs):
+        async with ctx.session_maker() as other:
+            await other.execute(update(PrintQueueItem).where(PrintQueueItem.id == item.id).values(**values))
+            await other.commit()
+        return True
+
+    return patch.object(scheduler, "_check_previous_success", AsyncMock(side_effect=edit))
+
+
+@pytest.mark.asyncio
+async def test_a_mapping_set_meanwhile_is_kept(one_printer):  # noqa: F811
+    """The pass would clear the bogus all-unresolved mapping it loaded."""
+    item_id = await _pinned_item(one_printer, ams_mapping="[-1]")
+
+    async with _scheduler(one_printer, idle=True) as scheduler:
+        with _edit_during_the_pass(one_printer, scheduler, ams_mapping="[2]"):
+            await scheduler.check_queue()
+
+    async with one_printer.session_maker() as db:
+        assert (await db.get(PrintQueueItem, item_id)).ams_mapping == "[2]"
+
+
+@pytest.mark.asyncio
+async def test_an_item_moved_meanwhile_is_not_sent_to_its_old_printer(one_printer):  # noqa: F811
+    """Nor given a mapping worked out for its old printer: the pass would
+    clear the bogus all-unresolved one it loaded."""
+    item_id = await _pinned_item(one_printer, ams_mapping="[-1]")
+    async with one_printer.session_maker() as db:
+        other = Printer(name="Other", serial_number="OTHER", ip_address="10.0.0.2", access_code="code", model="X1C")
+        db.add(other)
+        await db.commit()
+        other_id = other.id
+
+    async with _scheduler(one_printer, idle=True) as scheduler:
+        with _edit_during_the_pass(one_printer, scheduler, printer_id=other_id):
+            await scheduler.check_queue()
+            assert not scheduler._inflight
+
+    async with one_printer.session_maker() as db:
+        item = await db.get(PrintQueueItem, item_id)
+    assert (item.status, item.printer_id, item.ams_mapping) == ("pending", other_id, "[-1]")
+
+
+@pytest.mark.asyncio
+async def test_the_claim_requires_the_printer_the_item_was_selected_for(one_printer):  # noqa: F811
+    """A move landing after the re-read: the dispatch's claim still refuses it."""
+    from backend.app.services.print_scheduler import PrintScheduler
+
+    item_id = await one_printer.add_item()
+    scheduler = PrintScheduler()
+
+    async with one_printer.session_maker() as db:
+        assert not await scheduler._claim_for_dispatch(db, item_id, one_printer.printer_id + 1)
+        assert await scheduler._claim_for_dispatch(db, item_id, one_printer.printer_id)
+
+
+@pytest.mark.asyncio
+async def test_the_power_on_wait_holds_no_transaction(one_printer):  # noqa: F811
+    """RC-C8: 30-180 s of waiting for a printer to boot with the pass's
+    transaction open pins a connection (PostgreSQL) or a stale snapshot."""
+    from unittest.mock import MagicMock
+
+    from backend.app.services import print_scheduler as scheduler_module
+    from backend.app.services.print_scheduler import PrintScheduler
+
+    scheduler = PrintScheduler()
+    order: list[str] = []
+    plug = MagicMock(name="plug")
+    plug.name = "Plug"
+    service = MagicMock(
+        get_status=AsyncMock(return_value={"reachable": True, "state": "ON"}), turn_on=AsyncMock(return_value=True)
+    )
+
+    async def sleep(_seconds):
+        order.append("sleep")
+
+    async with one_printer.session_maker() as db:
+        real_commit = db.commit
+
+        async def commit():
+            order.append("commit")
+            await real_commit()
+
+        with (
+            patch.object(scheduler_module.smart_plug_manager, "get_service_for_plug", AsyncMock(return_value=service)),
+            patch.object(scheduler_module.asyncio, "sleep", sleep),
+            patch.object(scheduler_module.printer_manager, "connect_printer", AsyncMock(return_value=True)),
+            patch.object(db, "commit", commit),
+        ):
+            assert await scheduler._power_on_and_wait(plug, one_printer.printer_id, db)
+
+    assert order[:2] == ["commit", "sleep"]
+
+
+@pytest.mark.asyncio
+async def test_an_item_cancelled_meanwhile_is_left_as_cancelled(one_printer):  # noqa: F811
+    item_id = await _pinned_item(one_printer, ams_mapping="[-1]")
+
+    async with _scheduler(one_printer, idle=True) as scheduler:
+        with _edit_during_the_pass(one_printer, scheduler, status="cancelled"):
+            await scheduler.check_queue()
+
+    async with one_printer.session_maker() as db:
+        item = await db.get(PrintQueueItem, item_id)
+    assert (item.status, item.ams_mapping) == ("cancelled", "[-1]")
+
+
+@pytest.mark.asyncio
+async def test_another_plate_chosen_meanwhile_is_not_mapped_from_the_old_one(one_printer):  # noqa: F811
+    """The new mapping belongs to the new plate; checked against the old
+    plate's filaments it would be thrown away."""
+    item_id = await _pinned_item(one_printer, ams_mapping="[-1]")
+
+    async with _scheduler(one_printer, idle=True) as scheduler:
+        with _edit_during_the_pass(one_printer, scheduler, plate_id=2, ams_mapping="[3]"):
+            await scheduler.check_queue()
+            assert not scheduler._inflight
+
+    async with one_printer.session_maker() as db:
+        item = await db.get(PrintQueueItem, item_id)
+    assert (item.plate_id, item.ams_mapping) == (2, "[3]")
+
+
+@pytest.mark.asyncio
+async def test_a_model_job_pinned_meanwhile_is_not_sent_to_the_printer_it_was_offered(one_printer):  # noqa: F811
+    item_id = await _pinned_item(one_printer)
+    async with one_printer.session_maker() as db:
+        other = Printer(name="Other", serial_number="OTHER", ip_address="10.0.0.2", access_code="code", model="X1C")
+        db.add(other)
+        await db.commit()
+        other_id = other.id
+        await db.execute(
+            update(PrintQueueItem).where(PrintQueueItem.id == item_id).values(printer_id=None, target_model="X1C")
+        )
+        await db.commit()
+
+    async with _scheduler(one_printer, idle=True) as scheduler:
+        with _edit_during_the_pass(one_printer, scheduler, printer_id=other_id, target_model=None):
+            await scheduler.check_queue()
+            assert not scheduler._inflight
+
+    async with one_printer.session_maker() as db:
+        item = await db.get(PrintQueueItem, item_id)
+    assert (item.status, item.printer_id, item.target_model) == ("pending", other_id, None)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "edit",
+    [
+        {"plate_id": 2},
+        {"filament_overrides": '[{"slot_id": 1, "type": "PETG"}]'},
+        {"skip_filament_check": True},
+        {"target_model": "P1S"},
+        {"target_location": "Lab"},
+    ],
+)
+async def test_any_edit_to_what_the_dispatch_is_worked_out_from_waits_a_round(one_printer, edit):  # noqa: F811
+    await _pinned_item(one_printer)
+
+    async with _scheduler(one_printer, idle=True) as scheduler:
+        with _edit_during_the_pass(one_printer, scheduler, **edit):
+            await scheduler.check_queue()
+            assert not scheduler._inflight