crafty_client/
saga.rs

1//! Cross-shard saga coordinator (Tier 2 Phase 4 — framework compensation).
2
3use std::future::Future;
4use std::pin::Pin;
5use std::sync::atomic::{AtomicU32, Ordering};
6use std::sync::{Arc, Mutex};
7
8use crafty_proto::{decode, encode};
9use serde::{Deserialize, Serialize};
10
11use crate::{ClientError, KeyedClient};
12
13/// One forward step with a compensating keyed write on the same shard.
14#[derive(Debug, Clone, PartialEq, Eq)]
15pub struct SagaStep {
16    /// Shard routing key (same shard for forward + compensate).
17    pub key: Vec<u8>,
18    /// Application-encoded forward command.
19    pub command: Vec<u8>,
20    /// Application-encoded compensate command (must be idempotent).
21    pub compensate: Vec<u8>,
22}
23
24/// Ordered cross-shard write plan executed by [`run_saga`].
25#[derive(Debug, Clone, PartialEq, Eq)]
26pub struct SagaPlan {
27    /// Unique saga id (journal key / idempotency scope).
28    pub saga_id: Vec<u8>,
29    /// Steps executed in order; compensators run in reverse on forward failure.
30    pub steps: Vec<SagaStep>,
31}
32
33/// Outcome of a saga run.
34#[derive(Debug)]
35pub enum SagaOutcome {
36    /// Every forward step committed.
37    Completed(Vec<Vec<u8>>),
38    /// Forward failed at `failed_step`; compensators ran for `[0, failed_step)`.
39    Compensated {
40        /// Responses from successful forward steps before the failure.
41        forward_responses: Vec<Vec<u8>>,
42        /// Index of the step that failed.
43        failed_step: usize,
44        /// How many compensate commands succeeded (reverse order).
45        compensated_steps: usize,
46        /// Why the forward step failed.
47        forward_error: ClientError,
48    },
49}
50
51/// Why a saga could not finish cleanly.
52#[derive(Debug, thiserror::Error)]
53pub enum SagaError {
54    /// A compensate command failed after partial forward progress.
55    #[error(
56        "forward failed at step {failed_step} ({forward_completed} committed); \
57         compensation failed at step {compensate_failed_at}: {source}"
58    )]
59    CompensationFailed {
60        /// Forward step that failed.
61        failed_step: usize,
62        /// Forward steps that committed before the failure.
63        forward_completed: usize,
64        /// Compensate step index that failed.
65        compensate_failed_at: usize,
66        /// Forward responses collected before compensation.
67        forward_responses: Vec<Vec<u8>>,
68        #[source]
69        /// Client error from the failed compensate RPC.
70        source: ClientError,
71    },
72    /// Journal persistence failed (saga aborted before mutating client state further).
73    #[error("saga journal error: {0}")]
74    Journal(#[from] SagaJournalError),
75    /// [`resume_saga`] found no record for `plan.saga_id`.
76    #[error("no journal record for this saga")]
77    NotFound,
78    /// Catalog generation changed mid-saga (dynamic catalog expansion).
79    #[error("catalog version changed during saga (pinned {pinned}, current {current})")]
80    CatalogVersionChanged {
81        /// Catalog version pinned when the saga started.
82        pinned: u32,
83        /// Current catalog version observed mid-run.
84        current: u32,
85    },
86}
87
88/// Journal persistence failure.
89#[derive(Debug, thiserror::Error)]
90pub enum SagaJournalError {
91    /// Encode/decode of journal records failed.
92    #[error("codec: {0}")]
93    Codec(String),
94    /// Backend refused the write.
95    #[error("backend: {0}")]
96    Backend(String),
97}
98
99/// Lifecycle events for metrics / logging (see cross-shard-transactions ADR).
100#[derive(Debug, Clone, Copy, PartialEq, Eq)]
101pub enum SagaEvent {
102    /// All forward steps committed.
103    Completed {
104        /// Number of steps executed.
105        steps: usize,
106    },
107    /// Forward failed and compensators ran successfully.
108    Compensated {
109        /// Forward steps that committed before failure.
110        forward_steps: usize,
111        /// Compensate steps that succeeded.
112        compensated: usize,
113    },
114    /// Compensation failed — saga may be stuck; operator intervention required.
115    Stuck {
116        /// Forward step that failed.
117        failed_step: usize,
118        /// Compensate step that failed.
119        compensate_failed_at: usize,
120    },
121}
122
123/// Optional hooks for [`run_saga`].
124#[derive(Default)]
125pub struct RunSagaOpts<'a> {
126    /// Durable (or in-memory) saga journal.
127    pub journal: Option<&'a dyn SagaJournal>,
128    /// Pin catalog version in the journal (dynamic catalog mid-saga).
129    pub catalog_version: Option<u32>,
130    /// Live catalog generation checked before each forward step.
131    pub catalog_version_live: Option<Arc<AtomicU32>>,
132    /// Metrics / logging callback.
133    pub on_event: Option<&'a (dyn Fn(SagaEvent) + Send + Sync)>,
134}
135
136/// Hooks for [`resume_saga`] (journal required).
137pub struct ResumeSagaOpts<'a> {
138    /// Durable saga journal to load progress from.
139    pub journal: &'a dyn SagaJournal,
140    /// Live catalog generation checked before each resumed forward step.
141    pub catalog_version_live: Option<Arc<AtomicU32>>,
142    /// Metrics / logging callback.
143    pub on_event: Option<&'a (dyn Fn(SagaEvent) + Send + Sync)>,
144}
145
146/// Durable saga progress (journal value).
147#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
148pub struct SagaJournalRecord {
149    /// Saga identifier.
150    pub saga_id: Vec<u8>,
151    /// Latest phase.
152    pub phase: SagaJournalPhase,
153    /// Forward steps committed so far.
154    pub completed_steps: u32,
155    /// Catalog version pinned at start, if any.
156    pub catalog_version: Option<u32>,
157    /// Forward step that failed before compensation (if any).
158    #[serde(default)]
159    pub failed_step: Option<u32>,
160    /// Compensate step index that failed when phase is [`SagaJournalPhase::Stuck`].
161    #[serde(default)]
162    pub compensate_failed_at: Option<u32>,
163}
164
165/// Journal phase for resume / observability.
166#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
167pub enum SagaJournalPhase {
168    /// Forward execution in progress.
169    Running,
170    /// All forward steps committed.
171    Completed,
172    /// Running compensators after a forward failure.
173    Compensating,
174    /// Compensators finished (success or partial — see `completed_steps`).
175    Compensated,
176    /// Compensation could not complete.
177    Stuck,
178}
179
180/// Object-safe saga journal (Redis, group-0 side channel, in-memory tests).
181pub trait SagaJournal: Send + Sync {
182    /// Persist saga start.
183    fn on_started<'a>(
184        &'a self,
185        saga_id: &'a [u8],
186        steps: usize,
187        catalog_version: Option<u32>,
188    ) -> Pin<Box<dyn Future<Output = Result<(), SagaJournalError>> + Send + 'a>>;
189
190    /// Persist a committed forward step.
191    fn on_step_committed<'a>(
192        &'a self,
193        saga_id: &'a [u8],
194        step: usize,
195    ) -> Pin<Box<dyn Future<Output = Result<(), SagaJournalError>> + Send + 'a>>;
196
197    /// Persist saga completion.
198    fn on_completed<'a>(
199        &'a self,
200        saga_id: &'a [u8],
201    ) -> Pin<Box<dyn Future<Output = Result<(), SagaJournalError>> + Send + 'a>>;
202
203    /// Persist compensation start after forward failure.
204    fn on_compensation_started<'a>(
205        &'a self,
206        saga_id: &'a [u8],
207        failed_step: usize,
208    ) -> Pin<Box<dyn Future<Output = Result<(), SagaJournalError>> + Send + 'a>>;
209
210    /// Persist successful compensation.
211    fn on_compensated<'a>(
212        &'a self,
213        saga_id: &'a [u8],
214        compensated_steps: usize,
215    ) -> Pin<Box<dyn Future<Output = Result<(), SagaJournalError>> + Send + 'a>>;
216
217    /// Persist stuck saga (compensation failure).
218    fn on_stuck<'a>(
219        &'a self,
220        saga_id: &'a [u8],
221        failed_step: usize,
222        compensate_failed_at: usize,
223    ) -> Pin<Box<dyn Future<Output = Result<(), SagaJournalError>> + Send + 'a>>;
224
225    /// Load the latest journal record for `saga_id`, if any.
226    fn load<'a>(
227        &'a self,
228        saga_id: &'a [u8],
229    ) -> Pin<
230        Box<dyn Future<Output = Result<Option<SagaJournalRecord>, SagaJournalError>> + Send + 'a>,
231    >;
232}
233
234/// In-memory journal for unit tests.
235#[derive(Default)]
236pub struct InMemorySagaJournal {
237    records: Mutex<Vec<SagaJournalRecord>>,
238}
239
240impl InMemorySagaJournal {
241    /// Snapshot persisted records.
242    ///
243    /// # Panics
244    /// Panics if the journal lock is poisoned.
245    #[must_use]
246    pub fn records(&self) -> Vec<SagaJournalRecord> {
247        self.records.lock().expect("lock").clone()
248    }
249
250    fn upsert(&self, saga_id: &[u8], f: impl FnOnce(&mut SagaJournalRecord)) {
251        let mut guard = self.records.lock().expect("lock");
252        if let Some(rec) = guard.iter_mut().find(|r| r.saga_id == saga_id) {
253            f(rec);
254            return;
255        }
256        let mut rec = SagaJournalRecord {
257            saga_id: saga_id.to_vec(),
258            phase: SagaJournalPhase::Running,
259            completed_steps: 0,
260            catalog_version: None,
261            failed_step: None,
262            compensate_failed_at: None,
263        };
264        f(&mut rec);
265        guard.push(rec);
266    }
267}
268
269impl SagaJournal for InMemorySagaJournal {
270    fn on_started<'a>(
271        &'a self,
272        saga_id: &'a [u8],
273        _steps: usize,
274        catalog_version: Option<u32>,
275    ) -> Pin<Box<dyn Future<Output = Result<(), SagaJournalError>> + Send + 'a>> {
276        Box::pin(async move {
277            self.upsert(saga_id, |rec| {
278                rec.phase = SagaJournalPhase::Running;
279                rec.completed_steps = 0;
280                rec.catalog_version = catalog_version;
281                rec.failed_step = None;
282                rec.compensate_failed_at = None;
283            });
284            Ok(())
285        })
286    }
287
288    fn on_step_committed<'a>(
289        &'a self,
290        saga_id: &'a [u8],
291        step: usize,
292    ) -> Pin<Box<dyn Future<Output = Result<(), SagaJournalError>> + Send + 'a>> {
293        Box::pin(async move {
294            self.upsert(saga_id, |rec| {
295                rec.completed_steps = u32::try_from(step).expect("step index fits u32") + 1;
296            });
297            Ok(())
298        })
299    }
300
301    fn on_completed<'a>(
302        &'a self,
303        saga_id: &'a [u8],
304    ) -> Pin<Box<dyn Future<Output = Result<(), SagaJournalError>> + Send + 'a>> {
305        Box::pin(async move {
306            self.upsert(saga_id, |rec| rec.phase = SagaJournalPhase::Completed);
307            Ok(())
308        })
309    }
310
311    fn on_compensation_started<'a>(
312        &'a self,
313        saga_id: &'a [u8],
314        failed_step: usize,
315    ) -> Pin<Box<dyn Future<Output = Result<(), SagaJournalError>> + Send + 'a>> {
316        Box::pin(async move {
317            self.upsert(saga_id, |rec| {
318                rec.phase = SagaJournalPhase::Compensating;
319                rec.failed_step = Some(u32::try_from(failed_step).expect("step index fits u32"));
320            });
321            Ok(())
322        })
323    }
324
325    fn on_compensated<'a>(
326        &'a self,
327        saga_id: &'a [u8],
328        _compensated_steps: usize,
329    ) -> Pin<Box<dyn Future<Output = Result<(), SagaJournalError>> + Send + 'a>> {
330        Box::pin(async move {
331            self.upsert(saga_id, |rec| {
332                rec.phase = SagaJournalPhase::Compensated;
333                rec.compensate_failed_at = None;
334            });
335            Ok(())
336        })
337    }
338
339    fn on_stuck<'a>(
340        &'a self,
341        saga_id: &'a [u8],
342        failed_step: usize,
343        compensate_failed_at: usize,
344    ) -> Pin<Box<dyn Future<Output = Result<(), SagaJournalError>> + Send + 'a>> {
345        Box::pin(async move {
346            self.upsert(saga_id, |rec| {
347                rec.phase = SagaJournalPhase::Stuck;
348                rec.failed_step = Some(u32::try_from(failed_step).expect("step index fits u32"));
349                rec.compensate_failed_at =
350                    Some(u32::try_from(compensate_failed_at).expect("step index fits u32"));
351            });
352            Ok(())
353        })
354    }
355
356    fn load<'a>(
357        &'a self,
358        saga_id: &'a [u8],
359    ) -> Pin<
360        Box<dyn Future<Output = Result<Option<SagaJournalRecord>, SagaJournalError>> + Send + 'a>,
361    > {
362        Box::pin(async move {
363            let guard = self.records.lock().expect("lock");
364            Ok(guard.iter().find(|r| r.saga_id == saga_id).cloned())
365        })
366    }
367}
368
369/// Encode a [`SagaJournalRecord`] for external stores.
370///
371/// # Errors
372/// Returns [`SagaJournalError::Codec`] when encoding fails.
373pub fn encode_journal_record(record: &SagaJournalRecord) -> Result<Vec<u8>, SagaJournalError> {
374    encode(record).map_err(|e| SagaJournalError::Codec(e.to_string()))
375}
376
377/// Decode a [`SagaJournalRecord`].
378///
379/// # Errors
380/// Returns [`SagaJournalError::Codec`] when decoding fails.
381pub fn decode_journal_record(bytes: &[u8]) -> Result<SagaJournalRecord, SagaJournalError> {
382    decode(bytes).map_err(|e| SagaJournalError::Codec(e.to_string()))
383}
384
385fn check_catalog_version(opts: &RunSagaOpts<'_>) -> Result<(), SagaError> {
386    if let (Some(pinned), Some(live)) = (opts.catalog_version, &opts.catalog_version_live) {
387        let current = live.load(Ordering::SeqCst);
388        if current != pinned {
389            return Err(SagaError::CatalogVersionChanged { pinned, current });
390        }
391    }
392    Ok(())
393}
394
395async fn run_forward<C: KeyedClient>(
396    client: &C,
397    plan: &SagaPlan,
398    opts: &RunSagaOpts<'_>,
399    start_step: usize,
400    mut responses: Vec<Vec<u8>>,
401) -> Result<(Vec<Vec<u8>>, Option<(usize, ClientError)>), SagaError> {
402    for (step, item) in plan.steps.iter().enumerate().skip(start_step) {
403        check_catalog_version(opts)?;
404        match client
405            .propose_keyed(item.key.clone(), item.command.clone())
406            .await
407        {
408            Ok(bytes) => {
409                responses.push(bytes);
410                if let Some(journal) = opts.journal {
411                    journal.on_step_committed(&plan.saga_id, step).await?;
412                }
413            }
414            Err(forward_error) => {
415                return Ok((responses, Some((step, forward_error))));
416            }
417        }
418    }
419    Ok((responses, None))
420}
421
422#[allow(clippy::too_many_arguments)]
423async fn run_compensation<C: KeyedClient>(
424    client: &C,
425    plan: &SagaPlan,
426    opts: &RunSagaOpts<'_>,
427    failed_step: usize,
428    forward_responses: Vec<Vec<u8>>,
429    forward_error: ClientError,
430    from_rev: usize,
431    record_compensation_start: bool,
432) -> Result<SagaOutcome, SagaError> {
433    if record_compensation_start && let Some(journal) = opts.journal {
434        journal
435            .on_compensation_started(&plan.saga_id, failed_step)
436            .await?;
437    }
438
439    let mut compensated_steps = 0usize;
440    if failed_step == 0 {
441        if let Some(journal) = opts.journal {
442            journal.on_compensated(&plan.saga_id, 0).await?;
443        }
444        if let Some(on) = opts.on_event {
445            on(SagaEvent::Compensated {
446                forward_steps: 0,
447                compensated: 0,
448            });
449        }
450        return Ok(SagaOutcome::Compensated {
451            forward_responses,
452            failed_step: 0,
453            compensated_steps: 0,
454            forward_error,
455        });
456    }
457
458    for rev in (0..=from_rev).rev() {
459        let back = &plan.steps[rev];
460        if let Err(source) = client
461            .propose_keyed(back.key.clone(), back.compensate.clone())
462            .await
463        {
464            if let Some(journal) = opts.journal {
465                journal.on_stuck(&plan.saga_id, failed_step, rev).await?;
466            }
467            if let Some(on) = opts.on_event {
468                on(SagaEvent::Stuck {
469                    failed_step,
470                    compensate_failed_at: rev,
471                });
472            }
473            return Err(SagaError::CompensationFailed {
474                failed_step,
475                forward_completed: failed_step,
476                compensate_failed_at: rev,
477                forward_responses,
478                source,
479            });
480        }
481        compensated_steps += 1;
482    }
483
484    if let Some(journal) = opts.journal {
485        journal
486            .on_compensated(&plan.saga_id, compensated_steps)
487            .await?;
488    }
489    if let Some(on) = opts.on_event {
490        on(SagaEvent::Compensated {
491            forward_steps: failed_step,
492            compensated: compensated_steps,
493        });
494    }
495    Ok(SagaOutcome::Compensated {
496        forward_responses,
497        failed_step,
498        compensated_steps,
499        forward_error,
500    })
501}
502
503fn run_opts_from_record<'a>(
504    journal: &'a dyn SagaJournal,
505    record: &SagaJournalRecord,
506    catalog_version_live: Option<Arc<AtomicU32>>,
507    on_event: Option<&'a (dyn Fn(SagaEvent) + Send + Sync)>,
508) -> RunSagaOpts<'a> {
509    RunSagaOpts {
510        journal: Some(journal),
511        catalog_version: record.catalog_version,
512        catalog_version_live,
513        on_event,
514    }
515}
516
517/// Execute `plan` forward; on failure run compensators in reverse for committed steps.
518///
519/// **Not** serializable atomicity — see `docs/decisions/multi-raft.md#cross-shard-transactions`.
520///
521/// # Errors
522/// [`SagaError::CompensationFailed`] when a compensate command fails.
523/// [`SagaError::Journal`] when the journal hook returns an error.
524pub async fn run_saga<C: KeyedClient>(
525    client: &C,
526    plan: &SagaPlan,
527    opts: RunSagaOpts<'_>,
528) -> Result<SagaOutcome, SagaError> {
529    if plan.steps.is_empty() {
530        if let Some(on) = opts.on_event {
531            on(SagaEvent::Completed { steps: 0 });
532        }
533        return Ok(SagaOutcome::Completed(Vec::new()));
534    }
535
536    if let Some(journal) = opts.journal {
537        if let Some(record) = journal.load(&plan.saga_id).await?
538            && record.phase == SagaJournalPhase::Completed
539        {
540            if let Some(on) = opts.on_event {
541                on(SagaEvent::Completed {
542                    steps: plan.steps.len(),
543                });
544            }
545            return Ok(SagaOutcome::Completed(Vec::new()));
546        }
547
548        journal
549            .on_started(&plan.saga_id, plan.steps.len(), opts.catalog_version)
550            .await?;
551    }
552
553    let (responses, failure) = run_forward(client, plan, &opts, 0, Vec::new()).await?;
554    if let Some((failed_step, forward_error)) = failure {
555        let from_rev = failed_step.saturating_sub(1);
556        return run_compensation(
557            client,
558            plan,
559            &opts,
560            failed_step,
561            responses,
562            forward_error,
563            from_rev,
564            true,
565        )
566        .await;
567    }
568
569    if let Some(journal) = opts.journal {
570        journal.on_completed(&plan.saga_id).await?;
571    }
572    if let Some(on) = opts.on_event {
573        on(SagaEvent::Completed {
574            steps: plan.steps.len(),
575        });
576    }
577    Ok(SagaOutcome::Completed(responses))
578}
579
580/// Continue a saga from its durable journal record.
581///
582/// # Errors
583/// [`SagaError::NotFound`] when no journal record exists.
584/// Same errors as [`run_saga`] for forward/compensation failures.
585pub async fn resume_saga<C: KeyedClient>(
586    client: &C,
587    plan: &SagaPlan,
588    opts: ResumeSagaOpts<'_>,
589) -> Result<SagaOutcome, SagaError> {
590    let Some(record) = opts.journal.load(&plan.saga_id).await? else {
591        return Err(SagaError::NotFound);
592    };
593
594    let run_opts = run_opts_from_record(
595        opts.journal,
596        &record,
597        opts.catalog_version_live.clone(),
598        opts.on_event,
599    );
600
601    match record.phase {
602        SagaJournalPhase::Completed => {
603            if let Some(on) = opts.on_event {
604                on(SagaEvent::Completed {
605                    steps: plan.steps.len(),
606                });
607            }
608            Ok(SagaOutcome::Completed(Vec::new()))
609        }
610        SagaJournalPhase::Compensated => Ok(SagaOutcome::Compensated {
611            forward_responses: Vec::new(),
612            failed_step: record.failed_step.unwrap_or(0) as usize,
613            compensated_steps: 0,
614            forward_error: ClientError::Server("resumed compensated saga".into()),
615        }),
616        SagaJournalPhase::Running => {
617            let start = record.completed_steps as usize;
618            let (responses, failure) =
619                run_forward(client, plan, &run_opts, start, Vec::new()).await?;
620            if let Some((failed_step, forward_error)) = failure {
621                let from_rev = failed_step.saturating_sub(1);
622                return run_compensation(
623                    client,
624                    plan,
625                    &run_opts,
626                    failed_step,
627                    responses,
628                    forward_error,
629                    from_rev,
630                    true,
631                )
632                .await;
633            }
634            if let Some(journal) = run_opts.journal {
635                journal.on_completed(&plan.saga_id).await?;
636            }
637            if let Some(on) = opts.on_event {
638                on(SagaEvent::Completed {
639                    steps: plan.steps.len(),
640                });
641            }
642            Ok(SagaOutcome::Completed(responses))
643        }
644        SagaJournalPhase::Compensating | SagaJournalPhase::Stuck => {
645            let failed_step = record.failed_step.ok_or(SagaError::NotFound)? as usize;
646            let from_rev = match record.phase {
647                SagaJournalPhase::Stuck => {
648                    record.compensate_failed_at.ok_or(SagaError::NotFound)? as usize
649                }
650                _ => failed_step.saturating_sub(1),
651            };
652            run_compensation(
653                client,
654                plan,
655                &run_opts,
656                failed_step,
657                Vec::new(),
658                ClientError::Server("resumed compensation".into()),
659                from_rev,
660                record.phase == SagaJournalPhase::Compensating,
661            )
662            .await
663        }
664    }
665}
666
667#[cfg(test)]
668mod tests {
669    use std::sync::Arc;
670    use std::sync::atomic::{AtomicU32, Ordering};
671
672    use crafty_net::{Route, Transport, TransportError, decode_body, encode_body};
673    use crafty_proto::{ClientRequest, ClientResponse, NodeId};
674
675    use super::*;
676    use crate::{RemoteClient, RetryPolicy};
677
678    struct SagaScript {
679        forward_ok: u32,
680        compensate_ok: u32,
681        forward_calls: Arc<AtomicU32>,
682        compensate_calls: Arc<AtomicU32>,
683    }
684
685    impl Transport for SagaScript {
686        fn send(
687            &self,
688            _peer: NodeId,
689            _route: Route,
690            body: crafty_net::transport::Body,
691        ) -> crafty_net::transport::BoxFuture<
692            'static,
693            Result<crafty_net::transport::Body, TransportError>,
694        > {
695            let request = match decode_body::<ClientRequest>(&body) {
696                Ok(r) => r,
697                Err(e) => {
698                    return Box::pin(async move { Err(TransportError::Wire(e)) });
699                }
700            };
701            let forward_ok = self.forward_ok;
702            let compensate_ok = self.compensate_ok;
703            let forward_calls = Arc::clone(&self.forward_calls);
704            let compensate_calls = Arc::clone(&self.compensate_calls);
705            Box::pin(async move {
706                match request {
707                    ClientRequest::ProposeKeyed { command, .. } => {
708                        if command.first() == Some(&0xFF) {
709                            let n = compensate_calls.fetch_add(1, Ordering::Relaxed);
710                            if n >= compensate_ok {
711                                return Err(TransportError::Unreachable(NodeId(1)));
712                            }
713                        } else {
714                            let n = forward_calls.fetch_add(1, Ordering::Relaxed);
715                            if n >= forward_ok {
716                                return Err(TransportError::Unreachable(NodeId(1)));
717                            }
718                        }
719                        encode_body(&ClientResponse::Ok(b"ok".to_vec()))
720                            .map_err(TransportError::Wire)
721                    }
722                    _ => Err(TransportError::Io("unexpected request".into())),
723                }
724            })
725        }
726    }
727
728    fn client(script: Arc<SagaScript>) -> RemoteClient {
729        RemoteClient::new(script, [NodeId(1)]).with_retry(RetryPolicy {
730            max_attempts: 1,
731            ..RetryPolicy::default()
732        })
733    }
734
735    fn two_step_plan() -> SagaPlan {
736        SagaPlan {
737            saga_id: b"transfer-1".to_vec(),
738            steps: vec![
739                SagaStep {
740                    key: b"shard-a".to_vec(),
741                    command: vec![1],
742                    compensate: vec![0xFF, 1],
743                },
744                SagaStep {
745                    key: b"shard-b".to_vec(),
746                    command: vec![2],
747                    compensate: vec![0xFF, 2],
748                },
749            ],
750        }
751    }
752
753    #[tokio::test]
754    async fn saga_completes_all_forward_steps() {
755        let script = Arc::new(SagaScript {
756            forward_ok: 2,
757            compensate_ok: 0,
758            forward_calls: Arc::new(AtomicU32::new(0)),
759            compensate_calls: Arc::new(AtomicU32::new(0)),
760        });
761        let client = client(Arc::clone(&script));
762        let journal = InMemorySagaJournal::default();
763        let outcome = run_saga(
764            &client,
765            &two_step_plan(),
766            RunSagaOpts {
767                journal: Some(&journal),
768                ..RunSagaOpts::default()
769            },
770        )
771        .await
772        .expect("saga completes");
773        assert!(matches!(outcome, SagaOutcome::Completed(_)));
774        assert_eq!(script.compensate_calls.load(Ordering::Relaxed), 0);
775        assert!(
776            journal
777                .records()
778                .iter()
779                .any(|r| r.phase == SagaJournalPhase::Completed)
780        );
781    }
782
783    #[tokio::test]
784    async fn saga_compensates_after_second_forward_fails() {
785        let script = Arc::new(SagaScript {
786            forward_ok: 1,
787            compensate_ok: 1,
788            forward_calls: Arc::new(AtomicU32::new(0)),
789            compensate_calls: Arc::new(AtomicU32::new(0)),
790        });
791        let client = client(Arc::clone(&script));
792        let journal = InMemorySagaJournal::default();
793        let outcome = run_saga(
794            &client,
795            &two_step_plan(),
796            RunSagaOpts {
797                journal: Some(&journal),
798                ..RunSagaOpts::default()
799            },
800        )
801        .await
802        .expect("compensated saga");
803        let SagaOutcome::Compensated {
804            failed_step,
805            compensated_steps,
806            ..
807        } = outcome
808        else {
809            panic!("expected compensated, got {outcome:?}");
810        };
811        assert_eq!(failed_step, 1);
812        assert_eq!(compensated_steps, 1);
813        assert_eq!(script.compensate_calls.load(Ordering::Relaxed), 1);
814        assert!(
815            journal
816                .records()
817                .iter()
818                .any(|r| r.phase == SagaJournalPhase::Compensated)
819        );
820    }
821
822    #[tokio::test]
823    async fn saga_stuck_when_compensation_fails() {
824        let script = Arc::new(SagaScript {
825            forward_ok: 1,
826            compensate_ok: 0,
827            forward_calls: Arc::new(AtomicU32::new(0)),
828            compensate_calls: Arc::new(AtomicU32::new(0)),
829        });
830        let client = client(Arc::clone(&script));
831        let err = run_saga(&client, &two_step_plan(), RunSagaOpts::default())
832            .await
833            .unwrap_err();
834        assert!(matches!(err, SagaError::CompensationFailed { .. }));
835    }
836
837    #[tokio::test]
838    async fn resume_saga_continues_forward_after_partial_journal() {
839        let script = Arc::new(SagaScript {
840            forward_ok: 2,
841            compensate_ok: 0,
842            forward_calls: Arc::new(AtomicU32::new(0)),
843            compensate_calls: Arc::new(AtomicU32::new(0)),
844        });
845        let client = client(Arc::clone(&script));
846        let journal = InMemorySagaJournal::default();
847        let plan = two_step_plan();
848        journal
849            .on_started(&plan.saga_id, plan.steps.len(), None)
850            .await
851            .expect("seed");
852        journal
853            .on_step_committed(&plan.saga_id, 0)
854            .await
855            .expect("seed step");
856
857        let outcome = resume_saga(
858            &client,
859            &plan,
860            ResumeSagaOpts {
861                journal: &journal,
862                catalog_version_live: None,
863                on_event: None,
864            },
865        )
866        .await
867        .expect("resume completes");
868        assert!(matches!(outcome, SagaOutcome::Completed(_)));
869        assert_eq!(script.forward_calls.load(Ordering::Relaxed), 1);
870        assert!(
871            journal
872                .records()
873                .iter()
874                .any(|r| r.phase == SagaJournalPhase::Completed)
875        );
876    }
877
878    #[tokio::test]
879    async fn resume_saga_retries_stuck_compensation() {
880        let script = Arc::new(SagaScript {
881            forward_ok: 0,
882            compensate_ok: 1,
883            forward_calls: Arc::new(AtomicU32::new(0)),
884            compensate_calls: Arc::new(AtomicU32::new(0)),
885        });
886        let client = client(Arc::clone(&script));
887        let journal = InMemorySagaJournal::default();
888        let plan = two_step_plan();
889        journal
890            .on_started(&plan.saga_id, plan.steps.len(), None)
891            .await
892            .expect("seed");
893        journal
894            .on_step_committed(&plan.saga_id, 0)
895            .await
896            .expect("seed step");
897        journal
898            .on_compensation_started(&plan.saga_id, 1)
899            .await
900            .expect("seed compensating");
901        journal
902            .on_stuck(&plan.saga_id, 1, 0)
903            .await
904            .expect("seed stuck");
905
906        let outcome = resume_saga(
907            &client,
908            &plan,
909            ResumeSagaOpts {
910                journal: &journal,
911                catalog_version_live: None,
912                on_event: None,
913            },
914        )
915        .await
916        .expect("resume compensation");
917        assert!(matches!(outcome, SagaOutcome::Compensated { .. }));
918        assert_eq!(script.compensate_calls.load(Ordering::Relaxed), 1);
919        assert!(
920            journal
921                .records()
922                .iter()
923                .any(|r| r.phase == SagaJournalPhase::Compensated)
924        );
925    }
926
927    #[tokio::test]
928    async fn saga_rejects_catalog_version_change() {
929        let script = Arc::new(SagaScript {
930            forward_ok: 1,
931            compensate_ok: 0,
932            forward_calls: Arc::new(AtomicU32::new(0)),
933            compensate_calls: Arc::new(AtomicU32::new(0)),
934        });
935        let client = client(Arc::clone(&script));
936        let journal = InMemorySagaJournal::default();
937        let live = Arc::new(AtomicU32::new(2));
938        let err = run_saga(
939            &client,
940            &SagaPlan {
941                saga_id: b"transfer-2".to_vec(),
942                steps: vec![SagaStep {
943                    key: b"shard-a".to_vec(),
944                    command: vec![1],
945                    compensate: vec![0xFF, 1],
946                }],
947            },
948            RunSagaOpts {
949                journal: Some(&journal),
950                catalog_version: Some(1),
951                catalog_version_live: Some(live),
952                ..RunSagaOpts::default()
953            },
954        )
955        .await
956        .unwrap_err();
957        assert!(matches!(
958            err,
959            SagaError::CatalogVersionChanged {
960                pinned: 1,
961                current: 2
962            }
963        ));
964    }
965
966    #[tokio::test]
967    async fn saga_is_idempotent_when_journal_completed() {
968        let script = Arc::new(SagaScript {
969            forward_ok: 0,
970            compensate_ok: 0,
971            forward_calls: Arc::new(AtomicU32::new(0)),
972            compensate_calls: Arc::new(AtomicU32::new(0)),
973        });
974        let client = client(Arc::clone(&script));
975        let journal = InMemorySagaJournal::default();
976        journal.on_started(b"done", 1, Some(1)).await.expect("seed");
977        journal.on_completed(b"done").await.expect("seed complete");
978
979        let outcome = run_saga(
980            &client,
981            &SagaPlan {
982                saga_id: b"done".to_vec(),
983                steps: vec![SagaStep {
984                    key: b"k".to_vec(),
985                    command: vec![1],
986                    compensate: vec![0xFF],
987                }],
988            },
989            RunSagaOpts {
990                journal: Some(&journal),
991                ..RunSagaOpts::default()
992            },
993        )
994        .await
995        .expect("idempotent replay");
996        assert!(matches!(outcome, SagaOutcome::Completed(_)));
997        assert_eq!(script.forward_calls.load(Ordering::Relaxed), 0);
998    }
999}
crafty_client/saga.rs

crafty_client/
saga.rs
