mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-10-03 20:56:18 +02:00
maziggy
ce807fb1cc
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.
610 lines
27 KiB
Python
610 lines
27 KiB
Python
"""Regression tests for ``_watchdog_print_start``.
|
||
|
||
The watchdog reverts queue items to ``pending`` when a dispatched print never
|
||
lands on the printer (half-broken MQTT session — #887/#936/#967). H2D firmware
|
||
can sit at ``FINISH`` for 50+ seconds after accepting a ``project_file``
|
||
command before flipping ``gcode_state`` to ``PREPARE``, which used to trip the
|
||
state-only watchdog and cause the scheduler to revert the item; the subsequent
|
||
successful dispatch then looked like a reprint of the just-finished job (#1078).
|
||
|
||
The fix: treat ``subtask_id`` advancing past the pre-dispatch value as an
|
||
equivalent "command landed" signal, and raise the timeout from 45 s to 90 s as
|
||
belt-and-braces for slow transitions that also don't emit an early subtask_id
|
||
tick.
|
||
"""
|
||
|
||
from types import SimpleNamespace
|
||
from unittest.mock import AsyncMock, MagicMock, patch
|
||
|
||
import pytest
|
||
|
||
from backend.app.models.print_queue import PrintQueueItem
|
||
from backend.app.services.print_scheduler import DISPATCH_MAX_ATTEMPTS, PrintScheduler
|
||
|
||
|
||
@pytest.fixture
|
||
async def db_session():
|
||
"""In-memory SQLite with one ``printing`` queue item at id=1."""
|
||
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
||
|
||
import backend.app.models # noqa: F401 — populate Base.metadata
|
||
from backend.app.core.database import Base
|
||
|
||
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 with session_maker() as db:
|
||
db.add(PrintQueueItem(id=1, printer_id=42, archive_id=99, status="printing"))
|
||
await db.commit()
|
||
|
||
try:
|
||
yield session_maker
|
||
finally:
|
||
await engine.dispose()
|
||
|
||
|
||
def _status(state: str, subtask_id: str | None = None, gcode_file: str | None = None):
|
||
"""Minimal stand-in for PrinterState — only the fields the watchdog reads."""
|
||
return SimpleNamespace(state=state, subtask_id=subtask_id, gcode_file=gcode_file)
|
||
|
||
|
||
class TestWatchdogExitsEarlyOnPickup:
|
||
"""The watchdog must NOT revert when the printer has clearly picked up the job."""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_exits_on_state_change(self, db_session):
|
||
"""State transitioning away from pre_state is the primary "accepted" signal."""
|
||
get_status = MagicMock(return_value=_status("RUNNING", "OLD_SUBTASK"))
|
||
with (
|
||
patch("backend.app.services.print_scheduler.printer_manager.get_status", get_status),
|
||
patch("backend.app.services.print_scheduler.async_session", db_session),
|
||
):
|
||
await PrintScheduler._watchdog_print_start(
|
||
queue_item_id=1,
|
||
printer_id=42,
|
||
pre_state="FINISH",
|
||
pre_subtask_id="OLD_SUBTASK",
|
||
timeout=0.3,
|
||
poll_interval=0.05,
|
||
)
|
||
|
||
# Item should remain "printing" — watchdog recognised the pickup.
|
||
async with db_session() as db:
|
||
item = await db.get(PrintQueueItem, 1)
|
||
assert item.status == "printing"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_h2d_finish_to_running_via_subtask_id_then_active_state(self, db_session):
|
||
"""Regression for #1078 (preserved through the two-phase rewrite for #1678):
|
||
|
||
H2D keeps state=FINISH for ~50 s after accepting project_file, but
|
||
subtask_id flips to our new submission_id almost immediately. The
|
||
watchdog must NOT revert on the basis of state staying at FINISH —
|
||
Phase A exits on the subtask_id advance, Phase B then keeps watching
|
||
and exits SUCCESS as soon as the printer transitions to PREPARE /
|
||
RUNNING within the longer Phase B window.
|
||
"""
|
||
# First poll: state still FINISH, subtask_id advanced (Phase A → B).
|
||
# Second poll: state has flipped to RUNNING (Phase B success).
|
||
get_status = MagicMock(
|
||
side_effect=[
|
||
_status("FINISH", "NEW_SUBTASK_12345"),
|
||
_status("RUNNING", "NEW_SUBTASK_12345"),
|
||
]
|
||
+ [_status("RUNNING", "NEW_SUBTASK_12345")] * 10,
|
||
)
|
||
with (
|
||
patch("backend.app.services.print_scheduler.printer_manager.get_status", get_status),
|
||
patch("backend.app.services.print_scheduler.async_session", db_session),
|
||
):
|
||
await PrintScheduler._watchdog_print_start(
|
||
queue_item_id=1,
|
||
printer_id=42,
|
||
pre_state="FINISH",
|
||
pre_subtask_id="OLD_SUBTASK_99999",
|
||
timeout=0.3,
|
||
phase_b_timeout=0.3,
|
||
poll_interval=0.05,
|
||
)
|
||
|
||
async with db_session() as db:
|
||
item = await db.get(PrintQueueItem, 1)
|
||
assert item.status == "printing", (
|
||
"Phase A exit on subtask_id advance + Phase B observing the "
|
||
"active-state transition is the H2D success path — watchdog "
|
||
"must keep the item 'printing' (#1078)"
|
||
)
|
||
|
||
|
||
class TestWatchdogRevertsWhenStuck:
|
||
"""Genuine half-broken sessions still need the revert + reconnect recovery."""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_reverts_when_neither_state_nor_subtask_id_changes(self, db_session):
|
||
"""Both signals unchanged across the full timeout → revert to pending
|
||
and force MQTT reconnect (the #967 recovery path)."""
|
||
get_status = MagicMock(return_value=_status("FINISH", "OLD_SUBTASK"))
|
||
client = MagicMock()
|
||
get_client = MagicMock(return_value=client)
|
||
|
||
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),
|
||
):
|
||
await PrintScheduler._watchdog_print_start(
|
||
queue_item_id=1,
|
||
printer_id=42,
|
||
pre_state="FINISH",
|
||
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 == "pending"
|
||
assert item.started_at is None
|
||
|
||
client.force_reconnect_stale_session.assert_called_once()
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_reverts_on_finish_to_idle_user_dismissed_prompt(self, db_session):
|
||
"""Regression for #1370: when pre_state is FINISH and the printer
|
||
transitions to IDLE during the watchdog window, that's the user
|
||
dismissing a post-print prompt — NOT acceptance of our project_file.
|
||
|
||
The bundle in #1370 showed exactly this: queue item dispatched while
|
||
printer was in FINISH (residual from a previous print), command sent
|
||
but silently rejected by firmware, then the user manually cleared
|
||
the screen prompt so the printer moved to IDLE. The original
|
||
``state != pre_state`` check returned early on this transition and
|
||
the queue row was left stuck in 'printing' indefinitely, blocking
|
||
all future dispatches to that printer.
|
||
|
||
The watchdog now only treats transitions into the active-print
|
||
state set (PREPARE / SLICING / RUNNING / PAUSE) as a valid "command
|
||
landed" signal.
|
||
"""
|
||
get_status = MagicMock(return_value=_status("IDLE", "OLD_SUBTASK"))
|
||
client = MagicMock()
|
||
get_client = MagicMock(return_value=client)
|
||
|
||
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),
|
||
):
|
||
await PrintScheduler._watchdog_print_start(
|
||
queue_item_id=1,
|
||
printer_id=42,
|
||
pre_state="FINISH",
|
||
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 == "pending", (
|
||
"FINISH -> IDLE is the user dismissing a screen prompt, not "
|
||
"the printer accepting project_file — item must be reverted "
|
||
"to 'pending' so the scheduler can retry (#1370)"
|
||
)
|
||
assert item.started_at is None
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_does_not_revert_on_pickup_via_active_state(self, db_session):
|
||
"""Counterpart to the #1370 fix: transitions into the active-print
|
||
state set ARE a valid "command landed" signal. PREPARE / SLICING /
|
||
RUNNING / PAUSE all keep the item in 'printing'.
|
||
"""
|
||
for active_state in ("PREPARE", "SLICING", "RUNNING", "PAUSE"):
|
||
async with db_session() as db:
|
||
item = await db.get(PrintQueueItem, 1)
|
||
item.status = "printing"
|
||
item.started_at = None
|
||
await db.commit()
|
||
|
||
get_status = MagicMock(return_value=_status(active_state, "OLD_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", (
|
||
f"transition IDLE -> {active_state} must be treated as a "
|
||
f"valid 'command landed' signal — watchdog must not revert"
|
||
)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_default_timeout_is_90_seconds(self):
|
||
"""The default timeout must cover slow H2D FINISH→PREPARE transitions
|
||
(~50 s observed). A 45 s default would trip on the exact scenario the
|
||
subtask_id check is guarding against, leaving no fallback for printers
|
||
that don't echo subtask_id."""
|
||
import inspect
|
||
|
||
sig = inspect.signature(PrintScheduler._watchdog_print_start)
|
||
assert sig.parameters["timeout"].default == 90.0
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_default_phase_b_timeout_is_180_seconds(self):
|
||
"""Phase B (subtask_id advanced, waiting for active state) must
|
||
comfortably exceed the H2D FINISH→PREPARE delay (~50 s observed)
|
||
before declaring a printer-side wedge. 180 s gives ~3.5× headroom
|
||
and reverts the queue item in well under the previous 2-hour
|
||
expected_print TTL (#1678)."""
|
||
import inspect
|
||
|
||
sig = inspect.signature(PrintScheduler._watchdog_print_start)
|
||
assert sig.parameters["phase_b_timeout"].default == 180.0
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_reverts_when_subtask_advanced_but_state_never_active(self, db_session):
|
||
"""Regression for #1678: P1S on old firmware, power-cycled mid-print,
|
||
cloud+LAN re-auth dance in flight. Printer accepts project_file
|
||
(gcode_file updates, subtask_id advances to our submission id) but
|
||
never transitions from IDLE/FINISH to PREPARE/RUNNING. The pre-fix
|
||
watchdog returned SUCCESS as soon as subtask_id advanced and the
|
||
queue item stayed in 'printing' until container restart. Phase B now
|
||
keeps watching; if the active-state transition never arrives, the
|
||
item reverts to 'pending' so the user can retry without restarting.
|
||
"""
|
||
get_status = MagicMock(
|
||
return_value=_status("IDLE", "NEW_SUBTASK_12345", gcode_file="/new.3mf"),
|
||
)
|
||
client = MagicMock() # NOT None — must verify reconnect isn't called
|
||
get_client = MagicMock(return_value=client)
|
||
|
||
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),
|
||
):
|
||
await PrintScheduler._watchdog_print_start(
|
||
queue_item_id=1,
|
||
printer_id=42,
|
||
pre_state="IDLE",
|
||
pre_subtask_id="OLD_SUBTASK_99999",
|
||
pre_gcode_file="/old.3mf",
|
||
timeout=0.2,
|
||
phase_b_timeout=0.2,
|
||
poll_interval=0.05,
|
||
)
|
||
|
||
async with db_session() as db:
|
||
item = await db.get(PrintQueueItem, 1)
|
||
assert item.status == "pending", (
|
||
"subtask_id advanced (Phase A → B) but state never reached an "
|
||
"active value — printer-side wedge; the queue item must be "
|
||
"reverted to 'pending' (#1678)"
|
||
)
|
||
assert item.started_at is None
|
||
|
||
# File landed (subtask_id advance proves this), so a forced reconnect
|
||
# would trigger 0500_4003 mid-parse (#1150) — skip.
|
||
client.force_reconnect_stale_session.assert_not_called()
|
||
|
||
|
||
class TestWatchdogFallbackBehaviour:
|
||
"""Backwards-compat and defensive behaviour around missing data."""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_pre_subtask_id_none_falls_back_to_state_only(self, db_session):
|
||
"""When we never captured a pre-dispatch subtask_id (e.g. printer just
|
||
connected), the watchdog must still work on the state signal alone —
|
||
and still revert when state stays unchanged, so half-broken sessions
|
||
are still recovered."""
|
||
get_status = MagicMock(return_value=_status("FINISH", "SOMETHING"))
|
||
get_client = MagicMock(return_value=None)
|
||
|
||
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),
|
||
):
|
||
await PrintScheduler._watchdog_print_start(
|
||
queue_item_id=1,
|
||
printer_id=42,
|
||
pre_state="FINISH",
|
||
pre_subtask_id=None,
|
||
timeout=0.2,
|
||
poll_interval=0.05,
|
||
)
|
||
|
||
async with db_session() as db:
|
||
item = await db.get(PrintQueueItem, 1)
|
||
assert item.status == "pending"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_current_subtask_id_none_does_not_trigger_early_exit(self, db_session):
|
||
"""If the printer transiently reports subtask_id=None (e.g. during
|
||
reconnect), that must not be treated as "changed" — otherwise the
|
||
watchdog would exit early without a real pickup signal and leave the
|
||
item stuck in "printing" after a genuinely broken session."""
|
||
get_status = MagicMock(return_value=_status("FINISH", None))
|
||
get_client = MagicMock(return_value=None)
|
||
|
||
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),
|
||
):
|
||
await PrintScheduler._watchdog_print_start(
|
||
queue_item_id=1,
|
||
printer_id=42,
|
||
pre_state="FINISH",
|
||
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 == "pending"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_printer_disconnected_returns_without_reverting(self, db_session):
|
||
"""If the printer drops during the watchdog window, don't touch the DB —
|
||
the reconnect path will sort the queue state out."""
|
||
get_status = MagicMock(return_value=None)
|
||
|
||
with (
|
||
patch("backend.app.services.print_scheduler.printer_manager.get_status", get_status),
|
||
patch("backend.app.services.print_scheduler.async_session", db_session),
|
||
):
|
||
await PrintScheduler._watchdog_print_start(
|
||
queue_item_id=1,
|
||
printer_id=42,
|
||
pre_state="FINISH",
|
||
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"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_no_revert_if_item_already_completed(self, db_session):
|
||
"""If the print completed between watchdog arm-time and timeout (item is
|
||
no longer "printing"), the watchdog must not clobber whatever status it
|
||
ended up in — #967 race guard. Additionally it must NOT run the MQTT
|
||
session-recovery path (forced reconnect): when on_print_complete has
|
||
already moved the row, the print clearly landed on the printer and a
|
||
forced reconnect on a healthy session would break ongoing prints on
|
||
the same printer.
|
||
"""
|
||
# Move item on to "completed" before the watchdog fires.
|
||
async with db_session() as db:
|
||
item = await db.get(PrintQueueItem, 1)
|
||
item.status = "completed"
|
||
await db.commit()
|
||
|
||
get_status = MagicMock(return_value=_status("FINISH", "OLD_SUBTASK"))
|
||
client = MagicMock() # NOT None — must verify reconnect isn't called
|
||
get_client = MagicMock(return_value=client)
|
||
|
||
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),
|
||
):
|
||
await PrintScheduler._watchdog_print_start(
|
||
queue_item_id=1,
|
||
printer_id=42,
|
||
pre_state="FINISH",
|
||
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 == "completed" # untouched
|
||
|
||
client.force_reconnect_stale_session.assert_not_called()
|
||
|
||
|
||
class TestGcodeFileDiscriminator:
|
||
"""#1150 vs #887/#936: skip the forced reconnect when gcode_file changed
|
||
(project_file landed, slow parse — reconnecting causes 0500_4003).
|
||
Reconnect when gcode_file is unchanged (publish dropped — half-broken
|
||
session needs the original recovery)."""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_skips_reconnect_when_gcode_file_changed(self, db_session):
|
||
get_status = MagicMock(
|
||
return_value=_status("FINISH", "OLD_SUBTASK", gcode_file="/new.3mf"),
|
||
)
|
||
client = MagicMock()
|
||
get_client = MagicMock(return_value=client)
|
||
|
||
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),
|
||
):
|
||
await PrintScheduler._watchdog_print_start(
|
||
queue_item_id=1,
|
||
printer_id=42,
|
||
pre_state="FINISH",
|
||
pre_subtask_id="OLD_SUBTASK",
|
||
pre_gcode_file="/old.3mf",
|
||
timeout=0.2,
|
||
poll_interval=0.05,
|
||
)
|
||
|
||
# Item still reverts (the user-facing failure stays correct), but the
|
||
# MQTT session is left intact so the slow printer can finish parsing.
|
||
async with db_session() as db:
|
||
item = await db.get(PrintQueueItem, 1)
|
||
assert item.status == "pending"
|
||
client.force_reconnect_stale_session.assert_not_called()
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_reconnects_when_gcode_file_unchanged(self, db_session):
|
||
get_status = MagicMock(
|
||
return_value=_status("FINISH", "OLD_SUBTASK", gcode_file="/old.3mf"),
|
||
)
|
||
client = MagicMock()
|
||
get_client = MagicMock(return_value=client)
|
||
|
||
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),
|
||
):
|
||
await PrintScheduler._watchdog_print_start(
|
||
queue_item_id=1,
|
||
printer_id=42,
|
||
pre_state="FINISH",
|
||
pre_subtask_id="OLD_SUBTASK",
|
||
pre_gcode_file="/old.3mf",
|
||
timeout=0.2,
|
||
poll_interval=0.05,
|
||
)
|
||
|
||
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
|