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

nasaga_runtime_core/
lib.rs
