zenkey_fleet/query.rs
1//! The fan-in query discipline (RFC 05 §2.1) — moved verbatim from
2//! zenctl's `bus.rs`; this stays the single chokepoint for fleet GETs.
3
4use std::time::Duration;
5
6use anyhow::{Context, Result};
7use zenkey::grammar::with_base;
8use zenkey::{RegistrySlice, parse_slice};
9use zenoh::Session;
10use zenoh::query::{ConsolidationMode, QueryTarget};
11
12/// How a producer answered a procedure call.
13pub enum Answer {
14 /// A value reply — RFC 05 §3: "a reply always indicates success".
15 /// Carried as zenoh's refcounted buffer: cloning is a refcount bump,
16 /// and consumers decode via `reader()`/`to_bytes()` (a `Cow` — it
17 /// copies only when the payload arrived fragmented). Report §14's
18 /// zero-copy discipline: the old `to_bytes().to_vec()` double copy per
19 /// reply is retired.
20 Value(zenoh::bytes::ZBytes),
21 /// An error reply (`reply_err`), carrying the `{error, message}` envelope
22 /// when it parses. RFC 05 §3: "an error always indicates failure".
23 Error { name: String, message: String },
24}
25
26/// One host's answer, attributed to the origin that actually replied.
27pub struct FleetAnswer {
28 pub origin: String,
29 pub answer: Answer,
30}
31
32/// Call a procedure and collect **every** reply, attributed by origin.
33///
34/// The three things RFC 05 §2.1 requires, in the one place they cannot be
35/// forgotten:
36///
37/// 1. **target = All.** The default `BestMatching` short-circuits to a single
38/// queryable the moment any matching one is declared `complete` — "one
39/// storage config away from silently collapsing the fleet to one reply".
40/// 2. **consolidation = None.** Default consolidation keeps one reply *per
41/// reply key*; belt-and-braces against a producer that wrongly echoes the
42/// wildcard selector instead of replying on its own concrete key.
43/// 3. **Attribution by the reply's own key**, never by the key we asked on —
44/// that is what makes `*`-origin fan-out legible.
45///
46/// Silence is deliberately *not* interpreted here (RFC 05 §3.1: "no reply" is
47/// not one condition). Callers that need a verdict join this against the
48/// liveliness roster; see `cmd::doctor`.
49pub async fn fleet_get(
50 session: &Session,
51 base: &str,
52 key: &str,
53 payload: Option<Vec<u8>>,
54 timeout: Duration,
55) -> Result<Vec<FleetAnswer>> {
56 let mut builder = session
57 .get(key)
58 .target(QueryTarget::All)
59 .consolidation(ConsolidationMode::None)
60 .timeout(timeout);
61 if let Some(body) = payload {
62 builder = builder.payload(body);
63 }
64 let replies = builder
65 .await
66 .map_err(|e| anyhow::anyhow!("{e}"))
67 .with_context(|| format!("query failed: {key}"))?;
68
69 let mut out = Vec::new();
70 while let Ok(reply) = replies.recv_async().await {
71 match reply.result() {
72 Ok(sample) => {
73 let origin = origin_of(base, sample.key_expr().as_str());
74 out.push(FleetAnswer {
75 origin,
76 answer: Answer::Value(sample.payload().clone()),
77 });
78 }
79 Err(err) => {
80 // The error envelope is `{ "error": "<name>", "message": "…" }`
81 // (RFC 05 §3), with reserved names like `error/not-found`. If it
82 // does not parse we still surface the bytes — an unreadable
83 // refusal is still a refusal.
84 let bytes = err.payload().to_bytes();
85 let (name, message) = match serde_json::from_slice::<serde_json::Value>(&bytes) {
86 Ok(v) => (
87 v.get("error")
88 .and_then(|e| e.as_str())
89 .unwrap_or("error/unparsed")
90 .to_string(),
91 v.get("message")
92 .and_then(|m| m.as_str())
93 .unwrap_or_default()
94 .to_string(),
95 ),
96 Err(_) => (
97 "error/unparsed".to_string(),
98 String::from_utf8_lossy(&bytes).to_string(),
99 ),
100 };
101 // An error reply has no sample, so no concrete key to attribute
102 // by; zenoh does not surface the responder here.
103 out.push(FleetAnswer {
104 origin: "?".to_string(),
105 answer: Answer::Error { name, message },
106 });
107 }
108 }
109 }
110 Ok(out)
111}
112
113/// The origin chunk of a wire key, via the grammar (never by index — RFC 03
114/// §1.1: positions are relative to the configured base).
115fn origin_of(base: &str, key: &str) -> String {
116 zenkey::grammar::parse_full(base, key)
117 .map(|k| k.origin.chunk().to_string())
118 .unwrap_or_else(|| "?".to_string())
119}
120
121/// Discover every live producer's registry slice **from the bus**, with nothing
122/// compiled in (RFC 08 §6: "generic explorer tooling … needs no compiled-in
123/// registry").
124///
125/// Every producer MUST serve its registry slice as TOML on
126/// `@rpc/<producer>/introspect`. This fans one wildcard-producer `introspect`
127/// GET across the fleet — `<base>/v1/*/@rpc/*/introspect` — and parses each
128/// reply. It is the same introspect+`parse_slice` path `doctor` walks, minus
129/// the compiled-in diff: here the served slice *is* the answer.
130///
131/// A reply that does not parse is reported to stderr and skipped, never fatal:
132/// one malformed producer must not blind the tool to every other producer's
133/// slice. The tuple's first element is the producer (or service) base name the
134/// slice declares (`slice.name`), matching the compiled path's producer column.
135///
136/// Note: a verbatim service origin (`@catalog`) is unmatchable by the `*` of a
137/// fleet selector (grammar property D4), so pure-producer discovery does not
138/// enumerate services — `doctor` asks those by name.
139pub async fn fleet_registry(
140 session: &Session,
141 base: &str,
142 timeout: Duration,
143) -> Result<Vec<(String, RegistrySlice)>> {
144 Ok(fleet_registry_raw(session, base, timeout)
145 .await?
146 .into_iter()
147 .map(|(slice, _)| (slice.name.clone(), slice))
148 .collect())
149}
150
151/// As [`fleet_registry`], additionally yielding each reply's raw TOML text
152/// (the artifact the slice cache persists).
153pub async fn fleet_registry_raw(
154 session: &Session,
155 base: &str,
156 timeout: Duration,
157) -> Result<Vec<(RegistrySlice, String)>> {
158 // This session is un-namespaced on purpose (RFC 09 §5), so it must
159 // spell the base itself — exactly as `service call` composes its full key.
160 let key = with_base(base, zenkey::selector::fleet_rpc("*", &["introspect"]));
161 let answers = fleet_get(session, base, &key, None, timeout).await?;
162
163 let mut slices = Vec::new();
164 for answer in answers {
165 let Answer::Value(bytes) = answer.answer else {
166 continue;
167 };
168 let served_toml = String::from_utf8_lossy(&bytes.to_bytes()).to_string();
169 match parse_slice(&served_toml) {
170 Ok(slice) => slices.push((slice, served_toml)),
171 Err(e) => tracing::warn!(
172 origin = %answer.origin,
173 "introspect reply did not parse, skipping: {e}"
174 ),
175 }
176 }
177 Ok(slices)
178}
179
180/// One state sample from a snapshot GET.
181#[derive(Debug, Clone)]
182pub struct StateSample {
183 /// Full wire key.
184 pub key: String,
185 /// HLC timestamp, when the deployment stamps samples (RFC 04 §4
186 /// requires it for LWW to be meaningful — its absence is itself a
187 /// doctor-grade observation).
188 pub timestamp: Option<zenoh::time::Timestamp>,
189 pub payload_len: usize,
190}
191
192/// GET the current state under a selector with the fan-in discipline
193/// (target All, consolidation None) — the doctor's freshness check
194/// (RFC 04 §1.2) consumes the timestamps. Same chokepoint posture as
195/// [`fleet_get`]: no subcommand issues a raw `session.get`.
196pub async fn state_snapshot(
197 session: &Session,
198 selector: &str,
199 timeout: Duration,
200) -> Result<Vec<StateSample>> {
201 let replies = session
202 .get(selector)
203 .target(QueryTarget::All)
204 .consolidation(ConsolidationMode::None)
205 .timeout(timeout)
206 .await
207 .map_err(|e| anyhow::anyhow!("{e}"))
208 .with_context(|| format!("state snapshot failed: {selector}"))?;
209 let mut out = Vec::new();
210 while let Ok(reply) = replies.recv_async().await {
211 let Ok(sample) = reply.result() else { continue };
212 out.push(StateSample {
213 key: sample.key_expr().as_str().to_string(),
214 timestamp: sample.timestamp().copied(),
215 payload_len: sample.payload().len(),
216 });
217 }
218 Ok(out)
219}