ursula_stream/
response.rs

1use serde::Deserialize;
2use serde::Serialize;
3use ursula_shard::BucketStreamId;
4
5use crate::model::ProducerRequest;
6use crate::record_index::StreamRecordRange;
7
8#[derive(Debug, Clone, PartialEq, Eq)]
9pub enum StreamResponse {
10    BucketCreated {
11        bucket_id: String,
12    },
13    BucketAlreadyExists {
14        bucket_id: String,
15    },
16    BucketDeleted {
17        bucket_id: String,
18    },
19    BucketQuotaSet {
20        bucket_id: String,
21    },
22    Created {
23        stream_id: BucketStreamId,
24        next_offset: u64,
25        closed: bool,
26    },
27    AlreadyExists {
28        next_offset: u64,
29        closed: bool,
30        content_type: String,
31        stream_ttl_seconds: Option<u64>,
32        stream_expires_at_ms: Option<u64>,
33    },
34    Appended {
35        offset: u64,
36        next_offset: u64,
37        closed: bool,
38        deduplicated: bool,
39        producer: Option<ProducerRequest>,
40    },
41    Closed {
42        next_offset: u64,
43        deduplicated: bool,
44        producer: Option<ProducerRequest>,
45    },
46    Deleted,
47    ColdFlushed {
48        hot_start_offset: u64,
49    },
50    ColdCompacted {
51        compacted_chunks: u64,
52        compacted_bytes: u64,
53    },
54    SnapshotPublished {
55        snapshot_offset: u64,
56        snapshot_digest: String,
57        record_range: Option<StreamRecordRange>,
58    },
59    RetentionAdvanced {
60        retained_offset: u64,
61        record_range: Option<StreamRecordRange>,
62    },
63    Accessed {
64        changed: bool,
65        expired: bool,
66    },
67    AttrsUpdated {
68        changed: bool,
69    },
70    ColdGcAcked {
71        removed: u64,
72    },
73    /// A whole-bucket purge accepted by [`StreamCommand::PurgeBucket`].
74    ///
75    /// [`StreamCommand::PurgeBucket`]: crate::StreamCommand::PurgeBucket
76    BucketPurged {
77        bucket_id: String,
78        removed_streams: u64,
79    },
80    /// A whole-group state import accepted by [`StreamCommand::ImportSnapshot`].
81    ///
82    /// [`StreamCommand::ImportSnapshot`]: crate::StreamCommand::ImportSnapshot
83    SnapshotImported {
84        buckets: u64,
85        streams: u64,
86    },
87    Error {
88        code: StreamErrorCode,
89        message: String,
90        next_offset: Option<u64>,
91        context: Vec<StreamErrorContext>,
92    },
93}
94
95#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
96pub enum StreamErrorCode {
97    InvalidBucketId,
98    InvalidStreamId,
99    BucketNotFound,
100    BucketNotEmpty,
101    StreamNotFound,
102    StreamGone,
103    StreamAlreadyExistsConflict,
104    MissingContentType,
105    ContentTypeMismatch,
106    EmptyAppend,
107    StreamClosed,
108    StreamSeqConflict,
109    InvalidProducer,
110    ProducerEpochStale,
111    ProducerSeqConflict,
112    InvalidRetention,
113    OffsetOutOfRange,
114    InvalidColdFlush,
115    InvalidSnapshot,
116    SnapshotNotFound,
117    SnapshotConflict,
118    InvalidStreamAttrs,
119    InvalidRecordBoundaries,
120    RecordPreconditionFailed,
121    /// A state import targeted a group that already holds buckets or streams.
122    ImportConflict,
123    /// A state import payload failed snapshot validation.
124    ImportInvalid,
125    QuotaExceeded,
126}
127
128#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
129pub enum StreamErrorContext {
130    StreamClosed,
131    StaleColdFlushCandidate,
132    ProducerEpochStale {
133        current_epoch: u64,
134    },
135    ProducerSeqConflict {
136        expected_seq: u64,
137        received_seq: u64,
138    },
139    RecordTailMismatch {
140        current_record: u64,
141    },
142}
143
144impl StreamResponse {
145    pub(crate) fn error(code: StreamErrorCode, message: impl Into<String>) -> Self {
146        Self::error_with_context(code, message, Vec::new())
147    }
148
149    pub(crate) fn error_with_context(
150        code: StreamErrorCode,
151        message: impl Into<String>,
152        context: Vec<StreamErrorContext>,
153    ) -> Self {
154        Self::Error {
155            code,
156            message: message.into(),
157            next_offset: None,
158            context,
159        }
160    }
161
162    pub(crate) fn error_with_next_offset(
163        code: StreamErrorCode,
164        message: impl Into<String>,
165        next_offset: u64,
166    ) -> Self {
167        Self::error_with_next_offset_and_context(code, message, next_offset, Vec::new())
168    }
169
170    pub(crate) fn error_with_next_offset_and_context(
171        code: StreamErrorCode,
172        message: impl Into<String>,
173        next_offset: u64,
174        context: Vec<StreamErrorContext>,
175    ) -> Self {
176        Self::Error {
177            code,
178            message: message.into(),
179            next_offset: Some(next_offset),
180            context,
181        }
182    }
183}
ursula_stream/response.rs

ursula_stream/
response.rs
