eth: retry sync after peer-set replacements (#2391)

Author: vbhattaccmu

eth/peerset.go

@@ -60,6 +60,7 @@ type peerSet struct {
 	peers     map[string]*ethPeer // Peers connected on the `eth` protocol
 	snapPeers int                 // Number of `snap` compatible peers for connection prioritization
 	witPeers  int                 // Number of `wit` compatible peers for connection prioritization
+	revision  uint64              // Peer-set revision
 
 	snapWait map[string]chan *snap.Peer // Peers connected on `eth` waiting for their snap extension
 	snapPend map[string]*snap.Peer      // Peers connected on the `snap` protocol, but not yet on `eth`
@@ -267,6 +268,7 @@ func (ps *peerSet) registerPeer(peer *eth.Peer, extSnap *snap.Peer, extWit *wit.
 	}
 
 	ps.peers[id] = eth
+	ps.revision++
 
 	return nil
 }
@@ -283,6 +285,7 @@ func (ps *peerSet) unregisterPeer(id string) error {
 	}
 
 	delete(ps.peers, id)
+	ps.revision++
 
 	if peer.snapExt != nil {
 		ps.snapPeers--
@@ -448,6 +451,13 @@ func (ps *peerSet) len() int {
 	return len(ps.peers)
 }
 
+func (ps *peerSet) currentRevision() uint64 {
+	ps.lock.RLock()
+	defer ps.lock.RUnlock()
+
+	return ps.revision
+}
+
 // snapLen returns if the current number of `snap` peers in the set.
 func (ps *peerSet) snapLen() int {
 	ps.lock.RLock()

eth/peerset_test.go

@@ -2,18 +2,224 @@ package eth
 
 import (
 	"crypto/rand"
+	"errors"
 	"math/big"
 	"testing"
 	"time"
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/eth/protocols/eth"
+	"github.com/ethereum/go-ethereum/eth/protocols/snap"
 	"github.com/ethereum/go-ethereum/eth/protocols/wit"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/p2p"
 	"github.com/ethereum/go-ethereum/p2p/enode"
 )
 
+func TestPeerSetExtensions(t *testing.T) {
+	t.Run("snap", func(t *testing.T) {
+		caps := []p2p.Cap{{Name: eth.ProtocolName, Version: eth.ETH69}, {Name: snap.ProtocolName, Version: snap.SNAP1}}
+		testPeerSetExtension(t, caps, false, errSnapWithoutEth, newPeerSetSnapPeer, (*peerSet).registerSnapExtension, (*peerSet).waitSnapExtension)
+	})
+	t.Run("wit", func(t *testing.T) {
+		caps := []p2p.Cap{{Name: eth.ProtocolName, Version: eth.ETH69}, {Name: wit.ProtocolName, Version: wit.WIT2}}
+		testPeerSetExtension(t, caps, true, errWitWithoutEth, newPeerSetWitPeer, (*peerSet).registerWitExtension, (*peerSet).waitWitExtension)
+	})
+}
+
+func TestPeerSetRevision(t *testing.T) {
+	ps := newPeerSet()
+	peer := newPeerSetEthPeer(t, enode.ID{1}, nil)
+
+	if err := ps.unregisterPeer(peer.ID()); !errors.Is(err, errPeerNotRegistered) {
+		t.Fatalf("unexpected unregister error: %v", err)
+	}
+	if revision := ps.currentRevision(); revision != 0 {
+		t.Fatalf("failed unregister changed revision to %d", revision)
+	}
+	if err := ps.registerPeer(peer, nil, nil); err != nil {
+		t.Fatal(err)
+	}
+	if err := ps.registerPeer(peer, nil, nil); !errors.Is(err, errPeerAlreadyRegistered) {
+		t.Fatalf("unexpected duplicate registration error: %v", err)
+	}
+	if revision := ps.currentRevision(); revision != 1 {
+		t.Fatalf("registration changed revision to %d", revision)
+	}
+	if err := ps.unregisterPeer(peer.ID()); err != nil {
+		t.Fatal(err)
+	}
+	if revision := ps.currentRevision(); revision != 2 {
+		t.Fatalf("unregister changed revision to %d", revision)
+	}
+	ps.close()
+	if err := ps.registerPeer(peer, nil, nil); !errors.Is(err, errPeerSetClosed) {
+		t.Fatalf("unexpected closed set error: %v", err)
+	}
+	if revision := ps.currentRevision(); revision != 2 {
+		t.Fatalf("failed registration changed revision to %d", revision)
+	}
+}
+
+type extensionResult[T comparable] struct {
+	peer T
+	err  error
+}
+
+func testPeerSetExtension[T comparable](
+	t *testing.T,
+	caps []p2p.Cap,
+	witness bool,
+	incompatibleError error,
+	newExtension func(*testing.T, enode.ID, []p2p.Cap) T,
+	register func(*peerSet, T) error,
+	wait func(*peerSet, *eth.Peer) (T, error),
+) {
+	t.Helper()
+
+	t.Run("extension first", func(t *testing.T) {
+		ps := newPeerSet()
+		defer ps.close()
+
+		id := enode.ID{1}
+		main := newPeerSetEthPeer(t, id, caps)
+		ext := newExtension(t, id, caps)
+		if err := register(ps, ext); err != nil {
+			t.Fatal(err)
+		}
+		if err := register(ps, ext); !errors.Is(err, errPeerAlreadyRegistered) {
+			t.Fatalf("unexpected duplicate registration error: %v", err)
+		}
+		got, err := wait(ps, main)
+		if err != nil || got != ext {
+			t.Fatalf("extension mismatch: got %v, want %v, err %v", got, ext, err)
+		}
+	})
+
+	t.Run("main first", func(t *testing.T) {
+		ps := newPeerSet()
+		defer ps.close()
+
+		id := enode.ID{2}
+		main := newPeerSetEthPeer(t, id, caps)
+		ext := newExtension(t, id, caps)
+		result := make(chan extensionResult[T], 1)
+		go func() {
+			peer, err := wait(ps, main)
+			result <- extensionResult[T]{peer, err}
+		}()
+		waitForPeerSetWaiter(t, ps, id.String(), witness)
+		if err := register(ps, ext); err != nil {
+			t.Fatal(err)
+		}
+		if got := <-result; got.err != nil || got.peer != ext {
+			t.Fatalf("extension mismatch: got %v, want %v, err %v", got.peer, ext, got.err)
+		}
+	})
+
+	t.Run("closed", func(t *testing.T) {
+		ps := newPeerSet()
+		id := enode.ID{3}
+		main := newPeerSetEthPeer(t, id, caps)
+		result := make(chan extensionResult[T], 1)
+		go func() {
+			peer, err := wait(ps, main)
+			result <- extensionResult[T]{peer, err}
+		}()
+		waitForPeerSetWaiter(t, ps, id.String(), witness)
+		ps.close()
+		if got := <-result; !errors.Is(got.err, errPeerSetClosed) {
+			t.Fatalf("unexpected close error: %v", got.err)
+		}
+	})
+
+	t.Run("registered", func(t *testing.T) {
+		ps := newPeerSet()
+		defer ps.close()
+
+		id := enode.ID{4}
+		main := newPeerSetEthPeer(t, id, caps)
+		ext := newExtension(t, id, caps)
+		if err := ps.registerPeer(main, nil, nil); err != nil {
+			t.Fatal(err)
+		}
+		if err := register(ps, ext); !errors.Is(err, errPeerAlreadyRegistered) {
+			t.Fatalf("unexpected extension registration error: %v", err)
+		}
+		if _, err := wait(ps, main); !errors.Is(err, errPeerAlreadyRegistered) {
+			t.Fatalf("unexpected extension wait error: %v", err)
+		}
+	})
+
+	t.Run("incompatible", func(t *testing.T) {
+		ps := newPeerSet()
+		defer ps.close()
+
+		id := enode.ID{5}
+		ext := newExtension(t, id, caps[1:])
+		if err := register(ps, ext); !errors.Is(err, incompatibleError) {
+			t.Fatalf("unexpected extension registration error: %v", err)
+		}
+		main := newPeerSetEthPeer(t, id, caps[:1])
+		var zero T
+		if got, err := wait(ps, main); err != nil || got != zero {
+			t.Fatalf("unexpected extension: got %v, err %v", got, err)
+		}
+	})
+}
+
+func newPeerSetEthPeer(t *testing.T, id enode.ID, caps []p2p.Cap) *eth.Peer {
+	t.Helper()
+	peer, rw := newPeerSetProtocolPeer(t, id, caps)
+	result := eth.NewPeer(eth.ETH69, peer, rw, nil)
+	t.Cleanup(result.Close)
+	return result
+}
+
+func newPeerSetSnapPeer(t *testing.T, id enode.ID, caps []p2p.Cap) *snap.Peer {
+	t.Helper()
+	peer, rw := newPeerSetProtocolPeer(t, id, caps)
+	return snap.NewPeer(snap.SNAP1, peer, rw)
+}
+
+func newPeerSetWitPeer(t *testing.T, id enode.ID, caps []p2p.Cap) *wit.Peer {
+	t.Helper()
+	peer, rw := newPeerSetProtocolPeer(t, id, caps)
+	result := wit.NewPeer(wit.WIT2, peer, rw, log.New())
+	t.Cleanup(result.Close)
+	return result
+}
+
+func newPeerSetProtocolPeer(t *testing.T, id enode.ID, caps []p2p.Cap) (*p2p.Peer, p2p.MsgReadWriter) {
+	t.Helper()
+	app, net := p2p.MsgPipe()
+	t.Cleanup(func() {
+		app.Close()
+		net.Close()
+	})
+	return p2p.NewPeer(id, "test", caps), net
+}
+
+func waitForPeerSetWaiter(t *testing.T, ps *peerSet, id string, witness bool) {
+	t.Helper()
+	deadline := time.Now().Add(time.Second)
+	for time.Now().Before(deadline) {
+		ps.lock.RLock()
+		_, snapReady := ps.snapWait[id]
+		_, witReady := ps.witWait[id]
+		ps.lock.RUnlock()
+		ready := snapReady
+		if witness {
+			ready = witReady
+		}
+		if ready {
+			return
+		}
+		time.Sleep(time.Millisecond)
+	}
+	t.Fatal("extension waiter was not registered")
+}
+
 func TestPeerSetForgetTransactions(t *testing.T) {
 	t.Parallel()
 

eth/sync.go

@@ -61,8 +61,8 @@ type chainSyncer struct {
 	peerEventCh chan struct{}
 	doneCh      chan error // non-nil when sync is running
 
-	peersUnavailableUntil   time.Time
-	peersUnavailableAtCount int
+	peersUnavailableUntil time.Time
+	observedPeerRevision  uint64
 }
 
 // chainSyncOp is a scheduled sync operation.
@@ -112,6 +112,7 @@ func (cs *chainSyncer) loop() {
 	retry := newResettableTimer()
 	defer retry.stop()
 
+	cs.observedPeerRevision = cs.handler.peers.currentRevision()
 	for {
 		if op, wait := cs.nextSyncOp(); op != nil {
 			retry.stop()
@@ -143,7 +144,6 @@ func (cs *chainSyncer) onSyncDone(err error) {
 
 	if errors.Is(err, downloader.ErrPeersUnavailable) || errors.Is(err, downloader.ErrPeerBackedOff) || errors.Is(err, whitelist.ErrNoRemote) {
 		cs.peersUnavailableUntil = time.Now().Add(forceSyncCycle)
-		cs.peersUnavailableAtCount = cs.handler.peers.len()
 	} else {
 		cs.peersUnavailableUntil = time.Time{}
 	}
@@ -159,9 +159,11 @@ func (cs *chainSyncer) onSyncDone(err error) {
 }
 
 func (cs *chainSyncer) onPeerEvent() {
-	if !cs.peersUnavailableUntil.IsZero() && cs.handler.peers.len() != cs.peersUnavailableAtCount {
+	revision := cs.handler.peers.currentRevision()
+	if !cs.peersUnavailableUntil.IsZero() && revision != cs.observedPeerRevision {
 		cs.peersUnavailableUntil = time.Time{}
 	}
+	cs.observedPeerRevision = revision
 }
 
 func (cs *chainSyncer) shutdown() {

eth/sync_test.go

@@ -195,6 +195,7 @@ func TestChainSyncerCooldownSurvivesBlockAnnounce(t *testing.T) {
 	if err := handler.downloader.RegisterPeer(peer.ID(), eth.ETH68, &ethPeer{Peer: peer}); err != nil {
 		t.Fatal(err)
 	}
+	syncer.onPeerEvent()
 
 	syncer.onSyncDone(downloader.ErrPeersUnavailable)
 	if syncer.peersUnavailableUntil.IsZero() {
@@ -210,9 +211,50 @@ func TestChainSyncerCooldownSurvivesBlockAnnounce(t *testing.T) {
 	if err := handler.downloader.RegisterPeer(peer2.ID(), eth.ETH68, &ethPeer{Peer: peer2}); err != nil {
 		t.Fatal(err)
 	}
+	if err := handler.peers.unregisterPeer(peer.ID()); err != nil {
+		t.Fatal(err)
+	}
+	if handler.peers.len() != 1 {
+		t.Fatalf("peer replacement should preserve peer count, have %d", handler.peers.len())
+	}
+	syncer.onSyncDone(downloader.ErrPeersUnavailable)
 	syncer.onPeerEvent()
 	if !syncer.peersUnavailableUntil.IsZero() {
-		t.Fatal("a genuine peer-set change must clear the cooldown")
+		t.Fatal("a peer replacement must clear the cooldown")
+	}
+}
+
+func TestChainSyncerLoopInitializesPeerRevision(t *testing.T) {
+	handler, cleanup := newChainSyncerTestHandler(t)
+	defer cleanup()
+	handler.maxPeers = defaultMinSyncPeers
+
+	peer := registerPeerWithTD(t, handler.peers, 1_000_000)
+	if err := handler.downloader.RegisterPeer(peer.ID(), eth.ETH68, &ethPeer{Peer: peer}); err != nil {
+		t.Fatal(err)
+	}
+
+	syncer := handler.chainSync
+	syncer.peersUnavailableUntil = time.Now().Add(time.Hour)
+	handler.wg.Add(1)
+	done := make(chan struct{})
+	go func() {
+		syncer.loop()
+		close(done)
+	}()
+
+	if !syncer.handlePeerEvent() {
+		t.Fatal("chain syncer stopped before processing the peer event")
+	}
+	close(handler.quitSync)
+
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		t.Fatal("chain syncer did not stop")
+	}
+	if syncer.peersUnavailableUntil.IsZero() {
+		t.Fatal("an initial peer event with no peer-set change must not clear the cooldown")
 	}
 }