fix(db): failed commits drop the response's background work; read-only finalization keeps loaded rows; savepoint rollback clears the write marker

CHANGELOG.md

@@ -62,6 +62,18 @@ All notable changes to this project are documented in this file. The format is b
   one error per field, scoped by `X-Inertia-Error-Bag`, instead of rendering
   the 422 "invalid parameters" Error page over the form. GETs and non-Inertia
   requests, and invalid path/query values, keep the 422 page / JSON.
+- **Behaviour change (#807):** a failed request commit aborts the original
+  response and its background tasks before returning a JSON 500, preventing
+  follow-up writes from persisting after the failure. The 500 is still sent
+  when inner middleware or a dependency teardown swallows the abort.
+- **Behaviour change (#808):** read-only request finalization releases the
+  connection without expiring loaded ORM attributes, keeping them usable in
+  streaming bodies and background tasks without implicit async IO.
+- **Behaviour change (#809):** write markers follow savepoint release and
+  rollback; requests whose only writes were rolled back no longer commit or
+  run outer `on_commit` callbacks. A commit issued mid-request still runs the
+  request's `on_commit` callbacks at finalization, and `mark_written` no longer
+  begins a transaction.
 - Bulk-DML audit prefetch preserves statement options, including loader criteria,
   so it records only the rows selected by the UPDATE or DELETE (#810).
 - ORM restores now emit `restored` audit entries with translated action labels

docs/database/sessions.md

@@ -1,24 +1,33 @@
 # Session lifecycle
 
-Each request opens exactly one `AsyncSession`. The framework commits only when there are pending writes; read-only requests rollback. **Service code should not call `session.commit()` directly.**
+Each request opens exactly one `AsyncSession`. The framework commits only when there are pending writes; read-only requests end their transaction with a rollback that preserves loaded attributes. **Service code should not call `session.commit()` directly.**
 
 ## The `get_db` dependency
 
-```python
-# simple_module_db.deps
-async def get_db(request: Request) -> AsyncGenerator[AsyncSession, None]:
-    factory = request.app.state.sm.db.session_factory
-    async with factory() as session:
-        try:
-            yield session
-            if _has_pending_writes(session):
-                await session.commit()
-            else:
-                await session.rollback()
-        except Exception:
-            await session.rollback()
-            raise
-```
+`get_db` yields a request session and enlists it with
+`CommitBeforeResponseMiddleware`. At `http.response.start`, before headers or
+body bytes leave the server, the middleware commits pending writes or closes a
+read-only transaction. Closing rolls back and releases the connection without
+expiring or detaching loaded ORM objects. A streaming body or background task
+can read already-loaded attributes without database IO; unloaded attributes
+still require an explicit async load.
+Those objects keep the values they were loaded with: a later query in the same
+session does not refresh them unless it uses `populate_existing()`.
+
+If commit fails, the framework rolls back and unwinds the original response,
+including its streaming body and background tasks, then sends a JSON 500.
+Those tasks do not run for a failed request commit (#807).
+
+The dependency finalizes again on exit to persist writes made during a
+successful stream or background task. The session remains usable after
+read-only finalization, including for later writes (#808). Without the
+middleware, finalization happens when the dependency exits.
+
+Read-only finalization closes the transaction (`SessionTransaction.close()`)
+rather than calling `session.rollback()`. Session listeners see
+`after_transaction_end` for each closed transaction, but **not**
+`after_rollback` or `after_soft_rollback`; a listener that must observe the end
+of every request transaction should hook `after_transaction_end`.
 
 Usage in endpoints:
 
@@ -51,28 +60,36 @@ OrderServiceDep = Annotated[OrderService, Depends(_order_service)]
 
 ## The `has_writes` flag
 
-To distinguish "this request wrote something" from "this request only read", an `after_flush` event listener sets `session.info["has_writes"] = True`. The dependency checks:
-
-```python
-if session.info.get("has_writes") or session.new or session.dirty or session.deleted:
-    await session.commit()
-else:
-    await session.rollback()
-```
-
-Why both? `session.new/dirty/deleted` show **pending** writes that haven't flushed yet; `has_writes` catches the case where writes flushed but were already synced. Together they cover every path to "we touched the DB with intent to change it."
+ORM flushes and session-executed INSERT/UPDATE/DELETE statements mark the
+current transaction as written. Raw SQL writes require `mark_written(session)`.
+The finalizer checks both this marker and pending `new`/`dirty`/`deleted`
+objects, so an already-flushed write still commits.
+
+Write markers follow savepoint boundaries: releasing a savepoint merges its
+marker into its parent; rolling it back discards only its marker. A released
+child's writes still belong to its parent's rollback. If all writes were
+rolled back, finalization takes the read-only path and discards outer
+`on_commit` callbacks (#809). Callbacks registered inside a rolled-back
+savepoint are also discarded, while callbacks outside it survive if other
+writes commit. A full rollback clears all write markers and callbacks.
+
+A commit issued mid-request (fastapi-users commits inside `create` and
+`update`) keeps the session on the write path: finalization still commits, and
+`on_commit` callbacks queued before or after that commit run then. A later
+savepoint rollback does not take that back. `mark_written` never begins a
+transaction; with none open it only sets the flag.
 
 ## Why not always commit?
 
 - **Cheaper.** A rollback on a clean session is a single Postgres message; a commit with no writes still goes through two-phase round-trips on some drivers.
 - **Observability.** The write/read distinction is useful in profiling and logs — you can answer "how many requests mutated data?" without guessing.
-- **Safety.** If a GET handler accidentally stages a write (rare, but happens with `session.merge`), the rollback prevents silent persistence.
+- A handler that stages a write, including a GET handler, takes the write path and commits. Read-only finalization is not a guard against accidental writes.
 
 ## Why service code shouldn't commit
 
-The per-request session is **owned** by the dependency. If your service calls `commit()` mid-request, two problems arise:
+The per-request transaction is **owned** by the framework. If your service calls `commit()` mid-request, two problems arise:
 
-1. The dependency commits again on exit — harmless but confusing.
+1. It bypasses commit-before-response finalization and its post-commit callbacks.
 2. If a later step in the request fails, the partial writes are already persisted and can't be rolled back.
 
 Instead, **flush** when you need a DB-assigned value (e.g. an auto-generated `id`):
@@ -196,7 +213,7 @@ For endpoint tests, use `client` or `authenticated_client` — they share the sa
 | Situation | Call |
 |---|---|
 | Inside an endpoint handler / service, need DB-assigned id | `await session.flush()` |
-| Inside an endpoint handler / service, done with work | nothing — dependency commits on exit |
+| Inside an endpoint handler / service, done with work | nothing — middleware commits before response start |
 | Outside a request (worker, CLI, lifespan) | explicit `async with session.begin():` |
 | A handler that reads-only but wants to persist an audit row | stage the row; the dependency detects pending writes and commits |
 | A handler that must explicitly discard partial changes | `raise`ing propagates to the dependency, which rolls back |

docs/framework-conventions.md

@@ -316,7 +316,7 @@ Each request opens one session:
   await session.execute(text("UPDATE orders_order SET ..."))
   mark_written(session)
   ```
-- Read-only requests exit via rollback — cheaper, and keeps the session out of write-side profiling.
+- Read-only requests exit via rollback (loaded objects are not expired) — cheaper, and keeps the session out of write-side profiling.
 - Exceptions always rollback.
 
 Service code should not call `session.commit()` directly. Flush for intermediate reads if you need DB-assigned values, then let the framework commit. The injected

framework/db/simple_module_db/deps.py

@@ -20,7 +20,8 @@ async def get_db(request: Request) -> AsyncGenerator[RequestSession, None]:
     """Yield an async database session, auto-closing on exit.
 
     Commits only when the session has pending writes (``new``, ``dirty``,
-    or ``deleted`` objects); read-only handlers exit via ``rollback``.
+    or ``deleted`` objects); read-only handlers close their transaction
+    without expiring loaded objects.
 
     The commit itself normally happens in
     :class:`~simple_module_db.transaction.CommitBeforeResponseMiddleware`,

framework/db/simple_module_db/listeners.py

@@ -24,6 +24,7 @@
     _is_purge,
     _mark_dml_written,
     _mark_session_written,
+    attach_write_listeners,
     hard_delete,
     mark_written,
 )
@@ -140,6 +141,7 @@ def attach_session_listeners(session_class: type[Session]) -> None:
         event.listen(session_class, name, fn)
     # The legacy bulk_* methods reach no session event: guard them directly.
     install_bulk_guard(session_class)
+    attach_write_listeners(session_class)
     setattr(session_class, _LISTENERS_ATTACHED, True)
 
 

framework/db/simple_module_db/transaction.py

@@ -16,28 +16,31 @@
 seeds, provisioning scripts, integration tests, single-run k8s jobs — gets the
 guarantee without changing a line of endpoint code.
 
-A session is finalized exactly once. Whichever runs first wins and the other
-becomes a no-op, so the framework stays correct when the middleware is absent
-(a bare ``get_db``, a WebSocket, a test calling the dependency directly) and on
-the error path, where FastAPI unwinds the dependency — rolling it back —
-*before* the error response is sent.
+A session's pending work is finalized once. Whichever runs first wins and the
+other becomes a no-op (``finalize_session`` re-arms for later writes), so the
+framework stays correct when the middleware is absent (a bare ``get_db``, a
+WebSocket, a test calling the dependency directly) and on the error path, where
+FastAPI unwinds the dependency — rolling it back — *before* the error response
+is sent.
 """
 
 from __future__ import annotations
 
+import contextlib
 import json
 import logging
 import time
 from collections.abc import MutableMapping
 from typing import Any
 
 from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm import Session
 
 from simple_module_db.callbacks import (
     discard_on_commit_callbacks,
     run_on_commit_callbacks,
 )
-from simple_module_db.listeners import SESSION_HAS_WRITES_KEY
+from simple_module_db.writes import SESSION_HAS_WRITES_KEY, clear_written
 
 logger = logging.getLogger("simple_module.db")
 
@@ -52,6 +55,16 @@
 _INTERNAL_ERROR_BODY = json.dumps({"detail": "Internal Server Error"}).encode()
 
 
+class _ResponseAbortedError(Exception):
+    """Unwind the original response before replacing it after commit failure."""
+
+
+def _close_read_only_transaction(session: Session) -> None:
+    # Transaction.close rolls back and releases connections without expiring ORM state.
+    while transaction := session.get_nested_transaction() or session.get_transaction():
+        transaction.close()
+
+
 def _elapsed_ms(session: AsyncSession) -> float:
     start = session.info.get(SESSION_START_KEY)
     return round((time.perf_counter() - start) * 1000, 2) if start else 0.0
@@ -60,7 +73,7 @@ def _elapsed_ms(session: AsyncSession) -> float:
 def _has_pending(session: AsyncSession) -> bool:
     """Whether ``session`` holds work that still needs committing.
 
-    ``has_writes`` is stamped by the after_flush listener and survives the flush
+    ``has_writes`` is stamped by flushes and session DML (see ``writes``) and survives the flush
     emptying ``session.new``/``.dirty``/``.deleted`` — the raw collections alone
     would report a flushed-but-uncommitted write as read-only.
     """
@@ -76,7 +89,7 @@ def _settle(session: AsyncSession) -> None:
     would try to re-commit a session whose work has already landed.
     """
     session.info[_FINALIZED_KEY] = True
-    session.info.pop(SESSION_HAS_WRITES_KEY, None)
+    clear_written(session)
 
 
 async def finalize_session(session: AsyncSession) -> None:
@@ -91,10 +104,9 @@ async def finalize_session(session: AsyncSession) -> None:
     never commit. So each call commits whatever is pending *at that moment*, and
     ``get_db``'s exit code still runs afterwards to catch the rest.
 
-    Read-only handlers exit via ``rollback`` — one round-trip cheaper than
-    ``commit``, and it keeps read-only queries from showing up as writes in
-    query logs / ``pg_stat_statements``. A repeat call with nothing new pending
-    returns immediately rather than paying for a second rollback.
+    Read-only handlers close their transaction, rolling back and releasing the
+    connection without expiring loaded objects needed by streams or tasks.
+    A repeat call with nothing new pending returns immediately.
 
     On commit failure the session is rolled back before the error propagates,
     so the caller never has to reason about a half-finalized session.
@@ -104,7 +116,7 @@ async def finalize_session(session: AsyncSession) -> None:
             return
         _settle(session)
         discard_on_commit_callbacks(session)
-        await session.rollback()
+        await session.run_sync(_close_read_only_transaction)
         logger.debug(
             "db.session.read_only",
             extra={"operation": "read_only_rollback", "db_duration_ms": _elapsed_ms(session)},
@@ -178,14 +190,15 @@ async def __call__(self, scope, receive, send) -> None:
 
         sessions: list[AsyncSession] = []
         scope.setdefault("state", {})[REQUEST_SESSIONS_KEY] = sessions
+
         aborted = False
 
         async def send_wrapper(message) -> None:
             nonlocal aborted
             if aborted:
-                # The response we replaced is still streaming its body into a
-                # channel we've already closed off. Drop it.
-                return
+                # Inner code swallowed the abort and kept sending (an error
+                # handler's own response, more body): refuse it again.
+                raise _ResponseAbortedError
             if message["type"] == "http.response.start" and sessions:
                 try:
                     # A request normally has exactly one session (FastAPI caches
@@ -197,15 +210,21 @@ async def send_wrapper(message) -> None:
                     for session in sessions:
                         await finalize_session(session)
                 except Exception:
-                    aborted = True
                     logger.exception(
                         "db.session.commit_failed", extra={"operation": "commit_failed"}
                     )
-                    await _send_internal_error(send)
-                    return
+                    aborted = True
+                    raise _ResponseAbortedError from None
             await send(message)
 
-        await self.app(scope, receive, send_wrapper)
+        # ``suppress`` also strips the signal out of the ExceptionGroup a
+        # StreamingResponse may wrap it in (Python 3.12+).
+        with contextlib.suppress(_ResponseAbortedError):
+            await self.app(scope, receive, send_wrapper)
+        # Checked after a normal return too: an inner ``except Exception`` or a
+        # yield-dependency teardown can swallow the signal, leaving no response.
+        if aborted:
+            await _send_internal_error(send)
 
 
 async def _send_internal_error(send) -> None: