nasaga_runtime_core/
envelope.rs

1//! Saga command/result envelope：跨服务传输的受保护 header 与身份复验。
2//!
3//! envelope 至少携带 `saga_id`、workflow、definition version/digest、step、phase、
4//! attempt、`effect_id`、`command_id` 与 tenant。**身份不可信自报**：
5//! 消费方必须用 `verified()` 把文本字段解析回封闭类型，并重新派生 effect/command 身份
6//! 与自报值比对——派生是确定性的，比对失败即说明 envelope 被伪造、损坏或派生规则漂移，
7//! 一律拒绝进入业务路径。producer 身份认证（topic-to-owner 映射/端到端签名）发生在
8//! transport 接入层，本层假定 envelope 已通过该认证。
9
10use naoutbox_core::OutboxEvent;
11use nasaga_core::{
12    AttemptNo, CommandId, DefinitionVersion, EffectId, SagaId, StepAttemptStatus,
13    StepForwardStatus, StepName, StepPhase, TenantId, WorkflowName,
14};
15use serde::{Deserialize, Serialize};
16use uuid::Uuid;
17
18/// command envelope 的 Outbox event type；下游按此路由到 command topic。
19pub const COMMAND_EVENT_TYPE: &str = "saga.command";
20
21/// result envelope 的 Outbox event type；下游按此路由到 result topic。
22pub const RESULT_EVENT_TYPE: &str = "saga.result";
23
24/// Saga 事件的聚合类型；`aggregate_id = saga_id` 兼作 Kafka partition key——
25/// 顺序只是优化，正确性只依赖状态机、Inbox 与 CAS。
26pub const SAGA_AGGREGATE_TYPE: &str = "saga";
27
28/// 稳定原因码的最大字节数；与 `saga_instance.failure_code`、
29/// `saga_step.last_error_code` 的 VARCHAR(64) 持久化合同一致。
30const REASON_CODE_MAX_LEN: usize = 64;
31
32/// 业务作用：校验原因码能否安全进入持久化列、指标维度和跨服务 result envelope。
33///
34/// 参数说明：
35/// - `code`: 业务 handler 返回的稳定原因码。
36///
37/// 返回：非空、至多 64 字节且只含可移植标识符字符时返回真。
38pub(crate) fn valid_reason_code(code: &str) -> bool {
39    !code.is_empty()
40        && code.len() <= REASON_CODE_MAX_LEN
41        && code
42            .bytes()
43            .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'_' | b'-' | b'.'))
44}
45
46/// 派生 result 事件身份的固定命名空间（ASCII `nasasaga-v1-rsns`）。
47///
48/// 发布后**不得修改**：同一 command 的 result 事件身份必须跨进程稳定，
49/// 参与方事务重试才不会为同一结果占用两个 Inbox 去重身份。
50const RESULT_EVENT_NAMESPACE: Uuid = Uuid::from_bytes(*b"nasasaga-v1-rsns");
51
52/// 业务作用：以长度前缀方式拼接字段，消除字段边界歧义（与 nasaga-core 同规则）。
53///
54/// 参数说明：
55/// - `fields`: 按固定顺序给出的字段字节切片。
56///
57/// 返回：可送入 UUIDv5 的规范字节串。
58#[doc(hidden)]
59pub fn canonical_bytes(fields: &[&[u8]]) -> Vec<u8> {
60    let mut encoded = Vec::new();
61    for field in fields {
62        encoded.extend_from_slice(&(field.len() as u64).to_be_bytes());
63        encoded.extend_from_slice(field);
64    }
65    encoded
66}
67
68/// 业务作用：从命令身份确定性派生 result 事件身份，使参与方崩溃重试不产生第二个结果事件身份。
69///
70/// 参数说明：
71/// - `command_id`: 触发该结果的命令身份文本。
72///
73/// 返回：跨进程一致的 result 事件 id（UUID 文本）。
74pub fn derive_result_event_id(command_id: &str) -> String {
75    Uuid::new_v5(
76        &RESULT_EVENT_NAMESPACE,
77        &canonical_bytes(&[command_id.as_bytes(), b"result"]),
78    )
79    .to_string()
80}
81
82/// 业务作用：Orchestrator 发往参与方的 execute/cancel/compensate/resolve 命令 envelope。
83///
84/// 字段说明：`payload` 只在首步 execute 由启动请求携带；其余命令只携带 Saga 与 step
85/// 身份，参与方按 `saga_id + step` 从本地事实解析业务输入，天然避免敏感载荷跨服务复制。
86#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
87pub struct SagaCommandEnvelope {
88    /// 实例身份。
89    pub saga_id: String,
90    /// 规范化租户；无租户部署为固定 system tenant。
91    pub tenant_id: String,
92    /// workflow 名称。
93    pub workflow: String,
94    /// 实例固定的 definition 版本。
95    pub definition_version: u32,
96    /// 实例固定的 definition 摘要。
97    pub definition_digest: String,
98    /// 步骤名称。
99    pub step: String,
100    /// 执行阶段稳定名（execute/cancel/compensate/resolve）。
101    pub phase: String,
102    /// 尝试序号。
103    pub attempt: u32,
104    /// 跨 attempt 稳定的业务效果身份。
105    pub effect_id: String,
106    /// 本次尝试的命令身份；同一 attempt 传输重投复用。
107    pub command_id: String,
108    /// 仅由已审计管理操作签发的恢复身份；普通自动命令必须为空。
109    ///
110    /// Participant 只允许携带该字段的新命令重新打开 `HALTED` 查询或补偿 gate；因此
111    /// 自动 timeout/new attempt 不能解除人工冻结。旧格式 envelope 缺字段时按 `None`
112    /// 解码，支持滚动升级期间继续消费已发布消息。
113    #[serde(default, skip_serializing_if = "Option::is_none")]
114    pub recovery_operation_id: Option<String>,
115    /// 首步 execute 的业务输入；其余命令为空。
116    pub payload: Option<serde_json::Value>,
117    /// 原始业务正文；与兼容 JSON payload 互斥，字节与 schema 在重投中保持不变。
118    #[serde(default, skip_serializing_if = "Option::is_none")]
119    pub raw_payload: Option<nasaga_core::SagaPayload>,
120}
121
122/// 业务作用：参与方发回 Orchestrator 的结果 envelope。
123///
124/// 字段说明：`status` 是 attempt journal 词汇（四个 phase 终态的并集）；
125/// `terminal_status` 仅在取消屏障裁决 `ALREADY_TERMINAL` 时携带正向真实终态。
126#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
127pub struct SagaResultEnvelope {
128    /// 结果事件身份，进入 Orchestrator Inbox 去重。
129    pub event_id: String,
130    /// 实例身份。
131    pub saga_id: String,
132    /// 规范化租户。
133    pub tenant_id: String,
134    /// workflow 名称。
135    pub workflow: String,
136    /// definition 版本。
137    pub definition_version: u32,
138    /// definition 摘要。
139    pub definition_digest: String,
140    /// 步骤名称。
141    pub step: String,
142    /// 执行阶段稳定名。
143    pub phase: String,
144    /// 尝试序号。
145    pub attempt: u32,
146    /// 效果身份。
147    pub effect_id: String,
148    /// 命令身份。
149    pub command_id: String,
150    /// attempt 终态稳定名（SUCCEEDED/REJECTED/…/RESOLUTION_PENDING）。
151    pub status: String,
152    /// 取消裁决 `ALREADY_TERMINAL` 携带的正向真实终态稳定名。
153    pub terminal_status: Option<String>,
154    /// 稳定原因码；禁止携带敏感明细。
155    pub reason_code: Option<String>,
156}
157
158/// 业务作用：envelope 通过身份复验后的强类型视图，业务路径只消费本类型。
159#[derive(Debug, Clone)]
160pub struct VerifiedIdentity {
161    /// 实例身份。
162    pub saga_id: SagaId,
163    /// 租户身份。
164    pub tenant: TenantId,
165    /// workflow 名称。
166    pub workflow: WorkflowName,
167    /// definition 版本。
168    pub definition_version: DefinitionVersion,
169    /// definition 摘要。
170    pub definition_digest: String,
171    /// 步骤名称。
172    pub step: StepName,
173    /// 执行阶段。
174    pub phase: StepPhase,
175    /// 尝试序号。
176    pub attempt: AttemptNo,
177    /// 复验一致的效果身份。
178    pub effect: EffectId,
179    /// 复验一致的命令身份。
180    pub command: CommandId,
181}
182
183/// 业务作用：校验 envelope 的公共身份结构和确定性派生关系，在 Inbox claim 前拒绝身份字段不一致的命令。
184///
185/// 参数说明：
186/// - `saga_id`：命令声明的 Saga 实例身份。
187/// - `tenant_id`：命令声明的租户身份。
188/// - `workflow`：命令声明的工作流名。
189/// - `definition_version`：用于确定性身份派生的定义版本。
190/// - `definition_digest`：需为 64 字节小写十六进制的定义摘要。
191/// - `step`：命令声明的步骤名。
192/// - `phase`：需属于封闭集合的执行阶段。
193/// - `attempt`：用于命令身份派生的有效 attempt 序号。
194/// - `effect_id`：需要与本地派生结果精确比较的效果身份。
195/// - `command_id`：需要与本地派生结果精确比较的命令身份。
196///
197/// 返回：标识、阶段、attempt、摘要编码有效且重新派生的 effect/command 与声明一致时返回类型化身份，否则报受控原因。不会验证 transport 身份、摘要对应的真实定义或 producer 授权。
198#[allow(clippy::too_many_arguments)]
199fn verify_identity(
200    saga_id: &str,
201    tenant_id: &str,
202    workflow: &str,
203    definition_version: u32,
204    definition_digest: &str,
205    step: &str,
206    phase: &str,
207    attempt: u32,
208    effect_id: &str,
209    command_id: &str,
210) -> anyhow::Result<VerifiedIdentity> {
211    let saga_id = SagaId::new(saga_id).map_err(|v| anyhow::anyhow!("bad saga id: {}", v.code()))?;
212    let tenant =
213        TenantId::new(tenant_id).map_err(|v| anyhow::anyhow!("bad tenant: {}", v.code()))?;
214    let workflow =
215        WorkflowName::new(workflow).map_err(|v| anyhow::anyhow!("bad workflow: {}", v.code()))?;
216    let version = DefinitionVersion::new(definition_version)
217        .map_err(|v| anyhow::anyhow!("bad version: {}", v.code()))?;
218    let step = StepName::new(step).map_err(|v| anyhow::anyhow!("bad step: {}", v.code()))?;
219    let phase = StepPhase::parse(phase).ok_or_else(|| anyhow::anyhow!("bad phase"))?;
220    let attempt =
221        AttemptNo::new(attempt).map_err(|v| anyhow::anyhow!("bad attempt: {}", v.code()))?;
222    // definition digest 会参与实例合同复验；先限制为 SHA-256 小写十六进制，
223    // 防止大小写或非规范编码把同一摘要扩成多个传输身份。
224    if definition_digest.len() != 64
225        || !definition_digest
226            .bytes()
227            .all(|byte| byte.is_ascii_digit() || matches!(byte, b'a'..=b'f'))
228    {
229        anyhow::bail!("bad definition digest: expected 64 lowercase hex characters");
230    }
231    let effect = EffectId::derive(&saga_id, version, &step, phase);
232    let command = CommandId::derive(&effect, attempt);
233    // 身份复验是伪造与漂移的最后防线:自报 id 与确定性派生不一致的 envelope
234    // 不允许占用任何去重身份或触达业务 handler。
235    if effect.to_string() != effect_id || command.to_string() != command_id {
236        anyhow::bail!("envelope identity mismatch: derived ids differ from self-reported ids");
237    }
238    Ok(VerifiedIdentity {
239        saga_id,
240        tenant,
241        workflow,
242        definition_version: version,
243        definition_digest: definition_digest.to_string(),
244        step,
245        phase,
246        attempt,
247        effect,
248        command,
249    })
250}
251
252/// 业务作用：校验结果状态与 phase、event identity、terminal 附加字段之间的协议组合。
253///
254/// 参数说明：
255/// - `envelope`: 尚未进入 Inbox 的结果 envelope。
256/// - `identity`: 已完成公共身份派生复验的强类型视图。
257///
258/// 返回：组合属于封闭协议时返回 `Ok`；伪造 event id、跨 phase 状态或非法原因码返回错误，
259/// 调用方必须隔离消息且不得占用 Inbox 去重身份。
260fn verify_result_contract(
261    envelope: &SagaResultEnvelope,
262    identity: &VerifiedIdentity,
263) -> anyhow::Result<()> {
264    if envelope.event_id != derive_result_event_id(&envelope.command_id) {
265        anyhow::bail!("result event identity does not match command identity");
266    }
267    let status = envelope.parsed_status()?;
268    let phase_allows_status = match identity.phase {
269        StepPhase::Execute => matches!(
270            status,
271            StepAttemptStatus::Succeeded
272                | StepAttemptStatus::Rejected
273                | StepAttemptStatus::Unknown
274                | StepAttemptStatus::Halted
275        ),
276        StepPhase::Cancel => matches!(
277            status,
278            StepAttemptStatus::CancelConfirmed
279                | StepAttemptStatus::AlreadyTerminal
280                | StepAttemptStatus::ResolutionPending
281        ),
282        StepPhase::Compensate => matches!(
283            status,
284            StepAttemptStatus::Succeeded | StepAttemptStatus::Unknown | StepAttemptStatus::Halted
285        ),
286        StepPhase::Resolve => matches!(
287            status,
288            StepAttemptStatus::Succeeded
289                | StepAttemptStatus::Rejected
290                | StepAttemptStatus::Unknown
291                | StepAttemptStatus::Halted
292        ),
293    };
294    if !phase_allows_status {
295        anyhow::bail!("result status is not valid for its phase");
296    }
297
298    match (status, envelope.parsed_terminal()?) {
299        (
300            StepAttemptStatus::AlreadyTerminal,
301            Some(
302                StepForwardStatus::Succeeded
303                | StepForwardStatus::Rejected
304                | StepForwardStatus::Halted,
305            ),
306        ) => {}
307        (StepAttemptStatus::AlreadyTerminal, _) => {
308            anyhow::bail!("ALREADY_TERMINAL requires a settled terminal forward status")
309        }
310        (_, Some(_)) => anyhow::bail!("terminal_status is only valid for ALREADY_TERMINAL"),
311        (_, None) => {}
312    }
313
314    if let Some(code) = envelope.reason_code.as_deref() {
315        if !valid_reason_code(code) {
316            anyhow::bail!("reason_code is not a stable bounded identifier");
317        }
318    }
319    Ok(())
320}
321
322impl SagaCommandEnvelope {
323    /// 业务作用：合并兼容 JSON 与原始字节入口为唯一正文合同，拒绝双重正文来源。
324    ///
325    /// 参数说明: 无。
326    ///
327    /// 返回：原始正文或规范 JSON 字节；重复来源与非法正文返回类型化合同拒绝。
328    pub fn business_payload(&self) -> anyhow::Result<Option<nasaga_core::SagaPayload>> {
329        if self.payload.is_some() && self.raw_payload.is_some() {
330            return Err(crate::SagaCommandProcessingError::ContractInvalid.into());
331        }
332        let payload = self
333            .raw_payload
334            .clone()
335            .or_else(|| self.payload.clone().map(nasaga_core::SagaPayload::json));
336        if let Some(payload) = &payload {
337            payload
338                .validate()
339                .map_err(|_| crate::SagaCommandProcessingError::ContractInvalid)?;
340        }
341        Ok(payload)
342    }
343
344    /// 业务作用：在 Inbox 和业务解码前按本地 handler 声明复验媒体与 schema。
345    ///
346    /// 参数说明：
347    /// - `content_type`：本地编译期 descriptor 声明的精确媒体类型。
348    /// - `schema_id`：本地编译期 descriptor 声明的精确 schema 身份。
349    ///
350    /// 返回：business_payload 解析有效且正文缺省或媒体类型、schema ID 精确一致时成功，否则报错；不根据声明查找外部 schema 或验证其业务语义。
351    pub fn verify_payload_contract(
352        &self,
353        content_type: &str,
354        schema_id: &str,
355    ) -> anyhow::Result<()> {
356        if let Some(payload) = self.business_payload()? {
357            if payload.content_type != content_type || payload.schema_id != schema_id {
358                return Err(crate::SagaCommandProcessingError::ContractInvalid.into());
359            }
360        }
361        Ok(())
362    }
363
364    /// 业务作用：把已验证的正文转换为 handler 的类型化输入，二进制合同交付字节数组。
365    ///
366    /// 参数说明: 无。
367    ///
368    /// 返回：JSON 按原始文本直接解码，二进制保留全部字节；类型不匹配时返回合同拒绝。
369    pub fn decode_payload<T: serde::de::DeserializeOwned>(&self) -> anyhow::Result<T> {
370        let result = match self.business_payload()? {
371            None => serde_json::from_value(serde_json::Value::Null),
372            Some(payload) if payload.content_type == "application/json" => {
373                serde_json::from_slice(&payload.body)
374            }
375            Some(payload) => serde_json::from_value(serde_json::json!(payload.body)),
376        };
377        result.map_err(|_| crate::SagaCommandProcessingError::ContractInvalid.into())
378    }
379
380    /// 业务作用：解析并复验命令 envelope 的身份字段。
381    ///
382    /// 参数说明: 无。
383    ///
384    /// 返回：复验通过返回已证身份；否则返回错误，调用方走隔离路径而不是业务路径。
385    pub fn verified(&self) -> anyhow::Result<VerifiedIdentity> {
386        let identity = verify_identity(
387            &self.saga_id,
388            &self.tenant_id,
389            &self.workflow,
390            self.definition_version,
391            &self.definition_digest,
392            &self.step,
393            &self.phase,
394            self.attempt,
395            &self.effect_id,
396            &self.command_id,
397        )?;
398        if let Some(operation_id) = self.recovery_operation_id.as_deref() {
399            if operation_id.is_empty()
400                || operation_id.trim() != operation_id
401                || operation_id.len() > 190
402                || operation_id.chars().any(char::is_control)
403                || !matches!(identity.phase, StepPhase::Compensate | StepPhase::Resolve)
404            {
405                anyhow::bail!("recovery operation identity is invalid for the command phase");
406            }
407        }
408        Ok(identity)
409    }
410
411    /// 业务作用：在 Kafka 投递边界把命令身份错误转换成封闭分类。
412    ///
413    /// 参数说明: 无。
414    ///
415    /// 返回：身份合法时返回已证身份；否则只返回稳定的类型化分类，使 transport 不解析
416    /// 可能包含不可信字段的错误文本，也不让非法命令占用 Participant Inbox。
417    pub fn verified_for_delivery(&self) -> anyhow::Result<VerifiedIdentity> {
418        self.verified()
419            .map_err(|_| crate::SagaCommandProcessingError::IdentityInvalid.into())
420    }
421
422    /// 业务作用：把命令 envelope 序列化为 Outbox 事件，与触发它的状态迁移同事务发布。
423    ///
424    /// `event_id` 直接使用 `command_id`：同一 attempt 的传输重投必须复用同一投递身份，
425    /// dispatcher 重发不会产生新的去重身份。
426    ///
427    /// 参数说明: 无。
428    ///
429    /// 返回：可交给 `naoutbox` 的事件；序列化失败返回错误。
430    pub fn to_outbox_event(&self) -> anyhow::Result<OutboxEvent> {
431        // producer 也必须执行与 consumer 相同的身份复验；否则本地代码可以先把非法
432        // envelope 提交到 Outbox，直到远端消费时才形成永久重投/隔离毒消息。
433        self.verified()?;
434        Ok(OutboxEvent {
435            event_id: self.command_id.clone(),
436            aggregate_type: SAGA_AGGREGATE_TYPE.to_string(),
437            aggregate_id: self.saga_id.clone(),
438            event_type: COMMAND_EVENT_TYPE.to_string(),
439            payload: serde_json::to_vec(self)?,
440            traceparent: None,
441            // 租户归因取自已验真的 envelope 身份,与受信写入上下文同源;写入权威
442            // 仍是 append 时的 OutboxWriteContext。
443            tenant: self.tenant_id.clone(),
444        })
445    }
446}
447
448impl SagaResultEnvelope {
449    /// 业务作用：解析并复验结果 envelope 的身份字段。
450    ///
451    /// 参数说明: 无。
452    ///
453    /// 返回：复验通过返回已证身份；否则返回错误，调用方走隔离路径。
454    pub fn verified(&self) -> anyhow::Result<VerifiedIdentity> {
455        let identity = verify_identity(
456            &self.saga_id,
457            &self.tenant_id,
458            &self.workflow,
459            self.definition_version,
460            &self.definition_digest,
461            &self.step,
462            &self.phase,
463            self.attempt,
464            &self.effect_id,
465            &self.command_id,
466        )?;
467        verify_result_contract(self, &identity)?;
468        Ok(identity)
469    }
470
471    /// 业务作用：在 Kafka 投递边界把结果身份错误与合同错误转换成封闭分类。
472    ///
473    /// 参数说明: 无。
474    ///
475    /// 返回：身份和结果合同都合法时返回已证身份；否则只返回稳定的类型化分类，
476    /// 使 transport 无需解析可能含不可信字段的错误文本。
477    pub(crate) fn verified_for_delivery(&self) -> anyhow::Result<VerifiedIdentity> {
478        let identity = verify_identity(
479            &self.saga_id,
480            &self.tenant_id,
481            &self.workflow,
482            self.definition_version,
483            &self.definition_digest,
484            &self.step,
485            &self.phase,
486            self.attempt,
487            &self.effect_id,
488            &self.command_id,
489        )
490        .map_err(|_| crate::SagaResultProcessingError::IdentityInvalid)?;
491        verify_result_contract(self, &identity)
492            .map_err(|_| crate::SagaResultProcessingError::ContractInvalid)?;
493        Ok(identity)
494    }
495
496    /// 业务作用：解析结果状态列，收敛回 attempt journal 封闭词汇。
497    ///
498    /// 参数说明: 无。
499    ///
500    /// 返回：合法且非 `STARTED` 时返回状态；否则返回错误——结果必须是已裁决状态。
501    pub fn parsed_status(&self) -> anyhow::Result<StepAttemptStatus> {
502        let status = StepAttemptStatus::parse(&self.status)
503            .ok_or_else(|| anyhow::anyhow!("unknown result status"))?;
504        if status.is_in_flight() {
505            anyhow::bail!("result status must be settled");
506        }
507        Ok(status)
508    }
509
510    /// 业务作用：解析取消裁决携带的正向真实终态。
511    ///
512    /// 参数说明: 无。
513    ///
514    /// 返回：`ALREADY_TERMINAL` 结果返回其终态；其它结果返回 `None`；文本非法返回错误。
515    pub fn parsed_terminal(&self) -> anyhow::Result<Option<StepForwardStatus>> {
516        self.terminal_status
517            .as_deref()
518            .map(|raw| {
519                StepForwardStatus::parse(raw)
520                    .ok_or_else(|| anyhow::anyhow!("unknown terminal status"))
521            })
522            .transpose()
523    }
524
525    /// 业务作用：把结果 envelope 序列化为 Outbox 事件（参与方侧使用）。
526    ///
527    /// 参数说明: 无。
528    ///
529    /// 返回：可交给 `naoutbox` 的事件；序列化失败返回错误。
530    pub fn to_outbox_event(&self) -> anyhow::Result<OutboxEvent> {
531        // reason_code、phase/status 和确定性 event id 在发布前即封闭，避免生产者制造
532        // 自己的消费者必然拒绝、且已经可靠落入 Outbox 的毒消息。
533        self.verified()?;
534        Ok(OutboxEvent {
535            event_id: self.event_id.clone(),
536            aggregate_type: SAGA_AGGREGATE_TYPE.to_string(),
537            aggregate_id: self.saga_id.clone(),
538            event_type: RESULT_EVENT_TYPE.to_string(),
539            payload: serde_json::to_vec(self)?,
540            traceparent: None,
541            tenant: self.tenant_id.clone(),
542        })
543    }
544}
nasaga_runtime_core/envelope.rs

nasaga_runtime_core/
envelope.rs
