appcore_peer_rpc/
client.rs

1// =============================================================================
2//        #######
3//     ###       ###     F: client.rs
4//    ##   ## ##   ##    P: AppCore-Runtime
5//         ## ##
6//                       C: 2026/07/22 15:41:18 by dnettoRaw
7//    ##   ## ##   ##    U: 2026/08/02 12:48:56 by dnettoRaw
8//      ###########      S: 1.0.1-rc.8
9// =============================================================================
10
11use super::*;
12use crate::transport::http_status_error;
13
14/// Retry limits and exponential backoff bounds for peer requests.
15#[derive(Debug, Clone, PartialEq, Eq)]
16pub struct PeerRpcRetryPolicy {
17    /// Maximum attempts, including the initial request.
18    pub max_attempts: usize,
19    /// Delay before the first retry.
20    pub initial_backoff_ms: u64,
21    /// Maximum delay between attempts.
22    pub max_backoff_ms: u64,
23}
24
25impl Default for PeerRpcRetryPolicy {
26    fn default() -> Self {
27        Self {
28            max_attempts: 2,
29            initial_backoff_ms: 50,
30            max_backoff_ms: 500,
31        }
32    }
33}
34
35/// Runtime limits used by [`PeerRpcClient`].
36#[derive(Debug, Clone, PartialEq, Eq)]
37pub struct PeerRpcClientConfig {
38    /// Per-attempt network timeout.
39    pub request_timeout_ms: u64,
40    /// Lifetime assigned to signed request envelopes.
41    pub envelope_ttl_ms: u64,
42    /// Maximum accepted response body size.
43    pub max_response_bytes: usize,
44    /// Retry behavior for transport failures.
45    pub retry_policy: PeerRpcRetryPolicy,
46}
47
48impl Default for PeerRpcClientConfig {
49    fn default() -> Self {
50        Self {
51            request_timeout_ms: 5_000,
52            envelope_ttl_ms: 60_000,
53            max_response_bytes: 1_048_576,
54            retry_policy: PeerRpcRetryPolicy::default(),
55        }
56    }
57}
58
59/// HTTP request emitted through a [`PeerTransportProvider`].
60#[derive(Clone, PartialEq, Eq)]
61pub struct PeerRpcHttpRequest {
62    /// HTTP method.
63    pub method: String,
64    /// Stable peer RPC endpoint path.
65    pub path: String,
66    /// Encoded request body.
67    pub body: Vec<u8>,
68    /// Optional bearer credential; debug output always redacts it.
69    pub bearer_token: Option<String>,
70    /// Per-attempt network timeout.
71    pub timeout_ms: u64,
72    /// Maximum accepted response body size.
73    pub max_response_bytes: usize,
74}
75
76impl std::fmt::Debug for PeerRpcHttpRequest {
77    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
78        formatter
79            .debug_struct("PeerRpcHttpRequest")
80            .field("method", &self.method)
81            .field("path", &self.path)
82            .field("body_bytes", &self.body.len())
83            .field(
84                "bearer_token",
85                &self.bearer_token.as_ref().map(|_| "REDACTED"),
86            )
87            .field("timeout_ms", &self.timeout_ms)
88            .field("max_response_bytes", &self.max_response_bytes)
89            .finish()
90    }
91}
92
93/// Bounded response returned by a [`PeerTransportProvider`].
94#[derive(Clone, PartialEq, Eq)]
95pub struct PeerRpcHttpResponse {
96    /// HTTP status code.
97    pub status_code: u16,
98    /// Raw response body.
99    pub body: Vec<u8>,
100}
101
102impl std::fmt::Debug for PeerRpcHttpResponse {
103    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
104        formatter
105            .debug_struct("PeerRpcHttpResponse")
106            .field("status_code", &self.status_code)
107            .field("body_bytes", &self.body.len())
108            .finish()
109    }
110}
111
112/// Peer transport provider used by RPC clients to reach a peer endpoint.
113pub trait PeerTransportProvider: Send + Sync {
114    /// Sends one request to the peer base URL.
115    fn send(
116        &self,
117        base_url: &str,
118        request: PeerRpcHttpRequest,
119    ) -> Result<PeerRpcHttpResponse, PeerRpcError>;
120
121    /// Sends one request with cooperative cancellation.
122    fn send_cancellable(
123        &self,
124        base_url: &str,
125        request: PeerRpcHttpRequest,
126        cancellation: &CancellationToken,
127    ) -> Result<PeerRpcHttpResponse, PeerRpcError> {
128        if cancellation.is_cancelled() {
129            return Err(PeerRpcError::EndpointUnavailable);
130        }
131        self.send(base_url, request)
132    }
133}
134
135/// Authenticated client for stable peer query, command, and diagnostic endpoints.
136pub struct PeerRpcClient<T, I> {
137    source_identity: CoreIdentity,
138    config: PeerRpcClientConfig,
139    transport: T,
140    token_issuer: I,
141    cancellation: CancellationToken,
142}
143impl<T, I> PeerRpcClient<T, I>
144where
145    T: PeerTransportProvider,
146    I: PeerRpcTokenIssuer,
147{
148    /// Creates an authenticated peer RPC client.
149    pub fn new(
150        source_identity: CoreIdentity,
151        config: PeerRpcClientConfig,
152        transport: T,
153        token_issuer: I,
154    ) -> Self {
155        Self {
156            source_identity,
157            config,
158            transport,
159            token_issuer,
160            cancellation: CancellationToken::new(),
161        }
162    }
163
164    /// Replaces the shared cancellation token used by I/O and retry waits.
165    pub fn with_cancellation_token(mut self, cancellation: CancellationToken) -> Self {
166        self.cancellation = cancellation;
167        self
168    }
169
170    /// Cancels active official transport I/O and future retries.
171    pub fn cancel(&self) {
172        self.cancellation.cancel();
173    }
174
175    /// Reports whether this client has been cancelled.
176    pub fn is_cancelled(&self) -> bool {
177        self.cancellation.is_cancelled()
178    }
179
180    /// Executes a peer query.
181    pub fn query(
182        &self,
183        endpoint_url: &str,
184        request: PeerRpcOutboundRequest,
185    ) -> Result<PeerRpcResponse, PeerRpcError> {
186        self.call_peer(endpoint_url, PeerRpcCallKind::Query, request)
187    }
188
189    /// Executes a peer command.
190    pub fn command(
191        &self,
192        endpoint_url: &str,
193        request: PeerRpcOutboundRequest,
194    ) -> Result<PeerRpcResponse, PeerRpcError> {
195        self.call_peer(endpoint_url, PeerRpcCallKind::Command, request)
196    }
197
198    /// Reads the authenticated peer health endpoint.
199    pub fn health(&self, endpoint_url: &str) -> Result<PeerHealthResponse, PeerRpcError> {
200        let request_id = format!("health-{}-{}", std::process::id(), now_ms());
201        let token = self.token_issuer.issue_peer_token(
202            &request_id,
203            None,
204            now_ms(),
205            self.config.envelope_ttl_ms,
206        )?;
207        let response = self.send_with_retry(
208            endpoint_url,
209            PeerRpcHttpRequest {
210                method: "GET".to_string(),
211                path: PEER_HEALTH_PATH.to_string(),
212                body: Vec::new(),
213                bearer_token: Some(token),
214                timeout_ms: self.config.request_timeout_ms,
215                max_response_bytes: self.config.max_response_bytes,
216            },
217        )?;
218        serde_json::from_slice(&response.body)
219            .map_err(|error| PeerRpcError::InvalidResponse(error.to_string()))
220    }
221
222    /// Reads and returns the authenticated peer runtime manifest.
223    /// Loads the versioned public advertisement exposed by a peer.
224    pub fn advertisement(&self, endpoint_url: &str) -> Result<PeerAdvertisementV1, PeerRpcError> {
225        let request_id = format!("manifest-{}-{}", std::process::id(), now_ms());
226        let token = self.token_issuer.issue_peer_token(
227            &request_id,
228            None,
229            now_ms(),
230            self.config.envelope_ttl_ms,
231        )?;
232        let response = self.send_with_retry(
233            endpoint_url,
234            PeerRpcHttpRequest {
235                method: "GET".to_string(),
236                path: PEER_MANIFEST_PATH.to_string(),
237                body: Vec::new(),
238                bearer_token: Some(token),
239                timeout_ms: self.config.request_timeout_ms,
240                max_response_bytes: self.config.max_response_bytes,
241            },
242        )?;
243        let response = serde_json::from_slice::<PeerManifestResponse>(&response.body)
244            .map_err(|error| PeerRpcError::InvalidResponse(error.to_string()))?;
245        Ok(response.advertisement)
246    }
247
248    fn build_envelope(&self, request: &PeerRpcOutboundRequest) -> PeerRpcEnvelope {
249        let now = now_ms();
250        let trace_id = request
251            .trace
252            .as_ref()
253            .map(|trace| trace.trace_id.clone())
254            .unwrap_or_else(|| request.request_id.clone());
255        let mut envelope = PeerRpcEnvelope::new(
256            request.request_id.clone(),
257            trace_id,
258            self.source_identity.core_id.clone(),
259            request.target_core_id.clone(),
260            self.source_identity.tenant_id.clone(),
261            self.source_identity.cluster_id.clone(),
262            now,
263            now.saturating_add(self.config.envelope_ttl_ms.max(1)),
264            next_outbound_nonce(&request.request_id, now),
265            request.capability.clone(),
266            request.payload.clone(),
267            request.idempotency_key.clone(),
268            request.trace.clone(),
269        );
270        envelope.protocol_version = self.source_identity.protocol_version;
271        envelope
272    }
273
274    fn send_with_retry(
275        &self,
276        endpoint_url: &str,
277        request: PeerRpcHttpRequest,
278    ) -> Result<PeerRpcHttpResponse, PeerRpcError> {
279        let attempts = self.config.retry_policy.max_attempts.max(1);
280        let mut backoff_ms = self.config.retry_policy.initial_backoff_ms;
281        let mut last_error = PeerRpcError::EndpointUnavailable;
282        for attempt in 0..attempts {
283            match self
284                .transport
285                .send_cancellable(endpoint_url, request.clone(), &self.cancellation)
286            {
287                Ok(response) if (200..300).contains(&response.status_code) => return Ok(response),
288                Ok(response) => {
289                    last_error = http_status_error(response.status_code, response.body);
290                }
291                Err(error) => last_error = error,
292            }
293            if attempt + 1 < attempts {
294                if self
295                    .cancellation
296                    .wait_timeout(Duration::from_millis(backoff_ms))
297                {
298                    return Err(PeerRpcError::EndpointUnavailable);
299                }
300                backoff_ms = backoff_ms
301                    .saturating_mul(2)
302                    .min(self.config.retry_policy.max_backoff_ms);
303            }
304        }
305        Err(last_error)
306    }
307}
308
309impl<T, I> PeerRpcClientExecutor for PeerRpcClient<T, I>
310where
311    T: PeerTransportProvider,
312    I: PeerRpcTokenIssuer,
313{
314    fn call_peer(
315        &self,
316        endpoint_url: &str,
317        kind: PeerRpcCallKind,
318        request: PeerRpcOutboundRequest,
319    ) -> Result<PeerRpcResponse, PeerRpcError> {
320        let configured_attempts = self.config.retry_policy.max_attempts.max(1);
321        let attempts = if kind == PeerRpcCallKind::Command && request.idempotency_key.is_none() {
322            1
323        } else {
324            configured_attempts
325        };
326        let mut backoff_ms = self.config.retry_policy.initial_backoff_ms;
327        let mut last_error = PeerRpcError::EndpointUnavailable;
328
329        for attempt in 0..attempts {
330            if self.cancellation.is_cancelled() {
331                return Err(PeerRpcError::EndpointUnavailable);
332            }
333            match self.call_peer_once(endpoint_url, kind, &request) {
334                Ok(response) => return Ok(response),
335                Err(error) => {
336                    let retryable = peer_error_is_retryable(&error);
337                    last_error = error;
338                    if !retryable {
339                        break;
340                    }
341                }
342            }
343            if attempt + 1 < attempts {
344                if self
345                    .cancellation
346                    .wait_timeout(Duration::from_millis(backoff_ms))
347                {
348                    return Err(PeerRpcError::EndpointUnavailable);
349                }
350                backoff_ms = backoff_ms
351                    .saturating_mul(2)
352                    .min(self.config.retry_policy.max_backoff_ms);
353            }
354        }
355        Err(last_error)
356    }
357}
358
359impl<T, I> PeerRpcClient<T, I>
360where
361    T: PeerTransportProvider,
362    I: PeerRpcTokenIssuer,
363{
364    fn call_peer_once(
365        &self,
366        endpoint_url: &str,
367        kind: PeerRpcCallKind,
368        request: &PeerRpcOutboundRequest,
369    ) -> Result<PeerRpcResponse, PeerRpcError> {
370        let envelope = self.build_envelope(request);
371        let envelope_hash = envelope_signing_hash(&envelope);
372        let token = self.token_issuer.issue_peer_token(
373            &envelope.request_id,
374            Some(&envelope_hash),
375            now_ms(),
376            self.config.envelope_ttl_ms,
377        )?;
378        let body = serde_json::to_vec(&envelope)
379            .map_err(|error| PeerRpcError::InvalidEnvelope(error.to_string()))?;
380        let response = self.transport.send_cancellable(
381            endpoint_url,
382            PeerRpcHttpRequest {
383                method: "POST".to_string(),
384                path: match kind {
385                    PeerRpcCallKind::Query => PEER_QUERY_PATH,
386                    PeerRpcCallKind::Command => PEER_COMMAND_PATH,
387                }
388                .to_string(),
389                body,
390                bearer_token: Some(token),
391                timeout_ms: self.config.request_timeout_ms,
392                max_response_bytes: self.config.max_response_bytes,
393            },
394            &self.cancellation,
395        )?;
396        if !(200..300).contains(&response.status_code) {
397            return Err(http_status_error(response.status_code, response.body));
398        }
399        let response = serde_json::from_slice::<PeerRpcResponse>(&response.body)
400            .map_err(|error| PeerRpcError::InvalidResponse(error.to_string()))?;
401        if response.request_id != request.request_id {
402            return Err(PeerRpcError::InvalidResponse(
403                "peer response request_id mismatch".to_string(),
404            ));
405        }
406        Ok(response)
407    }
408}
409
410fn next_outbound_nonce(request_id: &str, now_ms: u64) -> String {
411    // appcore-norm: allow(global-state) reason: atomic sequence prevents process-local nonce reuse
412    static NONCE_COUNTER: AtomicU64 = AtomicU64::new(0);
413    let counter = NONCE_COUNTER.fetch_add(1, Ordering::Relaxed);
414    format!(
415        "{}-{}-{}-{}",
416        request_id,
417        now_ms,
418        std::process::id(),
419        counter
420    )
421}
422
423fn peer_error_is_retryable(error: &PeerRpcError) -> bool {
424    matches!(
425        error,
426        PeerRpcError::EndpointUnavailable | PeerRpcError::Transport(_)
427    )
428}
429
430pub(crate) fn now_ms() -> u64 {
431    SystemTime::now()
432        .duration_since(UNIX_EPOCH)
433        .map(|duration| duration.as_millis() as u64)
434        .unwrap_or(0)
435}
appcore_peer_rpc/client.rs

appcore_peer_rpc/
client.rs
