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

ursula_runtime/engine/
mod.rs
