Only send transactions to peers with open substream

bin/light-base/src/network_service.rs

@@ -739,7 +739,12 @@ impl<TPlat: Platform> NetworkService<TPlat> {
         let mut guarded = self.shared.guarded.lock().await;
 
         // TODO: collecting in a Vec :-/
-        for peer in guarded.network.peers_list().cloned().collect::<Vec<_>>() {
+        for peer in guarded
+            .network
+            .opened_transactions_substream(chain_index)
+            .cloned()
+            .collect::<Vec<_>>()
+        {
             if guarded
                 .network
                 .announce_transaction(&peer, chain_index, transaction)

bin/wasm-node/CHANGELOG.md

@@ -6,6 +6,7 @@
 
 - Fix occasional panic when connecting to a parachain with forks and/or missed slots. ([#2703](https://github.com/paritytech/smoldot/pull/2703))
 - Fix parachain initialization unnecessarily waiting for its corresponding relay chain initialization to be finished. ([#2705](https://github.com/paritytech/smoldot/pull/2705))
+- Fix panic when broadcasting a transaction to a peer while its connection is shutting down.
 
 ## 0.6.31 - 2022-08-30
 

src/libp2p/peers.rs

@@ -922,6 +922,20 @@ where
         )
     }
 
+    /// Returns the list of peers for which we have a fully established notifications protocol of
+    /// the given protocol.
+    pub fn opened_out_notifications(
+        &'_ self,
+        notifications_protocol_index: usize,
+    ) -> impl Iterator<Item = &'_ PeerId> + '_ {
+        // TODO: this is O(n)
+        self.peers_notifications_out
+            .iter()
+            .filter(move |((_, idx), _)| *idx == notifications_protocol_index)
+            .filter(|(_, state)| matches!(state.open, NotificationsOutOpenState::Open(_)))
+            .map(|((peer_idx, _), _)| &self.peers[*peer_idx].peer_id)
+    }
+
     /// Returns the state of the given connection.
     ///
     /// # Panic

src/network/service.rs

@@ -812,6 +812,16 @@ where
         )
     }
 
+    /// Returns the list of peers for which we have a fully established notifications protocol of
+    /// the given protocol.
+    pub fn opened_transactions_substream(
+        &'_ self,
+        chain_index: usize,
+    ) -> impl Iterator<Item = &'_ PeerId> + '_ {
+        self.inner
+            .opened_out_notifications(chain_index * NOTIFICATIONS_PROTOCOLS_PER_CHAIN + 1)
+    }
+
     ///
     ///
     /// Must be passed the SCALE-encoded transaction.