ursula_stream/
model.rs

1use serde::Deserialize;
2use serde::Serialize;
3use serde_json::Map;
4use serde_json::Value;
5use ursula_proto::ColdChunkRefV1;
6use ursula_proto::ExternalPayloadRefV1;
7use ursula_proto::ProducerRequestV1;
8use ursula_shard::BucketStreamId;
9
10pub const COLD_INDEX_PAGE_SPAN_BYTES: u64 = 64 * 1024 * 1024;
11
12#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
13pub enum StreamStatus {
14    Open,
15    Closed,
16}
17
18#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
19pub struct StreamMetadata {
20    pub stream_id: BucketStreamId,
21    pub content_type: String,
22    pub status: StreamStatus,
23    pub tail_offset: u64,
24    pub last_stream_seq: Option<String>,
25    pub stream_ttl_seconds: Option<u64>,
26    pub stream_expires_at_ms: Option<u64>,
27    pub created_at_ms: u64,
28    pub last_ttl_touch_at_ms: u64,
29}
30
31/// Maximum encoded JSON size of a stream attribute object. Attrs travel in
32/// every raft log entry, WAL record, and snapshot entry that carries them, so
33/// the limit keeps replicated state small.
34pub const MAX_STREAM_ATTRS_BYTES: usize = 16 * 1024;
35
36#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
37pub struct StreamAttrs {
38    #[serde(default, skip_serializing_if = "Option::is_none")]
39    pub title: Option<String>,
40    #[serde(default, skip_serializing_if = "Map::is_empty")]
41    pub metadata: Map<String, Value>,
42}
43
44impl StreamAttrs {
45    pub fn is_empty(&self) -> bool {
46        self.title.is_none() && self.metadata.is_empty()
47    }
48}
49
50pub type ProducerRequest = ProducerRequestV1;
51
52#[derive(Debug)]
53pub struct AppendStreamInput<'a> {
54    pub stream_id: BucketStreamId,
55    pub content_type: Option<&'a str>,
56    pub payload: &'a [u8],
57    pub close_after: bool,
58    pub stream_seq: Option<String>,
59    pub producer: Option<ProducerRequest>,
60    pub now_ms: u64,
61    pub record_match: Option<u64>,
62}
63
64#[derive(Debug)]
65pub(crate) struct AppendExternalInput<'a> {
66    pub(crate) stream_id: BucketStreamId,
67    pub(crate) content_type: Option<&'a str>,
68    pub(crate) payload: ExternalPayloadRef,
69    pub(crate) record_ends: Vec<u64>,
70    pub(crate) close_after: bool,
71    pub(crate) stream_seq: Option<String>,
72    pub(crate) producer: Option<ProducerRequest>,
73    pub(crate) now_ms: u64,
74    pub(crate) record_match: Option<u64>,
75}
76
77#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
78pub struct ProducerSnapshot {
79    pub producer_id: String,
80    pub producer_epoch: u64,
81    pub producer_seq: u64,
82    pub last_start_offset: u64,
83    pub last_next_offset: u64,
84    pub last_closed: bool,
85    pub last_items: Vec<ProducerAppendRecord>,
86}
87
88#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
89pub struct ProducerAppendRecord {
90    pub start_offset: u64,
91    pub next_offset: u64,
92    pub closed: bool,
93    #[serde(default)]
94    pub record_start: Option<u64>,
95    #[serde(default)]
96    pub record_next: Option<u64>,
97}
98
99#[derive(Debug, Clone, PartialEq, Eq)]
100pub(crate) struct ProducerState {
101    pub(crate) producer_epoch: u64,
102    pub(crate) producer_seq: u64,
103    pub(crate) last_start_offset: u64,
104    pub(crate) last_next_offset: u64,
105    pub(crate) last_closed: bool,
106    pub(crate) last_items: Vec<ProducerAppendRecord>,
107}
108
109#[derive(Debug, Clone, PartialEq, Eq)]
110pub struct StreamBatchAppend {
111    pub items: Vec<StreamBatchAppendItem>,
112    pub deduplicated: bool,
113}
114
115#[derive(Debug, Clone, PartialEq, Eq)]
116pub struct StreamBatchAppendItem {
117    pub offset: u64,
118    pub next_offset: u64,
119    pub closed: bool,
120    pub deduplicated: bool,
121}
122
123#[derive(Debug, Clone, PartialEq, Eq)]
124pub struct StreamRead {
125    pub offset: u64,
126    pub next_offset: u64,
127    pub content_type: String,
128    pub payload: Vec<u8>,
129    pub up_to_date: bool,
130    pub closed: bool,
131}
132
133pub type ColdChunkRef = ColdChunkRefV1;
134
135#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
136pub struct ObjectPayloadRef {
137    pub start_offset: u64,
138    pub end_offset: u64,
139    pub s3_path: String,
140    pub object_size: u64,
141}
142
143impl From<&ColdChunkRef> for ObjectPayloadRef {
144    fn from(chunk: &ColdChunkRef) -> Self {
145        Self {
146            start_offset: chunk.start_offset,
147            end_offset: chunk.end_offset,
148            s3_path: chunk.s3_path.clone(),
149            object_size: chunk.object_size,
150        }
151    }
152}
153
154pub type ExternalPayloadRef = ExternalPayloadRefV1;
155
156/// One unit of deferred cold-storage reclamation. Enqueued deterministically in
157/// the state machine when a stream's cold objects become unreferenced, drained
158/// asynchronously by the leader's background GC worker.
159#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
160pub struct ColdGcEntry {
161    pub seq: u64,
162    pub target: ColdGcTarget,
163}
164
165#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
166pub enum ColdGcTarget {
167    /// Every cold object owned by a fully removed stream. The whole
168    /// `{stream}/chunks/` prefix can be reclaimed at once.
169    Stream(BucketStreamId),
170    /// Specific cold object paths dropped while the stream lives on (snapshot
171    /// retention compaction).
172    Paths(Vec<String>),
173}
174
175#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
176pub struct HotPayloadSegment {
177    pub start_offset: u64,
178    pub end_offset: u64,
179    pub payload_start: usize,
180    pub payload_end: usize,
181}
182
183#[derive(Debug, Clone, PartialEq, Eq)]
184pub struct ColdFlushCandidate {
185    pub stream_id: BucketStreamId,
186    pub start_offset: u64,
187    pub end_offset: u64,
188    pub payload: Vec<u8>,
189}
190
191#[derive(Debug, Clone, PartialEq, Eq)]
192pub struct StreamReadColdSegment {
193    pub chunk: ColdChunkRef,
194    pub read_start_offset: u64,
195    pub len: usize,
196}
197
198#[derive(Debug, Clone, PartialEq, Eq)]
199pub struct StreamReadObjectSegment {
200    pub object: ObjectPayloadRef,
201    pub read_start_offset: u64,
202    pub len: usize,
203}
204
205#[derive(Debug, Clone, PartialEq, Eq)]
206pub struct StreamReadColdIndexSegment {
207    pub generation: u64,
208    pub page_id: u64,
209    pub read_start_offset: u64,
210    pub len: usize,
211}
212
213#[derive(Debug, Clone, PartialEq, Eq)]
214pub enum StreamReadSegment {
215    ColdIndex(StreamReadColdIndexSegment),
216    Object(StreamReadObjectSegment),
217    Hot(Vec<u8>),
218}
219
220#[derive(Debug, Clone, PartialEq, Eq)]
221pub struct StreamReadPlan {
222    pub offset: u64,
223    pub next_offset: u64,
224    pub content_type: String,
225    pub segments: Vec<StreamReadSegment>,
226    pub up_to_date: bool,
227    pub closed: bool,
228    pub retained_record_range: Option<crate::StreamRecordRange>,
229    pub record_range: Option<crate::StreamRecordRange>,
230}
231
232#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
233pub struct StreamMessageRecord {
234    pub start_offset: u64,
235    pub end_offset: u64,
236}
237
238#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
239pub struct StreamVisibleSnapshot {
240    pub offset: u64,
241    pub content_type: String,
242    pub payload: Vec<u8>,
243}
244
245#[derive(Debug, Clone, PartialEq, Eq)]
246pub struct StreamBootstrapPlan {
247    pub snapshot: Option<StreamVisibleSnapshot>,
248    pub updates: Vec<StreamMessageRecord>,
249    pub next_offset: u64,
250    pub content_type: String,
251    pub up_to_date: bool,
252    pub closed: bool,
253}
ursula_stream/model.rs

ursula_stream/
model.rs
