ursula_stream/
snapshot.rs

1use serde::Deserialize;
2use serde::Serialize;
3use ursula_shard::BucketStreamId;
4
5use crate::integrity::StreamIntegritySnapshot;
6use crate::model::ColdChunkRef;
7use crate::model::ColdGcEntry;
8use crate::model::HotPayloadSegment;
9use crate::model::ObjectPayloadRef;
10use crate::model::ProducerSnapshot;
11use crate::model::StreamAttrs;
12use crate::model::StreamMessageRecord;
13use crate::model::StreamMetadata;
14use crate::model::StreamVisibleSnapshot;
15
16#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
17pub struct StreamSnapshot {
18    pub buckets: Vec<String>,
19    pub streams: Vec<StreamSnapshotEntry>,
20    #[serde(default)]
21    pub pending_cold_gc: Vec<ColdGcEntry>,
22    #[serde(default)]
23    pub next_cold_gc_seq: u64,
24}
25
26#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
27pub struct StreamSnapshotEntry {
28    pub metadata: StreamMetadata,
29    #[serde(default)]
30    pub attrs: Option<StreamAttrs>,
31    pub hot_start_offset: u64,
32    pub payload: Vec<u8>,
33    pub hot_segments: Vec<HotPayloadSegment>,
34    #[serde(default)]
35    pub cold_frontier_offset: u64,
36    #[serde(default)]
37    pub cold_index_generation: u64,
38    pub cold_chunks: Vec<ColdChunkRef>,
39    pub external_segments: Vec<ObjectPayloadRef>,
40    pub message_records: Vec<StreamMessageRecord>,
41    pub integrity: StreamIntegritySnapshot,
42    pub visible_snapshot: Option<StreamVisibleSnapshot>,
43    pub producer_states: Vec<ProducerSnapshot>,
44}
45
46#[derive(Debug, Clone, PartialEq, Eq)]
47pub enum StreamSnapshotError {
48    DuplicateBucket(String),
49    DuplicateStream(BucketStreamId),
50    DuplicateProducer {
51        stream_id: BucketStreamId,
52        producer_id: String,
53    },
54    MissingBucket(BucketStreamId),
55    PayloadLengthMismatch {
56        stream_id: BucketStreamId,
57        tail_offset: u64,
58        payload_len: usize,
59    },
60    MessageBoundaryMismatch {
61        stream_id: BucketStreamId,
62    },
63    IntegrityMismatch {
64        stream_id: BucketStreamId,
65    },
66    SnapshotOffsetOutOfRange {
67        stream_id: BucketStreamId,
68        snapshot_offset: u64,
69        tail_offset: u64,
70    },
71}
72
73impl std::fmt::Display for StreamSnapshotError {
74    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
75        match self {
76            Self::DuplicateBucket(bucket_id) => {
77                write!(f, "snapshot contains duplicate bucket '{bucket_id}'")
78            }
79            Self::DuplicateStream(stream_id) => {
80                write!(f, "snapshot contains duplicate stream '{stream_id}'")
81            }
82            Self::DuplicateProducer {
83                stream_id,
84                producer_id,
85            } => write!(
86                f,
87                "snapshot stream '{stream_id}' contains duplicate producer '{producer_id}'"
88            ),
89            Self::MissingBucket(stream_id) => {
90                write!(
91                    f,
92                    "snapshot stream '{stream_id}' references a missing bucket"
93                )
94            }
95            Self::PayloadLengthMismatch {
96                stream_id,
97                tail_offset,
98                payload_len,
99            } => write!(
100                f,
101                "snapshot stream '{stream_id}' tail offset {tail_offset} does not match payload length {payload_len}"
102            ),
103            Self::MessageBoundaryMismatch { stream_id } => write!(
104                f,
105                "snapshot stream '{stream_id}' has inconsistent message boundaries"
106            ),
107            Self::IntegrityMismatch { stream_id } => write!(
108                f,
109                "snapshot stream '{stream_id}' has inconsistent integrity setsums"
110            ),
111            Self::SnapshotOffsetOutOfRange {
112                stream_id,
113                snapshot_offset,
114                tail_offset,
115            } => write!(
116                f,
117                "snapshot stream '{stream_id}' visible snapshot offset {snapshot_offset} is beyond tail offset {tail_offset}"
118            ),
119        }
120    }
121}
122
123impl std::error::Error for StreamSnapshotError {}
ursula_stream/snapshot.rs

ursula_stream/
snapshot.rs
