net: Disconnect peers which send us packets that we don't have dispatchers for.

src/error.rs

@@ -158,6 +158,9 @@ pub enum Error {
     #[error("Network operation failed")]
     NetworkOperationFailed,
 
+    #[error("Missing P2P message dispatcher")]
+    MissingDispatcher,
+
     #[cfg(feature = "arti-client")]
     #[error(transparent)]
     ArtiError(#[from] arti_client::Error),

src/net/channel.rs

@@ -288,7 +288,15 @@ impl Channel {
             });
 
             // Send result to our subscribers
-            self.message_subsystem.notify(&packet.command, &packet.payload).await;
+            match self.message_subsystem.notify(&packet.command, &packet.payload).await {
+                Ok(()) => {}
+                // If we're getting messages without dispatchers, it's spam.
+                Err(Error::MissingDispatcher) => {
+                    debug!(target: "net::channel::main_receive_loop()", "Stopping channel {:?}", self);
+                    return Err(Error::ChannelStopped)
+                }
+                Err(_) => unreachable!("You added a new error in notify()"),
+            }
         }
     }
 

src/net/message_subscriber.rs

@@ -241,17 +241,17 @@ impl MessageSubsystem {
 
     /// Transmits a payload to a dispatcher.
     /// Returns an error if the payload fails to transmit.
-    pub async fn notify(&self, command: &str, payload: &[u8]) {
+    pub async fn notify(&self, command: &str, payload: &[u8]) -> Result<()> {
         let Some(dispatcher) = self.dispatchers.lock().await.get(command).cloned() else {
             warn!(
                 target: "net::message_subscriber::notify",
                 "message_subscriber::notify: Command '{}' did not find a dispatcher",
                 command,
             );
-            return
+            return Err(Error::MissingDispatcher)
         };
 
-        dispatcher.trigger(payload).await;
+        Ok(dispatcher.trigger(payload).await)
     }
 
     /// Concurrently transmits an error message across dispatchers.