Failed in-transaction dispatch can be committed in SQLAlchemy backends #116

SQLAlchemyBackend and AsyncSQLAlchemyBackend flush event and outbox rows before dispatching in-transaction listeners.

If serialization/deserialization or a listener raises, append() fails as expected, but the pending rows remain in the SQLAlchemy session. A later, unrelated session.commit() persists an append that the caller observed as failed.

Reproduction

1. Configure a SQLAlchemy backend with an outbox and encryption.
2. Register an in-transaction listener for an event type.
3. Make listener dispatch fail, for example by making decryption raise.
4. Call append() and catch the exception.
5. Call session.commit().
6. Open a fresh session and load the stream.

Expected: the stream is empty; a failed append must not be persisted.

Actual: the stream contains the event. The outbox row can also be persisted.

This reproduces in both SQLAlchemyBackend and AsyncSQLAlchemyBackend.

Affected code

• event_sourcery_sqlalchemy/event_store.py — SqlAlchemyStorageStrategy.insert_events()
• event_sourcery_sqlalchemy/async_/event_store.py — AsyncSqlAlchemyStorageStrategy.insert_events()

Both implementations flush event/outbox rows before calling the dispatcher.

Proposed fix

Prepare and deserialize records for matching listeners before mutating the session, as the in-memory implementations already do:

1. Build RecordedRaw values.
2. Prepare dispatch, allowing failures to occur.
3. Create/update stream, insert events and outbox rows.
4. Invoke the already prepared listeners.

Alternatively, explicitly roll back or expunge all mutations before propagating a dispatch error. Preparing before mutation is preferable because it preserves the current atomicity model without taking ownership of the caller’s transaction.

Regression test

Add sync and async SQLAlchemy tests asserting that a dispatch/deserialization failure makes append() raise and that a subsequent session.commit() does not persist the event or outbox entry.