Reduce Saga Storage Overhead

🚀 Release notes: Reduce Saga Storage Overhead

📋 Summary

This release reduces saga storage load by introducing checkpoint commits: one session per saga run with explicit commits only at key points (after each step, after each compensation step, etc.) instead of committing after every storage call. Fewer commits, shorter lock hold time, and lower risk of deadlocks when using SQLAlchemy.

---

✨ What's New

🎯 Checkpoint commits & SagaStorageRun protocol

• New protocol SagaStorageRun (src/cqrs/saga/storage/protocol.py) — scoped “session” for a single saga. Methods do not commit; the caller calls commit() at checkpoints.
• New method ISagaStorage.create_run() — returns a context manager that yields a SagaStorageRun. Storages that don’t support it can raise NotImplementedError and execution falls back to the previous behaviour.
• SagaTransaction — when create_run() is available, the saga runs in one session and commits only at checkpoints (after create + RUNNING, after each step, after each compensation step, at completion/failure). Storages without create_run() keep working as before.

📦 Storage implementations

• Memory (MemorySagaStorage) — implements create_run() via _MemorySagaStorageRun; commit/rollback are no-ops but the protocol matches SQLAlchemy.
• SQLAlchemy (SqlAlchemySagaStorage) — implements create_run() via _SqlAlchemySagaStorageRun with one AsyncSession per run; mutations are committed only when the saga calls run.commit().

🔒 Deadlock mitigation

• load_saga_state(..., read_for_update=True) — when loading for recovery or exclusive update, the row can be locked (e.g. SELECT ... FOR UPDATE in SQLAlchemy). Together with checkpoint commits, this shortens lock duration and reduces deadlock risk.

♻️ Compensation

• SagaCompensator — new optional on_after_compensate_step callback, invoked after each successfully compensated step. When using a run, the saga passes run.commit so each compensation step is committed at a checkpoint.

📝 Documentation & types

• Docstrings added/updated for the storage protocol, SagaStorageRun, memory and SQLAlchemy storage, SagaTransaction, execution and recovery managers, and compensator. “Strict Backward Recovery” and checkpoint behaviour are described in code.

---

🐛 Fixes

• Deadlocks — checkpoint commits and read_for_update reduce long-held locks and concurrent-update conflicts.

---

🧪 Tests & infrastructure

Unit tests

• New tests/unit/test_saga/test_saga_storage_run.py — tests create_run() and checkpoint path (memory), and fallback when storage does not implement create_run().

Integration tests & databases

• PostgreSQL — integration tests now run against PostgreSQL (port 5433) in addition to MySQL.
• Test split — test_saga_storage_sqlalchemy_postgres.py / _mysql.py, test_saga_mediator_sqlalchemy_postgres.py / _mysql.py; conftest/fixtures extended with DATABASE_DSN_POSTGRESQL.

🔧 CI

• tests.yml — start and wait for PostgreSQL; set DATABASE_DSN_POSTGRESQL; run tests with MySQL and PostgreSQL.
• New workflow codspeed.yml — run benchmarks via CodSpeed (MySQL, PostgreSQL, Redis; pytest tests/benchmarks/ --codspeed).
• pytest-config.ini — added DATABASE_DSN_POSTGRESQL to env.

📦 Docker & dependencies

• docker-compose-test.yml — new service postgres_tests (PostgreSQL 15.4, port 5433).
• pyproject.toml — version 4.9.0; dev dependency pytest-codspeed==4.2.0 added.

⚡ Benchmarks

• Benchmarks in tests/benchmarks/ updated and extended (conftest, dataclasses/default for memory and SQLAlchemy) to measure the impact of the new storage behaviour.

---

✅ Compatibility

• Saga examples (saga.py, saga_recovery.py, saga_sqlalchemy_storage.py, saga_fallback.py, saga_recovery_scheduler.py, etc.) updated to the current storage API and checkpoint path where applicable.
• Backward compatibility — code using ISagaStorage without create_run() continues to work. The new path is used only when storage implements create_run() (Memory and SQLAlchemy do by default).

---

📌 Miscellaneous

• README.md — updated for structure and usage.
• Post-review: ruff format, pre-commit, and “banchmarks” → benchmarks rename.

---

🎉 Summary of benefits

⚡	Fewer DB commits per saga run
🔒	Shorter lock hold time and lower deadlock risk
✅	Backward compatible; custom storages can add create_run() like Memory/SQLAlchemy