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        pending_cold_gc_entries: u64,
80    },
81    /// A whole-group state import accepted by [`StreamCommand::ImportSnapshot`].
82    ///
83    /// [`StreamCommand::ImportSnapshot`]: crate::StreamCommand::ImportSnapshot
84    SnapshotImported {
85        buckets: u64,
86        streams: u64,
87    },
88    Error {
89        code: StreamErrorCode,
90        message: String,
91        next_offset: Option<u64>,
92        context: Vec<StreamErrorContext>,
93    },
94}
95
96#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
97pub enum StreamErrorCode {
98    InvalidBucketId,
99    InvalidStreamId,
100    BucketNotFound,
101    /// The bucket was permanently purged and its name is fenced from reuse.
102    BucketErased,
103    BucketNotEmpty,
104    StreamNotFound,
105    StreamGone,
106    StreamAlreadyExistsConflict,
107    MissingContentType,
108    ContentTypeMismatch,
109    EmptyAppend,
110    StreamClosed,
111    StreamSeqConflict,
112    InvalidProducer,
113    ProducerEpochStale,
114    ProducerSeqConflict,
115    InvalidRetention,
116    OffsetOutOfRange,
117    InvalidColdFlush,
118    InvalidSnapshot,
119    SnapshotNotFound,
120    SnapshotConflict,
121    InvalidStreamAttrs,
122    InvalidRecordBoundaries,
123    RecordPreconditionFailed,
124    /// A state import targeted a group that already holds buckets or streams.
125    ImportConflict,
126    /// A state import payload failed snapshot validation.
127    ImportInvalid,
128    QuotaExceeded,
129}
130
131#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
132pub enum StreamErrorContext {
133    StreamClosed,
134    StaleColdFlushCandidate,
135    ProducerEpochStale {
136        current_epoch: u64,
137    },
138    ProducerSeqConflict {
139        expected_seq: u64,
140        received_seq: u64,
141    },
142    RecordTailMismatch {
143        current_record: u64,
144    },
145}
146
147impl StreamResponse {
148    pub(crate) fn error(code: StreamErrorCode, message: impl Into<String>) -> Self {
149        Self::error_with_context(code, message, Vec::new())
150    }
151
152    pub(crate) fn error_with_context(
153        code: StreamErrorCode,
154        message: impl Into<String>,
155        context: Vec<StreamErrorContext>,
156    ) -> Self {
157        Self::Error {
158            code,
159            message: message.into(),
160            next_offset: None,
161            context,
162        }
163    }
164
165    pub(crate) fn error_with_next_offset(
166        code: StreamErrorCode,
167        message: impl Into<String>,
168        next_offset: u64,
169    ) -> Self {
170        Self::error_with_next_offset_and_context(code, message, next_offset, Vec::new())
171    }
172
173    pub(crate) fn error_with_next_offset_and_context(
174        code: StreamErrorCode,
175        message: impl Into<String>,
176        next_offset: u64,
177        context: Vec<StreamErrorContext>,
178    ) -> Self {
179        Self::Error {
180            code,
181            message: message.into(),
182            next_offset: Some(next_offset),
183            context,
184        }
185    }
186}
ursula_stream/response.rs

ursula_stream/
response.rs
