macula_rust/pool/
call.rs

1//! Calls and streams that reach a provider at its own station, and the DHT
2//! through the pool's links.
3//!
4//! A call resolves the procedure's advertisements from the DHT, keeps those
5//! the realm's pinned key authorizes (or, in a node's own namespace, those
6//! that node signed), and tries the freshest first: it dials the serving
7//! station the advertisement names, pinned by its node_id from the station's
8//! own station_endpoint record, and calls the provider there. It moves on to
9//! the next candidate only when a station cannot be reached within that
10//! candidate's share of the deadline, before anything is sent: once the CALL
11//! or STREAM_OPEN has gone out, under the whole deadline, its outcome is
12//! returned as it is, a timeout included, so one call reaches a provider at
13//! most once (macula's call_work and failure_scope/1). A candidate that
14//! answered is remembered until its advertisement expires.
15
16use std::future::Future;
17use std::sync::Arc;
18use std::time::Duration;
19
20use tokio::time::Instant;
21
22use crate::cbor::Value;
23use crate::frame::StreamMode;
24use crate::record::{self, RecordType, Trust, Verified};
25use crate::station_link::{self, Link, LinkError, Stream, DEFAULT_CALL_TIMEOUT};
26use crate::transport::Target;
27
28use super::{Pool, PoolError, PoolInner};
29
30/// No candidate gets less than a second of a call's time.
31const MIN_CANDIDATE_SHARE: Duration = Duration::from_secs(1);
32
33/// A call to a procedure: its realm and name, the provider to call (any
34/// trusted one when zero), the payload, how long to wait
35/// ([`DEFAULT_CALL_TIMEOUT`] when zero), and a UCAN and its proofs for a
36/// gated procedure.
37#[derive(Debug, Clone, PartialEq)]
38pub struct Call {
39    pub realm: [u8; 32],
40    pub procedure: String,
41    pub provider: [u8; 32],
42    pub payload: Value,
43    pub timeout: Duration,
44    pub token: Option<Vec<u8>>,
45    pub proofs: Vec<Vec<u8>>,
46}
47
48impl Default for Call {
49    fn default() -> Self {
50        Call {
51            realm: [0; 32],
52            procedure: String::new(),
53            provider: [0; 32],
54            payload: Value::Map(Vec::new()),
55            timeout: Duration::ZERO,
56            token: None,
57            proofs: Vec::new(),
58        }
59    }
60}
61
62/// A streaming session to open: its realm and name, the provider (any
63/// trusted one when zero), the mode, the open's payload, its deadline (the
64/// link's default when zero), and a UCAN and its proofs for a gated
65/// procedure. Its default mode is server_stream.
66#[derive(Debug, Clone, PartialEq)]
67pub struct StreamCall {
68    pub realm: [u8; 32],
69    pub procedure: String,
70    pub provider: [u8; 32],
71    pub mode: StreamMode,
72    pub payload: Value,
73    pub deadline: Duration,
74    pub token: Option<Vec<u8>>,
75    pub proofs: Vec<Vec<u8>>,
76}
77
78impl Default for StreamCall {
79    fn default() -> Self {
80        StreamCall {
81            realm: [0; 32],
82            procedure: String::new(),
83            provider: [0; 32],
84            mode: StreamMode::ServerStream,
85            payload: Value::Map(Vec::new()),
86            deadline: Duration::ZERO,
87            token: None,
88            proofs: Vec::new(),
89        }
90    }
91}
92
93/// A node serving a procedure, and the station it serves from.
94#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
95pub struct Provider {
96    pub node: [u8; 32],
97    pub station: [u8; 32],
98}
99
100/// A trusted advertisement: its provider, serving station and times.
101#[derive(Debug, Clone, Copy, PartialEq, Eq)]
102pub(super) struct Candidate {
103    provider: Provider,
104    expires_at: u64,
105    created_at: u64,
106}
107
108#[derive(Debug, Clone, PartialEq, Eq, Hash)]
109pub(super) struct ResolvedKey {
110    realm: [u8; 32],
111    procedure: String,
112    provider: [u8; 32],
113}
114
115impl Pool {
116    /// Calls a procedure at a provider that serves it, as macula 12 calls
117    /// one. A provider's ERROR comes back as
118    /// `PoolError::Link(LinkError::Provider { .. })`; when no candidate
119    /// answers, [`PoolError::NoProvider`] names each one tried.
120    pub async fn call(&self, c: Call) -> Result<Value, PoolError> {
121        let inner = &self.inner;
122        let realm_key = inner.realm_key_for(&c.realm, &c.procedure)?;
123        let timeout = if c.timeout.is_zero() {
124            DEFAULT_CALL_TIMEOUT
125        } else {
126            c.timeout
127        };
128        let deadline = Instant::now() + timeout;
129        let key = ResolvedKey {
130            realm: c.realm,
131            procedure: c.procedure.clone(),
132            provider: c.provider,
133        };
134        let candidates = bounded(deadline, inner.candidates(&key, realm_key)).await?;
135        let mut tried = Vec::new();
136        let count = candidates.len();
137        for (i, cand) in candidates.into_iter().enumerate() {
138            match inner
139                .reach(&cand, candidate_share(deadline, count - i))
140                .await
141            {
142                Ok(link) => {
143                    let outcome =
144                        bounded(deadline, inner.call_at(&link, &cand, &c, deadline)).await;
145                    return inner.settled(key, cand, outcome);
146                }
147                Err(PoolError::Closed) => return Err(PoolError::Closed),
148                Err(e) => {
149                    inner.forget(&key);
150                    tried.push((cand.provider, e));
151                    if Instant::now() >= deadline {
152                        break;
153                    }
154                }
155            }
156        }
157        Err(PoolError::NoProvider(tried))
158    }
159
160    /// Every provider whose advertisement of `procedure` in `realm` the
161    /// realm's pinned key authorizes, with the station each serves from,
162    /// freshest first.
163    pub async fn providers(
164        &self,
165        realm: &[u8; 32],
166        procedure: &str,
167    ) -> Result<Vec<Provider>, PoolError> {
168        let realm_key = self.inner.realm_key_for(realm, procedure)?;
169        let key = ResolvedKey {
170            realm: *realm,
171            procedure: procedure.to_string(),
172            provider: [0; 32],
173        };
174        let found = self.inner.resolve(&key, realm_key).await?;
175        Ok(found.into_iter().map(|c| c.provider).collect())
176    }
177
178    /// Opens a streaming session at a provider of the procedure, reached as
179    /// [`Pool::call`] reaches one: the next candidate only when a station
180    /// cannot be reached, and the link's own outcome is final. The stream is
181    /// open once its STREAM_OPEN is sent; a provider's or station's refusal
182    /// arrives on its first recv.
183    pub async fn open_stream(&self, c: StreamCall) -> Result<Stream, PoolError> {
184        let inner = &self.inner;
185        let realm_key = inner.realm_key_for(&c.realm, &c.procedure)?;
186        let deadline = Instant::now() + DEFAULT_CALL_TIMEOUT;
187        let key = ResolvedKey {
188            realm: c.realm,
189            procedure: c.procedure.clone(),
190            provider: c.provider,
191        };
192        let candidates = bounded(deadline, inner.candidates(&key, realm_key)).await?;
193        let mut tried = Vec::new();
194        let count = candidates.len();
195        for (i, cand) in candidates.into_iter().enumerate() {
196            match inner
197                .reach(&cand, candidate_share(deadline, count - i))
198                .await
199            {
200                Ok(link) => {
201                    let outcome = bounded(deadline, inner.open_at(&link, &cand, &c)).await;
202                    return inner.settled(key, cand, outcome);
203                }
204                Err(PoolError::Closed) => return Err(PoolError::Closed),
205                Err(e) => {
206                    inner.forget(&key);
207                    tried.push((cand.provider, e));
208                    if Instant::now() >= deadline {
209                        break;
210                    }
211                }
212            }
213        }
214        Err(PoolError::NoProvider(tried))
215    }
216
217    /// Where `station` is dialed, from the station_endpoint record the
218    /// station signed itself; one another key signed is refused.
219    pub async fn station_target(&self, station: &[u8; 32]) -> Result<Target, PoolError> {
220        self.inner.station_target(station).await
221    }
222
223    /// A link to `station`: one the pool holds, or a direct link it dials to
224    /// the address in the station's own endpoint record, pinned by its
225    /// node_id, within the default call timeout. A direct link that does not
226    /// come up on its first dial is not kept.
227    pub async fn link_to(&self, station: &[u8; 32]) -> Result<Link, PoolError> {
228        let deadline = Instant::now() + DEFAULT_CALL_TIMEOUT;
229        self.inner.link_to(station, deadline).await
230    }
231
232    /// The record under `key`, verified, from the first link that answers.
233    pub async fn find_record(&self, key: &[u8; 32]) -> Result<Verified, PoolError> {
234        self.inner
235            .first_answer(|l| async move { l.find_record(key).await })
236            .await
237    }
238
239    /// The records under `key` that verify, and how many did not, from the
240    /// first link that answers.
241    pub async fn find_records(&self, key: &[u8; 32]) -> Result<(Vec<Verified>, usize), PoolError> {
242        self.inner
243            .first_answer(|l| async move { l.find_records(key).await })
244            .await
245    }
246
247    /// The records of type `t` that verify, and how many did not, from the
248    /// first link that answers.
249    pub async fn find_records_by_type(
250        &self,
251        t: RecordType,
252    ) -> Result<(Vec<Verified>, usize), PoolError> {
253        self.inner
254            .first_answer(|l| async move { l.find_records_by_type(t).await })
255            .await
256    }
257
258    /// Puts a signed record through the first link whose station takes it.
259    pub async fn put_record(&self, wire: &[u8]) -> Result<(), PoolError> {
260        self.inner
261            .first_answer(|l| async move { l.put_record(wire).await })
262            .await
263    }
264}
265
266impl PoolInner {
267    fn remember(&self, key: ResolvedKey, cand: Candidate) {
268        self.lock().remember.insert(key, cand);
269    }
270
271    fn forget(&self, key: &ResolvedKey) {
272        self.lock().remember.remove(key);
273    }
274
275    /// The remembered candidate while it lives and its station is linked,
276    /// else the procedure's trusted advertisements from the DHT.
277    async fn candidates(
278        &self,
279        key: &ResolvedKey,
280        realm_key: Option<Vec<u8>>,
281    ) -> Result<Vec<Candidate>, PoolError> {
282        let remembered = self.lock().remember.get(key).copied();
283        if let Some(cand) = remembered {
284            if cand.expires_at as i64 > now_ms() && self.linked_to(&cand.provider.station).is_some()
285            {
286                return Ok(vec![cand]);
287            }
288        }
289        self.resolve(key, realm_key).await
290    }
291
292    /// The procedure's advertisements the realm key authorizes, by
293    /// `key.provider` when it is set, freshest first.
294    async fn resolve(
295        &self,
296        key: &ResolvedKey,
297        realm_key: Option<Vec<u8>>,
298    ) -> Result<Vec<Candidate>, PoolError> {
299        let slot = record::procedure_key(&key.realm, &key.procedure);
300        let (found, _) = self
301            .first_answer(|l| async move { l.find_records(&slot).await })
302            .await?;
303        let now = now_ms();
304        let trust = Trust {
305            profile: self.opts.identity.profile(),
306            realm_key,
307        };
308        let mut out: Vec<Candidate> = found
309            .iter()
310            .filter(|v| v.record().record_type == RecordType::PROCEDURE_ADVERTISEMENT)
311            .filter_map(|v| {
312                let ad = record::read_procedure_advertisement(v.record()).ok()?;
313                let wanted = ad.realm_id == key.realm
314                    && ad.procedure == key.procedure
315                    && (key.provider == [0; 32] || ad.advertiser_node == key.provider);
316                if !wanted || record::verify_authorization(v, &trust, now).is_err() {
317                    return None;
318                }
319                Some(Candidate {
320                    provider: Provider {
321                        node: ad.advertiser_node,
322                        station: ad.serving_station,
323                    },
324                    expires_at: v.record().expires_at,
325                    created_at: v.record().created_at,
326                })
327            })
328            .collect();
329        if out.is_empty() {
330            return Err(PoolError::NoProvider(Vec::new()));
331        }
332        out.sort_by_key(|c| std::cmp::Reverse(c.created_at));
333        Ok(out)
334    }
335
336    /// A link to the candidate's serving station, reached by `share`.
337    /// Nothing is sent.
338    async fn reach(self: &Arc<Self>, cand: &Candidate, share: Instant) -> Result<Link, PoolError> {
339        bounded(share, self.link_to(&cand.provider.station, share)).await
340    }
341
342    /// The outcome of a call or stream sent to `cand`, final whatever it is:
343    /// a CALL that went out is never sent again elsewhere. A candidate whose
344    /// provider answered is remembered, one that did not is forgotten.
345    fn settled<T>(
346        &self,
347        key: ResolvedKey,
348        cand: Candidate,
349        outcome: Result<T, PoolError>,
350    ) -> Result<T, PoolError> {
351        match &outcome {
352            Ok(_) | Err(PoolError::Link(LinkError::Provider { .. })) => self.remember(key, cand),
353            Err(_) => self.forget(&key),
354        }
355        outcome
356    }
357
358    /// Calls the candidate's provider on `link`, under what is left before
359    /// `deadline`.
360    async fn call_at(
361        &self,
362        link: &Link,
363        cand: &Candidate,
364        c: &Call,
365        deadline: Instant,
366    ) -> Result<Value, PoolError> {
367        let left = deadline.saturating_duration_since(Instant::now());
368        Ok(link
369            .call(station_link::Call {
370                realm: c.realm,
371                procedure: c.procedure.clone(),
372                target: cand.provider.node,
373                payload: c.payload.clone(),
374                timeout: left.max(Duration::from_millis(1)),
375                token: c.token.clone(),
376                proofs: c.proofs.clone(),
377            })
378            .await?)
379    }
380
381    /// Opens the stream at the candidate's provider on `link`.
382    async fn open_at(
383        &self,
384        link: &Link,
385        cand: &Candidate,
386        c: &StreamCall,
387    ) -> Result<Stream, PoolError> {
388        Ok(link
389            .open_stream(station_link::StreamCall {
390                realm: c.realm,
391                procedure: c.procedure.clone(),
392                target: cand.provider.node,
393                mode: c.mode,
394                payload: c.payload.clone(),
395                deadline: c.deadline,
396                token: c.token.clone(),
397                proofs: c.proofs.clone(),
398            })
399            .await?)
400    }
401
402    /// The link up now to `station`, if any.
403    fn linked_to(&self, station: &[u8; 32]) -> Option<Link> {
404        self.links()
405            .into_iter()
406            .find(|l| l.station_node_id() == *station)
407    }
408
409    pub(super) async fn link_to(
410        self: &Arc<Self>,
411        station: &[u8; 32],
412        deadline: Instant,
413    ) -> Result<Link, PoolError> {
414        if let Some(link) = self.linked_to(station) {
415            return Ok(link);
416        }
417        let (existing, direct) = {
418            let state = self.lock();
419            if state.closed {
420                return Err(PoolError::Closed);
421            }
422            let existing = state
423                .members
424                .iter()
425                .find(|m| m.target.expected_node_id == *station)
426                .cloned();
427            (existing, state.members.iter().filter(|m| m.direct).count())
428        };
429        let fresh = existing.is_none();
430        let member = match existing {
431            Some(m) => m,
432            None => {
433                if direct >= self.opts.max_direct_links {
434                    return Err(PoolError::DirectLinksFull);
435                }
436                let target = bounded(deadline, self.station_target(station)).await?;
437                self.start_member(target, true)
438            }
439        };
440        if let Some(link) = member.await_up(deadline, fresh).await {
441            return Ok(link);
442        }
443        if fresh {
444            self.drop_member(&member);
445        }
446        Err(PoolError::StationNotReached {
447            station: *station,
448            cause: member.last_error(),
449        })
450    }
451
452    async fn station_target(&self, station: &[u8; 32]) -> Result<Target, PoolError> {
453        let slot = record::station_endpoint_key(station);
454        let verified = self
455            .first_answer(|l| async move { l.find_record(&slot).await })
456            .await
457            .map_err(|e| match e {
458                PoolError::Link(link) => PoolError::NoStationEndpoint(Some(link)),
459                other => other,
460            })?;
461        let r = verified.record();
462        let signer = r.signed.as_ref().map(|s| s.key_id);
463        let endpoint = record::read_station_endpoint(r)
464            .map_err(|e| PoolError::NoStationEndpoint(Some(e.into())))?;
465        match (signer, endpoint.host_advertised.first()) {
466            (Some(signer), Some(host)) if signer == *station && endpoint.quic_port != 0 => {
467                Ok(Target {
468                    host: host.clone(),
469                    port: endpoint.quic_port,
470                    profile: self.opts.identity.profile(),
471                    expected_node_id: *station,
472                })
473            }
474            _ => Err(PoolError::NoStationEndpoint(None)),
475        }
476    }
477
478    /// Runs `ask` on the links in selection order and returns the first
479    /// answer, moving on only when a link could not carry the request: a
480    /// station's own answer, not_found included, is final.
481    pub(super) async fn first_answer<'a, T, F, Fut>(&self, ask: F) -> Result<T, PoolError>
482    where
483        F: Fn(Link) -> Fut,
484        Fut: Future<Output = Result<T, LinkError>> + 'a,
485    {
486        let links = self.links();
487        if links.is_empty() {
488            return Err(PoolError::NoLink(Vec::new()));
489        }
490        let mut errors = Vec::new();
491        for link in links {
492            match ask(link).await {
493                Err(e) if unreachable(&e) => errors.push(e),
494                answered => return answered.map_err(PoolError::Link),
495            }
496        }
497        Err(PoolError::NoLink(errors))
498    }
499}
500
501/// A failure of the link to carry a request, as opposed to the station's
502/// answer: no reply in time, or the link ended.
503fn unreachable(e: &LinkError) -> bool {
504    matches!(
505        e,
506        LinkError::CallTimeout
507            | LinkError::Closed
508            | LinkError::LivenessLost
509            | LinkError::Io(_)
510            | LinkError::Goodbye(_)
511            | LinkError::StatusExpired
512            | LinkError::BindingExpired
513    )
514}
515
516/// One candidate's part of what is left before `deadline` with `left`
517/// candidates to try, at least a second, as macula shares it, and never past
518/// `deadline` itself.
519fn candidate_share(deadline: Instant, left: usize) -> Instant {
520    let now = Instant::now();
521    let remaining = deadline.saturating_duration_since(now);
522    (now + (remaining / left.max(1) as u32).max(MIN_CANDIDATE_SHARE)).min(deadline)
523}
524
525/// `work` bounded by `deadline`: past it, the call timed out.
526async fn bounded<T>(
527    deadline: Instant,
528    work: impl Future<Output = Result<T, PoolError>>,
529) -> Result<T, PoolError> {
530    tokio::time::timeout_at(deadline, work)
531        .await
532        .unwrap_or(Err(PoolError::Link(LinkError::CallTimeout)))
533}
534
535fn now_ms() -> i64 {
536    crate::uuid_v7::now_ms() as i64
537}
macula_rust/pool/call.rs

macula_rust/pool/
call.rs
