iroh_gossip/
api.rs

1//! Public API for using iroh-gossip
2//!
3//! The API is usable both locally and over RPC.
4
5use std::{
6    collections::{BTreeSet, HashSet},
7    pin::Pin,
8    task::{Context, Poll},
9};
10
11use bytes::Bytes;
12use iroh_base::NodeId;
13use irpc::{channel::mpsc, rpc_requests, Client};
14use n0_future::{Stream, StreamExt, TryStreamExt};
15use nested_enum_utils::common_fields;
16use serde::{Deserialize, Serialize};
17use snafu::Snafu;
18
19use crate::proto::{DeliveryScope, TopicId};
20
21/// Default channel capacity for topic subscription channels (one per topic)
22const TOPIC_EVENTS_DEFAULT_CAP: usize = 2048;
23/// Channel capacity for topic command send channels.
24const TOPIC_COMMANDS_CAP: usize = 64;
25
26/// Input messages for the gossip actor.
27#[rpc_requests(message = RpcMessage, rpc_feature = "rpc")]
28#[derive(Debug, Serialize, Deserialize)]
29pub(crate) enum Request {
30    #[rpc(tx=mpsc::Sender<Event>, rx=mpsc::Receiver<Command>)]
31    Join(JoinRequest),
32}
33
34#[derive(Debug, Serialize, Deserialize)]
35pub(crate) struct JoinRequest {
36    pub topic_id: TopicId,
37    pub bootstrap: BTreeSet<NodeId>,
38}
39
40#[allow(missing_docs)]
41#[common_fields({
42    backtrace: Option<snafu::Backtrace>,
43    #[snafu(implicit)]
44    span_trace: n0_snafu::SpanTrace,
45})]
46#[derive(Debug, Snafu)]
47#[non_exhaustive]
48pub enum ApiError {
49    #[snafu(transparent)]
50    Rpc { source: irpc::Error },
51    /// The gossip topic was closed.
52    #[snafu(display("topic closed"))]
53    Closed {},
54}
55
56impl From<irpc::channel::SendError> for ApiError {
57    fn from(value: irpc::channel::SendError) -> Self {
58        irpc::Error::from(value).into()
59    }
60}
61
62impl From<irpc::channel::mpsc::RecvError> for ApiError {
63    fn from(value: irpc::channel::mpsc::RecvError) -> Self {
64        irpc::Error::from(value).into()
65    }
66}
67
68impl From<irpc::channel::oneshot::RecvError> for ApiError {
69    fn from(value: irpc::channel::oneshot::RecvError) -> Self {
70        irpc::Error::from(value).into()
71    }
72}
73
74/// API to control a [`Gossip`] instance.
75///
76/// This has methods to subscribe and join gossip topics, which return handles to publish
77/// and receive messages on topics.
78///
79/// [`Gossip`] derefs to [`GossipApi`], so all functions on [`GossipApi`] are directly callable
80/// from [`Gossip`].
81///
82/// Additionally, a [`GossipApi`] can be created by connecting to an RPC server. See [`Gossip::listen`]
83/// and [`GossipApi::connect`] (*requires the `rpc` feature).
84///
85/// [`Gossip`]: crate::net::Gossip
86/// [`Gossip::listen`]: crate::net::Gossip::listen
87#[derive(Debug, Clone)]
88pub struct GossipApi {
89    client: Client<Request>,
90}
91
92impl GossipApi {
93    #[cfg(feature = "net")]
94    pub(crate) fn local(tx: tokio::sync::mpsc::Sender<RpcMessage>) -> Self {
95        let local = irpc::LocalSender::<Request>::from(tx);
96        Self {
97            client: local.into(),
98        }
99    }
100
101    /// Connect to a remote as a RPC client.
102    #[cfg(feature = "rpc")]
103    pub fn connect(endpoint: quinn::Endpoint, addr: std::net::SocketAddr) -> Self {
104        let inner = irpc::Client::quinn(endpoint, addr);
105        Self { client: inner }
106    }
107
108    /// Listen on a quinn endpoint for incoming RPC connections.
109    #[cfg(all(feature = "rpc", feature = "net"))]
110    pub(crate) async fn listen(&self, endpoint: quinn::Endpoint) {
111        use irpc::rpc::{listen, RemoteService};
112
113        let local = self
114            .client
115            .as_local()
116            .expect("cannot listen on remote client");
117        let handler = Request::remote_handler(local);
118
119        listen::<Request>(endpoint, handler).await
120    }
121
122    /// Join a gossip topic with options.
123    ///
124    /// Returns a [`GossipTopic`] instantly. To wait for at least one connection to be established,
125    /// you can await [`GossipTopic::joined`].
126    ///
127    /// Messages will be queued until a first connection is available. If the internal channel becomes full,
128    /// the oldest messages will be dropped from the channel.
129    pub async fn subscribe_with_opts(
130        &self,
131        topic_id: TopicId,
132        opts: JoinOptions,
133    ) -> Result<GossipTopic, ApiError> {
134        let req = JoinRequest {
135            topic_id,
136            bootstrap: opts.bootstrap,
137        };
138        let (tx, rx) = self
139            .client
140            .bidi_streaming(req, TOPIC_COMMANDS_CAP, opts.subscription_capacity)
141            .await?;
142        Ok(GossipTopic::new(tx, rx))
143    }
144
145    /// Join a gossip topic with the default options and wait for at least one active connection.
146    pub async fn subscribe_and_join(
147        &self,
148        topic_id: TopicId,
149        bootstrap: Vec<NodeId>,
150    ) -> Result<GossipTopic, ApiError> {
151        let mut sub = self
152            .subscribe_with_opts(topic_id, JoinOptions::with_bootstrap(bootstrap))
153            .await?;
154        sub.joined().await?;
155        Ok(sub)
156    }
157
158    /// Join a gossip topic with the default options.
159    ///
160    /// Note that this will not wait for any bootstrap node to be available.
161    /// To ensure the topic is connected to at least one node, use [`GossipTopic::joined`]
162    /// or [`Self::subscribe_and_join`]
163    pub async fn subscribe(
164        &self,
165        topic_id: TopicId,
166        bootstrap: Vec<NodeId>,
167    ) -> Result<GossipTopic, ApiError> {
168        let sub = self
169            .subscribe_with_opts(topic_id, JoinOptions::with_bootstrap(bootstrap))
170            .await?;
171
172        Ok(sub)
173    }
174}
175
176/// Sender for a gossip topic.
177#[derive(Debug, Clone)]
178pub struct GossipSender(mpsc::Sender<Command>);
179
180impl GossipSender {
181    pub(crate) fn new(sender: mpsc::Sender<Command>) -> Self {
182        Self(sender)
183    }
184
185    /// Broadcasts a message to all nodes.
186    pub async fn broadcast(&self, message: Bytes) -> Result<(), ApiError> {
187        self.send(Command::Broadcast(message)).await?;
188        Ok(())
189    }
190
191    /// Broadcasts a message to our direct neighbors.
192    pub async fn broadcast_neighbors(&self, message: Bytes) -> Result<(), ApiError> {
193        self.send(Command::BroadcastNeighbors(message)).await?;
194        Ok(())
195    }
196
197    /// Joins a set of peers.
198    pub async fn join_peers(&self, peers: Vec<NodeId>) -> Result<(), ApiError> {
199        self.send(Command::JoinPeers(peers)).await?;
200        Ok(())
201    }
202
203    async fn send(&self, command: Command) -> Result<(), irpc::channel::SendError> {
204        self.0.send(command).await?;
205        Ok(())
206    }
207}
208
209/// Subscribed gossip topic.
210///
211/// This handle is a [`Stream`] of [`Event`]s from the topic, and can be used to send messages.
212///
213/// Once the [`GossipTopic`] is dropped, the network actor will leave the gossip topic.
214///
215/// It may be split into sender and receiver parts with [`Self::split`]. In this case, the topic will
216/// be left once both the [`GossipSender`] and [`GossipReceiver`] halves are dropped.
217#[derive(Debug)]
218pub struct GossipTopic {
219    sender: GossipSender,
220    receiver: GossipReceiver,
221}
222
223impl GossipTopic {
224    pub(crate) fn new(sender: mpsc::Sender<Command>, receiver: mpsc::Receiver<Event>) -> Self {
225        let sender = GossipSender::new(sender);
226        Self {
227            sender,
228            receiver: GossipReceiver::new(receiver),
229        }
230    }
231
232    /// Splits `self` into [`GossipSender`] and [`GossipReceiver`] parts.
233    pub fn split(self) -> (GossipSender, GossipReceiver) {
234        (self.sender, self.receiver)
235    }
236
237    /// Sends a message to all peers.
238    pub async fn broadcast(&mut self, message: Bytes) -> Result<(), ApiError> {
239        self.sender.broadcast(message).await
240    }
241
242    /// Sends a message to our direct neighbors in the swarm.
243    pub async fn broadcast_neighbors(&mut self, message: Bytes) -> Result<(), ApiError> {
244        self.sender.broadcast_neighbors(message).await
245    }
246
247    /// Waits until we are connected to at least one node.
248    ///
249    /// See [`GossipReceiver::joined`] for details.
250    pub async fn joined(&mut self) -> Result<(), ApiError> {
251        self.receiver.joined().await
252    }
253
254    /// Returns `true` if we are connected to at least one node.
255    pub fn is_joined(&self) -> bool {
256        self.receiver.is_joined()
257    }
258}
259
260impl Stream for GossipTopic {
261    type Item = Result<Event, ApiError>;
262
263    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
264        Pin::new(&mut self.receiver).poll_next(cx)
265    }
266}
267
268/// Receiver for gossip events on a topic.
269///
270/// This is a [`Stream`] of [`Event`]s emitted from the topic.
271#[derive(derive_more::Debug)]
272pub struct GossipReceiver {
273    #[debug("BoxStream")]
274    stream: Pin<Box<dyn Stream<Item = Result<Event, ApiError>> + Send + Sync + 'static>>,
275    neighbors: HashSet<NodeId>,
276}
277
278impl GossipReceiver {
279    pub(crate) fn new(events_rx: mpsc::Receiver<Event>) -> Self {
280        let stream = events_rx.into_stream().map_err(ApiError::from);
281        let stream = Box::pin(stream);
282        Self {
283            stream,
284            neighbors: Default::default(),
285        }
286    }
287
288    /// Lists our current direct neighbors.
289    pub fn neighbors(&self) -> impl Iterator<Item = NodeId> + '_ {
290        self.neighbors.iter().copied()
291    }
292
293    /// Waits until we are connected to at least one node.
294    ///
295    /// Progresses the event stream to the first [`Event::NeighborUp`] event.
296    ///
297    /// Note that this consumes this initial `NeighborUp` event. If you want to track
298    /// neighbors, use [`Self::neighbors`] after awaiting [`Self::joined`], and then
299    /// continue to track `NeighborUp` events on the event stream.
300    pub async fn joined(&mut self) -> Result<(), ApiError> {
301        while !self.is_joined() {
302            let _event = self.next().await.ok_or(ClosedSnafu.build())??;
303        }
304        Ok(())
305    }
306
307    /// Returns `true` if we are connected to at least one node.
308    pub fn is_joined(&self) -> bool {
309        !self.neighbors.is_empty()
310    }
311}
312
313impl Stream for GossipReceiver {
314    type Item = Result<Event, ApiError>;
315
316    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
317        let item = std::task::ready!(Pin::new(&mut self.stream).poll_next(cx));
318        if let Some(Ok(item)) = &item {
319            match item {
320                Event::NeighborUp(node_id) => {
321                    self.neighbors.insert(*node_id);
322                }
323                Event::NeighborDown(node_id) => {
324                    self.neighbors.remove(node_id);
325                }
326                _ => {}
327            }
328        }
329        Poll::Ready(item)
330    }
331}
332
333/// Events emitted from a gossip topic.
334///
335/// These are the events emitted from a [`GossipReceiver`].
336#[derive(Clone, PartialEq, Eq, PartialOrd, Ord, Debug, Serialize, Deserialize)]
337pub enum Event {
338    /// We have a new, direct neighbor in the swarm membership layer for this topic.
339    NeighborUp(NodeId),
340    /// We dropped direct neighbor in the swarm membership layer for this topic.
341    NeighborDown(NodeId),
342    /// We received a gossip message for this topic.
343    Received(Message),
344    /// We missed some messages because our [`GossipReceiver`] was not progressing fast enough.
345    Lagged,
346}
347
348impl From<crate::proto::Event<NodeId>> for Event {
349    fn from(event: crate::proto::Event<NodeId>) -> Self {
350        match event {
351            crate::proto::Event::NeighborUp(node_id) => Self::NeighborUp(node_id),
352            crate::proto::Event::NeighborDown(node_id) => Self::NeighborDown(node_id),
353            crate::proto::Event::Received(message) => Self::Received(Message {
354                content: message.content,
355                scope: message.scope,
356                delivered_from: message.delivered_from,
357            }),
358        }
359    }
360}
361
362/// A gossip message
363#[derive(Clone, PartialEq, Eq, PartialOrd, Ord, derive_more::Debug, Serialize, Deserialize)]
364pub struct Message {
365    /// The content of the message
366    #[debug("Bytes({})", self.content.len())]
367    pub content: Bytes,
368    /// The scope of the message.
369    /// This tells us if the message is from a direct neighbor or actual gossip.
370    pub scope: DeliveryScope,
371    /// The node that delivered the message. This is not the same as the original author.
372    pub delivered_from: NodeId,
373}
374
375/// Command for a gossip topic.
376#[derive(Serialize, Deserialize, derive_more::Debug, Clone)]
377pub enum Command {
378    /// Broadcasts a message to all nodes in the swarm.
379    Broadcast(#[debug("Bytes({})", _0.len())] Bytes),
380    /// Broadcasts a message to all direct neighbors.
381    BroadcastNeighbors(#[debug("Bytes({})", _0.len())] Bytes),
382    /// Connects to a set of peers.
383    JoinPeers(Vec<NodeId>),
384}
385
386/// Options for joining a gossip topic.
387#[derive(Serialize, Deserialize, Debug)]
388pub struct JoinOptions {
389    /// The initial bootstrap nodes.
390    pub bootstrap: BTreeSet<NodeId>,
391    /// The maximum number of messages that can be buffered in a subscription.
392    ///
393    /// If this limit is reached, the subscriber will receive a `Lagged` response,
394    /// the message will be dropped, and the subscriber will be closed.
395    ///
396    /// This is to prevent a single slow subscriber from blocking the dispatch loop.
397    /// If a subscriber is lagging, it should be closed and re-opened.
398    pub subscription_capacity: usize,
399}
400
401impl JoinOptions {
402    /// Creates [`JoinOptions`] with the provided bootstrap nodes and the default subscription
403    /// capacity.
404    pub fn with_bootstrap(nodes: impl IntoIterator<Item = NodeId>) -> Self {
405        Self {
406            bootstrap: nodes.into_iter().collect(),
407            subscription_capacity: TOPIC_EVENTS_DEFAULT_CAP,
408        }
409    }
410}
411
412#[cfg(test)]
413mod tests {
414    use crate::api::GossipTopic;
415
416    #[cfg(all(feature = "rpc", feature = "net"))]
417    #[tokio::test]
418    #[tracing_test::traced_test]
419    async fn test_rpc() -> n0_snafu::Result {
420        use iroh::{discovery::static_provider::StaticProvider, protocol::Router, RelayMap};
421        use n0_future::{time::Duration, StreamExt};
422        use n0_snafu::{Error, Result, ResultExt};
423        use rand_chacha::rand_core::SeedableRng;
424
425        use crate::{
426            api::{Event, GossipApi},
427            net::{test::create_endpoint, Gossip},
428            proto::TopicId,
429            ALPN,
430        };
431
432        let mut rng = rand_chacha::ChaCha12Rng::seed_from_u64(1);
433        let (relay_map, _relay_url, _guard) = iroh::test_utils::run_relay_server().await.unwrap();
434
435        async fn create_gossip_endpoint(
436            rng: &mut rand_chacha::ChaCha12Rng,
437            relay_map: RelayMap,
438        ) -> Result<(Router, Gossip)> {
439            let endpoint = create_endpoint(rng, relay_map, None).await?;
440            let gossip = Gossip::builder().spawn(endpoint.clone());
441            let router = Router::builder(endpoint)
442                .accept(ALPN, gossip.clone())
443                .spawn();
444            Ok((router, gossip))
445        }
446
447        let topic_id = TopicId::from_bytes([0u8; 32]);
448
449        // create our gossip node
450        let (router, gossip) = create_gossip_endpoint(&mut rng, relay_map.clone()).await?;
451
452        // create a second node so that we can test actually joining
453        let (node2_id, node2_addr, node2_task) = {
454            let (router, gossip) = create_gossip_endpoint(&mut rng, relay_map.clone()).await?;
455            let node_addr = router.endpoint().node_addr();
456            let node_id = router.endpoint().node_id();
457            let task = tokio::task::spawn(async move {
458                let mut topic = gossip.subscribe_and_join(topic_id, vec![]).await?;
459                topic.broadcast(b"hello".to_vec().into()).await?;
460                Result::<_, Error>::Ok(router)
461            });
462            (node_id, node_addr, task)
463        };
464
465        // create static provider to add node addr manually
466        let static_provider = StaticProvider::new();
467        static_provider.add_node_info(node2_addr);
468
469        router.endpoint().discovery().add(static_provider);
470
471        // expose the gossip node over RPC
472        let (rpc_server_endpoint, rpc_server_cert) =
473            irpc::util::make_server_endpoint("127.0.0.1:0".parse().unwrap())
474                .map_err(Error::anyhow)?;
475        let rpc_server_addr = rpc_server_endpoint.local_addr().e()?;
476        let rpc_server_task = tokio::task::spawn(async move {
477            gossip.listen(rpc_server_endpoint).await;
478        });
479
480        // connect to the RPC node with a new client
481        let rpc_client_endpoint =
482            irpc::util::make_client_endpoint("127.0.0.1:0".parse().unwrap(), &[&rpc_server_cert])
483                .map_err(Error::anyhow)?;
484        let rpc_client = GossipApi::connect(rpc_client_endpoint, rpc_server_addr);
485
486        // join via RPC
487        let recv = async move {
488            let mut topic = rpc_client
489                .subscribe_and_join(topic_id, vec![node2_id])
490                .await?;
491            // wait for a message
492            while let Some(event) = topic.try_next().await? {
493                match event {
494                    Event::Received(message) => {
495                        assert_eq!(&message.content[..], b"hello");
496                        break;
497                    }
498                    Event::Lagged => panic!("unexpected lagged event"),
499                    _ => {}
500                }
501            }
502            Result::<_, Error>::Ok(())
503        };
504
505        // timeout to not hang in case of failure
506        tokio::time::timeout(Duration::from_secs(10), recv)
507            .await
508            .e()??;
509
510        // shutdown
511        rpc_server_task.abort();
512        router.shutdown().await.e()?;
513        let router2 = node2_task.await.e()??;
514        router2.shutdown().await.e()?;
515        Ok(())
516    }
517
518    #[test]
519    fn ensure_gossip_topic_is_sync() {
520        #[allow(unused)]
521        fn get() -> GossipTopic {
522            unimplemented!()
523        }
524        #[allow(unused)]
525        fn check(_t: impl Sync) {}
526        #[allow(unused)]
527        fn foo() {
528            check(get());
529        }
530    }
531}
iroh_gossip/api.rs

iroh_gossip/
api.rs
