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();
242 let mut described_by_origin: Vec<crate::model::decode::DescribedSchema> = Vec::new();
247 let mut undescribed = 0usize;
248 let describe_opts = GetOpts::new(spec.timeout);
252 for slice in slice_set.slices() {
253 let key = rpc_key(base, slice, "describe")?;
254 let answers = fleet_get(fleet, &key, &describe_opts).await?;
255 let before = described_by_origin.len();
256 for a in answers {
257 let origin = a.origin;
258 let Answer::Value(bytes) = a.answer else {
259 continue;
260 };
261 let cow = bytes.to_bytes();
262 let Some(set) = std::str::from_utf8(&cow)
263 .ok()
264 .and_then(|t| zenkey::schema::SchemaSet::parse(t).ok())
265 else {
266 continue;
267 };
268 if described_by_origin.len() == before {
269 described.push((slice.name.clone(), set.clone()));
270 }
271 described_by_origin.push(crate::model::decode::DescribedSchema {
272 origin,
273 producer: slice.name.clone(),
274 set,
275 });
276 }
277 if described_by_origin.len() == before {
281 undescribed += 1;
282 }
283 }
284 for gap in crate::model::decode::totality_gaps(&described, &slice_set) {
287 findings.push(finding(
288 DoctorSeverity::Error,
289 CheckId::DescribeTotality,
290 gap.producer.clone(),
291 format!(
292 "describe is not total — missing: {}",
293 gap.missing.join(", ")
294 ),
295 Some("RFC 08 §7"),
296 ));
297 }
298 for drift in crate::model::decode::schema_drift(&described_by_origin) {
299 let servers: Vec<String> = drift
300 .servers
301 .iter()
302 .map(|s| match s.hash.as_option() {
306 Some(h) => format!("{}@{} ({h})", s.producer, s.origin),
307 None => format!("{}@{} (no identity served)", s.producer, s.origin),
308 })
309 .collect();
310 let (severity, evidence) = match drift.verdict {
315 DriftVerdict::Disagree => (
316 DoctorSeverity::Error,
317 format!("served with different schemas by {}", servers.join(", ")),
318 ),
319 DriftVerdict::Unjudgeable => (
320 DoctorSeverity::Warning,
321 format!(
322 "agreement cannot be judged — {} served no schema identity: {} \
323 (RFC 09 §5.1 O4; the hash exists for exactly this, RFC 08 §7)",
324 drift
325 .servers
326 .iter()
327 .filter(|s| s.hash.is_not_asked())
328 .count(),
329 servers.join(", ")
330 ),
331 ),
332 };
333 findings.push(finding(
334 severity,
335 CheckId::SchemaDrift,
336 drift.type_name.clone(),
337 evidence,
338 Some("RFC 08 §7"),
339 ));
340 }
341 if undescribed > 0 {
342 findings.push(finding(
343 DoctorSeverity::Info,
344 CheckId::DescribeMissing,
345 "fleet",
346 format!(
347 "{undescribed} producer(s) serve no describe (a SHOULD; generic tools \
348 render their payloads structurally)"
349 ),
350 Some("RFC 08 §7"),
351 ));
352 }
353
354 if spec.deep {
356 let now = std::time::SystemTime::now();
357 let mut unstamped = 0usize;
358 for slice in slice_set.slices() {
359 for subject in &slice.subjects {
360 let (Some(ttl), true) = (subject.ttl_s, subject.class.is(&zenkey::Class::State))
361 else {
362 continue;
363 };
364 let Ok(pattern) = zenkey::pattern::SubjectPattern::parse(&subject.path) else {
365 continue;
366 };
367 let selector = match &slice.service_origin {
368 Some(origin) => with_base(
369 base,
370 format!("v1/{origin}/state/{}", pattern.selector_tail()),
371 ),
372 None => with_base(
373 base,
374 format!("v1/*/state/{}/{}", slice.name, pattern.selector_tail()),
375 ),
376 };
377 let samples = state_snapshot(session, &selector, spec.timeout, spec.sample).await?;
378 let (family_findings, family_unstamped) = judge_state_samples(&samples, ttl, now);
379 findings.extend(family_findings);
380 unstamped += family_unstamped;
381 }
382 }
383 if unstamped > 0 {
384 findings.push(finding(
385 DoctorSeverity::Warning,
386 CheckId::UnstampedState,
387 "fleet",
388 format!(
389 "{unstamped} state sample(s) carry no HLC timestamp — the deployment \
390 lacks timestamping, which LWW requires; freshness is unjudgeable \
391 for them"
392 ),
393 Some("RFC 04 §4"),
394 ));
395 }
396 let storages = crate::storages(session, spec.timeout)
397 .await
398 .unwrap_or_default();
399 let coverage = crate::state_coverage(&slice_set, base, &storages);
400 let uncovered: Vec<&crate::CoverageRow> = coverage
401 .iter()
402 .filter(|r| r.coverage == crate::Coverage::Uncovered)
403 .collect();
404 if !uncovered.is_empty() {
405 findings.push(finding(
406 DoctorSeverity::Info,
407 CheckId::StorageCoverage,
408 "fleet",
409 format!(
410 "{} state famil(y|ies) have no storage coverage (volatile seeding \
411 may ride the advanced-pub/sub cache): {}",
412 uncovered.len(),
413 uncovered
414 .iter()
415 .map(|r| format!("{}/{}", r.producer, r.path))
416 .collect::<Vec<_>>()
417 .join(", ")
418 ),
419 Some("RFC 04 §3.5"),
420 ));
421 }
422 }
423
424 let observation = match spec.listen {
426 Some(window) => {
427 let store = crate::model::decode::SchemaStore::new(base, spec.timeout);
428 for (producer, set) in &described {
433 store.insert(producer, set.clone());
434 }
435 let _sealed = store.seal();
442 let (listen_findings, summary) =
443 observe_traffic(fleet, &slice_set, &store, &described, window).await?;
444 findings.extend(listen_findings);
445 Some(summary)
446 }
447 None => None,
448 };
449
450 Ok(DoctorReport {
451 findings,
452 synced: locals.is_some().then_some(synced).into(),
456 introspect_answered: answered,
457 live_producers: live,
458 describe_served: described.len(),
459 describe_missing: undescribed,
460 routers: routers.len(),
461 router_version,
462 deep: spec.deep,
463 observation,
464 })
465}
466
467const DECODE_BUDGET: u8 = 2;
470
471const SAME_FINDING: &str = "more key(s) with the same finding";
473
474fn emit_capped(
483 findings: &mut Vec<DoctorFinding>,
484 ex: Examples<DoctorFinding>,
485 check: CheckId,
486 tail: &str,
487) {
488 let more = ex.more(tail);
489
490 findings.extend(ex.into_vec());
491
492 if let Some(evidence) = more {
493 findings.push(finding(
494 DoctorSeverity::Info,
495 check,
496 "fleet",
497 evidence,
498 None,
499 ));
500 }
501}
502
503struct RateWindow {
506 cap: Option<u64>,
510 seen: u64,
512}
513
514async fn observe_traffic(
519 fleet: &crate::Fleet<'_>,
520 slices: &crate::model::registry::SliceSet,
521 store: &crate::model::decode::SchemaStore,
522 described: &[(String, zenkey::schema::SchemaSet)],
523 window: Duration,
524) -> Result<(Vec<DoctorFinding>, crate::report::ObservationSummary)> {
525 use std::collections::BTreeMap;
526
527 let (session, base) = (fleet.session(), fleet.base());
528
529 let scopes = crate::judge::common::data_plane_scopes(base, slices);
534
535 let monitor = crate::Monitor::start(session, crate::MonitorSpec::default()).await?;
536 let mut events = monitor.events();
537 let monitor = monitor.watching(&scopes).await?;
540 let started = tokio::time::Instant::now();
541 let deadline = started + window;
542
543 let mut fields =
547 crate::judge::field::FieldObservation::new(crate::judge::field::DEFAULT_MAX_PATHS);
548 let mut samples: u64 = 0;
549 let mut dropped: u64 = 0;
550 let mut synthetic: u64 = 0;
551 let mut facts_cache = crate::model::facts::FactsCache::default();
554 let mut decode_budget: BTreeMap<String, u8> = BTreeMap::new();
555 let mut unregistered: BTreeMap<String, u64> = BTreeMap::new();
557 let mut qos_bad: BTreeMap<String, (String, u64, u64)> = BTreeMap::new();
558 let mut undecodable: BTreeMap<String, (String, u64)> = BTreeMap::new();
559 let mut invalid: BTreeMap<String, (String, u64)> = BTreeMap::new();
560 let mut event_counts: BTreeMap<(String, String), RateWindow> = BTreeMap::new();
563 let mut foreign_stampers: BTreeMap<String, u64> = BTreeMap::new();
565
566 let window_over = tokio::time::sleep_until(deadline);
572 tokio::pin!(window_over);
573 loop {
574 let item = tokio::select! {
575 item = events.recv() => item,
576 () = &mut window_over => break,
577 };
578 match item {
579 Some(crate::StreamItem::Event(crate::FleetEvent::Sample(s))) => {
580 samples += 1;
581 if let Some(att) = &s.attachment
582 && is_synthetic_marker(&att.to_bytes())
583 {
584 synthetic += 1;
585 }
586 if let Some(crate::StampProvenance::Foreign { stamper }) = s.stamped_by {
591 *foreign_stampers.entry(stamper.to_string()).or_default() += 1;
592 }
593 let bytes = s.payload.to_bytes();
598 if bytes.len() > crate::model::decode::OBSERVE_LIMIT {
599 fields.observe_unread(&s.key);
600 } else {
601 let doc = crate::model::decode::structural_value(&bytes);
602 fields.observe(&s.key, started.elapsed().as_secs_f64(), doc.as_ref());
603 }
604 facts_cache.ensure(base, &s.key, Some(slices));
605 let facts = facts_cache.get(&s.key).expect("just ensured this key");
606 match &facts.registration {
607 crate::model::facts::Registration::Unregistered => {
608 *unregistered.entry(s.key.clone()).or_default() += 1;
609 }
610 crate::model::facts::Registration::Registered(sf) => {
611 if let (Some(profile), Some(declared)) = (sf.declared_qos(), &sf.qos) {
612 let entry = qos_bad
613 .entry(s.key.clone())
614 .or_insert_with(|| (declared.token().to_string(), 0, 0));
615 entry.2 += 1;
616 if !s.qos_matches(profile) {
617 entry.1 += 1;
618 }
619 }
620 if let (Some(rate), crate::model::facts::KeyShape::V1(v)) =
621 (&sf.rate, &facts.shape)
622 && v.class == "events"
623 {
624 let family = match &v.producer {
625 Some(p) => format!("{p}/{}", sf.path),
626 None => format!("{}/{}", v.origin, sf.path),
627 };
628 event_counts
634 .entry((family, rate.token()))
635 .or_insert_with(|| RateWindow {
636 cap: rate.cap_per_hour(),
637 seen: 0,
638 })
639 .seen += 1;
640 }
641 let budget = decode_budget.entry(s.key.clone()).or_default();
642 if *budget < DECODE_BUDGET {
643 *budget += 1;
644 let d = crate::model::decode::decode_sample(
652 fleet,
653 store,
654 Some(slices),
655 &s.key,
656 Some(&s.encoding),
657 &s.payload.to_bytes(),
658 )
659 .await;
660 match d.verdict {
661 crate::Verdict::NotValidated(
662 zenkey::schema::validate::NotValidated::Undecodable,
663 ) => {
664 let e = undecodable.entry(s.key.clone()).or_insert_with(|| {
665 (
666 d.decode_error
667 .unwrap_or_else(|| "does not decode".into()),
668 0,
669 )
670 });
671 e.1 += 1;
672 }
673 crate::Verdict::Invalid(errors) => {
674 let e = invalid
675 .entry(s.key.clone())
676 .or_insert_with(|| (errors.join("; "), 0));
677 e.1 += 1;
678 }
679 _ => {}
680 }
681 }
682 }
683 _ => {}
684 }
685 }
686 Some(crate::StreamItem::Dropped(n)) => dropped += n,
687 Some(_) => continue,
688 None => break,
689 }
690 }
691 let keys_seen = facts_cache.len();
692 monitor.shutdown().await?;
693
694 let budgets =
698 crate::judge::budget::BudgetObservation::observe(base, slices, facts_cache.keys());
699
700 let window_s = window.as_secs_f64();
701 let mut findings = Vec::new();
702
703 let mut ex = Examples::new(FINDING_CAP);
704 for (key, (error, n)) in &undecodable {
705 ex.push_with(|| {
706 finding(
707 DoctorSeverity::Error,
708 CheckId::PayloadUndecodable,
709 key.clone(),
710 format!(
711 "payload does not decode as its declared type: {error} ({n} sample(s) tried)"
712 ),
713 Some("RFC 08 §7"),
714 )
715 });
716 }
717 emit_capped(&mut findings, ex, CheckId::PayloadUndecodable, SAME_FINDING);
718 let mut ex = Examples::new(FINDING_CAP);
719 for (key, (violations, n)) in &invalid {
720 ex.push_with(|| {
721 finding(
722 DoctorSeverity::Error,
723 CheckId::PayloadInvalid,
724 key.clone(),
725 format!("payload violates the served schema: {violations} ({n} sample(s) tried)"),
726 Some("RFC 08 §7"),
727 )
728 });
729 }
730 emit_capped(&mut findings, ex, CheckId::PayloadInvalid, SAME_FINDING);
731 findings.extend(judge_qos_observed(&qos_bad));
732 if !foreign_stampers.is_empty() {
733 let mut named: Vec<String> = foreign_stampers
734 .iter()
735 .map(|(zid, n)| format!("{zid} ({n} sample(s))"))
736 .collect();
737 named.sort();
738 findings.push(finding(
739 DoctorSeverity::Info,
740 CheckId::TimestampStampedElsewhere,
741 "fleet".to_string(),
742 format!(
743 "HLCs on this bus are stamped by {} node(s) that are not the publishing \
744 session — a deployment with router-side timestamping, which is legal and \
745 common. Latency measured from these stamps is stamper→observer, not \
746 publisher→observer: {}",
747 foreign_stampers.len(),
748 named.join(", ")
749 ),
750 Some("RFC 09 §5.1 O7"),
751 ));
752 }
753 let mut ex = Examples::new(FINDING_CAP);
754 for (key, n) in &unregistered {
755 ex.push_with(|| {
756 finding(
757 DoctorSeverity::Warning,
758 CheckId::UnregisteredTraffic,
759 key.clone(),
760 format!(
761 "{n} sample(s) on a subject the producer's slice does not declare — \
762 for a conforming producer, a subject that is not registered does not exist"
763 ),
764 Some("RFC 08 §2"),
765 )
766 });
767 }
768 emit_capped(
769 &mut findings,
770 ex,
771 CheckId::UnregisteredTraffic,
772 SAME_FINDING,
773 );
774 if window <= Duration::from_secs(3600) {
778 for ((family, rate), RateWindow { cap, seen: count }) in &event_counts {
779 let Some(cap) = cap else {
780 continue;
781 };
782 if count > cap {
783 findings.push(finding(
784 DoctorSeverity::Warning,
785 CheckId::RateOverDeclared,
786 family.clone(),
787 format!(
788 "{count} event(s) in {window_s:.0}s exceeds the declared \
789 `{rate}` cap ({cap}/h)"
790 ),
791 Some("RFC 04 §1.3"),
792 ));
793 }
794 }
795 }
796
797 findings.extend(judge_cardinality(slices, &budgets, window_s));
798
799 let field_ctx = field_context_from(slices, described, &facts_cache);
803 findings.extend(crate::judge::field::judge_fields(
804 &fields, window_s, &field_ctx,
805 ));
806
807 Ok((
808 findings,
809 crate::report::ObservationSummary {
810 window_s,
811 scopes,
812 samples,
813 keys_seen,
814 dropped,
815 synthetic_marked: synthetic,
816 field_paths_dropped: fields.dropped_paths(),
819 facts_evicted: facts_cache.evicted(),
820 },
821 ))
822}
823
824fn field_context_from(
828 slices: &crate::model::registry::SliceSet,
829 described: &[(String, zenkey::schema::SchemaSet)],
830 facts: &crate::model::facts::FactsCache,
831) -> std::collections::BTreeMap<String, crate::judge::field::KeyFieldContext> {
832 use std::collections::BTreeMap;
833
834 let mut declared_cache: BTreeMap<(String, String), Option<crate::judge::field::DeclaredPaths>> =
835 BTreeMap::new();
836 let mut ctx = BTreeMap::new();
837 for (key, f) in facts.iter() {
838 let mut c = crate::judge::field::KeyFieldContext::default();
839 if let crate::model::facts::Registration::Registered(sf) = &f.registration {
840 c.ttl_s = sf.ttl_s;
841 c.type_name = Some(sf.type_name.clone());
842 if let Some(producer) = crate::judge::common::producer_of(f, Some(slices))
843 && !sf.type_name.is_empty()
844 {
845 let declared = declared_cache
846 .entry((producer.clone(), sf.type_name.clone()))
847 .or_insert_with(|| {
848 described
849 .iter()
850 .find(|(name, _)| *name == producer)
851 .and_then(|(_, set)| set.get(&sf.type_name))
852 .and_then(|schema| schema.json_document())
853 .and_then(crate::judge::field::DeclaredPaths::from_json_schema)
854 });
855 c.declared = declared.clone();
856 }
857 }
858 ctx.insert(key.to_string(), c);
859 }
860 ctx
861}
862
863fn judge_qos_observed(
873 qos_bad: &std::collections::BTreeMap<String, (String, u64, u64)>,
874) -> Vec<DoctorFinding> {
875 let mut findings = Vec::new();
876
877 let mut ex = Examples::new(FINDING_CAP);
878
879 for (key, (declared, bad, total)) in qos_bad.iter().filter(|(_, (_, bad, _))| *bad > 0) {
880 ex.push_with(|| {
881 finding(
882 DoctorSeverity::Warning,
883 CheckId::QosObservedMismatch,
884 key.clone(),
885 format!(
886 "{bad} of {total} sample(s) did not ride the declared {declared} — this \
887 is what actually rode: an interceptor MAY rewrite QoS, so it is a \
888 deviation, not proof of the publisher"
889 ),
890 Some("RFC 04 §3"),
891 )
892 });
893 }
894 emit_capped(
895 &mut findings,
896 ex,
897 CheckId::QosObservedMismatch,
898 SAME_FINDING,
899 );
900 findings
901}
902
903fn judge_introspect_coverage(
920 roster: &std::collections::BTreeMap<String, Vec<String>>,
921 locals: Option<&[RegistrySlice]>,
922 answered: usize,
923) -> Option<DoctorFinding> {
924 let live: usize = roster.values().map(Vec::len).sum();
925
926 let (in_scope, scope) = match locals {
927 None => (
928 live,
929 "scope: the whole roster (fleet-wide wildcard sweep)".to_string(),
930 ),
931 Some(locals) => {
932 let named = |origin: &str, producer: &str| {
933 let base_name = zenkey::grammar::Producer::parse_chunk(producer)
934 .map(|p| p.name().to_string())
935 .unwrap_or_else(|_| producer.to_string());
936 locals.iter().any(|l| {
937 l.name == base_name
938 || l.service_origin.as_ref().map(Declared::token) == Some(origin)
939 })
940 };
941 let in_scope: usize = roster
942 .iter()
943 .map(|(origin, producers)| producers.iter().filter(|p| named(origin, p)).count())
944 .sum();
945 let mut names: Vec<&str> = locals.iter().map(|l| l.name.as_str()).collect();
946 names.sort_unstable();
947 names.dedup();
948 let not_asked = live - in_scope;
949 (
950 in_scope,
951 format!(
952 "scope: the producer(s) the local registry names ({}); {} other \
953 live producer(s) were not asked and are not counted (O4)",
954 names.join(", "),
955 not_asked
956 ),
957 )
958 }
959 };
960 (answered < in_scope).then(|| {
961 finding(
962 DoctorSeverity::Error,
963 CheckId::IntrospectCoverage,
964 "fleet",
965 format!(
966 "{} of {} live producer(s) in scope did not answer introspect — \
967 alive ⇒ callable, so this is a finding, not a boot race; {scope}",
968 in_scope - answered,
969 in_scope
970 ),
971 Some("RFC 04 §5"),
972 )
973 })
974}
975
976fn judge_state_samples(
981 samples: &[crate::StateSample],
982 ttl: i64,
983 now: std::time::SystemTime,
984) -> (Vec<DoctorFinding>, usize) {
985 let mut findings = Vec::new();
986
987 let mut unstamped = 0usize;
988
989 for sample in samples {
990 match sample.timestamp {
991 Some(ts) => {
992 let stamped = ts.get_time().to_system_time();
993 if let Ok(age) = now.duration_since(stamped)
994 && age.as_secs() as i64 > ttl
995 {
996 findings.push(finding(
997 DoctorSeverity::Error,
998 CheckId::StaleState,
999 sample.key.clone(),
1000 format!(
1001 "{}s old against ttl {ttl}s (refresh <= ttl/2)",
1002 age.as_secs()
1003 ),
1004 Some("RFC 04 §1.2"),
1005 ));
1006 }
1007 }
1008 None => unstamped += 1,
1009 }
1010 }
1011 (findings, unstamped)
1012}
1013
1014fn judge_cardinality(
1031 slices: &crate::model::registry::SliceSet,
1032 observed: &crate::judge::budget::BudgetObservation,
1033 window_s: f64,
1034) -> Vec<DoctorFinding> {
1035 let mut findings = Vec::new();
1036
1037 let mut over: Examples<DoctorFinding> = Examples::new(FINDING_CAP);
1038
1039 for slice in slices.slices() {
1040 for s in &slice.subjects {
1041 if !s.path.contains('{') {
1042 continue; }
1044 if s.path.contains("...") {
1045 let seen: usize = observed
1046 .family(&slice.name, &s.path)
1047 .map(|origins| origins.values().map(|keys| keys.len()).sum())
1048 .unwrap_or(0);
1049 findings.push(finding(
1050 DoctorSeverity::Info,
1051 CheckId::CardinalityOverDeclared,
1052 format!("{}/{}", slice.name, s.path),
1053 format!(
1054 "exempt: rest-variable — a `{{var...}}` family is unbounded by \
1055 construction, so its declared cardinality is not a bound this \
1056 check can pass or fail; {seen} distinct key(s) observed in \
1057 {window_s:.0}s"
1058 ),
1059 Some("RFC 08 §6.1"),
1060 ));
1061 continue;
1062 }
1063 let Some(declared) = s.cardinality else {
1064 continue; };
1067 let Some(origins) = observed.family(&slice.name, &s.path) else {
1068 continue; };
1070 for (origin, keys) in origins {
1071 if keys.len() as i64 <= declared {
1072 continue; }
1074 let examples = Examples::collect(
1075 crate::judge::common::EXPANSION_CAP,
1076 keys.iter().map(String::as_str),
1077 );
1078 let subject = if origin.starts_with('@') {
1079 format!("{origin}/{}", s.path)
1080 } else {
1081 format!("{origin}/{}/{}", slice.name, s.path)
1082 };
1083 over.push_with(|| {
1084 finding(
1085 DoctorSeverity::Warning,
1086 CheckId::CardinalityOverDeclared,
1087 subject,
1088 format!(
1089 "{} distinct key(s) observed in {window_s:.0}s exceed the \
1090 declared cardinality {declared} — e.g. {}. A bounded window \
1091 observes a lower bound: the population is at least this",
1092 keys.len(),
1093 examples.as_slice().join(", ")
1094 ),
1095 Some("RFC 04 §1.2"),
1096 )
1097 });
1098 }
1099 }
1100 }
1101 emit_capped(
1102 &mut findings,
1103 over,
1104 CheckId::CardinalityOverDeclared,
1105 "more origin famil(y|ies) over their declared cardinality",
1106 );
1107 findings
1108}
1109
1110#[cfg(test)]
1111mod tests {
1112 use super::*;
1113
1114 const BOUNDED: &str = r#"
1115 [registry]
1116 version = "1.0"
1117 app = "t"
1118 convention = 1
1119 [producer]
1120 name = "sysinfo"
1121 [[subject]]
1122 path = "disk/{mount}/used"
1123 class = "telemetry"
1124 type = "Point"
1125 cardinality = 16
1126 "#;
1127
1128 #[test]
1132 fn cardinality_over_declared_fires_with_count_and_examples() {
1133 let slices = crate::model::registry::SliceSet::from_toml_for_tests(BOUNDED);
1134 let keys: Vec<String> = (0..40)
1135 .map(|i| format!("v1/h-aaaaaaaaaaaa/telemetry/sysinfo/disk/m{i:02}/used"))
1136 .collect();
1137 let obs = crate::judge::budget::BudgetObservation::observe(
1138 "",
1139 &slices,
1140 keys.iter().map(String::as_str),
1141 );
1142 let findings = judge_cardinality(&slices, &obs, 10.0);
1143 assert_eq!(findings.len(), 1, "{findings:?}");
1144 let f = &findings[0];
1145 assert_eq!(f.check, CheckId::CardinalityOverDeclared);
1146 assert_eq!(f.severity, DoctorSeverity::Warning);
1147 assert_eq!(f.subject, "h-aaaaaaaaaaaa/sysinfo/disk/{mount}/used");
1148 assert!(f.evidence.contains("40 distinct key(s)"), "{}", f.evidence);
1149 assert!(f.evidence.contains("cardinality 16"), "{}", f.evidence);
1150 assert!(f.evidence.contains("10s"), "the window is stated");
1151 assert!(
1152 f.evidence.contains("disk/m00/used"),
1153 "examples are named: {}",
1154 f.evidence
1155 );
1156 }
1157
1158 #[test]
1162 fn cardinality_under_declared_is_not_a_finding() {
1163 let slices = crate::model::registry::SliceSet::from_toml_for_tests(BOUNDED);
1164 let keys = [
1165 "v1/h-aaaaaaaaaaaa/telemetry/sysinfo/disk/root/used",
1166 "v1/h-aaaaaaaaaaaa/telemetry/sysinfo/disk/var/used",
1167 "v1/h-bbbbbbbbbbbb/telemetry/sysinfo/disk/root/used",
1169 ];
1170 let obs = crate::judge::budget::BudgetObservation::observe("", &slices, keys);
1171 assert!(judge_cardinality(&slices, &obs, 5.0).is_empty());
1172 }
1173
1174 #[test]
1178 fn rest_variable_families_are_exempt_and_say_so() {
1179 let toml = r#"
1180 [registry]
1181 version = "1.0"
1182 app = "t"
1183 convention = 1
1184 [producer]
1185 name = "gnmi"
1186 [[subject]]
1187 path = "{device}/{path...}"
1188 class = "telemetry"
1189 type = "Point"
1190 cardinality = 2
1191 "#;
1192 let slices = crate::model::registry::SliceSet::from_toml_for_tests(toml);
1193 let keys: Vec<String> = (0..5)
1194 .map(|i| format!("v1/h-aaaaaaaaaaaa/telemetry/gnmi/sw1/if/eth{i}/rx"))
1195 .collect();
1196 let obs = crate::judge::budget::BudgetObservation::observe(
1197 "",
1198 &slices,
1199 keys.iter().map(String::as_str),
1200 );
1201 let findings = judge_cardinality(&slices, &obs, 5.0);
1202 assert_eq!(findings.len(), 1, "{findings:?}");
1203 let f = &findings[0];
1204 assert_eq!(f.severity, DoctorSeverity::Info, "an exemption, not a pass");
1205 assert!(
1206 f.evidence.starts_with("exempt: rest-variable"),
1207 "{}",
1208 f.evidence
1209 );
1210 assert!(f.evidence.contains("5 distinct key(s)"), "{}", f.evidence);
1211 let quiet = judge_cardinality(&slices, &Default::default(), 5.0);
1214 assert_eq!(quiet.len(), 1);
1215 assert!(quiet[0].evidence.starts_with("exempt: rest-variable"));
1216 }
1217
1218 #[test]
1225 fn qos_mismatch_cap_bounds_violators_not_map_entries() {
1226 let mut qos_bad: std::collections::BTreeMap<String, (String, u64, u64)> =
1227 std::collections::BTreeMap::new();
1228 for i in 0..30u32 {
1229 let bad = if i < 5 { 0 } else { 1 };
1231 qos_bad.insert(
1232 format!("v1/h-a/telemetry/x/k{i:02}"),
1233 ("tel".into(), bad, 10),
1234 );
1235 }
1236 let findings = judge_qos_observed(&qos_bad);
1237 let per_key: Vec<&DoctorFinding> = findings
1238 .iter()
1239 .filter(|f| f.severity == DoctorSeverity::Warning)
1240 .collect();
1241 assert_eq!(per_key.len(), FINDING_CAP, "the cap bounds the findings");
1242 assert!(
1243 per_key.iter().all(|f| f.evidence.starts_with("1 of 10")),
1244 "only violators become findings: {findings:#?}"
1245 );
1246 assert!(
1247 per_key.iter().any(|f| f.subject.ends_with("k24")),
1248 "violators past the first {FINDING_CAP} map entries (k20..k24) are \
1249 not dropped: {findings:#?}"
1250 );
1251 let note = findings
1252 .iter()
1253 .find(|f| f.severity == DoctorSeverity::Info)
1254 .expect("a remainder note");
1255 assert_eq!(
1256 note.evidence, "… and 5 more key(s) with the same finding",
1257 "the note counts violators (25 − 20), not map entries"
1258 );
1259
1260 let few: std::collections::BTreeMap<String, (String, u64, u64)> = qos_bad
1262 .iter()
1263 .take(10)
1264 .map(|(k, v)| (k.clone(), v.clone()))
1265 .collect();
1266 let findings = judge_qos_observed(&few);
1267 assert_eq!(findings.len(), 5, "{findings:#?}");
1268 assert!(
1269 findings
1270 .iter()
1271 .all(|f| f.severity == DoctorSeverity::Warning)
1272 );
1273 }
1274
1275 fn roster_of(entries: &[(&str, &[&str])]) -> std::collections::BTreeMap<String, Vec<String>> {
1276 entries
1277 .iter()
1278 .map(|(origin, producers)| {
1279 (
1280 origin.to_string(),
1281 producers.iter().map(|p| p.to_string()).collect(),
1282 )
1283 })
1284 .collect()
1285 }
1286
1287 fn slice_named(name: &str) -> RegistrySlice {
1288 zenkey::parse_slice(&format!(
1289 "[registry]\nversion = \"1.0\"\napp = \"t\"\nconvention = 1\n\
1290 [producer]\nname = \"{name}\"\n"
1291 ))
1292 .expect("fixture slice parses")
1293 }
1294
1295 #[test]
1300 fn a_partial_registry_does_not_count_unasked_producers_against_coverage() {
1301 let roster = roster_of(&[("h-aaaaaaaaaaaa", &["sysinfo", "extra"])]);
1302 let locals = [slice_named("sysinfo")];
1303 assert_eq!(
1304 judge_introspect_coverage(&roster, Some(&locals), 1),
1305 None,
1306 "the un-asked producer is out of scope, not silent"
1307 );
1308 }
1309
1310 #[test]
1315 fn introspect_coverage_evidence_states_its_scope() {
1316 let roster = roster_of(&[
1317 ("h-aaaaaaaaaaaa", &["sysinfo", "extra"]),
1318 ("h-bbbbbbbbbbbb", &["sysinfo-2"]),
1319 ]);
1320 let locals = [slice_named("sysinfo")];
1321 let f = judge_introspect_coverage(&roster, Some(&locals), 1).expect("a finding");
1322 assert_eq!(f.check, CheckId::IntrospectCoverage);
1323 assert!(f.evidence.contains("1 of 2"), "{}", f.evidence);
1324 assert!(
1325 f.evidence.contains("the local registry names (sysinfo)"),
1326 "{}",
1327 f.evidence
1328 );
1329 assert!(
1330 f.evidence
1331 .contains("1 other live producer(s) were not asked"),
1332 "{}",
1333 f.evidence
1334 );
1335 }
1336
1337 #[test]
1340 fn a_service_slice_scopes_its_origin_into_coverage() {
1341 let roster = roster_of(&[("@catalog", &["catalog"]), ("h-aaaaaaaaaaaa", &["extra"])]);
1342 let locals = [zenkey::parse_slice(
1343 "[registry]\nversion = \"1.0\"\napp = \"t\"\nconvention = 1\n\
1344 [service]\nname = \"catalog\"\norigin = \"@catalog\"\n",
1345 )
1346 .expect("service slice parses")];
1347 assert_eq!(judge_introspect_coverage(&roster, Some(&locals), 1), None);
1348 let f = judge_introspect_coverage(&roster, Some(&locals), 0).expect("a finding");
1349 assert!(f.evidence.contains("1 of 1"), "{}", f.evidence);
1350 }
1351
1352 #[test]
1354 fn the_wildcard_sweep_judges_the_whole_roster() {
1355 let roster = roster_of(&[("h-aaaaaaaaaaaa", &["sysinfo", "extra"])]);
1356 let f = judge_introspect_coverage(&roster, None, 1).expect("a finding");
1357 assert!(f.evidence.contains("1 of 2"), "{}", f.evidence);
1358 assert!(f.evidence.contains("whole roster"), "{}", f.evidence);
1359 assert_eq!(judge_introspect_coverage(&roster, None, 2), None);
1360 }
1361
1362 #[test]
1363 fn freshness_judgement_is_pure_and_ttl_bound() {
1364 let now = std::time::SystemTime::now();
1365 let fresh_ts = zenoh::time::Timestamp::new(
1366 zenoh::time::NTP64::from(now.duration_since(std::time::UNIX_EPOCH).unwrap()),
1367 zenoh::time::TimestampId::rand(),
1368 );
1369 let stale_ts = zenoh::time::Timestamp::new(
1370 zenoh::time::NTP64::from(
1371 now.duration_since(std::time::UNIX_EPOCH).unwrap() - Duration::from_secs(120),
1372 ),
1373 zenoh::time::TimestampId::rand(),
1374 );
1375 let samples = vec![
1376 crate::StateSample {
1377 key: "b/v1/h-aaaaaaaaaaaa/state/p/health".into(),
1378 timestamp: Some(fresh_ts),
1379 payload_len: 2,
1380 },
1381 crate::StateSample {
1382 key: "b/v1/h-bbbbbbbbbbbb/state/p/health".into(),
1383 timestamp: Some(stale_ts),
1384 payload_len: 2,
1385 },
1386 crate::StateSample {
1387 key: "b/v1/h-cccccccccccc/state/p/health".into(),
1388 timestamp: None,
1389 payload_len: 2,
1390 },
1391 ];
1392 let (findings, unstamped) = judge_state_samples(&samples, 30, now);
1393 assert_eq!(
1394 findings.len(),
1395 1,
1396 "only the stale stamped sample is a finding"
1397 );
1398 assert_eq!(findings[0].check, CheckId::StaleState);
1399 assert!(findings[0].subject.contains("h-bbbbbbbbbbbb"));
1400 assert_eq!(unstamped, 1, "the unstamped sample is counted, not judged");
1401 }
1402}