nasaga_runtime_core/
transaction.rs

1//! Saga 运行时的显式本地事务裁决适配。
2
3use std::future::Future;
4
5use nasaga_backend::{SagaBackend, SagaTransactionRunError, SagaTransactionRunner};
6
7/// 业务作用：保留 Saga 本地事务的封闭失败阶段，使 transport 不解析错误文本决定 ACK。
8///
9/// `CommitRejected`/`CommitUncertain`/`RollbackFailed` 均禁止进入有界 DLT 预算：
10/// 前者需要原消息重试完成业务，后两者则无法声称数据库已回滚。
11#[derive(Debug, Clone, Copy, PartialEq, Eq)]
12pub enum SagaTransactionError {
13    /// 内层失败把事务标记为 rollback-only，最外层已确认回滚。
14    RollbackOnly,
15    /// 数据库明确拒绝 COMMIT，事务未提交但原输入不得 ACK。
16    CommitRejected,
17    /// COMMIT 请求后无法确定是否已持久。
18    CommitUncertain,
19    /// 物理回滚失败，不能声称本次无副作用。
20    RollbackFailed,
21    /// 事务开始前或执行内核的基础设施故障。
22    Infrastructure,
23}
24
25impl SagaTransactionError {
26    /// 业务作用：判断该阶段是否必须保留原输入且不消耗 DLT 预算。
27    ///
28    /// 参数说明: 无。
29    ///
30    /// 返回：COMMIT 拒绝/不确定或回滚失败返回真；已确认回滚的其他基础故障返回假。
31    pub fn requires_unbounded_redelivery(self) -> bool {
32        matches!(
33            self,
34            Self::CommitRejected | Self::CommitUncertain | Self::RollbackFailed
35        )
36    }
37}
38
39impl std::fmt::Display for SagaTransactionError {
40    /// 业务作用：输出不含 SQL、连接串或 payload 的稳定事务阶段。
41    ///
42    /// 参数说明：
43    /// - `formatter`: 标准格式化输出目标。
44    ///
45    /// 返回：文本写入成功返回 `Ok`；格式化失败返回对应错误。
46    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
47        formatter.write_str(match self {
48            Self::RollbackOnly => "Saga transaction rollback-only",
49            Self::CommitRejected => "Saga transaction commit rejected",
50            Self::CommitUncertain => "Saga transaction commit uncertain",
51            Self::RollbackFailed => "Saga transaction rollback failed",
52            Self::Infrastructure => "Saga transaction infrastructure failed",
53        })
54    }
55}
56
57impl std::error::Error for SagaTransactionError {}
58
59/// 业务作用：在指定 datasource 上执行 Saga 原子步骤，并保留 COMMIT 不确定与回滚失败分类。
60///
61/// 参数说明：
62/// - `backend`：包含同源 store、Inbox、Outbox 及事务执行器的完整后端。
63/// - `body`：在后端事务上下文内执行并返回领域值的 future。
64///
65/// 返回：原样返回 backend 事务执行器成功值，失败映射为封闭 Saga 事务分类；提交不确定不能视为回滚完成。本适配不自行比较 Inbox/Outbox 数据源，依赖已装配 backend 的同源合同。
66pub(crate) async fn run_for<B, T, F>(backend: &B, body: F) -> anyhow::Result<T>
67where
68    B: SagaBackend,
69    F: Future<Output = anyhow::Result<T>> + Send,
70    T: Send,
71{
72    backend
73        .transaction_runner()
74        .run(Box::pin(body))
75        .await
76        .map_err(map_error)
77}
78
79/// 业务作用：在事务体开始、每次恢复轮询及交还提交裁决前复验执行资格，等待期间失权则回滚全部事务事实。
80///
81/// 参数说明：
82/// - `backend`：固定事务域
83/// - `authorize`：同步验证本次操作的冻结资格
84/// - `body`：只能产生同事务数据库写入。
85///
86/// 返回：资格持续有效且提交确认后返回领域值；失权使事务体返回错误并走正常回滚。
87/// 已在有效资格下发出的 COMMIT 仍按数据库收据裁决，不因确认回包较晚而宣称回滚。
88pub(crate) async fn run_authorized_for<B, T, F>(
89    backend: &B,
90    authorize: &(dyn Fn() -> anyhow::Result<()> + Send + Sync),
91    body: F,
92) -> anyhow::Result<T>
93where
94    B: SagaBackend,
95    F: Future<Output = anyhow::Result<T>> + Send,
96    T: Send,
97{
98    run_for(backend, async {
99        let outcome = {
100            let mut body = Box::pin(body);
101            std::future::poll_fn(|context| {
102                // BEGIN、连接池或行锁都可能消耗租期；恢复数据库操作前必须重新取得同一资格。
103                if let Err(error) = authorize() {
104                    return std::task::Poll::Ready(Err(error));
105                }
106                body.as_mut().poll(context)
107            })
108            .await
109        }?;
110        // 先释放事务体的连接借用，再决定提交；失权不能留下 Inbox、journal、实例、Outbox 或 timer 的半份事实。
111        authorize()?;
112        Ok(outcome)
113    })
114    .await
115}
116
117/// 业务作用：把 natx 的封闭事务阶段映射成 Saga 对外错误，同时保留领域回滚原始错误。
118///
119/// 参数说明：
120/// - `error`: natx 返回的显式事务阶段错误。
121///
122/// 返回：领域回滚返回原错误；其余分支返回可 downcast 的封闭
123/// [`SagaTransactionError`]，供消费循环精确决定不 ACK。
124fn map_error(error: SagaTransactionRunError<anyhow::Error>) -> anyhow::Error {
125    match error {
126        SagaTransactionRunError::Rollback(error) => error,
127        SagaTransactionRunError::RollbackOnly => SagaTransactionError::RollbackOnly.into(),
128        SagaTransactionRunError::CommitRejected => SagaTransactionError::CommitRejected.into(),
129        SagaTransactionRunError::CommitUncertain => SagaTransactionError::CommitUncertain.into(),
130        SagaTransactionRunError::RollbackFailed => SagaTransactionError::RollbackFailed.into(),
131        SagaTransactionRunError::Infrastructure => SagaTransactionError::Infrastructure.into(),
132    }
133}
nasaga_runtime_core/transaction.rs

nasaga_runtime_core/
transaction.rs
