zenkey-fleet 0.11.1

Fleet engine for keyspace-v2 Zenoh tooling: disciplined fan-in queries, liveliness roster, registry-slice sets, schema-aware decode, live key-tree monitoring — the shared core of zenctl and zengui
Documentation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
//! Schema-driven payload synthesis (#162) — the datagen half of `zenctl gen`.
//!
//! Synthesis produces a **JSON value** for every schema kind; the kind's own
//! encoder (`DecoderRegistry::encode`, the same seam `zenctl pub` writes
//! through) turns it into wire bytes. That keeps this module codec-free: it
//! never frames bytes, it only answers "what instance would this schema
//! accept?".
//!
//! Deterministic on purpose: a `(seed, tick)` pair always yields the same
//! instance (spray's seeded-sine precedent) — a generator whose runs cannot
//! be reproduced cannot be used to bisect a consumer bug. Numeric leaves
//! wander on a sine per field so plots move; everything else is stable.

use serde_json::{Map, Value, json};
use zenkey::schema::TypeSchema;

use crate::model::jsonschema::resolve_ref;

/// How deep nested objects/arrays are followed before giving up — a cyclic
/// or pathological schema degrades to a placeholder, not a stack overflow.
///
/// Raised from 6 with `$ref` following (#384): a resolved reference costs a
/// level, and `schemars` hoists every nested named type into `$defs`, so the
/// old cap was spent on indirection rather than on nesting. A cycle still
/// terminates here — the cap is what stops it, since a `$ref` chain has no
/// other bottom.
const DEPTH_CAP: usize = 16;

/// A deterministic instance generator.
#[derive(Debug, Clone, Copy)]
pub struct Synth {
    pub seed: u64,
}

/// A cheap deterministic hash for per-field phase offsets (FNV-1a) — not
/// cryptographic, just stable across runs and platforms.
fn fnv(s: &str) -> u64 {
    let mut h: u64 = 0xcbf29ce484222325;
    for b in s.bytes() {
        h ^= u64::from(b);
        h = h.wrapping_mul(0x100000001b3);
    }
    h
}

impl Synth {
    pub fn new(seed: u64) -> Synth {
        Synth { seed }
    }

    /// A wandering numeric value: a sine over `tick`, phase-offset by the
    /// field's name so sibling fields do not move in lockstep.
    fn wander(&self, field: &str, tick: u64, min: f64, max: f64) -> f64 {
        let phase = (fnv(field) ^ self.seed) % 628 /* 2π·100 */;
        let x = (tick as f64) / 10.0 + (phase as f64) / 100.0;
        let mid = f64::midpoint(min, max);
        let amp = (max - min) / 2.0;
        mid + amp * x.sin()
    }

    /// Synthesize an instance for a schema entry. `None` means this kind
    /// cannot be synthesized here (unknown kind) — the caller degrades with
    /// a stated note, never silently.
    pub fn instance(&self, schema: &TypeSchema, tick: u64) -> Option<Value> {
        match schema.kind_str() {
            zenkey::schema::SchemaKind::JSON_SCHEMA => schema
                .json_document()
                .map(|doc| self.json_schema_value(doc, doc, "", tick, 0)),
            zenkey::schema::SchemaKind::CDR => {
                let fields = schema.cdr_fields()?;
                let types = schema.cdr_types();
                Some(self.cdr_fields_value(fields, types, tick, 0))
            }
            #[cfg(feature = "decode-protobuf")]
            zenkey::schema::SchemaKind::PROTOBUF => self.protobuf_value(schema, tick),
            _ => None,
        }
    }

