ursula_stream/
command.rs

1use std::fmt;
2
3use bytes::Bytes;
4use serde::Deserialize;
5use serde::Serialize;
6use ursula_shard::BucketStreamId;
7
8use crate::model::ColdChunkRef;
9use crate::model::ExternalPayloadRef;
10use crate::model::ProducerRequest;
11use crate::model::StreamAttrs;
12
13#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
14pub enum StreamCommand {
15    CreateBucket {
16        bucket_id: String,
17    },
18    DeleteBucket {
19        bucket_id: String,
20    },
21    CreateStream {
22        stream_id: BucketStreamId,
23        content_type: String,
24        initial_payload: Bytes,
25        close_after: bool,
26        stream_seq: Option<String>,
27        producer: Option<ProducerRequest>,
28        stream_ttl_seconds: Option<u64>,
29        stream_expires_at_ms: Option<u64>,
30        // `default` keeps pre-attrs replicated records decodable.
31        #[serde(default)]
32        attrs: Option<StreamAttrs>,
33        now_ms: u64,
34    },
35    CreateExternal {
36        stream_id: BucketStreamId,
37        content_type: String,
38        initial_payload: ExternalPayloadRef,
39        #[serde(default)]
40        record_ends: Vec<u64>,
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        // `default` keeps pre-attrs replicated records decodable.
47        #[serde(default)]
48        attrs: Option<StreamAttrs>,
49        now_ms: u64,
50    },
51    Append {
52        stream_id: BucketStreamId,
53        content_type: Option<String>,
54        payload: Bytes,
55        close_after: bool,
56        stream_seq: Option<String>,
57        producer: Option<ProducerRequest>,
58        now_ms: u64,
59        record_match: Option<u64>,
60    },
61    AppendExternal {
62        stream_id: BucketStreamId,
63        content_type: Option<String>,
64        payload: ExternalPayloadRef,
65        #[serde(default)]
66        record_ends: Vec<u64>,
67        close_after: bool,
68        stream_seq: Option<String>,
69        producer: Option<ProducerRequest>,
70        now_ms: u64,
71        record_match: Option<u64>,
72    },
73    AppendBatch {
74        stream_id: BucketStreamId,
75        content_type: Option<String>,
76        payloads: Vec<Bytes>,
77        producer: Option<ProducerRequest>,
78        now_ms: u64,
79    },
80    PublishSnapshot {
81        stream_id: BucketStreamId,
82        snapshot_offset: u64,
83        content_type: String,
84        payload: Bytes,
85        now_ms: u64,
86    },
87    TouchStreamAccess {
88        stream_id: BucketStreamId,
89        now_ms: u64,
90        renew_ttl: bool,
91    },
92    UpdateStreamAttrs {
93        stream_id: BucketStreamId,
94        attrs: Option<StreamAttrs>,
95        now_ms: u64,
96    },
97    FlushCold {
98        stream_id: BucketStreamId,
99        chunk: ColdChunkRef,
100    },
101    Close {
102        stream_id: BucketStreamId,
103        stream_seq: Option<String>,
104        producer: Option<ProducerRequest>,
105        now_ms: u64,
106    },
107    DeleteStream {
108        stream_id: BucketStreamId,
109    },
110    /// Confirms the leader's background worker has physically reclaimed every
111    /// queued cold-GC entry with `seq <= up_to_seq`; removes them from the
112    /// replicated queue. Idempotent under replay.
113    AckColdGc {
114        up_to_seq: u64,
115    },
116}
117
118impl fmt::Display for StreamCommand {
119    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
120        match self {
121            Self::CreateBucket { bucket_id } => write!(f, "create_bucket:{bucket_id}"),
122            Self::DeleteBucket { bucket_id } => write!(f, "delete_bucket:{bucket_id}"),
123            Self::CreateStream { stream_id, .. } => write!(f, "create_stream:{stream_id}"),
124            Self::CreateExternal {
125                stream_id,
126                initial_payload,
127                ..
128            } => write!(
129                f,
130                "create_external:{stream_id}:{} bytes",
131                initial_payload.payload_len
132            ),
133            Self::Append {
134                stream_id, payload, ..
135            } => write!(f, "append:{stream_id}:{} bytes", payload.len()),
136            Self::AppendExternal {
137                stream_id, payload, ..
138            } => write!(
139                f,
140                "append_external:{stream_id}:{} bytes",
141                payload.payload_len
142            ),
143            Self::AppendBatch {
144                stream_id,
145                payloads,
146                ..
147            } => write!(f, "append_batch:{stream_id}:{} items", payloads.len()),
148            Self::PublishSnapshot {
149                stream_id,
150                snapshot_offset,
151                payload,
152                ..
153            } => write!(
154                f,
155                "publish_snapshot:{stream_id}:{snapshot_offset}:{} bytes",
156                payload.len()
157            ),
158            Self::TouchStreamAccess {
159                stream_id,
160                renew_ttl,
161                ..
162            } => write!(f, "touch_stream_access:{stream_id}:renew_ttl={renew_ttl}"),
163            Self::UpdateStreamAttrs { stream_id, .. } => {
164                write!(f, "update_stream_attrs:{stream_id}")
165            }
166            Self::FlushCold { stream_id, chunk } => write!(
167                f,
168                "flush_cold:{stream_id}:{}..{}",
169                chunk.start_offset, chunk.end_offset
170            ),
171            Self::Close { stream_id, .. } => write!(f, "close_stream:{stream_id}"),
172            Self::DeleteStream { stream_id } => write!(f, "delete_stream:{stream_id}"),
173            Self::AckColdGc { up_to_seq } => write!(f, "ack_cold_gc:up_to_seq={up_to_seq}"),
174        }
175    }
176}
ursula_stream/command.rs

ursula_stream/
command.rs
