macula_rust/station_link/
serve.rs

1//! Serving, as macula 12's provider does it. A procedure is served under an
2//! org namespace: the realm's org directory names the org's key, the org's
3//! procedure delegation names this node, both are found in the DHT, and the
4//! signed procedure_advertisement carries them to the station in an
5//! ADVERTISE, checked against the realm key before it goes out. Or it is
6//! served in this node's own namespace, `~<node_id>/<name>` (D25 item 6): its
7//! advertisement carries no authorization, its signature alone authorizes it,
8//! and no realm key or record in the DHT is needed. The station routes CALLs
9//! for the procedure to this link; each is admitted once per (caller,
10//! request_id) and answered with a RESULT or ERROR signed by this node.
11//! UNADVERTISE carries a tombstone of the advertisement.
12//!
13//! Only open procedures are served: a gated one needs a post-quantum UCAN
14//! verifier, which this crate does not have.
15
16use std::future::Future;
17use std::pin::Pin;
18use std::sync::{Arc, Mutex, Weak};
19use std::time::Duration;
20
21use tokio::sync::watch;
22
23use crate::cbor::{self, Value};
24use crate::frame::{self, StreamMode, VerifiedRequest};
25use crate::record::{
26    self, Authorization, ProcedureAdvertisementOptions, Reason, Record, RecordError,
27    TombstoneOptions, Trust,
28};
29
30use super::admission::Verdict;
31use super::framing::MAX_FRAME_BYTES;
32use super::stream::StreamHandler;
33use super::{now_ms, Inner, Link, LinkError};
34
35/// The provider codes of a served procedure's ERRORs: a handler's own
36/// refusal, a handler that panicked, a procedure this link does not serve, a
37/// copy of a request still running, and a result the wire cannot carry.
38const CODE_HANDLER_ERROR: &str = "handler_error";
39const CODE_HANDLER_CRASHED: &str = "temporary_relay_failure";
40const CODE_UNKNOWN_PROCEDURE: &str = "unknown_next_peer";
41pub(super) const CODE_REQUEST_COPY: &str = "request_copy";
42const CODE_PAYLOAD_TOO_LARGE: &str = "payload_too_large";
43const CODE_UNSENDABLE: &str = "unknown_error";
44
45/// An ERROR's detail is at most 256 bytes, cut on a character boundary.
46const MAX_DETAIL_BYTES: usize = 256;
47
48/// macula's default and longest advertisement lifetime.
49const MAX_ADVERTISEMENT_TTL: Duration = Duration::from_secs(5 * 60);
50/// How soon a failed renewal is tried again, while the advertisement it
51/// replaces still lives.
52const REFRESH_RETRY: Duration = Duration::from_secs(10);
53
54/// A future a handler returns.
55pub type BoxFuture<T> = Pin<Box<dyn Future<Output = T> + Send + 'static>>;
56
57/// Answers a [`Request`] with a result payload, or an error whose text the
58/// caller receives as a handler_error's detail. A handler still running at
59/// the request's deadline is dropped and answered handler_error.
60pub type Handler = Arc<dyn Fn(Request) -> BoxFuture<Result<Value, String>> + Send + Sync>;
61
62/// A [`Handler`] from an async closure.
63pub fn handler<F, Fut>(f: F) -> Handler
64where
65    F: Fn(Request) -> Fut + Send + Sync + 'static,
66    Fut: Future<Output = Result<Value, String>> + Send + 'static,
67{
68    Arc::new(move |r| Box::pin(f(r)))
69}
70
71/// A CALL a served procedure answers: the caller (the key id its signature
72/// verified under), what it asked for, and its deadline in unix
73/// milliseconds.
74#[derive(Debug, Clone, PartialEq)]
75pub struct Request {
76    pub caller: [u8; 32],
77    pub realm: [u8; 32],
78    pub procedure: String,
79    pub payload: Value,
80    pub token: Option<Vec<u8>>,
81    pub proofs: Option<Vec<Vec<u8>>>,
82    pub deadline_ms: u64,
83}
84
85/// A procedure to serve: its realm and name, exactly one of a unary handler
86/// and a stream offer, and, for an org procedure, the realm key the org
87/// directory must be signed with, as the realm's members pin it; a procedure
88/// in this node's own namespace needs none.
89#[derive(Clone)]
90pub struct Offer {
91    pub realm: [u8; 32],
92    pub procedure: String,
93    pub handler: Option<Handler>,
94    pub stream: Option<StreamOffer>,
95    pub realm_key: Option<Vec<u8>>,
96}
97
98impl Offer {
99    /// A unary procedure, with no realm key.
100    pub fn unary(realm: [u8; 32], procedure: &str, handler: Handler) -> Offer {
101        Offer {
102            realm,
103            procedure: procedure.to_string(),
104            handler: Some(handler),
105            stream: None,
106            realm_key: None,
107        }
108    }
109
110    /// A streaming procedure of `mode`, with no realm key.
111    pub fn stream(
112        realm: [u8; 32],
113        procedure: &str,
114        mode: StreamMode,
115        handler: StreamHandler,
116    ) -> Offer {
117        Offer {
118            realm,
119            procedure: procedure.to_string(),
120            handler: None,
121            stream: Some(StreamOffer { mode, handler }),
122            realm_key: None,
123        }
124    }
125}
126
127/// A streaming procedure's mode and handler. The advertisement is the one a
128/// unary procedure sends, which names no mode: a STREAM_OPEN of another mode
129/// is refused mode_mismatch.
130#[derive(Clone)]
131pub struct StreamOffer {
132    pub mode: StreamMode,
133    pub handler: StreamHandler,
134}
135
136/// A procedure a link serves, as the link holds it.
137pub(super) type ServedEntry = Arc<ServedInner>;
138
139pub(super) struct ServedInner {
140    link: Weak<Inner>,
141    key: ([u8; 32], String),
142    pub(super) offer: Offer,
143    latest: Mutex<Record>,
144    err: Mutex<Option<LinkError>>,
145    done_tx: watch::Sender<bool>,
146}
147
148/// A procedure this link serves, until [`Served::stop`] or the link ends, or
149/// until its advertisement lapses because its authorization could not be
150/// found again. Cloning it shares the serving.
151#[derive(Clone)]
152pub struct Served {
153    inner: Arc<ServedInner>,
154}
155
156impl Link {
157    /// Advertises `o`'s procedure on the link and answers its CALLs until
158    /// stopped. For an org procedure it resolves the org directory and this
159    /// node's procedure delegation from the DHT; a procedure in this node's
160    /// own namespace needs neither, and another node's namespace is refused.
161    /// It signs the advertisement with this node's identity key, naming the
162    /// connected station as the serving station and living no longer than
163    /// the records it carries nor 5 minutes, and checks its authorization
164    /// (against the realm key for an org procedure) before sending it in an
165    /// ADVERTISE and putting it in the DHT. The advertisement is renewed at
166    /// half its lifetime.
167    pub async fn serve(&self, o: Offer) -> Result<Served, LinkError> {
168        let own = record::in_own_namespace(&o.procedure);
169        if o.handler.is_some() == o.stream.is_some() || (!own && o.realm_key.is_none()) {
170            return Err(LinkError::InvalidOffer);
171        }
172        if !matches!(record::procedure_org(&o.procedure), Ok(Some(_))) {
173            return Err(LinkError::NoOrg);
174        }
175        let (advertisement, wire) = self.advertisement(&o, MAX_ADVERTISEMENT_TTL).await?;
176        let key = (o.realm, o.procedure.clone());
177        let (done_tx, _) = watch::channel(false);
178        let served = Arc::new(ServedInner {
179            link: Arc::downgrade(&self.inner),
180            key: key.clone(),
181            offer: o,
182            latest: Mutex::new(advertisement),
183            err: Mutex::new(None),
184            done_tx,
185        });
186        {
187            let mut state = self.inner.lock();
188            if let Some(e) = &state.ended {
189                return Err(e.clone());
190            }
191            if state.served.contains_key(&key) {
192                return Err(LinkError::AlreadyServed);
193            }
194            state.served.insert(key, served.clone());
195        }
196        if let Err(e) = self.announce(&wire).await {
197            served.end(e.clone());
198            return Err(e);
199        }
200        tokio::spawn(renew(served.clone()));
201        Ok(Served { inner: served })
202    }
203
204    /// `o`'s signed procedure_advertisement and its wire form, living at most
205    /// `max_ttl`: with the authorization resolved from the DHT for an org
206    /// procedure, with none in this node's own namespace.
207    async fn advertisement(
208        &self,
209        o: &Offer,
210        max_ttl: Duration,
211    ) -> Result<(Record, Vec<u8>), LinkError> {
212        let inner = &self.inner;
213        let max_ttl_ms = max_ttl.as_millis() as u64;
214        // This node opens no sealed request, so its advertisement names no
215        // KEM key: callers reach it in the clear.
216        let opts = if record::in_own_namespace(&o.procedure) {
217            ProcedureAdvertisementOptions {
218                authorization: Authorization::None,
219                kem_key: None,
220                ttl_ms: max_ttl_ms,
221            }
222        } else {
223            let org = record::procedure_org(&o.procedure)?.ok_or(LinkError::NoOrg)?;
224            let directory = self
225                .find_record(&record::org_directory_key(&o.realm, org))
226                .await?;
227            let named = record::read_org_directory(directory.record())?;
228            let delegation = self
229                .find_record(&record::procedure_delegation_key(
230                    &named.org_key,
231                    &inner.self_id,
232                ))
233                .await?;
234            let now = now_ms();
235            let ttl = (max_ttl_ms as i64)
236                .min(directory.record().expires_at as i64 - now)
237                .min(delegation.record().expires_at as i64 - now);
238            if ttl <= 0 {
239                return Err(RecordError::AuthorizationOutlived.into());
240            }
241            ProcedureAdvertisementOptions {
242                authorization: Authorization::Delegation {
243                    org_directory: record::encode(directory.record())?,
244                    procedure_delegation: record::encode(delegation.record())?,
245                },
246                kem_key: None,
247                ttl_ms: ttl as u64,
248            }
249        };
250        let unsigned = record::new_procedure_advertisement(
251            &inner.self_id,
252            &o.realm,
253            &o.procedure,
254            &inner.station.node_id,
255            &opts,
256        )?;
257        // Signed, then checked as a caller will check it before it is sent
258        // anywhere.
259        let signed = record::sign(&unsigned, &inner.key)?;
260        let wire = record::encode(&signed)?;
261        let now = now_ms();
262        let verified = record::verify(&wire, inner.profile, now)?;
263        record::verify_authorization(
264            &verified,
265            &Trust {
266                profile: inner.profile,
267                realm_key: o.realm_key.clone(),
268            },
269            now,
270        )?;
271        Ok((signed, wire))
272    }
273
274    /// Sends an advertisement to the station in an ADVERTISE, which routes
275    /// CALLs through it, and puts it in the DHT, where a caller resolving
276    /// the procedure finds it, as macula's advertise_direct does both.
277    async fn announce(&self, wire: &[u8]) -> Result<(), LinkError> {
278        self.inner
279            .send_control(&frame::advertise_frame(wire))
280            .await?;
281        self.put_record(wire).await
282    }
283}
284
285impl Served {
286    /// Withdraws the advertisement with an UNADVERTISE carrying its
287    /// tombstone, signed by this node, puts the tombstone in the
288    /// advertisement's DHT slot, and stops answering the procedure's CALLs.
289    /// Stopping a procedure no longer served does nothing.
290    pub async fn stop(&self) -> Result<(), LinkError> {
291        if self.inner.is_done() {
292            return Ok(());
293        }
294        self.inner.end(LinkError::Stopped);
295        let Some(inner) = self.inner.link.upgrade() else {
296            return Ok(());
297        };
298        let latest = self
299            .inner
300            .latest
301            .lock()
302            .unwrap_or_else(|p| p.into_inner())
303            .clone();
304        let tombstone =
305            record::new_tombstone(&latest, Reason::Shutdown, &TombstoneOptions::default())?;
306        let wire = record::encode(&record::sign(&tombstone, &inner.key)?)?;
307        inner.send_control(&frame::unadvertise_frame(&wire)).await?;
308        Link { inner }.put_record(&wire).await
309    }
310
311    /// Waits until the procedure is no longer served, and says why.
312    pub async fn done(&self) -> LinkError {
313        let mut done = self.inner.done_tx.subscribe();
314        let _ = done.wait_for(|ended| *ended).await;
315        self.error().unwrap_or(LinkError::Stopped)
316    }
317
318    /// Why the procedure is no longer served: [`LinkError::Stopped`] after
319    /// stop, the link's error when it ended, or the renewal's failure; `None`
320    /// while served.
321    pub fn error(&self) -> Option<LinkError> {
322        self.inner
323            .err
324            .lock()
325            .unwrap_or_else(|p| p.into_inner())
326            .clone()
327    }
328}
329
330impl ServedInner {
331    fn is_done(&self) -> bool {
332        *self.done_tx.borrow()
333    }
334
335    /// Ends the serving once, with `err`: the link no longer routes the
336    /// procedure's CALLs to its handler.
337    pub(super) fn end(self: &Arc<Self>, err: LinkError) {
338        {
339            let mut held = self.err.lock().unwrap_or_else(|p| p.into_inner());
340            if held.is_some() {
341                return;
342            }
343            *held = Some(err);
344        }
345        if let Some(inner) = self.link.upgrade() {
346            let mut state = inner.lock();
347            if state
348                .served
349                .get(&self.key)
350                .is_some_and(|s| Arc::ptr_eq(s, self))
351            {
352                state.served.remove(&self.key);
353            }
354        }
355        let _ = self.done_tx.send_replace(true);
356    }
357}
358
359/// Sends a fresh advertisement at half the current one's lifetime. A renewal
360/// that fails is tried again every [`REFRESH_RETRY`] while the current one
361/// lives; when it lapses unrenewed, the procedure is no longer served and its
362/// error says why.
363async fn renew(served: Arc<ServedInner>) {
364    let mut current = served
365        .latest
366        .lock()
367        .unwrap_or_else(|p| p.into_inner())
368        .clone();
369    let mut wait = half_life(&current);
370    let mut last_err: Option<LinkError> = None;
371    let mut stopped = served.done_tx.subscribe();
372    loop {
373        let Some(mut link_done) = served.link.upgrade().map(|l| l.done_rx.clone()) else {
374            return;
375        };
376        tokio::select! {
377            _ = stopped.wait_for(|ended| *ended) => return,
378            _ = link_done.wait_for(|ended| *ended) => {
379                let err = served.link.upgrade().and_then(|l| l.lock().ended.clone()).unwrap_or(LinkError::Closed);
380                served.end(err);
381                return;
382            }
383            _ = tokio::time::sleep(wait) => {}
384        }
385        if now_ms() >= current.expires_at as i64 {
386            served.end(last_err.unwrap_or(LinkError::Stopped));
387            return;
388        }
389        let Some(inner) = served.link.upgrade() else {
390            return;
391        };
392        let link = Link { inner };
393        let renewed = async {
394            let (advertisement, wire) = link
395                .advertisement(&served.offer, MAX_ADVERTISEMENT_TTL)
396                .await?;
397            link.announce(&wire).await?;
398            Ok::<_, LinkError>(advertisement)
399        }
400        .await;
401        match renewed {
402            Ok(advertisement) => {
403                *served.latest.lock().unwrap_or_else(|p| p.into_inner()) = advertisement.clone();
404                wait = half_life(&advertisement);
405                current = advertisement;
406            }
407            Err(e) => {
408                last_err = Some(e);
409                let left = (current.expires_at as i64 - now_ms()).max(0) as u64;
410                wait = REFRESH_RETRY.min(Duration::from_millis(left));
411            }
412        }
413    }
414}
415
416fn half_life(r: &Record) -> Duration {
417    Duration::from_millis(r.expires_at.saturating_sub(r.created_at) / 2)
418}
419
420/// Answers a CALL the station routed to this link. A request that does not
421/// verify, or targets another node, gets no reply and is counted. One that
422/// verifies is judged by the admission, then answered by its handler, or by
423/// an ERROR naming why not.
424pub(super) fn called(inner: &Arc<Inner>, v: &Value) {
425    let Ok(request) = frame::verify_request(v, inner.profile) else {
426        inner.count("unverified_call");
427        return;
428    };
429    if request.target != inner.self_id {
430        inner.count("call_for_another_node");
431        return;
432    }
433    let inner = inner.clone();
434    match inner.admission.admit(&request, &inner.share, now_ms()) {
435        Verdict::Refused(code) => {
436            let reply = provider_error(&inner, &request, code, None);
437            tokio::spawn(async move { send_reply(&inner, reply).await });
438        }
439        Verdict::Copy(None) => {
440            let reply = provider_error(&inner, &request, CODE_REQUEST_COPY, None);
441            tokio::spawn(async move { send_reply(&inner, reply).await });
442        }
443        Verdict::Copy(Some(stored)) => {
444            tokio::spawn(async move {
445                let _ = inner.control.write(&stored, MAX_FRAME_BYTES).await;
446            });
447        }
448        Verdict::New => {
449            tokio::spawn(answer(inner, request));
450        }
451    }
452}
453
454/// Runs the request's handler and sends its signed reply, storing it for the
455/// request's copies.
456async fn answer(inner: Arc<Inner>, request: VerifiedRequest) {
457    let handler = inner
458        .lock()
459        .served
460        .get(&(request.realm, request.procedure.clone()))
461        .and_then(|s| s.offer.handler.clone());
462    let reply = match handler {
463        None => provider_error(&inner, &request, CODE_UNKNOWN_PROCEDURE, None),
464        Some(handler) => handled(&inner, handler, &request).await,
465    };
466    let Ok(encoded) = cbor::encode(&reply) else {
467        inner.count("unencodable_reply");
468        return;
469    };
470    inner.admission.store(&request, encoded.clone());
471    let _ = inner.control.write(&encoded, MAX_FRAME_BYTES).await;
472}
473
474/// The handler's answer to `request` as a signed reply: its result, its
475/// refusal as handler_error, a panic as temporary_relay_failure, or a result
476/// the wire cannot carry as payload_too_large or unknown_error.
477async fn handled(inner: &Inner, handler: Handler, request: &VerifiedRequest) -> Value {
478    let running = tokio::spawn(handler(Request {
479        caller: request.caller,
480        realm: request.realm,
481        procedure: request.procedure.clone(),
482        payload: request.payload.clone(),
483        token: request.token.clone(),
484        proofs: request.proofs.clone(),
485        deadline_ms: request.deadline,
486    }));
487    let abort = running.abort_handle();
488    let left = (request.deadline as i64 - now_ms()).max(0) as u64;
489    let outcome = tokio::time::timeout(Duration::from_millis(left), running).await;
490    match outcome {
491        Err(_) => {
492            abort.abort();
493            provider_error(
494                inner,
495                request,
496                CODE_HANDLER_ERROR,
497                Some("the request's deadline passed"),
498            )
499        }
500        Ok(Err(_panicked)) => provider_error(inner, request, CODE_HANDLER_CRASHED, None),
501        Ok(Ok(Err(refusal))) => provider_error(
502            inner,
503            request,
504            CODE_HANDLER_ERROR,
505            Some(bounded_detail(&refusal)),
506        ),
507        Ok(Ok(Ok(payload))) => match frame::sign_result(request, &payload, None, &inner.key) {
508            Ok(signed) => signed,
509            Err(_) if cbor::encode(&payload).is_ok_and(|e| e.len() > frame::MAX_FRAME_BYTES) => {
510                provider_error(inner, request, CODE_PAYLOAD_TOO_LARGE, None)
511            }
512            Err(_) => provider_error(inner, request, CODE_UNSENDABLE, None),
513        },
514    }
515}
516
517/// This node's signed ERROR for `request`. The request verified with this
518/// node as its target, so signing cannot fail on the key; a code or detail
519/// out of bounds is a bug in this module.
520fn provider_error(
521    inner: &Inner,
522    request: &VerifiedRequest,
523    code: &str,
524    detail: Option<&str>,
525) -> Value {
526    frame::sign_provider_error(request, code, detail, None, &inner.key)
527        .unwrap_or_else(|e| panic!("station_link: a provider error that does not sign: {e}"))
528}
529
530async fn send_reply(inner: &Inner, reply: Value) {
531    let _ = inner.write_control(&reply).await;
532}
533
534/// `text` cut to 256 bytes on a character boundary.
535pub(super) fn bounded_detail(text: &str) -> &str {
536    if text.len() <= MAX_DETAIL_BYTES {
537        return text;
538    }
539    let mut cut = MAX_DETAIL_BYTES;
540    while !text.is_char_boundary(cut) {
541        cut -= 1;
542    }
543    &text[..cut]
544}
macula_rust/station_link/serve.rs

macula_rust/station_link/
serve.rs
