macula_rust/
pool.rs

1//! A macula 12 node's set of station links, as macula's client pool keeps
2//! them: one link to each seed station, every seed pinned by its node_id,
3//! all links of one node sharing its identity key, statement issuer, request
4//! admission, publication seq and event dedup. A link that ends is dialed
5//! again after the respawn delay and given back the node's subscriptions and
6//! served procedures.
7//!
8//! Calls reach providers directly, as macula 12 calls them: the procedure's
9//! advertisements are resolved from the DHT and checked against the realm key
10//! the pool pins for the realm, the serving station an advertisement names is
11//! dialed (pinned by its node_id, from its own station_endpoint record) and
12//! the provider called there. Station procedures (`_dht.*`) go to the pool's
13//! links.
14//!
15//! Station discovery beyond the seeds is not here: macula discovers stations
16//! through mcl-stations/list_stations, which this pool does not call.
17
18mod call;
19mod content;
20mod member;
21mod pubsub;
22mod serve;
23
24pub use crate::station_link::Confidentiality;
25pub use crate::station_link::{ConfidentialityError, ConfidentialityReason, Report, ReportError};
26pub use call::{Call, Provider, StreamCall};
27pub use content::{content_procedure_bound, ContentOptions, CONTENT_PROCEDURE};
28pub use pubsub::Subscription;
29pub use serve::{Offer, Served};
30
31use std::collections::HashMap;
32use std::fmt;
33use std::sync::{Arc, Mutex, MutexGuard};
34use std::time::Duration;
35
36use crate::node_key::{carried_key_well_formed, NodeKey, Purpose};
37use crate::seal::Keyring;
38use crate::statement_issuer::{IssuerError, StatementIssuer};
39use crate::station_link::{
40    Admission, AdmissionLimits, EventDedup, Link, LinkError, PublicationSeq,
41};
42use crate::transport::Target;
43
44use member::Member;
45
46/// macula's defaults and caps for a pool's bounds.
47pub const DEFAULT_REPLICATION_FACTOR: usize = 2;
48pub const DEFAULT_RESPAWN_DELAY: Duration = Duration::from_secs(1);
49pub const DEFAULT_MAX_SEEDS: usize = 16;
50pub const DEFAULT_MAX_DIRECT_LINKS: usize = 8;
51pub const DEFAULT_CONNECT_TIMEOUT: Duration = Duration::from_secs(30);
52const MAX_LINK_LIMIT: usize = 64;
53
54/// Why a pool, or one of its operations, failed.
55#[derive(Debug, Clone, PartialEq, Eq)]
56pub enum PoolError {
57    /// A pool given no seed station.
58    NoSeeds,
59    /// A seed without the station's node_id, as host:port: a pool dials
60    /// only stations it can check.
61    SeedNotPinned(String),
62    /// More seeds than `max_seeds`.
63    TooManySeeds { given: usize, max: usize },
64    /// A realm key that is not a well-formed key of the pool's profile, and
65    /// its realm.
66    RealmTrustInvalid([u8; 32]),
67    /// An option out of its range, or a key that is no identity key.
68    InvalidOpts(String),
69    /// No link came up within the connect timeout, or none is up to carry an
70    /// operation; each link's last error.
71    NoLink(Vec<LinkError>),
72    /// An operation on a closed pool.
73    Closed,
74    /// A realm the pool pins no key for: nothing in it is served or trusted.
75    NoRealmKey,
76    /// No provider the pinned realm key authorizes answered: each candidate
77    /// tried and why it failed; none when there was no candidate at all.
78    NoProvider(Vec<(Provider, PoolError)>),
79    /// A serving station with no endpoint record it signed itself.
80    NoStationEndpoint(Option<LinkError>),
81    /// A serving station not yet linked while `max_direct_links` direct
82    /// links are held.
83    DirectLinksFull,
84    /// A station that could not be linked, and the last dial's error.
85    StationNotReached {
86        station: [u8; 32],
87        cause: Option<LinkError>,
88    },
89    /// A procedure no link would serve, and each link's error.
90    NotServed(Vec<LinkError>),
91    /// A link's own failure, or a provider's or station's answer.
92    Link(LinkError),
93    /// Content no node announces in the realm, or a sharer that does not
94    /// hold it.
95    NotShared,
96    /// Content every announcing sharer failed to give: each sharer's node
97    /// and why.
98    ContentUnavailable(Vec<([u8; 32], PoolError)>),
99    /// A block, manifest or whole that does not match the content id it was
100    /// asked for by, and which.
101    ContentMismatch(String),
102    /// Content over the fetch's bounds, and how large.
103    ContentTooLarge(String),
104    /// An answer a sharer never gives, and what it was.
105    ContentReply(String),
106    /// A call or an open that could not be kept confidential, so nothing
107    /// was sent.
108    Confidentiality(ConfidentialityError),
109}
110
111impl fmt::Display for PoolError {
112    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
113        match self {
114            PoolError::Link(e) => write!(f, "{e}"),
115            PoolError::Confidentiality(e) => write!(f, "{e}"),
116            PoolError::NoProvider(tried) if tried.is_empty() => {
117                f.write_str("no trusted provider advertises the procedure")
118            }
119            PoolError::NoProvider(tried) => {
120                f.write_str("no trusted provider answered:")?;
121                write_providers_tried(f, tried)
122            }
123            PoolError::ContentUnavailable(tried) => {
124                f.write_str("no sharer gave the content:")?;
125                write_sharers_tried(f, tried)
126            }
127            other => write!(f, "{other:?}"),
128        }
129    }
130}
131
132impl std::error::Error for PoolError {}
133
134impl From<LinkError> for PoolError {
135    fn from(e: LinkError) -> Self {
136        PoolError::Link(e)
137    }
138}
139
140/// Each provider tried, at its station, and why it failed.
141fn write_providers_tried(
142    f: &mut fmt::Formatter<'_>,
143    tried: &[(Provider, PoolError)],
144) -> fmt::Result {
145    for (p, e) in tried {
146        write!(f, " [{} at {}: {e}]", short(&p.node), short(&p.station))?;
147    }
148    Ok(())
149}
150
151/// Each sharer tried and why it failed.
152fn write_sharers_tried(f: &mut fmt::Formatter<'_>, tried: &[([u8; 32], PoolError)]) -> fmt::Result {
153    for (node, e) in tried {
154        write!(f, " [{}: {e}]", short(node))?;
155    }
156    Ok(())
157}
158
159fn short(id: &[u8; 32]) -> String {
160    id[..4].iter().map(|b| format!("{b:02x}")).collect()
161}
162
163/// A station to link to: where it is dialed and the node_id it must prove.
164#[derive(Debug, Clone, PartialEq, Eq)]
165pub struct Seed {
166    pub host: String,
167    pub port: u16,
168    pub node_id: [u8; 32],
169}
170
171/// The order calls, publications and DHT operations try the pool's links in.
172#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
173pub enum LinkSelection {
174    /// The links in seed order.
175    #[default]
176    FirstSuccess,
177    /// A fresh random order each time.
178    Random,
179}
180
181/// A link coming up, or ending or failing to dial with its error.
182#[derive(Debug, Clone, PartialEq, Eq)]
183pub struct LinkEvent {
184    pub station: [u8; 32],
185    pub direct: bool,
186    pub up: bool,
187    pub error: Option<LinkError>,
188}
189
190/// A pool's configuration. [`Opts::new`] gives macula's defaults; a zero
191/// bound also means its default.
192#[derive(Clone)]
193pub struct Opts {
194    /// The node's identity key; its profile is the pool's.
195    pub identity: Arc<NodeKey>,
196    /// Each realm's key as carried: an advertisement in a realm is trusted
197    /// only when its authorization verifies against it, and an org
198    /// procedure is served only in a realm it names.
199    pub realm_trust: HashMap<[u8; 32], Vec<u8>>,
200    pub replication_factor: usize,
201    pub respawn_delay: Duration,
202    pub max_seeds: usize,
203    pub max_direct_links: usize,
204    /// How long [`Pool::connect`] waits for a first link.
205    pub connect_timeout: Duration,
206    /// Bounds on the requests the node's served procedures take; `None` is
207    /// macula's defaults, with the cap one share per link the pool may hold.
208    pub admission: Option<AdmissionLimits>,
209    pub link_selection: LinkSelection,
210    /// Gives the node a KEM keyring and names its current key in the
211    /// advertisements of procedures served confidentially (macula 13, E2E
212    /// design amendment A1), so callers seal to it. Off by default: switch it
213    /// on only once every station runs macula 12.11 or later and every
214    /// caller runs 13. Without it the node opens no sealed request.
215    pub kem_advertise: bool,
216    /// Hears every link coming up and going down, on a task of its own.
217    pub on_link_event: Option<Arc<dyn Fn(LinkEvent) + Send + Sync>>,
218    /// Hears each failure to reissue the node's status statements or rotate
219    /// its CONNECT key; `None` writes it to stderr. Left failing, the links
220    /// end when their statements lapse.
221    pub on_issuer_error: Option<Arc<dyn Fn(IssuerError) + Send + Sync>>,
222}
223
224impl Opts {
225    /// macula's defaults for `identity`, trusting no realm.
226    pub fn new(identity: Arc<NodeKey>) -> Opts {
227        Opts {
228            identity,
229            realm_trust: HashMap::new(),
230            replication_factor: DEFAULT_REPLICATION_FACTOR,
231            respawn_delay: DEFAULT_RESPAWN_DELAY,
232            max_seeds: DEFAULT_MAX_SEEDS,
233            max_direct_links: DEFAULT_MAX_DIRECT_LINKS,
234            connect_timeout: DEFAULT_CONNECT_TIMEOUT,
235            admission: None,
236            link_selection: LinkSelection::FirstSuccess,
237            kem_advertise: false,
238            on_link_event: None,
239            on_issuer_error: None,
240        }
241    }
242}
243
244/// A node's station links. Cloning it shares the pool; the pool closes when
245/// [`Pool::close`] is called or its last handle is dropped.
246#[derive(Clone)]
247pub struct Pool {
248    inner: Arc<PoolInner>,
249}
250
251pub(crate) struct PoolInner {
252    opts: Opts,
253    self_id: [u8; 32],
254    issuer: StatementIssuer,
255    publication_seq: Arc<PublicationSeq>,
256    admission: Arc<Admission>,
257    dedup: Arc<EventDedup>,
258    /// One node identity, one keyring, shared by every link.
259    keyring: Option<Arc<Keyring>>,
260    state: Mutex<State>,
261    ticks: tokio::task::JoinHandle<()>,
262    content: content::Sharer,
263}
264
265struct State {
266    members: Vec<Arc<Member>>,
267    subs: HashMap<u64, Arc<pubsub::SubInner>>,
268    served: HashMap<u64, Arc<serve::ServedInner>>,
269    remember: HashMap<call::ResolvedKey, call::Candidate>,
270    closed: bool,
271}
272
273/// What a closing pool ends: its links and its subscriptions.
274type Closing = (Vec<Arc<Member>>, HashMap<u64, Arc<pubsub::SubInner>>);
275
276/// One of the pool's links.
277#[derive(Debug, Clone, PartialEq, Eq)]
278pub struct LinkStatus {
279    pub station: [u8; 32],
280    pub host: String,
281    pub port: u16,
282    pub direct: bool,
283    pub up: bool,
284}
285
286impl Pool {
287    /// Checks `seeds` and `opts`, dials every seed, and returns once one link
288    /// is up, or [`PoolError::NoLink`] with each link's last error when the
289    /// connect timeout passes first. Links not yet up keep dialing.
290    pub async fn connect(seeds: Vec<Seed>, opts: Opts) -> Result<Pool, PoolError> {
291        let opts = checked(&seeds, opts)?;
292        let self_id = opts
293            .identity
294            .node_id()
295            .map_err(|e| PoolError::InvalidOpts(e.to_string()))?;
296        let issuer = StatementIssuer::with_wall_clock(opts.identity.clone())
297            .map_err(|e| PoolError::InvalidOpts(e.to_string()))?;
298        let on_error = opts.on_issuer_error.clone();
299        let ticks = issuer.spawn_ticks(move |e| match &on_error {
300            Some(f) => f(e),
301            None => eprintln!("macula-rust pool: the statement issuer failed: {e}"),
302        });
303        let admission = opts.admission.expect("checked fills the admission limits");
304        let keyring = match opts.kem_advertise {
305            true => Some(Arc::new(
306                Keyring::system(opts.identity.profile())
307                    .map_err(|e| PoolError::InvalidOpts(e.to_string()))?,
308            )),
309            false => None,
310        };
311        let inner = Arc::new(PoolInner {
312            self_id,
313            issuer,
314            publication_seq: Arc::default(),
315            admission: Arc::new(Admission::new(admission)),
316            dedup: Arc::default(),
317            keyring,
318            state: Mutex::new(State {
319                members: Vec::new(),
320                subs: HashMap::new(),
321                served: HashMap::new(),
322                remember: HashMap::new(),
323                closed: false,
324            }),
325            ticks,
326            content: content::Sharer::default(),
327            opts,
328        });
329        let pool = Pool { inner };
330        for seed in &seeds {
331            pool.inner.start_member(pool.target(seed), false);
332        }
333        let deadline = tokio::time::Instant::now() + pool.inner.opts.connect_timeout;
334        if let Err(e) = pool.inner.await_up(deadline).await {
335            pool.close().await;
336            return Err(e);
337        }
338        Ok(pool)
339    }
340
341    fn target(&self, seed: &Seed) -> Target {
342        Target {
343            host: seed.host.clone(),
344            port: seed.port,
345            profile: self.inner.opts.identity.profile(),
346            expected_node_id: seed.node_id,
347        }
348    }
349
350    /// The node_id the pool links as.
351    pub fn node_id(&self) -> [u8; 32] {
352        self.inner.self_id
353    }
354
355    /// Every link the pool holds, seeds first.
356    pub fn status(&self) -> Vec<LinkStatus> {
357        let members = self.inner.lock().members.clone();
358        members
359            .iter()
360            .map(|m| LinkStatus {
361                station: m.target.expected_node_id,
362                host: m.target.host.clone(),
363                port: m.target.port,
364                direct: m.direct,
365                up: m.current().is_some(),
366            })
367            .collect()
368    }
369
370    /// Ends every link with a GOODBYE and every subscription. It withdraws
371    /// nothing: an advertisement lapses with its link.
372    pub async fn close(&self) {
373        let Some((members, subs)) = self.inner.mark_closed() else {
374            return;
375        };
376        self.inner.ticks.abort();
377        for m in &members {
378            m.retire();
379        }
380        for m in &members {
381            m.stopped().await;
382        }
383        for sub in subs.into_values() {
384            let _ = sub.end().await;
385        }
386    }
387}
388
389impl fmt::Debug for Pool {
390    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
391        f.debug_struct("Pool")
392            .field("node_id", &short(&self.inner.self_id))
393            .field("links", &self.status())
394            .finish()
395    }
396}
397
398impl PoolInner {
399    fn lock(&self) -> MutexGuard<'_, State> {
400        self.state.lock().unwrap_or_else(|p| p.into_inner())
401    }
402
403    /// Marks the pool closed, forgets what it serves, and hands back its
404    /// links and subscriptions to end; nothing when it was closed already.
405    fn mark_closed(&self) -> Option<Closing> {
406        let mut state = self.lock();
407        if state.closed {
408            return None;
409        }
410        state.closed = true;
411        state.served.clear();
412        Some((
413            std::mem::take(&mut state.members),
414            std::mem::take(&mut state.subs),
415        ))
416    }
417
418    /// The links up now, in the pool's selection order.
419    fn links(&self) -> Vec<Link> {
420        let members = self.lock().members.clone();
421        let mut up: Vec<Link> = members.iter().filter_map(|m| m.current()).collect();
422        if self.opts.link_selection == LinkSelection::Random {
423            shuffle(&mut up);
424        }
425        up
426    }
427
428    /// Waits until a link is up, or `deadline` passes.
429    async fn await_up(self: &Arc<Self>, deadline: tokio::time::Instant) -> Result<(), PoolError> {
430        while self.links().is_empty() {
431            self.before_deadline(deadline)?;
432            tokio::time::sleep(Duration::from_millis(10)).await;
433        }
434        Ok(())
435    }
436
437    /// Nothing while `deadline` is ahead; past it, [`PoolError::NoLink`] with
438    /// each link's last error.
439    fn before_deadline(&self, deadline: tokio::time::Instant) -> Result<(), PoolError> {
440        if tokio::time::Instant::now() < deadline {
441            return Ok(());
442        }
443        let members = self.lock().members.clone();
444        Err(PoolError::NoLink(
445            members.iter().filter_map(|m| m.last_error()).collect(),
446        ))
447    }
448
449    fn event(&self, e: LinkEvent) {
450        let Some(f) = self.opts.on_link_event.clone() else {
451            return;
452        };
453        tokio::spawn(async move { f(e) });
454    }
455
456    /// The key a procedure's authorization is checked against: the realm's
457    /// pinned key, or none for a procedure in a node's own namespace, which
458    /// its advertisement's signature alone authorizes.
459    fn realm_key_for(
460        &self,
461        realm: &[u8; 32],
462        procedure: &str,
463    ) -> Result<Option<Vec<u8>>, PoolError> {
464        if crate::record::in_own_namespace(procedure) {
465            return Ok(None);
466        }
467        self.opts
468            .realm_trust
469            .get(realm)
470            .cloned()
471            .map(Some)
472            .ok_or(PoolError::NoRealmKey)
473    }
474}
475
476impl Drop for PoolInner {
477    /// A pool whose last handle is dropped stops dialing and closes its links.
478    fn drop(&mut self) {
479        self.ticks.abort();
480        let state = self.state.get_mut().unwrap_or_else(|p| p.into_inner());
481        for m in &state.members {
482            m.retire();
483        }
484    }
485}
486
487/// Refuses, before anything is dialed, what macula's pool refuses, and fills
488/// in the defaults.
489fn checked(seeds: &[Seed], mut opts: Opts) -> Result<Opts, PoolError> {
490    if opts.identity.purpose() != Purpose::Identity {
491        return Err(PoolError::InvalidOpts("an identity key is required".into()));
492    }
493    let profile = opts.identity.profile();
494    for (realm, key) in &opts.realm_trust {
495        if !carried_key_well_formed(key, profile) {
496            return Err(PoolError::RealmTrustInvalid(*realm));
497        }
498    }
499    for (name, limit) in [
500        ("max_seeds", opts.max_seeds),
501        ("max_direct_links", opts.max_direct_links),
502        ("replication_factor", opts.replication_factor),
503    ] {
504        if limit > MAX_LINK_LIMIT {
505            return Err(PoolError::InvalidOpts(format!(
506                "{name} of {limit}, outside 1 to {MAX_LINK_LIMIT}"
507            )));
508        }
509    }
510    let or_default = |v: usize, d: usize| if v == 0 { d } else { v };
511    opts.max_seeds = or_default(opts.max_seeds, DEFAULT_MAX_SEEDS);
512    opts.max_direct_links = or_default(opts.max_direct_links, DEFAULT_MAX_DIRECT_LINKS);
513    opts.replication_factor = or_default(opts.replication_factor, DEFAULT_REPLICATION_FACTOR);
514    if opts.respawn_delay.is_zero() {
515        opts.respawn_delay = DEFAULT_RESPAWN_DELAY;
516    }
517    if opts.connect_timeout.is_zero() {
518        opts.connect_timeout = DEFAULT_CONNECT_TIMEOUT;
519    }
520    let admission = opts.admission.unwrap_or_else(|| {
521        let mut limits = AdmissionLimits::default();
522        limits.cap = limits.share * (opts.max_seeds + opts.max_direct_links);
523        limits
524    });
525    admission
526        .validate()
527        .map_err(|e| PoolError::InvalidOpts(e.to_string()))?;
528    opts.admission = Some(admission);
529    if seeds.is_empty() {
530        return Err(PoolError::NoSeeds);
531    }
532    if seeds.len() > opts.max_seeds {
533        return Err(PoolError::TooManySeeds {
534            given: seeds.len(),
535            max: opts.max_seeds,
536        });
537    }
538    if let Some(unpinned) = seeds.iter().find(|s| s.node_id == [0; 32]) {
539        return Err(PoolError::SeedNotPinned(format!(
540            "{}:{}",
541            unpinned.host, unpinned.port
542        )));
543    }
544    Ok(opts)
545}
546
547/// Fisher-Yates over the system's randomness.
548fn shuffle<T>(items: &mut [T]) {
549    for i in (1..items.len()).rev() {
550        let mut r = [0u8; 8];
551        if aws_lc_rs::rand::fill(&mut r).is_err() {
552            return;
553        }
554        let j = (u64::from_le_bytes(r) % (i as u64 + 1)) as usize;
555        items.swap(i, j);
556    }
557}
macula_rust/pool.rs

macula_rust/
pool.rs
