feat: use broadcast channel for event listeners

[fgimenez]

EventListeners implements a multi producer multi consumer queue where each sent value is seen by all consumers.
To achieve this EventListeners allocates a std::Vec to be filled with tokio::sync::UnboundedSender every time EventListeners::new_listener is called.

As every value sent via EventListeners is cloned to each UnboundedReceiver and the channels are unbounded this is prone to unlimited memory growth and eventual OOM attacks.

To prevent this, in this PR tokio's tokio::sync::broadcast multi producer multi consumer queue is used instead.

For now the size of all the broadcast channels is set to 1000, would be good to measure how much is needed for each. Pending adding metrics as suggested in this comment #8193 (comment) will be done in a follow up.

[emhane]

check out these types we have

reth/crates/metrics/src/common/mpsc.rs

Lines 36 to 43 in ef01d50

	/// A wrapper type around [UnboundedSender](mpsc::UnboundedSender) that updates metrics on send.
	#[derive(Debug)]
	pub struct UnboundedMeteredSender<T> {
	/// The [UnboundedSender](mpsc::UnboundedSender) that this wraps around
	sender: mpsc::UnboundedSender<T>,
	/// Holds metrics for this type
	metrics: MeteredSenderMetrics,
	}

reth/crates/metrics/src/common/mpsc.rs

Lines 75 to 82 in ef01d50

	/// A wrapper type around [Receiver](mpsc::UnboundedReceiver) that updates metrics on receive.
	#[derive(Debug)]
	pub struct UnboundedMeteredReceiver<T> {
	/// The [Sender](mpsc::Sender) that this wraps around
	receiver: mpsc::UnboundedReceiver<T>,
	/// Holds metrics for this type
	metrics: MeteredReceiverMetrics,
	}

we use them so far only for the channel NetworkManager->TransactionsManager for incoming transaction gossip, but I'd like to see more of them in the codebase.

we have a panel for observing this channel
https://reth.paradigm.xyz/d/d47d679c-c3b8-40b6-852d-cbfaa2dcdb37/reth---transaction-pool?orgId=1&refresh=30s&viewPanel=95

[fgimenez]

@emhane awesome thx! will check how to include something similar for the broadcast channels, the metrics can be very useful to assign the proper size to each

[emhane]

I see, you don't need a new BeaconEngineMessage to subscribe to the broadcast stream because you clone the sender and pass it to the handle, neat

[emhane]

how about implementing FromIterator here, that will work I think

[Rjected]

how about implementing FromIterator here, that will work I think

the map is provided by streamext, and streams are not (sync) iterators, so I'm not sure Fromiterator is the right fit here

[emhane]

https://docs.rs/tokio-stream/latest/tokio_stream/trait.FromStream.html

[Rjected]

rip, would be nice but it's sealed, hopefully soon ™️ in stable

[mattsse]

overall I think this great.
I don't think we'll have any issues with this for engine/pipeline events because those basically just for reporting and it's fine to drop some.

My main concern is the network transaction task which is more likely to drop messages but it relies on networkevents for peer tracking for example.
although 1k messages should be fine, I'd feel more comfortable if we could bump the default capacity to 2k and add a metric for when we lag in the tx task. maybe we should emit peer added/removed separately, but we should still proceed with this.

I'd also like a new function/stream variant that does not return results but rather skips the lag error, this would make the API easier in some places, ref

reth/crates/storage/provider/src/traits/chain.rs

Lines 37 to 43 in cb658ca

	/// A Stream of [CanonStateNotification].
	#[derive(Debug)]
	#[pin_project::pin_project]
	pub struct CanonStateNotificationStream {
	#[pin]
	st: BroadcastStream<CanonStateNotification>,
	}

we could move this stream type to our tokio util crate

we also need this for the txpool channels which is mostlikely the most critical part because exposed over RPC.

[mattsse]

I'm slightly concerned about this, because now we're no longer guaranteed delivery of all network events which can result in wrong peer tracking, for example session closed, although 1000 messages should be sufficient

[emhane]

I think dropping NetworkEvent::SessionEstablished and NetworkEvent::PeerAdded is recoverable, but not sure if dropping NetworkEvent::SessionClosed and NetworkEvent::PeerRemoved can lead to memory leak. depends on if all data structures that are updated accordingly are bounded.

[fgimenez]

I'd also like a new function/stream variant that does not return results but rather skips the lag error,

makes total sense, done ptal

[mattsse]

this is great!
broadcast is def better for this