dtmrs_core/
lib.rs

1//! 类型与状态机 —— 纯逻辑，不碰 I/O。
2//!
3//! 分布式事务的 bug 绝大多数在状态迁移上，所以把这层从存储和网络里隔离出来，
4//! 可以纯单元测试覆盖。
5
6pub mod dialect;
7
8pub use dialect::Backend;
9
10use serde::{Deserialize, Serialize};
11use std::fmt;
12
13/// 分支被调用后的结论。**这四态的区分是整个系统的命门。**
14#[derive(Debug, Clone, Copy, PartialEq, Eq)]
15pub enum BranchResult {
16    /// HTTP 200 —— 成功
17    Success,
18    /// HTTP 409 —— 业务**明确**要求回滚。只有这个才触发补偿
19    Failure,
20    /// HTTP 425 —— 还在处理中，别当失败
21    Ongoing,
22    /// 网络错误、超时、5xx —— 结果**未知**
23    ///
24    /// 绝不能当成失败：超时的时候对方可能已经成功了，贸然补偿会造成不一致。
25    /// 正确做法是重试，直到拿到 Success 或 Failure。
26    Unknown,
27}
28
29impl BranchResult {
30    /// 从 HTTP 状态码 + 响应体判定。响应体里的 `dtm_result` 字段优先于状态码，
31    /// 这样业务方用 200 返回 `{"dtm_result":"FAILURE"}` 也能表达失败。
32    pub fn from_http(status: u16, body: &str) -> Self {
33        if body.contains("FAILURE") {
34            return Self::Failure;
35        }
36        if body.contains("ONGOING") {
37            return Self::Ongoing;
38        }
39        match status {
40            200..=299 => Self::Success,
41            409 => Self::Failure,
42            425 => Self::Ongoing,
43            _ => Self::Unknown,
44        }
45    }
46
47    /// 从 gRPC 状态码判定。取值是 gRPC 规范里的标准编号，跟 DTM 的
48    /// `dtmgrpc` 对齐，这样两边的业务服务可以互换。
49    ///
50    /// # 这个映射为什么是这几个码
51    ///
52    /// HTTP 那边靠 409/425 表达「明确失败」和「还在处理」，gRPC 没有这两个码，
53    /// 得从 16 个标准码里各挑一个**不会被基础设施误用**的：
54    ///
55    /// | gRPC 码 | 语义 | 对应 HTTP |
56    /// |---|---|---|
57    /// | `OK`(0) | 成功 | 200 |
58    /// | `ABORTED`(10) | 业务**明确**要求回滚 | 409 |
59    /// | `FAILED_PRECONDITION`(9) | 还在处理，别当失败 | 425 |
60    /// | 其它全部 | 结果**未知**，重试 | 5xx / 超时 |
61    ///
62    /// 关键在最后一行。`UNAVAILABLE`(14)、`DEADLINE_EXCEEDED`(4)、`INTERNAL`(13)
63    /// 这些**都算未知**而不是失败 —— 它们恰恰是网络抖动和超时会产生的码，
64    /// 而超时的时候对方可能已经成功了。这跟 HTTP 侧「超时不等于失败」是同一条命门。
65    ///
66    /// 特别注意 `CANCELLED`(1) 和 `DEADLINE_EXCEEDED`(4)：调用方自己取消/超时
67    /// 产生的码，绝不能当成业务失败 —— 那是**我们这边**放弃了，不是对方拒绝了。
68    pub fn from_grpc(code: i32) -> Self {
69        match code {
70            GRPC_OK => Self::Success,
71            GRPC_ABORTED => Self::Failure,
72            GRPC_FAILED_PRECONDITION => Self::Ongoing,
73            // 包括 CANCELLED / DEADLINE_EXCEEDED / UNAVAILABLE / INTERNAL / …
74            // 一律按未知处理：只重试，不回滚
75            _ => Self::Unknown,
76        }
77    }
78}
79
80/// gRPC 标准状态码。只列用得上的三个，其余一律走 `_ => Unknown`。
81pub const GRPC_OK: i32 = 0;
82/// 业务明确要求回滚。gRPC 侧的 409
83pub const GRPC_ABORTED: i32 = 10;
84/// 还在处理中。gRPC 侧的 425
85pub const GRPC_FAILED_PRECONDITION: i32 = 9;
86
87#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
88#[serde(rename_all = "lowercase")]
89pub enum TransType {
90    Saga,
91    Tcc,
92    Msg,
93    Xa,
94    /// 步骤由**用户函数在运行时决定**，靠重放 + 结果记忆化做崩溃恢复。
95    /// 见 [`workflow_advance`]
96    Workflow,
97}
98
99impl fmt::Display for TransType {
100    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
101        let s = match self {
102            Self::Saga => "saga",
103            Self::Tcc => "tcc",
104            Self::Msg => "msg",
105            Self::Xa => "xa",
106            Self::Workflow => "workflow",
107        };
108        f.write_str(s)
109    }
110}
111
112impl TransType {
113    pub fn parse(s: &str) -> Option<Self> {
114        match s {
115            "saga" => Some(Self::Saga),
116            "tcc" => Some(Self::Tcc),
117            "msg" => Some(Self::Msg),
118            "xa" => Some(Self::Xa),
119            "workflow" => Some(Self::Workflow),
120            _ => None,
121        }
122    }
123}
124
125#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
126#[serde(rename_all = "lowercase")]
127pub enum GlobalStatus {
128    /// 仅二阶段消息用：TC 收到了但还不知道该不该执行
129    Prepared,
130    /// 可以推进
131    Submitted,
132    /// 需要回滚，正在逆序补偿
133    Aborting,
134    /// 终态
135    Succeed,
136    /// 终态
137    Failed,
138}
139
140impl GlobalStatus {
141    pub fn as_str(&self) -> &'static str {
142        match self {
143            Self::Prepared => "prepared",
144            Self::Submitted => "submitted",
145            Self::Aborting => "aborting",
146            Self::Succeed => "succeed",
147            Self::Failed => "failed",
148        }
149    }
150
151    pub fn parse(s: &str) -> Option<Self> {
152        match s {
153            "prepared" => Some(Self::Prepared),
154            "submitted" => Some(Self::Submitted),
155            "aborting" => Some(Self::Aborting),
156            "succeed" => Some(Self::Succeed),
157            "failed" => Some(Self::Failed),
158            _ => None,
159        }
160    }
161
162    /// 终态不再被 cron 调度
163    pub fn is_final(&self) -> bool {
164        matches!(self, Self::Succeed | Self::Failed)
165    }
166}
167
168#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
169#[serde(rename_all = "lowercase")]
170pub enum BranchStatus {
171    Prepared,
172    Succeed,
173    Failed,
174}
175
176impl BranchStatus {
177    pub fn as_str(&self) -> &'static str {
178        match self {
179            Self::Prepared => "prepared",
180            Self::Succeed => "succeed",
181            Self::Failed => "failed",
182        }
183    }
184    pub fn parse(s: &str) -> Option<Self> {
185        match s {
186            "prepared" => Some(Self::Prepared),
187            "succeed" => Some(Self::Succeed),
188            "failed" => Some(Self::Failed),
189            _ => None,
190        }
191    }
192}
193
194/// 分支操作类型。跟 DTM 的字符串保持一致，方便客户端互通。
195#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
196#[serde(rename_all = "lowercase")]
197pub enum BranchOp {
198    Action,
199    Compensate,
200    Try,
201    Confirm,
202    Cancel,
203    /// XA 的二阶段提交
204    Commit,
205    Rollback,
206}
207
208impl BranchOp {
209    pub fn as_str(&self) -> &'static str {
210        match self {
211            Self::Action => "action",
212            Self::Compensate => "compensate",
213            Self::Try => "try",
214            Self::Confirm => "confirm",
215            Self::Cancel => "cancel",
216            Self::Commit => "commit",
217            Self::Rollback => "rollback",
218        }
219    }
220
221    pub fn parse(s: &str) -> Option<Self> {
222        match s {
223            "action" => Some(Self::Action),
224            "compensate" => Some(Self::Compensate),
225            "try" => Some(Self::Try),
226            "confirm" => Some(Self::Confirm),
227            "cancel" => Some(Self::Cancel),
228            "commit" => Some(Self::Commit),
229            "rollback" => Some(Self::Rollback),
230            _ => None,
231        }
232    }
233
234    /// 补偿类操作对应的**正向**操作。屏障判空回滚要用。
235    pub fn origin_op(&self) -> Option<BranchOp> {
236        match self {
237            Self::Cancel => Some(Self::Try),
238            Self::Compensate => Some(Self::Action),
239            Self::Rollback => Some(Self::Action),
240            _ => None,
241        }
242    }
243
244    pub fn is_compensating(&self) -> bool {
245        self.origin_op().is_some()
246    }
247}
248
249/// 一个 SAGA 步骤：正向动作 + 对应补偿 + 这一步的业务数据
250#[derive(Debug, Clone, Serialize, Deserialize)]
251pub struct SagaStep {
252    pub action: String,
253    pub compensate: String,
254    /// 发给这一步分支的请求体。**每步各自独立** ——
255    /// 扣款那步要金额，发货那步要地址，它们本来就不该收到同一份数据。
256    ///
257    /// 留空则发 `{}`。`#[serde(default)]` 保证 0.2 及更早版本落库的
258    /// payload（没有这个字段）仍然能解出来，不会让老事务推不动。
259    #[serde(default)]
260    pub payload: String,
261}
262
263impl SagaStep {
264    /// 不带业务数据的一步（分支只靠 gid/branch_id/op 做幂等就够时用）
265    pub fn new(action: &str, compensate: &str) -> Self {
266        Self {
267            action: action.to_string(),
268            compensate: compensate.to_string(),
269            payload: String::new(),
270        }
271    }
272
273    /// 带业务数据的一步
274    pub fn with_payload(action: &str, compensate: &str, payload: &str) -> Self {
275        Self {
276            action: action.to_string(),
277            compensate: compensate.to_string(),
278            payload: payload.to_string(),
279        }
280    }
281}
282
283/// 推进全局事务后，状态机给出的下一步指令
284#[derive(Debug, Clone, PartialEq, Eq)]
285pub enum Advance {
286    /// 调这个分支
287    Call { index: usize, op: BranchOp },
288    /// 全部完成，落终态
289    Finish(GlobalStatus),
290    /// 有分支 Ongoing/Unknown，本轮到此为止，等下次 cron
291    Wait,
292    /// 跑一遍用户的 workflow 函数（只有 [`TransType::Workflow`] 会出现）。
293    ///
294    /// 单独一个变体而不是复用 `Call`：workflow 的正向走向是**用户函数**决定的，
295    /// 不是状态机决定的。混进 `Call` 会让人以为这里也能算出「下一个分支是谁」。
296    RunWorkflow,
297}
298
299/// SAGA 推进决策。**不碰 I/O，所以可以穷举测试。**
300///
301/// `actions[i]` / `compensates[i]` 是第 i 步两个分支各自的当前状态。
302pub fn saga_advance(
303    status: GlobalStatus,
304    actions: &[BranchStatus],
305    compensates: &[BranchStatus],
306) -> Advance {
307    debug_assert_eq!(actions.len(), compensates.len());
308    match status {
309        GlobalStatus::Submitted => {
310            // 正向：按序找第一个还没成功的
311            for (i, st) in actions.iter().enumerate() {
312                match st {
313                    BranchStatus::Succeed => continue,
314                    BranchStatus::Prepared => {
315                        return Advance::Call {
316                            index: i,
317                            op: BranchOp::Action,
318                        }
319                    }
320                    // 有分支被判失败，本该已经转 aborting；防御性处理
321                    BranchStatus::Failed => return Advance::Finish(GlobalStatus::Aborting),
322                }
323            }
324            Advance::Finish(GlobalStatus::Succeed)
325        }
326        GlobalStatus::Aborting => {
327            // 逆序补偿。**所有分支都补**，不管它的 action 成没成功 ——
328            // action 超时但实际成功的情况必须靠补偿兜住，多余的补偿由屏障空转掉。
329            for i in (0..compensates.len()).rev() {
330                if compensates[i] == BranchStatus::Prepared {
331                    return Advance::Call {
332                        index: i,
333                        op: BranchOp::Compensate,
334                    };
335                }
336            }
337            Advance::Finish(GlobalStatus::Failed)
338        }
339        GlobalStatus::Prepared => Advance::Wait,
340        s => Advance::Finish(s),
341    }
342}
343
344/// 重试退避策略。
345///
346/// 原来这两个值是写死的（10s 起、300s 封顶）。真实业务差异很大：
347/// 秒杀场景希望几百毫秒就重试，跨境对接可能希望几分钟才重试一次。
348#[derive(Debug, Clone, Copy, PartialEq, Eq)]
349pub struct RetryPolicy {
350    /// 第一次重试等多久（秒）
351    pub initial: i64,
352    /// 退避上限（秒）。每次翻倍，到这里为止
353    pub max: i64,
354}
355
356impl Default for RetryPolicy {
357    fn default() -> Self {
358        // 保持跟 0.2 一致的默认值，不配置的人行为不变
359        Self {
360            initial: 10,
361            max: 300,
362        }
363    }
364}
365
366impl RetryPolicy {
367    /// 从环境变量读，非法值一律退回默认 —— **绝不能因为配置写错就让推进器起不来**
368    pub fn from_env() -> Self {
369        let d = Self::default();
370        let get = |k: &str, fallback: i64| {
371            std::env::var(k)
372                .ok()
373                .and_then(|v| v.parse::<i64>().ok())
374                .filter(|v| *v > 0)
375                .unwrap_or(fallback)
376        };
377        let initial = get("DTMRS_RETRY_INTERVAL", d.initial);
378        let max = get("DTMRS_RETRY_MAX_INTERVAL", d.max);
379        // 上限比初始值还小是配置错误，取两者较大的，别让退避反向增长
380        Self {
381            initial,
382            max: max.max(initial),
383        }
384    }
385}
386
387/// 指数退避：`initial` → ×2 → … → 封顶 `max`
388pub fn next_interval_with(cur: i64, p: RetryPolicy) -> i64 {
389    if cur <= 0 {
390        return p.initial;
391    }
392    (cur * 2).min(p.max)
393}
394
395/// 用默认策略退避（10s 起、300s 封顶）
396pub fn next_interval(cur: i64) -> i64 {
397    next_interval_with(cur, RetryPolicy::default())
398}
399
400#[cfg(test)]
401mod tests {
402    use super::*;
403    use BranchStatus::{Failed, Prepared, Succeed};
404
405    #[test]
406    fn 超时不能当失败() {
407        // 这条错了就会造成数据不一致：对方可能已经成功了
408        assert_eq!(BranchResult::from_http(504, ""), BranchResult::Unknown);
409        assert_eq!(BranchResult::from_http(500, ""), BranchResult::Unknown);
410        // 只有明确的 409 / FAILURE 才算失败
411        assert_eq!(BranchResult::from_http(409, ""), BranchResult::Failure);
412        assert_eq!(
413            BranchResult::from_http(200, r#"{"dtm_result":"FAILURE"}"#),
414            BranchResult::Failure
415        );
416        assert_eq!(BranchResult::from_http(425, ""), BranchResult::Ongoing);
417        assert_eq!(BranchResult::from_http(200, "ok"), BranchResult::Success);
418    }
419
420    #[test]
421    fn grpc只有aborted才算失败() {
422        assert_eq!(BranchResult::from_grpc(0), BranchResult::Success);
423        // 业务明确要求回滚 —— gRPC 侧唯一能触发补偿的码
424        assert_eq!(BranchResult::from_grpc(10), BranchResult::Failure);
425        assert_eq!(BranchResult::from_grpc(9), BranchResult::Ongoing);
426
427        // 穷举 gRPC 全部 16 个标准码：除了这三个，一律是 Unknown。
428        // 这条错了就会数据不一致 —— UNAVAILABLE / DEADLINE_EXCEEDED 正是
429        // 网络抖动和超时产生的码，当成失败去回滚，对方可能其实已经成功了。
430        for code in 0..=15 {
431            let want = match code {
432                0 => BranchResult::Success,
433                10 => BranchResult::Failure,
434                9 => BranchResult::Ongoing,
435                _ => BranchResult::Unknown,
436            };
437            assert_eq!(BranchResult::from_grpc(code), want, "gRPC 码 {code} 判错了");
438        }
439        // 几个最容易写错的，单独钉一遍
440        assert_eq!(
441            BranchResult::from_grpc(1),
442            BranchResult::Unknown,
443            "CANCELLED 是我们自己放弃，不是对方拒绝"
444        );
445        assert_eq!(
446            BranchResult::from_grpc(4),
447            BranchResult::Unknown,
448            "DEADLINE_EXCEEDED 绝不能当失败"
449        );
450        assert_eq!(
451            BranchResult::from_grpc(14),
452            BranchResult::Unknown,
453            "UNAVAILABLE 绝不能当失败"
454        );
455        // 不认识的码（未来扩展 / 对方乱返）也必须是 Unknown
456        assert_eq!(BranchResult::from_grpc(99), BranchResult::Unknown);
457        assert_eq!(BranchResult::from_grpc(-1), BranchResult::Unknown);
458    }
459
460    #[test]
461    fn grpc与http的判定语义一致() {
462        // 同一个业务意图，两种协议必须得到同一个结论 —— 否则同一个服务
463        // 换协议接入就会有不同的回滚行为
464        for (http, grpc) in [(200u16, 0i32), (409, 10), (425, 9), (500, 13), (503, 14)] {
465            assert_eq!(
466                BranchResult::from_http(http, ""),
467                BranchResult::from_grpc(grpc),
468                "HTTP {http} 与 gRPC {grpc} 应当判定一致"
469            );
470        }
471    }
472
473    #[test]
474    fn 正向按序推进() {
475        let a = [Prepared, Prepared];
476        let c = [Prepared, Prepared];
477        assert_eq!(
478            saga_advance(GlobalStatus::Submitted, &a, &c),
479            Advance::Call {
480                index: 0,
481                op: BranchOp::Action
482            }
483        );
484        let a = [Succeed, Prepared];
485        assert_eq!(
486            saga_advance(GlobalStatus::Submitted, &a, &c),
487            Advance::Call {
488                index: 1,
489                op: BranchOp::Action
490            }
491        );
492        let a = [Succeed, Succeed];
493        assert_eq!(
494            saga_advance(GlobalStatus::Submitted, &a, &c),
495            Advance::Finish(GlobalStatus::Succeed)
496        );
497    }
498
499    #[test]
500    fn 补偿必须逆序() {
501        let a = [Succeed, Failed];
502        let c = [Prepared, Prepared];
503        // 先补第 1 步（后执行的先回滚）
504        assert_eq!(
505            saga_advance(GlobalStatus::Aborting, &a, &c),
506            Advance::Call {
507                index: 1,
508                op: BranchOp::Compensate
509            }
510        );
511        let c = [Prepared, Succeed];
512        assert_eq!(
513            saga_advance(GlobalStatus::Aborting, &a, &c),
514            Advance::Call {
515                index: 0,
516                op: BranchOp::Compensate
517            }
518        );
519        let c = [Succeed, Succeed];
520        assert_eq!(
521            saga_advance(GlobalStatus::Aborting, &a, &c),
522            Advance::Finish(GlobalStatus::Failed)
523        );
524    }
525
526    #[test]
527    fn 没跑过的分支也要补偿() {
528        // action 全没成功，补偿照样得发 —— 因为 action 可能超时但实际成功了。
529        // 多余的补偿由子事务屏障空转掉，这是安全的一侧。
530        let a = [Prepared, Prepared];
531        let c = [Prepared, Prepared];
532        assert_eq!(
533            saga_advance(GlobalStatus::Aborting, &a, &c),
534            Advance::Call {
535                index: 1,
536                op: BranchOp::Compensate
537            }
538        );
539    }
540
541    #[test]
542    fn 终态不再推进() {
543        for s in [GlobalStatus::Succeed, GlobalStatus::Failed] {
544            assert!(s.is_final());
545            assert_eq!(saga_advance(s, &[], &[]), Advance::Finish(s));
546        }
547    }
548
549    #[test]
550    fn 补偿操作能找到正向操作() {
551        assert_eq!(BranchOp::Compensate.origin_op(), Some(BranchOp::Action));
552        assert_eq!(BranchOp::Cancel.origin_op(), Some(BranchOp::Try));
553        assert_eq!(BranchOp::Action.origin_op(), None);
554        assert!(BranchOp::Compensate.is_compensating());
555        assert!(!BranchOp::Try.is_compensating());
556    }
557
558    #[test]
559    fn 退避有上限() {
560        assert_eq!(next_interval(0), 10);
561        assert_eq!(next_interval(10), 20);
562        assert_eq!(next_interval(200), 300);
563        assert_eq!(next_interval(300), 300);
564    }
565
566    #[test]
567    fn 退避策略可配且默认值不变() {
568        // 不配置的人行为必须跟 0.2 完全一致
569        let d = RetryPolicy::default();
570        assert_eq!((d.initial, d.max), (10, 300));
571
572        // 秒杀那种想快速重试的
573        let fast = RetryPolicy { initial: 1, max: 5 };
574        assert_eq!(next_interval_with(0, fast), 1);
575        assert_eq!(next_interval_with(1, fast), 2);
576        assert_eq!(next_interval_with(4, fast), 5, "封顶");
577        assert_eq!(next_interval_with(5, fast), 5);
578    }
579
580    #[test]
581    fn 上限小于初始值时不让退避反向增长() {
582        // 配置写反了（max < initial）不能导致「重试间隔越来越短」
583        let p = RetryPolicy {
584            initial: 60,
585            max: 10,
586        };
587        let fixed = RetryPolicy {
588            initial: p.initial,
589            max: p.max.max(p.initial),
590        };
591        assert_eq!(next_interval_with(0, fixed), 60);
592        assert_eq!(next_interval_with(60, fixed), 60, "不该缩到 10");
593    }
594
595    #[test]
596    fn 步骤的payload默认为空且能带数据() {
597        let a = SagaStep::new("http://a", "http://c");
598        assert_eq!(a.payload, "");
599        let b = SagaStep::with_payload("http://a", "http://c", r#"{"amount":100}"#);
600        assert_eq!(b.payload, r#"{"amount":100}"#);
601
602        // **老数据必须还能解**：0.2 落库的 payload 里没有 payload 字段
603        let old: SagaStep =
604            serde_json::from_str(r#"{"action":"http://a","compensate":"http://c"}"#)
605                .expect("老格式必须能解，否则升级后存量事务全推不动");
606        assert_eq!(old.payload, "");
607    }
608}
609
610/// TCC 推进决策。
611///
612/// # 跟 SAGA 的关键差别
613///
614/// SAGA 的正向分支（action）返回 FAILURE 会触发逆序补偿。
615/// **TCC 的 confirm 返回 FAILURE 绝不能触发 cancel** —— try 阶段资源已经预留成功、
616/// 全局也已经决定提交了，这时候去 cancel 会把已确认的事务撤掉，造成不一致。
617/// confirm 失败的唯一正确处理是**无限重试 + 报警等人介入**。
618///
619/// 所以这个函数在 Submitted 阶段永远不会返回 `Finish(Aborting)`。
620///
621/// Try 阶段不在这里 —— TCC 的 try 是**客户端自己驱动**的（这也是 TCC 要
622/// `registerBranch` 接口的原因），TC 只负责 confirm/cancel。
623pub fn tcc_advance(
624    status: GlobalStatus,
625    confirms: &[BranchStatus],
626    cancels: &[BranchStatus],
627) -> Advance {
628    debug_assert_eq!(confirms.len(), cancels.len());
629    match status {
630        // 客户端还在跑 try，TC 不插手
631        GlobalStatus::Prepared => Advance::Wait,
632        GlobalStatus::Submitted => {
633            for (i, st) in confirms.iter().enumerate() {
634                match st {
635                    BranchStatus::Succeed => continue,
636                    // Failed 也要继续重试 —— 见上面注释，绝不转 aborting
637                    BranchStatus::Prepared | BranchStatus::Failed => {
638                        return Advance::Call {
639                            index: i,
640                            op: BranchOp::Confirm,
641                        }
642                    }
643                }
644            }
645            Advance::Finish(GlobalStatus::Succeed)
646        }
647        GlobalStatus::Aborting => {
648            // 逆序 cancel。全部 cancel，空回滚由屏障负责
649            for i in (0..cancels.len()).rev() {
650                if cancels[i] != BranchStatus::Succeed {
651                    return Advance::Call {
652                        index: i,
653                        op: BranchOp::Cancel,
654                    };
655                }
656            }
657            Advance::Finish(GlobalStatus::Failed)
658        }
659        s => Advance::Finish(s),
660    }
661}
662
663/// 二阶段消息推进决策。
664///
665/// # 这个模式解决什么
666///
667/// "本地事务 + 可靠消息" —— 取代 RocketMQ 那类事务消息，不需要 MQ。
668/// 流程：`prepare` 落库 → 业务提交本地事务 → `submit`。
669/// 如果进程在两者之间崩了，TC 会回查业务方（`query_prepared`）问这单到底成没成。
670///
671/// # 没有补偿
672///
673/// msg 只保证"最终一定送达"，分支必须最终成功（幂等 + 无限重试）。
674/// 所以正向分支返回 FAILURE **不触发补偿**（压根没有补偿分支），
675/// 只能重试。真要放弃只能靠 `query_prepared` 回答 FAILURE 让整单作废。
676pub fn msg_advance(status: GlobalStatus, actions: &[BranchStatus]) -> Advance {
677    match status {
678        // 等 cron 去回查 query_prepared
679        GlobalStatus::Prepared => Advance::Wait,
680        GlobalStatus::Submitted => {
681            for (i, st) in actions.iter().enumerate() {
682                if *st != BranchStatus::Succeed {
683                    return Advance::Call {
684                        index: i,
685                        op: BranchOp::Action,
686                    };
687                }
688            }
689            Advance::Finish(GlobalStatus::Succeed)
690        }
691        // 回查得到 FAILURE：整单作废，没有补偿可做
692        GlobalStatus::Aborting => Advance::Finish(GlobalStatus::Failed),
693        s => Advance::Finish(s),
694    }
695}
696
697#[cfg(test)]
698mod tcc_msg_tests {
699    use super::*;
700    use BranchStatus::{Failed, Prepared, Succeed};
701
702    #[test]
703    fn tcc的try阶段tc不插手() {
704        assert_eq!(
705            tcc_advance(GlobalStatus::Prepared, &[Prepared], &[Prepared]),
706            Advance::Wait
707        );
708    }
709
710    #[test]
711    fn tcc按序confirm() {
712        let c = [Prepared, Prepared];
713        let x = [Prepared, Prepared];
714        assert_eq!(
715            tcc_advance(GlobalStatus::Submitted, &c, &x),
716            Advance::Call {
717                index: 0,
718                op: BranchOp::Confirm
719            }
720        );
721        assert_eq!(
722            tcc_advance(GlobalStatus::Submitted, &[Succeed, Prepared], &x),
723            Advance::Call {
724                index: 1,
725                op: BranchOp::Confirm
726            }
727        );
728        assert_eq!(
729            tcc_advance(GlobalStatus::Submitted, &[Succeed, Succeed], &x),
730            Advance::Finish(GlobalStatus::Succeed)
731        );
732    }
733
734    #[test]
735    fn confirm失败绝不能触发cancel() {
736        // 这是 TCC 最容易写错的地方：try 已成功、已决定提交，
737        // 这时候 cancel 会把已确认的事务撤掉 —— 必须重试而不是回滚
738        let c = [Succeed, Failed];
739        let x = [Prepared, Prepared];
740        assert_eq!(
741            tcc_advance(GlobalStatus::Submitted, &c, &x),
742            Advance::Call {
743                index: 1,
744                op: BranchOp::Confirm
745            },
746            "confirm 失败要继续重试 confirm，不能转 cancel"
747        );
748        // 穷举：Submitted 阶段永远不会返回 Aborting
749        for a in [Prepared, Succeed, Failed] {
750            for b in [Prepared, Succeed, Failed] {
751                let r = tcc_advance(GlobalStatus::Submitted, &[a, b], &x);
752                assert_ne!(r, Advance::Finish(GlobalStatus::Aborting));
753                assert_ne!(r, Advance::Finish(GlobalStatus::Failed));
754            }
755        }
756    }
757
758    #[test]
759    fn tcc逆序cancel() {
760        let c = [Prepared, Prepared];
761        assert_eq!(
762            tcc_advance(GlobalStatus::Aborting, &c, &[Prepared, Prepared]),
763            Advance::Call {
764                index: 1,
765                op: BranchOp::Cancel
766            }
767        );
768        assert_eq!(
769            tcc_advance(GlobalStatus::Aborting, &c, &[Prepared, Succeed]),
770            Advance::Call {
771                index: 0,
772                op: BranchOp::Cancel
773            }
774        );
775        assert_eq!(
776            tcc_advance(GlobalStatus::Aborting, &c, &[Succeed, Succeed]),
777            Advance::Finish(GlobalStatus::Failed)
778        );
779        // cancel 失败也要重试，不能就这么算了
780        assert_eq!(
781            tcc_advance(GlobalStatus::Aborting, &c, &[Succeed, Failed]),
782            Advance::Call {
783                index: 1,
784                op: BranchOp::Cancel
785            }
786        );
787    }
788
789    #[test]
790    fn msg等回查而不是自己推() {
791        assert_eq!(
792            msg_advance(GlobalStatus::Prepared, &[Prepared]),
793            Advance::Wait
794        );
795    }
796
797    #[test]
798    fn msg只往前不补偿() {
799        assert_eq!(
800            msg_advance(GlobalStatus::Submitted, &[Prepared, Prepared]),
801            Advance::Call {
802                index: 0,
803                op: BranchOp::Action
804            }
805        );
806        // 分支失败也只能重试 —— msg 没有补偿分支
807        assert_eq!(
808            msg_advance(GlobalStatus::Submitted, &[Failed]),
809            Advance::Call {
810                index: 0,
811                op: BranchOp::Action
812            }
813        );
814        assert_eq!(
815            msg_advance(GlobalStatus::Submitted, &[Succeed, Succeed]),
816            Advance::Finish(GlobalStatus::Succeed)
817        );
818        // 回查得到 FAILURE → 整单作废，无补偿可做
819        assert_eq!(
820            msg_advance(GlobalStatus::Aborting, &[Succeed]),
821            Advance::Finish(GlobalStatus::Failed)
822        );
823    }
824}
825
826/// XA 推进决策。
827///
828/// # 跟 TCC 同一条铁律
829///
830/// 分支一旦 `PREPARE TRANSACTION` 成功、全局又决定了提交，**commit 失败绝不能
831/// 转成 rollback** —— 别的分支可能已经 COMMIT PREPARED 了，这时候回滚就是
832/// 一半提交一半回滚。只能无限重试 + 报警。
833///
834/// 所以跟 `tcc_advance` 一样，Submitted 阶段永远不返回 `Finish(Aborting)`。
835///
836/// # XA 独有的危险
837///
838/// 已 prepare 未解决的事务会**一直持有锁**，在 Postgres 里还会阻塞 VACUUM
839/// 导致事务 ID 回卷风险。所以 XA 的 commit/rollback 必须最终送达 ——
840/// 这比 SAGA/TCC 的"补偿没跑成"严重得多。运维上要监控 `pg_prepared_xacts`。
841pub fn xa_advance(
842    status: GlobalStatus,
843    commits: &[BranchStatus],
844    rollbacks: &[BranchStatus],
845) -> Advance {
846    debug_assert_eq!(commits.len(), rollbacks.len());
847    match status {
848        // 客户端还在各分支上跑业务 SQL + PREPARE，TC 不插手
849        GlobalStatus::Prepared => Advance::Wait,
850        GlobalStatus::Submitted => {
851            for (i, st) in commits.iter().enumerate() {
852                if *st != BranchStatus::Succeed {
853                    return Advance::Call {
854                        index: i,
855                        op: BranchOp::Commit,
856                    };
857                }
858            }
859            Advance::Finish(GlobalStatus::Succeed)
860        }
861        GlobalStatus::Aborting => {
862            for i in (0..rollbacks.len()).rev() {
863                if rollbacks[i] != BranchStatus::Succeed {
864                    return Advance::Call {
865                        index: i,
866                        op: BranchOp::Rollback,
867                    };
868                }
869            }
870            Advance::Finish(GlobalStatus::Failed)
871        }
872        s => Advance::Finish(s),
873    }
874}
875
876/// workflow 推进决策。
877///
878/// # 这个模式跟前四种的结构性差别
879///
880/// SAGA / TCC / msg / XA 的步骤都是**提前声明**的，所以状态机能算出「下一步调谁」。
881/// workflow 反过来：步骤是**用户函数在运行时决定**的 —— 可以有 `if`、有循环、
882/// 有依赖前一步返回值的分叉。这是它存在的全部理由，也是它没法被本函数算出来的原因。
883///
884/// 所以这里只拥有三件事（这三件仍然可以穷举测试）：
885///
886/// 1. 什么时候该跑那个函数（Submitted → [`Advance::RunWorkflow`]）
887/// 2. 什么时候该补偿、按什么顺序（Aborting → 逆序）
888/// 3. 什么时候落终态
889///
890/// 「下一步调谁」交给用户函数，靠**重放 + 结果记忆化**保证崩溃后不重做。
891///
892/// # 补偿为什么能复用逆序那套
893///
894/// 用户函数每跑到一个分支就**动态登记**它的补偿（跟 TCC 的 `registerBranch`
895/// 一个形状），登记的行落在 `trans_branch_op` 里。所以回滚阶段跟 SAGA 完全一样：
896/// 逆序扫补偿行。区别只是这些行是运行时长出来的，不是提交时一次性写好的。
897///
898/// # 只补偿「登记过」的分支
899///
900/// 跟 SAGA「补偿所有分支」看着不同，其实是同一条规则：没跑到的分支压根没登记，
901/// 也就没有副作用要收拾。**关键在于补偿必须先于正向动作登记** ——
902/// 这样即使正向动作超时或进程当场崩了，补偿也已经在库里了，不会漏。
903/// 这跟 TCC「必须先 registerBranch 再调 try」是同一条教训。
904pub fn workflow_advance(status: GlobalStatus, compensates: &[BranchStatus]) -> Advance {
905    match status {
906        // workflow 没有 prepare 阶段，出现就是数据有问题，别乱动
907        GlobalStatus::Prepared => Advance::Wait,
908        GlobalStatus::Submitted => Advance::RunWorkflow,
909        GlobalStatus::Aborting => {
910            // 逆序补偿。失败的也要重试 —— 补偿没跑成就是真的漏了副作用
911            for i in (0..compensates.len()).rev() {
912                if compensates[i] != BranchStatus::Succeed {
913                    return Advance::Call {
914                        index: i,
915                        op: BranchOp::Compensate,
916                    };
917                }
918            }
919            Advance::Finish(GlobalStatus::Failed)
920        }
921        s => Advance::Finish(s),
922    }
923}
924
925#[cfg(test)]
926mod workflow_tests {
927    use super::*;
928    use BranchStatus::{Failed, Prepared, Succeed};
929
930    #[test]
931    fn submitted就是去跑函数() {
932        assert_eq!(
933            workflow_advance(GlobalStatus::Submitted, &[]),
934            Advance::RunWorkflow
935        );
936        // 已经登记了几个分支也一样 —— 该不该跑下一步是函数自己的事，
937        // 重放时靠记忆化跳过已完成的
938        assert_eq!(
939            workflow_advance(GlobalStatus::Submitted, &[Succeed, Prepared]),
940            Advance::RunWorkflow
941        );
942    }
943
944    #[test]
945    fn 回滚时逆序补偿() {
946        assert_eq!(
947            workflow_advance(GlobalStatus::Aborting, &[Prepared, Prepared]),
948            Advance::Call {
949                index: 1,
950                op: BranchOp::Compensate
951            },
952            "后执行的先回滚"
953        );
954        assert_eq!(
955            workflow_advance(GlobalStatus::Aborting, &[Prepared, Succeed]),
956            Advance::Call {
957                index: 0,
958                op: BranchOp::Compensate
959            }
960        );
961        assert_eq!(
962            workflow_advance(GlobalStatus::Aborting, &[Succeed, Succeed]),
963            Advance::Finish(GlobalStatus::Failed)
964        );
965        // 补偿失败要接着重试，不能就这么算了 —— 那是真的漏了副作用
966        assert_eq!(
967            workflow_advance(GlobalStatus::Aborting, &[Succeed, Failed]),
968            Advance::Call {
969                index: 1,
970                op: BranchOp::Compensate
971            }
972        );
973    }
974
975    #[test]
976    fn 一个分支都没登记就回滚是直接失败() {
977        // 函数第一步就要求回滚，还没来得及登记任何补偿 —— 没有副作用要收拾
978        assert_eq!(
979            workflow_advance(GlobalStatus::Aborting, &[]),
980            Advance::Finish(GlobalStatus::Failed)
981        );
982    }
983
984    #[test]
985    fn 终态不再推进() {
986        for s in [GlobalStatus::Succeed, GlobalStatus::Failed] {
987            assert_eq!(workflow_advance(s, &[]), Advance::Finish(s));
988        }
989    }
990
991    #[test]
992    fn workflow是一种事务类型() {
993        assert_eq!(TransType::parse("workflow"), Some(TransType::Workflow));
994        assert_eq!(TransType::Workflow.to_string(), "workflow");
995    }
996}
997
998#[cfg(test)]
999mod xa_tests {
1000    use super::*;
1001    use BranchStatus::{Failed, Prepared, Succeed};
1002
1003    #[test]
1004    fn xa的prepare阶段tc不插手() {
1005        // 各分支的业务 SQL + PREPARE TRANSACTION 都是客户端自己做的
1006        assert_eq!(
1007            xa_advance(GlobalStatus::Prepared, &[Prepared], &[Prepared]),
1008            Advance::Wait
1009        );
1010    }
1011
1012    #[test]
1013    fn xa按序commit() {
1014        let r = [Prepared, Prepared];
1015        assert_eq!(
1016            xa_advance(GlobalStatus::Submitted, &[Prepared, Prepared], &r),
1017            Advance::Call {
1018                index: 0,
1019                op: BranchOp::Commit
1020            }
1021        );
1022        assert_eq!(
1023            xa_advance(GlobalStatus::Submitted, &[Succeed, Prepared], &r),
1024            Advance::Call {
1025                index: 1,
1026                op: BranchOp::Commit
1027            }
1028        );
1029        assert_eq!(
1030            xa_advance(GlobalStatus::Submitted, &[Succeed, Succeed], &r),
1031            Advance::Finish(GlobalStatus::Succeed)
1032        );
1033    }
1034
1035    #[test]
1036    fn commit失败绝不能转rollback() {
1037        // 别的分支可能已经 COMMIT PREPARED 了，这时候回滚就是一半提交一半回滚
1038        let r = [Prepared, Prepared];
1039        assert_eq!(
1040            xa_advance(GlobalStatus::Submitted, &[Succeed, Failed], &r),
1041            Advance::Call {
1042                index: 1,
1043                op: BranchOp::Commit
1044            },
1045            "commit 失败要继续重试 commit"
1046        );
1047        // 穷举：Submitted 阶段永远不会走向回滚或失败
1048        for a in [Prepared, Succeed, Failed] {
1049            for b in [Prepared, Succeed, Failed] {
1050                let got = xa_advance(GlobalStatus::Submitted, &[a, b], &r);
1051                assert_ne!(got, Advance::Finish(GlobalStatus::Aborting));
1052                assert_ne!(got, Advance::Finish(GlobalStatus::Failed));
1053            }
1054        }
1055    }
1056
1057    #[test]
1058    fn xa逆序rollback且失败也要重试() {
1059        let c = [Prepared, Prepared];
1060        assert_eq!(
1061            xa_advance(GlobalStatus::Aborting, &c, &[Prepared, Prepared]),
1062            Advance::Call {
1063                index: 1,
1064                op: BranchOp::Rollback
1065            }
1066        );
1067        assert_eq!(
1068            xa_advance(GlobalStatus::Aborting, &c, &[Succeed, Succeed]),
1069            Advance::Finish(GlobalStatus::Failed)
1070        );
1071        // rollback 失败也不能就这么算了 —— 那会留下永久持锁的 prepared 事务
1072        assert_eq!(
1073            xa_advance(GlobalStatus::Aborting, &c, &[Succeed, Failed]),
1074            Advance::Call {
1075                index: 1,
1076                op: BranchOp::Rollback
1077            }
1078        );
1079    }
1080
1081    #[test]
1082    fn commit操作没有反向映射() {
1083        // XA 的 commit 不是补偿类操作，屏障不该给它做空回滚判定
1084        assert_eq!(BranchOp::Commit.origin_op(), None);
1085        assert!(!BranchOp::Commit.is_compensating());
1086        assert_eq!(BranchOp::parse("commit"), Some(BranchOp::Commit));
1087        assert_eq!(BranchOp::Commit.as_str(), "commit");
1088    }
1089}
dtmrs_core/lib.rs

dtmrs_core/
lib.rs
