Update Saga documentation

Author: Вадим Козыревский

docs/saga/recovery.md

@@ -39,6 +39,19 @@ Recovery steps:
 2. Reconstruct context object from persisted data
 3. Resume execution from last completed step or complete compensation
 
+### Concurrency Safety (Row Locking)
+
+In a distributed environment with multiple replicas, multiple recovery workers might attempt to recover the same incomplete saga simultaneously.
+
+To prevent race conditions, `recover_saga` uses **Row Locking**:
+
+1.  It calls `storage.load_saga_state(saga_id, read_for_update=True)`.
+2.  For SQL databases, this executes `SELECT ... FOR UPDATE`.
+3.  This acquires a database-level lock on the saga row.
+4.  If another worker tries to recover the same saga, it will block until the first worker completes or releases the lock.
+
+This ensures that **only one worker** can actively recover and execute a specific saga at any given time.
+
 ## Recovery Scenarios
 
 ### Interrupted Forward Execution
@@ -47,7 +60,7 @@ Recovery steps:
 **Recovery:** Skips completed steps, resumes from last completed step
 
 ```python
-status, _ = await storage.load_saga_state(saga_id)  # RUNNING
+status, _, _ = await storage.load_saga_state(saga_id)  # RUNNING
 await recover_saga(saga, saga_id, OrderContext)
 # Skips completed steps, continues execution
 ```
@@ -58,7 +71,7 @@ await recover_saga(saga, saga_id, OrderContext)
 **Recovery:** Completes compensation in reverse order
 
 ```python
-status, _ = await storage.load_saga_state(saga_id)  # COMPENSATING
+status, _, _ = await storage.load_saga_state(saga_id)  # COMPENSATING
 try:
     await recover_saga(saga, saga_id, OrderContext)
 except RuntimeError:

docs/saga/storage.md

@@ -16,10 +16,10 @@ Two storage implementations:
 ```python
 class ISagaStorage(abc.ABC):
     async def create_saga(saga_id, name, context) -> None
-    async def update_context(saga_id, context) -> None
+    async def update_context(saga_id, context, current_version: int | None = None) -> None
     async def update_status(saga_id, status) -> None
     async def log_step(saga_id, step_name, action, status, details=None) -> None
-    async def load_saga_state(saga_id) -> tuple[SagaStatus, dict]
+    async def load_saga_state(saga_id, *, read_for_update: bool = False) -> tuple[SagaStatus, dict, int]
     async def get_step_history(saga_id) -> list[SagaLogEntry]
 ```
 
@@ -46,7 +46,7 @@ mediator = bootstrap.bootstrap(
 )
 
 # Access storage data
-status, context_data = await storage.load_saga_state(saga_id)
+status, context_data, version = await storage.load_saga_state(saga_id)
 history = await storage.get_step_history(saga_id)
 ```
 
@@ -58,7 +58,7 @@ history = await storage.get_step_history(saga_id)
 
 ## SQLAlchemy Storage
 
-Database-backed implementation for production.
+Database-backed implementation for production. It uses a session factory to manage transactions internally, ensuring that every step is committed immediately ("checkpointing").
 
 ### Database Schema
 
@@ -67,6 +67,7 @@ Database-backed implementation for production.
 - `id` (UUID) - Primary key
 - `status` (VARCHAR) - PENDING, RUNNING, COMPENSATING, COMPLETED, FAILED
 - `context` (JSON)
+- `version` (INTEGER) - Optimistic locking version (default: 1)
 - `created_at`, `updated_at` (TIMESTAMP)
 
 **saga_logs:**
@@ -83,27 +84,31 @@ Database-backed implementation for production.
 
 ### Usage
 
+The storage requires an `async_sessionmaker` to create short-lived sessions for each operation.
+
 ```python
 import uuid
 import cqrs
 from cqrs.saga import bootstrap
 from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
 from cqrs.saga.storage.sqlalchemy import SqlAlchemySagaStorage, Base
 
-# Setup
-engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db")
-SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
+# Setup Engine with connection pool
+engine = create_async_engine(
+    "postgresql+asyncpg://user:pass@localhost/db",
+    pool_size=20,
+    max_overflow=10,
+)
+
+# Create session factory (factory, NOT session instance)
+session_factory = async_sessionmaker(engine, expire_on_commit=False)
 
 # Initialize tables (run once)
 async with engine.begin() as conn:
     await conn.run_sync(Base.metadata.create_all)
 
-# Create storage
-async def create_storage() -> SqlAlchemySagaStorage:
-    return SqlAlchemySagaStorage(SessionLocal())
-
-# Usage
-storage = await create_storage()
+# Initialize storage with session FACTORY
+storage = SqlAlchemySagaStorage(session_factory)
 
 # Register saga in SagaMap
 def saga_mapper(mapper: cqrs.SagaMap) -> None:
@@ -120,12 +125,47 @@ mediator = bootstrap.bootstrap(
 context = OrderContext(...)
 saga_id = uuid.uuid4()
 
+# The storage will automatically commit each step to the database
 async for step_result in mediator.stream(context, saga_id=saga_id):
     print(f"Step: {step_result.step_type.__name__}")
-
-await storage.session.commit()
 ```
 
+### Transaction Management
+
+**SqlAlchemySagaStorage** handles transactions automatically:
+
+1.  Each method (`create_saga`, `log_step`, etc.) opens a new session.
+2.  The operation is performed.
+3.  `session.commit()` is called immediately.
+4.  The session is closed.
+
+This design ensures:
+
+- **Crash Safety:** Even if the application crashes mid-saga, all completed steps are safely persisted.
+- **Connection Efficiency:** Connections are returned to the pool immediately after each operation.
+- **Isolation:** Saga storage operations don't interfere with your business logic transactions.
+
+### Concurrency Control
+
+The storage implementation provides two mechanisms to handle concurrency in distributed environments:
+
+#### 1. Optimistic Locking (Versioning)
+
+To prevent "lost updates" when multiple steps might update the context simultaneously (though sagas typically execute sequentially), the `version` column is used.
+
+- `update_context` accepts an optional `current_version`.
+- If provided, the storage checks if `version == current_version`.
+- If matched, it updates the context and increments the version (`version + 1`).
+- If not matched, it raises `SagaConcurrencyError`, indicating the state was modified by another process.
+
+#### 2. Row Locking (Recovery Safety)
+
+When recovering a saga (e.g., after a crash), it is critical that only one worker picks up the saga to avoid duplicate execution.
+
+- `load_saga_state(..., read_for_update=True)` uses `SELECT ... FOR UPDATE` (in SQL databases).
+- This acquires a row-level lock on the saga execution record.
+- Other workers attempting to lock the same saga will wait or fail, ensuring exclusive access during the recovery process.
+
 ## Choosing Storage
 
 **Memory Storage:**
@@ -138,10 +178,11 @@ await storage.session.commit()
 - ✅ Production, multi-process
 - ✅ Recovery after restarts
 - ✅ Audit trail
+- ✅ Robust transaction management
 
 ## Best Practices
 
-1. **Use persistent storage in production** — Memory storage loses data on restart
-2. **Create indexes** — Index `saga_id` and `created_at` for better performance
-3. **Handle session lifecycle** — Commit and close SQLAlchemy sessions properly
-4. **Monitor storage size** — Archive old saga logs periodically
+1.  **Use persistent storage in production** — Memory storage loses data on restart
+2.  **Configure Connection Pool** — Set appropriate `pool_size` and `max_overflow` for your load.
+3.  **Create indexes** — Index `saga_id` and `created_at` for better performance
+4.  **Monitor storage size** — Archive old saga logs periodically