nasaga_runtime_core/
participant.rs

1//! 参与方 adapter 支撑：`#[saga]` 宏生成代码调用的完整本地事务 wrapper。
2//!
3//! 固定事务序：身份复验（envelope 伪造不进业务路径）→ 开启本地事务 →
4//! Inbox claim(`command_id`) → 按 `effect_id` 竞争 participant step gate → 业务 handler
5//! → 本地业务结果 + gate 落账 → 结果 Outbox → COMMIT → transport 在 COMMIT 后 ACK。
6//! `Retryable` 与任何基础设施失败返回 `Err` 使事务回滚且**不得 ACK**。
7//!
8//! 当前覆盖 `local_fenceable` 的 execute/cancel/compensate、`resolve_only` Poll，以及
9//! `externally_cancellable` 的类型化 cancel + Poll resolve。外部取消必须以稳定 `effect_id`
10//! 调用远端幂等接口；本地提交失败后的重投不能形成第二份外部效果。
11
12use nainbox_core::{InboxClaim, InboxStore};
13use naoutbox_core::DurableOutboxAppend;
14use natelemetry::TraceContext;
15use std::collections::{BTreeMap, BTreeSet};
16
17use nasaga_backend::{
18    CancelAdjudication, CompensationAdmission, ExecuteAdmission, ExternalCancelAdmission,
19    ParticipantGateKey, ResolutionAdmission, SagaBackend, SagaBackendFactory, SagaParticipantStore,
20};
21use nasaga_core::{
22    CancelOutcome, CompensationOutcome, DefinitionVersion, EffectId, SagaCancelStep, SagaContext,
23    SagaExecutionError, SagaOutcome, SagaResolveStep, SagaStep, ServiceIdentity, StepAttemptStatus,
24    StepCancelStatus, StepCompensationStatus, StepForwardStatus, StepPhase, StepResolutionStatus,
25    TerminalForwardOutcome, WorkflowName,
26};
27use serde::de::DeserializeOwned;
28
29use crate::envelope::{
30    derive_result_event_id, valid_reason_code, SagaCommandEnvelope, SagaResultEnvelope,
31};
32
33/// 业务作用：抽象宏生成的类型化步骤分发入口，供 hosted command consumer 绑定具体 Service。
34///
35/// 实现只能由 `#[saga]` 按已发布 descriptor 生成；它必须先执行精确 workflow/version/step
36/// 门禁，再进入 [`ParticipantRuntime`] 的 Inbox、gate、业务和结果 Outbox 事务序。
37pub trait SagaCommandService<B: SagaBackend>: Send + Sync + 'static {
38    /// 业务作用：把一条已由 transport 认证并绑定目标 route 的命令交给完整参与方事务 wrapper。
39    ///
40    /// 参数说明：
41    /// - `runtime`: 本服务的参与方运行时。
42    /// - `envelope`: 已认证且 route 精确匹配的 command envelope。
43    /// - `producer`: transport 从可信来源映射出的 Orchestrator 逻辑身份。
44    ///
45    /// 返回：本地事务 COMMIT 后返回可 ACK 结论；合同、业务瞬态或基础设施错误返回错误。
46    fn handle_saga_command<'a>(
47        &'a self,
48        runtime: &'a ParticipantRuntime<B>,
49        envelope: &'a SagaCommandEnvelope,
50        producer: &'a ServiceIdentity,
51    ) -> impl std::future::Future<Output = anyhow::Result<ParticipantHandled>> + Send + 'a;
52
53    /// 业务作用：携带命令收据链路上下文的类型化分发入口；参与方结果事件由收据派生
54    /// 子上下文，保持跨服务 trace 连续。
55    ///
56    /// 默认实现忽略收据上下文并委托给
57    /// [`handle_saga_command`](Self::handle_saga_command)，旧版本宏展开在滚动升级期间
58    /// 继续满足本 trait；当前宏展开会覆写本方法并把 trace 传入事务 wrapper。
59    ///
60    /// 参数说明：
61    /// - `runtime`: 本服务的参与方运行时。
62    /// - `envelope`: 已认证且 route 精确匹配的 command envelope。
63    /// - `producer`: transport 从可信来源映射出的 Orchestrator 逻辑身份。
64    /// - `receipt_trace`: 收据中已校验的链路上下文；`None` 表示收据未携带。
65    ///
66    /// 返回：语义与 [`handle_saga_command`](Self::handle_saga_command) 一致。
67    fn handle_saga_command_traced<'a>(
68        &'a self,
69        runtime: &'a ParticipantRuntime<B>,
70        envelope: &'a SagaCommandEnvelope,
71        producer: &'a ServiceIdentity,
72        receipt_trace: Option<&'a TraceContext>,
73    ) -> impl std::future::Future<Output = anyhow::Result<ParticipantHandled>> + Send + 'a {
74        let _ = receipt_trace;
75        self.handle_saga_command(runtime, envelope, producer)
76    }
77}
78
79/// 业务作用：参与方命令处理的可 ACK 结论。
80///
81/// 全部分支都表示"事务已 COMMIT，可以 ACK"；`Retryable` 与基础设施失败不会出现在
82/// 这里——它们以 `Err` 返回并回滚。
83#[derive(Debug, Clone, Copy, PartialEq, Eq)]
84pub enum ParticipantHandled {
85    /// 业务 handler 真实执行并提交了结论。
86    Executed(StepAttemptStatus),
87    /// Inbox 命中重复命令；零副作用。
88    Duplicate,
89    /// 取消屏障已建立，execute 被本地状态拒绝；零正向效果，无结果事件。
90    Suppressed,
91    /// 该效果已有终态（新 attempt 重投）；已按既有终态重发结果事件，零业务效果。
92    Replayed(StepAttemptStatus),
93    /// 补偿合同违规（本地无正向成功效果）；已提交 `Halted` 违规事实并需告警。
94    ContractViolation,
95}
96
97/// 业务作用：冻结一个 workflow definition 可向本 Participant 发命令的逻辑 Orchestrator。
98///
99/// 信任投影同时锚定 workflow、version 与 digest；共享 Participant 的多个业务域不能只凭
100/// “某个已登记 Orchestrator”身份跨流程发起 execute、cancel、compensate 或 resolve。
101#[derive(Debug, Clone, PartialEq, Eq)]
102pub struct ParticipantCommandTrust {
103    producer: ServiceIdentity,
104    workflow: WorkflowName,
105    definition_version: DefinitionVersion,
106    definition_digest: Option<String>,
107}
108
109impl ParticipantCommandTrust {
110    /// 业务作用：从受信 deployment/definition projection 构造一条精确 command 授权边。
111    ///
112    /// 参数说明：
113    /// - `producer`: 该 definition 的逻辑 Orchestrator 身份。
114    /// - `workflow`: 投影绑定的 workflow 名称。
115    /// - `definition_version`: 投影绑定的不可变 definition 版本。
116    /// - `definition_digest`: 投影绑定的 64 位小写十六进制摘要。
117    ///
118    /// 返回：摘要编码合法时返回冻结投影；非法摘要在 consumer Ready 前拒绝启动。
119    pub fn new(
120        producer: ServiceIdentity,
121        workflow: WorkflowName,
122        definition_version: DefinitionVersion,
123        definition_digest: impl Into<String>,
124    ) -> anyhow::Result<Self> {
125        let definition_digest = definition_digest.into();
126        if !valid_definition_digest(&definition_digest) {
127            anyhow::bail!("participant command trust requires a canonical definition digest");
128        }
129        Ok(Self {
130            producer,
131            workflow,
132            definition_version,
133            definition_digest: Some(definition_digest),
134        })
135    }
136
137    /// 业务作用：为动态 Catalog 参与方构造按 workflow/version 限定的 Orchestrator 授权边。 participant 只发布自身 capability，不持有完整流程 seal；命令仍需通过 transport 身份认证、 envelope 派生复验与宏生成的精确步骤门禁，完整 digest 由 Orchestrator 的持久实例权威负责。
138    ///
139    /// 参数说明：
140    /// - `producer`：允许协调该能力的服务身份。
141    /// - `workflow`：本地能力支持的工作流名。
142    /// - `definition_version`：本地能力支持的定义版本。
143    ///
144    /// 返回：返回限制 producer、workflow 和定义版本且不绑定摘要的信任边，不执行 transport 认证或读取 Catalog；后续门禁仍需验证步骤与持久定义权威。
145    pub fn for_definition_version(
146        producer: ServiceIdentity,
147        workflow: WorkflowName,
148        definition_version: DefinitionVersion,
149    ) -> Self {
150        Self {
151            producer,
152            workflow,
153            definition_version,
154            definition_digest: None,
155        }
156    }
157}
158
159/// 业务作用：参与方运行时——持有 store/inbox/outbox 句柄与本服务的 Inbox consumer 名。
160///
161/// 无进程内业务状态；并发安全由 gate 行锁与 Inbox 唯一键承担。
162pub struct ParticipantRuntime<B: SagaBackend> {
163    backend: B,
164    consumer: String,
165    trusted_command_producers: BTreeSet<ServiceIdentity>,
166    command_trust: BTreeMap<(WorkflowName, DefinitionVersion, Option<String>), ServiceIdentity>,
167}
168
169/// 业务作用：表示已命中至少一条 Participant 信任投影的 command producer 能力。
170///
171/// 非 Kafka transport 应在校验 mTLS、签名或独占通道后创建本视图；其后所有 phase 调用都复用
172/// 同一受信身份，避免某个分支遗漏 producer 参数又退回未认证入口。
173pub struct AuthenticatedParticipantRuntime<'a, B: SagaBackend> {
174    runtime: &'a ParticipantRuntime<B>,
175    producer: &'a ServiceIdentity,
176}
177
178impl<B> ParticipantRuntime<B>
179where
180    B: SagaBackend + SagaBackendFactory,
181{
182    /// 业务作用：构造参与方运行时。
183    ///
184    /// 参数说明：
185    /// - `consumer`: 本服务的小写 canonical Inbox consumer 名称（各服务唯一，去重命名空间）。
186    /// - `command_trust`: 从受信 definition/deployment 投影得到的精确 Orchestrator 授权边。
187    ///
188    /// 返回：consumer 满足 128 字节小写 canonical Inbox key 时返回不建连的运行时；
189    /// 非法去重命名空间或空信任投影返回错误，避免未认证 command 获得业务入口。
190    pub fn new(
191        consumer: impl Into<String>,
192        command_trust: impl IntoIterator<Item = ParticipantCommandTrust>,
193    ) -> anyhow::Result<Self> {
194        Self::build(consumer, command_trust, B::default_backend()?)
195    }
196
197    /// 业务作用：构造绑定命名 datasource 的参与方运行时。
198    ///
199    /// 参数说明：
200    /// - `consumer`: 本服务的 canonical Inbox consumer 名称。
201    /// - `command_trust`: 从受信 definition/deployment 投影得到的 Orchestrator 授权边。
202    /// - `datasource`: Inbox、gate、业务事实与 result Outbox 共享的数据源。
203    ///
204    /// 返回：身份、信任投影与 datasource 合法时返回同源 runtime；任一边界不成立时在 I/O 前拒绝。
205    pub fn with_datasource(
206        consumer: impl Into<String>,
207        command_trust: impl IntoIterator<Item = ParticipantCommandTrust>,
208        datasource: impl AsRef<str>,
209    ) -> anyhow::Result<Self> {
210        Self::build(
211            consumer,
212            command_trust,
213            B::backend_for(datasource.as_ref())?,
214        )
215    }
216
217    /// 业务作用：把已验证的参与方配置与完整后端组合为唯一运行句柄。
218    ///
219    /// 参数说明：
220    /// - `consumer`：是 Inbox 去重命名空间
221    /// - `command_trust`：是授权投影
222    /// - `backend`：绑定同源持久角色。
223    ///
224    /// 返回：边界均合法时返回运行时；任一身份或投影冲突时在 I/O 前拒绝。
225    pub fn build(
226        consumer: impl Into<String>,
227        command_trust: impl IntoIterator<Item = ParticipantCommandTrust>,
228        backend: B,
229    ) -> anyhow::Result<Self> {
230        let consumer = consumer.into();
231        if consumer.is_empty()
232            || consumer.trim() != consumer
233            || consumer.len() > 128
234            || consumer.contains('\0')
235            || !consumer.bytes().all(|byte| {
236                byte.is_ascii_lowercase()
237                    || byte.is_ascii_digit()
238                    || matches!(byte, b'_' | b'-' | b'.')
239            })
240        {
241            anyhow::bail!("participant consumer must satisfy the Inbox key contract");
242        }
243        let mut trusted_command_producers = BTreeSet::new();
244        let mut projected_trust = BTreeMap::new();
245        for trust in command_trust {
246            trusted_command_producers.insert(trust.producer.clone());
247            let key = (
248                trust.workflow,
249                trust.definition_version,
250                trust.definition_digest,
251            );
252            if projected_trust.insert(key, trust.producer).is_some() {
253                anyhow::bail!(
254                    "participant command trust contains a duplicate definition projection"
255                );
256            }
257        }
258        if projected_trust.is_empty() {
259            anyhow::bail!("participant requires at least one command trust projection");
260        }
261        Ok(Self {
262            backend,
263            consumer,
264            trusted_command_producers,
265            command_trust: projected_trust,
266        })
267    }
268
269    /// 业务作用：读取参与方原子处理链绑定的 datasource 身份。
270    ///
271    /// 参数说明: 无。
272    ///
273    /// 返回：Inbox、Saga gate 与 Outbox 共用的不可变 qualifier。
274    pub fn datasource_ref(&self) -> &natx_core::DatasourceRef {
275        self.backend.datasource_ref()
276    }
277
278    /// 业务作用：把 transport 已认证 producer 收窄成只能调用 authenticated phase API 的能力视图。
279    ///
280    /// 参数说明：
281    /// - `producer`: transport 从 mTLS/SASL principal、独占 topic 或签名 key 映射出的身份。
282    ///
283    /// 返回：身份至少拥有一条启动投影时返回能力视图；否则在解析 envelope 与接触数据库前拒绝；
284    /// 每个 phase 仍会按该 envelope 的 workflow/version/digest 再做精确授权。
285    pub fn authenticate<'a>(
286        &'a self,
287        producer: &'a ServiceIdentity,
288    ) -> anyhow::Result<AuthenticatedParticipantRuntime<'a, B>> {
289        if !self.trusted_command_producers.contains(producer) {
290            return Err(crate::SagaCommandProcessingError::ProducerUnauthorized.into());
291        }
292        Ok(AuthenticatedParticipantRuntime {
293            runtime: self,
294            producer,
295        })
296    }
297
298    /// 业务作用：读取单一受信 command producer，供 Application 自动装配单参与方 transport 身份绑定。
299    ///
300    /// 该入口只在全部 workflow projection 都归属于同一逻辑 Orchestrator 时成立；多 producer 参与方
301    /// 必须由业务显式选择当前 listener principal 对应的 producer，不能由容器猜测授权边。
302    ///
303    /// 参数说明: 无。
304    ///
305    /// 返回：恰有一个受信 producer 时返回其稳定逻辑身份；存在多个 producer 时返回配置错误。
306    pub fn single_trusted_command_producer(&self) -> anyhow::Result<ServiceIdentity> {
307        if self.trusted_command_producers.len() != 1 {
308            anyhow::bail!(
309                "participant transport requires an explicit producer when trust contains multiple producers"
310            );
311        }
312        self.trusted_command_producers
313            .iter()
314            .next()
315            .cloned()
316            .ok_or_else(|| anyhow::anyhow!("participant command trust must not be empty"))
317    }
318
319    /// 业务作用：为自定义 connector 预检 command envelope 与 producer 的精确授权投影。
320    ///
321    /// 该预检不替代 phase handler 内的强制复验；它用于在解析业务 payload、获取连接或触碰 Inbox
322    /// 之前给 HTTP/gRPC adapter 一个统一的 fail-closed 入口。
323    ///
324    /// 参数说明：
325    /// - `envelope`: transport 已校验完整性、但尚未获业务授权的原始 command envelope。
326    /// - `producer`: transport 从签名、mTLS 或独占 route 映射出的逻辑 Orchestrator。
327    ///
328    /// 返回：身份派生与 workflow/version/digest/producer 投影全部匹配时成功；否则类型化拒绝。
329    pub fn verify_authenticated_command(
330        &self,
331        envelope: &SagaCommandEnvelope,
332        producer: &ServiceIdentity,
333    ) -> anyhow::Result<()> {
334        self.verified_command_identity(envelope, producer).map(drop)
335    }
336
337    /// 业务作用：execute 命令的完整本地事务 wrapper——gate 准入、业务执行、结果发布。
338    ///
339    /// 参数说明：
340    /// - `service`: 业务步骤实现。
341    /// - `envelope`: 已通过 transport 层 producer 认证的命令 envelope。
342    /// - `producer`: transport 认证并映射出的 Orchestrator 逻辑身份。
343    /// - `allow_unknown`: descriptor 声明的 `allow_unknown`；纯本地步骤返回 `Unknown`
344    ///   会按合同违规提交 `Halted`（不能进入无界等待）。
345    ///
346    /// 返回：可 ACK 的处理结论（COMMIT 已确认）；`Retryable`、身份复验失败或基础设施
347    /// 失败返回错误（已回滚，不得 ACK）。
348    pub async fn handle_authenticated_execute<S>(
349        &self,
350        service: &S,
351        envelope: &SagaCommandEnvelope,
352        producer: &ServiceIdentity,
353        allow_unknown: bool,
354    ) -> anyhow::Result<ParticipantHandled>
355    where
356        S: SagaStep,
357        S::Command: DeserializeOwned,
358    {
359        self.handle_authenticated_execute_traced(service, envelope, producer, None, allow_unknown)
360            .await
361    }
362
363    /// 业务作用：携带命令收据链路上下文的 execute 事务 wrapper——结果事件由收据派生
364    /// 子上下文，保持参与方到编排端的 trace 连续。
365    ///
366    /// 参数说明：
367    /// - `service`: 业务步骤实现。
368    /// - `envelope`: 已通过 transport 层 producer 认证的命令 envelope。
369    /// - `producer`: transport 认证并映射出的 Orchestrator 逻辑身份。
370    /// - `receipt_trace`: transport 收据显式解析出的链路上下文；`None` 表示未携带。
371    /// - `allow_unknown`: descriptor 声明的 `allow_unknown`。
372    ///
373    /// 返回：语义与 [`handle_authenticated_execute`](Self::handle_authenticated_execute)
374    /// 完全一致；trace 不改变事务序与可 ACK 边界。
375    pub async fn handle_authenticated_execute_traced<S>(
376        &self,
377        service: &S,
378        envelope: &SagaCommandEnvelope,
379        producer: &ServiceIdentity,
380        receipt_trace: Option<&TraceContext>,
381        allow_unknown: bool,
382    ) -> anyhow::Result<ParticipantHandled>
383    where
384        S: SagaStep,
385        S::Command: DeserializeOwned,
386    {
387        let identity = self.verified_command_identity(envelope, producer)?;
388        if identity.phase != StepPhase::Execute {
389            return Err(crate::SagaCommandProcessingError::ContractInvalid.into());
390        }
391        let _latency =
392            crate::latency::LatencyGuard::new(crate::latency::LatencyStage::ParticipantTransaction);
393        crate::transaction::run_for(&self.backend, async {
394            if matches!(
395                self.backend
396                    .inbox()
397                    .claim(&self.consumer, &envelope.command_id)
398                    .await?,
399                InboxClaim::Duplicate
400            ) {
401                return Ok(ParticipantHandled::Duplicate);
402            }
403            let gate = ParticipantGateKey {
404                saga_id: &identity.saga_id,
405                step: &identity.step,
406                tenant: &identity.tenant,
407                workflow: &identity.workflow,
408                definition_version: identity.definition_version,
409                definition_digest: &envelope.definition_digest,
410            };
411            match self
412                .backend
413                .store()
414                .admit_execute(&gate, &identity.effect)
415                .await?
416            {
417                // 取消屏障已建立:只提交 Inbox claim,零正向效果;不伪造 StepRejected,
418                // 取消事实由先前同事务写出的 CancelConfirmed Outbox 证明。
419                ExecuteAdmission::Suppressed => Ok(ParticipantHandled::Suppressed),
420                // 新 attempt 重投命中既有终态:原响应可能丢失,按既有终态以本 attempt
421                // 的身份重发结果事件,零业务效果。
422                ExecuteAdmission::AlreadyTerminal(terminal) => {
423                    let status = forward_to_attempt_status(terminal)?;
424                    self.emit_result(receipt_trace, envelope, status, None, None)
425                        .await?;
426                    Ok(ParticipantHandled::Replayed(status))
427                }
428                ExecuteAdmission::Admitted => {
429                    let context = self.context_of(&identity, envelope)?;
430                    let command: S::Command = envelope.decode_payload()?;
431                    match crate::latency::measure_handler(service.execute(&context, command)).await
432                    {
433                        Ok(SagaOutcome::Succeeded(_)) => {
434                            self.backend
435                                .store()
436                                .settle_execute(
437                                    &identity.saga_id,
438                                    &identity.step,
439                                    StepForwardStatus::Succeeded,
440                                )
441                                .await?;
442                            self.emit_result(
443                                receipt_trace,
444                                envelope,
445                                StepAttemptStatus::Succeeded,
446                                None,
447                                None,
448                            )
449                            .await?;
450                            Ok(ParticipantHandled::Executed(StepAttemptStatus::Succeeded))
451                        }
452                        Ok(SagaOutcome::Rejected {
453                            code,
454                        }) => {
455                            // 确定性拒绝是可提交结论:拒绝事实与 Inbox、结果事件原子提交。
456                            self.backend
457                                .store()
458                                .settle_execute(
459                                    &identity.saga_id,
460                                    &identity.step,
461                                    StepForwardStatus::Rejected,
462                                )
463                                .await?;
464                            self.emit_result(
465                                receipt_trace,
466                                envelope,
467                                StepAttemptStatus::Rejected,
468                                None,
469                                Some(safe_reason_code(code)),
470                            )
471                            .await?;
472                            Ok(ParticipantHandled::Executed(StepAttemptStatus::Rejected))
473                        }
474                        Ok(SagaOutcome::Unknown {
475                            code,
476                        }) => {
477                            if allow_unknown {
478                                self.backend
479                                    .store()
480                                    .settle_execute(
481                                        &identity.saga_id,
482                                        &identity.step,
483                                        StepForwardStatus::Unknown,
484                                    )
485                                    .await?;
486                                self.emit_result(
487                                    receipt_trace,
488                                    envelope,
489                                    StepAttemptStatus::Unknown,
490                                    None,
491                                    Some(safe_reason_code(code)),
492                                )
493                                .await?;
494                                Ok(ParticipantHandled::Executed(StepAttemptStatus::Unknown))
495                            } else {
496                                // 纯本地步骤谎称 Unknown = 合同违规:提交 Halted 冻结事实
497                                // 并告警,不能让"未知"进入无界等待。
498                                self.backend
499                                    .store()
500                                    .settle_execute(
501                                        &identity.saga_id,
502                                        &identity.step,
503                                        StepForwardStatus::Halted,
504                                    )
505                                    .await?;
506                                self.emit_result(
507                                    receipt_trace,
508                                    envelope,
509                                    StepAttemptStatus::Halted,
510                                    None,
511                                    Some("unknown_not_allowed"),
512                                )
513                                .await?;
514                                Ok(ParticipantHandled::Executed(StepAttemptStatus::Halted))
515                            }
516                        }
517                        Ok(SagaOutcome::Halted {
518                            code,
519                        }) => {
520                            self.backend
521                                .store()
522                                .settle_execute(
523                                    &identity.saga_id,
524                                    &identity.step,
525                                    StepForwardStatus::Halted,
526                                )
527                                .await?;
528                            self.emit_result(
529                                receipt_trace,
530                                envelope,
531                                StepAttemptStatus::Halted,
532                                None,
533                                Some(safe_reason_code(code)),
534                            )
535                            .await?;
536                            Ok(ParticipantHandled::Executed(StepAttemptStatus::Halted))
537                        }
538                        // 本次尝试没有可提交结论:回滚全部本地写并保留输入消息重投。
539                        Err(SagaExecutionError::Retryable {
540                            code,
541                        }) => {
542                            anyhow::bail!("retryable execution error: {code}")
543                        }
544                    }
545                }
546            }
547        })
548        .await
549    }
550
551    /// 业务作用：`local_fenceable` 步骤的取消命令 wrapper——在 gate 行锁上完成取消屏障裁决。
552    ///
553    /// 参数说明：
554    /// - `envelope`: 已认证的取消命令 envelope。
555    /// - `producer`: transport 认证并映射出的 Orchestrator 逻辑身份。
556    ///
557    /// 返回：可 ACK 的裁决结论（Confirmed/AlreadyTerminal/ResolutionPending 之一已
558    /// 连同结果事件提交）；基础设施失败返回错误（已回滚，不得 ACK）。
559    pub async fn handle_authenticated_cancel_local(
560        &self,
561        envelope: &SagaCommandEnvelope,
562        producer: &ServiceIdentity,
563    ) -> anyhow::Result<ParticipantHandled> {
564        self.handle_authenticated_cancel_local_traced(envelope, producer, None)
565            .await
566    }
567
568    /// 业务作用：携带命令收据链路上下文的本地可围栏取消事务 wrapper；结果事件由收据
569    /// 派生子上下文，语义与 [`handle_authenticated_cancel_local`](Self::handle_authenticated_cancel_local) 一致。
570    ///
571    /// 参数说明：
572    /// - `envelope`: 已通过 transport owner/route 认证的 cancel command。
573    /// - `producer`: transport 认证并映射出的 Orchestrator 逻辑身份。
574    /// - `receipt_trace`: transport 收据显式解析出的链路上下文；`None` 表示未携带。
575    ///
576    /// 返回：真实裁决与结果事件 COMMIT 后返回可 ACK 结论；失败返回错误并回滚。
577    pub async fn handle_authenticated_cancel_local_traced(
578        &self,
579        envelope: &SagaCommandEnvelope,
580        producer: &ServiceIdentity,
581        receipt_trace: Option<&TraceContext>,
582    ) -> anyhow::Result<ParticipantHandled> {
583        let identity = self.verified_command_identity(envelope, producer)?;
584        if identity.phase != StepPhase::Cancel {
585            return Err(crate::SagaCommandProcessingError::ContractInvalid.into());
586        }
587        let _latency =
588            crate::latency::LatencyGuard::new(crate::latency::LatencyStage::ParticipantTransaction);
589        crate::transaction::run_for(&self.backend, async {
590            if matches!(
591                self.backend
592                    .inbox()
593                    .claim(&self.consumer, &envelope.command_id)
594                    .await?,
595                InboxClaim::Duplicate
596            ) {
597                return Ok(ParticipantHandled::Duplicate);
598            }
599            let gate = ParticipantGateKey {
600                saga_id: &identity.saga_id,
601                step: &identity.step,
602                tenant: &identity.tenant,
603                workflow: &identity.workflow,
604                definition_version: identity.definition_version,
605                definition_digest: &envelope.definition_digest,
606            };
607            // 取消与 execute 竞争同一效果身份:execute effect 由参与方本地派生,
608            // 不信任命令自带的任何执行身份。
609            let execute_effect = EffectId::derive(
610                &identity.saga_id,
611                identity.definition_version,
612                &identity.step,
613                StepPhase::Execute,
614            );
615            let (status, terminal) = match self
616                .backend
617                .store()
618                .adjudicate_cancel(&gate, &execute_effect, &identity.effect)
619                .await?
620            {
621                CancelAdjudication::Confirmed => (StepAttemptStatus::CancelConfirmed, None),
622                CancelAdjudication::AlreadyTerminal(forward) => {
623                    (StepAttemptStatus::AlreadyTerminal, Some(forward))
624                }
625                CancelAdjudication::ResolutionPending => {
626                    (StepAttemptStatus::ResolutionPending, None)
627                }
628            };
629            self.emit_result(receipt_trace, envelope, status, terminal, None)
630                .await?;
631            Ok(ParticipantHandled::Executed(status))
632        })
633        .await
634    }
635
636    /// 业务作用：externally-cancellable 步骤的类型化取消事务 wrapper。
637    ///
638    /// 固定顺序为 Inbox claim → gate 行锁/取消准入 → 外部 cancel handler → 真实裁决落账 →
639    /// result Outbox → COMMIT。handler 调用期间 gate 权威持续持有，使并发 execute 不可能越过
640    /// 已确认取消；`ResolutionPending` 如实保留 Unknown，交 Poll resolver 收敛。
641    ///
642    /// 参数说明：
643    /// - `service`: 同时托管步骤取消能力的业务 Service。
644    /// - `envelope`: 已通过 transport owner/route 认证的 cancel command。
645    /// - `producer`: transport 认证并映射出的 Orchestrator 逻辑身份。
646    ///
647    /// 返回：真实裁决与结果事件 COMMIT 后返回可 ACK 结论；`Retryable`、payload 合同、失权
648    /// 或基础设施失败返回错误并回滚，transport 不得 ACK。
649    pub async fn handle_authenticated_cancel_external<C>(
650        &self,
651        service: &C,
652        envelope: &SagaCommandEnvelope,
653        producer: &ServiceIdentity,
654    ) -> anyhow::Result<ParticipantHandled>
655    where
656        C: SagaCancelStep,
657        C::Command: DeserializeOwned,
658    {
659        self.handle_authenticated_cancel_external_traced(service, envelope, producer, None)
660            .await
661    }
662
663    /// 业务作用：携带命令收据链路上下文的外部取消事务 wrapper；结果事件由收据派生
664    /// 子上下文，语义与 [`handle_authenticated_cancel_external`](Self::handle_authenticated_cancel_external) 一致。
665    ///
666    /// 参数说明：
667    /// - `service`: 同时托管步骤取消能力的业务 Service。
668    /// - `envelope`: 已通过 transport owner/route 认证的 cancel command。
669    /// - `producer`: transport 认证并映射出的 Orchestrator 逻辑身份。
670    /// - `receipt_trace`: transport 收据显式解析出的链路上下文；`None` 表示未携带。
671    ///
672    /// 返回：真实裁决与结果事件 COMMIT 后返回可 ACK 结论；失败返回错误并回滚。
673    pub async fn handle_authenticated_cancel_external_traced<C>(
674        &self,
675        service: &C,
676        envelope: &SagaCommandEnvelope,
677        producer: &ServiceIdentity,
678        receipt_trace: Option<&TraceContext>,
679    ) -> anyhow::Result<ParticipantHandled>
680    where
681        C: SagaCancelStep,
682        C::Command: DeserializeOwned,
683    {
684        let identity = self.verified_command_identity(envelope, producer)?;
685        if identity.phase != StepPhase::Cancel {
686            return Err(crate::SagaCommandProcessingError::ContractInvalid.into());
687        }
688        let _latency =
689            crate::latency::LatencyGuard::new(crate::latency::LatencyStage::ParticipantTransaction);
690        crate::transaction::run_for(&self.backend, async {
691            if matches!(
692                self.backend
693                    .inbox()
694                    .claim(&self.consumer, &envelope.command_id)
695                    .await?,
696                InboxClaim::Duplicate
697            ) {
698                return Ok(ParticipantHandled::Duplicate);
699            }
700            let gate = ParticipantGateKey {
701                saga_id: &identity.saga_id,
702                step: &identity.step,
703                tenant: &identity.tenant,
704                workflow: &identity.workflow,
705                definition_version: identity.definition_version,
706                definition_digest: &envelope.definition_digest,
707            };
708            let execute_effect = EffectId::derive(
709                &identity.saga_id,
710                identity.definition_version,
711                &identity.step,
712                StepPhase::Execute,
713            );
714            match self
715                .backend
716                .store()
717                .admit_external_cancel(&gate, &execute_effect, &identity.effect)
718                .await?
719            {
720                ExternalCancelAdmission::AlreadyAdjudicated(adjudication) => {
721                    // 已提交裁决只能重放 result；再次调用外部取消接口可能把幂等缺陷放大为
722                    // 误取消别的业务请求，因此 gate 事实优先于新 delivery attempt。
723                    let (status, terminal) = cancel_adjudication_result(adjudication);
724                    self.emit_result(receipt_trace, envelope, status, terminal, None)
725                        .await?;
726                    Ok(ParticipantHandled::Replayed(status))
727                }
728                ExternalCancelAdmission::Admitted => {
729                    // cancel 操作自己的 effect 用于取消请求幂等；目标 effect 必须指向原
730                    // execute，否则业务 API 会稳定地取消一条不存在的 cancel-phase 请求。
731                    let context = self
732                        .context_of(&identity, envelope)?
733                        .with_target_phase(StepPhase::Execute);
734                    let command: C::Command = envelope.decode_payload()?;
735                    match crate::latency::measure_handler(service.cancel(&context, command)).await {
736                        Ok(outcome) => {
737                            let (forward, cancel, status, terminal, reason) =
738                                external_cancel_result(outcome);
739                            let reason = reason.map(safe_reason_code);
740                            // 外部真实裁决、gate 与 result Outbox 同事务提交；先发 result 会让
741                            // Orchestrator 相信一个在参与方崩溃后并不存在的取消屏障。
742                            self.backend
743                                .store()
744                                .settle_external_cancel(
745                                    &identity.saga_id,
746                                    &identity.step,
747                                    forward,
748                                    cancel,
749                                )
750                                .await?;
751                            self.emit_result(receipt_trace, envelope, status, terminal, reason)
752                                .await?;
753                            Ok(ParticipantHandled::Executed(status))
754                        }
755                        Err(SagaExecutionError::Retryable {
756                            code,
757                        }) => {
758                            // 未获真实裁决时回滚 PENDING gate、Inbox 与所有业务写，让同一
759                            // command 保持可重投；绝不能把调用失败当作取消成功。
760                            anyhow::bail!("retryable external cancel error: {code}")
761                        }
762                    }
763                }
764            }
765        })
766        .await
767    }
768
769    /// 业务作用：补偿命令的完整本地事务 wrapper——按 `effect_id` 幂等撤销正向效果。
770    ///
771    /// 参数说明：
772    /// - `service`: 业务步骤实现。
773    /// - `envelope`: 已认证的补偿命令 envelope。
774    /// - `producer`: transport 认证并映射出的 Orchestrator 逻辑身份。
775    ///
776    /// 返回：可 ACK 的处理结论；补偿已成功时按幂等成功重发结果；本地无正向成功效果
777    /// 时提交 `Halted` 合同违规事实（[`ParticipantHandled::ContractViolation`]）；
778    /// `Retryable` 或基础设施失败返回错误（已回滚，不得 ACK）。
779    pub async fn handle_authenticated_compensate<S>(
780        &self,
781        service: &S,
782        envelope: &SagaCommandEnvelope,
783        producer: &ServiceIdentity,
784    ) -> anyhow::Result<ParticipantHandled>
785    where
786        S: SagaStep,
787        S::Compensation: DeserializeOwned,
788    {
789        self.handle_authenticated_compensate_traced(service, envelope, producer, None)
790            .await
791    }
792
793    /// 业务作用：携带命令收据链路上下文的补偿事务 wrapper；结果事件由收据派生子上下文，
794    /// 语义与 [`handle_authenticated_compensate`](Self::handle_authenticated_compensate) 一致。
795    ///
796    /// 参数说明：
797    /// - `service`: 业务步骤实现。
798    /// - `envelope`: 已通过 transport owner/route 认证的 compensate command。
799    /// - `producer`: transport 认证并映射出的 Orchestrator 逻辑身份。
800    /// - `receipt_trace`: transport 收据显式解析出的链路上下文；`None` 表示未携带。
801    ///
802    /// 返回：补偿裁决与结果事件 COMMIT 后返回可 ACK 结论；失败返回错误并回滚。
803    pub async fn handle_authenticated_compensate_traced<S>(
804        &self,
805        service: &S,
806        envelope: &SagaCommandEnvelope,
807        producer: &ServiceIdentity,
808        receipt_trace: Option<&TraceContext>,
809    ) -> anyhow::Result<ParticipantHandled>
810    where
811        S: SagaStep,
812        S::Compensation: DeserializeOwned,
813    {
814        let identity = self.verified_command_identity(envelope, producer)?;
815        if identity.phase != StepPhase::Compensate {
816            return Err(crate::SagaCommandProcessingError::ContractInvalid.into());
817        }
818        let _latency =
819            crate::latency::LatencyGuard::new(crate::latency::LatencyStage::ParticipantTransaction);
820        crate::transaction::run_for(&self.backend, async {
821            if matches!(
822                self.backend
823                    .inbox()
824                    .claim(&self.consumer, &envelope.command_id)
825                    .await?,
826                InboxClaim::Duplicate
827            ) {
828                return Ok(ParticipantHandled::Duplicate);
829            }
830            let gate = ParticipantGateKey {
831                saga_id: &identity.saga_id,
832                step: &identity.step,
833                tenant: &identity.tenant,
834                workflow: &identity.workflow,
835                definition_version: identity.definition_version,
836                definition_digest: &envelope.definition_digest,
837            };
838            match self
839                .backend
840                .store()
841                .admit_compensation(
842                    &gate,
843                    &identity.effect,
844                    envelope.recovery_operation_id.as_deref(),
845                )
846                .await?
847            {
848                // 既有补偿事实必须原样重放。尤其 UNKNOWN 可能是退款已生效但响应丢失，
849                // 自动再次调用 handler 会把“查询未知”升级成二次退款事故。
850                CompensationAdmission::AlreadySettled(terminal) => {
851                    let status = compensation_to_attempt_status(terminal)?;
852                    let reason = match terminal {
853                        StepCompensationStatus::Unknown => Some("compensation_outcome_unknown"),
854                        StepCompensationStatus::Halted => Some("compensation_already_halted"),
855                        _ => None,
856                    };
857                    self.emit_result(receipt_trace, envelope, status, None, reason)
858                        .await?;
859                    Ok(ParticipantHandled::Replayed(status))
860                }
861                // 协议破坏:补偿只会对已记账成功的步骤发出。提交违规事实并告警,
862                // 不自旋等待正向命令,也不进入无限重试。
863                CompensationAdmission::MissingForwardEffect => {
864                    self.emit_result(
865                        receipt_trace,
866                        envelope,
867                        StepAttemptStatus::Halted,
868                        None,
869                        Some("compensation_contract_violation"),
870                    )
871                    .await?;
872                    Ok(ParticipantHandled::ContractViolation)
873                }
874                CompensationAdmission::Admitted => {
875                    let context = self.context_of(&identity, envelope)?;
876                    // 补偿输入来自持久化事实(默认形态 (a)):命令只带身份,
877                    // 业务按 saga_id + step 自行解析要撤销的内容。
878                    let command: S::Compensation = envelope.decode_payload()?;
879                    match crate::latency::measure_handler(service.compensate(&context, command))
880                        .await
881                    {
882                        Ok(CompensationOutcome::Succeeded) => {
883                            self.backend
884                                .store()
885                                .settle_compensation(
886                                    &identity.saga_id,
887                                    &identity.step,
888                                    StepCompensationStatus::Succeeded,
889                                )
890                                .await?;
891                            self.emit_result(
892                                receipt_trace,
893                                envelope,
894                                StepAttemptStatus::Succeeded,
895                                None,
896                                None,
897                            )
898                            .await?;
899                            Ok(ParticipantHandled::Executed(StepAttemptStatus::Succeeded))
900                        }
901                        Ok(CompensationOutcome::Unknown {
902                            code,
903                        }) => {
904                            self.backend
905                                .store()
906                                .settle_compensation(
907                                    &identity.saga_id,
908                                    &identity.step,
909                                    StepCompensationStatus::Unknown,
910                                )
911                                .await?;
912                            self.emit_result(
913                                receipt_trace,
914                                envelope,
915                                StepAttemptStatus::Unknown,
916                                None,
917                                Some(safe_reason_code(code)),
918                            )
919                            .await?;
920                            Ok(ParticipantHandled::Executed(StepAttemptStatus::Unknown))
921                        }
922                        Ok(CompensationOutcome::Halted {
923                            code,
924                        }) => {
925                            self.backend
926                                .store()
927                                .settle_compensation(
928                                    &identity.saga_id,
929                                    &identity.step,
930                                    StepCompensationStatus::Halted,
931                                )
932                                .await?;
933                            self.emit_result(
934                                receipt_trace,
935                                envelope,
936                                StepAttemptStatus::Halted,
937                                None,
938                                Some(safe_reason_code(code)),
939                            )
940                            .await?;
941                            Ok(ParticipantHandled::Executed(StepAttemptStatus::Halted))
942                        }
943                        Err(SagaExecutionError::Retryable {
944                            code,
945                        }) => {
946                            anyhow::bail!("retryable compensation error: {code}")
947                        }
948                    }
949                }
950            }
951        })
952        .await
953    }
954
955    /// 业务作用：运行 `poll` 解决命令的类型化本地事务，把未知外部效果收敛为唯一裁决。
956    ///
957    /// Inbox claim、gate 方向裁决、查询结论与结果 Outbox 在同一事务提交；`Unknown` 只保持
958    /// `PENDING` 供下一有界 attempt 继续查询，确定终态以后新 attempt 只能重放，避免重复调用
959    /// 可能带 void/cancel-inquiry 副作用的裁决型接口。
960    ///
961    /// 参数说明：
962    /// - `service`: 实现 [`SagaResolveStep`] 的类型化查询服务。
963    /// - `envelope`: 已通过 transport producer 认证的 resolve 命令。
964    /// - `producer`: transport 认证并映射出的 Orchestrator 逻辑身份。
965    ///
966    /// 返回：事务提交后返回可 ACK 结论；`Retryable`、身份复验、方向合同或基础设施失败
967    /// 返回错误并回滚，transport 不得 ACK。
968    pub async fn handle_authenticated_resolve<R>(
969        &self,
970        service: &R,
971        envelope: &SagaCommandEnvelope,
972        producer: &ServiceIdentity,
973    ) -> anyhow::Result<ParticipantHandled>
974    where
975        R: SagaResolveStep,
976        R::Command: DeserializeOwned,
977    {
978        self.handle_authenticated_resolve_traced(service, envelope, producer, None)
979            .await
980    }
981
982    /// 业务作用：携带命令收据链路上下文的解决查询事务 wrapper；结果事件由收据派生
983    /// 子上下文，语义与 [`handle_authenticated_resolve`](Self::handle_authenticated_resolve) 一致。
984    ///
985    /// 参数说明：
986    /// - `service`: 托管解决查询能力的业务 Service。
987    /// - `envelope`: 已通过 transport owner/route 认证的 resolve command。
988    /// - `producer`: transport 认证并映射出的 Orchestrator 逻辑身份。
989    /// - `receipt_trace`: transport 收据显式解析出的链路上下文；`None` 表示未携带。
990    ///
991    /// 返回：解决裁决与结果事件 COMMIT 后返回可 ACK 结论；失败返回错误并回滚。
992    pub async fn handle_authenticated_resolve_traced<R>(
993        &self,
994        service: &R,
995        envelope: &SagaCommandEnvelope,
996        producer: &ServiceIdentity,
997        receipt_trace: Option<&TraceContext>,
998    ) -> anyhow::Result<ParticipantHandled>
999    where
1000        R: SagaResolveStep,
1001        R::Command: DeserializeOwned,
1002    {
1003        let identity = self.verified_command_identity(envelope, producer)?;
1004        if identity.phase != StepPhase::Resolve {
1005            return Err(crate::SagaCommandProcessingError::ContractInvalid.into());
1006        }
1007        let _latency =
1008            crate::latency::LatencyGuard::new(crate::latency::LatencyStage::ParticipantTransaction);
1009        crate::transaction::run_for(&self.backend, async {
1010            if matches!(
1011                self.backend
1012                    .inbox()
1013                    .claim(&self.consumer, &envelope.command_id)
1014                    .await?,
1015                InboxClaim::Duplicate
1016            ) {
1017                return Ok(ParticipantHandled::Duplicate);
1018            }
1019            let gate = ParticipantGateKey {
1020                saga_id: &identity.saga_id,
1021                step: &identity.step,
1022                tenant: &identity.tenant,
1023                workflow: &identity.workflow,
1024                definition_version: identity.definition_version,
1025                definition_digest: &envelope.definition_digest,
1026            };
1027            match self
1028                .backend
1029                .store()
1030                .admit_resolution(
1031                    &gate,
1032                    &identity.effect,
1033                    envelope.recovery_operation_id.as_deref(),
1034                )
1035                .await?
1036            {
1037                ResolutionAdmission::AlreadySettled {
1038                    status,
1039                    ..
1040                } => {
1041                    let attempt_status = resolution_to_attempt_status(status)?;
1042                    self.emit_result(receipt_trace, envelope, attempt_status, None, None)
1043                        .await?;
1044                    Ok(ParticipantHandled::Replayed(attempt_status))
1045                }
1046                ResolutionAdmission::MissingUnknownEffect => {
1047                    // 没有未知业务事实时不得调用查询 handler：凭空查询可能命中其它请求，
1048                    // 也会掩盖 command 链断裂。提交 Halted 供 Orchestrator 转人工审计。
1049                    self.emit_result(
1050                        receipt_trace,
1051                        envelope,
1052                        StepAttemptStatus::Halted,
1053                        None,
1054                        Some("resolution_contract_violation"),
1055                    )
1056                    .await?;
1057                    Ok(ParticipantHandled::ContractViolation)
1058                }
1059                ResolutionAdmission::Admitted(target) => {
1060                    // gate 已判定本次查询裁决正向还是补偿；把目标 phase/effect 明确交给
1061                    // resolver，禁止业务仅凭 resolve effect 猜测并查错远端请求。
1062                    let target_phase = match target {
1063                        nasaga_backend::ResolutionTarget::Forward => StepPhase::Execute,
1064                        nasaga_backend::ResolutionTarget::Compensation => StepPhase::Compensate,
1065                    };
1066                    let context = self
1067                        .context_of(&identity, envelope)?
1068                        .with_target_phase(target_phase);
1069                    let command: R::Command = envelope.decode_payload()?;
1070                    match crate::latency::measure_handler(service.resolve(&context, command)).await
1071                    {
1072                        Ok(SagaOutcome::Succeeded(_)) => {
1073                            self.backend
1074                                .store()
1075                                .settle_resolution(
1076                                    &identity.saga_id,
1077                                    &identity.step,
1078                                    target,
1079                                    StepResolutionStatus::Succeeded,
1080                                )
1081                                .await?;
1082                            self.emit_result(
1083                                receipt_trace,
1084                                envelope,
1085                                StepAttemptStatus::Succeeded,
1086                                None,
1087                                None,
1088                            )
1089                            .await?;
1090                            Ok(ParticipantHandled::Executed(StepAttemptStatus::Succeeded))
1091                        }
1092                        Ok(SagaOutcome::Rejected {
1093                            code,
1094                        }) => {
1095                            // Rejected 只能由 handler 在排除迟到生效后给出；本层原样可靠提交，
1096                            // 不把普通“查无记录”自行升级成确定拒绝。
1097                            self.backend
1098                                .store()
1099                                .settle_resolution(
1100                                    &identity.saga_id,
1101                                    &identity.step,
1102                                    target,
1103                                    StepResolutionStatus::Rejected,
1104                                )
1105                                .await?;
1106                            self.emit_result(
1107                                receipt_trace,
1108                                envelope,
1109                                StepAttemptStatus::Rejected,
1110                                None,
1111                                Some(safe_reason_code(code)),
1112                            )
1113                            .await?;
1114                            Ok(ParticipantHandled::Executed(StepAttemptStatus::Rejected))
1115                        }
1116                        Ok(SagaOutcome::Unknown {
1117                            code,
1118                        }) => {
1119                            self.backend
1120                                .store()
1121                                .settle_resolution(
1122                                    &identity.saga_id,
1123                                    &identity.step,
1124                                    target,
1125                                    StepResolutionStatus::Pending,
1126                                )
1127                                .await?;
1128                            self.emit_result(
1129                                receipt_trace,
1130                                envelope,
1131                                StepAttemptStatus::Unknown,
1132                                None,
1133                                Some(safe_reason_code(code)),
1134                            )
1135                            .await?;
1136                            Ok(ParticipantHandled::Executed(StepAttemptStatus::Unknown))
1137                        }
1138                        Ok(SagaOutcome::Halted {
1139                            code,
1140                        }) => {
1141                            self.backend
1142                                .store()
1143                                .settle_resolution(
1144                                    &identity.saga_id,
1145                                    &identity.step,
1146                                    target,
1147                                    StepResolutionStatus::Halted,
1148                                )
1149                                .await?;
1150                            self.emit_result(
1151                                receipt_trace,
1152                                envelope,
1153                                StepAttemptStatus::Halted,
1154                                None,
1155                                Some(safe_reason_code(code)),
1156                            )
1157                            .await?;
1158                            Ok(ParticipantHandled::Executed(StepAttemptStatus::Halted))
1159                        }
1160                        Err(SagaExecutionError::Retryable {
1161                            code,
1162                        }) => {
1163                            // 没有裁决时回滚 Inbox 与 PENDING 写入，原 command 保持可重投。
1164                            anyhow::bail!("retryable resolution error: {code}")
1165                        }
1166                    }
1167                }
1168            }
1169        })
1170        .await
1171    }
1172
1173    /// 业务作用：复验 command 身份并把 producer 绑定到该 definition 的精确信任投影。
1174    ///
1175    /// 这里必须先完成身份派生，再按 workflow/version/digest 查授权；任一失败都发生在获取数据库
1176    /// 连接和 Inbox claim 前，避免跨业务域 Orchestrator 触发未授权外部效果。
1177    ///
1178    /// 参数说明：
1179    /// - `envelope`: 待复验的 command envelope。
1180    /// - `producer`: transport 认证得到的逻辑 Orchestrator。
1181    ///
1182    /// 返回：envelope 身份和精确信任投影一致时返回已证身份；否则返回类型化拒绝。
1183    fn verified_command_identity(
1184        &self,
1185        envelope: &SagaCommandEnvelope,
1186        producer: &ServiceIdentity,
1187    ) -> anyhow::Result<crate::envelope::VerifiedIdentity> {
1188        let identity = envelope
1189            .verified()
1190            .map_err(|_| crate::SagaCommandProcessingError::IdentityInvalid)?;
1191        self.verify_command_producer(producer, &identity)?;
1192        // 本地 descriptor 是正文 schema 的权威；缺少声明时只允许兼容 JSON，不能信任命令自报 schema。
1193        let descriptor = crate::COLLECTED_SAGA_STEPS.iter().find(|descriptor| {
1194            descriptor.workflow == envelope.workflow
1195                && descriptor.definition_version == envelope.definition_version
1196                && descriptor.step == envelope.step
1197        });
1198        let (content_type, schema_id) = descriptor.map_or(("application/json", ""), |descriptor| {
1199            (
1200                descriptor.payload_content_type,
1201                descriptor.payload_schema_id,
1202            )
1203        });
1204        envelope.verify_payload_contract(content_type, schema_id)?;
1205        Ok(identity)
1206    }
1207
1208    /// 业务作用：把 transport 认证身份绑定到 envelope 指向的 workflow definition 投影。
1209    ///
1210    /// 参数说明：
1211    /// - `producer`: mTLS/SASL principal、独占 topic 或端到端签名映射出的逻辑身份。
1212    /// - `identity`: 已完成派生复验的 command 身份。
1213    ///
1214    /// 返回：producer 与精确 projection 一致时成功；不存在或被另一 Orchestrator 所有时类型化拒绝。
1215    fn verify_command_producer(
1216        &self,
1217        producer: &ServiceIdentity,
1218        identity: &crate::envelope::VerifiedIdentity,
1219    ) -> anyhow::Result<()> {
1220        let key = (
1221            identity.workflow.clone(),
1222            identity.definition_version,
1223            Some(identity.definition_digest.clone()),
1224        );
1225        let version_key = (identity.workflow.clone(), identity.definition_version, None);
1226        if self.command_trust.get(&key) != Some(producer)
1227            && self.command_trust.get(&version_key) != Some(producer)
1228        {
1229            return Err(crate::SagaCommandProcessingError::ProducerUnauthorized.into());
1230        }
1231        Ok(())
1232    }
1233
1234    /// 业务作用：由已证身份构造只读业务上下文，身份由框架派生而不是业务自造。
1235    ///
1236    /// 参数说明：
1237    /// - `identity`: envelope 已证身份。
1238    /// - `envelope`: 原始 envelope（提供 digest 文本）。
1239    ///
1240    /// 返回：可传给业务 handler 的上下文。
1241    fn context_of(
1242        &self,
1243        identity: &crate::envelope::VerifiedIdentity,
1244        envelope: &SagaCommandEnvelope,
1245    ) -> anyhow::Result<SagaContext> {
1246        Ok(SagaContext::new(
1247            identity.saga_id.clone(),
1248            identity.tenant.clone(),
1249            identity.workflow.clone(),
1250            identity.definition_version,
1251            envelope.definition_digest.clone(),
1252            identity.step.clone(),
1253            identity.phase,
1254            identity.attempt,
1255        )
1256        .with_payload(envelope.business_payload()?))
1257    }
1258
1259    /// 业务作用：把处理结论以稳定身份写入结果 Outbox，与业务写同事务发布。
1260    ///
1261    /// result 事件 id 由命令身份确定性派生：参与方事务重试不会为同一结果占用两个
1262    /// 去重身份。命令收据携带 trace 时，结果事件从它派生子上下文——参与方是链路
1263    /// 中的一跳而不是终点，编排端由此续接同一 trace。
1264    ///
1265    /// 参数说明：
1266    /// - `receipt_trace`: 命令收据中已校验的链路上下文；`None` 表示收据未携带。
1267    /// - `envelope`: 触发本结果的命令 envelope。
1268    /// - `status`: 处理结论。
1269    /// - `terminal`: 取消裁决 `ALREADY_TERMINAL` 携带的正向真实终态。
1270    /// - `reason_code`: 稳定原因码。
1271    ///
1272    /// 返回：写入成功返回 `Ok`。
1273    async fn emit_result(
1274        &self,
1275        receipt_trace: Option<&TraceContext>,
1276        envelope: &SagaCommandEnvelope,
1277        status: StepAttemptStatus,
1278        terminal: Option<StepForwardStatus>,
1279        reason_code: Option<&str>,
1280    ) -> anyhow::Result<()> {
1281        let result = SagaResultEnvelope {
1282            event_id: derive_result_event_id(&envelope.command_id),
1283            saga_id: envelope.saga_id.clone(),
1284            tenant_id: envelope.tenant_id.clone(),
1285            workflow: envelope.workflow.clone(),
1286            definition_version: envelope.definition_version,
1287            definition_digest: envelope.definition_digest.clone(),
1288            step: envelope.step.clone(),
1289            phase: envelope.phase.clone(),
1290            attempt: envelope.attempt,
1291            effect_id: envelope.effect_id.clone(),
1292            command_id: envelope.command_id.clone(),
1293            status: status.as_str().to_string(),
1294            terminal_status: terminal.map(|value| value.as_str().to_string()),
1295            reason_code: reason_code.map(str::to_string),
1296        };
1297        // 业务 reason 已在 handler 分支收敛为安全码；这里仍用完整 producer 合同
1298        // 拦截框架状态/身份漂移。这类确定性编程错误必须回滚并隔离，不得
1299        // 伪装成数据库瞬态故障耗尽重试预算。
1300        let event = result
1301            .to_outbox_event()
1302            .map_err(|_| crate::SagaCommandProcessingError::ContractInvalid)?;
1303        // 结果事件从命令收据派生子上下文:同一 trace-id 下参与方处理是新的一跳,
1304        // 编排端消费该结果时以此续接;收据未携带时按无上下文投递,不阻塞结果。
1305        let event = match receipt_trace {
1306            Some(trace) => {
1307                event.with_traceparent(trace.child(natelemetry::random_span_id()).to_traceparent())
1308            }
1309            None => event,
1310        };
1311        // 租户归因取命令合同里的已认证租户(与本地 gate 同事务写入的同一身份),
1312        // 供参与方 Outbox 层配额与对账使用;不读取 payload 自报身份。
1313        let write_context = naoutbox_core::OutboxWriteContext::new(envelope.tenant_id.as_str())
1314            .map_err(|_| crate::SagaCommandProcessingError::ContractInvalid)?;
1315        self.backend
1316            .outbox()
1317            .append_transactional_with_context(&write_context, &event)
1318            .await?;
1319        Ok(())
1320    }
1321}
1322
1323impl<B> AuthenticatedParticipantRuntime<'_, B>
1324where
1325    B: SagaBackend + SagaBackendFactory,
1326{
1327    /// 业务作用：以已绑定 producer 执行 execute 的完整本地事务。
1328    ///
1329    /// 参数说明：
1330    /// - `service`: 业务步骤实现。
1331    /// - `envelope`: 已由同一 transport 收据认证的命令。
1332    /// - `allow_unknown`: definition 是否允许正向 Unknown。
1333    ///
1334    /// 返回：事务提交后返回可 ACK 结论；否则返回拒绝或可重试错误。
1335    pub async fn execute<S>(
1336        &self,
1337        service: &S,
1338        envelope: &SagaCommandEnvelope,
1339        allow_unknown: bool,
1340    ) -> anyhow::Result<ParticipantHandled>
1341    where
1342        S: SagaStep,
1343        S::Command: DeserializeOwned,
1344    {
1345        self.runtime
1346            .handle_authenticated_execute(service, envelope, self.producer, allow_unknown)
1347            .await
1348    }
1349
1350    /// 业务作用：以已绑定 producer 执行本地可围栏取消裁决。
1351    ///
1352    /// 参数说明：
1353    /// - `envelope`: 已由同一 transport 收据认证的取消命令。
1354    ///
1355    /// 返回：取消屏障与结果提交后返回可 ACK 结论；否则返回拒绝或基础设施错误。
1356    pub async fn cancel_local(
1357        &self,
1358        envelope: &SagaCommandEnvelope,
1359    ) -> anyhow::Result<ParticipantHandled> {
1360        self.runtime
1361            .handle_authenticated_cancel_local(envelope, self.producer)
1362            .await
1363    }
1364
1365    /// 业务作用：以已绑定 producer 执行类型化外部取消裁决。
1366    ///
1367    /// 参数说明：
1368    /// - `service`: 外部取消服务实现。
1369    /// - `envelope`: 已认证取消命令。
1370    ///
1371    /// 返回：外部裁决、gate 与结果原子提交后返回可 ACK 结论；否则返回错误。
1372    pub async fn cancel_external<C>(
1373        &self,
1374        service: &C,
1375        envelope: &SagaCommandEnvelope,
1376    ) -> anyhow::Result<ParticipantHandled>
1377    where
1378        C: SagaCancelStep,
1379        C::Command: DeserializeOwned,
1380    {
1381        self.runtime
1382            .handle_authenticated_cancel_external(service, envelope, self.producer)
1383            .await
1384    }
1385
1386    /// 业务作用：以已绑定 producer 执行补偿或受审计 HALTED 恢复。
1387    ///
1388    /// 参数说明：
1389    /// - `service`: 补偿业务实现。
1390    /// - `envelope`: 已认证补偿命令。
1391    ///
1392    /// 返回：补偿事实与结果提交后返回可 ACK 结论；否则返回拒绝或可重试错误。
1393    pub async fn compensate<S>(
1394        &self,
1395        service: &S,
1396        envelope: &SagaCommandEnvelope,
1397    ) -> anyhow::Result<ParticipantHandled>
1398    where
1399        S: SagaStep,
1400        S::Compensation: DeserializeOwned,
1401    {
1402        self.runtime
1403            .handle_authenticated_compensate(service, envelope, self.producer)
1404            .await
1405    }
1406
1407    /// 业务作用：以已绑定 producer 执行 Unknown 查询或受审计 HALTED resolve 恢复。
1408    ///
1409    /// 参数说明：
1410    /// - `service`: 类型化查询实现。
1411    /// - `envelope`: 已认证 resolve 命令。
1412    ///
1413    /// 返回：查询裁决与结果提交后返回可 ACK 结论；否则返回拒绝或可重试错误。
1414    pub async fn resolve<R>(
1415        &self,
1416        service: &R,
1417        envelope: &SagaCommandEnvelope,
1418    ) -> anyhow::Result<ParticipantHandled>
1419    where
1420        R: SagaResolveStep,
1421        R::Command: DeserializeOwned,
1422    {
1423        self.runtime
1424            .handle_authenticated_resolve(service, envelope, self.producer)
1425            .await
1426    }
1427}
1428
1429/// 业务作用：校验信任投影中的 definition digest 使用唯一的 canonical 文本表示。
1430///
1431/// 参数说明：
1432/// - `digest`: 待校验的摘要文本。
1433///
1434/// 返回：恰好 64 字符且仅含小写十六进制字符时为 `true`。
1435fn valid_definition_digest(digest: &str) -> bool {
1436    digest.len() == 64
1437        && digest
1438            .bytes()
1439            .all(|byte| byte.is_ascii_digit() || matches!(byte, b'a'..=b'f'))
1440}
1441
1442/// 业务作用：把业务 handler 返回的非规范原因码收敛为稳定低敏码，避免已发生外部效果后
1443/// 因 result 合同校验失败回滚 gate/Inbox，继而在下一 attempt 重复调用外部系统。
1444///
1445/// 参数说明：
1446/// - `code`: handler 返回的静态原因码；不得记录其原文，避免把误填敏感文本写入日志。
1447///
1448/// 返回：合法时原样返回；非法时返回可持久化的统一合同违规码。
1449fn safe_reason_code(code: &'static str) -> &'static str {
1450    if valid_reason_code(code) {
1451        code
1452    } else {
1453        tracing::error!(
1454            reason_code_valid = false,
1455            "Saga 参与方原因码不符合传输合同，已安全降级并保留业务裁决"
1456        );
1457        "participant_reason_code_invalid"
1458    }
1459}
1460
1461/// 业务作用：把 gate 的正向终态映射为 attempt 结果词汇，用于终态重放。
1462///
1463/// 参数说明：
1464/// - `terminal`: gate 行的正向终态。
1465///
1466/// 返回：对应的 attempt 状态；`PENDING`/`CANCELLED`/`TIMED_OUT` 不是可重放终态，返回错误。
1467fn forward_to_attempt_status(terminal: StepForwardStatus) -> anyhow::Result<StepAttemptStatus> {
1468    match terminal {
1469        StepForwardStatus::Succeeded => Ok(StepAttemptStatus::Succeeded),
1470        StepForwardStatus::Rejected => Ok(StepAttemptStatus::Rejected),
1471        StepForwardStatus::Unknown => Ok(StepAttemptStatus::Unknown),
1472        StepForwardStatus::Halted => Ok(StepAttemptStatus::Halted),
1473        _ => anyhow::bail!("gate terminal status is not replayable"),
1474    }
1475}
1476
1477/// 业务作用：把 store 中既有取消裁决映射为可重发的 result envelope 字段。
1478///
1479/// 参数说明：
1480/// - `adjudication`: gate 已提交的取消裁决。
1481///
1482/// 返回：对应 attempt 状态与 `ALREADY_TERMINAL` 正向终态。
1483fn cancel_adjudication_result(
1484    adjudication: CancelAdjudication,
1485) -> (StepAttemptStatus, Option<StepForwardStatus>) {
1486    match adjudication {
1487        CancelAdjudication::Confirmed => (StepAttemptStatus::CancelConfirmed, None),
1488        CancelAdjudication::AlreadyTerminal(forward) => {
1489            (StepAttemptStatus::AlreadyTerminal, Some(forward))
1490        }
1491        CancelAdjudication::ResolutionPending => (StepAttemptStatus::ResolutionPending, None),
1492    }
1493}
1494
1495/// 业务作用：把类型化外部取消真实裁决投影为 participant gate 与 result 协议字段。
1496///
1497/// 参数说明：
1498/// - `outcome`: `SagaCancelStep` 返回的封闭裁决。
1499///
1500/// 返回：一致的 forward/cancel 状态、attempt 结果、可选正向终态与稳定原因码。
1501#[allow(clippy::type_complexity)]
1502fn external_cancel_result(
1503    outcome: CancelOutcome,
1504) -> (
1505    StepForwardStatus,
1506    StepCancelStatus,
1507    StepAttemptStatus,
1508    Option<StepForwardStatus>,
1509    Option<&'static str>,
1510) {
1511    match outcome {
1512        CancelOutcome::CancelConfirmed => (
1513            StepForwardStatus::Cancelled,
1514            StepCancelStatus::Confirmed,
1515            StepAttemptStatus::CancelConfirmed,
1516            None,
1517            None,
1518        ),
1519        CancelOutcome::AlreadyTerminal(terminal) => {
1520            let (forward, reason) = match terminal {
1521                TerminalForwardOutcome::Succeeded => (StepForwardStatus::Succeeded, None),
1522                TerminalForwardOutcome::Rejected {
1523                    code,
1524                } => (StepForwardStatus::Rejected, Some(code)),
1525                TerminalForwardOutcome::Halted {
1526                    code,
1527                } => (StepForwardStatus::Halted, Some(code)),
1528            };
1529            (
1530                forward,
1531                StepCancelStatus::AlreadyTerminal,
1532                StepAttemptStatus::AlreadyTerminal,
1533                Some(forward),
1534                reason,
1535            )
1536        }
1537        CancelOutcome::ResolutionPending {
1538            code,
1539        } => (
1540            StepForwardStatus::Unknown,
1541            StepCancelStatus::ResolutionPending,
1542            StepAttemptStatus::ResolutionPending,
1543            None,
1544            Some(code),
1545        ),
1546    }
1547}
1548
1549/// 业务作用：把参与方 gate 的补偿既有事实映射为 attempt 结果，用于安全重放。
1550///
1551/// 参数说明：
1552/// - `terminal`: gate 中已提交的补偿状态。
1553///
1554/// 返回：`Succeeded/Unknown/Halted` 返回同语义 attempt 状态；`None/Pending` 不是可重放
1555/// 事实，返回错误并拒绝伪造结果。
1556fn compensation_to_attempt_status(
1557    terminal: StepCompensationStatus,
1558) -> anyhow::Result<StepAttemptStatus> {
1559    match terminal {
1560        StepCompensationStatus::Succeeded => Ok(StepAttemptStatus::Succeeded),
1561        StepCompensationStatus::Unknown => Ok(StepAttemptStatus::Unknown),
1562        StepCompensationStatus::Halted => Ok(StepAttemptStatus::Halted),
1563        StepCompensationStatus::None | StepCompensationStatus::Pending => {
1564            anyhow::bail!("compensation gate status is not replayable")
1565        }
1566    }
1567}
1568
1569/// 业务作用：把参与方 gate 的解决终态映射为 attempt 结果，供确定裁决幂等重放。
1570///
1571/// 参数说明：
1572/// - `status`: gate 中已提交的解决状态。
1573///
1574/// 返回：`Succeeded/Rejected/Halted` 返回同语义 attempt 状态；`None/Pending` 尚未形成
1575/// 确定裁决，返回错误并拒绝伪造结果。
1576fn resolution_to_attempt_status(status: StepResolutionStatus) -> anyhow::Result<StepAttemptStatus> {
1577    match status {
1578        StepResolutionStatus::Succeeded => Ok(StepAttemptStatus::Succeeded),
1579        StepResolutionStatus::Rejected => Ok(StepAttemptStatus::Rejected),
1580        StepResolutionStatus::Halted => Ok(StepAttemptStatus::Halted),
1581        StepResolutionStatus::None | StepResolutionStatus::Pending => {
1582            anyhow::bail!("resolution gate status is not replayable")
1583        }
1584    }
1585}
nasaga_runtime_core/participant.rs

nasaga_runtime_core/
participant.rs
