cow_sdk_orderbook/
request.rs

1use std::{future::Future, sync::Arc, time::Duration};
2
3use cow_sdk_core::transport::policy::{
4    AttemptOutcome as RetryOutcome, LimiterKey, RequestRateLimiter, RetryPolicy, RetrySignal,
5    retry_after_from_headers, run_with_retry,
6};
7use cow_sdk_core::{HttpTransport, Redacted, TransportError};
8use http::header::{ACCEPT, CONTENT_TYPE, HeaderMap};
9use serde::de::DeserializeOwned;
10use serde_json::Value;
11use thiserror::Error;
12
13use crate::error::OrderbookError;
14
15/// Shared dyn-compatible [`HttpTransport`] handle threaded through orderbook
16/// request helpers.
17pub(crate) type SharedTransport = Arc<dyn HttpTransport + Send + Sync>;
18
19/// HTTP verb used by the typed orderbook transport.
20///
21/// Internal helper; the SDK chooses the method per request shape. Classified
22/// as `sdk-local-state` in the workspace enum policy manifest.
23#[derive(Debug, Clone, Copy, PartialEq, Eq)]
24pub enum HttpMethod {
25    /// `GET`.
26    Get,
27    /// `POST`.
28    Post,
29    /// `DELETE`.
30    Delete,
31    /// `PUT`.
32    Put,
33}
34
35/// Decoded transport response body.
36///
37/// Internal wrapper around payloads decoded by the typed transport. Classified
38/// as `sdk-local-state` in the workspace enum policy manifest.
39#[allow(
40    clippy::derive_partial_eq_without_eq,
41    reason = "the `Json(serde_json::Value)` variant cannot implement `Eq` because `serde_json::Value` does not implement `Eq`"
42)]
43#[derive(Debug, Clone, PartialEq)]
44pub enum ResponseBody {
45    /// JSON payload.
46    Json(Value),
47    /// Plain-text payload.
48    Text(String),
49    /// Empty response body.
50    Empty,
51}
52
53/// Structured non-2xx error returned by the orderbook transport layer.
54#[derive(Debug, Clone, PartialEq, Error)]
55#[error("{message}")]
56pub struct OrderbookApiError {
57    /// HTTP status code.
58    pub status: u16,
59    /// HTTP status text.
60    pub status_text: Redacted<String>,
61    /// Decoded response body captured from the error response.
62    pub body: Redacted<ResponseBody>,
63    message: Redacted<String>,
64    /// Server-suggested backoff parsed from the `Retry-After` response header,
65    /// when one was present on the failing response.
66    retry_after: Option<Duration>,
67}
68
69impl OrderbookApiError {
70    /// Creates a typed API error from status metadata and a decoded body.
71    ///
72    /// The resulting error carries no `Retry-After` hint; attach one parsed
73    /// from the response headers with [`OrderbookApiError::with_retry_after`].
74    #[must_use]
75    pub fn new(status: u16, status_text: impl Into<String>, body: ResponseBody) -> Self {
76        let status_text = status_text.into();
77        let message = match &body {
78            ResponseBody::Json(Value::Object(map)) => map
79                .get("description")
80                .or_else(|| map.get("error"))
81                .and_then(Value::as_str)
82                .map_or_else(|| status_text.clone(), ToOwned::to_owned),
83            ResponseBody::Json(Value::String(text)) => text.clone(),
84            ResponseBody::Text(text) if !text.is_empty() => text.clone(),
85            _ => status_text.clone(),
86        };
87
88        Self {
89            status,
90            status_text: Redacted::new(status_text),
91            body: Redacted::new(body),
92            message: Redacted::new(message),
93            retry_after: None,
94        }
95    }
96
97    /// Returns this error annotated with a parsed `Retry-After` backoff hint.
98    ///
99    /// `retry_after` is the resolved delay from the failing response's
100    /// `Retry-After` header (see
101    /// [`cow_sdk_core::transport::policy::retry_after_from_headers`]), or [`None`]
102    /// when the server sent no hint.
103    #[must_use]
104    pub const fn with_retry_after(mut self, retry_after: Option<Duration>) -> Self {
105        self.retry_after = retry_after;
106        self
107    }
108
109    /// Returns the server-suggested backoff parsed from the `Retry-After`
110    /// response header, when one was present on the failing response.
111    #[must_use]
112    pub const fn retry_after(&self) -> Option<Duration> {
113        self.retry_after
114    }
115}
116
117/// Low-level request description used by the transport helpers.
118#[derive(Debug, Clone, PartialEq, Eq)]
119pub struct FetchParams {
120    /// Relative API path appended to the resolved base URL.
121    pub path: String,
122    /// HTTP method used for the request.
123    pub method: HttpMethod,
124    /// Query pairs encoded onto the request URL.
125    pub query: Vec<(String, String)>,
126    /// Optional JSON request body.
127    pub body: Option<Value>,
128}
129
130impl FetchParams {
131    /// Creates a request descriptor from a path and method.
132    #[must_use]
133    pub fn new(path: impl Into<String>, method: HttpMethod) -> Self {
134        Self {
135            path: path.into(),
136            method,
137            query: Vec::new(),
138            body: None,
139        }
140    }
141
142    /// Returns a copy of this descriptor with an additional query parameter.
143    #[must_use]
144    pub fn with_query(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
145        self.query.push((key.into(), value.into()));
146        self
147    }
148
149    /// Returns a copy of this descriptor with a JSON request body.
150    #[must_use]
151    pub fn with_body(mut self, body: Value) -> Self {
152        self.body = Some(body);
153        self
154    }
155}
156
157/// Fully decoded HTTP response captured by low-level transport helpers.
158#[derive(Debug, Clone, PartialEq, Eq)]
159pub struct ResponseEnvelope {
160    /// HTTP status code.
161    pub status: u16,
162    /// HTTP status text.
163    pub status_text: String,
164    /// Response content type, when present.
165    pub content_type: Option<String>,
166    /// Raw response bytes.
167    pub body: Vec<u8>,
168}
169
170impl ResponseEnvelope {
171    /// Creates a JSON response envelope.
172    ///
173    /// # Panics
174    ///
175    /// Panics only if serializing an in-memory [`Value`] into a `Vec<u8>`
176    /// unexpectedly fails.
177    #[must_use]
178    pub fn json(status: u16, value: &Value) -> Self {
179        Self {
180            status,
181            status_text: canonical_status_text(status),
182            content_type: Some("application/json".to_owned()),
183            // SAFETY: serde_json::Value serializes to JSON bytes without
184            // relying on caller-controlled type implementations.
185            body: serde_json::to_vec(value).expect("test JSON serialization must succeed"),
186        }
187    }
188
189    /// Creates a plain-text response envelope.
190    #[must_use]
191    pub fn text(status: u16, body: impl Into<String>) -> Self {
192        Self {
193            status,
194            status_text: canonical_status_text(status),
195            content_type: Some("text/plain".to_owned()),
196            body: body.into().into_bytes(),
197        }
198    }
199
200    /// Creates an empty response envelope.
201    #[must_use]
202    pub fn empty(status: u16) -> Self {
203        Self {
204            status,
205            status_text: canonical_status_text(status),
206            content_type: None,
207            body: Vec::new(),
208        }
209    }
210
211    fn decoded_body(&self) -> ResponseBody {
212        if self.status == 204 || self.body.is_empty() {
213            return ResponseBody::Empty;
214        }
215
216        let prefer_json = self.content_type.as_deref().is_none_or(|content_type| {
217            content_type
218                .to_ascii_lowercase()
219                .starts_with("application/json")
220        });
221
222        if prefer_json && let Ok(value) = serde_json::from_slice::<Value>(&self.body) {
223            return ResponseBody::Json(value);
224        }
225
226        ResponseBody::Text(String::from_utf8_lossy(&self.body).into_owned())
227    }
228}
229
230#[derive(Debug, Clone, Copy, PartialEq, Eq)]
231enum ResponseKind {
232    Json,
233    Text,
234    Empty,
235}
236
237impl ResponseKind {
238    const fn accept_header(self) -> &'static str {
239        match self {
240            Self::Text => "text/plain, application/json",
241            Self::Json | Self::Empty => "application/json",
242        }
243    }
244}
245
246enum AttemptOutcome {
247    Response(ResponseEnvelope),
248    HttpError {
249        response: ResponseEnvelope,
250        headers: Vec<(String, String)>,
251    },
252}
253
254struct RequestExecution<'a> {
255    transport: &'a SharedTransport,
256    base_url: &'a str,
257    params: &'a FetchParams,
258    timeout: Option<Duration>,
259    additional_headers: Option<HeaderMap>,
260}
261
262/// Executes a JSON request with an optional per-request timeout override.
263///
264/// # Errors
265///
266/// Returns [`OrderbookError`] when request execution fails, the API returns a
267/// non-success response, or the success body cannot be decoded as JSON.
268pub async fn request_json_with_timeout<T>(
269    transport: &SharedTransport,
270    base_url: &str,
271    params: &FetchParams,
272    policy: &RetryPolicy,
273    rate_limiter: &RequestRateLimiter,
274    timeout: Option<Duration>,
275    additional_headers: Option<HeaderMap>,
276) -> Result<T, OrderbookError>
277where
278    T: DeserializeOwned,
279{
280    request_with(
281        RequestExecution {
282            transport,
283            base_url,
284            params,
285            timeout,
286            additional_headers,
287        },
288        policy,
289        rate_limiter,
290        ResponseKind::Json,
291        decode_success_body::<T>,
292    )
293    .await
294}
295
296/// Executes a text request with an optional per-request timeout override.
297///
298/// # Errors
299///
300/// Returns [`OrderbookError`] when request execution fails, the API returns a
301/// non-success response, or the success body cannot be decoded as UTF-8 text.
302pub async fn request_text_with_timeout(
303    transport: &SharedTransport,
304    base_url: &str,
305    params: &FetchParams,
306    policy: &RetryPolicy,
307    rate_limiter: &RequestRateLimiter,
308    timeout: Option<Duration>,
309    additional_headers: Option<HeaderMap>,
310) -> Result<String, OrderbookError> {
311    request_with(
312        RequestExecution {
313            transport,
314            base_url,
315            params,
316            timeout,
317            additional_headers,
318        },
319        policy,
320        rate_limiter,
321        ResponseKind::Text,
322        decode_text_body,
323    )
324    .await
325}
326
327/// Executes an empty-body request with an optional per-request timeout override.
328///
329/// # Errors
330///
331/// Returns [`OrderbookError`] when request execution fails or the API returns a
332/// non-success response.
333pub async fn request_empty_with_timeout(
334    transport: &SharedTransport,
335    base_url: &str,
336    params: &FetchParams,
337    policy: &RetryPolicy,
338    rate_limiter: &RequestRateLimiter,
339    timeout: Option<Duration>,
340    additional_headers: Option<HeaderMap>,
341) -> Result<(), OrderbookError> {
342    request_with(
343        RequestExecution {
344            transport,
345            base_url,
346            params,
347            timeout,
348            additional_headers,
349        },
350        policy,
351        rate_limiter,
352        ResponseKind::Empty,
353        |_| Ok(()),
354    )
355    .await
356}
357
358/// Executes an abstract JSON-producing attempt with retry and rate-limit policy.
359///
360/// # Errors
361///
362/// Returns [`OrderbookError`] when all attempts fail, the API returns a
363/// non-success response, or the success body cannot be decoded as JSON.
364pub async fn execute_json_with<T, F, Fut>(
365    policy: &RetryPolicy,
366    rate_limiter: &RequestRateLimiter,
367    mut attempt: F,
368) -> Result<T, OrderbookError>
369where
370    T: DeserializeOwned,
371    F: FnMut() -> Fut,
372    Fut: Future<Output = Result<ResponseEnvelope, (cow_sdk_core::TransportErrorClass, String)>>,
373{
374    execute_with(
375        None,
376        policy,
377        rate_limiter,
378        move || {
379            let future = attempt();
380            async move { future.await.map(AttemptOutcome::Response) }
381        },
382        decode_success_body::<T>,
383    )
384    .await
385}
386
387/// Executes an abstract empty-body attempt with retry and rate-limit policy.
388///
389/// # Errors
390///
391/// Returns [`OrderbookError`] when all attempts fail or the API returns a
392/// non-success response.
393pub async fn execute_empty_with<F, Fut>(
394    policy: &RetryPolicy,
395    rate_limiter: &RequestRateLimiter,
396    mut attempt: F,
397) -> Result<(), OrderbookError>
398where
399    F: FnMut() -> Fut,
400    Fut: Future<Output = Result<ResponseEnvelope, (cow_sdk_core::TransportErrorClass, String)>>,
401{
402    execute_with(
403        None,
404        policy,
405        rate_limiter,
406        move || {
407            let future = attempt();
408            async move { future.await.map(AttemptOutcome::Response) }
409        },
410        |_| Ok(()),
411    )
412    .await
413}
414
415async fn request_with<T, D>(
416    request: RequestExecution<'_>,
417    policy: &RetryPolicy,
418    rate_limiter: &RequestRateLimiter,
419    response_kind: ResponseKind,
420    decode_success: D,
421) -> Result<T, OrderbookError>
422where
423    D: Fn(&ResponseEnvelope) -> Result<T, OrderbookError>,
424{
425    let url = format!("{}{}", request.base_url, request.params.path);
426    let limiter_url = url::Url::parse(&url).map_err(|error| OrderbookError::Transport {
427        class: cow_sdk_core::TransportErrorClass::Builder,
428        detail: Redacted::new(format!(
429            "could not parse request URL for rate limiting: {error}"
430        )),
431    })?;
432    let transport = Arc::clone(request.transport);
433    let params = request.params.clone();
434    let timeout = request.timeout;
435    let additional_headers = request.additional_headers;
436
437    execute_with(
438        Some(&limiter_url),
439        policy,
440        rate_limiter,
441        || {
442            send_request(
443                Arc::clone(&transport),
444                url.clone(),
445                params.clone(),
446                timeout,
447                response_kind,
448                additional_headers.clone(),
449            )
450        },
451        decode_success,
452    )
453    .await
454}
455
456async fn send_request(
457    transport: SharedTransport,
458    url: String,
459    params: FetchParams,
460    timeout: Option<Duration>,
461    response_kind: ResponseKind,
462    additional_headers: Option<HeaderMap>,
463) -> Result<AttemptOutcome, (cow_sdk_core::TransportErrorClass, String)> {
464    let full_url = match append_query_string(&url, &params.query) {
465        Ok(url) => url,
466        Err(message) => {
467            return Err((cow_sdk_core::TransportErrorClass::Builder, message));
468        }
469    };
470
471    let body_string = match params.body.as_ref() {
472        Some(value) => match serde_json::to_string(value) {
473            Ok(body) => body,
474            Err(error) => {
475                return Err((
476                    cow_sdk_core::TransportErrorClass::Builder,
477                    format!("could not serialize request body: {error}"),
478                ));
479            }
480        },
481        None => String::new(),
482    };
483
484    let header_pairs = request_header_pairs(response_kind, additional_headers);
485
486    let result = match params.method {
487        HttpMethod::Get => transport.get(&full_url, &header_pairs, timeout).await,
488        HttpMethod::Post => {
489            transport
490                .post(&full_url, &body_string, &header_pairs, timeout)
491                .await
492        }
493        HttpMethod::Put => {
494            transport
495                .put(&full_url, &body_string, &header_pairs, timeout)
496                .await
497        }
498        HttpMethod::Delete => {
499            transport
500                .delete(&full_url, &body_string, &header_pairs, timeout)
501                .await
502        }
503    };
504
505    match result {
506        Ok(response) => {
507            let status = response.status();
508            let content_type = response
509                .header(CONTENT_TYPE.as_str())
510                .map(ToOwned::to_owned);
511            let success = (200..300).contains(&status);
512            let headers: Vec<(String, String)> = if success {
513                Vec::new()
514            } else {
515                response
516                    .headers()
517                    .iter()
518                    .map(|(name, value)| (name.clone(), value.as_inner().clone()))
519                    .collect()
520            };
521            let envelope = ResponseEnvelope {
522                status,
523                status_text: canonical_status_text(status),
524                content_type,
525                body: response.into_body().into_bytes(),
526            };
527            if success {
528                Ok(AttemptOutcome::Response(envelope))
529            } else {
530                // A conforming transport never returns `Ok` for a non-2xx
531                // status; normalize a misbehaving custom transport onto the
532                // HTTP-error outcome so retry classification and
533                // `Retry-After` handling stay uniform.
534                Ok(AttemptOutcome::HttpError {
535                    response: envelope,
536                    headers,
537                })
538            }
539        }
540        Err(TransportError::HttpStatus {
541            status,
542            headers,
543            body,
544        }) => {
545            let content_type = headers
546                .iter()
547                .find(|(name, _)| name.eq_ignore_ascii_case(CONTENT_TYPE.as_str()))
548                .map(|(_, value)| value.as_inner().clone());
549            Ok(AttemptOutcome::HttpError {
550                response: ResponseEnvelope {
551                    status,
552                    status_text: canonical_status_text(status),
553                    content_type,
554                    body: body.into_inner().into_bytes(),
555                },
556                headers: headers
557                    .into_iter()
558                    .map(|(name, value)| (name, value.into_inner()))
559                    .collect(),
560            })
561        }
562        Err(TransportError::Transport { class, detail }) => Err((class, detail.into_inner())),
563        Err(TransportError::Configuration { message }) => Err((
564            cow_sdk_core::TransportErrorClass::Builder,
565            message.into_inner(),
566        )),
567        Err(other) => Err((cow_sdk_core::TransportErrorClass::Other, other.to_string())),
568    }
569}
570
571fn append_query_string(url: &str, query: &[(String, String)]) -> Result<String, String> {
572    if query.is_empty() {
573        return Ok(url.to_owned());
574    }
575    url::Url::parse_with_params(
576        url,
577        query
578            .iter()
579            .map(|(key, value)| (key.as_str(), value.as_str())),
580    )
581    .map(String::from)
582    .map_err(|error| format!("could not encode query parameters: {error}"))
583}
584
585fn request_header_pairs(
586    response_kind: ResponseKind,
587    additional_headers: Option<HeaderMap>,
588) -> Vec<(String, String)> {
589    let mut pairs = Vec::with_capacity(2 + additional_headers.as_ref().map_or(0, HeaderMap::len));
590    pairs.push((ACCEPT.to_string(), response_kind.accept_header().to_owned()));
591    pairs.push((CONTENT_TYPE.to_string(), "application/json".to_owned()));
592    if let Some(extra) = additional_headers {
593        for (name, value) in &extra {
594            let Ok(value_str) = value.to_str() else {
595                continue;
596            };
597            pairs.push((name.as_str().to_owned(), value_str.to_owned()));
598        }
599    }
600    pairs
601}
602
603async fn execute_with<T, F, Fut, D>(
604    limiter_url: Option<&url::Url>,
605    policy: &RetryPolicy,
606    rate_limiter: &RequestRateLimiter,
607    mut attempt: F,
608    decode_success: D,
609) -> Result<T, OrderbookError>
610where
611    F: FnMut() -> Fut,
612    Fut: Future<Output = Result<AttemptOutcome, (cow_sdk_core::TransportErrorClass, String)>>,
613    D: Fn(&ResponseEnvelope) -> Result<T, OrderbookError>,
614{
615    // The shared driver in `cow_sdk_core::transport::policy` owns the retry loop,
616    // rate-limit acquisition, backoff, `Retry-After` clock, and retry telemetry.
617    // The closure performs one dispatch and classifies the result into the
618    // unified outcome space; the success envelope is decoded after the driver
619    // returns so a decode failure stays terminal and is never retried.
620    let limiter_key = limiter_url.map_or(LimiterKey::Global, LimiterKey::PerUrl);
621    let response = run_with_retry::<ResponseEnvelope, OrderbookError, _, _>(
622        policy,
623        rate_limiter,
624        limiter_key,
625        |attempt_index| {
626            let future = attempt();
627            async move {
628                #[cfg(feature = "tracing")]
629                record_span_attempts(attempt_index);
630                #[cfg(not(feature = "tracing"))]
631                let _ = attempt_index;
632
633                match future.await {
634                    Ok(AttemptOutcome::Response(response))
635                        if (200..300).contains(&response.status) =>
636                    {
637                        #[cfg(feature = "tracing")]
638                        record_span_status(response.status);
639                        RetryOutcome::Success(response)
640                    }
641                    Ok(outcome) => {
642                        let (response, headers) = match outcome {
643                            AttemptOutcome::Response(response) => (response, Vec::new()),
644                            AttemptOutcome::HttpError { response, headers } => (response, headers),
645                        };
646                        #[cfg(feature = "tracing")]
647                        record_span_status(response.status);
648                        let status = response.status;
649                        let body = response.decoded_body();
650                        // Resolve the server `Retry-After` hint while the
651                        // response headers are still in scope, then attach it to
652                        // the terminal error so a caller can read the suggested
653                        // backoff after the retry budget is exhausted. The retry
654                        // driver computes its own clock-injected backoff and does
655                        // not depend on this value.
656                        let retry_after = retry_after_from_headers(&headers);
657                        let error = OrderbookApiError::new(status, response.status_text, body)
658                            .with_retry_after(retry_after);
659                        RetryOutcome::Failure {
660                            error: error.into(),
661                            signal: RetrySignal::HttpStatus { status, headers },
662                        }
663                    }
664                    Err((class, detail)) => RetryOutcome::Failure {
665                        error: OrderbookError::Transport {
666                            class,
667                            detail: Redacted::new(detail),
668                        },
669                        signal: RetrySignal::Transport { class },
670                    },
671                }
672            }
673        },
674    )
675    .await?;
676
677    decode_success(&response)
678}
679
680#[cfg(feature = "tracing")]
681fn record_span_attempts(attempt_index: usize) {
682    let attempts = u64::try_from(attempt_index).unwrap_or(u64::MAX);
683    tracing::Span::current().record("attempts", attempts);
684}
685
686#[cfg(feature = "tracing")]
687fn record_span_status(status: u16) {
688    tracing::Span::current().record("status", u64::from(status));
689}
690
691fn decode_success_body<T>(response: &ResponseEnvelope) -> Result<T, OrderbookError>
692where
693    T: DeserializeOwned,
694{
695    serde_json::from_slice::<T>(&response.body).map_err(OrderbookError::from)
696}
697
698fn decode_text_body(response: &ResponseEnvelope) -> Result<String, OrderbookError> {
699    String::from_utf8(response.body.clone()).map_err(|error| OrderbookError::Transport {
700        class: cow_sdk_core::TransportErrorClass::Decode,
701        detail: Redacted::new(error.to_string()),
702    })
703}
704
705fn canonical_status_text(status: u16) -> String {
706    http::StatusCode::from_u16(status)
707        .ok()
708        .and_then(|status| status.canonical_reason().map(ToOwned::to_owned))
709        .unwrap_or_else(|| "Unknown Status".to_owned())
710}
cow_sdk_orderbook/request.rs

cow_sdk_orderbook/
request.rs
