ursula_stream/
model.rs

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

ursula_stream/
model.rs
