spatialrust_viewer/
replay.rs

1//! Portable state for the one-command deterministic replay demo.
2//!
3//! The state records the replay trace and its admission decision without
4//! depending on rosbag2, SQLite, or a renderer. Adapters can populate it from
5//! a bounded episode, while native/Web frontends can render the same trace and
6//! fail-closed mapping gate.
7
8use std::collections::BTreeSet;
9
10use crate::{StudioSource, ViewerError, ViewerResult};
11
12/// Current serialized one-command replay demo state schema version.
13pub const REPLAY_DEMO_STATE_VERSION: u32 = 1;
14
15/// One deterministic sample emitted by an episode replayer.
16#[derive(Clone, Debug, PartialEq, Eq)]
17#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
18#[cfg_attr(feature = "serde", serde(deny_unknown_fields))]
19pub struct ReplaySample {
20    /// Zero-based sequence number in deterministic replay order.
21    pub sequence: u64,
22    /// Logical source topic for the replayed record.
23    pub topic: String,
24    /// PointCloud2 header timestamp retained by the replay.
25    pub stamp_nanos: u64,
26    /// Number of points in the replayed record.
27    pub point_count: u64,
28    /// Topics paired with this record by the bounded sync window.
29    pub paired_topics: Vec<String>,
30}
31
32impl ReplaySample {
33    /// Creates one validated deterministic replay sample.
34    pub fn try_new(
35        sequence: u64,
36        topic: impl Into<String>,
37        stamp_nanos: u64,
38        point_count: u64,
39        paired_topics: Vec<String>,
40    ) -> ViewerResult<Self> {
41        let sample =
42            Self { sequence, topic: topic.into(), stamp_nanos, point_count, paired_topics };
43        sample.validate()?;
44        Ok(sample)
45    }
46
47    /// Validates topic identity, point count, and paired-topic uniqueness.
48    pub fn validate(&self) -> ViewerResult<()> {
49        if self.topic.trim().is_empty() || self.point_count == 0 {
50            return Err(ViewerError::InvalidState(
51                "replay samples require a topic and at least one point".into(),
52            ));
53        }
54        let mut topics = BTreeSet::new();
55        for paired_topic in &self.paired_topics {
56            if paired_topic.trim().is_empty()
57                || paired_topic == &self.topic
58                || !topics.insert(paired_topic)
59            {
60                return Err(ViewerError::InvalidState(
61                    "replay sample paired topics must be non-empty, distinct, and external".into(),
62                ));
63            }
64        }
65        Ok(())
66    }
67}
68
69/// Bounded episode and replay counters shown by the replay dashboard.
70#[derive(Clone, Debug, PartialEq, Eq)]
71#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
72#[cfg_attr(feature = "serde", serde(deny_unknown_fields))]
73pub struct ReplaySummary {
74    /// Number of records admitted into the bounded episode.
75    pub episode_record_count: u64,
76    /// Point count admitted into the bounded episode.
77    pub episode_point_count: u64,
78    /// Conservative allocated column bytes retained by the episode.
79    pub episode_byte_count: u64,
80    /// Number of records emitted by deterministic replay.
81    pub replayed_record_count: u64,
82    /// Number of primary/secondary bundles matched by the sync window.
83    pub matched_bundle_count: u64,
84    /// Largest matched timestamp delta in nanoseconds.
85    pub max_matched_delta_ns: u64,
86    /// Configured maximum timestamp delta in nanoseconds.
87    pub max_delta_ns: u64,
88    /// Wall time spent in bounded ingest and replay, in nanoseconds.
89    pub replay_wall_ns: u64,
90    /// Largest source allocation observed while reading the bag.
91    pub peak_source_bytes: u64,
92    /// Whether a second deterministic walk produced the same ordered trace.
93    pub deterministic_order_verified: bool,
94    /// Human-readable timestamp basis for the replay.
95    pub time_basis: String,
96    /// Whether a source-bound clock/frame calibration was applied.
97    pub calibration_applied: bool,
98}
99
100/// Bounded inventory for one point-cloud topic in the replay episode.
101#[derive(Clone, Debug, PartialEq, Eq)]
102#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
103#[cfg_attr(feature = "serde", serde(deny_unknown_fields))]
104pub struct ReplayTopic {
105    /// ROS topic name.
106    pub name: String,
107    /// Total messages present in the source bag for this topic.
108    pub bag_message_count: u64,
109    /// Number of bounded records retained from this topic.
110    pub retained_record_count: u64,
111    /// Number of points retained from this topic.
112    pub retained_point_count: u64,
113    /// Frame IDs observed in the retained records.
114    pub frame_ids: Vec<String>,
115}
116
117impl ReplayTopic {
118    /// Creates and validates one topic inventory.
119    pub fn try_new(
120        name: impl Into<String>,
121        bag_message_count: u64,
122        retained_record_count: u64,
123        retained_point_count: u64,
124        frame_ids: Vec<String>,
125    ) -> ViewerResult<Self> {
126        let topic = Self {
127            name: name.into(),
128            bag_message_count,
129            retained_record_count,
130            retained_point_count,
131            frame_ids,
132        };
133        topic.validate()?;
134        Ok(topic)
135    }
136
137    /// Validates topic identity, retained counters, and frame IDs.
138    pub fn validate(&self) -> ViewerResult<()> {
139        if self.name.trim().is_empty()
140            || (self.retained_record_count > 0 && self.retained_point_count == 0)
141        {
142            return Err(ViewerError::InvalidState(
143                "replay topic inventory has invalid identity or retained counters".into(),
144            ));
145        }
146        let mut frames = BTreeSet::new();
147        for frame_id in &self.frame_ids {
148            if frame_id.trim().is_empty() || !frames.insert(frame_id) {
149                return Err(ViewerError::InvalidState(
150                    "replay topic frame IDs must be non-empty and unique".into(),
151                ));
152            }
153        }
154        Ok(())
155    }
156}
157
158impl ReplaySummary {
159    /// Creates and validates bounded replay counters.
160    #[allow(clippy::too_many_arguments)]
161    pub fn try_new(
162        episode_record_count: u64,
163        episode_point_count: u64,
164        episode_byte_count: u64,
165        replayed_record_count: u64,
166        matched_bundle_count: u64,
167        max_matched_delta_ns: u64,
168        max_delta_ns: u64,
169        replay_wall_ns: u64,
170        peak_source_bytes: u64,
171        deterministic_order_verified: bool,
172        time_basis: impl Into<String>,
173        calibration_applied: bool,
174    ) -> ViewerResult<Self> {
175        let summary = Self {
176            episode_record_count,
177            episode_point_count,
178            episode_byte_count,
179            replayed_record_count,
180            matched_bundle_count,
181            max_matched_delta_ns,
182            max_delta_ns,
183            replay_wall_ns,
184            peak_source_bytes,
185            deterministic_order_verified,
186            time_basis: time_basis.into(),
187            calibration_applied,
188        };
189        summary.validate()?;
190        Ok(summary)
191    }
192
193    /// Validates counter ordering, sync bounds, and the timestamp basis.
194    pub fn validate(&self) -> ViewerResult<()> {
195        if self.time_basis.trim().is_empty() {
196            return Err(ViewerError::InvalidState("replay time basis must not be empty".into()));
197        }
198        if self.replayed_record_count > self.episode_record_count {
199            return Err(ViewerError::InvalidState(
200                "replayed record count cannot exceed episode record count".into(),
201            ));
202        }
203        if self.matched_bundle_count > self.replayed_record_count {
204            return Err(ViewerError::InvalidState(
205                "matched bundle count cannot exceed replayed record count".into(),
206            ));
207        }
208        if self.max_delta_ns == 0 && self.max_matched_delta_ns != 0 {
209            return Err(ViewerError::InvalidState(
210                "a non-zero matched delta requires a positive sync window".into(),
211            ));
212        }
213        if self.max_matched_delta_ns > self.max_delta_ns {
214            return Err(ViewerError::InvalidState(
215                "matched timestamp delta exceeds the configured sync window".into(),
216            ));
217        }
218        if self.calibration_applied && !self.deterministic_order_verified {
219            return Err(ViewerError::InvalidState(
220                "calibration cannot be applied to an unverified replay order".into(),
221            ));
222        }
223        Ok(())
224    }
225}
226
227/// Checksummed output artifact associated with a replay state.
228#[derive(Clone, Debug, PartialEq, Eq)]
229#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
230#[cfg_attr(feature = "serde", serde(deny_unknown_fields))]
231pub struct ReplayArtifact {
232    /// Logical role such as `state`, `dashboard`, or `manifest`.
233    pub role: String,
234    /// Absolute output path or source path represented by this receipt.
235    pub path: String,
236    /// Observed byte count.
237    pub size_bytes: u64,
238    /// Lowercase SHA-256 checksum.
239    pub sha256: String,
240}
241
242impl ReplayArtifact {
243    /// Creates and validates one replay artifact receipt.
244    pub fn try_new(
245        role: impl Into<String>,
246        path: impl Into<String>,
247        size_bytes: u64,
248        sha256: impl Into<String>,
249    ) -> ViewerResult<Self> {
250        let artifact =
251            Self { role: role.into(), path: path.into(), size_bytes, sha256: sha256.into() };
252        artifact.validate()?;
253        Ok(artifact)
254    }
255
256    /// Validates artifact identity and checksum shape.
257    pub fn validate(&self) -> ViewerResult<()> {
258        if self.role.trim().is_empty() || self.path.trim().is_empty() || self.size_bytes == 0 {
259            return Err(ViewerError::InvalidState(
260                "replay artifacts require a role, path, and non-zero size".into(),
261            ));
262        }
263        validate_sha256("replay artifact", &self.sha256)
264    }
265}
266
267/// Portable state emitted by the one-command deterministic replay demo.
268#[derive(Clone, Debug, PartialEq, Eq)]
269#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
270#[cfg_attr(feature = "serde", serde(deny_unknown_fields))]
271pub struct ReplayDemoState {
272    /// Serialized replay state schema version.
273    pub version: u32,
274    /// User-facing dashboard title.
275    pub title: String,
276    /// Checksummed input identity.
277    pub source: StudioSource,
278    /// Bounded topic inventory used to build the episode.
279    pub topics: Vec<ReplayTopic>,
280    /// Bounded episode and replay metrics.
281    pub summary: ReplaySummary,
282    /// Ordered replay trace samples.
283    pub samples: Vec<ReplaySample>,
284    /// Checksummed state/dashboard/manifest artifacts.
285    pub artifacts: Vec<ReplayArtifact>,
286    /// Whether deterministic replay has passed its admission checks.
287    pub replay_ready: bool,
288    /// Whether downstream mapping is admitted; calibration is never implicit.
289    pub mapping_admitted: bool,
290    /// Fail-closed reasons for mapping or replay admission.
291    pub blockers: Vec<String>,
292}
293
294impl ReplayDemoState {
295    /// Creates a replay state and derives replay and mapping admission.
296    pub fn try_new(
297        title: impl Into<String>,
298        source: StudioSource,
299        topics: Vec<ReplayTopic>,
300        summary: ReplaySummary,
301        samples: Vec<ReplaySample>,
302        artifacts: Vec<ReplayArtifact>,
303        blockers: Vec<String>,
304    ) -> ViewerResult<Self> {
305        let sample_count = u64::try_from(samples.len()).map_err(|_| {
306            ViewerError::InvalidState("replay sample count does not fit in u64".into())
307        })?;
308        let replay_ready = source.identity_matches
309            && summary.deterministic_order_verified
310            && summary.replayed_record_count > 0
311            && summary.replayed_record_count == summary.episode_record_count
312            && summary.replayed_record_count == sample_count;
313        let mapping_admitted = replay_ready && summary.calibration_applied;
314        let state = Self {
315            version: REPLAY_DEMO_STATE_VERSION,
316            title: title.into(),
317            source,
318            topics,
319            summary,
320            samples,
321            artifacts,
322            replay_ready,
323            mapping_admitted,
324            blockers,
325        };
326        state.validate()?;
327        Ok(state)
328    }
329
330    /// Validates the replay trace and all cross-panel admission invariants.
331    pub fn validate(&self) -> ViewerResult<()> {
332        if self.version != REPLAY_DEMO_STATE_VERSION {
333            return Err(ViewerError::InvalidState(format!(
334                "unsupported replay demo state version {}",
335                self.version
336            )));
337        }
338        if self.title.trim().is_empty() {
339            return Err(ViewerError::InvalidState("replay demo title must not be empty".into()));
340        }
341        self.source.validate()?;
342        let mut topic_names = BTreeSet::new();
343        for topic in &self.topics {
344            topic.validate()?;
345            if !topic_names.insert(&topic.name) {
346                return Err(ViewerError::InvalidState(
347                    "replay topic inventory contains a duplicate topic".into(),
348                ));
349            }
350        }
351        self.summary.validate()?;
352
353        for (expected_sequence, sample) in self.samples.iter().enumerate() {
354            sample.validate()?;
355            if sample.sequence != u64::try_from(expected_sequence).unwrap_or(u64::MAX) {
356                return Err(ViewerError::InvalidState(
357                    "replay samples must use contiguous zero-based sequence numbers".into(),
358                ));
359            }
360        }
361
362        let mut artifact_roles = BTreeSet::new();
363        let mut artifact_paths = BTreeSet::new();
364        for artifact in &self.artifacts {
365            artifact.validate()?;
366            if !artifact_roles.insert(&artifact.role) || !artifact_paths.insert(&artifact.path) {
367                return Err(ViewerError::InvalidState(
368                    "replay artifacts must have unique roles and paths".into(),
369                ));
370            }
371        }
372
373        let sample_count = u64::try_from(self.samples.len()).unwrap_or(u64::MAX);
374        let calculated_replay_ready = self.source.identity_matches
375            && self.summary.deterministic_order_verified
376            && self.summary.replayed_record_count > 0
377            && self.summary.replayed_record_count == self.summary.episode_record_count
378            && self.summary.replayed_record_count == sample_count;
379        if self.replay_ready != calculated_replay_ready {
380            return Err(ViewerError::InvalidState(
381                "replay_ready disagrees with source, summary, or trace admission".into(),
382            ));
383        }
384        let calculated_mapping = self.replay_ready && self.summary.calibration_applied;
385        if self.mapping_admitted != calculated_mapping {
386            return Err(ViewerError::InvalidState(
387                "mapping_admitted disagrees with replay and calibration admission".into(),
388            ));
389        }
390        if self.mapping_admitted && !self.blockers.is_empty() {
391            return Err(ViewerError::InvalidState(
392                "admitted replay mapping cannot contain blockers".into(),
393            ));
394        }
395        if !self.mapping_admitted && self.blockers.is_empty() {
396            return Err(ViewerError::InvalidState(
397                "blocked replay mapping must expose at least one blocker".into(),
398            ));
399        }
400        if self.blockers.iter().any(|blocker| blocker.trim().is_empty()) {
401            return Err(ViewerError::InvalidState(
402                "replay blockers must not contain empty messages".into(),
403            ));
404        }
405        Ok(())
406    }
407}
408
409fn validate_sha256(label: &str, value: &str) -> ViewerResult<()> {
410    if value.len() != 64
411        || !value.bytes().all(|byte| byte.is_ascii_hexdigit())
412        || value.bytes().any(|byte| byte.is_ascii_uppercase())
413    {
414        return Err(ViewerError::InvalidState(format!(
415            "{label} SHA-256 must be 64 lowercase hexadecimal characters"
416        )));
417    }
418    Ok(())
419}
420
421#[cfg(test)]
422mod tests {
423    use super::*;
424
425    const SHA: &str = "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef";
426
427    fn source(identity_matches: bool) -> StudioSource {
428        StudioSource::try_new(
429            "canonical bag",
430            "/media/input.db3",
431            SHA,
432            if identity_matches {
433                SHA
434            } else {
435                "fedcba9876543210fedcba9876543210fedcba9876543210fedcba9876543210"
436            },
437            identity_matches,
438        )
439        .unwrap()
440    }
441
442    fn summary(calibration_applied: bool) -> ReplaySummary {
443        ReplaySummary::try_new(
444            2,
445            8,
446            96,
447            2,
448            1,
449            4,
450            10,
451            20,
452            128,
453            true,
454            "PointCloud2 header stamp; no clock calibration applied",
455            calibration_applied,
456        )
457        .unwrap()
458    }
459
460    fn samples() -> Vec<ReplaySample> {
461        vec![
462            ReplaySample::try_new(0, "/front", 10, 4, vec!["/rear".into()]).unwrap(),
463            ReplaySample::try_new(1, "/rear", 14, 4, vec!["/front".into()]).unwrap(),
464        ]
465    }
466
467    #[test]
468    fn blocked_mapping_state_roundtrips_with_serde() {
469        let state = ReplayDemoState::try_new(
470            "Replay demo",
471            source(true),
472            vec![ReplayTopic::try_new("/front", 2, 1, 4, vec!["front_frame".into()]).unwrap()],
473            summary(false),
474            samples(),
475            Vec::new(),
476            vec!["clock calibration not applied".into()],
477        )
478        .unwrap();
479        assert!(state.replay_ready);
480        assert!(!state.mapping_admitted);
481        #[cfg(feature = "serde")]
482        {
483            let json = serde_json::to_string(&state).unwrap();
484            assert_eq!(serde_json::from_str::<ReplayDemoState>(&json).unwrap(), state);
485        }
486    }
487
488    #[test]
489    fn source_mismatch_cannot_be_admitted() {
490        let state = ReplayDemoState::try_new(
491            "Replay demo",
492            source(false),
493            vec![ReplayTopic::try_new("/front", 2, 1, 4, vec!["front_frame".into()]).unwrap()],
494            summary(false),
495            samples(),
496            Vec::new(),
497            vec!["input SHA-256 mismatch".into()],
498        )
499        .unwrap();
500        assert!(!state.replay_ready);
501        assert!(!state.mapping_admitted);
502    }
503
504    #[test]
505    fn non_contiguous_trace_is_rejected() {
506        let mut trace = samples();
507        trace[1].sequence = 3;
508        assert!(ReplayDemoState::try_new(
509            "Replay demo",
510            source(true),
511            vec![ReplayTopic::try_new("/front", 2, 1, 4, vec!["front_frame".into()]).unwrap()],
512            summary(false),
513            trace,
514            Vec::new(),
515            vec!["trace is invalid".into()],
516        )
517        .is_err());
518    }
519
520    #[test]
521    fn applied_calibration_requires_verified_order() {
522        assert!(ReplaySummary::try_new(1, 1, 1, 1, 0, 0, 1, 1, 1, false, "header stamp", true,)
523            .is_err());
524    }
525}
spatialrust_viewer/replay.rs

spatialrust_viewer/
replay.rs
