ursula_runtime/
command.rs

1use std::fmt;
2
3use serde::Deserialize;
4use serde::Serialize;
5use ursula_shard::ShardPlacement;
6use ursula_stream::StreamCommand;
7use ursula_stream::StreamSnapshot;
8
9use crate::request::AppendBatchRequest;
10use crate::request::AppendExternalRequest;
11use crate::request::AppendRequest;
12use crate::request::CloseStreamRequest;
13use crate::request::CreateStreamExternalRequest;
14use crate::request::CreateStreamRequest;
15use crate::request::DeleteStreamRequest;
16use crate::request::FlushColdRequest;
17use crate::request::PublishSnapshotRequest;
18use crate::request::StreamAppendCount;
19use crate::request::UpdateStreamAttrsRequest;
20
21#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
22pub struct GroupSnapshot {
23    pub placement: ShardPlacement,
24    pub group_commit_index: u64,
25    pub stream_snapshot: StreamSnapshot,
26    pub stream_append_counts: Vec<StreamAppendCount>,
27}
28
29/// Replicated group-level write envelope around the canonical
30/// [`StreamCommand`]: either one per-stream command, or an atomic batch of
31/// them applied as a single raft entry. This enum (serde-encoded) is the raft
32/// log payload; there is no separate wire mirror.
33#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
34#[expect(
35    clippy::large_enum_variant,
36    reason = "commands are transient write-path values; boxing `Stream` would \
37              add a heap allocation to every replicated write"
38)]
39pub enum GroupWriteCommand {
40    Stream(StreamCommand),
41    Batch { commands: Vec<StreamCommand> },
42}
43
44impl From<StreamCommand> for GroupWriteCommand {
45    fn from(command: StreamCommand) -> Self {
46        Self::Stream(command)
47    }
48}
49
50impl From<CreateStreamRequest> for StreamCommand {
51    fn from(request: CreateStreamRequest) -> Self {
52        Self::CreateStream {
53            stream_id: request.stream_id,
54            content_type: request.content_type,
55            initial_payload: request.initial_payload,
56            close_after: request.close_after,
57            stream_seq: request.stream_seq,
58            producer: request.producer,
59            stream_ttl_seconds: request.stream_ttl_seconds,
60            stream_expires_at_ms: request.stream_expires_at_ms,
61            attrs: request.attrs,
62            now_ms: request.now_ms,
63        }
64    }
65}
66
67impl From<CreateStreamExternalRequest> for StreamCommand {
68    fn from(request: CreateStreamExternalRequest) -> Self {
69        Self::CreateExternal {
70            stream_id: request.stream_id,
71            content_type: request.content_type,
72            initial_payload: request.initial_payload,
73            record_ends: request.record_ends,
74            close_after: request.close_after,
75            stream_seq: request.stream_seq,
76            producer: request.producer,
77            stream_ttl_seconds: request.stream_ttl_seconds,
78            stream_expires_at_ms: request.stream_expires_at_ms,
79            attrs: request.attrs,
80            now_ms: request.now_ms,
81        }
82    }
83}
84
85impl From<UpdateStreamAttrsRequest> for StreamCommand {
86    fn from(request: UpdateStreamAttrsRequest) -> Self {
87        Self::UpdateStreamAttrs {
88            stream_id: request.stream_id,
89            attrs: request.attrs,
90            now_ms: request.now_ms,
91        }
92    }
93}
94
95impl From<AppendRequest> for StreamCommand {
96    fn from(request: AppendRequest) -> Self {
97        Self::Append {
98            stream_id: request.stream_id,
99            content_type: Some(request.content_type),
100            payload: request.payload,
101            close_after: request.close_after,
102            stream_seq: request.stream_seq,
103            producer: request.producer,
104            now_ms: request.now_ms,
105            record_match: request.record_match,
106        }
107    }
108}
109
110impl From<AppendExternalRequest> for StreamCommand {
111    fn from(request: AppendExternalRequest) -> Self {
112        Self::AppendExternal {
113            stream_id: request.stream_id,
114            content_type: Some(request.content_type),
115            payload: request.payload,
116            record_ends: request.record_ends,
117            close_after: request.close_after,
118            stream_seq: request.stream_seq,
119            producer: request.producer,
120            now_ms: request.now_ms,
121            record_match: request.record_match,
122        }
123    }
124}
125
126impl From<AppendBatchRequest> for StreamCommand {
127    fn from(request: AppendBatchRequest) -> Self {
128        Self::AppendBatch {
129            stream_id: request.stream_id,
130            content_type: Some(request.content_type),
131            payloads: request.payloads,
132            producer: request.producer,
133            now_ms: request.now_ms,
134        }
135    }
136}
137
138impl From<PublishSnapshotRequest> for StreamCommand {
139    fn from(request: PublishSnapshotRequest) -> Self {
140        Self::PublishSnapshot {
141            stream_id: request.stream_id,
142            snapshot_offset: request.snapshot_offset,
143            content_type: request.content_type,
144            payload: request.payload,
145            now_ms: request.now_ms,
146        }
147    }
148}
149
150impl From<CloseStreamRequest> for StreamCommand {
151    fn from(request: CloseStreamRequest) -> Self {
152        Self::Close {
153            stream_id: request.stream_id,
154            stream_seq: request.stream_seq,
155            producer: request.producer,
156            now_ms: request.now_ms,
157        }
158    }
159}
160
161impl From<DeleteStreamRequest> for StreamCommand {
162    fn from(request: DeleteStreamRequest) -> Self {
163        Self::DeleteStream {
164            stream_id: request.stream_id,
165        }
166    }
167}
168
169impl From<FlushColdRequest> for StreamCommand {
170    fn from(request: FlushColdRequest) -> Self {
171        Self::FlushCold {
172            stream_id: request.stream_id,
173            chunk: request.chunk,
174        }
175    }
176}
177
178macro_rules! group_write_from_request {
179    ($($request:ty),+ $(,)?) => {
180        $(impl From<$request> for GroupWriteCommand {
181            fn from(request: $request) -> Self {
182                Self::Stream(StreamCommand::from(request))
183            }
184        })+
185    };
186}
187
188group_write_from_request!(
189    CreateStreamRequest,
190    CreateStreamExternalRequest,
191    UpdateStreamAttrsRequest,
192    AppendRequest,
193    AppendExternalRequest,
194    AppendBatchRequest,
195    PublishSnapshotRequest,
196    CloseStreamRequest,
197    DeleteStreamRequest,
198    FlushColdRequest,
199);
200
201impl fmt::Display for GroupWriteCommand {
202    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
203        match self {
204            Self::Stream(command) => command.fmt(f),
205            Self::Batch { commands } => write!(f, "batch:{} commands", commands.len()),
206        }
207    }
208}
ursula_runtime/command.rs

ursula_runtime/
command.rs
