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    /// Bounded exact response history for delayed retries. Missing in legacy
87    /// snapshots, which are restored with the last response as the sole
88    /// receipt.
89    #[serde(default)]
90    pub receipts: Vec<ProducerReceipt>,
91}
92
93#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
94pub struct ProducerReceipt {
95    pub producer_seq: u64,
96    pub start_offset: u64,
97    pub next_offset: u64,
98    pub closed: bool,
99    pub items: Vec<ProducerAppendRecord>,
100}
101
102#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
103pub struct ProducerAppendRecord {
104    pub start_offset: u64,
105    pub next_offset: u64,
106    pub closed: bool,
107    #[serde(default)]
108    pub record_start: Option<u64>,
109    #[serde(default)]
110    pub record_next: Option<u64>,
111}
112
113#[derive(Debug, Clone, PartialEq, Eq)]
114pub(crate) struct ProducerState {
115    pub(crate) producer_epoch: u64,
116    pub(crate) producer_seq: u64,
117    pub(crate) last_start_offset: u64,
118    pub(crate) last_next_offset: u64,
119    pub(crate) last_closed: bool,
120    pub(crate) last_items: Vec<ProducerAppendRecord>,
121    pub(crate) receipts: Vec<ProducerReceipt>,
122}
123
124#[derive(Debug, Clone, PartialEq, Eq)]
125pub struct StreamBatchAppend {
126    pub items: Vec<StreamBatchAppendItem>,
127    pub deduplicated: bool,
128}
129
130#[derive(Debug, Clone, PartialEq, Eq)]
131pub struct StreamBatchAppendItem {
132    pub offset: u64,
133    pub next_offset: u64,
134    pub closed: bool,
135    pub deduplicated: bool,
136}
137
138#[derive(Debug, Clone, PartialEq, Eq)]
139pub struct StreamRead {
140    pub offset: u64,
141    pub next_offset: u64,
142    pub content_type: String,
143    pub payload: Vec<u8>,
144    pub up_to_date: bool,
145    pub closed: bool,
146}
147
148pub type ColdChunkRef = ColdChunkRefV1;
149
150#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
151pub struct ObjectPayloadRef {
152    pub start_offset: u64,
153    pub end_offset: u64,
154    pub s3_path: String,
155    pub object_size: u64,
156    #[serde(default)]
157    pub object_offset: u64,
158}
159
160impl From<&ColdChunkRef> for ObjectPayloadRef {
161    fn from(chunk: &ColdChunkRef) -> Self {
162        Self {
163            start_offset: chunk.start_offset,
164            end_offset: chunk.end_offset,
165            s3_path: chunk.s3_path.clone(),
166            object_size: chunk.object_size,
167            object_offset: chunk.object_offset,
168        }
169    }
170}
171
172pub type ExternalPayloadRef = ExternalPayloadRefV1;
173
174/// One unit of deferred cold-storage reclamation. Enqueued deterministically in
175/// the state machine when a stream's cold objects become unreferenced, drained
176/// asynchronously by the leader's background GC worker.
177#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
178pub struct ColdGcEntry {
179    pub seq: u64,
180    /// Earliest wall-clock timestamp at which the physical object may be
181    /// reclaimed. Zero preserves the immediate behavior of legacy entries.
182    #[serde(default)]
183    pub not_before_ms: u64,
184    pub target: ColdGcTarget,
185}
186
187#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
188pub enum ColdGcTarget {
189    /// Every cold object owned by a fully removed stream. The whole
190    /// `{stream}/chunks/` prefix can be reclaimed at once.
191    Stream(BucketStreamId),
192    /// Specific cold object paths dropped while the stream lives on (snapshot
193    /// retention compaction).
194    Paths(Vec<String>),
195}
196
197#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
198pub struct HotPayloadSegment {
199    pub start_offset: u64,
200    pub end_offset: u64,
201    pub payload_start: usize,
202    pub payload_end: usize,
203}
204
205#[derive(Debug, Clone, PartialEq, Eq)]
206pub struct ColdFlushCandidate {
207    pub stream_id: BucketStreamId,
208    pub start_offset: u64,
209    pub end_offset: u64,
210    pub payload: Vec<u8>,
211    pub payload_digest: String,
212}
213
214#[derive(Debug, Clone, PartialEq, Eq)]
215pub struct StreamReadColdSegment {
216    pub chunk: ColdChunkRef,
217    pub read_start_offset: u64,
218    pub len: usize,
219}
220
221#[derive(Debug, Clone, PartialEq, Eq)]
222pub struct StreamReadObjectSegment {
223    pub object: ObjectPayloadRef,
224    pub read_start_offset: u64,
225    pub len: usize,
226}
227
228#[derive(Debug, Clone, PartialEq, Eq)]
229pub struct StreamReadColdIndexSegment {
230    pub generation: u64,
231    pub page_id: u64,
232    pub read_start_offset: u64,
233    pub len: usize,
234}
235
236#[derive(Debug, Clone, PartialEq, Eq)]
237pub enum StreamReadSegment {
238    ColdIndex(StreamReadColdIndexSegment),
239    Object(StreamReadObjectSegment),
240    Hot(Vec<u8>),
241}
242
243#[derive(Debug, Clone, PartialEq, Eq)]
244pub struct StreamReadPlan {
245    pub offset: u64,
246    pub next_offset: u64,
247    pub content_type: String,
248    pub segments: Vec<StreamReadSegment>,
249    pub up_to_date: bool,
250    pub closed: bool,
251    pub retained_record_range: Option<crate::StreamRecordRange>,
252    pub record_range: Option<crate::StreamRecordRange>,
253}
254
255#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
256pub struct StreamMessageRecord {
257    pub start_offset: u64,
258    pub end_offset: u64,
259}
260
261#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
262pub struct StreamVisibleSnapshot {
263    pub offset: u64,
264    pub content_type: String,
265    pub payload: Vec<u8>,
266    /// BLAKE3 digest over the content type and payload. Empty only when
267    /// decoding legacy snapshots; restore recomputes it.
268    #[serde(default)]
269    pub digest: String,
270}
271
272#[derive(Debug, Clone, PartialEq, Eq)]
273pub struct StreamBootstrapPlan {
274    pub snapshot: Option<StreamVisibleSnapshot>,
275    pub updates: Vec<StreamMessageRecord>,
276    pub next_offset: u64,
277    pub content_type: String,
278    pub up_to_date: bool,
279    pub closed: bool,
280}
281
282/// Per-bucket committed usage inside one Raft group's replicated state.
283///
284/// `committed_append_bytes` and `committed_records` are monotonic: they count
285/// accepted (non-deduplicated) appends and survive restarts through the
286/// snapshot. `retained_bytes` and `stream_count` are gauges derived from live
287/// stream state and are recomputed from the restored slots, so drift cannot
288/// accumulate across snapshot cycles. A bucket-wide total is the sum of this
289/// value across every Raft group.
290#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
291pub struct BucketUsage {
292    pub committed_append_bytes: u64,
293    pub committed_records: u64,
294    pub retained_bytes: u64,
295    pub stream_count: u64,
296}
297
298/// One bucket's usage as reported by a group or persisted in a snapshot.
299#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
300pub struct BucketUsageSnapshot {
301    pub bucket_id: String,
302    pub usage: BucketUsage,
303}
304
305/// Per-bucket data-plane quota stored in replicated state. `None` means
306/// unlimited. Every Raft group stores the same record and enforces it
307/// against its own local [`BucketUsage`], so the cluster-wide bound is
308/// `limit × group_count`: an abuse backstop, not exact tenant accounting.
309/// Exact tenant-level enforcement happens at the gateway from aggregated
310/// usage.
311#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
312pub struct BucketQuota {
313    pub max_streams: Option<u64>,
314    pub max_retained_bytes: Option<u64>,
315}
316
317impl BucketQuota {
318    /// A quota with no limits carries no information; setting it clears the
319    /// stored record.
320    pub fn is_unlimited(&self) -> bool {
321        self.max_streams.is_none() && self.max_retained_bytes.is_none()
322    }
323}
324
325/// One bucket's quota as stored in a group or persisted in a snapshot.
326#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
327pub struct BucketQuotaSnapshot {
328    pub bucket_id: String,
329    pub quota: BucketQuota,
330}
ursula_stream/model.rs

ursula_stream/
model.rs
