fix: cancel-safe get_db + drop sqlalchemy.pool cancellation noise

@Carter3DP's support package showed bambuddy.log filling with two
  distinct cascades on long uploads:

    ERROR sqlalchemy.pool   Exception terminating connection ...
                            CancelledError: Cancelled via cancel scope
                            ... by starlette.middleware.base
                            .BaseHTTPMiddleware.__call__.call_next
    ERROR sqlalchemy.pool   The garbage collector is trying to clean up
                            non-checked-in connection ... will be
                            terminated.
    WARN  backend.app.main  Runtime tracking commit failed:
                            (sqlite3.OperationalError) database is locked

  Single root cause. Starlette's BaseHTTPMiddleware (used under the hood
  by every @app.middleware("http") decorator) cancels the inner task
  scope when a client disconnects mid-request — common on long
  multipart uploads where the client times out before the server's
  response. Pre-fix get_db only caught Exception, but CancelledError
  is BaseException, so cancellation skipped the rollback path entirely.
  The SQLite write lock stayed held until GC reclaimed the connection
  ages later, blocking every other writer in the meantime. On Postgres
  the leak shape is identical; the symptom would be "QueuePool limit
  ... overflow" instead of "database is locked".

  (1) get_db now catches BaseException so CancelledError triggers
      rollback. Both rollback() and close() are wrapped in
      asyncio.shield so the cleanup completes even when the await
      itself is being cancelled by the same cancel scope. SQLite write
      lock is released promptly; connection returns to the pool instead
      of leaking until GC.

  (2) CancelledPoolNoiseFilter (new filter on sqlalchemy.pool) drops
      the residual records that pre-existing pools still emit during
      their own cleanup. Two patterns suppressed:
        - "Exception terminating connection ..." with a CancelledError
          anywhere in the exc_info chain (walks __cause__/__context__
          with a seen-set guard against pathological cycles)
        - "The garbage collector is trying to clean up non-checked-in
          connection ..." (always symptomatic of cancellation; never
          independently actionable)
      Real pool problems — broken connections, OSError on terminate,
      pool exhaustion — keep flowing because they carry a different
      exception chain or a different message prefix.

  13 regression tests across test_get_db_cancel_safety.py (commit on
  clean exit, rollback on regular Exception, rollback on CancelledError,
  close runs even if rollback raises, close failure on clean exit
  doesn't propagate, rollback + close both go through asyncio.shield)
  and test_cancelled_pool_filter.py (drops cancellation-driven
  terminate, drops GC-cleanup, keeps real OSError terminate, keeps
  terminate without exc_info, keeps unrelated pool messages, drops
  chained-cause CancelledError, defensive guard against self-referential
  cause chains).

  Applies to SQLite and PostgreSQL — get_db is dialect-agnostic and
  the filtered messages come from base sqlalchemy.pool not from any
  specific dialect.

CHANGELOG.md

@@ -35,6 +35,8 @@ All notable changes to Bambuddy will be documented in this file.
 - **Per-request trace ID column on every log line, plumbed through HTTP access log + application logs + response headers** — Builds on the new uvicorn-access-log-into-bambuddy.log change below: the access line tells you *who* called an endpoint, but until now there was no way to tie that line to the application records emitted on the server side while handling that request. A new FastAPI middleware (`trace_id_middleware` in `main.py`, sourced from `backend.app.core.trace`) stamps each request with a fresh 8-char hex ID (or honours a sane inbound `X-Trace-Id` header for cross-system correlation), stores it in a `ContextVar` so any code in the request's call stack can read it, echoes it on the response as `X-Trace-Id`, and a new `TraceIDFilter` injects it into every `LogRecord` so the format string `[%(trace_id)s]` resolves to the right ID for the right request. ContextVars (rather than `request.state`) are the right plumbing here because asyncio copies the current context into every `asyncio.create_task`, so background work spawned from inside a request inherits the trace ID without explicit threading; the logging filter has no access to the FastAPI request object regardless. Records emitted outside any request scope (startup, MQTT callbacks, scheduler) get a stable `-` placeholder so the column stays visually aligned and missing values are obvious in `grep`. Inbound `X-Trace-Id` is hard-validated against a strict whitelist (`[A-Za-z0-9_-]+`, max 64 chars) before being honoured — a hostile or buggy caller cannot smuggle log-injection payloads (newlines, control chars, megabyte blobs) into `bambuddy.log` via the trace-ID column; values that fail the gate silently trigger a freshly minted server-side ID rather than failing the request. Middleware is decorated AFTER `auth_middleware` on purpose: Starlette stacks `@app.middleware` decorators LIFO so the last-decorated runs first inbound, making trace stamp the OUTERMOST layer — auth log lines and every record emitted on the way down to and back from the route handler all carry the same ID. Output now looks like `2026-04-26 09:51:39,152 INFO [uvicorn.access] [a4f3b1e7] 192.168.1.42:54812 - "POST /api/v1/printers/1/print/stop HTTP/1.1" 200` paired with the route handler's `2026-04-26 09:51:39,158 INFO [bambu_mqtt] [a4f3b1e7] [SERIAL] Sent stop print command` — one `grep a4f3b1e7` away from the full causality chain. 30 new tests across `tests/unit/test_trace.py` (placeholder when no request scope, filter copies ContextVar value onto records, ID propagates into spawned tasks via asyncio context copy, concurrent requests don't leak IDs into each other, generator produces unique hex IDs, hostile payloads rejected by validator, max-length boundary, dash/underscore variants accepted) plus `tests/integration/test_trace_middleware.py` (X-Trace-Id header echoed on response, body and header IDs match, each request gets a unique ID, generator format stays short hex, safe inbound IDs honoured, hostile inbound IDs replaced, overlong inbound IDs replaced, ContextVar reset cleanly after request).
 
 ### Fixed
+- **`bambuddy.log` filling with `Exception terminating connection ... CancelledError` + `database is locked` cascades on long uploads** ([#1112](https://github.com/maziggy/bambuddy/issues/1112) follow-up, surfaced by @Carter3DP's support package) — Two-part fix to a single root cause: Starlette's `BaseHTTPMiddleware` (which FastAPI's `@app.middleware("http")` decorator uses under the hood) cancels the inner task scope when a client disconnects mid-request — common on long multipart uploads where the client times out before the server's response. Pre-fix `get_db` only caught `Exception`, but `CancelledError` is a `BaseException`, so cancellation skipped the rollback path entirely; the SQLite write lock stayed held until the connection was eventually GC'd, producing the `(sqlite3.OperationalError) database is locked` cascade against `runtime_seconds` updates and other tight-loop writers in @Carter3DP's log. Postgres users would see pool exhaustion / "QueuePool limit overflow" instead of file-level lock contention, but the leak shape is identical. **(1)** `get_db` now catches `BaseException` so `CancelledError` triggers rollback, and wraps both `rollback()` and `close()` in `asyncio.shield` so the cleanup completes even when the await itself is being cancelled by the same cancel scope. The SQLite write lock is released promptly; the connection returns to the pool instead of leaking until GC. **(2)** A `CancelledPoolNoiseFilter` (new `logging_filters.py` filter, attached to `sqlalchemy.pool`) drops the residual log noise that pre-existing pools still emit during their own cleanup — both the `Exception terminating connection ... CancelledError` records (matched on prefix + cancellation-driven `exc_info`, including chained `__cause__`/`__context__`) and the symptomatic `garbage collector is trying to clean up non-checked-in connection` records. Real pool problems — broken connections, network hiccups, exhaustion — keep flowing because they carry a different exception chain or a different message prefix; verified by `test_keeps_terminate_with_real_oserror` and `test_keeps_unrelated_pool_message`. 13 new regression tests across `test_get_db_cancel_safety.py` (commit on clean exit, rollback on regular `Exception`, rollback on `CancelledError` — the actual #1112 fix, close runs even if rollback raises, close failure on clean exit doesn't propagate, both rollback + close go through `asyncio.shield`) and `test_cancelled_pool_filter.py` (drops cancellation-driven terminate, drops GC-cleanup, keeps real `OSError` terminate, keeps terminate without `exc_info`, keeps unrelated pool messages, drops chained-cause `CancelledError`, defensive guard against self-referential cause chains). Applies to SQLite and PostgreSQL — `get_db` is dialect-agnostic and the filtered messages come from base `sqlalchemy.pool` not from any specific dialect.
+
 - **Windows install: `bambuddy.log` filling with `WinError 10054 — _ProactorBasePipeTransport._call_connection_lost` tracebacks** ([#1113](https://github.com/maziggy/bambuddy/issues/1113), reported by @cadtoolbox) — Cosmetic-but-noisy. When a printer / MQTT broker / camera RSTs a TCP socket instead of FINing it (offline X1Es in @cadtoolbox's setup, network gear that drops idle TCP, the printer firmware's own watchdog), Windows asyncio's Proactor cleanup path tries `socket.shutdown(SHUT_RDWR)` on the already-dead socket and hits `WinError 10054`. Application-layer reconnect logic (paho-mqtt, httpx) handles the actual disconnect fine — paho retries, MQTT comes back, telemetry resumes — so the traceback is pure asyncio bookkeeping noise, but it fired multiple times per minute on @cadtoolbox's 9-printer setup with 5 offline X1Es and was the first thing in the sanitized log. Adds a custom `loop.set_exception_handler` (new `backend/app/core/asyncio_handlers.py`) installed on Windows only that pattern-matches the specific `_call_connection_lost` cleanup-RST signature (three signals together: `sys.platform == "win32"`, the exception is `ConnectionResetError`, and the asyncio message string contains `_call_connection_lost`) and downgrades it to DEBUG. Real `ConnectionResetError`s raised inside application coroutines (different message string) and other Proactor cleanup errors (`BrokenPipeError`, `ConnectionAbortedError` — same callback site, distinct signal worth keeping visible) all pass through to `loop.default_exception_handler` unchanged. Linux / macOS use the Selector event loop and never hit this codepath, so `install_proactor_reset_filter()` is an explicit no-op there with a `False` return — verified by `test_install_is_no_op_on_non_windows`. 9 unit tests in `test_asyncio_handlers.py` cover: discriminator matches the exact reported signature, rejects unrelated `ConnectionResetError`s, rejects `BrokenPipeError` even on the same callback site, rejects when no exception object is present, install is platform-gated, install wires the handler onto the loop, suppression doesn't reach the default handler, and unrelated exceptions still hit the default handler. Wired from `lifespan` startup before any task can spawn that might trip it.
 
 - **Auto-Print G-code Injection: start snippet landed before printer startup, and `{placeholder}` substitution was silently broken** ([#422](https://github.com/maziggy/bambuddy/issues/422) follow-up) — Two compounding bugs surfaced by @pleite (Swapmod) and @DevScarabyte (multi-height test prints) on the initial #422 ship: **(1)** Start snippets were prepended to the entire `plate_X.gcode` content, which placed them *before* the printer's bed-heat / homing / nozzle-prime sequence — so a Swapmod start snippet that assumed nozzle-at-temp ran on a cold printer. The injection now anchors at `; MACHINE_START_GCODE_END` (the marker sitting at the bottom of every Bambu/Orca slicer's `MACHINE_START_GCODE` block, after `M109` wait-for-temp), matching where a slicer-side custom-start-gcode would land. Files without the marker (older slicer versions) keep the prepend behaviour as a fallback with a warning log. **(2)** Slicer-style placeholders like `G1 Z{max_layer_z} F600` were written verbatim to the output gcode — the printer firmware then parsed `Z{max_layer_z}` as `Z1` and crashed the head into the print on a 60mm-tall model (a real safety issue: prints damaged, top glass + AMS pushed up off the printer when the model was taller than the hard-coded park height). Added a header parser that reads the 3MF's `; HEADER_BLOCK_START..END` block (lowercased keys, `[units]` suffix stripped, spaces → underscores) and a Prusa-style `{name}` substitution pass that runs over both start and end snippets before injection. Supported placeholders: `{max_layer_z}` / `{max_print_height}` (top-layer Z), `{total_layer_number}` / `{total_layers}`, `{total_filament_weight}`, `{total_filament_length}`, plus any other normalised header key from the source file. Unknown placeholders are left in the snippet verbatim with a warning log — a typo never silently expands to an empty string and the firmware never receives a malformed `Z` parameter. 16 new regression tests in `test_gcode_injection.py` covering: start snippet anchored to the marker (printer startup runs first, snippet sits between `M109 S220` and the marker, file head untouched), missing-marker fallback path, end snippet still appended at EOF, `{max_layer_z}` resolved through the alias map, direct-key substitution from the normalised header, unknown-placeholder pass-through, and direct unit tests for each new helper (`_parse_3mf_gcode_header`, `_substitute_placeholders`, `_inject_start_at_marker`). Wiki page documents the supported placeholder list with a safety warning specifically calling out `{max_layer_z}` for park moves.

backend/app/core/database.py

@@ -141,11 +141,25 @@ async def get_db() -> AsyncSession:
         try:
             yield session
             await session.commit()
-        except Exception:
-            await session.rollback()
+        except BaseException:
+            # Catch BaseException (not just Exception) so CancelledError —
+            # raised when Starlette's BaseHTTPMiddleware cancels the inner
+            # task scope on client disconnect — also triggers rollback.
+            # `asyncio.shield` keeps the rollback running to completion
+            # even when the await itself gets cancelled, so the SQLite
+            # write lock is released promptly instead of being held until
+            # the connection is GC'd ages later (which was producing the
+            # "database is locked" cascade in #1112's support package).
+            try:
+                await asyncio.shield(session.rollback())
+            except BaseException:  # noqa: BLE001 — rollback failure must not mask the original
+                pass
             raise
         finally:
-            await session.close()
+            try:
+                await asyncio.shield(session.close())
+            except BaseException:  # noqa: BLE001 — close failure must not mask the original
+                pass
 
 
 async def init_db():

backend/app/core/logging_filters.py

@@ -1,13 +1,16 @@
 """Logging filters for the Bambuddy log pipeline.
 
-Currently houses a single filter that keeps only state-changing HTTP methods
-in the file-side uvicorn access log. See ``WriteRequestsOnlyFilter`` for the
-why; this lives in its own module so the test suite can import it without
-pulling in ``backend.app.main``'s entire startup graph.
+Holds two filters: ``WriteRequestsOnlyFilter`` keeps the file-side
+uvicorn access log focused on state-changing HTTP methods, and
+``CancelledPoolNoiseFilter`` drops SQLAlchemy connection-pool log noise
+caused by Starlette's ``BaseHTTPMiddleware`` cancellation propagation
+(see the filter's docstring for details). Both live here so tests can
+import them without pulling in ``backend.app.main``'s startup graph.
 """
 
 from __future__ import annotations
 
+import asyncio
 import logging
 
 
@@ -44,3 +47,63 @@ class WriteRequestsOnlyFilter(logging.Filter):
     def filter(self, record: logging.LogRecord) -> bool:  # noqa: A003 — stdlib API name
         message = record.getMessage()
         return any(token in message for token in self._WRITE_VERB_TOKENS)
+
+
+class CancelledPoolNoiseFilter(logging.Filter):
+    """Drop SQLAlchemy connection-pool log records driven by request cancellation.
+
+    Starlette's ``BaseHTTPMiddleware`` (used under the hood by FastAPI's
+    ``@app.middleware("http")`` decorator) cancels the inner task scope when a
+    client disconnects mid-request. The cancellation propagates into
+    SQLAlchemy's connection-pool cleanup and surfaces as two distinct ERROR
+    records — both expected on disconnect, neither actionable for the user:
+
+    1. ``Exception terminating connection ... CancelledError`` — fires every
+       time ``do_terminate`` is interrupted by the same cancel scope that's
+       unwinding the request. The ``CancelledError`` traceback always
+       attributes the cancel to ``BaseHTTPMiddleware.call_next``.
+
+    2. ``The garbage collector is trying to clean up non-checked-in
+       connection`` — fires later when the GC reclaims the session that
+       couldn't return its connection to the pool because of (1). It's
+       symptomatic of the cancellation, not a separate bug.
+
+    These pile up under heavy upload load (long multipart uploads where the
+    client times out before the server's response). Real connection-pool
+    issues — pool exhaustion, broken connections from network hiccups, etc.
+    — surface through DIFFERENT messages and a non-cancellation
+    ``exc_info`` chain, so they keep flowing through this filter unchanged.
+
+    Attach to ``logging.getLogger("sqlalchemy.pool")`` (and only there).
+    """
+
+    _GC_CLEANUP_PREFIX = "The garbage collector is trying to clean up non-checked-in connection"
+    _TERMINATE_PREFIX = "Exception terminating connection"
+
+    @staticmethod
+    def _has_cancelled_in_chain(exc: BaseException | None) -> bool:
+        """True if `exc` is `CancelledError` or has one in its cause chain."""
+        seen: set[int] = set()
+        cur: BaseException | None = exc
+        while cur is not None and id(cur) not in seen:
+            seen.add(id(cur))
+            if isinstance(cur, asyncio.CancelledError):
+                return True
+            cur = cur.__cause__ or cur.__context__
+        return False
+
+    def filter(self, record: logging.LogRecord) -> bool:  # noqa: A003 — stdlib API name
+        message = record.getMessage()
+        # GC-cleanup records have no exc_info — match by prefix only. Always
+        # symptomatic of the cancellation cascade, never independently useful.
+        if message.startswith(self._GC_CLEANUP_PREFIX):
+            return False
+        # Terminate-connection records carry a traceback; only drop those
+        # that are cancellation-driven. A real terminate failure (broken
+        # connection, network hiccup) keeps a non-CancelledError exc_info
+        # chain and surfaces normally.
+        if message.startswith(self._TERMINATE_PREFIX) and record.exc_info:
+            exc = record.exc_info[1]
+            if self._has_cancelled_in_chain(exc):
+                return False
+        return True

backend/app/main.py

@@ -288,7 +288,10 @@ if app_settings.log_to_file:
     # for exactly this reason. Filtered to write methods only
     # (POST/PUT/PATCH/DELETE) so the high-volume status-poll GETs from the
     # frontend don't churn the rotation window faster than it's useful.
-    from backend.app.core.logging_filters import WriteRequestsOnlyFilter
+    from backend.app.core.logging_filters import (
+        CancelledPoolNoiseFilter,
+        WriteRequestsOnlyFilter,
+    )
 
     uvicorn_access_logger = logging.getLogger("uvicorn.access")
     uvicorn_access_logger.addHandler(file_handler)
@@ -299,6 +302,13 @@ if app_settings.log_to_file:
     # ID column as the application logs they correlate with.
     uvicorn_access_logger.addFilter(TraceIDFilter())
 
+    # Drop SQLAlchemy connection-pool log noise that's caused by Starlette's
+    # BaseHTTPMiddleware cancelling the inner task scope on client
+    # disconnect (#1112). The cancel-safe `get_db` already prevents the
+    # underlying transaction leak; this filter only suppresses the residual
+    # log records that pre-existing pools still emit during their cleanup.
+    logging.getLogger("sqlalchemy.pool").addFilter(CancelledPoolNoiseFilter())
+
 # Reduce noise from third-party libraries in production
 if not app_settings.debug:
     logging.getLogger("sqlalchemy.engine").setLevel(logging.WARNING)

backend/tests/unit/test_cancelled_pool_filter.py

@@ -0,0 +1,76 @@
+"""Tests for the SQLAlchemy connection-pool cancellation noise filter (#1112)."""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+
+from backend.app.core.logging_filters import CancelledPoolNoiseFilter
+
+
+def _make_record(message: str, *, exc: BaseException | None = None) -> logging.LogRecord:
+    """Build a `LogRecord` carrying `message` (no positional args) and
+    optionally an `exc_info` tuple holding `exc`."""
+    record = logging.LogRecord(
+        name="sqlalchemy.pool.impl.AsyncAdaptedQueuePool",
+        level=logging.ERROR,
+        pathname=__file__,
+        lineno=0,
+        msg=message,
+        args=(),
+        exc_info=(type(exc), exc, exc.__traceback__) if exc is not None else None,
+    )
+    return record
+
+
+class TestCancelledPoolNoiseFilter:
+    """Drops the cancellation cascade, keeps real pool errors visible."""
+
+    def test_drops_terminate_with_cancelled_exc(self):
+        cancel = asyncio.CancelledError("Cancelled via cancel scope")
+        record = _make_record("Exception terminating connection <ABC>", exc=cancel)
+        assert CancelledPoolNoiseFilter().filter(record) is False
+
+    def test_drops_gc_cleanup_record(self):
+        # GC cleanup messages have no exc_info attached — match by prefix.
+        record = _make_record("The garbage collector is trying to clean up non-checked-in connection <ABC>")
+        assert CancelledPoolNoiseFilter().filter(record) is False
+
+    def test_keeps_terminate_with_real_oserror(self):
+        """A genuine connection-terminate failure (network hiccup, broken
+        socket) carries a non-cancellation exc_info chain. That's a real
+        problem the user should see — must NOT be dropped."""
+        oserr = OSError("broken pipe")
+        record = _make_record("Exception terminating connection <ABC>", exc=oserr)
+        assert CancelledPoolNoiseFilter().filter(record) is True
+
+    def test_keeps_terminate_without_exc_info(self):
+        """If for any reason `exc_info` is missing on a terminate record,
+        keep it — only filter when we have positive evidence it's the
+        cancellation cascade."""
+        record = _make_record("Exception terminating connection <ABC>")
+        assert CancelledPoolNoiseFilter().filter(record) is True
+
+    def test_keeps_unrelated_pool_message(self):
+        """Other pool messages (pool size warnings, etc.) keep flowing."""
+        record = _make_record("Pool size has been exceeded; will spawn overflow")
+        assert CancelledPoolNoiseFilter().filter(record) is True
+
+    def test_drops_when_cancelled_is_in_cause_chain(self):
+        """Real-world traceback: SQLAlchemy wraps the CancelledError in a
+        chained exception. The filter walks `__cause__`/`__context__` so a
+        chained CancelledError still counts."""
+        cancel = asyncio.CancelledError()
+        wrapper = RuntimeError("terminate failed")
+        wrapper.__cause__ = cancel
+        record = _make_record("Exception terminating connection <ABC>", exc=wrapper)
+        assert CancelledPoolNoiseFilter().filter(record) is False
+
+    def test_handles_self_referential_cause_chain(self):
+        """Defensive: malformed exception chains (rare but possible) must
+        not loop forever — the `seen` set guards against it."""
+        a = RuntimeError("a")
+        a.__cause__ = a  # pathological
+        record = _make_record("Exception terminating connection <ABC>", exc=a)
+        # Doesn't loop, doesn't raise, returns True (no CancelledError found).
+        assert CancelledPoolNoiseFilter().filter(record) is True

backend/tests/unit/test_get_db_cancel_safety.py

@@ -0,0 +1,163 @@
+"""Tests for `get_db` cancel-safety (#1112).
+
+Starlette's BaseHTTPMiddleware cancels the inner task scope when a
+client disconnects mid-request. Pre-fix `get_db` only caught `Exception`
+(not `BaseException`), so `CancelledError` skipped the rollback path —
+the SQLite write lock stayed held until the connection was eventually
+GC'd, producing the "database is locked" cascade in @Carter3DP's
+support package on #1112.
+
+The fix:
+  1. Catch `BaseException` so `CancelledError` triggers rollback.
+  2. `asyncio.shield` rollback + close so the cleanup completes even
+     when the await is cancelled by the same cancel scope.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from backend.app.core import database
+
+
+class _FakeSession:
+    """Minimal async-context-manager stand-in for `AsyncSession`.
+
+    Records which lifecycle methods were invoked so tests can assert on
+    the cleanup order without a real engine / DB file.
+    """
+
+    def __init__(self):
+        self.commit = AsyncMock(name="commit")
+        self.rollback = AsyncMock(name="rollback")
+        self.close = AsyncMock(name="close")
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, exc_type, exc, tb):
+        return False  # don't suppress
+
+
+@pytest.fixture
+def fake_session_factory(monkeypatch):
+    """Patch `database.async_session` to yield a fresh `_FakeSession`."""
+    session = _FakeSession()
+    monkeypatch.setattr(database, "async_session", lambda: session)
+    return session
+
+
+async def _consume_get_db(action):
+    """Drive `get_db` like FastAPI's dependency machinery does:
+    enter the async generator, run `action(session)`, then advance to
+    completion. Returns the entered session."""
+    gen = database.get_db()
+    session = await gen.__anext__()
+    try:
+        await action(session)
+    except StopAsyncIteration:
+        return session
+    # Advance to the end so the generator's finally runs.
+    try:
+        await gen.__anext__()
+    except StopAsyncIteration:
+        pass
+    return session
+
+
+class TestCancelSafety:
+    """Pin the cancel-safety contract end-to-end."""
+
+    @pytest.mark.asyncio
+    async def test_commit_on_clean_exit(self, fake_session_factory):
+        session = fake_session_factory
+
+        async def noop(_s):
+            pass
+
+        await _consume_get_db(noop)
+
+        session.commit.assert_awaited_once()
+        session.rollback.assert_not_awaited()
+        session.close.assert_awaited_once()
+
+    @pytest.mark.asyncio
+    async def test_rollback_on_regular_exception(self, fake_session_factory):
+        session = fake_session_factory
+
+        gen = database.get_db()
+        await gen.__anext__()
+        with pytest.raises(ValueError):
+            await gen.athrow(ValueError("route handler bug"))
+
+        session.commit.assert_not_awaited()
+        session.rollback.assert_awaited_once()
+        session.close.assert_awaited_once()
+
+    @pytest.mark.asyncio
+    async def test_rollback_on_cancelled_error(self, fake_session_factory):
+        """The actual #1112 fix: CancelledError must NOT skip the rollback.
+        Pre-fix `except Exception` caught nothing because CancelledError
+        is a BaseException, not an Exception."""
+        session = fake_session_factory
+
+        gen = database.get_db()
+        await gen.__anext__()
+        with pytest.raises(asyncio.CancelledError):
+            await gen.athrow(asyncio.CancelledError("client disconnected"))
+
+        session.commit.assert_not_awaited()
+        session.rollback.assert_awaited_once()
+        session.close.assert_awaited_once()
+
+    @pytest.mark.asyncio
+    async def test_close_runs_even_if_rollback_raises(self, fake_session_factory):
+        """A failing rollback (broken connection during cancellation) must
+        not prevent `close` from running — otherwise the pool would never
+        reclaim the connection."""
+        session = fake_session_factory
+        session.rollback.side_effect = OSError("broken pipe during rollback")
+
+        gen = database.get_db()
+        await gen.__anext__()
+        with pytest.raises(asyncio.CancelledError):
+            await gen.athrow(asyncio.CancelledError())
+
+        session.rollback.assert_awaited_once()
+        session.close.assert_awaited_once()
+
+    @pytest.mark.asyncio
+    async def test_close_failure_does_not_propagate(self, fake_session_factory):
+        """A failing close on the clean-exit path must not raise out of
+        `get_db` — the request already succeeded."""
+        session = fake_session_factory
+        session.close.side_effect = OSError("close failed")
+
+        async def noop(_s):
+            pass
+
+        # Must not raise.
+        await _consume_get_db(noop)
+
+        session.commit.assert_awaited_once()
+        session.close.assert_awaited_once()
+
+    @pytest.mark.asyncio
+    async def test_rollback_uses_shield(self, fake_session_factory):
+        """Cancellation arriving DURING rollback must not abort the
+        rollback — `asyncio.shield` keeps it running. Verify the call
+        path goes through `shield` so future refactors don't silently
+        drop the protection."""
+        # The fixture wires the fake session into `database.async_session`;
+        # we don't need the local handle here.
+        with patch.object(asyncio, "shield", wraps=asyncio.shield) as shield:
+            gen = database.get_db()
+            await gen.__anext__()
+            with pytest.raises(asyncio.CancelledError):
+                await gen.athrow(asyncio.CancelledError())
+
+        # rollback + close both shielded.
+        assert shield.call_count == 2