ursula_runtime/
request.rs

1use std::sync::Arc;
2
3use bytes::Bytes;
4use serde::Deserialize;
5use serde::Serialize;
6use ursula_shard::BucketStreamId;
7use ursula_shard::ShardPlacement;
8use ursula_stream::ColdChunkRef;
9use ursula_stream::ExternalPayloadRef;
10use ursula_stream::ProducerRequest;
11use ursula_stream::StreamAttrs;
12use ursula_stream::StreamIntegritySnapshot;
13use ursula_stream::StreamReadPlan;
14use ursula_stream::StreamReadSegment;
15use ursula_stream::StreamRecordRange;
16
17use crate::cold_index::ColdIndexPageCache;
18use crate::cold_index::ColdStoreColdIndexPageStore;
19use crate::cold_store::ColdStoreHandle;
20use crate::cold_store::DEFAULT_CONTENT_TYPE;
21use crate::engine::GroupEngineError;
22use crate::engine::in_memory::InMemoryGroupEngine;
23use crate::error::RuntimeError;
24
25#[derive(Debug, Clone, PartialEq, Eq)]
26pub struct CreateStreamRequest {
27    pub stream_id: BucketStreamId,
28    pub content_type: String,
29    pub content_type_explicit: bool,
30    pub initial_payload: Bytes,
31    pub close_after: bool,
32    pub stream_seq: Option<String>,
33    pub producer: Option<ProducerRequest>,
34    pub stream_ttl_seconds: Option<u64>,
35    pub stream_expires_at_ms: Option<u64>,
36    pub attrs: Option<StreamAttrs>,
37    pub now_ms: u64,
38}
39
40#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
41pub struct CreateStreamExternalRequest {
42    pub stream_id: BucketStreamId,
43    pub content_type: String,
44    pub initial_payload: ExternalPayloadRef,
45    #[serde(default)]
46    pub record_ends: Vec<u64>,
47    pub close_after: bool,
48    pub stream_seq: Option<String>,
49    pub producer: Option<ProducerRequest>,
50    pub stream_ttl_seconds: Option<u64>,
51    pub stream_expires_at_ms: Option<u64>,
52    pub attrs: Option<StreamAttrs>,
53    pub now_ms: u64,
54}
55
56impl CreateStreamExternalRequest {
57    pub fn from_create_request(
58        request: CreateStreamRequest,
59        initial_payload: ExternalPayloadRef,
60        record_ends: Vec<u64>,
61    ) -> Self {
62        Self {
63            stream_id: request.stream_id,
64            content_type: request.content_type,
65            initial_payload,
66            record_ends,
67            close_after: request.close_after,
68            stream_seq: request.stream_seq,
69            producer: request.producer,
70            stream_ttl_seconds: request.stream_ttl_seconds,
71            stream_expires_at_ms: request.stream_expires_at_ms,
72            attrs: request.attrs,
73            now_ms: request.now_ms,
74        }
75    }
76}
77
78impl CreateStreamRequest {
79    pub fn canonical_record_ends(&self) -> Vec<u64> {
80        ursula_stream::canonical_json_record_ends(&self.content_type, &self.initial_payload)
81            .unwrap_or_default()
82    }
83
84    pub fn new(stream_id: BucketStreamId, content_type: impl Into<String>) -> Self {
85        Self {
86            stream_id,
87            content_type: content_type.into(),
88            content_type_explicit: true,
89            initial_payload: Bytes::new(),
90            close_after: false,
91            stream_seq: None,
92            producer: None,
93            stream_ttl_seconds: None,
94            stream_expires_at_ms: None,
95            attrs: None,
96            now_ms: 0,
97        }
98    }
99}
100
101#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
102pub struct CreateStreamResponse {
103    pub placement: ShardPlacement,
104    pub next_offset: u64,
105    pub closed: bool,
106    pub already_exists: bool,
107    pub group_commit_index: u64,
108    pub record_range: Option<StreamRecordRange>,
109}
110
111#[derive(Debug, Clone, PartialEq, Eq)]
112pub struct HeadStreamRequest {
113    pub stream_id: BucketStreamId,
114    pub now_ms: u64,
115}
116
117#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
118pub struct HeadStreamResponse {
119    pub placement: ShardPlacement,
120    pub content_type: String,
121    pub tail_offset: u64,
122    pub cold_hot_start_offset: u64,
123    pub closed: bool,
124    pub stream_ttl_seconds: Option<u64>,
125    pub stream_expires_at_ms: Option<u64>,
126    pub snapshot_offset: Option<u64>,
127    pub snapshot_digest: Option<String>,
128    pub retained_offset: u64,
129    pub integrity: StreamIntegritySnapshot,
130    pub record_range: Option<StreamRecordRange>,
131}
132
133#[derive(Debug, Clone, PartialEq, Eq)]
134pub struct GetStreamAttrsRequest {
135    pub stream_id: BucketStreamId,
136    pub now_ms: u64,
137}
138
139#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
140pub struct GetStreamAttrsResponse {
141    pub placement: ShardPlacement,
142    pub attrs: Option<StreamAttrs>,
143}
144
145#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
146pub struct UpdateStreamAttrsRequest {
147    pub stream_id: BucketStreamId,
148    pub attrs: Option<StreamAttrs>,
149    pub now_ms: u64,
150}
151
152#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
153pub struct UpdateStreamAttrsResponse {
154    pub placement: ShardPlacement,
155    pub changed: bool,
156    pub group_commit_index: u64,
157}
158
159#[derive(Debug, Clone, PartialEq, Eq)]
160pub struct ReadStreamRequest {
161    pub stream_id: BucketStreamId,
162    pub offset: u64,
163    pub max_len: usize,
164    pub now_ms: u64,
165    pub record: Option<u64>,
166    pub max_records: Option<u64>,
167}
168
169impl ReadStreamRequest {
170    pub(crate) fn same_wait_plan(&self, other: &Self) -> bool {
171        self.stream_id == other.stream_id
172            && self.offset == other.offset
173            && self.max_len == other.max_len
174            && self.record == other.record
175            && self.max_records == other.max_records
176    }
177}
178
179#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
180pub struct ReadStreamResponse {
181    pub placement: ShardPlacement,
182    pub offset: u64,
183    pub next_offset: u64,
184    pub content_type: String,
185    #[serde(with = "serde_bytes")]
186    pub payload: Vec<u8>,
187    pub up_to_date: bool,
188    pub closed: bool,
189    pub retained_record_range: Option<StreamRecordRange>,
190    pub record_range: Option<StreamRecordRange>,
191}
192
193pub enum GroupReadStreamBody {
194    Materialized(Vec<u8>),
195    Planned {
196        stream_id: BucketStreamId,
197        plan: StreamReadPlan,
198        cold_store: Option<ColdStoreHandle>,
199        cold_index_cache: Option<Arc<ColdIndexPageCache<ColdStoreColdIndexPageStore>>>,
200    },
201    #[cfg(test)]
202    Blocking {
203        entered: Arc<crate::rt::sync::Notify>,
204        materialized: Arc<crate::rt::sync::Notify>,
205        release: Arc<crate::rt::sync::Notify>,
206        payload: Vec<u8>,
207    },
208}
209
210pub struct GroupReadStreamParts {
211    pub placement: ShardPlacement,
212    pub offset: u64,
213    pub next_offset: u64,
214    pub content_type: String,
215    pub up_to_date: bool,
216    pub closed: bool,
217    pub retained_record_range: Option<StreamRecordRange>,
218    pub record_range: Option<StreamRecordRange>,
219    pub body: GroupReadStreamBody,
220}
221
222impl GroupReadStreamParts {
223    pub fn from_response(response: ReadStreamResponse) -> Self {
224        Self {
225            placement: response.placement,
226            offset: response.offset,
227            next_offset: response.next_offset,
228            content_type: response.content_type,
229            up_to_date: response.up_to_date,
230            closed: response.closed,
231            retained_record_range: response.retained_record_range,
232            record_range: response.record_range,
233            body: GroupReadStreamBody::Materialized(response.payload),
234        }
235    }
236
237    pub fn from_plan(
238        placement: ShardPlacement,
239        stream_id: BucketStreamId,
240        plan: StreamReadPlan,
241        cold_store: Option<ColdStoreHandle>,
242        cold_index_cache: Option<Arc<ColdIndexPageCache<ColdStoreColdIndexPageStore>>>,
243    ) -> Self {
244        Self {
245            placement,
246            offset: plan.offset,
247            next_offset: plan.next_offset,
248            content_type: plan.content_type.clone(),
249            up_to_date: plan.up_to_date,
250            closed: plan.closed,
251            retained_record_range: plan.retained_record_range,
252            record_range: plan.record_range,
253            body: GroupReadStreamBody::Planned {
254                stream_id,
255                plan,
256                cold_store,
257                cold_index_cache,
258            },
259        }
260    }
261
262    pub async fn into_response(self) -> Result<ReadStreamResponse, GroupEngineError> {
263        let payload = match &self.body {
264            GroupReadStreamBody::Materialized(payload) => payload.clone(),
265            GroupReadStreamBody::Planned {
266                stream_id,
267                plan,
268                cold_store,
269                cold_index_cache,
270            } => {
271                InMemoryGroupEngine::read_payload_from_plan(
272                    cold_store.as_ref(),
273                    cold_index_cache.as_ref(),
274                    stream_id,
275                    plan,
276                )
277                .await?
278            }
279            #[cfg(test)]
280            GroupReadStreamBody::Blocking {
281                entered,
282                materialized,
283                release,
284                payload,
285            } => {
286                entered.notify_one();
287                materialized.notify_one();
288                release.notified().await;
289                payload.clone()
290            }
291        };
292        Ok(ReadStreamResponse {
293            placement: self.placement,
294            offset: self.offset,
295            next_offset: self.next_offset,
296            content_type: self.content_type,
297            payload,
298            up_to_date: self.up_to_date,
299            closed: self.closed,
300            retained_record_range: self.retained_record_range,
301            record_range: self.record_range,
302        })
303    }
304
305    pub fn payload_is_empty(&self) -> bool {
306        match &self.body {
307            GroupReadStreamBody::Materialized(payload) => payload.is_empty(),
308            GroupReadStreamBody::Planned { plan, .. } => {
309                plan.segments.iter().all(|segment| match segment {
310                    StreamReadSegment::Hot(payload) => payload.is_empty(),
311                    StreamReadSegment::ColdIndex(segment) => segment.len == 0,
312                    StreamReadSegment::Object(segment) => segment.len == 0,
313                })
314            }
315            #[cfg(test)]
316            GroupReadStreamBody::Blocking { payload, .. } => payload.is_empty(),
317        }
318    }
319}
320
321#[derive(Debug, Clone, PartialEq, Eq)]
322pub struct PublishSnapshotRequest {
323    pub stream_id: BucketStreamId,
324    pub snapshot_offset: u64,
325    pub content_type: String,
326    pub payload: Bytes,
327    pub expected_digest: Option<String>,
328    pub now_ms: u64,
329}
330
331#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
332pub struct PublishSnapshotResponse {
333    pub placement: ShardPlacement,
334    pub snapshot_offset: u64,
335    pub snapshot_digest: String,
336    pub group_commit_index: u64,
337    pub record_range: Option<StreamRecordRange>,
338}
339
340#[derive(Debug, Clone, PartialEq, Eq)]
341pub struct AdvanceRetentionRequest {
342    pub stream_id: BucketStreamId,
343    pub retained_offset: u64,
344    pub now_ms: u64,
345}
346
347#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
348pub struct AdvanceRetentionResponse {
349    pub placement: ShardPlacement,
350    pub retained_offset: u64,
351    pub group_commit_index: u64,
352    pub record_range: Option<StreamRecordRange>,
353}
354
355/// Sets or clears one bucket's data-plane quota record on a group. The
356/// caller replicates the same request to every group.
357#[derive(Debug, Clone, PartialEq, Eq)]
358pub struct SetBucketQuotaRequest {
359    pub bucket_id: String,
360    pub max_streams: Option<u64>,
361    pub max_retained_bytes: Option<u64>,
362}
363
364#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
365pub struct SetBucketQuotaResponse {
366    pub placement: ShardPlacement,
367    pub group_commit_index: u64,
368}
369
370#[derive(Debug, Clone, PartialEq, Eq)]
371pub struct ImportGroupStateRequest {
372    pub snapshot: Box<ursula_stream::StreamSnapshot>,
373}
374
375#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
376pub struct ImportGroupStateResponse {
377    pub placement: ShardPlacement,
378    pub buckets: u64,
379    pub streams: u64,
380    pub group_commit_index: u64,
381}
382
383#[derive(Debug, Clone, PartialEq, Eq)]
384pub struct ReadSnapshotRequest {
385    pub stream_id: BucketStreamId,
386    pub snapshot_offset: Option<u64>,
387    pub now_ms: u64,
388}
389
390#[derive(Debug, Clone, PartialEq, Eq)]
391pub struct ReadSnapshotResponse {
392    pub placement: ShardPlacement,
393    pub snapshot_offset: u64,
394    pub next_offset: u64,
395    pub content_type: String,
396    pub snapshot_digest: String,
397    pub payload: Vec<u8>,
398    pub up_to_date: bool,
399    pub record_range: Option<StreamRecordRange>,
400}
401
402#[derive(Debug, Clone, PartialEq, Eq)]
403pub struct DeleteSnapshotRequest {
404    pub stream_id: BucketStreamId,
405    pub snapshot_offset: u64,
406    pub now_ms: u64,
407}
408
409#[derive(Debug, Clone, PartialEq, Eq)]
410pub struct BootstrapStreamRequest {
411    pub stream_id: BucketStreamId,
412    pub now_ms: u64,
413}
414
415#[derive(Debug, Clone, PartialEq, Eq)]
416pub struct BootstrapUpdate {
417    pub start_offset: u64,
418    pub next_offset: u64,
419    pub content_type: String,
420    pub payload: Vec<u8>,
421}
422
423#[derive(Debug, Clone, PartialEq, Eq)]
424pub struct BootstrapStreamResponse {
425    pub placement: ShardPlacement,
426    pub snapshot_offset: Option<u64>,
427    pub snapshot_content_type: String,
428    pub snapshot_payload: Vec<u8>,
429    pub updates: Vec<BootstrapUpdate>,
430    pub next_offset: u64,
431    pub up_to_date: bool,
432    pub closed: bool,
433    pub record_range: Option<StreamRecordRange>,
434}
435
436#[derive(Debug, Clone, PartialEq, Eq)]
437pub struct CloseStreamRequest {
438    pub stream_id: BucketStreamId,
439    pub stream_seq: Option<String>,
440    pub producer: Option<ProducerRequest>,
441    pub now_ms: u64,
442}
443
444#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
445pub struct CloseStreamResponse {
446    pub placement: ShardPlacement,
447    pub next_offset: u64,
448    pub group_commit_index: u64,
449    pub deduplicated: bool,
450    pub record_range: Option<StreamRecordRange>,
451}
452
453#[derive(Debug, Clone, PartialEq, Eq)]
454pub struct DeleteStreamRequest {
455    pub stream_id: BucketStreamId,
456}
457
458#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
459pub struct DeleteStreamResponse {
460    pub placement: ShardPlacement,
461    pub group_commit_index: u64,
462}
463
464#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
465pub struct AckColdGcResponse {
466    pub placement: ShardPlacement,
467    pub removed: u64,
468    pub group_commit_index: u64,
469}
470
471#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
472pub struct PurgeBucketResponse {
473    pub placement: ShardPlacement,
474    pub removed_streams: u64,
475    pub group_commit_index: u64,
476}
477
478#[derive(Debug, Clone, PartialEq, Eq)]
479pub struct FlushColdRequest {
480    pub stream_id: BucketStreamId,
481    pub chunk: ColdChunkRef,
482}
483
484#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
485pub struct FlushColdResponse {
486    pub placement: ShardPlacement,
487    pub hot_start_offset: u64,
488    pub group_commit_index: u64,
489}
490
491#[derive(Debug, Clone, PartialEq, Eq)]
492pub struct CompactColdRequest {
493    pub stream_id: BucketStreamId,
494    pub old_chunks: Vec<ColdChunkRef>,
495    pub replacement: ColdChunkRef,
496    pub gc_not_before_ms: u64,
497}
498
499#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
500pub struct CompactColdResponse {
501    pub placement: ShardPlacement,
502    pub compacted_chunks: u64,
503    pub compacted_bytes: u64,
504    pub group_commit_index: u64,
505}
506
507#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
508pub struct TouchStreamAccessResponse {
509    pub placement: ShardPlacement,
510    pub changed: bool,
511    pub expired: bool,
512    pub group_commit_index: u64,
513}
514
515#[derive(Debug, Clone, PartialEq, Eq)]
516pub struct PlanColdFlushRequest {
517    pub stream_id: BucketStreamId,
518    pub min_hot_bytes: usize,
519    pub max_flush_bytes: usize,
520}
521
522#[derive(Debug, Clone, PartialEq, Eq)]
523pub struct PlanGroupColdFlushRequest {
524    pub min_hot_bytes: usize,
525    pub max_flush_bytes: usize,
526    /// Maximum aggregate payload bytes returned by one group planning pass.
527    pub max_batch_bytes: usize,
528}
529
530#[derive(Debug, Clone, PartialEq, Eq)]
531pub struct ColdHotBacklog {
532    pub stream_id: BucketStreamId,
533    pub stream_hot_bytes: u64,
534    pub group_hot_bytes: u64,
535}
536
537#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
538pub struct ColdWriteAdmission {
539    pub max_hot_bytes_per_group: Option<u64>,
540}
541
542impl ColdWriteAdmission {
543    pub(crate) fn is_enabled(self) -> bool {
544        self.max_hot_bytes_per_group.is_some()
545    }
546}
547
548#[derive(Debug, Clone, PartialEq, Eq)]
549pub struct AppendRequest {
550    pub stream_id: BucketStreamId,
551    pub content_type: String,
552    pub payload: Bytes,
553    pub close_after: bool,
554    pub stream_seq: Option<String>,
555    pub producer: Option<ProducerRequest>,
556    pub now_ms: u64,
557    pub record_match: Option<u64>,
558}
559
560#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
561pub struct AppendExternalRequest {
562    pub stream_id: BucketStreamId,
563    pub content_type: String,
564    pub payload: ExternalPayloadRef,
565    #[serde(default)]
566    pub record_ends: Vec<u64>,
567    pub close_after: bool,
568    pub stream_seq: Option<String>,
569    pub producer: Option<ProducerRequest>,
570    pub now_ms: u64,
571    pub record_match: Option<u64>,
572}
573
574impl AppendExternalRequest {
575    pub fn from_append_request(
576        request: AppendRequest,
577        payload: ExternalPayloadRef,
578        record_ends: Vec<u64>,
579    ) -> Self {
580        Self {
581            stream_id: request.stream_id,
582            content_type: request.content_type,
583            payload,
584            record_ends,
585            close_after: request.close_after,
586            stream_seq: request.stream_seq,
587            producer: request.producer,
588            now_ms: request.now_ms,
589            record_match: request.record_match,
590        }
591    }
592}
593
594impl AppendRequest {
595    pub fn canonical_record_ends(&self) -> Vec<u64> {
596        ursula_stream::canonical_json_record_ends(&self.content_type, &self.payload)
597            .unwrap_or_default()
598    }
599
600    pub fn new(stream_id: BucketStreamId, payload_len: u64) -> Self {
601        Self {
602            stream_id,
603            content_type: DEFAULT_CONTENT_TYPE.to_owned(),
604            payload: Bytes::from(vec![
605                0;
606                usize::try_from(payload_len)
607                    .expect("payload_len fits usize")
608            ]),
609            close_after: false,
610            stream_seq: None,
611            producer: None,
612            now_ms: 0,
613            record_match: None,
614        }
615    }
616
617    pub fn from_bytes(stream_id: BucketStreamId, payload: impl Into<Bytes>) -> Self {
618        Self {
619            stream_id,
620            content_type: DEFAULT_CONTENT_TYPE.to_owned(),
621            payload: payload.into(),
622            close_after: false,
623            stream_seq: None,
624            producer: None,
625            now_ms: 0,
626            record_match: None,
627        }
628    }
629
630    pub fn payload_len(&self) -> u64 {
631        u64::try_from(self.payload.len()).expect("payload len fits u64")
632    }
633}
634
635#[derive(Debug, Clone, PartialEq, Eq)]
636pub struct AppendBatchRequest {
637    pub stream_id: BucketStreamId,
638    pub content_type: String,
639    pub payloads: Vec<Bytes>,
640    pub producer: Option<ProducerRequest>,
641    pub now_ms: u64,
642}
643
644impl AppendBatchRequest {
645    pub fn new<P>(stream_id: BucketStreamId, payloads: Vec<P>) -> Self
646    where P: Into<Bytes> {
647        Self {
648            stream_id,
649            content_type: DEFAULT_CONTENT_TYPE.to_owned(),
650            payloads: payloads.into_iter().map(Into::into).collect(),
651            producer: None,
652            now_ms: 0,
653        }
654    }
655}
656
657#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
658pub struct AppendResponse {
659    pub placement: ShardPlacement,
660    pub start_offset: u64,
661    pub next_offset: u64,
662    pub stream_append_count: u64,
663    pub group_commit_index: u64,
664    pub closed: bool,
665    pub deduplicated: bool,
666    pub producer: Option<ProducerRequest>,
667    pub record_range: Option<StreamRecordRange>,
668    #[serde(default)]
669    pub stream_hot_bytes: u64,
670    #[serde(default)]
671    pub group_hot_bytes: u64,
672}
673
674#[derive(Debug, Clone, PartialEq, Eq)]
675pub struct AppendBatchResponse {
676    pub placement: ShardPlacement,
677    pub items: Vec<Result<AppendResponse, RuntimeError>>,
678}
679
680#[derive(Debug, Clone, PartialEq, Eq)]
681pub struct AppendTransactionRequest {
682    pub operations: Vec<AppendRequest>,
683}
684
685impl AppendTransactionRequest {
686    pub fn payload_bytes(&self) -> u64 {
687        self.operations.iter().fold(0_u64, |total, operation| {
688            total.saturating_add(operation.payload_len())
689        })
690    }
691}
692
693#[derive(Debug, Clone, PartialEq, Eq)]
694pub struct AppendTransactionResponse {
695    pub placement: ShardPlacement,
696    pub items: Vec<AppendResponse>,
697}
698
699#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
700pub struct StreamAppendCount {
701    pub stream_id: BucketStreamId,
702    pub append_count: u64,
703}
ursula_runtime/request.rs

ursula_runtime/
request.rs
