Fix deadlocks

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

.github/workflows/codspeed.yml

@@ -46,6 +46,19 @@ jobs:
           echo "MySQL did not become ready in time"
           exit 1
 
+      - name: Wait for PostgreSQL
+        run: |
+          for i in $(seq 1 30); do
+            if docker compose -f docker-compose-test.yml exec -T postgres_tests pg_isready -h localhost -U cqrs -q 2>/dev/null; then
+              echo "PostgreSQL is ready"
+              exit 0
+            fi
+            echo "Waiting for PostgreSQL... ($i/30)"
+            sleep 2
+          done
+          echo "PostgreSQL did not become ready in time"
+          exit 1
+
       - name: Wait for Redis
         run: |
           for i in $(seq 1 15); do

.github/workflows/tests.yml

@@ -69,6 +69,7 @@ jobs:
         vermin --target=3.10- --violations --eval-annotations --backport typing_extensions --exclude=venv --exclude=build --exclude=.git --exclude=.venv src examples tests
 
   test:
+    name: test (py ${{ matrix.python-version }})
     runs-on: ubuntu-latest
     strategy:
       fail-fast: false
@@ -105,6 +106,19 @@ jobs:
         echo "MySQL did not become ready in time"
         exit 1
 
+    - name: Wait for PostgreSQL
+      run: |
+        for i in $(seq 1 30); do
+          if docker compose -f docker-compose-test.yml exec -T postgres_tests pg_isready -h localhost -U cqrs -q 2>/dev/null; then
+            echo "PostgreSQL is ready"
+            exit 0
+          fi
+          echo "Waiting for PostgreSQL... ($i/30)"
+          sleep 2
+        done
+        echo "PostgreSQL did not become ready in time"
+        exit 1
+
     - name: Wait for Redis
       run: |
         for i in $(seq 1 15); do
@@ -119,6 +133,10 @@ jobs:
         exit 1
 
     - name: Run all tests with coverage
+      env:
+        DATABASE_DSN: mysql+asyncmy://cqrs:cqrs@localhost:3307/test_cqrs
+        DATABASE_DSN_MYSQL: mysql+asyncmy://cqrs:cqrs@localhost:3307/test_cqrs
+        DATABASE_DSN_POSTGRESQL: postgresql+asyncpg://cqrs:cqrs@localhost:5433/cqrs
       run: |
         pytest -c ./tests/pytest-config.ini --cov=src --cov-report=xml --cov-report=term -o cache_dir=/tmp/pytest_cache ./tests/unit ./tests/integration
 

docker-compose-dev.yml

@@ -15,6 +15,16 @@ services:
     command: --init-file /data/application/init.sql
     volumes:
       - ./tests/init_database.sql:/data/application/init.sql
+  postgres_dev:
+    image: postgres:latest
+    hostname: postgres-dev
+    restart: always
+    environment:
+      POSTGRES_USER: cqrs
+      POSTGRES_PASSWORD: cqrs
+      POSTGRES_DB: cqrs
+    ports:
+      - "5433:5432"
   kafka0:
     image: confluentinc/cp-kafka:7.2.1
     hostname: kafka0

docker-compose-test.yml

@@ -15,6 +15,16 @@ services:
     command: --init-file /data/application/init.sql
     volumes:
       - ./tests/init_database.sql:/data/application/init.sql
+  postgres_tests:
+    image: postgres:latest
+    hostname: postgres-test
+    restart: always
+    environment:
+      POSTGRES_USER: cqrs
+      POSTGRES_PASSWORD: cqrs
+      POSTGRES_DB: cqrs
+    ports:
+      - "5433:5432"
   redis_tests:
     image: redis:7.2
     hostname: redis

examples/saga_recovery_scheduler.py

