prns_runtime/runtime/
request_endpoints.rs

1use crate::engine::InstantMillis;
2use crate::identity::IdentityHash;
3use crate::routing::links::request::{
4    packed_binary_len, write_packed_binary_header, RequestId, MAX_PACKED_BINARY_HEADER_LEN,
5};
6use crate::routing::links::LinkId;
7use crate::routing::request_handlers::{RequestPathHash, RequestPolicy};
8use crate::units::RttMillis;
9use crate::wire::DestinationHash;
10
11#[derive(Debug, Clone, Copy, PartialEq, Eq)]
12pub enum RequestEndpointPolicy {
13    AllowNone,
14    AllowAll,
15    AllowList(&'static [IdentityHash]),
16}
17
18impl RequestEndpointPolicy {
19    #[must_use]
20    pub fn engine_policy(self) -> RequestPolicy {
21        match self {
22            RequestEndpointPolicy::AllowNone => RequestPolicy::AllowNone,
23            RequestEndpointPolicy::AllowAll => RequestPolicy::AllowAll,
24            RequestEndpointPolicy::AllowList(_) => RequestPolicy::AllowList,
25        }
26    }
27
28    /// The identities to admit at registration — non-empty only for [`RequestEndpointPolicy::AllowList`].
29    #[must_use]
30    pub fn seed_list(self) -> &'static [IdentityHash] {
31        match self {
32            RequestEndpointPolicy::AllowList(list) => list,
33            _ => &[],
34        }
35    }
36}
37
38#[derive(Debug, Clone, Copy, PartialEq, Eq)]
39pub enum Decline {
40    /// Send no confirmation at all. This will contribute to a timeout on the Link if not handled yourself.
41    ///
42    /// See [`respond_token`](RequestContext::respond_token)
43    Ignore,
44    CloseLink,
45    ResponseTooLarge,
46}
47
48pub trait ResponseSink {
49    fn put_packed(&mut self, bytes: &[u8]) -> Result<(), ResponseCapacityExceeded>;
50
51    fn put_bytes(&mut self, bytes: &[u8]) -> Result<(), ResponseCapacityExceeded>;
52
53    fn put_static_bytes(&mut self, bytes: &'static [u8]) -> Result<(), ResponseCapacityExceeded> {
54        self.put_bytes(bytes)
55    }
56
57    fn put_static_file(
58        &mut self,
59        _name: &'static str,
60        _bytes: &'static [u8],
61    ) -> Result<(), ResponseCapacityExceeded> {
62        Err(ResponseCapacityExceeded)
63    }
64
65    #[cfg(feature = "std")]
66    fn put_open_bytes(
67        &mut self,
68        _file: std::fs::File,
69        _byte_len: u64,
70    ) -> Result<(), ResponseCapacityExceeded> {
71        Err(ResponseCapacityExceeded)
72    }
73
74    #[cfg(feature = "std")]
75    fn put_open_file(
76        &mut self,
77        _name: &str,
78        _file: std::fs::File,
79        _byte_len: u64,
80    ) -> Result<(), ResponseCapacityExceeded> {
81        Err(ResponseCapacityExceeded)
82    }
83}
84
85#[derive(Debug, Clone, Copy, PartialEq, Eq)]
86pub struct ResponseCapacityExceeded;
87
88#[cfg(feature = "alloc")]
89impl ResponseSink for alloc::vec::Vec<u8> {
90    fn put_packed(&mut self, bytes: &[u8]) -> Result<(), ResponseCapacityExceeded> {
91        self.extend_from_slice(bytes);
92        Ok(())
93    }
94
95    fn put_bytes(&mut self, bytes: &[u8]) -> Result<(), ResponseCapacityExceeded> {
96        let mut header = [0u8; MAX_PACKED_BINARY_HEADER_LEN];
97        let header_len = write_packed_binary_header(bytes.len(), &mut header)
98            .map_err(|_| ResponseCapacityExceeded)?;
99        self.reserve(header_len + bytes.len());
100        self.extend_from_slice(&header[..header_len]);
101        self.extend_from_slice(bytes);
102        Ok(())
103    }
104}
105
106impl<const N: usize> ResponseSink for heapless::Vec<u8, N> {
107    fn put_packed(&mut self, bytes: &[u8]) -> Result<(), ResponseCapacityExceeded> {
108        self.extend_from_slice(bytes)
109            .map_err(|_| ResponseCapacityExceeded)
110    }
111
112    fn put_bytes(&mut self, bytes: &[u8]) -> Result<(), ResponseCapacityExceeded> {
113        let packed_len = packed_binary_len(bytes.len()).ok_or(ResponseCapacityExceeded)?;
114        if self.capacity() - self.len() < packed_len {
115            return Err(ResponseCapacityExceeded);
116        }
117        let mut header = [0u8; MAX_PACKED_BINARY_HEADER_LEN];
118        let header_len = write_packed_binary_header(bytes.len(), &mut header)
119            .map_err(|_| ResponseCapacityExceeded)?;
120        self.extend_from_slice(&header[..header_len])
121            .map_err(|_| ResponseCapacityExceeded)?;
122        self.extend_from_slice(bytes)
123            .map_err(|_| ResponseCapacityExceeded)
124    }
125}
126
127/// Only needed if you don't respond to the request inside your [`handle`](RequestEndpoint::handle) function.
128/// See [`respond_token`](RequestContext::respond_token).
129#[derive(Debug, Clone, Copy, PartialEq, Eq)]
130pub struct RespondToken {
131    pub link_id: LinkId,
132    pub request_id: RequestId,
133    /// The link's measured round trip when the request arrived.
134    pub rtt: RttMillis,
135}
136
137pub struct InboundRequest<'a> {
138    pub destination: DestinationHash,
139    pub data: &'a [u8],
140    pub requester: Option<IdentityHash>,
141    pub requested_at: InstantMillis,
142    respond_token: RespondToken,
143}
144
145impl<'a> InboundRequest<'a> {
146    #[must_use]
147    pub fn new(
148        destination: DestinationHash,
149        link_id: LinkId,
150        request_id: RequestId,
151        requester: Option<IdentityHash>,
152        requested_at: InstantMillis,
153        rtt: RttMillis,
154        data: &'a [u8],
155    ) -> Self {
156        Self {
157            destination,
158            data,
159            requester,
160            requested_at,
161            respond_token: RespondToken {
162                link_id,
163                request_id,
164                rtt,
165            },
166        }
167    }
168
169    #[must_use]
170    pub fn respond_token(&self) -> RespondToken {
171        self.respond_token
172    }
173}
174
175pub struct RequestContext<'a, S> {
176    pub state: &'a S,
177    pub destination: DestinationHash,
178    pub data: &'a [u8],
179    pub requester: Option<IdentityHash>,
180    pub requested_at: InstantMillis,
181    respond_token: RespondToken,
182    sink: &'a mut dyn ResponseSink,
183}
184
185impl<S> RequestContext<'_, S> {
186    /// Send a normal application response exactly as supplied.
187    ///
188    /// This is the default for text, protocol messages, and arbitrary byte payloads:
189    ///
190    /// ```ignore
191    /// context.respond("pong")
192    /// ```
193    ///
194    /// RNS calls this payload "packed", but it does not require MessagePack. Use
195    /// [`respond_messagepack_bytes`](Self::respond_messagepack_bytes) only when the peer
196    /// specifically expects a MessagePack `bin` value, and the file methods only for
197    /// named-file/resource semantics.
198    pub fn respond(&mut self, data: impl AsRef<[u8]>) -> Result<(), Decline> {
199        self.sink
200            .put_packed(data.as_ref())
201            .map_err(|_| Decline::ResponseTooLarge)
202    }
203
204    /// Legacy Reticulum spelling for [`respond`](Self::respond).
205    #[deprecated(note = "use RequestContext::respond for exact application payloads")]
206    #[doc(hidden)]
207    pub fn respond_packed(&mut self, bytes: &[u8]) -> Result<(), Decline> {
208        self.respond(bytes)
209    }
210
211    /// Encode `bytes` as one MessagePack `bin` value before responding.
212    pub fn respond_messagepack_bytes(&mut self, bytes: &[u8]) -> Result<(), Decline> {
213        self.sink
214            .put_bytes(bytes)
215            .map_err(|_| Decline::ResponseTooLarge)
216    }
217
218    /// Legacy ambiguous spelling for
219    /// [`respond_messagepack_bytes`](Self::respond_messagepack_bytes).
220    #[deprecated(
221        note = "use RequestContext::respond_messagepack_bytes for a MessagePack bin value"
222    )]
223    #[doc(hidden)]
224    pub fn respond_bytes(&mut self, bytes: &[u8]) -> Result<(), Decline> {
225        self.respond_messagepack_bytes(bytes)
226    }
227
228    /// Encode static bytes as one MessagePack `bin` value without first copying the source.
229    pub fn respond_static_messagepack_bytes(
230        &mut self,
231        bytes: &'static [u8],
232    ) -> Result<(), Decline> {
233        self.sink
234            .put_static_bytes(bytes)
235            .map_err(|_| Decline::ResponseTooLarge)
236    }
237
238    /// Legacy ambiguous spelling for
239    /// [`respond_static_messagepack_bytes`](Self::respond_static_messagepack_bytes).
240    #[deprecated(
241        note = "use RequestContext::respond_static_messagepack_bytes for a static MessagePack bin value"
242    )]
243    #[doc(hidden)]
244    pub fn respond_static_bytes(&mut self, bytes: &'static [u8]) -> Result<(), Decline> {
245        self.respond_static_messagepack_bytes(bytes)
246    }
247
248    /// Respond with a Reticulum Resource whose metadata names the file for native clients.
249    ///
250    /// The bytes remain borrowed from static storage; resource segmentation copies only the
251    /// current transfer window into the outgoing resource buffer.
252    pub fn respond_static_file(
253        &mut self,
254        name: &'static str,
255        bytes: &'static [u8],
256    ) -> Result<(), Decline> {
257        self.sink
258            .put_static_file(name, bytes)
259            .map_err(|_| Decline::ResponseTooLarge)
260    }
261
262    /// Respond with bytes from an already-open regular file.
263    ///
264    /// Host runtimes keep the handle open until its response lane is available, then read it in
265    /// bounded segments. Opening and validating the handle before calling this method keeps path
266    /// policy in the application and avoids retaining the complete payload per queued request.
267    #[cfg(feature = "std")]
268    #[doc(hidden)]
269    pub fn respond_open_bytes(
270        &mut self,
271        file: std::fs::File,
272        byte_len: u64,
273    ) -> Result<(), Decline> {
274        self.sink
275            .put_open_bytes(file, byte_len)
276            .map_err(|_| Decline::ResponseTooLarge)
277    }
278
279    /// Respond with an already-open regular file and Reticulum filename metadata.
280    ///
281    /// The host runtime streams the file after acquiring the response lane, so queued requests
282    /// retain one handle and a small descriptor rather than a copy of the file.
283    #[cfg(feature = "std")]
284    #[doc(hidden)]
285    pub fn respond_open_file(
286        &mut self,
287        name: &str,
288        file: std::fs::File,
289        byte_len: u64,
290    ) -> Result<(), Decline> {
291        self.sink
292            .put_open_file(name, file, byte_len)
293            .map_err(|_| Decline::ResponseTooLarge)
294    }
295
296    pub fn write_packed(&mut self, bytes: &[u8]) -> Result<&mut Self, ResponseCapacityExceeded> {
297        self.sink.put_packed(bytes)?;
298        Ok(self)
299    }
300
301    /// The token to answer this request later. You can keep it, return `Err(Decline::Ignore)` now, and answer from another task through the platform command handle.
302    ///
303    /// In this context, "keeping it" usually means capturing it somewhere in your AppState
304    #[must_use]
305    pub fn respond_token(&self) -> RespondToken {
306        self.respond_token
307    }
308}
309
310/// What a requester names to reach a [`RequestEndpoint`]: the stable hash of its `ENDPOINT_ID` string.
311pub type RequestEndpointId = RequestPathHash;
312
313#[allow(async_fn_in_trait)]
314pub trait RequestEndpoint<AppState = ()> {
315    /// You can use whatever string value you like (it's hashed and truncated so the wire length will be stable), but it's common convention to use URL/filesystem-like syntax, e.g., "/example/thing"
316    const ENDPOINT_ID: &'static str;
317    const POLICY: RequestEndpointPolicy;
318    async fn handle(context: RequestContext<'_, AppState>) -> Result<(), Decline>;
319}
320
321/// A compile-time set of endpoints, produced by [`request_endpoints!`](crate::request_endpoints); you probably want
322/// that macro rather than this trait directly.
323#[allow(async_fn_in_trait)]
324pub trait RequestEndpointSet<S> {
325    const REGISTRATIONS: &'static [(&'static str, RequestEndpointPolicy)];
326    async fn dispatch(cx: RequestContext<'_, S>, path_hash: RequestPathHash)
327        -> Result<(), Decline>;
328}
329
330/// The empty route set — what [`request_endpoints!`](crate::request_endpoints) with no arms hands back, and what a node
331/// that serves no requests carries. It registers nothing and declines every request as `Ignore`.
332impl<S> RequestEndpointSet<S> for () {
333    const REGISTRATIONS: &'static [(&'static str, RequestEndpointPolicy)] = &[];
334    async fn dispatch(
335        _cx: RequestContext<'_, S>,
336        _path_hash: RequestPathHash,
337    ) -> Result<(), Decline> {
338        Err(Decline::Ignore)
339    }
340}
341
342/// The value [`request_endpoints!`](crate::request_endpoints) hands back when given no endpoints — the empty [`RequestEndpointSet`].
343/// A named constructor so the macro needn't expand to a bare `()`, which `clippy::unused_unit`
344/// flags at every call site.
345pub const fn no_request_endpoints() {}
346
347/// Route one request to the handler its `path_hash` selects, building the [`RequestContext`] over
348/// the app's shared `state` and the runner's grant `sink`. `RequestEndpointSet::dispatch` is a static fn, so
349/// the runner dispatches with only `&state` and the endpoint-set type `R` — no `Router` wrapper.
350pub async fn dispatch_request<'a, S, R: RequestEndpointSet<S>>(
351    state: &'a S,
352    path_hash: RequestPathHash,
353    request: InboundRequest<'a>,
354    sink: &'a mut dyn ResponseSink,
355) -> Result<(), Decline> {
356    let cx = RequestContext {
357        state,
358        destination: request.destination,
359        data: request.data,
360        requester: request.requester,
361        requested_at: request.requested_at,
362        respond_token: request.respond_token(),
363        sink,
364    };
365    R::dispatch(cx, path_hash).await
366}
367
368/// Compose route types into a [`RequestEndpointSet`] value, e.g., `request_endpoints![Health, Echo, Status]`. Each arm awaits
369/// a concrete handler future, so the set is monomorphized. There's no boxing and it's`no_std`-clean.
370#[macro_export]
371macro_rules! request_endpoints {
372    () => {
373        $crate::runtime::request_endpoints::no_request_endpoints()
374    };
375    ($($endpoint:ty),+ $(,)?) => {{
376        struct RequestEndpointSetImpl;
377        impl<S> $crate::runtime::request_endpoints::RequestEndpointSet<S> for RequestEndpointSetImpl
378        where
379            $($endpoint: $crate::runtime::request_endpoints::RequestEndpoint<S>,)+
380        {
381            const REGISTRATIONS: &'static [(&'static str, $crate::runtime::request_endpoints::RequestEndpointPolicy)] = &[
382                $((
383                    <$endpoint as $crate::runtime::request_endpoints::RequestEndpoint<S>>::ENDPOINT_ID,
384                    <$endpoint as $crate::runtime::request_endpoints::RequestEndpoint<S>>::POLICY,
385                ),)+
386            ];
387
388            async fn dispatch(
389                cx: $crate::runtime::request_endpoints::RequestContext<'_, S>,
390                path_hash: $crate::routing::request_handlers::RequestPathHash,
391            ) -> ::core::result::Result<(), $crate::runtime::request_endpoints::Decline> {
392                $(
393                    if path_hash
394                        == $crate::routing::request_handlers::RequestPathHash::of(
395                            <$endpoint as $crate::runtime::request_endpoints::RequestEndpoint<S>>::ENDPOINT_ID,
396                        )
397                    {
398                        return <$endpoint as $crate::runtime::request_endpoints::RequestEndpoint<S>>::handle(cx).await;
399                    }
400                )+
401                ::core::result::Result::Err($crate::runtime::request_endpoints::Decline::Ignore)
402            }
403        }
404        RequestEndpointSetImpl
405    }};
406}
407
408#[cfg(test)]
409mod tests {
410    use super::*;
411
412    struct App {
413        greeting: &'static [u8],
414    }
415
416    struct Health;
417    impl RequestEndpoint<App> for Health {
418        const ENDPOINT_ID: &'static str = "/health";
419        const POLICY: RequestEndpointPolicy = RequestEndpointPolicy::AllowAll;
420        async fn handle(mut cx: RequestContext<'_, App>) -> Result<(), Decline> {
421            cx.respond("ok")
422        }
423    }
424
425    struct Greet;
426    impl RequestEndpoint<App> for Greet {
427        const ENDPOINT_ID: &'static str = "/greet";
428        const POLICY: RequestEndpointPolicy = RequestEndpointPolicy::AllowAll;
429        async fn handle(mut cx: RequestContext<'_, App>) -> Result<(), Decline> {
430            let greeting = cx.state.greeting;
431            cx.respond(greeting)
432        }
433    }
434
435    const ADMIN: IdentityHash = IdentityHash::new([0xAD; 16]);
436
437    struct Admin;
438    impl RequestEndpoint<App> for Admin {
439        const ENDPOINT_ID: &'static str = "/admin";
440        const POLICY: RequestEndpointPolicy = RequestEndpointPolicy::AllowList(&[ADMIN]);
441        async fn handle(_cx: RequestContext<'_, App>) -> Result<(), Decline> {
442            Err(Decline::CloseLink)
443        }
444    }
445
446    struct Ack;
447    impl RequestEndpoint<App> for Ack {
448        const ENDPOINT_ID: &'static str = "/ack";
449        const POLICY: RequestEndpointPolicy = RequestEndpointPolicy::AllowAll;
450        async fn handle(mut cx: RequestContext<'_, App>) -> Result<(), Decline> {
451            cx.respond([0u8; 0])
452        }
453    }
454
455    fn registrations<R: RequestEndpointSet<App>>(
456        _endpoints: R,
457    ) -> &'static [(&'static str, RequestEndpointPolicy)] {
458        R::REGISTRATIONS
459    }
460
461    #[test]
462    fn the_endpoint_set_is_the_registration_set_the_recipe_stands_up() {
463        let registrations = registrations(crate::request_endpoints![Health, Greet, Admin, Ack]);
464        assert_eq!(registrations.len(), 4);
465        assert_eq!(
466            registrations[0],
467            ("/health", RequestEndpointPolicy::AllowAll)
468        );
469        assert_eq!(registrations[2].0, "/admin");
470        assert_eq!(registrations[2].1.engine_policy(), RequestPolicy::AllowList);
471        assert_eq!(registrations[2].1.seed_list(), &[ADMIN]);
472        assert_eq!(registrations[0].1.engine_policy(), RequestPolicy::AllowAll);
473        assert!(registrations[0].1.seed_list().is_empty());
474    }
475
476    #[test]
477    fn messagepack_binary_sinks_frame_atomically() {
478        let mut exact = heapless::Vec::<u8, 7>::new();
479        exact.put_bytes(b"hello").unwrap();
480        assert_eq!(exact.as_slice(), &[0xC4, 5, b'h', b'e', b'l', b'l', b'o']);
481
482        let mut short = heapless::Vec::<u8, 6>::new();
483        assert_eq!(short.put_bytes(b"hello"), Err(ResponseCapacityExceeded));
484        assert!(short.is_empty());
485    }
486
487    #[cfg(feature = "alloc")]
488    #[test]
489    fn dispatch_endpoints_by_path_then_answers_or_declines() {
490        futures_executor::block_on(async {
491            async fn dispatch<R: RequestEndpointSet<App>>(
492                _endpoints: &R,
493                state: &App,
494                path: &str,
495                sink: &mut dyn ResponseSink,
496            ) -> Result<(), Decline> {
497                let request = InboundRequest::new(
498                    DestinationHash::new([3; 16]),
499                    LinkId::new([1; 16]),
500                    RequestId([2; 16]),
501                    None,
502                    InstantMillis(0),
503                    RttMillis::new(0),
504                    b"",
505                );
506                dispatch_request::<App, R>(state, RequestPathHash::of(path), request, sink).await
507            }
508
509            let endpoints = crate::request_endpoints![Health, Greet, Admin, Ack];
510            let state = App { greeting: b"hi" };
511
512            let mut greet = std::vec::Vec::new();
513            assert_eq!(
514                dispatch(&endpoints, &state, "/greet", &mut greet).await,
515                Ok(())
516            );
517            assert_eq!(greet.as_slice(), b"hi");
518
519            let mut health = std::vec::Vec::new();
520            assert_eq!(
521                dispatch(&endpoints, &state, "/health", &mut health).await,
522                Ok(())
523            );
524            assert_eq!(health.as_slice(), b"ok");
525
526            let mut ack = std::vec::Vec::new();
527            assert_eq!(dispatch(&endpoints, &state, "/ack", &mut ack).await, Ok(()));
528            assert!(ack.is_empty());
529
530            let mut admin = std::vec::Vec::new();
531            assert_eq!(
532                dispatch(&endpoints, &state, "/admin", &mut admin).await,
533                Err(Decline::CloseLink)
534            );
535
536            let mut miss = std::vec::Vec::new();
537            assert_eq!(
538                dispatch(&endpoints, &state, "/nope", &mut miss).await,
539                Err(Decline::Ignore)
540            );
541        });
542    }
543}
prns_runtime/runtime/request_endpoints.rs

prns_runtime/runtime/
request_endpoints.rs
