nasaga_runtime_core/
lib.rs

1//! Saga backend 中立状态机与 wrapper 共享的进程级运行状态。
2//!
3//! Orchestrator、Participant、timer、补偿、恢复与 fencing 裁决只在本 crate
4//! 保留一份，MySQL 与 PostgreSQL wrapper 只注入 [`nasaga_backend::SagaBackend`]。
5//! 数据库事实由各 backend 读取；transport、配额与管理动作指标全进程只计一次。
6//!
7//! [`Orchestrator::handle_authenticated_result_authorized_traced`] 让宿主把冻结的结果资格带入状态事务。
8//! 同一次资格在异步恢复、实例锁后及事务交还前持续复验；失权回滚完整事务并以
9//! [`SagaResultProcessingError::AuthorityUnavailable`] 保留原事件重投。宿主必须用不可复用的安全
10//! 发布代际识别 A→B→A，不能只比较当前材料摘要。COMMIT 已发出后的结局仍按 backend 收据裁决。
11
12#![forbid(unsafe_code)]
13#![deny(missing_docs)]
14
15pub use nasaga_core::{SagaPayload, SagaPayloadContract, SagaPayloadError};
16mod latency;
17pub use latency::{
18    render_saga_latency_metrics, saga_latency_snapshot, SagaLatencySnapshot, SAGA_LATENCY_BUCKETS,
19};
20mod catalog;
21mod envelope;
22mod management;
23mod observability;
24mod orchestrator;
25mod participant;
26mod registry;
27mod scheduled;
28mod timers;
29mod transaction;
30#[cfg(feature = "kafka")]
31mod transport;
32mod transport_auth;
33#[cfg(feature = "grpc-transport")]
34mod transport_grpc;
35#[cfg(feature = "redis-stream")]
36mod transport_redis;
37#[cfg(any(
38    feature = "kafka",
39    feature = "redis-stream",
40    feature = "grpc-transport"
41))]
42mod transport_shared;
43
44pub use catalog::{
45    select_capability_route, select_capability_routes, validate_capability_route_contract,
46    validate_redis_stream_route, CapabilityDescriptor, CapabilityReceipt, DefinitionActivationGate,
47    DefinitionArtifact, DefinitionCancelMode, DefinitionCatalogError, DefinitionCompensation,
48    DefinitionLifecycle, DefinitionLifecycleOperation, DefinitionPublishDisposition,
49    DefinitionRecord, DefinitionResolution, DefinitionResolutionMode, DefinitionStepArtifact,
50    DefinitionTimeoutPolicy, DynamicCatalogSnapshot, RegisteredCapability,
51    CATALOG_OBJECT_KEY_MAX_LEN,
52};
53pub use envelope::{
54    canonical_bytes, derive_result_event_id, SagaCommandEnvelope, SagaResultEnvelope,
55    VerifiedIdentity, COMMAND_EVENT_TYPE, RESULT_EVENT_TYPE, SAGA_AGGREGATE_TYPE,
56};
57pub use management::{
58    SagaAuditPage, SagaAuditPageCursor, SagaAuditRecord, SagaAuditTrail, SagaManagementContext,
59    SagaManagementError, SagaManagementExpectation, SagaManagementPermission,
60};
61pub use observability::SagaOperationalMetrics;
62pub use orchestrator::{
63    HandleOutcome, Orchestrator, OrchestratorConfig, SagaConcurrencyError, StartOutcome,
64    StartSagaError, StartSagaRequest, TenantActionRate, TimerOutcome,
65};
66pub use participant::{
67    AuthenticatedParticipantRuntime, ParticipantCommandTrust, ParticipantHandled,
68    ParticipantRuntime, SagaCommandService,
69};
70pub use registry::DefinitionRegistry;
71pub use scheduled::{
72    derive_scheduled_business_key, ScheduledBatchReport, ScheduledBatchSpec, ScheduledItem,
73};
74pub use timers::{
75    derive_timer_id, KIND_CANCEL_TIMEOUT, KIND_COMPENSATE_TIMEOUT,
76    KIND_COMPENSATION_RESOLUTION_BUDGET, KIND_FORWARD_RESOLUTION_BUDGET, KIND_INSTANCE_DEADLINE,
77    KIND_RESOLUTION_BUDGET, KIND_RESOLVE_TIMEOUT, KIND_STEP_TIMEOUT,
78};
79pub use transaction::SagaTransactionError;
80#[cfg(feature = "kafka")]
81pub use transport::{
82    SagaCommandRoute, SagaKafkaCommandConsumer, SagaKafkaCommandConsumerConfig,
83    SagaKafkaResultConsumer, SagaKafkaResultConsumerConfig,
84};
85pub use transport_auth::{
86    render_saga_http_command_dlt_metric, SagaHttpCredentialSource, SagaHttpCredentials,
87    SagaHttpMessageAuthError, SagaHttpMessageAuthFailure, SagaHttpMessageAuthenticator,
88    SagaHttpReplayGuard, SagaHttpReplayMetricAggregate, SagaHttpReplayMetrics, SagaHttpReplayPlane,
89    SagaHttpSignedMessage,
90};
91#[cfg(feature = "grpc-transport")]
92pub use transport_grpc::{
93    orchestrator_proto, outbox_disposition_of, proto as grpc_proto, SagaGrpcBindingError,
94    SagaGrpcCommandServer, SagaGrpcCommandTransportService, SagaGrpcPeerBinding,
95    SagaGrpcPeerIdentity, SagaGrpcReceipt, SagaGrpcResultServer, SagaGrpcResultTransportService,
96};
97#[cfg(feature = "redis-stream")]
98pub use transport_redis::{
99    publisher_duplicate_hints_total, safe_trim_by_group_frontier, stream_group_backlog,
100    verify_stream_transport_ready, SagaRedisStreamCommandConsumer, SagaRedisStreamPublisher,
101    SagaRedisStreamResultConsumer, SagaStreamAuth, SagaStreamConsumerConfig, SagaStreamPoller,
102    SagaStreamVerificationKey, StreamPollReport,
103};
104#[cfg(any(
105    feature = "kafka",
106    feature = "redis-stream",
107    feature = "grpc-transport"
108))]
109pub use transport_shared::{ParticipantCommandHandler, SagaCommandHandler, SagaResultHandler};
110
111use nasaga_core::{CancelMode, Compensation, ResolutionMode};
112
113/// 业务作用：用封闭类型标记 Saga command 的不可恢复投递错误。
114#[derive(Debug, Clone, Copy, PartialEq, Eq)]
115#[non_exhaustive]
116pub enum SagaCommandProcessingError {
117    /// transport 认证出的 producer 不在参与方信任投影中。
118    ProducerUnauthorized,
119    /// envelope 身份、派生身份或编码不合法。
120    IdentityInvalid,
121    /// command route 无权承载目标 workflow 步骤。
122    RouteUnauthorized,
123    /// phase、取消能力或 payload 违反已发布步骤合同。
124    ContractInvalid,
125}
126
127impl SagaCommandProcessingError {
128    /// 业务作用：返回可安全写入 DLT header 的稳定原因码。
129    ///
130    /// 参数说明: 无。
131    ///
132    /// 返回：不含 envelope、payload 或底层错误的低基数原因码。
133    pub const fn dead_letter_reason(self) -> &'static str {
134        match self {
135            Self::ProducerUnauthorized => "saga_command_producer_unauthorized",
136            Self::IdentityInvalid => "saga_command_identity_invalid",
137            Self::RouteUnauthorized => "saga_command_route_unauthorized",
138            Self::ContractInvalid => "saga_command_contract_invalid",
139        }
140    }
141
142    /// 业务作用：把未受信原因码收窄为封闭 command 错误。
143    ///
144    /// 参数说明：
145    /// - `reason`：是 connector 收到的候选原因码。
146    ///
147    /// 返回：精确命中白名单时返回对应类型；未知文本返回 `None`。
148    pub fn from_dead_letter_reason(reason: &str) -> Option<Self> {
149        match reason {
150            "saga_command_producer_unauthorized" => Some(Self::ProducerUnauthorized),
151            "saga_command_identity_invalid" => Some(Self::IdentityInvalid),
152            "saga_command_route_unauthorized" => Some(Self::RouteUnauthorized),
153            "saga_command_contract_invalid" => Some(Self::ContractInvalid),
154            _ => None,
155        }
156    }
157}
158
159impl std::fmt::Display for SagaCommandProcessingError {
160    /// 业务作用：输出稳定、脱敏的 command 错误分类。
161    ///
162    /// 参数说明：
163    /// - `formatter`：是标准格式化目标。
164    ///
165    /// 返回：分类文本写入成功返回 `Ok`；格式化失败返回对应错误。
166    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
167        formatter.write_str(match self {
168            Self::ProducerUnauthorized => "saga command producer is unauthorized",
169            Self::IdentityInvalid => "saga command identity is invalid",
170            Self::RouteUnauthorized => "saga command route is unauthorized",
171            Self::ContractInvalid => "saga command contract is invalid",
172        })
173    }
174}
175
176impl std::error::Error for SagaCommandProcessingError {}
177
178/// command DLT 接收端可信的全部稳定原因码。
179pub const SAGA_COMMAND_DEAD_LETTER_REASONS: &[&str] = &[
180    "saga_command_producer_unauthorized",
181    "saga_command_identity_invalid",
182    "saga_command_route_unauthorized",
183    "saga_command_contract_invalid",
184];
185
186/// 业务作用：用封闭类型标记 Saga result 的可信投递分类。
187#[derive(Debug, Clone, Copy, PartialEq, Eq)]
188#[non_exhaustive]
189pub enum SagaResultProcessingError {
190    /// 实例处于 `PAUSED`，恢复后必须继续处理同一结果。
191    Paused,
192    /// 本次结果操作的运行资格已经撤销或到期，须保留原事件等待重新准入。
193    AuthorityUnavailable,
194    /// transport 认证出的 producer 无权为目标步骤作证。
195    ProducerUnauthorized,
196    /// envelope 自报身份、派生身份或编码不合法。
197    IdentityInvalid,
198    /// envelope 与固定实例和 definition 合同不一致。
199    ContractInvalid,
200}
201
202impl SagaResultProcessingError {
203    /// 业务作用：返回可进入 DLT 的稳定 result 原因码。
204    ///
205    /// 参数说明: 无。
206    ///
207    /// 返回：暂停或失权返回 `None`；确定性协议错误返回脱敏原因码。
208    pub const fn dead_letter_reason(self) -> Option<&'static str> {
209        match self {
210            Self::Paused | Self::AuthorityUnavailable => None,
211            Self::ProducerUnauthorized => Some("saga_result_producer_unauthorized"),
212            Self::IdentityInvalid => Some("saga_result_identity_invalid"),
213            Self::ContractInvalid => Some("saga_result_contract_invalid"),
214        }
215    }
216}
217
218impl std::fmt::Display for SagaResultProcessingError {
219    /// 业务作用：输出稳定、脱敏的 result 错误分类。
220    ///
221    /// 参数说明：
222    /// - `formatter`：是标准格式化目标。
223    ///
224    /// 返回：分类文本写入成功返回 `Ok`；格式化失败返回对应错误。
225    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
226        formatter.write_str(match self {
227            Self::Paused => "saga result processing deferred while instance is paused",
228            Self::AuthorityUnavailable => "saga result processing authority is unavailable",
229            Self::ProducerUnauthorized => "saga result producer is unauthorized",
230            Self::IdentityInvalid => "saga result identity is invalid",
231            Self::ContractInvalid => "saga result contract is invalid",
232        })
233    }
234}
235
236impl std::error::Error for SagaResultProcessingError {}
237
238/// 业务作用：把 Participant command 失败收敛为 transport 可执行的保留、持续收敛或隔离动作。
239#[derive(Debug, Clone, Copy, PartialEq, Eq)]
240pub enum CommandDeliveryDisposition {
241    /// 普通瞬态失败，保留消息并消耗有界重试预算。
242    Retry,
243    /// 提交或回滚结果不确定，保留消息且不消耗隔离预算。
244    Defer,
245    /// 来源、身份或合同确定性非法，允许进入耐久 DLT。
246    DeadLetter,
247}
248
249/// 业务作用：限制同一 command 的普通瞬态重试次数，不影响结果不确定分支的持续收敛。
250#[derive(Debug, Clone, Copy, PartialEq, Eq)]
251pub struct CommandDeliveryPolicy {
252    /// 普通失败可重投的最大序号。
253    pub max_retries: u32,
254}
255
256impl Default for CommandDeliveryPolicy {
257    /// 业务作用：提供有界的默认 command 重投预算。
258    ///
259    /// 参数说明: 无。
260    ///
261    /// 返回：最多允许十六次普通失败重投的策略。
262    fn default() -> Self {
263        Self {
264            max_retries: 16,
265        }
266    }
267}
268
269impl CommandDeliveryPolicy {
270    /// 业务作用：结合失败类别与本次普通重试序号裁决 command 投递动作。
271    ///
272    /// 参数说明：
273    /// - `error`：是完整错误链
274    /// - `retry_attempt`：是从一开始的普通失败序号。
275    ///
276    /// 返回：确定性合同错误立即隔离，普通失败超预算后隔离，关键事务阶段持续保留。
277    pub fn decide(self, error: &anyhow::Error, retry_attempt: u32) -> CommandDeliveryDisposition {
278        let disposition = classify_command_delivery_error(error);
279        if disposition == CommandDeliveryDisposition::Retry && retry_attempt > self.max_retries {
280            CommandDeliveryDisposition::DeadLetter
281        } else {
282            disposition
283        }
284    }
285}
286
287/// 业务作用：从 command 错误链提取可安全跨 transport 传播的封闭 DLT 原因码。
288///
289/// 参数说明：
290/// - `error`：是 Participant 或入口门禁返回的完整错误链。
291///
292/// 返回：命中类型化确定性错误时返回低基数原因码；其它失败返回 `None`。
293pub fn command_dead_letter_reason(error: &anyhow::Error) -> Option<&'static str> {
294    error.chain().find_map(|cause| {
295        cause
296            .downcast_ref::<SagaCommandProcessingError>()
297            .map(|error| error.dead_letter_reason())
298    })
299}
300
301/// 业务作用：为 hosted command consumer 形成不依赖错误文本的重试与隔离分类。
302///
303/// 参数说明：
304/// - `error`：是 transport 身份门禁或 Participant 事务错误。
305///
306/// 返回：合同错误隔离，事务关键阶段持续保留，其它失败有界重试。
307pub fn classify_command_delivery_error(error: &anyhow::Error) -> CommandDeliveryDisposition {
308    if command_dead_letter_reason(error).is_some() {
309        CommandDeliveryDisposition::DeadLetter
310    } else if error.chain().any(|cause| {
311        cause
312            .downcast_ref::<SagaTransactionError>()
313            .is_some_and(|error| error.requires_unbounded_redelivery())
314    }) {
315        CommandDeliveryDisposition::Defer
316    } else {
317        CommandDeliveryDisposition::Retry
318    }
319}
320
321/// 业务作用：把 Orchestrator 结果处理失败收敛为 transport 可执行动作。
322#[derive(Debug, Clone, Copy, PartialEq, Eq)]
323pub enum ResultDeliveryDisposition {
324    /// 普通瞬态失败，保留消息并消耗有界预算。
325    Retry,
326    /// 暂停或事务结果不确定，保留消息且不消耗隔离预算。
327    Defer,
328    /// 确定性协议错误，允许保留原文后进入 DLT。
329    DeadLetter,
330}
331
332/// 业务作用：限制 result 普通瞬态重试次数，同时保留暂停与事务不确定的无限收敛语义。
333#[derive(Debug, Clone, Copy, PartialEq, Eq)]
334pub struct ResultDeliveryPolicy {
335    /// 普通失败可重投的最大序号。
336    pub max_retries: u32,
337}
338
339impl Default for ResultDeliveryPolicy {
340    /// 业务作用：提供有界的默认 result 重投预算。
341    ///
342    /// 参数说明: 无。
343    ///
344    /// 返回：最多允许十六次普通失败重投的策略。
345    fn default() -> Self {
346        Self {
347            max_retries: 16,
348        }
349    }
350}
351
352impl ResultDeliveryPolicy {
353    /// 业务作用：结合失败类别与普通失败序号裁决 result 投递动作。
354    ///
355    /// 参数说明：
356    /// - `error`：是完整错误链
357    /// - `retry_attempt`：是从一开始的普通失败序号。
358    ///
359    /// 返回：暂停、失权和事务不确定持续保留，普通失败超预算后隔离。
360    pub fn decide(self, error: &anyhow::Error, retry_attempt: u32) -> ResultDeliveryDisposition {
361        let disposition = classify_result_delivery_error(error);
362        if disposition == ResultDeliveryDisposition::Retry && retry_attempt > self.max_retries {
363            ResultDeliveryDisposition::DeadLetter
364        } else {
365            disposition
366        }
367    }
368}
369
370/// 业务作用：为 hosted result consumer 形成不依赖错误文本的重试、延后与隔离分类。
371///
372/// 参数说明：
373/// - `error`：是 Orchestrator 结果处理返回的完整错误链。
374///
375/// 返回：暂停、失权与事务关键阶段延后，确定性协议错误隔离，其它失败有界重试。
376pub fn classify_result_delivery_error(error: &anyhow::Error) -> ResultDeliveryDisposition {
377    if error.chain().any(|cause| {
378        cause
379            .downcast_ref::<SagaTransactionError>()
380            .is_some_and(|error| error.requires_unbounded_redelivery())
381    }) {
382        return ResultDeliveryDisposition::Defer;
383    }
384    let classified = error
385        .chain()
386        .find_map(|cause| cause.downcast_ref::<SagaResultProcessingError>().copied());
387    if matches!(
388        classified,
389        Some(SagaResultProcessingError::Paused | SagaResultProcessingError::AuthorityUnavailable)
390    ) {
391        ResultDeliveryDisposition::Defer
392    } else if classified.is_some_and(|error| error.dead_letter_reason().is_some()) {
393        ResultDeliveryDisposition::DeadLetter
394    } else {
395        ResultDeliveryDisposition::Retry
396    }
397}
398
399/// 业务作用：`#[saga]` 为每个本地步骤生成的静态合同投影。
400#[derive(Debug, Clone, Copy)]
401pub struct SagaStepDescriptor {
402    /// 本 handler 接受的规范媒体类型。
403    pub payload_content_type: &'static str,
404    /// 本 handler 支持的稳定 schema 身份。
405    pub payload_schema_id: &'static str,
406
407    /// workflow 名称。
408    pub workflow: &'static str,
409    /// definition 版本。
410    pub definition_version: u32,
411    /// 步骤名称。
412    pub step: &'static str,
413    /// 多事务域参与方使用的显式 datasource binding；单事务域可省略。
414    pub binding: Option<&'static str>,
415    /// 业务 Service 类型名。
416    pub service_type: &'static str,
417    /// 是否可补偿。
418    pub compensation: Compensation,
419    /// 取消形态。
420    pub cancel_mode: CancelMode,
421    /// 是否允许返回 `Unknown`。
422    pub allow_unknown: bool,
423    /// 本地托管的解决能力。
424    pub resolution_mode: Option<ResolutionMode>,
425    /// 源码位置，只用于启动拒绝时定位重复合同。
426    pub source: &'static str,
427}
428
429/// 当前 binary 内 `#[saga]` 收集到的全部本地步骤合同。
430#[linkme::distributed_slice]
431pub static COLLECTED_SAGA_STEPS: [SagaStepDescriptor];
432
433/// 业务作用：描述一个由 `#[saga_workflow]` 登记的只读流程定义工厂。
434///
435/// 工厂只返回业务合同，不持有 Orchestrator、transport 或 Application 生命周期；来源位置只在
436/// Ready 拒绝时帮助定位同一二进制内的冲突声明。
437#[derive(Clone, Copy)]
438pub struct SagaWorkflowDescriptor {
439    /// 返回一份已经完成业务字段构造、仍需进入运行时注册校验的流程定义。
440    pub factory: fn() -> anyhow::Result<nasaga_core::WorkflowDefinition>,
441    /// 声明所在源码位置，只用于启动拒绝的低敏定位。
442    pub source: &'static str,
443}
444
445/// 当前二进制内 `#[saga_workflow]` 收集到的全部流程定义工厂。
446#[linkme::distributed_slice]
447pub static COLLECTED_SAGA_WORKFLOWS: [SagaWorkflowDescriptor];
448
449/// 业务作用：调用当前二进制内全部流程工厂，并构造不可变、按定义键去重的注册表快照。
450///
451/// 收集顺序不具备业务含义；每份 definition 自身携带完整步骤顺序。同键同摘要按幂等声明处理，
452/// 同键不同摘要拒绝启动，不能由链接顺序决定哪一份合同生效。
453///
454/// 参数说明: 无。
455///
456/// 返回：全部工厂和 definition 校验通过时返回注册表；构造失败或摘要冲突时返回带声明位置的错误。
457pub fn collect_workflow_definitions() -> anyhow::Result<DefinitionRegistry> {
458    let mut registry = DefinitionRegistry::new();
459    for descriptor in COLLECTED_SAGA_WORKFLOWS {
460        let definition = (descriptor.factory)().map_err(|error| {
461            anyhow::anyhow!(
462                "saga workflow definition factory failed at {}: {}",
463                descriptor.source,
464                error
465            )
466        })?;
467        registry.register(definition).map_err(|error| {
468            anyhow::anyhow!(
469                "saga workflow definition is invalid at {}: {}",
470                descriptor.source,
471                error
472            )
473        })?;
474    }
475    Ok(registry)
476}
477
478/// 业务作用：在 Ready 前复验本地步骤合同唯一且与 definition 一致。
479///
480/// 参数说明：
481/// - `registry`：是启动期冻结的 workflow definition 集合。
482///
483/// 返回：本地投影无重复且已注册 definition 不漂移时返回 `Ok`；否则拒绝 Ready。
484pub fn verify_descriptors(registry: &DefinitionRegistry) -> anyhow::Result<()> {
485    let mut seen = std::collections::BTreeSet::new();
486    for descriptor in COLLECTED_SAGA_STEPS {
487        nasaga_core::SagaPayloadContract {
488            content_type: descriptor.payload_content_type.to_owned(),
489            schema_id: descriptor.payload_schema_id.to_owned(),
490        }
491        .validate()?;
492
493        let key = (
494            descriptor.workflow,
495            descriptor.definition_version,
496            descriptor.step,
497        );
498        // 同一 binary 不能让两个 handler 竞争同一步骤的业务效果。
499        if !seen.insert(key) {
500            anyhow::bail!(
501                "duplicate Saga step `{}` v{} `{}` ({})",
502                descriptor.workflow,
503                descriptor.definition_version,
504                descriptor.step,
505                descriptor.source
506            );
507        }
508        let workflow = nasaga_core::WorkflowName::new(descriptor.workflow)
509            .map_err(|value| anyhow::anyhow!("bad descriptor workflow: {}", value.code()))?;
510        let version = nasaga_core::DefinitionVersion::new(descriptor.definition_version)
511            .map_err(|value| anyhow::anyhow!("bad descriptor version: {}", value.code()))?;
512        let Some(definition) = registry.get(&workflow, version) else {
513            continue;
514        };
515        let step = nasaga_core::StepName::new(descriptor.step)
516            .map_err(|value| anyhow::anyhow!("bad descriptor step: {}", value.code()))?;
517        let Some(step_definition) = definition.step(&step) else {
518            anyhow::bail!(
519                "descriptor step `{}` is absent from workflow `{}` v{} ({})",
520                descriptor.step,
521                descriptor.workflow,
522                descriptor.definition_version,
523                descriptor.source
524            );
525        };
526        // 合同漂移在 Ready 前拒绝，避免已运行实例被不同的补偿或取消语义驱动。
527        if step_definition.payload_contract().content_type != descriptor.payload_content_type
528            || step_definition.payload_contract().schema_id != descriptor.payload_schema_id
529            || step_definition.compensation() != descriptor.compensation
530            || step_definition.cancel_mode() != descriptor.cancel_mode
531            || step_definition.resolution().allow_unknown() != descriptor.allow_unknown
532            || step_definition.resolution().mode() != descriptor.resolution_mode
533        {
534            anyhow::bail!(
535                "descriptor contract drift on step `{}` of `{}` v{} ({})",
536                descriptor.step,
537                descriptor.workflow,
538                descriptor.definition_version,
539                descriptor.source
540            );
541        }
542    }
543    Ok(())
544}
545
546/// 宏展开专用的内部再导出。
547#[doc(hidden)]
548pub mod __private {
549    pub use anyhow;
550    pub use linkme;
551    pub use nasaga_core as core;
552}
553
554use std::sync::atomic::{AtomicU64, Ordering};
555use std::time::Duration;
556
557static KAFKA_RESULT_PROCESSING_TOTAL: AtomicU64 = AtomicU64::new(0);
558static QUOTA_REJECTIONS_TOTAL: AtomicU64 = AtomicU64::new(0);
559static ACTION_RATE_REJECTIONS_TOTAL: AtomicU64 = AtomicU64::new(0);
560static KAFKA_RESULT_PROCESSING_MICROS: AtomicU64 = AtomicU64::new(0);
561static KAFKA_RESULT_RETRY_TOTAL: AtomicU64 = AtomicU64::new(0);
562static KAFKA_RESULT_DLT_REQUESTED_TOTAL: AtomicU64 = AtomicU64::new(0);
563static KAFKA_RESULT_ACK_TOTAL: AtomicU64 = AtomicU64::new(0);
564static KAFKA_RESULT_DUPLICATE_TOTAL: AtomicU64 = AtomicU64::new(0);
565static KAFKA_COMMAND_PROCESSING_TOTAL: AtomicU64 = AtomicU64::new(0);
566static KAFKA_COMMAND_PROCESSING_MICROS: AtomicU64 = AtomicU64::new(0);
567static KAFKA_COMMAND_RETRY_TOTAL: AtomicU64 = AtomicU64::new(0);
568static KAFKA_COMMAND_DLT_REQUESTED_TOTAL: AtomicU64 = AtomicU64::new(0);
569static KAFKA_COMMAND_ACK_TOTAL: AtomicU64 = AtomicU64::new(0);
570static KAFKA_COMMAND_DUPLICATE_TOTAL: AtomicU64 = AtomicU64::new(0);
571
572/// 业务作用：保存同一进程内只允许存在一份的 Saga transport 与治理计数快照。
573#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
574pub struct SagaProcessMetrics {
575    /// 按租户配额拒绝的创建请求累计数。
576    pub quota_rejections_total: u64,
577    /// 按租户频率拒绝的管理动作累计数。
578    pub action_rate_rejections_total: u64,
579    /// Kafka result handler 调用累计数。
580    pub kafka_result_processing_total: u64,
581    /// Kafka result handler 累计耗时微秒数。
582    pub kafka_result_processing_micros_sum: u64,
583    /// Kafka result 保留 offset 重投累计数。
584    pub kafka_result_retry_total: u64,
585    /// Kafka result 请求进入 durability-first DLT 的累计数。
586    pub kafka_result_dlt_requested_total: u64,
587    /// Kafka result 在本地提交后 ACK 的累计数。
588    pub kafka_result_ack_total: u64,
589    /// Kafka result 被 Inbox 吸收后 ACK 的累计数。
590    pub kafka_result_duplicate_total: u64,
591    /// Kafka command handler 调用累计数。
592    pub kafka_command_processing_total: u64,
593    /// Kafka command handler 累计耗时微秒数。
594    pub kafka_command_processing_micros_sum: u64,
595    /// Kafka command 因本地事务未提交而重投的累计数。
596    pub kafka_command_retry_total: u64,
597    /// Kafka command 请求进入 durability-first DLT 的累计数。
598    pub kafka_command_dlt_requested_total: u64,
599    /// Kafka command 在参与方提交后 ACK 的累计数。
600    pub kafka_command_ack_total: u64,
601    /// Kafka command 被 Inbox 吸收后 ACK 的累计数。
602    pub kafka_command_duplicate_total: u64,
603}
604
605/// 业务作用：读取 MySQL 与 PostgreSQL wrapper 共同维护的 process metrics 快照。
606///
607/// 参数说明: 无。
608///
609/// 返回：调用时刻各原子计数的低基数快照；不包含任何数据库事实或业务身份。
610pub fn process_metrics_snapshot() -> SagaProcessMetrics {
611    SagaProcessMetrics {
612        quota_rejections_total: QUOTA_REJECTIONS_TOTAL.load(Ordering::Relaxed),
613        action_rate_rejections_total: ACTION_RATE_REJECTIONS_TOTAL.load(Ordering::Relaxed),
614        kafka_result_processing_total: KAFKA_RESULT_PROCESSING_TOTAL.load(Ordering::Relaxed),
615        kafka_result_processing_micros_sum: KAFKA_RESULT_PROCESSING_MICROS.load(Ordering::Relaxed),
616        kafka_result_retry_total: KAFKA_RESULT_RETRY_TOTAL.load(Ordering::Relaxed),
617        kafka_result_dlt_requested_total: KAFKA_RESULT_DLT_REQUESTED_TOTAL.load(Ordering::Relaxed),
618        kafka_result_ack_total: KAFKA_RESULT_ACK_TOTAL.load(Ordering::Relaxed),
619        kafka_result_duplicate_total: KAFKA_RESULT_DUPLICATE_TOTAL.load(Ordering::Relaxed),
620        kafka_command_processing_total: KAFKA_COMMAND_PROCESSING_TOTAL.load(Ordering::Relaxed),
621        kafka_command_processing_micros_sum: KAFKA_COMMAND_PROCESSING_MICROS
622            .load(Ordering::Relaxed),
623        kafka_command_retry_total: KAFKA_COMMAND_RETRY_TOTAL.load(Ordering::Relaxed),
624        kafka_command_dlt_requested_total: KAFKA_COMMAND_DLT_REQUESTED_TOTAL
625            .load(Ordering::Relaxed),
626        kafka_command_ack_total: KAFKA_COMMAND_ACK_TOTAL.load(Ordering::Relaxed),
627        kafka_command_duplicate_total: KAFKA_COMMAND_DUPLICATE_TOTAL.load(Ordering::Relaxed),
628    }
629}
630
631/// 业务作用：累计一次按租户配额拒绝的创建请求。
632///
633/// 参数说明: 无。
634///
635/// 返回：无；原子计数不阻塞业务线程。
636pub fn record_quota_rejection() {
637    QUOTA_REJECTIONS_TOTAL.fetch_add(1, Ordering::Relaxed);
638}
639
640/// 业务作用：累计一次按租户频率拒绝的变更类管理动作。
641///
642/// 参数说明: 无。
643///
644/// 返回：无；原子计数不阻塞业务线程。
645pub fn record_action_rate_rejection() {
646    ACTION_RATE_REJECTIONS_TOTAL.fetch_add(1, Ordering::Relaxed);
647}
648
649/// 业务作用：记录一次 Kafka Saga result handler 的单调耗时。
650///
651/// 参数说明：
652/// - `elapsed`：是从 transport 门禁到本地事务结束的单调耗时。
653///
654/// 返回：无；次数原子递增，耗时使用饱和累计。
655pub fn record_kafka_processing(elapsed: Duration) {
656    KAFKA_RESULT_PROCESSING_TOTAL.fetch_add(1, Ordering::Relaxed);
657    saturating_add_duration(&KAFKA_RESULT_PROCESSING_MICROS, elapsed);
658}
659
660/// 业务作用：记录 Saga result 因本地事务未提交而保留 offset 重投。
661///
662/// 参数说明: 无。
663///
664/// 返回：无；原子计数不阻塞 transport。
665pub fn record_kafka_retry() {
666    KAFKA_RESULT_RETRY_TOTAL.fetch_add(1, Ordering::Relaxed);
667}
668
669/// 业务作用：记录 Saga result 已请求进入 durability-first DLT。
670///
671/// 参数说明: 无。
672///
673/// 返回：无；broker 最终持久结果由 transport 自身指标证明。
674pub fn record_kafka_dlt_requested() {
675    KAFKA_RESULT_DLT_REQUESTED_TOTAL.fetch_add(1, Ordering::Relaxed);
676}
677
678/// 业务作用：记录 Saga result 在本地提交后完成 ACK，并区分 Inbox 重复吸收。
679///
680/// 参数说明：
681/// - `duplicate`：表示本次是否未重复执行业务效果。
682///
683/// 返回：无；ACK 与重复计数原子更新。
684pub fn record_kafka_ack(duplicate: bool) {
685    KAFKA_RESULT_ACK_TOTAL.fetch_add(1, Ordering::Relaxed);
686    if duplicate {
687        KAFKA_RESULT_DUPLICATE_TOTAL.fetch_add(1, Ordering::Relaxed);
688    }
689}
690
691/// 业务作用：记录一次 Kafka Saga command handler 的单调耗时。
692///
693/// 参数说明：
694/// - `elapsed`：是从 transport 门禁到参与方事务结束的单调耗时。
695///
696/// 返回：无；次数原子递增，耗时使用饱和累计。
697pub fn record_kafka_command_processing(elapsed: Duration) {
698    KAFKA_COMMAND_PROCESSING_TOTAL.fetch_add(1, Ordering::Relaxed);
699    saturating_add_duration(&KAFKA_COMMAND_PROCESSING_MICROS, elapsed);
700}
701
702/// 业务作用：记录 Saga command 因参与方事务未提交而保留 offset 重投。
703///
704/// 参数说明: 无。
705///
706/// 返回：无；原子计数不阻塞 transport。
707pub fn record_kafka_command_retry() {
708    KAFKA_COMMAND_RETRY_TOTAL.fetch_add(1, Ordering::Relaxed);
709}
710
711/// 业务作用：记录 Saga command 已请求进入 durability-first DLT。
712///
713/// 参数说明: 无。
714///
715/// 返回：无；broker 最终持久结果由 transport 自身指标证明。
716pub fn record_kafka_command_dlt_requested() {
717    KAFKA_COMMAND_DLT_REQUESTED_TOTAL.fetch_add(1, Ordering::Relaxed);
718}
719
720/// 业务作用：记录 Saga command 在参与方提交后完成 ACK，并区分 Inbox 重复吸收。
721///
722/// 参数说明：
723/// - `duplicate`：表示本次是否未重复执行业务效果。
724///
725/// 返回：无；ACK 与重复计数原子更新。
726pub fn record_kafka_command_ack(duplicate: bool) {
727    KAFKA_COMMAND_ACK_TOTAL.fetch_add(1, Ordering::Relaxed);
728    if duplicate {
729        KAFKA_COMMAND_DUPLICATE_TOTAL.fetch_add(1, Ordering::Relaxed);
730    }
731}
732
733/// 业务作用：把单次耗时以微秒饱和累加到进程计数器。
734///
735/// 参数说明：
736/// - `target`：是目标原子计数
737/// - `elapsed`：是单调耗时。
738///
739/// 返回：无；溢出时保持 `u64::MAX`，不允许回绕成较小值。
740fn saturating_add_duration(target: &AtomicU64, elapsed: Duration) {
741    let micros = u64::try_from(elapsed.as_micros()).unwrap_or(u64::MAX);
742    let _ = target.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| {
743        Some(current.saturating_add(micros))
744    });
745}
nasaga_runtime_core/lib.rs

nasaga_runtime_core/
lib.rs
