fix(scheduler): cancel during queue dispatch actually cancels (#1853)

Symptom: user queued a batch of 10 prints, pressed Cancel on a pending
row, the print started anyway. Repeated consecutively. Support bundle
also showed 15x "sqlite3.OperationalError: database is locked" from the
sensor history recorder in the same 8-minute window.

Root cause is a check-then-act race in _start_print. check_queue takes
a snapshot of pending items, then _start_print does FTP delete + FTP
upload (5-30s) before the unconditional item.status = "printing";
db.commit() at line 2792. /cancel commits status='cancelled' in a
separate session during that window; the scheduler's stale in-memory
write overwrites it and start_print ships. The lock-contention finding
is the same shape from a different angle: _start_print did
await db.flush() at line 2555 (after item.archive_id set + library_file
delete) which opens the SQLite WAL writer lock and holds it through
the FTP upload, queueing every concurrent writer behind it including
the user's own cancel commit.

Three guards layered:

1) Atomic CAS at the pending->printing transition. UPDATE print_queue
   SET status='printing', started_at=NOW() WHERE id=:id AND
   status='pending'. rowcount==0 means user won; log abort, best-effort
   delete_file_async the file we just FTP'd up so it doesn't leak into
   the printer's BambuStudio file picker, send queue_item_failed WS
   event with reason="cancelled_mid_dispatch", return without calling
   printer_manager.start_print.

2) Early db.refresh(item) + bail right after the printer connectivity
   check. Saves the wasted FTP upload when the row was already
   cancelled before _start_print resumed. Defense in depth; guard 1
   catches the same case at the CAS point.

3) flush -> commit before the FTP block. The library-file-to-archive
   promotion's writes commit cleanly, WAL writer lock releases, sensor
   history and concurrent cancels stop queueing behind the scheduler.
   The flush-not-commit pattern was rolling back a pointer to an
   already-committed archive row, so the new behaviour matches reality
   (archive committed, pointer committed, FTP unblocked).
