relay_knowledge/adapters/
release_metadata.rs

1use std::time::Duration;
2
3use reqwest::{StatusCode, header};
4
5use crate::{
6    net::{
7        NetworkRuntime, http,
8        qos::{QosPolicy, QosRuntime},
9    },
10    ports::release_metadata::{
11        ReleaseMetadataError, ReleaseMetadataErrorKind, ReleaseMetadataFuture, ReleaseMetadataPort,
12        ReleaseMetadataRequest, ReleaseMetadataSession,
13    },
14    project::PROJECT_NAME,
15};
16
17const VERSION_CHECK_REQUEST_TIMEOUT: Duration = Duration::from_secs(3);
18
19struct HttpReleaseMetadataSession {
20    client: reqwest::Client,
21    qos: QosRuntime,
22    policy: QosPolicy,
23    max_response_bytes: u64,
24}
25
26impl ReleaseMetadataPort for NetworkRuntime {
27    fn open(&self) -> Result<Box<dyn ReleaseMetadataSession>, ReleaseMetadataError> {
28        let config = self.current();
29        let client =
30            http::outbound_json_client(&config.http).map_err(|error| ReleaseMetadataError {
31                kind: ReleaseMetadataErrorKind::ClientBuild,
32                message: error.to_string(),
33                retryable: false,
34            })?;
35
36        Ok(Box::new(HttpReleaseMetadataSession {
37            client,
38            qos: self.qos_runtime(),
39            policy: config.qos,
40            max_response_bytes: config.http.max_request_body_bytes,
41        }))
42    }
43}
44
45impl ReleaseMetadataSession for HttpReleaseMetadataSession {
46    fn fetch(&self, request: ReleaseMetadataRequest) -> ReleaseMetadataFuture<'_> {
47        Box::pin(async move {
48            let response = send_request(self, &request.url)
49                .await
50                .map_err(qos_transport_error)?;
51            validate_status(response.status())?;
52            read_bounded_body(response, self.max_response_bytes).await
53        })
54    }
55}
56
57async fn send_request(
58    session: &HttpReleaseMetadataSession,
59    url: &str,
60) -> Result<http::QosHttpResponse, http::QosHttpClientError> {
61    http::send_request_with_qos(
62        &session.qos,
63        &session.policy,
64        session
65            .client
66            .get(url)
67            .header(
68                header::USER_AGENT,
69                format!("{PROJECT_NAME}/{}", env!("CARGO_PKG_VERSION")),
70            )
71            .timeout(VERSION_CHECK_REQUEST_TIMEOUT),
72    )
73    .await
74}
75
76fn validate_status(status: StatusCode) -> Result<(), ReleaseMetadataError> {
77    if status.is_success() {
78        return Ok(());
79    }
80
81    Err(ReleaseMetadataError {
82        kind: ReleaseMetadataErrorKind::HttpStatus,
83        message: format!("release metadata request returned HTTP {}", status.as_u16()),
84        retryable: status.is_server_error()
85            || status == StatusCode::REQUEST_TIMEOUT
86            || status == StatusCode::TOO_MANY_REQUESTS,
87    })
88}
89
90async fn read_bounded_body(
91    mut response: http::QosHttpResponse,
92    max_response_bytes: u64,
93) -> Result<Vec<u8>, ReleaseMetadataError> {
94    if response
95        .content_length()
96        .is_some_and(|length| length > max_response_bytes)
97    {
98        return Err(response_too_large_error(max_response_bytes));
99    }
100
101    let limit = max_response_bytes.try_into().unwrap_or(usize::MAX);
102    let mut body = Vec::new();
103    while let Some(chunk) = response
104        .chunk()
105        .await
106        .map_err(|error| ReleaseMetadataError {
107            kind: ReleaseMetadataErrorKind::Transport,
108            message: error.to_string(),
109            retryable: true,
110        })?
111    {
112        append_bounded_body(&mut body, &chunk, limit)?;
113    }
114    Ok(body)
115}
116
117fn append_bounded_body(
118    body: &mut Vec<u8>,
119    chunk: &[u8],
120    max_response_bytes: usize,
121) -> Result<(), ReleaseMetadataError> {
122    let Some(next_len) = body
123        .len()
124        .checked_add(chunk.len())
125        .filter(|next_len| *next_len <= max_response_bytes)
126    else {
127        return Err(response_too_large_error(
128            max_response_bytes.try_into().unwrap_or(u64::MAX),
129        ));
130    };
131    body.reserve(next_len.saturating_sub(body.len()));
132    body.extend_from_slice(chunk);
133    Ok(())
134}
135
136fn qos_transport_error(error: http::QosHttpClientError) -> ReleaseMetadataError {
137    ReleaseMetadataError {
138        kind: if error.is_timeout() {
139            ReleaseMetadataErrorKind::NetworkTimeout
140        } else {
141            ReleaseMetadataErrorKind::Network
142        },
143        message: error.to_string(),
144        retryable: true,
145    }
146}
147
148fn response_too_large_error(max_response_bytes: u64) -> ReleaseMetadataError {
149    ReleaseMetadataError {
150        kind: ReleaseMetadataErrorKind::ResponseTooLarge,
151        message: format!("release metadata response exceeded {max_response_bytes} bytes"),
152        retryable: false,
153    }
154}
155
156#[cfg(test)]
157#[path = "release_metadata_tests.rs"]
158mod tests;
relay_knowledge/adapters/release_metadata.rs

relay_knowledge/adapters/
release_metadata.rs
