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

macula_rust/
pool.rs
