cl/sentinel: one status handshake per peer, and honour the ban on inbound connections

cl/sentinel/discovery.go

@@ -583,31 +583,57 @@ func (s *Sentinel) onConnection(_ network.Network, conn network.Conn) {
 			return
 		}
 
-		valid, err := s.handshaker.ValidatePeer(s.ctx, peerId)
-		if err != nil {
-			// Handshake transport error (stream reset, timeout, etc.) — keep the peer.
-			// The peer may still work for gossip even if status exchange failed.
-			log.Trace("[Sentinel] Handshake transport error (keeping connection)", "peer", peerId, "err", err)
-		}
+		s.exchangeStatus(peerId,
+			func() (bool, error) { return s.handshaker.ValidatePeer(s.ctx, peerId) },
+			s.closePeer,
+			func(id peer.ID) {
+				s.p2p.Host().Peerstore().RemovePeer(id)
+				s.closePeer(id)
+				s.peers.RemovePeer(id)
+			})
+	}()
+}
 
-		if !valid && err == nil {
-			// Handshake succeeded but fork digest mismatched — peer is on a different fork.
-			// Must disconnect to avoid receiving incompatible blocks.
-			log.Debug("[Sentinel] Fork mismatch, disconnecting peer", "peer", peerId)
-			s.p2p.Host().Peerstore().RemovePeer(peerId)
-			s.closePeer(peerId)
-			s.peers.RemovePeer(peerId)
-			return
-		}
+// exchangeStatus runs one status handshake for peerId and reports whether the peer was kept.
+//
+// Only one handshake per peer runs at a time: a burst of connection events would otherwise
+// start one each, all completing before the failure count could reach the pool's ban
+// threshold. The ban is read inside that serialized section, because a handshake that
+// completed while this event waited may have installed it — and ConnectWithPeer only
+// refuses banned peers on dials we initiate, not on events that arrive anyway.
+func (s *Sentinel) exchangeStatus(peerId peer.ID, validate func() (bool, error), closePeer, dropPeer func(peer.ID)) bool {
+	if !s.handshakeGate.tryAcquire(peerId) {
+		return false
+	}
+	defer s.handshakeGate.release(peerId)
 
-		if !valid {
-			// Handshake had a transport error AND returned invalid — keep anyway.
-			s.peers.RecordHandshakeFailure(peerId)
-		} else {
-			// we were able to successfully connect, so add this peer to our pool
-			s.peers.AddPeer(peerId)
+	if s.peers.BanStatus(peerId) {
+		closePeer(peerId)
+		return false
+	}
 
-			log.Trace("[Sentinel] Peer validated and added", "peer", peerId)
-		}
-	}()
+	valid, err := validate()
+	if err != nil {
+		// Handshake transport error (stream reset, timeout, etc.) — keep the peer.
+		// The peer may still work for gossip even if status exchange failed.
+		log.Trace("[Sentinel] Handshake transport error (keeping connection)", "peer", peerId, "err", err)
+	}
+
+	if !valid && err == nil {
+		// Handshake succeeded but fork digest mismatched — peer is on a different fork.
+		// Must disconnect to avoid receiving incompatible blocks.
+		log.Debug("[Sentinel] Fork mismatch, disconnecting peer", "peer", peerId)
+		dropPeer(peerId)
+		return false
+	}
+
+	if !valid {
+		// Handshake had a transport error AND returned invalid — keep anyway.
+		s.peers.RecordHandshakeFailure(peerId)
+		return true
+	}
+	// we were able to successfully connect, so add this peer to our pool
+	s.peers.AddPeer(peerId)
+	log.Trace("[Sentinel] Peer validated and added", "peer", peerId)
+	return true
 }

cl/sentinel/handshake_gate.go

