dtmrs_server/grpc/
server.rs

1//! TC 对外的 gRPC API。
2//!
3//! 这一层**只做协议转换**，所有判断都在 [`crate::api`] 里 —— HTTP 和 gRPC
4//! 共用同一份逻辑，不会出现「同一个请求走 HTTP 被拒、走 gRPC 却受理了」。
5//!
6//! 错误映射见 [`crate::api::ApiError`] 的表。
7
8use tonic::{Request, Response, Status};
9
10use super::pb;
11use crate::api::{Api, ApiError, RegisterBranch};
12use dtmrs_core::SagaStep;
13
14impl From<ApiError> for Status {
15    fn from(e: ApiError) -> Self {
16        match &e {
17            ApiError::BadRequest(m) => Status::invalid_argument(m.clone()),
18            ApiError::NotFound(m) => Status::not_found(m.clone()),
19            ApiError::Conflict(m) => Status::failed_precondition(m.clone()),
20            ApiError::Internal(m) => Status::internal(m.clone()),
21        }
22    }
23}
24
25pub struct TcService {
26    api: Api,
27}
28
29impl TcService {
30    pub fn new(api: Api) -> Self {
31        Self { api }
32    }
33
34    /// 包成 tonic 的 server，调用方直接挂到 `Server::builder().add_service(..)`
35    pub fn into_server(self) -> pb::tc_server::TcServer<Self> {
36        pb::tc_server::TcServer::new(self)
37    }
38
39    /// 带认证的版本。**必须和 HTTP 侧用同一个 `Auth`**，否则就出现
40    /// 「同一个请求走 HTTP 被拒、走 gRPC 却受理了」—— 这正是
41    /// 「绝对不能破坏的语义」里防的那种漂移。
42    ///
43    /// gRPC 没有 cookie 和登录页的概念，所以这里**只认 Bearer token**
44    /// （metadata 的 `authorization` 键）。管理台的会话 cookie 只在 HTTP 侧有意义。
45    pub fn into_server_with_auth(
46        self,
47        auth: std::sync::Arc<crate::auth::Auth>,
48    ) -> tonic::service::interceptor::InterceptedService<
49        pb::tc_server::TcServer<Self>,
50        impl tonic::service::Interceptor + Clone,
51    > {
52        pb::tc_server::TcServer::with_interceptor(self, move |req: tonic::Request<()>| {
53            let ok = req
54                .metadata()
55                .get("authorization")
56                .and_then(|v| v.to_str().ok())
57                .and_then(crate::auth::Auth::bearer)
58                .is_some_and(|t| auth.token_ok(t));
59            if ok {
60                Ok(req)
61            } else {
62                Err(tonic::Status::unauthenticated("需要 Bearer token"))
63            }
64        })
65    }
66}
67
68#[tonic::async_trait]
69impl pb::tc_server::Tc for TcService {
70    async fn new_gid(
71        &self,
72        _req: Request<pb::NewGidRequest>,
73    ) -> Result<Response<pb::NewGidReply>, Status> {
74        Ok(Response::new(pb::NewGidReply {
75            gid: self.api.new_gid(),
76        }))
77    }
78
79    async fn prepare(
80        &self,
81        req: Request<pb::PrepareRequest>,
82    ) -> Result<Response<pb::Empty>, Status> {
83        let r = req.into_inner();
84        // proto3 的 int64 没法区分「没传」和「传了 0」，所以用 0 表示走默认值。
85        // 宽限期本来也不该是 0 —— 那等于 prepare 完立刻回查，白问一次
86        let grace = if r.grace_secs > 0 {
87            Some(r.grace_secs)
88        } else {
89            None
90        };
91        self.api
92            .prepare(&r.gid, &r.trans_type, &r.actions, &r.query_prepared, grace)
93            .await?;
94        Ok(Response::new(pb::Empty {}))
95    }
96
97    async fn register_branch(
98        &self,
99        req: Request<pb::RegisterBranchRequest>,
100    ) -> Result<Response<pb::Empty>, Status> {
101        let r = req.into_inner();
102        self.api
103            .register_branch(&RegisterBranch {
104                gid: r.gid,
105                branch_id: r.branch_id,
106                confirm: r.confirm,
107                cancel: r.cancel,
108                r#try: r.r#try,
109                commit: r.commit,
110                rollback: r.rollback,
111            })
112            .await?;
113        Ok(Response::new(pb::Empty {}))
114    }
115
116    async fn submit(&self, req: Request<pb::SubmitRequest>) -> Result<Response<pb::Empty>, Status> {
117        let r = req.into_inner();
118        let tt = if r.trans_type.is_empty() {
119            "saga"
120        } else {
121            &r.trans_type
122        };
123        let steps: Vec<SagaStep> = r
124            .steps
125            .into_iter()
126            .map(|s| SagaStep {
127                action: s.action,
128                compensate: s.compensate,
129                payload: s.payload,
130            })
131            .collect();
132        self.api.submit(&r.gid, tt, &steps).await?;
133        Ok(Response::new(pb::Empty {}))
134    }
135
136    async fn abort(&self, req: Request<pb::AbortRequest>) -> Result<Response<pb::Empty>, Status> {
137        self.api.abort(&req.into_inner().gid).await?;
138        Ok(Response::new(pb::Empty {}))
139    }
140
141    async fn retry(&self, req: Request<pb::RetryRequest>) -> Result<Response<pb::Empty>, Status> {
142        self.api.retry(&req.into_inner().gid).await?;
143        Ok(Response::new(pb::Empty {}))
144    }
145
146    async fn query(
147        &self,
148        req: Request<pb::QueryRequest>,
149    ) -> Result<Response<pb::TransView>, Status> {
150        let v = self.api.query(&req.into_inner().gid).await?;
151        Ok(Response::new(pb::TransView {
152            gid: v.gid,
153            trans_type: v.trans_type,
154            status: v.status,
155            rollback_reason: v.rollback_reason,
156            create_time: v.create_time,
157            finish_time: v.finish_time,
158            branches: v
159                .branches
160                .into_iter()
161                .map(|b| pb::BranchView {
162                    branch_id: b.branch_id,
163                    op: b.op,
164                    url: b.url,
165                    status: b.status,
166                })
167                .collect(),
168        }))
169    }
170}
dtmrs_server/grpc/server.rs

dtmrs_server/grpc/
server.rs
