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 公共身份字段，把"自报身份"升级为"已证身份"。
184///
185/// 关键步骤：effect/command 由 `(saga, version, step, phase, attempt)` **重新派生**后与
186/// 自报值比对——不一致说明伪造、损坏或派生漂移，绝不进入 Inbox claim 或业务 handler，
187/// 否则攻击者可用自报 id 预先污染合法命令的去重身份。
188///
189/// 参数说明：
190/// - `saga_id`/`tenant_id`/`workflow`/`definition_version`/`definition_digest`/`step`/
191///   `phase`/`attempt`/`effect_id`/`command_id`: envelope 自报的身份字段。
192///
193/// 返回：全部字段合法且派生一致时返回已证身份；否则返回携带稳定原因的错误。
194#[allow(clippy::too_many_arguments)]
195fn verify_identity(
196    saga_id: &str,
197    tenant_id: &str,
198    workflow: &str,
199    definition_version: u32,
200    definition_digest: &str,
201    step: &str,
202    phase: &str,
203    attempt: u32,
204    effect_id: &str,
205    command_id: &str,
206) -> anyhow::Result<VerifiedIdentity> {
207    let saga_id = SagaId::new(saga_id).map_err(|v| anyhow::anyhow!("bad saga id: {}", v.code()))?;
208    let tenant =
209        TenantId::new(tenant_id).map_err(|v| anyhow::anyhow!("bad tenant: {}", v.code()))?;
210    let workflow =
211        WorkflowName::new(workflow).map_err(|v| anyhow::anyhow!("bad workflow: {}", v.code()))?;
212    let version = DefinitionVersion::new(definition_version)
213        .map_err(|v| anyhow::anyhow!("bad version: {}", v.code()))?;
214    let step = StepName::new(step).map_err(|v| anyhow::anyhow!("bad step: {}", v.code()))?;
215    let phase = StepPhase::parse(phase).ok_or_else(|| anyhow::anyhow!("bad phase"))?;
216    let attempt =
217        AttemptNo::new(attempt).map_err(|v| anyhow::anyhow!("bad attempt: {}", v.code()))?;
218    // definition digest 会参与实例合同复验；先限制为 SHA-256 小写十六进制，
219    // 防止大小写或非规范编码把同一摘要扩成多个传输身份。
220    if definition_digest.len() != 64
221        || !definition_digest
222            .bytes()
223            .all(|byte| byte.is_ascii_digit() || matches!(byte, b'a'..=b'f'))
224    {
225        anyhow::bail!("bad definition digest: expected 64 lowercase hex characters");
226    }
227    let effect = EffectId::derive(&saga_id, version, &step, phase);
228    let command = CommandId::derive(&effect, attempt);
229    // 身份复验是伪造与漂移的最后防线:自报 id 与确定性派生不一致的 envelope
230    // 不允许占用任何去重身份或触达业务 handler。
231    if effect.to_string() != effect_id || command.to_string() != command_id {
232        anyhow::bail!("envelope identity mismatch: derived ids differ from self-reported ids");
233    }
234    Ok(VerifiedIdentity {
235        saga_id,
236        tenant,
237        workflow,
238        definition_version: version,
239        definition_digest: definition_digest.to_string(),
240        step,
241        phase,
242        attempt,
243        effect,
244        command,
245    })
246}
247
248/// 业务作用：校验结果状态与 phase、event identity、terminal 附加字段之间的协议组合。
249///
250/// 参数说明：
251/// - `envelope`: 尚未进入 Inbox 的结果 envelope。
252/// - `identity`: 已完成公共身份派生复验的强类型视图。
253///
254/// 返回：组合属于封闭协议时返回 `Ok`；伪造 event id、跨 phase 状态或非法原因码返回错误，
255/// 调用方必须隔离消息且不得占用 Inbox 去重身份。
256fn verify_result_contract(
257    envelope: &SagaResultEnvelope,
258    identity: &VerifiedIdentity,
259) -> anyhow::Result<()> {
260    if envelope.event_id != derive_result_event_id(&envelope.command_id) {
261        anyhow::bail!("result event identity does not match command identity");
262    }
263    let status = envelope.parsed_status()?;
264    let phase_allows_status = match identity.phase {
265        StepPhase::Execute => matches!(
266            status,
267            StepAttemptStatus::Succeeded
268                | StepAttemptStatus::Rejected
269                | StepAttemptStatus::Unknown
270                | StepAttemptStatus::Halted
271        ),
272        StepPhase::Cancel => matches!(
273            status,
274            StepAttemptStatus::CancelConfirmed
275                | StepAttemptStatus::AlreadyTerminal
276                | StepAttemptStatus::ResolutionPending
277        ),
278        StepPhase::Compensate => matches!(
279            status,
280            StepAttemptStatus::Succeeded | StepAttemptStatus::Unknown | StepAttemptStatus::Halted
281        ),
282        StepPhase::Resolve => matches!(
283            status,
284            StepAttemptStatus::Succeeded
285                | StepAttemptStatus::Rejected
286                | StepAttemptStatus::Unknown
287                | StepAttemptStatus::Halted
288        ),
289    };
290    if !phase_allows_status {
291        anyhow::bail!("result status is not valid for its phase");
292    }
293
294    match (status, envelope.parsed_terminal()?) {
295        (
296            StepAttemptStatus::AlreadyTerminal,
297            Some(
298                StepForwardStatus::Succeeded
299                | StepForwardStatus::Rejected
300                | StepForwardStatus::Halted,
301            ),
302        ) => {}
303        (StepAttemptStatus::AlreadyTerminal, _) => {
304            anyhow::bail!("ALREADY_TERMINAL requires a settled terminal forward status")
305        }
306        (_, Some(_)) => anyhow::bail!("terminal_status is only valid for ALREADY_TERMINAL"),
307        (_, None) => {}
308    }
309
310    if let Some(code) = envelope.reason_code.as_deref() {
311        if !valid_reason_code(code) {
312            anyhow::bail!("reason_code is not a stable bounded identifier");
313        }
314    }
315    Ok(())
316}
317
318impl SagaCommandEnvelope {
319    /// 业务作用：合并兼容 JSON 与原始字节入口为唯一正文合同，拒绝双重正文来源。
320    ///
321    /// 参数说明: 无。
322    ///
323    /// 返回：原始正文或规范 JSON 字节；重复来源与非法正文返回类型化合同拒绝。
324    pub fn business_payload(&self) -> anyhow::Result<Option<nasaga_core::SagaPayload>> {
325        if self.payload.is_some() && self.raw_payload.is_some() {
326            return Err(crate::SagaCommandProcessingError::ContractInvalid.into());
327        }
328        let payload = self
329            .raw_payload
330            .clone()
331            .or_else(|| self.payload.clone().map(nasaga_core::SagaPayload::json));
332        if let Some(payload) = &payload {
333            payload
334                .validate()
335                .map_err(|_| crate::SagaCommandProcessingError::ContractInvalid)?;
336        }
337        Ok(payload)
338    }
339
340    /// 业务作用：在 Inbox 和业务解码前按本地 handler 声明复验媒体与 schema。
341    ///
342    /// 参数说明：content_type 与 schema_id 来自编译期 descriptor，不来自请求 header。
343    ///
344    /// 返回：无正文或合同完全一致时成功；任一漂移返回确定性合同拒绝。
345    pub fn verify_payload_contract(
346        &self,
347        content_type: &str,
348        schema_id: &str,
349    ) -> anyhow::Result<()> {
350        if let Some(payload) = self.business_payload()? {
351            if payload.content_type != content_type || payload.schema_id != schema_id {
352                return Err(crate::SagaCommandProcessingError::ContractInvalid.into());
353            }
354        }
355        Ok(())
356    }
357
358    /// 业务作用：把已验证的正文转换为 handler 的类型化输入，二进制合同交付字节数组。
359    ///
360    /// 参数说明: 无。
361    ///
362    /// 返回：JSON 按原始文本直接解码，二进制保留全部字节；类型不匹配时返回合同拒绝。
363    pub fn decode_payload<T: serde::de::DeserializeOwned>(&self) -> anyhow::Result<T> {
364        let result = match self.business_payload()? {
365            None => serde_json::from_value(serde_json::Value::Null),
366            Some(payload) if payload.content_type == "application/json" => {
367                serde_json::from_slice(&payload.body)
368            }
369            Some(payload) => serde_json::from_value(serde_json::json!(payload.body)),
370        };
371        result.map_err(|_| crate::SagaCommandProcessingError::ContractInvalid.into())
372    }
373
374    /// 业务作用：解析并复验命令 envelope 的身份字段。
375    ///
376    /// 参数说明: 无。
377    ///
378    /// 返回：复验通过返回已证身份；否则返回错误，调用方走隔离路径而不是业务路径。
379    pub fn verified(&self) -> anyhow::Result<VerifiedIdentity> {
380        let identity = verify_identity(
381            &self.saga_id,
382            &self.tenant_id,
383            &self.workflow,
384            self.definition_version,
385            &self.definition_digest,
386            &self.step,
387            &self.phase,
388            self.attempt,
389            &self.effect_id,
390            &self.command_id,
391        )?;
392        if let Some(operation_id) = self.recovery_operation_id.as_deref() {
393            if operation_id.is_empty()
394                || operation_id.trim() != operation_id
395                || operation_id.len() > 190
396                || operation_id.chars().any(char::is_control)
397                || !matches!(identity.phase, StepPhase::Compensate | StepPhase::Resolve)
398            {
399                anyhow::bail!("recovery operation identity is invalid for the command phase");
400            }
401        }
402        Ok(identity)
403    }
404
405    /// 业务作用：在 Kafka 投递边界把命令身份错误转换成封闭分类。
406    ///
407    /// 参数说明: 无。
408    ///
409    /// 返回：身份合法时返回已证身份；否则只返回稳定的类型化分类，使 transport 不解析
410    /// 可能包含不可信字段的错误文本，也不让非法命令占用 Participant Inbox。
411    pub fn verified_for_delivery(&self) -> anyhow::Result<VerifiedIdentity> {
412        self.verified()
413            .map_err(|_| crate::SagaCommandProcessingError::IdentityInvalid.into())
414    }
415
416    /// 业务作用：把命令 envelope 序列化为 Outbox 事件，与触发它的状态迁移同事务发布。
417    ///
418    /// `event_id` 直接使用 `command_id`：同一 attempt 的传输重投必须复用同一投递身份，
419    /// dispatcher 重发不会产生新的去重身份。
420    ///
421    /// 参数说明: 无。
422    ///
423    /// 返回：可交给 `naoutbox` 的事件；序列化失败返回错误。
424    pub fn to_outbox_event(&self) -> anyhow::Result<OutboxEvent> {
425        // producer 也必须执行与 consumer 相同的身份复验；否则本地代码可以先把非法
426        // envelope 提交到 Outbox，直到远端消费时才形成永久重投/隔离毒消息。
427        self.verified()?;
428        Ok(OutboxEvent {
429            event_id: self.command_id.clone(),
430            aggregate_type: SAGA_AGGREGATE_TYPE.to_string(),
431            aggregate_id: self.saga_id.clone(),
432            event_type: COMMAND_EVENT_TYPE.to_string(),
433            payload: serde_json::to_vec(self)?,
434            traceparent: None,
435            // 租户归因取自已验真的 envelope 身份,与受信写入上下文同源;写入权威
436            // 仍是 append 时的 OutboxWriteContext。
437            tenant: self.tenant_id.clone(),
438        })
439    }
440}
441
442impl SagaResultEnvelope {
443    /// 业务作用：解析并复验结果 envelope 的身份字段。
444    ///
445    /// 参数说明: 无。
446    ///
447    /// 返回：复验通过返回已证身份；否则返回错误，调用方走隔离路径。
448    pub fn verified(&self) -> anyhow::Result<VerifiedIdentity> {
449        let identity = verify_identity(
450            &self.saga_id,
451            &self.tenant_id,
452            &self.workflow,
453            self.definition_version,
454            &self.definition_digest,
455            &self.step,
456            &self.phase,
457            self.attempt,
458            &self.effect_id,
459            &self.command_id,
460        )?;
461        verify_result_contract(self, &identity)?;
462        Ok(identity)
463    }
464
465    /// 业务作用：在 Kafka 投递边界把结果身份错误与合同错误转换成封闭分类。
466    ///
467    /// 参数说明: 无。
468    ///
469    /// 返回：身份和结果合同都合法时返回已证身份；否则只返回稳定的类型化分类，
470    /// 使 transport 无需解析可能含不可信字段的错误文本。
471    pub(crate) fn verified_for_delivery(&self) -> anyhow::Result<VerifiedIdentity> {
472        let identity = verify_identity(
473            &self.saga_id,
474            &self.tenant_id,
475            &self.workflow,
476            self.definition_version,
477            &self.definition_digest,
478            &self.step,
479            &self.phase,
480            self.attempt,
481            &self.effect_id,
482            &self.command_id,
483        )
484        .map_err(|_| crate::SagaResultProcessingError::IdentityInvalid)?;
485        verify_result_contract(self, &identity)
486            .map_err(|_| crate::SagaResultProcessingError::ContractInvalid)?;
487        Ok(identity)
488    }
489
490    /// 业务作用：解析结果状态列，收敛回 attempt journal 封闭词汇。
491    ///
492    /// 参数说明: 无。
493    ///
494    /// 返回：合法且非 `STARTED` 时返回状态；否则返回错误——结果必须是已裁决状态。
495    pub fn parsed_status(&self) -> anyhow::Result<StepAttemptStatus> {
496        let status = StepAttemptStatus::parse(&self.status)
497            .ok_or_else(|| anyhow::anyhow!("unknown result status"))?;
498        if status.is_in_flight() {
499            anyhow::bail!("result status must be settled");
500        }
501        Ok(status)
502    }
503
504    /// 业务作用：解析取消裁决携带的正向真实终态。
505    ///
506    /// 参数说明: 无。
507    ///
508    /// 返回：`ALREADY_TERMINAL` 结果返回其终态；其它结果返回 `None`；文本非法返回错误。
509    pub fn parsed_terminal(&self) -> anyhow::Result<Option<StepForwardStatus>> {
510        self.terminal_status
511            .as_deref()
512            .map(|raw| {
513                StepForwardStatus::parse(raw)
514                    .ok_or_else(|| anyhow::anyhow!("unknown terminal status"))
515            })
516            .transpose()
517    }
518
519    /// 业务作用：把结果 envelope 序列化为 Outbox 事件（参与方侧使用）。
520    ///
521    /// 参数说明: 无。
522    ///
523    /// 返回：可交给 `naoutbox` 的事件；序列化失败返回错误。
524    pub fn to_outbox_event(&self) -> anyhow::Result<OutboxEvent> {
525        // reason_code、phase/status 和确定性 event id 在发布前即封闭，避免生产者制造
526        // 自己的消费者必然拒绝、且已经可靠落入 Outbox 的毒消息。
527        self.verified()?;
528        Ok(OutboxEvent {
529            event_id: self.event_id.clone(),
530            aggregate_type: SAGA_AGGREGATE_TYPE.to_string(),
531            aggregate_id: self.saga_id.clone(),
532            event_type: RESULT_EVENT_TYPE.to_string(),
533            payload: serde_json::to_vec(self)?,
534            traceparent: None,
535            tenant: self.tenant_id.clone(),
536        })
537    }
538}
nasaga_runtime_core/envelope.rs

nasaga_runtime_core/
envelope.rs
