From 2a439ba45240d6bb584d64f0fd0e9a190c7a8358 Mon Sep 17 00:00:00 2001 From: rjl493456442 Date: Thu, 6 Aug 2026 20:36:01 +0800 Subject: [PATCH] eth/downloader: implement BAL downloading (#35386) This PR implements the BAL downloader. Once the Amsterdam fork is enabled, BALs are scheduled for download for BAL-eligible blocks. Unlike mandatory components such as block bodies, BALs are optional and are downloaded on a best-effort basis. If a block's essential components are ready for delivery before its BAL has been retrieved, the block will be delivered without the BAL. --- core/rawdb/accessors_chain.go | 17 +- eth/downloader/downloader.go | 61 ++++- eth/downloader/downloader_test.go | 280 ++++++++++++++++++++- eth/downloader/fetchers_concurrent.go | 2 +- eth/downloader/fetchers_concurrent_bals.go | 114 +++++++++ eth/downloader/metrics.go | 5 + eth/downloader/peer.go | 62 ++++- eth/downloader/queue.go | 267 +++++++++++++++++++- eth/downloader/queue_test.go | 116 ++++++++- eth/downloader/resultstore.go | 4 +- eth/downloader/skeleton_test.go | 4 + 11 files changed, 897 insertions(+), 35 deletions(-) create mode 100644 eth/downloader/fetchers_concurrent_bals.go diff --git a/core/rawdb/accessors_chain.go b/core/rawdb/accessors_chain.go index e82a15877e..5822f27924 100644 --- a/core/rawdb/accessors_chain.go +++ b/core/rawdb/accessors_chain.go @@ -768,12 +768,17 @@ func writeAncientBlock(op ethdb.AncientWriteOp, block *types.Block, header *type if err := op.Append(ChainFreezerReceiptTable, num, receipts); err != nil { return fmt.Errorf("can't append block %d receipts: %v", num, err) } - // The assumption is held that BAL of ancient block is no longer available - // (it may still reachable, but it's not worthwhile to even retrieve it - // from the network). A nil entry is stored in the BAL table as the absence - // placeholder. - if err := op.AppendRaw(ChainFreezerBALTable, num, nil); err != nil { - return fmt.Errorf("can't append block %d bals: %v", num, err) + // Block access lists are only retrieved (best effort) for blocks close to + // the head of the network chain; a nil entry is stored as the absence + // placeholder for anything the network no longer serves. + if list := block.AccessList(); list != nil { + if err := op.Append(ChainFreezerBALTable, num, list); err != nil { + return fmt.Errorf("can't append block %d bals: %v", num, err) + } + } else { + if err := op.AppendRaw(ChainFreezerBALTable, num, nil); err != nil { + return fmt.Errorf("can't append block %d bals: %v", num, err) + } } return nil } diff --git a/eth/downloader/downloader.go b/eth/downloader/downloader.go index d1acb7795b..351e01d765 100644 --- a/eth/downloader/downloader.go +++ b/eth/downloader/downloader.go @@ -44,6 +44,7 @@ var ( MaxBlockFetch = 128 // Number of blocks to be fetched per retrieval request MaxHeaderFetch = 192 // Number of block headers to be fetched per retrieval request MaxReceiptFetch = 256 // Number of transaction receipts to allow fetching per request + MaxBALFetch = 128 // Number of block access lists to allow fetching per request maxQueuedHeaders = 32 * 1024 // [eth/62] Maximum number of headers to queue for import (DOS protection) maxHeadersProcess = 2048 // Number of header download results to import at once into the chain @@ -65,6 +66,7 @@ var ( errInvalidChain = errors.New("retrieved hash chain is invalid") errInvalidBody = errors.New("retrieved block body is invalid") errInvalidReceipt = errors.New("retrieved receipt is invalid") + errInvalidBAL = errors.New("retrieved block access list is invalid") errCancelStateFetch = errors.New("state data download canceled (requested)") errCancelContentProcessing = errors.New("content processing canceled (requested)") errCanceled = errors.New("syncing canceled (requested)") @@ -156,6 +158,7 @@ type Downloader struct { // Testing hooks bodyFetchHook func([]*types.Header) // Method to call upon starting a block body fetch receiptFetchHook func([]*types.Header) // Method to call upon starting a receipt fetch + balFetchHook func([]*types.Header) // Method to call upon starting a block access list fetch chainInsertHook func([]*fetchResult) // Method to call upon inserting a chain of blocks (possibly in multiple invocations) // Progress reporting metrics @@ -408,7 +411,7 @@ func (d *Downloader) synchronise(beaconPing chan struct{}) (err error) { d.queue.Reset(blockCacheMaxItems, blockCacheInitialItems) d.peers.Reset() - for _, ch := range []chan bool{d.queue.blockWakeCh, d.queue.receiptWakeCh} { + for _, ch := range []chan bool{d.queue.blockWakeCh, d.queue.receiptWakeCh, d.queue.balWakeCh} { select { case <-ch: default: @@ -628,6 +631,7 @@ func (d *Downloader) syncToHead() (err error) { func() error { return d.fetchHeaders(origin + 1) }, // Headers are always retrieved func() error { return d.fetchBodies(chainOffset) }, // Bodies are retrieved during normal and snap sync func() error { return d.fetchReceipts(chainOffset) }, // Receipts are retrieved during snap sync + func() error { return d.fetchBALs(chainOffset) }, // Access lists are retrieved best effort for the chain tail func() error { return d.processHeaders(origin + 1) }, } if mode == ethconfig.SnapSync { @@ -742,6 +746,19 @@ func (d *Downloader) fetchReceipts(from uint64) error { return err } +// fetchBALs iteratively downloads the scheduled block access lists, taking any +// available peers, reserving a chunk of access lists for each, waiting for +// delivery and also periodically checking for timeouts. Access lists are a +// best-effort component: blocks are imported without one if it does not arrive +// by the time all their mandatory components are downloaded. +func (d *Downloader) fetchBALs(from uint64) error { + log.Debug("Downloading block access lists", "origin", from) + err := d.concurrentFetch((*balQueue)(d)) + + log.Debug("Block access list download terminated", "err", err) + return err +} + // processHeaders takes batches of retrieved headers from an input channel and // keeps processing and scheduling them into the header chain and downloader's // queue until the stream ends or a failure occurs. @@ -749,6 +766,8 @@ func (d *Downloader) processHeaders(origin uint64) error { var ( mode = d.getMode() timer = time.NewTimer(time.Second) + + lastBALCutoffUpdate time.Time // Timestamp of the last access list cutoff refresh ) defer timer.Stop() @@ -761,7 +780,7 @@ func (d *Downloader) processHeaders(origin uint64) error { // Terminate header processing if we synced up if task == nil || len(task.headers) == 0 { // Notify everyone that headers are fully processed - for _, ch := range []chan bool{d.queue.blockWakeCh, d.queue.receiptWakeCh} { + for _, ch := range []chan bool{d.queue.blockWakeCh, d.queue.receiptWakeCh, d.queue.balWakeCh} { select { case ch <- false: case <-d.cancelCh: @@ -769,6 +788,26 @@ func (d *Downloader) processHeaders(origin uint64) error { } return nil } + // Restrict block access list retrieval to the immutability window + // below the head of the network chain. Access lists further back + // are not guaranteed to be retained by the network, so fetching + // them is not even attempted. + // + // Resolving the skeleton bounds hits the database, so only refresh + // the cutoff occasionally. Staleness is harmless: the cutoff only + // moves up as the head progresses, and an outdated one merely + // schedules a few extra blocks at the edge of the window, whose + // access lists are attempted and dropped on failure anyway. + if time.Since(lastBALCutoffUpdate) > time.Minute { + if latest, _, _, err := d.skeleton.Bounds(); err == nil { + if head := latest.Number.Uint64(); head > fullMaxForkAncestry { + d.queue.SetBALCutoff(head - fullMaxForkAncestry) + } else { + d.queue.SetBALCutoff(0) + } + } + lastBALCutoffUpdate = time.Now() + } // Otherwise split the chunk of headers into batches and process them headers, hashes, scheduled := task.headers, task.hashes, false @@ -840,7 +879,7 @@ func (d *Downloader) processHeaders(origin uint64) error { // Signal the downloader of the availability of new tasks if scheduled { - for _, ch := range []chan bool{d.queue.blockWakeCh, d.queue.receiptWakeCh} { + for _, ch := range []chan bool{d.queue.blockWakeCh, d.queue.receiptWakeCh, d.queue.balWakeCh} { select { case ch <- true: default: @@ -886,6 +925,13 @@ func (d *Downloader) importBlockResults(results []*fetchResult) error { blocks := make([]*types.Block, len(results)) for i, result := range results { blocks[i] = types.NewBlockWithHeader(result.Header).WithBody(result.body()) + + // Attach the access list if it was retrieved from the network. The + // content hash was already verified against the header on delivery; + // blocks lacking one have theirs computed locally during execution. + if list := result.BAL(); list != nil { + blocks[i] = blocks[i].WithAccessListUnsafe(list) + } } // Downloaded blocks are always regarded as trusted after the // transition. Because the downloaded chain is guided by the @@ -1090,6 +1136,12 @@ func (d *Downloader) commitSnapSyncData(results []*fetchResult, stateSync *state for i, result := range results { blocks[i] = types.NewBlockWithHeader(result.Header).WithBody(result.body()) receipts[i] = result.Receipts + + // Attach the access list if it was retrieved from the network, so it + // gets persisted alongside the block data. + if list := result.BAL(); list != nil { + blocks[i] = blocks[i].WithAccessListUnsafe(list) + } } if index, err := d.blockchain.InsertReceiptChain(blocks, receipts, d.ancientLimit); err != nil { log.Debug("Downloaded item processing failed", "number", results[index].Header.Number, "hash", results[index].Header.Hash(), "err", err) @@ -1100,6 +1152,9 @@ func (d *Downloader) commitSnapSyncData(results []*fetchResult, stateSync *state func (d *Downloader) commitPivotBlock(result *fetchResult) error { block := types.NewBlockWithHeader(result.Header).WithBody(result.body()) + if list := result.BAL(); list != nil { + block = block.WithAccessListUnsafe(list) + } log.Debug("Committing snap sync pivot as new head", "number", block.Number(), "hash", block.Hash()) // Commit the pivot block as the new head, will require full sync from here on diff --git a/eth/downloader/downloader_test.go b/eth/downloader/downloader_test.go index 5a1b6b94c6..88444ec8a1 100644 --- a/eth/downloader/downloader_test.go +++ b/eth/downloader/downloader_test.go @@ -17,6 +17,7 @@ package downloader import ( + "bytes" "math/big" "sync" "sync/atomic" @@ -25,10 +26,13 @@ import ( "github.com/ethereum/go-ethereum" "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/consensus" + "github.com/ethereum/go-ethereum/consensus/beacon" "github.com/ethereum/go-ethereum/consensus/ethash" "github.com/ethereum/go-ethereum/core" "github.com/ethereum/go-ethereum/core/rawdb" "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/eth/ethconfig" "github.com/ethereum/go-ethereum/eth/protocols/eth" "github.com/ethereum/go-ethereum/eth/protocols/snap" @@ -62,6 +66,17 @@ func newTesterWithNotification(t *testing.T, mode ethconfig.SyncMode, success fu // newTesterWithSnap is like newTesterWithNotification but selects the snap/2 // state syncer when snapV2 is set. func newTesterWithSnap(t *testing.T, mode ethconfig.SyncMode, success func(), snapV2 bool) *downloadTester { + gspec := &core.Genesis{ + Config: params.TestChainConfig, + Alloc: types.GenesisAlloc{testAddress: {Balance: big.NewInt(1000000000000000)}}, + BaseFee: big.NewInt(params.InitialBaseFee), + } + return newTesterWithGenesis(t, mode, success, snapV2, gspec, ethash.NewFaker()) +} + +// newTesterWithGenesis is like newTesterWithSnap, but the local chain is built +// from an arbitrary genesis specification and consensus engine. +func newTesterWithGenesis(t *testing.T, mode ethconfig.SyncMode, success func(), snapV2 bool, gspec *core.Genesis, engine consensus.Engine) *downloadTester { db, err := rawdb.Open(rawdb.NewMemoryDatabase(), rawdb.OpenOptions{}) if err != nil { panic(err) @@ -69,12 +84,7 @@ func newTesterWithSnap(t *testing.T, mode ethconfig.SyncMode, success func(), sn t.Cleanup(func() { db.Close() }) - gspec := &core.Genesis{ - Config: params.TestChainConfig, - Alloc: types.GenesisAlloc{testAddress: {Balance: big.NewInt(1000000000000000)}}, - BaseFee: big.NewInt(params.InitialBaseFee), - } - chain, err := core.NewBlockChain(db, gspec, ethash.NewFaker(), nil) + chain, err := core.NewBlockChain(db, gspec, engine, nil) if err != nil { panic(err) } @@ -96,14 +106,22 @@ func (dl *downloadTester) terminate() { // newPeer registers a new block download source into the downloader. func (dl *downloadTester) newPeer(id string, version uint, blocks []*types.Block) *downloadTesterPeer { + return dl.newPeerWithChain(id, version, newTestBlockchain(blocks), nil) +} + +// newPeerWithChain registers a new block download source into the downloader, +// serving content from the given pre-assembled chain. An optional gate can be +// specified to delay body deliveries until the respective access lists arrive. +func (dl *downloadTester) newPeerWithChain(id string, version uint, chain *core.BlockChain, gate *balGate) *downloadTesterPeer { dl.lock.Lock() defer dl.lock.Unlock() peer := &downloadTesterPeer{ dl: dl, id: id, - chain: newTestBlockchain(blocks), + chain: chain, withholdBodies: make(map[common.Hash]struct{}), + balGate: gate, dropped: make(chan error, 1), } dl.peers[id] = peer @@ -130,13 +148,66 @@ func (dl *downloadTester) dropPeer(id string) { type downloadTesterPeer struct { dl *downloadTester withholdBodies map[common.Hash]struct{} - corruptBodies bool // if set, the peer serves incorrect blocks + corruptBodies bool // if set, the peer serves incorrect blocks + balGate *balGate // if set, body deliveries wait for the access lists id string chain *core.BlockChain dropped chan error // signaled when res.Done receives an error } +// balGate delays the body delivery of a set of blocks until their access lists +// were delivered into the downloader's queue. Access lists are a best-effort +// component that the queue never waits on, so without external serialization a +// test cannot assert their attachment deterministically. +type balGate struct { + lock sync.Mutex + cond *sync.Cond + pending map[common.Hash]struct{} // blocks whose access list was not yet delivered +} + +func newBALGate(hashes []common.Hash) *balGate { + g := &balGate{ + pending: make(map[common.Hash]struct{}), + } + g.cond = sync.NewCond(&g.lock) + for _, hash := range hashes { + g.pending[hash] = struct{}{} + } + return g +} + +// served flags the access lists of the given blocks as delivered. +func (g *balGate) served(hashes []common.Hash) { + g.lock.Lock() + defer g.lock.Unlock() + + for _, hash := range hashes { + delete(g.pending, hash) + } + g.cond.Broadcast() +} + +// wait blocks until the access lists of all the given blocks were delivered. +func (g *balGate) wait(hashes []common.Hash) { + g.lock.Lock() + defer g.lock.Unlock() + + for { + blocked := false + for _, hash := range hashes { + if _, ok := g.pending[hash]; ok { + blocked = true + break + } + } + if !blocked { + return + } + g.cond.Wait() + } +} + func unmarshalRlpHeaders(rlpdata []rlp.RawValue) []*types.Header { var headers = make([]*types.Header, len(rlpdata)) for i, data := range rlpdata { @@ -247,6 +318,9 @@ func (dlp *downloadTesterPeer) RequestBodies(hashes []common.Hash, sink chan *et } txsHashes[i] = hash uncleHashes[i] = types.CalcUncleHash(body.Uncles) + if body.Withdrawals != nil { + withdrawalHashes[i] = types.DeriveSha(types.Withdrawals(body.Withdrawals), hasher) + } } if dlp.corruptBodies { for i := range txsHashes { @@ -268,6 +342,11 @@ func (dlp *downloadTesterPeer) RequestBodies(hashes []common.Hash, sink chan *et Done: make(chan error), } go func() { + // If gated, hold back the bodies until the access lists of the + // requested blocks were delivered + if dlp.balGate != nil { + dlp.balGate.wait(hashes) + } sink <- res if err := <-res.Done; err != nil { select { @@ -312,6 +391,54 @@ func (dlp *downloadTesterPeer) RequestReceipts(hashes []common.Hash, gasUsed []u return res.Req, nil } +// RequestBALs constructs a getBlockAccessLists method associated with a +// particular peer in the download tester. The returned function can be used to +// retrieve batches of block access lists from the particularly requested peer. +func (dlp *downloadTesterPeer) RequestBALs(hashes []common.Hash, sink chan *eth.Response) (*eth.Request, error) { + var ( + bals = make([]rlp.RawValue, 0, len(hashes)) + served = make([]common.Hash, 0, len(hashes)) + ) + for _, hash := range hashes { + data := dlp.chain.GetAccessListRLP(hash) + if len(data) == 0 { + // The signal for a missing access list is the empty string + bals = append(bals, rlp.EmptyString) + continue + } + bals = append(bals, data) + served = append(served, hash) + } + // compute the content hashes, zero hash for unavailable entries + meta := make([]common.Hash, len(bals)) + for i, data := range bals { + if bytes.Equal(data, rlp.EmptyString) { + continue + } + meta[i] = crypto.Keccak256Hash(data) + } + // deliver the response right away + resp := eth.BlockAccessListResponse(bals) + res := ð.Response{ + Req: ð.Request{Peer: dlp.id}, + Res: &resp, + Meta: meta, + Time: 1, + Done: make(chan error, 1), + } + go func() { + sink <- res + + // If gated, unblock the body deliveries of the served blocks, but only + // after the access lists were fully processed by the queue + if dlp.balGate != nil { + <-res.Done + dlp.balGate.served(served) + } + }() + return res.Req, nil +} + // ID retrieves the peer's unique identifier. func (dlp *downloadTesterPeer) ID() string { return dlp.id @@ -461,6 +588,143 @@ func testCanonSync(t *testing.T, protocol uint, mode SyncMode, snapV2 bool) { } } +// makeBALChain constructs a post-merge, Amsterdam-enabled chain whose blocks +// all carry a block access list commitment, along with the genesis needed to +// sync it. Every block contains a transaction so that no block has an empty +// body (empty-body blocks complete without a network retrieval, voiding any +// delivery ordering imposed by the tests). +func makeBALChain(n int) (*core.Genesis, []*types.Block) { + config := *params.MergedTestChainConfig + config.AmsterdamTime = new(uint64) + + gspec := &core.Genesis{ + Config: &config, + Alloc: types.GenesisAlloc{ + testAddress: {Balance: new(big.Int).Mul(big.NewInt(1000), big.NewInt(params.Ether))}, + params.BeaconRootsAddress: {Nonce: 1, Code: params.BeaconRootsCode, Balance: common.Big0}, + params.HistoryStorageAddress: {Nonce: 1, Code: params.HistoryStorageCode, Balance: common.Big0}, + params.WithdrawalQueueAddress: {Nonce: 1, Code: params.WithdrawalQueueCode, Balance: common.Big0}, + params.ConsolidationQueueAddress: {Nonce: 1, Code: params.ConsolidationQueueCode, Balance: common.Big0}, + params.BuilderDepositAddress: {Nonce: 1, Code: params.BuilderDepositCode, Balance: common.Big0}, + params.BuilderExitAddress: {Nonce: 1, Code: params.BuilderExitCode, Balance: common.Big0}, + }, + BaseFee: big.NewInt(params.InitialBaseFee), + Difficulty: common.Big0, + } + signer := types.LatestSigner(&config) + _, blocks, _ := core.GenerateChainWithGenesis(gspec, beacon.New(ethash.NewFaker()), n, func(i int, block *core.BlockGen) { + // The chain maker only executes the EIP-4788 system call when the + // beacon root is set explicitly; without it, the generated access + // lists would not match the ones computed at import. + block.SetParentBeaconRoot(common.Hash{}) + + tx, err := types.SignTx(types.NewTransaction(block.TxNonce(testAddress), common.Address{0x01}, big.NewInt(1000), params.TxGas, block.BaseFee(), nil), signer, testKey) + if err != nil { + panic(err) + } + block.AddTx(tx) + }) + return gspec, blocks +} + +// Tests that block access lists are retrieved from eth/71+ peers during sync +// and end up attached to the imported blocks; and that syncing against peers +// which cannot (or do not) serve access lists still completes, importing the +// blocks without them. +func TestBALSynchronisationFull(t *testing.T) { testBALSync(t, FullSync, eth.ETH71) } +func TestBALSynchronisationSnap(t *testing.T) { testBALSync(t, SnapSync, eth.ETH71) } +func TestBALSynchronisationLegacyPeer(t *testing.T) { testBALSync(t, FullSync, eth.ETH69) } + +func testBALSync(t *testing.T, mode SyncMode, protocol uint) { + gspec, blocks := makeBALChain(96) // long enough for a snap sync pivot below the head + + success := make(chan struct{}) + tester := newTesterWithGenesis(t, mode, func() { close(success) }, false, gspec, beacon.New(ethash.NewFaker())) + defer tester.terminate() + + // Assemble the serving chain, executing the blocks to persist their access + // lists. + peerChain, err := core.NewBlockChain(rawdb.NewMemoryDatabase(), gspec, beacon.New(ethash.NewFaker()), nil) + if err != nil { + t.Fatalf("failed to create peer chain: %v", err) + } + defer peerChain.Stop() + + if _, err := peerChain.InsertChain(blocks); err != nil { + t.Fatalf("failed to assemble peer chain: %v", err) + } + // Collect the blocks whose access lists the downloader should retrieve + var eligible []common.Hash + for _, block := range blocks { + if hash := block.Header().BlockAccessListHash; hash != nil && *hash != types.EmptyBlockAccessListHash { + eligible = append(eligible, block.Hash()) + } + } + if len(eligible) != len(blocks) { + t.Fatalf("expected all %d blocks to commit to an access list, got %d", len(blocks), len(eligible)) + } + // Access lists are best effort and never block the delivery of an otherwise + // completed block. To assert their attachment deterministically, gate the + // body deliveries of modern peers on the access lists arriving first. + var gate *balGate + if protocol >= eth.ETH71 { + gate = newBALGate(eligible) + } + tester.newPeerWithChain("peer", protocol, peerChain, gate) + + // Track which imported blocks had an access list attached + var ( + attachLock sync.Mutex + attached = make(map[uint64]bool) + ) + tester.downloader.chainInsertHook = func(results []*fetchResult) { + attachLock.Lock() + defer attachLock.Unlock() + for _, result := range results { + if result.BAL() != nil { + attached[result.Header.Number.Uint64()] = true + } + } + } + // Synchronise with the peer and make sure all relevant data was retrieved. + // In snap mode, announce a finalized block too, directing the chain segment + // below it straight into the ancient store: downloaded access lists must + // end up retrievable through that path as well. + var final *types.Header + if mode == SnapSync { + final = blocks[len(blocks)/3].Header() + } + if err := tester.downloader.BeaconSync(blocks[len(blocks)-1].Header(), final); err != nil { + t.Fatalf("failed to beacon-sync chain: %v", err) + } + select { + case <-success: + assertOwnChain(t, tester, len(blocks)+1) + case <-time.NewTimer(15 * time.Second).C: + t.Fatalf("failed to sync chain in fifteen seconds") + } + attachLock.Lock() + defer attachLock.Unlock() + + for _, block := range blocks { + if protocol >= eth.ETH71 { + // Modern peer: the access list must have been downloaded, attached + // to the imported block and persisted locally + if !attached[block.NumberU64()] { + t.Errorf("block %d: no access list attached at import", block.NumberU64()) + } + if have, want := tester.chain.GetAccessListRLP(block.Hash()), peerChain.GetAccessListRLP(block.Hash()); !bytes.Equal(have, want) { + t.Errorf("block %d: persisted access list mismatch: have %x, want %x", block.NumberU64(), have, want) + } + } else { + // Legacy peer: blocks must have been imported without access lists + if attached[block.NumberU64()] { + t.Errorf("block %d: unexpected access list attached at import", block.NumberU64()) + } + } + } +} + // Tests that if a large batch of blocks are being downloaded, it is throttled // until the cached blocks are retrieved. func TestThrottlingFull(t *testing.T) { testThrottling(t, eth.ETH69, FullSync) } diff --git a/eth/downloader/fetchers_concurrent.go b/eth/downloader/fetchers_concurrent.go index 51bf3404bd..ff8eecf892 100644 --- a/eth/downloader/fetchers_concurrent.go +++ b/eth/downloader/fetchers_concurrent.go @@ -362,7 +362,7 @@ func (d *Downloader) concurrentFetch(queue typedQueue) error { // validityErrorOfRequest returns err if it is related to block validity, and nil otherwise. func validityErrorOfRequest(err error) error { - if errors.Is(err, errInvalidBody) || errors.Is(err, errInvalidReceipt) { + if errors.Is(err, errInvalidBody) || errors.Is(err, errInvalidReceipt) || errors.Is(err, errInvalidBAL) { return err } return nil diff --git a/eth/downloader/fetchers_concurrent_bals.go b/eth/downloader/fetchers_concurrent_bals.go new file mode 100644 index 0000000000..737f45b17e --- /dev/null +++ b/eth/downloader/fetchers_concurrent_bals.go @@ -0,0 +1,114 @@ +// Copyright 2026 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library 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. +// +// The go-ethereum library 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 the go-ethereum library. If not, see . + +package downloader + +import ( + "time" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/eth/protocols/eth" + "github.com/ethereum/go-ethereum/log" +) + +// balQueue implements typedQueue and is a type adapter between the generic +// concurrent fetcher and the downloader. It fetches EIP-7928 block access +// lists on a best effort basis: only eth/71+ peers can serve them and blocks +// are imported without one if it doesn't arrive in time. +type balQueue Downloader + +// waker returns a notification channel that gets pinged in case more access +// list fetches have been queued up, so the fetcher might assign it to idle peers. +func (q *balQueue) waker() chan bool { + return q.queue.balWakeCh +} + +// pending returns the number of access lists that are currently queued for +// fetching by the concurrent downloader. +func (q *balQueue) pending() int { + return q.queue.PendingBALs() +} + +// capacity is responsible for calculating how many access lists a particular +// peer is estimated to be able to retrieve within the allotted round trip time. +func (q *balQueue) capacity(peer *peerConnection, rtt time.Duration) int { + if peer.version < eth.ETH71 { + return 0 + } + return peer.BALCapacity(rtt) +} + +// updateCapacity is responsible for updating how many access lists a particular +// peer is estimated to be able to retrieve in a unit time. +func (q *balQueue) updateCapacity(peer *peerConnection, items int, span time.Duration) { + peer.UpdateBALRate(items, span) +} + +// reserve is responsible for allocating a requested number of pending access +// lists from the download queue to the specified peer. Peers below eth/71 +// cannot serve access lists and are never assigned any. +func (q *balQueue) reserve(peer *peerConnection, items int) (*fetchRequest, bool, bool) { + if peer.version < eth.ETH71 { + return nil, false, false + } + return q.queue.ReserveBALs(peer, items) +} + +// unreserve is responsible for removing the current access list retrieval +// allocation assigned to a specific peer and placing it back into the pool to +// allow reassigning to some other peer. +func (q *balQueue) unreserve(peer string) int { + fails := q.queue.ExpireBALs(peer) + if fails > 2 { + log.Trace("Access list delivery timed out", "peer", peer) + } else { + log.Debug("Access list delivery stalling", "peer", peer) + } + return fails +} + +// request is responsible for converting a generic fetch request into an access +// list one and sending it to the remote peer for fulfillment. +func (q *balQueue) request(peer *peerConnection, req *fetchRequest, resCh chan *eth.Response) (*eth.Request, error) { + peer.log.Trace("Requesting new batch of access lists", "count", len(req.Headers), "from", req.Headers[0].Number) + if q.balFetchHook != nil { + q.balFetchHook(req.Headers) + } + hashes := make([]common.Hash, 0, len(req.Headers)) + for _, header := range req.Headers { + hashes = append(hashes, header.Hash()) + } + return peer.peer.RequestBALs(hashes, resCh) +} + +// deliver is responsible for taking a generic response packet from the +// concurrent fetcher, unpacking the access list data and delivering it to the +// downloader's queue. +func (q *balQueue) deliver(peer *peerConnection, packet *eth.Response) (int, error) { + bals := *packet.Res.(*eth.BlockAccessListResponse) + hashes := packet.Meta.([]common.Hash) // {keccak256 hash per entry, zero hash if unavailable} + + accepted, err := q.queue.DeliverBALs(peer.id, bals, hashes) + switch { + case err == nil && len(bals) == 0: + peer.log.Trace("Requested access lists delivered") + case err == nil: + peer.log.Trace("Delivered new batch of access lists", "count", len(bals), "accepted", accepted) + default: + peer.log.Debug("Failed to deliver retrieved access lists", "err", err) + } + return accepted, err +} diff --git a/eth/downloader/metrics.go b/eth/downloader/metrics.go index caa8dfb9ee..a70eb17e35 100644 --- a/eth/downloader/metrics.go +++ b/eth/downloader/metrics.go @@ -37,6 +37,11 @@ var ( receiptDropMeter = metrics.NewRegisteredMeter("eth/downloader/receipts/drop", nil) receiptTimeoutMeter = metrics.NewRegisteredMeter("eth/downloader/receipts/timeout", nil) + balInMeter = metrics.NewRegisteredMeter("eth/downloader/bals/in", nil) + balReqTimer = metrics.NewRegisteredTimer("eth/downloader/bals/req", nil) + balDropMeter = metrics.NewRegisteredMeter("eth/downloader/bals/drop", nil) + balTimeoutMeter = metrics.NewRegisteredMeter("eth/downloader/bals/timeout", nil) + throttleCounter = metrics.NewRegisteredCounter("eth/downloader/throttle", nil) // snapPeerSkipMeter tracks snap peers skipped by the state syncer because diff --git a/eth/downloader/peer.go b/eth/downloader/peer.go index d20bda69e9..beba39eea0 100644 --- a/eth/downloader/peer.go +++ b/eth/downloader/peer.go @@ -44,8 +44,9 @@ var ( type peerConnection struct { id string // Unique identifier of the peer - rates *msgrate.Tracker // Tracker to hone in on the number of items retrievable per second - lacking map[common.Hash]struct{} // Set of hashes not to request (didn't have previously) + rates *msgrate.Tracker // Tracker to hone in on the number of items retrievable per second + lacking map[common.Hash]struct{} // Set of hashes not to request (didn't have previously) + lackingBAL map[common.Hash]struct{} // Set of hashes not to request the access lists for (didn't have previously) peer Peer @@ -61,16 +62,18 @@ type Peer interface { RequestBodies([]common.Hash, chan *eth.Response) (*eth.Request, error) RequestReceipts([]common.Hash, []uint64, []uint64, chan *eth.Response) (*eth.Request, error) + RequestBALs([]common.Hash, chan *eth.Response) (*eth.Request, error) } // newPeerConnection creates a new downloader peer. func newPeerConnection(id string, version uint, peer Peer, logger log.Logger) *peerConnection { return &peerConnection{ - id: id, - lacking: make(map[common.Hash]struct{}), - peer: peer, - version: version, - log: logger, + id: id, + lacking: make(map[common.Hash]struct{}), + lackingBAL: make(map[common.Hash]struct{}), + peer: peer, + version: version, + log: logger, } } @@ -80,6 +83,7 @@ func (p *peerConnection) Reset() { defer p.lock.Unlock() p.lacking = make(map[common.Hash]struct{}) + p.lackingBAL = make(map[common.Hash]struct{}) } // UpdateHeaderRate updates the peer's estimated header retrieval throughput with @@ -100,6 +104,12 @@ func (p *peerConnection) UpdateReceiptRate(delivered int, elapsed time.Duration) p.rates.Update(eth.ReceiptsMsg, elapsed, delivered) } +// UpdateBALRate updates the peer's estimated block access list retrieval +// throughput with the current measurement. +func (p *peerConnection) UpdateBALRate(delivered int, elapsed time.Duration) { + p.rates.Update(eth.BlockAccessListsMsg, elapsed, delivered) +} + // HeaderCapacity retrieves the peer's header download allowance based on its // previously discovered throughput. func (p *peerConnection) HeaderCapacity(targetRTT time.Duration) int { @@ -130,6 +140,16 @@ func (p *peerConnection) ReceiptCapacity(targetRTT time.Duration) int { return cap } +// BALCapacity retrieves the peer's block access list download allowance based +// on its previously discovered throughput. +func (p *peerConnection) BALCapacity(targetRTT time.Duration) int { + cap := p.rates.Capacity(eth.BlockAccessListsMsg, targetRTT) + if cap > MaxBALFetch { + cap = MaxBALFetch + } + return cap +} + // MarkLacking appends a new entity to the set of items (blocks, receipts, states) // that a peer is known not to have (i.e. have been requested before). If the // set reaches its maximum allowed capacity, items are randomly dropped off. @@ -156,6 +176,34 @@ func (p *peerConnection) Lacks(hash common.Hash) bool { return ok } +// MarkLackingBAL appends a new block hash to the set of blocks whose access +// list the peer is known not to have. Access lists are tracked separately from +// the other block components, since a peer missing a block's access list may +// well possess its body and receipts. If the set reaches its maximum allowed +// capacity, items are randomly dropped off. +func (p *peerConnection) MarkLackingBAL(hash common.Hash) { + p.lock.Lock() + defer p.lock.Unlock() + + for len(p.lackingBAL) >= maxLackingHashes { + for drop := range p.lackingBAL { + delete(p.lackingBAL, drop) + break + } + } + p.lackingBAL[hash] = struct{}{} +} + +// LacksBAL retrieves whether the access list of a block is on the peer's +// lacking list (i.e. whether we know that the peer does not have it). +func (p *peerConnection) LacksBAL(hash common.Hash) bool { + p.lock.RLock() + defer p.lock.RUnlock() + + _, ok := p.lackingBAL[hash] + return ok +} + // peeringEvent is sent on the peer event feed when a remote peer connects or // disconnects. type peeringEvent struct { diff --git a/eth/downloader/queue.go b/eth/downloader/queue.go index 585906b8bd..a93ef14411 100644 --- a/eth/downloader/queue.go +++ b/eth/downloader/queue.go @@ -20,6 +20,7 @@ package downloader import ( + "bytes" "errors" "fmt" "sync" @@ -29,6 +30,7 @@ import ( "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/common/prque" "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/core/types/bal" "github.com/ethereum/go-ethereum/eth/ethconfig" "github.com/ethereum/go-ethereum/eth/protocols/eth" "github.com/ethereum/go-ethereum/log" @@ -39,6 +41,7 @@ import ( const ( bodyType = uint(0) receiptType = uint(1) + balType = uint(2) ) var ( @@ -46,6 +49,16 @@ var ( blockCacheInitialItems = 2048 // Initial number of blocks to start fetching, before we know the sizes of the blocks blockCacheMemory = 256 * 1024 * 1024 // Maximum amount of memory to use for block caching blockCacheSizeWeight = 0.1 // Multiplier to approximate the average block size based on past ones + + // balCacheMemory is the memory allowance for block access lists attached + // to cached results, budgeted separately from blockCacheMemory: access + // lists are attached best effort (present for some blocks, absent for + // others), so folding them into the per-block size expectation would + // make it swing wildly and shrink the block fetch batches whenever lists + // happen to arrive. Instead their exact attached-but-undelivered bytes + // are tracked, and once over the allowance the *access list* retrieval + // throttles itself while block throughput stays untouched. + balCacheMemory = 256 * 1024 * 1024 ) var ( @@ -71,9 +84,41 @@ type fetchResult struct { Transactions types.Transactions Receipts rlp.RawValue Withdrawals types.Withdrawals + + // accessList is the optional EIP-7928 block access list, retrieved on a + // best effort basis for blocks close to the head of the network chain. + accessList atomic.Pointer[balAttachment] } -func newFetchResult(header *types.Header, snapSync bool) *fetchResult { +// balAttachment couples a delivered block access list with its encoded size. +type balAttachment struct { + list *bal.BlockAccessList + size common.StorageSize +} + +// SetBAL attaches a downloaded block access list along with its encoded size. +func (f *fetchResult) SetBAL(list *bal.BlockAccessList, size common.StorageSize) { + f.accessList.Store(&balAttachment{list: list, size: size}) +} + +// BAL returns the attached block access list, or nil if none arrived in time. +func (f *fetchResult) BAL() *bal.BlockAccessList { + if attach := f.accessList.Load(); attach != nil { + return attach.list + } + return nil +} + +// BALSize returns the encoded size of the attached block access list, or zero +// if none arrived in time. +func (f *fetchResult) BALSize() common.StorageSize { + if attach := f.accessList.Load(); attach != nil { + return attach.size + } + return 0 +} + +func newFetchResult(header *types.Header, snapSync bool, fetchBAL bool) *fetchResult { item := &fetchResult{ Header: header, } @@ -90,6 +135,9 @@ func newFetchResult(header *types.Header, snapSync bool) *fetchResult { item.pending.Store(item.pending.Load() | (1 << receiptType)) } } + if fetchBAL { + item.pending.Store(item.pending.Load() | (1 << balType)) + } return item } @@ -109,9 +157,11 @@ func (f *fetchResult) SetBodyDone() { } } -// AllDone checks if item is done. +// AllDone checks if item is done. The block access list is a best-effort +// component and never holds back the delivery of an otherwise completed +// block: blocks are handed over without one if it hasn't arrived in time. func (f *fetchResult) AllDone() bool { - return f.pending.Load() == 0 + return f.pending.Load()&((1<= q.balCutoff +} + // InFlightBlocks retrieves whether there are block fetch requests currently in // flight. func (q *queue) InFlightBlocks() bool { @@ -280,6 +384,16 @@ func (q *queue) Schedule(headers []*types.Header, hashes []common.Hash, from uin q.receiptTaskQueue.Push(header, -int64(header.Number.Uint64())) } } + // Queue for best-effort access list retrieval if the block is recent + // enough for the network to still serve its access list + if q.balEligible(header) { + if _, ok := q.balTaskPool[hash]; ok { + log.Warn("Header already scheduled for access list fetch", "number", header.Number, "hash", hash) + } else { + q.balTaskPool[hash] = header + q.balTaskQueue.Push(header, -int64(header.Number.Uint64())) + } + } inserts++ q.headerHead = hash from++ @@ -317,6 +431,13 @@ func (q *queue) Results(block bool) []*fetchResult { } // Regardless if closed or not, we can still deliver whatever we have results := q.resultCache.GetCompleted(maxResultsProcess) + + // Access lists are a best-effort component: any retrieval task for a block + // that has been delivered upstream (with or without one) is obsolete, drop + // them to unblock the access list fetcher's termination. + if len(results) > 0 { + q.pruneBALTasks(results[len(results)-1].Header.Number.Uint64()) + } for _, result := range results { // Recalculate the result item weights to prevent memory exhaustion size := result.Header.Size() @@ -328,6 +449,7 @@ func (q *queue) Results(block bool) []*fetchResult { size += common.StorageSize(tx.Size()) } size += common.StorageSize(result.Withdrawals.Size()) + q.balBytes.Add(-int64(result.BALSize())) q.resultSize = common.StorageSize(blockCacheSizeWeight)*size + (1-common.StorageSize(blockCacheSizeWeight))*q.resultSize } @@ -337,7 +459,7 @@ func (q *queue) Results(block bool) []*fetchResult { throttleThreshold = q.resultCache.SetThrottleThreshold(throttleThreshold) // With results removed from the cache, wake throttled fetchers - for _, ch := range []chan bool{q.blockWakeCh, q.receiptWakeCh} { + for _, ch := range []chan bool{q.blockWakeCh, q.receiptWakeCh, q.balWakeCh} { select { case ch <- true: default: @@ -365,10 +487,28 @@ func (q *queue) stats() []interface{} { return []interface{}{ "receiptTasks", q.receiptTaskQueue.Size(), "blockTasks", q.blockTaskQueue.Size(), + "balTasks", q.balTaskQueue.Size(), "itemSize", q.resultSize, } } +// pruneBALTasks drops all queued access list retrieval tasks at or below the +// given block number. Since blocks are delivered upstream without waiting for +// their access lists, tasks below the delivery point serve no purpose anymore. +func (q *queue) pruneBALTasks(delivered uint64) { + q.lock.Lock() + defer q.lock.Unlock() + + for !q.balTaskQueue.Empty() { + header, _ := q.balTaskQueue.Peek() + if header.Number.Uint64() > delivered { + break + } + q.balTaskQueue.PopItem() + delete(q.balTaskPool, header.Hash()) + } +} + // ReserveBodies reserves a set of body fetches for the given peer, skipping any // previously failed downloads. Beside the next batch of needed fetches, it also // returns a flag whether empty blocks were queued requiring processing. @@ -389,6 +529,20 @@ func (q *queue) ReserveReceipts(p *peerConnection, count int) (*fetchRequest, bo return q.reserveHeaders(p, count, q.receiptTaskPool, q.receiptTaskQueue, q.receiptPendPool, receiptType) } +// ReserveBALs reserves a set of block access list fetches for the given peer, +// skipping any previously failed downloads. +func (q *queue) ReserveBALs(p *peerConnection, count int) (*fetchRequest, bool, bool) { + // Throttle the access list retrieval itself once the attached-but-not-yet + // delivered lists exhaust their own memory allowance. + if q.balBytes.Load() > int64(balCacheMemory) { + return nil, false, true + } + q.lock.Lock() + defer q.lock.Unlock() + + return q.reserveHeaders(p, count, q.balTaskPool, q.balTaskQueue, q.balPendPool, balType) +} + // reserveHeaders reserves a set of data download operations for a given peer, // skipping any previously failed ones. This method is a generic version used // by the individual special reservation functions. @@ -417,6 +571,14 @@ func (q *queue) reserveHeaders(p *peerConnection, count int, taskPool map[common skip := make([]*types.Header, 0) progress := false throttled := false + + // Access list availability is tracked separately from the other block + // components: a peer missing a block's access list may well have its + // body and receipts. + lacks := p.Lacks + if kind == balType { + lacks = p.LacksBAL + } for len(send) < count && !taskQueue.Empty() { // the task queue will pop items in order, so the highest prio block // is also the lowest block number. @@ -425,7 +587,7 @@ func (q *queue) reserveHeaders(p *peerConnection, count int, taskPool map[common // we can ask the resultcache if this header is within the // "prioritized" segment of blocks. If it is not, we need to throttle - stale, throttle, item, err := q.resultCache.AddFetch(header, q.mode == ethconfig.SnapSync) + stale, throttle, item, err := q.resultCache.AddFetch(header, q.mode == ethconfig.SnapSync, q.balEligible(header)) if stale { // Don't put back in the task queue, this item has already been // delivered upstream @@ -459,7 +621,7 @@ func (q *queue) reserveHeaders(p *peerConnection, count int, taskPool map[common // Remove it from the task queue taskQueue.PopItem() // Otherwise unless the peer is known not to have the data, add to the retrieve list - if p.Lacks(header.Hash()) { + if lacks(header.Hash()) { skip = append(skip, header) } else { send = append(send, header) @@ -505,6 +667,12 @@ func (q *queue) Revoke(peerID string) { } delete(q.receiptPendPool, peerID) } + if request, ok := q.balPendPool[peerID]; ok { + for _, header := range request.Headers { + q.balTaskQueue.Push(header, -int64(header.Number.Uint64())) + } + delete(q.balPendPool, peerID) + } } // ExpireBodies checks for in flight block body requests that exceeded a timeout @@ -527,6 +695,17 @@ func (q *queue) ExpireReceipts(peer string) int { return q.expire(peer, q.receiptPendPool, q.receiptTaskQueue) } +// ExpireBALs checks for in flight block access list requests that exceeded a +// timeout allowance, canceling them and returning the responsible peers for +// penalisation. +func (q *queue) ExpireBALs(peer string) int { + q.lock.Lock() + defer q.lock.Unlock() + + balTimeoutMeter.Mark(1) + return q.expire(peer, q.balPendPool, q.balTaskQueue) +} + // expire is the generic check that moves a specific expired task from a pending // pool back into a task pool. The syntax on the passed taskQueue is a bit weird // as we would need a generic expire method to handle both types, but that is not @@ -642,6 +821,82 @@ func (q *queue) DeliverReceipts(id string, receiptList []rlp.RawValue, receiptLi receiptReqTimer, receiptInMeter, receiptDropMeter, len(receiptList), validate, reconstruct) } +// DeliverBALs injects a block access list retrieval response into the results +// queue. Unlike bodies and receipts, access lists are a best-effort component: +// entries the remote peer does not possess are handed back to the task queue +// for retrieval from other peers, and blocks whose access lists do not arrive +// in time are delivered upstream without one. The hashes parameter carries the +// keccak256 hash of each raw entry (the zero hash for unavailable entries), +// precomputed by the protocol layer. +func (q *queue) DeliverBALs(id string, bals []rlp.RawValue, hashes []common.Hash) (int, error) { + q.lock.Lock() + defer q.lock.Unlock() + + request := q.balPendPool[id] + if request == nil { + balDropMeter.Mark(int64(len(bals))) + return 0, errNoFetchesPending + } + delete(q.balPendPool, id) + + balReqTimer.UpdateSince(request.Time) + balInMeter.Mark(int64(len(bals))) + + // If no data items were retrieved, mark them all as unavailable for the + // origin peer + if len(bals) == 0 { + for _, header := range request.Headers { + request.Peer.MarkLackingBAL(header.Hash()) + } + } + var ( + accepted int + failure error + ) + for i, header := range request.Headers { + // Should the response be invalid at some point, return all the + // remaining tasks to the queue for retrieval from other peers + if failure != nil || i >= len(bals) { + q.balTaskQueue.Push(header, -int64(header.Number.Uint64())) + continue + } + hash := header.Hash() + + // The empty string signals that the peer does not possess this access + // list (an empty list is itself a valid access list); leave the task + // queued for other peers to have a go at it. + if bytes.Equal(bals[i], rlp.EmptyString) { + request.Peer.MarkLackingBAL(hash) + q.balTaskQueue.Push(header, -int64(header.Number.Uint64())) + continue + } + // Validate the content against the hash committed in the header and + // decode it. Anything invalid is a protocol violation. + if header.BlockAccessListHash == nil || hashes[i] != *header.BlockAccessListHash { + failure = errInvalidBAL + q.balTaskQueue.Push(header, -int64(header.Number.Uint64())) + continue + } + list := new(bal.BlockAccessList) + if err := rlp.DecodeBytes(bals[i], list); err != nil { + failure = fmt.Errorf("%w: %v", errInvalidBAL, err) + q.balTaskQueue.Push(header, -int64(header.Number.Uint64())) + continue + } + // Attach the access list to the fetch result if the block was not yet + // delivered upstream; late arrivals are simply dropped. + if res, stale, err := q.resultCache.GetDeliverySlot(header.Number.Uint64()); err == nil && !stale && res != nil { + res.SetBAL(list, common.StorageSize(len(bals[i]))) + res.SetBALDone() + q.balBytes.Add(int64(len(bals[i]))) + accepted++ + } + delete(q.balTaskPool, hash) + } + balDropMeter.Mark(int64(len(bals) - accepted)) + return accepted, failure +} + // deliver injects a data retrieval response into the results queue. // // Note, this method expects the queue lock to be already held for writing. The diff --git a/eth/downloader/queue_test.go b/eth/downloader/queue_test.go index c7e8a0d1d6..19e58c44d2 100644 --- a/eth/downloader/queue_test.go +++ b/eth/downloader/queue_test.go @@ -17,6 +17,7 @@ package downloader import ( + "errors" "fmt" "log/slog" "math/big" @@ -30,6 +31,8 @@ import ( "github.com/ethereum/go-ethereum/consensus/ethash" "github.com/ethereum/go-ethereum/core" "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/core/types/bal" + "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/eth/protocols/eth" "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/params" @@ -89,8 +92,9 @@ func (chain *chainData) Len() int { func dummyPeer(id string) *peerConnection { p := &peerConnection{ - id: id, - lacking: make(map[common.Hash]struct{}), + id: id, + lacking: make(map[common.Hash]struct{}), + lackingBAL: make(map[common.Hash]struct{}), } return p } @@ -263,6 +267,114 @@ func TestEmptyBlocks(t *testing.T) { } } +// TestBlockAccessLists tests the scheduling and delivery of the best-effort +// block access list component: only blocks above the configured cutoff are +// scheduled, block delivery is never held back by outstanding access lists, +// and delivered lists are validated against the header commitment. +func TestBlockAccessLists(t *testing.T) { + // Create an access list along with its header commitment + list := bal.BlockAccessList{{Address: common.Address{0x01}}} + enc, err := rlp.EncodeToBytes(&list) + if err != nil { + t.Fatal(err) + } + balHash := crypto.Keccak256Hash(enc) + + // Assemble a chain of empty-body headers committing to the access list + var ( + headers = make([]*types.Header, 10) + hashes = make([]common.Hash, 10) + parent common.Hash + ) + for i := range headers { + headers[i] = &types.Header{ + ParentHash: parent, + Number: big.NewInt(int64(i + 1)), + Difficulty: big.NewInt(1), + TxHash: types.EmptyTxsHash, + UncleHash: types.EmptyUncleHash, + ReceiptHash: types.EmptyReceiptsHash, + BlockAccessListHash: &balHash, + } + hashes[i] = headers[i].Hash() + parent = hashes[i] + } + q := newQueue(16, 16) + q.Prepare(1, FullSync) + + // Only blocks 6..10 are within the retrieval window + q.SetBALCutoff(6) + q.Schedule(headers, hashes, 1) + if got, exp := q.PendingBALs(), 5; got != exp { + t.Errorf("wrong pending access list count, got %d, exp %d", got, exp) + } + // All the bodies are empty, so reserving them creates the fetch results, + // which must complete without waiting for any access lists + peer := dummyPeer("peer-1") + if req, _, _ := q.ReserveBodies(peer, 10); req != nil { + t.Fatal("there should be no body fetch tasks remaining") + } + if got, exp := q.resultCache.countCompleted(), 10; got != exp { + t.Errorf("wrong processable count, got %d, exp %d", got, exp) + } + // Reserve the first two access lists and deliver one of them, with the + // remote peer signalling that it does not possess the other + req, _, _ := q.ReserveBALs(peer, 2) + if got, exp := len(req.Headers), 2; got != exp { + t.Fatalf("expected %d requests, got %d", exp, got) + } + if got, exp := req.Headers[0].Number.Uint64(), uint64(6); got != exp { + t.Fatalf("expected header %d, got %d", exp, got) + } + accepted, err := q.DeliverBALs(peer.id, []rlp.RawValue{enc, rlp.EmptyString}, []common.Hash{balHash, {}}) + if accepted != 1 || err != nil { + t.Fatalf("unexpected delivery result, accepted %d, err %v", accepted, err) + } + if got, exp := q.balBytes.Load(), int64(len(enc)); got != exp { + t.Errorf("wrong attached access list bytes, got %d, exp %d", got, exp) + } + // The unavailable entry should be returned to the task queue and the peer + // marked as not possessing it + if got, exp := q.PendingBALs(), 4; got != exp { + t.Errorf("wrong pending access list count, got %d, exp %d", got, exp) + } + if !peer.LacksBAL(hashes[6]) { + t.Errorf("undelivered access list not marked as lacking") + } + // An access list not matching the header commitment must be rejected and + // the task handed back for retrieval from a different peer + peer2 := dummyPeer("peer-2") + if req, _, _ = q.ReserveBALs(peer2, 1); req == nil { + t.Fatal("expected access list fetch task") + } + badEnc, _ := rlp.EncodeToBytes(&bal.BlockAccessList{{Address: common.Address{0x02}}}) + accepted, err = q.DeliverBALs(peer2.id, []rlp.RawValue{badEnc}, []common.Hash{crypto.Keccak256Hash(badEnc)}) + if accepted != 0 || !errors.Is(err, errInvalidBAL) { + t.Fatalf("unexpected delivery result, accepted %d, err %v", accepted, err) + } + if got, exp := q.PendingBALs(), 4; got != exp { + t.Errorf("wrong pending access list count, got %d, exp %d", got, exp) + } + // Retrieve the completed blocks: the delivered access list must be attached + // and all remaining tasks pruned as obsolete + results := q.Results(false) + if got, exp := len(results), 10; got != exp { + t.Fatalf("wrong result count, got %d, exp %d", got, exp) + } + for i, result := range results { + if list := result.BAL(); (list != nil) != (i == 5) { + t.Errorf("block %d: unexpected access list attachment: %v", i+1, list) + } + } + if got, exp := q.PendingBALs(), 0; got != exp { + t.Errorf("wrong pending access list count after delivery, got %d, exp %d", got, exp) + } + // The access list memory allowance must have drained with the delivery + if got, exp := q.balBytes.Load(), int64(0); got != exp { + t.Errorf("wrong attached access list bytes after delivery, got %d, exp %d", got, exp) + } +} + // XTestDelivery does some more extensive testing of events that happen, // blocks that become known and peers that make reservations and deliveries. // disabled since it's not really a unit-test, but can be executed to test diff --git a/eth/downloader/resultstore.go b/eth/downloader/resultstore.go index 36c382fefc..e27df98b64 100644 --- a/eth/downloader/resultstore.go +++ b/eth/downloader/resultstore.go @@ -76,7 +76,7 @@ func (r *resultStore) SetThrottleThreshold(threshold uint64) uint64 { // throttled - if true, the store is at capacity, this particular header is not prio now // item - the result to store data into // err - any error that occurred -func (r *resultStore) AddFetch(header *types.Header, snapSync bool) (stale, throttled bool, item *fetchResult, err error) { +func (r *resultStore) AddFetch(header *types.Header, snapSync bool, fetchBAL bool) (stale, throttled bool, item *fetchResult, err error) { r.lock.Lock() defer r.lock.Unlock() @@ -86,7 +86,7 @@ func (r *resultStore) AddFetch(header *types.Header, snapSync bool) (stale, thro return stale, throttled, item, err } if item == nil { - item = newFetchResult(header, snapSync) + item = newFetchResult(header, snapSync, fetchBAL) r.items[index] = item } return stale, throttled, item, err diff --git a/eth/downloader/skeleton_test.go b/eth/downloader/skeleton_test.go index 9962d4f1a3..d10a07dfbe 100644 --- a/eth/downloader/skeleton_test.go +++ b/eth/downloader/skeleton_test.go @@ -212,6 +212,10 @@ func (p *skeletonTestPeer) RequestReceipts([]common.Hash, []uint64, []uint64, ch panic("skeleton sync must not request receipts") } +func (p *skeletonTestPeer) RequestBALs([]common.Hash, chan *eth.Response) (*eth.Request, error) { + panic("skeleton sync must not request block access lists") +} + // Tests various sync initializations based on previous leftovers in the database // and announced heads. func TestSkeletonSyncInit(t *testing.T) {