fix(queue): claim a queue item before dispatch so it can't be reassigned mid-upload (#2615)

A queue row stays status='pending' for the whole FTP upload; status only flips
to 'printing' at the end. The edit routes only blocked non-pending rows, so a
PATCH during the upload window was accepted while the in-flight dispatch kept
using its snapshotted printer -- splitting the queue row from the archive /
expected-print / physical command across two printers, and enabling a duplicate
dispatch on restart. The #1853 CAS guards cancellation, not reassignment.

Add a dispatching_at claim, stamped atomically (WHERE status='pending' AND
dispatching_at IS NULL) before any slow I/O and cleared on every exit. While
held, the single-item PATCH returns 409 (re-checked just before the write),
bulk edits skip the row, and the scheduler won't re-select it. Startup
reconciliation clears claims orphaned by a crash mid-dispatch. The row stays
pending throughout, so no status/UI/completion/reconciliation path changes.

New column print_queue.dispatching_at (nullable, dialect-safe DDL). Covered by
scheduler tests (claim exclusivity, non-pending rejection, release-on-exit,
skip-already-claimed, startup stale-clear) and API tests (reassign 409,
printer_id unchanged, bulk skip, unclaimed row still edits).

backend/app/api/routes/print_queue.py

@@ -774,7 +774,10 @@ async def bulk_update_queue_items(
     skipped_count = 0
 
     for item in items:
-        if item.status != "pending":
+        # Skip non-pending rows and rows a dispatch worker has claimed (#2615) —
+        # editing a claimed row mid-upload would split it from the in-flight
+        # dispatch, so it's excluded from the bulk change (cancel to move it).
+        if item.status != "pending" or item.dispatching_at is not None:
             skipped_count += 1
             continue
 
@@ -1082,6 +1085,14 @@ async def update_queue_item(
     if item.status != "pending":
         raise HTTPException(400, "Can only update pending items")
 
+    # Dispatch claim (#2615): the row is pending but a scheduler worker has
+    # already claimed it and is uploading to its printer. Editing now (e.g.
+    # reassigning printer_id) would split the queue row from the in-flight
+    # archive/expected-print/physical command. Reject until dispatch finishes;
+    # to move it, cancel first (the coordinated escape) and re-queue.
+    if item.dispatching_at is not None:
+        raise HTTPException(409, "Item is being dispatched — cancel it first to make changes")
+
     update_data = data.model_dump(exclude_unset=True)
 
     # Normalize target_model if being updated
@@ -1153,6 +1164,16 @@ async def update_queue_item(
             json.dumps(update_data["nozzle_mapping"]) if update_data["nozzle_mapping"] else None
         )
 
+    # Re-check the dispatch claim right before mutating (#2615). Several awaited
+    # validations ran since the guard above, and a scheduler worker may have
+    # claimed the row in that gap. A fresh read (item isn't dirty yet, so no
+    # autoflush races the check) narrows the window to effectively nothing.
+    claimed = (
+        await db.execute(select(PrintQueueItem.dispatching_at).where(PrintQueueItem.id == item_id))
+    ).scalar_one_or_none()
+    if claimed is not None:
+        raise HTTPException(409, "Item is being dispatched — cancel it first to make changes")
+
     for field, value in update_data.items():
         setattr(item, field, value)
 

backend/app/core/database.py

@@ -1521,6 +1521,22 @@ async def run_migrations(conn):
         except (OperationalError, ProgrammingError):
             pass  # Already applied
 
+    # Migration: Add dispatching_at claim column to print_queue (#2615). Nullable
+    # timestamp; the type differs by dialect (SQLite DATETIME vs Postgres
+    # TIMESTAMP) so an existing-DB upgrade doesn't hit "type datetime does not
+    # exist" on Postgres. On a fresh DB create_all() already built the column, so
+    # the ALTER is swallowed as "already exists".
+    #
+    # Placed AFTER the print_queue_new2 table-recreate above: that recreate
+    # (SQLite-only, and only on ancient DBs whose archive_id is still NOT NULL)
+    # rebuilds print_queue from an explicit column list that doesn't carry this
+    # column, so adding it earlier would let the recreate silently drop it. Adding
+    # it here means it survives that path.
+    if is_sqlite():
+        await _safe_execute(conn, "ALTER TABLE print_queue ADD COLUMN dispatching_at DATETIME")
+    else:
+        await _safe_execute(conn, "ALTER TABLE print_queue ADD COLUMN dispatching_at TIMESTAMP")
+
     # Migration: Add HA energy sensor entity columns to smart_plugs
     await _safe_execute(conn, "ALTER TABLE smart_plugs ADD COLUMN ha_power_entity VARCHAR(100)")
     await _safe_execute(conn, "ALTER TABLE smart_plugs ADD COLUMN ha_energy_today_entity VARCHAR(100)")

backend/app/models/print_queue.py

@@ -111,6 +111,16 @@ class PrintQueueItem(Base):
     # Status: pending, printing, completed, failed, skipped, cancelled
     status: Mapped[str] = mapped_column(String(20), default="pending")
 
+    # Dispatch claim (#2615). Set atomically by the scheduler the moment it
+    # begins dispatching this row and cleared when dispatch ends. The row stays
+    # `status='pending'` throughout the (slow) FTP upload, which left a window
+    # where a concurrent PATCH could reassign printer_id mid-upload and split the
+    # queue row from the archive/expected-print/physical command. While this is
+    # set the edit routes reject changes (409) and the scheduler won't re-select
+    # the row. Startup reconciliation clears any left over by a crash mid-dispatch
+    # (no coroutine survives a restart), so a stale claim never wedges an item.
+    dispatching_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
+
     # Cleared by the per-printer "Resume after failure" action (#1818) so the
     # scheduler's `_check_previous_success` lookback skips this row. Without
     # this, a single `failed` or `aborted` print poisoned every later

backend/app/services/print_scheduler.py

@@ -286,6 +286,8 @@ class PrintScheduler:
         self._running = True
         logger.info("Print scheduler started")
 
+        await self._clear_stale_dispatch_claims()
+
         while self._running:
             dispatched = False
             try:
@@ -297,6 +299,25 @@ class PrintScheduler:
             # not stall behind the idle interval; otherwise sleep normally (#2555).
             await asyncio.sleep(self._fast_check_interval if dispatched else self._check_interval)
 
+    async def _clear_stale_dispatch_claims(self) -> None:
+        """Clear dispatch claims left behind by a crash/restart mid-upload (#2615).
+
+        A claim is only ever held by a live dispatch coroutine, and no coroutine
+        survives a process restart — so every ``dispatching_at`` present at startup
+        is stale. Clearing them lets those still-pending rows be re-selected for a
+        fresh, consistent dispatch instead of being wedged out of the selection
+        query forever. Called once at the top of ``run()``."""
+        try:
+            async with async_session() as db:
+                res = await db.execute(
+                    update(PrintQueueItem).where(PrintQueueItem.dispatching_at.is_not(None)).values(dispatching_at=None)
+                )
+                await db.commit()
+                if res.rowcount:
+                    logger.info("Cleared %d stale dispatch claim(s) at startup (#2615)", res.rowcount)
+        except Exception as exc:
+            logger.error("Failed to clear stale dispatch claims at startup: %s", exc)
+
     def stop(self):
         """Stop the scheduler."""
         self._running = False
@@ -320,6 +341,11 @@ class PrintScheduler:
                 result = await db.execute(
                     select(PrintQueueItem)
                     .where(PrintQueueItem.status == "pending")
+                    # Never re-select a row a dispatch worker has already claimed
+                    # (#2615) — belt-and-suspenders with the _inflight exclusion
+                    # below, and the guard that lets an orphaned claim be ignored
+                    # until startup reconciliation clears it.
+                    .where(PrintQueueItem.dispatching_at.is_(None))
                     # archive/library_file are read by the cross-model gate
                     # (#2578); eager-load once per pass instead of a lazy-load
                     # (which would raise in async) per item.
@@ -339,6 +365,8 @@ class PrintScheduler:
                 result = await db.execute(
                     select(PrintQueueItem)
                     .where(PrintQueueItem.status == "pending")
+                    # Skip rows already claimed by a dispatch worker (#2615).
+                    .where(PrintQueueItem.dispatching_at.is_(None))
                     .options(
                         selectinload(PrintQueueItem.archive),
                         selectinload(PrintQueueItem.library_file),
@@ -880,11 +908,56 @@ class PrintScheduler:
         transfer's duration.
         """
         async with 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)
+            # Claim the row for dispatch BEFORE reading the printer snapshot or
+            # touching any slow I/O (#2615). The claim is an atomic CAS on
+            # (status='pending', dispatching_at IS NULL); while it's held the edit
+            # 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):
+                logger.info(
+                    "Queue item %s not claimable for dispatch (cancelled, removed, or already claimed) — skipping",
+                    item_id,
+                )
                 return
-            await self._start_print(item_db, item)
+            try:
+                item = await item_db.get(PrintQueueItem, item_id)
+                if not item:
+                    logger.info("Queue item %s vanished after claim — skipping", item_id)
+                    return
+                await self._start_print(item_db, item)
+            finally:
+                # Release the claim on every exit. Once dispatch has finished the
+                # row's status carries the lock (printing/failed/cancelled are all
+                # != pending), so the token is only needed for the duration of the
+                # upload. A row left pending (e.g. busy-printer deferral) becomes
+                # dispatchable again on the next tick.
+                await self._clear_dispatch_claim(item_db, item_id)
+
+    async def _claim_for_dispatch(self, db: AsyncSession, item_id: int) -> 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(
+            update(PrintQueueItem)
+            .where(PrintQueueItem.id == item_id)
+            .where(PrintQueueItem.status == "pending")
+            .where(PrintQueueItem.dispatching_at.is_(None))
+            .values(dispatching_at=datetime.now(timezone.utc))
+        )
+        await db.commit()
+        return res.rowcount > 0
+
+    async def _clear_dispatch_claim(self, db: AsyncSession, item_id: int) -> None:
+        """Clear the dispatch claim (#2615). Best-effort: a failure here must not
+        mask the dispatch outcome, and startup reconciliation clears any leftover."""
+        try:
+            await db.execute(update(PrintQueueItem).where(PrintQueueItem.id == item_id).values(dispatching_at=None))
+            await db.commit()
+        except Exception as exc:
+            logger.warning("Queue item %s: failed to clear dispatch claim: %s", item_id, exc)
 
     async def _find_idle_printer_for_model(
         self,

backend/tests/integration/test_print_queue_api.py

@@ -333,6 +333,57 @@ class TestPrintQueueAPI:
         assert result["bed_levelling"] is False
         assert result["timelapse"] is True
 
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_reassign_rejected_while_dispatching(
+        self, async_client: AsyncClient, queue_item_factory, printer_factory, db_session
+    ):
+        """#2615: a claimed (in-flight) row rejects edits with 409, so its printer
+        can't be reassigned out from under the running FTP upload."""
+        from datetime import datetime, timezone
+
+        item = await queue_item_factory(dispatching_at=datetime.now(timezone.utc))
+        other = await printer_factory()
+        original_printer_id = item.printer_id
+
+        response = await async_client.patch(f"/api/v1/queue/{item.id}", json={"printer_id": other.id})
+        assert response.status_code == 409
+
+        await db_session.refresh(item)
+        assert item.printer_id == original_printer_id, "printer_id must not change on a dispatching row"
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_bulk_update_skips_dispatching_item(
+        self, async_client: AsyncClient, queue_item_factory, printer_factory, db_session
+    ):
+        """#2615: bulk edits skip a claimed row rather than splitting it."""
+        from datetime import datetime, timezone
+
+        item = await queue_item_factory(dispatching_at=datetime.now(timezone.utc))
+        other = await printer_factory()
+        original_printer_id = item.printer_id
+
+        response = await async_client.patch("/api/v1/queue/bulk", json={"item_ids": [item.id], "printer_id": other.id})
+        assert response.status_code == 200
+        body = response.json()
+        assert body["skipped_count"] == 1
+        assert body["updated_count"] == 0
+
+        await db_session.refresh(item)
+        assert item.printer_id == original_printer_id
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_update_allowed_on_unclaimed_pending_item(
+        self, async_client: AsyncClient, queue_item_factory, db_session
+    ):
+        """Regression guard: a normal pending row (no claim) still edits fine."""
+        item = await queue_item_factory()
+        response = await async_client.patch(f"/api/v1/queue/{item.id}", json={"plate_id": 7})
+        assert response.status_code == 200
+        assert response.json()["plate_id"] == 7
+
     @pytest.mark.asyncio
     @pytest.mark.integration
     async def test_get_queue_item(self, async_client: AsyncClient, queue_item_factory, db_session):

backend/tests/unit/test_scheduler_reassign_race_2615.py

@@ -0,0 +1,138 @@
+"""Reassign-during-dispatch race regression (#2615).
+
+A queue row stays ``status='pending'`` for the whole (slow) FTP upload — status
+only flips to ``printing`` at the very end. That left a window where a PATCH
+could reassign ``printer_id`` mid-upload while the in-flight dispatch kept using
+the old printer, splitting the queue row from the archive / expected-print /
+physical command. The fix is a ``dispatching_at`` claim, stamped atomically
+before any slow I/O, that the edit routes reject on and the scheduler won't
+re-select. These tests cover the claim primitives, the guaranteed release, and
+the startup reconciliation that clears a claim orphaned by a crash mid-dispatch.
+"""
+
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, patch
+
+import pytest
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+
+import backend.app.models  # noqa: F401 - populate Base.metadata
+import backend.app.services.print_scheduler as scheduler_module
+from backend.app.core.database import Base
+from backend.app.models.print_queue import PrintQueueItem
+from backend.app.models.printer import Printer
+from backend.app.services.print_scheduler import PrintScheduler
+
+
+@pytest.fixture
+async def ctx():
+    engine = create_async_engine("sqlite+aiosqlite:///:memory:", echo=False)
+    async with engine.begin() as conn:
+        await conn.run_sync(Base.metadata.create_all)
+    sm = async_sessionmaker(engine, expire_on_commit=False)
+
+    async with sm() as db:
+        printer = Printer(name="P", serial_number="S", ip_address="127.0.0.1", access_code="c", model="X1C")
+        db.add(printer)
+        await db.flush()
+        item = PrintQueueItem(printer_id=printer.id, status="pending")
+        db.add(item)
+        await db.commit()
+        item_id = item.id
+
+    try:
+        yield SimpleNamespace(sm=sm, item_id=item_id, printer_id=printer.id)
+    finally:
+        await engine.dispose()
+
+
+async def _get(ctx, item_id=None):
+    async with ctx.sm() as db:
+        return await db.get(PrintQueueItem, item_id or ctx.item_id)
+
+
+@pytest.mark.asyncio
+async def test_claim_stamps_pending_row_and_is_exclusive(ctx):
+    sched = PrintScheduler()
+    async with ctx.sm() as db:
+        assert await sched._claim_for_dispatch(db, ctx.item_id) is True
+    assert (await _get(ctx)).dispatching_at is not None
+
+    # A second claim on an already-claimed row loses.
+    async with ctx.sm() as db:
+        assert await sched._claim_for_dispatch(db, ctx.item_id) is False
+
+
+@pytest.mark.asyncio
+async def test_claim_fails_on_non_pending_row(ctx):
+    sched = PrintScheduler()
+    async with ctx.sm() as db:
+        item = await db.get(PrintQueueItem, ctx.item_id)
+        item.status = "printing"
+        await db.commit()
+    async with ctx.sm() as db:
+        assert await sched._claim_for_dispatch(db, ctx.item_id) is False
+    assert (await _get(ctx)).dispatching_at is None
+
+
+@pytest.mark.asyncio
+async def test_clear_releases_the_claim(ctx):
+    sched = PrintScheduler()
+    async with ctx.sm() as db:
+        await sched._claim_for_dispatch(db, ctx.item_id)
+    async with ctx.sm() as db:
+        await sched._clear_dispatch_claim(db, ctx.item_id)
+    assert (await _get(ctx)).dispatching_at is None
+
+
+@pytest.mark.asyncio
+async def test_dispatch_one_claims_then_releases_around_start_print(ctx):
+    sched = PrintScheduler()
+    seen = {}
+
+    async def fake_start_print(db, item):
+        # Observe the claim is held while dispatch runs.
+        row = await db.get(PrintQueueItem, item.id)
+        seen["claimed_during"] = row.dispatching_at is not None
+
+    with (
+        patch.object(scheduler_module, "async_session", ctx.sm),
+        patch.object(sched, "_start_print", side_effect=fake_start_print) as sp,
+    ):
+        await sched._dispatch_one(ctx.item_id)
+
+    assert seen["claimed_during"] is True, "claim must be held while dispatch runs"
+    sp.assert_awaited_once()
+    # Released on exit so a deferred (still-pending) row can re-dispatch.
+    assert (await _get(ctx)).dispatching_at is None
+
+
+@pytest.mark.asyncio
+async def test_dispatch_one_skips_an_already_claimed_row(ctx):
+    sched = PrintScheduler()
+    # Pre-claim the row (as if another worker owns it).
+    async with ctx.sm() as db:
+        await sched._claim_for_dispatch(db, ctx.item_id)
+
+    with (
+        patch.object(scheduler_module, "async_session", ctx.sm),
+        patch.object(sched, "_start_print", new=AsyncMock()) as sp,
+    ):
+        await sched._dispatch_one(ctx.item_id)
+
+    sp.assert_not_called()  # claim lost → no dispatch
+    # And it must NOT clear the other worker's claim.
+    assert (await _get(ctx)).dispatching_at is not None
+
+
+@pytest.mark.asyncio
+async def test_startup_reconciliation_clears_stale_claims(ctx):
+    sched = PrintScheduler()
+    async with ctx.sm() as db:
+        await sched._claim_for_dispatch(db, ctx.item_id)
+    assert (await _get(ctx)).dispatching_at is not None
+
+    with patch.object(scheduler_module, "async_session", ctx.sm):
+        await sched._clear_stale_dispatch_claims()
+
+    assert (await _get(ctx)).dispatching_at is None, "a claim orphaned by a restart must be cleared"