zenkey_fleet/model/
impact.rs

1//! Impact attribution — a pure function of the edge set, the firing set
2//! and the down set (#389, RFC 06 §5.6), and the origin → entity join it
3//! needs (RFC 06 §5.1). No session, no clock, no private state: a
4//! notifier runs it on its tick, a `.zrec` replay runs it on a recording.
5//!
6//! - **Only containment kinds propagate** ([`EdgeKind::propagates`](crate::report::EdgeKind::propagates));
7//!   `l2_adjacent` and any unknown kind are inert.
8//! - **A root is a down entity with no down containment ancestor**; every
9//!   other affected site the walk reaches is a symptom carrying its
10//!   nearest root.
11//! - **The walk is bounded** — a depth cap and a visited set, because the
12//!   graph is built from claims by independent sensors and a cycle is a
13//!   reachable input — and what the bound cost is reported, never
14//!   absorbed (RFC 13 §3 O6).
15//! - **The output is ordered**: roots and symptoms by entity id, so two
16//!   consumers with the same inputs render the same thing.
17
18use std::collections::{BTreeMap, BTreeSet, VecDeque};
19
20use crate::report::{AliasDoc, EdgeDoc, EntityDoc, ImpactReport, Root, Symptom};
21
22/// What [`attribute`] reads.
23#[derive(Debug, Clone, Copy)]
24pub struct ImpactInputs<'a> {
25    /// The catalog's resolved edges, as observed.
26    pub edges: &'a [EdgeDoc],
27    /// Entities with a firing alert (or any firing notice) on them.
28    pub firing: &'a BTreeSet<String>,
29    /// Entities the consumer judged down — every member origin's alive
30    /// token gone (RFC 06 §5.6: downness is the consumer's decision).
31    pub down: &'a BTreeSet<String>,
32}
33
34/// The largest depth cap a consumer should accept: the reference
35/// implementation's, and what a notifier's config check enforces.
36pub const MAX_DEPTH_CAP: usize = 4;
37
38type Adjacency<'a> = BTreeMap<&'a str, BTreeSet<&'a str>>;
39
40/// Roots and symptoms over the containment graph, walking at most
41/// `depth_cap` edges from any entity.
42pub fn attribute(inputs: &ImpactInputs<'_>, depth_cap: usize) -> ImpactReport {
43    let mut children: Adjacency<'_> = BTreeMap::new();
44    let mut parents: Adjacency<'_> = BTreeMap::new();
45    for e in inputs.edges {
46        if !e.kind.propagates() {
47            continue;
48        }
49        let (Some(from), Some(to)) = (e.from.entity_id(), e.to.entity_id()) else {
50            // An `External` end has no liveliness to lose and cannot be in
51            // `down`: an edge touching one neither carries nor roots impact.
52            continue;
53        };
54        if from == to {
55            // A self-edge would make every entity its own ancestor.
56            continue;
57        }
58        children.entry(from).or_default().insert(to);
59        parents.entry(to).or_default().insert(from);
60    }
61
62    let mut walks_capped = 0u64;
63    let mut cycles_seen = 0u64;
64
65    // Roots: down, with no down ancestor within the cap.
66    let roots: Vec<&str> = inputs
67        .down
68        .iter()
69        .map(String::as_str)
70        .filter(|id| !has_down_ancestor(id, &parents, inputs.down, depth_cap, &mut walks_capped))
71        .collect();
72
73    // Nearest root wins; ties by root id ascending (roots iterate sorted,
74    // and a strictly nearer candidate is the only thing that replaces one).
75    let mut best: BTreeMap<&str, (usize, &str)> = BTreeMap::new();
76    let mut out_roots = Vec::with_capacity(roots.len());
77    for root in &roots {
78        let reach = descendants(
79            root,
80            &children,
81            depth_cap,
82            &mut walks_capped,
83            &mut cycles_seen,
84        );
85        for (id, depth) in &reach {
86            let cand = (*depth, *root);
87            best.entry(id)
88                .and_modify(|cur| {
89                    if cand < *cur {
90                        *cur = cand;
91                    }
92                })
93                .or_insert(cand);
94        }
95        out_roots.push(Root {
96            entity: (*root).to_string(),
97            reached: reach.keys().map(|s| (*s).to_string()).collect(),
98        });
99    }
100
101    // Symptoms: every firing or down site a root reaches that is not itself
102    // a root. A down entity under a down host is as much a symptom as an
103    // alert there — "vm-apps is gone", not nine messages about what it ran.
104    let mut symptoms = Vec::new();
105    for site in inputs.firing.iter().chain(inputs.down.iter()) {
106        let id = site.as_str();
107        if roots.contains(&id) {
108            continue;
109        }
110        if let Some((_, root)) = best.get(id) {
111            symptoms.push(Symptom {
112                entity: id.to_string(),
113                explained_by: (*root).to_string(),
114            });
115        }
116    }
117    symptoms.sort();
118    symptoms.dedup();
119
120    ImpactReport {
121        roots: out_roots,
122        symptoms,
123        walks_capped,
124        cycles_seen,
125        depth_cap,
126    }
127}
128
129/// Whether any containment ancestor within `depth_cap` is itself down.
130fn has_down_ancestor(
131    id: &str,
132    parents: &Adjacency<'_>,
133    down: &BTreeSet<String>,
134    depth_cap: usize,
135    walks_capped: &mut u64,
136) -> bool {
137    let mut seen: BTreeSet<&str> = BTreeSet::from([id]);
138    let mut queue: VecDeque<(&str, usize)> = VecDeque::from([(id, 0usize)]);
139    let mut capped = false;
140    while let Some((cur, depth)) = queue.pop_front() {
141        let Some(next) = parents.get(cur) else {
142            continue;
143        };
144        if depth >= depth_cap {
145            if next.iter().any(|p| !seen.contains(p)) {
146                capped = true;
147            }
148            continue;
149        }
150        for p in next {
151            if !seen.insert(p) {
152                continue;
153            }
154            if down.contains(*p) {
155                return true;
156            }
157            queue.push_back((p, depth + 1));
158        }
159    }
160    if capped {
161        *walks_capped += 1;
162    }
163    false
164}
165
166/// Every entity reachable downstream within `depth_cap`, with the shortest
167/// depth at which it was reached. The root is excluded; an edge back to it
168/// is a cycle, counted once per walk.
169fn descendants<'a>(
170    root: &'a str,
171    children: &Adjacency<'a>,
172    depth_cap: usize,
173    walks_capped: &mut u64,
174    cycles_seen: &mut u64,
175) -> BTreeMap<&'a str, usize> {
176    let mut out: BTreeMap<&str, usize> = BTreeMap::new();
177    let mut seen: BTreeSet<&str> = BTreeSet::from([root]);
178    let mut queue: VecDeque<(&str, usize)> = VecDeque::from([(root, 0usize)]);
179    let (mut capped, mut cycled) = (false, false);
180    while let Some((cur, depth)) = queue.pop_front() {
181        let Some(next) = children.get(cur) else {
182            continue;
183        };
184        if depth >= depth_cap {
185            if next.iter().any(|c| !seen.contains(c)) {
186                capped = true;
187            }
188            continue;
189        }
190        for c in next {
191            if *c == root {
192                cycled = true;
193                continue;
194            }
195            if !seen.insert(c) {
196                continue;
197            }
198            out.insert(c, depth + 1);
199            queue.push_back((c, depth + 1));
200        }
201    }
202    if capped {
203        *walks_capped += 1;
204    }
205    if cycled {
206        *cycles_seen += 1;
207    }
208    out
209}
210
211/// The RFC 06 §5.1 join, origin → entity: `origins[]` first, the legacy
212/// `host_id` second, then one alias hop so a merged entity resolves to its
213/// survivor. Deterministic under input order: the smallest matching entity
214/// id wins when the documents disagree.
215pub fn entity_of(origin: &str, entities: &[EntityDoc], aliases: &[AliasDoc]) -> Option<String> {
216    let by_origins = entities
217        .iter()
218        .filter(|e| e.origins.iter().any(|o| o == origin))
219        .map(|e| e.entity_id.as_str())
220        .min();
221    let by_host_id = || {
222        entities
223            .iter()
224            .filter(|e| e.host_id.as_deref() == Some(origin))
225            .map(|e| e.entity_id.as_str())
226            .min()
227    };
228    let id = by_origins.or_else(by_host_id)?;
229    let resolved = aliases
230        .iter()
231        .filter(|a| a.old_id == id)
232        .map(|a| a.entity_id.as_str())
233        .min()
234        .unwrap_or(id);
235    Some(resolved.to_string())
236}
237
238#[cfg(test)]
239mod tests {
240    use super::*;
241    use crate::report::{EdgeEnd, EdgeKind};
242
243    fn edge(kind: EdgeKind, from: &str, to: &str) -> EdgeDoc {
244        EdgeDoc {
245            edge_id: format!("e-{}-{from}-{to}", kind.as_str()),
246            kind,
247            from: EdgeEnd::Entity {
248                entity_id: from.into(),
249            },
250            to: EdgeEnd::Entity {
251                entity_id: to.into(),
252            },
253            attrs: BTreeMap::new(),
254            observers: Vec::new(),
255            last_updated: None,
256        }
257    }
258
259    fn set(ids: &[&str]) -> BTreeSet<String> {
260        ids.iter().map(|s| s.to_string()).collect()
261    }
262
263    /// `A hosts B runs C`, A down, an alert on C: A is the root, C a
264    /// symptom explained by A — and B, merely reached, is not a symptom of
265    /// anything because nothing fires there.
266    #[test]
267    fn a_chain_attributes_the_leaf_alert_to_the_down_host() {
268        let edges = [
269            edge(EdgeKind::Hosts, "A", "B"),
270            edge(EdgeKind::Runs, "B", "C"),
271        ];
272        let r = attribute(
273            &ImpactInputs {
274                edges: &edges,
275                firing: &set(&["C"]),
276                down: &set(&["A"]),
277            },
278            MAX_DEPTH_CAP,
279        );
280        assert_eq!(
281            r.roots,
282            vec![Root {
283                entity: "A".into(),
284                reached: vec!["B".into(), "C".into()]
285            }]
286        );
287        assert_eq!(
288            r.symptoms,
289            vec![Symptom {
290                entity: "C".into(),
291                explained_by: "A".into()
292            }]
293        );
294        assert_eq!((r.walks_capped, r.cycles_seen, r.depth_cap), (0, 0, 4));
295
296        // B down as well: B is a symptom of A, not a second root.
297        let r = attribute(
298            &ImpactInputs {
299                edges: &edges,
300                firing: &set(&["C"]),
301                down: &set(&["A", "B"]),
302            },
303            MAX_DEPTH_CAP,
304        );
305        assert_eq!(r.roots.len(), 1);
306        assert_eq!(
307            r.symptoms
308                .iter()
309                .map(|s| s.entity.as_str())
310                .collect::<Vec<_>>(),
311            vec!["B", "C"]
312        );
313    }
314
315    /// A symmetric kind never propagates: two neighbours, one down, the
316    /// other's alert stays a cause of its own. Nor does an unknown kind, or
317    /// an edge into an `External`.
318    #[test]
319    fn l2_adjacent_and_unknown_kinds_never_propagate() {
320        let mut edges = vec![
321            edge(EdgeKind::L2Adjacent, "A", "B"),
322            edge(EdgeKind::Other("teleports".into()), "A", "C"),
323        ];
324        edges.push(EdgeDoc {
325            to: EdgeEnd::External {
326                ip: Some("1.1.1.1".into()),
327                mac: None,
328                name: None,
329            },
330            ..edge(EdgeKind::Probes, "A", "x")
331        });
332        let r = attribute(
333            &ImpactInputs {
334                edges: &edges,
335                firing: &set(&["B", "C"]),
336                down: &set(&["A"]),
337            },
338            MAX_DEPTH_CAP,
339        );
340        assert_eq!(r.roots[0].reached, Vec::<String>::new());
341        assert!(r.symptoms.is_empty(), "{:?}", r.symptoms);
342    }
343
344    /// Two hosts each claiming to host the other: the walk terminates, and
345    /// says it saw the cycle.
346    #[test]
347    fn a_cycle_terminates_and_is_counted_once() {
348        let edges = [
349            edge(EdgeKind::Hosts, "A", "B"),
350            edge(EdgeKind::Hosts, "B", "A"),
351        ];
352        let r = attribute(
353            &ImpactInputs {
354                edges: &edges,
355                firing: &set(&["B"]),
356                down: &set(&["A"]),
357            },
358            MAX_DEPTH_CAP,
359        );
360        assert_eq!(r.cycles_seen, 1);
361        assert_eq!(r.roots[0].entity, "A");
362        assert_eq!(r.symptoms[0].explained_by, "A");
363    }
364
365    /// A chain five deep under the cap of four: the walk stops at four,
366    /// reports the cut, and the fifth is not a symptom of anything.
367    #[test]
368    fn a_walk_past_the_depth_cap_is_cut_and_reported() {
369        let edges = [
370            edge(EdgeKind::Hosts, "A", "B"),
371            edge(EdgeKind::Runs, "B", "C"),
372            edge(EdgeKind::Runs, "C", "D"),
373            edge(EdgeKind::Runs, "D", "E"),
374            edge(EdgeKind::Runs, "E", "F"),
375        ];
376        let r = attribute(
377            &ImpactInputs {
378                edges: &edges,
379                firing: &set(&["E", "F"]),
380                down: &set(&["A"]),
381            },
382            4,
383        );
384        assert_eq!(r.walks_capped, 1);
385        assert_eq!(r.roots[0].reached, vec!["B", "C", "D", "E"]);
386        assert_eq!(
387            r.symptoms
388                .iter()
389                .map(|s| s.entity.as_str())
390                .collect::<Vec<_>>(),
391            vec!["E"]
392        );
393    }
394
395    /// The same inputs in any order render the same report; the nearest
396    /// root wins, ties by id.
397    #[test]
398    fn the_output_is_deterministic_under_input_permutation() {
399        let edges = [
400            edge(EdgeKind::Hosts, "H2", "V"),
401            edge(EdgeKind::Hosts, "H1", "V"),
402            edge(EdgeKind::GatewayOf, "G", "H1"),
403            edge(EdgeKind::Runs, "V", "C"),
404        ];
405        let inputs = |edges: &[EdgeDoc]| {
406            attribute(
407                &ImpactInputs {
408                    edges,
409                    firing: &set(&["C", "V"]),
410                    down: &set(&["H2", "H1", "G"]),
411                },
412                MAX_DEPTH_CAP,
413            )
414        };
415        let a = inputs(&edges);
416        let mut reversed = edges.to_vec();
417        reversed.reverse();
418        let b = inputs(&reversed);
419        assert_eq!(a, b);
420        assert_eq!(
421            a.roots
422                .iter()
423                .map(|r| r.entity.as_str())
424                .collect::<Vec<_>>(),
425            vec!["G", "H2"],
426            "H1 is under G; roots are ordered"
427        );
428        let v = a.symptoms.iter().find(|s| s.entity == "V").unwrap();
429        assert_eq!(v.explained_by, "H2", "H2 is one hop away, G is two");
430    }
431
432    /// `origins[]` first, `host_id` second, one alias hop — and nothing for
433    /// an origin no document names.
434    #[test]
435    fn entity_of_joins_by_origins_then_host_id_then_alias() {
436        let entities = [
437            EntityDoc {
438                entity_id: "ent-new".into(),
439                origins: vec!["h-aaaaaaaaaaaa".into()],
440                host_id: None,
441                hostname: None,
442                rest: BTreeMap::new(),
443            },
444            EntityDoc {
445                entity_id: "ent-old".into(),
446                origins: vec![],
447                host_id: Some("h-bbbbbbbbbbbb".into()),
448                hostname: None,
449                rest: BTreeMap::new(),
450            },
451        ];
452        let aliases = [AliasDoc {
453            old_id: "ent-old".into(),
454            entity_id: "ent-merged".into(),
455            rest: BTreeMap::new(),
456        }];
457        assert_eq!(
458            entity_of("h-aaaaaaaaaaaa", &entities, &aliases).as_deref(),
459            Some("ent-new")
460        );
461        assert_eq!(
462            entity_of("h-bbbbbbbbbbbb", &entities, &aliases).as_deref(),
463            Some("ent-merged"),
464            "host_id match, then the alias re-points it"
465        );
466        assert_eq!(entity_of("h-cccccccccccc", &entities, &aliases), None);
467    }
468}
zenkey_fleet/model/impact.rs

zenkey_fleet/model/
impact.rs
