spvirit-codec 0.3.0

PVAccess protocol encode/decode and connection state tracking.
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
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
//! Monitor delta decoding: changed bitset, value, overrun bitset.
//!
//! The pvAccess specification puts a MONITOR update's changed bitset first,
//! then the delta data, then the overrun bitset. That order is what
//! [`PvdDecoder::decode_monitor_update`] implements, and it is what every live
//! connection in this workspace uses.
//!
//! Some implementations disagree about where the overrun bitset goes (or omit
//! it entirely). [`PvdDecoder::decode_monitor_update_lenient`] tries every
//! known layout and reports which one it matched; it exists for mid-stream
//! packet captures, where the introspection was never seen and the peer's
//! layout has to be inferred.

use crate::error::{DecodeError, DecodeResult};
use crate::spvd_decode::{DecodedValue, FieldType, PvdDecoder, StructureDesc};

/// A decoded MONITOR update: the delta value plus both bitsets.
#[derive(Debug, Clone)]
pub struct MonitorUpdate {
    pub value: DecodedValue,
    /// Raw changed bitset. Bit 0 is the whole structure; field bits start at 1.
    pub changed: Vec<u8>,
    /// Raw overrun bitset, same numbering. A set bit means the server dropped
    /// at least one update for that field before this one.
    pub overrun: Vec<u8>,
    /// Bytes consumed from the body.
    pub consumed: usize,
    /// Bit-indexed field paths for this update's introspection: index 0 is
    /// `"<whole structure>"`, index *n* is the field addressed by bit *n*.
    ///
    /// Captured at decode time so a callback can name the set bits without
    /// also being handed the [`StructureDesc`].
    pub paths: Vec<String>,
}

/// Which wire layout a lenient decode matched.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum MonitorLayout {
    /// changed bitset, data, overrun bitset. The pvAccess specification order.
    SpecOrder,
    /// changed bitset, overrun bitset, data.
    OverrunBeforeData,
    /// changed bitset, data, with no overrun bitset.
    ChangedOnly,
}

impl MonitorUpdate {
    /// True if any overrun bit is set.
    pub fn has_overrun(&self) -> bool {
        self.overrun.iter().any(|b| *b != 0)
    }

    /// Dotted paths of the fields whose overrun bits are set.
    ///
    /// Bit 0 yields the literal `"<whole structure>"`. Takes the descriptor
    /// explicitly; [`MonitorUpdate::overrun_paths`] uses the paths the decoder
    /// already captured.
    pub fn overrun_fields(&self, desc: &StructureDesc) -> Vec<String> {
        select_paths(&bit_paths(desc), &self.overrun)
    }

    /// Dotted paths of the fields marked changed in this update.
    ///
    /// Bit 0 yields the literal `"<whole structure>"`.
    pub fn changed_paths(&self) -> Vec<String> {
        select_paths(&self.paths, &self.changed)
    }

    /// Dotted paths of the fields whose overrun bits are set — the server
    /// dropped at least one earlier update for each of them.
    pub fn overrun_paths(&self) -> Vec<String> {
        select_paths(&self.paths, &self.overrun)
    }
}

/// Bit-indexed paths for a descriptor: bit 0 is the whole structure, field
/// bits follow in `flatten_field_paths` order.
pub(crate) fn bit_paths(desc: &StructureDesc) -> Vec<String> {
    let mut paths = vec!["<whole structure>".to_string()];
    flatten_field_paths(desc, "", &mut paths);
    paths
}

/// The subset of `paths` whose bit is set in `bits`.
fn select_paths(paths: &[String], bits: &[u8]) -> Vec<String> {
    paths
        .iter()
        .enumerate()
        .filter(|(bit, _)| {
            let byte = bit / 8;
            byte < bits.len() && (bits[byte] & (1 << (bit % 8))) != 0
        })
        .map(|(_, path)| path.clone())
        .collect()
}

/// Depth-first, self-then-nested. Must stay in step with
/// `count_structure_fields` in `spvd_decode.rs`, which numbers the bits.
fn flatten_field_paths(desc: &StructureDesc, prefix: &str, out: &mut Vec<String>) {
    for field in &desc.fields {
        let path = if prefix.is_empty() {
            field.name.clone()
        } else {
            format!("{prefix}.{}", field.name)
        };
        out.push(path.clone());
        if let FieldType::Structure(nested) = &field.field_type {
            flatten_field_paths(nested, &path, out);
        }
    }
}

