zenkey_fleet/report/consumers.rs
1//! Consumers and blast radius (#224): who *declares* a reader of a subject,
2//! and what changing that subject would reach.
3//!
4//! Every shape here is a join over the admin space (RFC 09 §5.1): declared
5//! subscribers and queriers with their verbatim keyexprs, the origin
6//! attachments the tokens back, and the topology's node roster. None of it
7//! is matching status — RFC 12 §9 defers foreign matching permanently, and
8//! a declaration is evidence that a session *asked for* a key, never proof
9//! that anything reads it. The wording rule follows from that: nothing in
10//! these reports says "listening", "matching" or "unmatched".
11
12use serde::Serialize;
13
14use super::admin::{CoverageRow, EntityKind};
15
16/// The consumers of one target selector, as the admin space declared them.
17#[derive(Debug, Clone, Serialize)]
18pub struct ConsumersReport {
19 /// The wire selector the declarations were related to.
20 pub target: String,
21 /// The admin selectors actually put to the bus (RFC 13 §3 O5) —
22 /// coverage is exactly this list.
23 pub asked: Vec<String>,
24 /// This session's own zid: the tool's own declarations appear in its
25 /// own results, and are named rather than hidden.
26 pub self_zid: String,
27 /// Whether any admin space answered, and how many. Flattened so the
28 /// discriminator rides at the top of the document beside the rows.
29 #[serde(flatten)]
30 pub admin: AdminAnswer,
31 /// One row per declaring session per declaration, ranked by relation.
32 /// Empty under [`AdminAnswer::NotAvailable`] is *not asked*; empty under
33 /// [`AdminAnswer::Answered`] is *nothing declared in the answering admin
34 /// spaces* — the renderers say which.
35 pub rows: Vec<ConsumerRow>,
36 /// Admin replies past the bound, not kept (RFC 13 §3 O6). A declaration
37 /// past the bound is a consumer this report cannot show.
38 pub reply_elided: u64,
39}
40
41/// Whether the admin space answered at all (RFC 13 §3 O4).
42///
43/// zenoh's `adminspace.enabled` defaults to *false* (routers ship with it
44/// on; a pure peer mesh has none), so a sweep nobody answered is *not
45/// asked*, never *nobody declared anything*. The two must not share a
46/// spelling anywhere, which is why this is an enum and not an `answered: 0`.
47#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
48#[serde(tag = "admin", rename_all = "snake_case")]
49pub enum AdminAnswer {
50 /// `answered` admin spaces served root docs; `nodes` is the topology's
51 /// roster, heard-of nodes included — the sessions behind an admin space
52 /// that did not answer are not in the rows.
53 Answered { answered: usize, nodes: usize },
54 /// No admin space answered any sweep.
55 NotAvailable,
56}
57
58/// One declared reader, per session.
59#[derive(Debug, Clone, Serialize)]
60pub struct ConsumerRow {
61 /// The session that declared it — or, under
62 /// [`Attribution::ReportedOnly`], the admin space that reported it.
63 pub zid: String,
64 /// `router` | `peer` | `client`, when the topology sweep heard of the
65 /// zid. Absent when it did not — unknown, not "no kind".
66 #[serde(skip_serializing_if = "Option::is_none")]
67 pub whatami: Option<String>,
68 /// The origins whose alive tokens the admin space attributes to this
69 /// session (#131). Several is legitimate (one process, several
70 /// producers); empty means *session only, unattributed*.
71 #[serde(skip_serializing_if = "Vec::is_empty")]
72 pub origins: Vec<String>,
73 pub attribution: Attribution,
74 /// `subscriber` or `querier` — the two entity kinds that read.
75 pub kind: EntityKind,
76 /// The declared key expression, verbatim.
77 pub keyexpr: String,
78 pub relation: Relation,
79 /// This tool's own session. Shown, and named.
80 #[serde(skip_serializing_if = "std::ops::Not::not")]
81 pub is_self: bool,
82 /// The declaration reaches the whole base (`**`, or a prefix that
83 /// includes every `v1` key): it intersects everything, and is shown as
84 /// such rather than as a consumer of this subject in particular.
85 #[serde(skip_serializing_if = "std::ops::Not::not")]
86 pub total_wildcard: bool,
87}
88
89/// How a row's zid was obtained.
90#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
91#[serde(rename_all = "snake_case")]
92pub enum Attribution {
93 /// The admin `sources` named the declaring session.
94 Session,
95 /// The sources named nobody: the zid is the reporting admin space's,
96 /// and the declaration is only known to exist behind it.
97 ReportedOnly,
98}
99
100/// How a declared keyexpr relates to the target, by key algebra
101/// (`zenoh-keyexpr`'s `includes`/`intersects`). Ranked: the ordering is the
102/// row order, most specific first.
103#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize)]
104#[serde(rename_all = "snake_case")]
105pub enum Relation {
106 /// The declaration and the target are the same set.
107 Exact,
108 /// The target includes the declaration: it hears a subset.
109 Narrower,
110 /// The declaration includes the target, and is not the whole base.
111 Wider,
112 /// The two overlap without either including the other.
113 Intersects,
114 /// The declaration includes the whole base (`**`).
115 Total,
116}
117
118impl Relation {
119 /// The wire token, for renderers that write it into prose.
120 pub fn as_str(self) -> &'static str {
121 match self {
122 Relation::Exact => "exact",
123 Relation::Narrower => "narrower",
124 Relation::Wider => "wider",
125 Relation::Intersects => "intersects",
126 Relation::Total => "total",
127 }
128 }
129}
130
131/// The blast radius of one declared subject (#224): its consumers, its
132/// storage coverage, what else declares on it, and its ledger entry.
133///
134/// Not `ImpactReport` — that name is the zenwatch attribution's
135/// (RFC 06 §5.6), a different question about a different plane.
136#[derive(Debug, Clone, Serialize)]
137pub struct SubjectImpact {
138 pub producer: String,
139 /// The subject path as the registry spells it (`disk/{mount}/used`).
140 pub path: String,
141 /// The declared class chunk; `*` when the path is known only from the
142 /// `[[deprecated]]` ledger, whose entries carry no class.
143 pub class: String,
144 /// The wire selector the subject's family resolves to under the base.
145 pub selector: String,
146 pub consumers: ConsumersReport,
147 /// The RFC 04 §2 coverage rows for this subject. `None` = the storage
148 /// sweep was not made (no admin space answered, so an empty list would
149 /// read as "uncovered"); `Some([])` = asked, and the subject is not a
150 /// state family — coverage does not apply.
151 #[serde(skip_serializing_if = "Option::is_none")]
152 pub coverage: Option<Vec<CoverageRow>>,
153 /// Distinct sessions declaring a publisher intersecting the selector.
154 /// Absent when no admin space answered (not asked, O4).
155 #[serde(skip_serializing_if = "Option::is_none")]
156 pub declared_publishers: Option<usize>,
157 /// Distinct sessions declaring a queryable intersecting the selector.
158 #[serde(skip_serializing_if = "Option::is_none")]
159 pub declared_queryables: Option<usize>,
160 /// The subject's `[[deprecated]]` entry, when the ledger carries one.
161 #[serde(skip_serializing_if = "Option::is_none")]
162 pub deprecated: Option<DeprecationFact>,
163}
164
165/// What the registry ledger says about a retired subject (RFC 08 §3).
166#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
167pub struct DeprecationFact {
168 #[serde(skip_serializing_if = "Option::is_none")]
169 pub since: Option<String>,
170 #[serde(skip_serializing_if = "Option::is_none")]
171 pub replaced_by: Option<String>,
172}
173
174#[cfg(test)]
175mod tests {
176 use super::*;
177 use serde_json::json;
178
179 fn row() -> ConsumerRow {
180 ConsumerRow {
181 zid: "eeff0011".into(),
182 whatami: Some("peer".into()),
183 origins: vec!["h-3fa9c2d41b7e".into()],
184 attribution: Attribution::Session,
185 kind: EntityKind::Subscriber,
186 keyexpr: "v1/h-3fa9c2d41b7e/state/sysinfo/health".into(),
187 relation: Relation::Narrower,
188 is_self: false,
189 total_wildcard: false,
190 }
191 }
192
193 /// The two admin answers never share a spelling: `not_available`
194 /// carries no counts at all, and `answered` always carries both.
195 #[test]
196 fn the_admin_answer_is_a_discriminator_not_a_zero() {
197 let mut r = ConsumersReport {
198 target: "v1/*/state/sysinfo/health".into(),
199 asked: vec!["@/*/*".into()],
200 self_zid: "ffffffff".into(),
201 admin: AdminAnswer::Answered {
202 answered: 1,
203 nodes: 2,
204 },
205 rows: vec![],
206 reply_elided: 0,
207 };
208 let v = serde_json::to_value(&r).unwrap();
209 assert_eq!(v["admin"], "answered");
210 assert_eq!(v["answered"], 1);
211 assert_eq!(v["nodes"], 2);
212
213 r.admin = AdminAnswer::NotAvailable;
214 let v = serde_json::to_value(&r).unwrap();
215 assert_eq!(v["admin"], "not_available");
216 assert!(v.get("answered").is_none(), "{v}");
217 assert!(v.get("nodes").is_none(), "{v}");
218 }
219
220 /// `origins` is absent when empty (session only, unattributed) and
221 /// `is_self`/`total_wildcard` are absent when false — news, not
222 /// non-news (O4).
223 #[test]
224 fn a_row_omits_what_is_not_news() {
225 let v = serde_json::to_value(row()).unwrap();
226 assert_eq!(
227 v,
228 json!({
229 "zid": "eeff0011",
230 "whatami": "peer",
231 "origins": ["h-3fa9c2d41b7e"],
232 "attribution": "session",
233 "kind": "subscriber",
234 "keyexpr": "v1/h-3fa9c2d41b7e/state/sysinfo/health",
235 "relation": "narrower",
236 })
237 );
238 let bare = ConsumerRow {
239 whatami: None,
240 origins: vec![],
241 attribution: Attribution::ReportedOnly,
242 is_self: true,
243 total_wildcard: true,
244 relation: Relation::Total,
245 keyexpr: "**".into(),
246 ..row()
247 };
248 let v = serde_json::to_value(bare).unwrap();
249 assert!(v.get("origins").is_none(), "{v}");
250 assert!(v.get("whatami").is_none(), "{v}");
251 assert_eq!(v["is_self"], true);
252 assert_eq!(v["total_wildcard"], true);
253 assert_eq!(v["attribution"], "reported_only");
254 assert_eq!(v["relation"], "total");
255 }
256
257 /// Relations rank most-specific first — the row order.
258 #[test]
259 fn relations_rank_specific_first() {
260 assert!(Relation::Exact < Relation::Narrower);
261 assert!(Relation::Narrower < Relation::Wider);
262 assert!(Relation::Wider < Relation::Intersects);
263 assert!(Relation::Intersects < Relation::Total);
264 }
265}