1use std::time::Duration;
12
13use crate::{Error, Result};
14use zenkey::grammar::with_base;
15use zenkey::{Declared, RegistrySlice};
16
17use crate::bus::query::{Answer, GetOpts, RepeatingRegistry, fleet_get, state_snapshot};
18use crate::judge::common::{FINDING_CAP, is_synthetic_marker};
19use crate::model::examples::Examples;
20use crate::report::{CheckId, DoctorFinding, DoctorReport, DoctorSeverity, DriftVerdict};
21
22#[derive(Debug, Clone)]
24pub struct DoctorSpec {
25 pub deep: bool,
28 pub sample: Option<usize>,
32 pub timeout: Duration,
34 pub listen: Option<Duration>,
40}
41
42fn finding(
43 severity: DoctorSeverity,
44 check: CheckId,
45 subject: impl Into<String>,
46 evidence: impl Into<String>,
47 citation: Option<&str>,
48) -> DoctorFinding {
49 DoctorFinding {
50 severity,
51 check,
52 subject: subject.into(),
53 evidence: evidence.into(),
54 citation: citation.map(str::to_string),
55 }
56}
57
58fn rpc_key(base: &str, slice: &RegistrySlice, procedure: &str) -> Result<String> {
62 Ok(match &slice.service_origin {
63 Some(origin) => {
64 let o = origin.known().ok_or_else(|| {
70 Error::malformed(
71 format!("slice {}", slice.name),
72 format!("carries {:?} as a service origin", origin.token()),
73 )
74 })?;
75 with_base(base, zenkey::selector::service_rpc(o, &[procedure]))
76 }
77 None => with_base(base, zenkey::selector::fleet_rpc(&slice.name, &[procedure])),
78 })
79}
80
81pub async fn run_doctor(
94 fleet: &crate::Fleet<'_>,
95 locals: Option<&crate::model::registry::SliceSet>,
96 spec: &DoctorSpec,
97) -> Result<DoctorReport> {
98 let (session, base) = (fleet.session(), fleet.base());
99
100 let locals = locals.filter(|set| !set.slices().is_empty());
104 let roster = crate::bus::roster::roster(fleet, spec.timeout).await?;
105
106 let mut findings: Vec<DoctorFinding> = Vec::new();
107 let mut synced: Vec<String> = Vec::new();
108 let mut answered = 0usize;
109
110 for local in locals.iter().flat_map(|set| set.slices()) {
112 let key = rpc_key(base, local, "introspect")?;
113 let answers = fleet_get(fleet, &key, &GetOpts::new(spec.timeout)).await?;
114 for answer in &answers {
115 let Answer::Value(bytes) = &answer.answer else {
116 continue;
117 };
118 answered += 1;
119 let served_toml = bytes.to_bytes();
120 let served_toml = String::from_utf8_lossy(&served_toml);
121 let served = match zenkey::parse_slice(&served_toml) {
122 Ok(s) => s,
123 Err(e) => {
124 findings.push(finding(
125 DoctorSeverity::Error,
126 CheckId::SliceParse,
127 format!("{}/{}", answer.origin, local.name),
128 format!("served slice does not parse: {e}"),
129 Some("RFC 08 §6"),
130 ));
131 continue;
132 }
133 };
134 let diff = zenkey::slice::diff(&served, local);
135 if diff.is_empty() {
136 synced.push(format!(
137 "{}/{} (registry {})",
138 answer.origin, local.name, served.version
139 ));
140 } else {
141 for f in &diff {
142 findings.push(finding(
143 DoctorSeverity::Error,
144 CheckId::SliceSync,
145 format!("{}/{}", answer.origin, local.name),
146 f.summary(),
147 Some("RFC 08 §6"),
148 ));
149 }
150 }
151 }
152 }
153
154 let sweep = if locals.is_none() {
157 let repeating = RepeatingRegistry::declare(fleet, spec.timeout).await?;
158 let slices: Vec<RegistrySlice> = repeating
159 .fetch()
160 .await?
161 .into_iter()
162 .map(|(s, _)| s)
163 .collect();
164 repeating.undeclare().await?;
165 Some(slices)
166 } else {
167 None
168 };
169
170 if let Some(slices) = &sweep {
173 answered = slices.len();
174 }
175
176 let live: usize = roster.values().map(Vec::len).sum();
184 findings.extend(judge_introspect_coverage(
185 &roster,
186 locals.map(crate::model::registry::SliceSet::slices),
187 answered,
188 ));
189
190 let routers = crate::routers(session, spec.timeout)
192 .await
193 .unwrap_or_default();
194 let mut router_version = None;
195 if routers.is_empty() {
196 findings.push(finding(
197 DoctorSeverity::Info,
198 CheckId::AdminUnreachable,
199 "mesh",
200 "no routers answered @/*/router (peer-only mesh, or the admin space is \
201 disabled) — storage/version checks skipped",
202 None,
203 ));
204 } else {
205 let versions: std::collections::BTreeSet<&str> = routers
206 .iter()
207 .filter_map(|r| r.version.as_deref())
208 .collect();
209 if versions.len() > 1 {
210 findings.push(finding(
211 DoctorSeverity::Error,
212 CheckId::RouterVersionSkew,
213 "mesh",
214 format!("router version skew across the mesh: {versions:?}"),
215 None,
216 ));
217 } else {
218 router_version = versions.iter().next().map(|v| v.to_string());
219 }
220 }
221
222 let slice_set: std::borrow::Cow<'_, crate::model::registry::SliceSet> = match sweep {
226 Some(slices) => {
227 std::borrow::Cow::Owned(crate::model::registry::SliceSet::from_slices(slices))
228 }
229 None => match locals {
232 Some(set) => std::borrow::Cow::Borrowed(set),
233 None => std::borrow::Cow::Owned(crate::model::registry::SliceSet::default()),
234 },
235 };
236 let mut described: Vec<(String, zenkey::schema::SchemaSet)> = Vec::new();
237 let mut undescribed = 0usize;
238 for slice in slice_set.slices() {
239 let key = rpc_key(base, slice, "describe")?;
240 let answers = fleet_get(fleet, &key, &GetOpts::new(spec.timeout)).await?;
241 let set = answers.into_iter().find_map(|a| match a.answer {
242 Answer::Value(bytes) => {
243 let cow = bytes.to_bytes();
244 std::str::from_utf8(&cow)
245 .ok()
246 .and_then(|t| zenkey::schema::SchemaSet::parse(t).ok())
247 }
248 Answer::Error { .. } => None,
249 });
250 match set {
251 Some(set) => described.push((slice.name.clone(), set)),
252 None => undescribed += 1,
253 }
254 }
255 for gap in crate::model::decode::totality_gaps(&described, &slice_set) {
258 findings.push(finding(
259 DoctorSeverity::Error,
260 CheckId::DescribeTotality,
261 gap.producer.clone(),
262 format!(
263 "describe is not total — missing: {}",
264 gap.missing.join(", ")
265 ),
266 Some("RFC 08 §7"),
267 ));
268 }
269 for drift in crate::model::decode::schema_drift(&described) {
270 let servers: Vec<String> = drift
271 .servers
272 .iter()
273 .map(|s| match s.hash.as_option() {
274 Some(h) => format!("{} ({h})", s.producer),
275 None => format!("{} (no identity served)", s.producer),
276 })
277 .collect();
278 let (severity, evidence) = match drift.verdict {
283 DriftVerdict::Disagree => (
284 DoctorSeverity::Error,
285 format!("served with different schemas by {}", servers.join(", ")),
286 ),
287 DriftVerdict::Unjudgeable => (
288 DoctorSeverity::Warning,
289 format!(
290 "agreement cannot be judged — {} served no schema identity: {} \
291 (RFC 09 §5.1 O4; the hash exists for exactly this, RFC 08 §7)",
292 drift
293 .servers
294 .iter()
295 .filter(|s| s.hash.is_not_asked())
296 .count(),
297 servers.join(", ")
298 ),
299 ),
300 };
301 findings.push(finding(
302 severity,
303 CheckId::SchemaDrift,
304 drift.type_name.clone(),
305 evidence,
306 Some("RFC 08 §7"),
307 ));
308 }
309 if undescribed > 0 {
310 findings.push(finding(
311 DoctorSeverity::Info,
312 CheckId::DescribeMissing,
313 "fleet",
314 format!(
315 "{undescribed} producer(s) serve no describe (a SHOULD; generic tools \
316 render their payloads structurally)"
317 ),
318 Some("RFC 08 §7"),
319 ));
320 }
321
322 if spec.deep {
324 let now = std::time::SystemTime::now();
325 let mut unstamped = 0usize;
326 for slice in slice_set.slices() {
327 for subject in &slice.subjects {
328 let (Some(ttl), true) = (subject.ttl_s, subject.class.is(&zenkey::Class::State))
329 else {
330 continue;
331 };
332 let Ok(pattern) = zenkey::pattern::SubjectPattern::parse(&subject.path) else {
333 continue;
334 };
335 let selector = match &slice.service_origin {
336 Some(origin) => with_base(
337 base,
338 format!("v1/{origin}/state/{}", pattern.selector_tail()),
339 ),
340 None => with_base(
341 base,
342 format!("v1/*/state/{}/{}", slice.name, pattern.selector_tail()),
343 ),
344 };
345 let samples = state_snapshot(session, &selector, spec.timeout, spec.sample).await?;
346 let (family_findings, family_unstamped) = judge_state_samples(&samples, ttl, now);
347 findings.extend(family_findings);
348 unstamped += family_unstamped;
349 }
350 }
351 if unstamped > 0 {
352 findings.push(finding(
353 DoctorSeverity::Warning,
354 CheckId::UnstampedState,
355 "fleet",
356 format!(
357 "{unstamped} state sample(s) carry no HLC timestamp — the deployment \
358 lacks timestamping, which LWW requires; freshness is unjudgeable \
359 for them"
360 ),
361 Some("RFC 04 §4"),
362 ));
363 }
364 let storages = crate::storages(session, spec.timeout)
365 .await
366 .unwrap_or_default();
367 let coverage = crate::state_coverage(&slice_set, base, &storages);
368 let uncovered: Vec<&crate::CoverageRow> = coverage
369 .iter()
370 .filter(|r| r.coverage == crate::Coverage::Uncovered)
371 .collect();
372 if !uncovered.is_empty() {
373 findings.push(finding(
374 DoctorSeverity::Info,
375 CheckId::StorageCoverage,
376 "fleet",
377 format!(
378 "{} state famil(y|ies) have no storage coverage (volatile seeding \
379 may ride the advanced-pub/sub cache): {}",
380 uncovered.len(),
381 uncovered
382 .iter()
383 .map(|r| format!("{}/{}", r.producer, r.path))
384 .collect::<Vec<_>>()
385 .join(", ")
386 ),
387 Some("RFC 04 §3.5"),
388 ));
389 }
390 }
391
392 let observation = match spec.listen {
394 Some(window) => {
395 let store = crate::model::decode::SchemaStore::new(base, spec.timeout);
396 for (producer, set) in &described {
401 store.insert(producer, set.clone());
402 }
403 let _sealed = store.seal();
410 let (listen_findings, summary) =
411 observe_traffic(fleet, &slice_set, &store, &described, window).await?;
412 findings.extend(listen_findings);
413 Some(summary)
414 }
415 None => None,
416 };
417
418 Ok(DoctorReport {
419 findings,
420 synced: locals.is_some().then_some(synced).into(),
424 introspect_answered: answered,
425 live_producers: live,
426 describe_served: described.len(),
427 describe_missing: undescribed,
428 routers: routers.len(),
429 router_version,
430 deep: spec.deep,
431 observation,
432 })
433}
434
435const DECODE_BUDGET: u8 = 2;
438
439const SAME_FINDING: &str = "more key(s) with the same finding";
441
442fn emit_capped(
451 findings: &mut Vec<DoctorFinding>,
452 ex: Examples<DoctorFinding>,
453 check: CheckId,
454 tail: &str,
455) {
456 let more = ex.more(tail);
457
458 findings.extend(ex.into_vec());
459
460 if let Some(evidence) = more {
461 findings.push(finding(
462 DoctorSeverity::Info,
463 check,
464 "fleet",
465 evidence,
466 None,
467 ));
468 }
469}
470
471struct RateWindow {
474 cap: Option<u64>,
478 seen: u64,
480}
481
482async fn observe_traffic(
487 fleet: &crate::Fleet<'_>,
488 slices: &crate::model::registry::SliceSet,
489 store: &crate::model::decode::SchemaStore,
490 described: &[(String, zenkey::schema::SchemaSet)],
491 window: Duration,
492) -> Result<(Vec<DoctorFinding>, crate::report::ObservationSummary)> {
493 use std::collections::BTreeMap;
494
495 let (session, base) = (fleet.session(), fleet.base());
496
497 let scopes = crate::judge::common::data_plane_scopes(base, slices);
502
503 let monitor = crate::Monitor::start(session, crate::MonitorSpec::default()).await?;
504 let mut events = monitor.events();
505 let monitor = monitor.watching(&scopes).await?;
508 let started = tokio::time::Instant::now();
509 let deadline = started + window;
510
511 let mut fields =
515 crate::judge::field::FieldObservation::new(crate::judge::field::DEFAULT_MAX_PATHS);
516 let mut samples: u64 = 0;
517 let mut dropped: u64 = 0;
518 let mut synthetic: u64 = 0;
519 let mut facts_cache = crate::model::facts::FactsCache::default();
522 let mut decode_budget: BTreeMap<String, u8> = BTreeMap::new();
523 let mut unregistered: BTreeMap<String, u64> = BTreeMap::new();
525 let mut qos_bad: BTreeMap<String, (String, u64, u64)> = BTreeMap::new();
526 let mut undecodable: BTreeMap<String, (String, u64)> = BTreeMap::new();
527 let mut invalid: BTreeMap<String, (String, u64)> = BTreeMap::new();
528 let mut event_counts: BTreeMap<(String, String), RateWindow> = BTreeMap::new();
531 let mut foreign_stampers: BTreeMap<String, u64> = BTreeMap::new();
533
534 let window_over = tokio::time::sleep_until(deadline);
540 tokio::pin!(window_over);
541 loop {
542 let item = tokio::select! {
543 item = events.recv() => item,
544 () = &mut window_over => break,
545 };
546 match item {
547 Some(crate::StreamItem::Event(crate::FleetEvent::Sample(s))) => {
548 samples += 1;
549 if let Some(att) = &s.attachment
550 && is_synthetic_marker(&att.to_bytes())
551 {
552 synthetic += 1;
553 }
554 if let Some(crate::StampProvenance::Foreign { stamper }) = s.stamped_by {
559 *foreign_stampers.entry(stamper.to_string()).or_default() += 1;
560 }
561 let bytes = s.payload.to_bytes();
566 if bytes.len() > crate::model::decode::OBSERVE_LIMIT {
567 fields.observe_unread(&s.key);
568 } else {
569 let doc = crate::model::decode::structural_value(&bytes);
570 fields.observe(&s.key, started.elapsed().as_secs_f64(), doc.as_ref());
571 }
572 facts_cache.ensure(base, &s.key, Some(slices));
573 let facts = facts_cache.get(&s.key).expect("just ensured this key");
574 match &facts.registration {
575 crate::model::facts::Registration::Unregistered => {
576 *unregistered.entry(s.key.clone()).or_default() += 1;
577 }
578 crate::model::facts::Registration::Registered(sf) => {
579 if let (Some(profile), Some(declared)) = (sf.declared_qos(), &sf.qos) {
580 let entry = qos_bad
581 .entry(s.key.clone())
582 .or_insert_with(|| (declared.token().to_string(), 0, 0));
583 entry.2 += 1;
584 if !s.qos_matches(profile) {
585 entry.1 += 1;
586 }
587 }
588 if let (Some(rate), crate::model::facts::KeyShape::V1(v)) =
589 (&sf.rate, &facts.shape)
590 && v.class == "events"
591 {
592 let family = match &v.producer {
593 Some(p) => format!("{p}/{}", sf.path),
594 None => format!("{}/{}", v.origin, sf.path),
595 };
596 event_counts
602 .entry((family, rate.token()))
603 .or_insert_with(|| RateWindow {
604 cap: rate.cap_per_hour(),
605 seen: 0,
606 })
607 .seen += 1;
608 }
609 let budget = decode_budget.entry(s.key.clone()).or_default();
610 if *budget < DECODE_BUDGET {
611 *budget += 1;
612 let d = crate::model::decode::decode_sample(
620 fleet,
621 store,
622 Some(slices),
623 &s.key,
624 Some(&s.encoding),
625 &s.payload.to_bytes(),
626 )
627 .await;
628 match d.verdict {
629 crate::Verdict::NotValidated(
630 zenkey::schema::validate::NotValidated::Undecodable,
631 ) => {
632 let e = undecodable.entry(s.key.clone()).or_insert_with(|| {
633 (
634 d.decode_error
635 .unwrap_or_else(|| "does not decode".into()),
636 0,
637 )
638 });
639 e.1 += 1;
640 }
641 crate::Verdict::Invalid(errors) => {
642 let e = invalid
643 .entry(s.key.clone())
644 .or_insert_with(|| (errors.join("; "), 0));
645 e.1 += 1;
646 }
647 _ => {}
648 }
649 }
650 }
651 _ => {}
652 }
653 }
654 Some(crate::StreamItem::Dropped(n)) => dropped += n,
655 Some(_) => continue,
656 None => break,
657 }
658 }
659 let keys_seen = facts_cache.len();
660 monitor.shutdown().await?;
661
662 let budgets =
666 crate::judge::budget::BudgetObservation::observe(base, slices, facts_cache.keys());
667
668 let window_s = window.as_secs_f64();
669 let mut findings = Vec::new();
670
671 let mut ex = Examples::new(FINDING_CAP);
672 for (key, (error, n)) in &undecodable {
673 ex.push_with(|| {
674 finding(
675 DoctorSeverity::Error,
676 CheckId::PayloadUndecodable,
677 key.clone(),
678 format!(
679 "payload does not decode as its declared type: {error} ({n} sample(s) tried)"
680 ),
681 Some("RFC 08 §7"),
682 )
683 });
684 }
685 emit_capped(&mut findings, ex, CheckId::PayloadUndecodable, SAME_FINDING);
686 let mut ex = Examples::new(FINDING_CAP);
687 for (key, (violations, n)) in &invalid {
688 ex.push_with(|| {
689 finding(
690 DoctorSeverity::Error,
691 CheckId::PayloadInvalid,
692 key.clone(),
693 format!("payload violates the served schema: {violations} ({n} sample(s) tried)"),
694 Some("RFC 08 §7"),
695 )
696 });
697 }
698 emit_capped(&mut findings, ex, CheckId::PayloadInvalid, SAME_FINDING);
699 findings.extend(judge_qos_observed(&qos_bad));
700 if !foreign_stampers.is_empty() {
701 let mut named: Vec<String> = foreign_stampers
702 .iter()
703 .map(|(zid, n)| format!("{zid} ({n} sample(s))"))
704 .collect();
705 named.sort();
706 findings.push(finding(
707 DoctorSeverity::Info,
708 CheckId::TimestampStampedElsewhere,
709 "fleet".to_string(),
710 format!(
711 "HLCs on this bus are stamped by {} node(s) that are not the publishing \
712 session — a deployment with router-side timestamping, which is legal and \
713 common. Latency measured from these stamps is stamper→observer, not \
714 publisher→observer: {}",
715 foreign_stampers.len(),
716 named.join(", ")
717 ),
718 Some("RFC 09 §5.1 O7"),
719 ));
720 }
721 let mut ex = Examples::new(FINDING_CAP);
722 for (key, n) in &unregistered {
723 ex.push_with(|| {
724 finding(
725 DoctorSeverity::Warning,
726 CheckId::UnregisteredTraffic,
727 key.clone(),
728 format!(
729 "{n} sample(s) on a subject the producer's slice does not declare — \
730 for a conforming producer, a subject that is not registered does not exist"
731 ),
732 Some("RFC 08 §2"),
733 )
734 });
735 }
736 emit_capped(
737 &mut findings,
738 ex,
739 CheckId::UnregisteredTraffic,
740 SAME_FINDING,
741 );
742 if window <= Duration::from_secs(3600) {
746 for ((family, rate), RateWindow { cap, seen: count }) in &event_counts {
747 let Some(cap) = cap else {
748 continue;
749 };
750 if count > cap {
751 findings.push(finding(
752 DoctorSeverity::Warning,
753 CheckId::RateOverDeclared,
754 family.clone(),
755 format!(
756 "{count} event(s) in {window_s:.0}s exceeds the declared \
757 `{rate}` cap ({cap}/h)"
758 ),
759 Some("RFC 04 §1.3"),
760 ));
761 }
762 }
763 }
764
765 findings.extend(judge_cardinality(slices, &budgets, window_s));
766
767 let field_ctx = field_context_from(slices, described, &facts_cache);
771 findings.extend(crate::judge::field::judge_fields(
772 &fields, window_s, &field_ctx,
773 ));
774
775 Ok((
776 findings,
777 crate::report::ObservationSummary {
778 window_s,
779 scopes,
780 samples,
781 keys_seen,
782 dropped,
783 synthetic_marked: synthetic,
784 field_paths_dropped: fields.dropped_paths(),
787 facts_evicted: facts_cache.evicted(),
788 },
789 ))
790}
791
792fn field_context_from(
796 slices: &crate::model::registry::SliceSet,
797 described: &[(String, zenkey::schema::SchemaSet)],
798 facts: &crate::model::facts::FactsCache,
799) -> std::collections::BTreeMap<String, crate::judge::field::KeyFieldContext> {
800 use std::collections::BTreeMap;
801
802 let mut declared_cache: BTreeMap<(String, String), Option<crate::judge::field::DeclaredPaths>> =
803 BTreeMap::new();
804 let mut ctx = BTreeMap::new();
805 for (key, f) in facts.iter() {
806 let mut c = crate::judge::field::KeyFieldContext::default();
807 if let crate::model::facts::Registration::Registered(sf) = &f.registration {
808 c.ttl_s = sf.ttl_s;
809 c.type_name = Some(sf.type_name.clone());
810 if let Some(producer) = crate::judge::common::producer_of(f, Some(slices))
811 && !sf.type_name.is_empty()
812 {
813 let declared = declared_cache
814 .entry((producer.clone(), sf.type_name.clone()))
815 .or_insert_with(|| {
816 described
817 .iter()
818 .find(|(name, _)| *name == producer)
819 .and_then(|(_, set)| set.get(&sf.type_name))
820 .and_then(|schema| schema.json_document())
821 .and_then(crate::judge::field::DeclaredPaths::from_json_schema)
822 });
823 c.declared = declared.clone();
824 }
825 }
826 ctx.insert(key.to_string(), c);
827 }
828 ctx
829}
830
831fn judge_qos_observed(
841 qos_bad: &std::collections::BTreeMap<String, (String, u64, u64)>,
842) -> Vec<DoctorFinding> {
843 let mut findings = Vec::new();
844
845 let mut ex = Examples::new(FINDING_CAP);
846
847 for (key, (declared, bad, total)) in qos_bad.iter().filter(|(_, (_, bad, _))| *bad > 0) {
848 ex.push_with(|| {
849 finding(
850 DoctorSeverity::Warning,
851 CheckId::QosObservedMismatch,
852 key.clone(),
853 format!(
854 "{bad} of {total} sample(s) did not ride the declared {declared} — this \
855 is what actually rode: an interceptor MAY rewrite QoS, so it is a \
856 deviation, not proof of the publisher"
857 ),
858 Some("RFC 04 §3"),
859 )
860 });
861 }
862 emit_capped(
863 &mut findings,
864 ex,
865 CheckId::QosObservedMismatch,
866 SAME_FINDING,
867 );
868 findings
869}
870
871fn judge_introspect_coverage(
888 roster: &std::collections::BTreeMap<String, Vec<String>>,
889 locals: Option<&[RegistrySlice]>,
890 answered: usize,
891) -> Option<DoctorFinding> {
892 let live: usize = roster.values().map(Vec::len).sum();
893
894 let (in_scope, scope) = match locals {
895 None => (
896 live,
897 "scope: the whole roster (fleet-wide wildcard sweep)".to_string(),
898 ),
899 Some(locals) => {
900 let named = |origin: &str, producer: &str| {
901 let base_name = zenkey::grammar::Producer::parse_chunk(producer)
902 .map(|p| p.name().to_string())
903 .unwrap_or_else(|_| producer.to_string());
904 locals.iter().any(|l| {
905 l.name == base_name
906 || l.service_origin.as_ref().map(Declared::token) == Some(origin)
907 })
908 };
909 let in_scope: usize = roster
910 .iter()
911 .map(|(origin, producers)| producers.iter().filter(|p| named(origin, p)).count())
912 .sum();
913 let mut names: Vec<&str> = locals.iter().map(|l| l.name.as_str()).collect();
914 names.sort_unstable();
915 names.dedup();
916 let not_asked = live - in_scope;
917 (
918 in_scope,
919 format!(
920 "scope: the producer(s) the local registry names ({}); {} other \
921 live producer(s) were not asked and are not counted (O4)",
922 names.join(", "),
923 not_asked
924 ),
925 )
926 }
927 };
928 (answered < in_scope).then(|| {
929 finding(
930 DoctorSeverity::Error,
931 CheckId::IntrospectCoverage,
932 "fleet",
933 format!(
934 "{} of {} live producer(s) in scope did not answer introspect — \
935 alive ⇒ callable, so this is a finding, not a boot race; {scope}",
936 in_scope - answered,
937 in_scope
938 ),
939 Some("RFC 04 §5"),
940 )
941 })
942}
943
944fn judge_state_samples(
949 samples: &[crate::StateSample],
950 ttl: i64,
951 now: std::time::SystemTime,
952) -> (Vec<DoctorFinding>, usize) {
953 let mut findings = Vec::new();
954
955 let mut unstamped = 0usize;
956
957 for sample in samples {
958 match sample.timestamp {
959 Some(ts) => {
960 let stamped = ts.get_time().to_system_time();
961 if let Ok(age) = now.duration_since(stamped)
962 && age.as_secs() as i64 > ttl
963 {
964 findings.push(finding(
965 DoctorSeverity::Error,
966 CheckId::StaleState,
967 sample.key.clone(),
968 format!(
969 "{}s old against ttl {ttl}s (refresh <= ttl/2)",
970 age.as_secs()
971 ),
972 Some("RFC 04 §1.2"),
973 ));
974 }
975 }
976 None => unstamped += 1,
977 }
978 }
979 (findings, unstamped)
980}
981
982fn judge_cardinality(
999 slices: &crate::model::registry::SliceSet,
1000 observed: &crate::judge::budget::BudgetObservation,
1001 window_s: f64,
1002) -> Vec<DoctorFinding> {
1003 let mut findings = Vec::new();
1004
1005 let mut over: Examples<DoctorFinding> = Examples::new(FINDING_CAP);
1006
1007 for slice in slices.slices() {
1008 for s in &slice.subjects {
1009 if !s.path.contains('{') {
1010 continue; }
1012 if s.path.contains("...") {
1013 let seen: usize = observed
1014 .family(&slice.name, &s.path)
1015 .map(|origins| origins.values().map(|keys| keys.len()).sum())
1016 .unwrap_or(0);
1017 findings.push(finding(
1018 DoctorSeverity::Info,
1019 CheckId::CardinalityOverDeclared,
1020 format!("{}/{}", slice.name, s.path),
1021 format!(
1022 "exempt: rest-variable — a `{{var...}}` family is unbounded by \
1023 construction, so its declared cardinality is not a bound this \
1024 check can pass or fail; {seen} distinct key(s) observed in \
1025 {window_s:.0}s"
1026 ),
1027 Some("RFC 08 §6.1"),
1028 ));
1029 continue;
1030 }
1031 let Some(declared) = s.cardinality else {
1032 continue; };
1035 let Some(origins) = observed.family(&slice.name, &s.path) else {
1036 continue; };
1038 for (origin, keys) in origins {
1039 if keys.len() as i64 <= declared {
1040 continue; }
1042 let examples = Examples::collect(
1043 crate::judge::common::EXPANSION_CAP,
1044 keys.iter().map(String::as_str),
1045 );
1046 let subject = if origin.starts_with('@') {
1047 format!("{origin}/{}", s.path)
1048 } else {
1049 format!("{origin}/{}/{}", slice.name, s.path)
1050 };
1051 over.push_with(|| {
1052 finding(
1053 DoctorSeverity::Warning,
1054 CheckId::CardinalityOverDeclared,
1055 subject,
1056 format!(
1057 "{} distinct key(s) observed in {window_s:.0}s exceed the \
1058 declared cardinality {declared} — e.g. {}. A bounded window \
1059 observes a lower bound: the population is at least this",
1060 keys.len(),
1061 examples.as_slice().join(", ")
1062 ),
1063 Some("RFC 04 §1.2"),
1064 )
1065 });
1066 }
1067 }
1068 }
1069 emit_capped(
1070 &mut findings,
1071 over,
1072 CheckId::CardinalityOverDeclared,
1073 "more origin famil(y|ies) over their declared cardinality",
1074 );
1075 findings
1076}
1077
1078#[cfg(test)]
1079mod tests {
1080 use super::*;
1081
1082 const BOUNDED: &str = r#"
1083 [registry]
1084 version = "1.0"
1085 app = "t"
1086 convention = 1
1087 [producer]
1088 name = "sysinfo"
1089 [[subject]]
1090 path = "disk/{mount}/used"
1091 class = "telemetry"
1092 type = "Point"
1093 cardinality = 16
1094 "#;
1095
1096 #[test]
1100 fn cardinality_over_declared_fires_with_count_and_examples() {
1101 let slices = crate::model::registry::SliceSet::from_toml_for_tests(BOUNDED);
1102 let keys: Vec<String> = (0..40)
1103 .map(|i| format!("v1/h-aaaaaaaaaaaa/telemetry/sysinfo/disk/m{i:02}/used"))
1104 .collect();
1105 let obs = crate::judge::budget::BudgetObservation::observe(
1106 "",
1107 &slices,
1108 keys.iter().map(String::as_str),
1109 );
1110 let findings = judge_cardinality(&slices, &obs, 10.0);
1111 assert_eq!(findings.len(), 1, "{findings:?}");
1112 let f = &findings[0];
1113 assert_eq!(f.check, CheckId::CardinalityOverDeclared);
1114 assert_eq!(f.severity, DoctorSeverity::Warning);
1115 assert_eq!(f.subject, "h-aaaaaaaaaaaa/sysinfo/disk/{mount}/used");
1116 assert!(f.evidence.contains("40 distinct key(s)"), "{}", f.evidence);
1117 assert!(f.evidence.contains("cardinality 16"), "{}", f.evidence);
1118 assert!(f.evidence.contains("10s"), "the window is stated");
1119 assert!(
1120 f.evidence.contains("disk/m00/used"),
1121 "examples are named: {}",
1122 f.evidence
1123 );
1124 }
1125
1126 #[test]
1130 fn cardinality_under_declared_is_not_a_finding() {
1131 let slices = crate::model::registry::SliceSet::from_toml_for_tests(BOUNDED);
1132 let keys = [
1133 "v1/h-aaaaaaaaaaaa/telemetry/sysinfo/disk/root/used",
1134 "v1/h-aaaaaaaaaaaa/telemetry/sysinfo/disk/var/used",
1135 "v1/h-bbbbbbbbbbbb/telemetry/sysinfo/disk/root/used",
1137 ];
1138 let obs = crate::judge::budget::BudgetObservation::observe("", &slices, keys);
1139 assert!(judge_cardinality(&slices, &obs, 5.0).is_empty());
1140 }
1141
1142 #[test]
1146 fn rest_variable_families_are_exempt_and_say_so() {
1147 let toml = r#"
1148 [registry]
1149 version = "1.0"
1150 app = "t"
1151 convention = 1
1152 [producer]
1153 name = "gnmi"
1154 [[subject]]
1155 path = "{device}/{path...}"
1156 class = "telemetry"
1157 type = "Point"
1158 cardinality = 2
1159 "#;
1160 let slices = crate::model::registry::SliceSet::from_toml_for_tests(toml);
1161 let keys: Vec<String> = (0..5)
1162 .map(|i| format!("v1/h-aaaaaaaaaaaa/telemetry/gnmi/sw1/if/eth{i}/rx"))
1163 .collect();
1164 let obs = crate::judge::budget::BudgetObservation::observe(
1165 "",
1166 &slices,
1167 keys.iter().map(String::as_str),
1168 );
1169 let findings = judge_cardinality(&slices, &obs, 5.0);
1170 assert_eq!(findings.len(), 1, "{findings:?}");
1171 let f = &findings[0];
1172 assert_eq!(f.severity, DoctorSeverity::Info, "an exemption, not a pass");
1173 assert!(
1174 f.evidence.starts_with("exempt: rest-variable"),
1175 "{}",
1176 f.evidence
1177 );
1178 assert!(f.evidence.contains("5 distinct key(s)"), "{}", f.evidence);
1179 let quiet = judge_cardinality(&slices, &Default::default(), 5.0);
1182 assert_eq!(quiet.len(), 1);
1183 assert!(quiet[0].evidence.starts_with("exempt: rest-variable"));
1184 }
1185
1186 #[test]
1193 fn qos_mismatch_cap_bounds_violators_not_map_entries() {
1194 let mut qos_bad: std::collections::BTreeMap<String, (String, u64, u64)> =
1195 std::collections::BTreeMap::new();
1196 for i in 0..30u32 {
1197 let bad = if i < 5 { 0 } else { 1 };
1199 qos_bad.insert(
1200 format!("v1/h-a/telemetry/x/k{i:02}"),
1201 ("tel".into(), bad, 10),
1202 );
1203 }
1204 let findings = judge_qos_observed(&qos_bad);
1205 let per_key: Vec<&DoctorFinding> = findings
1206 .iter()
1207 .filter(|f| f.severity == DoctorSeverity::Warning)
1208 .collect();
1209 assert_eq!(per_key.len(), FINDING_CAP, "the cap bounds the findings");
1210 assert!(
1211 per_key.iter().all(|f| f.evidence.starts_with("1 of 10")),
1212 "only violators become findings: {findings:#?}"
1213 );
1214 assert!(
1215 per_key.iter().any(|f| f.subject.ends_with("k24")),
1216 "violators past the first {FINDING_CAP} map entries (k20..k24) are \
1217 not dropped: {findings:#?}"
1218 );
1219 let note = findings
1220 .iter()
1221 .find(|f| f.severity == DoctorSeverity::Info)
1222 .expect("a remainder note");
1223 assert_eq!(
1224 note.evidence, "… and 5 more key(s) with the same finding",
1225 "the note counts violators (25 − 20), not map entries"
1226 );
1227
1228 let few: std::collections::BTreeMap<String, (String, u64, u64)> = qos_bad
1230 .iter()
1231 .take(10)
1232 .map(|(k, v)| (k.clone(), v.clone()))
1233 .collect();
1234 let findings = judge_qos_observed(&few);
1235 assert_eq!(findings.len(), 5, "{findings:#?}");
1236 assert!(
1237 findings
1238 .iter()
1239 .all(|f| f.severity == DoctorSeverity::Warning)
1240 );
1241 }
1242
1243 fn roster_of(entries: &[(&str, &[&str])]) -> std::collections::BTreeMap<String, Vec<String>> {
1244 entries
1245 .iter()
1246 .map(|(origin, producers)| {
1247 (
1248 origin.to_string(),
1249 producers.iter().map(|p| p.to_string()).collect(),
1250 )
1251 })
1252 .collect()
1253 }
1254
1255 fn slice_named(name: &str) -> RegistrySlice {
1256 zenkey::parse_slice(&format!(
1257 "[registry]\nversion = \"1.0\"\napp = \"t\"\nconvention = 1\n\
1258 [producer]\nname = \"{name}\"\n"
1259 ))
1260 .expect("fixture slice parses")
1261 }
1262
1263 #[test]
1268 fn a_partial_registry_does_not_count_unasked_producers_against_coverage() {
1269 let roster = roster_of(&[("h-aaaaaaaaaaaa", &["sysinfo", "extra"])]);
1270 let locals = [slice_named("sysinfo")];
1271 assert_eq!(
1272 judge_introspect_coverage(&roster, Some(&locals), 1),
1273 None,
1274 "the un-asked producer is out of scope, not silent"
1275 );
1276 }
1277
1278 #[test]
1283 fn introspect_coverage_evidence_states_its_scope() {
1284 let roster = roster_of(&[
1285 ("h-aaaaaaaaaaaa", &["sysinfo", "extra"]),
1286 ("h-bbbbbbbbbbbb", &["sysinfo-2"]),
1287 ]);
1288 let locals = [slice_named("sysinfo")];
1289 let f = judge_introspect_coverage(&roster, Some(&locals), 1).expect("a finding");
1290 assert_eq!(f.check, CheckId::IntrospectCoverage);
1291 assert!(f.evidence.contains("1 of 2"), "{}", f.evidence);
1292 assert!(
1293 f.evidence.contains("the local registry names (sysinfo)"),
1294 "{}",
1295 f.evidence
1296 );
1297 assert!(
1298 f.evidence
1299 .contains("1 other live producer(s) were not asked"),
1300 "{}",
1301 f.evidence
1302 );
1303 }
1304
1305 #[test]
1308 fn a_service_slice_scopes_its_origin_into_coverage() {
1309 let roster = roster_of(&[("@catalog", &["catalog"]), ("h-aaaaaaaaaaaa", &["extra"])]);
1310 let locals = [zenkey::parse_slice(
1311 "[registry]\nversion = \"1.0\"\napp = \"t\"\nconvention = 1\n\
1312 [service]\nname = \"catalog\"\norigin = \"@catalog\"\n",
1313 )
1314 .expect("service slice parses")];
1315 assert_eq!(judge_introspect_coverage(&roster, Some(&locals), 1), None);
1316 let f = judge_introspect_coverage(&roster, Some(&locals), 0).expect("a finding");
1317 assert!(f.evidence.contains("1 of 1"), "{}", f.evidence);
1318 }
1319
1320 #[test]
1322 fn the_wildcard_sweep_judges_the_whole_roster() {
1323 let roster = roster_of(&[("h-aaaaaaaaaaaa", &["sysinfo", "extra"])]);
1324 let f = judge_introspect_coverage(&roster, None, 1).expect("a finding");
1325 assert!(f.evidence.contains("1 of 2"), "{}", f.evidence);
1326 assert!(f.evidence.contains("whole roster"), "{}", f.evidence);
1327 assert_eq!(judge_introspect_coverage(&roster, None, 2), None);
1328 }
1329
1330 #[test]
1331 fn freshness_judgement_is_pure_and_ttl_bound() {
1332 let now = std::time::SystemTime::now();
1333 let fresh_ts = zenoh::time::Timestamp::new(
1334 zenoh::time::NTP64::from(now.duration_since(std::time::UNIX_EPOCH).unwrap()),
1335 zenoh::time::TimestampId::rand(),
1336 );
1337 let stale_ts = zenoh::time::Timestamp::new(
1338 zenoh::time::NTP64::from(
1339 now.duration_since(std::time::UNIX_EPOCH).unwrap() - Duration::from_secs(120),
1340 ),
1341 zenoh::time::TimestampId::rand(),
1342 );
1343 let samples = vec![
1344 crate::StateSample {
1345 key: "b/v1/h-aaaaaaaaaaaa/state/p/health".into(),
1346 timestamp: Some(fresh_ts),
1347 payload_len: 2,
1348 },
1349 crate::StateSample {
1350 key: "b/v1/h-bbbbbbbbbbbb/state/p/health".into(),
1351 timestamp: Some(stale_ts),
1352 payload_len: 2,
1353 },
1354 crate::StateSample {
1355 key: "b/v1/h-cccccccccccc/state/p/health".into(),
1356 timestamp: None,
1357 payload_len: 2,
1358 },
1359 ];
1360 let (findings, unstamped) = judge_state_samples(&samples, 30, now);
1361 assert_eq!(
1362 findings.len(),
1363 1,
1364 "only the stale stamped sample is a finding"
1365 );
1366 assert_eq!(findings[0].check, CheckId::StaleState);
1367 assert!(findings[0].subject.contains("h-bbbbbbbbbbbb"));
1368 assert_eq!(unstamped, 1, "the unstamped sample is counted, not judged");
1369 }
1370}