1use std::collections::BTreeMap;
24
25use zenoh::key_expr::keyexpr;
26
27use crate::model::registry::SliceSet;
28use crate::report::{
29 Asked, CheckFinding, CheckKind, Deployment, GarbageCollection, HistoryMode, Judgement,
30 Persistence, PlanWarning, PlannedStorage, PlannedVolume, Refusal, RegistryFacts, Replication,
31 StorageCheck, StorageClass, StorageExplain, StorageInfo, StoragePlan, Taker, TakerRelation,
32 WarningKind,
33};
34
35pub const DEFAULT_LIFESPAN_S: i64 = 86_400;
38pub const DEFAULT_GC_PERIOD_S: u64 = 30;
40pub const DEFAULT_GC_MARGIN: f64 = 2.0;
42
43pub const CHECK_ASKED: &str = "@/*/router/**/storage_manager/storages/**";
46
47fn capability(plugin: &str) -> Option<(Persistence, Option<HistoryMode>)> {
51 match plugin {
52 "memory" => Some((Persistence::Volatile, Some(HistoryMode::Latest))),
53 "fs" | "rocksdb" => Some((Persistence::Durable, Some(HistoryMode::Latest))),
54 "influxdb" => Some((Persistence::Durable, Some(HistoryMode::All))),
55 "redb" => Some((Persistence::Durable, None)),
56 _ => None,
57 }
58}
59
60fn default_replication() -> BTreeMap<String, serde_json::Value> {
63 [
64 ("interval", serde_json::json!(10.0)),
65 ("sub_intervals", serde_json::json!(5)),
66 ("hot", serde_json::json!(6)),
67 ("warm", serde_json::json!(30)),
68 ("propagation_delay", serde_json::json!(250)),
69 ]
70 .into_iter()
71 .map(|(k, v)| (k.to_string(), v))
72 .collect()
73}
74
75pub fn literal_prefix(key_expr: &str) -> String {
81 key_expr
82 .split('/')
83 .take_while(|c| !c.contains('*') && !c.contains('$'))
84 .collect::<Vec<_>>()
85 .join("/")
86}
87
88struct Family {
90 producer: String,
91 path: String,
92 is_state: bool,
93 ttl_s: Option<i64>,
94 selector: String,
95}
96
97fn families(slices: &SliceSet, base: &str) -> Vec<Family> {
100 let mut out = Vec::new();
101 for slice in slices.slices() {
102 for subject in &slice.subjects {
103 let Ok(pattern) = zenkey::pattern::SubjectPattern::parse(&subject.path) else {
104 continue;
105 };
106 let class = subject.class.token();
107 let selector = match &slice.service_origin {
108 Some(origin) => zenkey::grammar::with_base(
109 base,
110 format!("v1/{origin}/{class}/{}", pattern.selector_tail()),
111 ),
112 None => zenkey::grammar::with_base(
113 base,
114 format!("v1/*/{class}/{}/{}", slice.name, pattern.selector_tail()),
115 ),
116 };
117 out.push(Family {
118 producer: slice.name.clone(),
119 path: subject.path.clone(),
120 is_state: subject.class.is(&zenkey::Class::State),
121 ttl_s: subject.ttl_s,
122 selector,
123 });
124 }
125 }
126 out
127}
128
129fn longest_ttl<'f>(fams: impl Iterator<Item = &'f Family>) -> Option<(i64, String)> {
132 fams.filter(|f| f.is_state)
133 .filter_map(|f| f.ttl_s.map(|t| (t, format!("{}/{}", f.producer, f.path))))
134 .max_by(|a, b| a.0.cmp(&b.0).then_with(|| b.1.cmp(&a.1)))
135}
136
137fn warn(kind: WarningKind, cite: &str, text: impl Into<String>) -> PlanWarning {
138 PlanWarning {
139 kind,
140 text: text.into(),
141 cite: cite.to_string(),
142 }
143}
144
145fn refuse_storage(
146 name: &str,
147 key_expr: Option<String>,
148 cite: &str,
149 reason: impl Into<String>,
150) -> Refusal {
151 Refusal {
152 storage: Some(name.to_string()),
153 volume: None,
154 key_expr,
155 reason: reason.into(),
156 cite: cite.to_string(),
157 }
158}
159
160pub fn plan_storages(
168 slices: Option<&SliceSet>,
169 fallback_base: &str,
170 deployment: &Deployment,
171) -> StoragePlan {
172 let base = deployment
173 .base
174 .clone()
175 .unwrap_or_else(|| fallback_base.to_string());
176 let fams: Option<Vec<Family>> = slices.map(|s| families(s, &base));
177 let registry = match (slices, &fams) {
178 (Some(s), Some(f)) => {
179 let longest = longest_ttl(f.iter());
180 Asked::Asked(RegistryFacts {
181 slices: s.slices().len(),
182 max_ttl_s: longest.as_ref().map(|l| l.0),
183 ttl_source: longest.map(|l| l.1),
184 })
185 }
186 _ => Asked::NotAsked,
187 };
188
189 let mut refusals = Vec::new();
190
191 let mut volumes = Vec::new();
193 let mut refused_volumes: Vec<String> = Vec::new();
194 for (id, spec) in &deployment.volumes {
195 let mut warnings = Vec::new();
196 let (persistence, history) = match capability(&spec.plugin) {
197 Some((p, Some(fixed))) => match spec.history {
198 Some(h) if h != fixed => {
199 refusals.push(Refusal {
200 storage: None,
201 volume: Some(id.clone()),
202 key_expr: None,
203 reason: format!(
204 "plugin {:?} offers {} history only; it cannot be declared \
205 history = {:?}",
206 spec.plugin,
207 fixed.as_str(),
208 h.as_str()
209 ),
210 cite: "RFC 09 §2.1".into(),
211 });
212 refused_volumes.push(id.clone());
213 continue;
214 }
215 _ => (Some(p), fixed),
216 },
217 Some((p, None)) => match spec.history {
218 Some(h) => (Some(p), h),
219 None => {
220 refusals.push(Refusal {
221 storage: None,
222 volume: Some(id.clone()),
223 key_expr: None,
224 reason: format!(
225 "plugin {:?} offers both history modes and the choice is per \
226 volume: declare history = \"latest\" or \"all\" (one volume \
227 per mode from the same plugin)",
228 spec.plugin
229 ),
230 cite: "RFC 09 §2.1".into(),
231 });
232 refused_volumes.push(id.clone());
233 continue;
234 }
235 },
236 None => match spec.history {
237 Some(h) => {
238 warnings.push(warn(
239 WarningKind::UnknownPlugin,
240 "RFC 09 §2.1",
241 format!(
242 "plugin {:?} is not in the capability table; its history = \
243 {:?} is taken as declared, not verified",
244 spec.plugin,
245 h.as_str()
246 ),
247 ));
248 (None, h)
249 }
250 None => {
251 refusals.push(Refusal {
252 storage: None,
253 volume: Some(id.clone()),
254 key_expr: None,
255 reason: format!(
256 "plugin {:?} is not in the capability table, so its history \
257 mode cannot be inferred: declare history = \"latest\" or \"all\"",
258 spec.plugin
259 ),
260 cite: "RFC 09 §2.1".into(),
261 });
262 refused_volumes.push(id.clone());
263 continue;
264 }
265 },
266 };
267 volumes.push(PlannedVolume {
268 id: id.clone(),
269 plugin: spec.plugin.clone(),
270 history,
271 persistence,
272 params: spec.params.clone(),
273 warnings,
274 });
275 }
276
277 let mut storages: Vec<PlannedStorage> = Vec::new();
279 for (name, spec) in &deployment.storages {
280 let (class, key_expr) = match (spec.class, spec.selector.as_deref()) {
282 (Some(c), None) => (Some(c), zenkey::grammar::with_base(&base, c.selector())),
283 (None, Some(sel)) => (None, zenkey::grammar::with_base(&base, sel)),
284 (Some(_), Some(_)) => {
285 refusals.push(refuse_storage(
286 name,
287 None,
288 "RFC 04 §4",
289 "declares both class and selector — the class derives the selector, so \
290 name one or the other",
291 ));
292 continue;
293 }
294 (None, None) => {
295 refusals.push(refuse_storage(
296 name,
297 None,
298 "RFC 04 §4",
299 "declares neither class nor selector — nothing says what it stores",
300 ));
301 continue;
302 }
303 };
304 let Ok(ke) = keyexpr::new(key_expr.as_str()) else {
305 refusals.push(refuse_storage(
306 name,
307 Some(key_expr.clone()),
308 "RFC 03 §2",
309 format!("{key_expr:?} is not a valid key expression"),
310 ));
311 continue;
312 };
313
314 let Some(volume) = volumes.iter().find(|v| v.id == spec.volume) else {
316 let reason = if refused_volumes.contains(&spec.volume) {
317 format!("its volume {:?} was refused (see above)", spec.volume)
318 } else {
319 format!(
320 "names volume {:?}, which [volumes] does not declare",
321 spec.volume
322 )
323 };
324 refusals.push(refuse_storage(name, Some(key_expr), "RFC 09 §2", reason));
325 continue;
326 };
327 let history = volume.history;
328
329 let replication = match &spec.replication {
331 Replication::Enabled(false) => None,
332 Replication::Enabled(true) => Some(default_replication()),
333 Replication::Params(p) => Some(p.clone()),
334 };
335 if replication.is_some() && history == HistoryMode::All {
336 refusals.push(refuse_storage(
337 name,
338 Some(key_expr),
339 "RFC 09 §2.2",
340 format!(
341 "declares replication on volume {:?}, whose history mode is all — \
342 the storage manager refuses to start such a storage, so this plan \
343 refuses it first (anti-entropy aligns one value per key; an all-mode \
344 volume cannot participate)",
345 volume.id
346 ),
347 ));
348 continue;
349 }
350
351 let mut warnings = Vec::new();
352 if let Some(p) = &replication
353 && let (Some(interval), Some(delay)) = (
354 p.get("interval").and_then(serde_json::Value::as_f64),
355 p.get("propagation_delay")
356 .and_then(serde_json::Value::as_f64),
357 )
358 && delay / 1000.0 >= interval / 2.0
359 {
360 warnings.push(warn(
361 WarningKind::ReplicationParams,
362 "RFC 09 §2.2",
363 format!(
364 "propagation_delay {delay} ms is not below interval/2 = {} ms — \
365 divergent or inconsistent replication parameters cause digest \
366 storms, not errors",
367 interval * 500.0
368 ),
369 ));
370 }
371
372 let covered: Option<Vec<&Family>> = fams.as_ref().map(|f| {
374 f.iter()
375 .filter(|fam| keyexpr::new(fam.selector.as_str()).is_ok_and(|fk| ke.intersects(fk)))
376 .collect()
377 });
378 if let Some(c) = &covered
379 && c.is_empty()
380 {
381 let reason = format!(
382 "the registry declares no subject under {key_expr:?} — empty coverage is \
383 a finding, not a plan (a storage that captures nothing is either a \
384 typo or a registry gap; either way it is not this deployment's)"
385 );
386 refusals.push(refuse_storage(name, Some(key_expr), "RFC 13 §3", reason));
387 continue;
388 }
389
390 let margin = spec.gc_margin.unwrap_or(DEFAULT_GC_MARGIN);
392 let longest = covered
393 .as_ref()
394 .and_then(|c| longest_ttl(c.iter().copied()));
395 let (lifespan_s, derivation) = match (spec.gc_lifespan_s, &covered, &longest) {
396 (Some(explicit), _, Some((ttl, src))) => {
397 if explicit < *ttl {
398 warnings.push(warn(
399 WarningKind::LifespanBelowTtl,
400 "RFC 09 §2.3",
401 format!(
402 "gc_lifespan_s {explicit} is below the longest covered ttl_s \
403 {ttl} ({src}): a delete must stay observable ≥ ttl_s (RFC 04 \
404 §1.2), else a slow replica may resurrect a retired key"
405 ),
406 ));
407 }
408 (
409 explicit,
410 format!(
411 "declared gc_lifespan_s {explicit} (longest covered ttl_s {ttl}, {src})"
412 ),
413 )
414 }
415 (Some(explicit), Some(_), None) => (
416 explicit,
417 format!("declared gc_lifespan_s {explicit} (no state subject under this selector)"),
418 ),
419 (Some(explicit), None, _) => (
420 explicit,
421 format!("declared gc_lifespan_s {explicit} (no registry: unverified)"),
422 ),
423 (None, Some(_), Some((ttl, src))) => {
424 let lifespan = (*ttl as f64 * margin).ceil() as i64;
425 (
426 lifespan,
427 format!("max ttl_s {ttl} ({src}) × {margin:?} = {lifespan} s"),
428 )
429 }
430 (None, Some(_), None) => (
431 DEFAULT_LIFESPAN_S,
432 format!(
433 "no state subject under this selector: RFC 09 §2.3 default \
434 {DEFAULT_LIFESPAN_S} s"
435 ),
436 ),
437 (None, None, _) => (
438 DEFAULT_LIFESPAN_S,
439 format!("no registry: RFC 09 §2.3 default {DEFAULT_LIFESPAN_S} s, unverified"),
440 ),
441 };
442
443 let mut complete = spec.complete;
445 if complete
446 && !(replication.is_some()
447 && history == HistoryMode::Latest
448 && class == Some(StorageClass::State))
449 {
450 complete = false;
451 let why = if class != Some(StorageClass::State) {
452 "it is not the fully covering latest storage (class state)"
453 } else if history != HistoryMode::Latest {
454 "its volume is not latest-mode"
455 } else {
456 "it is not replicated"
457 };
458 warnings.push(warn(
459 WarningKind::CompleteRefused,
460 "RFC 09 §2.2",
461 format!(
462 "complete = true refused: {why} — complete is right only on a \
463 replicated, fully covering latest storage, where it lets the router \
464 answer any state GET from the nearest replica; emitted as false"
465 ),
466 ));
467 }
468
469 match volume.plugin.as_str() {
471 "influxdb" => warnings.push(warn(
472 WarningKind::RetentionIsTheDatabases,
473 "RFC 09 §2.3",
474 "retention is the database's policy (an InfluxDB retention policy), not \
475 zenoh config: garbage_collection prunes metadata and never drops a value, \
476 so this storage's data grows until the database prunes it — size the \
477 volume against the write rate",
478 )),
479 "redb" => match (history, &spec.retention) {
480 (HistoryMode::All, None) => warnings.push(warn(
481 WarningKind::RetentionRequired,
482 "RFC 09 §2.1",
483 "an all-mode redb storage that declares no retention policy refuses to \
484 start — add a retention block bounding age, bytes or per-key samples \
485 (a loud config error is recoverable in seconds; a full disk is not)",
486 )),
487 (HistoryMode::Latest, Some(_)) => warnings.push(warn(
488 WarningKind::RetentionPointless,
489 "RFC 09 §2.1",
490 "a retention block on a latest-mode volume is refused at startup: there \
491 is no history to prune, and it would report passes while reclaiming \
492 nothing",
493 )),
494 _ => {}
495 },
496 "memory" if class.is_some_and(StorageClass::seeds) => warnings.push(warn(
497 WarningKind::VolatileSeed,
498 "RFC 09 §2.1",
499 "a volatile volume under a seed-bearing class: gone on router restart, \
500 so late joiners lose their seed until state refreshes",
501 )),
502 _ => {}
503 }
504
505 storages.push(PlannedStorage {
506 name: name.clone(),
507 class,
508 strip_prefix: literal_prefix(&key_expr),
509 key_expr,
510 volume: volume.id.clone(),
511 history,
512 replication,
513 complete,
514 garbage_collection: GarbageCollection {
515 period_s: spec.gc_period_s.unwrap_or(DEFAULT_GC_PERIOD_S),
516 lifespan_s,
517 derivation,
518 },
519 retention: spec.retention.clone(),
520 params: spec.params.clone(),
521 covers: match &covered {
522 Some(c) => Asked::Asked(c.len()),
523 None => Asked::NotAsked,
524 },
525 warnings,
526 });
527 }
528
529 let mut overlaps: Vec<(usize, usize)> = Vec::new();
531 for i in 0..storages.len() {
532 for j in (i + 1)..storages.len() {
533 let (a, b) = (&storages[i], &storages[j]);
534 if let (Ok(ka), Ok(kb)) = (
535 keyexpr::new(a.key_expr.as_str()),
536 keyexpr::new(b.key_expr.as_str()),
537 ) && ka.intersects(kb)
538 {
539 overlaps.push((i, j));
540 }
541 }
542 }
543 for (i, j) in overlaps {
544 for (this, other) in [(i, j), (j, i)] {
545 let text = format!(
546 "overlaps {} ({}): a GET under both selectors is answered by both, \
547 duplicate and possibly divergent — accept it (subscribers are unaffected; \
548 GET consumers consolidate) or carve one selector out of the other",
549 storages[other].name, storages[other].key_expr
550 );
551 storages[this]
552 .warnings
553 .push(warn(WarningKind::Overlap, "RFC 09 §2", text));
554 }
555 }
556
557 StoragePlan {
558 base,
559 registry,
560 volumes,
561 storages,
562 refusals,
563 }
564}
565
566fn json5_key(k: &str) -> String {
570 let bare = !k.is_empty()
571 && !k.starts_with(|c: char| c.is_ascii_digit())
572 && k.chars().all(|c| c.is_ascii_alphanumeric() || c == '_');
573 if bare {
574 k.to_string()
575 } else {
576 serde_json::to_string(k).expect("a string serializes")
577 }
578}
579
580fn json5_value(v: &serde_json::Value) -> String {
582 match v {
583 serde_json::Value::Object(m) => {
584 let inner: Vec<String> = m
585 .iter()
586 .map(|(k, v)| format!("{}: {}", json5_key(k), json5_value(v)))
587 .collect();
588 format!("{{ {} }}", inner.join(", "))
589 }
590 other => serde_json::to_string(other).expect("a value serializes"),
591 }
592}
593
594fn json5_object<'a>(
596 head: impl IntoIterator<Item = (&'a str, serde_json::Value)>,
597 params: &BTreeMap<String, serde_json::Value>,
598) -> String {
599 let mut parts: Vec<String> = head
600 .into_iter()
601 .map(|(k, v)| format!("{}: {}", json5_key(k), json5_value(&v)))
602 .collect();
603 parts.extend(
604 params
605 .iter()
606 .map(|(k, v)| format!("{}: {}", json5_key(k), json5_value(v))),
607 );
608 if parts.is_empty() {
609 "{}".to_string()
610 } else {
611 format!("{{ {} }}", parts.join(", "))
612 }
613}
614
615pub fn to_json5(plan: &StoragePlan) -> String {
623 use std::fmt::Write as _;
624 let mut out = String::new();
625 let _ = writeln!(
626 out,
627 "// zenohd storage_manager block — generated by `zenctl storage gen` (RFC 09 §2)."
628 );
629 let _ = write!(out, "// base {:?}; ", plan.base);
630 match plan.registry.as_option() {
631 Some(r) => {
632 let _ = write!(out, "registry: {} slice(s)", r.slices);
633 match (&r.max_ttl_s, &r.ttl_source) {
634 (Some(t), Some(src)) => {
635 let _ = write!(out, ", longest state ttl_s {t} ({src})");
636 }
637 _ => {
638 let _ = write!(out, ", no state subject declared");
639 }
640 }
641 let _ = writeln!(out, ".");
642 }
643 None => {
644 let _ = writeln!(
645 out,
646 "registry: not asked — every lifespan below is RFC 09 §2.3's default \
647 {DEFAULT_LIFESPAN_S} s, unverified against any ttl_s."
648 );
649 }
650 }
651 let _ = writeln!(
652 out,
653 "// Merge under the router config's `plugins`; every non-memory volume needs its \
654 backend plugin installed, version-matched to the router."
655 );
656 let _ = writeln!(out, "plugins: {{");
657 let _ = writeln!(out, " storage_manager: {{");
658
659 let _ = writeln!(out, " volumes: {{");
661 for v in &plan.volumes {
662 for w in &v.warnings {
663 let _ = writeln!(out, " // ! {}: {} ({})", w.kind_str(), w.text, w.cite);
664 }
665 let mut head: Vec<(&str, serde_json::Value)> = Vec::new();
666 if v.id != v.plugin {
667 head.push(("backend", serde_json::json!(v.plugin)));
668 }
669 if capability(&v.plugin).is_some_and(|(_, fixed)| fixed.is_none()) {
672 head.push(("history", serde_json::json!(v.history.as_str())));
673 }
674 let pair = match v.persistence {
675 Some(p) => format!("{} · {}", p.as_str(), v.history.as_str()),
676 None => format!(
677 "? · {} (plugin not in the capability table)",
678 v.history.as_str()
679 ),
680 };
681 let _ = writeln!(
682 out,
683 " {}: {}, // {pair} (RFC 09 §2.1)",
684 json5_key(&v.id),
685 json5_object(head, &v.params)
686 );
687 }
688 let _ = writeln!(out, " }},");
689
690 let _ = writeln!(out, " storages: {{");
692 for r in &plan.refusals {
693 let what = match (&r.storage, &r.volume) {
694 (Some(s), _) => format!("storage {s}"),
695 (None, Some(v)) => format!("volume {v}"),
696 (None, None) => "entry".to_string(),
697 };
698 let _ = writeln!(out, " // REFUSED {what}: {} ({})", r.reason, r.cite);
699 }
700 for s in &plan.storages {
701 let what = match s.class {
702 Some(c) => format!("class {}", c.as_str()),
703 None => "selector override".to_string(),
704 };
705 let covers = match s.covers.as_option() {
706 Some(n) => format!("; {n} declared subject(s) under it"),
707 None => String::new(),
708 };
709 let _ = writeln!(out, " // {}: {what}{covers}", s.name);
710 for w in &s.warnings {
711 let _ = writeln!(out, " // ! {}: {} ({})", w.kind_str(), w.text, w.cite);
712 }
713 let _ = writeln!(out, " {}: {{", json5_key(&s.name));
714 let _ = writeln!(
715 out,
716 " key_expr: {},",
717 json5_value(&serde_json::json!(s.key_expr))
718 );
719 let _ = writeln!(
720 out,
721 " strip_prefix: {}, // derived: the literal leftmost run of key_expr",
722 json5_value(&serde_json::json!(s.strip_prefix))
723 );
724 let _ = writeln!(
725 out,
726 " volume: {},",
727 json5_object([("id", serde_json::json!(s.volume))], &s.params)
728 );
729 if let Some(rep) = &s.replication {
730 let _ = writeln!(
731 out,
732 " replication: {}, // identical on every replica (RFC 09 §2.2)",
733 json5_object([], rep)
734 );
735 }
736 if s.complete {
737 let _ = writeln!(
738 out,
739 " complete: true, // replicated, fully covering latest storage (RFC 09 §2.2)"
740 );
741 }
742 if let Some(ret) = &s.retention {
743 let _ = writeln!(
744 out,
745 " retention: {}, // the backend's own policy (RFC 09 §2.1)",
746 json5_value(ret)
747 );
748 }
749 let _ = writeln!(
750 out,
751 " garbage_collection: {{ period: {}, lifespan: {} }}, // {} (RFC 09 §2.3)",
752 s.garbage_collection.period_s,
753 s.garbage_collection.lifespan_s,
754 s.garbage_collection.derivation
755 );
756 let _ = writeln!(out, " }},");
757 }
758 let _ = writeln!(out, " }},");
759 let _ = writeln!(out, " }},");
760 let _ = writeln!(out, "}}");
761 out
762}
763
764impl PlanWarning {
765 pub fn kind_str(&self) -> String {
767 match serde_json::to_value(self.kind).expect("a kind serializes") {
768 serde_json::Value::String(s) => s,
769 _ => unreachable!("a unit variant serializes to a string"),
770 }
771 }
772}
773
774fn observed_lifespan(raw: &serde_json::Value) -> Option<f64> {
781 let gc = raw
782 .get("garbage_collection")
783 .or_else(|| raw.get("garbage_collection_config"))?;
784 let lifespan = gc.get("lifespan")?;
785 lifespan
786 .as_f64()
787 .or_else(|| lifespan.get("secs").and_then(serde_json::Value::as_f64))
788}
789
790pub fn check_storages(plan: &StoragePlan, observed: &[StorageInfo]) -> StorageCheck {
798 let mut findings = Vec::new();
799 let mut unjudged = Vec::new();
800 if observed.is_empty() {
801 return StorageCheck {
802 base: plan.base.clone(),
803 asked: CHECK_ASKED.into(),
804 planned: plan.storages.len(),
805 observed: 0,
806 findings,
807 unjudged,
808 judgement: Judgement::Unobservable {
809 reason: "the admin space answered no storages — a peer-only mesh, a router \
810 without the storage manager, or the admin space is disabled; there \
811 is nothing to compare the plan against"
812 .into(),
813 },
814 };
815 }
816 for p in &plan.storages {
817 let rows: Vec<&StorageInfo> = observed.iter().filter(|o| o.name == p.name).collect();
818 if rows.is_empty() {
819 findings.push(CheckFinding {
820 kind: CheckKind::Missing,
821 storage: p.name.clone(),
822 zid: None,
823 planned: Some(p.key_expr.clone()),
824 observed: None,
825 });
826 continue;
827 }
828 for o in rows {
829 let mut differs =
830 |kind: CheckKind, planned: &str, observed: Option<&str>, field| match observed {
831 Some(v) if v == planned => {}
832 Some(v) => findings.push(CheckFinding {
833 kind,
834 storage: p.name.clone(),
835 zid: Some(o.zid.clone()),
836 planned: Some(planned.to_string()),
837 observed: Some(v.to_string()),
838 }),
839 None => unjudged.push(format!(
840 "{}@{}: the admin document does not carry {field}",
841 p.name, o.zid
842 )),
843 };
844 differs(
845 CheckKind::KeyExprDiffers,
846 &p.key_expr,
847 o.key_expr.as_deref(),
848 "key_expr",
849 );
850 differs(
851 CheckKind::StripPrefixDiffers,
852 &p.strip_prefix,
853 o.strip_prefix.as_deref(),
854 "strip_prefix",
855 );
856 differs(
857 CheckKind::VolumeDiffers,
858 &p.volume,
859 o.volume.as_deref(),
860 "volume",
861 );
862 match observed_lifespan(&o.raw) {
863 Some(l) if l < p.garbage_collection.lifespan_s as f64 => {
864 findings.push(CheckFinding {
865 kind: CheckKind::LifespanBelowMinimum,
866 storage: p.name.clone(),
867 zid: Some(o.zid.clone()),
868 planned: Some(p.garbage_collection.lifespan_s.to_string()),
869 observed: Some(l.to_string()),
870 });
871 }
872 Some(_) => {}
873 None => unjudged.push(format!(
874 "{}@{}: the admin document does not carry garbage_collection.lifespan",
875 p.name, o.zid
876 )),
877 }
878 }
879 }
880 for o in observed {
881 if !plan.storages.iter().any(|p| p.name == o.name) {
882 findings.push(CheckFinding {
883 kind: CheckKind::Extra,
884 storage: o.name.clone(),
885 zid: Some(o.zid.clone()),
886 planned: None,
887 observed: o.key_expr.clone(),
888 });
889 }
890 }
891 let judgement = if findings.is_empty() {
892 Judgement::NotEstablished {
893 reason: format!(
894 "every planned storage runs as planned on {} observed row(s)",
895 observed.len()
896 ),
897 }
898 } else {
899 Judgement::Established
900 };
901 StorageCheck {
902 base: plan.base.clone(),
903 asked: CHECK_ASKED.into(),
904 planned: plan.storages.len(),
905 observed: observed.len(),
906 findings,
907 unjudged,
908 judgement,
909 }
910}
911
912pub fn explain(plan: &StoragePlan, key: &str) -> StorageExplain {
916 let mut takers = Vec::new();
917 let mut refused_takers = Vec::new();
918 let Ok(k) = keyexpr::new(key) else {
919 return StorageExplain {
920 key: key.to_string(),
921 base: plan.base.clone(),
922 takers,
923 refused_takers,
924 none_reason: Some(format!("{key:?} is not a valid key expression (RFC 03 §2)")),
925 };
926 };
927 for s in &plan.storages {
928 let Ok(ke) = keyexpr::new(s.key_expr.as_str()) else {
929 continue;
930 };
931 let relation = if ke.includes(k) {
932 TakerRelation::Includes
933 } else if ke.intersects(k) {
934 TakerRelation::Intersects
935 } else {
936 continue;
937 };
938 let basis = match s.class {
939 Some(c) => format!("class {} under base {:?}", c.as_str(), plan.base),
940 None => format!("a selector override under base {:?}", plan.base),
941 };
942 let why = match relation {
943 TakerRelation::Includes => format!(
944 "{basis}: {} includes every key {key} names; stored under strip_prefix {:?} on volume {} ({})",
945 s.key_expr,
946 s.strip_prefix,
947 s.volume,
948 s.history.as_str()
949 ),
950 TakerRelation::Intersects => format!(
951 "{basis}: {} intersects {key} — some keys under it land here, not all",
952 s.key_expr
953 ),
954 };
955 takers.push(Taker {
956 storage: s.name.clone(),
957 key_expr: s.key_expr.clone(),
958 class: s.class,
959 relation,
960 why,
961 });
962 }
963 for r in &plan.refusals {
964 if let (Some(name), Some(ke)) = (&r.storage, &r.key_expr)
965 && keyexpr::new(ke.as_str()).is_ok_and(|ke| ke.includes(k))
966 {
967 refused_takers.push(name.clone());
968 }
969 }
970 let none_reason = takers.is_empty().then(|| {
971 let at_origin = zenkey::grammar::strip_base(&plan.base, key)
972 .and_then(|rel| rel.split('/').nth(1).map(|c| c.starts_with('@')))
973 .unwrap_or(false);
974 let mut reason = String::from("no planned storage's selector includes it");
975 if at_origin {
976 reason.push_str(
977 " — its origin is an @-chunk, and `*` never matches one: a service's state \
978 needs its own explicit storage (RFC 03 §4 D4)",
979 );
980 }
981 if !refused_takers.is_empty() {
982 reason.push_str(&format!(
983 "; refused storage(s) {} would have",
984 refused_takers.join(", ")
985 ));
986 }
987 reason
988 });
989 StorageExplain {
990 key: key.to_string(),
991 base: plan.base.clone(),
992 takers,
993 refused_takers,
994 none_reason,
995 }
996}
997
998#[cfg(test)]
999mod tests {
1000 use super::*;
1001 use crate::report::{StorageSpec, VolumeSpec};
1002
1003 fn fixture_registry() -> SliceSet {
1004 let dir =
1005 std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("../fixture-tests/registry");
1006 SliceSet::from_dirs(&[dir]).expect("the fixture registry reads")
1007 }
1008
1009 fn volume(plugin: &str, history: Option<HistoryMode>) -> VolumeSpec {
1010 VolumeSpec {
1011 plugin: plugin.into(),
1012 history,
1013 params: BTreeMap::new(),
1014 }
1015 }
1016
1017 fn storage(class: StorageClass, volume: &str) -> StorageSpec {
1018 StorageSpec {
1019 class: Some(class),
1020 volume: volume.into(),
1021 ..Default::default()
1022 }
1023 }
1024
1025 fn reference() -> Deployment {
1027 let mut d = Deployment {
1028 base: Some("zensight".into()),
1029 ..Default::default()
1030 };
1031 d.volumes.insert("fs".into(), volume("fs", None));
1032 d.volumes
1033 .insert("influxdb".into(), volume("influxdb", None));
1034 d.storages.insert("latest".into(), {
1035 let mut s = storage(StorageClass::State, "fs");
1036 s.replication = Replication::Enabled(true);
1037 s.complete = true;
1038 s
1039 });
1040 d.storages.insert(
1041 "timeseries".into(),
1042 storage(StorageClass::Telemetry, "influxdb"),
1043 );
1044 d.storages
1045 .insert("catalog".into(), storage(StorageClass::Catalog, "fs"));
1046 d.storages.insert(
1047 "pdns_history".into(),
1048 storage(StorageClass::CatalogPdns, "influxdb"),
1049 );
1050 d
1051 }
1052
1053 fn by_name<'p>(plan: &'p StoragePlan, name: &str) -> &'p PlannedStorage {
1054 plan.storages
1055 .iter()
1056 .find(|s| s.name == name)
1057 .unwrap_or_else(|| panic!("{name} planned; refusals: {:?}", plan.refusals))
1058 }
1059
1060 #[test]
1061 fn the_literal_prefix_stops_at_the_first_wildcard() {
1062 assert_eq!(literal_prefix("zensight/v1/*/state/**"), "zensight/v1");
1063 assert_eq!(
1064 literal_prefix("zensight/v1/@catalog/state/pdns/**"),
1065 "zensight/v1/@catalog/state/pdns"
1066 );
1067 assert_eq!(literal_prefix("v1/*/state/**"), "v1");
1068 assert_eq!(literal_prefix("**"), "");
1069 assert_eq!(literal_prefix("a/b$*/c"), "a");
1070 }
1071
1072 #[test]
1075 fn lifespans_are_derived_from_the_registry_and_shown() {
1076 let plan = plan_storages(Some(&fixture_registry()), "", &reference());
1077 assert!(plan.refusals.is_empty(), "{:?}", plan.refusals);
1078 let facts = plan.registry.as_option().expect("asked");
1079 assert_eq!(facts.max_ttl_s, Some(31_536_000));
1080 assert_eq!(facts.ttl_source.as_deref(), Some("catalog/alias/{old_id}"));
1081
1082 let latest = by_name(&plan, "latest");
1083 assert_eq!(latest.key_expr, "zensight/v1/*/state/**");
1084 assert_eq!(latest.strip_prefix, "zensight/v1");
1085 assert_eq!(latest.garbage_collection.lifespan_s, 1800);
1086 assert_eq!(
1087 latest.garbage_collection.derivation,
1088 "max ttl_s 900 (gnmi/artifact/{kind}) × 2.0 = 1800 s"
1090 );
1091 assert!(latest.complete, "replicated, latest-mode, class state");
1092 assert!(latest.replication.is_some());
1093
1094 let catalog = by_name(&plan, "catalog");
1095 assert_eq!(catalog.strip_prefix, "zensight/v1/@catalog/state");
1096 assert_eq!(catalog.garbage_collection.lifespan_s, 63_072_000);
1097
1098 let pdns = by_name(&plan, "pdns_history");
1099 assert_eq!(pdns.strip_prefix, "zensight/v1/@catalog/state/pdns");
1100 assert_eq!(pdns.garbage_collection.lifespan_s, 172_800);
1101 assert!(
1102 pdns.garbage_collection
1103 .derivation
1104 .contains("catalog/pdns/{ip_slug}")
1105 );
1106
1107 let ts = by_name(&plan, "timeseries");
1108 assert_eq!(ts.garbage_collection.lifespan_s, DEFAULT_LIFESPAN_S);
1109 assert!(
1110 ts.garbage_collection
1111 .derivation
1112 .contains("no state subject")
1113 );
1114 assert!(
1115 ts.warnings
1116 .iter()
1117 .any(|w| w.kind == WarningKind::RetentionIsTheDatabases && w.cite == "RFC 09 §2.3")
1118 );
1119 }
1120
1121 #[test]
1124 fn catalog_and_pdns_history_overlap_and_state_does_not() {
1125 let plan = plan_storages(Some(&fixture_registry()), "", &reference());
1126 let overlaps = |name: &str| -> Vec<String> {
1127 by_name(&plan, name)
1128 .warnings
1129 .iter()
1130 .filter(|w| w.kind == WarningKind::Overlap)
1131 .map(|w| w.text.clone())
1132 .collect()
1133 };
1134 assert!(overlaps("catalog")[0].contains("overlaps pdns_history"));
1135 assert!(overlaps("pdns_history")[0].contains("overlaps catalog"));
1136 assert!(overlaps("latest").is_empty(), "`*` never matches @catalog");
1137 }
1138
1139 #[test]
1142 fn replication_on_an_all_mode_volume_is_refused() {
1143 let mut d = reference();
1144 d.storages.get_mut("timeseries").unwrap().replication = Replication::Enabled(true);
1145 let plan = plan_storages(Some(&fixture_registry()), "", &d);
1146 assert!(plan.storages.iter().all(|s| s.name != "timeseries"));
1147 let r = plan
1148 .refusals
1149 .iter()
1150 .find(|r| r.storage.as_deref() == Some("timeseries"))
1151 .expect("refused");
1152 assert_eq!(r.cite, "RFC 09 §2.2");
1153 assert_eq!(r.key_expr.as_deref(), Some("zensight/v1/*/telemetry/**"));
1154 }
1155
1156 #[test]
1159 fn complete_is_refused_off_the_replicated_latest_storage() {
1160 let mut d = reference();
1161 d.storages.get_mut("catalog").unwrap().complete = true;
1162 d.storages.get_mut("latest").unwrap().replication = Replication::Enabled(false);
1163 let plan = plan_storages(Some(&fixture_registry()), "", &d);
1164 for name in ["catalog", "latest"] {
1165 let s = by_name(&plan, name);
1166 assert!(!s.complete);
1167 assert!(
1168 s.warnings
1169 .iter()
1170 .any(|w| w.kind == WarningKind::CompleteRefused && w.cite == "RFC 09 §2.2"),
1171 "{name}: {:?}",
1172 s.warnings
1173 );
1174 }
1175 }
1176
1177 #[test]
1180 fn undeclared_volumes_and_empty_coverage_are_refused() {
1181 let mut d = reference();
1182 d.storages
1183 .insert("events".into(), storage(StorageClass::Events, "influxdb"));
1184 d.storages
1185 .insert("stray".into(), storage(StorageClass::State, "nope"));
1186 d.volumes.insert("redb".into(), volume("redb", None));
1187 let plan = plan_storages(Some(&fixture_registry()), "", &d);
1188 let refused = |name: &str| {
1189 plan.refusals
1190 .iter()
1191 .find(|r| r.storage.as_deref() == Some(name) || r.volume.as_deref() == Some(name))
1192 .unwrap_or_else(|| panic!("{name} refused: {:?}", plan.refusals))
1193 };
1194 assert!(refused("events").reason.contains("declares no subject"));
1195 assert!(refused("stray").reason.contains("does not declare"));
1196 assert!(refused("redb").reason.contains("per volume"));
1197 assert_eq!(plan.storages.len(), 4);
1198 }
1199
1200 #[test]
1203 fn redb_retention_is_judged_by_the_volumes_mode() {
1204 let mut d = reference();
1205 d.volumes.insert(
1206 "redb-history".into(),
1207 volume("redb", Some(HistoryMode::All)),
1208 );
1209 d.volumes
1210 .insert("redb".into(), volume("redb", Some(HistoryMode::Latest)));
1211 d.storages.insert(
1212 "timeseries".into(),
1213 storage(StorageClass::Telemetry, "redb-history"),
1214 );
1215 d.storages.insert("catalog".into(), {
1216 let mut s = storage(StorageClass::Catalog, "redb");
1217 s.retention = Some(serde_json::json!({"max_age_s": 60}));
1218 s
1219 });
1220 let plan = plan_storages(Some(&fixture_registry()), "", &d);
1221 assert!(
1222 by_name(&plan, "timeseries")
1223 .warnings
1224 .iter()
1225 .any(|w| w.kind == WarningKind::RetentionRequired)
1226 );
1227 assert!(
1228 by_name(&plan, "catalog")
1229 .warnings
1230 .iter()
1231 .any(|w| w.kind == WarningKind::RetentionPointless)
1232 );
1233 d.volumes
1236 .insert("fs".into(), volume("fs", Some(HistoryMode::All)));
1237 let plan = plan_storages(Some(&fixture_registry()), "", &d);
1238 assert!(
1239 plan.refusals
1240 .iter()
1241 .any(|r| r.volume.as_deref() == Some("fs"))
1242 );
1243 assert!(
1244 plan.refusals
1245 .iter()
1246 .any(|r| r.storage.as_deref() == Some("latest") && r.reason.contains("refused"))
1247 );
1248 }
1249
1250 #[test]
1253 fn without_a_registry_the_plan_degrades_and_says_so() {
1254 let mut d = reference();
1255 d.storages
1256 .insert("events".into(), storage(StorageClass::Events, "influxdb"));
1257 let plan = plan_storages(None, "", &d);
1258 assert!(plan.registry.is_not_asked());
1259 assert!(plan.refusals.is_empty());
1260 let latest = by_name(&plan, "latest");
1261 assert_eq!(latest.garbage_collection.lifespan_s, DEFAULT_LIFESPAN_S);
1262 assert_eq!(
1263 latest.garbage_collection.derivation,
1264 "no registry: RFC 09 §2.3 default 86400 s, unverified"
1265 );
1266 assert!(latest.covers.is_not_asked());
1267 assert!(to_json5(&plan).contains("registry: not asked"));
1268 }
1269
1270 #[test]
1273 fn the_base_falls_back_and_the_empty_base_is_the_identity() {
1274 let mut d = reference();
1275 d.base = None;
1276 let plan = plan_storages(None, "", &d);
1277 assert_eq!(by_name(&plan, "latest").key_expr, "v1/*/state/**");
1278 assert_eq!(by_name(&plan, "latest").strip_prefix, "v1");
1279 let plan = plan_storages(None, "acme", &d);
1280 assert_eq!(by_name(&plan, "latest").key_expr, "acme/v1/*/state/**");
1281 }
1282
1283 #[test]
1286 fn the_json5_carries_the_block_and_its_comments() {
1287 let plan = plan_storages(Some(&fixture_registry()), "", &reference());
1288 let doc = to_json5(&plan);
1289 assert!(doc.contains("plugins: {\n storage_manager: {\n volumes: {"));
1290 assert!(doc.contains(" fs: {}, // durable · latest (RFC 09 §2.1)"));
1291 assert!(doc.contains(" key_expr: \"zensight/v1/*/state/**\","));
1292 assert!(doc.contains(" strip_prefix: \"zensight/v1\","));
1293 assert!(
1294 doc.contains("garbage_collection: { period: 30, lifespan: 1800 }, // max ttl_s 900")
1295 );
1296 assert!(doc.contains("replication: { hot: 6, interval: 10.0, propagation_delay: 250, sub_intervals: 5, warm: 30 }"));
1297 assert!(doc.contains(" complete: true,"));
1298 assert!(doc.contains(" // ! overlap: overlaps pdns_history"));
1299 assert!(doc.contains(" // ! retention_is_the_databases:"));
1300 assert!(
1301 !doc.contains("redb"),
1302 "nothing said of redb's retention on other rows"
1303 );
1304 assert_eq!(json5_key("redb-history"), "\"redb-history\"");
1306 assert_eq!(json5_key("fs"), "fs");
1307 let refused = {
1308 let mut d = reference();
1309 d.storages
1310 .insert("events".into(), storage(StorageClass::Events, "influxdb"));
1311 to_json5(&plan_storages(Some(&fixture_registry()), "", &d))
1312 };
1313 assert!(refused.contains(" // REFUSED storage events:"));
1314 }
1315
1316 fn observed(
1317 name: &str,
1318 key_expr: &str,
1319 strip: &str,
1320 volume: &str,
1321 lifespan: Option<i64>,
1322 ) -> StorageInfo {
1323 let mut raw = serde_json::json!({"key_expr": key_expr});
1324 if let Some(l) = lifespan {
1325 raw["garbage_collection"] = serde_json::json!({"period": 30, "lifespan": l});
1326 }
1327 StorageInfo {
1328 zid: "aabbccdd".into(),
1329 name: name.into(),
1330 key_expr: Some(key_expr.into()),
1331 strip_prefix: Some(strip.into()),
1332 volume: Some(volume.into()),
1333 raw,
1334 }
1335 }
1336
1337 #[test]
1340 fn the_check_diffs_the_plan_against_what_runs() {
1341 let plan = plan_storages(Some(&fixture_registry()), "", &reference());
1342
1343 let empty = check_storages(&plan, &[]);
1344 assert!(empty.judgement.is_unobservable());
1345 assert_eq!(crate::judgement_exit_code(&empty.judgement), 2);
1346
1347 let clean = vec![
1348 observed(
1349 "latest",
1350 "zensight/v1/*/state/**",
1351 "zensight/v1",
1352 "fs",
1353 Some(1800),
1354 ),
1355 observed(
1356 "timeseries",
1357 "zensight/v1/*/telemetry/**",
1358 "zensight/v1",
1359 "influxdb",
1360 Some(86400),
1361 ),
1362 observed(
1363 "catalog",
1364 "zensight/v1/@catalog/state/**",
1365 "zensight/v1/@catalog/state",
1366 "fs",
1367 Some(63_072_000),
1368 ),
1369 observed(
1370 "pdns_history",
1371 "zensight/v1/@catalog/state/pdns/**",
1372 "zensight/v1/@catalog/state/pdns",
1373 "influxdb",
1374 Some(172_800),
1375 ),
1376 ];
1377 let c = check_storages(&plan, &clean);
1378 assert!(c.findings.is_empty(), "{:?}", c.findings);
1379 assert_eq!(crate::judgement_exit_code(&c.judgement), 0);
1380 assert_eq!(c.asked, CHECK_ASKED);
1381
1382 let drifted = vec![
1383 observed(
1385 "latest",
1386 "zensight/v1/*/state/**",
1387 "zensight",
1388 "fs",
1389 Some(600),
1390 ),
1391 observed(
1393 "timeseries",
1394 "zensight/v1/**/telemetry/**",
1395 "zensight/v1",
1396 "memory",
1397 Some(86400),
1398 ),
1399 observed(
1401 "catalog",
1402 "zensight/v1/@catalog/state/**",
1403 "zensight/v1/@catalog/state",
1404 "fs",
1405 None,
1406 ),
1407 observed("blobs", "zensight/v1/*/@blob/**", "zensight/v1", "fs", None),
1409 ];
1410 let c = check_storages(&plan, &drifted);
1411 let kinds: Vec<(CheckKind, &str)> = c
1412 .findings
1413 .iter()
1414 .map(|f| (f.kind, f.storage.as_str()))
1415 .collect();
1416 assert_eq!(
1417 kinds,
1418 [
1419 (CheckKind::StripPrefixDiffers, "latest"),
1420 (CheckKind::LifespanBelowMinimum, "latest"),
1421 (CheckKind::Missing, "pdns_history"),
1422 (CheckKind::KeyExprDiffers, "timeseries"),
1423 (CheckKind::VolumeDiffers, "timeseries"),
1424 (CheckKind::Extra, "blobs"),
1425 ]
1426 );
1427 assert_eq!(
1428 c.unjudged,
1429 ["catalog@aabbccdd: the admin document does not carry garbage_collection.lifespan"]
1430 );
1431 assert_eq!(crate::judgement_exit_code(&c.judgement), 1);
1432
1433 assert_eq!(
1435 observed_lifespan(
1436 &serde_json::json!({"garbage_collection_config": {"lifespan": {"secs": 7, "nanos": 0}}})
1437 ),
1438 Some(7.0)
1439 );
1440 }
1441
1442 #[test]
1445 fn explain_names_the_taker_or_the_reason_there_is_none() {
1446 let mut d = reference();
1447 d.storages.remove("catalog");
1448 d.storages
1449 .insert("events".into(), storage(StorageClass::Events, "influxdb"));
1450 let plan = plan_storages(Some(&fixture_registry()), "", &d);
1451
1452 let e = explain(&plan, "zensight/v1/h-3fa9c2d41b7e/state/sysinfo/health");
1453 assert_eq!(e.takers.len(), 1);
1454 assert_eq!(e.takers[0].storage, "latest");
1455 assert_eq!(e.takers[0].relation, TakerRelation::Includes);
1456 assert!(
1457 e.takers[0]
1458 .why
1459 .contains("class state under base \"zensight\"")
1460 );
1461 assert!(e.none_reason.is_none());
1462
1463 let e = explain(&plan, "zensight/v1/@catalog/state/entity/x");
1464 assert!(e.takers.is_empty());
1465 assert!(e.none_reason.as_deref().unwrap().contains("RFC 03 §4 D4"));
1466
1467 let e = explain(
1468 &plan,
1469 "zensight/v1/h-3fa9c2d41b7e/events/netring/capture/01J",
1470 );
1471 assert!(e.takers.is_empty());
1472 assert_eq!(e.refused_takers, ["events"]);
1473 assert!(
1474 e.none_reason
1475 .as_deref()
1476 .unwrap()
1477 .contains("refused storage(s) events would have")
1478 );
1479
1480 let e = explain(&plan, "zensight/v1/*/state/**");
1481 assert_eq!(e.takers[0].relation, TakerRelation::Includes);
1482 let e = explain(&plan, "zensight/v1/**");
1483 assert!(
1484 e.takers
1485 .iter()
1486 .all(|t| t.relation == TakerRelation::Intersects)
1487 );
1488
1489 let e = explain(&plan, "a//b");
1490 assert!(
1491 e.none_reason
1492 .as_deref()
1493 .unwrap()
1494 .contains("not a valid key expression")
1495 );
1496 }
1497}