macula_rust/
station_link.rs

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

macula_rust/
station_link.rs
