helix_im/
diagnostics.rs

1//! 业务事实的只读旁路；不分配Effect、不更改业务状态、不读取时钟或执行I/O。
2use crate::module::ImModule;
3
4#[derive(Default)]
5pub(crate) struct Session {
6    pub login: String,
7    pub device: String,
8    pub now_ms: u64,
9    next_gap: u64,
10    gaps: std::collections::BTreeMap<crate::state::ChannelId, Gap>,
11}
12struct Gap {
13    id: u64,
14    from: u64,
15    through: u64,
16    started_ms: u64,
17}
18#[derive(Default)]
19pub(crate) struct Observation<'a> {
20    pub event: &'static str,
21    pub domain: &'static str,
22    pub stage: &'static str,
23    pub path: &'static str,
24    pub result: &'static str,
25    pub reason: &'static str,
26    pub channel: &'a str,
27    pub business_event_id: &'a str,
28    pub traceparent: &'a str,
29    pub batch_id: &'a str,
30    pub seq: Option<u64>,
31    pub from: Option<u64>,
32    pub next: Option<u64>,
33    pub cursor: Option<u64>,
34    pub expected: Option<u64>,
35    pub received: Option<u64>,
36    pub declared_from: Option<u64>,
37    pub declared_to: Option<u64>,
38    pub count: usize,
39    pub corr: Option<u64>,
40    pub has_more: bool,
41    pub gap: Option<u64>,
42    pub elapsed: f64,
43    pub sync_session_id: &'a str,
44    pub pending_operations: Option<u64>,
45    pub barrier_mask: Option<u64>,
46    pub scheduler_inflight: Option<u64>,
47    pub scheduler_pending: Option<u64>,
48    pub batch_persist_inflight: Option<u64>,
49    pub page_index: Option<u64>,
50    pub total_pages: Option<u64>,
51    pub started_at_ms: Option<u64>,
52    pub completed_at_ms: Option<u64>,
53}
54struct OptionalNumber(Option<u64>);
55impl std::fmt::Display for OptionalNumber {
56    /// 缺失值保持空字段，由后台省略；不伪填序号0。
57    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
58        match self.0 {
59            Some(v) => write!(f, "{v}"),
60            None => Ok(()),
61        }
62    }
63}
64impl ImModule {
65    /// 宿主在注册模块前注入诊断会话；不改变认证身份或业务协议。
66    pub fn with_diagnostic_session(mut self, login: String, device: String) -> Self {
67        self.diagnostics = Session {
68            login,
69            device,
70            ..Default::default()
71        };
72        self
73    }
74    /// 仅从已认证config取身份；固定标量借用，无正文clone/JSON/额外存储扫描。
75    pub(crate) fn diagnose(&self, record: Observation<'_>) {
76        if self.diagnostics.login.is_empty() {
77            return;
78        }
79        tracing::info!(target:"helix::diagnostics",
80            sync_session_id=record.sync_session_id,page_index=%OptionalNumber(record.page_index),total_pages=%OptionalNumber(record.total_pages),
81            started_at_ms=%OptionalNumber(record.started_at_ms),completed_at_ms=%OptionalNumber(record.completed_at_ms),
82            event=record.event,tenant_id=self.config.company_id.as_str(),user_id=self.config.auth_user_id.as_str(),
83            login_attempt_id=self.diagnostics.login.as_str(),device_session_id=self.diagnostics.device.as_str(),
84            batch_id=record.batch_id,business_event_id=record.business_event_id,traceparent=record.traceparent,channel_id=record.channel,stage=record.stage,path=record.path,result=record.result,reason=record.reason,
85            seq_domain=if record.domain.is_empty() {"channel_cursor"} else {record.domain},event_seq=%OptionalNumber(record.seq),from_seq=%OptionalNumber(record.from),
86            next_seq=%OptionalNumber(record.next),local_contiguous_seq=%OptionalNumber(record.cursor),
87            expected_seq=%OptionalNumber(record.expected),received_seq=%OptionalNumber(record.received),
88            declared_from_seq=%OptionalNumber(record.declared_from),declared_to_seq=%OptionalNumber(record.declared_to),
89            corr=%OptionalNumber(record.corr),count=record.count as u64,has_more=record.has_more,expected_known=false,evidence_complete=false,
90            gap_id=%GapKey{session:self.diagnostics.login.as_str(),channel:record.channel,id:record.gap},elapsed_seconds=record.elapsed,
91            pending_operations=%OptionalNumber(record.pending_operations),barrier_mask=%OptionalNumber(record.barrier_mask),
92            scheduler_inflight=%OptionalNumber(record.scheduler_inflight),scheduler_pending=%OptionalNumber(record.scheduler_pending),
93            batch_persist_inflight=%OptionalNumber(record.batch_persist_inflight)
94        );
95    }
96    /// 只读取已存在的单群cursor，表示本次观察的已提交边界，不计算max(seq)。
97    pub(crate) fn diagnose_checkpoint(
98        &mut self,
99        channel: crate::state::ChannelId,
100        source: &'static str,
101    ) {
102        if let Some(cursor) = self
103            .state
104            .channels
105            .get(&channel)
106            .map(|state| state.cursor.value().0)
107        {
108            self.diagnose(Observation {
109                event: "channel_checkpoint",
110                stage: source,
111                channel: channel.as_str(),
112                cursor: Some(cursor),
113                ..Default::default()
114            });
115            if self
116                .diagnostics
117                .gaps
118                .get(&channel)
119                .is_some_and(|gap| cursor >= gap.through)
120            {
121                if let Some(gap) = self.diagnostics.gaps.remove(&channel) {
122                    self.diagnose(Observation {
123                        event: "gap_terminal",
124                        result: "cursor_caught_up",
125                        reason: "requires_persistence_reconciliation",
126                        channel: channel.as_str(),
127                        gap: Some(gap.id),
128                        expected: Some(gap.from),
129                        received: gap.through.checked_add(1),
130                        cursor: Some(cursor),
131                        elapsed: self.diagnostics.now_ms.saturating_sub(gap.started_ms) as f64
132                            / 1000.0,
133                        ..Default::default()
134                    });
135                }
136            }
137        }
138    }
139    /// Episode仅用于观测；有界映射不参与补偿或提交决策，满时显式报告截断。
140    fn diagnose_gap(
141        &mut self,
142        channel: crate::state::ChannelId,
143        from: u64,
144        through: u64,
145        reason: &'static str,
146    ) {
147        if from > through {
148            return;
149        }
150        if self.diagnostics.gaps.len() >= 1024 && !self.diagnostics.gaps.contains_key(&channel) {
151            self.diagnose(Observation {
152                event: "diagnostic_truncated",
153                reason: "capacity_limit",
154                count: 1,
155                ..Default::default()
156            });
157            return;
158        }
159        let existed = self.diagnostics.gaps.contains_key(&channel);
160        if !existed {
161            self.diagnostics.next_gap += 1;
162            self.diagnostics.gaps.insert(
163                channel,
164                Gap {
165                    id: self.diagnostics.next_gap,
166                    from,
167                    through,
168                    started_ms: self.diagnostics.now_ms,
169                },
170            );
171        }
172        let Some(gap) = self.diagnostics.gaps.get_mut(&channel) else {
173            return;
174        };
175        gap.through = gap.through.max(through);
176        gap.from = gap.from.min(from);
177        let (id, from, through) = (gap.id, gap.from, gap.through);
178        self.diagnose(Observation {
179            event: if existed { "gap_updated" } else { "gap_opened" },
180            reason,
181            channel: channel.as_str(),
182            gap: Some(id),
183            expected: Some(from),
184            received: through.checked_add(1),
185            ..Default::default()
186        });
187    }
188    /// 已解析WS只记录标量，接收与业务接受/提交分开；pong区间原样保留不更改补偿。
189    pub(crate) fn diagnose_inbound(&mut self, frame: &crate::ws::WsFrame) {
190        if self.diagnostics.login.is_empty() {
191            return;
192        }
193        let data = frame.data();
194        let channel = data
195            .and_then(|v| {
196                v.get("channelId")
197                    .or_else(|| v.get("channel_id"))
198                    .or_else(|| {
199                        (frame.action().ok() == Some("increment_channel"))
200                            .then(|| v.get("id"))
201                            .flatten()
202                    })
203                    .or_else(|| {
204                        v.get("post")
205                            .and_then(|p| p.get("channel_id").or_else(|| p.get("channelId")))
206                    })
207            })
208            .and_then(serde_json::Value::as_str)
209            .unwrap_or("");
210        let seq = frame.event_seq().map(|value| value.0);
211        if matches!(
212            frame.action().ok(),
213            Some("increment_channel" | "increment_channel_end")
214        ) {
215            self.diagnose(Observation {
216                event: if frame.action().ok() == Some("increment_channel") {
217                    "channel_inventory_item_received"
218                } else {
219                    "channel_inventory_end_received"
220                },
221                stage: "ws_received",
222                result: "received",
223                channel,
224                batch_id: data
225                    .and_then(|value| value.get("batchId"))
226                    .and_then(serde_json::Value::as_str)
227                    .unwrap_or(""),
228                traceparent: frame
229                    .root()
230                    .get("tracing")
231                    .and_then(|value| value.get("traceparent"))
232                    .and_then(serde_json::Value::as_str)
233                    .unwrap_or(""),
234                ..Default::default()
235            });
236        }
237        let cursor = crate::state::ChannelId::from_str(channel)
238            .and_then(|id| self.state.channels.get(&id))
239            .map(|ch| ch.cursor.value().0);
240        if !channel.is_empty() && seq.is_some() {
241            self.diagnose(Observation {
242                domain: if data
243                    .is_some_and(|v| v.get("stream_seq").is_some() || v.get("streamSeq").is_some())
244                {
245                    "stream_seq"
246                } else {
247                    "legacy_event_seq"
248                },
249                event: "delivery_stage",
250                stage: "ws_received",
251                result: if cursor.zip(seq).is_some_and(|(c, s)| s <= c) {
252                    "already_committed"
253                } else {
254                    "received"
255                },
256                business_event_id: data
257                    .and_then(|d| d.get("event_id").or_else(|| d.get("eventId")))
258                    .and_then(serde_json::Value::as_str)
259                    .unwrap_or(""),
260                traceparent: frame
261                    .root()
262                    .get("tracing")
263                    .and_then(|v| v.get("traceparent"))
264                    .and_then(serde_json::Value::as_str)
265                    .unwrap_or(""),
266                path: "live_ws",
267                channel,
268                seq,
269                cursor,
270                ..Default::default()
271            });
272        }
273        if let (Some(seq), Some(cursor), Some(channel)) =
274            (seq, cursor, crate::state::ChannelId::from_str(channel))
275        {
276            if seq > cursor.saturating_add(1) {
277                self.diagnose_gap(
278                    channel,
279                    cursor.saturating_add(1),
280                    seq - 1,
281                    "pending_authority_check",
282                );
283            }
284        }
285        if let Some(gaps) = data
286            .and_then(|v| v.get("gaps"))
287            .and_then(serde_json::Value::as_array)
288        {
289            for gap in gaps.iter().take(128) {
290                let Some(channel) = gap.get("channelId").and_then(serde_json::Value::as_str) else {
291                    continue;
292                };
293                let number = |name| {
294                    gap.get(name).and_then(|v| {
295                        v.as_u64()
296                            .or_else(|| v.as_str().and_then(|s| s.parse().ok()))
297                    })
298                };
299                let from = number("fromSeq");
300                let to = number("maxSeq");
301                self.diagnose(Observation {
302                    event: "gap_observed",
303                    reason: "server_reported_range",
304                    path: "offline_recovery",
305                    channel,
306                    declared_from: from,
307                    declared_to: to,
308                    ..Default::default()
309                });
310            }
311            if gaps.len() > 128 {
312                self.diagnose(Observation {
313                    event: "diagnostic_truncated",
314                    reason: "capacity_limit",
315                    count: gaps.len() - 128,
316                    ..Default::default()
317                });
318            }
319        }
320    }
321}
322
323struct GapKey<'a> {
324    session: &'a str,
325    channel: &'a str,
326    id: Option<u64>,
327}
328impl std::fmt::Display for GapKey<'_> {
329    /// Session与群限定episode身份；不放入任何时序标签。
330    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
331        match self.id {
332            Some(id) => write!(f, "{}:{}:{id}", self.session, self.channel),
333            None => Ok(()),
334        }
335    }
336}
337
338#[cfg(test)]
339mod tests {
340    use super::*;
341    use crate::state::{ChannelId, Seq};
342    /// 精确大序号缺口只维护有界观测状态；游标追上必须单独记录且不改业务cursor。
343    #[test]
344    fn gap_episode_preserves_exact_range_and_waits_for_committed_cursor() {
345        let channel = ChannelId::from_str("chfixx00000000000000000001").unwrap();
346        let mut module = ImModule::new(Default::default())
347            .with_diagnostic_session("login".into(), "device".into());
348        module.register_channel(channel, 9007199254740992);
349        module.diagnostics.now_ms = 100;
350        let frame=crate::ws::WsFrame::parse(br#"{"action":"post","data":{"channelId":"chfixx00000000000000000001","streamSeq":9007199254740995}}"#).unwrap();
351        module.diagnose_inbound(&frame);
352        let gap = module.diagnostics.gaps.get(&channel).unwrap();
353        assert_eq!(
354            (gap.from, gap.through),
355            (9007199254740993, 9007199254740994)
356        );
357        let id = gap.id;
358        module.diagnose_inbound(&frame);
359        assert_eq!(module.diagnostics.gaps.get(&channel).unwrap().id, id);
360        assert_eq!(module.cursor_for(channel), Some(9007199254740992));
361        module.diagnose_checkpoint(channel, "test_readonly");
362        assert_eq!(module.diagnostics.gaps.len(), 1);
363        module
364            .state
365            .channels
366            .get_mut(&channel)
367            .unwrap()
368            .cursor
369            .try_advance(Seq(9007199254740995));
370        module.diagnose_checkpoint(channel, "test_committed");
371        assert!(module.diagnostics.gaps.is_empty());
372    }
373}
helix_im/diagnostics.rs

helix_im/
diagnostics.rs
