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    /// Require the owning Raft leader's applied state instead of permitting a
168    /// local follower read. Recovery paths use this after an acknowledged
169    /// write; ordinary catch-up consumers keep the cheaper follower-local
170    /// behavior.
171    pub leader_only: bool,
172}
173
174impl ReadStreamRequest {
175    pub(crate) fn same_wait_plan(&self, other: &Self) -> bool {
176        self.stream_id == other.stream_id
177            && self.offset == other.offset
178            && self.max_len == other.max_len
179            && self.record == other.record
180            && self.max_records == other.max_records
181            && self.leader_only == other.leader_only
182    }
183}
184
185#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
186pub struct ReadStreamResponse {
187    pub placement: ShardPlacement,
188    pub offset: u64,
189    pub next_offset: u64,
190    pub content_type: String,
191    #[serde(with = "serde_bytes")]
192    pub payload: Vec<u8>,
193    pub up_to_date: bool,
194    pub closed: bool,
195    pub retained_record_range: Option<StreamRecordRange>,
196    pub record_range: Option<StreamRecordRange>,
197}
198
199pub enum GroupReadStreamBody {
200    Materialized(Vec<u8>),
201    Planned {
202        stream_id: BucketStreamId,
203        plan: StreamReadPlan,
204        cold_store: Option<ColdStoreHandle>,
205        cold_index_cache: Option<Arc<ColdIndexPageCache<ColdStoreColdIndexPageStore>>>,
206    },
207    #[cfg(test)]
208    Blocking {
209        entered: Arc<crate::rt::sync::Notify>,
210        materialized: Arc<crate::rt::sync::Notify>,
211        release: Arc<crate::rt::sync::Notify>,
212        payload: Vec<u8>,
213    },
214}
215
216pub struct GroupReadStreamParts {
217    pub placement: ShardPlacement,
218    pub offset: u64,
219    pub next_offset: u64,
220    pub content_type: String,
221    pub up_to_date: bool,
222    pub closed: bool,
223    pub retained_record_range: Option<StreamRecordRange>,
224    pub record_range: Option<StreamRecordRange>,
225    pub body: GroupReadStreamBody,
226}
227
228impl GroupReadStreamParts {
229    pub fn from_response(response: ReadStreamResponse) -> Self {
230        Self {
231            placement: response.placement,
232            offset: response.offset,
233            next_offset: response.next_offset,
234            content_type: response.content_type,
235            up_to_date: response.up_to_date,
236            closed: response.closed,
237            retained_record_range: response.retained_record_range,
238            record_range: response.record_range,
239            body: GroupReadStreamBody::Materialized(response.payload),
240        }
241    }
242
243    pub fn from_plan(
244        placement: ShardPlacement,
245        stream_id: BucketStreamId,
246        plan: StreamReadPlan,
247        cold_store: Option<ColdStoreHandle>,
248        cold_index_cache: Option<Arc<ColdIndexPageCache<ColdStoreColdIndexPageStore>>>,
249    ) -> Self {
250        Self {
251            placement,
252            offset: plan.offset,
253            next_offset: plan.next_offset,
254            content_type: plan.content_type.clone(),
255            up_to_date: plan.up_to_date,
256            closed: plan.closed,
257            retained_record_range: plan.retained_record_range,
258            record_range: plan.record_range,
259            body: GroupReadStreamBody::Planned {
260                stream_id,
261                plan,
262                cold_store,
263                cold_index_cache,
264            },
265        }
266    }
267
268    pub async fn into_response(self) -> Result<ReadStreamResponse, GroupEngineError> {
269        let payload = match &self.body {
270            GroupReadStreamBody::Materialized(payload) => payload.clone(),
271            GroupReadStreamBody::Planned {
272                stream_id,
273                plan,
274                cold_store,
275                cold_index_cache,
276            } => {
277                InMemoryGroupEngine::read_payload_from_plan(
278                    cold_store.as_ref(),
279                    cold_index_cache.as_ref(),
280                    stream_id,
281                    plan,
282                )
283                .await?
284            }
285            #[cfg(test)]
286            GroupReadStreamBody::Blocking {
287                entered,
288                materialized,
289                release,
290                payload,
291            } => {
292                entered.notify_one();
293                materialized.notify_one();
294                release.notified().await;
295                payload.clone()
296            }
297        };
298        Ok(ReadStreamResponse {
299            placement: self.placement,
300            offset: self.offset,
301            next_offset: self.next_offset,
302            content_type: self.content_type,
303            payload,
304            up_to_date: self.up_to_date,
305            closed: self.closed,
306            retained_record_range: self.retained_record_range,
307            record_range: self.record_range,
308        })
309    }
310
311    pub fn payload_is_empty(&self) -> bool {
312        match &self.body {
313            GroupReadStreamBody::Materialized(payload) => payload.is_empty(),
314            GroupReadStreamBody::Planned { plan, .. } => {
315                plan.segments.iter().all(|segment| match segment {
316                    StreamReadSegment::Hot(payload) => payload.is_empty(),
317                    StreamReadSegment::ColdIndex(segment) => segment.len == 0,
318                    StreamReadSegment::Object(segment) => segment.len == 0,
319                })
320            }
321            #[cfg(test)]
322            GroupReadStreamBody::Blocking { payload, .. } => payload.is_empty(),
323        }
324    }
325}
326
327#[derive(Debug, Clone, PartialEq, Eq)]
328pub struct PublishSnapshotRequest {
329    pub stream_id: BucketStreamId,
330    pub snapshot_offset: u64,
331    pub content_type: String,
332    pub payload: Bytes,
333    pub expected_digest: Option<String>,
334    pub now_ms: u64,
335}
336
337#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
338pub struct PublishSnapshotResponse {
339    pub placement: ShardPlacement,
340    pub snapshot_offset: u64,
341    pub snapshot_digest: String,
342    pub group_commit_index: u64,
343    pub record_range: Option<StreamRecordRange>,
344}
345
346#[derive(Debug, Clone, PartialEq, Eq)]
347pub struct AdvanceRetentionRequest {
348    pub stream_id: BucketStreamId,
349    pub retained_offset: u64,
350    pub now_ms: u64,
351}
352
353#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
354pub struct AdvanceRetentionResponse {
355    pub placement: ShardPlacement,
356    pub retained_offset: u64,
357    pub group_commit_index: u64,
358    pub record_range: Option<StreamRecordRange>,
359}
360
361/// Sets or clears one bucket's data-plane quota record on a group. The
362/// caller replicates the same request to every group.
363#[derive(Debug, Clone, PartialEq, Eq)]
364pub struct SetBucketQuotaRequest {
365    pub bucket_id: String,
366    pub max_streams: Option<u64>,
367    pub max_retained_bytes: Option<u64>,
368}
369
370#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
371pub struct SetBucketQuotaResponse {
372    pub placement: ShardPlacement,
373    pub group_commit_index: u64,
374}
375
376#[derive(Debug, Clone, PartialEq, Eq)]
377pub struct ImportGroupStateRequest {
378    pub snapshot: Box<ursula_stream::StreamSnapshot>,
379}
380
381#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
382pub struct ImportGroupStateResponse {
383    pub placement: ShardPlacement,
384    pub buckets: u64,
385    pub streams: u64,
386    pub group_commit_index: u64,
387}
388
389#[derive(Debug, Clone, PartialEq, Eq)]
390pub struct ReadSnapshotRequest {
391    pub stream_id: BucketStreamId,
392    pub snapshot_offset: Option<u64>,
393    pub now_ms: u64,
394}
395
396#[derive(Debug, Clone, PartialEq, Eq)]
397pub struct ReadSnapshotResponse {
398    pub placement: ShardPlacement,
399    pub snapshot_offset: u64,
400    pub next_offset: u64,
401    pub content_type: String,
402    pub snapshot_digest: String,
403    pub payload: Vec<u8>,
404    pub up_to_date: bool,
405    pub record_range: Option<StreamRecordRange>,
406}
407
408#[derive(Debug, Clone, PartialEq, Eq)]
409pub struct DeleteSnapshotRequest {
410    pub stream_id: BucketStreamId,
411    pub snapshot_offset: u64,
412    pub now_ms: u64,
413}
414
415#[derive(Debug, Clone, PartialEq, Eq)]
416pub struct BootstrapStreamRequest {
417    pub stream_id: BucketStreamId,
418    pub now_ms: u64,
419}
420
421#[derive(Debug, Clone, PartialEq, Eq)]
422pub struct BootstrapUpdate {
423    pub start_offset: u64,
424    pub next_offset: u64,
425    pub content_type: String,
426    pub payload: Vec<u8>,
427}
428
429#[derive(Debug, Clone, PartialEq, Eq)]
430pub struct BootstrapStreamResponse {
431    pub placement: ShardPlacement,
432    pub snapshot_offset: Option<u64>,
433    pub snapshot_content_type: String,
434    pub snapshot_payload: Vec<u8>,
435    pub updates: Vec<BootstrapUpdate>,
436    pub next_offset: u64,
437    pub up_to_date: bool,
438    pub closed: bool,
439    pub record_range: Option<StreamRecordRange>,
440}
441
442#[derive(Debug, Clone, PartialEq, Eq)]
443pub struct CloseStreamRequest {
444    pub stream_id: BucketStreamId,
445    pub stream_seq: Option<String>,
446    pub producer: Option<ProducerRequest>,
447    pub now_ms: u64,
448}
449
450#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
451pub struct CloseStreamResponse {
452    pub placement: ShardPlacement,
453    pub next_offset: u64,
454    pub group_commit_index: u64,
455    pub deduplicated: bool,
456    pub record_range: Option<StreamRecordRange>,
457}
458
459#[derive(Debug, Clone, PartialEq, Eq)]
460pub struct DeleteStreamRequest {
461    pub stream_id: BucketStreamId,
462}
463
464#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
465pub struct DeleteStreamResponse {
466    pub placement: ShardPlacement,
467    pub group_commit_index: u64,
468}
469
470#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
471pub struct AckColdGcResponse {
472    pub placement: ShardPlacement,
473    pub removed: u64,
474    pub group_commit_index: u64,
475}
476
477#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
478pub struct PurgeBucketResponse {
479    pub placement: ShardPlacement,
480    pub removed_streams: u64,
481    /// Rolling-upgrade compatibility for <=0.4.5 voters, which did not return
482    /// a bucket-specific cold-GC count. Missing is mapped to maximally pending,
483    /// never to zero, so a mixed-version cluster cannot forge absence proof.
484    /// Remove after the minimum supported rolling source is >=0.4.6.
485    #[serde(default = "unknown_pending_cold_gc_entries")]
486    pub pending_cold_gc_entries: u64,
487    pub group_commit_index: u64,
488}
489
490const fn unknown_pending_cold_gc_entries() -> u64 {
491    u64::MAX
492}
493
494#[derive(Debug, Clone, PartialEq, Eq)]
495pub struct FlushColdRequest {
496    pub stream_id: BucketStreamId,
497    pub chunk: ColdChunkRef,
498}
499
500#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
501pub struct FlushColdResponse {
502    pub placement: ShardPlacement,
503    pub hot_start_offset: u64,
504    pub group_commit_index: u64,
505}
506
507#[derive(Debug, Clone, PartialEq, Eq)]
508pub struct CompactColdRequest {
509    pub stream_id: BucketStreamId,
510    pub old_chunks: Vec<ColdChunkRef>,
511    pub replacement: ColdChunkRef,
512    pub gc_not_before_ms: u64,
513}
514
515#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
516pub struct CompactColdResponse {
517    pub placement: ShardPlacement,
518    pub compacted_chunks: u64,
519    pub compacted_bytes: u64,
520    pub group_commit_index: u64,
521}
522
523#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
524pub struct TouchStreamAccessResponse {
525    pub placement: ShardPlacement,
526    pub changed: bool,
527    pub expired: bool,
528    pub group_commit_index: u64,
529}
530
531#[derive(Debug, Clone, PartialEq, Eq)]
532pub struct PlanColdFlushRequest {
533    pub stream_id: BucketStreamId,
534    pub min_hot_bytes: usize,
535    pub max_flush_bytes: usize,
536}
537
538#[derive(Debug, Clone, PartialEq, Eq)]
539pub struct PlanGroupColdFlushRequest {
540    pub min_hot_bytes: usize,
541    pub max_flush_bytes: usize,
542    /// Maximum aggregate payload bytes returned by one group planning pass.
543    pub max_batch_bytes: usize,
544}
545
546#[derive(Debug, Clone, PartialEq, Eq)]
547pub struct ColdHotBacklog {
548    pub stream_id: BucketStreamId,
549    pub stream_hot_bytes: u64,
550    pub group_hot_bytes: u64,
551}
552
553#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
554pub struct ColdWriteAdmission {
555    pub max_hot_bytes_per_group: Option<u64>,
556}
557
558impl ColdWriteAdmission {
559    pub(crate) fn is_enabled(self) -> bool {
560        self.max_hot_bytes_per_group.is_some()
561    }
562}
563
564#[derive(Debug, Clone, PartialEq, Eq)]
565pub struct AppendRequest {
566    pub stream_id: BucketStreamId,
567    pub content_type: String,
568    pub payload: Bytes,
569    pub close_after: bool,
570    pub stream_seq: Option<String>,
571    pub producer: Option<ProducerRequest>,
572    pub now_ms: u64,
573    pub record_match: Option<u64>,
574}
575
576#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
577pub struct AppendExternalRequest {
578    pub stream_id: BucketStreamId,
579    pub content_type: String,
580    pub payload: ExternalPayloadRef,
581    #[serde(default)]
582    pub record_ends: Vec<u64>,
583    pub close_after: bool,
584    pub stream_seq: Option<String>,
585    pub producer: Option<ProducerRequest>,
586    pub now_ms: u64,
587    pub record_match: Option<u64>,
588}
589
590impl AppendExternalRequest {
591    pub fn from_append_request(
592        request: AppendRequest,
593        payload: ExternalPayloadRef,
594        record_ends: Vec<u64>,
595    ) -> Self {
596        Self {
597            stream_id: request.stream_id,
598            content_type: request.content_type,
599            payload,
600            record_ends,
601            close_after: request.close_after,
602            stream_seq: request.stream_seq,
603            producer: request.producer,
604            now_ms: request.now_ms,
605            record_match: request.record_match,
606        }
607    }
608}
609
610impl AppendRequest {
611    pub fn canonical_record_ends(&self) -> Vec<u64> {
612        ursula_stream::canonical_json_record_ends(&self.content_type, &self.payload)
613            .unwrap_or_default()
614    }
615
616    pub fn new(stream_id: BucketStreamId, payload_len: u64) -> Self {
617        Self {
618            stream_id,
619            content_type: DEFAULT_CONTENT_TYPE.to_owned(),
620            payload: Bytes::from(vec![
621                0;
622                usize::try_from(payload_len)
623                    .expect("payload_len fits usize")
624            ]),
625            close_after: false,
626            stream_seq: None,
627            producer: None,
628            now_ms: 0,
629            record_match: None,
630        }
631    }
632
633    pub fn from_bytes(stream_id: BucketStreamId, payload: impl Into<Bytes>) -> Self {
634        Self {
635            stream_id,
636            content_type: DEFAULT_CONTENT_TYPE.to_owned(),
637            payload: payload.into(),
638            close_after: false,
639            stream_seq: None,
640            producer: None,
641            now_ms: 0,
642            record_match: None,
643        }
644    }
645
646    pub fn payload_len(&self) -> u64 {
647        u64::try_from(self.payload.len()).expect("payload len fits u64")
648    }
649}
650
651#[derive(Debug, Clone, PartialEq, Eq)]
652pub struct AppendBatchRequest {
653    pub stream_id: BucketStreamId,
654    pub content_type: String,
655    pub payloads: Vec<Bytes>,
656    pub producer: Option<ProducerRequest>,
657    pub now_ms: u64,
658}
659
660impl AppendBatchRequest {
661    pub fn new<P>(stream_id: BucketStreamId, payloads: Vec<P>) -> Self
662    where P: Into<Bytes> {
663        Self {
664            stream_id,
665            content_type: DEFAULT_CONTENT_TYPE.to_owned(),
666            payloads: payloads.into_iter().map(Into::into).collect(),
667            producer: None,
668            now_ms: 0,
669        }
670    }
671}
672
673#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
674pub struct AppendResponse {
675    pub placement: ShardPlacement,
676    pub start_offset: u64,
677    pub next_offset: u64,
678    pub stream_append_count: u64,
679    pub group_commit_index: u64,
680    pub closed: bool,
681    pub deduplicated: bool,
682    pub producer: Option<ProducerRequest>,
683    pub record_range: Option<StreamRecordRange>,
684    #[serde(default)]
685    pub stream_hot_bytes: u64,
686    #[serde(default)]
687    pub group_hot_bytes: u64,
688}
689
690#[derive(Debug, Clone, PartialEq, Eq)]
691pub struct AppendBatchResponse {
692    pub placement: ShardPlacement,
693    pub items: Vec<Result<AppendResponse, RuntimeError>>,
694}
695
696#[derive(Debug, Clone, PartialEq, Eq)]
697pub struct AppendTransactionRequest {
698    pub operations: Vec<AppendRequest>,
699}
700
701impl AppendTransactionRequest {
702    pub fn payload_bytes(&self) -> u64 {
703        self.operations.iter().fold(0_u64, |total, operation| {
704            total.saturating_add(operation.payload_len())
705        })
706    }
707}
708
709#[derive(Debug, Clone, PartialEq, Eq)]
710pub struct AppendTransactionResponse {
711    pub placement: ShardPlacement,
712    pub items: Vec<AppendResponse>,
713}
714
715#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
716pub struct StreamAppendCount {
717    pub stream_id: BucketStreamId,
718    pub append_count: u64,
719}
720
721#[cfg(test)]
722mod compatibility_tests {
723    use super::PurgeBucketResponse;
724
725    #[test]
726    fn legacy_purge_response_is_never_interpreted_as_absence_proof() {
727        let response: PurgeBucketResponse = serde_json::from_value(serde_json::json!({
728            "placement": {"core_id": 0, "shard_id": 0, "raft_group_id": 7},
729            "removed_streams": 0,
730            "group_commit_index": 8
731        }))
732        .expect("decode <=0.4.5 purge response");
733
734        assert_eq!(response.pending_cold_gc_entries, u64::MAX);
735    }
736}
ursula_runtime/request.rs

ursula_runtime/
request.rs
