ursula_stream/
command.rs

1use serde::Deserialize;
2use serde::Serialize;
3use ursula_shard::BucketStreamId;
4
5use crate::model::ColdChunkRef;
6use crate::model::ExternalPayloadRef;
7use crate::model::ProducerRequest;
8use crate::model::StreamAttrs;
9
10#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
11pub enum StreamCommand {
12    CreateBucket {
13        bucket_id: String,
14    },
15    DeleteBucket {
16        bucket_id: String,
17    },
18    CreateStream {
19        stream_id: BucketStreamId,
20        content_type: String,
21        initial_payload: Vec<u8>,
22        close_after: bool,
23        stream_seq: Option<String>,
24        producer: Option<ProducerRequest>,
25        stream_ttl_seconds: Option<u64>,
26        stream_expires_at_ms: Option<u64>,
27        forked_from: Option<BucketStreamId>,
28        fork_offset: Option<u64>,
29        // `default` keeps pre-attrs WAL records decodable.
30        #[serde(default)]
31        attrs: Option<StreamAttrs>,
32        now_ms: u64,
33    },
34    CreateExternal {
35        stream_id: BucketStreamId,
36        content_type: String,
37        initial_payload: ExternalPayloadRef,
38        close_after: bool,
39        stream_seq: Option<String>,
40        producer: Option<ProducerRequest>,
41        stream_ttl_seconds: Option<u64>,
42        stream_expires_at_ms: Option<u64>,
43        forked_from: Option<BucketStreamId>,
44        fork_offset: Option<u64>,
45        // `default` keeps pre-attrs WAL records decodable.
46        #[serde(default)]
47        attrs: Option<StreamAttrs>,
48        now_ms: u64,
49    },
50    Append {
51        stream_id: BucketStreamId,
52        content_type: Option<String>,
53        payload: Vec<u8>,
54        close_after: bool,
55        stream_seq: Option<String>,
56        producer: Option<ProducerRequest>,
57        now_ms: u64,
58    },
59    AppendExternal {
60        stream_id: BucketStreamId,
61        content_type: Option<String>,
62        payload: ExternalPayloadRef,
63        close_after: bool,
64        stream_seq: Option<String>,
65        producer: Option<ProducerRequest>,
66        now_ms: u64,
67    },
68    AppendBatch {
69        stream_id: BucketStreamId,
70        content_type: Option<String>,
71        payloads: Vec<Vec<u8>>,
72        producer: Option<ProducerRequest>,
73        now_ms: u64,
74    },
75    PublishSnapshot {
76        stream_id: BucketStreamId,
77        snapshot_offset: u64,
78        content_type: String,
79        payload: Vec<u8>,
80        now_ms: u64,
81    },
82    TouchStreamAccess {
83        stream_id: BucketStreamId,
84        now_ms: u64,
85        renew_ttl: bool,
86    },
87    UpdateStreamAttrs {
88        stream_id: BucketStreamId,
89        attrs: Option<StreamAttrs>,
90        now_ms: u64,
91    },
92    AddForkRef {
93        stream_id: BucketStreamId,
94        now_ms: u64,
95    },
96    ReleaseForkRef {
97        stream_id: BucketStreamId,
98    },
99    FlushCold {
100        stream_id: BucketStreamId,
101        chunk: ColdChunkRef,
102    },
103    Close {
104        stream_id: BucketStreamId,
105        stream_seq: Option<String>,
106        producer: Option<ProducerRequest>,
107        now_ms: u64,
108    },
109    DeleteStream {
110        stream_id: BucketStreamId,
111    },
112    /// Confirms the leader's background worker has physically reclaimed every
113    /// queued cold-GC entry with `seq <= up_to_seq`; removes them from the
114    /// replicated queue. Idempotent under replay.
115    AckColdGc {
116        up_to_seq: u64,
117    },
118}
ursula_stream/command.rs

ursula_stream/
command.rs
