use crate::module::ImModule;
#[derive(Default)]
pub(crate) struct Session {
pub login: String,
pub device: String,
pub now_ms: u64,
next_gap: u64,
gaps: std::collections::BTreeMap<crate::state::ChannelId, Gap>,
}
struct Gap {
id: u64,
from: u64,
through: u64,
started_ms: u64,
}
#[derive(Default)]
pub(crate) struct Observation<'a> {
pub event: &'static str,
pub domain: &'static str,
pub stage: &'static str,
pub path: &'static str,
pub result: &'static str,
pub reason: &'static str,
pub channel: &'a str,
pub business_event_id: &'a str,
pub traceparent: &'a str,
pub batch_id: &'a str,
pub seq: Option<u64>,
pub from: Option<u64>,
pub next: Option<u64>,
pub cursor: Option<u64>,
pub expected: Option<u64>,
pub received: Option<u64>,
pub declared_from: Option<u64>,
pub declared_to: Option<u64>,
pub count: usize,
pub corr: Option<u64>,
pub has_more: bool,
pub gap: Option<u64>,
pub elapsed: f64,
pub sync_session_id: &'a str,
pub pending_operations: Option<u64>,
pub barrier_mask: Option<u64>,
pub scheduler_inflight: Option<u64>,
pub scheduler_pending: Option<u64>,
pub batch_persist_inflight: Option<u64>,
pub page_index: Option<u64>,
pub total_pages: Option<u64>,
pub started_at_ms: Option<u64>,
pub completed_at_ms: Option<u64>,
}
struct OptionalNumber(Option<u64>);
impl std::fmt::Display for OptionalNumber {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self.0 {
Some(v) => write!(f, "{v}"),
None => Ok(()),
}
}
}
impl ImModule {
pub fn with_diagnostic_session(mut self, login: String, device: String) -> Self {
self.diagnostics = Session {
login,
device,
..Default::default()
};
self
}
pub(crate) fn diagnose(&self, record: Observation<'_>) {
if self.diagnostics.login.is_empty() {
return;
}
tracing::info!(target:"helix::diagnostics",
sync_session_id=record.sync_session_id,page_index=%OptionalNumber(record.page_index),total_pages=%OptionalNumber(record.total_pages),
started_at_ms=%OptionalNumber(record.started_at_ms),completed_at_ms=%OptionalNumber(record.completed_at_ms),
event=record.event,tenant_id=self.config.company_id.as_str(),user_id=self.config.auth_user_id.as_str(),
login_attempt_id=self.diagnostics.login.as_str(),device_session_id=self.diagnostics.device.as_str(),
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,
seq_domain=if record.domain.is_empty() {"channel_cursor"} else {record.domain},event_seq=%OptionalNumber(record.seq),from_seq=%OptionalNumber(record.from),
next_seq=%OptionalNumber(record.next),local_contiguous_seq=%OptionalNumber(record.cursor),
expected_seq=%OptionalNumber(record.expected),received_seq=%OptionalNumber(record.received),
declared_from_seq=%OptionalNumber(record.declared_from),declared_to_seq=%OptionalNumber(record.declared_to),
corr=%OptionalNumber(record.corr),count=record.count as u64,has_more=record.has_more,expected_known=false,evidence_complete=false,
gap_id=%GapKey{session:self.diagnostics.login.as_str(),channel:record.channel,id:record.gap},elapsed_seconds=record.elapsed,
pending_operations=%OptionalNumber(record.pending_operations),barrier_mask=%OptionalNumber(record.barrier_mask),
scheduler_inflight=%OptionalNumber(record.scheduler_inflight),scheduler_pending=%OptionalNumber(record.scheduler_pending),
batch_persist_inflight=%OptionalNumber(record.batch_persist_inflight)
);
}
pub(crate) fn diagnose_checkpoint(
&mut self,
channel: crate::state::ChannelId,
source: &'static str,
) {
if let Some(cursor) = self
.state
.channels
.get(&channel)
.map(|state| state.cursor.value().0)
{
self.diagnose(Observation {
event: "channel_checkpoint",
stage: source,
channel: channel.as_str(),
cursor: Some(cursor),
..Default::default()
});
if self
.diagnostics
.gaps
.get(&channel)
.is_some_and(|gap| cursor >= gap.through)
{
if let Some(gap) = self.diagnostics.gaps.remove(&channel) {
self.diagnose(Observation {
event: "gap_terminal",
result: "cursor_caught_up",
reason: "requires_persistence_reconciliation",
channel: channel.as_str(),
gap: Some(gap.id),
expected: Some(gap.from),
received: gap.through.checked_add(1),
cursor: Some(cursor),
elapsed: self.diagnostics.now_ms.saturating_sub(gap.started_ms) as f64
/ 1000.0,
..Default::default()
});
}
}
}
}
fn diagnose_gap(
&mut self,
channel: crate::state::ChannelId,
from: u64,
through: u64,
reason: &'static str,
) {
if from > through {
return;
}
if self.diagnostics.gaps.len() >= 1024 && !self.diagnostics.gaps.contains_key(&channel) {
self.diagnose(Observation {
event: "diagnostic_truncated",
reason: "capacity_limit",
count: 1,
..Default::default()
});
return;
}
let existed = self.diagnostics.gaps.contains_key(&channel);
if !existed {
self.diagnostics.next_gap += 1;
self.diagnostics.gaps.insert(
channel,
Gap {
id: self.diagnostics.next_gap,
from,
through,
started_ms: self.diagnostics.now_ms,
},
);
}
let Some(gap) = self.diagnostics.gaps.get_mut(&channel) else {
return;
};
gap.through = gap.through.max(through);
gap.from = gap.from.min(from);
let (id, from, through) = (gap.id, gap.from, gap.through);
self.diagnose(Observation {
event: if existed { "gap_updated" } else { "gap_opened" },
reason,
channel: channel.as_str(),
gap: Some(id),
expected: Some(from),
received: through.checked_add(1),
..Default::default()
});
}
pub(crate) fn diagnose_inbound(&mut self, frame: &crate::ws::WsFrame) {
if self.diagnostics.login.is_empty() {
return;
}
let data = frame.data();
let channel = data
.and_then(|v| {
v.get("channelId")
.or_else(|| v.get("channel_id"))
.or_else(|| {
(frame.action().ok() == Some("increment_channel"))
.then(|| v.get("id"))
.flatten()
})
.or_else(|| {
v.get("post")
.and_then(|p| p.get("channel_id").or_else(|| p.get("channelId")))
})
})
.and_then(serde_json::Value::as_str)
.unwrap_or("");
let seq = frame.event_seq().map(|value| value.0);
if matches!(
frame.action().ok(),
Some("increment_channel" | "increment_channel_end")
) {
self.diagnose(Observation {
event: if frame.action().ok() == Some("increment_channel") {
"channel_inventory_item_received"
} else {
"channel_inventory_end_received"
},
stage: "ws_received",
result: "received",
channel,
batch_id: data
.and_then(|value| value.get("batchId"))
.and_then(serde_json::Value::as_str)
.unwrap_or(""),
traceparent: frame
.root()
.get("tracing")
.and_then(|value| value.get("traceparent"))
.and_then(serde_json::Value::as_str)
.unwrap_or(""),
..Default::default()
});
}
let cursor = crate::state::ChannelId::from_str(channel)
.and_then(|id| self.state.channels.get(&id))
.map(|ch| ch.cursor.value().0);
if !channel.is_empty() && seq.is_some() {
self.diagnose(Observation {
domain: if data
.is_some_and(|v| v.get("stream_seq").is_some() || v.get("streamSeq").is_some())
{
"stream_seq"
} else {
"legacy_event_seq"
},
event: "delivery_stage",
stage: "ws_received",
result: if cursor.zip(seq).is_some_and(|(c, s)| s <= c) {
"already_committed"
} else {
"received"
},
business_event_id: data
.and_then(|d| d.get("event_id").or_else(|| d.get("eventId")))
.and_then(serde_json::Value::as_str)
.unwrap_or(""),
traceparent: frame
.root()
.get("tracing")
.and_then(|v| v.get("traceparent"))
.and_then(serde_json::Value::as_str)
.unwrap_or(""),
path: "live_ws",
channel,
seq,
cursor,
..Default::default()
});
}
if let (Some(seq), Some(cursor), Some(channel)) =
(seq, cursor, crate::state::ChannelId::from_str(channel))
{
if seq > cursor.saturating_add(1) {
self.diagnose_gap(
channel,
cursor.saturating_add(1),
seq - 1,
"pending_authority_check",
);
}
}
if let Some(gaps) = data
.and_then(|v| v.get("gaps"))
.and_then(serde_json::Value::as_array)
{
for gap in gaps.iter().take(128) {
let Some(channel) = gap.get("channelId").and_then(serde_json::Value::as_str) else {
continue;
};
let number = |name| {
gap.get(name).and_then(|v| {
v.as_u64()
.or_else(|| v.as_str().and_then(|s| s.parse().ok()))
})
};
let from = number("fromSeq");
let to = number("maxSeq");
self.diagnose(Observation {
event: "gap_observed",
reason: "server_reported_range",
path: "offline_recovery",
channel,
declared_from: from,
declared_to: to,
..Default::default()
});
}
if gaps.len() > 128 {
self.diagnose(Observation {
event: "diagnostic_truncated",
reason: "capacity_limit",
count: gaps.len() - 128,
..Default::default()
});
}
}
}
}
struct GapKey<'a> {
session: &'a str,
channel: &'a str,
id: Option<u64>,
}
impl std::fmt::Display for GapKey<'_> {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self.id {
Some(id) => write!(f, "{}:{}:{id}", self.session, self.channel),
None => Ok(()),
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::state::{ChannelId, Seq};
#[test]
fn gap_episode_preserves_exact_range_and_waits_for_committed_cursor() {
let channel = ChannelId::from_str("chfixx00000000000000000001").unwrap();
let mut module = ImModule::new(Default::default())
.with_diagnostic_session("login".into(), "device".into());
module.register_channel(channel, 9007199254740992);
module.diagnostics.now_ms = 100;
let frame=crate::ws::WsFrame::parse(br#"{"action":"post","data":{"channelId":"chfixx00000000000000000001","streamSeq":9007199254740995}}"#).unwrap();
module.diagnose_inbound(&frame);
let gap = module.diagnostics.gaps.get(&channel).unwrap();
assert_eq!(
(gap.from, gap.through),
(9007199254740993, 9007199254740994)
);
let id = gap.id;
module.diagnose_inbound(&frame);
assert_eq!(module.diagnostics.gaps.get(&channel).unwrap().id, id);
assert_eq!(module.cursor_for(channel), Some(9007199254740992));
module.diagnose_checkpoint(channel, "test_readonly");
assert_eq!(module.diagnostics.gaps.len(), 1);
module
.state
.channels
.get_mut(&channel)
.unwrap()
.cursor
.try_advance(Seq(9007199254740995));
module.diagnose_checkpoint(channel, "test_committed");
assert!(module.diagnostics.gaps.is_empty());
}
}