cloud_sdk/operation/
prepared.rs

1//! Prepared operation storage, endpoint binding, and execution.
2
3use core::fmt;
4
5use cloud_sdk_sanitization::sanitize_bytes;
6
7use crate::authentication::{
8    AsyncAuthenticatedTransport, AuthenticatedRequest, AuthenticationScopePolicy,
9    BlockingAuthenticatedTransport, drive_async_authenticated,
10};
11use crate::operation::{
12    CheckedResponseGuard, OperationId, OperationImpact, OperationMetadata, RequestIdPolicy,
13    ResponsePolicy, ResponsePolicyError,
14};
15use crate::transport::{
16    BoundTransport, EndpointIdentity, EndpointPolicy, RawResponsePolicy, RequestHeaders,
17    ResponseBuffer, TransportRequest,
18};
19use crate::{ProviderId, ProviderMarker, ServiceId, ServiceMarker};
20
21mod error;
22mod storage;
23use error::{EndpointCheckError, map_endpoint_error};
24pub use error::{PreparedExecutionError, PreparedRequestPolicyError};
25pub use storage::{PreparationStorage, PrepareOperation};
26
27/// Whether one prepared request body can be sent again byte-for-byte.
28#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
29pub enum BodyReplayability {
30    /// The body source cannot guarantee an identical subsequent read.
31    NotReplayable,
32    /// The complete body is an immutable byte snapshot for the request lifetime.
33    Replayable,
34}
35
36/// Provider service and immutable endpoint trust policy.
37#[derive(Clone, Copy, Debug, Eq, PartialEq)]
38pub struct ProviderService<'endpoint> {
39    provider_id: ProviderId,
40    service_id: ServiceId,
41    endpoint_policy: EndpointPolicy<'endpoint>,
42}
43
44impl<'endpoint> ProviderService<'endpoint> {
45    /// Binds validated provider and service IDs to an endpoint trust policy.
46    #[must_use]
47    pub const fn new(
48        provider_id: ProviderId,
49        service_id: ServiceId,
50        endpoint_policy: EndpointPolicy<'endpoint>,
51    ) -> Self {
52        Self {
53            provider_id,
54            service_id,
55            endpoint_policy,
56        }
57    }
58
59    /// Binds a provider-owned service marker to an endpoint trust policy.
60    #[must_use]
61    pub const fn from_marker<S: ServiceMarker>(endpoint_policy: EndpointPolicy<'endpoint>) -> Self {
62        Self::new(<S::Provider as ProviderMarker>::ID, S::ID, endpoint_policy)
63    }
64
65    /// Returns the canonical provider namespace.
66    #[must_use]
67    pub const fn provider_id(self) -> ProviderId {
68        self.provider_id
69    }
70
71    /// Returns the canonical provider-owned service namespace.
72    #[must_use]
73    pub const fn service_id(self) -> ServiceId {
74        self.service_id
75    }
76
77    /// Returns the immutable endpoint trust policy.
78    #[must_use]
79    pub const fn endpoint_policy(self) -> EndpointPolicy<'endpoint> {
80        self.endpoint_policy
81    }
82}
83
84/// Complete request, endpoint, operation metadata, and response policy.
85#[derive(Clone, Copy)]
86pub struct PreparedRequest<'request> {
87    request: TransportRequest<'request>,
88    service: ProviderService<'request>,
89    metadata: OperationMetadata,
90    response_policy: ResponsePolicy,
91    authentication_policy: AuthenticationScopePolicy<'request>,
92    raw_response_policy: RawResponsePolicy<'request>,
93    operation_id: Option<OperationId>,
94    body_replayability: BodyReplayability,
95}
96
97impl<'request> PreparedRequest<'request> {
98    /// Creates a complete prepared request after checking cross-policy invariants.
99    ///
100    /// # Errors
101    ///
102    /// Returns [`PreparedRequestPolicyError::MissingRequestIdHeader`] when
103    /// operation metadata protects or retains request IDs but the raw response
104    /// policy does not admit `x-request-id`. Returns
105    /// [`PreparedRequestPolicyError::ReadOnlyMethodMismatch`] when read-only
106    /// metadata is paired with any method other than `GET` or `HEAD`.
107    pub fn new(
108        request: TransportRequest<'request>,
109        service: ProviderService<'request>,
110        metadata: OperationMetadata,
111        response_policy: ResponsePolicy,
112        authentication_policy: AuthenticationScopePolicy<'request>,
113        raw_response_policy: RawResponsePolicy<'request>,
114    ) -> Result<Self, PreparedRequestPolicyError> {
115        if matches!(metadata.impact(), OperationImpact::ReadOnly)
116            && !request.method().permits_direct_read_only()
117        {
118            return Err(PreparedRequestPolicyError::ReadOnlyMethodMismatch);
119        }
120        if metadata.request_id_policy() != RequestIdPolicy::Discard
121            && !raw_response_policy.admits_header("x-request-id")
122        {
123            return Err(PreparedRequestPolicyError::MissingRequestIdHeader);
124        }
125        Ok(Self {
126            request,
127            service,
128            metadata,
129            response_policy,
130            authentication_policy,
131            raw_response_policy,
132            operation_id: None,
133            body_replayability: if request.body().is_empty() {
134                BodyReplayability::Replayable
135            } else {
136                BodyReplayability::NotReplayable
137            },
138        })
139    }
140
141    /// Binds a validated provider operation identifier to this request.
142    #[must_use]
143    pub const fn with_operation_id(mut self, operation_id: OperationId) -> Self {
144        self.operation_id = Some(operation_id);
145        self
146    }
147
148    /// Marks the immutable prepared body snapshot as byte-for-byte replayable.
149    ///
150    /// Providers must call this only after preparation has completed and when
151    /// the borrowed body bytes cannot change for the prepared request lifetime.
152    #[must_use]
153    pub const fn with_replayable_body(mut self) -> Self {
154        self.body_replayability = BodyReplayability::Replayable;
155        self
156    }
157
158    /// Returns the validated transport request.
159    #[must_use]
160    pub const fn transport_request(self) -> TransportRequest<'request> {
161        self.request
162    }
163
164    /// Returns the bound provider service.
165    #[must_use]
166    pub const fn service(self) -> ProviderService<'request> {
167        self.service
168    }
169
170    /// Returns complete safety and retry metadata.
171    #[must_use]
172    pub const fn metadata(self) -> OperationMetadata {
173        self.metadata
174    }
175
176    /// Returns complete checked-response policy.
177    #[must_use]
178    pub const fn response_policy(self) -> ResponsePolicy {
179        self.response_policy
180    }
181
182    /// Returns the complete provider-owned authentication-scope policy.
183    #[must_use]
184    pub const fn authentication_policy(self) -> AuthenticationScopePolicy<'request> {
185        self.authentication_policy
186    }
187
188    /// Returns the complete status-class raw response policy.
189    #[must_use]
190    pub const fn raw_response_policy(self) -> RawResponsePolicy<'request> {
191        self.raw_response_policy
192    }
193
194    /// Returns the request with its mandatory authentication and raw wire policy.
195    #[must_use]
196    pub(crate) const fn authenticated_request(self) -> AuthenticatedRequest<'request, 'request> {
197        AuthenticatedRequest::new(
198            self.request,
199            self.authentication_policy,
200            &self.raw_response_policy,
201        )
202    }
203
204    /// Returns the provider operation identifier when one was bound.
205    #[must_use]
206    pub const fn operation_id(self) -> Option<OperationId> {
207        self.operation_id
208    }
209
210    /// Returns the explicit request-body replay capability.
211    #[must_use]
212    pub const fn body_replayability(self) -> BodyReplayability {
213        self.body_replayability
214    }
215
216    pub(crate) fn with_request_headers<'headers>(
217        self,
218        headers: RequestHeaders<'headers>,
219    ) -> PreparedRequest<'headers>
220    where
221        'request: 'headers,
222    {
223        let request: TransportRequest<'headers> = self.request;
224        PreparedRequest {
225            request: request.with_headers(headers),
226            service: self.service,
227            metadata: self.metadata,
228            response_policy: self.response_policy,
229            authentication_policy: self.authentication_policy,
230            raw_response_policy: self.raw_response_policy,
231            operation_id: self.operation_id,
232            body_replayability: self.body_replayability,
233        }
234    }
235
236    pub(crate) fn has_same_retry_policy(&self, other: &Self) -> bool {
237        self.service == other.service
238            && self.metadata == other.metadata
239            && self.response_policy == other.response_policy
240            && self.authentication_policy == other.authentication_policy
241            && self.raw_response_policy == other.raw_response_policy
242            && self.operation_id == other.operation_id
243            && self.body_replayability == other.body_replayability
244            && self.has_same_header_policy(other)
245    }
246
247    fn has_same_header_policy(&self, other: &Self) -> bool {
248        let left = self.request.headers().as_slice();
249        let right = other.request.headers().as_slice();
250        left.len() == right.len()
251            && left
252                .iter()
253                .zip(right)
254                .all(|(left, right)| left.sensitivity() == right.sensitivity())
255    }
256
257    /// Applies the complete prepared response policy without executing transport.
258    pub fn validate_response<'buffer>(
259        self,
260        response: ResponseBuffer<'buffer>,
261    ) -> Result<CheckedResponseGuard<'buffer>, ResponsePolicyError> {
262        self.response_policy
263            .validate(response, self.metadata.request_id_policy())
264    }
265
266    /// Applies operation-owned metadata policy before provider error decoding.
267    ///
268    /// This is the error-status counterpart to [`Self::validate_response`].
269    /// It extracts and protects, discards, or admits retention of the provider
270    /// request identifier without applying success-status or body policy.
271    pub fn apply_response_metadata_policy(
272        self,
273        response: &mut ResponseBuffer<'_>,
274    ) -> Result<(), ResponsePolicyError> {
275        super::policy::apply_request_id_policy(response, self.metadata.request_id_policy())
276    }
277
278    /// Verifies endpoint identity, executes once, and validates the response.
279    pub fn execute_blocking<'buffer, T>(
280        self,
281        transport: &T,
282        response_storage: &'buffer mut [u8],
283        response_header_storage: &'buffer mut [u8],
284    ) -> Result<CheckedResponseGuard<'buffer>, PreparedExecutionError<T::Error>>
285    where
286        T: BlockingAuthenticatedTransport + BoundTransport,
287    {
288        let response = self.send_blocking(transport, response_storage, response_header_storage)?;
289        self.response_policy
290            .validate(response, self.metadata.request_id_policy())
291            .map_err(PreparedExecutionError::ResponsePolicy)
292    }
293
294    pub(crate) fn send_blocking<'buffer, T>(
295        self,
296        transport: &T,
297        response_storage: &'buffer mut [u8],
298        response_header_storage: &'buffer mut [u8],
299    ) -> Result<ResponseBuffer<'buffer>, PreparedExecutionError<T::Error>>
300    where
301        T: BlockingAuthenticatedTransport + BoundTransport,
302    {
303        if self.requires_execution_permit() {
304            sanitize_bytes(response_storage);
305            sanitize_bytes(response_header_storage);
306            return Err(PreparedExecutionError::AuthorizationRequired);
307        }
308        self.send_blocking_authorized(transport, None, response_storage, response_header_storage)
309    }
310
311    pub(crate) fn execute_blocking_authorized<'buffer, T>(
312        self,
313        transport: &T,
314        confirmed_endpoint: Option<EndpointIdentity<'_>>,
315        response_storage: &'buffer mut [u8],
316        response_header_storage: &'buffer mut [u8],
317    ) -> Result<CheckedResponseGuard<'buffer>, PreparedExecutionError<T::Error>>
318    where
319        T: BlockingAuthenticatedTransport + BoundTransport,
320    {
321        let response = self.send_blocking_authorized(
322            transport,
323            confirmed_endpoint,
324            response_storage,
325            response_header_storage,
326        )?;
327        self.response_policy
328            .validate(response, self.metadata.request_id_policy())
329            .map_err(PreparedExecutionError::ResponsePolicy)
330    }
331
332    pub(crate) fn send_blocking_authorized<'buffer, T>(
333        self,
334        transport: &T,
335        confirmed_endpoint: Option<EndpointIdentity<'_>>,
336        response_storage: &'buffer mut [u8],
337        response_header_storage: &'buffer mut [u8],
338    ) -> Result<ResponseBuffer<'buffer>, PreparedExecutionError<T::Error>>
339    where
340        T: BlockingAuthenticatedTransport + BoundTransport,
341    {
342        let mut response = ResponseBuffer::new(
343            response_storage,
344            self.raw_response_policy.max_body_bytes(),
345            response_header_storage,
346        );
347        self.verify_endpoint(transport, confirmed_endpoint)
348            .map_err(map_endpoint_error)?;
349        transport
350            .send_authenticated(self.authenticated_request(), response.writer())
351            .map_err(PreparedExecutionError::Transport)?;
352        Ok(response)
353    }
354
355    /// Async equivalent of [`Self::execute_blocking`] without owning an executor.
356    pub async fn execute_async<'transport, 'buffer, T>(
357        &'transport self,
358        transport: &'transport T,
359        response_storage: &'buffer mut [u8],
360        response_header_storage: &'buffer mut [u8],
361    ) -> Result<CheckedResponseGuard<'buffer>, PreparedExecutionError<T::Error>>
362    where
363        T: AsyncAuthenticatedTransport + BoundTransport,
364        'request: 'transport,
365    {
366        let response = self
367            .send_async(transport, response_storage, response_header_storage)
368            .await?;
369        self.response_policy
370            .validate(response, self.metadata.request_id_policy())
371            .map_err(PreparedExecutionError::ResponsePolicy)
372    }
373
374    pub(crate) async fn send_async<'transport, 'buffer, T>(
375        &'transport self,
376        transport: &'transport T,
377        response_storage: &'buffer mut [u8],
378        response_header_storage: &'buffer mut [u8],
379    ) -> Result<ResponseBuffer<'buffer>, PreparedExecutionError<T::Error>>
380    where
381        T: AsyncAuthenticatedTransport + BoundTransport,
382        'request: 'transport,
383    {
384        if self.requires_execution_permit() {
385            sanitize_bytes(response_storage);
386            sanitize_bytes(response_header_storage);
387            return Err(PreparedExecutionError::AuthorizationRequired);
388        }
389        self.send_async_authorized(transport, None, response_storage, response_header_storage)
390            .await
391    }
392
393    pub(crate) async fn execute_async_authorized<'transport, 'buffer, T>(
394        &'transport self,
395        transport: &'transport T,
396        confirmed_endpoint: Option<EndpointIdentity<'_>>,
397        response_storage: &'buffer mut [u8],
398        response_header_storage: &'buffer mut [u8],
399    ) -> Result<CheckedResponseGuard<'buffer>, PreparedExecutionError<T::Error>>
400    where
401        T: AsyncAuthenticatedTransport + BoundTransport,
402        'request: 'transport,
403    {
404        let response = self
405            .send_async_authorized(
406                transport,
407                confirmed_endpoint,
408                response_storage,
409                response_header_storage,
410            )
411            .await?;
412        self.response_policy
413            .validate(response, self.metadata.request_id_policy())
414            .map_err(PreparedExecutionError::ResponsePolicy)
415    }
416
417    pub(crate) async fn send_async_authorized<'transport, 'buffer, T>(
418        &'transport self,
419        transport: &'transport T,
420        confirmed_endpoint: Option<EndpointIdentity<'_>>,
421        response_storage: &'buffer mut [u8],
422        response_header_storage: &'buffer mut [u8],
423    ) -> Result<ResponseBuffer<'buffer>, PreparedExecutionError<T::Error>>
424    where
425        T: AsyncAuthenticatedTransport + BoundTransport,
426        'request: 'transport,
427    {
428        let mut response = ResponseBuffer::new(
429            response_storage,
430            self.raw_response_policy.max_body_bytes(),
431            response_header_storage,
432        );
433        self.verify_endpoint(transport, confirmed_endpoint)
434            .map_err(map_endpoint_error)?;
435        drive_async_authenticated(transport, self.authenticated_request(), response.writer())
436            .await
437            .map_err(|error| match error {
438                crate::transport::AsyncExecutionError::Transport(error) => {
439                    PreparedExecutionError::Transport(error)
440                }
441                crate::transport::AsyncExecutionError::Response(error) => {
442                    PreparedExecutionError::ResponseWriter(error)
443                }
444            })?;
445        Ok(response)
446    }
447
448    pub(crate) const fn requires_execution_permit(self) -> bool {
449        !self.request.method().permits_direct_read_only()
450            || !matches!(self.metadata.impact(), OperationImpact::ReadOnly)
451            || matches!(self.metadata.cost_intent(), super::CostIntent::MayIncurCost)
452    }
453
454    fn verify_endpoint<T>(
455        self,
456        transport: &T,
457        confirmed_endpoint: Option<EndpointIdentity<'_>>,
458    ) -> Result<(), EndpointCheckError>
459    where
460        T: BoundTransport,
461    {
462        let actual = transport
463            .endpoint_identity()
464            .map_err(EndpointCheckError::Invalid)?;
465        match confirmed_endpoint {
466            Some(expected) if actual == expected => Ok(()),
467            Some(_) => Err(EndpointCheckError::Mismatch),
468            None => self
469                .service
470                .endpoint_policy
471                .verify(actual)
472                .map_err(|_| EndpointCheckError::Mismatch),
473        }
474    }
475}
476
477impl fmt::Debug for PreparedRequest<'_> {
478    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
479        formatter
480            .debug_struct("PreparedRequest")
481            .field("request", &self.request)
482            .field("service", &self.service)
483            .field("metadata", &self.metadata)
484            .field("response_policy", &self.response_policy)
485            .field("authentication_policy", &self.authentication_policy)
486            .field("raw_response_policy", &self.raw_response_policy)
487            .field("operation_id", &self.operation_id)
488            .field("body_replayability", &self.body_replayability)
489            .finish()
490    }
491}
cloud_sdk/operation/prepared.rs

cloud_sdk/operation/
prepared.rs
