ursula_runtime/engine/
mod.rs

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

ursula_runtime/engine/
mod.rs