This commit is contained in:
maziggy
2026-06-29 12:59:09 +02:00
parent 14665f4281
commit c32bc82dd4
3 changed files with 312 additions and 4 deletions
+2
View File
File diff suppressed because one or more lines are too long
+73 -4
View File
@@ -7,7 +7,7 @@ import time
from datetime import datetime, timezone
from pathlib import Path
from sqlalchemy import func, select
from sqlalchemy import func, select, update
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload
@@ -2486,6 +2486,22 @@ class PrintScheduler:
await self._power_off_if_needed(db, item)
return
# Cancel-while-dispatching race (#1853): the scheduler's snapshot of
# `items` was taken at the top of check_queue, but the user can /cancel
# any pending row in the gap before we reach this point. Re-read the
# row and bail out cleanly instead of starting an FTP upload for a row
# that's already cancelled. The atomic CAS at the pending→printing
# transition (below, before start_print) is the load-bearing guard;
# this is the early-exit optimisation that avoids wasted FTP I/O.
await db.refresh(item)
if item.status != "pending":
logger.info(
"Queue item %s no longer pending (status=%s) — aborting dispatch",
item.id,
item.status,
)
return
# Determine source: archive or library file
archive = None
library_file = None
@@ -2552,7 +2568,12 @@ class PrintScheduler:
await db.delete(library_file)
file_path = settings.base_dir / archive.file_path
filename = archive.filename
await db.flush()
# Commit, not flush — flush opens the SQLite write
# transaction (item.archive_id update + library_file
# delete) and would hold the WAL writer lock through the
# FTP upload below, causing "database is locked" cascades
# for sensor history + concurrent cancels (#1853).
await db.commit()
logger.info(
"Queue item %s: Created archive %s from library file %s",
item.id,
@@ -2789,9 +2810,57 @@ class PrintScheduler:
# If we crash after this commit but before start_print(), the item will be
# in "printing" status without actually printing - but that's safer than
# accidentally reprinting the same file hours later.
item.status = "printing"
item.started_at = datetime.now(timezone.utc)
#
# Atomic CAS (#1853): a user pressing /cancel mid-dispatch (between the
# initial pending read at the top of check_queue and this point) flips
# the row to "cancelled" in a separate session. Without the WHERE
# status='pending' clause, the unconditional update here would silently
# overwrite that cancellation and we'd ship the MQTT start_print below
# — printer obeys, user sees "I pressed cancel and the print started".
# rowcount==0 means the user won the race; bail out, best-effort delete
# the file we just uploaded, do NOT send start_print.
now_utc = datetime.now(timezone.utc)
cas = await db.execute(
update(PrintQueueItem)
.where(PrintQueueItem.id == item.id)
.where(PrintQueueItem.status == "pending")
.values(status="printing", started_at=now_utc)
)
await db.commit()
if cas.rowcount == 0:
logger.info(
"Queue item %s no longer pending at print-command time "
"(cancelled or removed mid-dispatch) — aborting before MQTT send (#1853)",
item.id,
)
try:
await delete_file_async(
printer.ip_address,
printer.access_code,
remote_path,
socket_timeout=ftp_timeout,
printer_model=printer.model,
)
except Exception as cleanup_err:
logger.debug(
"Queue item %s: best-effort cleanup of uploaded file failed: %s",
item.id,
cleanup_err,
)
try:
await ws_manager.send_queue_item_failed(
user_id=toast_uid,
queue_item_id=item.id,
printer_id=item.printer_id,
reason="cancelled_mid_dispatch",
)
except Exception:
pass
return
# Sync the in-memory item so subsequent code that reads item.status /
# item.started_at sees the values we just persisted.
item.status = "printing"
item.started_at = now_utc
for cleanup_path in cleanup_disk_paths:
try:
@@ -0,0 +1,237 @@
"""Cancel-during-dispatch race regression (#1853).
The user reported: queued a batch of 10 prints, pressed Cancel on a pending
item, the print started anyway. Root cause is a check-then-act race in
``_start_print``: the snapshot of pending items is taken at the top of
``check_queue``, then ``_start_print`` does FTP delete + FTP upload (5-30 s)
before flipping the row to ``"printing"`` and sending MQTT. If the user wins
the race and ``/cancel`` lands during that window, the scheduler's stale
in-memory write of ``status="printing"`` silently overwrites the cancellation.
Three guards exercised here:
* Early refresh after the connectivity check — bails before FTP I/O if the
row is already cancelled.
* Atomic CAS at the pending→printing transition — UPDATE WHERE
status='pending'; rowcount==0 means user won, do NOT send MQTT.
* Best-effort delete of the file we just FTP'd up when the CAS aborts.
"""
from contextlib import ExitStack
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock, 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.archive import PrintArchive
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 queue_factory(tmp_path):
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)
case_counter = 0
async def make_case(*, status="pending"):
nonlocal case_counter
case_counter += 1
base_dir = tmp_path / f"case-{case_counter}"
base_dir.mkdir()
archive_rel = Path("archives") / f"job-{case_counter}.3mf"
archive_abs = base_dir / archive_rel
archive_abs.parent.mkdir(parents=True, exist_ok=True)
archive_abs.write_bytes(b"archive payload")
async with session_maker() as db:
printer = Printer(
name=f"Printer {case_counter}",
serial_number=f"SERIAL-{case_counter}",
ip_address="127.0.0.1",
access_code="access-code",
model="X1C",
)
db.add(printer)
await db.flush()
archive = PrintArchive(
printer_id=printer.id,
filename=f"job-{case_counter}.3mf",
file_path=str(archive_rel),
file_size=archive_abs.stat().st_size,
content_hash=None,
thumbnail_path=None,
timelapse_path=None,
print_time_seconds=120,
status="completed",
)
db.add(archive)
await db.flush()
item = PrintQueueItem(
printer_id=printer.id,
archive_id=archive.id,
status=status,
bed_levelling=True,
flow_cali=False,
vibration_cali=True,
layer_inspect=False,
timelapse=False,
use_ams=True,
nozzle_offset_cali=True,
)
db.add(item)
await db.commit()
return SimpleNamespace(
session_maker=session_maker,
base_dir=base_dir,
archive_path=archive_abs,
printer_id=printer.id,
archive_id=archive.id,
queue_item_id=item.id,
upload=AsyncMock(return_value=True),
start_print=MagicMock(return_value=True),
delete_file=AsyncMock(return_value=True),
)
try:
yield make_case
finally:
await engine.dispose()
async def _dispatch(ctx, *, upload_side_effect=None):
scheduler = PrintScheduler()
if upload_side_effect is not None:
ctx.upload.side_effect = upload_side_effect
patches = [
patch.object(scheduler_module.settings, "base_dir", ctx.base_dir),
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", ctx.start_print),
patch("backend.app.services.print_scheduler.printer_manager.set_awaiting_plate_clear", MagicMock()),
patch(
"backend.app.services.print_scheduler.get_ftp_retry_settings",
AsyncMock(return_value=(False, 0, 0, 1.0)),
),
patch("backend.app.services.print_scheduler.delete_file_async", ctx.delete_file),
patch("backend.app.services.print_scheduler.upload_file_async", ctx.upload),
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",
AsyncMock(),
),
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, "_propagate_owner_to_printer_manager", AsyncMock()),
patch.object(scheduler, "_power_off_if_needed", AsyncMock()),
patch.object(scheduler, "_preheat_and_soak", AsyncMock()),
]
with ExitStack() as stack:
for patcher in patches:
stack.enter_context(patcher)
async with ctx.session_maker() as db:
item = await db.get(PrintQueueItem, ctx.queue_item_id)
await scheduler._start_print(db, item)
async def _final_status(ctx):
async with ctx.session_maker() as db:
item = await db.get(PrintQueueItem, ctx.queue_item_id)
return item.status, item.started_at
@pytest.mark.asyncio
async def test_cancel_during_ftp_upload_aborts_before_mqtt(queue_factory):
"""User wins the race during the FTP upload — CAS must detect & bail.
This is the headline #1853 scenario: snapshot saw pending, FTP upload
starts, user clicks Cancel, /cancel commits ``cancelled`` to the row,
FTP finishes successfully, scheduler reaches the CAS. CAS rowcount must
be 0; ``printer_manager.start_print`` must NOT be called; row must stay
``cancelled``; uploaded file must be deleted from the printer's SD.
"""
ctx = await queue_factory()
async def cancel_mid_upload(*args, **kwargs):
# Simulate /cancel landing in a separate session while FTP is in
# flight. The endpoint commits status='cancelled' then returns 200.
async with ctx.session_maker() as other_db:
other_item = await other_db.get(PrintQueueItem, ctx.queue_item_id)
other_item.status = "cancelled"
await other_db.commit()
return True
await _dispatch(ctx, upload_side_effect=cancel_mid_upload)
status, started_at = await _final_status(ctx)
assert status == "cancelled", "CAS overwrote the user's cancellation"
assert started_at is None, "started_at must not be stamped on a cancelled row"
ctx.start_print.assert_not_called()
# Two delete calls — the pre-upload sweep and the post-CAS cleanup.
assert ctx.delete_file.await_count == 2
@pytest.mark.asyncio
async def test_cancel_before_ftp_upload_skips_dispatch(queue_factory):
"""Early-refresh path: row was cancelled before _start_print resumed.
Mirrors the case where ``/cancel`` lands between the ``check_queue``
snapshot and the time ``_start_print`` runs. The early ``db.refresh``
after the connectivity check sees ``cancelled`` and returns immediately
— no FTP upload, no MQTT send, row unchanged.
"""
ctx = await queue_factory()
# Flip to cancelled before _start_print runs; the in-memory snapshot
# the scheduler holds still reads 'pending', exactly the bug shape.
async with ctx.session_maker() as other_db:
item = await other_db.get(PrintQueueItem, ctx.queue_item_id)
item.status = "cancelled"
await other_db.commit()
await _dispatch(ctx)
status, started_at = await _final_status(ctx)
assert status == "cancelled"
assert started_at is None
ctx.upload.assert_not_awaited()
ctx.start_print.assert_not_called()
@pytest.mark.asyncio
async def test_happy_path_still_dispatches(queue_factory):
"""Sanity: no cancel, no race — pending row flips to printing, MQTT fires.
Regression guard so the CAS doesn't accidentally block normal dispatch
on a row that was always pending.
"""
ctx = await queue_factory()
await _dispatch(ctx)
status, started_at = await _final_status(ctx)
assert status == "printing"
assert started_at is not None
ctx.upload.assert_awaited_once()
ctx.start_print.assert_called_once()