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

macula_rust/
pool.rs
