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

nasaga_runtime_core/
participant.rs