@@ -561,7 +561,7 @@ async def recovery_loop(
         except asyncio.CancelledError:
             logger.info("Recovery loop cancelled.")
             raise
-        except Exception as e:
+        except Exception:
             logger.exception(f"Recovery iteration failed: {traceback.format_exc()}")
 
         if max_iterations is not None and iteration >= max_iterations:
@@ -618,7 +618,7 @@ async def create_interrupted_saga(storage: MemorySagaStorage) -> uuid.UUID:
 async def main() -> None:
     """
     Run the saga recovery scheduler demo and display its outcome.
-    
+
     Sets up an in-memory saga storage, creates a simulated interrupted saga and marks it stale, runs the recovery loop for three iterations (using the module's recovery_loop and recovery configuration constants), then loads and prints the final saga state.
     """
     print("\n" + "=" * 70)
@@ -660,4 +660,4 @@ async def main() -> None:
 
 
 if __name__ == "__main__":
-    asyncio.run(main())
+    asyncio.run(main())

examples/saga_sqlalchemy_storage.py

@@ -207,4 +207,4 @@ def saga_mapper(mapper: cqrs.SagaMap) -> None:
 
 
 if __name__ == "__main__":
-    asyncio.run(main())
+    asyncio.run(main())

pyproject.toml

@@ -53,6 +53,7 @@ dev = [
   "pytest-env==0.6.2",
   "cryptography==42.0.2",
   "asyncmy==0.2.9",
+  "asyncpg>=0.29.0",
   "redis>=5.0.0",
   # Circuit breaker for tests
   "aiobreaker>=0.3.0"  # from aiobreaker

src/cqrs/saga/compensation.py

@@ -27,7 +27,7 @@ def __init__(
     ) -> None:
         """
         Create a SagaCompensator configured to perform compensation of completed saga steps with retry and optional post-step callback.
-        
+
         Parameters:
             saga_id: Identifier of the saga.
             context: Saga execution context passed to step compensation handlers.
@@ -51,12 +51,12 @@ async def compensate_steps(
     ) -> None:
         """
         Compensates completed saga steps in reverse order, applying retry logic and recording step statuses.
-        
+
         Compensates each handler from last to first, skipping steps already recorded as compensated in the saga history. Updates the saga status to COMPENSATING at the start and logs per-step statuses (STARTED, COMPLETED, FAILED) in storage. After a step completes, the optional on_after_compensate_step callback (if provided) is awaited. If any step fails after all retry attempts, the saga is marked as FAILED. If no completed steps are provided, no compensation is attempted and the saga is marked as FAILED.
-        
+
         Parameters:
             completed_steps (list[SagaStepHandler[ContextT, typing.Any]]): Handlers corresponding to steps that completed during the saga; these will be compensated in reverse order.
-        
+
         Returns:
             None
         """
@@ -184,4 +184,4 @@ async def _compensate_step_with_retry(
 
         # If we get here, all retries failed
         if last_exception:
-            raise last_exception
+            raise last_exception

src/cqrs/saga/saga.py

@@ -157,10 +157,10 @@ async def __aiter__(
     ) -> typing.AsyncIterator[SagaStepResult[ContextT, typing.Any]]:
         """
         Execute saga steps sequentially and yield each step result.
-        
+
         Implements the Strict Backward Recovery strategy: if the saga is in COMPENSATING or FAILED status, forward execution is never resumed. When the underlying storage provides create_run(), execution is performed within a per-saga run with checkpoint commits; otherwise the legacy run-less path is used.
         Returns:
-        	AsyncIterator[SagaStepResult[ContextT, typing.Any]]: An async iterator that yields the result for each executed saga step in order.
+                AsyncIterator[SagaStepResult[ContextT, typing.Any]]: An async iterator that yields the result for each executed saga step in order.
         """
         try:
             run_cm = self._storage.create_run()
@@ -226,13 +226,13 @@ async def _execute(
     ) -> typing.AsyncIterator[SagaStepResult[ContextT, typing.Any]]:
         """
         Execute the saga's configured steps, using the provided storage run for checkpointed operations when available, and perform recovery and compensation as required.
-        
+
         Parameters:
             run (SagaStorageRun | None): Optional per-saga storage run. When provided, the run is used for loading saga state, creating run-scoped managers/executors, and committing at checkpoint boundaries. When None, the transaction's internal managers and executors are used.
-        
+
         Returns:
             Async iterator that yields SagaStepResult values for each step that completes; each yielded result will include the transaction's saga_id.
-        
+
         Raises:
             RuntimeError: If the saga was recovered in COMPENSATING or FAILED state and compensation was completed, forward execution is not allowed.
         """
@@ -258,6 +258,8 @@ async def _execute(
                 self._saga.__class__.__name__,
                 self._context,
             )
+            if run is not None:
+                await run.commit()
             await state_manager.update_status(SagaStatus.RUNNING)
             if run is not None:
                 await run.commit()
@@ -308,10 +310,14 @@ async def _execute(
 
                 completed_step_names = await recovery_manager.load_completed_step_names()
             except ValueError:
+                if run is not None:
+                    await run.rollback()
                 await state_manager.create_saga(
                     self._saga.__class__.__name__,
                     self._context,
                 )
+                if run is not None:
+                    await run.commit()
                 await state_manager.update_status(SagaStatus.RUNNING)
                 if run is not None:
                     await run.commit()
@@ -390,7 +396,7 @@ async def _execute(
     async def _compensate(self) -> None:
         """
         Mark the transaction as compensated and run compensation for all completed steps in reverse order.
-        
+
         Sets an internal flag to prevent repeated compensation and delegates to the compensator which applies the configured retry behavior.
         """
         # Prevent double compensation
@@ -519,4 +525,4 @@ def transaction(
             compensation_retry_count=compensation_retry_count,
             compensation_retry_delay=compensation_retry_delay,
             compensation_retry_backoff=compensation_retry_backoff,
-        )
+        )

src/cqrs/saga/storage/protocol.py

@@ -22,7 +22,7 @@ async def create_saga(
     ) -> None:
         """
         Create a new saga execution record with initial PENDING status and version 1.
-        
+
         Parameters:
             saga_id (uuid.UUID): Unique identifier for the saga (primary key).
             name (str): Human-friendly name used for diagnostics and filtering.
@@ -37,13 +37,13 @@ async def update_context(
     ) -> None:
         """
         Persist a snapshot of the saga's execution context, optionally using optimistic locking.
-        
+
         Parameters:
             saga_id (uuid.UUID): Identifier of the saga to update.
             context (dict[str, Any]): JSON-serializable context object to store as the new snapshot.
             current_version (int | None): If provided, perform an optimistic-locking update that succeeds only
                 if the stored version matches this value; on success the stored version is incremented.
-        
+
         Raises:
             SagaConcurrencyError: If `current_version` is provided and does not match the stored version.
         """
@@ -55,11 +55,11 @@ async def update_status(
     ) -> None:
         """
         Set the global status for the saga identified by `saga_id`.
-        
+
         Parameters:
             saga_id (uuid.UUID): Identifier of the saga to update.
             status (SagaStatus): New global status to persist (for example RUNNING, COMPLETED, COMPENSATING).
-        
+
         Notes:
             This operation does not commit the storage session; the caller must call `commit()` on the active run or session to persist the change.
         """
@@ -74,7 +74,7 @@ async def log_step(
     ) -> None:
         """
         Append a step transition to the saga's execution log.
-        
+
         Parameters:
             saga_id (uuid.UUID): Identifier of the saga whose log will be appended.
             step_name (str): Logical name of the step (used for diagnostics and replay).
@@ -91,28 +91,30 @@ async def load_saga_state(
     ) -> tuple[SagaStatus, dict[str, typing.Any], int]:
         """
         Load the current saga execution state.
-        
+
         Parameters:
             saga_id (uuid.UUID): Identifier of the saga to load.
             read_for_update (bool): If True, acquire a database lock for update to prevent concurrent modifications.
-        
+
         Returns:
             tuple[SagaStatus, dict[str, Any], int]: A tuple containing the saga's global status, the latest persisted context (JSON-serializable), and the current optimistic-locking version number.
         """
+        ...
 
     async def get_step_history(
         self,
         saga_id: uuid.UUID,
     ) -> list[SagaLogEntry]:
         """
         Retrieve the chronological step log for a saga.
-        
+
         Parameters:
             saga_id (uuid.UUID): Identifier of the saga whose step history to retrieve.
-        
+
         Returns:
             list[SagaLogEntry]: Ordered list of step log entries for the saga, from oldest to newest.
         """
+        ...
 
     async def commit(self) -> None:
         """
@@ -343,13 +345,13 @@ def create_run(
     ) -> contextlib.AbstractAsyncContextManager[SagaStorageRun]:
         """
         Create a scoped async run context for a single saga execution session with checkpointed commits.
-        
+
         The context manager yields a SagaStorageRun that provides the same mutation/read methods as the storage but does not commit automatically; the caller must call commit() or rollback() at desired checkpoints.
-        
+
         Returns:
             contextlib.AbstractAsyncContextManager[SagaStorageRun]: Async context manager yielding a SagaStorageRun session.
-        
+
         Raises:
             NotImplementedError: If the storage backend does not support scoped runs.
         """
-        raise NotImplementedError("This storage does not support create_run()")
+        raise NotImplementedError("This storage does not support create_run()")