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..a227a451 --- /dev/null +++ b/submit_ce/api/save_participant.py @@ -0,0 +1,191 @@ +"""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_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.""" + + 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 failed, and whether the rollback succeeded. + + 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 + ``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 + + 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.""" + + 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_save_failed`` 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_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. + + Exceptions raised here are logged and never mask the original + exception, which propagates to the caller unchanged. + """ diff --git a/submit_ce/domain/event/process.py b/submit_ce/domain/event/process.py index 40b2a428..558ed064 100644 --- a/submit_ce/domain/event/process.py +++ b/submit_ce/domain/event/process.py @@ -140,7 +140,7 @@ class BuildSourcePackage(EventWithSideEffect): NAME = "build source package" NAMED = "built source package" - def validate(self, submission: Submission) -> None: + def validate_pre_lock(self, submission: Submission) -> None: """The submission must exist to have a source package built.""" if not submission.submission_id: raise InvalidEvent( diff --git a/submit_ce/domain/event/tests/test_oversize_detection.py b/submit_ce/domain/event/tests/test_oversize_detection.py index 252ebb22..ebc1981e 100644 --- a/submit_ce/domain/event/tests/test_oversize_detection.py +++ b/submit_ce/domain/event/tests/test_oversize_detection.py @@ -50,9 +50,18 @@ def delete_all_source_files(self, sid): def get_workspace(self, sid): return self.workspace + def delete_source_package(self, sid): + pass + def delete_preflight(self, sid): pass + def delete_user_decisions(self, sid): + pass + + def delete_directives(self, sid): + pass + def delete_preview(self, sid): pass diff --git a/submit_ce/implementations/legacy_implementation/__init__.py b/submit_ce/implementations/legacy_implementation/__init__.py index f9bd63ae..65dd4f7c 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,128 @@ 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_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_save_failed(ctx, failure) + except Exception: + logger.exception("on_save_failed handler raised 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 +271,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 +289,38 @@ 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: - consequent_event, after = db.store_event(session, event, before, after) - committed.append(consequent_event) + ctx.phase = SavePhase.EVENT_PERSIST + 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) + 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 +330,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..98855730 --- /dev/null +++ b/submit_ce/ui/tests/test_save_participants.py @@ -0,0 +1,404 @@ +"""Tests for `SaveParticipant` phases fired by ``SubmitApi.save()``. + +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 Event, 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 _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. + + ``raise_in`` makes the named phase raise `_Boom`. ``failures`` collects + the `SaveFailure` passed to ``on_save_failed``. + """ + + 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_save_failed(self, ctx: SaveContext, failure: SaveFailure) -> None: + self.failures.append(failure) + self._fire("on_save_failed") + + +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 + + +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") + + +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_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) + + 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_save_failed", "b"), ("on_save_failed", "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_save_failed 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_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_save_failed 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_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_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_save_failed") + _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_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")) + + +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, + (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) + + 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_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): + """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