    /// Walk a draft 2020-12 document conservatively: satisfy `type`,
    /// `required` (by emitting every declared property), `enum`/`const`,
    /// combinators, `$ref`, and numeric bounds. Unknown or empty schemas get
    /// a wandering number — `{}` accepts anything.
    ///
    /// `root` is the whole document, carried so `$ref` can be resolved
    /// against it; `doc` is the subschema being satisfied.
    fn json_schema_value(
        &self,
        root: &Value,
        doc: &Value,
        field: &str,
        tick: u64,
        depth: usize,
    ) -> Value {
        if depth > DEPTH_CAP {
            return Value::Null;
        }
        if let Some(c) = doc.get("const") {
            return c.clone();
        }
        if let Some(e) = doc.get("enum").and_then(Value::as_array)
            && let Some(first) = e.first()
        {
            return first.clone();
        }
        // `$ref` into `$defs` is where `schemars` puts every nested named
        // type, so a walk that does not follow it synthesizes a wandering
        // number where a struct belongs (#384). Unresolvable falls through
        // to the conservative default below, as an unknown schema does.
        if let Some(pointer) = doc.get("$ref").and_then(Value::as_str)
            && let Some(target) = resolve_ref(root, pointer)
        {
            return self.json_schema_value(root, target, field, tick, depth + 1);
        }
        // `allOf` composes one shape out of several, so an instance must
        // satisfy every member: merge them.
        if let Some(members) = doc.get("allOf").and_then(Value::as_array) {
            let mut merged = Map::new();
            for member in members {
                if let Value::Object(o) =
                    self.json_schema_value(root, member, field, tick, depth + 1)
                {
                    merged.extend(o);
                }
            }
            return Value::Object(merged);
        }
        // `oneOf`/`anyOf` alternate, and one instance satisfies one branch —
        // so the first is as good a pick as any. Deliberately unlike the
        // declared-path walk in `judge::field`, which must union every
        // branch: a *surface* is all the shapes allowed, an *instance* is one.
        for branch in ["oneOf", "anyOf"] {
            if let Some(b) = doc.get(branch).and_then(Value::as_array)
                && let Some(first) = b.first()
            {
                return self.json_schema_value(root, first, field, tick, depth + 1);
            }
        }
        let ty = doc.get("type").and_then(Value::as_str).unwrap_or("number");
        match ty {
            "object" => {
                let mut out = Map::new();
                if let Some(props) = doc.get("properties").and_then(Value::as_object) {
                    for (name, sub) in props {
                        out.insert(
                            name.clone(),
                            self.json_schema_value(root, sub, name, tick, depth + 1),
                        );
                    }
                }
                Value::Object(out)
            }
            "array" => {
                let n = doc
                    .get("minItems")
                    .and_then(Value::as_u64)
                    .unwrap_or(1)
                    .max(1);
                let item = doc.get("items").cloned().unwrap_or(json!({}));
                Value::Array(
                    (0..n)
                        .map(|i| self.json_schema_value(root, &item, field, tick + i, depth + 1))
                        .collect(),
                )
            }
            "string" => Value::String(format!(
                "{}-{}",
                if field.is_empty() { "s" } else { field },
                tick % 10
            )),
            "boolean" => Value::Bool(tick.is_multiple_of(2)),
            "integer" => {
                let (min, max) = bounds(doc, 0.0, 100.0);
                json!(self.wander(field, tick, min, max).round() as i64)
            }
            "null" => Value::Null,
            // "number" and anything else numeric-shaped.
            _ => {
                let (min, max) = bounds(doc, 0.0, 100.0);
                json!(self.wander(field, tick, min, max))
            }
        }
    }

    /// The `cdr` kind's compact field list (RFC 08 §7.1): positional
    /// `[{name, type}]` with a local `types` table for composites.
    fn cdr_fields_value(
        &self,
        fields: &Value,
        types: Option<&Map<String, Value>>,
        tick: u64,
        depth: usize,
    ) -> Value {
        if depth > DEPTH_CAP {
            return Value::Null;
        }
        let Some(list) = fields.as_array() else {
            return Value::Null;
        };
        let mut out = Map::new();
        for f in list {
            let Some(name) = f.get("name").and_then(Value::as_str) else {
                continue;
            };
            let ty = f.get("type").cloned().unwrap_or(Value::Null);
            out.insert(
                name.to_string(),
                self.cdr_value(&ty, types, name, tick, depth),
            );
        }
        Value::Object(out)
    }

