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

ursula_stream/
snapshot.rs
