cloud_sdk/transport/streaming/
policy.rs

1//! Complete per-operation streaming policy.
2
3/// Global ceiling for one stream's actual transferred bytes.
4pub const MAX_STREAM_BYTES: u64 = 1_125_899_906_842_624;
5/// Global ceiling for one source chunk.
6pub const MAX_STREAM_CHUNK_BYTES: usize = 16_777_216;
7/// Global ceiling for chunks observed in one stream attempt.
8pub const MAX_STREAM_CHUNKS: u32 = 16_777_216;
9/// Global ceiling for source, sink, and waiting observations in one attempt.
10pub const MAX_STREAM_OBSERVATIONS: u32 = 67_108_864;
11/// Global ceiling for consecutive observations that transfer no bytes.
12pub const MAX_CONSECUTIVE_ZERO_PROGRESS: u16 = 4_096;
13
14/// Streaming operation shape.
15#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
16pub enum StreamKind {
17    /// One finite request-body upload.
18    FiniteUpload,
19    /// One finite response-body download.
20    FiniteDownload,
21    /// An event download whose lifetime is bounded by caller cancellation and
22    /// the observation policy.
23    CallerCancelledEvent,
24}
25
26/// Wire framing responsibility.
27#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
28pub enum StreamFraming {
29    /// The exact stream length is known before execution.
30    Declared(u64),
31    /// The executor owns framing for an explicitly unknown-length stream.
32    ExecutorOwned,
33}
34
35/// Visibility of accepted bytes before successful completion.
36#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
37pub enum StreamSinkMode {
38    /// Accepted bytes remain hidden and the sink can roll them back on abort.
39    Transactional,
40    /// Accepted bytes may be externally visible and cannot be rolled back.
41    Direct,
42}
43
44/// Invalid hard streaming limits.
45#[derive(Clone, Copy, Debug, Eq, PartialEq)]
46pub enum StreamLimitsError {
47    /// The byte limit is zero.
48    ByteLimitZero,
49    /// The byte limit exceeds the global ceiling.
50    ByteLimitTooLarge,
51    /// The chunk-size limit is zero.
52    ChunkBytesZero,
53    /// The chunk-size limit exceeds the global ceiling.
54    ChunkBytesTooLarge,
55    /// The chunk-count limit is zero.
56    ChunkLimitZero,
57    /// The chunk-count limit exceeds the global ceiling.
58    ChunkLimitTooLarge,
59    /// The observation limit is zero or cannot admit every allowed chunk.
60    ObservationLimitTooSmall,
61    /// The observation limit exceeds the global ceiling.
62    ObservationLimitTooLarge,
63    /// The zero-progress tolerance exceeds its global or observation bound.
64    ZeroProgressLimitTooLarge,
65}
66
67impl_static_error!(StreamLimitsError,
68    Self::ByteLimitZero => "stream byte limit is zero",
69    Self::ByteLimitTooLarge => "stream byte limit exceeds the global ceiling",
70    Self::ChunkBytesZero => "stream chunk-size limit is zero",
71    Self::ChunkBytesTooLarge => "stream chunk-size limit exceeds the global ceiling",
72    Self::ChunkLimitZero => "stream chunk limit is zero",
73    Self::ChunkLimitTooLarge => "stream chunk limit exceeds the global ceiling",
74    Self::ObservationLimitTooSmall => "stream observation limit cannot admit every chunk",
75    Self::ObservationLimitTooLarge => "stream observation limit exceeds the global ceiling",
76    Self::ZeroProgressLimitTooLarge => "stream zero-progress limit is too large",
77);
78
79/// Hard byte, chunk, observation, and zero-progress limits.
80#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
81pub struct StreamLimits {
82    max_bytes: u64,
83    max_chunk_bytes: usize,
84    max_chunks: u32,
85    max_observations: u32,
86    max_consecutive_zero_progress: u16,
87}
88
89impl StreamLimits {
90    /// Creates complete nonzero limits beneath global ceilings.
91    pub const fn new(
92        max_bytes: u64,
93        max_chunk_bytes: usize,
94        max_chunks: u32,
95        max_observations: u32,
96        max_consecutive_zero_progress: u16,
97    ) -> Result<Self, StreamLimitsError> {
98        if max_bytes == 0 {
99            return Err(StreamLimitsError::ByteLimitZero);
100        }
101        if max_bytes > MAX_STREAM_BYTES {
102            return Err(StreamLimitsError::ByteLimitTooLarge);
103        }
104        if max_chunk_bytes == 0 {
105            return Err(StreamLimitsError::ChunkBytesZero);
106        }
107        if max_chunk_bytes > MAX_STREAM_CHUNK_BYTES {
108            return Err(StreamLimitsError::ChunkBytesTooLarge);
109        }
110        if max_chunks == 0 {
111            return Err(StreamLimitsError::ChunkLimitZero);
112        }
113        if max_chunks > MAX_STREAM_CHUNKS {
114            return Err(StreamLimitsError::ChunkLimitTooLarge);
115        }
116        if max_observations < max_chunks {
117            return Err(StreamLimitsError::ObservationLimitTooSmall);
118        }
119        if max_observations > MAX_STREAM_OBSERVATIONS {
120            return Err(StreamLimitsError::ObservationLimitTooLarge);
121        }
122        if max_consecutive_zero_progress > MAX_CONSECUTIVE_ZERO_PROGRESS
123            || max_consecutive_zero_progress as u32 > max_observations
124        {
125            return Err(StreamLimitsError::ZeroProgressLimitTooLarge);
126        }
127        Ok(Self {
128            max_bytes,
129            max_chunk_bytes,
130            max_chunks,
131            max_observations,
132            max_consecutive_zero_progress,
133        })
134    }
135
136    /// Returns the actual-byte ceiling.
137    #[must_use]
138    pub const fn max_bytes(self) -> u64 {
139        self.max_bytes
140    }
141
142    /// Returns the per-chunk byte ceiling.
143    #[must_use]
144    pub const fn max_chunk_bytes(self) -> usize {
145        self.max_chunk_bytes
146    }
147
148    /// Returns the source-chunk ceiling.
149    #[must_use]
150    pub const fn max_chunks(self) -> u32 {
151        self.max_chunks
152    }
153
154    /// Returns the source, sink, and waiting observation ceiling.
155    #[must_use]
156    pub const fn max_observations(self) -> u32 {
157        self.max_observations
158    }
159
160    /// Returns the admitted consecutive zero-progress observations.
161    #[must_use]
162    pub const fn max_consecutive_zero_progress(self) -> u16 {
163        self.max_consecutive_zero_progress
164    }
165}
166
167/// Incoherent stream policy.
168#[derive(Clone, Copy, Debug, Eq, PartialEq)]
169pub enum StreamPolicyError {
170    /// The declared length exceeds the operation byte limit.
171    DeclaredLengthTooLarge,
172    /// Event streams require executor-owned unknown-length framing.
173    EventRequiresExecutorFraming,
174}
175
176impl_static_error!(StreamPolicyError,
177    Self::DeclaredLengthTooLarge => "declared stream length exceeds the operation limit",
178    Self::EventRequiresExecutorFraming => "event stream requires executor-owned framing",
179);
180
181/// Complete stream behavior fixed before the first observation.
182#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
183pub struct StreamPolicy {
184    kind: StreamKind,
185    framing: StreamFraming,
186    sink_mode: StreamSinkMode,
187    limits: StreamLimits,
188}
189
190impl StreamPolicy {
191    /// Creates a coherent stream policy without permissive defaults.
192    pub const fn new(
193        kind: StreamKind,
194        framing: StreamFraming,
195        sink_mode: StreamSinkMode,
196        limits: StreamLimits,
197    ) -> Result<Self, StreamPolicyError> {
198        if let StreamFraming::Declared(length) = framing
199            && length > limits.max_bytes
200        {
201            return Err(StreamPolicyError::DeclaredLengthTooLarge);
202        }
203        if matches!(kind, StreamKind::CallerCancelledEvent)
204            && !matches!(framing, StreamFraming::ExecutorOwned)
205        {
206            return Err(StreamPolicyError::EventRequiresExecutorFraming);
207        }
208        Ok(Self {
209            kind,
210            framing,
211            sink_mode,
212            limits,
213        })
214    }
215
216    /// Returns the operation shape.
217    #[must_use]
218    pub const fn kind(self) -> StreamKind {
219        self.kind
220    }
221
222    /// Returns the explicit wire framing policy.
223    #[must_use]
224    pub const fn framing(self) -> StreamFraming {
225        self.framing
226    }
227
228    /// Returns partial-byte visibility policy.
229    #[must_use]
230    pub const fn sink_mode(self) -> StreamSinkMode {
231        self.sink_mode
232    }
233
234    /// Returns all hard limits.
235    #[must_use]
236    pub const fn limits(self) -> StreamLimits {
237        self.limits
238    }
239}
240
241pub(super) const fn partial_state(
242    mode: StreamSinkMode,
243    write_attempted: bool,
244) -> super::StreamPartialState {
245    if !write_attempted {
246        super::StreamPartialState::Clean
247    } else {
248        match mode {
249            StreamSinkMode::Transactional => super::StreamPartialState::RollbackRequired,
250            StreamSinkMode::Direct => super::StreamPartialState::Dirty,
251        }
252    }
253}
cloud_sdk/transport/streaming/policy.rs

cloud_sdk/transport/streaming/
policy.rs
