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

nasaga_runtime_core/
lib.rs
