commonware_p2p/lib.rs
1//! Communicate with authenticated peers over encrypted connections.
2//!
3//! # Status
4//!
5//! Stability varies by primitive. See [README](https://github.com/commonwarexyz/monorepo#stability) for details.
6
7#![doc(
8 html_logo_url = "https://commonware.xyz/imgs/rustdoc_logo.svg",
9 html_favicon_url = "https://commonware.xyz/favicon.ico"
10)]
11
12use commonware_macros::{stability_mod, stability_scope};
13
14stability_mod!(ALPHA, pub mod simulated);
15
16stability_scope!(BETA {
17 use commonware_actor::{Feedback, Unreliable};
18 use commonware_cryptography::PublicKey;
19 use commonware_runtime::{IoBuf, IoBufs};
20 use commonware_utils::{
21 channel::{mpsc, ring},
22 ordered::{Map, Set},
23 };
24 use std::{error::Error as StdError, fmt::Debug, future::Future, time::SystemTime};
25
26 mod sizing;
27 pub mod authenticated;
28 pub mod types;
29 pub mod utils;
30
31 pub use types::{Address, Ingress};
32
33 /// Tuple representing a message received from a given public key.
34 ///
35 /// This message is guaranteed to adhere to the configuration of the channel and
36 /// will already be decrypted and authenticated.
37 pub type Message<P> = (P, IoBuf);
38
39 /// Alias for identifying communication channels.
40 pub type Channel = u64;
41
42 /// Enum indicating the set of recipients to send a message to.
43 #[derive(Clone, Debug)]
44 pub enum Recipients<P: PublicKey> {
45 All,
46 Some(Vec<P>),
47 One(P),
48 }
49
50 /// Interface for sending messages to a set of recipients without rate-limiting restrictions.
51 pub trait UnlimitedSender: Clone + Send + Sync + 'static {
52 /// Public key type used to identify recipients.
53 type PublicKey: PublicKey;
54
55 /// Sends a message to a set of recipients.
56 ///
57 /// # Offline Recipients
58 ///
59 /// If a recipient is offline at the time a message is sent, the message
60 /// will be dropped. It is up to the application to handle retries (if
61 /// necessary).
62 ///
63 /// # Panics
64 ///
65 /// Panics if `message` exceeds the sender's configured maximum application payload size.
66 ///
67 /// # Returns
68 ///
69 /// Feedback from submitting the message for delivery.
70 /// [`Unreliable`] indicates that local submission may be rejected under backpressure.
71 /// [`Feedback::accepted`] does not guarantee that the recipient will receive the message.
72 fn send(
73 &mut self,
74 recipients: Recipients<Self::PublicKey>,
75 message: impl Into<IoBufs> + Send,
76 priority: bool,
77 ) -> Unreliable<Feedback>;
78 }
79
80 /// Interface for constructing a [`CheckedSender`] from a set of [`Recipients`],
81 /// filtering out any that are currently rate-limited.
82 pub trait LimitedSender: Clone + Send + Sync + 'static {
83 /// Public key type used to identify recipients.
84 type PublicKey: PublicKey;
85
86 /// The type of [`CheckedSender`] returned after checking recipients.
87 type Checked<'a>: CheckedSender<PublicKey = Self::PublicKey> + Send
88 where
89 Self: 'a;
90
91 /// Checks which recipients are within their rate limit and returns a
92 /// [`CheckedSender`] for sending to them.
93 ///
94 /// # Rate Limiting
95 ///
96 /// Recipients that exceed their rate limit will be filtered out. The
97 /// returned [`CheckedSender`] will only send to non-limited recipients.
98 ///
99 /// # Returns
100 ///
101 /// A [`CheckedSender`] containing only the recipients that are not
102 /// currently rate-limited, or an error with the earliest instant at which
103 /// all recipients will be available if all are rate-limited.
104 fn check(
105 &mut self,
106 recipients: Recipients<Self::PublicKey>,
107 ) -> Result<Self::Checked<'_>, SystemTime>;
108 }
109
110 /// Interface for sending messages to [`Recipients`] that are not currently rate-limited.
111 pub trait CheckedSender: Send {
112 /// Public key type used to identify [`Recipients`].
113 type PublicKey: PublicKey;
114
115 /// Returns the recipients retained by the check.
116 fn recipients(&self) -> Vec<Self::PublicKey>;
117
118 /// Sends a message to the pre-checked recipients.
119 ///
120 /// # Offline Recipients
121 ///
122 /// If a recipient is offline at the time a message is sent, the message
123 /// will be dropped. It is up to the application to handle retries (if
124 /// necessary).
125 ///
126 /// # Panics
127 ///
128 /// Panics if `message` exceeds the sender's configured maximum application payload size.
129 ///
130 /// # Returns
131 ///
132 /// Feedback from submitting the message for delivery.
133 /// [`Unreliable`] indicates that local submission may be rejected under backpressure.
134 /// [`Feedback::accepted`] does not guarantee that the recipient will receive the message.
135 fn send(self, message: impl Into<IoBufs> + Send, priority: bool) -> Unreliable<Feedback>;
136 }
137
138 /// Interface for sending messages to a set of recipients.
139 pub trait Sender: LimitedSender {
140 /// Sends a message to a set of recipients.
141 ///
142 /// # Offline Recipients
143 ///
144 /// If a recipient is offline at the time a message is sent, the message
145 /// will be dropped. It is up to the application to handle retries (if
146 /// necessary).
147 ///
148 /// # Rate Limiting
149 ///
150 /// Recipients that exceed their rate limit will be skipped. The message is
151 /// still sent to non-limited recipients.
152 ///
153 /// # Panics
154 ///
155 /// Panics if `message` exceeds the sender's configured maximum application payload size.
156 ///
157 /// # Returns
158 ///
159 /// The recipients we will attempt to send to. Returns an
160 /// empty list if all recipients are rate-limited, the sender has closed, or the send is
161 /// not accepted.
162 fn send(
163 &mut self,
164 recipients: Recipients<Self::PublicKey>,
165 message: impl Into<IoBufs> + Send,
166 priority: bool,
167 ) -> Vec<Self::PublicKey> {
168 self.check(recipients).map_or_else(
169 |_| Vec::new(),
170 |checked_sender| {
171 let recipients = checked_sender.recipients();
172 let feedback = checked_sender.send(message, priority);
173 if feedback.accepted() {
174 recipients
175 } else {
176 Vec::new()
177 }
178 },
179 )
180 }
181 }
182
183 // Blanket implementation of `Sender` for all `LimitedSender`s.
184 impl<S: LimitedSender> Sender for S {}
185
186 /// Interface for receiving messages from arbitrary recipients.
187 pub trait Receiver: Debug + Send + 'static {
188 /// Error that can occur when receiving a message.
189 type Error: Debug + StdError + Send + Sync;
190
191 /// Public key type used to identify recipients.
192 type PublicKey: PublicKey;
193
194 /// Receive a message from an arbitrary recipient.
195 fn recv(
196 &mut self,
197 ) -> impl Future<Output = Result<Message<Self::PublicKey>, Self::Error>> + Send;
198 }
199
200 /// Notification sent to subscribers when a peer set changes.
201 #[derive(Clone, Debug)]
202 pub struct PeerSetUpdate<P: PublicKey> {
203 /// The index of the peer set that changed.
204 pub index: u64,
205 /// The primary and secondary peers in the new set.
206 pub latest: TrackedPeers<P>,
207 /// Union of primary and secondary peers across all tracked peer sets.
208 pub all: TrackedPeers<P>,
209 }
210
211 /// Alias for the subscription type returned by [`Provider::subscribe`].
212 pub type PeerSetSubscription<P> = mpsc::UnboundedReceiver<PeerSetUpdate<P>>;
213
214 /// Alias for the subscription type returned by [`Blocker::blocked`].
215 ///
216 /// Each value is the full set of peers this node currently blocks.
217 pub type BlockedSubscription<P> = ring::Receiver<Set<P>>;
218
219 /// Primary and secondary peers provided together to [`Manager::track`].
220 ///
221 /// The same public key may appear in both `primary` and `secondary`. [`Manager::track`]
222 /// deduplicates overlapping keys, storing them as primary only.
223 #[derive(Clone, Debug, PartialEq, Eq)]
224 pub struct TrackedPeers<P: PublicKey> {
225 /// Peers eligible for primary-only policies.
226 pub primary: Set<P>,
227 /// Peers eligible for secondary-only policies.
228 pub secondary: Set<P>,
229 }
230
231 impl<P: PublicKey> TrackedPeers<P> {
232 pub const fn new(primary: Set<P>, secondary: Set<P>) -> Self {
233 Self { primary, secondary }
234 }
235
236 pub fn primary(primary: Set<P>) -> Self {
237 Self::new(primary, Set::default())
238 }
239
240 /// Returns the deduplicated union of primary and secondary peers.
241 pub fn union(self) -> Set<P> {
242 Set::from_iter_dedup(self.primary.into_iter().chain(self.secondary))
243 }
244 }
245
246 impl<P: PublicKey> From<Set<P>> for TrackedPeers<P> {
247 fn from(primary: Set<P>) -> Self {
248 Self::primary(primary)
249 }
250 }
251
252 impl<P: PublicKey> Default for TrackedPeers<P> {
253 fn default() -> Self {
254 Self::new(Set::default(), Set::default())
255 }
256 }
257
258 /// Primary and secondary peers provided together to [`AddressableManager::track`].
259 ///
260 /// The same public key may appear in both maps. [`AddressableManager::track`]
261 /// deduplicates overlapping keys, storing them as primary only.
262 #[derive(Clone, Debug)]
263 pub struct AddressableTrackedPeers<P: PublicKey> {
264 /// Addresses for peers eligible for primary-only policies.
265 pub primary: Map<P, Address>,
266 /// Addresses for peers eligible for secondary-only policies.
267 pub secondary: Map<P, Address>,
268 }
269
270 impl<P: PublicKey> AddressableTrackedPeers<P> {
271 pub const fn new(primary: Map<P, Address>, secondary: Map<P, Address>) -> Self {
272 Self { primary, secondary }
273 }
274
275 pub fn primary(primary: Map<P, Address>) -> Self {
276 Self::new(primary, Map::default())
277 }
278 }
279
280 impl<P: PublicKey> From<Map<P, Address>> for AddressableTrackedPeers<P> {
281 fn from(primary: Map<P, Address>) -> Self {
282 Self::primary(primary)
283 }
284 }
285
286 /// Interface for reading peer set information.
287 pub trait Provider: Debug + Clone + Send + 'static {
288 /// Public key type used to identify peers.
289 type PublicKey: PublicKey;
290
291 /// Fetch the primary and secondary peers tracked at the given ID.
292 fn peer_set(
293 &mut self,
294 id: u64,
295 ) -> impl Future<Output = Option<TrackedPeers<Self::PublicKey>>> + Send;
296
297 /// Subscribe to notifications when new peer sets are added.
298 ///
299 /// Returns a receiver of [`PeerSetUpdate`] notifications. Each update's
300 /// `latest` reflects how [`Manager::track`] stored the set: a peer listed in
301 /// both roles appears only under `latest.primary`. The `all` field aggregates
302 /// across tracked sets with the same rule (secondary excludes keys present as primary).
303 fn subscribe(
304 &mut self,
305 ) -> impl Future<Output = PeerSetSubscription<Self::PublicKey>> + Send;
306 }
307
308 /// Interface for managing peer set membership (where peer addresses are not known).
309 pub trait Manager: Provider {
310 /// Track a primary and secondary peer set with the given ID.
311 ///
312 /// The peer set ID passed to this function should be strictly managed, ideally matching the epoch
313 /// of the consensus engine. It must be monotonically increasing as new peer sets are
314 /// tracked.
315 ///
316 /// For good connectivity, all peers must track the same peer sets at the same ID.
317 ///
318 /// Callers may pass either a list of primary peers or a [`TrackedPeers`] value containing both primary and secondary peers.
319 ///
320 /// Overlapping keys in [`TrackedPeers`] are allowed; they are deduplicated as primary only.
321 ///
322 /// ## Active Peers
323 ///
324 /// The most recently registered peer set (highest ID) is considered the
325 /// active set. Implementations use the active set to decide which peers to
326 /// maintain connections with and which to disconnect from.
327 ///
328 /// ## Primary vs Secondary Peers
329 ///
330 /// In p2p networks, there are often two tiers of peers: ones that help "drive progress" and ones that want to
331 /// "follow that progress" (but not contribute to it). We call the former "primary" and the latter "secondary".
332 /// When both are tracked, mechanisms favor "primary" peers but continue to replicate data to "secondary" peers (
333 /// often both gossiping data to them and answering requests from them).
334 fn track<R>(&mut self, id: u64, peers: R) -> Feedback
335 where
336 R: Into<TrackedPeers<Self::PublicKey>> + Send;
337 }
338
339 /// Interface for managing peer set membership (where peer addresses are known).
340 pub trait AddressableManager: Provider {
341 /// Track a primary peer set and secondary peers with the given ID.
342 ///
343 /// The peer set ID passed to this function should be strictly managed, ideally matching the epoch
344 /// of the consensus engine. It must be monotonically increasing as new peer sets are
345 /// tracked.
346 ///
347 /// For good connectivity, all peers must track the same peer sets at the same ID.
348 ///
349 /// Callers may pass either a list of primary peers or a [`AddressableTrackedPeers`] value containing
350 /// both primary and secondary peers.
351 ///
352 /// The same key may appear in both maps; see [`AddressableTrackedPeers`].
353 ///
354 /// ## Active Peers
355 ///
356 /// The most recently registered peer set (highest ID) is considered the
357 /// active set. Implementations use the active set to decide which peers to
358 /// maintain connections with and which to disconnect from.
359 ///
360 /// ## Primary vs Secondary Peers
361 ///
362 /// In p2p networks, there are often two tiers of peers: ones that help "drive progress" and ones that want to
363 /// "follow that progress" (but not contribute to it). We call the former "primary" and the latter "secondary".
364 /// When both are tracked, mechanisms favor "primary" peers but continue to replicate data to "secondary" peers (
365 /// often both gossiping data to them and answering requests from them).
366 fn track<R>(&mut self, id: u64, peers: R) -> Feedback
367 where
368 R: Into<AddressableTrackedPeers<Self::PublicKey>> + Send;
369
370 /// Update addresses for multiple peers without creating a new peer set.
371 ///
372 /// For each primary or secondary peer with a changed address:
373 /// - Any existing connection to the peer is severed (it was on the old IP)
374 /// - The listener's allowed IPs are updated to reflect the new egress IP
375 /// - Future connections will use the new address
376 fn overwrite(&mut self, peers: Map<Self::PublicKey, Address>) -> Feedback;
377 }
378
379 /// Interface for blocking other peers.
380 pub trait Blocker: Clone + Send + 'static {
381 /// Public key type used to identify peers.
382 type PublicKey: PublicKey;
383
384 /// Block a peer, disconnecting them if currently connected and preventing future connections.
385 fn block(&mut self, peer: Self::PublicKey) -> Feedback;
386
387 /// Subscribe to the set of peers this node currently blocks.
388 ///
389 /// The subscription yields the current set once the request is processed,
390 /// then a new set whenever a peer is blocked or unblocked. An unread set is
391 /// replaced by the next one, so a reader always sees the latest state.
392 fn blocked(&mut self) -> BlockedSubscription<Self::PublicKey>;
393 }
394});
395
396/// Logs a warning and blocks a peer in a single call.
397///
398/// This macro combines a [`tracing::warn!`] with a [`Blocker::block`] call
399/// to ensure consistent logging at every block site. The peer is always
400/// included as a `peer` field in the log output.
401///
402/// # Examples
403///
404/// ```ignore
405/// block!(self.blocker, sender, "invalid message");
406/// block!(self.blocker, sender, ?err, "invalid ack signature");
407/// block!(self.blocker, sender, %view, "blocking peer for epoch mismatch");
408/// ```
409#[cfg(not(any(
410 commonware_stability_GAMMA,
411 commonware_stability_DELTA,
412 commonware_stability_EPSILON,
413 commonware_stability_RESERVED
414)))] // BETA
415#[macro_export]
416macro_rules! block {
417 ($blocker:expr, $peer:expr, $($arg:tt)+) => {
418 let peer = $peer;
419 tracing::warn!(peer = ?peer, $($arg)+);
420 #[allow(clippy::disallowed_methods)]
421 $blocker.block(peer)
422 };
423}
424
425/// Block a peer without logging.
426#[allow(
427 clippy::disallowed_methods,
428 reason = "test helper that bypasses the block! macro"
429)]
430#[cfg(test)]
431pub fn block_peer<B: Blocker>(blocker: &mut B, peer: B::PublicKey) -> Feedback {
432 blocker.block(peer)
433}