commonware_glue/stateful/probe/actor/
mod.rs

1use super::{
2    mailbox::{Mailbox, Message},
3    sample::Sample,
4};
5use commonware_actor::mailbox::Receiver as ActorReceiver;
6use commonware_consensus::{marshal::core::Variant, simplex::scheme::Scheme, types::Epoch};
7use commonware_cryptography::{PublicKey, certificate::Provider};
8use commonware_p2p::{Blocker, Receiver, Sender};
9use commonware_parallel::Strategy;
10use commonware_runtime::{Clock, ContextCell, Handle, Metrics, Spawner, spawn_cell};
11use commonware_utils::NonZeroDuration;
12use discovery::Discovery;
13use rand_core::CryptoRng;
14use std::num::NonZeroUsize;
15
16mod discovery;
17mod service;
18
19/// Configuration for the [`Probe`] actor.
20pub struct Config<E, D, T, P, B>
21where
22    E: Spawner + CryptoRng + Clock + Metrics,
23    D: Provider<Scope = Epoch>,
24    T: Strategy,
25    P: PublicKey,
26    B: Blocker<PublicKey = P>,
27{
28    /// The runtime context.
29    pub context: E,
30    /// Provider of epoch-specific certificate schemes for finalization verification.
31    pub provider: D,
32    /// The strategy to use for signature verification.
33    pub strategy: T,
34    /// The mailbox capacity.
35    pub capacity: NonZeroUsize,
36    /// Blocker used to block malicious peers.
37    pub blocker: B,
38    /// Finalizations below this epoch are ignored when discovering a floor. Discovery requests are
39    /// sent to this epoch's participants.
40    pub minimum_epoch: Epoch,
41    /// How long to wait for enough finalization replies before clearing the pending
42    /// responses and re-requesting.
43    pub retry_timeout: NonZeroDuration,
44}
45
46/// Discovers a sync floor by adopting the highest finalization from a peer sample.
47///
48/// The actor is a two-phase state machine. It starts in discovery, waits until a subscriber needs
49/// a floor, then solicits and samples peers' finalizations without answering any of its own. Once a
50/// marshal is attached, it hands off to service, answering peers' requests from that marshal and
51/// never issuing outbound requests. A source node that never needed a floor attaches a marshal
52/// without consuming one and enters service without soliciting peers.
53pub struct Probe<E, S, D, V, T, P, B>
54where
55    E: Spawner + CryptoRng + Clock + Metrics,
56    S: Scheme<V::Commitment, PublicKey = P>,
57    D: Provider<Scope = Epoch, Scheme = S>,
58    V: Variant,
59    T: Strategy,
60    P: PublicKey,
61    B: Blocker<PublicKey = P>,
62{
63    context: ContextCell<E>,
64    mailbox: ActorReceiver<Message<S, V>>,
65    provider: D,
66    strategy: T,
67    blocker: B,
68    minimum_epoch: Epoch,
69    retry_timeout: NonZeroDuration,
70}
71
72impl<E, S, D, V, T, P, B> Probe<E, S, D, V, T, P, B>
73where
74    E: Spawner + CryptoRng + Clock + Metrics,
75    S: Scheme<V::Commitment, PublicKey = P>,
76    D: Provider<Scope = Epoch, Scheme = S>,
77    V: Variant,
78    T: Strategy,
79    P: PublicKey,
80    B: Blocker<PublicKey = P>,
81{
82    /// Create a probe actor and mailbox.
83    pub fn new(config: Config<E, D, T, P, B>) -> (Self, Mailbox<S, V>) {
84        let (sender, receiver) =
85            commonware_actor::mailbox::new(config.context.child("mailbox"), config.capacity);
86        let mailbox = Mailbox::new(sender);
87        (
88            Self {
89                context: ContextCell::new(config.context),
90                mailbox: receiver,
91                provider: config.provider,
92                strategy: config.strategy,
93                blocker: config.blocker,
94                minimum_epoch: config.minimum_epoch,
95                retry_timeout: config.retry_timeout,
96            },
97            mailbox,
98        )
99    }
100
101    /// Start the probe actor.
102    pub fn start(
103        mut self,
104        net: (impl Sender<PublicKey = P>, impl Receiver<PublicKey = P>),
105    ) -> Handle<()> {
106        spawn_cell!(self.context, self.run(net))
107    }
108
109    async fn run(
110        self,
111        (mut sender, mut receiver): (impl Sender<PublicKey = P>, impl Receiver<PublicKey = P>),
112    ) {
113        Discovery {
114            context: self.context,
115            mailbox: self.mailbox,
116            provider: self.provider,
117            strategy: self.strategy,
118            blocker: self.blocker,
119            retry_timeout: self.retry_timeout,
120            sample: Sample::new(self.minimum_epoch),
121            floor_subscribers: Vec::new(),
122        }
123        .run(&mut sender, &mut receiver)
124        .await;
125    }
126}
commonware_glue/stateful/probe/actor/mod.rs

commonware_glue/stateful/probe/actor/
mod.rs
