rullst_core/
rpc.rs

1//! Versioned transport primitives for `#[server_function]`.
2//!
3//! This module owns serialization, size limits, correlation, and redacted
4//! failures. Authentication, authorization, tenant scope, idempotency, and
5//! domain validation remain server-side application policy. Production routes
6//! must be composed with Rullst's security baseline and the application's
7//! identity/authorization layers.
8
9use crate::client_contract::{ClientContractError, FailureCode, FailureDetail};
10
11/// A typed result returned by every bounded Rullst server function.
12pub type RpcResult<T> = Result<T, RpcFailure>;
13
14/// Machine-readable RPC failure with no arbitrary provider or debug message.
15#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
16#[error("RPC failed with code `{}`", code.as_str())]
17#[non_exhaustive]
18pub struct RpcFailure {
19    code: FailureCode,
20    retryable: bool,
21}
22
23impl RpcFailure {
24    /// Creates an application failure from the bounded lowercase dotted-code
25    /// grammar, such as `counter.limit_reached`.
26    pub fn application(
27        code: impl Into<String>,
28        retryable: bool,
29    ) -> Result<Self, ClientContractError> {
30        Ok(Self {
31            code: FailureCode::new(code)?,
32            retryable,
33        })
34    }
35
36    /// Returns the stable failure code.
37    pub fn code(&self) -> &str {
38        self.code.as_str()
39    }
40
41    /// Whether a retry may be appropriate. This never authorizes replaying a
42    /// state-changing operation.
43    pub const fn retryable(&self) -> bool {
44        self.retryable
45    }
46
47    pub(crate) fn framework(code: &'static str, retryable: bool) -> Self {
48        Self {
49            code: FailureCode::framework(code),
50            retryable,
51        }
52    }
53
54    #[cfg(target_arch = "wasm32")]
55    pub(crate) fn from_detail(detail: &FailureDetail) -> Self {
56        Self {
57            code: detail.code().clone(),
58            retryable: detail.retryable(),
59        }
60    }
61
62    #[cfg(not(target_arch = "wasm32"))]
63    fn detail(&self) -> FailureDetail {
64        FailureDetail::new(self.code.clone(), self.retryable)
65    }
66}
67
68#[cfg(not(target_arch = "wasm32"))]
69mod server {
70    use super::{RpcFailure, RpcResult};
71    use crate::Router;
72    use crate::client_contract::{
73        CURRENT_CLIENT_CONTRACT_VERSION, ClientContractError, ClientContractPolicy, RequestId,
74        ServerFailure, ServerResponse,
75    };
76    use axum::body::{Body, to_bytes};
77    use axum::extract::Request;
78    use axum::http::{StatusCode, header};
79    use axum::response::{IntoResponse, Response};
80    use serde::Serialize;
81    use serde::de::DeserializeOwned;
82    use std::future::Future;
83    use std::time::{SystemTime, UNIX_EPOCH};
84
85    /// Builds the single POST route generated for a server function.
86    ///
87    /// The returned router intentionally has no implicit identity policy. Merge
88    /// it into the application before installing the production security,
89    /// session, authentication, tenant, authorization, and rate-limit layers.
90    pub fn route<H, T>(path: &'static str, handler: H) -> Router
91    where
92        T: 'static,
93        H: axum::handler::Handler<T, ()>,
94    {
95        Router::new().route(path, axum::routing::post(handler))
96    }
97
98    /// Decodes one bounded request, invokes the generated typed adapter, and
99    /// emits the matching versioned success or failure envelope.
100    pub async fn handle_request<Req, Output, Invoke, InvokeFuture>(
101        request: Request,
102        invoke: Invoke,
103    ) -> Response
104    where
105        Req: DeserializeOwned + Send + 'static,
106        Output: Serialize,
107        Invoke: FnOnce(Req) -> InvokeFuture,
108        InvokeFuture: Future<Output = RpcResult<Output>>,
109    {
110        let content_type = request
111            .headers()
112            .get(header::CONTENT_TYPE)
113            .and_then(|value| value.to_str().ok())
114            .unwrap_or_default();
115        if !is_json_content_type(content_type) {
116            return failure_response(
117                StatusCode::UNSUPPORTED_MEDIA_TYPE,
118                server_request_id(),
119                RpcFailure::framework("rpc.content_type_required", false),
120            );
121        }
122
123        let (_, body) = request.into_parts();
124        let policy = ClientContractPolicy::default();
125        let body = match to_bytes(body, policy.max_body_bytes()).await {
126            Ok(body) => body,
127            Err(_) => {
128                return failure_response(
129                    StatusCode::PAYLOAD_TOO_LARGE,
130                    server_request_id(),
131                    RpcFailure::framework("rpc.request_too_large", false),
132                );
133            }
134        };
135        let envelope = match policy.decode_request::<Req>(&body) {
136            Ok(envelope) => envelope,
137            Err(error) => {
138                let (status, failure) = request_contract_failure(&error);
139                return failure_response(status, client_request_id(&body), failure);
140            }
141        };
142        let version = envelope.version();
143        let request_id = envelope.request_id().clone();
144
145        match invoke(envelope.into_payload()).await {
146            Ok(data) => {
147                let response =
148                    ServerResponse::new(version, request_id.clone(), now_epoch_ms(), data);
149                match policy.encode_response(&response) {
150                    Ok(body) => json_response(StatusCode::OK, body),
151                    Err(_) => failure_response(
152                        StatusCode::INTERNAL_SERVER_ERROR,
153                        request_id,
154                        RpcFailure::framework("rpc.response_encoding", false),
155                    ),
156                }
157            }
158            Err(failure) => failure_response(StatusCode::UNPROCESSABLE_ENTITY, request_id, failure),
159        }
160    }
161
162    fn request_contract_failure(error: &ClientContractError) -> (StatusCode, RpcFailure) {
163        match error {
164            ClientContractError::BodyTooLarge { .. } => (
165                StatusCode::PAYLOAD_TOO_LARGE,
166                RpcFailure::framework("rpc.request_too_large", false),
167            ),
168            ClientContractError::UnsupportedVersion { .. } => (
169                StatusCode::CONFLICT,
170                RpcFailure::framework("rpc.version_unsupported", false),
171            ),
172            _ => (
173                StatusCode::BAD_REQUEST,
174                RpcFailure::framework("rpc.request_invalid", false),
175            ),
176        }
177    }
178
179    fn failure_response(
180        status: StatusCode,
181        request_id: RequestId,
182        failure: RpcFailure,
183    ) -> Response {
184        let policy = ClientContractPolicy::default();
185        let response = ServerFailure::new(
186            CURRENT_CLIENT_CONTRACT_VERSION,
187            request_id,
188            now_epoch_ms(),
189            failure.detail(),
190        );
191        match policy.encode_failure(&response) {
192            Ok(body) => json_response(status, body),
193            Err(_) => (
194                StatusCode::INTERNAL_SERVER_ERROR,
195                [(header::CONTENT_TYPE, "application/json")],
196                Body::from(
197                    r#"{"contract":"rullst.client","version":1,"error":{"code":"rpc.internal","retryable":false}}"#,
198                ),
199            )
200                .into_response(),
201        }
202    }
203
204    fn json_response(status: StatusCode, body: Vec<u8>) -> Response {
205        (status, [(header::CONTENT_TYPE, "application/json")], body).into_response()
206    }
207
208    /// Echoes the caller's validated `request_id` when a rejected envelope
209    /// still carries one (for example an unsupported version or a payload of
210    /// the wrong shape), so the client correlates the failure and sees its real
211    /// code instead of `rpc.correlation_mismatch`. Otherwise a fresh
212    /// framework identifier is used.
213    fn client_request_id(body: &[u8]) -> RequestId {
214        #[derive(serde::Deserialize)]
215        struct Correlation {
216            request_id: RequestId,
217        }
218        serde_json::from_slice::<Correlation>(body)
219            .map(|correlation| correlation.request_id)
220            .unwrap_or_else(|_| server_request_id())
221    }
222
223    fn server_request_id() -> RequestId {
224        let value = format!("rpc_{}", uuid::Uuid::new_v4().simple());
225        RequestId::framework(value)
226    }
227
228    fn now_epoch_ms() -> u64 {
229        SystemTime::now()
230            .duration_since(UNIX_EPOCH)
231            .ok()
232            .and_then(|duration| u64::try_from(duration.as_millis()).ok())
233            .unwrap_or(0)
234    }
235
236    fn is_json_content_type(content_type: &str) -> bool {
237        content_type
238            .split(';')
239            .next()
240            .is_some_and(|media_type| media_type.trim().eq_ignore_ascii_case("application/json"))
241    }
242}
243
244#[cfg(not(target_arch = "wasm32"))]
245pub use server::{handle_request, route};
246
247#[cfg(all(test, not(target_arch = "wasm32")))]
248#[allow(clippy::unwrap_used, clippy::expect_used)]
249mod tests {
250    use super::*;
251    use crate::client_contract::ClientContractPolicy;
252
253    async fn failure_for(body: &str) -> (u16, String, String) {
254        let request = axum::http::Request::post("/api/rpc/test")
255            .header("content-type", "application/json")
256            .body(axum::body::Body::from(body.to_string()))
257            .unwrap();
258        let response =
259            handle_request::<u32, u32, _, _>(request, |value| async move { Ok(value) }).await;
260        let status = response.status().as_u16();
261        let bytes = axum::body::to_bytes(response.into_body(), 64 * 1024)
262            .await
263            .unwrap();
264        let failure = ClientContractPolicy::default()
265            .decode_failure(&bytes)
266            .expect("a framework failure envelope");
267        (
268            status,
269            failure.request_id().as_str().to_string(),
270            failure.error().code().as_str().to_string(),
271        )
272    }
273
274    #[tokio::test]
275    async fn rejected_envelopes_echo_the_client_request_id() {
276        let (status, id, code) = failure_for(
277            r#"{"contract":"rullst.client","version":99,"request_id":"rpc_client1","payload":1}"#,
278        )
279        .await;
280        assert_eq!(
281            (status, id.as_str(), code.as_str()),
282            (409, "rpc_client1", "rpc.version_unsupported")
283        );
284
285        let (status, id, code) = failure_for(
286            r#"{"contract":"rullst.client","version":1,"request_id":"rpc_client2","payload":"x"}"#,
287        )
288        .await;
289        assert_eq!(
290            (status, id.as_str(), code.as_str()),
291            (400, "rpc_client2", "rpc.request_invalid")
292        );
293
294        // Without a valid identifier the server still answers with its own.
295        let (status, id, code) = failure_for(r#"{"request_id":"has space"}"#).await;
296        assert_eq!((status, code.as_str()), (400, "rpc.request_invalid"));
297        assert!(id.starts_with("rpc_") && id != "has space", "{id}");
298    }
299}
rullst_core/rpc.rs

rullst_core/
rpc.rs
