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

ursula_runtime/engine/
mod.rs