    fn cdr_value(
        &self,
        ty: &Value,
        types: Option<&Map<String, Value>>,
        field: &str,
        tick: u64,
        depth: usize,
    ) -> Value {
        if depth > DEPTH_CAP {
            return Value::Null;
        }
        match ty {
            Value::String(name) => match name.as_str() {
                "bool" => Value::Bool(tick.is_multiple_of(2)),
                "string" => Value::String(format!("{field}-{}", tick % 10)),
                "float32" | "float" | "float64" | "double" => {
                    json!(self.wander(field, tick, 0.0, 100.0))
                }
                // The full primitive vocabulary of the cdr kind (RFC 08
                // §7.1's IDL-flavoured aliases included).
                "int8" | "char" | "int16" | "short" | "int32" | "long" | "int64" | "long long" => {
                    json!(self.wander(field, tick, 0.0, 100.0).round() as i64)
                }
                "uint8" | "byte" | "octet" | "uint16" | "unsigned short" | "uint32"
                | "unsigned long" | "uint64" | "unsigned long long" => {
                    json!(self.wander(field, tick, 0.0, 100.0).round().abs() as u64)
                }
                // A named composite from the local table.
                other => match types.and_then(|t| t.get(other)) {
                    Some(composite) => {
                        let fields = composite.get("fields").unwrap_or(composite);
                        self.cdr_fields_value(fields, types, tick, depth + 1)
                    }
                    None => Value::Null,
                },
            },
            // {"array": {"of": T, "len": n}} / {"sequence": {"of": T}}
            Value::Object(o) => {
                if let Some(arr) = o.get("array") {
                    let n = arr.get("len").and_then(Value::as_u64).unwrap_or(1).max(1);
                    let of = arr.get("of").cloned().unwrap_or(Value::Null);
                    Value::Array(
                        (0..n)
                            .map(|i| self.cdr_value(&of, types, field, tick + i, depth + 1))
                            .collect(),
                    )
                } else if let Some(seq) = o.get("sequence") {
                    let of = seq.get("of").cloned().unwrap_or(Value::Null);
                    Value::Array(vec![self.cdr_value(&of, types, field, tick, depth + 1)])
                } else {
                    Value::Null
                }
            }
            _ => Value::Null,
        }
    }

    /// Protobuf: field names and kinds off the served descriptor; the value
    /// is JSON in prost-reflect's serde dialect, which `store.encode`
    /// deserializes into a `DynamicMessage`.
    #[cfg(feature = "decode-protobuf")]
    fn protobuf_value(&self, schema: &TypeSchema, tick: u64) -> Option<Value> {
        use prost_reflect::{DescriptorPool, Kind};
        let fds = schema.protobuf_descriptor_set()?;
        let message = schema.protobuf_message()?;
        let pool = DescriptorPool::decode(fds.as_slice()).ok()?;
        let desc = pool.get_message_by_name(message)?;
        fn message_value(
            synth: &Synth,
            desc: &prost_reflect::MessageDescriptor,
            tick: u64,
            depth: usize,
        ) -> Value {
            if depth > DEPTH_CAP {
                return Value::Object(Map::new());
            }
            let mut out = Map::new();
            for field in desc.fields() {
                let name = field.json_name().to_string();
                let v = match field.kind() {
                    Kind::Double | Kind::Float => json!(synth.wander(&name, tick, 0.0, 100.0)),
                    Kind::Int32
                    | Kind::Int64
                    | Kind::Sint32
                    | Kind::Sint64
                    | Kind::Sfixed32
                    | Kind::Sfixed64 => {
                        json!(synth.wander(&name, tick, 0.0, 100.0).round() as i64)
                    }
                    Kind::Uint32 | Kind::Uint64 | Kind::Fixed32 | Kind::Fixed64 => {
                        json!(synth.wander(&name, tick, 0.0, 100.0).round().abs() as u64)
                    }
                    Kind::Bool => Value::Bool(tick.is_multiple_of(2)),
                    Kind::String => Value::String(format!("{name}-{}", tick % 10)),
                    Kind::Bytes => Value::String(String::new()),
                    Kind::Enum(e) => e
                        .values()
                        .next()
                        .map(|v| Value::String(v.name().to_string()))
                        .unwrap_or(Value::Null),
                    Kind::Message(m) => message_value(synth, &m, tick, depth + 1),
                };
                let v = if field.is_list() {
                    Value::Array(vec![v])
                } else {
                    v
                };
                out.insert(name, v);
            }
            Value::Object(out)
        }
        Some(message_value(self, &desc, tick, 0))
    }
}

fn bounds(doc: &Value, dmin: f64, dmax: f64) -> (f64, f64) {
    let min = doc.get("minimum").and_then(Value::as_f64).unwrap_or(dmin);
    let max = doc
        .get("maximum")
        .and_then(Value::as_f64)
        .unwrap_or_else(|| dmax.max(min + 1.0));
    (min, max.max(min))
}

#[cfg(test)]
mod tests {
    use super::*;
    use zenkey::schema::WireEncoding;
    use zenkey::schema::decode::DecoderRegistry;

