Fixes after review

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

README.md

@@ -87,7 +87,7 @@ class JoinMeetingCommandHandler(RequestHandler[JoinMeetingCommand, None]):
 
       def __init__(self, meetings_api: MeetingAPIProtocol) -> None:
           self._meetings_api = meetings_api
-          self.events: list[Event] = []
+          self._events: list[Event] = []
 
       @property
       def events(self) -> typing.List[events.Event]:
@@ -115,7 +115,7 @@ class ReadMeetingQueryHandler(RequestHandler[ReadMeetingQuery, ReadMeetingQueryR
 
       def __init__(self, meetings_api: MeetingAPIProtocol) -> None:
           self._meetings_api = meetings_api
-          self.events: list[Event] = []
+          self._events: list[Event] = []
 
       @property
       def events(self) -> typing.List[events.Event]:
@@ -323,7 +323,7 @@ def init_queries(mapper: requests.RequestMap) -> None:
     mapper.bind(queries.ReadMeetingQuery, query_handlers.ReadMeetingQueryHandler)
 
 def init_events(mapper: events.EventMap) -> None:
-    mapper.bind(events.NotificationEvent[events_models.NotificationMeetingRoomClosed], event_handlers.MeetingRoomClosedNotificationHandler)
+    mapper.bind(events.NotificationEvent[event_models.NotificationMeetingRoomClosed], event_handlers.MeetingRoomClosedNotificationHandler)
     mapper.bind(events.NotificationEvent[event_models.ECSTMeetingRoomClosed], event_handlers.UpdateMeetingRoomReadModelHandler)
 ```
 

examples/saga_recovery_scheduler.py

@@ -79,6 +79,7 @@
 import dataclasses
 import datetime
 import logging
+import traceback
 import typing
 import uuid
 
@@ -263,12 +264,7 @@ async def create_shipment(
         shipment_id = f"shipment_{order_id}"
         tracking_number = f"TRACK{self._tracking_counter:08d}"
         self._shipments[shipment_id] = tracking_number
-        logger.info(
-            "  ✓ Created shipment %s for order %s (tracking: %s)",
-            shipment_id,
-            order_id,
-            tracking_number,
-        )
+        logger.info(f"  ✓ Created shipment {shipment_id} for order {order_id} (tracking: {tracking_number})")
         return shipment_id, tracking_number
 
     async def cancel_shipment(self, shipment_id: str) -> None:
@@ -515,19 +511,19 @@ async def run_recovery_iteration(
     processed = 0
     for saga_id in ids:
         try:
-            logger.info("Recovering saga %s...", saga_id)
+            logger.info(f"Recovering saga {saga_id}...")
             await recover_saga(saga, saga_id, context_builder, container, storage)
-            logger.info("Saga %s recovered successfully.", saga_id)
+            logger.info(f"Saga {saga_id} recovered successfully.")
             processed += 1
         except RuntimeError as e:
             if "recovered in" in str(e) and "state" in str(e):
-                logger.info("Saga %s recovery completed compensation: %s", saga_id, e)
+                logger.info(f"Saga {saga_id} recovery completed compensation: {traceback.format_exc()}")
                 processed += 1
             else:
-                logger.exception("Saga %s recovery failed: %s", saga_id, e)
+                logger.exception(f"Saga {saga_id} recovery failed: {traceback.format_exc()}")
                 processed += 1
-        except Exception as e:
-            logger.exception("Saga %s recovery failed: %s", saga_id, e)
+        except Exception:
+            logger.exception(f"Saga {saga_id} recovery failed: {traceback.format_exc()}")
             processed += 1
     return processed
 
@@ -551,25 +547,25 @@ async def recovery_loop(
     iteration = 0
     while True:
         iteration += 1
-        logger.info("Recovery iteration %s", iteration)
+        logger.info(f"Recovery iteration {iteration}")
         try:
             processed = await run_recovery_iteration(
                 storage,
                 saga,
                 OrderContext,
             )
             if processed > 0:
-                logger.info("Processed %s saga(s) this iteration.", processed)
+                logger.info(f"Processed {processed} saga(s) this iteration.")
             else:
                 logger.debug("No sagas to recover.")
         except asyncio.CancelledError:
             logger.info("Recovery loop cancelled.")
             raise
         except Exception as e:
-            logger.exception("Recovery iteration failed: %s", e)
+            logger.exception(f"Recovery iteration failed: {traceback.format_exc()}")
 
         if max_iterations is not None and iteration >= max_iterations:
-            logger.info("Reached max_iterations=%s, stopping.", max_iterations)
+            logger.info(f"Reached max_iterations={max_iterations}, stopping.")
             break
         await asyncio.sleep(interval_seconds)
 

examples/saga_sqlalchemy_storage.py

@@ -140,13 +140,13 @@ async def setup_database(engine: AsyncEngine) -> None:
 
 
 async def main() -> None:
-    # 1. Create SQLAlchemy Engine with Connection Pool
-    # SQLAlchemy creates a pool by default (QueuePool for most dialects, SingletonThreadPool for SQLite)
     """
     Run a demonstration that executes an OrderSaga using an async SQLAlchemy engine and persistent SqlAlchemySagaStorage.
-    
+
     Initializes a pooled async SQLAlchemy engine and schema, creates a session factory and SqlAlchemySagaStorage, bootstraps a mediator with a DI container and saga mapper, runs an OrderSaga while streaming step results to stdout, and then reloads and prints the persisted saga state and step history before disposing the engine.
     """
+    # 1. Create SQLAlchemy Engine with Connection Pool
+    # SQLAlchemy creates a pool by default (QueuePool for most dialects, SingletonThreadPool for SQLite)
     engine = create_async_engine(
         DB_URL,
         echo=False,  # Set to True to see SQL queries

src/cqrs/saga/compensation.py

@@ -105,9 +105,6 @@ async def compensate_steps(
                     "compensate",
                     SagaStepStatus.COMPLETED,
                 )
-                if self._on_after_compensate_step is not None:
-                    await self._on_after_compensate_step()
-
             except Exception as compensation_error:
                 await self._storage.log_step(
                     self._saga_id,
@@ -118,6 +115,19 @@ async def compensate_steps(
                 )
                 # Store both step and error for better error reporting
                 compensation_errors.append((step, compensation_error))
+                continue
+
+            # Callback only after successful step compensation; failures are not treated as step failure
+            if self._on_after_compensate_step is not None:
+                try:
+                    await self._on_after_compensate_step()
+                except Exception as callback_error:
+                    logger.error(
+                        "on_after_compensate_step failed (e.g. run.commit): %s",
+                        callback_error,
+                        exc_info=callback_error,
+                    )
+                    raise
 
         # If compensation failed after all retries
         if compensation_errors:

src/cqrs/saga/saga.py

@@ -86,6 +86,9 @@ def __init__(
         self._completed_steps: list[SagaStepHandler[ContextT, typing.Any]] = []
         self._error: BaseException | None = None
         self._compensated: bool = False
+        self._comp_retry_count = compensation_retry_count
+        self._comp_retry_delay = compensation_retry_delay
+        self._comp_retry_backoff = compensation_retry_backoff
 
         self._saga_id = saga_id or uuid.uuid4()
         self._is_new_saga = saga_id is None
@@ -112,9 +115,9 @@ def __init__(
             self._saga_id,
             context,
             storage,
-            compensation_retry_count,
-            compensation_retry_delay,
-            compensation_retry_backoff,
+            self._comp_retry_count,
+            self._comp_retry_delay,
+            self._comp_retry_backoff,
         )
 
     @property
@@ -172,6 +175,51 @@ async def __aiter__(
             async for step_result in self._execute(None):
                 yield step_result
 
+    def _build_run_scoped_components(
+        self,
+        run: SagaStorageRun,
+    ) -> tuple[
+        SagaStateManager,
+        SagaRecoveryManager,
+        SagaStepExecutor[ContextT],
+        FallbackStepExecutor[ContextT],
+        SagaCompensator[ContextT],
+    ]:
+        """Build state manager, recovery manager, executors, and compensator for a storage run (checkpoint commits)."""
+        state_manager = SagaStateManager(self._saga_id, run)
+        recovery_manager = SagaRecoveryManager(
+            self._saga_id,
+            run,
+            self._container,
+            self._saga.steps,
+        )
+        step_executor = SagaStepExecutor(
+            self._context,
+            self._container,
+            state_manager,
+        )
+        fallback_executor = FallbackStepExecutor(
+            self._context,
+            self._container,
+            state_manager,
+        )
+        compensator = SagaCompensator(
+            self._saga_id,
+            self._context,
+            run,
+            self._comp_retry_count,
+            self._comp_retry_delay,
+            self._comp_retry_backoff,
+            on_after_compensate_step=run.commit,
+        )
+        return (
+            state_manager,
+            recovery_manager,
+            step_executor,
+            fallback_executor,
+            compensator,
+        )
+
     async def _execute(
         self,
         run: SagaStorageRun | None,
@@ -189,32 +237,13 @@ async def _execute(
             RuntimeError: If the saga was recovered in COMPENSATING or FAILED state and compensation was completed, forward execution is not allowed.
         """
         if run is not None:
-            state_manager = SagaStateManager(self._saga_id, run)
-            recovery_manager = SagaRecoveryManager(
-                self._saga_id,
-                run,
-                self._container,
-                self._saga.steps,
-            )
-            step_executor = SagaStepExecutor(
-                self._context,
-                self._container,
-                state_manager,
-            )
-            fallback_executor = FallbackStepExecutor(
-                self._context,
-                self._container,
+            (
                 state_manager,
-            )
-            compensator = SagaCompensator(
-                self._saga_id,
-                self._context,
-                run,
-                self._compensator._retry_count,
-                self._compensator._retry_delay,
-                self._compensator._retry_backoff,
-                on_after_compensate_step=run.commit,
-            )
+                recovery_manager,
+                step_executor,
+                fallback_executor,
+                compensator,
+            ) = self._build_run_scoped_components(run)
         else:
             state_manager = self._state_manager
             recovery_manager = self._recovery_manager

src/cqrs/saga/storage/protocol.py

@@ -19,21 +19,23 @@ async def create_saga(
         saga_id: uuid.UUID,
         name: str,
         context: dict[str, typing.Any],
-    ) -> None: """
+    ) -> 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.
             context (dict[str, Any]): JSON-serializable initial saga context to persist.
         """
-        ...
+
     async def update_context(
         self,
         saga_id: uuid.UUID,
         context: dict[str, typing.Any],
         current_version: int | None = None,
-    ) -> None: """
+    ) -> None:
+        """
         Persist a snapshot of the saga's execution context, optionally using optimistic locking.
         
         Parameters:
@@ -45,12 +47,13 @@ async def update_context(
         Raises:
             SagaConcurrencyError: If `current_version` is provided and does not match the stored version.
         """
-        ...
+
     async def update_status(
         self,
         saga_id: uuid.UUID,
         status: SagaStatus,
-    ) -> None: """
+    ) -> None:
+        """
         Set the global status for the saga identified by `saga_id`.
         
         Parameters:
@@ -60,15 +63,16 @@ async def update_status(
         Notes:
             This operation does not commit the storage session; the caller must call `commit()` on the active run or session to persist the change.
         """
-        ...
+
     async def log_step(
         self,
         saga_id: uuid.UUID,
         step_name: str,
         action: typing.Literal["act", "compensate"],
         status: SagaStepStatus,
         details: str | None = None,
-    ) -> None: """
+    ) -> None:
+        """
         Append a step transition to the saga's execution log.
         
         Parameters:
@@ -78,13 +82,14 @@ async def log_step(
             status (SagaStepStatus): The step transition status to record (e.g., started, completed, failed, compensated).
             details (str | None): Optional human-readable details or diagnostics about the transition.
         """
-        ...
+
     async def load_saga_state(
         self,
         saga_id: uuid.UUID,
         *,
         read_for_update: bool = False,
-    ) -> tuple[SagaStatus, dict[str, typing.Any], int]: """
+    ) -> tuple[SagaStatus, dict[str, typing.Any], int]:
+        """
         Load the current saga execution state.
         
         Parameters:
@@ -94,11 +99,12 @@ async def load_saga_state(
         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]: """
+    ) -> list[SagaLogEntry]:
+        """
         Retrieve the chronological step log for a saga.
         
         Parameters:
@@ -107,19 +113,20 @@ async def get_step_history(
         Returns:
             list[SagaLogEntry]: Ordered list of step log entries for the saga, from oldest to newest.
         """
-        ...
-    async def commit(self) -> None: """
-Finalize the storage run by persisting and committing all pending changes made during this session.
-
-This method makes the run's checkpointed changes durable; the caller is responsible for invoking commit at logical checkpoints to persist session state.
-"""
-...
-    async def rollback(self) -> None: """
-Abort the current storage run and revert any uncommitted changes in the session.
-
-This releases the run's transactional state without persisting pending updates so that the storage remains as it was before the run began.
-"""
-...
+
+    async def commit(self) -> None:
+        """
+        Finalize the storage run by persisting and committing all pending changes made during this session.
+
+        This method makes the run's checkpointed changes durable; the caller is responsible for invoking commit at logical checkpoints to persist session state.
+        """
+
+    async def rollback(self) -> None:
+        """
+        Abort the current storage run and revert any uncommitted changes in the session.
+
+        This releases the run's transactional state without persisting pending updates so that the storage remains as it was before the run began.
+        """
 
 
 class ISagaStorage(abc.ABC):