spatialrust_sync/
episode_builder.rs

1//! Bounded construction of deterministic in-memory episodes.
2
3use spatialrust_records::record_storage_bytes;
4
5use crate::{MemoryEpisode, StampedRecord, SyncError, SyncResult};
6
7/// Hard limits applied while collecting an episode.
8///
9/// The byte limit accounts for the allocated scalar capacity of every record,
10/// which is deliberately conservative for external or untrusted inputs.
11#[derive(Clone, Copy, Debug, PartialEq, Eq)]
12pub struct EpisodeLimits {
13    /// Maximum number of stamped records.
14    pub max_records: u64,
15    /// Maximum total point count across all records.
16    pub max_points: u64,
17    /// Maximum allocated column-storage bytes across all records.
18    pub max_bytes: u64,
19}
20
21impl EpisodeLimits {
22    /// Creates episode limits; [`MemoryEpisodeBuilder::try_new`] validates them.
23    #[must_use]
24    pub const fn new(max_records: u64, max_points: u64, max_bytes: u64) -> Self {
25        Self { max_records, max_points, max_bytes }
26    }
27
28    fn validate(self) -> SyncResult<Self> {
29        for (name, value) in [
30            ("max_records", self.max_records),
31            ("max_points", self.max_points),
32            ("max_bytes", self.max_bytes),
33        ] {
34            if value == 0 {
35                return Err(SyncError::InvalidConfiguration(format!(
36                    "episode {name} must be greater than zero"
37                )));
38            }
39        }
40        Ok(self)
41    }
42}
43
44impl Default for EpisodeLimits {
45    fn default() -> Self {
46        Self::new(4_096, 16_777_216, 512 * 1024 * 1024)
47    }
48}
49
50/// Bounded collector for a [`MemoryEpisode`].
51///
52/// Records are retained only after all three limits have been checked. Call
53/// [`MemoryEpisodeBuilder::finish`] to sort them into the episode's stable
54/// timestamp/topic order.
55#[derive(Debug)]
56pub struct MemoryEpisodeBuilder {
57    limits: EpisodeLimits,
58    records: Vec<StampedRecord>,
59    points: u64,
60    bytes: u64,
61}
62
63impl MemoryEpisodeBuilder {
64    /// Creates an empty builder with validated hard limits.
65    pub fn try_new(limits: EpisodeLimits) -> SyncResult<Self> {
66        Ok(Self { limits: limits.validate()?, records: Vec::new(), points: 0, bytes: 0 })
67    }
68
69    /// Adds one stamped record if all configured limits remain satisfied.
70    pub fn push(&mut self, stamped: StampedRecord) -> SyncResult<()> {
71        let record_points = u64::try_from(stamped.record.cloud().len())
72            .map_err(|_| SyncError::InvalidConfiguration("episode point count overflow".into()))?;
73        let record_bytes = record_storage_bytes(&stamped.record)?;
74        let current_records = u64::try_from(self.records.len())
75            .map_err(|_| SyncError::InvalidConfiguration("episode record count overflow".into()))?;
76        let next_points = self.points.checked_add(record_points).ok_or_else(|| {
77            SyncError::InvalidConfiguration("episode point count overflow".into())
78        })?;
79        let next_bytes = self
80            .bytes
81            .checked_add(record_bytes)
82            .ok_or_else(|| SyncError::InvalidConfiguration("episode byte count overflow".into()))?;
83
84        check_limit("records", 1, current_records, self.limits.max_records)?;
85        check_limit("points", record_points, self.points, self.limits.max_points)?;
86        check_limit("bytes", record_bytes, self.bytes, self.limits.max_bytes)?;
87
88        self.records.push(stamped);
89        self.points = next_points;
90        self.bytes = next_bytes;
91        Ok(())
92    }
93
94    /// Returns the number of records currently retained.
95    #[must_use]
96    pub fn len(&self) -> usize {
97        self.records.len()
98    }
99
100    /// Returns whether the builder contains no records.
101    #[must_use]
102    pub fn is_empty(&self) -> bool {
103        self.records.is_empty()
104    }
105
106    /// Returns the total point count currently retained.
107    #[must_use]
108    pub const fn points(&self) -> u64 {
109        self.points
110    }
111
112    /// Returns the conservative allocated column-storage total.
113    #[must_use]
114    pub const fn bytes(&self) -> u64 {
115        self.bytes
116    }
117
118    /// Returns the hard limits used by this builder.
119    #[must_use]
120    pub const fn limits(&self) -> EpisodeLimits {
121        self.limits
122    }
123
124    /// Finishes the bounded collection as a deterministic in-memory episode.
125    #[must_use]
126    pub fn finish(self) -> MemoryEpisode {
127        MemoryEpisode::from_records(self.records)
128    }
129}
130
131fn check_limit(resource: &'static str, requested: u64, current: u64, limit: u64) -> SyncResult<()> {
132    let next = current.checked_add(requested).ok_or(SyncError::EpisodeLimitExceeded {
133        resource,
134        requested,
135        current,
136        limit,
137    })?;
138    if next > limit {
139        return Err(SyncError::EpisodeLimitExceeded { resource, requested, current, limit });
140    }
141    Ok(())
142}
143
144#[cfg(test)]
145mod tests {
146    use super::{EpisodeLimits, MemoryEpisodeBuilder};
147    use crate::{ClockDomain, StampedRecord, StampedTime, TopicId};
148    use spatialrust_core::{
149        PointBuffer, PointBufferSet, PointCloud, SpatialMetadata, StandardSchemas, Timestamp,
150    };
151    use spatialrust_records::{SchemaVersion, SpatialRecord};
152
153    fn sample(topic: &str, timestamp: u64, point_count: usize) -> StampedRecord {
154        let mut buffers = PointBufferSet::new();
155        buffers.insert("x", PointBuffer::from_f32(vec![1.0; point_count]));
156        buffers.insert("y", PointBuffer::from_f32(vec![0.0; point_count]));
157        buffers.insert("z", PointBuffer::from_f32(vec![0.0; point_count]));
158        let cloud = PointCloud::try_from_parts(
159            StandardSchemas::point_xyz(),
160            buffers,
161            SpatialMetadata::default(),
162        )
163        .unwrap();
164        let record =
165            SpatialRecord::try_from_cloud("point", SchemaVersion::new(1, 0), cloud).unwrap();
166        StampedRecord::new(
167            TopicId::new(topic),
168            StampedTime::exact("test", ClockDomain::External, Timestamp::from_nanos(timestamp)),
169            record,
170        )
171    }
172
173    #[test]
174    fn enforces_limits_before_retaining_record() {
175        let mut builder = MemoryEpisodeBuilder::try_new(EpisodeLimits::new(1, 4, 100)).unwrap();
176        builder.push(sample("lidar", 20, 1)).unwrap();
177        let error = builder.push(sample("lidar", 10, 2)).unwrap_err();
178        assert!(error.to_string().contains("episode records limit exceeded"));
179        assert_eq!(builder.len(), 1);
180        assert_eq!(builder.points(), 1);
181        assert_eq!(builder.bytes(), 12);
182    }
183
184    #[test]
185    fn finish_uses_episode_deterministic_order() {
186        let mut builder = MemoryEpisodeBuilder::try_new(EpisodeLimits::new(4, 4, 48)).unwrap();
187        builder.push(sample("lidar", 20, 1)).unwrap();
188        builder.push(sample("camera", 10, 1)).unwrap();
189        let episode = builder.finish();
190        assert_eq!(episode.records()[0].topic.as_str(), "camera");
191        assert_eq!(episode.records()[1].topic.as_str(), "lidar");
192    }
193}
spatialrust_sync/episode_builder.rs

spatialrust_sync/
episode_builder.rs
