spatialrust_records/
streaming.rs

1//! Bounded streaming configuration, memory accounting, receipts, and workloads.
2
3use std::collections::BTreeMap;
4use std::num::NonZeroUsize;
5use std::sync::{
6    atomic::{AtomicBool, AtomicU64, Ordering},
7    Arc,
8};
9
10use crate::{RecordsError, RecordsResult};
11
12/// Default number of points requested from a streaming source.
13pub const DEFAULT_STREAM_CHUNK_POINTS: usize = 16_384;
14/// Default hard limit for explicitly tracked streaming memory (256 MiB).
15pub const DEFAULT_STREAM_MEMORY_BUDGET_BYTES: u64 = 256 * 1024 * 1024;
16/// Stable identifier for the versioned JSON receipt contract.
17pub const STREAMING_RECEIPT_SCHEMA: &str = "spatialrust.streaming.receipt";
18/// Current streaming receipt schema version.
19pub const STREAMING_RECEIPT_VERSION: u32 = 1;
20
21/// Hard limit for memory explicitly owned by a streaming execution.
22#[derive(Clone, Copy, Debug, PartialEq, Eq)]
23pub struct MemoryBudget {
24    limit_bytes: u64,
25}
26
27impl MemoryBudget {
28    /// Creates a non-zero hard memory limit.
29    pub fn new(limit_bytes: u64) -> RecordsResult<Self> {
30        if limit_bytes == 0 {
31            return Err(RecordsError::InvalidConfiguration(
32                "streaming memory budget must be positive".into(),
33            ));
34        }
35        Ok(Self { limit_bytes })
36    }
37
38    /// Returns the hard limit in bytes.
39    #[must_use]
40    pub const fn limit_bytes(self) -> u64 {
41        self.limit_bytes
42    }
43}
44
45impl Default for MemoryBudget {
46    fn default() -> Self {
47        Self { limit_bytes: DEFAULT_STREAM_MEMORY_BUDGET_BYTES }
48    }
49}
50
51/// Ordering guarantee requested from a streaming source or pipeline.
52#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
53pub enum StreamOrdering {
54    /// Preserve the deterministic order defined by the source.
55    #[default]
56    Source,
57    /// Permit reordering while requiring deterministic output for identical inputs.
58    Deterministic,
59}
60
61/// Common bounded-stream execution options.
62#[derive(Clone, Debug, PartialEq, Eq)]
63pub struct StreamOptions {
64    chunk_points: NonZeroUsize,
65    memory_budget: MemoryBudget,
66    prefetch_chunks: usize,
67    ordering: StreamOrdering,
68}
69
70impl StreamOptions {
71    /// Creates options with an explicit positive chunk size and hard memory budget.
72    pub fn new(chunk_points: usize, memory_budget: MemoryBudget) -> RecordsResult<Self> {
73        let chunk_points = NonZeroUsize::new(chunk_points).ok_or_else(|| {
74            RecordsError::InvalidConfiguration("stream chunk_points must be positive".into())
75        })?;
76        Ok(Self {
77            chunk_points,
78            memory_budget,
79            prefetch_chunks: 0,
80            ordering: StreamOrdering::Source,
81        })
82    }
83
84    /// Requests bounded source prefetch. Zero disables prefetch.
85    #[must_use]
86    pub const fn with_prefetch_chunks(mut self, prefetch_chunks: usize) -> Self {
87        self.prefetch_chunks = prefetch_chunks;
88        self
89    }
90
91    /// Selects the required ordering contract.
92    #[must_use]
93    pub const fn with_ordering(mut self, ordering: StreamOrdering) -> Self {
94        self.ordering = ordering;
95        self
96    }
97
98    /// Returns the requested maximum points per source chunk.
99    #[must_use]
100    pub const fn chunk_points(&self) -> usize {
101        self.chunk_points.get()
102    }
103
104    /// Returns the hard memory budget.
105    #[must_use]
106    pub const fn memory_budget(&self) -> MemoryBudget {
107        self.memory_budget
108    }
109
110    /// Returns the maximum number of prefetched chunks.
111    #[must_use]
112    pub const fn prefetch_chunks(&self) -> usize {
113        self.prefetch_chunks
114    }
115
116    /// Returns the required ordering contract.
117    #[must_use]
118    pub const fn ordering(&self) -> StreamOrdering {
119        self.ordering
120    }
121}
122
123impl Default for StreamOptions {
124    fn default() -> Self {
125        Self {
126            chunk_points: NonZeroUsize::new(DEFAULT_STREAM_CHUNK_POINTS)
127                .expect("default stream chunk size is non-zero"),
128            memory_budget: MemoryBudget::default(),
129            prefetch_chunks: 0,
130            ordering: StreamOrdering::Source,
131        }
132    }
133}
134
135#[derive(Debug)]
136struct MemoryTrackerInner {
137    limit_bytes: u64,
138    current_bytes: AtomicU64,
139    peak_bytes: AtomicU64,
140}
141
142/// Concurrent, exact accounting for memory explicitly owned by a stream.
143///
144/// Reservations fail before the configured hard limit is exceeded. Dropping a
145/// [`MemoryReservation`] releases its bytes, including during error unwinding.
146#[derive(Clone, Debug)]
147pub struct MemoryTracker {
148    inner: Arc<MemoryTrackerInner>,
149}
150
151impl MemoryTracker {
152    /// Creates a tracker for `budget`.
153    #[must_use]
154    pub fn new(budget: MemoryBudget) -> Self {
155        Self {
156            inner: Arc::new(MemoryTrackerInner {
157                limit_bytes: budget.limit_bytes,
158                current_bytes: AtomicU64::new(0),
159                peak_bytes: AtomicU64::new(0),
160            }),
161        }
162    }
163
164    /// Atomically reserves bytes or fails without changing the current count.
165    pub fn try_reserve(&self, bytes: u64) -> RecordsResult<MemoryReservation> {
166        let limit = self.inner.limit_bytes;
167        let result =
168            self.inner.current_bytes.fetch_update(Ordering::AcqRel, Ordering::Acquire, |current| {
169                current.checked_add(bytes).filter(|next| *next <= limit)
170            });
171        match result {
172            Ok(previous) => {
173                let current = previous + bytes;
174                self.inner.peak_bytes.fetch_max(current, Ordering::AcqRel);
175                Ok(MemoryReservation { tracker: self.clone(), bytes })
176            }
177            Err(current) => {
178                Err(RecordsError::MemoryBudgetExceeded { requested: bytes, current, limit })
179            }
180        }
181    }
182
183    /// Returns an atomic snapshot of current, peak, and limit bytes.
184    #[must_use]
185    pub fn snapshot(&self) -> MemorySnapshot {
186        MemorySnapshot {
187            current_bytes: self.inner.current_bytes.load(Ordering::Acquire),
188            peak_bytes: self.inner.peak_bytes.load(Ordering::Acquire),
189            limit_bytes: self.inner.limit_bytes,
190        }
191    }
192}
193
194/// Point-in-time memory accounting values.
195#[derive(Clone, Copy, Debug, PartialEq, Eq)]
196pub struct MemorySnapshot {
197    /// Bytes currently reserved.
198    pub current_bytes: u64,
199    /// Maximum simultaneously reserved bytes.
200    pub peak_bytes: u64,
201    /// Configured hard limit.
202    pub limit_bytes: u64,
203}
204
205/// RAII token that releases an exact tracked byte count on drop.
206#[derive(Debug)]
207pub struct MemoryReservation {
208    tracker: MemoryTracker,
209    bytes: u64,
210}
211
212impl MemoryReservation {
213    /// Returns the reserved byte count.
214    #[must_use]
215    pub const fn bytes(&self) -> u64 {
216        self.bytes
217    }
218
219    /// Releases the unused tail of a conservative reservation.
220    pub fn shrink_to(&mut self, bytes: u64) -> RecordsResult<()> {
221        if bytes > self.bytes {
222            return Err(RecordsError::InvalidConfiguration(format!(
223                "cannot grow a memory reservation from {} to {bytes} bytes",
224                self.bytes
225            )));
226        }
227        let released = self.bytes - bytes;
228        if released > 0 {
229            let previous = self.tracker.inner.current_bytes.fetch_sub(released, Ordering::AcqRel);
230            debug_assert!(previous >= released, "memory reservation accounting underflow");
231            self.bytes = bytes;
232        }
233        Ok(())
234    }
235}
236
237impl Drop for MemoryReservation {
238    fn drop(&mut self) {
239        let previous = self.tracker.inner.current_bytes.fetch_sub(self.bytes, Ordering::AcqRel);
240        debug_assert!(previous >= self.bytes, "memory reservation accounting underflow");
241    }
242}
243
244/// Cloneable cooperative cancellation state checked at chunk boundaries.
245#[derive(Clone, Debug, Default)]
246pub struct CancellationToken {
247    cancelled: Arc<AtomicBool>,
248}
249
250impl CancellationToken {
251    /// Requests cancellation. Repeated calls are harmless.
252    pub fn cancel(&self) {
253        self.cancelled.store(true, Ordering::Release);
254    }
255
256    /// Returns whether cancellation has been requested.
257    #[must_use]
258    pub fn is_cancelled(&self) -> bool {
259        self.cancelled.load(Ordering::Acquire)
260    }
261
262    /// Returns [`RecordsError::Cancelled`] after cancellation is observed.
263    pub fn check(&self) -> RecordsResult<()> {
264        if self.is_cancelled() {
265            Err(RecordsError::Cancelled)
266        } else {
267            Ok(())
268        }
269    }
270}
271
272/// Direction of an explicit host/device transfer.
273#[cfg_attr(feature = "receipt-json", derive(serde::Serialize, serde::Deserialize))]
274#[cfg_attr(feature = "receipt-json", serde(rename_all = "snake_case"))]
275#[cfg_attr(feature = "receipt-json", serde(deny_unknown_fields))]
276#[derive(Clone, Copy, Debug, PartialEq, Eq)]
277pub enum StreamingTransferDirection {
278    /// Host memory to a device.
279    HostToDevice,
280    /// Device memory to the host.
281    DeviceToHost,
282    /// Explicit device-to-device movement.
283    DeviceToDevice,
284}
285
286/// One named, explicit transfer included in a streaming receipt.
287#[cfg_attr(feature = "receipt-json", derive(serde::Serialize, serde::Deserialize))]
288#[cfg_attr(feature = "receipt-json", serde(deny_unknown_fields))]
289#[derive(Clone, Debug, PartialEq, Eq)]
290pub struct StreamingTransferReceipt {
291    /// Stable operation name.
292    pub name: String,
293    /// Transfer direction.
294    pub direction: StreamingTransferDirection,
295    /// Bytes moved.
296    pub bytes: u64,
297}
298
299/// Timing and allocation counters for one named execution phase.
300#[cfg_attr(feature = "receipt-json", derive(serde::Serialize, serde::Deserialize))]
301#[cfg_attr(feature = "receipt-json", serde(deny_unknown_fields))]
302#[derive(Clone, Debug, PartialEq, Eq)]
303pub struct StreamingPhaseReceipt {
304    /// Elapsed wall-clock nanoseconds.
305    pub elapsed_ns: u64,
306    /// Explicitly allocated bytes attributed to the phase.
307    pub allocated_bytes: u64,
308}
309
310/// Versioned, deterministic accounting receipt for one streaming execution.
311#[cfg_attr(feature = "receipt-json", derive(serde::Serialize, serde::Deserialize))]
312#[cfg_attr(feature = "receipt-json", serde(deny_unknown_fields))]
313#[derive(Clone, Debug, PartialEq, Eq)]
314pub struct StreamingReceipt {
315    schema: String,
316    version: u32,
317    source_id: String,
318    input_points: u64,
319    output_points: u64,
320    chunks_read: u64,
321    chunks_written: u64,
322    bytes_read: u64,
323    bytes_written: u64,
324    peak_tracked_bytes: u64,
325    spilled_bytes: u64,
326    phases: BTreeMap<String, StreamingPhaseReceipt>,
327    transfers: Vec<StreamingTransferReceipt>,
328}
329
330impl StreamingReceipt {
331    /// Creates an empty v1 receipt for a non-empty source identifier.
332    pub fn new(source_id: impl Into<String>) -> RecordsResult<Self> {
333        let source_id = source_id.into();
334        if source_id.trim().is_empty() {
335            return Err(RecordsError::InvalidReceipt("source_id must not be empty".into()));
336        }
337        Ok(Self {
338            schema: STREAMING_RECEIPT_SCHEMA.into(),
339            version: STREAMING_RECEIPT_VERSION,
340            source_id,
341            input_points: 0,
342            output_points: 0,
343            chunks_read: 0,
344            chunks_written: 0,
345            bytes_read: 0,
346            bytes_written: 0,
347            peak_tracked_bytes: 0,
348            spilled_bytes: 0,
349            phases: BTreeMap::new(),
350            transfers: Vec::new(),
351        })
352    }
353
354    /// Records one input chunk using checked counters.
355    pub fn record_input_chunk(&mut self, points: u64, bytes: u64) -> RecordsResult<()> {
356        checked_add(&mut self.input_points, points, "input_points")?;
357        checked_add(&mut self.bytes_read, bytes, "bytes_read")?;
358        checked_add(&mut self.chunks_read, 1, "chunks_read")
359    }
360
361    /// Records one output chunk using checked counters.
362    pub fn record_output_chunk(&mut self, points: u64, bytes: u64) -> RecordsResult<()> {
363        checked_add(&mut self.output_points, points, "output_points")?;
364        checked_add(&mut self.bytes_written, bytes, "bytes_written")?;
365        checked_add(&mut self.chunks_written, 1, "chunks_written")
366    }
367
368    /// Records bytes written to explicit temporary spill storage.
369    pub fn record_spill(&mut self, bytes: u64) -> RecordsResult<()> {
370        checked_add(&mut self.spilled_bytes, bytes, "spilled_bytes")
371    }
372
373    /// Captures the peak from an exact memory tracker.
374    pub fn capture_memory(&mut self, tracker: &MemoryTracker) {
375        self.peak_tracked_bytes = self.peak_tracked_bytes.max(tracker.snapshot().peak_bytes);
376    }
377
378    /// Inserts or replaces one named phase receipt.
379    pub fn record_phase(
380        &mut self,
381        name: impl Into<String>,
382        elapsed_ns: u64,
383        allocated_bytes: u64,
384    ) -> RecordsResult<()> {
385        let name = name.into();
386        if name.trim().is_empty() {
387            return Err(RecordsError::InvalidReceipt("phase name must not be empty".into()));
388        }
389        self.phases.insert(name, StreamingPhaseReceipt { elapsed_ns, allocated_bytes });
390        Ok(())
391    }
392
393    /// Appends one named explicit transfer.
394    pub fn record_transfer(
395        &mut self,
396        name: impl Into<String>,
397        direction: StreamingTransferDirection,
398        bytes: u64,
399    ) -> RecordsResult<()> {
400        let name = name.into();
401        if name.trim().is_empty() {
402            return Err(RecordsError::InvalidReceipt("transfer name must not be empty".into()));
403        }
404        self.transfers.push(StreamingTransferReceipt { name, direction, bytes });
405        Ok(())
406    }
407
408    /// Validates schema identity and all name constraints.
409    pub fn validate(&self) -> RecordsResult<()> {
410        if self.schema != STREAMING_RECEIPT_SCHEMA || self.version != STREAMING_RECEIPT_VERSION {
411            return Err(RecordsError::InvalidReceipt(format!(
412                "expected {STREAMING_RECEIPT_SCHEMA} v{STREAMING_RECEIPT_VERSION}, found {} v{}",
413                self.schema, self.version
414            )));
415        }
416        if self.source_id.trim().is_empty() {
417            return Err(RecordsError::InvalidReceipt("source_id must not be empty".into()));
418        }
419        if self.phases.keys().any(|name| name.trim().is_empty()) {
420            return Err(RecordsError::InvalidReceipt("phase name must not be empty".into()));
421        }
422        if self.transfers.iter().any(|transfer| transfer.name.trim().is_empty()) {
423            return Err(RecordsError::InvalidReceipt("transfer name must not be empty".into()));
424        }
425        Ok(())
426    }
427
428    /// Serializes this receipt using the versioned JSON contract.
429    #[cfg(feature = "receipt-json")]
430    pub fn to_json(&self) -> RecordsResult<String> {
431        self.validate()?;
432        serde_json::to_string_pretty(self)
433            .map_err(|error| RecordsError::InvalidReceipt(error.to_string()))
434    }
435
436    /// Parses and validates a versioned JSON receipt.
437    #[cfg(feature = "receipt-json")]
438    pub fn from_json(json: &str) -> RecordsResult<Self> {
439        let receipt: Self = serde_json::from_str(json)
440            .map_err(|error| RecordsError::InvalidReceipt(error.to_string()))?;
441        receipt.validate()?;
442        Ok(receipt)
443    }
444
445    /// Returns the receipt schema version.
446    #[must_use]
447    pub const fn version(&self) -> u32 {
448        self.version
449    }
450
451    /// Returns the stable source identifier.
452    #[must_use]
453    pub fn source_id(&self) -> &str {
454        &self.source_id
455    }
456
457    /// Returns input points observed so far.
458    #[must_use]
459    pub const fn input_points(&self) -> u64 {
460        self.input_points
461    }
462
463    /// Returns output points observed so far.
464    #[must_use]
465    pub const fn output_points(&self) -> u64 {
466        self.output_points
467    }
468
469    /// Returns input chunks observed so far.
470    #[must_use]
471    pub const fn chunks_read(&self) -> u64 {
472        self.chunks_read
473    }
474
475    /// Returns output chunks observed so far.
476    #[must_use]
477    pub const fn chunks_written(&self) -> u64 {
478        self.chunks_written
479    }
480
481    /// Returns input bytes observed so far.
482    #[must_use]
483    pub const fn bytes_read(&self) -> u64 {
484        self.bytes_read
485    }
486
487    /// Returns output bytes observed so far.
488    #[must_use]
489    pub const fn bytes_written(&self) -> u64 {
490        self.bytes_written
491    }
492
493    /// Returns the maximum explicitly tracked live memory.
494    #[must_use]
495    pub const fn peak_tracked_bytes(&self) -> u64 {
496        self.peak_tracked_bytes
497    }
498
499    /// Returns bytes written to temporary spill storage.
500    #[must_use]
501    pub const fn spilled_bytes(&self) -> u64 {
502        self.spilled_bytes
503    }
504
505    /// Returns deterministic phase receipts ordered by name.
506    #[must_use]
507    pub const fn phases(&self) -> &BTreeMap<String, StreamingPhaseReceipt> {
508        &self.phases
509    }
510
511    /// Returns explicit transfers in execution order.
512    #[must_use]
513    pub fn transfers(&self) -> &[StreamingTransferReceipt] {
514        &self.transfers
515    }
516}
517
518fn checked_add(target: &mut u64, value: u64, name: &str) -> RecordsResult<()> {
519    *target =
520        target.checked_add(value).ok_or_else(|| RecordsError::ReceiptOverflow(name.to_owned()))?;
521    Ok(())
522}
523
524/// One canonical scale/chunk/budget combination for reproducible comparisons.
525#[derive(Clone, Copy, Debug, PartialEq, Eq)]
526pub struct StreamingWorkload {
527    /// Stable workload identifier.
528    pub id: &'static str,
529    /// Number of generated or selected input points.
530    pub point_count: u64,
531    /// Requested maximum points per chunk.
532    pub chunk_points: usize,
533    /// Hard tracked-memory limit.
534    pub memory_budget_bytes: u64,
535}
536
537const STREAMING_WORKLOADS: [StreamingWorkload; 9] = [
538    workload("stream-1m-16k", 1_000_000, 16_384, 64),
539    workload("stream-1m-64k", 1_000_000, 65_536, 64),
540    workload("stream-1m-256k", 1_000_000, 262_144, 128),
541    workload("stream-10m-16k", 10_000_000, 16_384, 64),
542    workload("stream-10m-64k", 10_000_000, 65_536, 64),
543    workload("stream-10m-256k", 10_000_000, 262_144, 128),
544    workload("stream-100m-16k", 100_000_000, 16_384, 64),
545    workload("stream-100m-64k", 100_000_000, 65_536, 64),
546    workload("stream-100m-256k", 100_000_000, 262_144, 128),
547];
548
549const fn workload(
550    id: &'static str,
551    point_count: u64,
552    chunk_points: usize,
553    memory_mib: u64,
554) -> StreamingWorkload {
555    StreamingWorkload {
556        id,
557        point_count,
558        chunk_points,
559        memory_budget_bytes: memory_mib * 1024 * 1024,
560    }
561}
562
563/// Returns the canonical 1M/10M/100M streaming workload matrix.
564#[must_use]
565pub const fn canonical_streaming_workloads() -> &'static [StreamingWorkload] {
566    &STREAMING_WORKLOADS
567}
568
569#[cfg(test)]
570mod tests {
571    use std::sync::{Arc, Barrier};
572    use std::thread;
573
574    #[cfg(feature = "receipt-json")]
575    use super::STREAMING_RECEIPT_SCHEMA;
576    use super::{
577        canonical_streaming_workloads, CancellationToken, MemoryBudget, MemoryTracker,
578        StreamOptions, StreamingReceipt, StreamingTransferDirection,
579    };
580    use crate::RecordsError;
581
582    #[test]
583    fn memory_budget_is_fail_closed_and_drop_releases() {
584        let tracker = MemoryTracker::new(MemoryBudget::new(100).unwrap());
585        let first = tracker.try_reserve(60).unwrap();
586        assert_eq!(first.bytes(), 60);
587        let error = tracker.try_reserve(41).unwrap_err();
588        assert!(matches!(
589            error,
590            RecordsError::MemoryBudgetExceeded { requested: 41, current: 60, limit: 100 }
591        ));
592        assert_eq!(tracker.snapshot().current_bytes, 60);
593        drop(first);
594        assert_eq!(tracker.snapshot().current_bytes, 0);
595        assert_eq!(tracker.snapshot().peak_bytes, 60);
596    }
597
598    #[test]
599    fn reservation_can_shrink_but_never_grow() {
600        let tracker = MemoryTracker::new(MemoryBudget::new(100).unwrap());
601        let mut reservation = tracker.try_reserve(80).unwrap();
602        reservation.shrink_to(30).unwrap();
603        assert_eq!(reservation.bytes(), 30);
604        assert_eq!(tracker.snapshot().current_bytes, 30);
605        assert!(reservation.shrink_to(31).is_err());
606        assert_eq!(tracker.snapshot().current_bytes, 30);
607    }
608
609    #[test]
610    fn concurrent_reservations_never_exceed_limit() {
611        let tracker = MemoryTracker::new(MemoryBudget::new(100).unwrap());
612        let barrier = Arc::new(Barrier::new(3));
613        let mut handles = Vec::new();
614        for _ in 0..2 {
615            let tracker = tracker.clone();
616            let barrier = barrier.clone();
617            handles.push(thread::spawn(move || {
618                barrier.wait();
619                let reservation = tracker.try_reserve(60);
620                barrier.wait();
621                reservation
622            }));
623        }
624        barrier.wait();
625        barrier.wait();
626        let reservations =
627            handles.into_iter().map(|handle| handle.join().unwrap()).collect::<Vec<_>>();
628        assert_eq!(reservations.iter().filter(|result| result.is_ok()).count(), 1);
629        assert!(tracker.snapshot().peak_bytes <= 100);
630    }
631
632    #[test]
633    fn cancellation_is_clone_visible() {
634        let token = CancellationToken::default();
635        let worker = token.clone();
636        token.cancel();
637        assert!(worker.is_cancelled());
638        assert!(matches!(worker.check(), Err(RecordsError::Cancelled)));
639    }
640
641    #[test]
642    fn options_reject_zero_chunk_size() {
643        let error = StreamOptions::new(0, MemoryBudget::default()).unwrap_err();
644        assert!(matches!(error, RecordsError::InvalidConfiguration(_)));
645    }
646
647    #[test]
648    fn receipt_accounts_named_work_without_hidden_transfers() {
649        let tracker = MemoryTracker::new(MemoryBudget::new(1024).unwrap());
650        let reservation = tracker.try_reserve(512).unwrap();
651        let mut receipt = StreamingReceipt::new("synthetic://one-chunk").unwrap();
652        receipt.record_input_chunk(10, 120).unwrap();
653        receipt.record_output_chunk(4, 48).unwrap();
654        receipt.record_phase("transform", 99, 512).unwrap();
655        receipt
656            .record_transfer("explicit-upload", StreamingTransferDirection::HostToDevice, 120)
657            .unwrap();
658        receipt.capture_memory(&tracker);
659        drop(reservation);
660        receipt.validate().unwrap();
661        assert_eq!(receipt.input_points(), 10);
662        assert_eq!(receipt.output_points(), 4);
663        assert_eq!(receipt.peak_tracked_bytes(), 512);
664        assert_eq!(receipt.transfers().len(), 1);
665    }
666
667    #[test]
668    fn receipt_counter_overflow_is_denied_without_wrapping() {
669        let mut receipt = StreamingReceipt::new("synthetic://overflow").unwrap();
670        receipt.input_points = u64::MAX;
671        assert!(matches!(
672            receipt.record_input_chunk(1, 0),
673            Err(RecordsError::ReceiptOverflow(name)) if name == "input_points"
674        ));
675        assert_eq!(receipt.input_points(), u64::MAX);
676    }
677
678    #[cfg(feature = "receipt-json")]
679    #[test]
680    fn json_receipt_roundtrip_is_versioned() {
681        let mut receipt = StreamingReceipt::new("copc://autzen").unwrap();
682        receipt.record_input_chunk(32, 384).unwrap();
683        let json = receipt.to_json().unwrap();
684        let value: serde_json::Value = serde_json::from_str(&json).unwrap();
685        assert_eq!(value["schema"], STREAMING_RECEIPT_SCHEMA);
686        assert_eq!(value["version"], 1);
687        assert_eq!(StreamingReceipt::from_json(&json).unwrap(), receipt);
688    }
689
690    #[cfg(feature = "receipt-json")]
691    #[test]
692    fn json_receipt_rejects_unknown_fields_and_versions() {
693        let receipt = StreamingReceipt::new("copc://autzen").unwrap();
694        let mut value = serde_json::to_value(&receipt).unwrap();
695        value["unexpected"] = serde_json::json!(true);
696        assert!(StreamingReceipt::from_json(&value.to_string()).is_err());
697
698        let mut value = serde_json::to_value(&receipt).unwrap();
699        value["version"] = serde_json::json!(2);
700        assert!(StreamingReceipt::from_json(&value.to_string()).is_err());
701    }
702
703    #[test]
704    fn canonical_workloads_cover_scale_and_chunk_matrix() {
705        let workloads = canonical_streaming_workloads();
706        assert_eq!(workloads.len(), 9);
707        for points in [1_000_000, 10_000_000, 100_000_000] {
708            let chunks = workloads
709                .iter()
710                .filter(|workload| workload.point_count == points)
711                .map(|workload| workload.chunk_points)
712                .collect::<Vec<_>>();
713            assert_eq!(chunks, [16_384, 65_536, 262_144]);
714        }
715    }
716}
spatialrust_records/streaming.rs

spatialrust_records/
streaming.rs