impl PvdDecoder {
    /// Decode a MONITOR update in specification order: changed bitset, then
    /// the delta data, then the overrun bitset.
    pub fn decode_monitor_update(
        &self,
        data: &[u8],
        desc: &StructureDesc,
    ) -> DecodeResult<MonitorUpdate> {
        let (changed, mut offset) = self.read_bitset(data, 0)?;
        let (value, consumed) =
            self.decode_structure_with_bitset_body(&data[offset..], desc, &changed)?;
        offset += consumed;
        let (overrun, next) = self.read_bitset(data, offset)?;
        Ok(MonitorUpdate {
            value,
            changed,
            overrun,
            consumed: next,
            paths: bit_paths(desc),
        })
    }

    /// Try all three known layouts and report which one won.
    ///
    /// For mid-stream packet captures, where the introspection was missed and
    /// the peer's layout is unknown. Live connections should use
    /// [`PvdDecoder::decode_monitor_update`].
    pub fn decode_monitor_update_lenient(
        &self,
        data: &[u8],
        desc: &StructureDesc,
    ) -> DecodeResult<(MonitorUpdate, MonitorLayout)> {
        let candidates = [
            (
                MonitorLayout::SpecOrder,
                self.decode_monitor_update(data, desc),
            ),
            (
                MonitorLayout::OverrunBeforeData,
                self.decode_overrun_before_data(data, desc),
            ),
            (
                MonitorLayout::ChangedOnly,
                self.decode_changed_only(data, desc),
            ),
        ];

        let mut best: Option<(MonitorUpdate, MonitorLayout, i32)> = None;
        let mut last_err = DecodeError::Malformed("no monitor layout matched");
        for (layout, result) in candidates {
            match result {
                Ok(update) => {
                    let score = score_decoded(&update.value);
                    let better = match &best {
                        None => true,
                        Some((prev, _, prev_score)) => {
                            score > *prev_score
                                || (score == *prev_score && update.consumed > prev.consumed)
                        }
                    };
                    if better {
                        best = Some((update, layout, score));
                    }
                }
                Err(e) => last_err = e,
            }
        }

        best.map(|(u, l, _)| (u, l)).ok_or(last_err)
    }

    /// changed bitset, overrun bitset, data.
    fn decode_overrun_before_data(
        &self,
        data: &[u8],
        desc: &StructureDesc,
    ) -> DecodeResult<MonitorUpdate> {
        let (changed, offset) = self.read_bitset(data, 0)?;
        let (overrun, mut offset) = self.read_bitset(data, offset)?;
        let (value, consumed) =
            self.decode_structure_with_bitset_body(&data[offset..], desc, &changed)?;
        offset += consumed;
        Ok(MonitorUpdate {
            value,
            changed,
            overrun,
            consumed: offset,
            paths: bit_paths(desc),
        })
    }

    /// changed bitset, data, nothing else.
    fn decode_changed_only(
        &self,
        data: &[u8],
        desc: &StructureDesc,
    ) -> DecodeResult<MonitorUpdate> {
        let (changed, mut offset) = self.read_bitset(data, 0)?;
        let (value, consumed) =
            self.decode_structure_with_bitset_body(&data[offset..], desc, &changed)?;
        offset += consumed;
        Ok(MonitorUpdate {
            value,
            changed,
            overrun: Vec::new(),
            consumed: offset,
            paths: bit_paths(desc),
        })
    }

    /// Read a size-prefixed bitset starting at `offset`; returns it and the
    /// offset just past it.
    fn read_bitset(&self, data: &[u8], offset: usize) -> DecodeResult<(Vec<u8>, usize)> {
        if offset > data.len() {
            return Err(DecodeError::Truncated {
                needed: offset,
                available: data.len(),
            });
        }
        let (size, consumed) = self.decode_size(&data[offset..])?;
        let start = offset + consumed;
        let end = start + size;
        if end > data.len() {
            return Err(DecodeError::Truncated {
                needed: end,
                available: data.len(),
            });
        }
        Ok((data[start..end].to_vec(), end))
    }
}

