ursula_stream/
snapshot.rs1use 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 {}