macula_rust/station_link/
call.rs

1//! Calls on a link: a CALL signed with the link's identity key, answered by
2//! the RESULT or ERROR that verifies for it: a provider reply signed by its
3//! target, or a relay error the connected station signed. A reply that does
4//! not verify is counted and ignored, and the call keeps waiting, as macula's
5//! link does. On a v4 link the liveness probe is a call too.
6
7use std::sync::{Arc, Weak};
8use std::time::Duration;
9
10use tokio::sync::oneshot;
11
12use crate::cbor::Value;
13use crate::frame::{self, ReplyType, RequestSpec, VerifiedRequest};
14use crate::handshake::VERSION_5;
15
16use super::{now_ms, Inner, Link, LinkError};
17
18/// macula's default timeout for a call.
19pub const DEFAULT_CALL_TIMEOUT: Duration = Duration::from_secs(5);
20/// The longest a call waits: the far edge of a provider's deadline window.
21pub const MAX_CALL_TIMEOUT: Duration = Duration::from_secs(10 * 60);
22
23const LIVENESS_EVERY: Duration = Duration::from_secs(30);
24const LIVENESS_TIMEOUT: Duration = Duration::from_secs(30);
25const LIVENESS_PROCEDURE: &str = "_macula.ping";
26
27/// One request: the realm and procedure, the node it targets (the station
28/// itself when zero, as for `_dht.*`; the provider's node_id otherwise), its
29/// payload, how long to wait (the default when zero), and a UCAN token and
30/// its delegation chain's proofs for a gated procedure.
31#[derive(Debug, Clone, PartialEq)]
32pub struct Call {
33    pub realm: [u8; 32],
34    pub procedure: String,
35    pub target: [u8; 32],
36    pub payload: Value,
37    pub timeout: Duration,
38    pub token: Option<Vec<u8>>,
39    pub proofs: Vec<Vec<u8>>,
40}
41
42impl Default for Call {
43    fn default() -> Self {
44        Call {
45            realm: [0; 32],
46            procedure: String::new(),
47            target: [0; 32],
48            payload: Value::Map(Vec::new()),
49            timeout: Duration::ZERO,
50            token: None,
51            proofs: Vec::new(),
52        }
53    }
54}
55
56/// A call waiting for its reply.
57pub(super) struct Pending {
58    pub(super) request: VerifiedRequest,
59    pub(super) outcome: oneshot::Sender<Result<Value, LinkError>>,
60}
61
62impl Link {
63    /// Signs `c` as a CALL, sends it, and waits for the RESULT or ERROR that
64    /// verifies for it.
65    pub async fn call(&self, c: Call) -> Result<Value, LinkError> {
66        call(&self.inner, c).await
67    }
68}
69
70pub(super) async fn call(inner: &Arc<Inner>, c: Call) -> Result<Value, LinkError> {
71    let timeout = if c.timeout.is_zero() {
72        DEFAULT_CALL_TIMEOUT
73    } else {
74        c.timeout.min(MAX_CALL_TIMEOUT)
75    };
76    let target = if c.target == [0; 32] {
77        inner.station.node_id
78    } else {
79        c.target
80    };
81    let mut request_id = [0u8; 16];
82    aws_lc_rs::rand::fill(&mut request_id).map_err(|_| LinkError::Io("no randomness".into()))?;
83    let signed = frame::sign_call(
84        &RequestSpec {
85            request_id,
86            realm: c.realm,
87            procedure: c.procedure,
88            target,
89            deadline: (now_ms() + timeout.as_millis() as i64) as u64,
90            payload: c.payload,
91            mode: None,
92            token: c.token,
93            proofs: c.proofs,
94            source_route: None,
95            retry_budget: None,
96        },
97        &inner.key,
98    )?;
99    let request = frame::verify_request(&signed, inner.profile)?;
100    let (outcome_tx, outcome) = oneshot::channel();
101    {
102        let mut state = inner.lock();
103        if let Some(e) = &state.ended {
104            return Err(e.clone());
105        }
106        state.pending.insert(
107            request_id,
108            Pending {
109                request,
110                outcome: outcome_tx,
111            },
112        );
113    }
114    let forget = || {
115        inner.lock().pending.remove(&request_id);
116    };
117    if let Err(e) = inner.write_control(&signed).await {
118        forget();
119        return Err(e);
120    }
121    let answered = tokio::time::timeout(timeout, outcome).await;
122    forget();
123    match answered {
124        Ok(Ok(outcome)) => outcome,
125        Ok(Err(_)) => Err(inner.lock().ended.clone().unwrap_or(LinkError::Closed)),
126        Err(_) => Err(LinkError::CallTimeout),
127    }
128}
129
130/// A RESULT or ERROR matched to its pending call by the ids it claims, and
131/// handed on only once it verifies for that call's request.
132pub(super) fn replied(inner: &Arc<Inner>, v: &Value) {
133    let Ok((request_id, _)) = frame::claimed_reply_ids(v) else {
134        inner.count("malformed_reply");
135        return;
136    };
137    let request = match inner.lock().pending.get(&request_id) {
138        Some(p) => p.request.clone(),
139        None => {
140            inner.count("unmatched_reply");
141            return;
142        }
143    };
144    let Some(outcome) = verified_outcome(inner, v, &request) else {
145        inner.count("unverified_reply");
146        return;
147    };
148    if let Some(p) = inner.lock().pending.remove(&request_id) {
149        let _ = p.outcome.send(outcome);
150    }
151}
152
153fn verified_outcome(
154    inner: &Inner,
155    v: &Value,
156    request: &VerifiedRequest,
157) -> Option<Result<Value, LinkError>> {
158    if v.get("reply").is_some() {
159        let reply = frame::verify_reply(v, request, inner.profile).ok()?;
160        return Some(match reply.frame_type {
161            ReplyType::Result => Ok(reply.payload.unwrap_or(Value::Null)),
162            ReplyType::Error => Err(LinkError::Provider {
163                responded_by: reply.responded_by,
164                code: reply.code.unwrap_or_default(),
165                detail: reply.detail,
166            }),
167        });
168    }
169    let relayed =
170        frame::verify_relay_error(v, request, inner.profile, &inner.station.node_id).ok()?;
171    Some(Err(LinkError::Relay {
172        reported_by: relayed.reported_by,
173        code: relayed.code,
174    }))
175}
176
177/// A liveness probe every 30 seconds; two misses in a row end the link. On v5
178/// the probe is a liveness_ping, answered by the station's connection with a
179/// liveness_pong of its nonce, and nothing is signed. On v4 it is a
180/// `_macula.ping` call: any verified answer counts, a provider's or a relay
181/// error from the station alike, as it proves the station is there.
182pub(super) async fn probe(link: Weak<Inner>) {
183    let mut misses = 0;
184    loop {
185        let Some(mut done) = link.upgrade().map(|l| l.done_rx.clone()) else {
186            return;
187        };
188        tokio::select! {
189            _ = done.wait_for(|ended| *ended) => return,
190            _ = tokio::time::sleep(LIVENESS_EVERY) => {}
191        }
192        let Some(inner) = link.upgrade() else { return };
193        let outcome = match inner.version {
194            VERSION_5 => probe_v5(&inner).await,
195            _ => call(
196                &inner,
197                Call {
198                    procedure: LIVENESS_PROCEDURE.to_string(),
199                    timeout: LIVENESS_TIMEOUT,
200                    ..Call::default()
201                },
202            )
203            .await
204            .map(|_| ()),
205        };
206        misses = if matches!(outcome, Err(LinkError::CallTimeout)) {
207            misses + 1
208        } else {
209            0
210        };
211        if misses >= 2 {
212            inner.end(LinkError::LivenessLost);
213            return;
214        }
215    }
216}
217
218/// One v5 liveness probe: answered when a liveness_pong of its nonce arrives
219/// within [`LIVENESS_TIMEOUT`], [`LinkError::CallTimeout`] when none does.
220async fn probe_v5(inner: &Inner) -> Result<(), LinkError> {
221    let mut nonce = [0u8; frame::LIVENESS_NONCE_SIZE];
222    aws_lc_rs::rand::fill(&mut nonce).map_err(|_| LinkError::Io("no randomness".into()))?;
223    let mut pongs = inner.pongs_rx.lock().await;
224    // A pong that arrived after its probe's deadline must not take the place
225    // of this probe's.
226    while pongs.try_recv().is_ok() {}
227    inner
228        .send_control(&frame::liveness_ping_frame(&nonce))
229        .await?;
230    let mut done = inner.done_rx.clone();
231    let answered = async {
232        while let Some(pong) = pongs.recv().await {
233            if pong == nonce {
234                return Ok(());
235            }
236        }
237        Err(LinkError::Closed)
238    };
239    tokio::select! {
240        _ = done.wait_for(|ended| *ended) => Err(LinkError::Closed),
241        outcome = tokio::time::timeout(LIVENESS_TIMEOUT, answered) => {
242            outcome.unwrap_or(Err(LinkError::CallTimeout))
243        }
244    }
245}
macula_rust/station_link/call.rs

macula_rust/station_link/
call.rs
