cloud_sdk_testkit/
stream.rs

1//! Deterministic bounded streaming sources and sinks.
2
3use cloud_sdk::buffer::sanitize_bytes;
4use cloud_sdk::transport::{
5    AsyncStreamSink, AsyncStreamSource, BlockingStreamSink, BlockingStreamSource,
6    StreamPartialState, StreamRead, StreamReplayability,
7};
8
9/// Maximum chunks in one borrowed stream fixture.
10pub const MAX_STREAM_FIXTURE_CHUNKS: usize = 1_024;
11
12/// Invalid fixture or deterministic fixture I/O failure.
13#[derive(Clone, Copy, Debug, Eq, PartialEq)]
14pub enum StreamFixtureError {
15    /// The fixture contains too many chunks.
16    TooManyChunks,
17    /// One source chunk exceeds supplied scratch storage.
18    SourceScratchTooSmall,
19    /// Sink storage cannot accept the next deterministic write.
20    SinkStorageTooSmall,
21    /// The configured maximum sink write is zero.
22    ZeroWriteLimit,
23    /// Fault observation indices are one-based.
24    ZeroFaultIndex,
25    /// The source reached its configured injected failure.
26    InjectedSourceFault,
27    /// The sink reached its configured injected failure.
28    InjectedSinkFault,
29    /// Alternating stream patterns require a nonempty data chunk.
30    EmptyPatternData,
31}
32
33impl_static_error!(StreamFixtureError,
34    Self::TooManyChunks => "stream fixture contains too many chunks",
35    Self::SourceScratchTooSmall => "stream fixture scratch storage is too small",
36    Self::SinkStorageTooSmall => "stream fixture sink storage is too small",
37    Self::ZeroWriteLimit => "stream fixture sink write limit is zero",
38    Self::ZeroFaultIndex => "stream fixture fault index must be nonzero",
39    Self::InjectedSourceFault => "stream fixture injected a source failure",
40    Self::InjectedSinkFault => "stream fixture injected a sink failure",
41    Self::EmptyPatternData => "alternating stream pattern data is empty",
42);
43
44/// Borrowed ordered chunks for one deterministic finite source.
45pub struct StreamFixtureSource<'fixture> {
46    chunks: &'fixture [&'fixture [u8]],
47    index: usize,
48    observations: usize,
49    replayability: StreamReplayability<'fixture>,
50    fault_at_observation: Option<usize>,
51}
52
53impl<'fixture> StreamFixtureSource<'fixture> {
54    /// Creates one bounded source. Empty borrowed slices represent explicit
55    /// empty chunks rather than end-of-stream.
56    pub const fn new(chunks: &'fixture [&'fixture [u8]]) -> Result<Self, StreamFixtureError> {
57        if chunks.len() > MAX_STREAM_FIXTURE_CHUNKS {
58            return Err(StreamFixtureError::TooManyChunks);
59        }
60        Ok(Self {
61            chunks,
62            index: 0,
63            observations: 0,
64            replayability: StreamReplayability::NotReplayable,
65            fault_at_observation: None,
66        })
67    }
68
69    /// Creates a bounded source with an explicit replay capability.
70    pub const fn with_replayability(
71        chunks: &'fixture [&'fixture [u8]],
72        replayability: StreamReplayability<'fixture>,
73    ) -> Result<Self, StreamFixtureError> {
74        if chunks.len() > MAX_STREAM_FIXTURE_CHUNKS {
75            return Err(StreamFixtureError::TooManyChunks);
76        }
77        Ok(Self {
78            chunks,
79            index: 0,
80            observations: 0,
81            replayability,
82            fault_at_observation: None,
83        })
84    }
85
86    /// Injects a source failure at one one-based read observation.
87    pub const fn with_fault_at_observation(
88        mut self,
89        observation: usize,
90    ) -> Result<Self, StreamFixtureError> {
91        if observation == 0 {
92            return Err(StreamFixtureError::ZeroFaultIndex);
93        }
94        self.fault_at_observation = Some(observation);
95        Ok(self)
96    }
97
98    /// Returns source observations including the final end marker.
99    #[must_use]
100    pub const fn observations(&self) -> usize {
101        self.observations
102    }
103
104    fn read(&mut self, output: &mut [u8]) -> Result<StreamRead, StreamFixtureError> {
105        self.observations = self.observations.saturating_add(1);
106        if self.fault_at_observation == Some(self.observations) {
107            return Err(StreamFixtureError::InjectedSourceFault);
108        }
109        let Some(chunk) = self.chunks.get(self.index) else {
110            return Ok(StreamRead::End);
111        };
112        let target = output
113            .get_mut(..chunk.len())
114            .ok_or(StreamFixtureError::SourceScratchTooSmall)?;
115        target.copy_from_slice(chunk);
116        self.index = self.index.saturating_add(1);
117        Ok(StreamRead::Chunk(chunk.len()))
118    }
119}
120
121/// Non-terminating deterministic source pattern for hard-limit tests.
122#[derive(Clone, Copy, Debug, Eq, PartialEq)]
123pub enum StreamPattern<'fixture> {
124    /// Every observation is an explicit empty chunk.
125    EndlessEmpty,
126    /// Observations alternate between an empty chunk and borrowed data.
127    AlternatingEmptyData(&'fixture [u8]),
128}
129
130/// Non-terminating stream source used to verify cancellation and hard bounds.
131pub struct StreamPatternSource<'fixture> {
132    pattern: StreamPattern<'fixture>,
133    observations: usize,
134}
135
136impl<'fixture> StreamPatternSource<'fixture> {
137    /// Creates a deterministic non-terminating source pattern.
138    pub const fn new(pattern: StreamPattern<'fixture>) -> Result<Self, StreamFixtureError> {
139        if matches!(pattern, StreamPattern::AlternatingEmptyData(data) if data.is_empty()) {
140            return Err(StreamFixtureError::EmptyPatternData);
141        }
142        Ok(Self {
143            pattern,
144            observations: 0,
145        })
146    }
147
148    /// Returns the number of source observations.
149    #[must_use]
150    pub const fn observations(&self) -> usize {
151        self.observations
152    }
153
154    fn read(&mut self, output: &mut [u8]) -> Result<StreamRead, StreamFixtureError> {
155        self.observations = self.observations.saturating_add(1);
156        match self.pattern {
157            StreamPattern::EndlessEmpty => Ok(StreamRead::Chunk(0)),
158            StreamPattern::AlternatingEmptyData(_) if self.observations % 2 == 1 => {
159                Ok(StreamRead::Chunk(0))
160            }
161            StreamPattern::AlternatingEmptyData(data) => {
162                let target = output
163                    .get_mut(..data.len())
164                    .ok_or(StreamFixtureError::SourceScratchTooSmall)?;
165                target.copy_from_slice(data);
166                Ok(StreamRead::Chunk(data.len()))
167            }
168        }
169    }
170}
171
172impl BlockingStreamSource for StreamPatternSource<'_> {
173    type Error = StreamFixtureError;
174
175    fn replayability(&self) -> StreamReplayability<'_> {
176        StreamReplayability::NotReplayable
177    }
178
179    fn read_chunk(&mut self, output: &mut [u8]) -> Result<StreamRead, Self::Error> {
180        self.read(output)
181    }
182}
183
184impl AsyncStreamSource for StreamPatternSource<'_> {
185    type Error = StreamFixtureError;
186
187    fn replayability(&self) -> StreamReplayability<'_> {
188        StreamReplayability::NotReplayable
189    }
190
191    async fn read_chunk<'operation>(
192        &'operation mut self,
193        output: &'operation mut [u8],
194    ) -> Result<StreamRead, Self::Error> {
195        self.read(output)
196    }
197}
198
199impl BlockingStreamSource for StreamFixtureSource<'_> {
200    type Error = StreamFixtureError;
201
202    fn replayability(&self) -> StreamReplayability<'_> {
203        self.replayability
204    }
205
206    fn read_chunk(&mut self, output: &mut [u8]) -> Result<StreamRead, Self::Error> {
207        self.read(output)
208    }
209}
210
211impl AsyncStreamSource for StreamFixtureSource<'_> {
212    type Error = StreamFixtureError;
213
214    fn replayability(&self) -> StreamReplayability<'_> {
215        self.replayability
216    }
217
218    async fn read_chunk<'operation>(
219        &'operation mut self,
220        output: &'operation mut [u8],
221    ) -> Result<StreamRead, Self::Error> {
222        self.read(output)
223    }
224}
225
226/// Caller-buffered deterministic sink with configurable short writes.
227pub struct StreamFixtureSink<'storage> {
228    output: &'storage mut [u8],
229    initialized_len: usize,
230    max_write_bytes: usize,
231    writes: usize,
232    committed: bool,
233    aborted: Option<StreamPartialState>,
234    fault_at_write: Option<usize>,
235}
236
237impl<'storage> StreamFixtureSink<'storage> {
238    /// Creates a sink and clears all caller storage before first use.
239    pub fn new(
240        output: &'storage mut [u8],
241        max_write_bytes: usize,
242    ) -> Result<Self, StreamFixtureError> {
243        sanitize_bytes(output);
244        if max_write_bytes == 0 {
245            return Err(StreamFixtureError::ZeroWriteLimit);
246        }
247        Ok(Self {
248            output,
249            initialized_len: 0,
250            max_write_bytes,
251            writes: 0,
252            committed: false,
253            aborted: None,
254            fault_at_write: None,
255        })
256    }
257
258    /// Injects a sink failure at one one-based write attempt.
259    pub const fn with_fault_at_write(mut self, write: usize) -> Result<Self, StreamFixtureError> {
260        if write == 0 {
261            return Err(StreamFixtureError::ZeroFaultIndex);
262        }
263        self.fault_at_write = Some(write);
264        Ok(self)
265    }
266
267    /// Returns initialized sink bytes.
268    #[must_use]
269    pub fn bytes(&self) -> &[u8] {
270        self.output.get(..self.initialized_len).unwrap_or_default()
271    }
272
273    /// Returns deterministic sink write count.
274    #[must_use]
275    pub const fn writes(&self) -> usize {
276        self.writes
277    }
278
279    /// Reports whether the successful stream was committed.
280    #[must_use]
281    pub const fn is_committed(&self) -> bool {
282        self.committed
283    }
284
285    /// Returns the last incomplete-state abort classification.
286    #[must_use]
287    pub const fn aborted_with(&self) -> Option<StreamPartialState> {
288        self.aborted
289    }
290
291    fn write(&mut self, input: &[u8]) -> Result<usize, StreamFixtureError> {
292        let next_write = self.writes.saturating_add(1);
293        if self.fault_at_write == Some(next_write) {
294            return Err(StreamFixtureError::InjectedSinkFault);
295        }
296        let accepted = core::cmp::min(input.len(), self.max_write_bytes);
297        let end = self
298            .initialized_len
299            .checked_add(accepted)
300            .ok_or(StreamFixtureError::SinkStorageTooSmall)?;
301        let target = self
302            .output
303            .get_mut(self.initialized_len..end)
304            .ok_or(StreamFixtureError::SinkStorageTooSmall)?;
305        let source = input
306            .get(..accepted)
307            .ok_or(StreamFixtureError::SinkStorageTooSmall)?;
308        target.copy_from_slice(source);
309        self.initialized_len = end;
310        self.writes = self.writes.saturating_add(1);
311        Ok(accepted)
312    }
313
314    fn commit_inner(&mut self) {
315        self.committed = true;
316    }
317
318    fn abort_inner(&mut self, partial: StreamPartialState) {
319        self.aborted = Some(partial);
320        if matches!(partial, StreamPartialState::RollbackRequired) {
321            sanitize_bytes(self.output);
322            self.initialized_len = 0;
323        }
324    }
325}
326
327impl BlockingStreamSink for StreamFixtureSink<'_> {
328    type Error = StreamFixtureError;
329
330    fn write_chunk(&mut self, input: &[u8]) -> Result<usize, Self::Error> {
331        self.write(input)
332    }
333
334    fn commit(&mut self) -> Result<(), Self::Error> {
335        self.commit_inner();
336        Ok(())
337    }
338
339    fn abort(&mut self, partial: StreamPartialState) {
340        self.abort_inner(partial);
341    }
342}
343
344impl AsyncStreamSink for StreamFixtureSink<'_> {
345    type Error = StreamFixtureError;
346
347    async fn write_chunk<'operation>(
348        &'operation mut self,
349        input: &'operation [u8],
350    ) -> Result<usize, Self::Error> {
351        self.write(input)
352    }
353
354    async fn commit(&mut self) -> Result<(), Self::Error> {
355        self.commit_inner();
356        Ok(())
357    }
358
359    fn abort(&mut self, partial: StreamPartialState) {
360        self.abort_inner(partial);
361    }
362}
cloud_sdk_testkit/stream.rs

cloud_sdk_testkit/
stream.rs
