cloud_sdk/pagination/header_cursor/
execution.rs

1use core::fmt;
2
3use cloud_sdk_sanitization::sanitize_bytes;
4
5use crate::authentication::BlockingAuthenticatedTransport;
6use crate::operation::{CheckedResponseGuard, PreparedExecutionError, PreparedRequest};
7use crate::transport::{
8    BoundTransport, EndpointIdentity, MAX_REQUEST_HEADERS, RequestHeader, RequestHeaders,
9};
10
11use super::{DecodedHeaderCursor, HeaderCursorPolicy};
12use crate::pagination::{
13    CursorDigest, CursorHistory, PaginationCursor, PaginationError, PaginationLimits,
14};
15
16mod asynchronous;
17mod local_async;
18
19/// Cursor validation or prepared-request execution failure.
20pub enum HeaderCursorExecutionError<E> {
21    /// Cursor policy, provenance, storage, or header validation failed.
22    Pagination(PaginationError),
23    /// The exact retained prepared request failed to execute.
24    Prepared(PreparedExecutionError<E>),
25}
26
27impl<E> fmt::Debug for HeaderCursorExecutionError<E> {
28    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
29        match self {
30            Self::Pagination(error) => formatter.debug_tuple("Pagination").field(error).finish(),
31            Self::Prepared(_) => formatter.write_str("Prepared([redacted])"),
32        }
33    }
34}
35
36impl<E> fmt::Display for HeaderCursorExecutionError<E> {
37    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
38        formatter.write_str(match self {
39            Self::Pagination(_) => "header cursor validation failed",
40            Self::Prepared(_) => "header cursor prepared request failed",
41        })
42    }
43}
44
45impl<E> core::error::Error for HeaderCursorExecutionError<E> {}
46
47/// Exact prepared-request context retained for one header-cursor traversal.
48#[derive(Clone, Copy)]
49pub struct HeaderCursorSession<'request, 'policy> {
50    pub(super) policy: HeaderCursorPolicy<'policy>,
51    pub(super) prepared: PreparedRequest<'request>,
52}
53
54impl<'policy> HeaderCursorPolicy<'policy> {
55    /// Binds pagination to one complete prepared request before any dispatch.
56    pub fn bind<'request>(
57        self,
58        prepared: PreparedRequest<'request>,
59    ) -> Result<HeaderCursorSession<'request, 'policy>, PaginationError> {
60        if prepared.operation_id() != Some(self.operation_id()) {
61            return Err(PaginationError::OperationMismatch);
62        }
63        if !prepared
64            .raw_response_policy()
65            .admits_header(self.next_response().as_str())
66        {
67            return Err(PaginationError::ResponseHeaderNotAdmitted);
68        }
69        Ok(HeaderCursorSession {
70            policy: self,
71            prepared,
72        })
73    }
74}
75
76impl HeaderCursorSession<'_, '_> {
77    /// Returns the operation retained by this exact request context.
78    #[must_use]
79    pub const fn operation_id(&self) -> crate::operation::OperationId {
80        self.policy.operation_id()
81    }
82
83    /// Executes the initial blocking request and decodes only its own response.
84    #[allow(clippy::too_many_arguments)]
85    pub fn execute_blocking<'response, 'cursor, 'endpoint, T>(
86        &self,
87        transport: &'endpoint T,
88        response_storage: &'response mut [u8],
89        response_header_storage: &'response mut [u8],
90        decimal_scratch: &mut [u8],
91        transfer_scratch: &mut [u8],
92        cursor_destination: &'cursor mut [u8],
93        limits: PaginationLimits,
94    ) -> Result<
95        HeaderCursorPage<'response, 'cursor, 'endpoint, '_, '_, '_>,
96        HeaderCursorExecutionError<T::Error>,
97    >
98    where
99        T: BlockingAuthenticatedTransport + BoundTransport,
100    {
101        clear_execution_buffers(
102            response_storage,
103            response_header_storage,
104            decimal_scratch,
105            transfer_scratch,
106            cursor_destination,
107        );
108        let endpoint = transport.endpoint_identity().map_err(|error| {
109            HeaderCursorExecutionError::Prepared(PreparedExecutionError::EndpointIdentity(error))
110        })?;
111        execute_blocking(
112            self,
113            None,
114            endpoint,
115            transport,
116            response_storage,
117            response_header_storage,
118            decimal_scratch,
119            transfer_scratch,
120            cursor_destination,
121            limits,
122        )
123    }
124}
125
126impl fmt::Debug for HeaderCursorSession<'_, '_> {
127    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
128        formatter
129            .debug_struct("HeaderCursorSession")
130            .field("policy", &self.policy)
131            .field("prepared", &"[bound]")
132            .finish()
133    }
134}
135
136/// Checked response and provenance-bound next-page state.
137pub struct HeaderCursorPage<'response, 'cursor, 'endpoint, 'session, 'request, 'policy> {
138    response: CheckedResponseGuard<'response>,
139    next: HeaderCursorNext<'cursor, 'endpoint, 'session, 'request, 'policy>,
140}
141
142impl<'response, 'cursor, 'endpoint, 'session, 'request, 'policy>
143    HeaderCursorPage<'response, 'cursor, 'endpoint, 'session, 'request, 'policy>
144{
145    /// Separates the checked response from its next-page state.
146    #[must_use]
147    pub fn into_parts(
148        self,
149    ) -> (
150        CheckedResponseGuard<'response>,
151        HeaderCursorNext<'cursor, 'endpoint, 'session, 'request, 'policy>,
152    ) {
153        (self.response, self.next)
154    }
155}
156
157impl fmt::Debug for HeaderCursorPage<'_, '_, '_, '_, '_, '_> {
158    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
159        formatter
160            .debug_struct("HeaderCursorPage")
161            .field("response", &self.response)
162            .field("next", &self.next)
163            .finish()
164    }
165}
166
167/// Provenance-bound continuation state from one executed prepared request.
168pub enum HeaderCursorNext<'cursor, 'endpoint, 'session, 'request, 'policy> {
169    /// The next-cursor header was absent, so traversal is complete.
170    Complete,
171    /// The exact request context and bounded cursor are retained together.
172    Continue(HeaderCursorContinuation<'cursor, 'endpoint, 'session, 'request, 'policy>),
173}
174
175impl HeaderCursorNext<'_, '_, '_, '_, '_> {
176    /// Reports whether the provider declared a terminal page.
177    #[must_use]
178    pub const fn is_complete(&self) -> bool {
179        matches!(self, Self::Complete)
180    }
181}
182
183impl fmt::Debug for HeaderCursorNext<'_, '_, '_, '_, '_> {
184    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
185        match self {
186            Self::Complete => formatter.write_str("HeaderCursorNext::Complete"),
187            Self::Continue(_) => formatter.write_str("HeaderCursorNext::Continue([redacted])"),
188        }
189    }
190}
191
192/// Cleanup-owning cursor inseparable from its original prepared request.
193///
194/// Continuations expose execution only; another request cannot be attached.
195///
196/// ```compile_fail
197/// use cloud_sdk::operation::PreparedRequest;
198/// use cloud_sdk::pagination::HeaderCursorContinuation;
199///
200/// fn rebind(
201///     continuation: HeaderCursorContinuation<'_, '_, '_, '_, '_>,
202///     replacement: PreparedRequest<'_>,
203/// ) {
204///     continuation.bind(replacement);
205/// }
206/// ```
207pub struct HeaderCursorContinuation<'cursor, 'endpoint, 'session, 'request, 'policy> {
208    pub(super) session: &'session HeaderCursorSession<'request, 'policy>,
209    pub(super) cursor: PaginationCursor<'cursor>,
210    pub(super) endpoint: EndpointIdentity<'endpoint>,
211}
212
213impl<'cursor, 'endpoint, 'session, 'request, 'policy>
214    HeaderCursorContinuation<'cursor, 'endpoint, 'session, 'request, 'policy>
215{
216    /// Returns the provider operation that produced this continuation.
217    #[must_use]
218    pub const fn operation_id(&self) -> crate::operation::OperationId {
219        self.session.operation_id()
220    }
221
222    /// Checks and transactionally records the exact cursor in caller history.
223    pub fn observe_history(
224        &self,
225        history: &mut CursorHistory<'_>,
226        digest: CursorDigest,
227    ) -> Result<(), PaginationError> {
228        history.observe(&self.cursor, digest)
229    }
230
231    /// Executes the next blocking request using the retained request context.
232    #[allow(clippy::too_many_arguments)]
233    pub fn execute_blocking<'response, 'next, T>(
234        &self,
235        transport: &T,
236        response_storage: &'response mut [u8],
237        response_header_storage: &'response mut [u8],
238        decimal_scratch: &mut [u8],
239        transfer_scratch: &mut [u8],
240        cursor_destination: &'next mut [u8],
241        limits: PaginationLimits,
242    ) -> Result<
243        HeaderCursorPage<'response, 'next, 'endpoint, 'session, 'request, 'policy>,
244        HeaderCursorExecutionError<T::Error>,
245    >
246    where
247        T: BlockingAuthenticatedTransport + BoundTransport,
248    {
249        clear_execution_buffers(
250            response_storage,
251            response_header_storage,
252            decimal_scratch,
253            transfer_scratch,
254            cursor_destination,
255        );
256        let endpoint = transport.endpoint_identity().map_err(|error| {
257            HeaderCursorExecutionError::Prepared(PreparedExecutionError::EndpointIdentity(error))
258        })?;
259        if endpoint != self.endpoint {
260            return Err(HeaderCursorExecutionError::Pagination(
261                PaginationError::EndpointMismatch,
262            ));
263        }
264        execute_blocking(
265            self.session,
266            Some(&self.cursor),
267            self.endpoint,
268            transport,
269            response_storage,
270            response_header_storage,
271            decimal_scratch,
272            transfer_scratch,
273            cursor_destination,
274            limits,
275        )
276    }
277}
278
279impl fmt::Debug for HeaderCursorContinuation<'_, '_, '_, '_, '_> {
280    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
281        formatter
282            .debug_struct("HeaderCursorContinuation")
283            .field("operation", &self.operation_id())
284            .field("request", &"[bound]")
285            .field("cursor", &"[redacted]")
286            .finish()
287    }
288}
289
290pub(super) fn clear_execution_buffers(
291    response_storage: &mut [u8],
292    response_header_storage: &mut [u8],
293    decimal_scratch: &mut [u8],
294    transfer_scratch: &mut [u8],
295    cursor_destination: &mut [u8],
296) {
297    sanitize_bytes(response_storage);
298    sanitize_bytes(response_header_storage);
299    sanitize_bytes(decimal_scratch);
300    sanitize_bytes(transfer_scratch);
301    sanitize_bytes(cursor_destination);
302}
303
304#[allow(clippy::too_many_arguments)]
305fn execute_blocking<'response, 'cursor, 'endpoint, 'session, 'request, 'policy, T>(
306    session: &'session HeaderCursorSession<'request, 'policy>,
307    cursor: Option<&PaginationCursor<'_>>,
308    endpoint: EndpointIdentity<'endpoint>,
309    transport: &T,
310    response_storage: &'response mut [u8],
311    response_header_storage: &'response mut [u8],
312    decimal_scratch: &mut [u8],
313    transfer_scratch: &mut [u8],
314    cursor_destination: &'cursor mut [u8],
315    limits: PaginationLimits,
316) -> Result<
317    HeaderCursorPage<'response, 'cursor, 'endpoint, 'session, 'request, 'policy>,
318    HeaderCursorExecutionError<T::Error>,
319>
320where
321    T: BlockingAuthenticatedTransport + BoundTransport,
322{
323    let response = session
324        .policy
325        .with_request_headers(cursor, decimal_scratch, |pagination| {
326            with_merged_request(&session.prepared, pagination, |prepared| {
327                prepared.execute_blocking(transport, response_storage, response_header_storage)
328            })
329        })
330        .map_err(HeaderCursorExecutionError::Pagination)?
331        .map_err(HeaderCursorExecutionError::Pagination)?
332        .map_err(HeaderCursorExecutionError::Prepared)?;
333    finish_page(
334        session,
335        endpoint,
336        response,
337        transfer_scratch,
338        cursor_destination,
339        limits,
340    )
341}
342
343pub(super) fn finish_page<'response, 'cursor, 'endpoint, 'session, 'request, 'policy, E>(
344    session: &'session HeaderCursorSession<'request, 'policy>,
345    endpoint: EndpointIdentity<'endpoint>,
346    response: CheckedResponseGuard<'response>,
347    transfer_scratch: &mut [u8],
348    cursor_destination: &'cursor mut [u8],
349    limits: PaginationLimits,
350) -> Result<
351    HeaderCursorPage<'response, 'cursor, 'endpoint, 'session, 'request, 'policy>,
352    HeaderCursorExecutionError<E>,
353> {
354    let next = session
355        .policy
356        .decode_next(
357            response.response_headers(),
358            transfer_scratch,
359            cursor_destination,
360            limits,
361        )
362        .map_err(HeaderCursorExecutionError::Pagination)?;
363    let next = match next {
364        DecodedHeaderCursor::Complete => HeaderCursorNext::Complete,
365        DecodedHeaderCursor::Continue(cursor) => {
366            HeaderCursorNext::Continue(HeaderCursorContinuation {
367                session,
368                cursor,
369                endpoint,
370            })
371        }
372    };
373    Ok(HeaderCursorPage { response, next })
374}
375
376pub(super) fn with_merged_request<'request, R>(
377    prepared: &PreparedRequest<'request>,
378    pagination: RequestHeaders<'_>,
379    inspect: impl FnOnce(PreparedRequest<'_>) -> R,
380) -> Result<R, PaginationError> {
381    let base = prepared.transport_request().headers().as_slice();
382    let extra = pagination.as_slice();
383    let count = base
384        .len()
385        .checked_add(extra.len())
386        .ok_or(PaginationError::RequestHeaderConflict)?;
387    if count > MAX_REQUEST_HEADERS {
388        return Err(PaginationError::RequestHeaderConflict);
389    }
390    let first = extra
391        .first()
392        .copied()
393        .ok_or(PaginationError::InvalidHeaderState)?;
394    let mut entries: [RequestHeader<'_>; MAX_REQUEST_HEADERS] = [first; MAX_REQUEST_HEADERS];
395    entries
396        .get_mut(..base.len())
397        .ok_or(PaginationError::RequestHeaderConflict)?
398        .copy_from_slice(base);
399    entries
400        .get_mut(base.len()..count)
401        .ok_or(PaginationError::RequestHeaderConflict)?
402        .copy_from_slice(extra);
403    let selected = entries
404        .get(..count)
405        .ok_or(PaginationError::RequestHeaderConflict)?;
406    let headers =
407        RequestHeaders::new(selected).map_err(|_| PaginationError::RequestHeaderConflict)?;
408    Ok(inspect((*prepared).with_request_headers(headers)))
409}
410
411#[cfg(test)]
412mod tests {
413    use super::clear_execution_buffers;
414
415    #[test]
416    fn predispatch_cleanup_clears_every_caller_buffer() {
417        let mut response = [0xa5; 3];
418        let mut headers = [0xa5; 5];
419        let mut decimal = [0xa5; 7];
420        let mut transfer = [0xa5; 11];
421        let mut cursor = [0xa5; 13];
422        clear_execution_buffers(
423            &mut response,
424            &mut headers,
425            &mut decimal,
426            &mut transfer,
427            &mut cursor,
428        );
429        assert_eq!(response, [0; 3]);
430        assert_eq!(headers, [0; 5]);
431        assert_eq!(decimal, [0; 7]);
432        assert_eq!(transfer, [0; 11]);
433        assert_eq!(cursor, [0; 13]);
434    }
435}
cloud_sdk/pagination/header_cursor/execution.rs

cloud_sdk/pagination/header_cursor/
execution.rs
