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/// - `datasource`: Saga store、Inbox 与 Outbox 共同绑定的数据源身份。
63/// - `body`: 在同源 ambient transaction 内执行并返回领域结果的 Future。
64///
65/// 返回：数据库确认提交后返回领域值；跨源、回滚、提交不确定或基础设施失败返回封闭分类。
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/// 参数说明：`backend` 固定事务域，`authorize` 同步验证本次操作的冻结资格，`body` 只能产生同事务数据库写入。
82///
83/// 返回：资格持续有效且提交确认后返回领域值；失权使事务体返回错误并走正常回滚。
84/// 已在有效资格下发出的 COMMIT 仍按数据库收据裁决，不因确认回包较晚而宣称回滚。
85pub(crate) async fn run_authorized_for<B, T, F>(
86    backend: &B,
87    authorize: &(dyn Fn() -> anyhow::Result<()> + Send + Sync),
88    body: F,
89) -> anyhow::Result<T>
90where
91    B: SagaBackend,
92    F: Future<Output = anyhow::Result<T>> + Send,
93    T: Send,
94{
95    run_for(backend, async {
96        let outcome = {
97            let mut body = Box::pin(body);
98            std::future::poll_fn(|context| {
99                // BEGIN、连接池或行锁都可能消耗租期；恢复数据库操作前必须重新取得同一资格。
100                if let Err(error) = authorize() {
101                    return std::task::Poll::Ready(Err(error));
102                }
103                body.as_mut().poll(context)
104            })
105            .await
106        }?;
107        // 先释放事务体的连接借用，再决定提交；失权不能留下实例、首步 Outbox 或 timer 的半份事实。
108        authorize()?;
109        Ok(outcome)
110    })
111    .await
112}
113
114/// 业务作用：把 natx 的封闭事务阶段映射成 Saga 对外错误，同时保留领域回滚原始错误。
115///
116/// 参数说明：
117/// - `error`: natx 返回的显式事务阶段错误。
118///
119/// 返回：领域回滚返回原错误；其余分支返回可 downcast 的封闭
120/// [`SagaTransactionError`]，供消费循环精确决定不 ACK。
121fn map_error(error: SagaTransactionRunError<anyhow::Error>) -> anyhow::Error {
122    match error {
123        SagaTransactionRunError::Rollback(error) => error,
124        SagaTransactionRunError::RollbackOnly => SagaTransactionError::RollbackOnly.into(),
125        SagaTransactionRunError::CommitRejected => SagaTransactionError::CommitRejected.into(),
126        SagaTransactionRunError::CommitUncertain => SagaTransactionError::CommitUncertain.into(),
127        SagaTransactionRunError::RollbackFailed => SagaTransactionError::RollbackFailed.into(),
128        SagaTransactionRunError::Infrastructure => SagaTransactionError::Infrastructure.into(),
129    }
130}
nasaga_runtime_core/transaction.rs

nasaga_runtime_core/
transaction.rs
