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`, `committed_records`, and
285/// `committed_write_units` are monotonic: they count
286/// committed (non-deduplicated) appends and survive restarts through the
287/// snapshot. `retained_bytes` and `stream_count` are gauges derived from live
288/// stream state and are recomputed from the restored slots, so drift cannot
289/// accumulate across snapshot cycles. Deleting or purging a bucket zeros the
290/// gauges but retains the monotonic counters: otherwise committed writes can
291/// disappear before an asynchronous accounting reader observes them. A bucket
292/// recreated under the same ID continues the counters. A bucket-wide total is
293/// the sum of this value across every Raft group.
294#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
295pub struct BucketUsage {
296    pub committed_append_bytes: u64,
297    pub committed_records: u64,
298    /// Committed write operations measured in the unit declared by the usage
299    /// API, rounded up per operation and never below one. The API describes
300    /// the unit separately so this state is a usage fact rather than a price
301    /// name. Kept in replicated state because a gateway cannot distinguish
302    /// "commit succeeded, response was lost" from "the write never committed".
303    #[serde(default, alias = "committed_write_units_10kib")]
304    pub committed_write_units: u64,
305    pub retained_bytes: u64,
306    pub stream_count: u64,
307}
308
309/// One bucket's usage as reported by a group or persisted in a snapshot.
310#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
311pub struct BucketUsageSnapshot {
312    pub bucket_id: String,
313    pub usage: BucketUsage,
314}
315
316/// Per-bucket data-plane quota stored in replicated state. `None` means
317/// unlimited. Every Raft group stores the same record and enforces it
318/// against its own local [`BucketUsage`], so the cluster-wide bound is
319/// `limit × group_count`: an abuse backstop, not exact tenant accounting.
320/// Exact tenant-level enforcement happens at the gateway from aggregated
321/// usage.
322#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
323pub struct BucketQuota {
324    pub max_streams: Option<u64>,
325    pub max_retained_bytes: Option<u64>,
326}
327
328impl BucketQuota {
329    /// A quota with no limits carries no information; setting it clears the
330    /// stored record.
331    pub fn is_unlimited(&self) -> bool {
332        self.max_streams.is_none() && self.max_retained_bytes.is_none()
333    }
334}
335
336/// One bucket's quota as stored in a group or persisted in a snapshot.
337#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
338pub struct BucketQuotaSnapshot {
339    pub bucket_id: String,
340    pub quota: BucketQuota,
341}
ursula_stream/model.rs

ursula_stream/
model.rs
