nanotify_core/
queue.rs

1use crate::{ConfigError, EventKind, Notification, NotifyErrorKind};
2use nametrics_core::atomic::{add, AtomicHistogram, LATENCY_SECONDS};
3use nametrics_core::{
4    LegacyMetricsSource, MetricDescriptor, MetricKind, MetricSample, MetricValue,
5};
6use std::collections::BTreeSet;
7use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
8use std::sync::Arc;
9use std::time::Duration;
10use tokio::sync::mpsc;
11
12static ENQUEUED: MetricDescriptor = MetricDescriptor {
13    name: "nanotify_enqueued_total",
14    help: "进入通知队列的离散事件累计数。",
15    unit: "",
16    kind: MetricKind::Counter,
17    label_names: &["provider", "event"],
18    histogram_bounds: &[],
19};
20static DROPPED: MetricDescriptor = MetricDescriptor {
21    name: "nanotify_dropped_total",
22    help: "因队列或停机门禁丢弃的通知累计数。",
23    unit: "",
24    kind: MetricKind::Counter,
25    label_names: &["provider", "event", "reason"],
26    histogram_bounds: &[],
27};
28static DELIVERIES: MetricDescriptor = MetricDescriptor {
29    name: "nanotify_deliveries_total",
30    help: "以最终结果分类的通知投递累计数。",
31    unit: "",
32    kind: MetricKind::Counter,
33    label_names: &["provider", "event", "outcome"],
34    histogram_bounds: &[],
35};
36static DURATION: MetricDescriptor = MetricDescriptor {
37    name: "nanotify_delivery_duration_seconds",
38    help: "包含有限重试和等待的通知投递时长。",
39    unit: "seconds",
40    kind: MetricKind::Histogram,
41    label_names: &["provider", "outcome"],
42    histogram_bounds: LATENCY_SECONDS,
43};
44static DEPTH: MetricDescriptor = MetricDescriptor {
45    name: "nanotify_queue_depth",
46    help: "等待 worker 领取的通知数量。",
47    unit: "",
48    kind: MetricKind::Gauge,
49    label_names: &[],
50    histogram_bounds: &[],
51};
52static CAPACITY: MetricDescriptor = MetricDescriptor {
53    name: "nanotify_queue_capacity",
54    help: "当前进程通知队列的固定容量。",
55    unit: "",
56    kind: MetricKind::Gauge,
57    label_names: &[],
58    histogram_bounds: &[],
59};
60static DESCRIPTORS: &[&MetricDescriptor] = &[
61    &ENQUEUED,
62    &DROPPED,
63    &DELIVERIES,
64    &DURATION,
65    &DEPTH,
66    &CAPACITY,
67];
68const EVENTS: [EventKind; 3] = [
69    EventKind::SlowSql,
70    EventKind::ExecutionError,
71    EventKind::AcquireTimeout,
72];
73const DROP_REASONS: [&str; 3] = ["queue_full", "stopping", "shutdown_timeout"];
74const OUTCOMES: [&str; 10] = [
75    "accepted",
76    "timeout",
77    "unavailable",
78    "rate_limited",
79    "rejected",
80    "authentication",
81    "invalid_request",
82    "other",
83    "panic",
84    "shutdown",
85];
86
87/// 一次 envelope 投递的最终结果，不以尝试次数重复计数。
88#[derive(Debug, Clone, Copy, PartialEq, Eq)]
89#[repr(usize)]
90pub enum DeliveryOutcome {
91    /// 适配器确认下游受理，不证明最终收件。
92    Accepted,
93    /// 投递超过期限，不能仅凭超时认定远端未受理。
94    Timeout,
95    /// 通知服务暂不可用。
96    Unavailable,
97    /// 通知服务拒绝了超过限额的请求。
98    RateLimited,
99    /// 通知服务明确拒绝受理。
100    Rejected,
101    /// 通知服务认证失败。
102    Authentication,
103    /// 请求不符合通知服务接口要求。
104    InvalidRequest,
105    /// 其它稳定分类之外的投递失败。
106    Other,
107    /// 投递或其析构因 panic 展开退出。
108    Panic,
109    /// 投递在宿主停机收口时被取消。
110    Shutdown,
111}
112
113impl From<NotifyErrorKind> for DeliveryOutcome {
114    /// 业务作用：把渠道错误映射到统一投递指标。
115    ///
116    /// 参数说明：
117    /// - `kind`：是稳定渠道失败分类。
118    ///
119    /// 返回：不会携带原始错误的固定结果。
120    fn from(kind: NotifyErrorKind) -> Self {
121        match kind {
122            NotifyErrorKind::Timeout => Self::Timeout,
123            NotifyErrorKind::Unavailable => Self::Unavailable,
124            NotifyErrorKind::RateLimited => Self::RateLimited,
125            NotifyErrorKind::Rejected => Self::Rejected,
126            NotifyErrorKind::Authentication => Self::Authentication,
127            NotifyErrorKind::InvalidRequest => Self::InvalidRequest,
128            NotifyErrorKind::Other => Self::Other,
129        }
130    }
131}
132
133struct ProviderMetrics {
134    id: String,
135    enqueued: [AtomicU64; 3],
136    dropped: [[AtomicU64; 3]; 3],
137    deliveries: [[AtomicU64; 10]; 3],
138    duration: [AtomicHistogram; 10],
139}
140
141/// 通知源的固定原子单元；仅抓取时构造 label 和样本。
142pub struct QueueMetrics {
143    providers: Vec<ProviderMetrics>,
144    capacity: usize,
145    depth: AtomicU64,
146}
147
148impl QueueMetrics {
149    /// 业务作用：为指标注册事务提供精确最坏系列数。
150    ///
151    /// 参数说明：无。
152    ///
153    /// 返回：全部 provider、事件、结果和固定桶展开后的总系列数。
154    pub fn series_budget(&self) -> usize {
155        self.providers.len() * (3 + 9 + 30 + 10 * (LATENCY_SECONDS.len() + 3)) + 2
156    }
157
158    /// 业务作用：在一个 envelope 结束时写入唯一投递结果与总耗时。
159    ///
160    /// 参数说明：
161    /// - `provider`：是冻结下标
162    /// - `event`：为事件
163    /// - `outcome`：为最终结果
164    /// - `elapsed`：包含重试。
165    ///
166    /// 返回：无；未知下标不写入，业务结果不受影响。
167    pub fn record_delivery(
168        &self,
169        provider: usize,
170        event: EventKind,
171        outcome: DeliveryOutcome,
172        elapsed: Duration,
173    ) {
174        if let Some(slot) = self.providers.get(provider) {
175            add(&slot.deliveries[event as usize][outcome as usize], 1);
176            slot.duration[outcome as usize].observe(elapsed);
177        }
178    }
179}
180
181impl LegacyMetricsSource for QueueMetrics {
182    /// 业务作用：声明此源拥有的静态通知指标合同。
183    ///
184    /// 参数说明：无。
185    ///
186    /// 返回：固定 descriptor 集合。
187    fn descriptors(&self) -> &'static [&'static MetricDescriptor] {
188        DESCRIPTORS
189    }
190
191    /// 业务作用：从原子单元生成可供 Prometheus 和 OTLP 共用的有界快照。
192    ///
193    /// 参数说明：无。
194    ///
195    /// 返回：全部冻结维度的当前累计样本。
196    fn snapshot(&self) -> Option<Vec<MetricSample>> {
197        let mut samples = Vec::new();
198        for provider in &self.providers {
199            for event in EVENTS {
200                let labels = vec![
201                    ("provider", provider.id.clone()),
202                    ("event", event.as_str().into()),
203                ];
204                samples.push(MetricSample {
205                    name: ENQUEUED.name,
206                    labels: labels.clone(),
207                    value: MetricValue::Counter(
208                        provider.enqueued[event as usize].load(Ordering::Relaxed),
209                    ),
210                });
211                for (index, reason) in DROP_REASONS.iter().enumerate() {
212                    let mut labels = labels.clone();
213                    labels.push(("reason", (*reason).into()));
214                    samples.push(MetricSample {
215                        name: DROPPED.name,
216                        labels,
217                        value: MetricValue::Counter(
218                            provider.dropped[event as usize][index].load(Ordering::Relaxed),
219                        ),
220                    });
221                }
222                for (index, outcome) in OUTCOMES.iter().enumerate() {
223                    let mut labels = labels.clone();
224                    labels.push(("outcome", (*outcome).into()));
225                    samples.push(MetricSample {
226                        name: DELIVERIES.name,
227                        labels,
228                        value: MetricValue::Counter(
229                            provider.deliveries[event as usize][index].load(Ordering::Relaxed),
230                        ),
231                    });
232                }
233            }
234            for (index, outcome) in OUTCOMES.iter().enumerate() {
235                samples.push(MetricSample {
236                    name: DURATION.name,
237                    labels: vec![
238                        ("provider", provider.id.clone()),
239                        ("outcome", (*outcome).into()),
240                    ],
241                    value: provider.duration[index].snapshot(),
242                });
243            }
244        }
245        samples.push(MetricSample {
246            name: DEPTH.name,
247            labels: vec![],
248            value: MetricValue::Gauge(self.depth.load(Ordering::Relaxed) as f64),
249        });
250        samples.push(MetricSample {
251            name: CAPACITY.name,
252            labels: vec![],
253            value: MetricValue::Gauge(self.capacity as f64),
254        });
255        Some(samples)
256    }
257
258    /// 业务作用：由统一 MetricHub 的结构化出口渲染，不建立第二份文本状态。
259    ///
260    /// 参数说明：
261    /// - `output`：是宿主的输出缓冲区。
262    ///
263    /// 返回：无；结构化快照由 MetricHub 统一渲染。
264    fn render_prometheus(&self, output: &mut String) {
265        nametrics_core::atomic::render_snapshot(
266            DESCRIPTORS,
267            &self.snapshot().unwrap_or_default(),
268            output,
269        );
270    }
271}
272
273struct QueueState {
274    accepting: AtomicBool,
275    metrics: Arc<QueueMetrics>,
276}
277
278/// 启动期解析的具体队列路由；热路径不解析 provider 名称，不执行用户回调。
279#[derive(Clone)]
280pub struct AlertRoute {
281    sender: mpsc::Sender<Envelope>,
282    state: Arc<QueueState>,
283    provider: usize,
284    requires_initialized_notify: bool,
285}
286
287/// 非阻塞入队结果；不包含可以反向传播到业务调用的错误。
288#[derive(Debug, Clone, Copy, PartialEq, Eq)]
289pub enum EnqueueOutcome {
290    /// 通知已进入本地有界队列，尚未确认下游受理。
291    Enqueued,
292    /// 队列容量不足，新通知被丢弃。
293    Full,
294    /// 通知生产权已关闭，新通知被拒绝。
295    Stopping,
296    /// 进程默认通知实现尚未安装，本次通知不入队。
297    Ignored,
298}
299
300impl AlertRoute {
301    /// 业务作用：尝试移交通知所有权，满队列采用 drop-new 并记录稳定丢弃原因。
302    ///
303    /// 参数说明：
304    /// - `notification`：是规则命中后才构造的有界消息。
305    ///
306    /// 返回：入队、容量不足、停止或未初始化时忽略；本方法从不等待消费者或业务实现。
307    pub fn try_send(&self, notification: Notification) -> EnqueueOutcome {
308        // 未初始化的进程通知直接忽略，不占队列、不记录投递失败，也不在稍后初始化后补发旧事件。
309        if self.requires_initialized_notify && crate::get().is_none() {
310            return EnqueueOutcome::Ignored;
311        }
312        let event = notification.fields().event;
313        let slot = &self.state.metrics.providers[self.provider];
314        // 停机先关闭生产权，旧 Stream 的终态仍能安全丢弃通知而不延长业务停机。
315        if !self.state.accepting.load(Ordering::Acquire) {
316            add(&slot.dropped[event as usize][1], 1);
317            return EnqueueOutcome::Stopping;
318        }
319        match self.sender.try_reserve() {
320            Ok(permit) => {
321                if !self.state.accepting.load(Ordering::Acquire) {
322                    add(&slot.dropped[event as usize][1], 1);
323                    return EnqueueOutcome::Stopping;
324                }
325                add(&self.state.metrics.depth, 1);
326                add(&slot.enqueued[event as usize], 1);
327                permit.send(Envelope {
328                    provider: self.provider,
329                    notification,
330                    queued: true,
331                    metrics: self.state.metrics.clone(),
332                });
333                EnqueueOutcome::Enqueued
334            }
335            Err(mpsc::error::TrySendError::Full(_)) => {
336                add(&slot.dropped[event as usize][0], 1);
337                EnqueueOutcome::Full
338            }
339            Err(mpsc::error::TrySendError::Closed(_)) => {
340                add(&slot.dropped[event as usize][1], 1);
341                EnqueueOutcome::Stopping
342            }
343        }
344    }
345}
346
347/// 已冻结 provider 目录的队列生产能力。
348#[derive(Clone)]
349pub struct NotificationProducer {
350    sender: mpsc::Sender<Envelope>,
351    state: Arc<QueueState>,
352}
353
354impl NotificationProducer {
355    /// 业务作用：在启动期把名称解析为可在热路径直接使用的队列路由。
356    ///
357    /// 参数说明：
358    /// - `provider_ref`：必须来自经过校验的冻结配置。
359    ///
360    /// 返回：对应路由；未知 provider 返回 None。
361    pub fn route(&self, provider_ref: &str) -> Option<AlertRoute> {
362        self.state
363            .metrics
364            .providers
365            .iter()
366            .position(|slot| slot.id == provider_ref)
367            .map(|provider| AlertRoute {
368                sender: self.sender.clone(),
369                state: self.state.clone(),
370                provider,
371                requires_initialized_notify: false,
372            })
373    }
374
375    /// 业务作用：取得使用进程通知实现的非阻塞路由，未初始化时通知不进入队列。
376    ///
377    /// 参数说明：无。
378    ///
379    /// 返回：目录包含默认路由时返回生产能力；其它目录返回 None，不增建 provider 或指标维度。
380    pub fn default_route(&self) -> Option<AlertRoute> {
381        self.route(crate::DEFAULT_PROVIDER_ID).map(|mut route| {
382            route.requires_initialized_notify = true;
383            route
384        })
385    }
386    /// 业务作用：先撤销所有生产路由的接受权，再由宿主排空已接受消息。
387    ///
388    /// 参数说明：无。
389    ///
390    /// 返回：无；重复关闭没有额外副作用。
391    pub fn close(&self) {
392        self.state.accepting.store(false, Ordering::Release);
393    }
394    /// 业务作用：交出只持有原子单元的统一指标源。
395    ///
396    /// 参数说明：无。
397    ///
398    /// 返回：与生产端、worker 同源的指标句柄。
399    pub fn metrics(&self) -> Arc<QueueMetrics> {
400        self.state.metrics.clone()
401    }
402}
403
404/// 消费端唯一拥有的待投递消息。
405pub struct Envelope {
406    /// 启动期冻结的 provider 目录索引。
407    pub provider: usize,
408    /// 已施加文本边界、由 worker 独占投递责任的通知。
409    pub notification: Notification,
410    queued: bool,
411    metrics: Arc<QueueMetrics>,
412}
413
414impl Drop for Envelope {
415    /// 业务作用：即使关闭与已预留生产者并发，也为未被领取的消息回收深度并记录丢弃。
416    ///
417    /// 参数说明：无。
418    ///
419    /// 返回：无；已被 worker 领取的消息不重复计数。
420    fn drop(&mut self) {
421        if self.queued {
422            self.metrics.depth.fetch_sub(1, Ordering::Relaxed);
423            add(
424                &self.metrics.providers[self.provider].dropped
425                    [self.notification.fields().event as usize][2],
426                1,
427            );
428        }
429    }
430}
431
432/// 单消费者接收权，宿主可在不销毁旧 route 的情况下关闭并排空。
433pub struct NotificationReceiver {
434    receiver: mpsc::Receiver<Envelope>,
435    state: Arc<QueueState>,
436}
437
438impl NotificationReceiver {
439    /// 业务作用：取走一条已接受消息并从等待深度中扣除。
440    ///
441    /// 参数说明：无。
442    ///
443    /// 返回：消息或已关闭且排空的 None。
444    pub async fn recv(&mut self) -> Option<Envelope> {
445        let mut message = self.receiver.recv().await;
446        if let Some(message) = &mut message {
447            message.queued = false;
448            self.state.metrics.depth.fetch_sub(1, Ordering::Relaxed);
449        }
450        message
451    }
452    /// 业务作用：关闭生产与接收入口，保留现有消息供有限排空。
453    ///
454    /// 参数说明：无。
455    ///
456    /// 返回：无；关闭后不再接受新通知。
457    pub fn close(&mut self) {
458        self.state.accepting.store(false, Ordering::Release);
459        self.receiver.close();
460    }
461    /// 业务作用：在停机预算耗尽时计数并释放尚未投递的消息。
462    ///
463    /// 参数说明：无。
464    ///
465    /// 返回：本次放弃的消息数。
466    pub fn discard_remaining(&mut self) -> usize {
467        self.close();
468        let mut count = 0;
469        while let Ok(message) = self.receiver.try_recv() {
470            drop(message);
471            count += 1;
472        }
473        count
474    }
475}
476
477impl Drop for NotificationReceiver {
478    /// 业务作用：异常退出也撤销旧路由并登记尚未消费的通知，避免悬空生产端。
479    ///
480    /// 参数说明：无。
481    ///
482    /// 返回：无；只进行有限队列清理和原子计数。
483    fn drop(&mut self) {
484        self.discard_remaining();
485    }
486}
487
488/// 只负责有界内存队列的工厂，不创建任务、线程或网络客户端。
489pub struct NotificationQueue;
490
491impl NotificationQueue {
492    /// 业务作用：按已引用 provider 的冻结目录预分配通知指标和有界队列。
493    ///
494    /// 参数说明：
495    /// - `provider_ids`：为唯一合法标识
496    /// - `capacity`：为 1..=65536 的队列容量。
497    ///
498    /// 返回：生产端与唯一接收端；非法目录或容量不分配运行资源。
499    pub fn bounded(
500        provider_ids: Vec<String>,
501        capacity: usize,
502    ) -> Result<(NotificationProducer, NotificationReceiver), ConfigError> {
503        if !(1..=65536).contains(&capacity) || provider_ids.is_empty() || provider_ids.len() > 32 {
504            return Err(ConfigError("queue"));
505        }
506        let mut seen = BTreeSet::new();
507        if provider_ids
508            .iter()
509            .any(|id| !valid_provider_id(id) || !seen.insert(id))
510        {
511            return Err(ConfigError("providers"));
512        }
513        let providers = provider_ids
514            .into_iter()
515            .map(|id| ProviderMetrics {
516                id,
517                enqueued: std::array::from_fn(|_| AtomicU64::new(0)),
518                dropped: std::array::from_fn(|_| std::array::from_fn(|_| AtomicU64::new(0))),
519                deliveries: std::array::from_fn(|_| std::array::from_fn(|_| AtomicU64::new(0))),
520                duration: std::array::from_fn(|_| AtomicHistogram::new(LATENCY_SECONDS)),
521            })
522            .collect();
523        let state = Arc::new(QueueState {
524            accepting: AtomicBool::new(true),
525            metrics: Arc::new(QueueMetrics {
526                providers,
527                capacity,
528                depth: AtomicU64::new(0),
529            }),
530        });
531        let (sender, receiver) = mpsc::channel(capacity);
532        Ok((
533            NotificationProducer {
534                sender,
535                state: state.clone(),
536            },
537            NotificationReceiver {
538                receiver,
539                state,
540            },
541        ))
542    }
543}
544
545/// 业务作用：限制 provider 目录和指标标签的字符集与长度。
546///
547/// 参数说明：
548/// - `id`：是普通配置中的渠道标识，不是渠道凭据。
549///
550/// 返回：仅当满足字母开头的 1..=64 字节 ASCII 标识时为真。
551pub fn valid_provider_id(id: &str) -> bool {
552    (1..=64).contains(&id.len())
553        && id.as_bytes()[0].is_ascii_alphabetic()
554        && id
555            .bytes()
556            .all(|ch| ch.is_ascii_alphanumeric() || matches!(ch, b'_' | b'.' | b'-'))
557}
nanotify_core/queue.rs

nanotify_core/
queue.rs
