matrix_sdk/http_client/
mod.rs

1// Copyright 2020 The Matrix.org Foundation C.I.C.
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::{
16    any::type_name,
17    fmt::Debug,
18    num::NonZeroUsize,
19    sync::{
20        atomic::{AtomicU64, Ordering},
21        Arc,
22    },
23    time::Duration,
24};
25
26use bytes::{Bytes, BytesMut};
27use bytesize::ByteSize;
28use eyeball::SharedObservable;
29use http::Method;
30use ruma::api::{
31    error::{FromHttpResponseError, IntoHttpError},
32    AuthScheme, OutgoingRequest, SendAccessToken, SupportedVersions,
33};
34use tokio::sync::{Semaphore, SemaphorePermit};
35use tracing::{debug, field::debug, instrument, trace};
36
37use crate::{config::RequestConfig, error::HttpError};
38
39#[cfg(not(target_family = "wasm"))]
40mod native;
41#[cfg(target_family = "wasm")]
42mod wasm;
43
44#[cfg(not(target_family = "wasm"))]
45pub(crate) use native::HttpSettings;
46
47pub(crate) const DEFAULT_REQUEST_TIMEOUT: Duration = Duration::from_secs(30);
48
49#[derive(Clone, Debug)]
50struct MaybeSemaphore(Arc<Option<Semaphore>>);
51
52#[allow(dead_code)] // false-positive lint: we never use it but only hold it for the drop
53struct MaybeSemaphorePermit<'a>(Option<SemaphorePermit<'a>>);
54
55impl MaybeSemaphore {
56    fn new(max: Option<NonZeroUsize>) -> Self {
57        let inner = max.map(|i| Semaphore::new(i.into()));
58        MaybeSemaphore(Arc::new(inner))
59    }
60
61    async fn acquire(&self) -> MaybeSemaphorePermit<'_> {
62        match self.0.as_ref() {
63            Some(inner) => {
64                // This can only ever error if the semaphore was closed,
65                // which we never do, so we can safely ignore any error case
66                MaybeSemaphorePermit(inner.acquire().await.ok())
67            }
68            None => MaybeSemaphorePermit(None),
69        }
70    }
71}
72
73#[derive(Clone, Debug)]
74pub(crate) struct HttpClient {
75    pub(crate) inner: reqwest::Client,
76    pub(crate) request_config: RequestConfig,
77    concurrent_request_semaphore: MaybeSemaphore,
78    next_request_id: Arc<AtomicU64>,
79}
80
81impl HttpClient {
82    pub(crate) fn new(inner: reqwest::Client, request_config: RequestConfig) -> Self {
83        HttpClient {
84            inner,
85            request_config,
86            concurrent_request_semaphore: MaybeSemaphore::new(
87                request_config.max_concurrent_requests,
88            ),
89            next_request_id: AtomicU64::new(0).into(),
90        }
91    }
92
93    fn get_request_id(&self) -> String {
94        let request_id = self.next_request_id.fetch_add(1, Ordering::SeqCst);
95        format!("REQ-{request_id}")
96    }
97
98    fn serialize_request<R>(
99        &self,
100        request: R,
101        config: RequestConfig,
102        homeserver: String,
103        access_token: Option<&str>,
104        supported_versions: &SupportedVersions,
105    ) -> Result<http::Request<Bytes>, IntoHttpError>
106    where
107        R: OutgoingRequest + Debug,
108    {
109        trace!(request_type = type_name::<R>(), "Serializing request");
110
111        let send_access_token = match access_token {
112            Some(access_token) => {
113                if config.force_auth {
114                    SendAccessToken::Always(access_token)
115                } else {
116                    SendAccessToken::IfRequired(access_token)
117                }
118            }
119            None => SendAccessToken::None,
120        };
121
122        let request = request
123            .try_into_http_request::<BytesMut>(&homeserver, send_access_token, supported_versions)?
124            .map(|body| body.freeze());
125
126        Ok(request)
127    }
128
129    #[allow(clippy::too_many_arguments)]
130    #[instrument(
131        skip(self, request, config, homeserver, access_token, supported_versions, send_progress),
132        fields(
133            uri,
134            method,
135            request_id,
136            request_size,
137            request_duration,
138            status,
139            response_size,
140            sentry_event_id
141        )
142    )]
143    pub async fn send<R>(
144        &self,
145        request: R,
146        config: Option<RequestConfig>,
147        homeserver: String,
148        access_token: Option<&str>,
149        supported_versions: &SupportedVersions,
150        send_progress: SharedObservable<TransmissionProgress>,
151    ) -> Result<R::IncomingResponse, HttpError>
152    where
153        R: OutgoingRequest + Debug,
154        HttpError: From<FromHttpResponseError<R::EndpointError>>,
155    {
156        let config = match config {
157            Some(config) => config,
158            None => self.request_config,
159        };
160
161        // Keep some local variables in a separate scope so the compiler doesn't include
162        // them in the future type. https://github.com/rust-lang/rust/issues/57478
163        let request = {
164            let request_id = self.get_request_id();
165            let span = tracing::Span::current();
166
167            // At this point in the code, the config isn't behind an Option anymore, that's
168            // why we record it here, instead of in the #[instrument] macro.
169            span.record("config", debug(config)).record("request_id", request_id);
170
171            let auth_scheme = R::METADATA.authentication;
172            match auth_scheme {
173                AuthScheme::AccessToken
174                | AuthScheme::AccessTokenOptional
175                | AuthScheme::AppserviceToken
176                | AuthScheme::AppserviceTokenOptional
177                | AuthScheme::None => {}
178                AuthScheme::ServerSignatures => {
179                    return Err(HttpError::NotClientRequest);
180                }
181            }
182
183            let request = self
184                .serialize_request(request, config, homeserver, access_token, supported_versions)
185                .map_err(HttpError::IntoHttp)?;
186
187            let method = request.method();
188
189            let mut uri_parts = request.uri().clone().into_parts();
190            if let Some(path_and_query) = &mut uri_parts.path_and_query {
191                *path_and_query =
192                    path_and_query.path().try_into().expect("path is valid PathAndQuery");
193            }
194            let uri = http::Uri::from_parts(uri_parts).expect("created from valid URI");
195
196            span.record("method", debug(method)).record("uri", uri.to_string());
197
198            // POST, PUT, PATCH are the only methods that are reasonably used
199            // in conjunction with request bodies
200            if [Method::POST, Method::PUT, Method::PATCH].contains(method) {
201                let request_size = request.body().len().try_into().unwrap_or(u64::MAX);
202                span.record(
203                    "request_size",
204                    ByteSize(request_size).display().si_short().to_string(),
205                );
206            }
207
208            request
209        };
210
211        // will be automatically dropped at the end of this function
212        let _handle = self.concurrent_request_semaphore.acquire().await;
213
214        // There's a bunch of state in send_request, factor out a pinned inner
215        // future to reduce this size of futures that await this function.
216        match Box::pin(self.send_request::<R>(request, config, send_progress)).await {
217            Ok(response) => {
218                debug!("Got response");
219                Ok(response)
220            }
221            Err(e) => {
222                debug!("Error while sending request: {e:?}");
223                Err(e)
224            }
225        }
226    }
227}
228
229/// Progress of sending or receiving a payload.
230#[derive(Clone, Copy, Debug, Default)]
231pub struct TransmissionProgress {
232    /// How many bytes were already transferred.
233    pub current: usize,
234    /// How many bytes there are in total.
235    pub total: usize,
236}
237
238async fn response_to_http_response(
239    mut response: reqwest::Response,
240) -> Result<http::Response<Bytes>, reqwest::Error> {
241    let status = response.status();
242
243    let mut http_builder = http::Response::builder().status(status);
244    let headers = http_builder.headers_mut().expect("Can't get the response builder headers");
245
246    for (k, v) in response.headers_mut().drain() {
247        if let Some(key) = k {
248            headers.insert(key, v);
249        }
250    }
251
252    let body = response.bytes().await?;
253
254    Ok(http_builder.body(body).expect("Can't construct a response using the given body"))
255}
256
257#[cfg(all(test, not(target_family = "wasm")))]
258mod tests {
259    use std::{
260        num::NonZeroUsize,
261        sync::{
262            atomic::{AtomicU8, Ordering},
263            Arc,
264        },
265        time::Duration,
266    };
267
268    use matrix_sdk_common::executor::spawn;
269    use matrix_sdk_test::{async_test, test_json};
270    use wiremock::{
271        matchers::{method, path},
272        Mock, Request, ResponseTemplate,
273    };
274
275    use crate::{
276        http_client::RequestConfig,
277        test_utils::{set_client_session, test_client_builder_with_server},
278    };
279
280    #[async_test]
281    async fn test_ensure_concurrent_request_limit_is_observed() {
282        let (client_builder, server) = test_client_builder_with_server().await;
283        let client = client_builder
284            .request_config(RequestConfig::default().max_concurrent_requests(NonZeroUsize::new(5)))
285            .build()
286            .await
287            .unwrap();
288
289        set_client_session(&client).await;
290
291        let counter = Arc::new(AtomicU8::new(0));
292        let inner_counter = counter.clone();
293
294        Mock::given(method("GET"))
295            .and(path("/_matrix/client/versions"))
296            .respond_with(ResponseTemplate::new(200).set_body_json(&*test_json::VERSIONS))
297            .mount(&server)
298            .await;
299
300        Mock::given(method("GET"))
301            .and(path("_matrix/client/r0/account/whoami"))
302            .respond_with(move |_req: &Request| {
303                inner_counter.fetch_add(1, Ordering::SeqCst);
304                // we stall the requests
305                ResponseTemplate::new(200).set_delay(Duration::from_secs(60))
306            })
307            .mount(&server)
308            .await;
309
310        let bg_task = spawn(async move {
311            futures_util::future::join_all((0..10).map(|_| client.whoami())).await
312        });
313
314        // give it some time to issue the requests
315        tokio::time::sleep(Duration::from_millis(300)).await;
316
317        assert_eq!(
318            counter.load(Ordering::SeqCst),
319            5,
320            "More requests passed than the limit we configured"
321        );
322        bg_task.abort();
323    }
324
325    #[async_test]
326    async fn test_ensure_no_max_concurrent_request_does_not_limit() {
327        let (client_builder, server) = test_client_builder_with_server().await;
328        let client = client_builder
329            .request_config(RequestConfig::default().max_concurrent_requests(None))
330            .build()
331            .await
332            .unwrap();
333
334        set_client_session(&client).await;
335
336        let counter = Arc::new(AtomicU8::new(0));
337        let inner_counter = counter.clone();
338
339        Mock::given(method("GET"))
340            .and(path("/_matrix/client/versions"))
341            .respond_with(ResponseTemplate::new(200).set_body_json(&*test_json::VERSIONS))
342            .mount(&server)
343            .await;
344
345        Mock::given(method("GET"))
346            .and(path("_matrix/client/r0/account/whoami"))
347            .respond_with(move |_req: &Request| {
348                inner_counter.fetch_add(1, Ordering::SeqCst);
349                ResponseTemplate::new(200).set_delay(Duration::from_secs(60))
350            })
351            .mount(&server)
352            .await;
353
354        let bg_task = spawn(async move {
355            futures_util::future::join_all((0..254).map(|_| client.whoami())).await
356        });
357
358        // give it some time to issue the requests
359        tokio::time::sleep(Duration::from_secs(1)).await;
360
361        assert_eq!(counter.load(Ordering::SeqCst), 254, "Not all requests passed through");
362        bg_task.abort();
363    }
364}