# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference) # Canonical Residual: https://corpusfetch.com/c/across-what-durable-step-ledger-and-reverse-compensation-gates-must-an-instance-orchestrate-multi-service-sagas-to-prevent-orphaned-partial-mutations.md # Remnant Warrant: https://corpusfetch.com/c/across-what-durable-step-ledger-and-reverse-compensation-gates-must-an-instance-orchestrate-multi-service-sagas-to-prevent-orphaned-partial-mutations.warrant.json # Markdown SHA-256: f499dd57885ad194c71f8bf890c0a47172256f024c3c48443b227e32016ab860 # Code AST SHA-256: bc7628f1c4527b6529a8ac7018c9cfc73f8eceed8a7b4c5500dfdf467d1a7698 # Verified Classes: SagaAbortedError, SagaStepSpec, DurableSagaOrchestrator # Defensive Guards: 1 fail-closed invariants (zero external deps) import json import sqlite3 import time from dataclasses import dataclass from typing import Any, Callable class SagaAbortedError(RuntimeError): '''Raised when a forward saga step fails and reverse LIFO compensation has been executed.''' def __init__(self, saga_id: str, failed_step: str, fully_compensated: bool, cause: BaseException) -> None: self.saga_id = saga_id self.failed_step = failed_step self.fully_compensated = fully_compensated self.cause = cause super().__init__( f"Saga {saga_id!r} failed at step {failed_step!r} " f"(fully_compensated={fully_compensated}): {cause}" ) @dataclass(frozen=True) class SagaStepSpec: ''' Defines a single saga step: - `forward_fn(context_dict, forward_idempotency_key) -> dict_updates` - `compensate_fn(context_dict, compensate_idempotency_key) -> None` (Must be idempotent and safe to call even if `forward_fn` timed out mid-flight). ''' name: str forward_fn: Callable[[dict[str, Any], str], dict[str, Any] | None] compensate_fn: Callable[[dict[str, Any], str], None] | None = None class DurableSagaOrchestrator: ''' Executes multi-step workflows with: 1. Write-ahead step intent logging (`STARTED` persisted BEFORE network call). 2. Deterministic forward and compensation idempotency keys (`{saga_id}:{step}:fwd|comp`). 3. Automatic reverse-order (LIFO) compensation across all STARTED/SUCCEEDED steps on failure. ''' def __init__( self, conn: sqlite3.Connection, *, compensation_max_retries: int = 3, compensation_backoff_sec: float = 0.05, ) -> None: self.conn = conn self.compensation_max_retries = max(1, compensation_max_retries) self.compensation_backoff_sec = max(0.0, compensation_backoff_sec) self._init_ledger() def _init_ledger(self) -> None: with self.conn: self.conn.execute( "CREATE TABLE IF NOT EXISTS saga_step_ledger (" " saga_id TEXT NOT NULL," " step_index INTEGER NOT NULL," " step_name TEXT NOT NULL," " status TEXT NOT NULL CHECK(status IN (" " 'STARTED', 'SUCCEEDED', 'COMPENSATED', 'COMPENSATION_FAILED'" " ))," " fwd_idempotency_key TEXT NOT NULL," " comp_idempotency_key TEXT NOT NULL," " context_snapshot_json TEXT NOT NULL," " error_detail TEXT," " updated_at_epoch REAL NOT NULL," " PRIMARY KEY (saga_id, step_index)" ")" ) def _record_step_state( self, saga_id: str, step_index: int, step_name: str, status: str, context: dict[str, Any], error_detail: str | None = None, ) -> tuple[str, str]: fwd_key = f"{saga_id}:{step_name}:fwd" comp_key = f"{saga_id}:{step_name}:comp" with self.conn: self.conn.execute( "INSERT INTO saga_step_ledger " "(saga_id, step_index, step_name, status, fwd_idempotency_key, " " comp_idempotency_key, context_snapshot_json, error_detail, updated_at_epoch) " "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) " "ON CONFLICT(saga_id, step_index) DO UPDATE SET " " status = excluded.status," " context_snapshot_json = excluded.context_snapshot_json," " error_detail = excluded.error_detail," " updated_at_epoch = excluded.updated_at_epoch", ( saga_id, step_index, step_name, status, fwd_key, comp_key, json.dumps(context, sort_keys=True), error_detail, time.time(), ), ) return fwd_key, comp_key def execute_saga( self, saga_id: str, steps: list[SagaStepSpec], initial_context: dict[str, Any] | None = None, ) -> dict[str, Any]: ''' Runs `steps` sequentially. If any step raises an exception, immediately runs reverse-order (LIFO) compensations for that step (since it was marked STARTED before the wire call) and all preceding SUCCEEDED steps. ''' ctx: dict[str, Any] = dict(initial_context or {}) executed_indices: list[int] = [] for idx, step in enumerate(steps): # Gate 2: Write-ahead STARTED state BEFORE invoking remote API! fwd_key, _ = self._record_step_state(saga_id, idx, step.name, "STARTED", ctx) executed_indices.append(idx) try: updates = step.forward_fn(ctx, fwd_key) if isinstance(updates, dict): ctx.update(updates) self._record_step_state(saga_id, idx, step.name, "SUCCEEDED", ctx) except BaseException as exc: # Gate 4: Trigger reverse (LIFO) compensation across all touched steps all_compensated = self._compensate_lifo(saga_id, steps, executed_indices, ctx) raise SagaAbortedError( saga_id=saga_id, failed_step=step.name, fully_compensated=all_compensated, cause=exc, ) from exc return ctx def _compensate_lifo( self, saga_id: str, steps: list[SagaStepSpec], executed_indices: list[int], ctx: dict[str, Any], ) -> bool: all_clean = True for idx in reversed(executed_indices): step = steps[idx] if step.compensate_fn is None: self._record_step_state(saga_id, idx, step.name, "COMPENSATED", ctx) continue comp_key = f"{saga_id}:{step.name}:comp" last_err: str | None = None succeeded = False for attempt in range(self.compensation_max_retries): try: step.compensate_fn(ctx, comp_key) succeeded = True break except Exception as comp_exc: last_err = str(comp_exc) if attempt + 1 < self.compensation_max_retries and self.compensation_backoff_sec > 0: time.sleep(self.compensation_backoff_sec * (2 ** attempt)) if succeeded: self._record_step_state(saga_id, idx, step.name, "COMPENSATED", ctx) else: all_clean = False self._record_step_state( saga_id, idx, step.name, "COMPENSATION_FAILED", ctx, error_detail=last_err ) return all_clean