ursula_runtime/
command.rs

1use std::fmt;
2
3use bytes::Bytes;
4use serde::Deserialize;
5use serde::Serialize;
6use ursula_shard::BucketStreamId;
7use ursula_shard::ShardPlacement;
8use ursula_stream::ColdChunkRef;
9use ursula_stream::ExternalPayloadRef;
10use ursula_stream::ProducerRequest;
11use ursula_stream::StreamAttrs;
12use ursula_stream::StreamSnapshot;
13
14use crate::request::AppendBatchRequest;
15use crate::request::AppendExternalRequest;
16use crate::request::AppendRequest;
17use crate::request::CloseStreamRequest;
18use crate::request::CreateStreamExternalRequest;
19use crate::request::CreateStreamRequest;
20use crate::request::DeleteStreamRequest;
21use crate::request::FlushColdRequest;
22use crate::request::PublishSnapshotRequest;
23use crate::request::StreamAppendCount;
24use crate::request::UpdateStreamAttrsRequest;
25
26#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
27pub struct GroupSnapshot {
28    pub placement: ShardPlacement,
29    pub group_commit_index: u64,
30    pub stream_snapshot: StreamSnapshot,
31    pub stream_append_counts: Vec<StreamAppendCount>,
32}
33
34#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
35#[serde(tag = "command", rename_all = "snake_case")]
36pub enum GroupWriteCommand {
37    CreateStream {
38        stream_id: BucketStreamId,
39        content_type: String,
40        initial_payload: Bytes,
41        close_after: bool,
42        stream_seq: Option<String>,
43        producer: Option<ProducerRequest>,
44        stream_ttl_seconds: Option<u64>,
45        stream_expires_at_ms: Option<u64>,
46        forked_from: Option<BucketStreamId>,
47        fork_offset: Option<u64>,
48        // `default` keeps pre-attrs WAL records decodable.
49        #[serde(default)]
50        attrs: Option<StreamAttrs>,
51        now_ms: u64,
52    },
53    CreateExternal {
54        stream_id: BucketStreamId,
55        content_type: String,
56        initial_payload: ExternalPayloadRef,
57        close_after: bool,
58        stream_seq: Option<String>,
59        producer: Option<ProducerRequest>,
60        stream_ttl_seconds: Option<u64>,
61        stream_expires_at_ms: Option<u64>,
62        forked_from: Option<BucketStreamId>,
63        fork_offset: Option<u64>,
64        // `default` keeps pre-attrs WAL records decodable.
65        #[serde(default)]
66        attrs: Option<StreamAttrs>,
67        now_ms: u64,
68    },
69    Append {
70        stream_id: BucketStreamId,
71        content_type: String,
72        payload: Bytes,
73        close_after: bool,
74        stream_seq: Option<String>,
75        producer: Option<ProducerRequest>,
76        now_ms: u64,
77    },
78    AppendExternal {
79        stream_id: BucketStreamId,
80        content_type: String,
81        payload: ExternalPayloadRef,
82        close_after: bool,
83        stream_seq: Option<String>,
84        producer: Option<ProducerRequest>,
85        now_ms: u64,
86    },
87    AppendBatch {
88        stream_id: BucketStreamId,
89        content_type: String,
90        payloads: Vec<Bytes>,
91        producer: Option<ProducerRequest>,
92        now_ms: u64,
93    },
94    PublishSnapshot {
95        stream_id: BucketStreamId,
96        snapshot_offset: u64,
97        content_type: String,
98        payload: Bytes,
99        now_ms: u64,
100    },
101    TouchStreamAccess {
102        stream_id: BucketStreamId,
103        now_ms: u64,
104        renew_ttl: bool,
105    },
106    UpdateStreamAttrs {
107        stream_id: BucketStreamId,
108        attrs: Option<StreamAttrs>,
109        now_ms: u64,
110    },
111    AddForkRef {
112        stream_id: BucketStreamId,
113        now_ms: u64,
114    },
115    ReleaseForkRef {
116        stream_id: BucketStreamId,
117    },
118    FlushCold {
119        stream_id: BucketStreamId,
120        chunk: ColdChunkRef,
121    },
122    CloseStream {
123        stream_id: BucketStreamId,
124        stream_seq: Option<String>,
125        producer: Option<ProducerRequest>,
126        now_ms: u64,
127    },
128    DeleteStream {
129        stream_id: BucketStreamId,
130    },
131    AckColdGc {
132        up_to_seq: u64,
133    },
134    Batch {
135        commands: Vec<GroupWriteCommand>,
136    },
137}
138
139impl From<CreateStreamRequest> for GroupWriteCommand {
140    fn from(request: CreateStreamRequest) -> Self {
141        Self::CreateStream {
142            stream_id: request.stream_id,
143            content_type: request.content_type,
144            initial_payload: request.initial_payload,
145            close_after: request.close_after,
146            stream_seq: request.stream_seq,
147            producer: request.producer,
148            stream_ttl_seconds: request.stream_ttl_seconds,
149            stream_expires_at_ms: request.stream_expires_at_ms,
150            forked_from: request.forked_from,
151            fork_offset: request.fork_offset,
152            attrs: request.attrs,
153            now_ms: request.now_ms,
154        }
155    }
156}
157
158impl From<&CreateStreamRequest> for GroupWriteCommand {
159    fn from(request: &CreateStreamRequest) -> Self {
160        Self::CreateStream {
161            stream_id: request.stream_id.clone(),
162            content_type: request.content_type.clone(),
163            initial_payload: request.initial_payload.clone(),
164            close_after: request.close_after,
165            stream_seq: request.stream_seq.clone(),
166            producer: request.producer.clone(),
167            stream_ttl_seconds: request.stream_ttl_seconds,
168            stream_expires_at_ms: request.stream_expires_at_ms,
169            forked_from: request.forked_from.clone(),
170            fork_offset: request.fork_offset,
171            attrs: request.attrs.clone(),
172            now_ms: request.now_ms,
173        }
174    }
175}
176
177impl From<CreateStreamExternalRequest> for GroupWriteCommand {
178    fn from(request: CreateStreamExternalRequest) -> Self {
179        Self::CreateExternal {
180            stream_id: request.stream_id,
181            content_type: request.content_type,
182            initial_payload: request.initial_payload,
183            close_after: request.close_after,
184            stream_seq: request.stream_seq,
185            producer: request.producer,
186            stream_ttl_seconds: request.stream_ttl_seconds,
187            stream_expires_at_ms: request.stream_expires_at_ms,
188            forked_from: request.forked_from,
189            fork_offset: request.fork_offset,
190            attrs: request.attrs,
191            now_ms: request.now_ms,
192        }
193    }
194}
195
196impl From<&CreateStreamExternalRequest> for GroupWriteCommand {
197    fn from(request: &CreateStreamExternalRequest) -> Self {
198        Self::CreateExternal {
199            stream_id: request.stream_id.clone(),
200            content_type: request.content_type.clone(),
201            initial_payload: request.initial_payload.clone(),
202            close_after: request.close_after,
203            stream_seq: request.stream_seq.clone(),
204            producer: request.producer.clone(),
205            stream_ttl_seconds: request.stream_ttl_seconds,
206            stream_expires_at_ms: request.stream_expires_at_ms,
207            forked_from: request.forked_from.clone(),
208            fork_offset: request.fork_offset,
209            attrs: request.attrs.clone(),
210            now_ms: request.now_ms,
211        }
212    }
213}
214
215impl From<UpdateStreamAttrsRequest> for GroupWriteCommand {
216    fn from(request: UpdateStreamAttrsRequest) -> Self {
217        Self::UpdateStreamAttrs {
218            stream_id: request.stream_id,
219            attrs: request.attrs,
220            now_ms: request.now_ms,
221        }
222    }
223}
224
225impl From<&UpdateStreamAttrsRequest> for GroupWriteCommand {
226    fn from(request: &UpdateStreamAttrsRequest) -> Self {
227        Self::UpdateStreamAttrs {
228            stream_id: request.stream_id.clone(),
229            attrs: request.attrs.clone(),
230            now_ms: request.now_ms,
231        }
232    }
233}
234
235impl From<AppendRequest> for GroupWriteCommand {
236    fn from(request: AppendRequest) -> Self {
237        Self::Append {
238            stream_id: request.stream_id,
239            content_type: request.content_type,
240            payload: request.payload,
241            close_after: request.close_after,
242            stream_seq: request.stream_seq,
243            producer: request.producer,
244            now_ms: request.now_ms,
245        }
246    }
247}
248
249impl From<&AppendRequest> for GroupWriteCommand {
250    fn from(request: &AppendRequest) -> Self {
251        Self::Append {
252            stream_id: request.stream_id.clone(),
253            content_type: request.content_type.clone(),
254            payload: request.payload.clone(),
255            close_after: request.close_after,
256            stream_seq: request.stream_seq.clone(),
257            producer: request.producer.clone(),
258            now_ms: request.now_ms,
259        }
260    }
261}
262
263impl From<AppendExternalRequest> for GroupWriteCommand {
264    fn from(request: AppendExternalRequest) -> Self {
265        Self::AppendExternal {
266            stream_id: request.stream_id,
267            content_type: request.content_type,
268            payload: request.payload,
269            close_after: request.close_after,
270            stream_seq: request.stream_seq,
271            producer: request.producer,
272            now_ms: request.now_ms,
273        }
274    }
275}
276
277impl From<&AppendExternalRequest> for GroupWriteCommand {
278    fn from(request: &AppendExternalRequest) -> Self {
279        Self::AppendExternal {
280            stream_id: request.stream_id.clone(),
281            content_type: request.content_type.clone(),
282            payload: request.payload.clone(),
283            close_after: request.close_after,
284            stream_seq: request.stream_seq.clone(),
285            producer: request.producer.clone(),
286            now_ms: request.now_ms,
287        }
288    }
289}
290
291impl From<AppendBatchRequest> for GroupWriteCommand {
292    fn from(request: AppendBatchRequest) -> Self {
293        Self::AppendBatch {
294            stream_id: request.stream_id,
295            content_type: request.content_type,
296            payloads: request.payloads,
297            producer: request.producer,
298            now_ms: request.now_ms,
299        }
300    }
301}
302
303impl From<&AppendBatchRequest> for GroupWriteCommand {
304    fn from(request: &AppendBatchRequest) -> Self {
305        Self::AppendBatch {
306            stream_id: request.stream_id.clone(),
307            content_type: request.content_type.clone(),
308            payloads: request.payloads.clone(),
309            producer: request.producer.clone(),
310            now_ms: request.now_ms,
311        }
312    }
313}
314
315impl From<PublishSnapshotRequest> for GroupWriteCommand {
316    fn from(request: PublishSnapshotRequest) -> Self {
317        Self::PublishSnapshot {
318            stream_id: request.stream_id,
319            snapshot_offset: request.snapshot_offset,
320            content_type: request.content_type,
321            payload: request.payload,
322            now_ms: request.now_ms,
323        }
324    }
325}
326
327impl From<&PublishSnapshotRequest> for GroupWriteCommand {
328    fn from(request: &PublishSnapshotRequest) -> Self {
329        Self::PublishSnapshot {
330            stream_id: request.stream_id.clone(),
331            snapshot_offset: request.snapshot_offset,
332            content_type: request.content_type.clone(),
333            payload: request.payload.clone(),
334            now_ms: request.now_ms,
335        }
336    }
337}
338
339impl From<CloseStreamRequest> for GroupWriteCommand {
340    fn from(request: CloseStreamRequest) -> Self {
341        Self::CloseStream {
342            stream_id: request.stream_id,
343            stream_seq: request.stream_seq,
344            producer: request.producer,
345            now_ms: request.now_ms,
346        }
347    }
348}
349
350impl From<&CloseStreamRequest> for GroupWriteCommand {
351    fn from(request: &CloseStreamRequest) -> Self {
352        Self::CloseStream {
353            stream_id: request.stream_id.clone(),
354            stream_seq: request.stream_seq.clone(),
355            producer: request.producer.clone(),
356            now_ms: request.now_ms,
357        }
358    }
359}
360
361impl From<DeleteStreamRequest> for GroupWriteCommand {
362    fn from(request: DeleteStreamRequest) -> Self {
363        Self::DeleteStream {
364            stream_id: request.stream_id,
365        }
366    }
367}
368
369impl From<&DeleteStreamRequest> for GroupWriteCommand {
370    fn from(request: &DeleteStreamRequest) -> Self {
371        Self::DeleteStream {
372            stream_id: request.stream_id.clone(),
373        }
374    }
375}
376
377impl From<FlushColdRequest> for GroupWriteCommand {
378    fn from(request: FlushColdRequest) -> Self {
379        Self::FlushCold {
380            stream_id: request.stream_id,
381            chunk: request.chunk,
382        }
383    }
384}
385
386impl From<&FlushColdRequest> for GroupWriteCommand {
387    fn from(request: &FlushColdRequest) -> Self {
388        Self::FlushCold {
389            stream_id: request.stream_id.clone(),
390            chunk: request.chunk.clone(),
391        }
392    }
393}
394
395impl fmt::Display for GroupWriteCommand {
396    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
397        match self {
398            Self::CreateStream { stream_id, .. } => {
399                write!(f, "create_stream:{stream_id}")
400            }
401            Self::CreateExternal {
402                stream_id,
403                initial_payload,
404                ..
405            } => {
406                write!(
407                    f,
408                    "create_external:{stream_id}:{} bytes",
409                    initial_payload.payload_len
410                )
411            }
412            Self::Append {
413                stream_id, payload, ..
414            } => {
415                write!(f, "append:{stream_id}:{} bytes", payload.len())
416            }
417            Self::AppendExternal {
418                stream_id, payload, ..
419            } => {
420                write!(
421                    f,
422                    "append_external:{stream_id}:{} bytes",
423                    payload.payload_len
424                )
425            }
426            Self::AppendBatch {
427                stream_id,
428                payloads,
429                ..
430            } => {
431                write!(f, "append_batch:{stream_id}:{} items", payloads.len())
432            }
433            Self::PublishSnapshot {
434                stream_id,
435                snapshot_offset,
436                payload,
437                ..
438            } => {
439                write!(
440                    f,
441                    "publish_snapshot:{stream_id}:{snapshot_offset}:{} bytes",
442                    payload.len()
443                )
444            }
445            Self::TouchStreamAccess {
446                stream_id,
447                renew_ttl,
448                ..
449            } => {
450                write!(f, "touch_stream_access:{stream_id}:renew_ttl={renew_ttl}")
451            }
452            Self::UpdateStreamAttrs { stream_id, .. } => {
453                write!(f, "update_stream_attrs:{stream_id}")
454            }
455            Self::AddForkRef { stream_id, .. } => {
456                write!(f, "add_fork_ref:{stream_id}")
457            }
458            Self::ReleaseForkRef { stream_id } => {
459                write!(f, "release_fork_ref:{stream_id}")
460            }
461            Self::FlushCold { stream_id, chunk } => {
462                write!(
463                    f,
464                    "flush_cold:{stream_id}:{}..{}",
465                    chunk.start_offset, chunk.end_offset
466                )
467            }
468            Self::CloseStream { stream_id, .. } => {
469                write!(f, "close_stream:{stream_id}")
470            }
471            Self::DeleteStream { stream_id } => {
472                write!(f, "delete_stream:{stream_id}")
473            }
474            Self::AckColdGc { up_to_seq } => {
475                write!(f, "ack_cold_gc:up_to_seq={up_to_seq}")
476            }
477            Self::Batch { commands } => {
478                write!(f, "batch:{} commands", commands.len())
479            }
480        }
481    }
482}
ursula_runtime/command.rs

ursula_runtime/
command.rs
