Saga Compensations Under Partial Failure: Idempotent Undo Paths That Run

// saga/state.ts
export type StepStatus = "pending" | "committed" | "compensated" | "failed";

export interface SagaStep {
 name: string;
 status: StepStatus;
 forwardIdempotencyKey: string;
 compensateIdempotencyKey: string;
 resultRef?: string; // ARN / paymentId / bookingId
}

export async function markCommitted(sagaId: string, step: string, resultRef: string) {
 // ✅ Conditional write — never advance if already compensating
 await ddb.update({
 Key: { pk: `saga#${sagaId}` },
 ConditionExpression: "attribute_exists(pk) AND #st = :running",
 UpdateExpression:
 "SET steps.#n.#s = :c, steps.#n.resultRef = :r, updatedAt = :t",
 ExpressionAttributeNames: { "#n": step, "#s": "status", "#st": "status" },
 ExpressionAttributeValues: {
 ":c": "committed",
 ":r": resultRef,
 ":running": "running",
 ":t": Date.now(),
 },
 });
}

// saga/compensate.ts
export async function compensatePayment(step: SagaStep) {
 if (!step.resultRef) return; // nothing committed
 await payments.refund({
 paymentId: step.resultRef,
 idempotencyKey: step.compensateIdempotencyKey, // ✅ stable forever
 reason: "saga_rollback",
 });
 // Refund API returns 200 on replay of same key — treat as success
}

export async function runCompensations(sagaId: string, steps: SagaStep[]) {
 // ✅ Reverse order of committed steps only
 for (const step of [...steps].reverse().filter((s) => s.status === "committed")) {
 try {
 await compensateDispatch[step.name](step);
 await markCompensated(sagaId, step.name);
 } catch (err) {
 await markCompensateFailed(sagaId, step.name, err);
 throw err; // leave remaining for redrive — do not invent success
 }
 }
}

# metrics/saga_comp.py
from aws_lambda_powertools import Metrics
from aws_lambda_powertools.metrics import MetricUnit

metrics = Metrics(namespace="Sagas")

def emit_compensate(step: str, ok: bool, latency_ms: float):
 metrics.add_metric("CompensateLatencyMs", MetricUnit.Milliseconds, latency_ms)
 metrics.add_metric(
 "CompensateSuccess" if ok else "CompensateFailure",
 MetricUnit.Count,
 1,
 )
 metrics.add_dimension(name="step", value=step)