http.rs 17.3%

14 of 81 lines run, 3 of 26 functions · run #260 on main · run missed

//! The HTTP client the cloud stores send their requests through: `ureq`
//! on rustls with the web roots, the client oxy-core itself uses, so the
//! stores cross-compile for every target oxy is released for. Each
//! request runs on the runtime's blocking pool; bodies are held whole,
//! which is how the stores use them (a bundle, a record, a listing).

use async_trait::async_trait;
use bytes::Bytes;
use http_body_util::{BodyExt, Full};
use object_store::client::{
    ClientOptions, HttpClient, HttpConnector, HttpError, HttpErrorKind, HttpRequest, HttpResponse,
    HttpResponseBody, HttpService,
};
use std::io::Read;
use std::time::Duration;

/// How long a request may take in all, and to connect.
const TIMEOUT: Duration = Duration::from_secs(600);
const CONNECT_TIMEOUT: Duration = Duration::from_secs(30);

/// A rustls configuration trusting the web roots.
fn tls() -> Result<std::sync::Arc<rustls::ClientConfig>, String> {
    oxy_core::protocol::tls_client_config(None).map_err(|e| e.to_string())
}

/// Make the agent every request of a store goes through.
pub fn agent() -> Result<ureq::Agent, String> {
    Ok(ureq::AgentBuilder::new()
        .tls_config(tls()?)
        .timeout_connect(CONNECT_TIMEOUT)
        .timeout(TIMEOUT)
        .build())
}

#[derive(Debug, Clone, Copy, Default)]
pub struct UreqConnector;

impl HttpConnector for UreqConnector {
    fn connect(&self, _options: &ClientOptions) -> object_store::Result<HttpClient> {
        let agent = agent().map_err(|e| object_store::Error::Generic {
            store: "ureq",
            source: e.into(),
        })?;
        Ok(HttpClient::new(UreqService { agent }))
    }
}

#[derive(Debug, Clone)]
struct UreqService {
    agent: ureq::Agent,
}

#[async_trait]
impl HttpService for UreqService {
    async fn call(&self, req: HttpRequest) -> Result<HttpResponse, HttpError> {
        let (parts, body) = req.into_parts();
        let bytes = body.collect().await?.to_bytes();
        let agent = self.agent.clone();
        tokio::task::spawn_blocking(move || send(&agent, parts, &bytes))
            .await
            .map_err(|e| HttpError::new(HttpErrorKind::Interrupted, e))?
    }
}

fn send(
    agent: &ureq::Agent,
    parts: http::request::Parts,
    bytes: &[u8],
) -> Result<HttpResponse, HttpError> {
    let url = parts.uri.to_string();
    let mut r = agent.request(parts.method.as_str(), &url);
    for (name, value) in &parts.headers {
        if let Ok(v) = value.to_str() {
            r = r.set(name.as_str(), v);
        }
    }
    let sent = if bytes.is_empty()
        && matches!(
            parts.method,
            http::Method::GET | http::Method::HEAD | http::Method::DELETE
        ) {
        r.call()
    } else {
        r.send_bytes(bytes)
    };
    let resp = match sent {
        Ok(resp) => resp,
        // A refusal is a response the store interprets (a 404, a 412).
        Err(ureq::Error::Status(_, resp)) => resp,
        Err(ureq::Error::Transport(t)) => {
            let kind = match t.kind() {
                ureq::ErrorKind::Dns
                | ureq::ErrorKind::ConnectionFailed
                | ureq::ErrorKind::ProxyConnect => HttpErrorKind::Connect,
                ureq::ErrorKind::Io => HttpErrorKind::Timeout,
                _ => HttpErrorKind::Request,
            };
            return Err(HttpError::new(kind, t));
        }
    };
    let status = http::StatusCode::from_u16(resp.status())
        .map_err(|e| HttpError::new(HttpErrorKind::Decode, e))?;
    let mut builder = http::Response::builder().status(status);
    for name in resp.headers_names() {
        for value in resp.all(&name) {
            builder = builder.header(name.as_str(), value);
        }
    }
    let mut body = Vec::new();
    resp.into_reader()
        .read_to_end(&mut body)
        .map_err(|e| HttpError::new(HttpErrorKind::Decode, e))?;
    let body = HttpResponseBody::new(
        Full::new(Bytes::from(body)).map_err(|never: std::convert::Infallible| match never {}),
    );
    builder
        .body(body)
        .map_err(|e| HttpError::new(HttpErrorKind::Decode, e))
}

