appcore_peer_rpc/
host.rs

1// =============================================================================
2//        #######
3//     ###       ###     F: host.rs
4//    ##   ## ##   ##    P: AppCore-Runtime
5//         ## ##
6//                       C: 2026/07/22 15:41:18 by dnettoRaw
7//    ##   ## ##   ##    U: 2026/07/23 23:50:45 by dnettoRaw
8//      ###########      S: 1.0.1-rc.8
9// =============================================================================
10
11use super::*;
12use crate::client::now_ms;
13use crate::transport::{gzip_decode_limited, gzip_if_beneficial};
14
15/// Shared immutable state used by the peer RPC HTTP router.
16#[derive(Clone)]
17pub struct PeerRpcHttpState {
18    pub(crate) manifest: DistributedCoreManifest,
19    pub(crate) validator: PeerRpcValidator,
20    pub(crate) dispatcher: Arc<dyn PeerRpcDispatcher>,
21    pub(crate) authenticator: Arc<dyn PeerRpcAuthenticator>,
22}
23
24/// HTTP host exposing the stable peer health, manifest, query, and command routes.
25pub struct PeerRpcHttpHost {
26    host: String,
27    port: u16,
28    router: Router,
29}
30impl PeerRpcHttpHost {
31    /// Creates a peer HTTP host with explicit validation, dispatch, and authentication.
32    pub fn new(
33        host: impl Into<String>,
34        port: u16,
35        manifest: DistributedCoreManifest,
36        validator: PeerRpcValidator,
37        dispatcher: Arc<dyn PeerRpcDispatcher>,
38        authenticator: Arc<dyn PeerRpcAuthenticator>,
39    ) -> Self {
40        let state = PeerRpcHttpState {
41            manifest,
42            validator,
43            dispatcher,
44            authenticator,
45        };
46        let router = Router::new()
47            .route(PEER_HEALTH_PATH, get(peer_health_handler))
48            .route(PEER_MANIFEST_PATH, get(peer_manifest_handler))
49            .route(PEER_QUERY_PATH, post(peer_query_handler))
50            .route(PEER_COMMAND_PATH, post(peer_command_handler))
51            .with_state(state);
52        Self {
53            host: host.into(),
54            port,
55            router,
56        }
57    }
58
59    /// Returns a clone of the configured Axum router.
60    pub fn router(&self) -> Router {
61        self.router.clone()
62    }
63
64    /// Runs the host until the shared shutdown flag is set.
65    pub fn run_until_shutdown(&self, shutdown: Arc<AtomicBool>) -> io::Result<()> {
66        let address = format!("{}:{}", self.host, self.port);
67        let router = self.router();
68        let runtime = tokio::runtime::Builder::new_current_thread()
69            .enable_all()
70            .build()
71            .map_err(io::Error::other)?;
72        runtime.block_on(async move {
73            let listener = tokio::net::TcpListener::bind(address).await?;
74            axum::serve(listener, router)
75                .with_graceful_shutdown(wait_for_shutdown(shutdown))
76                .await
77        })
78    }
79}
80
81async fn peer_health_handler(
82    State(state): State<PeerRpcHttpState>,
83    headers: HeaderMap,
84) -> Response {
85    match state
86        .authenticator
87        .authenticate(bearer_token(&headers), None, now_ms())
88    {
89        Ok(()) => {
90            let identity = &state.manifest.identity;
91            (
92                StatusCode::OK,
93                Json(PeerHealthResponse {
94                    ok: true,
95                    core_id: identity.core_id.clone(),
96                    tenant_id: identity.tenant_id.clone(),
97                    cluster_id: identity.cluster_id.clone(),
98                }),
99            )
100                .into_response()
101        }
102        Err(error) => peer_error_response("health", error),
103    }
104}
105
106async fn peer_manifest_handler(
107    State(state): State<PeerRpcHttpState>,
108    headers: HeaderMap,
109) -> Response {
110    match state
111        .authenticator
112        .authenticate(bearer_token(&headers), None, now_ms())
113    {
114        Ok(()) => (
115            StatusCode::OK,
116            Json(PeerManifestResponse {
117                advertisement: crate::advertisement::advertisement_from_manifest(&state.manifest),
118            }),
119        )
120            .into_response(),
121        Err(error) => peer_error_response("manifest", error),
122    }
123}
124
125async fn peer_query_handler(
126    State(state): State<PeerRpcHttpState>,
127    headers: HeaderMap,
128    body: Bytes,
129) -> Response {
130    let envelope = match decode_peer_envelope(&state, &headers, &body) {
131        Ok(envelope) => envelope,
132        Err(error) => return peer_error_response("invalid", error),
133    };
134    handle_peer_envelope(state, headers, envelope, PeerRpcKind::Query).await
135}
136
137async fn peer_command_handler(
138    State(state): State<PeerRpcHttpState>,
139    headers: HeaderMap,
140    body: Bytes,
141) -> Response {
142    let envelope = match decode_peer_envelope(&state, &headers, &body) {
143        Ok(envelope) => envelope,
144        Err(error) => return peer_error_response("invalid", error),
145    };
146    handle_peer_envelope(state, headers, envelope, PeerRpcKind::Command).await
147}
148
149pub(crate) fn decode_peer_envelope(
150    state: &PeerRpcHttpState,
151    headers: &HeaderMap,
152    body: &[u8],
153) -> Result<PeerRpcEnvelope, PeerRpcError> {
154    let max_bytes = state.validator.max_envelope_bytes();
155    let decoded = match headers.get(header::CONTENT_ENCODING) {
156        None => {
157            if body.len() > max_bytes {
158                return Err(PeerRpcError::PayloadTooLarge);
159            }
160            body.to_vec()
161        }
162        Some(value) if value.as_bytes().eq_ignore_ascii_case(b"gzip") => {
163            gzip_decode_limited(body, max_bytes)?
164        }
165        Some(_) => {
166            return Err(PeerRpcError::InvalidEnvelope(
167                "unsupported_content_encoding".to_string(),
168            ))
169        }
170    };
171    serde_json::from_slice(&decoded)
172        .map_err(|error| PeerRpcError::InvalidEnvelope(error.to_string()))
173}
174
175#[derive(Debug, Clone, Copy)]
176enum PeerRpcKind {
177    Query,
178    Command,
179}
180
181async fn handle_peer_envelope(
182    state: PeerRpcHttpState,
183    headers: HeaderMap,
184    envelope: PeerRpcEnvelope,
185    kind: PeerRpcKind,
186) -> Response {
187    let accepts_gzip = accepts_gzip(&headers);
188    let request_id = envelope.request_id.clone();
189    let now = now_ms();
190    if let Err(error) = state.authenticator.authenticate(
191        bearer_token(&headers),
192        Some(&envelope_signing_hash(&envelope)),
193        now,
194    ) {
195        return peer_error_response(&request_id, error);
196    }
197    if let Err(error) = state.validator.validate(&envelope, now) {
198        return peer_error_response(&request_id, error);
199    }
200    let dispatcher = Arc::clone(&state.dispatcher);
201    let response = tokio::task::spawn_blocking(move || match kind {
202        PeerRpcKind::Query => dispatcher.dispatch_peer_query(envelope),
203        PeerRpcKind::Command => dispatcher.dispatch_peer_command(envelope),
204    })
205    .await
206    .unwrap_or_else(|_| {
207        Err(PeerRpcError::Transport(
208            "peer dispatcher panicked".to_string(),
209        ))
210    });
211    match response {
212        Ok(response) => peer_json_response(StatusCode::OK, &response, accepts_gzip),
213        Err(error) => peer_error_response(&request_id, error),
214    }
215}
216
217fn peer_json_response<T>(status: StatusCode, value: &T, allow_gzip: bool) -> Response
218where
219    T: Serialize,
220{
221    if allow_gzip {
222        if let Ok(body) = serde_json::to_vec(value) {
223            if let Ok(Some(compressed)) = gzip_if_beneficial(&body) {
224                let mut response = (status, compressed).into_response();
225                response.headers_mut().insert(
226                    header::CONTENT_TYPE,
227                    HeaderValue::from_static("application/json"),
228                );
229                response
230                    .headers_mut()
231                    .insert(header::CONTENT_ENCODING, HeaderValue::from_static("gzip"));
232                return response;
233            }
234        }
235    }
236    (status, Json(value)).into_response()
237}
238
239pub(crate) fn accepts_gzip(headers: &HeaderMap) -> bool {
240    headers
241        .get(header::ACCEPT_ENCODING)
242        .and_then(|value| value.to_str().ok())
243        .map(|value| {
244            value.split(',').any(|encoding| {
245                let mut parts = encoding.trim().split(';');
246                let name = parts.next().unwrap_or_default().trim();
247                let disabled = parts.any(|parameter| {
248                    parameter
249                        .trim()
250                        .strip_prefix("q=")
251                        .map(|quality| quality.trim() == "0" || quality.trim() == "0.0")
252                        .unwrap_or(false)
253                });
254                name.eq_ignore_ascii_case("gzip") && !disabled
255            })
256        })
257        .unwrap_or(false)
258}
259
260fn peer_error_response(request_id: &str, error: PeerRpcError) -> Response {
261    let status = match error {
262        PeerRpcError::Unauthorized => StatusCode::UNAUTHORIZED,
263        PeerRpcError::Forbidden => StatusCode::FORBIDDEN,
264        PeerRpcError::EndpointUnavailable => StatusCode::SERVICE_UNAVAILABLE,
265        PeerRpcError::PayloadTooLarge => StatusCode::PAYLOAD_TOO_LARGE,
266        PeerRpcError::TenantMismatch
267        | PeerRpcError::ClusterMismatch
268        | PeerRpcError::TargetMismatch
269        | PeerRpcError::Expired
270        | PeerRpcError::NonceReplay
271        | PeerRpcError::InvalidBodyHash
272        | PeerRpcError::InvalidResponse(_)
273        | PeerRpcError::Transport(_)
274        | PeerRpcError::InvalidEnvelope(_) => StatusCode::BAD_REQUEST,
275        PeerRpcError::ProtocolMismatch => StatusCode::CONFLICT,
276        PeerRpcError::NonceCacheFull => StatusCode::SERVICE_UNAVAILABLE,
277    };
278    (
279        status,
280        Json(PeerRpcResponse::rejected(
281            request_id,
282            peer_error_code(&error),
283        )),
284    )
285        .into_response()
286}
287
288fn peer_error_code(error: &PeerRpcError) -> &'static str {
289    match error {
290        PeerRpcError::PayloadTooLarge => "payload_too_large",
291        PeerRpcError::Unauthorized => "unauthorized",
292        PeerRpcError::Forbidden => "forbidden",
293        PeerRpcError::EndpointUnavailable => "endpoint_unavailable",
294        PeerRpcError::TenantMismatch => "tenant_mismatch",
295        PeerRpcError::ClusterMismatch => "cluster_mismatch",
296        PeerRpcError::TargetMismatch => "target_mismatch",
297        PeerRpcError::ProtocolMismatch => "protocol_mismatch",
298        PeerRpcError::Expired => "expired",
299        PeerRpcError::NonceReplay => "nonce_replay",
300        PeerRpcError::NonceCacheFull => "nonce_cache_full",
301        PeerRpcError::InvalidBodyHash => "invalid_body_hash",
302        PeerRpcError::InvalidResponse(_) => "invalid_response",
303        PeerRpcError::Transport(_) => "transport",
304        PeerRpcError::InvalidEnvelope(_) => "invalid_envelope",
305    }
306}
307
308fn bearer_token(headers: &HeaderMap) -> Option<&str> {
309    headers
310        .get(axum::http::header::AUTHORIZATION)
311        .and_then(|value| value.to_str().ok())
312}
313
314async fn wait_for_shutdown(shutdown: Arc<AtomicBool>) {
315    loop {
316        if shutdown.load(Ordering::SeqCst) {
317            break;
318        }
319        tokio::time::sleep(Duration::from_millis(100)).await;
320    }
321}
appcore_peer_rpc/host.rs

appcore_peer_rpc/
host.rs
