aspero_engine/
transport.rs

1//! The engine's side of the [`HttpClient`] seam: a real client, and pacing.
2
3use std::sync::Arc;
4use std::time::Duration;
5
6use aspero_core::http::Method;
7use aspero_core::{Error, HttpClient, HttpRequest, HttpResponse};
8use async_trait::async_trait;
9use jiff::Timestamp;
10use tokio::sync::Mutex;
11use tokio::time::Instant;
12
13/// `HttpClient` over `reqwest` (rustls). One per provider, so each gets its
14/// own attempt timeout and connection pool.
15pub struct ReqwestClient {
16    client: reqwest::Client,
17}
18
19impl ReqwestClient {
20    pub fn new(user_agent: &str, attempt_timeout: Duration) -> Result<Self, Error> {
21        let client = reqwest::Client::builder()
22            .user_agent(user_agent)
23            .timeout(attempt_timeout)
24            .build()
25            .map_err(Error::network)?;
26        Ok(ReqwestClient { client })
27    }
28}
29
30#[async_trait]
31impl HttpClient for ReqwestClient {
32    async fn send(&self, request: HttpRequest) -> Result<HttpResponse, Error> {
33        let method = match request.method {
34            Method::Post => reqwest::Method::POST,
35            _ => reqwest::Method::GET,
36        };
37        let mut builder = self.client.request(method, &request.url);
38        for (name, value) in &request.headers {
39            builder = builder.header(name, value);
40        }
41        if let Some(body) = request.body {
42            builder = builder.body(body);
43        }
44        let resp = builder.send().await.map_err(transport_error)?;
45        let status = resp.status().as_u16();
46        let headers = resp
47            .headers()
48            .iter()
49            .filter_map(|(k, v)| Some((k.as_str().to_owned(), v.to_str().ok()?.to_owned())))
50            .collect();
51        let body = resp.bytes().await.map_err(transport_error)?.to_vec();
52        Ok(HttpResponse::new(status, headers, body, Timestamp::now()))
53    }
54}
55
56fn transport_error(e: reqwest::Error) -> Error {
57    if e.is_timeout() {
58        Error::Timeout
59    } else {
60        Error::network(e)
61    }
62}
63
64/// Spaces request *starts* at least `min_interval` apart, across all tasks
65/// sharing it. Slots are handed out in arrival order, so waiting is fair.
66///
67/// In-process only; the cross-process limiter (DESIGN §6) is deferred.
68///
69/// Logs each request (at `info`, target `aspero::http`) with the provider
70/// label, status, duration and time spent waiting for a slot.
71pub struct Paced {
72    label: String,
73    inner: Arc<dyn HttpClient>,
74    min_interval: Duration,
75    next_slot: Mutex<Option<Instant>>,
76}
77
78impl Paced {
79    pub fn new(
80        label: impl Into<String>,
81        inner: Arc<dyn HttpClient>,
82        min_interval: Duration,
83    ) -> Self {
84        Paced {
85            label: label.into(),
86            inner,
87            min_interval,
88            next_slot: Mutex::new(None),
89        }
90    }
91
92    /// Claims the next slot and returns when it starts.
93    async fn claim(&self) -> Instant {
94        let mut next = self.next_slot.lock().await;
95        let now = Instant::now();
96        let slot = next.map_or(now, |n| n.max(now));
97        *next = Some(slot + self.min_interval);
98        slot
99    }
100}
101
102#[async_trait]
103impl HttpClient for Paced {
104    async fn send(&self, request: HttpRequest) -> Result<HttpResponse, Error> {
105        let queued = Instant::now();
106        let slot = self.claim().await;
107        tokio::time::sleep_until(slot).await;
108        let started = Instant::now();
109        let url = loggable_url(&request.url);
110        let result = self.inner.send(request).await;
111        let ms = |since: Instant, until: Instant| until.duration_since(since).as_millis() as u64;
112        let (waited_ms, took_ms) = (ms(queued, started), ms(started, Instant::now()));
113        match &result {
114            Ok(resp) => tracing::info!(
115                target: "aspero::http",
116                provider = %self.label, status = resp.status, took_ms, waited_ms, "{url}"
117            ),
118            Err(e) => tracing::warn!(
119                target: "aspero::http",
120                provider = %self.label, error = %e, took_ms, waited_ms, "{url}"
121            ),
122        }
123        result
124    }
125}
126
127/// A URL for logs: path and query only, contact and key parameters
128/// redacted (logs get pasted into issues), long queries shortened.
129fn loggable_url(url: &str) -> String {
130    let path = url
131        .split_once("://")
132        .map_or(url, |(_, rest)| rest.find('/').map_or("/", |i| &rest[i..]));
133    let redacted: String = match path.split_once('?') {
134        None => path.to_owned(),
135        Some((p, query)) => {
136            let params: Vec<String> = query
137                .split('&')
138                .map(|kv| match kv.split_once('=') {
139                    Some((k, _)) if matches!(k, "mailto" | "api_key" | "key") => format!("{k}=…"),
140                    _ => kv.to_owned(),
141                })
142                .collect();
143            format!("{p}?{}", params.join("&"))
144        }
145    };
146    const MAX: usize = 140;
147    match redacted.char_indices().nth(MAX) {
148        Some((cut, _)) => format!("{}…", &redacted[..cut]),
149        None => redacted,
150    }
151}
152
153#[cfg(test)]
154mod tests {
155    use super::*;
156
157    struct Echo;
158
159    #[async_trait]
160    impl HttpClient for Echo {
161        async fn send(&self, _: HttpRequest) -> Result<HttpResponse, Error> {
162            Ok(HttpResponse::new(
163                200,
164                vec![],
165                vec![],
166                Timestamp::UNIX_EPOCH,
167            ))
168        }
169    }
170
171    #[tokio::test(start_paused = true)]
172    async fn paced_requests_start_min_interval_apart() {
173        let paced = Arc::new(Paced::new(
174            "test",
175            Arc::new(Echo),
176            Duration::from_millis(250),
177        ));
178        let start = Instant::now();
179        let mut tasks = Vec::new();
180        for _ in 0..4 {
181            let paced = paced.clone();
182            tasks.push(tokio::spawn(async move {
183                paced.send(HttpRequest::get("http://x")).await.unwrap();
184                start.elapsed().as_millis()
185            }));
186        }
187        let mut starts = Vec::new();
188        for t in tasks {
189            starts.push(t.await.unwrap());
190        }
191        starts.sort_unstable();
192        assert_eq!(starts, [0, 250, 500, 750]);
193    }
194
195    #[test]
196    fn logged_urls_hide_contacts_and_keys() {
197        assert_eq!(
198            loggable_url("https://api.crossref.org/works?query=x&mailto=me%40example.org&rows=3"),
199            "/works?query=x&mailto=…&rows=3"
200        );
201        assert_eq!(
202            loggable_url("https://doi.org/10.1038/nature14539"),
203            "/10.1038/nature14539"
204        );
205    }
206
207    #[tokio::test(start_paused = true)]
208    async fn idle_time_is_not_banked() {
209        let paced = Paced::new("test", Arc::new(Echo), Duration::from_millis(100));
210        paced.send(HttpRequest::get("http://x")).await.unwrap();
211        tokio::time::sleep(Duration::from_secs(5)).await;
212        let start = Instant::now();
213        paced.send(HttpRequest::get("http://x")).await.unwrap();
214        paced.send(HttpRequest::get("http://x")).await.unwrap();
215        assert_eq!(start.elapsed().as_millis(), 100);
216    }
217}
aspero_engine/transport.rs

aspero_engine/
transport.rs
