nanotify_core/
lib.rs

1//! 有界、渠道中立的离散事件通知合同。
2//!
3//! 队列生产端只向固定容量队列尝试入队，不调用第三方代码；宿主负责 worker 生命周期、
4//! 超时、并发和停止。文本在构造通知时清理控制字符并限制长度，不接收数据库对象或任意 JSON。
5//! 业务主动实现 [`Notify`] 并自行选择通知微服务的通信方式；本组件不提供渠道客户端、协议或机器人凭据配置。
6//! 业务用 [`init`] 主动安装进程实现；[`notify`] 在未初始化时忽略消息，有实现时直接调用。
7//! 直接调用不隐式获得队列、超时或异常隔离。受管 SQL 通知由宿主 worker 调用业务实现，
8//! [`NotifyReceipt::accepted`] 为真才表示适配器接受，不代表最终收件；有界队列不提供持久送达或跨副本去重。
9
10mod queue;
11pub use queue::*;
12mod registry;
13pub use registry::*;
14
15use serde::{Deserialize, Serialize};
16use std::fmt;
17use std::time::Duration;
18
19/// 进程级通知调度预算；未启用通知时仅保留配置，不创建运行资源。
20#[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize)]
21#[serde(default, deny_unknown_fields)]
22pub struct DispatcherConfig {
23    /// 等待 worker 接收的通知数量上限。
24    pub queue_capacity: usize,
25    /// 宿主同时执行的通知投递数量上限。
26    pub max_in_flight: usize,
27    /// worker 开始投递后覆盖全部尝试与退避的总预算，单位毫秒，不含排队时间。
28    pub delivery_timeout_ms: u64,
29    /// 停机后队列排空预算，单位毫秒，同时受宿主剩余期限约束。
30    pub shutdown_drain_timeout_ms: u64,
31    /// 同一通知最多尝试次数，包含首次投递；重试还须有安全证据。
32    pub max_attempts: usize,
33    /// 可安全重试时的初始退避，单位毫秒。
34    pub retry_initial_backoff_ms: u64,
35    /// 指数退避的上限，单位毫秒；服务端要求的最小等待仍须满足。
36    pub retry_max_backoff_ms: u64,
37}
38
39impl Default for DispatcherConfig {
40    /// 业务作用：提供所有宿主共用的有限队列与投递预算。
41    ///
42    /// 参数说明：无。
43    ///
44    /// 返回：关闭无限等待和无限重试的稳定默认值。
45    fn default() -> Self {
46        Self {
47            queue_capacity: 512,
48            max_in_flight: 4,
49            delivery_timeout_ms: 3000,
50            shutdown_drain_timeout_ms: 3000,
51            max_attempts: 1,
52            retry_initial_backoff_ms: 250,
53            retry_max_backoff_ms: 2000,
54        }
55    }
56}
57
58impl DispatcherConfig {
59    /// 业务作用：在分配队列或调用渠道前验证调度资源和时间上限。
60    ///
61    /// 参数说明：无。
62    ///
63    /// 返回：配置合法时成功，否则只返回静态字段名。
64    pub fn validate(&self) -> Result<(), ConfigError> {
65        for (valid, field) in [
66            (
67                (1..=65536).contains(&self.queue_capacity),
68                "dispatcher.queue_capacity",
69            ),
70            (
71                (1..=32).contains(&self.max_in_flight),
72                "dispatcher.max_in_flight",
73            ),
74            (
75                (100..=30000).contains(&self.delivery_timeout_ms),
76                "dispatcher.delivery_timeout_ms",
77            ),
78            (
79                self.shutdown_drain_timeout_ms <= 30000,
80                "dispatcher.shutdown_drain_timeout_ms",
81            ),
82            (
83                (1..=5).contains(&self.max_attempts),
84                "dispatcher.max_attempts",
85            ),
86            (
87                (50..=5000).contains(&self.retry_initial_backoff_ms),
88                "dispatcher.retry_initial_backoff_ms",
89            ),
90            (
91                (self.retry_initial_backoff_ms..=10000).contains(&self.retry_max_backoff_ms),
92                "dispatcher.retry_max_backoff_ms",
93            ),
94        ] {
95            if !valid {
96                return Err(ConfigError(field));
97            }
98        }
99        Ok(())
100    }
101}
102
103/// 只携带静态配置路径的安全错误。
104#[derive(Debug, Clone, Copy, PartialEq, Eq)]
105pub struct ConfigError(pub &'static str);
106
107impl fmt::Display for ConfigError {
108    /// 业务作用：报告失败字段而不显示配置材料。
109    ///
110    /// 参数说明：
111    /// - `formatter`：接收脱敏诊断。
112    ///
113    /// 返回：格式化成功或格式化器失败。
114    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
115        write!(formatter, "invalid notification configuration: {}", self.0)
116    }
117}
118impl std::error::Error for ConfigError {}
119
120/// 固定离散事件集合，也是通知指标的低基数维度。
121#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, Deserialize, Serialize)]
122#[serde(rename_all = "snake_case")]
123#[repr(usize)]
124pub enum EventKind {
125    #[default]
126    /// 数据库活跃时长达到慢 SQL 阈值。
127    SlowSql,
128    /// 数据库操作返回显式执行错误。
129    ExecutionError,
130    /// 获取数据库连接超过等待期限。
131    AcquireTimeout,
132}
133
134impl EventKind {
135    /// 业务作用：返回稳定事件名称供通知与指标共享。
136    ///
137    /// 参数说明：无。
138    ///
139    /// 返回：不包含调用参数的静态名称。
140    pub const fn as_str(self) -> &'static str {
141        match self {
142            Self::SlowSql => "slow_sql",
143            Self::ExecutionError => "execution_error",
144            Self::AcquireTimeout => "acquire_timeout",
145        }
146    }
147}
148
149/// 渠道中立的消息严重程度。
150#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, Deserialize, Serialize)]
151#[serde(rename_all = "snake_case")]
152pub enum Severity {
153    /// 业务信息通知，无需按错误处理。
154    Info,
155    #[default]
156    /// 需要关注的业务告警。
157    Warning,
158    /// 需要处理的执行失败。
159    Error,
160    /// 业务认定需要紧急响应的严重告警。
161    Critical,
162}
163
164/// 启动期冻结的通知来源身份；与指标出口使用同一进程标识。
165#[derive(Debug, Clone, Default, PartialEq, Eq)]
166pub struct NotificationIdentity {
167    /// 产生通知的稳定服务名。
168    pub service: String,
169    /// 产生通知的进程实例身份。
170    pub instance: String,
171    /// 可选部署环境身份。
172    pub environment: Option<String>,
173    /// 可选部署集群身份。
174    pub cluster: Option<String>,
175}
176
177impl Severity {
178    /// 业务作用：返回跨渠道一致的严重程度名称。
179    ///
180    /// 参数说明：无。
181    ///
182    /// 返回：固定的纯文本标识。
183    pub const fn as_str(self) -> &'static str {
184        match self {
185            Self::Info => "info",
186            Self::Warning => "warning",
187            Self::Error => "error",
188            Self::Critical => "critical",
189        }
190    }
191}
192
193/// 通知输入；构造后所有文本均由 [`Notification::new`] 施加统一边界。
194#[derive(Clone, Default)]
195pub struct NotificationFields {
196    /// 通知事件标识，供业务渠道关联，不自动提供跨副本去重。
197    pub id: String,
198    /// 触发本次通知的固定事件分类。
199    pub event: EventKind,
200    /// 业务规则指定的严重程度。
201    pub severity: Severity,
202    /// 事件发生时间，Unix epoch 毫秒。
203    pub occurred_at_unix_ms: u64,
204    /// 产生通知的稳定服务名。
205    pub service: String,
206    /// 产生通知的进程实例身份。
207    pub instance: String,
208    /// 可选部署环境身份。
209    pub environment: Option<String>,
210    /// 可选部署集群身份。
211    pub cluster: Option<String>,
212    /// 产生事件的数据库驱动标识。
213    pub driver: String,
214    /// 产生事件的数据源目录名称。
215    pub datasource: String,
216    /// 触发事件的数据库或连接操作类别。
217    pub operation: String,
218    /// 可选完整 Mapper 方法身份。
219    pub method: Option<String>,
220    /// 连接获取事件的可选用途标识。
221    pub purpose: Option<String>,
222    /// 事件关联的数据库活跃时长或连接等待时长，不含通知排队时间。
223    pub duration: Duration,
224    /// 操作的稳定结果分类，不应写入原始错误文本。
225    pub outcome: String,
226    /// 可选稳定错误分类，不包含数据库响应正文。
227    pub error_kind: Option<String>,
228    /// 显式允许携带的 SQL 模板，不得包含绑定参数或凭据。
229    pub prepared_sql: Option<String>,
230    /// 可选调用链标识，供渠道关联业务事件。
231    pub trace_id: Option<String>,
232    /// 该操作是否同时命中慢 SQL 阈值，即使主通知事件为执行错误也保留此事实。
233    pub slow: bool,
234}
235
236/// 拥有型且长度受限的通知；不实现输出消息内容的诊断格式化。
237#[derive(Clone)]
238pub struct Notification(NotificationFields);
239
240impl Notification {
241    /// 业务作用：在通知跨越异步队列前建立文本长度和控制字符边界。
242    ///
243    /// 参数说明：
244    /// - `fields`：是仅在规则命中后构造的通知输入，不得包含凭据或绑定参数。
245    ///
246    /// 返回：标识最多 128 字符、方法最多 512 字符、SQL 最多 4096 字符的只读通知。
247    pub fn new(mut fields: NotificationFields) -> Self {
248        for text in [
249            &mut fields.id,
250            &mut fields.service,
251            &mut fields.instance,
252            &mut fields.driver,
253            &mut fields.datasource,
254            &mut fields.operation,
255            &mut fields.outcome,
256        ] {
257            *text = bounded_text(text, 128);
258        }
259        for value in [
260            &mut fields.environment,
261            &mut fields.cluster,
262            &mut fields.purpose,
263            &mut fields.error_kind,
264            &mut fields.trace_id,
265        ]
266        .into_iter()
267        .flatten()
268        {
269            *value = bounded_text(value, 128);
270        }
271        if let Some(value) = &mut fields.method {
272            *value = bounded_text(value, 512);
273        }
274        if let Some(value) = &mut fields.prepared_sql {
275            *value = bounded_text(value, 4096);
276        }
277        Self(fields)
278    }
279
280    /// 业务作用：允许 transport 读取已经通过边界约束的消息。
281    ///
282    /// 参数说明：无。
283    ///
284    /// 返回：不可修改的通知内容，调用方不得写入普通日志。
285    pub fn fields(&self) -> &NotificationFields {
286        &self.0
287    }
288}
289
290/// 业务作用：按 Unicode 边界限制长度并消除控制与方向覆盖字符，防止消息结构注入。
291///
292/// 参数说明：
293/// - `text`：为待投递内容
294/// - `max_chars`：为字符数硬上限。
295///
296/// 返回：长度有界的纯文本；控制字符替换为空格，截断结尾以省略号标示。
297pub fn bounded_text(text: &str, max_chars: usize) -> String {
298    if max_chars == 0 {
299        return String::new();
300    }
301    let mut chars = text.chars();
302    let mut output = String::with_capacity(text.len().min(max_chars.saturating_mul(4)));
303    for _ in 0..max_chars {
304        let Some(ch) = chars.next() else {
305            return output;
306        };
307        output.push(
308            if ch.is_control() || matches!(ch, '\u{202a}'..='\u{202e}' | '\u{2066}'..='\u{2069}') {
309                ' '
310            } else {
311                ch
312            },
313        );
314    }
315    if chars.next().is_some() {
316        output.pop();
317        output.push('…');
318    }
319    output
320}
321
322/// 渠道受理回执，不携带原始响应正文；未受理不等同于 trait 返回错误。
323#[derive(Debug, Clone, PartialEq, Eq)]
324pub struct NotifyReceipt {
325    /// 业务适配器提供的可选消息标识，不应包含渠道凭据或响应正文。
326    pub message_id: Option<String>,
327    /// true 表示下游受理，false 在受管 dispatcher 中记为拒绝；均不证明最终用户收件。
328    pub accepted: bool,
329}
330
331/// 稳定失败分类；原始响应与 URL 不进入错误链。
332#[derive(Debug, Clone, Copy, PartialEq, Eq)]
333#[repr(usize)]
334pub enum NotifyErrorKind {
335    /// 投递超过期限，不能仅凭超时认定远端未受理。
336    Timeout,
337    /// 通知服务暂不可用。
338    Unavailable,
339    /// 通知服务拒绝了超过限额的请求。
340    RateLimited,
341    /// 通知服务明确拒绝受理。
342    Rejected,
343    /// 通知服务认证失败。
344    Authentication,
345    /// 请求不符合通知服务接口要求。
346    InvalidRequest,
347    /// 其它稳定分类之外的投递失败。
348    Other,
349}
350
351impl NotifyErrorKind {
352    /// 业务作用：把渠道失败映射为低基数诊断名称。
353    ///
354    /// 参数说明：无。
355    ///
356    /// 返回：稳定的静态失败名称。
357    pub const fn as_str(self) -> &'static str {
358        match self {
359            Self::Timeout => "timeout",
360            Self::Unavailable => "unavailable",
361            Self::RateLimited => "rate_limited",
362            Self::Rejected => "rejected",
363            Self::Authentication => "authentication",
364            Self::InvalidRequest => "invalid_request",
365            Self::Other => "other",
366        }
367    }
368}
369
370/// 明确的重试证据；结果未知不得自行重发。
371#[derive(Debug, Clone, Copy, PartialEq, Eq)]
372pub enum RetrySafety {
373    /// 缺少安全重试证据，禁止自动重发。
374    Never,
375    /// 连接未发送或服务明确拒绝，并允许在宿主预算内重新尝试。
376    Safe {
377        /// 下游要求的最小等待；省略时使用宿主退避策略。
378        retry_after: Option<Duration>,
379    },
380}
381
382/// 脱敏错误，只描述稳定分类及重试证据。
383#[derive(Debug, Clone, Copy, PartialEq, Eq)]
384pub struct NotifyError {
385    /// 不含凭据、URL 或响应正文的失败分类。
386    pub kind: NotifyErrorKind,
387    /// 是否有明确证据允许重试，默认不允许。
388    pub retry: RetrySafety,
389}
390
391impl NotifyError {
392    /// 业务作用：构造不能安全重发的通知失败。
393    ///
394    /// 参数说明：
395    /// - `kind`：是不含原始响应的分类。
396    ///
397    /// 返回：默认禁止重试的错误。
398    pub const fn new(kind: NotifyErrorKind) -> Self {
399        Self {
400            kind,
401            retry: RetrySafety::Never,
402        }
403    }
404    /// 业务作用：声明连接未发送或服务明确拒绝后的安全重试。
405    ///
406    /// 参数说明：
407    /// - `kind`：为分类
408    /// - `retry_after`：为服务要求的最小等待。
409    ///
410    /// 返回：携带重试证据的错误，仍受宿主总预算限制。
411    pub const fn retryable(kind: NotifyErrorKind, retry_after: Option<Duration>) -> Self {
412        Self {
413            kind,
414            retry: RetrySafety::Safe {
415                retry_after,
416            },
417        }
418    }
419}
420
421impl fmt::Display for NotifyError {
422    /// 业务作用：避免渠道凭据或响应正文进入普通错误链。
423    ///
424    /// 参数说明：
425    /// - `formatter`：接收稳定分类。
426    ///
427    /// 返回：格式化器的写入结果。
428    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
429        formatter.write_str(self.kind.as_str())
430    }
431}
432impl std::error::Error for NotifyError {}
433
434/// 受管 SQL 通知只在宿主管理的异步 worker 内调用，不得在 SQL 生产路径调用。
435/// 业务也可直接调用进程通知接口，此时由业务负责超时、异常隔离与执行边界。
436/// 业务实现负责调用通知微服务；协议、鉴权和下游消息发送由业务拥有，框架不连接具体消息渠道。
437/// Future 必须协作让出执行权且取消安全；同步阻塞或自行派生的后台任务不受宿主时间预算保护。
438#[async_trait::async_trait]
439pub trait Notify: Send + Sync {
440    /// 业务作用：投递一条已完成裁决和脱敏的通知，不参与业务事务。
441    ///
442    /// 参数说明：
443    /// - `notification`：是只读有界消息。
444    ///
445    /// 返回：包含是否受理的回执或稳定失败；受理不表示最终用户收到，未知结果不得声明为安全重试。
446    async fn notify(&self, notification: &Notification) -> Result<NotifyReceipt, NotifyError>;
447}
nanotify_core/lib.rs

nanotify_core/
lib.rs