@@ -0,0 +1,61 @@
+// Copyright 2026 The Erigon Authors
+// This file is part of Erigon.
+//
+// Erigon is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// Erigon is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with Erigon. If not, see <http://www.gnu.org/licenses/>.
+
+package sentinel
+
+import (
+	"sync"
+
+	"github.com/libp2p/go-libp2p/core/peer"
+)
+
+// handshakeGate admits one status handshake per peer at a time. Connection events are
+// handled on their own goroutine, so without this a burst of events for one peer starts a
+// handshake for each, and the three-strike ban cannot intervene because every attempt
+// completes before any of them raises the failure count to its threshold.
+type handshakeGate struct {
+	mu       sync.Mutex
+	inflight map[peer.ID]struct{}
+}
+
+func newHandshakeGate() *handshakeGate {
+	return &handshakeGate{inflight: make(map[peer.ID]struct{})}
+}
+
+// tryAcquire reports whether the caller may handshake this peer. A false return means
+// another attempt is already running and this event should be dropped, not queued: the
+// peer is about to be judged by the attempt already in flight.
+func (g *handshakeGate) tryAcquire(pid peer.ID) bool {
+	g.mu.Lock()
+	defer g.mu.Unlock()
+	if _, busy := g.inflight[pid]; busy {
+		return false
+	}
+	g.inflight[pid] = struct{}{}
+	return true
+}
+
+func (g *handshakeGate) release(pid peer.ID) {
+	g.mu.Lock()
+	defer g.mu.Unlock()
+	delete(g.inflight, pid)
+}
+
+func (g *handshakeGate) inFlight() int {
+	g.mu.Lock()
+	defer g.mu.Unlock()
+	return len(g.inflight)
+}

cl/sentinel/handshake_gate_test.go

@@ -0,0 +1,79 @@
+// Copyright 2026 The Erigon Authors
+// This file is part of Erigon.
+//
+// Erigon is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// Erigon is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with Erigon. If not, see <http://www.gnu.org/licenses/>.
+
+package sentinel
+
+import (
+	"sync"
+	"sync/atomic"
+	"testing"
+
+	"github.com/libp2p/go-libp2p/core/peer"
+	"github.com/stretchr/testify/require"
+)
+
+// Every connection event is handled on its own goroutine, so a burst of events for one
+// peer produced a burst of concurrent handshakes. The three-strike ban could not stop it:
+// all of them completed before any pushed the counter to 3. Measured on a gnosis node at
+// 101 attempts against a single peer in two minutes, until libp2p's own dialer refused.
+func TestHandshakeGateAdmitsOneAttemptPerPeerAtATime(t *testing.T) {
+	g := newHandshakeGate()
+	pid := peer.ID("peer-a")
+
+	require.True(t, g.tryAcquire(pid))
+	require.False(t, g.tryAcquire(pid), "a second concurrent attempt on the same peer must be refused")
+
+	g.release(pid)
+	require.True(t, g.tryAcquire(pid), "the peer is attemptable again once the first finishes")
+}
+
+// Serialising per peer must not serialise the whole node: unrelated peers still proceed.
+func TestHandshakeGateDoesNotBlockOtherPeers(t *testing.T) {
+	g := newHandshakeGate()
+
+	require.True(t, g.tryAcquire(peer.ID("peer-a")))
+	require.True(t, g.tryAcquire(peer.ID("peer-b")))
+	require.True(t, g.tryAcquire(peer.ID("peer-c")))
+}
+
+// The burst is the thing being fixed, so drive it concurrently rather than in sequence.
+func TestHandshakeGateUnderConcurrentBurst(t *testing.T) {
+	g := newHandshakeGate()
+	pid := peer.ID("peer-a")
+
+	var admitted atomic.Int64
+	var wg sync.WaitGroup
+	for range 64 {
+		wg.Go(func() {
+			if g.tryAcquire(pid) {
+				admitted.Add(1)
+			}
+		})
+	}
+	wg.Wait()
+
+	require.Equal(t, int64(1), admitted.Load(), "64 simultaneous events admitted %d attempts", admitted.Load())
+}
+
+func TestHandshakeGateForgetsReleasedPeers(t *testing.T) {
+	g := newHandshakeGate()
+	pid := peer.ID("peer-a")
+
+	require.True(t, g.tryAcquire(pid))
+	g.release(pid)
+
+	require.Zero(t, g.inFlight(), "a released peer must not be retained")
+}

cl/sentinel/handshake_sequence_test.go

