diff --git a/cl/sentinel/discovery.go b/cl/sentinel/discovery.go index 1035cfc6837..8bb71e14155 100644 --- a/cl/sentinel/discovery.go +++ b/cl/sentinel/discovery.go @@ -554,9 +554,16 @@ func (s *Sentinel) listenForPeers() { } func (s *Sentinel) onConnection(_ network.Network, conn network.Conn) { - go func() { - peerId := conn.RemotePeer() + peerId := conn.RemotePeer() + go s.handleNewConnection(peerId, func() (bool, error) { + return s.handshaker.ValidatePeer(s.ctx, peerId) + }) +} +// handleNewConnection admits or rejects a peer that has just connected, then runs its status +// handshake. Reports whether the peer was kept. +func (s *Sentinel) handleNewConnection(peerId peer.ID, validate func() (bool, error)) bool { + { // Check if this peer helps any underserved subnets (< minimumPeersPerSubnet) peerHelpsSubnets := false if nodeVal, ok := s.pidToEnr.Load(peerId); ok { @@ -580,34 +587,54 @@ func (s *Sentinel) onConnection(_ network.Network, conn network.Conn) { s.p2p.Host().Peerstore().RemovePeer(peerId) s.closePeer(peerId) s.peers.RemovePeer(peerId) - return + return false } + } - 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) - } + return s.exchangeStatus(peerId, s.peers.BanStatus, validate, + 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. +// The ban must be read while the slot is held: a handshake completing meanwhile may install it. +func (s *Sentinel) exchangeStatus(peerId peer.ID, banned func(peer.ID) bool, 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 banned(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 } diff --git a/cl/sentinel/handshake_gate.go b/cl/sentinel/handshake_gate.go new file mode 100644 index 00000000000..56c2a8d489f --- /dev/null +++ b/cl/sentinel/handshake_gate.go @@ -0,0 +1,66 @@ +// 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 . + +package sentinel + +import ( + "sync" + + "github.com/libp2p/go-libp2p/core/peer" +) + +// handshakeGate admits one status handshake per peer at a time, so a burst of connection +// events cannot outrun the pool's failure count before it reaches the ban 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 the +// event should be dropped, not queued: the attempt in flight will judge the peer. +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) +} + +// holds reports whether a handshake slot for pid is currently taken. +func (g *handshakeGate) holds(pid peer.ID) bool { + g.mu.Lock() + defer g.mu.Unlock() + _, busy := g.inflight[pid] + return busy +} diff --git a/cl/sentinel/handshake_gate_test.go b/cl/sentinel/handshake_gate_test.go new file mode 100644 index 00000000000..f4c0b85addd --- /dev/null +++ b/cl/sentinel/handshake_gate_test.go @@ -0,0 +1,77 @@ +// 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 . + +package sentinel + +import ( + "sync" + "sync/atomic" + "testing" + + "github.com/libp2p/go-libp2p/core/peer" + "github.com/stretchr/testify/require" +) + +// A burst of events for one peer must not become a burst of concurrent handshakes: the +// three-strike ban cannot stop them if they all complete before the counter reaches 3. +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") +} diff --git a/cl/sentinel/handshake_sequence_test.go b/cl/sentinel/handshake_sequence_test.go new file mode 100644 index 00000000000..530e0bc9c3b --- /dev/null +++ b/cl/sentinel/handshake_sequence_test.go @@ -0,0 +1,190 @@ +// 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 . + +package sentinel + +import ( + "errors" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/libp2p/go-libp2p/core/peer" + "github.com/stretchr/testify/require" + + "github.com/erigontech/erigon/cl/p2p" + "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) {} + +// banObserver wraps the pool's BanStatus and counts every read that happens without this +// peer's handshake slot being held. Reading the ban before admission is the defect under +// test, and it is invisible to a test that only checks outcomes. Violations are counted +// rather than asserted on the spot: this runs on handshake goroutines, where a failed +// require would Goexit and deadlock the test instead of reporting. +type banObserver struct { + s *Sentinel + violations atomic.Int64 +} + +func (b *banObserver) read(pid peer.ID) bool { + if !b.s.handshakeGate.holds(pid) { + b.violations.Add(1) + } + return b.s.peers.BanStatus(pid) +} + +func (b *banObserver) assertAlwaysHeld(t *testing.T) { + t.Helper() + require.Zero(t, b.violations.Load(), + "the ban must be read while this peer's handshake slot is held, not before admission") +} + +func waitForInFlight(t *testing.T, s *Sentinel, want int) { + t.Helper() + deadline := time.Now().Add(2 * time.Second) + for s.handshakeGate.inFlight() != want { + if time.Now().After(deadline) { + t.Fatalf("timed out waiting for %d handshake(s) in flight, have %d", want, s.handshakeGate.inFlight()) + } + time.Sleep(time.Millisecond) + } +} + +// The full sequence: one handshake paused mid-flight, a second event for the same peer +// arriving while it runs, an unrelated peer proceeding regardless, three failures reaching +// the pool ban, and the next event for the banned peer closing it without a handshake. +func TestHandshakeSequenceSerializesPerPeerAndReachesTheBan(t *testing.T) { + s := testSentinel() + obs := &banObserver{s: s} + banned := obs.read + pidA, pidB := peer.ID("peer-a"), peer.ID("peer-b") + + var validatedA, validatedB, closedA int + var mu sync.Mutex + countA := func() { mu.Lock(); validatedA++; mu.Unlock() } + + resume := make(chan struct{}) + entered := make(chan struct{}) + // Unblock the paused handshake unconditionally: an assertion below reaching FailNow + // before the explicit close would otherwise leave its goroutine parked forever. + var once sync.Once + release := func() { once.Do(func() { close(resume) }) } + t.Cleanup(release) + + var wg sync.WaitGroup + wg.Go(func() { + // First event for peer A: holds the slot until released. + s.exchangeStatus(pidA, banned, func() (bool, error) { + countA() + close(entered) + <-resume + return false, errors.New("stream reset") + }, noop, noop) + }) + + <-entered + waitForInFlight(t, s, 1) + + // A second event for the same peer, arriving while the first handshake is in flight. + require.False(t, s.exchangeStatus(pidA, banned, func() (bool, error) { + t.Error("a second handshake started for a peer already being handshaked") + return false, nil + }, noop, noop)) + + // An unrelated peer must not be held up by peer A's in-flight handshake. + require.True(t, s.exchangeStatus(pidB, banned, func() (bool, error) { + validatedB++ + return true, nil + }, noop, noop)) + require.Equal(t, 1, validatedB) + + release() + wg.Wait() + + // Two more failures reach the pool's three-strike ban. + for range 2 { + require.True(t, s.exchangeStatus(pidA, banned, func() (bool, error) { + countA() + return false, errors.New("stream reset") + }, noop, noop)) + } + require.Equal(t, 3, validatedA) + require.True(t, s.peers.BanStatus(pidA), "three handshake failures must ban the peer") + + // The next event closes the banned peer instead of handshaking it again. + require.False(t, s.exchangeStatus(pidA, banned, func() (bool, error) { + t.Error("a banned peer must not be handshaked") + return false, nil + }, func(peer.ID) { closedA++ }, noop)) + require.Equal(t, 3, validatedA) + require.Equal(t, 1, closedA) + require.Zero(t, s.handshakeGate.inFlight(), "every exit path must release the slot") + obs.assertAlwaysHeld(t) +} + +// The connection handler must route through the serialized path: a banned peer entering +// handleNewConnection is closed without its handshake being attempted. +func TestHandleNewConnectionRefusesABannedPeerWithoutHandshaking(t *testing.T) { + s := testSentinel() + s.p2p = stubP2P{host: newTestHost(t)} + s.cfg = &SentinelConfig{P2PConfig: p2p.P2PConfig{MaxPeerCount: 100}} + pidA := peer.ID("peer-a") + s.peers.SetBanStatus(pidA, true) + + require.False(t, s.handleNewConnection(pidA, func() (bool, error) { + t.Error("a banned peer must not be handshaked") + return false, nil + })) + require.Zero(t, s.handshakeGate.inFlight()) +} + +// A peer that is not banned reaches its handshake through the handler. +func TestHandleNewConnectionHandshakesAnUnbannedPeer(t *testing.T) { + s := testSentinel() + s.p2p = stubP2P{host: newTestHost(t)} + s.cfg = &SentinelConfig{P2PConfig: p2p.P2PConfig{MaxPeerCount: 100}} + + validated := 0 + require.True(t, s.handleNewConnection(peer.ID("peer-a"), func() (bool, error) { + validated++ + return true, nil + })) + require.Equal(t, 1, validated) +} + +// A completed handshake reporting the wrong fork is dropped and must not count toward the ban. +func TestExchangeStatusDropsAForkMismatchWithoutRecordingAFailure(t *testing.T) { + s := testSentinel() + pid := peer.ID("peer-a") + + dropped := 0 + obs := &banObserver{s: s} + require.False(t, s.exchangeStatus(pid, obs.read, + 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 fork-mismatch exit must release the slot") + obs.assertAlwaysHeld(t) +} diff --git a/cl/sentinel/sentinel.go b/cl/sentinel/sentinel.go index a844e405e76..5274a5c098e 100644 --- a/cl/sentinel/sentinel.go +++ b/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())) diff --git a/cl/sentinel/testhost_test.go b/cl/sentinel/testhost_test.go new file mode 100644 index 00000000000..63d3613779b --- /dev/null +++ b/cl/sentinel/testhost_test.go @@ -0,0 +1,47 @@ +// 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 . + +package sentinel + +import ( + "testing" + + "github.com/libp2p/go-libp2p" + pubsub "github.com/libp2p/go-libp2p-pubsub" + "github.com/libp2p/go-libp2p/core/host" + "github.com/libp2p/go-libp2p/core/metrics" + "github.com/stretchr/testify/require" + + "github.com/erigontech/erigon/p2p/discover" +) + +// stubP2P satisfies p2p.P2PManager with only the host the connection handler needs. +type stubP2P struct{ host host.Host } + +func (s stubP2P) Pubsub() *pubsub.PubSub { return nil } +func (s stubP2P) Host() host.Host { return s.host } +func (s stubP2P) BandwidthCounter() *metrics.BandwidthCounter { return nil } +func (s stubP2P) UDPv5Listener() *discover.UDPv5 { return nil } +func (s stubP2P) UpdateENRAttSubnets(subnetIndex int, on bool) {} +func (s stubP2P) UpdateENRSyncNets(subnetIndex int, on bool) {} + +func newTestHost(t *testing.T) host.Host { + t.Helper() + h, err := libp2p.New(libp2p.NoListenAddrs) + require.NoError(t, err) + t.Cleanup(func() { _ = h.Close() }) + return h +}