actordb_protocol/
rpc.rs

1//! Framework-neutral unary RPC boundary. Adapters own authentication and I/O.
2//! No adapter may transparently replay a mutation. Decoding must be bounded
3//! before allocating message collections, including opaque receipt payloads.
4
5use std::future::Future;
6use std::pin::Pin;
7use std::sync::Arc;
8use std::time::Instant;
9
10use crate::{client, RpcResponseShape};
11
12pub type RpcFuture<'a, T> = Pin<Box<dyn Future<Output = T> + Send + 'a>>;
13/// Adapter/service ownership carried until decoding, apply, or response encoding
14/// finishes. Dropping a caller must not discard an already admitted mutation.
15pub type RpcOwnership = Vec<Box<dyn Send + Sync + 'static>>;
16
17#[derive(Debug, Clone)]
18#[non_exhaustive]
19pub enum RpcError {
20    /// The adapter can prove that the operation was not dispatched.
21    NotDispatched(String),
22    /// Dispatch may have occurred; mutation callers must report uncertainty.
23    Failure(String),
24}
25impl std::fmt::Display for RpcError {
26    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
27        match self {
28            Self::NotDispatched(s) | Self::Failure(s) => f.write_str(s),
29        }
30    }
31}
32impl std::error::Error for RpcError {}
33
34#[derive(Debug, Clone)]
35pub struct RpcCallOptions {
36    pub deadline: Instant,
37    pub max_decoded_response_bytes: usize,
38    pub response_shape: RpcResponseShape,
39}
40
41/// A typed request plus the adapter's retained receive/admission ownership.
42pub struct RpcRequest {
43    request: client::Request,
44    ownership: RpcOwnership,
45}
46impl RpcRequest {
47    pub fn new(request: client::Request) -> Self {
48        Self {
49            request,
50            ownership: Vec::new(),
51        }
52    }
53    pub fn retain<T: Send + Sync + 'static>(&mut self, owner: T) {
54        self.ownership.push(Box::new(owner));
55    }
56    pub fn into_parts(self) -> (client::Request, RpcOwnership) {
57        (self.request, self.ownership)
58    }
59    pub fn from_parts(request: client::Request, ownership: RpcOwnership) -> Self {
60        Self { request, ownership }
61    }
62}
63impl std::ops::Deref for RpcRequest {
64    type Target = client::Request;
65    fn deref(&self) -> &Self::Target {
66        &self.request
67    }
68}
69
70/// A response retains server drain or client materialization ownership until
71/// the adapter/body or SDK conversion completes. Returned user values are then
72/// caller-owned and outside the RPC materialization allowance.
73pub struct RpcReply {
74    pub response: client::Response,
75    ownership: RpcOwnership,
76}
77impl RpcReply {
78    pub fn new(response: client::Response) -> Self {
79        Self {
80            response,
81            ownership: Vec::new(),
82        }
83    }
84    pub fn retain<T: Send + Sync + 'static>(&mut self, owner: T) {
85        self.ownership.push(Box::new(owner));
86    }
87    pub fn into_parts(self) -> (client::Response, RpcOwnership) {
88        (self.response, self.ownership)
89    }
90    pub fn from_parts(response: client::Response, ownership: RpcOwnership) -> Self {
91        Self {
92            response,
93            ownership,
94        }
95    }
96}
97impl std::fmt::Debug for RpcReply {
98    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
99        f.debug_struct("RpcReply").finish_non_exhaustive()
100    }
101}
102impl std::ops::Deref for RpcReply {
103    type Target = client::Response;
104    fn deref(&self) -> &Self::Target {
105        &self.response
106    }
107}
108impl std::ops::DerefMut for RpcReply {
109    fn deref_mut(&mut self) -> &mut Self::Target {
110        &mut self.response
111    }
112}
113
114/// Object-safe client adapter: gRPC, Thrift, or another unary RPC framework.
115pub trait RpcClient: Send + Sync + 'static {
116    fn call(
117        &self,
118        request: Arc<client::Request>,
119        options: RpcCallOptions,
120    ) -> RpcFuture<'_, Result<RpcReply, RpcError>>;
121}
122
123/// Object-safe server adapter target. Authentication/ACL decisions happen in
124/// the framework adapter before it calls this service, never in this trait.
125pub trait RpcService: Send + Sync + 'static {
126    fn call(&self, request: RpcRequest) -> RpcFuture<'_, Result<RpcReply, RpcError>>;
127}
actordb_protocol/rpc.rs

actordb_protocol/
rpc.rs
