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

ursula_stream/
model.rs
