RFC: Saga Pattern — automatic compensation on pipeline failure

## RFC: Saga Pattern — automatic compensation on pipeline failure

### Summary

Add support for the Saga Pattern, allowing each action in a pipeline to declare a compensating function. When an action fails (after retry exhaustion), the pipeline automatically executes compensations in reverse order for all previously completed actions, restoring the system to a consistent state.

---

### Motivation

Dotflow already supports **retry** for transient failures (network timeout, temporary unavailability). However, there is no mechanism for **business failures** — cases where retrying will never succeed and the correct behavior is to undo what was already done.

Example: a pipeline that charges a credit card, reserves inventory, and creates a shipment. If the shipment step fails, the payment must be refunded and the inventory released.

| Mechanism | When to use |
|-----------|-------------|
| Retry | Transient failures (network timeout, temporary unavailability) |
| Saga | Business failures (out of stock, card declined, invalid data) |

---

### Proposed API

```python
from dotflow import action

def refund_card(ctx):
    refund(ctx["transaction_id"])

def release_inventory(ctx):
    release(ctx["item_id"])

@action(name="charge_card", compensate=refund_card)
def charge_card(ctx):
    return {"transaction_id": "abc123"}

@action(name="reserve_inventory", compensate=release_inventory)
def reserve_inventory(ctx):
    return {"item_id": "xyz789"}

@action(name="create_shipment")
def create_shipment(ctx):
    # no compensate — this is the step that failed, nothing to undo
    ...
```

---

### Execution flow

```mermaid
flowchart LR
    A[Action 1] -->|success| B[Action 2]
    B -->|success| C[Action 3]
    C -->|FAIL| D[Start Saga]
    D --> E[Compensate Action 2]
    E --> F[Compensate Action 1]
    F --> G[Clean state]
```

```mermaid
sequenceDiagram
    participant P as Pipeline
    participant A1 as charge_card
    participant A2 as reserve_inventory
    participant A3 as create_shipment
    participant S as Saga

    P->>A1: execute
    A1-->>P: success (transaction_id)
    P->>A2: execute
    A2-->>P: success (item_id)
    P->>A3: execute
    A3-->>P: FAIL (after retry exhaustion)
    P->>S: trigger compensation
    S->>A2: release_inventory(ctx)
    A2-->>S: done
    S->>A1: refund_card(ctx)
    A1-->>S: done
    S-->>P: saga complete — clean state
```

---

### Behavior rules

1. Compensations run in **reverse order** (LIFO) — only for actions that completed successfully
2. Each compensation receives the context produced by its corresponding action
3. Compensation functions must be **idempotent** — running twice produces the same result
4. Compensation functions should have aggressive retry — a failed compensation leaves the system in an inconsistent state
5. Actions without a `compensate` parameter are skipped during rollback

---

### Task result with Saga

When Saga is triggered, the workflow result should include compensation status:

```json
{
    "workflow_id": "037dbfb0-5f19-4f30-970b-4b00840c58b3",
    "saga": {
        "triggered": true,
        "failed_task_id": 2,
        "compensations": [
            {
                "task_id": 1,
                "name": "release_inventory",
                "status": "Completed",
                "duration": 0.002
            },
            {
                "task_id": 0,
                "name": "refund_card",
                "status": "Completed",
                "duration": 0.015
            }
        ]
    },
    "tasks": ["..."]
}
```

---

### Open questions

- How should Saga interact with **Parallel** execution mode? Compensate all parallel actions that succeeded?
- Should there be a way to opt out of Saga at pipeline level (e.g., `DotFlow(saga=False)`)?
- Should failed compensations raise a `SagaCompensationError` with details of what could not be rolled back?
- Should Saga state be persisted via the storage provider for checkpoint/resume scenarios?
- Should compensations have their own configurable retry/timeout, or inherit from the original action?

---

### Related issues

- #124 (RFC: API contract specification)
- #123 (RFC: Multi-language SDK support)
- #248 (Enhancement: Pass previous_context between task groups)

Navigation Menu

Uh oh!

RFC: Saga Pattern — automatic compensation on pipeline failure #277