ursula_runtime/engine/
mod.rs

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

ursula_runtime/engine/
mod.rs
