cloud_sdk/transport/streaming/
replay.rs

1//! Exact bounded source identity for explicit streaming replay.
2
3use core::fmt;
4
5/// Maximum exact source-version identity length.
6pub const MAX_STREAM_SOURCE_ID_BYTES: usize = 256;
7
8/// Invalid source identity.
9#[derive(Clone, Copy, Debug, Eq, PartialEq)]
10pub enum StreamSourceIdError {
11    /// Source identities cannot be empty.
12    Empty,
13    /// The identity exceeds [`MAX_STREAM_SOURCE_ID_BYTES`].
14    TooLong,
15}
16
17impl_static_error!(StreamSourceIdError,
18    Self::Empty => "stream source identity is empty",
19    Self::TooLong => "stream source identity is too long",
20);
21
22/// Exact source version supplied by the source owner.
23#[derive(Clone, Copy, Eq, Hash, Ord, PartialEq, PartialOrd)]
24pub struct StreamSourceId<'a>(&'a [u8]);
25
26impl<'a> StreamSourceId<'a> {
27    /// Admits one nonempty bounded exact identity.
28    pub const fn new(value: &'a [u8]) -> Result<Self, StreamSourceIdError> {
29        if value.is_empty() {
30            return Err(StreamSourceIdError::Empty);
31        }
32        if value.len() > MAX_STREAM_SOURCE_ID_BYTES {
33            return Err(StreamSourceIdError::TooLong);
34        }
35        Ok(Self(value))
36    }
37
38    /// Returns exact identity bytes for caller-controlled comparison or hashing.
39    #[must_use]
40    pub const fn as_bytes(self) -> &'a [u8] {
41        self.0
42    }
43}
44
45impl fmt::Debug for StreamSourceId<'_> {
46    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
47        formatter.write_str("StreamSourceId([redacted])")
48    }
49}
50
51/// Whether a streaming body can be reproduced for another attempt.
52#[derive(Clone, Copy, Debug, Eq, PartialEq)]
53pub enum StreamReplayability<'a> {
54    /// The source cannot guarantee byte-for-byte reproduction.
55    NotReplayable,
56    /// The source promises stable bytes while this exact version remains current.
57    Replayable(StreamSourceId<'a>),
58}
59
60/// Streaming replay rejection.
61#[derive(Clone, Copy, Debug, Eq, PartialEq)]
62pub enum StreamReplayError {
63    /// Either attempt uses a non-replayable source.
64    NonReplayable,
65    /// The source version changed between attempts.
66    SourceChanged,
67}
68
69impl_static_error!(StreamReplayError,
70    Self::NonReplayable => "stream source is not replayable",
71    Self::SourceChanged => "stream source changed between attempts",
72);
73
74/// Validates that a later attempt uses the same explicit replayable source.
75pub fn validate_stream_replay(
76    initial: StreamReplayability<'_>,
77    replay: StreamReplayability<'_>,
78) -> Result<(), StreamReplayError> {
79    let (StreamReplayability::Replayable(initial), StreamReplayability::Replayable(replay)) =
80        (initial, replay)
81    else {
82        return Err(StreamReplayError::NonReplayable);
83    };
84    if initial.as_bytes() != replay.as_bytes() {
85        return Err(StreamReplayError::SourceChanged);
86    }
87    Ok(())
88}
cloud_sdk/transport/streaming/replay.rs

cloud_sdk/transport/streaming/
replay.rs