    /// The whole point: a synthesized instance survives the kind's own
    /// encoder — and for json-schema (with validate-json on in tests) that
    /// encoder *validates*, so this is a real conformance round trip.
    #[test]
    fn a_synthesized_json_instance_encodes_and_validates() {
        let schema = TypeSchema::json_schema(json!({
            "type": "object",
            "required": ["status", "load", "cores"],
            "properties": {
                "status": { "type": "string", "enum": ["ok", "degraded"] },
                "load": { "type": "number", "minimum": 0.0, "maximum": 1.0 },
                "cores": { "type": "integer", "minimum": 1, "maximum": 128 },
                "tags": { "type": "array", "items": { "type": "string" } },
                "nested": {
                    "type": "object",
                    "properties": { "up": { "type": "boolean" } },
                },
            },
        }));
        let registry = DecoderRegistry::new();
        let synth = Synth::new(42);
        for tick in 0..20 {
            let v = synth
                .instance(&schema, tick)
                .expect("json-schema synthesizes");
            let bytes = registry
                .encode(&schema, &v, &WireEncoding::Json)
                .unwrap_or_else(|e| panic!("tick {tick}: {v} refused: {e}"));
            let back = registry
                .decode(&schema, &WireEncoding::Json, &bytes)
                .unwrap();
            assert_eq!(
                back.verdict,
                zenkey::schema::validate::Verdict::Valid,
                "tick {tick}"
            );
        }
    }

    /// A tagged enum behind a `$ref` — the shape `schemars` emits for every
    /// nested named type, and the one #384 was about. Without following the
    /// reference the synthesizer produced a wandering *number* for `value`,
    /// which the validator on this same path then rejected: the round trip
    /// below is what fails if the resolver goes away.
    #[test]
    fn a_ref_into_defs_synthesizes_the_referenced_shape() {
        let schema = TypeSchema::json_schema(json!({
            "type": "object",
            "required": ["name", "value"],
            "properties": {
                "name": { "type": "string" },
                "value": { "$ref": "#/$defs/TelemetryValue" },
            },
            "$defs": {
                "TelemetryValue": {
                    "oneOf": [
                        { "type": "object",
                          "required": ["type", "value"],
                          "properties": {
                              "type": { "const": "counter", "type": "string" },
                              "value": { "type": "integer", "minimum": 0 } } },
                        { "type": "object",
                          "required": ["type", "value"],
                          "properties": {
                              "type": { "const": "gauge", "type": "string" },
                              "value": { "type": "number" } } },
                    ],
                },
            },
        }));
        let registry = DecoderRegistry::new();
        let synth = Synth::new(7);
        let v = synth.instance(&schema, 3).expect("json-schema synthesizes");
        assert!(
            v["value"].is_object(),
            "the reference resolved to the enum, not to a placeholder number: {v}"
        );
        assert_eq!(v["value"]["type"], "counter", "the first branch, satisfied");

        let bytes = registry
            .encode(&schema, &v, &WireEncoding::Json)
            .unwrap_or_else(|e| panic!("{v} refused: {e}"));
        let back = registry
            .decode(&schema, &WireEncoding::Json, &bytes)
            .unwrap();
        assert_eq!(back.verdict, zenkey::schema::validate::Verdict::Valid);
    }

    /// Same (seed, tick) → same instance; different tick → the numerics move.
    #[test]
    fn synthesis_is_deterministic_and_wanders() {
        let schema = TypeSchema::json_schema(json!({
            "type": "object",
            "properties": { "v": { "type": "number" } },
        }));
        let synth = Synth::new(7);
        assert_eq!(
            synth.instance(&schema, 3),
            synth.instance(&schema, 3),
            "reproducible runs are the point"
        );
        assert_ne!(synth.instance(&schema, 3), synth.instance(&schema, 4));
    }

    /// The cdr field list synthesizes an object its encoder accepts.
    #[cfg(feature = "decode-cdr")]
    #[test]
    fn a_synthesized_cdr_instance_encodes() {
        let schema = TypeSchema::cdr(json!({
            "fields": [
                { "name": "x", "type": "float64" },
                { "name": "n", "type": "uint32" },
                { "name": "label", "type": "string" },
            ],
        }));
        let registry = DecoderRegistry::new();
        let v = Synth::new(1).instance(&schema, 0).expect("cdr synthesizes");
        let bytes = registry
            .encode(&schema, &v, &WireEncoding::Cdr)
            .expect("the instance encodes");
        assert!(!bytes.is_empty());
    }

    /// An unknown kind is `None` — the caller states the degradation.
    #[test]
    fn an_unknown_kind_declines_instead_of_guessing() {
        let set = zenkey::schema::SchemaSet::parse(
            r#"{"schema_version":1,"app":"t",
                "types":{"W":{"kind":"cddl","hash":"sha256:00","spec":"x = int"}}}"#,
        )
        .unwrap();
        assert_eq!(Synth::new(0).instance(set.get("W").unwrap(), 0), None);
    }
}