diff --git a/.changeset/compared_header_chain_work_before_downloading_blocks_from_a_peer.md b/.changeset/compared_header_chain_work_before_downloading_blocks_from_a_peer.md new file mode 100644 index 00000000..3bdf4eac --- /dev/null +++ b/.changeset/compared_header_chain_work_before_downloading_blocks_from_a_peer.md @@ -0,0 +1,7 @@ +--- +default: major +--- + +# Compared header chain work before downloading blocks from a peer + +The syncer now walks a peer's headers until their chain either outweighs ours or runs out, and only downloads blocks once it knows the chain is worth adopting. Previously a peer stuck on a fork that would never outweigh our chain had its entire fork re-downloaded every sync interval. diff --git a/syncer/peer.go b/syncer/peer.go index e763aaf0..3c1fbb1c 100644 --- a/syncer/peer.go +++ b/syncer/peer.go @@ -140,20 +140,20 @@ func (p *Peer) DiscoverIP(timeout time.Duration) (string, error) { } // SendHeaders requests up to n headers from p, starting from the supplied -// index, which must be on the peer's best chain. The peer also returns the -// number of remaining headers left to sync. -func (p *Peer) SendHeaders(cs consensus.State, maxHeaders uint64, timeout time.Duration) ([]types.BlockHeader, uint64, error) { +// index, which must be on the peer's best chain. It also returns the state +// after applying them and the number of headers the peer still has. +func (p *Peer) SendHeaders(cs consensus.State, maxHeaders uint64, timeout time.Duration) ([]types.BlockHeader, consensus.State, uint64, error) { r := &gateway.RPCSendHeaders{Index: cs.Index, Max: maxHeaders} err := p.callRPC(r, timeout) if err == nil { for _, bh := range r.Headers { if err := consensus.ValidateHeader(cs, bh); err != nil { - return nil, 0, fmt.Errorf("peer sent invalid header %v: %w", bh.ID(), err) + return nil, consensus.State{}, 0, fmt.Errorf("peer sent invalid header %v: %w", bh.ID(), err) } cs = consensus.ApplyHeader(cs, bh, time.Time{}) } } - return r.Headers, r.Remaining, err + return r.Headers, cs, r.Remaining, err } // SendTransactions requests a subset of a block's transactions from the peer. diff --git a/syncer/syncer.go b/syncer/syncer.go index ec854d22..0c907d31 100644 --- a/syncer/syncer.go +++ b/syncer/syncer.go @@ -781,6 +781,154 @@ func (s *Syncer) peerLoop(ctx context.Context) error { return nil } +// a headerBatch is one SendHeaders response and the state it extends. +type headerBatch struct { + cs consensus.State + headers []types.BlockHeader +} + +// a peerChain is the result of walking a peer's header chain. +type peerChain struct { + fork consensus.State // state at the fork point + tip types.BlockHeader // last header needed to outweigh our chain + batchTips []types.BlockID // last header of each batch, to detect a reorg + remaining uint64 // headers the peer has beyond tip + heavier bool + retained []headerBatch // the walked batches, if kept +} + +// batches is the number of SendHeaders calls needed to reach tip. +func (pc peerChain) batches() int { return len(pc.batchTips) } + +// headersRetained reports whether every walked batch was kept. +func (pc peerChain) headersRetained() bool { return len(pc.retained) == len(pc.batchTips) } + +const ( + // retainedBatches bounds how many batches a walk keeps + retainedBatches = 2 + + // maxHeaderWalk bounds one peer's walk; a sync round waits for every peer, + // and a peer that cannot finish within it is dropped + maxHeaderWalk = 2 * time.Minute + + // maxChainSync backstops a wedged download; a sync round waits for every + // peer, and parallelSync only bounds individual requests + maxChainSync = 15 * time.Minute +) + +var ( + // errWalkAbandoned wraps the reasons a walk stops without a verdict; the + // peer stays eligible and is retried next tick + errWalkAbandoned = errors.New("header walk abandoned") + errPeerReorg = fmt.Errorf("%w: peer reorged", errWalkAbandoned) + + // errPeerHeaders marks a failure attributable to the peer, unlike a block + // download, which may have involved any peer + errPeerHeaders = errors.New("peer failed to serve headers") + errWalkTimeout = fmt.Errorf("%w: walk exceeded %v", errPeerHeaders, maxHeaderWalk) +) + +// walkPeerChain requests headers from p until its chain outweighs ts or it runs +// out of them, reporting whether the chain is worth downloading blocks for. +func (s *Syncer) walkPeerChain(p *Peer, hist [32]types.BlockID, ts consensus.State) (peerChain, error) { + // the deadline covers history probing too, and clamps each request + deadline := time.Now().Add(maxHeaderWalk) + sendHeaders := func(cs consensus.State) ([]types.BlockHeader, consensus.State, uint64, error) { + timeout := min(s.config.SendHeadersTimeout, time.Until(deadline)) + if timeout <= 0 { + return nil, consensus.State{}, 0, errWalkTimeout + } + headers, tip, remaining, err := p.SendHeaders(cs, s.config.MaxSendHeaders, timeout) + if err != nil && !time.Now().Before(deadline) { + return nil, consensus.State{}, 0, errWalkTimeout + } + return headers, tip, remaining, err + } + for _, id := range hist { + if id == (types.BlockID{}) { + // skip empty history entries which can occur when we don't have a + // full history of blocks. + continue + } + fork, ok := s.cm.State(id) + if !ok { + return peerChain{}, errors.New("missing state for history") + } + headers, tip, remaining, err := sendHeaders(fork) + if err != nil && strings.Contains(err.Error(), "EOF") { + continue // probably "index is not on our best chain" + } else if err != nil { + return peerChain{}, err + } + pc, cs := peerChain{fork: fork}, fork + for { + if len(headers) > 0 { + pc.tip = headers[len(headers)-1] + pc.batchTips = append(pc.batchTips, pc.tip.ID()) + if len(pc.batchTips) <= retainedBatches { + pc.retained = append(pc.retained, headerBatch{cs: cs, headers: headers}) + } else { + pc.retained = nil + } + } + pc.remaining = remaining + if tip.SufficientlyHeavierThan(ts) { + pc.heavier = true + return pc, nil + } else if remaining == 0 || len(headers) == 0 { + return pc, nil + } + cs = tip + if headers, tip, remaining, err = sendHeaders(cs); err != nil { + if strings.Contains(err.Error(), "EOF") { + return peerChain{}, errPeerReorg + } + return peerChain{}, err + } + } + } + return peerChain{}, errors.New("no common history") +} + +// syncChain downloads the blocks of the chain pc describes, returning how much +// was actually synced. p only serves headers the walk did not retain; blocks +// come from every eligible peer via parallelSync. +func (s *Syncer) syncChain(ctx context.Context, p *Peer, pc peerChain) (peerChain, error) { + ctx, cancel := context.WithTimeout(ctx, maxChainSync) + defer cancel() + + synced := peerChain{fork: pc.fork} + if pc.headersRetained() { + for _, b := range pc.retained { + if err := s.parallelSync(ctx, b.cs, b.headers); err != nil { + return synced, err + } + synced.tip = b.headers[len(b.headers)-1] + synced.batchTips = append(synced.batchTips, synced.tip.ID()) + } + synced.remaining = pc.remaining + return synced, nil + } + cs := pc.fork + for _, want := range pc.batchTips { + headers, tip, remaining, err := p.SendHeaders(cs, s.config.MaxSendHeaders, s.config.SendHeadersTimeout) + if err != nil && strings.Contains(err.Error(), "EOF") { + return synced, errPeerReorg + } else if err != nil { + return synced, fmt.Errorf("%w: %w", errPeerHeaders, err) + } else if len(headers) == 0 || headers[len(headers)-1].ID() != want { + // they are no longer serving the chain we judged heavier + return synced, errPeerReorg + } else if err := s.parallelSync(ctx, cs, headers); err != nil { + return synced, err + } + cs = tip + synced.tip, synced.remaining = headers[len(headers)-1], remaining + synced.batchTips = append(synced.batchTips, synced.tip.ID()) + } + return synced, nil +} + func (s *Syncer) syncLoop(ctx context.Context) error { ticker := time.NewTicker(s.config.SyncInterval) defer ticker.Stop() @@ -801,62 +949,47 @@ func (s *Syncer) syncLoop(ctx context.Context) error { } s.mu.Unlock() type resp struct { - peer *Peer - cs consensus.State - headers []types.BlockHeader - remaining uint64 - err error + peer *Peer + chain peerChain + err error } respChan := make(chan resp, len(peers)) hist, err := s.cm.History() if err != nil { return err // generally fatal } + tipState := s.cm.TipState() for _, p := range peers { go func(p *Peer) { - cs, headers, remaining, err := func() (consensus.State, []types.BlockHeader, uint64, error) { - for _, id := range hist { - if id == (types.BlockID{}) { - // skip empty history entries which can occur when - // we don't have a full history of blocks. - continue - } - cs, ok := s.cm.State(id) - if !ok { - return consensus.State{}, nil, 0, errors.New("missing state for history") - } - headers, remaining, err := p.SendHeaders(cs, s.config.MaxSendHeaders, s.config.SendHeadersTimeout) - if err != nil && strings.Contains(err.Error(), "EOF") { - continue // probably "index is not on our best chain" - } else if err != nil { - return consensus.State{}, nil, 0, err - } - return cs, headers, remaining, nil - } - return consensus.State{}, nil, 0, errors.New("no common history") - }() - respChan <- resp{peer: p, cs: cs, headers: headers, remaining: remaining, err: err} + pc, err := s.walkPeerChain(p, hist, tipState) + respChan <- resp{peer: p, chain: pc, err: err} }(p) } // sync each set of headers as they arrive seen := make(map[types.BlockID]bool) for range peers { - if r := <-respChan; r.err != nil { + if r := <-respChan; errors.Is(r.err, errWalkAbandoned) { + s.log.Debug("abandoned header walk", zap.Stringer("peer", r.peer), zap.Error(r.err)) + } else if r.err != nil { r.peer.setErr(r.err) - } else if len(r.headers) == 0 { + } else if !r.chain.heavier { + // a later relay from them triggers a resync + s.log.Debug("peer chain does not outweigh ours", zap.Stringer("peer", r.peer), zap.Int("batches", r.chain.batches())) r.peer.setSynced(true) - } else if id := r.headers[len(r.headers)-1].ID(); seen[id] { + } else if id := r.chain.tip.ID(); seen[id] { continue // already syncing these blocks from another peer } else { seen[id] = true - s.log.Debug("syncing blocks", zap.Stringer("peer", r.peer), zap.Stringer("start", r.cs.Index), zap.Int("n", len(r.headers))) - if err := s.parallelSync(ctx, r.cs, r.headers); err != nil { + s.log.Debug("syncing blocks", zap.Stringer("peer", r.peer), zap.Stringer("start", r.chain.fork.Index), zap.Int("batches", r.chain.batches())) + synced, err := s.syncChain(ctx, r.peer, r.chain) + if errors.Is(err, errPeerHeaders) { + r.peer.setErr(err) + } else if err != nil { s.log.Debug("sync failed", zap.Stringer("peer", r.peer), zap.Error(err)) - } else if r.remaining == 0 { - // peer sent all their headers; mark them as synced and - // relay their tip + } else if synced.batches() > 0 && synced.remaining == 0 { + // peer sent all their headers r.peer.setSynced(true) - go s.relayV2Header(r.headers[len(r.headers)-1], r.peer) + go s.relayV2Header(synced.tip, r.peer) } } } diff --git a/syncer/syncer_test.go b/syncer/syncer_test.go index af52d605..499b9e61 100644 --- a/syncer/syncer_test.go +++ b/syncer/syncer_test.go @@ -107,8 +107,9 @@ func TestSyncer(t *testing.T) { s2, cm2 := newTestSyncer(t, syncer.WithLogger(log.Named("syncer2"))) defer s2.Close() - // mine enough blocks to test both v1 and v2 regimes - testutil.MineBlocks(t, cm1, types.VoidAddress, int(cm1.TipState().Network.HardforkV2.RequireHeight+100)) + // mine enough blocks to cover both v1 and v2 regimes, and to split the + // download into enough requests that the bad peer is not starved of work + testutil.MineBlocks(t, cm1, types.VoidAddress, int(cm1.TipState().Network.HardforkV2.RequireHeight+1000)) if _, err := s1.Connect(context.Background(), s2.Addr()); err != nil { t.Fatal(err) @@ -151,8 +152,9 @@ func TestSyncWithBadPeer(t *testing.T) { s2, cm2 := newTestSyncer(t, syncer.WithLogger(log.Named("syncer2"))) defer s2.Close() - // mine enough blocks to test both v1 and v2 regimes - testutil.MineBlocks(t, cm1, types.VoidAddress, int(cm1.TipState().Network.HardforkV2.RequireHeight+100)) + // mine enough blocks to cover both v1 and v2 regimes, and to split the + // download into enough requests that the bad peer is not starved of work + testutil.MineBlocks(t, cm1, types.VoidAddress, int(cm1.TipState().Network.HardforkV2.RequireHeight+1000)) // simulate another peer, one that returns invalid blocks _, genesis := testutil.Network() @@ -186,7 +188,13 @@ func TestSyncWithBadPeer(t *testing.T) { if cm1.Tip() != cm2.Tip() { t.Fatalf("tips are not equal: %v != %v", cm1.Tip(), cm2.Tip()) } - // bad peer should be banned + // the ban propagates asynchronously, so wait for the peer to be dropped + for range 100 { + if len(s2.Peers()) == 1 { + break + } + time.Sleep(100 * time.Millisecond) + } if peers := s2.Peers(); len(peers) != 1 { t.Fatalf("expected 1 peer, got %v", peers) } else if peers[0].UniqueID() == badID { @@ -326,7 +334,7 @@ func TestSendHeaders(t *testing.T) { if err != nil { t.Fatal(err) } - headers, rem, err := p.SendHeaders(cs, 90, time.Second) + headers, _, rem, err := p.SendHeaders(cs, 90, time.Second) if err != nil { t.Fatal(err) } else if len(headers) != 90 { @@ -862,3 +870,168 @@ func TestMaxInflightRPCsBackpressureNotDropped(t *testing.T) { t.Fatalf("backpressured RPC failed after the slot was freed: %v", err) } } + +// forkManager serves a limited number of headers per request and counts +// requests, simulating a peer whose chain spans several sync batches. +type forkManager struct { + headerLimit uint64 + headerReqs atomic.Uint64 + blockReqs atomic.Uint64 + *chain.Manager +} + +func (fm *forkManager) Headers(index types.ChainIndex, maxHeaders uint64) ([]types.BlockHeader, uint64, error) { + fm.headerReqs.Add(1) + return fm.Manager.Headers(index, min(maxHeaders, fm.headerLimit)) +} + +func (fm *forkManager) BlocksForHistory(history []types.BlockID, maxBlocks uint64) ([]types.Block, uint64, error) { + fm.blockReqs.Add(1) + return fm.Manager.BlocksForHistory(history, maxBlocks) +} + +// newForkPeer starts a syncer serving at most headerLimit headers per request +// from a chain of the given length, mined to addr. Its own sync loop is +// disabled so it never adopts our chain. +func newForkPeer(t testing.TB, headerLimit uint64, addr types.Address, blocks int) (*syncer.Syncer, *forkManager) { + t.Helper() + + n, genesis := testutil.Network() + store, err := chain.NewDBStore(chain.NewMemDB(), n, genesis, nil) + if err != nil { + t.Fatal(err) + } + fm := &forkManager{headerLimit: headerLimit, Manager: chain.NewManager(store)} + testutil.MineBlocks(t, fm.Manager, addr, blocks) + + l, err := net.Listen("tcp", ":0") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { l.Close() }) + + s := syncer.New(l, fm, testutil.NewEphemeralPeerStore(), gateway.Header{ + GenesisID: genesis.ID(), + UniqueID: gateway.GenerateUniqueID(), + NetAddress: l.Addr().String(), + }, syncer.WithSyncInterval(time.Hour)) // effectively disabled + go s.Run() + t.Cleanup(func() { s.Close() }) + return s, fm +} + +// waitForTip blocks until cm reaches want. +func waitForTip(t testing.TB, cm *chain.Manager, want types.ChainIndex, msg string) { + t.Helper() + for range 100 { + if cm.Tip() == want { + return + } + time.Sleep(100 * time.Millisecond) + } + t.Fatalf("%s: expected tip %v, got %v", msg, want, cm.Tip()) +} + +// TestForkPeerNotResynced verifies that we stop syncing from a peer on a fork +// that never becomes our best chain: we walk their headers once, download no +// blocks, and don't walk them again every sync interval. +func TestForkPeerNotResynced(t *testing.T) { + log := zaptest.NewLogger(t) + + // s1 has the longer chain + s1, cm1 := newTestSyncer(t, syncer.WithLogger(log.Named("syncer1"))) + defer s1.Close() + testutil.MineBlocks(t, cm1, types.VoidAddress, 20) + + // s2 is on a shorter fork, serving 5 headers at a time so it always reports + // headers remaining + s2, fm := newForkPeer(t, 5, types.Address{1}, 10) + + s1Tip := cm1.Tip() + + if _, err := s1.Connect(context.Background(), s2.Addr()); err != nil { + t.Fatal(err) + } + + // wait for s1 to walk the fork peer's headers + for range 100 { + if fm.headerReqs.Load() > 0 { + break + } + time.Sleep(100 * time.Millisecond) + } + if fm.headerReqs.Load() == 0 { + t.Fatal("s1 never requested headers from the fork peer") + } + + // allow any in-flight rounds to settle + time.Sleep(time.Second) + settledHeaders := fm.headerReqs.Load() + time.Sleep(2 * time.Second) // ~20 sync intervals + if reqs := fm.headerReqs.Load(); reqs != settledHeaders { + t.Fatalf("s1 kept starting sync rounds with the fork peer: %v more header requests", reqs-settledHeaders) + } + + if reqs := fm.blockReqs.Load(); reqs != 0 { + t.Fatalf("s1 downloaded %v block batches from a chain that never outweighed ours", reqs) + } + + peers := s1.Peers() + if len(peers) != 1 { + t.Fatalf("expected 1 peer, got %v", len(peers)) + } else if !peers[0].Synced() { + t.Fatal("fork peer that did not extend our chain should be marked synced") + } + + if cm1.Tip() != s1Tip { + t.Fatalf("s1 tip should not have changed: expected %v, got %v", s1Tip, cm1.Tip()) + } + + // our tip moving is not a reason to ask again; only a relay from them is + testutil.MineBlocks(t, cm1, types.VoidAddress, 1) + time.Sleep(time.Second) // ~10 sync intervals + if reqs := fm.headerReqs.Load(); reqs != settledHeaders { + t.Fatalf("s1 re-synced the fork peer after its own tip moved: %v more header requests", reqs-settledHeaders) + } else if reqs := fm.blockReqs.Load(); reqs != 0 { + t.Fatalf("s1 downloaded %v block batches after its own tip moved", reqs) + } +} + +// TestSyncAcrossBatches verifies that we keep syncing from a peer with more +// headers than a single batch can carry. +func TestSyncAcrossBatches(t *testing.T) { + log := zaptest.NewLogger(t) + + // s1 starts at genesis + s1, cm1 := newTestSyncer(t, syncer.WithLogger(log.Named("syncer1"))) + defer s1.Close() + + // s2 has 20 blocks but serves 5 headers at a time, so catching up takes + // several batches + s2, fm := newForkPeer(t, 5, types.VoidAddress, 20) + + if _, err := s1.Connect(context.Background(), s2.Addr()); err != nil { + t.Fatal(err) + } + waitForTip(t, cm1, fm.Manager.Tip(), "s1 did not catch up") +} + +// TestHeavierForkAcrossBatches verifies that we adopt a peer's chain that is +// heavier than ours overall, even though its first batch of headers is not. +func TestHeavierForkAcrossBatches(t *testing.T) { + log := zaptest.NewLogger(t) + + // s1 has a short chain of its own + s1, cm1 := newTestSyncer(t, syncer.WithLogger(log.Named("syncer1"))) + defer s1.Close() + testutil.MineBlocks(t, cm1, types.VoidAddress, 10) + + // s2's fork is heavier, but its first 5-header batch is shorter than s1's + // chain and does not trigger a reorg + s2, fm := newForkPeer(t, 5, types.Address{1}, 30) + + if _, err := s1.Connect(context.Background(), s2.Addr()); err != nil { + t.Fatal(err) + } + waitForTip(t, cm1, fm.Manager.Tip(), "s1 did not adopt the heavier chain") +}