macula_rust/
station_link.rs

1//! A client's link to one macula 12 station, as macula_station_link and
2//! macula-go's stationlink are: a QUIC connection dialed to the station its
3//! target pins, one bidirectional control stream, the v4 handshake on it, then
4//! status statements both ways and every frame of the session.
5//!
6//! After HELLO the link sends its own status statement at every reissue of
7//! the node's statement issuer, and ends when the station's statement is five
8//! minutes past its expiry or the station's TLS binding reaches its
9//! not_after. In pq_hybrid every control frame is neighbour-signed with a
10//! sequence number per direction, from 0 after HELLO; a frame out of sequence
11//! ends the link. A liveness probe, a `_macula.ping` call every 30 seconds,
12//! ends the link after two misses in a row.
13
14mod admission;
15mod call;
16mod dht;
17mod framing;
18mod pubsub;
19mod serve;
20mod stream;
21
22pub use admission::{Admission, AdmissionLimits};
23pub use call::{Call, DEFAULT_CALL_TIMEOUT, MAX_CALL_TIMEOUT};
24pub use pubsub::{Event, EventDedup, Publication, PublicationSeq, SignedPublication, Subscription};
25pub use serve::{handler, BoxFuture, Handler, Offer, Request, Served, StreamOffer};
26pub use stream::{
27    stream_handler, Stream, StreamCall, StreamEvent, StreamHandler, DEFAULT_STREAM_DEADLINE,
28};
29
30use std::collections::HashMap;
31use std::fmt;
32use std::sync::atomic::{AtomicI64, Ordering};
33use std::sync::{Arc, Mutex, MutexGuard};
34use std::time::Duration;
35
36use sha2::{Digest, Sha384};
37use tokio::sync::watch;
38
39use crate::cbor::{self, Value};
40use crate::frame::{self, FrameError, NeighbourLink, NeighbourPeer};
41use crate::handshake::{self, ClientSession, HandshakeError, Peer, Station};
42use crate::node_key::NodeKey;
43use crate::profile::Profile;
44use crate::record::RecordError;
45use crate::statement_issuer::{IssuerError, StatementIssuer, StatementSubscription};
46use crate::transport::{self, DialError, Target};
47
48use framing::{read_frame, FrameWriter, HANDSHAKE_FRAME_BYTES, MAX_FRAME_BYTES};
49
50/// How long the handshake may take, as macula's 30 seconds.
51pub const HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(30);
52
53/// How long Close waits, after its GOODBYE, for the station to close the
54/// connection before closing it itself.
55const CLOSE_LINGER: Duration = Duration::from_secs(1);
56
57/// How long past a statement's expiry the link keeps a station whose next
58/// statement has not arrived: macula's five minutes.
59const STATUS_GRACE_MS: i64 = 5 * 60 * 1000;
60
61/// Why a link or one of its operations failed.
62#[derive(Debug, Clone, PartialEq, Eq)]
63pub enum LinkError {
64    /// The configuration cannot be dialed: a key of another profile than the
65    /// target's, or admission limits macula would not start with.
66    InvalidConfig(String),
67    /// The dial failed.
68    Dial(String),
69    /// The handshake refused, or was refused.
70    Handshake(HandshakeError),
71    /// The handshake did not finish within [`HANDSHAKE_TIMEOUT`].
72    HandshakeTimeout,
73    /// The issuer had no CONNECT material in force.
74    Issuer(IssuerError),
75    /// The QUIC connection or a stream failed.
76    Io(String),
77    /// A frame longer than the cap in force.
78    FrameTooLarge(usize),
79    /// A frame that could not be built, or one received that does not
80    /// verify.
81    Frame(FrameError),
82    /// A record that could not be built, or one found that does not verify.
83    Record(RecordError),
84    /// The station's status statement lapsed past the grace.
85    StatusExpired,
86    /// The station's TLS binding reached its not_after.
87    BindingExpired,
88    /// The link was closed by its owner.
89    Closed,
90    /// The station ended the link with a GOODBYE, for this reason.
91    Goodbye(String),
92    /// The station missed two liveness probes in a row.
93    LivenessLost,
94    /// No verified reply answered the call within its timeout.
95    CallTimeout,
96    /// A provider's ERROR for the call.
97    Provider {
98        responded_by: [u8; 32],
99        code: String,
100        detail: Option<String>,
101    },
102    /// A relay error the connected station reported for the call.
103    Relay { reported_by: [u8; 32], code: String },
104    /// A find_record the station answered not_found.
105    RecordNotFound,
106    /// A station reply to a DHT call in a shape that call never answers with.
107    UnexpectedReply(String),
108    /// An offer without exactly one handler, or an org procedure's without
109    /// the realm key.
110    InvalidOffer,
111    /// A procedure without a namespace, which nobody can authorize.
112    NoOrg,
113    /// A procedure already served on this link.
114    AlreadyServed,
115    /// A served procedure withdrawn by its owner.
116    Stopped,
117    /// A stream ended by a STREAM_ERROR: the peer's, the station's relay
118    /// error (`relay`), or this side's own abort.
119    Stream {
120        code: String,
121        message: String,
122        relay: bool,
123    },
124    /// A stream that ended normally.
125    EndOfStream,
126    /// A send on a stream this side has ended.
127    StreamClosed,
128    /// A STREAM_OPEN over the 1 MiB a peer reads of one.
129    StreamOpenTooLarge(usize),
130}
131
132impl fmt::Display for LinkError {
133    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
134        match self {
135            LinkError::Provider {
136                code,
137                detail: Some(d),
138                ..
139            } => write!(f, "the provider answered {code}: {d}"),
140            LinkError::Provider { code, .. } => write!(f, "the provider answered {code}"),
141            LinkError::Relay { code, .. } => {
142                write!(f, "the station could not relay the call: {code}")
143            }
144            LinkError::Stream { code, message, .. } if !message.is_empty() => {
145                write!(f, "stream error {code}: {message}")
146            }
147            LinkError::Stream { code, .. } => write!(f, "stream error {code}"),
148            LinkError::Handshake(e) => write!(f, "handshake: {e}"),
149            LinkError::Frame(e) => write!(f, "frame: {e}"),
150            LinkError::Record(e) => write!(f, "record: {e}"),
151            LinkError::Issuer(e) => write!(f, "{e}"),
152            LinkError::Goodbye(reason) => write!(f, "the station said goodbye: {reason}"),
153            other => write!(f, "{other:?}"),
154        }
155    }
156}
157
158impl std::error::Error for LinkError {}
159
160impl From<FrameError> for LinkError {
161    fn from(e: FrameError) -> Self {
162        LinkError::Frame(e)
163    }
164}
165
166impl From<RecordError> for LinkError {
167    fn from(e: RecordError) -> Self {
168        LinkError::Record(e)
169    }
170}
171
172impl From<HandshakeError> for LinkError {
173    fn from(e: HandshakeError) -> Self {
174        LinkError::Handshake(e)
175    }
176}
177
178impl From<DialError> for LinkError {
179    fn from(e: DialError) -> Self {
180        LinkError::Dial(e.to_string())
181    }
182}
183
184/// What a link is dialed with: the station to reach, the node's identity
185/// key, the statement issuer that holds its CONNECT key and statements, and
186/// the realm membership endorsement to present, empty for none. Links of one
187/// node share their publication seq, admission and dedup; a link given none
188/// makes its own.
189pub struct Config {
190    pub target: Target,
191    pub identity: Arc<NodeKey>,
192    pub issuer: StatementIssuer,
193    pub member_endorsement: Vec<u8>,
194    pub publication_seq: Option<Arc<PublicationSeq>>,
195    pub admission: Option<Arc<Admission>>,
196    pub dedup: Option<Arc<EventDedup>>,
197    /// This link's place in the admission: the station it dialed, host:port,
198    /// when `None`.
199    pub share: Option<String>,
200}
201
202impl Config {
203    /// A link's configuration with none of the shared parts.
204    pub fn new(target: Target, identity: Arc<NodeKey>, issuer: StatementIssuer) -> Config {
205        Config {
206            target,
207            identity,
208            issuer,
209            member_endorsement: Vec::new(),
210            publication_seq: None,
211            admission: None,
212            dedup: None,
213            share: None,
214        }
215    }
216}
217
218/// A handshaked link to one station. Cloning it shares the link.
219#[derive(Clone)]
220pub struct Link {
221    inner: Arc<Inner>,
222}
223
224struct Inner {
225    /// Distinct for every link a process dials.
226    serial: u64,
227    connection: quinn::Connection,
228    _endpoint: quinn::Endpoint,
229    control: FrameWriter,
230    profile: Profile,
231    key: Arc<NodeKey>,
232    self_id: [u8; 32],
233    station: Station,
234    station_capabilities: u64,
235    connection_hash: [u8; 48],
236    /// Orders a neighbour signature's seq with its write.
237    send_seq: tokio::sync::Mutex<u64>,
238    status_deadline: AtomicI64,
239    publication_seq: Arc<PublicationSeq>,
240    admission: Arc<Admission>,
241    dedup: Arc<EventDedup>,
242    share: String,
243    state: Mutex<State>,
244    done_tx: watch::Sender<bool>,
245    done_rx: watch::Receiver<bool>,
246}
247
248struct State {
249    ended: Option<LinkError>,
250    /// The owner is closing the link: however it then ends, it ends
251    /// [`LinkError::Closed`].
252    closing: bool,
253    unrouted: HashMap<String, u64>,
254    pending: HashMap<[u8; 16], call::Pending>,
255    subs: HashMap<([u8; 32], String), Vec<pubsub::SubscriberSlot>>,
256    served: HashMap<([u8; 32], String), serve::ServedEntry>,
257    streams: Vec<std::sync::Weak<stream::StreamInner>>,
258}
259
260impl Link {
261    /// Dials `cfg.target`, runs the v4 handshake as a client, and returns the
262    /// link once the station's HELLO accepts it, within
263    /// [`HANDSHAKE_TIMEOUT`].
264    pub async fn dial(cfg: Config) -> Result<Link, LinkError> {
265        if cfg.identity.profile() != cfg.target.profile {
266            return Err(LinkError::InvalidConfig(
267                "the identity key is of another profile than the target's".into(),
268            ));
269        }
270        if let Some(admission) = &cfg.admission {
271            admission.limits().validate()?;
272        }
273        tokio::time::timeout(HANDSHAKE_TIMEOUT, handshaken(cfg))
274            .await
275            .map_err(|_| LinkError::HandshakeTimeout)
276            .and_then(|linked| linked)
277    }
278
279    /// The node_id of the station the link reached.
280    pub fn station_node_id(&self) -> [u8; 32] {
281        self.inner.station.node_id
282    }
283
284    /// A number distinct for every link this process dials.
285    pub fn serial(&self) -> u64 {
286        self.inner.serial
287    }
288
289    /// The node_id this link connected as.
290    pub fn node_id(&self) -> [u8; 32] {
291        self.inner.self_id
292    }
293
294    /// The capability bits the station's HELLO announced.
295    pub fn station_capabilities(&self) -> u64 {
296        self.inner.station_capabilities
297    }
298
299    /// The profile the link runs.
300    pub fn profile(&self) -> Profile {
301        self.inner.profile
302    }
303
304    /// Why the link ended, or `None` while it runs.
305    pub fn error(&self) -> Option<LinkError> {
306        self.inner.lock().ended.clone()
307    }
308
309    /// Waits until the link has ended, and says why.
310    pub async fn done(&self) -> LinkError {
311        let mut done = self.inner.done_rx.clone();
312        let _ = done.wait_for(|ended| *ended).await;
313        self.error().unwrap_or(LinkError::Closed)
314    }
315
316    /// The frames received that nothing on this link handles, by what they
317    /// were.
318    pub fn unrouted(&self) -> HashMap<String, u64> {
319        self.inner.lock().unrouted.clone()
320    }
321
322    /// Sends a GOODBYE with `reason`, closes the control stream's sending
323    /// side, waits up to a second for the station to close the connection,
324    /// and ends the link. Closing an ended link does nothing.
325    pub async fn close(&self, reason: &str) -> Result<(), LinkError> {
326        {
327            let mut state = self.inner.lock();
328            if state.ended.is_some() {
329                return Ok(());
330            }
331            state.closing = true;
332        }
333        let goodbye = frame::goodbye_frame(reason, None)?;
334        let sent = self.inner.send_control(&goodbye).await;
335        if sent.is_ok() {
336            self.inner.control.finish().await;
337            let _ = tokio::time::timeout(CLOSE_LINGER, self.inner.connection.closed()).await;
338        }
339        self.inner.end(LinkError::Closed);
340        sent
341    }
342}
343
344async fn handshaken(cfg: Config) -> Result<Link, LinkError> {
345    let dialed = transport::dial_target(&cfg.target).await?;
346    let (send, mut recv) = dialed
347        .connection
348        .open_bi()
349        .await
350        .map_err(|e| LinkError::Io(format!("open the control stream: {e}")))?;
351    let control = FrameWriter::new(send);
352    control
353        .write(&handshake::opener(), HANDSHAKE_FRAME_BYTES)
354        .await?;
355    let challenge = read_frame(&mut recv, HANDSHAKE_FRAME_BYTES).await?;
356    let material = cfg.issuer.connect_material().map_err(LinkError::Issuer)?;
357    let (connect, station) = handshake::answer_challenge(
358        &challenge,
359        &ClientSession {
360            profile: cfg.target.profile,
361            expected_node_id: cfg.target.expected_node_id,
362            leaf: &dialed.leaf,
363            identity_key: cfg.identity.public_key(),
364            connect_key: &material.key,
365            connect_binding: &material.binding,
366            connect_status: &material.status,
367            capabilities: 0,
368            now_ms: now_ms(),
369            member_endorsement: cfg.member_endorsement.clone(),
370        },
371    )?;
372    control.write(&connect, HANDSHAKE_FRAME_BYTES).await?;
373    let hello = read_frame(&mut recv, HANDSHAKE_FRAME_BYTES).await?;
374    let capabilities = handshake::read_hello(&hello)?;
375    let self_id = cfg
376        .identity
377        .node_id()
378        .map_err(|e| LinkError::InvalidConfig(e.to_string()))?;
379    let statements = cfg
380        .issuer
381        .subscribe(&material.binding)
382        .map_err(LinkError::Issuer)?;
383    let (done_tx, done_rx) = watch::channel(false);
384    let share = cfg
385        .share
386        .unwrap_or_else(|| format!("{}:{}", cfg.target.host, cfg.target.port));
387    static SERIALS: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(1);
388    let inner = Arc::new(Inner {
389        serial: SERIALS.fetch_add(1, Ordering::Relaxed),
390        connection: dialed.connection,
391        _endpoint: dialed.endpoint,
392        control,
393        profile: cfg.target.profile,
394        key: cfg.identity,
395        self_id,
396        status_deadline: AtomicI64::new(station.status_expires_at + STATUS_GRACE_MS),
397        station_capabilities: capabilities,
398        connection_hash: Sha384::digest(&challenge).into(),
399        station,
400        send_seq: tokio::sync::Mutex::new(0),
401        publication_seq: cfg.publication_seq.unwrap_or_default(),
402        admission: cfg
403            .admission
404            .unwrap_or_else(|| Arc::new(Admission::new(AdmissionLimits::default()))),
405        dedup: cfg.dedup.unwrap_or_default(),
406        share,
407        state: Mutex::new(State {
408            ended: None,
409            closing: false,
410            unrouted: HashMap::new(),
411            pending: HashMap::new(),
412            subs: HashMap::new(),
413            served: HashMap::new(),
414            streams: Vec::new(),
415        }),
416        done_tx,
417        done_rx,
418    });
419    tokio::spawn(send_statements(Arc::downgrade(&inner), statements));
420    tokio::spawn(read_control(inner.clone(), recv));
421    tokio::spawn(stream::accept_streams(Arc::downgrade(&inner)));
422    tokio::spawn(call::probe(Arc::downgrade(&inner)));
423    tokio::spawn(watch_expiries(Arc::downgrade(&inner)));
424    tokio::spawn(watch_connection(Arc::downgrade(&inner)));
425    Ok(Link { inner })
426}
427
428impl Inner {
429    fn lock(&self) -> MutexGuard<'_, State> {
430        self.state
431            .lock()
432            .unwrap_or_else(|poisoned| poisoned.into_inner())
433    }
434
435    fn count(&self, what: &str) {
436        *self.lock().unrouted.entry(what.to_string()).or_default() += 1;
437    }
438
439    /// A version-2 frame on the control stream, neighbour-signed with the
440    /// next seq when the profile signs its type.
441    async fn send_control(&self, v: &Value) -> Result<(), LinkError> {
442        let mut seq = self.send_seq.lock().await;
443        let signed = frame::sign_neighbour(
444            v,
445            &self.key,
446            &NeighbourLink {
447                connection: self.connection_hash,
448                seq: *seq,
449            },
450        )?;
451        if frame::neighbour_signed(self.profile, &frame_type_of(v)) {
452            *seq += 1;
453        }
454        self.write_control(&signed).await
455    }
456
457    /// A frame on the control stream as it is: a data frame, which carries
458    /// its own end-to-end signature and no neighbour signature.
459    async fn write_control(&self, v: &Value) -> Result<(), LinkError> {
460        let encoded =
461            cbor::encode(v).map_err(|e| LinkError::Frame(FrameError::Payload(e.to_string())))?;
462        self.control.write(&encoded, MAX_FRAME_BYTES).await
463    }
464
465    /// Ends the link once, with `err`: pending calls, subscriptions, served
466    /// procedures and streams end with it, and the connection closes.
467    fn end(&self, err: LinkError) {
468        let (err, pending, subs, served, streams) = {
469            let mut state = self.lock();
470            if state.ended.is_some() {
471                return;
472            }
473            let err = if state.closing {
474                LinkError::Closed
475            } else {
476                err
477            };
478            state.ended = Some(err.clone());
479            (
480                err,
481                std::mem::take(&mut state.pending),
482                std::mem::take(&mut state.subs),
483                std::mem::take(&mut state.served),
484                std::mem::take(&mut state.streams),
485            )
486        };
487        for (_, p) in pending {
488            let _ = p.outcome.send(Err(err.clone()));
489        }
490        drop(subs);
491        for s in served.into_values() {
492            s.end(err.clone());
493        }
494        for s in streams.into_iter().filter_map(|w| w.upgrade()) {
495            stream::StreamInner::end(&s, Some(err.clone()));
496        }
497        self.connection.close(0u32.into(), b"link ended");
498        let _ = self.done_tx.send_replace(true);
499    }
500}
501
502/// Sends each statement the issuer reissues as a STATUS, until the link
503/// ends. A STATUS carries no neighbour signature and takes no seq.
504async fn send_statements(link: std::sync::Weak<Inner>, mut statements: StatementSubscription) {
505    loop {
506        let Some(done) = link.upgrade().map(|l| l.done_rx.clone()) else {
507            return;
508        };
509        let mut done = done;
510        let statement = tokio::select! {
511            _ = done.wait_for(|ended| *ended) => return,
512            statement = statements.recv() => statement,
513        };
514        let (Some(statement), Some(inner)) = (statement, link.upgrade()) else {
515            return;
516        };
517        if let Err(e) = inner
518            .control
519            .write(&handshake::status_frame(&statement), MAX_FRAME_BYTES)
520            .await
521        {
522            inner.end(e);
523            return;
524        }
525    }
526}
527
528/// Reads the station's frames until the link ends.
529async fn read_control(inner: Arc<Inner>, mut recv: quinn::RecvStream) {
530    let mut recv_seq = 0u64;
531    let mut done = inner.done_rx.clone();
532    loop {
533        let payload = tokio::select! {
534            _ = done.wait_for(|ended| *ended) => return,
535            payload = read_frame(&mut recv, MAX_FRAME_BYTES) => payload,
536        };
537        let outcome = match payload {
538            Ok(payload) => received(&inner, &payload, &mut recv_seq),
539            Err(e) => Err(e),
540        };
541        if let Err(e) = outcome {
542            inner.end(e);
543            return;
544        }
545    }
546}
547
548/// One frame from the station: a STATUS renews the station's statement,
549/// every other frame is opened from its neighbour signature at the next seq,
550/// and a GOODBYE ends the link.
551fn received(inner: &Arc<Inner>, payload: &[u8], recv_seq: &mut u64) -> Result<(), LinkError> {
552    let v = cbor::decode(payload).map_err(|_| LinkError::Frame(FrameError::Malformed))?;
553    let frame_type = frame_type_of(&v);
554    if frame_type == "status" {
555        let expires_at = handshake::read_status(
556            payload,
557            &Peer {
558                profile: inner.profile,
559                identity_key: inner.station.identity_key.clone(),
560                binding: inner.station.tls_binding.clone(),
561                now_ms: now_ms(),
562            },
563        )?;
564        inner
565            .status_deadline
566            .store(expires_at + STATUS_GRACE_MS, Ordering::SeqCst);
567        return Ok(());
568    }
569    let opened = frame::verify_neighbour(
570        &v,
571        &NeighbourPeer {
572            profile: inner.profile,
573            peer_key: inner.station.identity_key.clone(),
574            connection: inner.connection_hash,
575            seq: *recv_seq,
576        },
577    )?;
578    if frame::neighbour_signed(inner.profile, &frame_type) {
579        *recv_seq += 1;
580    }
581    match frame_type.as_str() {
582        "event" => pubsub::evented(inner, &opened),
583        "result" | "error" => call::replied(inner, &opened),
584        "call" => serve::called(inner, &opened),
585        "goodbye" => {
586            let reason = match opened.get("reason") {
587                Some(Value::Text(r)) => r.clone(),
588                _ => String::new(),
589            };
590            return Err(LinkError::Goodbye(reason));
591        }
592        _ => inner.count(&frame_type),
593    }
594    Ok(())
595}
596
597/// Ends the link when the station's statement lapses past the grace, or its
598/// TLS binding reaches its not_after.
599async fn watch_expiries(link: std::sync::Weak<Inner>) {
600    loop {
601        let Some(inner) = link.upgrade() else { return };
602        let now = now_ms();
603        if now >= inner.station.binding_not_after {
604            inner.end(LinkError::BindingExpired);
605            return;
606        }
607        let status_deadline = inner.status_deadline.load(Ordering::SeqCst);
608        if now >= status_deadline {
609            inner.end(LinkError::StatusExpired);
610            return;
611        }
612        let wait = (status_deadline.min(inner.station.binding_not_after) - now).clamp(1, 60_000);
613        let mut done = inner.done_rx.clone();
614        drop(inner);
615        tokio::select! {
616            _ = done.wait_for(|ended| *ended) => return,
617            _ = tokio::time::sleep(Duration::from_millis(wait as u64)) => {}
618        }
619    }
620}
621
622/// Ends the link when its QUIC connection closes.
623async fn watch_connection(link: std::sync::Weak<Inner>) {
624    let Some(connection) = link.upgrade().map(|l| l.connection.clone()) else {
625        return;
626    };
627    let cause = connection.closed().await;
628    if let Some(inner) = link.upgrade() {
629        inner.end(LinkError::Io(cause.to_string()));
630    }
631}
632
633fn frame_type_of(v: &Value) -> String {
634    match v.get("frame_type") {
635        Some(Value::Text(t)) => t.clone(),
636        _ => String::new(),
637    }
638}
639
640fn now_ms() -> i64 {
641    crate::uuid_v7::now_ms() as i64
642}
macula_rust/station_link.rs

macula_rust/
station_link.rs