/// Plausibility score for a candidate decode. Higher is better.
///
/// Moved verbatim from `epics_decode.rs`, where it drove the old
/// three-way "try everything" monitor decode. Only
/// [`PvdDecoder::decode_monitor_update_lenient`] uses it now.
fn score_decoded(value: &DecodedValue) -> i32 {
    let DecodedValue::Structure(fields) = value else {
        return -1;
    };

    let mut score = fields.len() as i32;

    let mut has_value = false;
    let mut has_alarm = false;
    let mut has_ts = false;

    for (name, val) in fields {
        match name.as_str() {
            "value" => {
                has_value = true;
                score += 4;
                match val {
                    DecodedValue::Array(items) => {
                        if items.is_empty() {
                            score -= 2;
                        } else {
                            score += 6 + (items.len().min(8) as i32);
                        }
                    }
                    DecodedValue::Structure(_) => score += 1,
                    _ => score += 2,
                }
            }
            "alarm" => {
                has_alarm = true;
                score += 2;
            }
            "timeStamp" => {
                has_ts = true;
                score += 2;
                if let DecodedValue::Structure(ts_fields) = val {
                    if let Some(secs) = ts_fields.iter().find_map(|(n, v)| {
                        if n == "secondsPastEpoch" {
                            if let DecodedValue::Int64(s) = v {
                                return Some(*s);
                            }
                        }
                        None
                    }) {
                        if (0..=4_000_000_000i64).contains(&secs) {
                            score += 2;
                        } else if secs.abs() > 10_000_000_000i64 {
                            score -= 2;
                        }
                    }
                }
            }
            "display" | "control" => {
                score += 1;
            }
            _ => {}
        }
    }

    if !has_value {
        score -= 2;
    }
    if !has_alarm {
        score -= 1;
    }
    if !has_ts {
        score -= 1;
    }

    score
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::spvd_decode::{FieldDesc, FieldType, PvdDecoder, StructureDesc, TypeCode};

    fn nt_scalar_desc() -> StructureDesc {
        let mut alarm = StructureDesc::new();
        alarm.fields.push(FieldDesc {
            name: "severity".to_string(),
            field_type: FieldType::Scalar(TypeCode::Int32),
        });

        let mut desc = StructureDesc::new();
        desc.fields.push(FieldDesc {
            name: "value".to_string(),
            field_type: FieldType::Scalar(TypeCode::Int32),
        });
        desc.fields.push(FieldDesc {
            name: "alarm".to_string(),
            field_type: FieldType::Structure(alarm),
        });
        desc
    }

    /// changed bitset (1 byte), value data, overrun bitset (1 byte).
    fn spec_order_body(changed: u8, value: i32, overrun: u8) -> Vec<u8> {
        let mut b = vec![1, changed];
        b.extend_from_slice(&value.to_le_bytes());
        b.extend_from_slice(&[1, overrun]);
        b
    }

    #[test]
    fn decodes_spec_order_and_reports_consumed() {
        let decoder = PvdDecoder::new(false);
        let desc = nt_scalar_desc();
        // bit 1 = "value".
        let body = spec_order_body(0b0000_0010, 42, 0);
        let update = decoder.decode_monitor_update(&body, &desc).unwrap();

        assert_eq!(update.changed, vec![0b0000_0010]);
        assert_eq!(update.overrun, vec![0]);
        assert!(!update.has_overrun());
        assert_eq!(update.consumed, body.len());
        let DecodedValue::Structure(fields) = &update.value else {
            panic!("expected a structure");
        };
        assert_eq!(fields.len(), 1);
        assert_eq!(fields[0].0, "value");
    }

    #[test]
    fn overrun_bits_resolve_to_field_paths() {
        let decoder = PvdDecoder::new(false);
        let desc = nt_scalar_desc();
        // Overrun on bit 1 ("value") and bit 3 ("alarm.severity").
        let body = spec_order_body(0b0000_0010, 42, 0b0000_1010);
        let update = decoder.decode_monitor_update(&body, &desc).unwrap();

        assert!(update.has_overrun());
        assert_eq!(
            update.overrun_fields(&desc),
            vec!["value", "alarm.severity"]
        );
    }

    #[test]
    fn bit_zero_overrun_reports_the_whole_structure() {
        let decoder = PvdDecoder::new(false);
        let desc = nt_scalar_desc();
        let body = spec_order_body(0b0000_0010, 42, 0b0000_0001);
        let update = decoder.decode_monitor_update(&body, &desc).unwrap();
        assert_eq!(update.overrun_fields(&desc), vec!["<whole structure>"]);
    }

    /// The bit numbering `overrun_fields` walks must be the same numbering
    /// `decode_structure_with_bitset_body` uses, which is
    /// `count_structure_fields`' self-then-nested depth-first order. If the
    /// two ever diverge, overrun bits map to the wrong field names.
    #[test]
    fn flatten_field_paths_agrees_with_count_structure_fields() {
        let mut leaf = StructureDesc::new();
        leaf.fields.push(FieldDesc {
            name: "deep".to_string(),
            field_type: FieldType::Scalar(TypeCode::Int32),
        });

        let mut mid = StructureDesc::new();
        mid.fields.push(FieldDesc {
            name: "a".to_string(),
            field_type: FieldType::Scalar(TypeCode::Int32),
        });
        mid.fields.push(FieldDesc {
            name: "leaf".to_string(),
            field_type: FieldType::Structure(leaf),
        });
        mid.fields.push(FieldDesc {
            name: "b".to_string(),
            field_type: FieldType::Scalar(TypeCode::Int32),
        });

        let mut root = StructureDesc::new();
        root.fields.push(FieldDesc {
            name: "value".to_string(),
            field_type: FieldType::Scalar(TypeCode::Int32),
        });
        root.fields.push(FieldDesc {
            name: "mid".to_string(),
            field_type: FieldType::Structure(mid),
        });
        root.fields.push(FieldDesc {
            name: "tail".to_string(),
            field_type: FieldType::Scalar(TypeCode::Int32),
        });

        let mut paths = Vec::new();
        flatten_field_paths(&root, "", &mut paths);

        assert_eq!(
            paths,
            vec![
                "value",
                "mid",
                "mid.a",
                "mid.leaf",
                "mid.leaf.deep",
                "mid.b",
                "tail",
            ]
        );
        assert_eq!(
            paths.len(),
            crate::spvd_decode::count_structure_fields(&root),
            "one path per bit, in the same order the bits are numbered"
        );

        // And the paths line up with the bits when read through an update:
        // bit 0 is the whole structure, so field bits start at 1.
        let mut overrun = vec![0u8; 2];
        let bit = 1 + 4; // "mid.leaf.deep"
        overrun[bit / 8] |= 1 << (bit % 8);
        let update = MonitorUpdate {
            value: DecodedValue::Structure(Vec::new()),
            changed: Vec::new(),
            overrun,
            consumed: 0,
            paths: bit_paths(&root),
        };
        assert_eq!(update.overrun_fields(&root), vec!["mid.leaf.deep"]);
        assert_eq!(update.overrun_paths(), vec!["mid.leaf.deep"]);
    }

    /// The shape the client monitor callback relies on: a decoded update can
    /// name its own changed and overrun bits without the caller holding the
    /// descriptor.
    #[test]
    fn monitor_update_reports_overrun_paths() {
        let decoder = PvdDecoder::new(false);
        let desc = nt_scalar_desc();
        // changed = bit 1 ("value"), overrun = bit 3 ("alarm.severity").
        let body = spec_order_body(0b0000_0010, 42, 0b0000_1000);
        let update = decoder.decode_monitor_update(&body, &desc).unwrap();

        assert_eq!(update.changed_paths(), vec!["value"]);
        assert_eq!(update.overrun_paths(), vec!["alarm.severity"]);
        assert!(update.has_overrun());
    }

    #[test]
    fn missing_overrun_bitset_is_truncated_not_silently_accepted() {
        let decoder = PvdDecoder::new(false);
        let desc = nt_scalar_desc();
        // changed bitset and value, then nothing.
        let mut body = vec![1, 0b0000_0010];
        body.extend_from_slice(&42i32.to_le_bytes());
        assert!(matches!(
            decoder.decode_monitor_update(&body, &desc).unwrap_err(),
            DecodeError::Truncated { .. }
        ));
    }

    #[test]
    fn lenient_identifies_the_spec_layout() {
        let decoder = PvdDecoder::new(false);
        let desc = nt_scalar_desc();
        let body = spec_order_body(0b0000_0010, 42, 0);
        let (_, layout) = decoder.decode_monitor_update_lenient(&body, &desc).unwrap();
        assert_eq!(layout, MonitorLayout::SpecOrder);
    }

    #[test]
    fn lenient_recovers_the_overrun_before_data_layout() {
        let decoder = PvdDecoder::new(false);
        let desc = nt_scalar_desc();
        // changed bitset, overrun bitset, then the value.
        let mut body = vec![1, 0b0000_0010, 1, 0];
        body.extend_from_slice(&42i32.to_le_bytes());

        // Strict must not accept it: it would read the value from the
        // overrun bitset's bytes.
        let strict = decoder.decode_monitor_update(&body, &desc);
        let strict_ok = strict.map(|u| u.value).ok();
        assert_ne!(
            strict_ok.as_ref().and_then(scalar_value_of),
            Some(42),
            "strict must not accidentally decode the non-spec layout"
        );

        let (update, layout) = decoder.decode_monitor_update_lenient(&body, &desc).unwrap();
        assert_eq!(layout, MonitorLayout::OverrunBeforeData);
        assert_eq!(scalar_value_of(&update.value), Some(42));
    }

    fn scalar_value_of(v: &DecodedValue) -> Option<i32> {
        let DecodedValue::Structure(fields) = v else {
            return None;
        };
        fields
            .iter()
            .find_map(|(name, val)| match (name.as_str(), val) {
                ("value", DecodedValue::Int32(n)) => Some(*n),
                _ => None,
            })
    }

    #[test]
    fn lenient_falls_back_to_changed_only() {
        let decoder = PvdDecoder::new(false);
        let desc = nt_scalar_desc();
        // changed bitset and value, no overrun bitset at all.
        let mut body = vec![1, 0b0000_0010];
        body.extend_from_slice(&42i32.to_le_bytes());
        let (update, layout) = decoder.decode_monitor_update_lenient(&body, &desc).unwrap();
        assert_eq!(layout, MonitorLayout::ChangedOnly);
        assert!(update.overrun.is_empty());
    }

    #[test]
    fn round_trips_an_encoded_delta() {
        use crate::spvd_encode::{compute_changed_bits, encode_nt_payload_delta, nt_payload_desc};
        use spvirit_types::{NtPayload, NtScalar, ScalarValue};

        let prev = NtPayload::Scalar(NtScalar::from_value(ScalarValue::F64(1.0)));
        let next = NtPayload::Scalar(NtScalar::from_value(ScalarValue::F64(3.5)));
        let desc = nt_payload_desc(&next);

        let (bitset, values) =
            encode_nt_payload_delta(&prev, &next, &desc, false).expect("value changed");

        // Spec order: changed bitset, data, then an empty overrun bitset.
        let mut body = bitset.clone();
        body.extend_from_slice(&values);
        body.extend_from_slice(&[0u8]); // zero-length overrun bitset

        let decoder = PvdDecoder::new(false);
        let update = decoder.decode_monitor_update(&body, &desc).unwrap();

        assert_eq!(update.consumed, body.len());
        assert!(!update.has_overrun());

        // The changed bitset the decoder reports is exactly what
        // compute_changed_bits produced for this pair.
        let bits =
            compute_changed_bits(&projection(&prev, &desc), &projection(&next, &desc), &desc)
                .expect("value changed");
        let mut expected = vec![0u8; bits.len().div_ceil(8)];
        for (i, b) in bits.iter().enumerate() {
            if *b {
                expected[i / 8] |= 1 << (i % 8);
            }
        }
        assert_eq!(update.changed, expected);
        assert!(bits[1], "bit 1 is 'value', which is what changed");

        // And the value round-trips.
        let DecodedValue::Structure(fields) = &update.value else {
            panic!("expected a structure");
        };
        let value = fields
            .iter()
            .find(|(n, _)| n == "value")
            .map(|(_, v)| v)
            .expect("value field present");
        match value {
            DecodedValue::Float64(v) => assert!((v - 3.5).abs() < 1e-9),
            other => panic!("unexpected value {other:?}"),
        }
    }

    /// `compute_changed_bits` compares two `DecodedValue`s; `encode_nt_payload_delta`
    /// projects each payload onto the descriptor first. Reproduce that
    /// projection by encoding and decoding against the same descriptor.
    fn projection(payload: &spvirit_types::NtPayload, desc: &StructureDesc) -> DecodedValue {
        let bytes = crate::spvd_encode::encode_nt_payload_values_for_desc(payload, desc, false);
        PvdDecoder::new(false)
            .decode_structure(&bytes, desc)
            .expect("projection")
            .0
    }
}