ursula_runtime/engine/
mod.rs

1pub mod in_memory;
2
3use std::borrow::Cow;
4use std::future::Future;
5use std::pin::Pin;
6use std::sync::Arc;
7
8use serde::Deserialize;
9use serde::Serialize;
10use ursula_shard::BucketStreamId;
11use ursula_shard::ShardPlacement;
12use ursula_stream::BucketUsageSnapshot;
13use ursula_stream::ColdFlushCandidate;
14use ursula_stream::ColdGcEntry;
15use ursula_stream::StreamCommand;
16use ursula_stream::StreamErrorCode;
17use ursula_stream::StreamErrorContext;
18
19use crate::command::GroupSnapshot;
20use crate::command::GroupWriteCommand;
21use crate::metrics::RaftSnapshotBuildSample;
22use crate::metrics::RaftWriteManySample;
23use crate::metrics::RuntimeMetricsInner;
24use crate::request::AckColdGcResponse;
25use crate::request::AdvanceRetentionRequest;
26use crate::request::AdvanceRetentionResponse;
27use crate::request::AppendBatchRequest;
28use crate::request::AppendExternalRequest;
29use crate::request::AppendRequest;
30use crate::request::AppendResponse;
31use crate::request::AppendTransactionRequest;
32use crate::request::AppendTransactionResponse;
33use crate::request::BootstrapStreamRequest;
34use crate::request::BootstrapStreamResponse;
35use crate::request::CloseStreamRequest;
36use crate::request::CloseStreamResponse;
37use crate::request::ColdHotBacklog;
38use crate::request::ColdWriteAdmission;
39use crate::request::CompactColdRequest;
40use crate::request::CompactColdResponse;
41use crate::request::CreateStreamExternalRequest;
42use crate::request::CreateStreamRequest;
43use crate::request::CreateStreamResponse;
44use crate::request::DeleteSnapshotRequest;
45use crate::request::DeleteStreamRequest;
46use crate::request::DeleteStreamResponse;
47use crate::request::FlushColdRequest;
48use crate::request::FlushColdResponse;
49use crate::request::GetStreamAttrsRequest;
50use crate::request::GetStreamAttrsResponse;
51use crate::request::GroupReadStreamParts;
52use crate::request::HeadStreamRequest;
53use crate::request::HeadStreamResponse;
54use crate::request::ImportGroupStateRequest;
55use crate::request::PlanColdFlushRequest;
56use crate::request::PlanGroupColdFlushRequest;
57use crate::request::PublishSnapshotRequest;
58use crate::request::PublishSnapshotResponse;
59use crate::request::PurgeBucketResponse;
60use crate::request::ReadSnapshotRequest;
61use crate::request::ReadSnapshotResponse;
62use crate::request::ReadStreamRequest;
63use crate::request::ReadStreamResponse;
64use crate::request::SetBucketQuotaRequest;
65use crate::request::SetBucketQuotaResponse;
66use crate::request::TouchStreamAccessResponse;
67use crate::request::UpdateStreamAttrsRequest;
68use crate::request::UpdateStreamAttrsResponse;
69
70pub type GroupAppendFuture<'a> =
71    Pin<Box<dyn Future<Output = Result<AppendResponse, GroupEngineError>> + Send + 'a>>;
72pub type GroupAppendBatchFuture<'a> =
73    Pin<Box<dyn Future<Output = Result<GroupAppendBatchResponse, GroupEngineError>> + Send + 'a>>;
74pub type GroupAppendTransactionFuture<'a> =
75    Pin<Box<dyn Future<Output = Result<AppendTransactionResponse, GroupEngineError>> + Send + 'a>>;
76pub type GroupFlushColdFuture<'a> =
77    Pin<Box<dyn Future<Output = Result<FlushColdResponse, GroupEngineError>> + Send + 'a>>;
78pub type GroupCompactColdFuture<'a> =
79    Pin<Box<dyn Future<Output = Result<CompactColdResponse, GroupEngineError>> + Send + 'a>>;
80pub type GroupPlanColdFlushFuture<'a> =
81    Pin<Box<dyn Future<Output = Result<Option<ColdFlushCandidate>, GroupEngineError>> + Send + 'a>>;
82pub type GroupPlanNextColdFlushBatchFuture<'a> =
83    Pin<Box<dyn Future<Output = Result<Vec<ColdFlushCandidate>, GroupEngineError>> + Send + 'a>>;
84pub type GroupColdHotBacklogFuture<'a> =
85    Pin<Box<dyn Future<Output = Result<ColdHotBacklog, GroupEngineError>> + Send + 'a>>;
86pub type GroupBucketUsageFuture<'a> =
87    Pin<Box<dyn Future<Output = Result<Vec<BucketUsageSnapshot>, GroupEngineError>> + Send + 'a>>;
88pub type GroupCreateStreamFuture<'a> =
89    Pin<Box<dyn Future<Output = Result<CreateStreamResponse, GroupEngineError>> + Send + 'a>>;
90pub type GroupHeadStreamFuture<'a> =
91    Pin<Box<dyn Future<Output = Result<HeadStreamResponse, GroupEngineError>> + Send + 'a>>;
92pub type GroupGetStreamAttrsFuture<'a> =
93    Pin<Box<dyn Future<Output = Result<GetStreamAttrsResponse, GroupEngineError>> + Send + 'a>>;
94pub type GroupReadStreamFuture<'a> =
95    Pin<Box<dyn Future<Output = Result<ReadStreamResponse, GroupEngineError>> + Send + 'a>>;
96pub type GroupReadStreamPartsFuture<'a> =
97    Pin<Box<dyn Future<Output = Result<GroupReadStreamParts, GroupEngineError>> + Send + 'a>>;
98pub type GroupRequireLiveReadOwnerFuture<'a> =
99    Pin<Box<dyn Future<Output = Result<(), GroupEngineError>> + Send + 'a>>;
100pub type GroupPublishSnapshotFuture<'a> =
101    Pin<Box<dyn Future<Output = Result<PublishSnapshotResponse, GroupEngineError>> + Send + 'a>>;
102pub type GroupAdvanceRetentionFuture<'a> =
103    Pin<Box<dyn Future<Output = Result<AdvanceRetentionResponse, GroupEngineError>> + Send + 'a>>;
104pub type GroupSetBucketQuotaFuture<'a> =
105    Pin<Box<dyn Future<Output = Result<SetBucketQuotaResponse, GroupEngineError>> + Send + 'a>>;
106pub type GroupReadSnapshotFuture<'a> =
107    Pin<Box<dyn Future<Output = Result<ReadSnapshotResponse, GroupEngineError>> + Send + 'a>>;
108pub type GroupDeleteSnapshotFuture<'a> =
109    Pin<Box<dyn Future<Output = Result<(), GroupEngineError>> + Send + 'a>>;
110pub type GroupBootstrapStreamFuture<'a> =
111    Pin<Box<dyn Future<Output = Result<BootstrapStreamResponse, GroupEngineError>> + Send + 'a>>;
112pub type GroupTouchStreamAccessFuture<'a> =
113    Pin<Box<dyn Future<Output = Result<TouchStreamAccessResponse, GroupEngineError>> + Send + 'a>>;
114pub type GroupUpdateStreamAttrsFuture<'a> =
115    Pin<Box<dyn Future<Output = Result<UpdateStreamAttrsResponse, GroupEngineError>> + Send + 'a>>;
116pub type GroupCloseStreamFuture<'a> =
117    Pin<Box<dyn Future<Output = Result<CloseStreamResponse, GroupEngineError>> + Send + 'a>>;
118pub type GroupDeleteStreamFuture<'a> =
119    Pin<Box<dyn Future<Output = Result<DeleteStreamResponse, GroupEngineError>> + Send + 'a>>;
120pub type GroupAckColdGcFuture<'a> =
121    Pin<Box<dyn Future<Output = Result<AckColdGcResponse, GroupEngineError>> + Send + 'a>>;
122pub type GroupPurgeBucketFuture<'a> =
123    Pin<Box<dyn Future<Output = Result<PurgeBucketResponse, GroupEngineError>> + Send + 'a>>;
124pub type GroupPlanColdGcFuture<'a> =
125    Pin<Box<dyn Future<Output = Result<Vec<ColdGcEntry>, GroupEngineError>> + Send + 'a>>;
126pub type GroupImportGroupStateFuture<'a> = Pin<
127    Box<
128        dyn Future<Output = Result<crate::request::ImportGroupStateResponse, GroupEngineError>>
129            + Send
130            + 'a,
131    >,
132>;
133pub type GroupSnapshotFuture<'a> =
134    Pin<Box<dyn Future<Output = Result<GroupSnapshot, GroupEngineError>> + Send + 'a>>;
135pub type GroupInstallSnapshotFuture<'a> =
136    Pin<Box<dyn Future<Output = Result<(), GroupEngineError>> + Send + 'a>>;
137pub type GroupShutdownFuture<'a> =
138    Pin<Box<dyn Future<Output = Result<(), GroupEngineError>> + Send + 'a>>;
139pub type GroupWriteFuture<'a> =
140    Pin<Box<dyn Future<Output = Result<GroupWriteResponse, GroupEngineError>> + Send + 'a>>;
141pub type GroupWriteBatchFuture<'a> = Pin<
142    Box<
143        dyn Future<
144                Output = Result<
145                    Vec<Result<GroupWriteResponse, GroupEngineError>>,
146                    GroupEngineError,
147                >,
148            > + Send
149            + 'a,
150    >,
151>;
152pub type GroupEngineCreateFuture<'a> =
153    Pin<Box<dyn Future<Output = Result<Box<dyn GroupEngine>, GroupEngineError>> + Send + 'a>>;
154
155#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
156pub struct GroupAppendBatchResponse {
157    pub placement: ShardPlacement,
158    pub items: Vec<Result<AppendResponse, GroupEngineError>>,
159}
160
161#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
162pub enum GroupWriteResponse {
163    CreateStream(CreateStreamResponse),
164    Append(AppendResponse),
165    AppendBatch(GroupAppendBatchResponse),
166    PublishSnapshot(PublishSnapshotResponse),
167    AdvanceRetention(AdvanceRetentionResponse),
168    SetBucketQuota(SetBucketQuotaResponse),
169    TouchStreamAccess(TouchStreamAccessResponse),
170    UpdateStreamAttrs(UpdateStreamAttrsResponse),
171    FlushCold(FlushColdResponse),
172    CompactCold(CompactColdResponse),
173    CloseStream(CloseStreamResponse),
174    DeleteStream(DeleteStreamResponse),
175    AckColdGc(AckColdGcResponse),
176    PurgeBucket(PurgeBucketResponse),
177    ImportGroupState(crate::request::ImportGroupStateResponse),
178    Batch(Vec<Result<GroupWriteResponse, GroupEngineError>>),
179}
180
181pub trait GroupEngine: Send + 'static {
182    fn accepts_local_writes(&self) -> bool {
183        true
184    }
185
186    fn create_stream<'a>(
187        &'a mut self,
188        request: CreateStreamRequest,
189        placement: ShardPlacement,
190        admission: ColdWriteAdmission,
191    ) -> GroupCreateStreamFuture<'a>;
192
193    fn create_stream_external<'a>(
194        &'a mut self,
195        request: CreateStreamExternalRequest,
196        _placement: ShardPlacement,
197    ) -> GroupCreateStreamFuture<'a> {
198        Box::pin(async move {
199            Err(GroupEngineError::new(format!(
200                "external stream create is not supported for stream '{}'",
201                request.stream_id
202            )))
203        })
204    }
205
206    fn head_stream<'a>(
207        &'a mut self,
208        request: HeadStreamRequest,
209        placement: ShardPlacement,
210    ) -> GroupHeadStreamFuture<'a>;
211
212    /// Per-bucket committed usage held by this group's replicated state.
213    ///
214    /// Served from local replica state, leader or follower: usage export
215    /// tolerates replication lag, and requiring leadership would make a
216    /// node-local aggregate fail whenever any group is led elsewhere.
217    /// Deliberately a required method — an engine that silently reported
218    /// nothing would underbill.
219    fn bucket_usage<'a>(&'a mut self, placement: ShardPlacement) -> GroupBucketUsageFuture<'a>;
220
221    fn get_stream_attrs<'a>(
222        &'a mut self,
223        request: GetStreamAttrsRequest,
224        _placement: ShardPlacement,
225    ) -> GroupGetStreamAttrsFuture<'a> {
226        Box::pin(async move {
227            Err(GroupEngineError::new(format!(
228                "stream attrs read is not supported for stream '{}'",
229                request.stream_id
230            )))
231        })
232    }
233
234    fn read_stream<'a>(
235        &'a mut self,
236        request: ReadStreamRequest,
237        placement: ShardPlacement,
238    ) -> GroupReadStreamFuture<'a>;
239
240    fn read_stream_parts<'a>(
241        &'a mut self,
242        request: ReadStreamRequest,
243        placement: ShardPlacement,
244    ) -> GroupReadStreamPartsFuture<'a> {
245        Box::pin(async move {
246            let response = self.read_stream(request, placement).await?;
247            Ok(GroupReadStreamParts::from_response(response))
248        })
249    }
250
251    fn require_local_live_read_owner<'a>(
252        &'a mut self,
253        _placement: ShardPlacement,
254    ) -> GroupRequireLiveReadOwnerFuture<'a> {
255        Box::pin(async { Ok(()) })
256    }
257
258    fn publish_snapshot<'a>(
259        &'a mut self,
260        request: PublishSnapshotRequest,
261        _placement: ShardPlacement,
262    ) -> GroupPublishSnapshotFuture<'a> {
263        Box::pin(async move {
264            Err(GroupEngineError::new(format!(
265                "snapshot publish is not supported for stream '{}'",
266                request.stream_id
267            )))
268        })
269    }
270
271    fn advance_retention<'a>(
272        &'a mut self,
273        request: AdvanceRetentionRequest,
274        _placement: ShardPlacement,
275    ) -> GroupAdvanceRetentionFuture<'a> {
276        Box::pin(async move {
277            Err(GroupEngineError::new(format!(
278                "retention advance is not supported for stream '{}'",
279                request.stream_id
280            )))
281        })
282    }
283
284    /// Restore path: replaces an empty group's state with a backup snapshot
285    /// as one replicated write. See `StreamCommand::ImportSnapshot`.
286    fn import_group_state<'a>(
287        &'a mut self,
288        _request: ImportGroupStateRequest,
289        placement: ShardPlacement,
290    ) -> GroupImportGroupStateFuture<'a> {
291        Box::pin(async move {
292            Err(GroupEngineError::new(format!(
293                "group state import is not supported for group {}",
294                placement.raft_group_id.0
295            )))
296        })
297    }
298
299    fn set_bucket_quota<'a>(
300        &'a mut self,
301        request: SetBucketQuotaRequest,
302        _placement: ShardPlacement,
303    ) -> GroupSetBucketQuotaFuture<'a> {
304        Box::pin(async move {
305            Err(GroupEngineError::new(format!(
306                "bucket quotas are not supported for bucket '{}'",
307                request.bucket_id
308            )))
309        })
310    }
311
312    fn read_snapshot<'a>(
313        &'a mut self,
314        request: ReadSnapshotRequest,
315        _placement: ShardPlacement,
316    ) -> GroupReadSnapshotFuture<'a> {
317        Box::pin(async move {
318            Err(GroupEngineError::new(format!(
319                "snapshot read is not supported for stream '{}'",
320                request.stream_id
321            )))
322        })
323    }
324
325    fn delete_snapshot<'a>(
326        &'a mut self,
327        request: DeleteSnapshotRequest,
328        _placement: ShardPlacement,
329    ) -> GroupDeleteSnapshotFuture<'a> {
330        Box::pin(async move {
331            Err(GroupEngineError::new(format!(
332                "snapshot delete is not supported for stream '{}'",
333                request.stream_id
334            )))
335        })
336    }
337
338    fn bootstrap_stream<'a>(
339        &'a mut self,
340        request: BootstrapStreamRequest,
341        _placement: ShardPlacement,
342    ) -> GroupBootstrapStreamFuture<'a> {
343        Box::pin(async move {
344            Err(GroupEngineError::new(format!(
345                "bootstrap is not supported for stream '{}'",
346                request.stream_id
347            )))
348        })
349    }
350
351    fn touch_stream_access<'a>(
352        &'a mut self,
353        stream_id: BucketStreamId,
354        now_ms: u64,
355        renew_ttl: bool,
356        placement: ShardPlacement,
357    ) -> GroupTouchStreamAccessFuture<'a>;
358
359    fn update_stream_attrs<'a>(
360        &'a mut self,
361        request: UpdateStreamAttrsRequest,
362        _placement: ShardPlacement,
363    ) -> GroupUpdateStreamAttrsFuture<'a> {
364        Box::pin(async move {
365            Err(GroupEngineError::new(format!(
366                "stream attrs update is not supported for stream '{}'",
367                request.stream_id
368            )))
369        })
370    }
371
372    fn close_stream<'a>(
373        &'a mut self,
374        request: CloseStreamRequest,
375        placement: ShardPlacement,
376    ) -> GroupCloseStreamFuture<'a>;
377
378    fn delete_stream<'a>(
379        &'a mut self,
380        request: DeleteStreamRequest,
381        placement: ShardPlacement,
382    ) -> GroupDeleteStreamFuture<'a>;
383
384    /// Replicated confirmation that cold-GC entries up to `up_to_seq` have been
385    /// physically reclaimed; pops them from the queue. Default unsupported.
386    fn ack_cold_gc<'a>(
387        &'a mut self,
388        _up_to_seq: u64,
389        _placement: ShardPlacement,
390    ) -> GroupAckColdGcFuture<'a> {
391        Box::pin(async { Err(GroupEngineError::new("cold GC ack is not supported")) })
392    }
393
394    /// Replicated tenant offboarding: removes every stream in the bucket, the
395    /// bucket, and its quota in this group. Monotonic aggregate usage remains
396    /// available to asynchronous accounting readers. Default unsupported.
397    fn purge_bucket<'a>(
398        &'a mut self,
399        _bucket_id: String,
400        _placement: ShardPlacement,
401    ) -> GroupPurgeBucketFuture<'a> {
402        Box::pin(async { Err(GroupEngineError::new("bucket purge is not supported")) })
403    }
404
405    /// Leader-local read of the front of the cold-GC queue for the background
406    /// worker to reclaim. Default returns an empty batch.
407    fn plan_cold_gc<'a>(
408        &'a mut self,
409        _max: usize,
410        _placement: ShardPlacement,
411    ) -> GroupPlanColdGcFuture<'a> {
412        Box::pin(async { Ok(Vec::new()) })
413    }
414
415    fn append<'a>(
416        &'a mut self,
417        request: AppendRequest,
418        placement: ShardPlacement,
419        admission: ColdWriteAdmission,
420    ) -> GroupAppendFuture<'a>;
421
422    fn append_external<'a>(
423        &'a mut self,
424        request: AppendExternalRequest,
425        _placement: ShardPlacement,
426    ) -> GroupAppendFuture<'a> {
427        Box::pin(async move {
428            Err(GroupEngineError::new(format!(
429                "external append is not supported for stream '{}'",
430                request.stream_id
431            )))
432        })
433    }
434
435    fn append_batch<'a>(
436        &'a mut self,
437        request: AppendBatchRequest,
438        placement: ShardPlacement,
439        admission: ColdWriteAdmission,
440    ) -> GroupAppendBatchFuture<'a>;
441
442    fn append_batch_many<'a>(
443        &'a mut self,
444        requests: Vec<AppendBatchRequest>,
445        placement: ShardPlacement,
446        admission: ColdWriteAdmission,
447    ) -> GroupWriteBatchFuture<'a> {
448        Box::pin(async move {
449            let mut responses = Vec::with_capacity(requests.len());
450            for request in requests {
451                let response = self
452                    .append_batch(request, placement, admission)
453                    .await
454                    .map(GroupWriteResponse::AppendBatch);
455                responses.push(response);
456            }
457            Ok(responses)
458        })
459    }
460
461    fn append_transaction<'a>(
462        &'a mut self,
463        _request: AppendTransactionRequest,
464        _placement: ShardPlacement,
465        _admission: ColdWriteAdmission,
466    ) -> GroupAppendTransactionFuture<'a> {
467        Box::pin(async {
468            Err(GroupEngineError::new(
469                "append transactions are not supported by this group engine",
470            ))
471        })
472    }
473
474    fn flush_cold<'a>(
475        &'a mut self,
476        request: FlushColdRequest,
477        _placement: ShardPlacement,
478    ) -> GroupFlushColdFuture<'a> {
479        Box::pin(async move {
480            Err(GroupEngineError::new(format!(
481                "cold flush is not supported for stream '{}'",
482                request.stream_id
483            )))
484        })
485    }
486
487    fn compact_cold<'a>(
488        &'a mut self,
489        request: CompactColdRequest,
490        _placement: ShardPlacement,
491    ) -> GroupCompactColdFuture<'a> {
492        Box::pin(async move {
493            Err(GroupEngineError::new(format!(
494                "cold compaction is not supported for stream '{}'",
495                request.stream_id
496            )))
497        })
498    }
499
500    fn plan_cold_flush<'a>(
501        &'a mut self,
502        request: PlanColdFlushRequest,
503        _placement: ShardPlacement,
504    ) -> GroupPlanColdFlushFuture<'a> {
505        Box::pin(async move {
506            Err(GroupEngineError::new(format!(
507                "cold flush planning is not supported for stream '{}'",
508                request.stream_id
509            )))
510        })
511    }
512
513    fn plan_next_cold_flush_batch<'a>(
514        &'a mut self,
515        _request: PlanGroupColdFlushRequest,
516        _placement: ShardPlacement,
517        _max_candidates: usize,
518    ) -> GroupPlanNextColdFlushBatchFuture<'a> {
519        Box::pin(async move {
520            Err(GroupEngineError::new(
521                "group cold flush planning is not supported",
522            ))
523        })
524    }
525
526    fn cold_hot_backlog<'a>(
527        &'a mut self,
528        stream_id: BucketStreamId,
529        _placement: ShardPlacement,
530    ) -> GroupColdHotBacklogFuture<'a> {
531        Box::pin(async move {
532            Err(GroupEngineError::new(format!(
533                "cold hot backlog is not supported for stream '{stream_id}'"
534            )))
535        })
536    }
537
538    fn snapshot<'a>(&'a mut self, placement: ShardPlacement) -> GroupSnapshotFuture<'a>;
539
540    fn install_snapshot<'a>(
541        &'a mut self,
542        snapshot: GroupSnapshot,
543    ) -> GroupInstallSnapshotFuture<'a>;
544
545    fn shutdown<'a>(&'a mut self) -> GroupShutdownFuture<'a> {
546        Box::pin(async { Ok(()) })
547    }
548
549    fn write_batch<'a>(
550        &'a mut self,
551        commands: Vec<GroupWriteCommand>,
552        placement: ShardPlacement,
553    ) -> GroupWriteBatchFuture<'a> {
554        Box::pin(async move {
555            let mut responses = Vec::with_capacity(commands.len());
556            for command in commands {
557                let response = match command {
558                    GroupWriteCommand::Stream(command) => {
559                        self.dispatch_stream_command(command, placement).await
560                    }
561                    GroupWriteCommand::Batch { commands } => {
562                        let mut batched = Vec::with_capacity(commands.len());
563                        for command in commands {
564                            batched.push(self.dispatch_stream_command(command, placement).await);
565                        }
566                        Ok(GroupWriteResponse::Batch(batched))
567                    }
568                    GroupWriteCommand::Transaction { .. } => Err(GroupEngineError::new(
569                        "append transactions require an atomic group engine",
570                    )),
571                };
572                responses.push(response);
573            }
574            Ok(responses)
575        })
576    }
577
578    /// Routes one canonical [`StreamCommand`] to the matching typed engine
579    /// method. This is the only spelling of the command-to-operation mapping;
580    /// engines that replicate commands wholesale (raft) bypass it.
581    fn dispatch_stream_command<'a>(
582        &'a mut self,
583        command: StreamCommand,
584        placement: ShardPlacement,
585    ) -> GroupWriteFuture<'a> {
586        Box::pin(async move {
587            match command {
588                StreamCommand::CreateStream {
589                    stream_id,
590                    content_type,
591                    initial_payload,
592                    close_after,
593                    stream_seq,
594                    producer,
595                    stream_ttl_seconds,
596                    stream_expires_at_ms,
597                    attrs,
598                    now_ms,
599                } => self
600                    .create_stream(
601                        CreateStreamRequest {
602                            stream_id,
603                            content_type,
604                            content_type_explicit: true,
605                            initial_payload,
606                            close_after,
607                            stream_seq,
608                            producer,
609                            stream_ttl_seconds,
610                            stream_expires_at_ms,
611                            attrs,
612                            now_ms,
613                        },
614                        placement,
615                        ColdWriteAdmission::default(),
616                    )
617                    .await
618                    .map(GroupWriteResponse::CreateStream),
619                StreamCommand::CreateExternal {
620                    stream_id,
621                    content_type,
622                    initial_payload,
623                    record_ends,
624                    close_after,
625                    stream_seq,
626                    producer,
627                    stream_ttl_seconds,
628                    stream_expires_at_ms,
629                    attrs,
630                    now_ms,
631                } => self
632                    .create_stream_external(
633                        CreateStreamExternalRequest {
634                            stream_id,
635                            content_type,
636                            initial_payload,
637                            record_ends,
638                            close_after,
639                            stream_seq,
640                            producer,
641                            stream_ttl_seconds,
642                            stream_expires_at_ms,
643                            attrs,
644                            now_ms,
645                        },
646                        placement,
647                    )
648                    .await
649                    .map(GroupWriteResponse::CreateStream),
650                StreamCommand::Append {
651                    stream_id,
652                    content_type,
653                    payload,
654                    close_after,
655                    stream_seq,
656                    producer,
657                    now_ms,
658                    record_match,
659                } => self
660                    .append(
661                        AppendRequest {
662                            stream_id,
663                            content_type: content_type.unwrap_or_default(),
664                            payload,
665                            close_after,
666                            stream_seq,
667                            producer,
668                            now_ms,
669                            record_match,
670                        },
671                        placement,
672                        ColdWriteAdmission::default(),
673                    )
674                    .await
675                    .map(GroupWriteResponse::Append),
676                StreamCommand::AppendExternal {
677                    stream_id,
678                    content_type,
679                    payload,
680                    record_ends,
681                    close_after,
682                    stream_seq,
683                    producer,
684                    now_ms,
685                    record_match,
686                } => self
687                    .append_external(
688                        AppendExternalRequest {
689                            stream_id,
690                            content_type: content_type.unwrap_or_default(),
691                            payload,
692                            record_ends,
693                            close_after,
694                            stream_seq,
695                            producer,
696                            now_ms,
697                            record_match,
698                        },
699                        placement,
700                    )
701                    .await
702                    .map(GroupWriteResponse::Append),
703                StreamCommand::AppendBatch {
704                    stream_id,
705                    content_type,
706                    payloads,
707                    producer,
708                    now_ms,
709                } => self
710                    .append_batch(
711                        AppendBatchRequest {
712                            stream_id,
713                            content_type: content_type.unwrap_or_default(),
714                            payloads,
715                            producer,
716                            now_ms,
717                        },
718                        placement,
719                        ColdWriteAdmission::default(),
720                    )
721                    .await
722                    .map(GroupWriteResponse::AppendBatch),
723                StreamCommand::PublishSnapshot {
724                    stream_id,
725                    snapshot_offset,
726                    content_type,
727                    payload,
728                    expected_digest,
729                    now_ms,
730                } => self
731                    .publish_snapshot(
732                        PublishSnapshotRequest {
733                            stream_id,
734                            snapshot_offset,
735                            content_type,
736                            payload,
737                            expected_digest,
738                            now_ms,
739                        },
740                        placement,
741                    )
742                    .await
743                    .map(GroupWriteResponse::PublishSnapshot),
744                StreamCommand::AdvanceRetention {
745                    stream_id,
746                    retained_offset,
747                    now_ms,
748                } => self
749                    .advance_retention(
750                        AdvanceRetentionRequest {
751                            stream_id,
752                            retained_offset,
753                            now_ms,
754                        },
755                        placement,
756                    )
757                    .await
758                    .map(GroupWriteResponse::AdvanceRetention),
759                StreamCommand::TouchStreamAccess {
760                    stream_id,
761                    now_ms,
762                    renew_ttl,
763                } => self
764                    .touch_stream_access(stream_id, now_ms, renew_ttl, placement)
765                    .await
766                    .map(GroupWriteResponse::TouchStreamAccess),
767                StreamCommand::UpdateStreamAttrs {
768                    stream_id,
769                    attrs,
770                    now_ms,
771                } => self
772                    .update_stream_attrs(
773                        UpdateStreamAttrsRequest {
774                            stream_id,
775                            attrs,
776                            now_ms,
777                        },
778                        placement,
779                    )
780                    .await
781                    .map(GroupWriteResponse::UpdateStreamAttrs),
782                StreamCommand::FlushCold { stream_id, chunk } => self
783                    .flush_cold(FlushColdRequest { stream_id, chunk }, placement)
784                    .await
785                    .map(GroupWriteResponse::FlushCold),
786                StreamCommand::CompactCold {
787                    stream_id,
788                    old_chunks,
789                    replacement,
790                    gc_not_before_ms,
791                } => self
792                    .compact_cold(
793                        CompactColdRequest {
794                            stream_id,
795                            old_chunks,
796                            replacement,
797                            gc_not_before_ms,
798                        },
799                        placement,
800                    )
801                    .await
802                    .map(GroupWriteResponse::CompactCold),
803                StreamCommand::Close {
804                    stream_id,
805                    stream_seq,
806                    producer,
807                    now_ms,
808                } => self
809                    .close_stream(
810                        CloseStreamRequest {
811                            stream_id,
812                            stream_seq,
813                            producer,
814                            now_ms,
815                        },
816                        placement,
817                    )
818                    .await
819                    .map(GroupWriteResponse::CloseStream),
820                StreamCommand::DeleteStream { stream_id } => self
821                    .delete_stream(DeleteStreamRequest { stream_id }, placement)
822                    .await
823                    .map(GroupWriteResponse::DeleteStream),
824                StreamCommand::AckColdGc { up_to_seq } => self
825                    .ack_cold_gc(up_to_seq, placement)
826                    .await
827                    .map(GroupWriteResponse::AckColdGc),
828                StreamCommand::PurgeBucket { bucket_id } => self
829                    .purge_bucket(bucket_id, placement)
830                    .await
831                    .map(GroupWriteResponse::PurgeBucket),
832                StreamCommand::ImportSnapshot { snapshot } => self
833                    .import_group_state(ImportGroupStateRequest { snapshot }, placement)
834                    .await
835                    .map(GroupWriteResponse::ImportGroupState),
836                StreamCommand::SetBucketQuota {
837                    bucket_id,
838                    max_streams,
839                    max_retained_bytes,
840                } => self
841                    .set_bucket_quota(
842                        SetBucketQuotaRequest {
843                            bucket_id,
844                            max_streams,
845                            max_retained_bytes,
846                        },
847                        placement,
848                    )
849                    .await
850                    .map(GroupWriteResponse::SetBucketQuota),
851                StreamCommand::CreateBucket { .. } | StreamCommand::DeleteBucket { .. } => Err(
852                    GroupEngineError::new("bucket commands are not valid group writes"),
853                ),
854            }
855        })
856    }
857}
858
859pub trait GroupEngineFactory: Send + Sync + 'static {
860    fn hosts_group(&self, _placement: ShardPlacement) -> bool {
861        true
862    }
863
864    fn create<'a>(
865        &'a self,
866        placement: ShardPlacement,
867        metrics: GroupEngineMetrics,
868    ) -> GroupEngineCreateFuture<'a>;
869}
870
871#[derive(Debug, Clone)]
872pub struct GroupEngineMetrics {
873    pub(crate) inner: Arc<RuntimeMetricsInner>,
874}
875
876impl GroupEngineMetrics {
877    pub fn record_wal_batch(
878        &self,
879        placement: ShardPlacement,
880        record_count: usize,
881        write_ns: u64,
882        sync_ns: u64,
883    ) {
884        self.inner.record_wal_batch(
885            placement.core_id,
886            placement.raft_group_id,
887            u64::try_from(record_count).expect("record count fits u64"),
888            write_ns,
889            sync_ns,
890        );
891    }
892
893    pub fn record_raft_write_many(
894        &self,
895        placement: ShardPlacement,
896        command_count: usize,
897        logical_command_count: usize,
898        response_count: usize,
899        submit_ns: u64,
900        response_ns: u64,
901    ) {
902        self.inner.record_raft_write_many(
903            placement.core_id,
904            placement.raft_group_id,
905            RaftWriteManySample {
906                command_count: u64::try_from(command_count).expect("command count fits u64"),
907                logical_command_count: u64::try_from(logical_command_count)
908                    .expect("logical command count fits u64"),
909                response_count: u64::try_from(response_count).expect("response count fits u64"),
910                submit_ns,
911                response_ns,
912            },
913        );
914    }
915
916    pub fn record_raft_apply_batch(
917        &self,
918        placement: ShardPlacement,
919        entry_count: usize,
920        apply_ns: u64,
921    ) {
922        self.inner.record_raft_apply_batch(
923            placement.core_id,
924            placement.raft_group_id,
925            u64::try_from(entry_count).expect("entry count fits u64"),
926            apply_ns,
927        );
928    }
929
930    pub fn record_raft_snapshot_build(
931        &self,
932        placement: ShardPlacement,
933        stream_count: usize,
934        body_bytes: usize,
935        pointer_bytes: usize,
936        build_ns: u64,
937        external_upload: bool,
938        inline_fallback: bool,
939    ) {
940        self.inner
941            .record_raft_snapshot_build(placement.raft_group_id, RaftSnapshotBuildSample {
942                streams: u64::try_from(stream_count).expect("stream count fits u64"),
943                body_bytes: u64::try_from(body_bytes).expect("snapshot body bytes fits u64"),
944                pointer_bytes: u64::try_from(pointer_bytes)
945                    .expect("snapshot pointer bytes fits u64"),
946                build_ns,
947                external_upload,
948                inline_fallback,
949            });
950    }
951}
952
953#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
954pub struct GroupLeaderHint {
955    pub node_id: Option<u64>,
956    pub address: Option<String>,
957}
958
959#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
960pub struct StreamEngineError {
961    message: String,
962    code: StreamErrorCode,
963    next_offset: Option<u64>,
964    #[serde(default, skip_serializing_if = "Vec::is_empty")]
965    context: Vec<StreamErrorContext>,
966}
967
968/// Infra error variants with structured fields render their human message on
969/// demand (`message`) instead of storing a denormalized copy alongside the
970/// fields. `Internal` is the exception: it carries free-form text with no
971/// structured source, so it keeps an owned `message`.
972#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, thiserror::Error)]
973pub enum GroupInfraError {
974    #[error("{message}")]
975    Internal { message: String },
976    #[error("ProtoDecode: protobuf raft payload missing {field}")]
977    ProtoDecode { field: String },
978    #[error(
979        "ColdBackpressure: stream '{stream_id}' would raise group hot bytes from {before_group_hot_bytes} to {after_group_hot_bytes}, above limit {limit}"
980    )]
981    ColdBackpressure {
982        stream_id: BucketStreamId,
983        before_group_hot_bytes: u64,
984        after_group_hot_bytes: u64,
985        limit: u64,
986    },
987    #[error(
988        "RaftUncommittedBackpressure: group uncommitted bytes {current} plus incoming {incoming} would exceed limit {limit}"
989    )]
990    RaftUncommittedBackpressure {
991        current: u64,
992        incoming: u64,
993        limit: u64,
994    },
995}
996
997impl GroupInfraError {
998    pub fn internal(message: impl Into<String>) -> Self {
999        Self::Internal {
1000            message: message.into(),
1001        }
1002    }
1003
1004    pub fn proto_decode(field: impl Into<String>) -> Self {
1005        Self::ProtoDecode {
1006            field: field.into(),
1007        }
1008    }
1009
1010    pub fn cold_backpressure(
1011        stream_id: BucketStreamId,
1012        before_group_hot_bytes: u64,
1013        after_group_hot_bytes: u64,
1014        limit: u64,
1015    ) -> Self {
1016        Self::ColdBackpressure {
1017            stream_id,
1018            before_group_hot_bytes,
1019            after_group_hot_bytes,
1020            limit,
1021        }
1022    }
1023
1024    pub fn raft_uncommitted_backpressure(current: u64, incoming: u64, limit: u64) -> Self {
1025        Self::RaftUncommittedBackpressure {
1026            current,
1027            incoming,
1028            limit,
1029        }
1030    }
1031
1032    pub fn message(&self) -> Cow<'_, str> {
1033        match self {
1034            Self::Internal { message } => Cow::Borrowed(message),
1035            other => Cow::Owned(other.to_string()),
1036        }
1037    }
1038
1039    pub fn is_cold_backpressure(&self) -> bool {
1040        matches!(self, Self::ColdBackpressure { .. })
1041    }
1042
1043    pub fn is_backpressure(&self) -> bool {
1044        matches!(
1045            self,
1046            Self::ColdBackpressure { .. } | Self::RaftUncommittedBackpressure { .. }
1047        )
1048    }
1049}
1050
1051#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, thiserror::Error)]
1052pub enum GroupEngineError {
1053    #[error("{}", .0.message)]
1054    Stream(StreamEngineError),
1055    #[error("{}", .0.message())]
1056    Infra(GroupInfraError),
1057    #[error("{message}")]
1058    ForwardToLeader {
1059        message: String,
1060        leader_hint: GroupLeaderHint,
1061    },
1062}
1063
1064impl GroupEngineError {
1065    pub fn new(message: impl Into<String>) -> Self {
1066        Self::Infra(GroupInfraError::internal(message))
1067    }
1068
1069    pub fn cold_backpressure(
1070        stream_id: BucketStreamId,
1071        before_group_hot_bytes: u64,
1072        after_group_hot_bytes: u64,
1073        limit: u64,
1074    ) -> Self {
1075        Self::Infra(GroupInfraError::cold_backpressure(
1076            stream_id,
1077            before_group_hot_bytes,
1078            after_group_hot_bytes,
1079            limit,
1080        ))
1081    }
1082
1083    pub fn raft_uncommitted_backpressure(current: u64, incoming: u64, limit: u64) -> Self {
1084        Self::Infra(GroupInfraError::raft_uncommitted_backpressure(
1085            current, incoming, limit,
1086        ))
1087    }
1088
1089    pub fn stream(code: StreamErrorCode, message: impl Into<String>) -> Self {
1090        Self::stream_with_next_offset(code, message, None)
1091    }
1092
1093    pub fn stream_with_next_offset(
1094        code: StreamErrorCode,
1095        message: impl Into<String>,
1096        next_offset: Option<u64>,
1097    ) -> Self {
1098        Self::stream_with_context(code, message, next_offset, vec![])
1099    }
1100
1101    pub fn stream_with_context(
1102        code: StreamErrorCode,
1103        message: impl Into<String>,
1104        next_offset: Option<u64>,
1105        context: Vec<StreamErrorContext>,
1106    ) -> Self {
1107        Self::Stream(StreamEngineError {
1108            message: format!("{code:?}: {}", message.into()),
1109            code,
1110            next_offset,
1111            context,
1112        })
1113    }
1114
1115    pub fn stream_from_replicated(
1116        message: impl Into<String>,
1117        code: StreamErrorCode,
1118        next_offset: Option<u64>,
1119        context: Vec<StreamErrorContext>,
1120    ) -> Self {
1121        Self::Stream(StreamEngineError {
1122            message: message.into(),
1123            code,
1124            next_offset,
1125            context,
1126        })
1127    }
1128
1129    pub fn forward_to_leader(
1130        message: impl Into<String>,
1131        node_id: Option<u64>,
1132        address: Option<String>,
1133    ) -> Self {
1134        Self::ForwardToLeader {
1135            message: message.into(),
1136            leader_hint: GroupLeaderHint { node_id, address },
1137        }
1138    }
1139
1140    pub fn message(&self) -> Cow<'_, str> {
1141        match self {
1142            Self::Stream(err) => Cow::Borrowed(&err.message),
1143            Self::Infra(err) => err.message(),
1144            Self::ForwardToLeader { message, .. } => Cow::Borrowed(message),
1145        }
1146    }
1147
1148    pub fn code(&self) -> Option<StreamErrorCode> {
1149        match self {
1150            Self::Stream(err) => Some(err.code),
1151            Self::Infra(_) | Self::ForwardToLeader { .. } => None,
1152        }
1153    }
1154
1155    pub fn stream_parts(
1156        &self,
1157    ) -> Option<(&str, StreamErrorCode, Option<u64>, &[StreamErrorContext])> {
1158        match self {
1159            Self::Stream(err) => Some((&err.message, err.code, err.next_offset, &err.context)),
1160            Self::Infra(_) | Self::ForwardToLeader { .. } => None,
1161        }
1162    }
1163
1164    pub fn next_offset(&self) -> Option<u64> {
1165        match self {
1166            Self::Stream(err) => err.next_offset,
1167            Self::Infra(_) | Self::ForwardToLeader { .. } => None,
1168        }
1169    }
1170
1171    pub fn context(&self) -> &[StreamErrorContext] {
1172        match self {
1173            Self::Stream(err) => &err.context,
1174            Self::Infra(_) | Self::ForwardToLeader { .. } => &[],
1175        }
1176    }
1177
1178    pub fn leader_hint(&self) -> Option<&GroupLeaderHint> {
1179        match self {
1180            Self::ForwardToLeader { leader_hint, .. } => Some(leader_hint),
1181            Self::Stream(_) | Self::Infra(_) => None,
1182        }
1183    }
1184
1185    pub fn infra(&self) -> Option<&GroupInfraError> {
1186        match self {
1187            Self::Infra(err) => Some(err),
1188            Self::Stream(_) | Self::ForwardToLeader { .. } => None,
1189        }
1190    }
1191
1192    pub fn is_cold_backpressure(&self) -> bool {
1193        self.infra()
1194            .is_some_and(GroupInfraError::is_cold_backpressure)
1195    }
1196
1197    pub fn is_backpressure(&self) -> bool {
1198        self.infra().is_some_and(GroupInfraError::is_backpressure)
1199    }
1200}
ursula_runtime/engine/mod.rs

ursula_runtime/engine/
mod.rs
