use super::mailbox::{Mailbox, Message}; use commonware_actor::mailbox::Receiver as ActorReceiver; use commonware_consensus::{marshal::core::Variant, simplex::scheme::Scheme, types::Epoch}; use commonware_cryptography::{certificate::Provider, PublicKey}; use commonware_p2p::{Blocker, Receiver, Sender}; use commonware_parallel::Strategy; use commonware_runtime::{spawn_cell, Clock, ContextCell, Handle, Metrics, Spawner}; use commonware_utils::NonZeroDuration; use discovery::Discovery; use rand_core::CryptoRng; use std::num::NonZeroUsize; mod discovery; mod service; /// Configuration for the [`Probe`] actor. pub struct Config where E: Spawner + CryptoRng + Clock + Metrics, D: Provider, T: Strategy, P: PublicKey, B: Blocker, { /// The runtime context. pub context: E, /// Provider of epoch-specific certificate schemes for finalization verification. pub provider: D, /// The strategy to use for signature verification. pub strategy: T, /// The mailbox capacity. pub capacity: NonZeroUsize, /// Blocker used to block peers that send invalid finalizations. pub blocker: B, /// Finalizations below this epoch are ignored when discovering a floor. pub minimum_epoch: Epoch, /// How long to wait for enough finalization replies before clearing the pending /// responses and re-requesting. pub retry_timeout: NonZeroDuration, } /// Discovers a sync floor by adopting the highest finalization from a peer sample. /// /// The actor is a two-phase state machine. It starts in discovery, waits until a subscriber needs /// a floor, then solicits and samples peers' finalizations without answering any of its own. Once a /// marshal is attached, it hands off to service, answering peers' requests from that marshal and /// never issuing outbound requests. A source node that never needed a floor attaches a marshal /// without consuming one and enters service without soliciting peers. pub struct Probe where E: Spawner + CryptoRng + Clock + Metrics, S: Scheme, D: Provider, V: Variant, T: Strategy, P: PublicKey, B: Blocker, { context: ContextCell, mailbox: ActorReceiver>, provider: D, strategy: T, blocker: B, minimum_epoch: Epoch, retry_timeout: NonZeroDuration, } impl Probe where E: Spawner + CryptoRng + Clock + Metrics, S: Scheme, D: Provider, V: Variant, T: Strategy, P: PublicKey, B: Blocker, { /// Create a probe actor and mailbox. pub fn new(config: Config) -> (Self, Mailbox) { let (sender, receiver) = commonware_actor::mailbox::new(config.context.child("mailbox"), config.capacity); let mailbox = Mailbox::new(sender); ( Self { context: ContextCell::new(config.context), mailbox: receiver, provider: config.provider, strategy: config.strategy, blocker: config.blocker, minimum_epoch: config.minimum_epoch, retry_timeout: config.retry_timeout, }, mailbox, ) } /// Start the probe actor. pub fn start( mut self, net: (impl Sender, impl Receiver), ) -> Handle<()> { spawn_cell!(self.context, self.run(net)) } async fn run( self, (mut sender, mut receiver): (impl Sender, impl Receiver), ) { Discovery { context: self.context, mailbox: self.mailbox, provider: self.provider, strategy: self.strategy, blocker: self.blocker, minimum_epoch: self.minimum_epoch, retry_timeout: self.retry_timeout, floor: None, floor_subscribers: Vec::new(), } .run(&mut sender, &mut receiver) .await; } }