wowlab_node/sentinel/
request.rs

1//! Typed execution policy for signed Sentinel HTTP requests.
2
3use std::{error::Error, pin::Pin, sync::Arc, time::Duration};
4
5use reqwest::StatusCode;
6use serde::de::DeserializeOwned;
7use wowlab_common::retry::ExponentialSchedule;
8
9use super::{RequestSigner, SentinelError, SignedHeaders};
10
11const INITIAL_RETRY_DELAY: Duration = Duration::from_millis(500);
12const BACKOFF_MULTIPLIER: u32 = 2;
13const MAX_RETRY_DELAY: Duration = Duration::from_secs(10);
14const HTTP_TOO_MANY_REQUESTS: u16 = 429;
15
16type BoxFuture<'a, T> = Pin<Box<dyn Future<Output = T> + Send + 'a>>;
17type BoxError = Box<dyn Error + Send + Sync>;
18
19#[derive(Clone, Copy, Debug, Eq, PartialEq)]
20pub(super) enum HttpMethod {
21    Get,
22    Post,
23}
24
25impl HttpMethod {
26    fn method_name(self) -> &'static str {
27        match self {
28            Self::Get => "GET",
29            Self::Post => "POST",
30        }
31    }
32}
33
34#[derive(Clone, Copy, Debug, Eq, PartialEq)]
35pub(super) enum RetryPolicy {
36    Safe { max_retries: u32 },
37    Idempotent { max_retries: u32 },
38}
39
40impl RetryPolicy {
41    fn max_retries(self) -> u32 {
42        match self {
43            Self::Safe { max_retries } | Self::Idempotent { max_retries } => max_retries,
44        }
45    }
46}
47
48#[derive(Clone, Eq, PartialEq)]
49pub(super) enum RequestBody {
50    Empty,
51    Json(Vec<u8>),
52    Protobuf(Vec<u8>),
53}
54
55impl RequestBody {
56    fn bytes(&self) -> &[u8] {
57        match self {
58            Self::Empty => &[],
59            Self::Json(bytes) | Self::Protobuf(bytes) => bytes,
60        }
61    }
62
63    fn content_type(&self, method: HttpMethod) -> Option<&'static str> {
64        match (method, self) {
65            (HttpMethod::Get, _) => None,
66            (HttpMethod::Post, Self::Empty | Self::Json(_)) => Some("application/json"),
67            (HttpMethod::Post, Self::Protobuf(_)) => Some("application/octet-stream"),
68        }
69    }
70
71    fn kind(&self) -> &'static str {
72        match self {
73            Self::Empty => "empty",
74            Self::Json(_) => "json",
75            Self::Protobuf(_) => "protobuf",
76        }
77    }
78}
79
80pub(super) struct NodeRequest<T> {
81    method: HttpMethod,
82    path: String,
83    body: RequestBody,
84    retry: RetryPolicy,
85    response: ResponsePolicy<T>,
86}
87
88struct ResponsePolicy<T> {
89    name: &'static str,
90    accepts_status: fn(StatusCode) -> bool,
91    decode: fn(&[u8]) -> Result<T, SentinelError>,
92}
93
94impl<T> ResponsePolicy<T>
95where
96    T: DeserializeOwned,
97{
98    fn json_2xx() -> Self {
99        Self {
100            name: "2xx_json",
101            accepts_status: is_success,
102            decode: decode_json,
103        }
104    }
105}
106
107impl<T> NodeRequest<T>
108where
109    T: DeserializeOwned,
110{
111    pub(super) fn get(path: impl Into<String>, max_retries: u32) -> Self {
112        Self {
113            method: HttpMethod::Get,
114            path: path.into(),
115            body: RequestBody::Empty,
116            retry: RetryPolicy::Safe { max_retries },
117            response: ResponsePolicy::json_2xx(),
118        }
119    }
120
121    pub(super) fn post_json(path: impl Into<String>, body: Vec<u8>, max_retries: u32) -> Self {
122        Self {
123            method: HttpMethod::Post,
124            path: path.into(),
125            body: RequestBody::Json(body),
126            retry: RetryPolicy::Idempotent { max_retries },
127            response: ResponsePolicy::json_2xx(),
128        }
129    }
130
131    pub(super) fn post_protobuf(path: impl Into<String>, body: Vec<u8>, max_retries: u32) -> Self {
132        Self {
133            method: HttpMethod::Post,
134            path: path.into(),
135            body: RequestBody::Protobuf(body),
136            retry: RetryPolicy::Idempotent { max_retries },
137            response: ResponsePolicy::json_2xx(),
138        }
139    }
140}
141
142impl<T> std::fmt::Debug for NodeRequest<T> {
143    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
144        f.debug_struct("NodeRequest")
145            .field("method", &self.method)
146            .field("signed_path", &signature_path(&self.path))
147            .field("body_kind", &self.body.kind())
148            .field("body_len", &self.body.bytes().len())
149            .field("retry", &self.retry)
150            .field("response", &self.response.name)
151            .finish_non_exhaustive()
152    }
153}
154
155pub(super) struct TransportRequest<'a> {
156    method: HttpMethod,
157    url: &'a str,
158    headers: SignedHeaders,
159    content_type: Option<&'static str>,
160    body: &'a [u8],
161}
162
163pub(super) struct TransportResponse {
164    status: StatusCode,
165    body: ResponseBody,
166}
167
168enum ResponseBody {
169    Reqwest(reqwest::Response),
170    #[cfg(test)]
171    Buffered(Vec<u8>),
172}
173
174impl TransportResponse {
175    #[cfg(test)]
176    pub(super) fn buffered(status: StatusCode, body: impl Into<Vec<u8>>) -> Self {
177        Self {
178            status,
179            body: ResponseBody::Buffered(body.into()),
180        }
181    }
182
183    async fn bytes(self) -> Result<Vec<u8>, SentinelError> {
184        match self.body {
185            ResponseBody::Reqwest(response) => {
186                let bytes = response.bytes().await?;
187
188                Ok(bytes.to_vec())
189            }
190            #[cfg(test)]
191            ResponseBody::Buffered(bytes) => Ok(bytes),
192        }
193    }
194
195    async fn error_text(self) -> String {
196        match self.body {
197            ResponseBody::Reqwest(response) => response
198                .text()
199                .await
200                .unwrap_or_else(|error| format!("Failed to read error response: {error}")),
201            #[cfg(test)]
202            ResponseBody::Buffered(bytes) => String::from_utf8_lossy(&bytes).into_owned(),
203        }
204    }
205}
206
207pub(super) struct TransportError {
208    source: BoxError,
209    retryable: bool,
210}
211
212impl TransportError {
213    fn new(source: reqwest::Error) -> Self {
214        let retryable = source.is_timeout() || source.is_connect() || source.is_request();
215
216        Self {
217            source: Box::new(source),
218            retryable,
219        }
220    }
221
222    #[cfg(test)]
223    pub(super) fn test(source: impl Error + Send + Sync + 'static, retryable: bool) -> Self {
224        Self {
225            source: Box::new(source),
226            retryable,
227        }
228    }
229}
230
231impl From<reqwest::Error> for TransportError {
232    fn from(error: reqwest::Error) -> Self {
233        Self::new(error)
234    }
235}
236
237pub(super) trait HttpTransport: Send + Sync {
238    fn send<'a>(
239        &'a self,
240        request: TransportRequest<'a>,
241    ) -> BoxFuture<'a, Result<TransportResponse, TransportError>>;
242}
243
244#[derive(Clone)]
245struct ReqwestTransport {
246    client: reqwest::Client,
247}
248
249impl HttpTransport for ReqwestTransport {
250    fn send<'a>(
251        &'a self,
252        request: TransportRequest<'a>,
253    ) -> BoxFuture<'a, Result<TransportResponse, TransportError>> {
254        Box::pin(async move {
255            let mut builder = match request.method {
256                HttpMethod::Get => self.client.get(request.url),
257                HttpMethod::Post => self.client.post(request.url).body(request.body.to_vec()),
258            }
259            .header("X-Node-Key", &request.headers.key)
260            .header("X-Node-Sig", &request.headers.signature)
261            .header("X-Node-Ts", &request.headers.timestamp);
262
263            if let Some(content_type) = request.content_type {
264                builder = builder.header("Content-Type", content_type);
265            }
266
267            let response = builder.send().await?;
268
269            Ok(TransportResponse {
270                status: response.status(),
271                body: ResponseBody::Reqwest(response),
272            })
273        })
274    }
275}
276
277pub(super) trait RetrySleeper: Send + Sync {
278    fn sleep(&self, delay: Duration) -> BoxFuture<'_, ()>;
279}
280
281#[derive(Debug)]
282struct TokioRetrySleeper;
283
284impl RetrySleeper for TokioRetrySleeper {
285    fn sleep(&self, delay: Duration) -> BoxFuture<'_, ()> {
286        Box::pin(tokio::time::sleep(delay))
287    }
288}
289
290#[derive(Clone)]
291pub(super) struct RequestExecutor {
292    transport: Arc<dyn HttpTransport>,
293    sleeper: Arc<dyn RetrySleeper>,
294    sentinel_url: String,
295    sentinel_host: String,
296    signer: Arc<dyn RequestSigner>,
297}
298
299pub(super) struct ExecutorTarget {
300    pub(super) sentinel_url: String,
301    pub(super) sentinel_host: String,
302}
303
304#[cfg(test)]
305pub(super) struct ExecutorDependencies {
306    pub(super) signer: Arc<dyn RequestSigner>,
307    pub(super) transport: Arc<dyn HttpTransport>,
308    pub(super) sleeper: Arc<dyn RetrySleeper>,
309}
310
311impl RequestExecutor {
312    pub(super) fn new(
313        target: ExecutorTarget,
314        client: reqwest::Client,
315        signer: Arc<dyn RequestSigner>,
316    ) -> Self {
317        Self {
318            transport: Arc::new(ReqwestTransport { client }),
319            sleeper: Arc::new(TokioRetrySleeper),
320            sentinel_url: target.sentinel_url,
321            sentinel_host: target.sentinel_host,
322            signer,
323        }
324    }
325
326    #[cfg(test)]
327    pub(super) fn with_dependencies(
328        target: ExecutorTarget,
329        dependencies: ExecutorDependencies,
330    ) -> Self {
331        Self {
332            transport: dependencies.transport,
333            sleeper: dependencies.sleeper,
334            sentinel_url: target.sentinel_url,
335            sentinel_host: target.sentinel_host,
336            signer: dependencies.signer,
337        }
338    }
339
340    pub(super) async fn execute<T>(&self, request: NodeRequest<T>) -> Result<T, SentinelError>
341    where
342        T: DeserializeOwned,
343    {
344        let response = self.execute_response(&request).await?;
345
346        if !(request.response.accepts_status)(response.status) {
347            return Err(SentinelError::api_status(
348                response.status,
349                response.error_text().await,
350            ));
351        }
352
353        let bytes = response.bytes().await?;
354
355        (request.response.decode)(&bytes)
356    }
357
358    async fn execute_response<T>(
359        &self,
360        request: &NodeRequest<T>,
361    ) -> Result<TransportResponse, SentinelError> {
362        let url = format!("{}{}", self.sentinel_url, request.path);
363        let signed_path = signature_path(&request.path);
364        let mut last_error = None;
365        let mut schedule =
366            ExponentialSchedule::new(INITIAL_RETRY_DELAY, MAX_RETRY_DELAY, BACKOFF_MULTIPLIER);
367
368        for attempt in 0..=request.retry.max_retries() {
369            if attempt > 0 {
370                let delay = schedule.next_delay();
371
372                log_retry(attempt, signed_path, delay);
373                self.sleeper.sleep(delay).await;
374            }
375
376            let headers = self.signer.sign_request(
377                request.method.method_name(),
378                &self.sentinel_host,
379                signed_path,
380                request.body.bytes(),
381            );
382            let transport_request = TransportRequest {
383                method: request.method,
384                url: &url,
385                headers,
386                content_type: request.body.content_type(request.method),
387                body: request.body.bytes(),
388            };
389
390            match self.transport.send(transport_request).await {
391                Ok(response) => {
392                    let status = response.status;
393
394                    if status.is_client_error() && status.as_u16() != HTTP_TOO_MANY_REQUESTS {
395                        return Ok(response);
396                    }
397
398                    if status.is_server_error() || status.as_u16() == HTTP_TOO_MANY_REQUESTS {
399                        log_server_retry(status, attempt, signed_path);
400                        // #t(block: rust_alloc_in_loop) one-off error message per retry, not accumulating
401                        last_error = Some(SentinelError::api_status(
402                            status,
403                            format!("Server returned {status}"),
404                        ));
405                        continue;
406                    }
407
408                    return Ok(response);
409                }
410                Err(error) => {
411                    if error.retryable {
412                        log_network_retry(error.source.as_ref(), attempt, signed_path);
413                        last_error = Some(SentinelError::http(error.source));
414                        continue;
415                    }
416
417                    return Err(SentinelError::http(error.source));
418                }
419            }
420        }
421
422        Err(last_error.unwrap_or_else(|| SentinelError::api("Max retries exceeded")))
423    }
424}
425
426impl std::fmt::Debug for RequestExecutor {
427    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
428        f.debug_struct("RequestExecutor")
429            .field("sentinel_url", &self.sentinel_url)
430            .field("sentinel_host", &self.sentinel_host)
431            .field("transport", &"<dyn HttpTransport>")
432            .field("sleeper", &"<dyn RetrySleeper>")
433            .field("signer", &"<dyn RequestSigner>")
434            .finish_non_exhaustive()
435    }
436}
437
438fn decode_json<T>(bytes: &[u8]) -> Result<T, SentinelError>
439where
440    T: DeserializeOwned,
441{
442    Ok(serde_json::from_slice(bytes)?)
443}
444
445fn is_success(status: StatusCode) -> bool {
446    status.is_success()
447}
448
449pub(super) fn signature_path(path: &str) -> &str {
450    path.split('?').next().unwrap_or(path)
451}
452
453fn log_retry(attempt: u32, path: &str, delay: Duration) {
454    tracing::debug!(attempt, path, ?delay, "Retrying Sentinel request");
455}
456
457fn log_server_retry(status: StatusCode, attempt: u32, path: &str) {
458    tracing::warn!(%status, attempt, path, "Sentinel server error; retrying");
459}
460
461fn log_network_retry(error: &(dyn Error + Send + Sync), attempt: u32, path: &str) {
462    tracing::warn!(%error, attempt, path, "Sentinel network error; retrying");
463}
464
465#[cfg(test)]
466mod tests;
wowlab_node/sentinel/request.rs

wowlab_node/sentinel/
request.rs
