Pump the gossip engine while waiting for the BEEFY runtime pallet (memory leak fix) (#11694)

* Pump the gossip engine while waiting for the BEEFY runtime pallet

This fixes a memory leak when the BEEFY gadget is turned on, but
the runtime doesn't actually use BEEFY.

* Implement `FusedFuture` for `GossipEngine`

* Fuse futures outside of loops

substrate/client/network-gossip/src/bridge.rs

@@ -53,6 +53,8 @@ pub struct GossipEngine<B: BlockT> {
 	message_sinks: HashMap<B::Hash, Vec<Sender<TopicNotification>>>,
 	/// Buffered messages (see [`ForwardingState`]).
 	forwarding_state: ForwardingState<B>,
+
+	is_terminated: bool,
 }
 
 /// A gossip engine receives messages from the network via the `network_event_stream` and forwards
@@ -94,6 +96,8 @@ impl<B: BlockT> GossipEngine<B> {
 			network_event_stream,
 			message_sinks: HashMap::new(),
 			forwarding_state: ForwardingState::Idle,
+
+			is_terminated: false,
 		}
 	}
 
@@ -214,7 +218,10 @@ impl<B: BlockT> Future for GossipEngine<B> {
 							Event::Dht(_) => {},
 						},
 						// The network event stream closed. Do the same for [`GossipValidator`].
-						Poll::Ready(None) => return Poll::Ready(()),
+						Poll::Ready(None) => {
+							self.is_terminated = true;
+							return Poll::Ready(())
+						},
 						Poll::Pending => break,
 					}
 				},
@@ -288,6 +295,12 @@ impl<B: BlockT> Future for GossipEngine<B> {
 	}
 }
 
+impl<B: BlockT> futures::future::FusedFuture for GossipEngine<B> {
+	fn is_terminated(&self) -> bool {
+		self.is_terminated
+	}
+}
+
 #[cfg(test)]
 mod tests {
 	use super::*;