/// The region that holds an S3 bucket, as S3 itself reports it in
/// `x-amz-bucket-region` on any request, even a refused one.
pub fn bucket_region(bucket: &str) -> Option<String> {
    let agent = ureq::AgentBuilder::new()
        .tls_config(tls().ok()?)
        .timeout(Duration::from_secs(30))
        .redirects(0)
        .build();
    // S3's wildcard certificate covers one label: dotted names go by path.
    let url = if bucket.contains('.') {
        format!("https://s3.amazonaws.com/{bucket}")
    } else {
        format!("https://{bucket}.s3.amazonaws.com")
    };
    let resp = match agent.head(&url).call() {
        Ok(r) | Err(ureq::Error::Status(_, r)) => r,
        Err(_) => return None,
    };
    resp.header("x-amz-bucket-region").map(str::to_string)
}

1		//! The HTTP client the cloud stores send their requests through: `ureq`
2		//! on rustls with the web roots, the client oxy-core itself uses, so the
3		//! stores cross-compile for every target oxy is released for. Each
4		//! request runs on the runtime's blocking pool; bodies are held whole,
5		//! which is how the stores use them (a bundle, a record, a listing).
6
7		use async_trait::async_trait;
8		use bytes::Bytes;
9		use http_body_util::{BodyExt, Full};
10		use object_store::client::{
11		ClientOptions, HttpClient, HttpConnector, HttpError, HttpErrorKind, HttpRequest, HttpResponse,
12		HttpResponseBody, HttpService,
13		};
14		use std::io::Read;
15		use std::time::Duration;
16
17		/// How long a request may take in all, and to connect.
18		const TIMEOUT: Duration = Duration::from_secs(600);
19		const CONNECT_TIMEOUT: Duration = Duration::from_secs(30);
20
21		/// A rustls configuration trusting the web roots.
22	2×	fn tls() -> Result<std::sync::Arc<rustls::ClientConfig>, String> {
23	2×	oxy_core::protocol::tls_client_config(None).map_err(\|e\| e.to_string())
24	2×	}
25
26		/// Make the agent every request of a store goes through.
27	2×	pub fn agent() -> Result<ureq::Agent, String> {
28	2×	Ok(ureq::AgentBuilder::new()
29	2×	.tls_config(tls()?)
30	2×	.timeout_connect(CONNECT_TIMEOUT)
31	2×	.timeout(TIMEOUT)
32	2×	.build())
33	2×	}
34
35		#[derive(Debug, Clone, Copy, Default)]
36		pub struct UreqConnector;
37
38		impl HttpConnector for UreqConnector {
39	2×	fn connect(&self, _options: &ClientOptions) -> object_store::Result<HttpClient> {
40	2×	let agent = agent().map_err(\|e\| object_store::Error::Generic {
41		store: "ureq",
42	0×	source: e.into(),
43	0×	})?;
44	2×	Ok(HttpClient::new(UreqService { agent }))
45	2×	}
46		}
47
48		#[derive(Debug, Clone)]
49		struct UreqService {
50		agent: ureq::Agent,
51		}
52
53		#[async_trait]
54		impl HttpService for UreqService {
55	0×	async fn call(&self, req: HttpRequest) -> Result<HttpResponse, HttpError> {
56		let (parts, body) = req.into_parts();
57		let bytes = body.collect().await?.to_bytes();
58		let agent = self.agent.clone();
59	0×	tokio::task::spawn_blocking(move \|\| send(&agent, parts, &bytes))
60		.await
61	0×	.map_err(\|e\| HttpError::new(HttpErrorKind::Interrupted, e))?
62	0×	}
63		}
64
65	0×	fn send(
66	0×	agent: &ureq::Agent,
67	0×	parts: http::request::Parts,
68	0×	bytes: &[u8],
69	0×	) -> Result<HttpResponse, HttpError> {
70	0×	let url = parts.uri.to_string();
71	0×	let mut r = agent.request(parts.method.as_str(), &url);
72	0×	for (name, value) in &parts.headers {
73	0×	if let Ok(v) = value.to_str() {
74	0×	r = r.set(name.as_str(), v);
75	0×	}
76		}
77	0×	let sent = if bytes.is_empty()
78	0×	&& matches!(
79	0×	parts.method,
80		http::Method::GET \| http::Method::HEAD \| http::Method::DELETE
81		) {
82	0×	r.call()
83		} else {
84	0×	r.send_bytes(bytes)
85		};
86	0×	let resp = match sent {
87	0×	Ok(resp) => resp,
88		// A refusal is a response the store interprets (a 404, a 412).
89	0×	Err(ureq::Error::Status(_, resp)) => resp,
90	0×	Err(ureq::Error::Transport(t)) => {
91	0×	let kind = match t.kind() {
92		ureq::ErrorKind::Dns
93		\| ureq::ErrorKind::ConnectionFailed
94	0×	\| ureq::ErrorKind::ProxyConnect => HttpErrorKind::Connect,
95	0×	ureq::ErrorKind::Io => HttpErrorKind::Timeout,
96	0×	_ => HttpErrorKind::Request,
97		};
98	0×	return Err(HttpError::new(kind, t));
99		}
100		};
101	0×	let status = http::StatusCode::from_u16(resp.status())
102	0×	.map_err(\|e\| HttpError::new(HttpErrorKind::Decode, e))?;
103	0×	let mut builder = http::Response::builder().status(status);
104	0×	for name in resp.headers_names() {
105	0×	for value in resp.all(&name) {
106	0×	builder = builder.header(name.as_str(), value);
107	0×	}
108		}
109	0×	let mut body = Vec::new();
110	0×	resp.into_reader()
111	0×	.read_to_end(&mut body)
112	0×	.map_err(\|e\| HttpError::new(HttpErrorKind::Decode, e))?;
113	0×	let body = HttpResponseBody::new(
114	0×	Full::new(Bytes::from(body)).map_err(\|never: std::convert::Infallible\| match never {}),
115		);
116	0×	builder
117	0×	.body(body)
118	0×	.map_err(\|e\| HttpError::new(HttpErrorKind::Decode, e))
119	0×	}
120
121		/// The region that holds an S3 bucket, as S3 itself reports it in
122		/// `x-amz-bucket-region` on any request, even a refused one.
123	0×	pub fn bucket_region(bucket: &str) -> Option<String> {
124	0×	let agent = ureq::AgentBuilder::new()
125	0×	.tls_config(tls().ok()?)
126	0×	.timeout(Duration::from_secs(30))
127	0×	.redirects(0)
128	0×	.build();
129		// S3's wildcard certificate covers one label: dotted names go by path.
130	0×	let url = if bucket.contains('.') {
131	0×	format!("https://s3.amazonaws.com/{bucket}")
132		} else {
133	0×	format!("https://{bucket}.s3.amazonaws.com")
134		};
135	0×	let resp = match agent.head(&url).call() {
136	0×	Ok(r) \| Err(ureq::Error::Status(_, r)) => r,
137	0×	Err(_) => return None,
138		};
139	0×	resp.header("x-amz-bucket-region").map(str::to_string)
140	0×	}

`/`	Search
`g` `c`	Code
`g` `h`	History
`g` `b`	Bookmarks
`g` `t`	Trust
`y`	Permalink to this state (files and trees)
`?`	This sheet

Coverage / crates/oxy-strata/src/http.rs 17.3%

Keyboard