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    /// Bucket-scoped erasure domain owning every byte referenced by `target`.
181    /// Physical packs never cross this boundary.
182    pub bucket_id: String,
183    /// Earliest wall-clock timestamp at which the physical object may be
184    /// reclaimed. Zero preserves the immediate behavior of legacy entries.
185    #[serde(default)]
186    pub not_before_ms: u64,
187    pub target: ColdGcTarget,
188}
189
190#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
191pub enum ColdGcTarget {
192    /// Every cold object owned by a fully removed stream. The whole
193    /// `{stream}/chunks/` prefix can be reclaimed at once.
194    Stream(BucketStreamId),
195    /// Specific cold object paths dropped while the stream lives on (snapshot
196    /// retention compaction).
197    Paths(Vec<String>),
198}
199
200#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
201pub struct HotPayloadSegment {
202    pub start_offset: u64,
203    pub end_offset: u64,
204    pub payload_start: usize,
205    pub payload_end: usize,
206}
207
208#[derive(Debug, Clone, PartialEq, Eq)]
209pub struct ColdFlushCandidate {
210    pub stream_id: BucketStreamId,
211    pub start_offset: u64,
212    pub end_offset: u64,
213    pub payload: Vec<u8>,
214    pub payload_digest: String,
215}
216
217#[derive(Debug, Clone, PartialEq, Eq)]
218pub struct StreamReadColdSegment {
219    pub chunk: ColdChunkRef,
220    pub read_start_offset: u64,
221    pub len: usize,
222}
223
224#[derive(Debug, Clone, PartialEq, Eq)]
225pub struct StreamReadObjectSegment {
226    pub object: ObjectPayloadRef,
227    pub read_start_offset: u64,
228    pub len: usize,
229}
230
231#[derive(Debug, Clone, PartialEq, Eq)]
232pub struct StreamReadColdIndexSegment {
233    pub generation: u64,
234    pub page_id: u64,
235    pub read_start_offset: u64,
236    pub len: usize,
237}
238
239#[derive(Debug, Clone, PartialEq, Eq)]
240pub enum StreamReadSegment {
241    ColdIndex(StreamReadColdIndexSegment),
242    Object(StreamReadObjectSegment),
243    Hot(Vec<u8>),
244}
245
246#[derive(Debug, Clone, PartialEq, Eq)]
247pub struct StreamReadPlan {
248    pub offset: u64,
249    pub next_offset: u64,
250    pub content_type: String,
251    pub segments: Vec<StreamReadSegment>,
252    pub up_to_date: bool,
253    pub closed: bool,
254    pub retained_record_range: Option<crate::StreamRecordRange>,
255    pub record_range: Option<crate::StreamRecordRange>,
256}
257
258#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
259pub struct StreamMessageRecord {
260    pub start_offset: u64,
261    pub end_offset: u64,
262}
263
264#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
265pub struct StreamVisibleSnapshot {
266    pub offset: u64,
267    pub content_type: String,
268    pub payload: Vec<u8>,
269    /// BLAKE3 digest over the content type and payload. Empty only when
270    /// decoding legacy snapshots; restore recomputes it.
271    #[serde(default)]
272    pub digest: String,
273}
274
275#[derive(Debug, Clone, PartialEq, Eq)]
276pub struct StreamBootstrapPlan {
277    pub snapshot: Option<StreamVisibleSnapshot>,
278    pub updates: Vec<StreamMessageRecord>,
279    pub next_offset: u64,
280    pub content_type: String,
281    pub up_to_date: bool,
282    pub closed: bool,
283}
284
285/// Per-bucket committed usage inside one Raft group's replicated state.
286///
287/// `committed_append_bytes`, `committed_records`, and
288/// `committed_write_units` are monotonic: they count
289/// committed (non-deduplicated) appends and survive restarts through the
290/// snapshot. `retained_bytes` and `stream_count` are gauges derived from live
291/// stream state and are recomputed from the restored slots, so drift cannot
292/// accumulate across snapshot cycles. Deleting or purging a bucket zeros the
293/// gauges but retains the monotonic counters: otherwise committed writes can
294/// disappear before an asynchronous accounting reader observes them. A bucket
295/// recreated under the same ID continues the counters. A bucket-wide total is
296/// the sum of this value across every Raft group.
297#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
298pub struct BucketUsage {
299    pub committed_append_bytes: u64,
300    pub committed_records: u64,
301    /// Committed write operations measured in the unit declared by the usage
302    /// API, rounded up per operation and never below one. The API describes
303    /// the unit separately so this state is a usage fact rather than a price
304    /// name. Kept in replicated state because a gateway cannot distinguish
305    /// "commit succeeded, response was lost" from "the write never committed".
306    #[serde(default, alias = "committed_write_units_10kib")]
307    pub committed_write_units: u64,
308    pub retained_bytes: u64,
309    pub stream_count: u64,
310}
311
312/// One bucket's usage as reported by a group or persisted in a snapshot.
313#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
314pub struct BucketUsageSnapshot {
315    pub bucket_id: String,
316    pub usage: BucketUsage,
317}
318
319/// Per-bucket data-plane quota stored in replicated state. `None` means
320/// unlimited. Every Raft group stores the same record and enforces it
321/// against its own local [`BucketUsage`], so the cluster-wide bound is
322/// `limit × group_count`: an abuse backstop, not exact tenant accounting.
323/// Exact tenant-level enforcement happens at the gateway from aggregated
324/// usage.
325#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
326pub struct BucketQuota {
327    pub max_streams: Option<u64>,
328    pub max_retained_bytes: Option<u64>,
329}
330
331impl BucketQuota {
332    /// A quota with no limits carries no information; setting it clears the
333    /// stored record.
334    pub fn is_unlimited(&self) -> bool {
335        self.max_streams.is_none() && self.max_retained_bytes.is_none()
336    }
337}
338
339/// One bucket's quota as stored in a group or persisted in a snapshot.
340#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
341pub struct BucketQuotaSnapshot {
342    pub bucket_id: String,
343    pub quota: BucketQuota,
344}
ursula_stream/model.rs

ursula_stream/
model.rs
