dtmrs_server/
api.rs

1//! TC 的对外操作，**与协议无关**。
2//!
3//! HTTP 和 gRPC 两套接口都只做「协议 ↔ 这一层」的转换，业务判断全在这里。
4//! 分成两处写迟早会漂移 —— 而这一层漂移的后果是「同一个请求走 HTTP 被拒、
5//! 走 gRPC 却受理了」，这种不一致在事务系统里是要命的。
6//!
7//! 错误用 [`ApiError`] 表达，由各协议层翻译成自己的表示：
8//!
9//! | ApiError | HTTP | gRPC |
10//! |---|---|---|
11//! | `BadRequest` | 400 | `INVALID_ARGUMENT` |
12//! | `NotFound` | 404 | `NOT_FOUND` |
13//! | `Conflict` | 200 + `dtm_result=FAILURE` | `FAILED_PRECONDITION` |
14//! | `Internal` | 500 | `INTERNAL` |
15//!
16//! `Conflict` 在 HTTP 上返回 200 是**刻意保留的历史行为**（已终结的事务再调
17//! abort），换成 4xx 会打破现有客户端。
18
19use crate::{msg_rows, saga_rows, tcc_rows};
20use dtmrs_core::{BranchOp, GlobalStatus, SagaStep, TransType};
21use dtmrs_store::Store;
22use serde::Serialize;
23
24#[derive(Debug, Clone, PartialEq, Eq)]
25pub enum ApiError {
26    BadRequest(String),
27    NotFound(String),
28    /// 请求本身合法，但当前状态下做不了
29    Conflict(String),
30    Internal(String),
31}
32
33impl ApiError {
34    pub fn message(&self) -> &str {
35        match self {
36            Self::BadRequest(m) | Self::NotFound(m) | Self::Conflict(m) | Self::Internal(m) => m,
37        }
38    }
39}
40
41impl std::fmt::Display for ApiError {
42    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
43        f.write_str(self.message())
44    }
45}
46
47pub type Result<T> = std::result::Result<T, ApiError>;
48
49fn internal(e: impl std::fmt::Display) -> ApiError {
50    ApiError::Internal(e.to_string())
51}
52
53#[derive(Debug, Clone, Serialize)]
54pub struct BranchView {
55    pub branch_id: String,
56    pub op: String,
57    pub url: String,
58    pub status: String,
59}
60
61#[derive(Debug, Clone, Serialize)]
62pub struct TransView {
63    pub gid: String,
64    pub trans_type: String,
65    pub status: String,
66    pub rollback_reason: String,
67    pub create_time: i64,
68    pub finish_time: Option<i64>,
69    pub branches: Vec<BranchView>,
70}
71
72/// 分支登记请求。TCC 用 confirm/cancel，XA 用 commit/rollback。
73#[derive(Debug, Clone, Default)]
74pub struct RegisterBranch {
75    pub gid: String,
76    pub branch_id: String,
77    pub confirm: String,
78    pub cancel: String,
79    pub r#try: String,
80    pub commit: String,
81    pub rollback: String,
82}
83
84#[derive(Clone)]
85pub struct Api {
86    pub store: Store,
87}
88
89impl Api {
90    pub fn new(store: Store) -> Self {
91        Self { store }
92    }
93
94    /// 时间戳 + 进程内计数。生产建议客户端直接用业务单号当 gid ——
95    /// 那样天然幂等，重试不会变成两笔
96    pub fn new_gid(&self) -> String {
97        use std::sync::atomic::{AtomicU64, Ordering};
98        static SEQ: AtomicU64 = AtomicU64::new(0);
99        let n = SEQ.fetch_add(1, Ordering::Relaxed);
100        format!("{}-{}", dtmrs_store::now(), n)
101    }
102
103    /// 提交。
104    ///
105    /// **重复提交同一个 gid 必须成功而不是报错** —— 客户端网络抖动重试时
106    /// 返回错误会让它以为没受理，然后换个 gid 再来一次，就成了两笔。
107    pub async fn submit(&self, gid: &str, trans_type: &str, steps: &[SagaStep]) -> Result<()> {
108        if gid.is_empty() {
109            return Err(ApiError::BadRequest("gid 不能为空".into()));
110        }
111        let Some(tt) = TransType::parse(trans_type) else {
112            return Err(ApiError::BadRequest("未知 trans_type".into()));
113        };
114
115        // tcc / msg / xa：prepare 已经建过事务，submit 只是把它推成 submitted
116        if let Ok(Some(g)) = self.store.get_global(gid).await {
117            if g.status == GlobalStatus::Prepared {
118                self.store
119                    .set_global_status(gid, GlobalStatus::Submitted, "")
120                    .await
121                    .map_err(internal)?;
122                let _ = self.store.schedule_now(gid).await;
123            }
124            // 已经提交过 —— 幂等返回成功
125            return Ok(());
126        }
127
128        match tt {
129            TransType::Saga => {
130                if steps.is_empty() {
131                    return Err(ApiError::BadRequest("saga 的 steps 不能为空".into()));
132                }
133                let (g, branches) = saga_rows(gid, steps);
134                self.store
135                    .create_global(&g, &branches)
136                    .await
137                    .map_err(internal)?;
138                Ok(())
139            }
140            TransType::Tcc | TransType::Msg | TransType::Xa => {
141                Err(ApiError::BadRequest("tcc/xa/msg 要先调 prepare".into()))
142            }
143            // workflow 的「步骤」是**代码**，没法表示成 URL 存进库里，
144            // 所以只能在嵌入式形态下提交（Embedded::workflow + submit_workflow）。
145            // 这不是暂未实现，是这个模式的本质决定的
146            TransType::Workflow => Err(ApiError::BadRequest(
147                "workflow 模式只能在嵌入式形态下提交（步骤是进程内的函数，不是 URL）".into(),
148            )),
149        }
150    }
151
152    /// 第一阶段。msg 建 prepared 事务 + 正向分支；tcc / xa 只建空事务。
153    pub async fn prepare(
154        &self,
155        gid: &str,
156        trans_type: &str,
157        actions: &[String],
158        query_prepared: &str,
159        grace_secs: Option<i64>,
160    ) -> Result<()> {
161        if gid.is_empty() {
162            return Err(ApiError::BadRequest("gid 不能为空".into()));
163        }
164        match TransType::parse(trans_type) {
165            Some(TransType::Msg) => {
166                if actions.is_empty() {
167                    return Err(ApiError::BadRequest("msg 的 actions 不能为空".into()));
168                }
169                if query_prepared.is_empty() {
170                    // 没有回查地址，客户端崩在 prepare 和 submit 之间就没人能
171                    // 决断这单了。猜「已提交」会重复扣款，猜「没提交」会丢单
172                    return Err(ApiError::BadRequest(
173                        "msg 必须提供 query_prepared，否则崩溃后无法决断".into(),
174                    ));
175                }
176                let (g, br) = msg_rows(gid, actions, query_prepared, grace_secs.unwrap_or(10));
177                self.store.create_global(&g, &br).await.map_err(internal)?;
178                Ok(())
179            }
180            Some(tt @ (TransType::Tcc | TransType::Xa)) => {
181                let mut g = tcc_rows(gid);
182                g.trans_type = tt;
183                self.store.create_global(&g, &[]).await.map_err(internal)?;
184                Ok(())
185            }
186            _ => Err(ApiError::BadRequest(
187                "prepare 支持 tcc / xa / msg；saga 直接 submit".into(),
188            )),
189        }
190    }
191
192    /// 分支登记。**必须先登记再做一阶段**：反过来的话一阶段成功但登记失败，
193    /// TC 就不知道有这个分支，回滚时不会处理它 —— TCC 是预留资源永久泄漏，
194    /// XA 更糟，会留下一个永久持锁的 prepared 事务。
195    pub async fn register_branch(&self, r: &RegisterBranch) -> Result<()> {
196        if r.gid.is_empty() || r.branch_id.is_empty() {
197            return Err(ApiError::BadRequest("gid / branch_id 不能为空".into()));
198        }
199        let tt = match self.store.get_global(&r.gid).await {
200            Ok(Some(g)) => g.trans_type,
201            Ok(None) => return Err(ApiError::NotFound("gid 不存在，先 prepare".into())),
202            Err(e) => return Err(internal(e)),
203        };
204
205        let mut ops = Vec::new();
206        match tt {
207            TransType::Tcc => {
208                if r.confirm.is_empty() || r.cancel.is_empty() {
209                    return Err(ApiError::BadRequest(
210                        "tcc 分支必须提供 confirm 和 cancel".into(),
211                    ));
212                }
213                ops.push((BranchOp::Confirm, r.confirm.clone()));
214                ops.push((BranchOp::Cancel, r.cancel.clone()));
215                if !r.r#try.is_empty() {
216                    ops.push((BranchOp::Try, r.r#try.clone()));
217                }
218            }
219            TransType::Xa => {
220                if r.commit.is_empty() || r.rollback.is_empty() {
221                    // 缺任一个都可能留下永久持锁的 prepared 事务
222                    return Err(ApiError::BadRequest(
223                        "xa 分支必须提供 commit 和 rollback".into(),
224                    ));
225                }
226                ops.push((BranchOp::Commit, r.commit.clone()));
227                ops.push((BranchOp::Rollback, r.rollback.clone()));
228            }
229            _ => return Err(ApiError::BadRequest("只有 tcc 和 xa 需要登记分支".into())),
230        }
231
232        self.store
233            .register_branch(&r.gid, &r.branch_id, &ops)
234            .await
235            .map_err(internal)
236    }
237
238    /// 主动中止，触发逆序补偿
239    pub async fn abort(&self, gid: &str) -> Result<()> {
240        match self.store.get_global(gid).await {
241            Ok(Some(g)) if !g.status.is_final() => {
242                self.store
243                    .set_global_status(gid, GlobalStatus::Aborting, "调用方主动中止")
244                    .await
245                    .map_err(internal)?;
246                let _ = self.store.schedule_now(gid).await;
247                Ok(())
248            }
249            Ok(Some(_)) => Err(ApiError::Conflict("事务已终结，无法中止".into())),
250            Ok(None) => Err(ApiError::NotFound("gid 不存在".into())),
251            Err(e) => Err(internal(e)),
252        }
253    }
254
255    pub async fn query(&self, gid: &str) -> Result<TransView> {
256        let g = self
257            .store
258            .get_global(gid)
259            .await
260            .map_err(internal)?
261            .ok_or_else(|| ApiError::NotFound("gid 不存在".into()))?;
262        let branches = self.store.list_branches(gid).await.map_err(internal)?;
263        Ok(TransView {
264            gid: g.gid,
265            trans_type: g.trans_type.to_string(),
266            status: g.status.as_str().into(),
267            rollback_reason: g.rollback_reason,
268            create_time: g.create_time,
269            finish_time: g.finish_time,
270            branches: branches
271                .into_iter()
272                .map(|b| BranchView {
273                    branch_id: b.branch_id,
274                    op: b.op.as_str().into(),
275                    url: b.url,
276                    status: b.status.as_str().into(),
277                })
278                .collect(),
279        })
280    }
281
282    pub async fn list_recent(&self, limit: i64) -> Vec<TransView> {
283        self.store
284            .list_recent(limit)
285            .await
286            .unwrap_or_default()
287            .into_iter()
288            .map(|g| TransView {
289                gid: g.gid,
290                trans_type: g.trans_type.to_string(),
291                status: g.status.as_str().into(),
292                rollback_reason: g.rollback_reason,
293                create_time: g.create_time,
294                finish_time: g.finish_time,
295                branches: Vec::new(),
296            })
297            .collect()
298    }
299}
dtmrs_server/api.rs

dtmrs_server/
api.rs
