olai_http/
service.rs

1use std::fmt::Debug;
2use std::future::Future;
3use std::pin::Pin;
4use std::sync::Arc;
5
6use tokio::runtime::Handle;
7
8/// An abstraction over HTTP request execution.
9///
10/// This trait allows decoupling request building (which uses `reqwest::RequestBuilder`)
11/// from request execution. The primary use case is [`SpawnService`], which spawns
12/// execution on a dedicated I/O runtime -- useful when the CPU runtime has I/O disabled.
13pub trait HttpService: Debug + Send + Sync + 'static {
14    /// Execute an HTTP request, returning the response.
15    ///
16    /// The error type is [`crate::Error`]. Implementations that perform the
17    /// actual network I/O surface transport failures as
18    /// [`crate::Error::ReqwestError`], which the retry layer inspects to decide
19    /// whether a failure is retryable. Other variants (e.g. a cancelled I/O
20    /// task) are treated as non-retryable.
21    fn call(
22        &self,
23        request: reqwest::Request,
24    ) -> Pin<Box<dyn Future<Output = crate::Result<reqwest::Response>> + Send + '_>>;
25}
26
27/// Default [`HttpService`] that delegates directly to [`reqwest::Client::execute`].
28#[derive(Debug, Clone)]
29pub struct ReqwestService(reqwest::Client);
30
31impl ReqwestService {
32    pub fn new(client: reqwest::Client) -> Self {
33        Self(client)
34    }
35}
36
37impl HttpService for ReqwestService {
38    fn call(
39        &self,
40        request: reqwest::Request,
41    ) -> Pin<Box<dyn Future<Output = crate::Result<reqwest::Response>> + Send + '_>> {
42        let client = self.0.clone();
43        Box::pin(async move { Ok(client.execute(request).await?) })
44    }
45}
46
47/// An [`HttpService`] that spawns each request on a separate tokio runtime.
48///
49/// This is useful when the calling runtime (e.g. a CPU-bound DataFusion runtime)
50/// may have I/O disabled. All HTTP I/O -- including credential refresh -- is
51/// routed through the provided runtime handle.
52///
53/// # Example
54///
55/// ```ignore
56/// use olai_http::CloudClient;
57///
58/// let io_runtime = tokio::runtime::Runtime::new().unwrap();
59/// let client = CloudClient::new_with_token("tok")
60///     .with_runtime(io_runtime.handle().clone());
61/// ```
62#[derive(Debug)]
63pub struct SpawnService {
64    inner: Arc<dyn HttpService>,
65    handle: Handle,
66}
67
68impl SpawnService {
69    pub fn new(inner: Arc<dyn HttpService>, handle: Handle) -> Self {
70        Self { inner, handle }
71    }
72}
73
74impl HttpService for SpawnService {
75    fn call(
76        &self,
77        request: reqwest::Request,
78    ) -> Pin<Box<dyn Future<Output = crate::Result<reqwest::Response>> + Send + '_>> {
79        let inner = Arc::clone(&self.inner);
80        let handle = self.handle.clone();
81        Box::pin(async move {
82            match handle.spawn(async move { inner.call(request).await }).await {
83                Ok(result) => result,
84                Err(join_err) if join_err.is_cancelled() => {
85                    // The spawned I/O task was cancelled, typically because the
86                    // I/O runtime is being shut down concurrently with an
87                    // in-flight request (a graceful-shutdown race). This is not a
88                    // bug in our code, so surface it as an error rather than
89                    // crashing the client.
90                    Err(crate::Error::Generic {
91                        source: Box::new(join_err),
92                    })
93                }
94                Err(join_err) => {
95                    // The spawned I/O task genuinely panicked. That indicates a
96                    // programming error (e.g. a bug in request execution), not a
97                    // transient network failure, so we re-panic with context
98                    // rather than silently swallowing the cause.
99                    panic!("I/O runtime task panicked: {join_err}")
100                }
101            }
102        })
103    }
104}
105
106/// Create an [`HttpService`] for the given client, optionally wrapping in [`SpawnService`].
107pub(crate) fn make_service(
108    client: reqwest::Client,
109    runtime: Option<&Handle>,
110) -> Arc<dyn HttpService> {
111    let base: Arc<dyn HttpService> = Arc::new(ReqwestService::new(client));
112    match runtime {
113        Some(handle) => Arc::new(SpawnService::new(base, handle.clone())),
114        None => base,
115    }
116}
olai_http/service.rs

olai_http/
service.rs
