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) {