@@ -0,0 +1,111 @@
+// Copyright 2026 The Erigon Authors
+// This file is part of Erigon.
+//
+// Erigon is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// Erigon is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with Erigon. If not, see <http://www.gnu.org/licenses/>.
+
+package sentinel
+
+import (
+	"errors"
+	"testing"
+
+	"github.com/libp2p/go-libp2p/core/peer"
+	"github.com/stretchr/testify/require"
+
+	"github.com/erigontech/erigon/cl/sentinel/peers"
+)
+
+func testSentinel() *Sentinel {
+	// The ban bookkeeping this exercises never touches the host.
+	return &Sentinel{peers: peers.NewPool(nil), handshakeGate: newHandshakeGate()}
+}
+
+func noop(peer.ID) {}
+
+// The pool bans a peer after three handshake failures. Every connection event must run
+// through the same serialized path, so the third failure is reached and the ban is applied
+// to the next event instead of another handshake being started.
+func TestExchangeStatusReachesTheBanAndThenRefusesThePeer(t *testing.T) {
+	s := testSentinel()
+	pid := peer.ID("peer-a")
+
+	validated, closed, dropped := 0, 0, 0
+	validate := func() (bool, error) {
+		validated++
+		return false, errors.New("stream reset")
+	}
+	onClose := func(peer.ID) { closed++ }
+	onDrop := func(peer.ID) { dropped++ }
+
+	for range 3 {
+		require.True(t, s.exchangeStatus(pid, validate, onClose, onDrop),
+			"a transport error keeps the peer: it may still serve gossip")
+	}
+	require.Equal(t, 3, validated)
+	require.True(t, s.peers.BanStatus(pid), "three handshake failures must ban the peer")
+
+	require.False(t, s.exchangeStatus(pid, validate, onClose, onDrop))
+	require.Equal(t, 3, validated, "a banned peer must not be handshaked again")
+	require.Equal(t, 1, closed)
+	require.Zero(t, dropped)
+	require.Zero(t, s.handshakeGate.inFlight(), "the gate must be released on the banned exit")
+}
+
+// A ban installed by a handshake that completed while this event was waiting must be seen:
+// the ban is read inside the serialized section, not before it.
+func TestExchangeStatusRefusesAPeerBannedWhileTheEventWaited(t *testing.T) {
+	s := testSentinel()
+	pid := peer.ID("peer-a")
+
+	s.peers.SetBanStatus(pid, true)
+	closed := 0
+	require.False(t, s.exchangeStatus(pid, func() (bool, error) {
+		t.Fatal("a banned peer must not be handshaked")
+		return false, nil
+	}, func(peer.ID) { closed++ }, noop))
+	require.Equal(t, 1, closed)
+}
+
+// Serializing per peer is the point: while one handshake is in flight, a second event for
+// the same peer must not start another.
+func TestExchangeStatusAdmitsOneHandshakePerPeerAtATime(t *testing.T) {
+	s := testSentinel()
+	pid := peer.ID("peer-a")
+
+	outer := 0
+	require.True(t, s.exchangeStatus(pid, func() (bool, error) {
+		outer++
+		require.False(t, s.exchangeStatus(pid, func() (bool, error) {
+			t.Fatal("a second handshake started while one was in flight")
+			return false, nil
+		}, noop, noop), "the concurrent event must be refused")
+		return true, nil
+	}, noop, noop))
+	require.Equal(t, 1, outer)
+	require.Zero(t, s.handshakeGate.inFlight())
+}
+
+// A completed handshake reporting the wrong fork is not a failure to retry: the peer is
+// dropped outright and must not count toward the ban.
+func TestExchangeStatusDropsAForkMismatchWithoutRecordingAFailure(t *testing.T) {
+	s := testSentinel()
+	pid := peer.ID("peer-a")
+
+	dropped := 0
+	require.False(t, s.exchangeStatus(pid, func() (bool, error) { return false, nil },
+		noop, func(peer.ID) { dropped++ }))
+	require.Equal(t, 1, dropped)
+	require.False(t, s.peers.BanStatus(pid))
+	require.Zero(t, s.handshakeGate.inFlight(), "the gate must be released on the fork-mismatch exit")
+}

cl/sentinel/sentinel.go

@@ -53,13 +53,14 @@ import (
 )
 
 type Sentinel struct {
-	started  bool
-	listener *discover.UDPv5 // this is us in the network.
-	ctx      context.Context
-	cancel   context.CancelFunc
-	cfg      *SentinelConfig
-	peers    *peers.Pool
-	p2p      p2p.P2PManager
+	started       bool
+	listener      *discover.UDPv5 // this is us in the network.
+	ctx           context.Context
+	cancel        context.CancelFunc
+	cfg           *SentinelConfig
+	peers         *peers.Pool
+	handshakeGate *handshakeGate
+	p2p           p2p.P2PManager
 
 	httpApi http.Handler
 
@@ -143,6 +144,7 @@ func New(
 
 	signal.Reset(syscall.SIGINT)
 	s.peers = peers.NewPool(s.p2p.Host())
+	s.handshakeGate = newHandshakeGate()
 
 	mux := chi.NewRouter()
 	mux.Get("/", httpreqresp.NewRequestHandler(s.p2p.Host()))