nstreams_core/
version.rs

1use crate::event::{EventVersion, StreamEvent};
2
3/// Tracks the last delivered version for gapless client-side verification.
4#[derive(Debug, Clone, Copy, PartialEq, Eq)]
5pub struct VersionedDelivery {
6    pub version: EventVersion,
7}
8
9impl VersionedDelivery {
10    pub fn first(version: EventVersion) -> Self {
11        Self { version }
12    }
13
14    /// Verify that `event` continues the monotonic version sequence.
15    pub fn verify_next(&self, event: &StreamEvent) -> crate::Result<EventVersion> {
16        let expected = self.version.saturating_add(1);
17        if event.version != expected {
18            return Err(crate::Error::VersionGap {
19                expected,
20                actual: event.version,
21            });
22        }
23        Ok(event.version)
24    }
25
26    /// Verify the first event in a replay sequence.
27    pub fn verify_initial(
28        last_delivered: Option<EventVersion>,
29        event: &StreamEvent,
30    ) -> crate::Result<EventVersion> {
31        match last_delivered {
32            Some(last) => VersionedDelivery { version: last }.verify_next(event),
33            None => Ok(event.version),
34        }
35    }
36}
37
38#[cfg(test)]
39mod tests {
40    use super::*;
41
42    #[test]
43    fn detects_version_gaps() {
44        let delivery = VersionedDelivery::first(10);
45        let event = StreamEvent {
46            namespace: "test".into(),
47            version: 12,
48            payload: vec![],
49            filter_value: None,
50        };
51        assert!(matches!(
52            delivery.verify_next(&event),
53            Err(crate::Error::VersionGap {
54                expected: 11,
55                actual: 12
56            })
57        ));
58    }
59}
nstreams_core/version.rs

nstreams_core/
version.rs
