introduce relay only mode

1 files changed, 65 insertions, 5 deletions

diff --git a/src/transport/libp2p.rs b/src/transport/libp2p.rs
index a6778c6..7351b16 100644
--- a/src/transport/libp2p.rs
+++ b/src/transport/libp2p.rs
@@ -146,6 +146,9 @@ pub(crate) struct Libp2pConfig {
     pub(crate) relays: Vec<Multiaddr>,
     /// Relay for other nodes.
     pub(crate) relay: bool,
+    /// Speak continuum (`PROTOCOL`): not on a node that only relays (`--relay-only`),
+    /// which nobody should take for a node.
+    pub(crate) continuum: bool,
     /// Find peers on the LAN (mDNS).
     pub(crate) mdns: bool,
     /// Use public infrastructure: the IPFS DHT, public relays.
@@ -716,9 +719,15 @@ impl Libp2pTransport {
             swarm.add_external_address(address.clone());
         }
         let mut control = swarm.behaviour().stream.new_control();
-        let incoming = control
-            .accept(PROTOCOL)
-            .map_err(|error| anyhow!("Accepting {PROTOCOL}: {error}"))?;
+        let incoming = if config.continuum {
+            Some(
+                control
+                    .accept(PROTOCOL)
+                    .map_err(|error| anyhow!("Accepting {PROTOCOL}: {error}"))?,
+            )
+        } else {
+            None
+        };
         let (commands, receiver) = mpsc::unbounded_channel();
         let mut book = Book {
             announced: config.announce,
@@ -751,7 +760,7 @@ impl Libp2pTransport {
             control,
             commands,
             book,
-            incoming: tokio::sync::Mutex::new(Some(incoming)),
+            incoming: tokio::sync::Mutex::new(incoming),
             stopped,
         })
     }
@@ -1694,7 +1703,7 @@ impl Transport for Libp2pTransport {
     /// time, in the order they were read; PINGs and PONGs at once (`is_probe`).
     async fn start_consumer(&self, handler: MessageHandler) -> Result<()> {
         let Some(mut incoming) = self.incoming.lock().await.take() else {
-            bail!("libp2p consumer already started");
+            bail!("libp2p consumer already started, or this node only relays");
         };
         let handler: Arc<HandlerFn> = Arc::from(handler);
         let (sender, mut received) = mpsc::channel::<(PeerId, ParsedMessage, usize)>(READS);
@@ -1781,6 +1790,7 @@ mod tests {
             announce: Vec::new(),
             relays,
             relay,
+            continuum: true,
             mdns: false,
             public: false,
             memory: true,
@@ -1926,6 +1936,7 @@ mod tests {
             announce: vec![announced.clone()],
             relays: Vec::new(),
             relay: false,
+            continuum: true,
             mdns: false,
             public: false,
             memory: true,
@@ -1988,6 +1999,55 @@ mod tests {
         assert!(a.connection(b.local_id()).is_some());
     }
 
+    /// A relay that is no node (`--relay-only`) relays, and takes no messages itself.
+    #[tokio::test]
+    async fn a_relay_that_is_no_node() {
+        let port = NEXT_PORT.fetch_add(1, Ordering::Relaxed);
+        let key = Keypair::generate_ed25519();
+        let r_peer = key.public().to_peer_id();
+        let r_listen: Multiaddr = format!("/memory/{port}").parse().unwrap();
+        let r = Libp2pTransport::start(Libp2pConfig {
+            key,
+            listen: vec![r_listen.clone()],
+            announce: Vec::new(),
+            relays: Vec::new(),
+            relay: true,
+            continuum: false,
+            mdns: false,
+            public: false,
+            memory: true,
+        })
+        .unwrap();
+        let r_address = r_listen.with_p2p(r_peer).unwrap();
+        let (b, _) = memory_node_with(false, vec![r_address.clone()], false);
+        let b = Arc::new(b);
+        let mut at_b = received(Arc::clone(&b));
+        assert!(
+            eventually(|| !b.local_addresses().is_empty()).await,
+            "b got no slot on the relay"
+        );
+        let (a, _) = memory_node();
+        let relayed = b.local_addresses()[0].clone();
+        assert_eq!(a.resolve_seed(&relayed).await.unwrap(), b.local_id());
+        a.send_message(b.local_id(), &status()).await.unwrap();
+        let (from, _) = at_b.recv().await.unwrap();
+        assert_eq!(from, a.local_id());
+
+        // The relay itself doesn't speak continuum: it takes no messages (a send to it
+        // isn't refused, though: the protocol goes with the first data, `swarm_config`,
+        // so the relay's refusal comes too late; nodes never send to it, as it never
+        // proves itself a node).
+        assert_eq!(
+            a.resolve_seed(&r_address.to_string()).await.unwrap(),
+            r.local_id()
+        );
+        assert!(
+            r.start_consumer(Box::new(|_, _, _| Box::pin(async { Ok(()) })))
+                .await
+                .is_err()
+        );
+    }
+
     /// A relay configured without its peer ID is found out, and used.
     #[tokio::test]
     async fn a_relay_configured_without_its_peer_id() {