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_wal_storage(
894        &self,
895        placement: ShardPlacement,
896        fsyncs: u64,
897        fsync_records: u64,
898        reclaims: u64,
899        reclaimed_bytes: u64,
900        reclaim_ns: u64,
901        physical_bytes: u64,
902    ) {
903        self.inner.record_wal_storage(
904            placement.core_id,
905            placement.raft_group_id,
906            fsyncs,
907            fsync_records,
908            reclaims,
909            reclaimed_bytes,
910            reclaim_ns,
911            physical_bytes,
912        );
913    }
914
915    pub fn record_wal_recovery(
916        &self,
917        placement: ShardPlacement,
918        recovery_ns: u64,
919        records: u64,
920        bytes: u64,
921        live_entries: u64,
922    ) {
923        self.inner.record_wal_recovery(
924            placement.core_id,
925            recovery_ns,
926            records,
927            bytes,
928            live_entries,
929        );
930    }
931
932    pub fn record_raft_write_many(
933        &self,
934        placement: ShardPlacement,
935        command_count: usize,
936        logical_command_count: usize,
937        response_count: usize,
938        submit_ns: u64,
939        response_ns: u64,
940    ) {
941        self.inner.record_raft_write_many(
942            placement.core_id,
943            placement.raft_group_id,
944            RaftWriteManySample {
945                command_count: u64::try_from(command_count).expect("command count fits u64"),
946                logical_command_count: u64::try_from(logical_command_count)
947                    .expect("logical command count fits u64"),
948                response_count: u64::try_from(response_count).expect("response count fits u64"),
949                submit_ns,
950                response_ns,
951            },
952        );
953    }
954
955    pub fn record_raft_apply_batch(
956        &self,
957        placement: ShardPlacement,
958        entry_count: usize,
959        apply_ns: u64,
960    ) {
961        self.inner.record_raft_apply_batch(
962            placement.core_id,
963            placement.raft_group_id,
964            u64::try_from(entry_count).expect("entry count fits u64"),
965            apply_ns,
966        );
967    }
968
969    pub fn record_raft_snapshot_build(
970        &self,
971        placement: ShardPlacement,
972        stream_count: usize,
973        body_bytes: usize,
974        pointer_bytes: usize,
975        build_ns: u64,
976        external_upload: bool,
977        inline_fallback: bool,
978    ) {
979        self.inner
980            .record_raft_snapshot_build(placement.raft_group_id, RaftSnapshotBuildSample {
981                streams: u64::try_from(stream_count).expect("stream count fits u64"),
982                body_bytes: u64::try_from(body_bytes).expect("snapshot body bytes fits u64"),
983                pointer_bytes: u64::try_from(pointer_bytes)
984                    .expect("snapshot pointer bytes fits u64"),
985                build_ns,
986                external_upload,
987                inline_fallback,
988            });
989    }
990}
991
992#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
993pub struct GroupLeaderHint {
994    pub node_id: Option<u64>,
995    pub address: Option<String>,
996}
997
998#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
999pub struct StreamEngineError {
1000    message: String,
1001    code: StreamErrorCode,
1002    next_offset: Option<u64>,
1003    #[serde(default, skip_serializing_if = "Vec::is_empty")]
1004    context: Vec<StreamErrorContext>,
1005}
1006
1007/// Infra error variants with structured fields render their human message on
1008/// demand (`message`) instead of storing a denormalized copy alongside the
1009/// fields. `Internal` is the exception: it carries free-form text with no
1010/// structured source, so it keeps an owned `message`.
1011#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, thiserror::Error)]
1012pub enum GroupInfraError {
1013    #[error("{message}")]
1014    Internal { message: String },
1015    #[error("ProtoDecode: protobuf raft payload missing {field}")]
1016    ProtoDecode { field: String },
1017    #[error(
1018        "ColdBackpressure: stream '{stream_id}' would raise group hot bytes from {before_group_hot_bytes} to {after_group_hot_bytes}, above limit {limit}"
1019    )]
1020    ColdBackpressure {
1021        stream_id: BucketStreamId,
1022        before_group_hot_bytes: u64,
1023        after_group_hot_bytes: u64,
1024        limit: u64,
1025    },
1026    #[error(
1027        "RaftUncommittedBackpressure: group uncommitted bytes {current} plus incoming {incoming} would exceed limit {limit}"
1028    )]
1029    RaftUncommittedBackpressure {
1030        current: u64,
1031        incoming: u64,
1032        limit: u64,
1033    },
1034}
1035
1036impl GroupInfraError {
1037    pub fn internal(message: impl Into<String>) -> Self {
1038        Self::Internal {
1039            message: message.into(),
1040        }
1041    }
1042
1043    pub fn proto_decode(field: impl Into<String>) -> Self {
1044        Self::ProtoDecode {
1045            field: field.into(),
1046        }
1047    }
1048
1049    pub fn cold_backpressure(
1050        stream_id: BucketStreamId,
1051        before_group_hot_bytes: u64,
1052        after_group_hot_bytes: u64,
1053        limit: u64,
1054    ) -> Self {
1055        Self::ColdBackpressure {
1056            stream_id,
1057            before_group_hot_bytes,
1058            after_group_hot_bytes,
1059            limit,
1060        }
1061    }
1062
1063    pub fn raft_uncommitted_backpressure(current: u64, incoming: u64, limit: u64) -> Self {
1064        Self::RaftUncommittedBackpressure {
1065            current,
1066            incoming,
1067            limit,
1068        }
1069    }
1070
1071    pub fn message(&self) -> Cow<'_, str> {
1072        match self {
1073            Self::Internal { message } => Cow::Borrowed(message),
1074            other => Cow::Owned(other.to_string()),
1075        }
1076    }
1077
1078    pub fn is_cold_backpressure(&self) -> bool {
1079        matches!(self, Self::ColdBackpressure { .. })
1080    }
1081
1082    pub fn is_backpressure(&self) -> bool {
1083        matches!(
1084            self,
1085            Self::ColdBackpressure { .. } | Self::RaftUncommittedBackpressure { .. }
1086        )
1087    }
1088}
1089
1090#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, thiserror::Error)]
1091pub enum GroupEngineError {
1092    #[error("{}", .0.message)]
1093    Stream(StreamEngineError),
1094    #[error("{}", .0.message())]
1095    Infra(GroupInfraError),
1096    #[error("{message}")]
1097    ForwardToLeader {
1098        message: String,
1099        leader_hint: GroupLeaderHint,
1100    },
1101}
1102
1103impl GroupEngineError {
1104    pub fn new(message: impl Into<String>) -> Self {
1105        Self::Infra(GroupInfraError::internal(message))
1106    }
1107
1108    pub fn cold_backpressure(
1109        stream_id: BucketStreamId,
1110        before_group_hot_bytes: u64,
1111        after_group_hot_bytes: u64,
1112        limit: u64,
1113    ) -> Self {
1114        Self::Infra(GroupInfraError::cold_backpressure(
1115            stream_id,
1116            before_group_hot_bytes,
1117            after_group_hot_bytes,
1118            limit,
1119        ))
1120    }
1121
1122    pub fn raft_uncommitted_backpressure(current: u64, incoming: u64, limit: u64) -> Self {
1123        Self::Infra(GroupInfraError::raft_uncommitted_backpressure(
1124            current, incoming, limit,
1125        ))
1126    }
1127
1128    pub fn stream(code: StreamErrorCode, message: impl Into<String>) -> Self {
1129        Self::stream_with_next_offset(code, message, None)
1130    }
1131
1132    pub fn stream_with_next_offset(
1133        code: StreamErrorCode,
1134        message: impl Into<String>,
1135        next_offset: Option<u64>,
1136    ) -> Self {
1137        Self::stream_with_context(code, message, next_offset, vec![])
1138    }
1139
1140    pub fn stream_with_context(
1141        code: StreamErrorCode,
1142        message: impl Into<String>,
1143        next_offset: Option<u64>,
1144        context: Vec<StreamErrorContext>,
1145    ) -> Self {
1146        Self::Stream(StreamEngineError {
1147            message: format!("{code:?}: {}", message.into()),
1148            code,
1149            next_offset,
1150            context,
1151        })
1152    }
1153
1154    pub fn stream_from_replicated(
1155        message: impl Into<String>,
1156        code: StreamErrorCode,
1157        next_offset: Option<u64>,
1158        context: Vec<StreamErrorContext>,
1159    ) -> Self {
1160        Self::Stream(StreamEngineError {
1161            message: message.into(),
1162            code,
1163            next_offset,
1164            context,
1165        })
1166    }
1167
1168    pub fn forward_to_leader(
1169        message: impl Into<String>,
1170        node_id: Option<u64>,
1171        address: Option<String>,
1172    ) -> Self {
1173        Self::ForwardToLeader {
1174            message: message.into(),
1175            leader_hint: GroupLeaderHint { node_id, address },
1176        }
1177    }
1178
1179    pub fn message(&self) -> Cow<'_, str> {
1180        match self {
1181            Self::Stream(err) => Cow::Borrowed(&err.message),
1182            Self::Infra(err) => err.message(),
1183            Self::ForwardToLeader { message, .. } => Cow::Borrowed(message),
1184        }
1185    }
1186
1187    pub fn code(&self) -> Option<StreamErrorCode> {
1188        match self {
1189            Self::Stream(err) => Some(err.code),
1190            Self::Infra(_) | Self::ForwardToLeader { .. } => None,
1191        }
1192    }
1193
1194    pub fn stream_parts(
1195        &self,
1196    ) -> Option<(&str, StreamErrorCode, Option<u64>, &[StreamErrorContext])> {
1197        match self {
1198            Self::Stream(err) => Some((&err.message, err.code, err.next_offset, &err.context)),
1199            Self::Infra(_) | Self::ForwardToLeader { .. } => None,
1200        }
1201    }
1202
1203    pub fn next_offset(&self) -> Option<u64> {
1204        match self {
1205            Self::Stream(err) => err.next_offset,
1206            Self::Infra(_) | Self::ForwardToLeader { .. } => None,
1207        }
1208    }
1209
1210    pub fn context(&self) -> &[StreamErrorContext] {
1211        match self {
1212            Self::Stream(err) => &err.context,
1213            Self::Infra(_) | Self::ForwardToLeader { .. } => &[],
1214        }
1215    }
1216
1217    pub fn leader_hint(&self) -> Option<&GroupLeaderHint> {
1218        match self {
1219            Self::ForwardToLeader { leader_hint, .. } => Some(leader_hint),
1220            Self::Stream(_) | Self::Infra(_) => None,
1221        }
1222    }
1223
1224    pub fn infra(&self) -> Option<&GroupInfraError> {
1225        match self {
1226            Self::Infra(err) => Some(err),
1227            Self::Stream(_) | Self::ForwardToLeader { .. } => None,
1228        }
1229    }
1230
1231    pub fn is_cold_backpressure(&self) -> bool {
1232        self.infra()
1233            .is_some_and(GroupInfraError::is_cold_backpressure)
1234    }
1235
1236    pub fn is_backpressure(&self) -> bool {
1237        self.infra().is_some_and(GroupInfraError::is_backpressure)
1238    }
1239}
ursula_runtime/engine/mod.rs

ursula_runtime/engine/
mod.rs
