p2p/sentry: PeersStreams calls Send on one PeerEvents stream from several goroutines

### What is wrong

`PeersStreams` in the sentry gRPC server ends up calling `Send` on the same `PeerEvents` stream from more than one goroutine. grpc-go does not allow this:

> It is safe to have a goroutine calling SendMsg and another goroutine calling RecvMsg on the same stream at the same time, but **it is not safe to call SendMsg on the same stream in different goroutines**.

(doc on `ServerStream.SendMsg`, `google.golang.org/grpc v1.83.2`; the generated `PeerEvent` `Send` reaches `SendMsg` through `GenericServerStream`)

There are three separate ways it happens. On top of that, `doBroadcast` keeps a lock held across a blocking `Send`, which is the same shape as #23542 in the txpool - and as described further down, the slow-subscriber case is what produces the concurrent sends at scale.

### Two broadcasts can send on one stream together

`doBroadcast` holds a **read** lock across `Send` (`p2p/sentry/sentry_grpc_server.go:1737`):

```go
func (s *PeersStreams) doBroadcast(reply *sentryproto.PeerEvent) (ids []uint, errs []error) {
s.mu.RLock()
defer s.mu.RUnlock()
for id, stream := range s.streams {
err := stream.Send(reply)
```

`RLock` is shared, so two callers run this loop at the same time over the same map and call `Send` on the same stream. And the callers really do run together - they are on the per-peer goroutine inside `p2p.Protocol.Run` (`sentry_grpc_server.go:729-730`):

```go
ss.sendNewPeerToClients(gointerfaces.ConvertHashToH512(peerID))
defer ss.sendGonePeerToClients(gointerfaces.ConvertHashToH512(peerID))
```

Two peers connecting or dropping at the same moment is enough.

### The replay sends on one stream from N goroutines

This one needs no concurrency from `Broadcast` at all (`sentry_grpc_server.go:1594-1604`):

```go
ss.rangePeers(func(peerInfo *PeerInfo) bool {
if pv := peerInfo.EthProtocol(); pv == 0 || pv != ss.ethVersion {
return true
}
eg.Go(func() error {
return server.Send(&sentryproto.PeerEvent{...})
})
```

One errgroup goroutine per replay-eligible peer - peers still in handshake (`protocol == 0`) or on a different ETH version are skipped - and every one of them sends on the same `server` stream.

### The replay overlaps the live stream

`PeerEvents` registers the stream before it replays (`sentry_grpc_server.go:1591`), so while the replay goroutines are sending, a `Broadcast` from a peer connect or disconnect can be sending on the same stream as well.

### Liveness, and how it feeds back into the violation

`SendMsg` can block once the stream's available flow-control capacity is exhausted - for example after a client stops reading while peer events keep coming. `doBroadcast` holds the read lock across that call, so while it is parked:

- `Add` and `remove` need the write lock, so new subscriptions and completed subscription teardown are blocked. Note the handler may well begin returning; it is its deferred `remove` that waits.
- other broadcasts can independently reach the same slow stream and block in `Send`; depending on map iteration order this can delay healthy subscribers, and it prevents those peer lifecycle calls from completing.
- `sendGonePeerToClients` is deferred inside the peer's `Run`, so a blocked disconnect notification prevents that peer goroutine from exiting. Multiple departing peers then accumulate.

Worth noticing that the second bullet is not only a liveness problem. Because `RLock` is shared, several broadcast goroutines can be parked inside `Send` on the *same* stream at once. So a single subscriber that stops reading does not merely stall things - it is precisely what produces the forbidden concurrent `SendMsg` calls, and the more peer churn there is, the more goroutines pile up on that one stream.

This is not theoretical, `PeerEvents` has real consumers: `execution/p2p/message_listener.go:199`, `p2p/sentry/sentry_multi_client/sentry_multi_client.go:132`, `p2p/sentry/libsentry/sentrymultiplexer.go:567`, `txnprovider/txpool/fetch.go:726`.

