From 7237280a4b8aac9ae94db7cd480fc85b2c50d036 Mon Sep 17 00:00:00 2001 From: "Brian D. Caruso" Date: Wed, 8 Jul 2026 14:25:07 -0400 Subject: [PATCH 1/4] This change adds a `SaveParticipant` abstraction. MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Goal to allow the SubmitAPI to have more options to add behavior that happens inside protected critical sections. Adds ability to have behavior in the save transaction with four phases of callbacks (`under_lock`, `before_commit`, `after_commit`, `on_rollback`), injected via constructor. `on_rollback` receives an explicit `SaveFailure` record (exception + `SavePhase` enum + current event/participant) built from a cursor the save loop maintains, so participants never reverse-engineer failure location from exception types. `SubmitApi.save()` runs a critical section (row lock → validate/execute events → persist → commit) that is currently a private detail of `LegacySubmitImplementation`. Composed implementations like `PubsubEventSubmitImplementation` can only act before/after the whole `save()` call. This is *outside* the transaction and nothing is notified of rollback. This changes provides several new places to tie in code that is run inside the transaction that makes up the protected critical section. Scope: **core machinery + tests only**. Pubsub stays a decorator; conversion in follow-up work. --- docs/plans/save_participants_plan.md | 119 +++++++ submit_ce/api/__init__.py | 7 +- submit_ce/api/save_participant.py | 171 ++++++++++ .../legacy_implementation/__init__.py | 181 ++++++++--- .../legacy_implementation/flask_impl.py | 2 + submit_ce/ui/backend.py | 5 +- submit_ce/ui/tests/test_save_participants.py | 292 ++++++++++++++++++ 7 files changed, 725 insertions(+), 52 deletions(-) create mode 100644 docs/plans/save_participants_plan.md create mode 100644 submit_ce/api/save_participant.py create mode 100644 submit_ce/ui/tests/test_save_participants.py diff --git a/docs/plans/save_participants_plan.md b/docs/plans/save_participants_plan.md new file mode 100644 index 00000000..8fb7e337 --- /dev/null +++ b/docs/plans/save_participants_plan.md @@ -0,0 +1,119 @@ +# SaveParticipant: transaction-phase callbacks for SubmitApi.save() + +## Context + +`SubmitApi.save()` runs a critical section (row lock → validate/execute events → persist → commit) that is currently a private detail of `LegacySubmitImplementation`. Composed implementations like `PubsubEventSubmitImplementation` can only act before/after the whole `save()` call — nothing can run *inside* the transaction (atomic audit/outbox writes, global invariant checks under the lock) and nothing is notified of rollback (relevant for the fs/db desync problem documented in `legacy_implementation/__init__.py:156-178`). + +This change adds a **SaveParticipant** abstraction: objects enlisted in the save transaction with four phase callbacks (`under_lock`, `before_commit`, `after_commit`, `on_rollback`), injected via constructor. `on_rollback` receives an explicit `SaveFailure` record (exception + `SavePhase` enum + current event/participant) built from a cursor the save loop maintains, so participants never reverse-engineer failure location from exception types. + +Scope decision (user-confirmed): **core machinery + tests only**. Pubsub stays a decorator; conversion is a follow-up. + +## New file: `submit_ce/api/save_participant.py` + +Follows codebase style: ABC with no-op defaults (like `EventWithSideEffect.validate_under_lock` in `domain/event/base.py:250`), `(str, Enum)` for the enum (like `SubmissionType` in `domain/submission.py:245`). Use `TYPE_CHECKING` import for `SubmitApi` to avoid a circular import with `api/submit.py`. + +```python +class SavePhase(str, Enum): + LOAD_LOCK = "load_lock" # _load(..., lock_row=True) + PARTICIPANT_UNDER_LOCK = "participant_under_lock" + EVENT_VALIDATE_UNDER_LOCK = "event_validate_under_lock" + EVENT_EXECUTE = "event_execute" # side effects; not rolled back by db + EVENT_APPLY = "event_apply" + EVENT_PERSIST = "event_persist" # db.store_event / store_withdrawal + PARTICIPANT_BEFORE_COMMIT = "participant_before_commit" + COMMIT = "commit" # session.commit() itself → indeterminate state + +@dataclass +class SaveContext: + api: "SubmitApi" + submission_id: Optional[str] + requested_events: Sequence[Event] + before: Optional[Submission] = None # set once loaded under lock + existing_events: List[Event] = field(default_factory=list) + after: Optional[Submission] = None # set as events apply + committed: List[Event] = field(default_factory=list) # includes consequence events + # cursor — updated by the save loop just before each risky call + phase: SavePhase = SavePhase.LOAD_LOCK + current_event: Optional[Event] = None + current_participant: Optional["SaveParticipant"] = None + +@dataclass(frozen=True) +class SaveFailure: + exc: BaseException + phase: SavePhase + event: Optional[Event] = None + participant: Optional["SaveParticipant"] = None + +class SaveParticipant(ABC): # all methods no-op defaults; docstrings define the contract + def under_lock(self, ctx: SaveContext) -> None: ... # raise → clean abort + def before_commit(self, ctx: SaveContext) -> None: ... # raise → rollback (executed side effects survive) + def after_commit(self, ctx: SaveContext) -> None: ... # exceptions logged & swallowed + def on_rollback(self, ctx: SaveContext, failure: SaveFailure) -> None: ... # never raises (logged) +``` + +Contract points to document in docstrings: +- Ordering: list order for `under_lock`/`before_commit`; **reverse** list order for `after_commit`/`on_rollback` (onion semantics). +- `on_rollback` fires only on actual rollback — NOT for `after_commit` failures. +- `phase == COMMIT` failure means indeterminate DB state (alert, don't compensate). +- `event.executed` set → side effects definitely ran; `phase == EVENT_EXECUTE` → that event's side effects in unknown partial state. +- No transaction handle is exposed on `SaveContext` (deliberately deferred until a use case needs it, e.g. an outbox participant); `before_commit` participants observe state and can veto by raising, but cannot yet write in the same transaction. + +Re-export from `submit_ce/api/__init__.py` alongside the existing exports. + +## Modify: `submit_ce/implementations/legacy_implementation/__init__.py` + +**`__init__` (line 79):** add `participants: Optional[List[SaveParticipant]] = None` param → `self.participants = list(participants or [])`. + +**`save()` (line 133):** restructure both branches (general + Withdraw) to: +1. Build a `SaveContext`. +2. Wrap the critical section in one `try/except BaseException` (single try, no nesting — cursor fields on `ctx` are updated by assignments just before each risky call). +3. General path: `_load(lock_row=True)` sets `ctx.before/existing_events` → fire `under_lock` participants (cursor: `PARTICIPANT_UNDER_LOCK` + participant) → delegate to `_save(ctx=...)`. +4. Except handler: `session.rollback()` explicitly, build `SaveFailure(exc, ctx.phase, ctx.current_event, ctx.current_participant)`, notify `on_rollback` in reverse order (each individually try/excepted with `logger.exception`, never masking), then bare `raise` — the original exception propagates unchanged so existing `except InvalidEvent` handling in controllers is untouched. + +**`_save()` (line 185):** takes `ctx`; updates cursor before each step in the queue loop: +- `ctx.current_event = event`; `EVENT_VALIDATE_UNDER_LOCK` before `validate_under_lock` (line 219), `EVENT_EXECUTE` before `execute` (line 221), `EVENT_APPLY` before `apply` (line 226), `EVENT_PERSIST` before `db.store_event` (line 228). Append committed events to `ctx.committed` (keep the local `committed` list behavior; `ctx.committed` can alias it). Set `ctx.after` as `after` advances; clear `ctx.current_event` when the queue drains. +- After the loop: fire `before_commit` participants (reverse order, cursor `PARTICIPANT_BEFORE_COMMIT` + participant), then cursor `COMMIT`, `session.commit()` (line 238). +- After successful commit: fire `after_commit` in reverse order, each wrapped in `try/except Exception: logger.exception(...)` — never fails a committed save. + +**`_save_withdrawal()` (line 241):** same treatment so withdrawals don't silently skip participants: `ctx.before = seed` after `load_latest_announced`, fire `under_lock`; cursor `EVENT_PERSIST` around `store_withdrawal`, `EVENT_EXECUTE` around `event.execute`; `before_commit` → `COMMIT` → commit (line 261) → `after_commit`. The shared `try/except`/rollback-notify lives in `save()`, covering this path too. + +Note: today rollback is implicit via the `with session:` context manager — the new explicit `session.rollback()` in the except handler is required so `on_rollback` participants observe post-rollback state. + +## Modify: `submit_ce/implementations/legacy_implementation/flask_impl.py` + +`FlaskSubmitImplementation.__init__`: add `participants=None`, pass through to `super().__init__`. + +## Wiring: `submit_ce/ui/backend.py` + +No functional change needed (default is empty list). Optionally pass `participants=[]` explicitly at the `FlaskSubmitImplementation(...)` construction (line 40) as the documented injection point — a one-line comment noting this is where participants get wired. + +## `NullImplementation` (`submit_ce/implementations/__init__.py:250`) + +Leave as-is (its trivial `save()` has no participants). Not part of the transactional contract; pragma no cover already. + +## Tests: new `submit_ce/ui/tests/test_save_participants.py` + +Mirror the pattern of `test_save_validate_under_lock.py` (app fixture, `sub_created`, throwaway probe classes, `current_app.api`). Inject participants by appending to `current_app.api.participants` (restore in a finally/fixture, since the `app` fixture may be session-scoped) or by constructing a fresh `FlaskSubmitImplementation(store=..., participants=[...])` inside the app context — prefer whichever the existing fixtures make cleanest; direct construction avoids cross-test leakage. + +A `_ProbeParticipant(SaveParticipant)` records `(phase_name, participant_id)` tuples into a shared list, with flags to raise in a chosen phase. + +Test cases: +1. **Happy-path ordering**: with two participants A, B and one side-effect event, calls are `A.under_lock, B.under_lock, , B.before_commit, A.before_commit, B.after_commit, A.after_commit` and event history grows by one. +2. **`under_lock` raise** → nothing committed (history unchanged), event `execute` never ran, `on_rollback` fired in reverse order with `failure.phase == PARTICIPANT_UNDER_LOCK` and `failure.participant is` the raiser; original exception type propagates. +3. **`before_commit` raise** → history unchanged (DB rolled back), `on_rollback` with `phase == PARTICIPANT_BEFORE_COMMIT`; note event `execute` DID run (assert probe event's calls) — documents the side-effect caveat. +4. **Event `validate_under_lock` raise** (reuse `_ProbeSideEffect`-style event) → `on_rollback` with `phase == EVENT_VALIDATE_UNDER_LOCK` and `failure.event is` the probe event; `InvalidEvent` propagates unchanged. +5. **`after_commit` raise** → save returns normally, event committed, `on_rollback` NOT called. +6. **`on_rollback` raise** → other participants' `on_rollback` still called; the original exception (not the participant's) propagates. +7. **`ctx` contents**: in `before_commit`, `ctx.after` is set and `ctx.committed` non-empty. +8. **Withdraw path** (new coverage): a participant fires on `save(Withdraw(...))` — model the setup on the withdraw controller test fixtures if usable (`published_submission` fixture); if setting one up proves heavy, cover at least under_lock/before_commit/after_commit firing. + +## Verification + +```bash +uv run pytest submit_ce/ui/tests/test_save_participants.py # new tests +uv run pytest submit_ce/ui/tests/test_save_validate_under_lock.py # existing lock-path behavior unchanged +uv run pytest submit_ce # full suite — save() is on every fixture path +uv run ruff check --output-format=github submit_ce +``` + +The full suite matters here: every stage fixture (`sub_created` … `published_submission`) goes through `save()`, so any regression in the restructured critical section surfaces broadly. diff --git a/submit_ce/api/__init__.py b/submit_ce/api/__init__.py index 1e219651..a247a18f 100644 --- a/submit_ce/api/__init__.py +++ b/submit_ce/api/__init__.py @@ -4,9 +4,14 @@ from .file_store import SubmissionFileStore from .submit import SubmitApi from .compile_service import CompileService +from .save_participant import SaveContext, SaveFailure, SaveParticipant, SavePhase __all__ = [ SubmissionFileStore, SubmitApi, - CompileService + CompileService, + SaveContext, + SaveFailure, + SaveParticipant, + SavePhase, ] diff --git a/submit_ce/api/save_participant.py b/submit_ce/api/save_participant.py new file mode 100644 index 00000000..6a1bd2f8 --- /dev/null +++ b/submit_ce/api/save_participant.py @@ -0,0 +1,171 @@ +"""Participants enlisted in the ``SubmitApi.save()`` transaction. + +``SubmitApi.save()`` implementations run a critical section: load and lock the +submission, validate and execute events, persist them, and commit. A +:class:`SaveParticipant` is an object enlisted in that transaction with +callbacks at defined phases. Unlike a decorator wrapping ``save()`` from the +outside, a participant shares the fate of the transaction: it can veto the +save by raising, and it is notified when the transaction rolls back. + +Participants are for infrastructure concerns about the save as a whole +(notification, audit, metrics). Behavior that is semantically part of a +submission's history belongs on the event lifecycle instead +(:meth:`EventWithSideEffect.validate_under_lock`, +:meth:`EventWithSideEffect.execute`, :meth:`Event.get_consequences`). + +Participants are injected via the implementation's constructor, not +registered dynamically. + +No transaction handle is exposed on :class:`SaveContext`. This is deliberate: +it is deferred until a use case needs it (e.g. an outbox participant that +writes rows in the same transaction). Until then ``before_commit`` +participants can observe state and veto by raising, but cannot write in the +same transaction. +""" +from abc import ABC +from dataclasses import dataclass, field +from enum import Enum +from typing import List, Optional, Sequence, TYPE_CHECKING + +from submit_ce.domain import Event, Submission + +if TYPE_CHECKING: # avoid circular import with submit_ce.api.submit + from submit_ce.api import SubmitApi + + +class SavePhase(str, Enum): + """Where the save critical section is (or was, when it failed). + + The save loop updates :attr:`SaveContext.phase` just before each risky + step, so a :class:`SaveFailure` can report exactly where an exception + escaped without participants reverse-engineering it from exception types. + """ + + LOAD_LOCK = "load_lock" + """Loading the submission with the row lock.""" + + PARTICIPANT_UNDER_LOCK = "participant_under_lock" + """A participant's :meth:`SaveParticipant.under_lock` is running.""" + + EVENT_VALIDATE_UNDER_LOCK = "event_validate_under_lock" + """An event's ``validate_under_lock()`` is running.""" + + EVENT_EXECUTE = "event_execute" + """An event's ``execute()`` (side effects) is running. Side effects are + NOT rolled back by the database; a failure here leaves that event's side + effects in an unknown partial state.""" + + EVENT_APPLY = "event_apply" + """An event's pure state projection is running.""" + + EVENT_PERSIST = "event_persist" + """The event is being written to the database.""" + + PARTICIPANT_BEFORE_COMMIT = "participant_before_commit" + """A participant's :meth:`SaveParticipant.before_commit` is running.""" + + COMMIT = "commit" + """``session.commit()`` itself. A failure here means the database state + is indeterminate (the commit may or may not have taken effect) — + participants should alert, not compensate.""" + + +@dataclass +class SaveContext: + """State of one ``save()`` transaction, passed to every participant phase. + + Fields are populated as the save proceeds; a participant sees the fields + that exist at its phase (e.g. ``after`` and ``committed`` are populated by + ``before_commit`` time). The cursor fields (``phase``, ``current_event``, + ``current_participant``) track where the save loop is and feed + :class:`SaveFailure` on rollback. + """ + + api: "SubmitApi" + submission_id: Optional[str] + requested_events: Sequence[Event] + """The events as passed by the caller, before consequences.""" + + before: Optional[Submission] = None + """Submission state as loaded under the lock. None when creating.""" + + existing_events: List[Event] = field(default_factory=list) + after: Optional[Submission] = None + """Projected submission state after the events applied so far.""" + + committed: List[Event] = field(default_factory=list) + """Events persisted this save, including consequence events.""" + + # cursor — updated by the save loop just before each risky call + phase: SavePhase = SavePhase.LOAD_LOCK + current_event: Optional[Event] = None + current_participant: Optional["SaveParticipant"] = None + + +@dataclass(frozen=True) +class SaveFailure: + """Why and where a save transaction rolled back. + + Passed to :meth:`SaveParticipant.on_rollback`. ``phase`` says where the + exception escaped; the context (``ctx.committed``, each event's + ``executed`` timestamp) says what had already happened. Both are needed: + alerting keys off ``phase``, cleanup keys off what executed. An event with + ``executed`` set definitely ran its side effects (which survive the DB + rollback); ``phase == EVENT_EXECUTE`` means ``event``'s side effects are + in an unknown partial state. + """ + + exc: BaseException + phase: SavePhase + event: Optional[Event] = None + """The event being processed when the exception escaped, if any.""" + + participant: Optional["SaveParticipant"] = None + """The participant that raised, if one did.""" + + +class SaveParticipant(ABC): + """A participant enlisted in the ``save()`` transaction. + + All methods default to no-ops; override only the phases you care about. + + Ordering: implementations call ``under_lock`` in participant list order + and ``before_commit``, ``after_commit`` and ``on_rollback`` in reverse + list order (onion semantics). + """ + + def under_lock(self, ctx: SaveContext) -> None: + """Called inside the transaction, after the submission row is loaded + and locked, before any event validates or executes. + + When the save creates a new submission there is no row to lock and + ``ctx.before`` is None. + + Raising here is the clean abort: nothing has executed or persisted, + the rollback is complete, and the exception propagates to the caller. + """ + + def before_commit(self, ctx: SaveContext) -> None: + """Called inside the transaction after all events (including + consequences) have been applied and persisted, just before commit. + + ``ctx.after`` and ``ctx.committed`` are populated. Raising rolls the + database transaction back — with the same caveat as + ``validate_under_lock``: already-executed event side effects are not + rolled back. + """ + + def after_commit(self, ctx: SaveContext) -> None: + """Called after the transaction committed successfully. + + The save has succeeded and the caller will be told so: exceptions + raised here are logged and swallowed, never propagated. + """ + + def on_rollback(self, ctx: SaveContext, failure: SaveFailure) -> None: + """Called after the transaction rolled back. + + Fires only on an actual rollback — not for ``after_commit`` failures. + Exceptions raised here are logged and never mask the original + exception, which propagates to the caller unchanged. + """ diff --git a/submit_ce/implementations/legacy_implementation/__init__.py b/submit_ce/implementations/legacy_implementation/__init__.py index f9bd63ae..ef8cc2c3 100644 --- a/submit_ce/implementations/legacy_implementation/__init__.py +++ b/submit_ce/implementations/legacy_implementation/__init__.py @@ -11,6 +11,8 @@ from submit_ce.api import SubmitApi from submit_ce.api.email_service import EmailService +from submit_ce.api.save_participant import (SaveContext, SaveFailure, + SaveParticipant, SavePhase) from submit_ce.api.file_store import SubmissionFileStore from submit_ce.domain.agent import Client, User from submit_ce.domain.config import SubmitConfig @@ -67,6 +69,9 @@ class LegacySubmitImplementation(SubmitApi): Configuration thta is relevant to the SubmitAPI. get_session : Callable[[], SqlalchemySession], optional Factory returning a SQLAlchemy session for database access. + participants : Optional[List[SaveParticipant]] + Participants enlisted in every ``save()`` transaction. See + :mod:`submit_ce.api.save_participant`. Notes ----- @@ -81,12 +86,14 @@ def __init__(self, compiler: CompileService, email_service: EmailService, config: SubmitConfig, - get_session:Callable[[],SqlalchemySession]): + get_session:Callable[[],SqlalchemySession], + participants: Optional[List[SaveParticipant]] = None): self.get_session = get_session self.compiler = compiler self.store = store self.email_service = email_service self.config = config or SubmitConfig() + self.participants = list(participants or []) def __repr__(self) -> str: return (f"{self.__class__.__name__}(" @@ -133,63 +140,117 @@ def _load(self, session: SqlalchemySession, submission_id: str, lock_row: bool = def save(self, *events: Event, submission_id: Optional[str] = None) -> Tuple[Submission, List[Event]]: if not events: raise NothingToDo() - if isinstance(events[0], Withdraw): + if isinstance(events[0], Withdraw) and len(events) > 1: # BDC I don't love how the Withdraw is handled. I'd perfer if it were done in a side effect - if len(events) > 1: - raise SaveError("Must save Withdraw as the only item in the list of Events") - else: - with self.get_session() as session: - return self._save_withdrawal(events[0], session) + raise SaveError("Must save Withdraw as the only item in the list of Events") + ctx = SaveContext(api=self, submission_id=submission_id, requested_events=events) with self.get_session() as session: - before: Optional[Submission] = None - existing_events: List[Event] = [] - if submission_id is not None: - """ - This is a critical section where: - 1. the submission is read from the db - 2. changes are made to submission including file changes via Event.execute - 3. the file state is written to the db, checksum, size, file type. - - The arXiv_submission row for the submission will be - locked. Legacy did not lock during file upload. - - There are at least these problems: - - 1. Correctness problem: If the files are uploaded, the state of - the files changes, these need to be written to the db, if there - is an exception before the db is written then the db and files - are out of sync. - - 2. Race condition problem: if the db submission row is not - locked, two processes can both upload at the same time which can - create a file system state that neither intended. - - (there may be other problems) - - We may need a different design for this. Maybe a immutable - upload space id? Maybe go to no file state info in the db? - - FAQ: - - What happens if the _load() locks but then there is an exception - during file upload or other times? - - The session will be rolled back and the files on the FS may not - match what is in the db for size and checksum. """ - - before, existing_events = self._load(session, submission_id, lock_row=True) - elif events[0].submission_id is None and not isinstance(events[0], CreateSubmission): - raise NoSuchSubmission('Unable to determine submission') - return self._save(*events, submission=before, session=session, existing_events=existing_events) + try: + if isinstance(events[0], Withdraw): + result = self._save_withdrawal(events[0], session, ctx) + else: + before: Optional[Submission] = None + existing_events: List[Event] = [] + if submission_id is not None: + """ + This is a critical section where: + 1. the submission is read from the db + 2. changes are made to submission including file changes via Event.execute + 3. the file state is written to the db, checksum, size, file type. + + The arXiv_submission row for the submission will be + locked. Legacy did not lock during file upload. + + There are at least these problems: + + 1. Correctness problem: If the files are uploaded, the state of + the files changes, these need to be written to the db, if there + is an exception before the db is written then the db and files + are out of sync. + + 2. Race condition problem: if the db submission row is not + locked, two processes can both upload at the same time which can + create a file system state that neither intended. + + (there may be other problems) + + We may need a different design for this. Maybe a immutable + upload space id? Maybe go to no file state info in the db? + + FAQ: + + What happens if the _load() locks but then there is an exception + during file upload or other times? + + The session will be rolled back and the files on the FS may not + match what is in the db for size and checksum. """ + + ctx.phase = SavePhase.LOAD_LOCK + before, existing_events = self._load(session, submission_id, lock_row=True) + ctx.before = before + ctx.existing_events = existing_events + elif events[0].submission_id is None and not isinstance(events[0], CreateSubmission): + raise NoSuchSubmission('Unable to determine submission') + self._participants_under_lock(ctx) + result = self._save(*events, submission=before, session=session, + existing_events=existing_events, ctx=ctx) + except BaseException as exc: + # Roll back explicitly (rather than relying on the session + # context manager) so on_rollback participants observe + # post-rollback state. The original exception propagates + # unchanged; participant failures are logged, never masking it. + session.rollback() + failure = SaveFailure(exc=exc, phase=ctx.phase, + event=ctx.current_event, + participant=ctx.current_participant) + for participant in reversed(self.participants): + try: + participant.on_rollback(ctx, failure) + except Exception: + logger.exception("on_rollback failed for participant %r", + participant) + raise + self._participants_after_commit(ctx) + return result + + def _participants_under_lock(self, ctx: SaveContext) -> None: + """Fire `under_lock` in participant list order, inside the transaction.""" + for participant in self.participants: + ctx.phase = SavePhase.PARTICIPANT_UNDER_LOCK + ctx.current_participant = participant + participant.under_lock(ctx) + ctx.current_participant = None + + def _participants_before_commit(self, ctx: SaveContext) -> None: + """Fire `before_commit` in reverse list order, inside the transaction.""" + for participant in reversed(self.participants): + ctx.phase = SavePhase.PARTICIPANT_BEFORE_COMMIT + ctx.current_participant = participant + participant.before_commit(ctx) + ctx.current_participant = None + + def _participants_after_commit(self, ctx: SaveContext) -> None: + """Fire `after_commit` in reverse list order. + + The save has already committed, so exceptions are logged and + swallowed — they must not fail a save the caller will be told + succeeded.""" + for participant in reversed(self.participants): + try: + participant.after_commit(ctx) + except Exception: + logger.exception("after_commit failed for participant %r", + participant) def _save(self, *events, submission: Submission, session, - existing_events: List[Event] + existing_events: List[Event], + ctx: SaveContext ) -> Tuple[Submission, List[Event]]: """Internal save for when submission is already read from the db.""" before = submission - committed: List[Event] = [] + committed: List[Event] = ctx.committed # alias: participants see it grow # A work-queue since events may imply consequent events (see # Event.consequences) that need to be processed in this same locked # session/transaction. They are inserted at the front of the queue so a @@ -199,6 +260,7 @@ def _save(self, *events, queue: List[Event] = list(events) while queue: event = queue.pop(0) + ctx.current_event = event if event.submission_id is None and before and before.submission_id is not None: event.submission_id = before.submission_id @@ -216,29 +278,37 @@ def _save(self, *events, # InvalidEvent here rolls the DB transaction back. # Any earlier EventWithSideEffect.execute() is not rolled back. # The caller's `except InvalidEvent` decides UX. + ctx.phase = SavePhase.EVENT_VALIDATE_UNDER_LOCK event.validate_under_lock(self, before) logger.debug('Execute event %s: %s', event.event_id, event.NAME) + ctx.phase = SavePhase.EVENT_EXECUTE event.execute(self, before) if not event.executed: event.executed = get_tzaware_utc_now() logger.debug('Apply event %s: %s', event.event_id, event.NAME) + ctx.phase = SavePhase.EVENT_APPLY after = event.apply(before) if not event.committed: + ctx.phase = SavePhase.EVENT_PERSIST consequent_event, after = db.store_event(session, event, before, after) committed.append(consequent_event) before = after # Prepare for the next event. + ctx.after = after # Queue any follow-on events implied by this one, given the new state. for consequence in reversed(event.get_consequences(after)): queue.insert(0, consequence) + ctx.current_event = None all_ = sorted(existing_events + committed, key=lambda e: e.created) + self._participants_before_commit(ctx) + ctx.phase = SavePhase.COMMIT session.commit() return after, list(all_) - def _save_withdrawal(self, event: Withdraw, session) \ + def _save_withdrawal(self, event: Withdraw, session, ctx: SaveContext) \ -> Tuple[Submission, List[Event]]: """Save a `Withdraw`, creating a new ``wdr`` submission. @@ -248,16 +318,27 @@ def _save_withdrawal(self, event: Withdraw, session) \ written to the new submission's workspace. """ event.created = datetime.now(UTC) + ctx.phase = SavePhase.LOAD_LOCK seed = db.to_submission(db.load_latest_announced(session, event.paper_id)) + ctx.before = seed + self._participants_under_lock(ctx) + ctx.current_event = event # Creates the wdr row and assigns the new submission id onto `after`. + ctx.phase = SavePhase.EVENT_PERSIST consequent, after = db.store_withdrawal(session, event, seed) # Now that the new id exists, write the `withdrawn` source file to it. + ctx.phase = SavePhase.EVENT_EXECUTE event.execute(self, after) if not event.executed: event.executed = get_tzaware_utc_now() + ctx.after = after + ctx.committed.append(consequent) + ctx.current_event = None + self._participants_before_commit(ctx) + ctx.phase = SavePhase.COMMIT session.commit() return after, [consequent] diff --git a/submit_ce/implementations/legacy_implementation/flask_impl.py b/submit_ce/implementations/legacy_implementation/flask_impl.py index 98217f6f..35e8b4a1 100644 --- a/submit_ce/implementations/legacy_implementation/flask_impl.py +++ b/submit_ce/implementations/legacy_implementation/flask_impl.py @@ -18,6 +18,7 @@ def __init__(self, compiler=None, email_service=None, config=None, + participants=None, ): from submit_ce.domain.config import SubmitConfig from submit_ce.implementations import NullEmailService @@ -27,4 +28,5 @@ def __init__(self, email_service=email_service or NullEmailService(), config=config or SubmitConfig(), get_session=flask_get_session, + participants=participants, ) diff --git a/submit_ce/ui/backend.py b/submit_ce/ui/backend.py index a804df18..d38ac4b5 100644 --- a/submit_ce/ui/backend.py +++ b/submit_ce/ui/backend.py @@ -41,7 +41,10 @@ def config_backend_api(settings: Settings) -> SubmitApi: store=store, compiler=CompileApiService(), email_service=email_service, - config=SubmitConfig.from_config(settings)) + config=SubmitConfig.from_config(settings), + # This is where SaveParticipants get wired into every save() + # transaction. See submit_ce.api.save_participant. + participants=[]) def email_service_from_settings(settings: Settings): diff --git a/submit_ce/ui/tests/test_save_participants.py b/submit_ce/ui/tests/test_save_participants.py new file mode 100644 index 00000000..5cbf9cc9 --- /dev/null +++ b/submit_ce/ui/tests/test_save_participants.py @@ -0,0 +1,292 @@ +"""Tests for `SaveParticipant` phases fired by ``SubmitApi.save()``. + +Participants are enlisted in the save transaction (see +``submit_ce.api.save_participant``): ``under_lock`` fires inside the locked +transaction before any event runs, ``before_commit`` after all events are +persisted but before commit, ``after_commit`` after a successful commit, and +``on_rollback`` (with a `SaveFailure` naming the failed `SavePhase`) when the +transaction rolls back. + +These tests exercise the real ``FlaskSubmitImplementation`` save path (via the +``app`` fixture) using throw-away participants and side-effect events, in the +style of ``test_save_validate_under_lock.py``. Participants are injected by +mutating ``current_app.api.participants``; the ``app`` fixture is +function-scoped so this does not leak between tests. +""" +from typing import List, Optional, Tuple + +import pytest +from flask import current_app + +from submit_ce.api.save_participant import (SaveContext, SaveFailure, + SaveParticipant, SavePhase) +from submit_ce.domain.agent import InternalClient +from submit_ce.domain.event.base import EventWithSideEffect +from submit_ce.domain.event.legacy import Withdraw +from submit_ce.domain.exceptions import InvalidEvent + + +class _Boom(RuntimeError): + """Distinct exception so tests can assert the original error propagates.""" + + +class _ProbeParticipant(SaveParticipant): + """Records ``(phase, participant_name)`` calls into a shared list. + + ``raise_in`` makes the named phase raise `_Boom`. ``failures`` collects + the `SaveFailure` passed to ``on_rollback``. + """ + + def __init__(self, name: str, calls: List[Tuple[str, str]], + raise_in: Optional[str] = None): + self.name = name + self.calls = calls + self.raise_in = raise_in + self.failures: List[SaveFailure] = [] + self.ctx_snapshots: List[dict] = [] + + def _fire(self, phase: str) -> None: + self.calls.append((phase, self.name)) + if self.raise_in == phase: + raise _Boom(f"{self.name} raised in {phase}") + + def under_lock(self, ctx: SaveContext) -> None: + self._fire("under_lock") + + def before_commit(self, ctx: SaveContext) -> None: + self.ctx_snapshots.append({ + "after": ctx.after, + "committed": list(ctx.committed), + }) + self._fire("before_commit") + + def after_commit(self, ctx: SaveContext) -> None: + self._fire("after_commit") + + def on_rollback(self, ctx: SaveContext, failure: SaveFailure) -> None: + self.failures.append(failure) + self._fire("on_rollback") + + +class _ProbeSideEffect(EventWithSideEffect): + """Side-effect event that records calls and can raise under the lock.""" + + NAME = "probe side effect" + NAMED = "probe side effect" + + should_block: bool = False + calls: List[str] = [] + + def validate_pre_lock(self, submission) -> None: + pass + + def validate_under_lock(self, api, submission) -> None: + self.calls.append("validate_under_lock") + if self.should_block: + raise InvalidEvent(self, "blocked under lock") + + def execute(self, api, submission) -> None: + self.calls.append("execute") + + def project(self, submission): + return submission + + +def _client() -> InternalClient: + return InternalClient(name="test_save_participants") + + +def _enlist(*participants: SaveParticipant) -> None: + current_app.api.participants = list(participants) + + +def test_phase_ordering_two_participants(app, authorized_user, sub_created): + """under_lock in list order; before_commit/after_commit in reverse order, + with event work in between; the event is committed.""" + with app.app_context(): + sid = str(sub_created.submission_id) + _, history_before = current_app.api.get_with_history(sid) + + calls: List[Tuple[str, str]] = [] + a = _ProbeParticipant("a", calls) + b = _ProbeParticipant("b", calls) + _enlist(a, b) + + probe = _ProbeSideEffect(creator=authorized_user, client=_client(), calls=[]) + current_app.api.save(probe, submission_id=sid) + + assert calls == [ + ("under_lock", "a"), ("under_lock", "b"), + ("before_commit", "b"), ("before_commit", "a"), + ("after_commit", "b"), ("after_commit", "a"), + ] + # Event work happened between under_lock and before_commit. + assert probe.calls == ["validate_under_lock", "execute"] + _, history_after = current_app.api.get_with_history(sid) + assert len(history_after) == len(history_before) + 1 + + +def test_under_lock_raise_aborts_cleanly(app, authorized_user, sub_created): + """Raising in under_lock is the clean abort: no event ran, nothing + committed, on_rollback fires with the participant phase and identity.""" + with app.app_context(): + sid = str(sub_created.submission_id) + _, history_before = current_app.api.get_with_history(sid) + + calls: List[Tuple[str, str]] = [] + a = _ProbeParticipant("a", calls) + b = _ProbeParticipant("b", calls, raise_in="under_lock") + _enlist(a, b) + + probe = _ProbeSideEffect(creator=authorized_user, client=_client(), calls=[]) + with pytest.raises(_Boom): + current_app.api.save(probe, submission_id=sid) + + assert probe.calls == [] # no event validated or executed + assert calls == [ + ("under_lock", "a"), ("under_lock", "b"), + ("on_rollback", "b"), ("on_rollback", "a"), # reverse order + ] + failure = a.failures[0] + assert failure.phase == SavePhase.PARTICIPANT_UNDER_LOCK + assert failure.participant is b + assert failure.event is None + assert isinstance(failure.exc, _Boom) + + _, history_after = current_app.api.get_with_history(sid) + assert len(history_after) == len(history_before) + + +def test_before_commit_raise_rolls_back_db(app, authorized_user, sub_created): + """Raising in before_commit rolls the DB back — but the event's side + effects already ran (the documented validate_under_lock-style caveat).""" + with app.app_context(): + sid = str(sub_created.submission_id) + _, history_before = current_app.api.get_with_history(sid) + + calls: List[Tuple[str, str]] = [] + a = _ProbeParticipant("a", calls, raise_in="before_commit") + _enlist(a) + + probe = _ProbeSideEffect(creator=authorized_user, client=_client(), calls=[]) + with pytest.raises(_Boom): + current_app.api.save(probe, submission_id=sid) + + # The side effects DID run; only the DB write was rolled back. + assert probe.calls == ["validate_under_lock", "execute"] + failure = a.failures[0] + assert failure.phase == SavePhase.PARTICIPANT_BEFORE_COMMIT + assert failure.participant is a + + _, history_after = current_app.api.get_with_history(sid) + assert len(history_after) == len(history_before) + + +def test_event_validate_under_lock_failure_reported(app, authorized_user, sub_created): + """An event rejected under the lock reaches on_rollback with the event + phase and the event itself; InvalidEvent propagates unchanged.""" + with app.app_context(): + sid = str(sub_created.submission_id) + calls: List[Tuple[str, str]] = [] + a = _ProbeParticipant("a", calls) + _enlist(a) + + probe = _ProbeSideEffect(creator=authorized_user, client=_client(), + should_block=True, calls=[]) + with pytest.raises(InvalidEvent): + current_app.api.save(probe, submission_id=sid) + + failure = a.failures[0] + assert failure.phase == SavePhase.EVENT_VALIDATE_UNDER_LOCK + assert failure.event is probe + assert failure.participant is None + assert isinstance(failure.exc, InvalidEvent) + + +def test_after_commit_raise_is_swallowed(app, authorized_user, sub_created): + """The save already committed: an after_commit failure is logged and + swallowed, save() returns normally and on_rollback does NOT fire.""" + with app.app_context(): + sid = str(sub_created.submission_id) + _, history_before = current_app.api.get_with_history(sid) + + calls: List[Tuple[str, str]] = [] + a = _ProbeParticipant("a", calls, raise_in="after_commit") + _enlist(a) + + probe = _ProbeSideEffect(creator=authorized_user, client=_client(), calls=[]) + current_app.api.save(probe, submission_id=sid) # must not raise + + assert a.failures == [] # no rollback happened + assert ("on_rollback", "a") not in calls + _, history_after = current_app.api.get_with_history(sid) + assert len(history_after) == len(history_before) + 1 + + +def test_on_rollback_raise_never_masks_original(app, authorized_user, sub_created): + """A participant blowing up in on_rollback is logged; the other + participants are still notified and the ORIGINAL exception propagates.""" + with app.app_context(): + sid = str(sub_created.submission_id) + calls: List[Tuple[str, str]] = [] + a = _ProbeParticipant("a", calls) + b = _ProbeParticipant("b", calls, raise_in="on_rollback") + _enlist(a, b) + + probe = _ProbeSideEffect(creator=authorized_user, client=_client(), + should_block=True, calls=[]) + with pytest.raises(InvalidEvent): # original, not b's _Boom + current_app.api.save(probe, submission_id=sid) + + # b raised in on_rollback, but a was still notified after it. + assert ("on_rollback", "b") in calls + assert ("on_rollback", "a") in calls + assert calls.index(("on_rollback", "b")) < calls.index(("on_rollback", "a")) + + +def test_before_commit_sees_final_state(app, authorized_user, sub_created): + """By before_commit the context carries the projected state and the + committed events of this save.""" + with app.app_context(): + sid = str(sub_created.submission_id) + calls: List[Tuple[str, str]] = [] + a = _ProbeParticipant("a", calls) + _enlist(a) + + probe = _ProbeSideEffect(creator=authorized_user, client=_client(), calls=[]) + current_app.api.save(probe, submission_id=sid) + + snapshot = a.ctx_snapshots[0] + assert snapshot["after"] is not None + assert len(snapshot["committed"]) >= 1 + + +def test_withdraw_path_fires_participants(app, authorized_user, + published_submission, mocker): + """The Withdraw branch of save() is a separate code path from _save; it + must fire the same participant phases.""" + with app.app_context(): + _, paper_id = published_submission + # The app fixture uses a NullFileStore whose write methods raise by + # design; Withdraw.execute writes the `withdrawn` source file. + mocker.patch.object(current_app.api.store, "delete_all_source_files") + mocker.patch.object(current_app.api.store, "store_source_file") + + calls: List[Tuple[str, str]] = [] + a = _ProbeParticipant("a", calls) + _enlist(a) + + cmd = Withdraw(creator=authorized_user, client=_client(), + paper_id=paper_id, + comment="Withdrawn because of a fatal flaw in section 3.", + abstract="This paper has been withdrawn by the authors.") + submission, committed = current_app.api.save(cmd) + + assert calls == [ + ("under_lock", "a"), + ("before_commit", "a"), + ("after_commit", "a"), + ] + snapshot = a.ctx_snapshots[0] + assert snapshot["after"] is not None + assert len(snapshot["committed"]) == 1 From a1e9f6f3809860e3608a4472ade6f4b640419f93 Mon Sep 17 00:00:00 2001 From: Brian Maltzan Date: Fri, 17 Jul 2026 15:25:58 -0400 Subject: [PATCH 2/4] Maybe reorder or catch? --- .../legacy_implementation/__init__.py | 16 +++++ submit_ce/ui/tests/test_save_participants.py | 62 +++++++++++++++++++ 2 files changed, 78 insertions(+) diff --git a/submit_ce/implementations/legacy_implementation/__init__.py b/submit_ce/implementations/legacy_implementation/__init__.py index ef8cc2c3..e8652dbc 100644 --- a/submit_ce/implementations/legacy_implementation/__init__.py +++ b/submit_ce/implementations/legacy_implementation/__init__.py @@ -199,7 +199,23 @@ def save(self, *events: Event, submission_id: Optional[str] = None) -> Tuple[Sub # context manager) so on_rollback participants observe # post-rollback state. The original exception propagates # unchanged; participant failures are logged, never masking it. + + session.rollback() + # PR #82 review: session.rollback() can itself raise + # (e.g. the DB connection dropped mid-transaction), masking the + # original exception and skipping the on_rollback loop below -- + # exactly the DB-failure case participants exist to catch. The + # guard below preserves the original error and still notifies + # participants. See the xfail test + # test_rollback_failure_keeps_original_error_and_fires_participants. + # try: + # session.rollback() + # except Exception: + # logger.exception("session.rollback() failed during save " + # "error handling; original error preserved") + + failure = SaveFailure(exc=exc, phase=ctx.phase, event=ctx.current_event, participant=ctx.current_participant) diff --git a/submit_ce/ui/tests/test_save_participants.py b/submit_ce/ui/tests/test_save_participants.py index 5cbf9cc9..5acb6567 100644 --- a/submit_ce/ui/tests/test_save_participants.py +++ b/submit_ce/ui/tests/test_save_participants.py @@ -30,6 +30,11 @@ class _Boom(RuntimeError): """Distinct exception so tests can assert the original error propagates.""" +class _RollbackBoom(RuntimeError): + """Simulates the DB connection dropping so ``session.rollback()`` itself + raises during save()'s except handler (PR #82 review finding #1).""" + + class _ProbeParticipant(SaveParticipant): """Records ``(phase, participant_name)`` calls into a shared list. @@ -244,6 +249,63 @@ def test_on_rollback_raise_never_masks_original(app, authorized_user, sub_create assert calls.index(("on_rollback", "b")) < calls.index(("on_rollback", "a")) +@pytest.mark.xfail( + reason="PR #82 review: unguarded session.rollback() in save()'s " + "except block masks the original error and skips on_rollback. Remove " + "this marker when the rollback is guarded.", + strict=True, + raises=_RollbackBoom, +) +def test_rollback_failure_keeps_original_error_and_fires_participants( + app, authorized_user, sub_created, monkeypatch): + """PR #82 review: if the save fails AND ``session.rollback()`` then + raises (e.g. the DB connection dropped mid-transaction), save() must still + + (a) propagate the ORIGINAL failure, not the rollback error, and + (b) fire ``on_rollback`` so audit/notification participants are not + silently skipped on exactly the DB-failure case they exist to catch. + + This currently FAILS: the unguarded ``session.rollback()`` in save()'s + except block raises ``_RollbackBoom``, which masks the original ``_Boom`` + and skips the ``on_rollback`` loop entirely. It should pass once the + rollback is guarded (rollback failure logged, original exception preserved, + participants still notified). + """ + with app.app_context(): + sid = str(sub_created.submission_id) + + calls: List[Tuple[str, str]] = [] + recorder = _ProbeParticipant("recorder", calls) + # `trigger` supplies the ORIGINAL failure, from inside the locked txn. + trigger = _ProbeParticipant("trigger", calls, raise_in="under_lock") + _enlist(recorder, trigger) + + api = current_app.api + real_get_session = api.get_session + + def get_session_with_failing_rollback(): + session = real_get_session() + + def boom_rollback(*args, **kwargs): + raise _RollbackBoom("connection dropped during rollback") + + session.rollback = boom_rollback + return session + + monkeypatch.setattr(api, "get_session", + get_session_with_failing_rollback) + + probe = _ProbeSideEffect(creator=authorized_user, client=_client(), calls=[]) + # The caller must see the real cause, not the rollback failure. + with pytest.raises(_Boom): + api.save(probe, submission_id=sid) + + # on_rollback must still fire despite rollback() raising. + assert ("on_rollback", "recorder") in calls + assert recorder.failures, "on_rollback participant was never notified" + assert isinstance(recorder.failures[0].exc, _Boom) + + def test_before_commit_sees_final_state(app, authorized_user, sub_created): """By before_commit the context carries the projected state and the committed events of this save.""" From a73190bdc4aafe948e0708ce57b37cf29d1595a8 Mon Sep 17 00:00:00 2001 From: "Brian D. Caruso" Date: Thu, 23 Jul 2026 14:53:10 -0400 Subject: [PATCH 3/4] Fixes bug that consequence failures reported as persistence failures MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit vent.get_consequences() is called while ctx.phase remains EVENT_PERSIST. That method deliberately raises for undeclared consequence types, so rollback participants receive an inaccurate SaveFailure.phase, contrary to the abstraction’s “exact failure location” contract. Adds an EVENT_CONSEQUENCES phase and set it before this call. Adds a test. --- submit_ce/api/save_participant.py | 3 + .../legacy_implementation/__init__.py | 5 +- submit_ce/ui/tests/test_save_participants.py | 67 ++++++++++++++++++- 3 files changed, 72 insertions(+), 3 deletions(-) diff --git a/submit_ce/api/save_participant.py b/submit_ce/api/save_participant.py index 6a1bd2f8..fd0e63bd 100644 --- a/submit_ce/api/save_participant.py +++ b/submit_ce/api/save_participant.py @@ -47,6 +47,9 @@ class SavePhase(str, Enum): PARTICIPANT_UNDER_LOCK = "participant_under_lock" """A participant's :meth:`SaveParticipant.under_lock` is running.""" + EVENT_CONSEQUENCES = "event_consequences" + """A runtime check that the Event only emits consequences of the type defined in the class.""" + EVENT_VALIDATE_UNDER_LOCK = "event_validate_under_lock" """An event's ``validate_under_lock()`` is running.""" diff --git a/submit_ce/implementations/legacy_implementation/__init__.py b/submit_ce/implementations/legacy_implementation/__init__.py index ef8cc2c3..e0b659a7 100644 --- a/submit_ce/implementations/legacy_implementation/__init__.py +++ b/submit_ce/implementations/legacy_implementation/__init__.py @@ -291,13 +291,14 @@ def _save(self, *events, after = event.apply(before) if not event.committed: ctx.phase = SavePhase.EVENT_PERSIST - consequent_event, after = db.store_event(session, event, before, after) - committed.append(consequent_event) + stored_event, after = db.store_event(session, event, before, after) + committed.append(stored_event) before = after # Prepare for the next event. ctx.after = after # Queue any follow-on events implied by this one, given the new state. + ctx.phase = SavePhase.EVENT_CONSEQUENCES for consequence in reversed(event.get_consequences(after)): queue.insert(0, consequence) diff --git a/submit_ce/ui/tests/test_save_participants.py b/submit_ce/ui/tests/test_save_participants.py index 5cbf9cc9..9e745e1c 100644 --- a/submit_ce/ui/tests/test_save_participants.py +++ b/submit_ce/ui/tests/test_save_participants.py @@ -21,7 +21,7 @@ from submit_ce.api.save_participant import (SaveContext, SaveFailure, SaveParticipant, SavePhase) from submit_ce.domain.agent import InternalClient -from submit_ce.domain.event.base import EventWithSideEffect +from submit_ce.domain.event.base import Event, EventWithSideEffect from submit_ce.domain.event.legacy import Withdraw from submit_ce.domain.exceptions import InvalidEvent @@ -92,6 +92,40 @@ def project(self, submission): return submission +class _UndeclaredConsequence(Event): + """A plain no-op event, used only as an undeclared consequence type.""" + + NAME = "undeclared consequence" + NAMED = "undeclared consequence" + + def validate_pre_lock(self, submission) -> None: + pass + + def project(self, submission): + return submission + + +class _ProbeEmitsUndeclaredConsequence(Event): + """An event whose `consequences()` returns a type missing from its own + `CONSEQUENCE_TYPES` — i.e. it violates the declared-consequence contract + that `Event.get_consequences()` enforces at runtime.""" + + NAME = "probe emits undeclared consequence" + NAMED = "probe emits undeclared consequence" + + # Deliberately does NOT declare _UndeclaredConsequence. + CONSEQUENCE_TYPES = frozenset() + + def validate_pre_lock(self, submission) -> None: + pass + + def project(self, submission): + return submission + + def consequences(self, submission): + return [_UndeclaredConsequence(creator=self.creator, client=self.client)] + + def _client() -> InternalClient: return InternalClient(name="test_save_participants") @@ -203,6 +237,37 @@ def test_event_validate_under_lock_failure_reported(app, authorized_user, sub_cr assert isinstance(failure.exc, InvalidEvent) +def test_event_consequences_failure_reported(app, authorized_user, sub_created): + """`Event.get_consequences()` raises `RuntimeError` when an event returns + a consequence type it did not declare in `CONSEQUENCE_TYPES` (the runtime + check in `Event.get_consequences`, domain/event/base.py:196-200). The save + loop must report this as `SavePhase.EVENT_CONSEQUENCES`, name the + offending event, and roll back — including the parent event's own write, + even though it was persisted earlier in the same transaction.""" + with app.app_context(): + sid = str(sub_created.submission_id) + _, history_before = current_app.api.get_with_history(sid) + + calls: List[Tuple[str, str]] = [] + a = _ProbeParticipant("a", calls) + _enlist(a) + + probe = _ProbeEmitsUndeclaredConsequence(creator=authorized_user, client=_client()) + with pytest.raises(RuntimeError, match="undeclared consequence"): + current_app.api.save(probe, submission_id=sid) + + failure = a.failures[0] + assert failure.phase == SavePhase.EVENT_CONSEQUENCES + assert failure.event is probe + assert failure.participant is None + assert isinstance(failure.exc, RuntimeError) + + # The parent event's own persisted write is rolled back too, even + # though get_consequences() raised after store_event() ran. + _, history_after = current_app.api.get_with_history(sid) + assert len(history_after) == len(history_before) + + def test_after_commit_raise_is_swallowed(app, authorized_user, sub_created): """The save already committed: an after_commit failure is logged and swallowed, save() returns normally and on_rollback does NOT fire.""" From 1f979a18f308360f8856cb94d3e94e5fe92561d8 Mon Sep 17 00:00:00 2001 From: "Brian D. Caruso" Date: Thu, 23 Jul 2026 15:56:48 -0400 Subject: [PATCH 4/4] Fixes problem where rollback might fail and mask errors --- submit_ce/api/save_participant.py | 29 ++++++-- .../legacy_implementation/__init__.py | 37 +++++----- submit_ce/ui/tests/test_save_participants.py | 67 +++++++------------ 3 files changed, 65 insertions(+), 68 deletions(-) diff --git a/submit_ce/api/save_participant.py b/submit_ce/api/save_participant.py index fd0e63bd..a227a451 100644 --- a/submit_ce/api/save_participant.py +++ b/submit_ce/api/save_participant.py @@ -107,9 +107,9 @@ class SaveContext: @dataclass(frozen=True) class SaveFailure: - """Why and where a save transaction rolled back. + """Why and where a save failed, and whether the rollback succeeded. - Passed to :meth:`SaveParticipant.on_rollback`. ``phase`` says where the + Passed to :meth:`SaveParticipant.on_save_failed`. ``phase`` says where the exception escaped; the context (``ctx.committed``, each event's ``executed`` timestamp) says what had already happened. Both are needed: alerting keys off ``phase``, cleanup keys off what executed. An event with @@ -120,6 +120,11 @@ class SaveFailure: exc: BaseException phase: SavePhase + + rolledback: bool + """If the db rollback successfully ran or not. If this is `False` the state of the db + is unknown, the data may or may not have been saved.""" + event: Optional[Event] = None """The event being processed when the exception escaped, if any.""" @@ -133,7 +138,7 @@ class SaveParticipant(ABC): All methods default to no-ops; override only the phases you care about. Ordering: implementations call ``under_lock`` in participant list order - and ``before_commit``, ``after_commit`` and ``on_rollback`` in reverse + and ``before_commit``, ``after_commit`` and ``on_save_failed`` in reverse list order (onion semantics). """ @@ -165,10 +170,22 @@ def after_commit(self, ctx: SaveContext) -> None: raised here are logged and swallowed, never propagated. """ - def on_rollback(self, ctx: SaveContext, failure: SaveFailure) -> None: - """Called after the transaction rolled back. + def on_save_failed(self, ctx: SaveContext, failure: SaveFailure) -> None: + """Called after the save failed and a rollback was attempted. + + Fires whenever the transaction did not commit — not for + ``after_commit`` failures, where the save already succeeded. + + A DB rollback is attempted before this runs, but it may not have + succeeded: check ``failure.rolledback``. When it is ``False`` the + rollback itself raised (e.g. the DB connection dropped) and the durable + DB state is unknown, so participants should alert rather than attempt + compensation. + + Because the rollback may have left the session unusable, participants + should not read or write to the db. Do out-of-band work here (log, + metric, external notification) rather than ``ctx`` database access. - Fires only on an actual rollback — not for ``after_commit`` failures. Exceptions raised here are logged and never mask the original exception, which propagates to the caller unchanged. """ diff --git a/submit_ce/implementations/legacy_implementation/__init__.py b/submit_ce/implementations/legacy_implementation/__init__.py index 59f2a123..65dd4f7c 100644 --- a/submit_ce/implementations/legacy_implementation/__init__.py +++ b/submit_ce/implementations/legacy_implementation/__init__.py @@ -196,34 +196,29 @@ def save(self, *events: Event, submission_id: Optional[str] = None) -> Tuple[Sub existing_events=existing_events, ctx=ctx) except BaseException as exc: # Roll back explicitly (rather than relying on the session - # context manager) so on_rollback participants observe - # post-rollback state. The original exception propagates - # unchanged; participant failures are logged, never masking it. - - - session.rollback() - # PR #82 review: session.rollback() can itself raise - # (e.g. the DB connection dropped mid-transaction), masking the - # original exception and skipping the on_rollback loop below -- - # exactly the DB-failure case participants exist to catch. The - # guard below preserves the original error and still notifies - # participants. See the xfail test - # test_rollback_failure_keeps_original_error_and_fires_participants. - # try: - # session.rollback() - # except Exception: - # logger.exception("session.rollback() failed during save " - # "error handling; original error preserved") - + # context manager) so on_save_failed participants observe + # post-rollback state. If the rollback itself raises, the + # failure is recorded (rolledback=False) and participants are + # still notified. The original exception propagates unchanged; + # participant failures are logged, never masking it. + + rolledback=False + try: + session.rollback() + rolledback=True + except Exception: + logger.exception("session.rollback() failed during save " + "error handling; original error preserved") failure = SaveFailure(exc=exc, phase=ctx.phase, + rolledback=rolledback, event=ctx.current_event, participant=ctx.current_participant) for participant in reversed(self.participants): try: - participant.on_rollback(ctx, failure) + participant.on_save_failed(ctx, failure) except Exception: - logger.exception("on_rollback failed for participant %r", + logger.exception("on_save_failed handler raised for participant %r", participant) raise self._participants_after_commit(ctx) diff --git a/submit_ce/ui/tests/test_save_participants.py b/submit_ce/ui/tests/test_save_participants.py index ad5f6812..98855730 100644 --- a/submit_ce/ui/tests/test_save_participants.py +++ b/submit_ce/ui/tests/test_save_participants.py @@ -1,12 +1,5 @@ """Tests for `SaveParticipant` phases fired by ``SubmitApi.save()``. -Participants are enlisted in the save transaction (see -``submit_ce.api.save_participant``): ``under_lock`` fires inside the locked -transaction before any event runs, ``before_commit`` after all events are -persisted but before commit, ``after_commit`` after a successful commit, and -``on_rollback`` (with a `SaveFailure` naming the failed `SavePhase`) when the -transaction rolls back. - These tests exercise the real ``FlaskSubmitImplementation`` save path (via the ``app`` fixture) using throw-away participants and side-effect events, in the style of ``test_save_validate_under_lock.py``. Participants are injected by @@ -39,7 +32,7 @@ class _ProbeParticipant(SaveParticipant): """Records ``(phase, participant_name)`` calls into a shared list. ``raise_in`` makes the named phase raise `_Boom`. ``failures`` collects - the `SaveFailure` passed to ``on_rollback``. + the `SaveFailure` passed to ``on_save_failed``. """ def __init__(self, name: str, calls: List[Tuple[str, str]], @@ -68,9 +61,9 @@ def before_commit(self, ctx: SaveContext) -> None: def after_commit(self, ctx: SaveContext) -> None: self._fire("after_commit") - def on_rollback(self, ctx: SaveContext, failure: SaveFailure) -> None: + def on_save_failed(self, ctx: SaveContext, failure: SaveFailure) -> None: self.failures.append(failure) - self._fire("on_rollback") + self._fire("on_save_failed") class _ProbeSideEffect(EventWithSideEffect): @@ -167,7 +160,7 @@ def test_phase_ordering_two_participants(app, authorized_user, sub_created): def test_under_lock_raise_aborts_cleanly(app, authorized_user, sub_created): """Raising in under_lock is the clean abort: no event ran, nothing - committed, on_rollback fires with the participant phase and identity.""" + committed, on_save_failed fires with the participant phase and identity.""" with app.app_context(): sid = str(sub_created.submission_id) _, history_before = current_app.api.get_with_history(sid) @@ -184,7 +177,7 @@ def test_under_lock_raise_aborts_cleanly(app, authorized_user, sub_created): assert probe.calls == [] # no event validated or executed assert calls == [ ("under_lock", "a"), ("under_lock", "b"), - ("on_rollback", "b"), ("on_rollback", "a"), # reverse order + ("on_save_failed", "b"), ("on_save_failed", "a"), # reverse order ] failure = a.failures[0] assert failure.phase == SavePhase.PARTICIPANT_UNDER_LOCK @@ -222,7 +215,7 @@ def test_before_commit_raise_rolls_back_db(app, authorized_user, sub_created): def test_event_validate_under_lock_failure_reported(app, authorized_user, sub_created): - """An event rejected under the lock reaches on_rollback with the event + """An event rejected under the lock reaches on_save_failed with the event phase and the event itself; InvalidEvent propagates unchanged.""" with app.app_context(): sid = str(sub_created.submission_id) @@ -275,7 +268,7 @@ def test_event_consequences_failure_reported(app, authorized_user, sub_created): def test_after_commit_raise_is_swallowed(app, authorized_user, sub_created): """The save already committed: an after_commit failure is logged and - swallowed, save() returns normally and on_rollback does NOT fire.""" + swallowed, save() returns normally and on_save_failed does NOT fire.""" with app.app_context(): sid = str(sub_created.submission_id) _, history_before = current_app.api.get_with_history(sid) @@ -288,19 +281,19 @@ def test_after_commit_raise_is_swallowed(app, authorized_user, sub_created): current_app.api.save(probe, submission_id=sid) # must not raise assert a.failures == [] # no rollback happened - assert ("on_rollback", "a") not in calls + assert ("on_save_failed", "a") not in calls _, history_after = current_app.api.get_with_history(sid) assert len(history_after) == len(history_before) + 1 -def test_on_rollback_raise_never_masks_original(app, authorized_user, sub_created): - """A participant blowing up in on_rollback is logged; the other +def test_on_save_failed_raise_never_masks_original(app, authorized_user, sub_created): + """A participant blowing up in on_save_failed is logged; the other participants are still notified and the ORIGINAL exception propagates.""" with app.app_context(): sid = str(sub_created.submission_id) calls: List[Tuple[str, str]] = [] a = _ProbeParticipant("a", calls) - b = _ProbeParticipant("b", calls, raise_in="on_rollback") + b = _ProbeParticipant("b", calls, raise_in="on_save_failed") _enlist(a, b) probe = _ProbeSideEffect(creator=authorized_user, client=_client(), @@ -308,33 +301,23 @@ def test_on_rollback_raise_never_masks_original(app, authorized_user, sub_create with pytest.raises(InvalidEvent): # original, not b's _Boom current_app.api.save(probe, submission_id=sid) - # b raised in on_rollback, but a was still notified after it. - assert ("on_rollback", "b") in calls - assert ("on_rollback", "a") in calls - assert calls.index(("on_rollback", "b")) < calls.index(("on_rollback", "a")) + # b raised in on_save_failed, but a was still notified after it. + assert ("on_save_failed", "b") in calls + assert ("on_save_failed", "a") in calls + assert calls.index(("on_save_failed", "b")) < calls.index(("on_save_failed", "a")) -@pytest.mark.xfail( - reason="PR #82 review: unguarded session.rollback() in save()'s " - "except block masks the original error and skips on_rollback. Remove " - "this marker when the rollback is guarded.", - strict=True, - raises=_RollbackBoom, -) def test_rollback_failure_keeps_original_error_and_fires_participants( app, authorized_user, sub_created, monkeypatch): """PR #82 review: if the save fails AND ``session.rollback()`` then raises (e.g. the DB connection dropped mid-transaction), save() must still - (a) propagate the ORIGINAL failure, not the rollback error, and - (b) fire ``on_rollback`` so audit/notification participants are not - silently skipped on exactly the DB-failure case they exist to catch. - - This currently FAILS: the unguarded ``session.rollback()`` in save()'s - except block raises ``_RollbackBoom``, which masks the original ``_Boom`` - and skips the ``on_rollback`` loop entirely. It should pass once the - rollback is guarded (rollback failure logged, original exception preserved, - participants still notified). + (a) propagate the ORIGINAL failure, not the rollback error, + (b) fire ``on_save_failed`` so audit/notification participants are not + silently skipped on exactly the DB-failure case they exist to catch, + and + (c) tell those participants the rollback did not succeed, via + ``SaveFailure.rolledback``, so they know the DB state is unknown. """ with app.app_context(): sid = str(sub_created.submission_id) @@ -365,10 +348,12 @@ def boom_rollback(*args, **kwargs): with pytest.raises(_Boom): api.save(probe, submission_id=sid) - # on_rollback must still fire despite rollback() raising. - assert ("on_rollback", "recorder") in calls - assert recorder.failures, "on_rollback participant was never notified" + # on_save_failed must still fire despite rollback() raising. + assert ("on_save_failed", "recorder") in calls + assert recorder.failures, "on_save_failed participant was never notified" assert isinstance(recorder.failures[0].exc, _Boom) + # The failed rollback is signalled so participants know DB state is unknown. + assert recorder.failures[0].rolledback is False def test_before_commit_sees_final_state(app, authorized_user, sub_created):