wowlab_node/
sentinel.rs

1//! Signed HTTP client for sentinel operations.
2
3use std::{error::Error, sync::Arc, time::Duration};
4
5use prost::Message;
6use reqwest::StatusCode;
7use uuid::Uuid;
8use wowlab_common::node_http::{
9    NodeChunkCompletionResponse, NodeRegistrationRequest, NodeRegistrationResponse,
10    NodeTokenResponse, NodeUnlinkResponse, NodeWorkContextResponse,
11};
12use wowlab_types::{
13    constants::{HTTP_CONNECT_TIMEOUT_SECS, HTTP_REQUEST_TIMEOUT_SECS, MAX_RETRY_ATTEMPTS},
14    proto::BatchChunkCompletion,
15    sensitive::Sensitive,
16};
17
18mod request;
19
20use request::{ExecutorTarget, NodeRequest, RequestExecutor};
21
22type BoxError = Box<dyn Error + Send + Sync>;
23
24#[derive(Debug)]
25pub(crate) struct SignedHeaders {
26    pub key: String,
27    pub signature: String,
28    pub timestamp: String,
29}
30
31pub(crate) trait RequestSigner: Send + Sync {
32    fn sign_request(&self, method: &str, host: &str, path: &str, body: &[u8]) -> SignedHeaders;
33}
34
35/// Failure communicating with the Sentinel service.
36#[derive(Debug, thiserror::Error)]
37#[error("{kind}")]
38pub struct SentinelError {
39    #[source]
40    kind: SentinelErrorKind,
41}
42
43#[derive(Debug, thiserror::Error)]
44enum SentinelErrorKind {
45    #[error("HTTP error: {0}")]
46    Http(#[source] BoxError),
47    #[error("API error: {message}")]
48    Api {
49        status: Option<StatusCode>,
50        message: String,
51    },
52    #[error("Failed to build HTTP client: {0}")]
53    ClientBuild(String),
54    #[cfg(feature = "supabase")]
55    #[error("Failed to build HTTP client: {0}")]
56    Supabase(#[source] wowlab_supabase::SupabaseError),
57    #[cfg(feature = "supabase")]
58    #[error("Failed to build HTTP client: {0}")]
59    GameDataCache(#[source] wowlab_engine_adapter_data::CacheError),
60    #[error("Invalid URL: {0}")]
61    InvalidUrl(String),
62    #[error("Serialization error: {0}")]
63    Serialization(#[source] serde_json::Error),
64    #[error("HTTP error: error decoding response body")]
65    Decode(#[source] serde_json::Error),
66}
67
68impl SentinelError {
69    pub(crate) fn client_build(message: impl Into<String>) -> Self {
70        Self::new(SentinelErrorKind::ClientBuild(message.into()))
71    }
72
73    #[cfg(feature = "supabase")]
74    pub(crate) fn supabase(source: wowlab_supabase::SupabaseError) -> Self {
75        Self::new(SentinelErrorKind::Supabase(source))
76    }
77
78    #[cfg(feature = "supabase")]
79    pub(crate) fn game_data_cache(source: wowlab_engine_adapter_data::CacheError) -> Self {
80        Self::new(SentinelErrorKind::GameDataCache(source))
81    }
82
83    fn api(message: impl Into<String>) -> Self {
84        Self::new(SentinelErrorKind::Api {
85            status: None,
86            message: message.into(),
87        })
88    }
89
90    fn api_status(status: StatusCode, message: impl Into<String>) -> Self {
91        Self::new(SentinelErrorKind::Api {
92            status: Some(status),
93            message: message.into(),
94        })
95    }
96
97    fn invalid_url(message: impl Into<String>) -> Self {
98        Self::new(SentinelErrorKind::InvalidUrl(message.into()))
99    }
100
101    fn serialization(error: serde_json::Error) -> Self {
102        Self::new(SentinelErrorKind::Serialization(error))
103    }
104
105    fn decode(error: serde_json::Error) -> Self {
106        Self::new(SentinelErrorKind::Decode(error))
107    }
108
109    fn http(source: BoxError) -> Self {
110        Self::new(SentinelErrorKind::Http(source))
111    }
112
113    fn new(kind: SentinelErrorKind) -> Self {
114        Self { kind }
115    }
116
117    fn status(&self) -> Option<StatusCode> {
118        match &self.kind {
119            SentinelErrorKind::Api { status, .. } => *status,
120            _ => None,
121        }
122    }
123}
124
125#[derive(Clone, Copy, Debug, Eq, PartialEq)]
126pub(crate) enum SentinelUnlinkOutcome {
127    Removed,
128    AlreadyAbsent,
129}
130
131impl From<reqwest::Error> for SentinelError {
132    fn from(error: reqwest::Error) -> Self {
133        Self::http(Box::new(error))
134    }
135}
136
137impl From<serde_json::Error> for SentinelError {
138    fn from(error: serde_json::Error) -> Self {
139        Self::decode(error)
140    }
141}
142
143#[cfg(feature = "supabase")]
144impl From<wowlab_supabase::SupabaseError> for SentinelError {
145    fn from(source: wowlab_supabase::SupabaseError) -> Self {
146        Self::supabase(source)
147    }
148}
149
150#[cfg(feature = "supabase")]
151impl From<wowlab_engine_adapter_data::CacheError> for SentinelError {
152    fn from(source: wowlab_engine_adapter_data::CacheError) -> Self {
153        Self::game_data_cache(source)
154    }
155}
156
157#[derive(Clone)]
158pub(crate) struct SentinelClient {
159    executor: RequestExecutor,
160}
161
162impl SentinelClient {
163    pub(crate) fn new(
164        sentinel_url: String,
165        signer: impl RequestSigner + 'static,
166    ) -> Result<Self, SentinelError> {
167        let http = match reqwest::Client::builder()
168            .timeout(Duration::from_secs(HTTP_REQUEST_TIMEOUT_SECS))
169            .connect_timeout(Duration::from_secs(HTTP_CONNECT_TIMEOUT_SECS))
170            .build()
171        {
172            Ok(http) => http,
173            Err(error) => return Err(SentinelError::client_build(error.to_string())),
174        };
175
176        let sentinel_host = reqwest::Url::parse(&sentinel_url)
177            .map_err(|error| SentinelError::invalid_url(error.to_string()))?
178            .host_str()
179            .ok_or_else(|| SentinelError::invalid_url("URL has no host"))?
180            .to_string();
181
182        Ok(Self {
183            executor: RequestExecutor::new(
184                ExecutorTarget {
185                    sentinel_url,
186                    sentinel_host,
187                },
188                http,
189                Arc::new(signer),
190            ),
191        })
192    }
193
194    pub(crate) async fn register_node(
195        &self,
196        registration: &NodeRegistrationRequest,
197    ) -> Result<NodeRegistrationResponse, SentinelError> {
198        let body = match serde_json::to_vec(registration) {
199            Ok(body) => body,
200            Err(error) => return Err(SentinelError::serialization(error)),
201        };
202
203        self.executor
204            .execute(NodeRequest::post_json(
205                "/nodes/register",
206                body,
207                MAX_RETRY_ATTEMPTS,
208            ))
209            .await
210    }
211
212    pub(crate) async fn refresh_token(&self) -> Result<Sensitive<String>, SentinelError> {
213        let resp: NodeTokenResponse = self
214            .executor
215            .execute(NodeRequest::post_json(
216                "/nodes/token",
217                Vec::new(),
218                MAX_RETRY_ATTEMPTS,
219            ))
220            .await?;
221
222        Ok(Sensitive::new(resp.into_beacon_token()))
223    }
224
225    pub(crate) async fn unlink(&self) -> Result<SentinelUnlinkOutcome, SentinelError> {
226        let response: Result<NodeUnlinkResponse, SentinelError> = self
227            .executor
228            .execute(NodeRequest::post_json(
229                "/nodes/unlink",
230                Vec::new(),
231                MAX_RETRY_ATTEMPTS,
232            ))
233            .await;
234        let response = match response {
235            Ok(response) => response,
236            Err(error) if error.status() == Some(StatusCode::NOT_FOUND) => {
237                return Ok(SentinelUnlinkOutcome::AlreadyAbsent);
238            }
239            Err(error) => return Err(error),
240        };
241
242        if !response.is_success() {
243            return Err(SentinelError::api("Unlink response reported failure"));
244        }
245
246        Ok(SentinelUnlinkOutcome::Removed)
247    }
248
249    pub(crate) async fn fetch_work_context(
250        &self,
251        job_id: Uuid,
252        hash_hex: &str,
253        claim_token: &str,
254    ) -> Result<NodeWorkContextResponse, SentinelError> {
255        let path = format!("/jobs/{job_id}/work_context?hash={hash_hex}&claim_token={claim_token}");
256
257        self.executor
258            .execute(NodeRequest::get(path, MAX_RETRY_ATTEMPTS))
259            .await
260    }
261
262    pub(crate) async fn complete_batch(
263        &self,
264        completion: &BatchChunkCompletion,
265    ) -> Result<(), SentinelError> {
266        let path = format!("/chunks/complete?job_id={}", completion.job_id);
267        let body = completion.encode_to_vec();
268        let _: NodeChunkCompletionResponse = self
269            .executor
270            .execute(NodeRequest::post_protobuf(path, body, MAX_RETRY_ATTEMPTS))
271            .await?;
272
273        Ok(())
274    }
275}
276
277impl std::fmt::Debug for SentinelClient {
278    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
279        f.debug_struct("SentinelClient")
280            .field("executor", &self.executor)
281            .finish_non_exhaustive()
282    }
283}
284
285#[cfg(test)]
286mod tests;
wowlab_node/sentinel.rs

wowlab_node/
sentinel.rs