### Suggested direction

#23730 added `grpcutil.StreamBroadcaster` for this exact shape. Each subscriber gets a bounded queue that only its own handler goroutine drains, so one goroutine does every `Send` on a stream and `Broadcast` only does non-blocking queue writes. That makes all three violations above structurally impossible rather than merely avoided, and it removes the send-under-lock stall.

But it cannot be a straight swap, and there are two traps worth writing down before someone attempts it.

**Do not replay first and then subscribe.** That looks like the obvious way to stop the replay racing the live stream, but it opens a window in which a peer event happening after the replay and before registration is lost. The current code registers first, which avoids that gap but permits the concurrent sends described above. What is needed is a bootstrap step:

1. register the subscriber and its queue first, so nothing is missed from that moment;
2. take a snapshot of the replay-eligible peers;
3. send the snapshot sequentially on the handler goroutine;
4. then fall into the normal drain loop, which delivers whatever queued during step 3.

Note `StreamBroadcaster.Subscribe` as it stands registers and immediately enters its drain loop with nothing in between, so this needs a small addition to that API, something like a `Subscribe` variant taking a bootstrap function that runs after registration and before the drain.

**Snapshot the peers before sending, not during.** `rangePeers` holds the peer store's read lock across the callback:

```go
func (ss *GrpcServer) rangePeers(f func(peerInfo *PeerInfo) bool) {
store := ss.peers.Load()
store.mu.RLock()
defer store.mu.RUnlock()
for _, peerInfo := range store.peers {
...
cont := f(peerInfo)
```

Today the callback only does `eg.Go(...)` and returns straight away, so the lock is held briefly. If the replay is made sequential inside this callback, every `Send` would run under the peer store read lock, and one slow client would then block peer store writers. So collect the matching peers into a slice under the lock, release it, and send afterwards.

One behaviour point to settle while doing this. With register-then-snapshot, a peer that is included in the snapshot but whose live `Connect` broadcast has not yet been queued can appear both in the snapshot and in the queue, so duplicate `Connect` events remain possible during bootstrap. The window is the gap in `Protocol.Run` between `getOrCreatePeer` inserting the peer under the store write lock and `sendNewPeerToClients` being called a few lines later. They are bounded, but they do not go away. The current code has the same duplicate `Connect` and concurrent-send ordering problem in the replay/live overlap. Duplicates are the safer failure mode compared to losing events, but it is worth confirming the consumers listed above are happy to treat `Connect` as idempotent.

### On testing

The existing replay test, `TestGrpcServer_PeerEvents_ReplayFiltersByVersion` in `p2p/sentry/sentry_grpc_server_test.go:1258`, cannot catch any of this: its `mockPeerEventsStream.Send` serialises itself with a `sync.Mutex`, whereas grpc-go requires the caller to avoid concurrent sends in the first place.

Please do not fix that by simply removing the mutex. Then the mock's own `events` append becomes an unsynchronised slice write, and `-race` would be reporting a bug in the test helper rather than the contract violation we care about.

What is needed is a dedicated stream fake that **asserts** the overlap - for instance an atomic in-flight counter plus a gate to hold one `Send` open, failing if a second `Send` is entered while the first has not returned - with event collection synchronised separately.

And do not rely on `-race` by itself. The grpc-go API explicitly forbids concurrent `SendMsg` calls, but an API-contract violation need not show up as a Go data race. So the test should use a stream fake that directly detects overlapping `Send` calls.

Beyond that, and in the same spirit as the tests in `node/gointerfaces/grpcutil/stream_broadcaster_test.go`: a client that stops reading must not stop delivery to the other subscribers, must not block registration or teardown, and must not keep a peer goroutine from exiting. A direct `PeersStreams` test can cover the slow-send, registration and removal cases deterministically; an integration test can show the peer-run teardown part.

Found while reviewing #23730.