dtmrs_server/
driver.rs

1//! 事务推进器：把状态机的决策落成真实的 HTTP 调用和状态更新。
2//!
3//! 这里是**唯一**会修改全局事务状态的地方，决策全部来自 `dtmrs_core::saga_advance`，
4//! 本文件只负责 I/O。这样状态迁移的正确性可以在 core 里纯单测覆盖。
5
6use crate::registry::{parse_target, BranchCtx, Registry, Target};
7use dtmrs_core::{
8    msg_advance, saga_advance, tcc_advance, xa_advance, Advance, BranchOp, BranchResult,
9    BranchStatus, GlobalStatus, SagaStep, TransType,
10};
11use dtmrs_store::{GlobalRow, Store};
12use std::sync::Arc;
13use std::time::Duration;
14use tracing::{info, warn};
15
16#[derive(Clone)]
17pub struct Driver {
18    pub store: Store,
19    pub http: reqwest::Client,
20    pub owner: String,
21    /// 租约时长（秒）。持租约的实例崩了，这么久之后别的实例接手
22    pub lease: i64,
23    /// 进程内分支注册表。嵌入式模式用，纯 HTTP 部署时是空表
24    pub registry: Arc<Registry>,
25    /// gRPC 分支调用器（带 channel 缓存）
26    #[cfg(feature = "grpc")]
27    pub grpc: crate::grpc::client::GrpcCaller,
28    /// workflow 函数注册表。跟 registry 一样，纯 HTTP 部署时是空表
29    pub workflows: Arc<crate::workflow::WorkflowRegistry>,
30}
31
32impl Driver {
33    pub fn new(store: Store, owner: String) -> Self {
34        Self {
35            store,
36            http: reqwest::Client::builder()
37                .timeout(Duration::from_secs(10))
38                .build()
39                .expect("build http client"),
40            owner,
41            lease: 30,
42            registry: Arc::new(Registry::new()),
43            #[cfg(feature = "grpc")]
44            grpc: crate::grpc::client::GrpcCaller::new(Duration::from_secs(10)),
45            workflows: Arc::new(crate::workflow::WorkflowRegistry::new()),
46        }
47    }
48
49    /// 挂上进程内分支注册表 —— 嵌入式模式的入口
50    pub fn with_registry(mut self, r: Arc<Registry>) -> Self {
51        self.registry = r;
52        self
53    }
54
55    /// 挂上 workflow 函数注册表
56    pub fn with_workflows(mut self, w: Arc<crate::workflow::WorkflowRegistry>) -> Self {
57        self.workflows = w;
58        self
59    }
60
61    /// 常驻循环：抢一个到期事务推一下，没活就睡
62    pub async fn run_forever(self, tick: Duration) {
63        loop {
64            match self.store.lock_one_due(&self.owner, self.lease).await {
65                Ok(Some(g)) => {
66                    if let Err(e) = self.process(&g).await {
67                        warn!(gid = %g.gid, error = %e, "推进出错，等下轮重试");
68                    }
69                }
70                Ok(None) => tokio::time::sleep(tick).await,
71                Err(e) => {
72                    warn!(error = %e, "取待办失败");
73                    tokio::time::sleep(tick).await;
74                }
75            }
76        }
77    }
78
79    /// 推进一个全局事务，直到它落终态或需要等待。
80    ///
81    /// 可以被重复调用（崩溃恢复就靠这个）—— 分支的幂等由客户端屏障保证。
82    pub async fn process(&self, g: &GlobalRow) -> anyhow::Result<()> {
83        match g.trans_type {
84            TransType::Saga => self.process_saga(g).await,
85            TransType::Tcc => self.process_tcc(g).await,
86            TransType::Msg => self.process_msg(g).await,
87            TransType::Xa => self.process_xa(g).await,
88            TransType::Workflow => self.process_workflow(g).await,
89        }
90    }
91
92    // ---------------- workflow ----------------
93
94    /// 跑用户的 workflow 函数，失败则逆序补偿它**已经登记过**的分支。
95    ///
96    /// 跟另外四种模式的差别见 [`dtmrs_core::workflow_advance`]：正向走向由
97    /// 用户函数决定，状态机只管「什么时候跑、什么时候补、按什么顺序补」。
98    async fn process_workflow(&self, g: &GlobalRow) -> anyhow::Result<()> {
99        let (name, input) = crate::workflow::decode_payload(&g.payload);
100        let mut status = g.status;
101
102        loop {
103            let rows = self.store.list_branches(&g.gid).await?;
104            let compensates = compensate_states(&rows);
105
106            match dtmrs_core::workflow_advance(status, &compensates) {
107                Advance::Finish(s) => {
108                    info!(gid = %g.gid, status = s.as_str(), "workflow 事务终结");
109                    self.store.set_global_status(&g.gid, s, "").await?;
110                    return Ok(());
111                }
112                Advance::Wait => return Ok(()),
113
114                Advance::RunWorkflow => {
115                    let Some(f) = self.workflows.get(&name) else {
116                        // 漏注册（新版本删了 workflow / 换了名字）。
117                        // **按结果未知处理** —— 这是部署问题，改回来重试就好，
118                        // 判失败会白白触发回滚
119                        warn!(gid = %g.gid, workflow = %name,
120                              "workflow 未注册，按结果未知处理（会重试，不回滚）");
121                        self.retry_later(g).await?;
122                        return Ok(());
123                    };
124                    let ctx =
125                        crate::workflow::WorkflowCtx::new(&g.gid, &input, self.store.clone(), rows);
126                    match f(ctx).await {
127                        Ok(()) => {
128                            info!(gid = %g.gid, workflow = %name, "workflow 跑完");
129                            self.store
130                                .set_global_status(&g.gid, GlobalStatus::Succeed, "")
131                                .await?;
132                            return Ok(());
133                        }
134                        Err(crate::workflow::WorkflowError::Rollback(reason)) => {
135                            info!(gid = %g.gid, workflow = %name, %reason, "workflow 要求回滚");
136                            self.store
137                                .set_global_status(&g.gid, GlobalStatus::Aborting, &reason)
138                                .await?;
139                            status = GlobalStatus::Aborting;
140                            continue;
141                        }
142                        Err(crate::workflow::WorkflowError::Diverged {
143                            branch_id: bid,
144                            recorded,
145                            got,
146                        }) => {
147                            // **绝不能继续**：按位置记忆化会张冠李戴，回滚也会补错对象。
148                            // 也不回滚 —— 我们已经不知道真实进度了，硬回滚更危险。
149                            // 停在这里等人：改回确定性的代码，重启就能接着推。
150                            warn!(gid = %g.gid, workflow = %name, branch = %bid,
151                                  %recorded, %got,
152                                  "workflow 重放走岔了，已停止推进，需要人工介入");
153                            self.retry_later(g).await?;
154                            return Ok(());
155                        }
156                        Err(e) => {
157                            // Retry / Internal：只重试，绝不回滚
158                            warn!(gid = %g.gid, workflow = %name, error = %e, "workflow 需要重试");
159                            self.retry_later(g).await?;
160                            return Ok(());
161                        }
162                    }
163                }
164
165                Advance::Call { index, op } => {
166                    let bid = branch_id(index);
167                    let Some(url) = url_of(&rows, &bid, op) else {
168                        // 补偿行不见了，不该发生
169                        warn!(gid = %g.gid, branch = %bid, "workflow 补偿地址缺失");
170                        self.retry_later(g).await?;
171                        return Ok(());
172                    };
173                    match self.call_branch(g, &bid, op, &url).await {
174                        BranchResult::Success => {
175                            self.store
176                                .set_branch_status(&g.gid, &bid, op, BranchStatus::Succeed)
177                                .await?;
178                        }
179                        // 补偿失败只能不停重试 —— 漏掉就是真的漏了副作用
180                        _ => {
181                            warn!(gid = %g.gid, branch = %bid, "workflow 补偿未成功，会重试");
182                            self.retry_later(g).await?;
183                            return Ok(());
184                        }
185                    }
186                }
187            }
188        }
189    }
190
191    // ---------------- SAGA ----------------
192
193    async fn process_saga(&self, g: &GlobalRow) -> anyhow::Result<()> {
194        let steps: Vec<SagaStep> = serde_json::from_str(&g.payload).unwrap_or_default();
195        if steps.is_empty() {
196            self.store
197                .set_global_status(&g.gid, GlobalStatus::Succeed, "")
198                .await?;
199            return Ok(());
200        }
201        let mut status = g.status;
202
203        loop {
204            let (actions, compensates) = self.branch_states(&g.gid, steps.len()).await?;
205            match saga_advance(status, &actions, &compensates) {
206                Advance::Finish(s) => {
207                    if s == GlobalStatus::Aborting {
208                        // 防御性分支：状态机发现有 failed 分支但全局还没转 aborting
209                        status = s;
210                        self.store
211                            .set_global_status(&g.gid, s, "分支已判失败")
212                            .await?;
213                        continue;
214                    }
215                    info!(gid = %g.gid, status = s.as_str(), "事务终结");
216                    self.store.set_global_status(&g.gid, s, "").await?;
217                    return Ok(());
218                }
219                Advance::Wait => return Ok(()),
220                // 只有 workflow 模式会出现，别的模式走到这里说明状态机接错了。
221                // **不 panic** —— 推进器是常驻的，崩了整个 TC 就停了
222                Advance::RunWorkflow => {
223                    warn!(gid = %g.gid, "非 workflow 事务收到 RunWorkflow 决策，跳过");
224                    return Ok(());
225                }
226                Advance::Call { index, op } => {
227                    let branch_id = branch_id(index);
228                    let url = match op {
229                        BranchOp::Action => &steps[index].action,
230                        _ => &steps[index].compensate,
231                    };
232                    match self.call_branch(g, &branch_id, op, url).await {
233                        BranchResult::Success => {
234                            self.store
235                                .set_branch_status(&g.gid, &branch_id, op, BranchStatus::Succeed)
236                                .await?;
237                        }
238                        BranchResult::Failure => {
239                            self.store
240                                .set_branch_status(&g.gid, &branch_id, op, BranchStatus::Failed)
241                                .await?;
242                            if op == BranchOp::Action {
243                                // 只有业务**明确**说失败才回滚
244                                info!(gid = %g.gid, branch = %branch_id, "分支要求回滚");
245                                status = GlobalStatus::Aborting;
246                                self.store
247                                    .set_global_status(
248                                        &g.gid,
249                                        GlobalStatus::Aborting,
250                                        &format!("分支 {branch_id} 返回 FAILURE"),
251                                    )
252                                    .await?;
253                            } else {
254                                // 补偿都失败了，只能不停重试 —— 这时候需要人介入
255                                warn!(gid = %g.gid, branch = %branch_id, "补偿失败，需要人工介入");
256                                self.retry_later(g).await?;
257                                return Ok(());
258                            }
259                        }
260                        BranchResult::Ongoing | BranchResult::Unknown => {
261                            // **绝不能当成失败**：对方可能已经成功了。退避重试。
262                            self.retry_later(g).await?;
263                            return Ok(());
264                        }
265                    }
266                }
267            }
268        }
269    }
270
271    // ---------------- TCC ----------------
272
273    /// TCC 的 try 阶段是**客户端驱动**的（客户端先 registerBranch 再调 try），
274    /// TC 只负责 confirm / cancel。所以分支的 URL 来自 `trans_branch_op` 表，
275    /// 不是全局 payload。
276    async fn process_tcc(&self, g: &GlobalRow) -> anyhow::Result<()> {
277        self.drive_two_phase(g, BranchOp::Confirm, BranchOp::Cancel, "TCC")
278            .await
279    }
280
281    // ---------------- XA ----------------
282
283    /// XA 的一阶段（业务 SQL + `PREPARE TRANSACTION`）由**客户端**做，
284    /// TC 只负责统一决定 `COMMIT PREPARED` 还是 `ROLLBACK PREPARED`。
285    ///
286    /// 形状跟 TCC 一样，只是 op 换成 commit/rollback。语义上那条铁律也一样：
287    /// **commit 失败绝不能转 rollback** —— 别的分支可能已经提交了。
288    ///
289    /// XA 独有的严重性：没解决的 prepared 事务会**永久持锁**，在 Postgres 里
290    /// 还阻塞 VACUUM。所以这里的重试比 SAGA 的补偿重试要紧得多。
291    async fn process_xa(&self, g: &GlobalRow) -> anyhow::Result<()> {
292        self.drive_two_phase(g, BranchOp::Commit, BranchOp::Rollback, "XA")
293            .await
294    }
295
296    /// TCC 和 XA 共用的二阶段推进：正向 op 全做完就成功，反向 op 全做完就失败，
297    /// **任一方向的失败都只重试，绝不改变方向**。
298    async fn drive_two_phase(
299        &self,
300        g: &GlobalRow,
301        fwd: BranchOp,
302        bwd: BranchOp,
303        label: &str,
304    ) -> anyhow::Result<()> {
305        let rows = self.store.list_branches(&g.gid).await?;
306        let n = rows
307            .iter()
308            .filter_map(|r| index_of(&r.branch_id))
309            .max()
310            .map(|m| m + 1)
311            .unwrap_or(0);
312        if n == 0 {
313            // 一个分支都没登记就 submit/abort 了 —— 空事务，直接落终态
314            let s = if g.status == GlobalStatus::Aborting {
315                GlobalStatus::Failed
316            } else {
317                GlobalStatus::Succeed
318            };
319            self.store.set_global_status(&g.gid, s, "").await?;
320            return Ok(());
321        }
322
323        let status = g.status;
324        loop {
325            let rows = self.store.list_branches(&g.gid).await?;
326            let (f, b) = split_by_op(&rows, n, fwd, bwd);
327            let adv = if fwd == BranchOp::Commit {
328                xa_advance(status, &f, &b)
329            } else {
330                tcc_advance(status, &f, &b)
331            };
332            match adv {
333                Advance::Finish(s) => {
334                    info!(gid = %g.gid, status = s.as_str(), mode = label, "事务终结");
335                    self.store.set_global_status(&g.gid, s, "").await?;
336                    return Ok(());
337                }
338                Advance::Wait => return Ok(()),
339                // 只有 workflow 模式会出现，别的模式走到这里说明状态机接错了。
340                // **不 panic** —— 推进器是常驻的，崩了整个 TC 就停了
341                Advance::RunWorkflow => {
342                    warn!(gid = %g.gid, "非 workflow 事务收到 RunWorkflow 决策，跳过");
343                    return Ok(());
344                }
345                Advance::Call { index, op } => {
346                    let bid = branch_id(index);
347                    let Some(url) = url_of(&rows, &bid, op) else {
348                        warn!(gid = %g.gid, branch = %bid, op = op.as_str(), mode = label,
349                              "分支没登记这个操作的 URL，无法调用");
350                        self.retry_later(g).await?;
351                        return Ok(());
352                    };
353                    match self.call_branch(g, &bid, op, &url).await {
354                        BranchResult::Success => {
355                            self.store
356                                .set_branch_status(&g.gid, &bid, op, BranchStatus::Succeed)
357                                .await?;
358                        }
359                        // 二阶段失败**绝不改变全局方向**：一阶段已经成功、
360                        // 方向已经定了，反向操作会造成一半提交一半回滚。
361                        // 唯一正确处理是无限重试 + 报警。
362                        BranchResult::Failure => {
363                            self.store
364                                .set_branch_status(&g.gid, &bid, op, BranchStatus::Failed)
365                                .await?;
366                            warn!(gid = %g.gid, branch = %bid, op = op.as_str(), mode = label,
367                                  "二阶段失败，会持续重试，需要人工介入");
368                            self.retry_later(g).await?;
369                            return Ok(());
370                        }
371                        BranchResult::Ongoing | BranchResult::Unknown => {
372                            self.retry_later(g).await?;
373                            return Ok(());
374                        }
375                    }
376                }
377            }
378        }
379    }
380
381    // ---------------- 二阶段消息 ----------------
382
383    /// 流程：`prepare` 落库 → 业务提交本地事务 → `submit`。
384    ///
385    /// 如果进程在这两步之间崩了，事务会一直停在 prepared。这时 TC 靠回查
386    /// `query_prepared` 问业务方"你那个本地事务到底提交了没有"，据此决定
387    /// 是往前推还是整单作废。**这是取代 MQ 事务消息的关键一环。**
388    async fn process_msg(&self, g: &GlobalRow) -> anyhow::Result<()> {
389        let mut status = g.status;
390
391        if status == GlobalStatus::Prepared {
392            if g.query_prepared.is_empty() {
393                // 没给回查地址就没法自动决断。不能瞎猜 —— 猜错要么丢单要么重复扣款
394                warn!(gid = %g.gid, "msg 事务没提供 query_prepared，无法回查，等人处理");
395                self.retry_later(g).await?;
396                return Ok(());
397            }
398            // 借用分支调用的通道做回查，branch_id 用 "00" 跟真实分支区分开
399            match self
400                .call_branch(g, "00", BranchOp::Action, &g.query_prepared)
401                .await
402            {
403                BranchResult::Success => {
404                    info!(gid = %g.gid, "回查：本地事务已提交 → 继续推进");
405                    self.store
406                        .set_global_status(&g.gid, GlobalStatus::Submitted, "")
407                        .await?;
408                    status = GlobalStatus::Submitted;
409                }
410                BranchResult::Failure => {
411                    // 业务方明确说"这单没提交" → 整单作废。msg 没有补偿分支，
412                    // 但也不需要：正向分支压根还没跑过
413                    info!(gid = %g.gid, "回查：本地事务未提交 → 整单作废");
414                    self.store
415                        .set_global_status(
416                            &g.gid,
417                            GlobalStatus::Failed,
418                            "回查得到 FAILURE：本地事务未提交",
419                        )
420                        .await?;
421                    return Ok(());
422                }
423                BranchResult::Ongoing | BranchResult::Unknown => {
424                    // 回查本身失败了，不能当作"没提交"。退避重试。
425                    self.retry_later(g).await?;
426                    return Ok(());
427                }
428            }
429        }
430
431        let steps: Vec<SagaStep> = serde_json::from_str(&g.payload).unwrap_or_default();
432        if steps.is_empty() {
433            self.store
434                .set_global_status(&g.gid, GlobalStatus::Succeed, "")
435                .await?;
436            return Ok(());
437        }
438        loop {
439            let (actions, _) = self.branch_states(&g.gid, steps.len()).await?;
440            match msg_advance(status, &actions) {
441                Advance::Finish(s) => {
442                    info!(gid = %g.gid, status = s.as_str(), "消息事务终结");
443                    self.store.set_global_status(&g.gid, s, "").await?;
444                    return Ok(());
445                }
446                Advance::Wait => return Ok(()),
447                // 只有 workflow 模式会出现，别的模式走到这里说明状态机接错了。
448                // **不 panic** —— 推进器是常驻的，崩了整个 TC 就停了
449                Advance::RunWorkflow => {
450                    warn!(gid = %g.gid, "非 workflow 事务收到 RunWorkflow 决策，跳过");
451                    return Ok(());
452                }
453                Advance::Call { index, op } => {
454                    let bid = branch_id(index);
455                    match self.call_branch(g, &bid, op, &steps[index].action).await {
456                        BranchResult::Success => {
457                            self.store
458                                .set_branch_status(&g.gid, &bid, op, BranchStatus::Succeed)
459                                .await?;
460                        }
461                        // msg 保证"最终一定送达"，没有补偿一说。失败只能重试。
462                        BranchResult::Failure | BranchResult::Ongoing | BranchResult::Unknown => {
463                            self.retry_later(g).await?;
464                            return Ok(());
465                        }
466                    }
467                }
468            }
469        }
470    }
471
472    async fn retry_later(&self, g: &GlobalRow) -> anyhow::Result<()> {
473        let iv = dtmrs_core::next_interval(g.next_cron_interval);
474        self.store.schedule_retry(&g.gid, iv).await?;
475        Ok(())
476    }
477
478    /// 取每一步的 action / compensate 分支当前状态，按步序对齐
479    async fn branch_states(
480        &self,
481        gid: &str,
482        n: usize,
483    ) -> anyhow::Result<(Vec<BranchStatus>, Vec<BranchStatus>)> {
484        let rows = self.store.list_branches(gid).await?;
485        let mut actions = vec![BranchStatus::Prepared; n];
486        let mut compensates = vec![BranchStatus::Prepared; n];
487        for r in rows {
488            let Some(i) = index_of(&r.branch_id) else {
489                continue;
490            };
491            if i >= n {
492                continue;
493            }
494            match r.op {
495                BranchOp::Action | BranchOp::Try => actions[i] = r.status,
496                BranchOp::Compensate | BranchOp::Cancel | BranchOp::Rollback => {
497                    compensates[i] = r.status
498                }
499                _ => {}
500            }
501        }
502        Ok((actions, compensates))
503    }
504
505    /// 调一个分支。`local://名字` 走进程内函数，`grpc://` 走 gRPC，其它走 HTTP。
506    async fn call_branch(
507        &self,
508        g: &GlobalRow,
509        branch_id: &str,
510        op: BranchOp,
511        url: &str,
512    ) -> BranchResult {
513        match parse_target(url) {
514            Target::Local(name) => self.call_local(g, branch_id, op, &name).await,
515            Target::Http(u) => self.call_http(g, branch_id, op, &u).await,
516            #[cfg(feature = "grpc")]
517            Target::Grpc(t) => {
518                self.grpc
519                    .call(
520                        &t,
521                        &g.gid,
522                        &g.trans_type.to_string(),
523                        branch_id,
524                        op.as_str(),
525                    )
526                    .await
527            }
528            // 编译时关掉了 grpc feature，却遇到 grpc:// 分支。
529            // 按「结果未知」处理（重试，不回滚）—— 这是构建配置问题，不是业务失败
530            #[cfg(not(feature = "grpc"))]
531            Target::Grpc(t) => {
532                warn!(gid = %g.gid, branch = %branch_id, endpoint = %t.endpoint,
533                      "遇到 grpc:// 分支但本次构建关掉了 grpc feature，按结果未知处理");
534                BranchResult::Unknown
535            }
536        }
537    }
538
539    /// 进程内调用：没有网络、没有序列化，一次函数调用。
540    async fn call_local(
541        &self,
542        g: &GlobalRow,
543        branch_id: &str,
544        op: BranchOp,
545        name: &str,
546    ) -> BranchResult {
547        let Some(h) = self.registry.get(name) else {
548            // 漏注册（比如新版本删了 handler）。**必须当 Unknown 而不是 Failure**：
549            // 判失败会触发回滚，而这其实是部署问题，改回来重试才对。
550            warn!(gid = %g.gid, branch = %branch_id, handler = name,
551                  "本地分支未注册，按结果未知处理（会重试，不回滚）");
552            return BranchResult::Unknown;
553        };
554        let ctx = BranchCtx {
555            gid: g.gid.clone(),
556            branch_id: branch_id.to_string(),
557            op,
558            trans_type: g.trans_type.to_string(),
559        };
560        let r = h(ctx).await;
561        info!(gid = %g.gid, branch = %branch_id, op = op.as_str(),
562              handler = name, result = ?r, "本地分支返回");
563        r
564    }
565
566    /// 远端调用。查询参数跟 DTM 保持一致，客户端屏障库可以直接复用。
567    async fn call_http(
568        &self,
569        g: &GlobalRow,
570        branch_id: &str,
571        op: BranchOp,
572        url: &str,
573    ) -> BranchResult {
574        let req = self
575            .http
576            .post(url)
577            .query(&[
578                ("gid", g.gid.as_str()),
579                ("trans_type", &g.trans_type.to_string()),
580                ("branch_id", branch_id),
581                ("op", op.as_str()),
582            ])
583            .header("content-type", "application/json")
584            .body(branch_payload(&g.payload));
585        match req.send().await {
586            Ok(resp) => {
587                let code = resp.status().as_u16();
588                let body = resp.text().await.unwrap_or_default();
589                let r = BranchResult::from_http(code, &body);
590                info!(gid = %g.gid, branch = %branch_id, op = op.as_str(), code, result = ?r, "分支返回");
591                r
592            }
593            Err(e) => {
594                // 超时/连不上 —— 结果未知，必须重试而不是回滚
595                warn!(gid = %g.gid, branch = %branch_id, error = %e, "分支不可达，结果未知");
596                BranchResult::Unknown
597            }
598        }
599    }
600}
601
602/// 分支号：第 0 步是 "01"，跟 DTM 一致
603pub fn branch_id(index: usize) -> String {
604    format!("{:02}", index + 1)
605}
606
607fn index_of(branch_id: &str) -> Option<usize> {
608    branch_id
609        .parse::<usize>()
610        .ok()
611        .and_then(|v| v.checked_sub(1))
612}
613
614/// 按 op 把分支行拆成两列（正向 / 反向），按步序对齐
615fn split_by_op(
616    rows: &[dtmrs_store::BranchRow],
617    n: usize,
618    fwd: BranchOp,
619    bwd: BranchOp,
620) -> (Vec<BranchStatus>, Vec<BranchStatus>) {
621    let mut a = vec![BranchStatus::Prepared; n];
622    let mut b = vec![BranchStatus::Prepared; n];
623    for r in rows {
624        let Some(i) = index_of(&r.branch_id) else {
625            continue;
626        };
627        if i >= n {
628            continue;
629        }
630        if r.op == fwd {
631            a[i] = r.status;
632        } else if r.op == bwd {
633            b[i] = r.status;
634        }
635    }
636    (a, b)
637}
638
639/// 按分支序取出各 compensate 行的状态，用于 workflow 的逆序补偿。
640///
641/// 跟 `split_by_op` 的区别：workflow 的分支数是**运行时长出来的**，
642/// 提交时并不知道有几个，所以长度得从已登记的行里推出来。
643fn compensate_states(rows: &[dtmrs_store::BranchRow]) -> Vec<BranchStatus> {
644    let n = rows
645        .iter()
646        .filter(|r| r.op == BranchOp::Compensate)
647        .filter_map(|r| index_of(&r.branch_id))
648        .max()
649        .map(|m| m + 1)
650        .unwrap_or(0);
651    // 没登记补偿的分支（比如纯查询步骤）留成 Succeed，逆序扫描时会跳过它 ——
652    // 本来就没有副作用要收拾
653    let mut v = vec![BranchStatus::Succeed; n];
654    for r in rows.iter().filter(|r| r.op == BranchOp::Compensate) {
655        if let Some(i) = index_of(&r.branch_id) {
656            if i < n {
657                v[i] = r.status;
658            }
659        }
660    }
661    v
662}
663
664fn url_of(rows: &[dtmrs_store::BranchRow], branch_id: &str, op: BranchOp) -> Option<String> {
665    rows.iter()
666        .find(|r| r.branch_id == branch_id && r.op == op)
667        .map(|r| r.url.clone())
668}
669
670/// MVP 阶段各分支共用同一份请求体。真实场景应该每步独立 payload，
671/// 那是第二版的事（见 DESIGN.md 范围表）。
672fn branch_payload(_global_payload: &str) -> String {
673    "{}".to_string()
674}
675
676#[cfg(test)]
677mod tests {
678    use super::*;
679
680    #[test]
681    fn 分支号与下标互转() {
682        assert_eq!(branch_id(0), "01");
683        assert_eq!(branch_id(9), "10");
684        assert_eq!(index_of("01"), Some(0));
685        assert_eq!(index_of("10"), Some(9));
686        assert_eq!(index_of("00"), None);
687        assert_eq!(index_of("xx"), None);
688    }
689}
dtmrs_server/driver.rs

dtmrs_server/
driver.rs
