fix(db): release scheduler/cloud/cover sessions across slow I/O, add LIFO pool (#2572)

Three more idle-in-transaction / thundering-herd paths from farm testing:

- print_scheduler: _start_print commits before the FTP delete/upload and
  _preheat_and_soak commits before the heat-soak wait, so the per-item
  session no longer sits idle-in-transaction across preheat + upload.
- cloud/filament-info: rollback the request transaction after the token
  read and before the sequential Bambu Cloud calls; single-flight
  concurrent misses for the same setting_id through one shared call.
- printers/cover: coalesce identical in-flight cover requests so followers
  serve from the cache the leader fills instead of duplicating the
  multi-path FTP + 3MF extraction.

Also adds pool_use_lifo (PostgreSQL default on, DB_POOL_USE_LIFO override,
shown in /system/db-pool) so a bursty farm keeps a small hot connection set.
This commit is contained in:
maziggy
2026-07-18 16:15:09 +02:00
parent 5afdaa83d1
commit 80687982c1
10 changed files with 559 additions and 60 deletions
+1
View File
@@ -8,6 +8,7 @@ All notable changes to Bambuddy will be documented in this file.
- **Orca Cloud profile sync now connects by approving a code instead of the copy-paste sign-in** — Connecting Bambuddy to Orca Cloud used to mean opening an OAuth sign-in in a new tab, watching it redirect to a `localhost` URL that fails to load, then copying that dead URL out of the address bar and pasting it back into Bambuddy. That dance existed only because Orca's auth backend (Supabase) accepts no redirect target other than `localhost`, and the deliberately-broken redirect page confused nearly everyone who reached it. OrcaSlicer has since shipped a first-class external-app pairing API (the OAuth 2.0 Device Authorization Grant, RFC 8628), so the flow is now: click **Connect**, approve a short code on your Orca Cloud settings page, and Bambuddy pairs itself — no redirect, no paste, no client secret, and it behaves identically from a LAN IP, `localhost`, or behind a reverse proxy. Bambuddy requests **read-only** access (it only lists and views your Orca Cloud profiles), keeps the pairing alive with the API's rotating refresh tokens (validated end-to-end against Orca's staging and production servers), and stores nothing beyond the issued token pair. The profile list and detail views are unchanged, so nothing downstream of the connect step looks different. The old paste-based sign-in and the email/password fallback are removed. Points at production Orca Cloud by default; `ORCA_CLOUD_API_BASE` overrides the endpoint for testing.
### Fixed
- **Three more idle-in-transaction / thundering-herd paths surfaced by continued farm testing (#2572, reporter @Jostxxl)** — On `origin/dev` with 93 printers and multiple concurrent UI clients the reporter timestamp-correlated the surviving pool pressure to three remaining paths, none of them auth-related. **(a) The scheduler held its per-item session across preheat and the FTP upload.** `_dispatch_selected` opens one `async_session` per queue item and hands it to `_start_print`, which reads the printer/archive rows up front and then runs the preheat/heat-soak wait and the FTP delete+upload — all on the transaction opened by those first `SELECT`s. One correlated session's last statement was a `settings` `SELECT` at the exact moment the log showed "Starting queue item" → preheat → "FTP upload started" for a 96 MB 3MF; the transaction stayed open for the whole transfer. (This refines the earlier note that the scheduler paths were "already bounded" — the per-item session itself was the hold.) `_start_print` now commits right before the FTP delete/upload, and `_preheat_and_soak` commits after its read phase and before the up-to-15-minute soak wait (both loops touch only `printer_manager` state and `asyncio.sleep`, no DB). `expire_on_commit=False` keeps the loaded rows readable; the status writes afterward (upload-failure path and the pending→printing CAS) transparently open a fresh transaction. **(b) `/cloud/filament-info` held its request session across sequential Bambu Cloud round-trips and single-flighted nothing.** The route took its session via `Depends(get_db)` (held for the whole request), read the stored token, then looped over the uncached ids issuing one external `get_setting_detail` HTTP call each — so the session sat idle-in-transaction across N cloud calls, and because the printer overview mounts one filament-info request per printer card, several browsers hit the same uncached preset at once and each issued its own cloud call. The route now releases the transaction (`rollback`) right after the token read and before the cloud loop (Phase 3's local-preset read reopens a fresh one), and concurrent misses for the same id single-flight through one shared cloud call. **(c) `/printers/{id}/cover` had no in-flight coalescing.** The connection was already released before the download, but simultaneous clients could all miss the cache and each run the full multi-path FTP lookup + 3MF extraction (one observed transfer pulled an 81 MB 3MF while real print uploads were in flight). Identical concurrent cover requests now coalesce: the first becomes the leader and the rest await it, then serve from the positive/negative cache it filled. Also adds `pool_use_lifo` (PostgreSQL default on, `DB_POOL_USE_LIFO` override, shown in `/system/db-pool`) so a bursty farm keeps a small hot connection set busy and lets excess overflow connections age out via `pool_recycle` instead of churning the whole pool. Covered by tests: the scheduler releasing its connection before both the FTP upload and the soak wait, the filament-info single-flight (concurrent misses share one cloud call, cache-hit skips cloud, a failed fetch leaves no stuck in-flight entry), and concurrent cover requests downloading once.
- **An "Any [model]" queue job dispatched from a Virtual Printer printed to the empty external spool and aborted at layer 0 (#2595, diagnosed by @Sawtaytoes, PR #2596)** — On a farm of identical X1Cs with different filaments loaded per AMS, the intended flow — VP in Queue mode, auto-dispatch, target **Any X1C**, force-colour-match picks the printer that has the right spool — sent the job to the correctly-matched printer and then failed: the printer ignored the AMS, pulled the empty external spool, and aborted with "not enough filament", even though the mapped slot was loaded (the same print via a specific printer, or straight from the slicer, worked). **Root cause.** A slicer talking to a Virtual Printer only ever sees the VP's external spool — a VP advertises no AMS — so the slicer sends `use_ams=false`, and VP intake stamps that onto the queue item. But an "Any [model]" item is colour-matched to a real printer *at dispatch*, resolving a real AMS slot in `ams_mapping`; the scheduler still forwarded the stale `use_ams=false`. The print-command builder only ever forced `use_ams` **off** (the all-external case) and never back **on**, so `use_ams=false` shipped alongside `ams_mapping=[<real tray>]` → external spool → abort. **Fix.** For single-nozzle printers the resolved mapping is now authoritative: a real AMS tray (0-253) forces `use_ams=true`; an explicit external selection (254/255) still forces it false; an unresolved `-1` mapping does neither (preserving the #2589 contract — it should have been recomputed upstream, and must not be silently promoted to AMS or downgraded to external). Dual-nozzle printers are untouched, where `use_ams` encodes nozzle routing rather than an on/off flag. Because the correction lives at the single command-builder choke point, it fixes the VP, queue, and manual paths alike. Covered by tests for the VP `false`+real-tray promotion, padded mappings, all-external staying off, unresolved `-1` staying put, the original all-external downgrade, and the dual-nozzle bypass.
- **Reconnecting or restarting inflated Stats → Total Print Time by hundreds of hours on large farms (#2592, reporter @Jostxxl)** — On the reporter's farm a restart pushed Total Print Time from ~1,500h to 3,215h. When a printer reconnects, the connected edge runs `reconcile_stale_active_prints`, which closes out every archive still stuck in `status="printing"` (missed completions, disconnects, restarts) by synthesising an aborted `on_print_complete`. That wrote a `PrintLogEntry` whose duration was `completed_at - started_at` — but for a reconciled archive the real end time is unknown: the print stopped somewhere during the disconnect, and `completed_at` is only the reconnect moment. So each stale archive banked its entire multi-day gap as print time (one row was 51.9h), and a printer with several stale archives contributed hundreds of fabricated hours at once. Worse, the Stats total *recomputed* `completed_at - started_at` whenever the stored duration was falsy, so storing NULL wouldn't have helped. Reconciled completions now log an explicit `duration_seconds = 0` (honest "no measured runtime") and the two Stats time paths trust a stored 0 instead of recomputing from the stale timestamps — legacy rows that never recorded a duration still fall back as before. Reconciled aborts also get a truthful `failure_reason` ("Stale - reconciled after reconnect, end time unknown") instead of being mislabelled "User cancelled". Genuine long prints are untouched: nothing is capped, a still-running >24h print is never treated as stale, and a real >24h run keeps its full measured duration. Re-running reconciliation is already idempotent (the archive flips to `aborted`, so it isn't re-selected). Existing inflated rows from before this fix are not auto-corrected — they're indistinguishable from real cancellations in the database, and a blanket cap would clobber genuine long prints; the reporter repaired his own rows by hand. Covered by tests for the multi-day reconcile, multiple stale archives per printer, a retained >24h print, and the Stats total ignoring reconciled time while still counting real runtime.
- **H2C prints intermittently recorded no filament and never deducted from inventory (#2582, reporter @gyrene2083)** — On an H2C (firmware `01.02.00.00`) filament usage sometimes wasn't deducted and the Print Log showed no filament for that print; the reporter confirmed the tell-tale detail — the failed print's archived `.3mf` didn't exist to download. Filament totals, the Print Log filament column, and the weight deduction all read the sliced 3MF's data, so when that file can't be pulled off the printer the print drops to the no-3MF fallback archive and every one of them comes up empty. The download itself was the failure: the H2C is the same H2 generation and the same firmware line as the P2S, whose FTPS data channel trips a vsFTPd + TLS 1.3 session-reuse bug on Python 3.13 (#1401) — and the X2D hit the sibling handshake variant (#1638). Both were fixed by capping that model's FTP control/data channel to TLS 1.2 via the per-model FTP profile registry, but the H2C had no entry and so ran on the Python-default TLS 1.3, leaving its 3MF downloads to fail the same way (intermittently, matching the "sometimes works, sometimes doesn't" report — the session-reuse race rather than a hard handshake failure). The H2C now gets the same `cap_tls_v1_2` profile as the P2S/X2D (with its `O1C`/`O1C2` SSDP codes mapped to it), so the sliced 3MF comes off the printer reliably and the slice data — filament total, Print Log filament, and the inventory deduction — is populated again. H2D is deliberately left on the default profile; it negotiates TLS 1.3 without this fault.
+98 -38
View File
@@ -4,6 +4,7 @@ Bambu Lab Cloud API Routes
Handles authentication and profile management with Bambu Cloud.
"""
import asyncio
import json
import logging
from datetime import datetime, timezone
@@ -737,6 +738,92 @@ _filament_cache: dict[str, dict] = {}
_filament_cache_time: float = 0
FILAMENT_CACHE_TTL = 300 # 5 minutes
# In-flight cloud lookups, keyed by setting_id (#2572). The printer overview
# mounts one filament-info request per printer card, so at farm scale several
# browsers ask for the same uncached preset within the same instant. Without
# coalescing each request issues its own Bambu Cloud round-trip for the same id
# (a thundering herd against a rate-limited API). The first caller to miss a
# given id becomes the leader and resolves it; concurrent callers await its
# future and reuse the result instead of duplicating the call.
_filament_inflight: dict[str, asyncio.Future] = {}
async def _fetch_one_cloud_filament(setting_id: str, cloud: BambuCloudService) -> dict | None:
"""Fetch a single filament preset from Bambu Cloud.
Returns ``{"name", "k"}`` on success (name may be empty when the preset
resolves but carries no display name), or ``None`` when the lookup fails.
Never raises — a 400 is the expected answer for many bare preset IDs and is
logged at DEBUG; anything else is a real fault logged at WARNING.
"""
try:
api_setting_id = _filament_id_to_setting_id(setting_id)
data = await cloud.get_setting_detail(api_setting_id)
setting = data.get("setting", {})
name = data.get("name", "")
k_value = setting.get("pressure_advance")
if k_value is not None:
try:
k_value = float(k_value)
except (ValueError, TypeError):
k_value = None
return {"name": name, "k": k_value}
except Exception as e:
# A 400 here is the *expected* answer, not a fault, and the local-preset
# fallback (Phase 3) exists to handle it (#2530). Two routine causes:
# * Many official presets are only addressable with a printer variant
# suffix — "GFSA00" resolves, "GFSL05" does not, only "GFSL05_07"
# (@BBL A1) does. The bare ID is all the AMS reports, so the lookup
# legitimately misses.
# * Personal presets ("P…") belong to the Bambu account that sliced the
# file; another account will never resolve them.
# Logging those at WARNING on every AMS tooltip refresh trains users to
# ignore the log. Anything else — expired token, 5xx, a connection
# failure — stays at WARNING because it is a fault.
expected_miss = isinstance(e, BambuCloudError) and e.status_code == 400
logger.log(
logging.DEBUG if expected_miss else logging.WARNING,
"Failed to get cloud preset %s (API ID: %s): %s",
setting_id,
_filament_id_to_setting_id(setting_id),
e,
)
return None
async def _resolve_cloud_filament(setting_id: str, cloud: BambuCloudService) -> dict | None:
"""Resolve one preset via Bambu Cloud, single-flighting concurrent misses (#2572).
Concurrent callers for the same ``setting_id`` share one cloud round-trip:
the first caller resolves it while the rest await the shared future. Returns
the info dict (also populating ``_filament_cache``) or ``None`` on failure.
"""
if setting_id in _filament_cache:
return _filament_cache[setting_id]
existing = _filament_inflight.get(setting_id)
if existing is not None:
# Another request is already fetching this id — reuse its result.
# shield() so our own cancellation can't cancel the shared leader.
try:
return await asyncio.shield(existing)
except Exception:
return None
fut: asyncio.Future = asyncio.get_event_loop().create_future()
_filament_inflight[setting_id] = fut
info: dict | None = None
try:
info = await _fetch_one_cloud_filament(setting_id, cloud)
return info
finally:
if info is not None:
_filament_cache[setting_id] = info
if not fut.done():
fut.set_result(info)
_filament_inflight.pop(setting_id, None)
# Built-in filament ID → name mapping (fallback when cloud API and local profiles
# don't have the entry). Based on Bambu Lab's known filament catalogue.
_BUILTIN_FILAMENT_NAMES: dict[str, str] = {
@@ -949,50 +1036,23 @@ async def get_filament_info(
# Phase 2: Try cloud for uncached IDs
if unresolved_ids:
cloud = await build_authenticated_cloud(db, current_user)
# Release the request's DB transaction before the sequential Bambu Cloud
# round-trips below (#2572). build_authenticated_cloud has read the
# stored token — the only DB access this phase needs — and nothing until
# Phase 3 touches the DB again. Without this the session sat "idle in
# transaction" for the full duration of N external HTTP calls, pinning a
# pooled connection per in-flight request. Phase 3's read transparently
# opens a fresh transaction on the same still-open session.
await db.rollback()
if cloud is not None and cloud.is_authenticated:
try:
still_unresolved: list[str] = []
for setting_id in unresolved_ids:
try:
api_setting_id = _filament_id_to_setting_id(setting_id)
data = await cloud.get_setting_detail(api_setting_id)
setting = data.get("setting", {})
name = data.get("name", "")
k_value = setting.get("pressure_advance")
if k_value is not None:
try:
k_value = float(k_value)
except (ValueError, TypeError):
k_value = None
info = {"name": name, "k": k_value}
_filament_cache[setting_id] = info
info = await _resolve_cloud_filament(setting_id, cloud)
if info is not None:
result[setting_id] = info
if not name:
still_unresolved.append(setting_id)
except Exception as e:
# A 400 here is the *expected* answer, not a fault, and Phase 3
# exists to handle it (#2530). Two routine causes:
# * Many official presets are only addressable with a printer
# variant suffix — "GFSA00" resolves, "GFSL05" does not, only
# "GFSL05_07" (@BBL A1) does. The bare ID is all the AMS
# reports, so the lookup legitimately misses.
# * Personal presets ("P…") belong to the Bambu account that
# sliced the file; another account will never resolve them.
# Logging those at WARNING on every AMS tooltip refresh trains
# users to ignore the log. Anything else — expired token, 5xx,
# a connection failure — stays at WARNING because it is a fault.
expected_miss = isinstance(e, BambuCloudError) and e.status_code == 400
logger.log(
logging.DEBUG if expected_miss else logging.WARNING,
"Failed to get cloud preset %s (API ID: %s): %s",
setting_id,
_filament_id_to_setting_id(setting_id),
e,
)
if info is None or not info.get("name"):
still_unresolved.append(setting_id)
unresolved_ids = still_unresolved
finally:
await cloud.close()
+58 -2
View File
@@ -963,6 +963,15 @@ _cover_cache: dict[int, dict[tuple[str, str], bytes]] = {}
# Cleared on print start alongside _cover_cache.
_cover_404_cache: dict[int, set[tuple[str, str]]] = {}
# In-flight cover downloads, keyed by (printer_id, subtask_name, view_key) (#2572).
# The farm dashboard mounts a cover tile per printer card, so several browsers
# request the same printer's cover in the same instant, all miss the cache, and
# each runs the full multi-path FTP lookup + 3MF extraction (one observed live
# transfer pulled an 81 MB 3MF while real print uploads were in flight). The
# first request to miss becomes the leader; concurrent requests await its future
# and then serve from the positive/negative cache it filled.
_cover_inflight: dict[tuple[int, str, str], asyncio.Future] = {}
def clear_cover_cache(printer_id: int) -> None:
"""Clear cached cover images for a printer. Call on print start to avoid stale thumbnails."""
@@ -1039,6 +1048,53 @@ async def get_printer_cover(
if printer_id in _cover_404_cache and cache_key in _cover_404_cache[printer_id]:
raise HTTPException(404, f"No cover available for '{subtask_name}' (cached)")
# Coalesce concurrent downloads for the same cover (#2572). The positive and
# negative caches were just checked above; if another request is already
# downloading this exact cover, wait for it and serve from the cache it fills
# instead of launching a duplicate multi-path FTP + 3MF extraction.
inflight_key = (printer_id, subtask_name, view_key)
leader = _cover_inflight.get(inflight_key)
if leader is not None:
# shield() so our own cancellation can't cancel the shared leader.
try:
await asyncio.shield(leader)
except Exception:
pass
if printer_id in _cover_cache and cache_key in _cover_cache[printer_id]:
return Response(content=_cover_cache[printer_id][cache_key], media_type="image/png")
if printer_id in _cover_404_cache and cache_key in _cover_404_cache[printer_id]:
raise HTTPException(404, f"No cover available for '{subtask_name}' (cached)")
# Leader finished without filling either cache (a transient 503) — fall
# through and try the download ourselves.
fut: asyncio.Future = asyncio.get_event_loop().create_future()
_cover_inflight[inflight_key] = fut
try:
image_data = await _produce_cover_image(printer, printer_id, subtask_name, view, view_key, plate_num, cache_key)
return Response(content=image_data, media_type="image/png")
finally:
if not fut.done():
fut.set_result(None)
_cover_inflight.pop(inflight_key, None)
async def _produce_cover_image(
printer: Printer,
printer_id: int,
subtask_name: str,
view: str | None,
view_key: str,
plate_num: int | None,
cache_key: tuple[str, str],
) -> bytes:
"""Download the active-print 3MF and extract its cover thumbnail (#2572).
Split out of ``get_printer_cover`` so concurrent requests for the same cover
can single-flight through it (see ``_cover_inflight``). Returns the PNG bytes
on success (also filling ``_cover_cache``) and raises ``HTTPException`` on
failure (filling ``_cover_404_cache`` for the definitive 404s). Does no DB
work — the caller already released the pooled connection before this runs.
"""
# Build possible 3MF filenames from subtask_name
# Bambu printers may store files as "name.gcode.3mf" (sliced via Bambu Studio)
# or just "name.3mf" (uploaded directly)
@@ -1202,7 +1258,7 @@ async def get_printer_cover(
if printer_id not in _cover_cache:
_cover_cache[printer_id] = {}
_cover_cache[printer_id][(subtask_name, view_key)] = image_data
return Response(content=image_data, media_type="image/png")
return image_data
except KeyError:
continue
@@ -1213,7 +1269,7 @@ async def get_printer_cover(
if printer_id not in _cover_cache:
_cover_cache[printer_id] = {}
_cover_cache[printer_id][(subtask_name, view_key)] = image_data
return Response(content=image_data, media_type="image/png")
return image_data
_cover_404_cache.setdefault(printer_id, set()).add(cache_key)
raise HTTPException(404, "No thumbnail found in 3MF file")
+5
View File
@@ -84,6 +84,11 @@ class Settings(BaseSettings):
db_max_overflow: int | None = Field(default=None, ge=0)
db_pool_timeout: int | None = Field(default=None, gt=0)
db_pool_recycle: int | None = Field(default=None, gt=0)
# LIFO checkout (PostgreSQL default on): reuse the most-recently-returned
# connection so a bursty farm keeps a small hot set busy and lets the excess
# overflow connections age out via pool_recycle instead of churning the whole
# pool. Override with DB_POOL_USE_LIFO. No effect on SQLite. (#2572)
db_pool_use_lifo: bool | None = Field(default=None)
# Logging
log_level: str = "INFO" # Override with LOG_LEVEL env var or DEBUG=true
+4
View File
@@ -51,6 +51,9 @@ def _resolve_pool_kwargs() -> dict:
"max_overflow": max_overflow,
"pool_pre_ping": True,
"pool_recycle": settings.db_pool_recycle if settings.db_pool_recycle is not None else 1800,
# LIFO checkout keeps a bursty farm on a small hot connection set and
# lets overflow connections recycle out during quiet spells (#2572).
"pool_use_lifo": settings.db_pool_use_lifo if settings.db_pool_use_lifo is not None else True,
}
if settings.db_pool_timeout is not None:
kwargs["pool_timeout"] = settings.db_pool_timeout
@@ -69,6 +72,7 @@ def _create_engine():
"pool_timeout": kwargs.get("pool_timeout", 30),
"pool_recycle": kwargs.get("pool_recycle", -1),
"pool_pre_ping": kwargs.get("pool_pre_ping", False),
"pool_use_lifo": kwargs.get("pool_use_lifo", False),
}
eng = create_async_engine(
+22
View File
@@ -2509,6 +2509,15 @@ class PrintScheduler:
except Exception as exc:
logger.warning("Queue item %s: preheat chamber M141 failed: %s", item.id, exc)
# Release the pooled DB connection before the (potentially many-minute)
# heat-soak wait below (#2572). Every setting this method needs is read
# above; the wait/soak loop only polls printer_manager state and sleeps —
# it never touches the DB. Without this the caller's transaction sat
# "idle in transaction" for the whole soak, pinning one pooled connection
# per preheating printer. expire_on_commit=False keeps item/printer
# readable afterwards; there are no pending writes to lose here.
await db.commit()
# Wait for convergence. Bed warm-up is fast (~5 min from cold); chamber
# via M141 takes a few minutes; chamber via bed radiation can take 20+.
# Poll every 3s — frequent enough for responsive logging without
@@ -3071,6 +3080,19 @@ class PrintScheduler:
f"retry_enabled={ftp_retry_enabled}, retry_count={ftp_retry_count}, timeout={ftp_timeout}"
)
# Release the pooled DB connection before the FTP delete/upload (#2572).
# Every read this method needs (printer, archive/library, preheat) is
# done, and the library-file branch already committed its archive
# creation. Without this the transaction opened by the first SELECT above
# stays "idle in transaction" for the entire upload — multiple seconds
# for a large 3MF — pinning one pooled connection per in-flight dispatch;
# a farm dispatching many jobs at once then exhausts the pool. This was
# correlated to an exact idle-in-transaction session on a 93-printer farm
# (reporter @Jostxxl). expire_on_commit=False keeps item/printer/archive
# readable; the status writes below (upload-failure path and the
# pending->printing CAS) transparently open a fresh transaction.
await db.commit()
# Delete existing file if present (avoids 553 error on overwrite)
try:
logger.debug("Queue item %s: Deleting existing file %s if present...", item.id, remote_path)
@@ -0,0 +1,94 @@
"""Concurrent cover requests for the same print coalesce into one download (#2572).
The farm dashboard mounts a cover tile per printer card, so several browsers
request the same printer's cover in the same instant. Before this fix each miss
ran the full multi-path FTP lookup + 3MF extraction independently (one observed
live transfer pulled an 81 MB 3MF while real uploads were in flight). Now the
first miss becomes the leader and the rest await its result, then serve from the
cache it filled.
"""
import asyncio
from types import SimpleNamespace
from unittest.mock import MagicMock, patch
import pytest
import backend.app.api.routes.printers as printers_mod
from backend.app.api.routes.printers import get_printer_cover
class _FakeSession:
def __init__(self, printer):
self._printer = printer
async def __aenter__(self):
return self
async def __aexit__(self, *exc):
return False
async def execute(self, *args, **kwargs):
return SimpleNamespace(scalar_one_or_none=lambda: self._printer)
@pytest.fixture(autouse=True)
def _clear_cover_state():
printers_mod._cover_cache.clear()
printers_mod._cover_404_cache.clear()
printers_mod._cover_inflight.clear()
yield
printers_mod._cover_cache.clear()
printers_mod._cover_404_cache.clear()
printers_mod._cover_inflight.clear()
@pytest.mark.asyncio
async def test_concurrent_cover_requests_download_once():
printer = SimpleNamespace(id=1, ip_address="127.0.0.1", access_code="x", model="X1C", name="P")
state = SimpleNamespace(subtask_name="job", state="RUNNING")
produce_calls = {"n": 0}
async def slow_produce(printer_row, printer_id, subtask_name, view, view_key, plate_num, cache_key):
produce_calls["n"] += 1
await asyncio.sleep(0.1) # hold leadership long enough for followers to attach
printers_mod._cover_cache.setdefault(printer_id, {})[cache_key] = b"PNGDATA"
return b"PNGDATA"
with (
patch("backend.app.core.database.async_session", lambda: _FakeSession(printer)),
patch.object(printers_mod.printer_manager, "get_status", MagicMock(return_value=state)),
patch.object(printers_mod, "resolve_plate_id", MagicMock(return_value=1)),
patch.object(printers_mod, "_produce_cover_image", slow_produce),
):
responses = await asyncio.gather(*[get_printer_cover(1, None, None) for _ in range(5)])
assert produce_calls["n"] == 1, "concurrent cover requests each ran their own FTP download"
assert {bytes(r.body) for r in responses} == {b"PNGDATA"}
@pytest.mark.asyncio
async def test_second_request_serves_from_positive_cache():
"""A follower arriving after the leader filled the cache serves it directly."""
printer = SimpleNamespace(id=1, ip_address="127.0.0.1", access_code="x", model="X1C", name="P")
state = SimpleNamespace(subtask_name="job", state="RUNNING")
produce_calls = {"n": 0}
async def produce(printer_row, printer_id, subtask_name, view, view_key, plate_num, cache_key):
produce_calls["n"] += 1
printers_mod._cover_cache.setdefault(printer_id, {})[cache_key] = b"PNGDATA"
return b"PNGDATA"
with (
patch("backend.app.core.database.async_session", lambda: _FakeSession(printer)),
patch.object(printers_mod.printer_manager, "get_status", MagicMock(return_value=state)),
patch.object(printers_mod, "resolve_plate_id", MagicMock(return_value=1)),
patch.object(printers_mod, "_produce_cover_image", produce),
):
first = await get_printer_cover(1, None, None)
second = await get_printer_cover(1, None, None)
assert produce_calls["n"] == 1 # second hit the positive cache
assert bytes(first.body) == bytes(second.body) == b"PNGDATA"
@@ -0,0 +1,87 @@
"""Filament-info cloud lookups single-flight concurrent misses (#2572).
The printer overview mounts one ``/cloud/filament-info`` request per printer
card, so at farm scale several browsers ask for the same uncached preset in the
same instant. Without coalescing each request issued its own Bambu Cloud
round-trip for the same id — a thundering herd against a rate-limited API.
``_resolve_cloud_filament`` makes the first caller the leader and has the rest
await its result.
"""
import asyncio
import pytest
import backend.app.api.routes.cloud as cloud_mod
from backend.app.api.routes.cloud import _resolve_cloud_filament
class _FakeCloud:
"""Counts cloud calls; blocks in get_setting_detail until released."""
def __init__(self, gate: asyncio.Event, *, fail: bool = False):
self.calls = 0
self._gate = gate
self._fail = fail
async def get_setting_detail(self, api_setting_id):
self.calls += 1
await self._gate.wait()
if self._fail:
raise RuntimeError("cloud down")
return {"name": "PLA Matte", "setting": {"pressure_advance": "0.021"}}
@pytest.fixture(autouse=True)
def _clear_state():
cloud_mod._filament_cache.clear()
cloud_mod._filament_inflight.clear()
yield
cloud_mod._filament_cache.clear()
cloud_mod._filament_inflight.clear()
@pytest.mark.asyncio
async def test_concurrent_misses_share_one_cloud_call():
gate = asyncio.Event()
cloud = _FakeCloud(gate)
leader = asyncio.create_task(_resolve_cloud_filament("GFSA00", cloud))
await asyncio.sleep(0) # let the leader register its in-flight future
follower = asyncio.create_task(_resolve_cloud_filament("GFSA00", cloud))
await asyncio.sleep(0) # let the follower attach to the leader's future
gate.set()
leader_info, follower_info = await asyncio.gather(leader, follower)
assert cloud.calls == 1, "the two concurrent misses did not coalesce"
assert leader_info == {"name": "PLA Matte", "k": 0.021}
assert follower_info == leader_info
assert cloud_mod._filament_cache["GFSA00"] == leader_info
assert "GFSA00" not in cloud_mod._filament_inflight # cleaned up
@pytest.mark.asyncio
async def test_cache_hit_skips_cloud_entirely():
gate = asyncio.Event()
gate.set()
cloud = _FakeCloud(gate)
cloud_mod._filament_cache["GFSA00"] = {"name": "cached", "k": None}
info = await _resolve_cloud_filament("GFSA00", cloud)
assert info == {"name": "cached", "k": None}
assert cloud.calls == 0
@pytest.mark.asyncio
async def test_failed_fetch_returns_none_and_leaves_no_inflight():
gate = asyncio.Event()
gate.set()
cloud = _FakeCloud(gate, fail=True)
info = await _resolve_cloud_filament("GFSA00", cloud)
assert info is None
assert "GFSA00" not in cloud_mod._filament_cache
assert "GFSA00" not in cloud_mod._filament_inflight
+20 -20
View File
@@ -88,7 +88,7 @@ def _ints(**values):
@pytest.mark.asyncio
async def test_global_disabled_inherit_skips(scheduler, item, archive):
"""preheat_enabled=False + item.preheat_override='inherit' → no heater dispatch."""
db = MagicMock()
db = AsyncMock()
client = _make_client()
with (
@@ -106,7 +106,7 @@ async def test_global_disabled_inherit_skips(scheduler, item, archive):
@pytest.mark.asyncio
async def test_item_override_off_bypasses_global_on(scheduler, item, archive):
"""preheat_enabled=True + item.preheat_override='off' → preheat suppressed."""
db = MagicMock()
db = AsyncMock()
client = _make_client()
item.preheat_override = "off"
@@ -126,7 +126,7 @@ async def test_item_override_off_bypasses_global_on(scheduler, item, archive):
async def test_item_override_on_runs_despite_global_off(scheduler, item, archive):
"""preheat_enabled=False + item.preheat_override='on' → preheat runs (bed
fires, chamber depends on the resolved target)."""
db = MagicMock()
db = AsyncMock()
client = _make_client()
item.preheat_override = "on"
item.preheat_chamber_target_override = 0 # explicit no-chamber so the assertion is sharp
@@ -154,7 +154,7 @@ async def test_item_override_on_runs_despite_global_off(scheduler, item, archive
async def test_chamber_target_override_beats_filament_map(scheduler, item, archive):
"""item.preheat_chamber_target_override is the highest-priority source —
a PLA-only print with an explicit 50°C override still heats the chamber."""
db = MagicMock()
db = AsyncMock()
client = _make_client()
item.preheat_chamber_target_override = 50
@@ -178,7 +178,7 @@ async def test_filament_map_picks_max_across_loaded_slots(scheduler, item, archi
"""Mixed PA + PLA load: PA=50 + PLA=0 → chamber target 50 (the max).
The "lowest common denominator" model is wrong here; PA's requirement
is the binding constraint."""
db = MagicMock()
db = AsyncMock()
client = _make_client()
with (
@@ -203,7 +203,7 @@ async def test_filament_map_picks_max_across_loaded_slots(scheduler, item, archi
async def test_pla_only_derives_zero_chamber_skips(scheduler, item, archive):
"""PLA-only print: filament-map lookup returns 0 → chamber phase skips
automatically without the user touching anything."""
db = MagicMock()
db = AsyncMock()
client = _make_client()
with (
@@ -226,7 +226,7 @@ async def test_unknown_filament_type_falls_to_default(scheduler, item, archive):
"""A loaded tray with a type not in the map uses the `default` entry —
keeps users with custom filament names safe (they get 0 by default,
can be tuned via the per-filament editor)."""
db = MagicMock()
db = AsyncMock()
client = _make_client()
with (
@@ -248,7 +248,7 @@ async def test_unknown_filament_type_falls_to_default(scheduler, item, archive):
async def test_custom_filament_targets_json_parses(scheduler, item, archive):
"""User-customised filament-target JSON overrides the bundled defaults —
raising PLA to 30°C makes a PLA-only print actually heat the chamber."""
db = MagicMock()
db = AsyncMock()
client = _make_client()
custom_map = '{"PLA": 30, "default": 0}'
@@ -270,7 +270,7 @@ async def test_custom_filament_targets_json_parses(scheduler, item, archive):
async def test_malformed_filament_targets_falls_back_to_defaults(scheduler, item, archive):
"""A corrupted JSON in the setting must not break the scheduler — log
and use bundled defaults."""
db = MagicMock()
db = AsyncMock()
client = _make_client()
with (
@@ -298,7 +298,7 @@ async def test_malformed_filament_targets_falls_back_to_defaults(scheduler, item
async def test_no_bed_temperature_in_archive_skips(scheduler, item):
"""Archive without bed_temperature metadata skips entirely rather than
guessing a default that might wreck a non-PLA print."""
db = MagicMock()
db = AsyncMock()
client = _make_client()
bare_archive = SimpleNamespace(bed_temperature=None)
@@ -320,7 +320,7 @@ async def test_no_bed_temperature_in_archive_skips(scheduler, item):
async def test_x1c_skips_m141_but_waits_passively(scheduler, item, archive):
"""X1C has a chamber sensor but no active heater — M141 must NOT fire even
when the filament map derives a non-zero target."""
db = MagicMock()
db = AsyncMock()
client = _make_client()
with (
@@ -343,7 +343,7 @@ async def test_x1c_skips_m141_but_waits_passively(scheduler, item, archive):
async def test_p1s_no_chamber_sensor_uses_soak_timer_only(scheduler, item, archive):
"""P1S has no chamber sensor — derived target is ignored for the wait
loop, only the soak timer applies."""
db = MagicMock()
db = AsyncMock()
client = _make_client()
with (
@@ -365,7 +365,7 @@ async def test_p1s_no_chamber_sensor_uses_soak_timer_only(scheduler, item, archi
@pytest.mark.asyncio
async def test_lost_client_skips_silently(scheduler, item, archive):
"""If the MQTT client drops, the helper returns without raising."""
db = MagicMock()
db = AsyncMock()
with (
patch.object(scheduler, "_get_bool_setting", AsyncMock(return_value=True)),
@@ -385,7 +385,7 @@ async def test_h2d_flips_airduct_to_heating_before_m141(scheduler, item, archive
chamber fan actively extracts the heat we're trying to put in and the
chamber never converges. Verify airduct=heating fires AND lands before
the chamber-target call so the heater starts in the right airflow regime."""
db = MagicMock()
db = AsyncMock()
client = _make_client()
call_order = []
client.set_airduct_mode.side_effect = lambda mode: call_order.append(("airduct", mode)) or True
@@ -417,7 +417,7 @@ async def test_x1c_skips_airduct_no_heater_no_call(scheduler, item, archive):
AND has_heater, so X1C gets neither call regardless. Important: a
spurious set_airduct on X1C wouldn't just be wasted MQTT — there's no
flap to set, so the firmware response would be undefined behaviour."""
db = MagicMock()
db = AsyncMock()
client = _make_client()
with (
@@ -452,7 +452,7 @@ async def test_get_preheat_filament_targets_defaults_when_missing(scheduler):
"""Empty / null setting → bundled defaults are used. _get_preheat_filament_targets
upper-cases the keys, so the bundled `default` becomes `DEFAULT` on the
returned dict — keep both spellings synced."""
db = MagicMock()
db = AsyncMock()
with patch.object(scheduler, "_get_setting", AsyncMock(return_value=None)):
targets = await scheduler._get_preheat_filament_targets(db)
# The bundled defaults dict is kept as-is on the "no setting" path, so
@@ -475,7 +475,7 @@ async def test_h2d_chamber_heat_switches_airduct_to_heating(scheduler, item, arc
with chamber_target > 0 must switch the airduct to heating BEFORE the
M141 dispatch — otherwise the open exhaust flap actively fights the
chamber heater and the chamber never converges."""
db = MagicMock()
db = AsyncMock()
client = _make_client()
with (
@@ -500,7 +500,7 @@ async def test_h2d_chamber_zero_switches_airduct_to_cooling(scheduler, item, arc
previously left in heating mode (from a prior ABS run) must switch
back to cooling. Otherwise PLA prints inherit ABS's closed-flap recirc
and run hot."""
db = MagicMock()
db = AsyncMock()
client = _make_client()
with (
@@ -524,7 +524,7 @@ async def test_h2d_airduct_already_correct_idempotent(scheduler, item, archive):
"""If the airduct is already in the desired mode, don't re-send
`set_airduct` — the firmware accepts it but it generates needless MQTT
chatter and could thrash the flap motor on rapid repeats."""
db = MagicMock()
db = AsyncMock()
client = _make_client()
with (
@@ -551,7 +551,7 @@ async def test_x1c_no_airduct_flap_never_fires_set_airduct(scheduler, item, arch
no-op. Regression guard: wiring this to `supports_chamber_temp` or
`supports_chamber_heater` alone would have leaked the command to
X1C/X1E or P2S inappropriately."""
db = MagicMock()
db = AsyncMock()
client = _make_client()
with (
@@ -0,0 +1,170 @@
"""The scheduler must release its pooled DB connection before long printer I/O (#2572).
``_dispatch_selected`` opens one ``async_session`` per queue item and hands it to
``_start_print``, which reads the printer/archive rows up front and then runs the
preheat/heat-soak wait and the FTP delete+upload. Before this fix the transaction
opened by those first reads stayed "idle in transaction" for the whole soak and
the entire multi-second 3MF upload, pinning one pooled connection per in-flight
dispatch. On a large farm dispatching many jobs at once that exhausted the pool —
a reporter (@Jostxxl) correlated one surviving idle-in-transaction session to
exactly this path by timestamp.
Two release points are verified:
* ``_start_print`` commits before the FTP delete/upload block.
* ``_preheat_and_soak`` commits after its read phase, before the (up-to-15-minute)
soak wait.
Both rely on ``expire_on_commit=False`` keeping the ORM rows readable afterwards.
"""
from contextlib import ExitStack
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from sqlalchemy import text
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 dispatch_case(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)
base_dir = tmp_path / "case"
base_dir.mkdir()
archive_rel = Path("archives") / "job.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="Printer",
serial_number="SERIAL",
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="job.3mf",
file_path=str(archive_rel),
file_size=archive_abs.stat().st_size,
status="completed",
)
db.add(archive)
await db.flush()
item = PrintQueueItem(printer_id=printer.id, archive_id=archive.id, status="pending")
db.add(item)
await db.commit()
ids = SimpleNamespace(printer_id=printer.id, archive_id=archive.id, item_id=item.id)
try:
yield SimpleNamespace(session_maker=session_maker, base_dir=base_dir, ids=ids)
finally:
await engine.dispose()
@pytest.mark.asyncio
async def test_connection_released_before_ftp_upload(dispatch_case):
"""At the moment the FTP upload starts, the caller's transaction is closed."""
scheduler = PrintScheduler()
observed: dict = {}
async with dispatch_case.session_maker() as db:
item = await db.get(PrintQueueItem, dispatch_case.ids.item_id)
async def record_txn_state(*args, **kwargs):
# Captured at the instant upload_file_async runs — must be outside a txn.
observed["in_transaction_at_upload"] = db.in_transaction()
return True
patches = [
patch.object(scheduler_module.settings, "base_dir", dispatch_case.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", MagicMock(return_value=True)),
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", AsyncMock(return_value=True)),
patch("backend.app.services.print_scheduler.upload_file_async", record_txn_state),
patch("backend.app.services.print_scheduler.cache_3mf_download", MagicMock()),
patch("backend.app.services.print_scheduler.spawn_background_task", MagicMock()),
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 p in patches:
stack.enter_context(p)
await scheduler._start_print(db, item)
assert observed.get("in_transaction_at_upload") is False, (
"the pooled connection was still idle-in-transaction when the FTP upload started"
)
@pytest.mark.asyncio
async def test_connection_released_before_preheat_soak(dispatch_case):
"""The preheat stage releases the caller's transaction before the soak wait."""
scheduler = PrintScheduler()
observed: dict = {}
item = SimpleNamespace(id=1, preheat_override="on", preheat_chamber_target_override=None)
printer = SimpleNamespace(id=7, model="P1S") # no chamber sensor → straight to soak
archive = SimpleNamespace(bed_temperature=60)
async with dispatch_case.session_maker() as db:
# Simulate the caller's earlier reads holding an open transaction.
await db.execute(text("SELECT 1"))
assert db.in_transaction()
async def record_sleep(_seconds):
observed["in_transaction_at_soak"] = db.in_transaction()
client = MagicMock()
client.set_bed_temperature = MagicMock(return_value=True)
with (
patch.object(scheduler, "_get_bool_setting", AsyncMock(return_value=True)),
patch.object(
scheduler,
"_get_int_setting",
AsyncMock(
side_effect=lambda _db, key, default: {
"preheat_max_wait_seconds": 0,
"preheat_soak_seconds": 300,
}.get(key, default)
),
),
patch.object(scheduler, "_get_preheat_filament_targets", AsyncMock(return_value={})),
patch("backend.app.services.print_scheduler.printer_manager") as pm,
patch("backend.app.services.print_scheduler.asyncio.sleep", record_sleep),
):
pm.get_client.return_value = client
pm.get_status.return_value = SimpleNamespace(
temperatures={"bed": 100, "chamber": 0}, airduct_mode=0, raw_data={}
)
await scheduler._preheat_and_soak(db, item, printer, archive)
assert observed.get("in_transaction_at_soak") is False, (
"the pooled connection was still idle-in-transaction during the heat-soak wait"
)