diff --git a/CHANGELOG.md b/CHANGELOG.md index 25668e7aa..621162eb3 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/docs/database/sessions.md b/docs/database/sessions.md index f63c911a1..f477e3b71 100644 --- a/docs/database/sessions.md +++ b/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 | diff --git a/docs/framework-conventions.md b/docs/framework-conventions.md index 609b38dae..1f0eed7e0 100644 --- a/docs/framework-conventions.md +++ b/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 diff --git a/framework/db/simple_module_db/deps.py b/framework/db/simple_module_db/deps.py index 4fbf39ffe..c2dc3cc7f 100644 --- a/framework/db/simple_module_db/deps.py +++ b/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`, diff --git a/framework/db/simple_module_db/listeners.py b/framework/db/simple_module_db/listeners.py index e13eec735..f0be66d84 100644 --- a/framework/db/simple_module_db/listeners.py +++ b/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) diff --git a/framework/db/simple_module_db/transaction.py b/framework/db/simple_module_db/transaction.py index 971872a25..8a3cae7f8 100644 --- a/framework/db/simple_module_db/transaction.py +++ b/framework/db/simple_module_db/transaction.py @@ -16,15 +16,17 @@ 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 @@ -32,12 +34,13 @@ 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: diff --git a/framework/db/simple_module_db/writes.py b/framework/db/simple_module_db/writes.py index 583c1467f..74690b92a 100644 --- a/framework/db/simple_module_db/writes.py +++ b/framework/db/simple_module_db/writes.py @@ -12,15 +12,25 @@ from __future__ import annotations +from sqlalchemy import event from sqlalchemy import inspect as sa_inspect from sqlalchemy.ext.asyncio import AsyncSession -from sqlalchemy.orm import ORMExecuteState, Session +from sqlalchemy.orm import ORMExecuteState, Session, SessionTransaction from simple_module_db.mixins import SoftDeleteMixin # Key on ``Session.info`` distinguishing read-only requests from write requests # after flush has cleared ``session.new/.dirty/.deleted``. SESSION_HAS_WRITES_KEY = "has_writes" +# Open transactions (root or savepoint) holding writes that have not committed. +_WRITE_TRANSACTIONS_KEY = "sm_db_write_transactions" +# Writes no open transaction owns: a commit the caller issued mid-request +# (fastapi-users commits inside ``update``/``create``) or a mark made with no +# transaction begun. Only ``clear_written`` or a full rollback drops it, so a +# later savepoint rollback cannot discard the on-commit work of durable writes. +_UNTRACKED_WRITES_KEY = "sm_db_untracked_writes" +# Class marker for the transaction listeners; see ``attach_write_listeners``. +_WRITE_LISTENERS_ATTACHED = "_sm_write_listeners_attached" # Key on ``Session.info`` holding ``id()`` of instances the caller asked to # hard-delete, consumed by the before_flush soft-delete rewrite. @@ -37,8 +47,76 @@ def mark_written(session: Session | AsyncSession) -> None: Without it that write is rolled back at request end, silently (GH #336). Accepts either an ``AsyncSession`` or the underlying sync ``Session``. + The mark belongs to the innermost open transaction: releasing a savepoint + hands it to the parent, rolling one back discards it. It never begins a + transaction; with none open the flag is simply set. """ - session.info[SESSION_HAS_WRITES_KEY] = True + sync_session = session.sync_session if isinstance(session, AsyncSession) else session + if not _class_wired(type(sync_session)) and not event.contains( + sync_session, "after_commit", _merge_written_transaction + ): + # A session class the framework never wired (a bare test session). + event.listen(sync_session, "after_commit", _merge_written_transaction) + event.listen(sync_session, "after_soft_rollback", _discard_written_transaction) + info = sync_session.info + transaction = sync_session.get_nested_transaction() or sync_session.get_transaction() + if transaction is None: + info[_UNTRACKED_WRITES_KEY] = True + else: + info.setdefault(_WRITE_TRANSACTIONS_KEY, set()).add(transaction) + info[SESSION_HAS_WRITES_KEY] = True + + +def clear_written(session: Session | AsyncSession) -> None: + """Forget every write mark: the session's work was finalized.""" + for key in (SESSION_HAS_WRITES_KEY, _WRITE_TRANSACTIONS_KEY, _UNTRACKED_WRITES_KEY): + session.info.pop(key, None) + + +def attach_write_listeners(session_class: type[Session]) -> None: + """Track write marks across commits and savepoint rollbacks; idempotent. + + Registered once per session class (``attach_session_listeners``); a + subclass inherits class-level session listeners, so it is skipped. + """ + if _class_wired(session_class): + return + event.listen(session_class, "after_commit", _merge_written_transaction) + event.listen(session_class, "after_soft_rollback", _discard_written_transaction) + setattr(session_class, _WRITE_LISTENERS_ATTACHED, True) + + +def _class_wired(session_class: type) -> bool: + return any(base.__dict__.get(_WRITE_LISTENERS_ATTACHED) for base in session_class.__mro__) + + +def _merge_written_transaction(session: Session) -> None: + # ``after_commit`` fires for the root and for released savepoints, while + # the committing transaction is still the current one. + transaction = session.get_nested_transaction() or session.get_transaction() + if transaction is None: + return + written: set[SessionTransaction] = session.info.get(_WRITE_TRANSACTIONS_KEY, set()) + if transaction not in written: + return + written.remove(transaction) + if transaction.parent is None: + # Durable now, but its on-commit work still waits for finalization. + session.info[_UNTRACKED_WRITES_KEY] = True + else: + written.add(transaction.parent) + + +def _discard_written_transaction(session: Session, transaction: SessionTransaction) -> None: + if transaction.parent is None: + clear_written(session) + return + written: set[SessionTransaction] = session.info.get(_WRITE_TRANSACTIONS_KEY, set()) + if transaction not in written: + return + written.remove(transaction) + if not written and not session.info.get(_UNTRACKED_WRITES_KEY): + session.info.pop(SESSION_HAS_WRITES_KEY, None) def _mark_session_written(session: Session, flush_context: object) -> None: @@ -48,7 +126,7 @@ def _mark_session_written(session: Session, flush_context: object) -> None: — flush empties those sets — so we stash a tag on ``session.info`` that survives the rest of the request. """ - session.info[SESSION_HAS_WRITES_KEY] = True + mark_written(session) def _mark_dml_written(execute_state: ORMExecuteState) -> None: @@ -65,7 +143,7 @@ def _mark_dml_written(execute_state: ORMExecuteState) -> None: pay for a commit. """ if execute_state.is_insert or execute_state.is_update or execute_state.is_delete: - execute_state.session.info[SESSION_HAS_WRITES_KEY] = True + mark_written(execute_state.session) async def hard_delete(session: AsyncSession, instance: object) -> None: diff --git a/framework/db/tests/test_db_response_finalization.py b/framework/db/tests/test_db_response_finalization.py new file mode 100644 index 000000000..102e593d7 --- /dev/null +++ b/framework/db/tests/test_db_response_finalization.py @@ -0,0 +1,243 @@ +"""Response-start finalization must preserve reads and suppress aborted work.""" + +from types import SimpleNamespace + +import httpx +import pytest +from _models import _TxnBase, _TxnThing +from fastapi import BackgroundTasks, Depends, FastAPI +from fastapi.responses import StreamingResponse +from simple_module_db import RequestSession +from simple_module_db.deps import get_db +from simple_module_db.listeners import register_listeners +from simple_module_db.session import init_db +from simple_module_db.transaction import CommitBeforeResponseMiddleware, finalize_session +from sqlalchemy import event, select + + +@pytest.fixture +async def state(tmp_path): + state = init_db(f"sqlite+aiosqlite:///{tmp_path / 'finalization.db'}") + register_listeners(state) + async with state.engine.begin() as connection: + await connection.run_sync(_TxnBase.metadata.create_all) + try: + yield state + finally: + await state.engine.dispose() + + +def build_app(state): + app = FastAPI() + app.state.sm = SimpleNamespace(db=state) + app.add_middleware(CommitBeforeResponseMiddleware) + return app + + +@pytest.mark.parametrize("streaming", [False, True]) +async def test_commit_failure_aborts_original_response(state, monkeypatch, streaming): + app = build_app(state) + ran = [] + real_commit = RequestSession.commit + + async def fail_once(session): + monkeypatch.setattr(RequestSession, "commit", real_commit) + raise RuntimeError("commit failed") + + monkeypatch.setattr(RequestSession, "commit", fail_once) + + @app.post("/") + async def endpoint(bg: BackgroundTasks, db: RequestSession = Depends(get_db)): + db.add(_TxnThing(name="endpoint")) + await db.flush() + + async def write(name): + ran.append(name) + db.add(_TxnThing(name=name)) + await db.flush() + + async def body(): + await write("stream") + yield b"original" + + bg.add_task(write, "background") + return StreamingResponse(body(), background=bg) if streaming else {} + + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), base_url="http://test" + ) as client: + response = await client.post("/") + assert response.status_code == 500 + assert response.json() == {"detail": "Internal Server Error"} + assert ran == [] + async with state.session_factory() as db: + assert (await db.execute(select(_TxnThing))).scalars().all() == [] + + +async def test_read_only_response_keeps_loaded_attributes_and_releases_connection(state): + async with state.session_factory() as db: + db.add(_TxnThing(name="hello")) + await db.commit() + app = build_app(state) + observed = [] + checked_out = [] + event.listen(state.engine.sync_engine, "checkout", lambda *args: checked_out.append(1)) + event.listen(state.engine.sync_engine, "checkin", lambda *args: checked_out.pop()) + + @app.get("/") + async def endpoint(bg: BackgroundTasks, db: RequestSession = Depends(get_db)): + row = (await db.execute(select(_TxnThing))).scalar_one() + + async def task(): + observed.append(row.name) + row.name = "updated in background" + await db.flush() + + async def body(): + assert checked_out == [] + assert not db.in_transaction() + yield row.name.encode() + + bg.add_task(task) + return StreamingResponse(body(), background=bg) + + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), base_url="http://test" + ) as client: + response = await client.get("/") + assert response.content == b"hello" + assert observed == ["hello"] + async with state.session_factory() as db: + assert (await db.execute(select(_TxnThing.name))).scalar_one() == "updated in background" + + +@pytest.mark.parametrize("write_kind", ["orm", "dml", "raw"]) +@pytest.mark.parametrize("outer_write", [False, True]) +async def test_savepoint_rollback_only_discards_its_write_marker(state, write_kind, outer_write): + from simple_module_db import mark_written + from sqlalchemy import insert, text + + observed = [] + async with state.session_factory() as db: + db.on_commit(lambda: observed.append("outer")) + if outer_write: + db.add(_TxnThing(name="kept")) + await db.flush() + with pytest.raises(ValueError): + async with db.begin_nested(): + db.on_commit(lambda: observed.append("rolled back")) + if write_kind == "orm": + db.add(_TxnThing(name="gone")) + await db.flush() + elif write_kind == "dml": + await db.execute(insert(_TxnThing).values(name="gone")) + else: + await db.execute(text("INSERT INTO txn_test_thing (name) VALUES ('gone')")) + mark_written(db) + raise ValueError("abort savepoint") + await finalize_session(db) + assert observed == (["outer"] if outer_write else []) + async with state.session_factory() as db: + assert (await db.execute(select(_TxnThing.name))).scalars().all() == ( + ["kept"] if outer_write else [] + ) + + +@pytest.mark.parametrize("rollback_parent", [False, True]) +async def test_released_child_write_belongs_to_parent_savepoint(state, rollback_parent): + observed = [] + async with state.session_factory() as db: + db.on_commit(lambda: observed.append("outer")) + parent = await db.begin_nested() + async with db.begin_nested(): + db.add(_TxnThing(name="child")) + await db.flush() + db.on_commit(lambda: observed.append("child")) + if rollback_parent: + await parent.rollback() + else: + await parent.commit() + await finalize_session(db) + assert observed == ([] if rollback_parent else ["outer", "child"]) + async with state.session_factory() as db: + assert (await db.execute(select(_TxnThing.name))).scalars().all() == ( + [] if rollback_parent else ["child"] + ) + + +async def test_read_only_finalize_closes_open_savepoints_and_releases_connection(state): + checked_out = [] + event.listen(state.engine.sync_engine, "checkout", lambda *args: checked_out.append(1)) + event.listen(state.engine.sync_engine, "checkin", lambda *args: checked_out.pop()) + async with state.session_factory() as db: + db.add(_TxnThing(name="seed")) + await db.commit() + async with state.session_factory() as db: + row = (await db.execute(select(_TxnThing))).scalar_one() + await db.begin_nested() + await db.begin_nested() + await db.execute(select(_TxnThing)) + db.on_commit(lambda: pytest.fail("discarded on read-only finalize")) + await finalize_session(db) + assert checked_out == [] + assert not db.in_transaction() + assert not db.sync_session.get_nested_transaction() + assert row.name == "seed" # still loaded, no implicit IO + # a later use autobegins cleanly and is rolled back by the next finalize + assert (await db.execute(select(_TxnThing.name))).scalar_one() == "seed" + assert db.in_transaction() + + +class _SwallowingMiddleware: + """An inner error handler that eats every exception, optionally answering itself.""" + + def __init__(self, app, *, answer: bool): + self.app, self.answer = app, answer + + async def __call__(self, scope, receive, send): + try: + await self.app(scope, receive, send) + except Exception: + if not self.answer: + return + try: + await send({"type": "http.response.start", "status": 503, "headers": []}) + await send({"type": "http.response.body", "body": b"inner"}) + except Exception: + return + + +@pytest.mark.parametrize("answer", [False, True]) +async def test_swallowed_abort_still_sends_one_internal_error(state, monkeypatch, answer): + async def fail(session): + raise RuntimeError("commit failed") + + monkeypatch.setattr(RequestSession, "commit", fail) + app = FastAPI() + app.state.sm = SimpleNamespace(db=state) + app.add_middleware(_SwallowingMiddleware, answer=answer) + app.add_middleware(CommitBeforeResponseMiddleware) + + @app.post("/") + async def endpoint(db: RequestSession = Depends(get_db)): + db.add(_TxnThing(name="endpoint")) + await db.flush() + return {} + + starts = [] + + async def counting(scope, receive, send): + async def counting_send(message): + if message["type"] == "http.response.start": + starts.append(message["status"]) + await send(message) + + await app(scope, receive, counting_send) + + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=counting), base_url="http://test" + ) as client: + response = await client.post("/") + assert response.status_code == 500 + assert response.json() == {"detail": "Internal Server Error"} + assert starts == [500] diff --git a/framework/db/tests/test_db_write_marks.py b/framework/db/tests/test_db_write_marks.py new file mode 100644 index 000000000..e6e59ee20 --- /dev/null +++ b/framework/db/tests/test_db_write_marks.py @@ -0,0 +1,101 @@ +"""Write marks across mid-request commits, savepoints and bare sessions (#809).""" + +import pytest +from _models import _TxnBase, _TxnThing +from simple_module_db import mark_written +from simple_module_db.listeners import register_listeners +from simple_module_db.session import init_db +from simple_module_db.transaction import finalize_session +from simple_module_db.writes import ( + SESSION_HAS_WRITES_KEY, + _merge_written_transaction, +) +from sqlalchemy import event, select, text +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker + + +@pytest.fixture +async def state(tmp_path): + state = init_db(f"sqlite+aiosqlite:///{tmp_path / 'marks.db'}") + register_listeners(state) + async with state.engine.begin() as connection: + await connection.run_sync(_TxnBase.metadata.create_all) + try: + yield state + finally: + await state.engine.dispose() + + +async def _names(state): + async with state.session_factory() as db: + return (await db.execute(select(_TxnThing.name))).scalars().all() + + +@pytest.mark.parametrize("register", ["before_commit", "after_commit"]) +async def test_mid_request_commit_still_runs_on_commit_callbacks(state, register): + """fastapi-users commits inside ``update``; the request's callbacks must still run.""" + observed = [] + async with state.session_factory() as db: + if register == "before_commit": + db.on_commit(lambda: observed.append("ran")) + db.add(_TxnThing(name="durable")) + await db.commit() + if register == "after_commit": + db.on_commit(lambda: observed.append("ran")) + # An unrelated savepoint rollback must not erase the durable write's mark. + with pytest.raises(ValueError): + async with db.begin_nested(): + db.add(_TxnThing(name="gone")) + await db.flush() + raise ValueError + await finalize_session(db) + assert SESSION_HAS_WRITES_KEY not in db.info + assert observed == ["ran"] + assert await _names(state) == ["durable"] + + +async def test_commit_without_writes_stays_read_only(state): + observed = [] + async with state.session_factory() as db: + await db.execute(select(_TxnThing)) + await db.commit() + db.on_commit(lambda: observed.append("ran")) + await finalize_session(db) + assert observed == [] + + +async def test_mark_written_does_not_begin_a_transaction(state): + async with state.session_factory() as db: + mark_written(db) + assert not db.in_transaction() + assert db.info[SESSION_HAS_WRITES_KEY] is True + async with db.begin(): # would raise "already begun" had mark_written begun one + await db.execute(text("INSERT INTO txn_test_thing (name) VALUES ('raw')")) + assert await _names(state) == ["raw"] + + +async def test_framework_sessions_use_class_listeners(state): + async with state.session_factory() as db: + db.add(_TxnThing(name="x")) + await db.flush() + assert not event.contains(db.sync_session, "after_commit", _merge_written_transaction) + assert event.contains(type(db.sync_session), "after_commit", _merge_written_transaction) + + +async def test_bare_session_tracks_savepoint_rollback(state): + """A session class the framework never wired still follows savepoints.""" + factory = async_sessionmaker(state.engine, class_=AsyncSession, expire_on_commit=False) + async with factory() as db: + await db.execute(select(_TxnThing)) + with pytest.raises(ValueError): + async with db.begin_nested(): + await db.execute(text("INSERT INTO txn_test_thing (name) VALUES ('gone')")) + mark_written(db) + raise ValueError + assert SESSION_HAS_WRITES_KEY not in db.info + async with db.begin_nested(): + await db.execute(text("INSERT INTO txn_test_thing (name) VALUES ('kept')")) + mark_written(db) + assert db.info[SESSION_HAS_WRITES_KEY] is True + await finalize_session(db) + assert await _names(state) == ["kept"] diff --git a/modules/file_storage/file_storage/visibility.py b/modules/file_storage/file_storage/visibility.py index 28f35037e..42dd93219 100644 --- a/modules/file_storage/file_storage/visibility.py +++ b/modules/file_storage/file_storage/visibility.py @@ -86,11 +86,13 @@ async def thumbnail(self, row: StoredFile, width: int) -> bytes: async def _release_connection(self, row: StoredFile) -> None: """End a read-only request transaction early, returning its connection. - ``row`` is detached first so its loaded attributes stay readable — the - rollback would otherwise expire it and the next access would try lazy - IO. A session holding writes is left alone: those commit (or roll back) - with the request as usual. ``finalize_session`` is re-armable, so any - later work in the request still opens a fresh transaction and commits. + Read-only finalization closes the transaction without expiring loaded + objects (#808), so ``row`` stays readable without lazy IO either way. + It is also detached, so no later work on this session can autoflush or + refresh it while the caller still holds it. A session holding writes is + left alone: those commit (or roll back) with the request as usual. + ``finalize_session`` is re-armable, so any later work in the request + still opens a fresh transaction and commits. """ db = self.db if db.info.get(SESSION_HAS_WRITES_KEY) or db.new or db.dirty or db.deleted: diff --git a/modules/tenants/tenants/_owner_guard.py b/modules/tenants/tenants/_owner_guard.py index a60b9f8e0..3e1eca125 100644 --- a/modules/tenants/tenants/_owner_guard.py +++ b/modules/tenants/tenants/_owner_guard.py @@ -15,7 +15,7 @@ from __future__ import annotations -from simple_module_db.listeners import SESSION_HAS_WRITES_KEY +from simple_module_db import mark_written from sqlalchemy import delete, exists, update from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import aliased @@ -64,6 +64,7 @@ async def remove_owner(db: AsyncSession, membership: Membership) -> bool: def _mark_written(db: AsyncSession) -> None: - # A Core-style statement is not a flush, so the request session would - # otherwise roll it back as a read-only request (#336). - db.sync_session.info[SESSION_HAS_WRITES_KEY] = True + # Goes through ``mark_written`` so the stamp follows savepoint rollbacks; a + # bare ``info[SESSION_HAS_WRITES_KEY] = True`` is dropped by an unrelated + # savepoint rollback (#809). + mark_written(db) diff --git a/skills/simple-module-database/SKILL.md b/skills/simple-module-database/SKILL.md index a6badacb7..c4396ddbd 100644 --- a/skills/simple-module-database/SKILL.md +++ b/skills/simple-module-database/SKILL.md @@ -60,10 +60,10 @@ The per-request session opens once, and **the dependency commits or rolls back f | Outcome | What `get_db` does | |---|---| | Request succeeded **and** session has pending writes | `commit()` (one transaction) | -| Request succeeded **and** session is read-only | `rollback()` (cheaper than commit on no-ops) | +| Request succeeded **and** session is read-only | closes the transaction (rollback; loaded objects stay usable, no expiry) | | Request raised | `rollback()` | -A pending-writes flag is set by an `after_flush` listener and by inspecting `session.new / dirty / deleted`. **Service code must not call `session.commit()`.** Manual commits double-commit on success and break the read-only optimization (which keeps read-only requests out of write-side profiling and pgBouncer transaction stats). +A pending-writes flag is set by flushes and session-executed INSERT/UPDATE/DELETE (follows savepoints: a rolled-back savepoint's writes no longer count) and by inspecting `session.new / dirty / deleted`. **Service code must not call `session.commit()`.** Manual commits double-commit on success and break the read-only optimization (which keeps read-only requests out of write-side profiling and pgBouncer transaction stats). `session.flush()` is fine and sometimes necessary — use it when you need a DB-assigned primary key mid-request: