feat(flare-workflow): implement saga rollback/compensation handlers

crates/flare-workflow/src/definition.rs

@@ -33,6 +33,13 @@ pub enum ValidationError {
     /// A cycle was detected in the workflow DAG.
     #[error("cycle detected involving step '{0}'")]
     CycleDetected(StepId),
+
+    /// A step registered a rollback handler on a mode that never journals a
+    /// full context snapshot on success, so there's nothing to reconstruct
+    /// a rollback context from. Only `Sequential`, `FanOut`, and `Loop`
+    /// support rollback.
+    #[error("step '{step}' registers a rollback handler on unsupported mode '{mode}'")]
+    RollbackUnsupportedMode { step: StepId, mode: &'static str },
 }
 
 /// Definition of a single step within a workflow.
@@ -58,6 +65,14 @@ pub struct StepDefinition<D: WorkflowData> {
     /// Named variable to store this step's output in for later `{{var}}`
     /// references (OpenFang semantics).
     pub output_var: Option<String>,
+    /// Saga compensation handler, invoked during rollback if the workflow
+    /// later fails. Only `Sequential`, `FanOut`, and `Loop` modes support
+    /// this — see `ValidationError::RollbackUnsupportedMode`.
+    pub rollback: Option<Arc<dyn StepExecutor<D>>>,
+    /// Retry policy for the rollback handler itself. Falls back to the
+    /// step's own `retry_policy` (then the workflow default) when unset.
+    /// Reuses the step's own `timeout`.
+    pub rollback_retry_policy: Option<RetryPolicy>,
 }
 
 impl<D: WorkflowData> Clone for StepDefinition<D> {
@@ -77,6 +92,8 @@ impl<D: WorkflowData> Clone for StepDefinition<D> {
             scheduled_at: self.scheduled_at,
             run_if: self.run_if.clone(),
             output_var: self.output_var.clone(),
+            rollback: self.rollback.clone(),
+            rollback_retry_policy: self.rollback_retry_policy.clone(),
         }
     }
 }
@@ -96,6 +113,8 @@ impl<D: WorkflowData> fmt::Debug for StepDefinition<D> {
             .field("delay", &self.delay)
             .field("scheduled_at", &self.scheduled_at)
             .field("run_if", &self.run_if.as_ref().map(|_| "<condition>"))
+            .field("rollback", &self.rollback.as_ref().map(|_| "<rollback>"))
+            .field("rollback_retry_policy", &self.rollback_retry_policy)
             .finish_non_exhaustive()
     }
 }
@@ -121,6 +140,8 @@ impl<D: WorkflowData> StepDefinition<D> {
             scheduled_at: None,
             run_if: None,
             output_var: None,
+            rollback: None,
+            rollback_retry_policy: None,
         }
     }
 
@@ -189,6 +210,22 @@ impl<D: WorkflowData> StepDefinition<D> {
         self
     }
 
+    /// Register a saga compensation handler, invoked during rollback if the
+    /// workflow later fails. `retry_policy` overrides this handler's own
+    /// retry policy; when `None`, it falls back to the step's own
+    /// `retry_policy`, then the workflow default. Only `Sequential`,
+    /// `FanOut`, and `Loop` step modes support rollback — registering it on
+    /// any other mode is rejected by `WorkflowDefinition::validate()`.
+    pub fn with_rollback(
+        mut self,
+        executor: Arc<dyn StepExecutor<D>>,
+        retry_policy: Option<RetryPolicy>,
+    ) -> Self {
+        self.rollback = Some(executor);
+        self.rollback_retry_policy = retry_policy;
+        self
+    }
+
     /// Iterator over all dependencies (both `depends_on` and `depends_on_any`).
     pub fn all_dependencies(&self) -> impl Iterator<Item = &StepId> {
         self.depends_on.iter().chain(self.depends_on_any.iter())
@@ -294,6 +331,27 @@ impl<D: WorkflowData> WorkflowDefinition<D> {
             }
         }
 
+        for step in &self.steps {
+            if step.rollback.is_none() {
+                continue;
+            }
+            // Only modes whose successful completion journals a full
+            // context snapshot can be reconstructed for a rollback handler.
+            let unsupported_mode = match &step.mode {
+                StepMode::Sequential | StepMode::FanOut | StepMode::Loop { .. } => None,
+                StepMode::Collect => Some("collect"),
+                StepMode::Conditional { .. } => Some("conditional"),
+                StepMode::Sleep { .. } => Some("sleep"),
+                StepMode::WaitEvent { .. } => Some("wait_event"),
+            };
+            if let Some(mode) = unsupported_mode {
+                return Err(ValidationError::RollbackUnsupportedMode {
+                    step: step.id.clone(),
+                    mode,
+                });
+            }
+        }
+
         let mut visited = HashSet::new();
         let mut rec_stack = HashSet::new();
 

crates/flare-workflow/src/engine.rs

@@ -142,7 +142,7 @@ pub struct WorkflowEngine<D: WorkflowData, S: StateStore<D> = InMemoryStore<D>>
     shutdown_tx: Arc<watch::Sender<bool>>,
     active_workflows: Arc<AtomicUsize>,
     /// Jitter factor applied to retry backoff delays (0.0-1.0).
-    jitter: f64,
+    pub(crate) jitter: f64,
     /// In-process completions for pending `WaitEvent` steps, keyed by
     /// `"{run_id}:{step_id}:{name}"`. A completed event is also journaled so
     /// it survives restart and pre-delivery races.
@@ -684,20 +684,13 @@ impl<D: WorkflowData, S: StateStore<D> + 'static> WorkflowEngine<D, S> {
                 } else {
                     "Workflow deadlocked: no steps ready and none running"
                 };
-                self.state_store
-                    .update(run_id, |s| {
-                        s.status = WorkflowStatus::Failed;
-                        s.error = Some(error_message.to_string());
-                    })
-                    .await?;
-                self.event_bus
-                    .publish(WorkflowEvent::WorkflowFailed {
-                        run_id,
-                        failed_step: failed_step
-                            .unwrap_or_else(|| StepId::new("internal_scheduler")),
-                        error: error_message.to_string(),
-                    })
-                    .await;
+                self.finish_workflow_failed(
+                    run_id,
+                    &definition,
+                    failed_step,
+                    error_message.to_string(),
+                )
+                .await?;
                 return Ok(());
             }
 
@@ -930,20 +923,14 @@ impl<D: WorkflowData, S: StateStore<D> + 'static> WorkflowEngine<D, S> {
             t.failed.iter().next().cloned()
         };
 
-        if let Some(ref step) = failed_step {
-            self.state_store
-                .update(run_id, |s| {
-                    s.status = WorkflowStatus::Failed;
-                    s.error = Some("One or more steps failed".to_string());
-                })
-                .await?;
-            self.event_bus
-                .publish(WorkflowEvent::WorkflowFailed {
-                    run_id,
-                    failed_step: step.clone(),
-                    error: "One or more steps failed".into(),
-                })
-                .await;
+        if let Some(step) = failed_step {
+            self.finish_workflow_failed(
+                run_id,
+                &definition,
+                Some(step),
+                "One or more steps failed".to_string(),
+            )
+            .await?;
         } else {
             let output = self
                 .state_store
@@ -975,6 +962,51 @@ impl<D: WorkflowData, S: StateStore<D> + 'static> WorkflowEngine<D, S> {
         Ok(())
     }
 
+    /// Settle a run to `WorkflowStatus::Failed`, running the saga rollback
+    /// phase first if any step in `definition` registered a `rollback`
+    /// handler. `failed_step` is the specific step whose failure triggered
+    /// the workflow's failure, if any (a genuine deadlock with no failed
+    /// step passes `None`). Zero overhead for workflows with no registered
+    /// rollbacks: the `.any(...)` check short-circuits before any journal
+    /// reads.
+    async fn finish_workflow_failed(
+        &self,
+        run_id: WorkflowRunId,
+        definition: &WorkflowDefinition<D>,
+        failed_step: Option<StepId>,
+        error_message: String,
+    ) -> WorkflowResult<()> {
+        if definition.steps.iter().any(|s| s.rollback.is_some()) {
+            self.run_rollback_phase(run_id, definition, failed_step.as_ref())
+                .await?;
+        }
+        // Status flips to `Failed` only here, after the rollback phase
+        // returns — a crash mid-unwind leaves the run `Running`, so
+        // `recover()` (which only resumes `Running`/`Pending` runs) picks it
+        // back up and `run_rollback_phase` resumes from whatever `Rollback`
+        // entries already exist.
+        self.state_store
+            .update(run_id, |s| {
+                s.status = WorkflowStatus::Failed;
+                // `run_rollback_phase` may have already folded a
+                // compensation-failure note into `s.error` above; append the
+                // primary failure reason rather than clobbering it.
+                s.error = Some(match s.error.take() {
+                    Some(rollback_note) => format!("{error_message}; {rollback_note}"),
+                    None => error_message.clone(),
+                });
+            })
+            .await?;
+        self.event_bus
+            .publish(WorkflowEvent::WorkflowFailed {
+                run_id,
+                failed_step: failed_step.unwrap_or_else(|| StepId::new("internal_scheduler")),
+                error: error_message,
+            })
+            .await;
+        Ok(())
+    }
+
     /// Execute a step with retry logic, appending the terminal result to the
     /// durable journal so recovery never re-executes completed steps.
     async fn execute_step_with_retry(
@@ -1194,7 +1226,7 @@ impl<D: WorkflowData, S: StateStore<D> + 'static> WorkflowEngine<D, S> {
         };
         let step_timeout = definition.get_timeout(step);
         let until_lower = until.to_lowercase();
-        let mut current_output = String::new();
+        let mut last_context: Option<WorkflowContext<D>> = None;
         let mut executed = 0u32;
 
         for iter in 1..=*max_iterations {
@@ -1244,11 +1276,9 @@ impl<D: WorkflowData, S: StateStore<D> + 'static> WorkflowEngine<D, S> {
                             }
                         })
                         .await?;
-                    current_output = out;
+                    last_context = Some(context.clone());
                     tracing::info!(run_id = %run_id, step = %step.id, iter, "Loop iteration completed");
-                    if !until_lower.is_empty()
-                        && current_output.to_lowercase().contains(&until_lower)
-                    {
+                    if !until_lower.is_empty() && out.to_lowercase().contains(&until_lower) {
                         break;
                     }
                 }
@@ -1287,14 +1317,23 @@ impl<D: WorkflowData, S: StateStore<D> + 'static> WorkflowEngine<D, S> {
             }
         }
 
-        // Journal the loop's terminal success (last output is the payload).
+        // Journal the loop's terminal success carrying the full serialized
+        // context (matching Sequential/FanOut's terminal `StepRun` entry) so
+        // a registered rollback handler can reconstruct this step's own
+        // output from the journal, same as any other supported mode.
+        let final_context = match last_context {
+            Some(ctx) => ctx,
+            None => self.state_store.load(run_id).await?.context,
+        };
+        let context_bytes = serde_json::to_vec(&final_context)
+            .map_err(|e| WorkflowError::Journal(format!("serialize context: {e}")))?;
         self.state_store
             .append_journal(
                 run_id,
                 JournalEntry::StepRun {
                     step_id: step.id.clone(),
                     attempt: executed,
-                    result: Some(EntryResult::Success(current_output.into_bytes())),
+                    result: Some(EntryResult::Success(context_bytes)),
                 },
             )
             .await?;

crates/flare-workflow/src/events.rs

@@ -59,6 +59,22 @@ pub enum WorkflowEvent {
     WorkflowCancelled {
         run_id: WorkflowRunId,
     },
+    /// A saga rollback (compensation) handler exhausted its retries. The
+    /// unwind continues regardless — see `RollbackCompleted` for the final
+    /// tally.
+    RollbackStepFailed {
+        run_id: WorkflowRunId,
+        step_id: StepId,
+        error: String,
+    },
+    /// The rollback phase for a failed run has finished (best-effort): every
+    /// eligible step's compensation was attempted, whether or not it
+    /// succeeded.
+    RollbackCompleted {
+        run_id: WorkflowRunId,
+        compensated: Vec<StepId>,
+        failed: Vec<StepId>,
+    },
 }
 
 /// Trait for subscribing to workflow events.
@@ -231,6 +247,20 @@ impl EventSubscriber for LoggingSubscriber {
             WorkflowEvent::WorkflowCancelled { run_id } => {
                 info!(run_id = %run_id, "Workflow cancelled");
             }
+            WorkflowEvent::RollbackStepFailed {
+                run_id,
+                step_id,
+                error,
+            } => {
+                warn!(run_id = %run_id, step_id = %step_id, error = error, "Rollback step failed");
+            }
+            WorkflowEvent::RollbackCompleted {
+                run_id,
+                compensated,
+                failed,
+            } => {
+                info!(run_id = %run_id, compensated = compensated.len(), failed = failed.len(), "Rollback phase completed");
+            }
         }
     }
 }

crates/flare-workflow/src/lib.rs

@@ -19,6 +19,7 @@ pub mod executor;
 pub mod journal;
 pub mod json;
 pub mod retry;
+pub mod rollback;
 pub mod sqlite_store;
 pub mod store;
 pub mod types;