fix(queue): withdraw an expected print when the command never goes out

feat(db): warn when the connection pool can outgrow the PostgreSQL server

    fix(mqtt): an unusable layer_num must not drop the printer connection

    test: patch settings.base_dir via monkeypatch so it unwinds on error

    test: restore the config module after reloading it

CHANGELOG.md

@@ -30,6 +30,9 @@ All notable changes to Bambuddy will be documented in this file.
 - **Debug logs now record what the printer reports between the last layer and the end of a print (#2547, reporter @anthonyma94)** — The finish photo wants a moment that Bambu firmware does not obviously announce: printing done, toolhead parked, filament unload not yet started. Bambuddy has been driving that capture from `stg_cur=22` ("Filament unloading"), which turns out to fire on no model at all — across 247 support bundles there is not a single stage-22 capture, including the window in which it was the only trigger in the code, where all 104 captures on A1, A1 Mini, H2C, H2D, P1S, P2S, X1C and X2D fell through to the after-the-fact fallback. Choosing a replacement was not possible from the bundles we had, because outside `stg_cur` and `mc_print_sub_stage` every stage and action field the printers send is dropped unread, and the most promising candidates (`print_real_action`, `mc_action`, `mc_stage`) are absent from A1, A1 Mini and P1S payloads entirely. With debug logging enabled, Bambuddy now dumps those raw fields for the window between the last object layer and the end of the print — opening on the first end-of-print signal (last layer reached, progress at 99+, or no remaining time), logging only what changed frame to frame, and closing on the state transition — so a single debug bundle per model can show whether any firmware marks that moment. Diagnostics only: nothing reads these values, they are printer telemetry with nothing identifying in them, and at normal log levels the probe does no work at all. Covered by tests for the window boundaries, the frame budget and the guarantee that the probe cannot break status ingest.
 
 ### Fixed
+- **A database hiccup during dispatch could leave a queue item stuck and the next print of that file filed under the wrong archive** — When PostgreSQL briefly refused a connection in the middle of dispatching a queue item, the dispatch failed part-way through and left two things behind. **Root cause.** Bambuddy tells itself to expect a print just *before* it sends the print command, because the printer can report the job before the send even returns — but nothing undid that expectation when the command was never sent. The entry does expire after two hours, which is far longer than it takes to react to a failed job by pressing print again: that reprint was folded into the old archive and inherited its filament mapping and plate instead of getting a fresh one. Separately, releasing the row's dispatch lock is best-effort and needed the same database that had just refused, so it gave up after one attempt and the item stayed invisible to the scheduler until a restart. Nothing was ever sent to the printer — the failure happens before the print command — so this cost a stuck item, not a wrong print. **Fix.** An expectation is now withdrawn whenever the print command doesn't go out, which also covers two cases that were silently leaking before: a job cancelled during dispatch, and a print command the printer rejects. The dispatch lock is retried rather than abandoned after one try, and any lock left behind is released on the next quiet moment instead of surviving until a restart. Covered by tests for the withdrawal being an exact inverse of the registration, for a confirmed print keeping its expectation, for two dispatches not disturbing each other, and for the lock recovering from both a brief and a sustained database outage.
+- **Bambuddy can be configured to want more database connections than PostgreSQL will give it** — The connection pool's own ceiling is 100 per worker process by default, while a stock PostgreSQL allows 100 in total and reserves 3 of those for administrators. Nothing checked the two against each other, so the mismatch only appeared as a failure somewhere unrelated once the connections ran out — which is how the dispatch failure above happened. **Fix.** Bambuddy now compares the two at startup and, when the pool could ask for more than the server allows, logs a warning naming both numbers, how many connections are already open, and the settings to change. Both figures are also included in the support bundle, so it is possible to tell a misconfigured limit apart from connections being held too long. Nothing is adjusted automatically: the pool is sized before any connection exists to ask the server with, and the right ceiling depends on how many worker processes you run and what else shares the server. Unchanged on SQLite, which has no such limit, and a server that declines the question is ignored rather than delaying startup. Covered by tests for the warning content, the reserved-slot arithmetic, the values reported to the bundle, and startup surviving a refused probe.
+- **An unusable layer number from a printer could drop its connection (#2702 follow-up)** — Reading the current layer from a status message assumed it would always be a number. Anything else raised an error out of the routine that reads those messages, and nothing above that point catches it, so the connection's listener stopped and the printer looked silent until the staleness check rebuilt it — losing not just the layer number but the print-start and print-finished detection carried in the same message. An unusable value is now ignored, holding the last known layer rather than substituting zero, which the firmware uses to signal a cancellation. This is the same containment applied to the layer *total* in this release, three lines away in the same routine. Covered by tests.
 - **The layer count stayed empty for a whole print, and First Layer Complete notifications read `1/0` (#2702, reporter @sn8key)** — On a P1S, a job started from Archives or through the Virtual Printer showed no layer information in the Print Status panel, and the Discord **First Layer Complete** notification went out as `1/0` instead of `1/33`. The count usually appeared some minutes into the print, which made it look intermittent, and the reporter noticed it filling in at the exact moment they submitted a bug report. **Root cause.** Bambuddy applies the printer's reported layer total as soon as it arrives, and separately clears that total when it detects a new print starting, so a leftover count from the previous job cannot become the denominator for this job's filament split. Both happen while processing the same status message, in that order — and the message announcing a new print is frequently the same one carrying the new print's layer total, so the value was stored and then immediately discarded. That is not merely late but unrecoverable: Bambu firmware sends only fields whose value has changed, so having published the total once, the printer never mentions it again. It could only come back in a full status refresh, which Bambuddy requests when it connects and when you press Force Refresh, but not during a print. Submitting a bug report happens to trigger one, which is why that appeared to fix it; and an install whose connection drops from time to time recovered by itself within minutes, which is why the fault looked random and why a *stable* connection made it worse rather than better. Everything that shows a layer count reads this single value, so the finish-photo trigger that fires on the last layer and the mid-print filament split were affected the same way. **Fix.** The new-print reset now keeps the total that arrived alongside the starting message, while still discarding anything left over from the previous job. If the starting message carried no total, Bambuddy asks for a full refresh once, and once more if layers are being laid down and there is still no total — by which point the printer certainly knows it. That is at most two extra messages per print, and never a per-layer retry. One incidental fix: an unusable layer total — `null`, or anything that isn't a number — previously raised an error out of the routine that reads status messages. Nothing above that point catches it, so the whole message was thrown away and the connection's listener stopped, leaving the printer to look silent until the staleness check rebuilt the connection. Such a value is now ignored, and the refresh described above then fetches the real total. Covered by tests for the same-message case, the case where the total arrives a message earlier, the previous job's total still being discarded, the one-shot bound, the refresh answer not re-triggering the reset, and the malformed values.
 - **Support bundles could contain a printer-status file that no tool could open (#2702)** — Every support bundle includes a redacted copy of the printer's last status message, so that a given model and firmware's actual field layout is available when diagnosing a report. For printers whose AMS flow-calibration factor had a particular number of decimal places, that file was not valid JSON. **Root cause.** Redaction ran over the finished text of the file, and one of its patterns recognises Bambu serial numbers by shape — a shape the digits of a decimal number can also take. A calibration value of `0.0199999995529652` came out as `0.[SERIAL]`, which is not a number, so the file failed to parse from that point on. **Fix.** Redaction now walks the structure and rewrites text values only, leaving numbers, true/false and empty values exactly as the printer reported them, so the file always parses. Nothing is redacted less thoroughly than before. Covered by tests using the value from the bundle that exposed this, and for the walk leaving keys, booleans and nested containers alone.
 - **Setting Spoolman options over the API with a true/false value returned a server error** — `PUT /settings/spoolman` accepts a free-form body, and sending the natural JSON form for a switch — `{"spoolman_enabled": true}` rather than `{"spoolman_enabled": "true"}` — came back as a 500 with nothing useful in it. The shipped UI always sends strings, so this only affected people driving Bambuddy from a script or a Home Assistant `rest_command`, which is exactly where a real boolean is the obvious thing to send. **Root cause.** Settings are stored as text and every reader compares them as text, but the submitted value went in untouched. Deciding whether Spoolman had just been switched on called a string operation on it, which a boolean does not have; and the raw boolean was also written straight to a text column, which SQLite quietly turns into 1/0 while PostgreSQL refuses it outright — so the stored result depended on which database the install used. **Fix.** Boolean-ish settings are now converted to a canonical `true`/`false` on the way in, accepting real booleans, `1`/`0`, and the usual spellings (`True`, `yes`, `on`) case-insensitively, since this is a documented API that scripts talk to. A value with no sensible reading, such as `"banana"`, now returns a 400 naming the field instead of being stored as-is and silently treated as off. Two details are preserved deliberately: a blank value still means "use the default" for the two options that default to on, and reading a stored value stays as strict as it has always been elsewhere in the codebase, so no existing row changes meaning. Text options are checked too, so a JSON object can no longer be stored as its own printed form. One incidental improvement: a value stored as `True` by an earlier API call showed as off in the UI, which compares case-sensitively, while the backend treated it as on — canonical storage removes that disagreement. Covered by tests across the accepted spellings, the rejected values, the blank-means-default behaviour, and the read path.

backend/app/core/database.py

@@ -27,6 +27,11 @@ def _set_sqlite_pragmas(dbapi_conn, connection_record):
 # /system/db-pool can report it without re-deriving the dialect defaults.
 _pool_config: dict = {}
 
+# What the PostgreSQL server itself will allow, read once at startup. None on
+# SQLite, or when the probe could not run. Reported by get_pool_status() so a
+# support bundle carries both sides of the comparison.
+_server_connection_limits: dict | None = None
+
 
 def _resolve_pool_kwargs() -> dict:
     """Build the pool kwargs for ``create_async_engine`` (issue #2572).
@@ -151,6 +156,10 @@ def get_pool_status() -> dict:
     return {
         "dialect": "sqlite" if is_sqlite() else "postgresql",
         "config": dict(_pool_config),
+        # Both sides of the ceiling-vs-server comparison, so a support bundle
+        # shows whether a TooManyConnectionsError was a misconfiguration or a
+        # genuine leak. None on SQLite or if the startup probe couldn't run.
+        "server_limits": dict(_server_connection_limits) if _server_connection_limits else None,
         **gauges,
     }
 
@@ -317,6 +326,107 @@ async def init_db():
     await seed_spool_catalog()
     await seed_color_catalog()
 
+    await check_pool_fits_server()
+
+
+async def check_pool_fits_server() -> None:
+    """Warn when the pool may ask PostgreSQL for more connections than it allows.
+
+    ``pool_size + max_overflow`` is the most connections one worker process will
+    ever open. If that exceeds what the server permits, the pool never reaches
+    its own limit and so never queues: it goes straight to the server, which
+    refuses with ``TooManyConnectionsError``. That surfaces wherever the next
+    connection happened to be needed — in the reported case, halfway through a
+    queue dispatch, which then left an expected-print registration and a dispatch
+    claim behind (#2702 follow-up).
+
+    The distinction is worth knowing when reading a log: SQLAlchemy's own
+    ``QueuePool limit ... timed out`` means the pool is the bottleneck (too much
+    concurrency, or connections held too long), whereas asyncpg's
+    ``TooManyConnectionsError`` means the pool's ceiling is above the server's.
+
+    Not clamped, deliberately. Pool sizes are fixed when the engine is created,
+    which happens at import — before any connection exists to ask the server
+    with — and ``engine`` / ``async_session`` are imported by name in ~150 places,
+    so swapping the engine afterwards would leave stale references. The correct
+    ceiling also depends on the worker count and on anything else sharing the
+    server, neither of which Bambuddy can see. So this reports the mismatch with
+    both numbers and the knobs to fix it, and leaves the choice to the operator.
+    """
+    global _server_connection_limits
+    if is_sqlite():
+        return
+
+    from sqlalchemy import text
+
+    in_use: int | None = None
+    try:
+        async with engine.connect() as conn:
+            max_conn = int((await conn.execute(text("SHOW max_connections"))).scalar_one())
+            reserved = int((await conn.execute(text("SHOW superuser_reserved_connections"))).scalar_one())
+            try:
+                in_use = int(
+                    (
+                        await conn.execute(
+                            text("SELECT count(*) FROM pg_stat_activity WHERE backend_type = 'client backend'")
+                        )
+                    ).scalar_one()
+                )
+            except Exception as exc:
+                # `pg_stat_activity.backend_type` is PostgreSQL 10+, and a
+                # restricted role sees fewer rows. The count is a nice-to-have
+                # for spotting other clients; the warning itself only needs the
+                # two settings above, so losing it must not cost the warning.
+                # Done last on purpose: a failed statement can abort the
+                # transaction, and nothing else uses this connection after it.
+                logger.debug("Could not count client backends: %s", exc)
+    except Exception as exc:
+        # A diagnostic must never be the reason startup fails. An older server
+        # or a restricted role may refuse these.
+        logger.debug("Could not read PostgreSQL connection limits: %s", exc)
+        return
+
+    available = max_conn - reserved
+    ceiling = _pool_config.get("pool_size", 0) + _pool_config.get("max_overflow", 0)
+    _server_connection_limits = {
+        "max_connections": max_conn,
+        "superuser_reserved_connections": reserved,
+        "available_to_bambuddy": available,
+        "client_backends_at_startup": in_use,
+        "pool_ceiling_per_worker": ceiling,
+    }
+
+    if ceiling > available:
+        in_use_note = (
+            f" {in_use} client connection(s) are open on the server right now, including "
+            "this one — a count well above 1 means something else shares it."
+            if in_use is not None
+            else ""
+        )
+        logger.warning(
+            "DB pool may exceed what PostgreSQL allows: this worker can open up to %d "
+            "connections (pool_size %d + max_overflow %d) but the server permits %d "
+            "(max_connections %d minus %d reserved for superusers).%s Exhaustion surfaces "
+            "as TooManyConnectionsError at whatever ran next, not as a pool timeout. "
+            "Lower DB_POOL_SIZE / DB_MAX_OVERFLOW, or raise the server's "
+            "max_connections — and account for every worker process and any other "
+            "client sharing this server.",
+            ceiling,
+            _pool_config.get("pool_size", 0),
+            _pool_config.get("max_overflow", 0),
+            available,
+            max_conn,
+            reserved,
+            in_use_note,
+        )
+    else:
+        logger.info(
+            "DB pool fits the server: up to %d connection(s) per worker, %d available (max_connections %d).",
+            ceiling,
+            available,
+            max_conn,
+        )
+
 
 # B2: Module-level counter exposing the number of rows skipped during the last
 # _migrate_encrypt_legacy_secrets() invocation. Surfaced via /encryption-status

backend/app/main.py

@@ -740,6 +740,51 @@ def register_expected_print(
     )
 
 
+def unregister_expected_print(printer_id: int, filename: str, archive_id: int) -> None:
+    """Undo :func:`register_expected_print` when the print never went out.
+
+    Registration has to happen *before* the MQTT print command, because the
+    printer can report the print before the line after the send executes. So
+    every path that registers and then fails to send — a cancel winning the
+    #1853 CAS race, a ``start_print()`` that returns False, or any exception in
+    between — leaves an expectation for a print that will never arrive.
+
+    The TTL sweep evicts those after two hours, which is far longer than it
+    takes a user to react to a failed dispatch by pressing print again: that
+    reprint would be folded into the *old* archive and take the stale
+    ``ams_mapping`` / ``plate_id`` with it. Hence the explicit inverse.
+
+    Mirrors the sweep's rules, including the one that is easy to get wrong:
+    ``_print_ams_mappings`` / ``_print_plate_ids`` are keyed by archive, not by
+    file, so they may only be dropped once no live key still points at that
+    archive.
+    """
+    keys = [(printer_id, filename)]
+    if filename.endswith(".3mf"):
+        base = filename[:-4]
+        keys.append((printer_id, base))
+        keys.append((printer_id, f"{base}.gcode"))
+
+    removed = False
+    for key in keys:
+        if _expected_prints.pop(key, None) is not None:
+            removed = True
+        _expected_print_creators.pop(key, None)
+        _expected_print_registered_at.pop(key, None)
+
+    if archive_id not in set(_expected_prints.values()):
+        _print_ams_mappings.pop(archive_id, None)
+        _print_plate_ids.pop(archive_id, None)
+
+    if removed:
+        logging.getLogger(__name__).info(
+            "Unregistered expected print: printer=%s, file=%s, archive=%s (print was never sent)",
+            printer_id,
+            filename,
+            archive_id,
+        )
+
+
 def _compute_run_filament_grams(
     status: str,
     archive_filament_used_grams: float | None,

backend/app/services/bambu_mqtt.py

@@ -3023,7 +3023,25 @@ class BambuMQTTClient:
                 )
 
         if "layer_num" in data:
-            new_layer = int(data["layer_num"])
+            try:
+                new_layer = int(data["layer_num"])
+            except (TypeError, ValueError):
+                # Contained for the same reason as `total_layer_num` above: an
+                # exception raised here escapes `_update_state` and paho
+                # re-raises it on the network thread. Losing this frame would
+                # also lose the print-start and completion detection further
+                # down, which is worse than losing a layer number.
+                #
+                # Held at the last known layer rather than substituted with 0:
+                # a fabricated 0 reads as the firmware's cancel reset, which
+                # would move `_last_valid_layer_num` and show layer 0 in the UI
+                # until the next good frame.
+                logger.debug(
+                    "[%s] ignoring unusable layer_num: %r",
+                    self.serial_number,
+                    data["layer_num"],
+                )
+                new_layer = self.state.layer_num
             old_layer = self.state.layer_num
             # Save last non-zero layer for usage tracking (firmware resets to 0 on cancel)
             if old_layer > 0:

backend/app/services/print_scheduler.py

@@ -280,17 +280,29 @@ class PrintScheduler:
         # event-loop thread, so this dict needs no lock.
         # item_id -> (task, printer_id)
         self._inflight: dict[int, tuple[asyncio.Task, int | None]] = {}
+        # Expected prints registered by `_start_print` that have not yet had a
+        # print command sent. Populated at registration, dropped once
+        # `start_print()` succeeds, and rolled back by `_dispatch_one` on every
+        # other exit. Same threading argument as `_inflight` above: one
+        # sequential caller, callbacks on the same loop, so no lock.
+        # item_id -> (printer_id, remote_filename, archive_id)
+        self._unconfirmed_expected_print: dict[int, tuple[int, str, int]] = {}
 
     async def run(self):
         """Main loop - check queue every interval."""
         self._running = True
         logger.info("Print scheduler started")
 
-        await self._clear_stale_dispatch_claims()
+        await self._clear_stale_dispatch_claims(at_startup=True)
 
         while self._running:
             dispatched = False
             try:
+                # No-op while any upload is in flight; on a quiet tick it releases
+                # a claim whose best-effort clear failed (e.g. the database was
+                # briefly unreachable), instead of leaving the row wedged until
+                # the next restart.
+                await self._clear_stale_dispatch_claims()
                 dispatched = await self.check_queue()
             except Exception as e:
                 logger.error("Scheduler error: %s", e)
@@ -299,14 +311,29 @@ 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).
+    async def _clear_stale_dispatch_claims(self, *, at_startup: bool = False) -> None:
+        """Clear dispatch claims with no live dispatch coroutine behind them (#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()``."""
+        A claim is only ever held by a live dispatch coroutine, so when this
+        process has nothing in ``_inflight`` every ``dispatching_at`` in the table
+        is stale. At startup that is trivially true — no coroutine survives a
+        restart. It is equally true on any later tick where no upload is running,
+        which is what makes this safe to repeat rather than only run once.
+
+        Repeating it matters because ``_clear_dispatch_claim`` is best-effort: if
+        the database is briefly unreachable at exactly the moment dispatch ends,
+        the claim survives and the row is wedged out of the selection query. That
+        used to last until the next restart (#2702 follow-up, seen when
+        PostgreSQL refused a connection mid-dispatch).
+
+        ``_inflight`` is populated when the task is spawned, before the coroutine
+        claims its row, and pruned by a done-callback that cannot run before the
+        coroutine's own ``finally`` — so "claim present, nothing in flight" has no
+        race window and needs no age threshold. A size-derived upload deadline
+        (``max(600s, size/25KB/s)``) has no safe fixed bound anyway.
+        """
+        if self._inflight:
+            return
         try:
             async with async_session() as db:
                 res = await db.execute(
@@ -314,9 +341,13 @@ class PrintScheduler:
                 )
                 await db.commit()
                 if res.rowcount:
-                    logger.info("Cleared %d stale dispatch claim(s) at startup (#2615)", res.rowcount)
+                    logger.info(
+                        "Cleared %d orphaned dispatch claim(s)%s (#2615)",
+                        res.rowcount,
+                        " at startup" if at_startup else "",
+                    )
         except Exception as exc:
-            logger.error("Failed to clear stale dispatch claims at startup: %s", exc)
+            logger.error("Failed to clear orphaned dispatch claims: %s", exc)
 
     def stop(self):
         """Stop the scheduler."""
@@ -929,6 +960,14 @@ class PrintScheduler:
                     return
                 await self._start_print(item_db, item)
             finally:
+                # Undo an expected-print registration whose print command never
+                # went out. One choke point covers every way `_start_print` can
+                # end without sending: a raised exception (a DB failure mid-
+                # dispatch is the reported case), an early return, a cancel
+                # winning the #1853 CAS, or `start_print()` returning False.
+                # A confirmed send removes the entry itself, so this is a no-op
+                # on the happy path.
+                self._rollback_unconfirmed_expected_print(item_id)
                 # 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
@@ -936,6 +975,30 @@ class PrintScheduler:
                 # dispatchable again on the next tick.
                 await self._clear_dispatch_claim(item_db, item_id)
 
+    def _rollback_unconfirmed_expected_print(self, item_id: int) -> None:
+        """Drop an expectation for a print command that was never sent.
+
+        Best-effort and never raises: this runs in the ``finally`` of dispatch,
+        where the interesting exception is usually the one already propagating.
+        """
+        pending = self._unconfirmed_expected_print.pop(item_id, None)
+        if pending is None:
+            return
+        printer_id, remote_filename, archive_id = pending
+        try:
+            from backend.app.main import unregister_expected_print
+
+            unregister_expected_print(printer_id, remote_filename, archive_id)
+        except Exception:
+            logger.warning(
+                "Queue item %s: failed to unregister expected print (printer=%s, file=%s, archive=%s)",
+                item_id,
+                printer_id,
+                remote_filename,
+                archive_id,
+                exc_info=True,
+            )
+
     async def _claim_for_dispatch(self, db: AsyncSession, item_id: int) -> bool:
         """Atomically stamp ``dispatching_at`` on a still-pending, unclaimed row.
 
@@ -954,12 +1017,39 @@ class PrintScheduler:
 
     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)
+        mask the dispatch outcome.
+
+        Retried, because the failure mode in practice is transient and narrow: a
+        database that is momentarily unreachable — PostgreSQL out of connection
+        slots is the observed case — refuses this write for a second or two while
+        the dispatch that just ended is still holding the row out of the selection
+        query. One attempt was enough to wedge the item; a couple of spaced
+        attempts clear it. Each attempt rolls back first, since a failed write
+        leaves the session needing it before it can be reused.
+
+        If every attempt fails, ``_clear_stale_dispatch_claims`` picks the row up
+        on the next quiet tick.
+        """
+        for attempt in range(1, 4):
+            try:
+                await db.execute(update(PrintQueueItem).where(PrintQueueItem.id == item_id).values(dispatching_at=None))
+                await db.commit()
+                return
+            except Exception as exc:
+                try:
+                    await db.rollback()
+                except Exception:
+                    pass
+                if attempt == 3:
+                    logger.warning(
+                        "Queue item %s: failed to clear dispatch claim after %d attempts: %s "
+                        "— a later quiet tick will release it",
+                        item_id,
+                        attempt,
+                        exc,
+                    )
+                    return
+                await asyncio.sleep(0.5 * attempt)
 
     async def _find_idle_printer_for_model(
         self,
@@ -3432,6 +3522,12 @@ class PrintScheduler:
                 created_by_id=item.created_by_id,
                 plate_id=item.plate_id,
             )
+            # Registration happens before the print command by necessity (the
+            # printer can report the print before the send returns), so record
+            # what to undo if we never get as far as sending. `_dispatch_one`
+            # rolls back anything still pending here on every exit — exception,
+            # early return, or cancel winning the CAS below.
+            self._unconfirmed_expected_print[item.id] = (item.printer_id, remote_filename, archive.id)
 
         # Propagate the queue item's owner into printer_manager so the
         # print-complete callback can credit the user in the PrintLogEntry
@@ -3556,6 +3652,10 @@ class PrintScheduler:
         )
 
         if started:
+            # The command is away, so the expectation is now legitimate and must
+            # survive. Anything still in this dict when _dispatch_one exits gets
+            # rolled back.
+            self._unconfirmed_expected_print.pop(item.id, None)
             logger.info("Queue item %s: Print started successfully - %s", item.id, filename)
             # No dispatch-toast event here: the legacy bg-dispatch path kept
             # status='processing' from upload start until the printer acked

backend/tests/integration/test_library_slice_api.py

@@ -93,15 +93,24 @@ async def _wait_for_job(client: AsyncClient, job_id: int, timeout: float = 5.0)
 
 
 @pytest.fixture
-async def slice_test_setup(db_session, tmp_path):
-    """Source LibraryFile + 3 LocalPresets + preferred_slicer=orcaslicer."""
+async def slice_test_setup(db_session, tmp_path, monkeypatch):
+    """Source LibraryFile + 3 LocalPresets + preferred_slicer=orcaslicer.
+
+    ``base_dir`` is patched via ``monkeypatch`` rather than assigned and
+    restored by hand. ``app_settings`` is a process-wide singleton, and the
+    hand-rolled version only restored after the ``yield`` — so anything raising
+    during setup (a commit, a refresh) left ``base_dir`` pointing at a
+    ``tmp_path`` that pytest then deleted, and every later test in that xdist
+    worker which reads it failed. That was the cause of intermittent failures
+    in ``TestLibraryPathHelpers`` and ``TestArchivePlatesDesignOverrides``,
+    which share nothing with this module but land in the same worker.
+    """
     storage_dir = tmp_path / "library" / "files"
     storage_dir.mkdir(parents=True, exist_ok=True)
     src_path = storage_dir / "Cube.stl"
     src_path.write_bytes(b"solid Cube\nendsolid\n")
 
-    original_base_dir = app_settings.base_dir
-    app_settings.base_dir = tmp_path
+    monkeypatch.setattr(app_settings, "base_dir", tmp_path)
 
     src_file = LibraryFile(
         filename="Cube.stl",
@@ -137,7 +146,6 @@ async def slice_test_setup(db_session, tmp_path):
         "tmp_path": tmp_path,
     }
 
-    app_settings.base_dir = original_base_dir
     slicer_api_module.set_shared_http_client(None)
 
 

backend/tests/unit/test_config_env_warnings.py

@@ -9,6 +9,35 @@ import logging
 import pytest
 
 
+@pytest.fixture(autouse=True)
+def _restore_config_module():
+    """Undo the ``importlib.reload`` these tests depend on.
+
+    Reloading ``backend.app.core.config`` re-executes it, so ``settings`` becomes
+    a *new* object built from the environment as it stands mid-test. Nothing put
+    the old one back. ``monkeypatch`` unwinds the env vars, not the reload.
+
+    The result is two live ``Settings`` instances in one process: every module
+    that did ``from ... config import settings`` at import time keeps the
+    original, while anything resolving ``config.settings`` afterwards gets the
+    replacement — and under xdist that split persisted for every later test in
+    the same worker. It surfaced as unrelated path assertions failing with a
+    ``base_dir`` from *this* module's tmp_path (``TestLibraryPathHelpers``,
+    ``TestUploadSourceThreeMF``, ``TestArchivePlatesDesignOverrides``,
+    ``TestSystemHealthAPI``), which is why it looked like a random flake and
+    moved between runs as the work distribution changed.
+
+    Snapshotting the whole module dict rather than just ``settings`` restores
+    object *identity*, which is what the two views have to agree on.
+    """
+    import backend.app.core.config as cfg_mod
+
+    saved = dict(cfg_mod.__dict__)
+    yield
+    cfg_mod.__dict__.clear()
+    cfg_mod.__dict__.update(saved)
+
+
 @pytest.mark.unit
 def test_unknown_mfa_env_var_logs_info(monkeypatch, caplog):
     """A typo'd MFA_* env var must be logged at INFO so operators see it."""

backend/tests/unit/test_dispatch_claim_recovery.py

@@ -0,0 +1,144 @@
+"""A dispatch claim must not survive the dispatch that held it (#2615, #2702).
+
+``dispatching_at`` holds a queue row out of the selection query for the duration
+of an upload. Clearing it is best-effort, and the observed failure was narrow:
+PostgreSQL refused a connection for a second or two at exactly the moment
+dispatch ended, the single clear attempt failed, and the row stayed invisible to
+the scheduler until the process restarted.
+
+Two independent recoveries, tested here: the clear retries, and a later tick
+releases any claim with no dispatch behind it.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+
+@pytest.fixture
+def scheduler():
+    from backend.app.services.print_scheduler import PrintScheduler
+
+    return PrintScheduler()
+
+
+def _session(fail_times: int) -> MagicMock:
+    """A session whose execute() fails `fail_times` times, then succeeds."""
+    db = MagicMock()
+    calls = {"n": 0}
+
+    async def execute(*_a, **_k):
+        calls["n"] += 1
+        if calls["n"] <= fail_times:
+            raise RuntimeError("remaining connection slots are reserved for roles with the SUPERUSER attribute")
+        return MagicMock(rowcount=1)
+
+    db.execute = AsyncMock(side_effect=execute)
+    db.commit = AsyncMock()
+    db.rollback = AsyncMock()
+    db._calls = calls
+    return db
+
+
+# ---------------------------------------------------------------------------
+# The retry
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+@pytest.mark.unit
+async def test_a_transient_failure_is_retried_and_the_claim_clears(scheduler):
+    """The reported case: one failed attempt used to wedge the row."""
+    db = _session(fail_times=1)
+
+    with patch("backend.app.services.print_scheduler.asyncio.sleep", new=AsyncMock()):
+        await scheduler._clear_dispatch_claim(db, 597)
+
+    assert db._calls["n"] == 2
+    assert db.commit.await_count == 1
+
+
+@pytest.mark.asyncio
+@pytest.mark.unit
+async def test_the_session_is_rolled_back_between_attempts(scheduler):
+    """A failed write leaves the session needing a rollback before reuse."""
+    db = _session(fail_times=1)
+
+    with patch("backend.app.services.print_scheduler.asyncio.sleep", new=AsyncMock()):
+        await scheduler._clear_dispatch_claim(db, 597)
+
+    assert db.rollback.await_count == 1
+
+
+@pytest.mark.asyncio
+@pytest.mark.unit
+async def test_retries_are_bounded_and_never_raise(scheduler):
+    """Dispatch's outcome must not be masked by this cleanup failing."""
+    db = _session(fail_times=99)
+
+    with patch("backend.app.services.print_scheduler.asyncio.sleep", new=AsyncMock()):
+        await scheduler._clear_dispatch_claim(db, 597)  # must not raise
+
+    assert db._calls["n"] == 3
+
+
+@pytest.mark.asyncio
+@pytest.mark.unit
+async def test_no_retry_when_the_first_attempt_works(scheduler):
+    """The happy path must not pay for the retry."""
+    db = _session(fail_times=0)
+
+    await scheduler._clear_dispatch_claim(db, 597)
+
+    assert db._calls["n"] == 1
+
+
+# ---------------------------------------------------------------------------
+# The quiet-tick sweep
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+@pytest.mark.unit
+async def test_the_sweep_does_nothing_while_an_upload_is_in_flight(scheduler):
+    """An in-flight dispatch owns its claim — clearing it would let a second
+    dispatch pick up the same row mid-upload, which is what #2615 prevents."""
+    scheduler._inflight[597] = (MagicMock(), 1)
+
+    with patch("backend.app.services.print_scheduler.async_session") as sess:
+        await scheduler._clear_stale_dispatch_claims()
+
+    sess.assert_not_called()
+
+
+@pytest.mark.asyncio
+@pytest.mark.unit
+async def test_the_sweep_releases_a_claim_with_nothing_in_flight(scheduler):
+    """`_inflight` is populated before the coroutine claims its row, and pruned
+    after its `finally` — so "claim present, nothing in flight" is orphaned."""
+    db = MagicMock()
+    db.execute = AsyncMock(return_value=MagicMock(rowcount=1))
+    db.commit = AsyncMock()
+    ctx = MagicMock()
+    ctx.__aenter__ = AsyncMock(return_value=db)
+    ctx.__aexit__ = AsyncMock(return_value=False)
+
+    with patch("backend.app.services.print_scheduler.async_session", return_value=ctx):
+        await scheduler._clear_stale_dispatch_claims()
+
+    assert db.execute.await_count == 1
+    assert db.commit.await_count == 1
+
+
+@pytest.mark.asyncio
+@pytest.mark.unit
+async def test_the_sweep_survives_a_database_that_is_still_down(scheduler):
+    """It runs every tick; a failure must not break the scheduler loop."""
+    ctx = MagicMock()
+    ctx.__aenter__ = AsyncMock(side_effect=RuntimeError("still refusing connections"))
+    ctx.__aexit__ = AsyncMock(return_value=False)
+
+    with patch("backend.app.services.print_scheduler.async_session", return_value=ctx):
+        await scheduler._clear_stale_dispatch_claims()  # must not raise

backend/tests/unit/test_expected_print_rollback.py

@@ -0,0 +1,263 @@
+"""A dispatch that never sends the print command must leave no expectation.
+
+``register_expected_print`` has to run *before* the MQTT command, because the
+printer can report the print before the send returns. So any path that
+registers and then fails to send leaves Bambuddy expecting a print that will
+never arrive: a cancel winning the #1853 CAS race, ``start_print()`` returning
+False, or an exception in between — a PostgreSQL connection failure mid-dispatch
+is the case that surfaced this (#2702 follow-up).
+
+The two-hour TTL sweep does eventually evict such an entry, but two hours is far
+longer than it takes someone to react to a failed dispatch by pressing print
+again. That reprint would be folded into the *old* archive and inherit its
+``ams_mapping`` and ``plate_id`` instead of creating a fresh one.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+
+@pytest.fixture
+def expected_print_tables():
+    """The module-level registries, emptied around each test."""
+    from backend.app import main
+
+    names = (
+        "_expected_prints",
+        "_expected_print_creators",
+        "_expected_print_registered_at",
+        "_print_ams_mappings",
+        "_print_plate_ids",
+    )
+    saved = {n: dict(getattr(main, n)) for n in names}
+    for n in names:
+        getattr(main, n).clear()
+    yield main
+    for n in names:
+        getattr(main, n).clear()
+        getattr(main, n).update(saved[n])
+
+
+# ---------------------------------------------------------------------------
+# unregister_expected_print is the exact inverse of register_expected_print
+# ---------------------------------------------------------------------------
+
+
+def test_unregister_leaves_every_registry_as_it_found_them(expected_print_tables):
+    """The strongest form: register then unregister is a round trip to empty."""
+    main = expected_print_tables
+
+    main.register_expected_print(1, "widget.3mf", 298, ams_mapping=[3, 6], created_by_id=7, plate_id=1)
+    assert main._expected_prints, "nothing registered — the test proves nothing"
+
+    main.unregister_expected_print(1, "widget.3mf", 298)
+
+    assert main._expected_prints == {}
+    assert main._expected_print_creators == {}
+    assert main._expected_print_registered_at == {}
+    assert main._print_ams_mappings == {}
+    assert main._print_plate_ids == {}
+
+
+def test_unregister_clears_the_filename_variants_too(expected_print_tables):
+    """Registration stores the name three ways; a partial undo still matches."""
+    main = expected_print_tables
+
+    main.register_expected_print(1, "widget.3mf", 298)
+    main.unregister_expected_print(1, "widget.3mf", 298)
+
+    for key in ((1, "widget.3mf"), (1, "widget"), (1, "widget.gcode")):
+        assert key not in main._expected_prints, f"{key} survived"
+
+
+def test_unregister_does_not_touch_another_printers_expectation(expected_print_tables):
+    main = expected_print_tables
+
+    main.register_expected_print(1, "widget.3mf", 298)
+    main.register_expected_print(2, "widget.3mf", 299)
+
+    main.unregister_expected_print(1, "widget.3mf", 298)
+
+    assert main._expected_prints[(2, "widget.3mf")] == 299
+
+
+def test_archive_keyed_tables_survive_while_another_file_still_points_at_them(
+    expected_print_tables,
+):
+    """Mirrors the TTL sweep's rule, which is the easy thing to get wrong.
+
+    ``_print_ams_mappings`` and ``_print_plate_ids`` are keyed by archive, not
+    by file. Two files can be registered against one archive, so dropping them
+    on the first unregister would strip usage-tracking data from a print that is
+    still expected.
+    """
+    main = expected_print_tables
+
+    main.register_expected_print(1, "plate1.3mf", 298, ams_mapping=[3], plate_id=1)
+    main.register_expected_print(1, "plate2.3mf", 298, ams_mapping=[3], plate_id=2)
+
+    main.unregister_expected_print(1, "plate1.3mf", 298)
+
+    assert main._print_ams_mappings.get(298) == [3]
+    assert 298 in main._print_plate_ids
+
+
+def test_unregistering_an_unknown_print_is_a_no_op(expected_print_tables):
+    """Runs from a ``finally``, so it must tolerate having nothing to do."""
+    main = expected_print_tables
+
+    main.unregister_expected_print(99, "never-registered.3mf", 1234)
+
+    assert main._expected_prints == {}
+
+
+# ---------------------------------------------------------------------------
+# The scheduler's rollback hook
+# ---------------------------------------------------------------------------
+
+
+def test_scheduler_rollback_undoes_a_recorded_registration(expected_print_tables):
+    main = expected_print_tables
+    from backend.app.services.print_scheduler import PrintScheduler
+
+    sched = PrintScheduler()
+    main.register_expected_print(1, "widget.3mf", 298, ams_mapping=[3, 6], plate_id=1)
+    sched._unconfirmed_expected_print[597] = (1, "widget.3mf", 298)
+
+    sched._rollback_unconfirmed_expected_print(597)
+
+    assert main._expected_prints == {}
+    assert sched._unconfirmed_expected_print == {}
+
+
+def test_scheduler_rollback_is_a_no_op_after_a_confirmed_send(expected_print_tables):
+    """A sent print's expectation must survive — the callback needs it."""
+    main = expected_print_tables
+    from backend.app.services.print_scheduler import PrintScheduler
+
+    sched = PrintScheduler()
+    main.register_expected_print(1, "widget.3mf", 298, ams_mapping=[3, 6])
+    sched._unconfirmed_expected_print[597] = (1, "widget.3mf", 298)
+    # What `_start_print` does once start_print() returns True.
+    sched._unconfirmed_expected_print.pop(597, None)
+
+    sched._rollback_unconfirmed_expected_print(597)
+
+    assert main._expected_prints[(1, "widget.3mf")] == 298
+    assert main._print_ams_mappings[298] == [3, 6]
+
+
+def test_scheduler_rollback_never_raises(expected_print_tables, monkeypatch):
+    """It runs in the ``finally`` of dispatch, usually with an exception already
+    propagating — it must not replace it with one of its own."""
+    from backend.app.services.print_scheduler import PrintScheduler
+
+    sched = PrintScheduler()
+    sched._unconfirmed_expected_print[597] = (1, "widget.3mf", 298)
+    monkeypatch.setattr(
+        expected_print_tables,
+        "unregister_expected_print",
+        lambda *a, **k: (_ for _ in ()).throw(RuntimeError("boom")),
+    )
+
+    sched._rollback_unconfirmed_expected_print(597)  # must not raise
+
+    assert sched._unconfirmed_expected_print == {}, "entry must be dropped even on failure"
+
+
+@pytest.mark.asyncio
+@pytest.mark.unit
+async def test_dispatch_withdraws_the_expectation_when_start_print_raises(expected_print_tables):
+    """End to end through `_dispatch_one`, on the reported failure.
+
+    A database error inside `_start_print` must leave no expectation behind, must
+    still release the claim, and must not be swallowed — the background-task
+    runner logs it, and hiding it here would turn a loud failure into a silent
+    one.
+    """
+    from unittest.mock import AsyncMock, MagicMock, patch
+
+    main = expected_print_tables
+    from backend.app.services.print_scheduler import PrintScheduler
+
+    sched = PrintScheduler()
+
+    async def fake_start_print(db, item):
+        # What `_start_print` does before the point the real one died.
+        main.register_expected_print(1, "widget.3mf", 298, ams_mapping=[3, 6], plate_id=1)
+        sched._unconfirmed_expected_print[597] = (1, "widget.3mf", 298)
+        raise RuntimeError("remaining connection slots are reserved for roles with the SUPERUSER attribute")
+
+    db = MagicMock()
+    db.get = AsyncMock(return_value=MagicMock(id=597))
+    ctx = MagicMock()
+    ctx.__aenter__ = AsyncMock(return_value=db)
+    ctx.__aexit__ = AsyncMock(return_value=False)
+
+    with (
+        patch("backend.app.services.print_scheduler.async_session", return_value=ctx),
+        patch.object(sched, "_claim_for_dispatch", AsyncMock(return_value=True)),
+        patch.object(sched, "_start_print", side_effect=fake_start_print),
+        patch.object(sched, "_clear_dispatch_claim", AsyncMock()) as clear,
+        pytest.raises(RuntimeError),
+    ):
+        await sched._dispatch_one(597)
+
+    assert main._expected_prints == {}, "expectation survived a dispatch that never sent a print"
+    assert main._print_ams_mappings == {}
+    assert main._print_plate_ids == {}
+    assert sched._unconfirmed_expected_print == {}
+    clear.assert_awaited_once_with(db, 597)
+
+
+@pytest.mark.asyncio
+@pytest.mark.unit
+async def test_dispatch_keeps_the_expectation_when_the_print_was_sent(expected_print_tables):
+    """The mirror image: a confirmed send must survive dispatch teardown, or the
+    print-complete callback would create a duplicate archive."""
+    from unittest.mock import AsyncMock, MagicMock, patch
+
+    main = expected_print_tables
+    from backend.app.services.print_scheduler import PrintScheduler
+
+    sched = PrintScheduler()
+
+    async def fake_start_print(db, item):
+        main.register_expected_print(1, "widget.3mf", 298, ams_mapping=[3, 6])
+        sched._unconfirmed_expected_print[597] = (1, "widget.3mf", 298)
+        sched._unconfirmed_expected_print.pop(597, None)  # start_print() returned True
+
+    db = MagicMock()
+    db.get = AsyncMock(return_value=MagicMock(id=597))
+    ctx = MagicMock()
+    ctx.__aenter__ = AsyncMock(return_value=db)
+    ctx.__aexit__ = AsyncMock(return_value=False)
+
+    with (
+        patch("backend.app.services.print_scheduler.async_session", return_value=ctx),
+        patch.object(sched, "_claim_for_dispatch", AsyncMock(return_value=True)),
+        patch.object(sched, "_start_print", side_effect=fake_start_print),
+        patch.object(sched, "_clear_dispatch_claim", AsyncMock()),
+    ):
+        await sched._dispatch_one(597)
+
+    assert main._expected_prints[(1, "widget.3mf")] == 298
+    assert main._print_ams_mappings[298] == [3, 6]
+
+
+def test_rollback_entries_are_per_item(expected_print_tables):
+    """Two dispatches in flight must not roll back each other's registration."""
+    main = expected_print_tables
+    from backend.app.services.print_scheduler import PrintScheduler
+
+    sched = PrintScheduler()
+    main.register_expected_print(1, "a.3mf", 1)
+    main.register_expected_print(2, "b.3mf", 2)
+    sched._unconfirmed_expected_print[10] = (1, "a.3mf", 1)
+    sched._unconfirmed_expected_print[11] = (2, "b.3mf", 2)
+
+    sched._rollback_unconfirmed_expected_print(10)
+
+    assert (1, "a.3mf") not in main._expected_prints
+    assert main._expected_prints[(2, "b.3mf")] == 2

backend/tests/unit/test_pool_fits_server.py

@@ -0,0 +1,166 @@
+"""The pool must not silently be allowed to outgrow the PostgreSQL server.
+
+``pool_size + max_overflow`` is the most connections one worker will open. When
+that exceeds what the server permits, the pool never hits its own limit and so
+never queues — it asks the server, which refuses with
+``TooManyConnectionsError`` at whatever happened to need a connection next. In
+the report behind this, that was the middle of a queue dispatch.
+
+The check is diagnostic, not corrective: pool sizes are fixed at engine creation
+(import time, before any connection exists to ask with), and the right ceiling
+depends on the worker count and on other clients sharing the server. So the
+contract under test is "says something accurate and loud, and never breaks
+startup".
+"""
+
+from __future__ import annotations
+
+import logging
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+
+def _engine_reporting(max_conn: int, reserved: int, in_use: int | None = 0) -> MagicMock:
+    """An engine whose connection answers the three probe queries in order.
+
+    ``in_use=None`` makes the third query fail, standing in for PostgreSQL < 10
+    where ``pg_stat_activity.backend_type`` does not exist.
+    """
+    conn = MagicMock()
+    conn.execute = AsyncMock(
+        side_effect=[
+            MagicMock(scalar_one=MagicMock(return_value=max_conn)),
+            MagicMock(scalar_one=MagicMock(return_value=reserved)),
+            (
+                MagicMock(scalar_one=MagicMock(return_value=in_use))
+                if in_use is not None
+                else RuntimeError('column "backend_type" does not exist')
+            ),
+        ]
+    )
+    ctx = MagicMock()
+    ctx.__aenter__ = AsyncMock(return_value=conn)
+    ctx.__aexit__ = AsyncMock(return_value=False)
+    engine = MagicMock()
+    engine.connect = MagicMock(return_value=ctx)
+    return engine
+
+
+async def _run_check(*, pool_size, max_overflow, max_conn, reserved, in_use=0, sqlite=False):
+    from backend.app.core import database
+
+    with (
+        patch.object(database, "is_sqlite", return_value=sqlite),
+        patch.object(database, "_pool_config", {"pool_size": pool_size, "max_overflow": max_overflow}),
+        patch.object(database, "engine", _engine_reporting(max_conn, reserved, in_use)),
+        patch.object(database, "_server_connection_limits", None),
+    ):
+        await database.check_pool_fits_server()
+        return database._server_connection_limits
+
+
+@pytest.mark.asyncio
+@pytest.mark.unit
+async def test_warns_when_the_ceiling_exceeds_what_the_server_allows(caplog):
+    """Bambuddy's own PostgreSQL default against a stock server: 100 vs 100-3."""
+    with caplog.at_level(logging.WARNING, logger="backend.app.core.database"):
+        await _run_check(pool_size=20, max_overflow=80, max_conn=100, reserved=3)
+
+    assert any(r.levelno == logging.WARNING for r in caplog.records)
+    msg = caplog.text
+    # The numbers an operator needs, and the knobs to change.
+    for expected in ("100", "97", "DB_POOL_SIZE", "DB_MAX_OVERFLOW", "max_connections"):
+        assert expected in msg, f"warning omits {expected!r}"
+
+
+@pytest.mark.asyncio
+@pytest.mark.unit
+async def test_silent_when_the_pool_fits(caplog):
+    with caplog.at_level(logging.WARNING, logger="backend.app.core.database"):
+        await _run_check(pool_size=20, max_overflow=80, max_conn=500, reserved=3)
+
+    assert not [r for r in caplog.records if r.levelno >= logging.WARNING]
+
+
+@pytest.mark.asyncio
+@pytest.mark.unit
+async def test_the_reserved_slots_count_against_the_budget(caplog):
+    """Exactly at max_connections is still too many — reserved slots are not ours."""
+    with caplog.at_level(logging.WARNING, logger="backend.app.core.database"):
+        await _run_check(pool_size=10, max_overflow=90, max_conn=100, reserved=3)
+
+    assert [r for r in caplog.records if r.levelno == logging.WARNING]
+
+
+@pytest.mark.asyncio
+@pytest.mark.unit
+async def test_both_sides_are_recorded_for_the_support_bundle():
+    limits = await _run_check(pool_size=20, max_overflow=80, max_conn=100, reserved=3, in_use=41)
+
+    assert limits == {
+        "max_connections": 100,
+        "superuser_reserved_connections": 3,
+        "available_to_bambuddy": 97,
+        "client_backends_at_startup": 41,
+        "pool_ceiling_per_worker": 100,
+    }
+
+
+@pytest.mark.asyncio
+@pytest.mark.unit
+async def test_sqlite_is_skipped_entirely():
+    """No such concept, and the probe SQL is PostgreSQL-only."""
+    limits = await _run_check(pool_size=20, max_overflow=200, max_conn=0, reserved=0, sqlite=True)
+
+    assert limits is None
+
+
+@pytest.mark.asyncio
+@pytest.mark.unit
+async def test_a_probe_failure_cannot_break_startup(caplog):
+    """A restricted role or an older server may refuse these queries."""
+    from backend.app.core import database
+
+    engine = MagicMock()
+    ctx = MagicMock()
+    ctx.__aenter__ = AsyncMock(side_effect=RuntimeError("permission denied"))
+    ctx.__aexit__ = AsyncMock(return_value=False)
+    engine.connect = MagicMock(return_value=ctx)
+
+    with (
+        patch.object(database, "is_sqlite", return_value=False),
+        patch.object(database, "_pool_config", {"pool_size": 20, "max_overflow": 80}),
+        patch.object(database, "engine", engine),
+        patch.object(database, "_server_connection_limits", None),
+        caplog.at_level(logging.WARNING, logger="backend.app.core.database"),
+    ):
+        await database.check_pool_fits_server()  # must not raise
+
+        assert database._server_connection_limits is None
+    assert not [r for r in caplog.records if r.levelno >= logging.WARNING]
+
+
+@pytest.mark.asyncio
+@pytest.mark.unit
+async def test_an_old_server_without_backend_type_still_gets_the_warning(caplog):
+    """`pg_stat_activity.backend_type` is PostgreSQL 10+; the docs recommend 14+
+    but asyncpg reaches back to 9.5. Losing that count must not cost the
+    warning, which only needs the two settings."""
+    with caplog.at_level(logging.WARNING, logger="backend.app.core.database"):
+        limits = await _run_check(pool_size=20, max_overflow=80, max_conn=100, reserved=3, in_use=None)
+
+    assert [r for r in caplog.records if r.levelno == logging.WARNING], "warning was lost with the count"
+    assert "100" in caplog.text and "97" in caplog.text
+    # The sentence about other clients is dropped rather than rendered as None.
+    assert "None client" not in caplog.text
+    assert limits["client_backends_at_startup"] is None
+    assert limits["max_connections"] == 100
+
+
+@pytest.mark.unit
+def test_get_pool_status_exposes_the_server_limits_key():
+    """The support bundle reads this; the key must exist even on SQLite."""
+    from backend.app.core.database import get_pool_status
+
+    assert "server_limits" in get_pool_status()