Skip to main content

zenkey_fleet/model/
storage.rs

1//! The storage planner (RFC 09 §2, #393): a deployment file plus the registry
2//! in, the router's `storage_manager` block out — with every derived number
3//! shown, every caveat cited, and every refusal named.
4//!
5//! RFC 09 §2 specifies the class-driven storages, each with a selector, a
6//! literal `strip_prefix`, a volume from the §2.1 capability table, and a
7//! `garbage_collection.lifespan` that must be ≥ the longest `ttl_s` in the
8//! registry (§2.3, the tombstone-visibility row of RFC 04 §1.2). The registry
9//! knows that number; until now nobody computed it, and the two things a
10//! human types wrong here — a `strip_prefix` that is not a literal prefix of
11//! its selector, and a lifespan chosen by feel — produce a router that starts
12//! happily and stores nothing, or resurrects a retired key from a slow
13//! replica.
14//!
15//! Pure, like everything in [`crate::model`]: values in hand, no session.
16//! [`plan_storages`] takes an *optional* registry and says what it could not
17//! verify without one rather than inventing a number (RFC 13 §3 O4);
18//! [`check_storages`] compares a plan against storages somebody else read
19//! off the admin space; [`explain`] answers "which storage takes this key"
20//! over the plan alone. [`to_json5`] is the one rendering of the plan that
21//! is not zenkey's — it is `zenohd`'s.
22
23use 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
35/// Zenoh's own default `garbage_collection.lifespan`, seconds (RFC 09 §2.3:
36/// "default 24 h").
37pub const DEFAULT_LIFESPAN_S: i64 = 86_400;
38/// Zenoh's own default `garbage_collection.period`, seconds.
39pub const DEFAULT_GC_PERIOD_S: u64 = 30;
40/// The default margin over the longest covered `ttl_s`.
41pub const DEFAULT_GC_MARGIN: f64 = 2.0;
42
43/// The admin selector `--check` reads storages from — the same one
44/// [`crate::storages`] sweeps, restated here so the report can cite it.
45pub const CHECK_ASKED: &str = "@/*/router/**/storage_manager/storages/**";
46
47/// RFC 09 §2.1's capability table: persistence, and the history mode when
48/// the plugin fixes it. `None` history = per volume (`redb`); `None` overall
49/// = a plugin this tool does not know.
50fn 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
60/// RFC 09 §2.2's example replication block — what `replication = true`
61/// means.
62fn 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
75/// The literal leftmost run of a key expression — the one `strip_prefix`
76/// Zenoh accepts (string prefix, no wildcards). `zensight/v1/*/state/**`
77/// → `zensight/v1`; `zensight/v1/@catalog/state/pdns/**` →
78/// `zensight/v1/@catalog/state/pdns`. Empty when the expression opens on a
79/// wildcard.
80pub 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
88/// One declared subject, as a wire family under the base.
89struct Family {
90    producer: String,
91    path: String,
92    is_state: bool,
93    ttl_s: Option<i64>,
94    selector: String,
95}
96
97/// Every declared subject of every class as the selector it occupies on the
98/// wire — the same composition `state_coverage` uses, all classes.
99fn 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
129/// The longest `ttl_s` among state families, and which one carries it —
130/// the alphabetically first of a tie, so the derivation is stable.
131fn 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
160/// Plan the storages of a deployment (#393).
161///
162/// `slices` is the registry when one was asked — `None` degrades every
163/// lifespan to RFC 09 §2.3's default and says so in the derivation, and
164/// skips the coverage refusal (no registry is not an empty registry, RFC 13
165/// §3 O4). `fallback_base` is the observer's resolved base, used when the
166/// deployment file names none.
167pub 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    // ── Volumes: the capability pair, per volume (RFC 09 §2.1 v1.28). ──
192    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    // ── Storages. ──
278    let mut storages: Vec<PlannedStorage> = Vec::new();
279    for (name, spec) in &deployment.storages {
280        // The selector: one of class and override, never both, never neither.
281        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        // The volume, and its mode — the fact every §2.2 decision reads.
315        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        // Replication — refused, not discovered, on an all-mode volume.
330        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        // Coverage: what the registry declares under this selector.
373        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        // The tombstone lifetime, derived (RFC 09 §2.3).
391        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        // `complete: true` — right in exactly one place (RFC 09 §2.2).
444        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        // The volume's row of the §2.1 table, and its caveats.
470        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    // ── Overlaps (RFC 09 §2's documented one is catalog vs pdns_history). ──
530    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
566// ── The zenohd rendering ──────────────────────────────────────────────────
567
568/// A JSON5 object key: bare where JSON5 allows it, quoted otherwise.
569fn 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
580/// A JSON value on one line — JSON is JSON5.
581fn 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
594/// One `{ id: "fs", dir: "latest" }`-shaped object from an id and params.
595fn 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
615/// The plan as the `plugins.storage_manager` block `zenohd` reads — JSON5,
616/// with the derivations and every warning as comments beside the storage
617/// they concern, and the refusals where the refused storage would have been.
618///
619/// Merge it under the router config's `plugins`; each non-memory volume needs
620/// its backend plugin installed and version-matched to the router (RFC 09
621/// §2).
622pub 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    // Volumes.
660    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        // `history` is a volume knob only where the plugin offers both modes;
670        // the fixed rows would refuse an unknown key.
671        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    // Storages.
691    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    /// The kind as its wire token.
766    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
774// ── --check ───────────────────────────────────────────────────────────────
775
776/// The running `garbage_collection.lifespan` of a storage, seconds, as far
777/// as the admin document says. Layouts vary by version — a bare number, or
778/// serde's `{ secs, nanos }` for a `Duration` — and an absent field is
779/// `None`, which is *unjudged* and not *agreeing*.
780fn 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
790/// Compare a plan against the storages a router admits to running (#393) —
791/// the configuration half of `storage list`'s coverage question.
792///
793/// `observed` is what [`crate::storages`] read off the admin space. Empty is
794/// **unobservable**, not clean: a peer-only mesh, a router without the
795/// storage manager and a disabled admin space all answer nothing, and none
796/// of them is a router running the plan.
797pub 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
912// ── --explain ─────────────────────────────────────────────────────────────
913
914/// Which planned storage(s) take `key`, and why (#393). Pure over the plan.
915pub 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    /// The RFC 09 §2 sketch, as a deployment.
1026    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    /// The headline: lifespans come from the registry, per storage, with the
1073    /// computation shown — and `strip_prefix` is derived.
1074    #[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            // Several subjects tie at 900; the alphabetically first names it.
1089            "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    /// The documented overlap, warned on both sides; and `*` versus
1122    /// `@catalog` is *not* one.
1123    #[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    /// RFC 09 §2.2: replication on an all-mode volume is a startup refusal,
1140    /// so the plan refuses first — and names the storage.
1141    #[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    /// `complete = true` anywhere but the one right place is emitted as
1157    /// `false` and said so.
1158    #[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    /// The other refusals: an undeclared volume, a class the registry has no
1178    /// subject for, a redb volume without a mode.
1179    #[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    /// redb's per-volume mode: retention is mandatory in all mode and refused
1201    /// in latest mode (RFC 09 §2.1).
1202    #[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        // And an `fs` volume declared all-mode is a refused volume, taking
1234        // its storages with it.
1235        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    /// No registry: the default lifespan, said to be unverified, and no
1251    /// coverage refusal — not asked is not empty (RFC 13 §3 O4).
1252    #[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    /// The base falls back to the observer's when the file names none; an
1271    /// empty base composes to the bus-root deployment.
1272    #[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    /// The JSON5 is the RFC 09 §2 sketch's shape, with the numbers filled in
1284    /// and the caveats beside the storage they concern.
1285    #[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        // A quoted key where JSON5 needs one, a bare one where it does not.
1305        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    /// `--check` over a hand-built admin reading: every finding kind, and the
1338    /// two non-verdicts (empty admin space; a field the layout omits).
1339    #[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            // Too short a lifespan, and the wrong prefix.
1384            observed(
1385                "latest",
1386                "zensight/v1/*/state/**",
1387                "zensight",
1388                "fs",
1389                Some(600),
1390            ),
1391            // Wrong selector, wrong volume.
1392            observed(
1393                "timeseries",
1394                "zensight/v1/**/telemetry/**",
1395                "zensight/v1",
1396                "memory",
1397                Some(86400),
1398            ),
1399            // Layout omits the gc block.
1400            observed(
1401                "catalog",
1402                "zensight/v1/@catalog/state/**",
1403                "zensight/v1/@catalog/state",
1404                "fs",
1405                None,
1406            ),
1407            // Not planned at all.
1408            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        // serde's Duration shape for the lifespan is read too.
1434        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    /// `--explain`: the taker and its reason; the @-origin key nobody takes;
1443    /// the key a refused storage would have taken.
1444    #[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}