appcore_distributed_contracts/peer_rpc/
v1.rs

1// =============================================================================
2//        #######
3//     ###       ###     F: v1.rs
4//    ##   ## ##   ##    P: AppCore-Runtime
5//         ## ##
6//                       C: 2026/07/22 13:21:42 by dnettoRaw
7//    ##   ## ##   ##    U: 2026/08/02 12:48:56 by dnettoRaw
8//      ###########      S: 1.0.1-rc.8
9// =============================================================================
10
11//! Peer RPC protocol version 1.
12
13use appcore_types::{CapabilityName, ClusterId, CoreId, ProtocolVersion, TenantId, TraceContext};
14use serde::{Deserialize, Serialize};
15use sha2::{Digest, Sha256};
16use std::collections::BTreeMap;
17use std::fmt::{Debug, Formatter};
18
19/// Version number of this peer RPC wire contract.
20pub const PEER_RPC_PROTOCOL_VERSION: u16 = 1;
21/// Public authenticated peer health endpoint.
22pub const PEER_HEALTH_PATH: &str = "/v1/peer/health";
23/// Public authenticated peer manifest endpoint.
24pub const PEER_MANIFEST_PATH: &str = "/v1/peer/manifest";
25/// Authenticated peer query endpoint.
26pub const PEER_QUERY_PATH: &str = "/v1/peer/query";
27/// Authenticated peer command endpoint.
28pub const PEER_COMMAND_PATH: &str = "/v1/peer/command";
29
30/// Authenticated peer request envelope.
31#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
32pub struct PeerRpcEnvelope {
33    /// Stable request identity.
34    pub request_id: String,
35    /// Trace identity propagated across cores.
36    pub trace_id: String,
37    /// Distributed protocol version.
38    #[serde(default)]
39    pub protocol_version: ProtocolVersion,
40    /// Core issuing the request.
41    pub source_core_id: CoreId,
42    /// Core expected to execute the request.
43    pub target_core_id: CoreId,
44    /// Tenant isolation boundary.
45    pub tenant_id: TenantId,
46    /// Cluster isolation boundary.
47    pub cluster_id: ClusterId,
48    /// Creation timestamp in milliseconds.
49    pub timestamp_ms: u64,
50    /// Expiration timestamp in milliseconds.
51    pub expires_at_ms: u64,
52    /// Single-use replay-protection value.
53    pub nonce: String,
54    /// Generic capability being invoked.
55    pub capability: CapabilityName,
56    /// Opaque application-owned payload.
57    pub payload: Vec<u8>,
58    /// Optional idempotency key for a mutating request.
59    pub idempotency_key: Option<String>,
60    /// SHA-256 digest of `payload`.
61    pub body_hash: String,
62    /// Optional structured trace context.
63    pub trace: Option<TraceContext>,
64}
65
66impl Debug for PeerRpcEnvelope {
67    fn fmt(&self, formatter: &mut Formatter<'_>) -> std::fmt::Result {
68        formatter
69            .debug_struct("PeerRpcEnvelope")
70            .field("request_id", &self.request_id)
71            .field("trace_id", &self.trace_id)
72            .field("protocol_version", &self.protocol_version)
73            .field("source_core_id", &self.source_core_id)
74            .field("target_core_id", &self.target_core_id)
75            .field("tenant_id", &self.tenant_id)
76            .field("cluster_id", &self.cluster_id)
77            .field("timestamp_ms", &self.timestamp_ms)
78            .field("expires_at_ms", &self.expires_at_ms)
79            .field("capability", &self.capability)
80            .field("payload_bytes", &self.payload.len())
81            .field("body_hash", &self.body_hash)
82            .field("has_idempotency_key", &self.idempotency_key.is_some())
83            .field("trace", &self.trace)
84            .finish()
85    }
86}
87
88impl PeerRpcEnvelope {
89    /// Creates a versioned envelope and binds its payload digest.
90    #[allow(clippy::too_many_arguments)]
91    pub fn new(
92        request_id: impl Into<String>,
93        trace_id: impl Into<String>,
94        source_core_id: CoreId,
95        target_core_id: CoreId,
96        tenant_id: TenantId,
97        cluster_id: ClusterId,
98        timestamp_ms: u64,
99        expires_at_ms: u64,
100        nonce: impl Into<String>,
101        capability: CapabilityName,
102        payload: Vec<u8>,
103        idempotency_key: Option<String>,
104        trace: Option<TraceContext>,
105    ) -> Self {
106        let body_hash = payload_hash(&payload);
107        Self {
108            request_id: request_id.into(),
109            trace_id: trace_id.into(),
110            protocol_version: ProtocolVersion::default(),
111            source_core_id,
112            target_core_id,
113            tenant_id,
114            cluster_id,
115            timestamp_ms,
116            expires_at_ms,
117            nonce: nonce.into(),
118            capability,
119            payload,
120            idempotency_key,
121            body_hash,
122            trace,
123        }
124    }
125}
126
127/// Response returned for a peer query or command.
128#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
129pub struct PeerRpcResponse {
130    /// Whether execution succeeded.
131    pub ok: bool,
132    /// Request identity echoed by the peer.
133    pub request_id: String,
134    /// Opaque application-owned response payload.
135    pub payload: Vec<u8>,
136    /// Controlled failure detail.
137    pub error: Option<String>,
138}
139
140impl Debug for PeerRpcResponse {
141    fn fmt(&self, formatter: &mut Formatter<'_>) -> std::fmt::Result {
142        formatter
143            .debug_struct("PeerRpcResponse")
144            .field("ok", &self.ok)
145            .field("request_id", &self.request_id)
146            .field("payload_bytes", &self.payload.len())
147            .field("has_error", &self.error.is_some())
148            .finish()
149    }
150}
151
152impl PeerRpcResponse {
153    /// Creates a successful response.
154    pub fn ok(request_id: impl Into<String>, payload: Vec<u8>) -> Self {
155        Self {
156            ok: true,
157            request_id: request_id.into(),
158            payload,
159            error: None,
160        }
161    }
162
163    /// Creates a controlled rejected response.
164    pub fn rejected(request_id: impl Into<String>, error: impl Into<String>) -> Self {
165        Self {
166            ok: false,
167            request_id: request_id.into(),
168            payload: Vec::new(),
169            error: Some(error.into()),
170        }
171    }
172}
173
174/// Provider-independent peer RPC failure.
175#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
176pub enum PeerRpcError {
177    /// The request or response exceeds its configured bound.
178    #[error("peer RPC payload is too large")]
179    PayloadTooLarge,
180    /// Authentication credentials are missing or invalid.
181    #[error("peer RPC request is unauthorized")]
182    Unauthorized,
183    /// Credentials are valid but do not authorize this request.
184    #[error("peer RPC request is forbidden")]
185    Forbidden,
186    /// No eligible peer endpoint is available.
187    #[error("peer RPC endpoint is unavailable")]
188    EndpointUnavailable,
189    /// Source and target tenants differ.
190    #[error("peer RPC tenant mismatch")]
191    TenantMismatch,
192    /// Source and target clusters differ.
193    #[error("peer RPC cluster mismatch")]
194    ClusterMismatch,
195    /// The request targets another core.
196    #[error("peer RPC target mismatch")]
197    TargetMismatch,
198    /// Source and target protocol versions are incompatible.
199    #[error("peer RPC protocol mismatch")]
200    ProtocolMismatch,
201    /// The envelope has expired.
202    #[error("peer RPC envelope expired")]
203    Expired,
204    /// The envelope nonce was already accepted.
205    #[error("peer RPC nonce replay")]
206    NonceReplay,
207    /// Replay protection reached its configured bound.
208    #[error("peer RPC nonce cache is full")]
209    NonceCacheFull,
210    /// The payload does not match the envelope digest.
211    #[error("peer RPC body hash is invalid")]
212    InvalidBodyHash,
213    /// The remote endpoint returned an invalid response.
214    #[error("invalid peer RPC response: {0}")]
215    InvalidResponse(String),
216    /// Transport execution failed.
217    #[error("peer RPC transport failed: {0}")]
218    Transport(String),
219    /// The incoming envelope is malformed.
220    #[error("invalid peer RPC envelope: {0}")]
221    InvalidEnvelope(String),
222}
223
224/// Kind of direct peer call.
225#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
226#[serde(rename_all = "snake_case")]
227pub enum PeerRpcCallKind {
228    /// Side-effect-free query.
229    Query,
230    /// Mutating or important command.
231    Command,
232}
233
234/// Provider-neutral request passed to a peer client executor.
235#[derive(Clone, PartialEq, Eq)]
236pub struct PeerRpcOutboundRequest {
237    /// Stable request identity.
238    pub request_id: String,
239    /// Core expected to execute the request.
240    pub target_core_id: CoreId,
241    /// Generic capability being invoked.
242    pub capability: CapabilityName,
243    /// Opaque application-owned payload.
244    pub payload: Vec<u8>,
245    /// Optional idempotency key.
246    pub idempotency_key: Option<String>,
247    /// Optional trace context.
248    pub trace: Option<TraceContext>,
249}
250
251impl Debug for PeerRpcOutboundRequest {
252    fn fmt(&self, formatter: &mut Formatter<'_>) -> std::fmt::Result {
253        formatter
254            .debug_struct("PeerRpcOutboundRequest")
255            .field("request_id", &self.request_id)
256            .field("target_core_id", &self.target_core_id)
257            .field("capability", &self.capability)
258            .field("payload_bytes", &self.payload.len())
259            .field("has_idempotency_key", &self.idempotency_key.is_some())
260            .field("trace", &self.trace)
261            .finish()
262    }
263}
264
265impl PeerRpcOutboundRequest {
266    /// Creates an outbound peer request.
267    pub fn new(
268        request_id: impl Into<String>,
269        target_core_id: CoreId,
270        capability: CapabilityName,
271        payload: Vec<u8>,
272        idempotency_key: Option<String>,
273        trace: Option<TraceContext>,
274    ) -> Self {
275        Self {
276            request_id: request_id.into(),
277            target_core_id,
278            capability,
279            payload,
280            idempotency_key,
281            trace,
282        }
283    }
284}
285
286/// Provider contract used to invoke a direct peer endpoint.
287pub trait PeerRpcClientExecutor: Send + Sync {
288    /// Executes one query or command against `endpoint_url`.
289    fn call_peer(
290        &self,
291        endpoint_url: &str,
292        kind: PeerRpcCallKind,
293        request: PeerRpcOutboundRequest,
294    ) -> Result<PeerRpcResponse, PeerRpcError>;
295}
296
297/// Response returned by the peer health endpoint.
298#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
299pub struct PeerHealthResponse {
300    /// Whether the peer is ready to receive calls.
301    pub ok: bool,
302    /// Peer core identity.
303    pub core_id: CoreId,
304    /// Peer tenant boundary.
305    pub tenant_id: TenantId,
306    /// Peer cluster boundary.
307    pub cluster_id: ClusterId,
308}
309
310/// Response returned by the peer manifest endpoint.
311#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
312pub struct PeerManifestResponse {
313    /// Versioned provider-independent peer advertisement.
314    pub advertisement: PeerAdvertisementV1,
315}
316
317/// Stable peer advertisement independent of internal core-manifest layout.
318#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
319pub struct PeerAdvertisementV1 {
320    /// Advertisement schema version.
321    pub schema_version: u16,
322    /// Distributed and Runtime identity fields required for compatibility.
323    pub identity: PeerIdentityV1,
324    /// Human-readable application name.
325    pub app_name: String,
326    /// Application version.
327    pub app_version: String,
328    /// Minimum compatible Runtime version.
329    pub runtime_min_version: String,
330    /// Optional maximum compatible Runtime version.
331    pub runtime_max_version: Option<String>,
332    /// Generic capabilities exposed by this peer.
333    pub capabilities: Vec<PeerCapabilityV1>,
334    /// Public network endpoints without credentials.
335    pub endpoints: Vec<PeerEndpointV1>,
336    /// Non-sensitive routing metadata.
337    pub metadata: BTreeMap<String, String>,
338}
339
340/// Identity fields carried by a V1 peer advertisement.
341#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
342pub struct PeerIdentityV1 {
343    /// Tenant isolation boundary.
344    pub tenant_id: String,
345    /// Cluster isolation boundary.
346    pub cluster_id: String,
347    /// Stable logical core identity.
348    pub core_id: String,
349    /// Unique running instance identity.
350    pub instance_id: String,
351    /// Generic core role.
352    pub kind: String,
353    /// Distributed protocol version.
354    pub protocol_version: u16,
355    /// Application identity.
356    pub app_id: String,
357    /// Compatible application family.
358    pub app_family: String,
359    /// Sync compatibility group.
360    pub sync_group: String,
361    /// Runtime contract version.
362    pub runtime_contract: u16,
363    /// Runtime node identity.
364    pub node_id: String,
365}
366
367/// Generic capability advertised by a peer.
368#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
369pub struct PeerCapabilityV1 {
370    /// Stable capability name.
371    pub name: String,
372    /// Capability contract version.
373    pub version: String,
374    /// `query`, `command`, or `stream`.
375    pub mode: String,
376    /// `local`, `cluster`, or `tenant`.
377    pub visibility: String,
378    /// Whether service leadership is required.
379    pub requires_leader: bool,
380    /// Whether the capability is read-only.
381    pub read_only: bool,
382    /// Whether mutating requests require idempotency.
383    pub idempotency_required: bool,
384}
385
386/// Public endpoint advertised by a peer.
387#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
388pub struct PeerEndpointV1 {
389    /// Logical endpoint name.
390    pub name: String,
391    /// Public endpoint URL.
392    pub url: String,
393    /// Transport protocol identifier.
394    pub protocol: String,
395    /// Non-sensitive endpoint metadata.
396    pub metadata: BTreeMap<String, String>,
397}
398
399fn payload_hash(payload: &[u8]) -> String {
400    let digest = Sha256::digest(payload);
401    let mut output = String::with_capacity(digest.len() * 2);
402    const HEX: &[u8; 16] = b"0123456789abcdef";
403    for byte in digest {
404        output.push(HEX[(byte >> 4) as usize] as char);
405        output.push(HEX[(byte & 0x0f) as usize] as char);
406    }
407    output
408}
409
410#[cfg(test)]
411mod tests {
412    use super::*;
413
414    #[test]
415    fn envelope_has_stable_v1_json_shape() {
416        let envelope = PeerRpcEnvelope::new(
417            "req-1",
418            "trace-1",
419            CoreId::new("core-a").unwrap(),
420            CoreId::new("core-b").unwrap(),
421            TenantId::new("tenant-a").unwrap(),
422            ClusterId::new("cluster-a").unwrap(),
423            10,
424            20,
425            "nonce-1",
426            CapabilityName::new("runtime.query").unwrap(),
427            b"hello".to_vec(),
428            None,
429            None,
430        );
431        let encoded = serde_json::to_value(envelope).unwrap();
432        let fixture: serde_json::Value =
433            serde_json::from_str(include_str!("../fixtures/peer-rpc-envelope-v1.json")).unwrap();
434        assert_eq!(encoded, fixture);
435    }
436
437    #[test]
438    fn peer_debug_omits_opaque_payloads_and_error_details() {
439        let marker = b"secret-marker-must-not-appear";
440        let envelope = PeerRpcEnvelope::new(
441            "req-1",
442            "trace-1",
443            CoreId::new("core-a").unwrap(),
444            CoreId::new("core-b").unwrap(),
445            TenantId::new("tenant-a").unwrap(),
446            ClusterId::new("cluster-a").unwrap(),
447            10,
448            20,
449            "nonce-secret-marker-must-not-appear",
450            CapabilityName::new("runtime.query").unwrap(),
451            marker.to_vec(),
452            Some("secret-marker-must-not-appear".to_string()),
453            None,
454        );
455        let response = PeerRpcResponse::rejected("req-1", "secret-marker-must-not-appear");
456        let outbound = PeerRpcOutboundRequest::new(
457            "req-1",
458            CoreId::new("core-b").unwrap(),
459            CapabilityName::new("runtime.query").unwrap(),
460            marker.to_vec(),
461            Some("secret-marker-must-not-appear".to_string()),
462            None,
463        );
464
465        assert!(!format!("{envelope:?}").contains("secret-marker-must-not-appear"));
466        assert!(!format!("{response:?}").contains("secret-marker-must-not-appear"));
467        assert!(!format!("{outbound:?}").contains("secret-marker-must-not-appear"));
468    }
469}
appcore_distributed_contracts/peer_rpc/v1.rs

appcore_distributed_contracts/peer_rpc/
v1.rs
