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

@@ -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
+	// Block access lists are only retrieved (best effort) for blocks close to
-	// (it may still reachable, but it's not worthwhile to even retrieve it
+	// the head of the network chain; a nil entry is stored as the absence
-	// from the network). A nil entry is stored in the BAL table as the absence
+	// placeholder for anything the network no longer serves.
-	// placeholder.
+	if list := block.AccessList(); list != nil {
-	if err := op.AppendRaw(ChainFreezerBALTable, num, nil); err != nil {
+		if err := op.Append(ChainFreezerBALTable, num, list); err != nil {
-		return fmt.Errorf("can't append block %d bals: %v", num, err)
+			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
 }

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

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{
+	chain, err := core.NewBlockChain(db, gspec, engine, nil)
-		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)
 	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 := &eth.Response{
+		Req:  &eth.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) }

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

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 <http://www.gnu.org/licenses/>.
+
+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
+}

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

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
+	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)
+	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,
+		id:         id,
-		lacking: make(map[common.Hash]struct{}),
+		lacking:    make(map[common.Hash]struct{}),
-		peer:    peer,
+		lackingBAL: make(map[common.Hash]struct{}),
-		version: version,
+		peer:       peer,
-		log:     logger,
+		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 {

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<<bodyType)|(1<<receiptType)) == 0
 }
 
 // SetReceiptsDone flags the receipts as finished.
@@ -121,6 +171,13 @@ func (f *fetchResult) SetReceiptsDone() {
 	}
 }
 
+// SetBALDone flags the block access list as finished.
+func (f *fetchResult) SetBALDone() {
+	if v := f.pending.Load(); (v & (1 << balType)) != 0 {
+		f.pending.Add(-4)
+	}
+}
+
 // Done checks if the given type is done already
 func (f *fetchResult) Done(kind uint) bool {
 	v := f.pending.Load()
@@ -143,8 +200,15 @@ type queue struct {
 	receiptPendPool  map[string]*fetchRequest           // Currently pending receipt retrieval operations
 	receiptWakeCh    chan bool                          // Channel to notify when receipt fetcher of new tasks
 
+	balTaskPool  map[common.Hash]*types.Header      // Pending block access list retrieval tasks, mapping hashes to headers
+	balTaskQueue *prque.Prque[int64, *types.Header] // Priority queue of the headers to fetch the access lists for
+	balPendPool  map[string]*fetchRequest           // Currently pending access list retrieval operations
+	balWakeCh    chan bool                          // Channel to notify the access list fetcher of new tasks
+	balCutoff    uint64                             // Minimum block number for which access lists are attempted (best effort window below the network head)
+
 	resultCache *resultStore       // Downloaded but not yet delivered fetch results
 	resultSize  common.StorageSize // Approximate size of a block (exponential moving average)
+	balBytes    atomic.Int64       // Exact encoded bytes of access lists attached to cached results
 
 	lock   *sync.RWMutex
 	active *sync.Cond
@@ -161,6 +225,8 @@ func newQueue(blockCacheLimit int, thresholdInitialSize int) *queue {
 		blockWakeCh:      make(chan bool, 1),
 		receiptTaskQueue: prque.New[int64, *types.Header](nil),
 		receiptWakeCh:    make(chan bool, 1),
+		balTaskQueue:     prque.New[int64, *types.Header](nil),
+		balWakeCh:        make(chan bool, 1),
 		active:           sync.NewCond(lock),
 		lock:             lock,
 	}
@@ -185,6 +251,12 @@ func (q *queue) Reset(blockCacheLimit int, thresholdInitialSize int) {
 	q.receiptTaskQueue.Reset()
 	q.receiptPendPool = make(map[string]*fetchRequest)
 
+	q.balTaskPool = make(map[common.Hash]*types.Header)
+	q.balTaskQueue.Reset()
+	q.balPendPool = make(map[string]*fetchRequest)
+	q.balCutoff = 0
+	q.balBytes.Store(0)
+
 	q.resultCache = newResultStore(blockCacheLimit)
 	q.resultCache.SetThrottleThreshold(uint64(thresholdInitialSize))
 }
@@ -214,6 +286,38 @@ func (q *queue) PendingReceipts() int {
 	return q.receiptTaskQueue.Size()
 }
 
+// PendingBALs retrieves the number of block access lists pending for retrieval.
+func (q *queue) PendingBALs() int {
+	q.lock.Lock()
+	defer q.lock.Unlock()
+
+	return q.balTaskQueue.Size()
+}
+
+// SetBALCutoff updates the minimum block number for which block access lists
+// are attempted to be downloaded. Access lists further below the head of the
+// network chain are not guaranteed to be retained by the network, so fetching
+// them is not even attempted.
+func (q *queue) SetBALCutoff(cutoff uint64) {
+	q.lock.Lock()
+	defer q.lock.Unlock()
+
+	q.balCutoff = cutoff
+}
+
+// balEligible reports whether the access list of the given block should be
+// scheduled for retrieval. Only post-Amsterdam blocks within the recency
+// window below the network head are attempted, and known-empty access lists
+// are not worth a network retrieval.
+//
+// Note, this method expects the queue lock to be already held.
+func (q *queue) balEligible(header *types.Header) bool {
+	if header.BlockAccessListHash == nil || *header.BlockAccessListHash == types.EmptyBlockAccessListHash {
+		return false
+	}
+	return header.Number.Uint64() >= 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

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,
+		id:         id,
-		lacking: make(map[common.Hash]struct{}),
+		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

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

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