commonware_p2p/simulated/
ingress.rs

1use super::{Error, Receiver, Sender};
2use crate::{authenticated::UnboundedMailbox, Address, Channel, PeerSetSubscription};
3use commonware_cryptography::PublicKey;
4use commonware_runtime::{Clock, Quota};
5use commonware_utils::{
6    channel::{fallible::FallibleExt, mpsc, oneshot, ring},
7    ordered::{Map, Set},
8};
9use rand_distr::Normal;
10use std::time::Duration;
11
12pub enum Message<P: PublicKey, E: Clock> {
13    Register {
14        channel: Channel,
15        public_key: P,
16        quota: Quota,
17        #[allow(clippy::type_complexity)]
18        result: oneshot::Sender<Result<(Sender<P, E>, Receiver<P>), Error>>,
19    },
20    Track {
21        id: u64,
22        peers: Set<P>,
23    },
24    PeerSet {
25        id: u64,
26        response: oneshot::Sender<Option<Set<P>>>,
27    },
28    Subscribe {
29        response: oneshot::Sender<PeerSetSubscription<P>>,
30    },
31    SubscribeConnected {
32        response: oneshot::Sender<ring::Receiver<Vec<P>>>,
33    },
34    LimitBandwidth {
35        public_key: P,
36        egress_cap: Option<usize>,
37        ingress_cap: Option<usize>,
38        result: oneshot::Sender<()>,
39    },
40    AddLink {
41        sender: P,
42        receiver: P,
43        sampler: Normal<f64>,
44        success_rate: f64,
45        result: oneshot::Sender<Result<(), Error>>,
46    },
47    RemoveLink {
48        sender: P,
49        receiver: P,
50        result: oneshot::Sender<Result<(), Error>>,
51    },
52    Block {
53        /// The public key of the peer sending the block request.
54        from: P,
55        /// The public key of the peer to block.
56        to: P,
57    },
58    Blocked {
59        result: oneshot::Sender<Result<Vec<(P, P)>, Error>>,
60    },
61}
62
63impl<P: PublicKey, E: Clock> std::fmt::Debug for Message<P, E> {
64    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
65        match self {
66            Self::Register { .. } => f.debug_struct("Register").finish_non_exhaustive(),
67            Self::Track { id, .. } => f
68                .debug_struct("Track")
69                .field("id", id)
70                .finish_non_exhaustive(),
71            Self::PeerSet { id, .. } => f
72                .debug_struct("PeerSet")
73                .field("id", id)
74                .finish_non_exhaustive(),
75            Self::Subscribe { .. } => f.debug_struct("Subscribe").finish_non_exhaustive(),
76            Self::SubscribeConnected { .. } => {
77                f.debug_struct("SubscribeConnected").finish_non_exhaustive()
78            }
79            Self::LimitBandwidth { .. } => f.debug_struct("LimitBandwidth").finish_non_exhaustive(),
80            Self::AddLink { .. } => f.debug_struct("AddLink").finish_non_exhaustive(),
81            Self::RemoveLink { .. } => f.debug_struct("RemoveLink").finish_non_exhaustive(),
82            Self::Block { from, to } => f
83                .debug_struct("Block")
84                .field("from", from)
85                .field("to", to)
86                .finish(),
87            Self::Blocked { .. } => f.debug_struct("Blocked").finish_non_exhaustive(),
88        }
89    }
90}
91
92/// Describes a connection between two peers.
93///
94/// Links are unidirectional (and must be set up in both directions
95/// for a bidirectional connection).
96#[derive(Clone)]
97pub struct Link {
98    /// Mean latency for the delivery of a message.
99    pub latency: Duration,
100
101    /// Standard deviation of the latency for the delivery of a message.
102    pub jitter: Duration,
103
104    /// Probability of a message being delivered successfully (in range \[0,1\]).
105    pub success_rate: f64,
106}
107
108/// Interface for modifying the simulated network.
109///
110/// At any point, peers can be added/removed and links
111/// between said peers can be modified.
112#[derive(Debug)]
113pub struct Oracle<P: PublicKey, E: Clock> {
114    sender: UnboundedMailbox<Message<P, E>>,
115}
116
117impl<P: PublicKey, E: Clock> Clone for Oracle<P, E> {
118    fn clone(&self) -> Self {
119        Self {
120            sender: self.sender.clone(),
121        }
122    }
123}
124
125impl<P: PublicKey, E: Clock> Oracle<P, E> {
126    /// Create a new instance of the oracle.
127    pub(crate) const fn new(sender: UnboundedMailbox<Message<P, E>>) -> Self {
128        Self { sender }
129    }
130
131    /// Create a new [Control] interface for some peer.
132    pub fn control(&self, me: P) -> Control<P, E> {
133        Control {
134            me,
135            sender: self.sender.clone(),
136        }
137    }
138
139    /// Create a new [Manager].
140    ///
141    /// Useful for mocking [crate::authenticated::discovery].
142    pub fn manager(&self) -> Manager<P, E> {
143        Manager {
144            oracle: self.clone(),
145        }
146    }
147
148    /// Create a new [SocketManager].
149    ///
150    /// Useful for mocking [crate::authenticated::lookup].
151    pub fn socket_manager(&self) -> SocketManager<P, E> {
152        SocketManager {
153            oracle: self.clone(),
154        }
155    }
156
157    /// Return a list of all blocked peers.
158    pub async fn blocked(&self) -> Result<Vec<(P, P)>, Error> {
159        self.sender
160            .0
161            .request(|result| Message::Blocked { result })
162            .await
163            .ok_or(Error::NetworkClosed)?
164    }
165
166    /// Set bandwidth limits for a peer.
167    ///
168    /// Bandwidth is specified for the peer's egress (upload) and ingress (download)
169    /// rates in bytes per second. Use `None` for unlimited bandwidth.
170    ///
171    /// Bandwidth can be specified before a peer is registered or linked.
172    pub async fn limit_bandwidth(
173        &self,
174        public_key: P,
175        egress_cap: Option<usize>,
176        ingress_cap: Option<usize>,
177    ) -> Result<(), Error> {
178        self.sender
179            .0
180            .request(|result| Message::LimitBandwidth {
181                public_key,
182                egress_cap,
183                ingress_cap,
184                result,
185            })
186            .await
187            .ok_or(Error::NetworkClosed)
188    }
189
190    /// Create a unidirectional link between two peers.
191    ///
192    /// Link can be called multiple times for the same sender/receiver. The latest
193    /// setting will be used.
194    ///
195    /// Link can be called before a peer is registered or bandwidth is specified.
196    pub async fn add_link(&self, sender: P, receiver: P, config: Link) -> Result<(), Error> {
197        // Sanity checks
198        if sender == receiver {
199            return Err(Error::LinkingSelf);
200        }
201        if config.success_rate < 0.0 || config.success_rate > 1.0 {
202            return Err(Error::InvalidSuccessRate(config.success_rate));
203        }
204
205        // Convert Duration to milliseconds as f64 for the Normal distribution
206        let latency_ms = config.latency.as_secs_f64() * 1000.0;
207        let jitter_ms = config.jitter.as_secs_f64() * 1000.0;
208
209        // Create distribution
210        let sampler = Normal::new(latency_ms, jitter_ms).unwrap();
211
212        self.sender
213            .0
214            .request(|result| Message::AddLink {
215                sender,
216                receiver,
217                sampler,
218                success_rate: config.success_rate,
219                result,
220            })
221            .await
222            .ok_or(Error::NetworkClosed)?
223    }
224
225    /// Remove a unidirectional link between two peers.
226    ///
227    /// If no link exists, this will return an error.
228    pub async fn remove_link(&self, sender: P, receiver: P) -> Result<(), Error> {
229        // Sanity checks
230        if sender == receiver {
231            return Err(Error::LinkingSelf);
232        }
233
234        self.sender
235            .0
236            .request(|result| Message::RemoveLink {
237                sender,
238                receiver,
239                result,
240            })
241            .await
242            .ok_or(Error::NetworkClosed)?
243    }
244
245    /// Set the peers for a given id.
246    fn track(&self, id: u64, peers: Set<P>) {
247        self.sender.0.send_lossy(Message::Track { id, peers });
248    }
249
250    /// Get the peers for a given id.
251    async fn peer_set(&self, id: u64) -> Option<Set<P>> {
252        self.sender
253            .0
254            .request(|response| Message::PeerSet { id, response })
255            .await
256            .flatten()
257    }
258
259    /// Subscribe to notifications when new peer sets are added.
260    async fn subscribe(&self) -> PeerSetSubscription<P> {
261        self.sender
262            .0
263            .request(|response| Message::Subscribe { response })
264            .await
265            .unwrap_or_else(|| {
266                let (_, rx) = mpsc::unbounded_channel();
267                rx
268            })
269    }
270}
271
272/// Implementation of [crate::Manager] for peers.
273///
274/// Useful for mocking [crate::authenticated::discovery].
275pub struct Manager<P: PublicKey, E: Clock> {
276    /// The oracle to send messages to.
277    oracle: Oracle<P, E>,
278}
279
280impl<P: PublicKey, E: Clock> std::fmt::Debug for Manager<P, E> {
281    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
282        f.debug_struct("Manager").finish_non_exhaustive()
283    }
284}
285
286impl<P: PublicKey, E: Clock> Clone for Manager<P, E> {
287    fn clone(&self) -> Self {
288        Self {
289            oracle: self.oracle.clone(),
290        }
291    }
292}
293
294impl<P: PublicKey, E: Clock> crate::Provider for Manager<P, E> {
295    type PublicKey = P;
296
297    async fn peer_set(&mut self, id: u64) -> Option<Set<Self::PublicKey>> {
298        self.oracle.peer_set(id).await
299    }
300
301    async fn subscribe(&mut self) -> PeerSetSubscription<Self::PublicKey> {
302        self.oracle.subscribe().await
303    }
304}
305
306impl<P: PublicKey, E: Clock> crate::Manager for Manager<P, E> {
307    async fn track(&mut self, id: u64, peers: Set<Self::PublicKey>) {
308        self.oracle.track(id, peers);
309    }
310}
311
312/// Implementation of [crate::AddressableManager] for peers with [Address]es.
313///
314/// Useful for mocking [crate::authenticated::lookup].
315///
316/// # Note on [Address]
317///
318/// Because addresses are never exposed in [crate::simulated],
319/// there is nothing to assert submitted data against. We thus consider
320/// all addresses to be valid.
321pub struct SocketManager<P: PublicKey, E: Clock> {
322    /// The oracle to send messages to.
323    oracle: Oracle<P, E>,
324}
325
326impl<P: PublicKey, E: Clock> std::fmt::Debug for SocketManager<P, E> {
327    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
328        f.debug_struct("SocketManager").finish_non_exhaustive()
329    }
330}
331
332impl<P: PublicKey, E: Clock> Clone for SocketManager<P, E> {
333    fn clone(&self) -> Self {
334        Self {
335            oracle: self.oracle.clone(),
336        }
337    }
338}
339
340impl<P: PublicKey, E: Clock> crate::Provider for SocketManager<P, E> {
341    type PublicKey = P;
342
343    async fn peer_set(&mut self, id: u64) -> Option<Set<Self::PublicKey>> {
344        self.oracle.peer_set(id).await
345    }
346
347    async fn subscribe(&mut self) -> PeerSetSubscription<P> {
348        self.oracle.subscribe().await
349    }
350}
351
352impl<P: PublicKey, E: Clock> crate::AddressableManager for SocketManager<P, E> {
353    async fn track(&mut self, id: u64, peers: Map<Self::PublicKey, Address>) {
354        // Ignore all addresses (simulated network doesn't use them)
355        self.oracle.track(id, peers.into_keys());
356    }
357
358    async fn overwrite(&mut self, _peers: Map<Self::PublicKey, Address>) {
359        // We consider all addresses to be valid, so this is a no-op
360    }
361}
362
363/// Individual control interface for a peer in the simulated network.
364#[derive(Debug)]
365pub struct Control<P: PublicKey, E: Clock> {
366    /// The public key of the peer this control interface is for.
367    me: P,
368
369    /// Sender for messages to the oracle.
370    sender: UnboundedMailbox<Message<P, E>>,
371}
372
373impl<P: PublicKey, E: Clock> Clone for Control<P, E> {
374    fn clone(&self) -> Self {
375        Self {
376            me: self.me.clone(),
377            sender: self.sender.clone(),
378        }
379    }
380}
381
382impl<P: PublicKey, E: Clock> Control<P, E> {
383    /// Register the communication interfaces for the peer over a given [Channel].
384    ///
385    /// # Rate Limiting
386    ///
387    /// The `quota` parameter specifies the rate limit for outbound messages to each peer.
388    /// Recipients that exceed their rate limit will be skipped when sending.
389    pub async fn register(
390        &self,
391        channel: Channel,
392        quota: Quota,
393    ) -> Result<(Sender<P, E>, Receiver<P>), Error> {
394        let public_key = self.me.clone();
395        self.sender
396            .0
397            .request(|result| Message::Register {
398                channel,
399                public_key,
400                quota,
401                result,
402            })
403            .await
404            .ok_or(Error::NetworkClosed)?
405    }
406}
407
408impl<P: PublicKey, E: Clock> crate::Blocker for Control<P, E> {
409    type PublicKey = P;
410
411    async fn block(&mut self, public_key: P) {
412        self.sender.0.send_lossy(Message::Block {
413            from: self.me.clone(),
414            to: public_key,
415        });
416    }
417}
commonware_p2p/simulated/ingress.rs

commonware_p2p/simulated/
ingress.rs
