execution: make block read aheader non-global (#21056)

# execution: make block read aheader non-global

## Summary

Replaces the package-level `globalReadAheader` in
`execution/exec/blocks_read_ahead.go` with a per-`Ethereum` instance.
Each backend now owns its own `*BlockReadAheader` (caches + warmup
`WaitGroup`) and threads it through to the consumers that need it.

## Why

The `globalReadAheader` is a process-wide singleton. The exec module's
`ValidateChain` calls `AddHeaderAndBody`, which spawns a fire-and-forget
warmup goroutine tracked on a shared `sync.WaitGroup`. The eth backend's
`Stop()` waits on that same WaitGroup before closing the chain DB. With
**multiple `Ethereum` instances in one process** (e.g. the new `evm
enginextest` runner that creates ~36 k testers in a single run, or any
other harness that spins up many backends), `Add(1)` from one tester
races with `Wait()` returning in another. Once the counter momentarily
hits zero, a concurrent `Add(1)` panics:

```
panic: sync: WaitGroup is reused before previous Wait has returned
```

This was reproducible at `--workers ≥ 16` running `evm enginextest`
against the full EEST `blockchain_tests_engine_x` set on tmpfs.
Single-Erigon-per-process production deployments never trigger it, which
is why it has been latent.

## Approach

Make the read-aheader a regular struct owned by each `Ethereum`. The
lifetime is the same as the eth backend, the WaitGroup never escapes the
backend, and the race vanishes.

## Files (13 changed, +69 / −58)

- `execution/exec/blocks_read_ahead.go` — exports `BlockReadAheader` and
`NewBlockReadAheader()`. Methods replace the package-level `*Global*`
wrappers (`AddHeaderAndBodyToGlobalReadAheader`,
`AddSendersToGlobalReadAheader`,
`ReadBodyWithTransactionsFromGlobalReadAheader`,
`ReadBlockWithSendersFromGlobalReadAheader`, `WaitForWarmup`). The
unused `WarmBodyFromGlobalReadAheader` is dropped.
- `node/eth/backend.go` — `Ethereum.readAheader` field, initialized in
`New` to `exec.NewBlockReadAheader()`, used by `Stop()` and threaded
into the staged-sync configs and `NewExecModule`.
- `execution/execmodule/exec_module.go` — `ExecModule.readAheader`
field; new arg on `NewExecModule`. `ValidateChain` now calls
`e.readAheader.AddHeaderAndBody(...)`.
- `execution/stagedsync/stage_senders.go` (+ `stage_senders_test.go`) —
`SendersCfg.readAheader`; new arg on `StageSendersCfg`; calls
`cfg.readAheader.{AddSenders, ReadBodyWithTransactions}`.
- `execution/stagedsync/stage_execute.go` —
`ExecuteBlockCfg.readAheader`; new arg on `StageExecuteBlocksCfg`.
- `execution/stagedsync/exec3.go`, `exec3_serial.go` — call
`te.cfg.readAheader.ReadBlockWithSenders(...)` /
`se.cfg.readAheader.ReadBlockWithSenders(...)`.
- `execution/stagedsync/stage_witness.go` — passes a fresh
`exec.NewBlockReadAheader()` (rewind-only path, doesn't share with the
live backend).
- `execution/stagedsync/stageloop/stageloop.go` — `NewDefaultStages` /
`NewPipelineStages` / `NewInMemoryExecution` accept `readAheader` and
plumb it into the senders + execute stage configs.
- `execution/execmodule/execmoduletester/exec_module_tester.go` —
constructs one `readAheader` and reuses it across the tester's stage
configs and `NewExecModule` call (mirroring how `Ethereum` shares a
single instance).
- `cmd/integration/commands/stages.go`, `state_stages.go` — pass fresh
`exec.NewBlockReadAheader()` instances at the integration-CLI call sites
(each invocation is its own short-lived process).

No runtime semantics change inside the read-aheader itself: same caches,
same one-warmup-at-a-time `atomic.Bool`, same `WaitForWarmup` behaviour.
The only change is that the WaitGroup is per-instance rather than
process-wide.

## Verification

- `make lint` clean.
- **`evm enginextest`** against the full EEST
`blockchain_tests_engine_x` set (`fixtures_develop` v5.4.0), `--workers
8` on `TMPDIR=/dev/shm`:

  | metric | value |
  |--------|-------|
  | Tests passed | **63920 / 63920** |
  | Wall time | 5:10 |
  | Peak RSS | 5.0 GB |

  No `WaitGroup` panic at any worker count tested (8, 16, 24).
- **`make test-all`** with
`ERIGON_EXECUTION_TESTS_TMPDIR=/mnt/erigon-ramdisk GOGC=80`: **222
packages green, 0 failures.** All packages this PR touches pass
directly:
  - `execution/exec` (0.10s)
  - `execution/execmodule` (0.83s), `execmoduletester` (0.29s)
- `execution/stagedsync` (7.18s), `stagedsync/bodydownload`,
`stagedsync/headerdownload`
- `execution/tests` (212.89s, contains the previously-flaky
`TestInvalidReceiptHashHighMgas`)
- Leak-loop tooling test (build tag `leak`, in the engine-x test runner
package) shows goroutine count flat at 3 across 15 iterations — the
read-aheader's warmup goroutine terminates cleanly per Close.

## Notes

- The integration commands (`cmd/integration/commands/stages.go`,
`state_stages.go`) and the `RewindStagesForWitness` helper construct
fresh `exec.NewBlockReadAheader()` instances rather than threading one
through. Each is a one-shot, short-lived process where the cache
wouldn't carry value across the call. If we ever want shared caching
there, those sites have a clear seam to do it.
- Production single-instance Erigon behaviour is unchanged: same caches,
same warmup semantics, same `Stop()` wait, same `dbg.ReadAhead` gate
inside `warmBody`.

Author: taratorio

cmd/integration/commands/stages.go

@@ -606,7 +606,7 @@ func stageSenders(db kv.TemporalRwDB, ctx context.Context, logger log.Logger) er
 		return err
 	}
 
-	cfg := stagedsync.StageSendersCfg(chainConfig, sync.Cfg(), false /* badBlockHalt */, tmpdir, pm, br, nil /* hd */)
+	cfg := stagedsync.StageSendersCfg(chainConfig, sync.Cfg(), false /* badBlockHalt */, tmpdir, pm, br, nil /* hd */, exec.NewBlockReadAheader())
 	if unwind > 0 {
 		if unwind > s.BlockNumber {
 			return errors.New("cannot unwind past 0")
@@ -680,7 +680,7 @@ func stageExec(db kv.TemporalRwDB, ctx context.Context, logger log.Logger) error
 	cfg := stagedsync.StageExecuteBlocksCfg(db, pm, batchSize, chainConfig, engine, vmConfig, notifications,
 		/*stateStream=*/ false,
 		/*badBlockHalt=*/ true,
-		dirs, br, nil, genesis, syncCfg, false /*experimentalBAL*/)
+		dirs, br, nil, genesis, syncCfg, false /*experimentalBAL*/, exec.NewBlockReadAheader())
 
 	if unwind > 0 {
 		if err := db.ViewTemporal(ctx, func(tx kv.TemporalTx) error {
@@ -1258,7 +1258,7 @@ func newSync(ctx context.Context, db kv.TemporalRwDB, builderConfig *buildercfg.
 	}
 	notifications := shards.NewNotifications(nil)
 	blockRetire := freezeblocks.NewBlockRetire(estimate.CompressSnapshot.Workers(), dirs, blockReader, blockWriter, db, heimdallStore, bridgeStore, chainConfig, &cfg, notifications.Events, blockSnapBuildSema, logger)
-	stageList := stageloop.NewDefaultStages(context.Background(), db, p2p.Config{}, &cfg, sentryControlServer, notifications, nil, blockReader, blockRetire, nil, nil)
+	stageList := stageloop.NewDefaultStages(context.Background(), db, p2p.Config{}, &cfg, sentryControlServer, notifications, nil, blockReader, blockRetire, nil, nil, exec.NewBlockReadAheader())
 	sync := stagedsync.New(cfg.Sync, stageList, stagedsync.DefaultUnwindOrder, stagedsync.DefaultPruneOrder, logger, stages.ModeApplyingBlocks)
 	return blockRetire, engine, vmConfig, sync
 }

cmd/integration/commands/state_stages.go

@@ -37,6 +37,7 @@ import (
 	"github.com/erigontech/erigon/db/state/execctx"
 	"github.com/erigontech/erigon/execution/builder/buildercfg"
 	chainspec "github.com/erigontech/erigon/execution/chain/spec"
+	"github.com/erigontech/erigon/execution/exec"
 	"github.com/erigontech/erigon/execution/stagedsync"
 	"github.com/erigontech/erigon/execution/stagedsync/stages"
 	"github.com/erigontech/erigon/execution/tracing/tracers/logger"
@@ -183,7 +184,7 @@ func syncBySmallSteps(db kv.TemporalRwDB, builderConfig buildercfg.BuilderConfig
 	}
 
 	br, _ := blocksIO(db, logger1)
-	execCfg := stagedsync.StageExecuteBlocksCfg(db, pm, batchSize, chainConfig, engine, vmConfig, notifications, false, true, dirs, br, nil, spec.Genesis, syncCfg, false)
+	execCfg := stagedsync.StageExecuteBlocksCfg(db, pm, batchSize, chainConfig, engine, vmConfig, notifications, false, true, dirs, br, nil, spec.Genesis, syncCfg, false, exec.NewBlockReadAheader())
 
 	execUntilFunc := func(execToBlock uint64) stagedsync.ExecFunc {
 		return func(badBlockUnwind bool, s *stagedsync.StageState, unwinder stagedsync.Unwinder, doms *execctx.SharedDomains, rwTx kv.TemporalRwTx, logger log.Logger) error {
@@ -369,7 +370,7 @@ func loopExec(db kv.TemporalRwDB, ctx context.Context, unwind uint64, logger log
 	initialCycle := false
 	br, _ := blocksIO(db, logger)
 	notifications := shards.NewNotifications(nil)
-	cfg := stagedsync.StageExecuteBlocksCfg(db, pm, batchSize, chainConfig, engine, vmConfig, notifications, false, true, dirs, br, nil, spec.Genesis, syncCfg, false)
+	cfg := stagedsync.StageExecuteBlocksCfg(db, pm, batchSize, chainConfig, engine, vmConfig, notifications, false, true, dirs, br, nil, spec.Genesis, syncCfg, false, exec.NewBlockReadAheader())
 
 	// set block limit of execute stage
 	sync.MockExecFunc(stages.Execution, func(badBlockUnwind bool, stageState *stagedsync.StageState, unwinder stagedsync.Unwinder, sd *execctx.SharedDomains, tx kv.TemporalRwTx, logger log.Logger) error {

execution/exec/blocks_read_ahead.go

@@ -22,7 +22,7 @@ import (
 	"github.com/erigontech/erigon/execution/types/accounts"
 )
 
-type blockReadAheader struct {
+type BlockReadAheader struct {
 	// keeps some caches for block themselves
 	headers *lru.Cache[common.Hash, *types.Header]
 	bodies  *lru.Cache[common.Hash, *types.Body]
@@ -33,7 +33,7 @@ type blockReadAheader struct {
 	warmWg  sync.WaitGroup
 }
 
-func newBlockReadAheader() *blockReadAheader {
+func NewBlockReadAheader() *BlockReadAheader {
 	headers, err := lru.New[common.Hash, *types.Header](4)
 	if err != nil {
 		panic(err)
@@ -46,16 +46,14 @@ func newBlockReadAheader() *blockReadAheader {
 	if err != nil {
 		panic(err)
 	}
-	return &blockReadAheader{
+	return &BlockReadAheader{
 		headers: headers,
 		bodies:  bodies,
 		senders: senders,
 	}
 }
 
-var globalReadAheader = newBlockReadAheader()
-
-func (bra *blockReadAheader) AddHeaderAndBody(ctx context.Context, db kv.RoDB, header *types.Header, body *types.Body) {
+func (bra *BlockReadAheader) AddHeaderAndBody(ctx context.Context, db kv.RoDB, header *types.Header, body *types.Body) {
 	blockHash := header.Hash()
 	bra.headers.Add(blockHash, header)
 	bra.bodies.Add(blockHash, body)
@@ -75,10 +73,10 @@ func (bra *blockReadAheader) AddHeaderAndBody(ctx context.Context, db kv.RoDB, h
 // WaitForWarmup blocks until any in-flight warmBody goroutine finishes or
 // the context is cancelled. Call before closing the database to avoid
 // waitTxsAllDoneOnClose hangs.
-func WaitForWarmup(ctx context.Context) {
+func (bra *BlockReadAheader) WaitForWarmup(ctx context.Context) {
 	done := make(chan struct{})
 	go func() {
-		globalReadAheader.warmWg.Wait()
+		bra.warmWg.Wait()
 		close(done)
 	}()
 	select {
@@ -87,26 +85,18 @@ func WaitForWarmup(ctx context.Context) {
 	}
 }
 
-func (bra *blockReadAheader) AddSenders(senders []byte, blockHash common.Hash) {
+func (bra *BlockReadAheader) AddSenders(senders []byte, blockHash common.Hash) {
 	if _, ok := bra.bodies.Get(blockHash); !ok {
 		return
 	}
 	bra.senders.Add(blockHash, common.Copy(senders))
 }
 
-func AddHeaderAndBodyToGlobalReadAheader(ctx context.Context, db kv.RoDB, header *types.Header, body *types.Body) {
-	globalReadAheader.AddHeaderAndBody(ctx, db, header, body)
-}
-
-func AddSendersToGlobalReadAheader(senders []byte, blockHash common.Hash) {
-	globalReadAheader.AddSenders(senders, blockHash)
-}
-
 // warmBody warms state for all transactions in a body using multiple workers.
 // It reads: To accounts, To account code, To account storage from access lists,
 // and block-level access lists. Each worker creates its own transaction.
 // Only one warmBody can run at a time - concurrent calls are no-ops.
-func (bra *blockReadAheader) warmBody(ctx context.Context, db kv.RoDB, header *types.Header, body *types.Body, workers int) {
+func (bra *BlockReadAheader) warmBody(ctx context.Context, db kv.RoDB, header *types.Header, body *types.Body, workers int) {
 	defer bra.warming.Store(false)
 
 	if !dbg.ReadAhead {
@@ -269,24 +259,20 @@ func (bra *blockReadAheader) warmBody(ctx context.Context, db kv.RoDB, header *t
 	wg.Wait()
 }
 
-func WarmBodyFromGlobalReadAheader(ctx context.Context, db kv.RoDB, header *types.Header, body *types.Body, workers int) {
-	globalReadAheader.warmBody(ctx, db, header, body, workers)
-}
-
-func ReadBodyWithTransactionsFromGlobalReadAheader(blockHash common.Hash) (*types.Body, bool) {
-	return globalReadAheader.bodies.Get(blockHash)
+func (bra *BlockReadAheader) ReadBodyWithTransactions(blockHash common.Hash) (*types.Body, bool) {
+	return bra.bodies.Get(blockHash)
 }
 
-func ReadBlockWithSendersFromGlobalReadAheader(blockHash common.Hash) (*types.Block, bool) {
-	header, ok := globalReadAheader.headers.Get(blockHash)
+func (bra *BlockReadAheader) ReadBlockWithSenders(blockHash common.Hash) (*types.Block, bool) {
+	header, ok := bra.headers.Get(blockHash)
 	if header == nil || !ok {
 		return nil, false
 	}
-	body, ok := globalReadAheader.bodies.Get(blockHash)
+	body, ok := bra.bodies.Get(blockHash)
 	if body == nil || !ok {
 		return nil, false
 	}
-	senders, ok := globalReadAheader.senders.Get(blockHash)
+	senders, ok := bra.senders.Get(blockHash)
 	if len(senders) == 0 || !ok {
 		return nil, false
 	}

execution/execmodule/exec_module.go

@@ -225,7 +225,8 @@ type ExecModule struct {
 	publishedSD    func() *execctx.SharedDomains // fallback for background commit
 
 	// stateCache is a cache for state data (accounts, storage, code)
-	stateCache *cache.StateCache
+	stateCache  *cache.StateCache
+	readAheader *exec.BlockReadAheader
 
 	stopNode func() error
 }
@@ -249,6 +250,7 @@ func NewExecModule(
 	fcuBackgroundPrune bool,
 	fcuBackgroundCommit bool,
 	onlySnapDownloadOnStart bool,
+	readAheader *exec.BlockReadAheader,
 	stopNode func() error,
 ) *ExecModule {
 	domainCache := cache.NewDefaultStateCache()
@@ -273,6 +275,7 @@ func NewExecModule(
 		fcuBackgroundCommit:     fcuBackgroundCommit,
 		onlySnapDownloadOnStart: onlySnapDownloadOnStart,
 		stateCache:              domainCache,
+		readAheader:             readAheader,
 		stopNode:                stopNode,
 	}
 
@@ -418,7 +421,7 @@ func (e *ExecModule) ValidateChain(ctx context.Context, blockHash common.Hash, b
 		if err != nil {
 			return ValidationResult{}, err
 		}
-		exec.AddHeaderAndBodyToGlobalReadAheader(ctx, e.db, header, body)
+		e.readAheader.AddHeaderAndBody(ctx, e.db, header, body)
 		currentBlockNumber = rawdb.ReadCurrentBlockNumber(overlay)
 	} else {
 		if err := e.db.View(ctx, func(tx kv.Tx) error {
@@ -431,7 +434,7 @@ func (e *ExecModule) ValidateChain(ctx context.Context, blockHash common.Hash, b
 			if err != nil {
 				return err
 			}
-			exec.AddHeaderAndBodyToGlobalReadAheader(ctx, e.db, header, body)
+			e.readAheader.AddHeaderAndBody(ctx, e.db, header, body)
 			currentBlockNumber = rawdb.ReadCurrentBlockNumber(tx)
 			return nil
 		}); err != nil {

execution/execmodule/execmoduletester/exec_module_tester.go

@@ -54,6 +54,7 @@ import (
 	dbstate "github.com/erigontech/erigon/db/state"
 	"github.com/erigontech/erigon/execution/builder"
 	"github.com/erigontech/erigon/execution/chain"
+	"github.com/erigontech/erigon/execution/exec"
 	"github.com/erigontech/erigon/execution/execmodule"
 	"github.com/erigontech/erigon/execution/execmodule/chainreader"
 	"github.com/erigontech/erigon/execution/protocol/rules"
@@ -616,6 +617,7 @@ func New(tb testing.TB, opts ...Option) *ExecModuleTester {
 		close(miningCancel)
 	}()
 
+	readAheader := exec.NewBlockReadAheader()
 	blkBuilder := builder.NewBuilder(
 		mock.Ctx,
 		mock.DB,
@@ -639,6 +641,7 @@ func New(tb testing.TB, opts ...Option) *ExecModuleTester {
 			gspec,
 			cfg.Sync,
 			false, /*experimentalBAL*/
+			readAheader,
 		),
 		nil, /*notifier*/
 		&vm.Config{},
@@ -659,7 +662,7 @@ func New(tb testing.TB, opts ...Option) *ExecModuleTester {
 			stagedsync.StageHeadersCfg(mock.sentriesClient.Hd, mock.ChainConfig, cfg.Sync, sendHeaderRequest, propagateNewBlockHashes, penalize, false /* noP2PDiscovery */, mock.BlockReader),
 			stagedsync.StageBlockHashesCfg(mock.Dirs.Tmp, blockWriter),
 			stagedsync.StageBodiesCfg(mock.sentriesClient.Bd, sendBodyRequest, penalize, blockPropagator, cfg.Sync.BodyDownloadTimeoutSeconds, mock.ChainConfig, mock.BlockReader, blockWriter),
-			stagedsync.StageSendersCfg(mock.ChainConfig, cfg.Sync, false /* badBlockHalt */, dirs.Tmp, pruneMode, mock.BlockReader, mock.sentriesClient.Hd),
+			stagedsync.StageSendersCfg(mock.ChainConfig, cfg.Sync, false /* badBlockHalt */, dirs.Tmp, pruneMode, mock.BlockReader, mock.sentriesClient.Hd, readAheader),
 			stagedsync.StageExecuteBlocksCfg(
 				mock.DB,
 				pruneMode,
@@ -676,6 +679,7 @@ func New(tb testing.TB, opts ...Option) *ExecModuleTester {
 				gspec,
 				cfg.Sync,
 				false, /*experimentalBAL*/
+				readAheader,
 			),
 			stagedsync.StageTxLookupCfg(pruneMode, dirs.Tmp, mock.BlockReader),
 			stagedsync.StageFinishCfg(),
@@ -695,13 +699,13 @@ func New(tb testing.TB, opts ...Option) *ExecModuleTester {
 	}
 
 	cfg.Genesis = gspec
-	pipelineStages := stageloop.NewPipelineStages(mock.Ctx, db, &cfg, mock.sentriesClient, mock.Notifications, snapDownloader, mock.BlockReader, blockRetire, tracer, nil)
+	pipelineStages := stageloop.NewPipelineStages(mock.Ctx, db, &cfg, mock.sentriesClient, mock.Notifications, snapDownloader, mock.BlockReader, blockRetire, tracer, nil, readAheader)
 	mock.posStagedSync = stagedsync.New(cfg.Sync, pipelineStages, stagedsync.PipelineUnwindOrder, stagedsync.PipelinePruneOrder, logger, stages.ModeApplyingBlocks)
 
 	// Create validation Sync and PipelineExecutor.
 	validationNotifications := shards.NewNotifications(nil)
 	validationSync := stageloop.NewInMemoryExecution(mock.Ctx, mock.DB, &cfg, mock.sentriesClient,
-		validationNotifications, mock.BlockReader, blockWriter, logger)
+		validationNotifications, mock.BlockReader, blockWriter, logger, readAheader)
 	dispatcher := execmodule.NewDispatcher(mock.ChainConfig, mock.Notifications.Events, mock.Notifications.StateChangesConsumer, logger)
 	pipelineExecutor := execmodule.NewPipelineExecutor(mock.posStagedSync, mock.DB, mock.BlockReader, mock.ChainConfig, mock.Engine, validationSync, validationNotifications, dispatcher, logger)
 
@@ -731,6 +735,7 @@ func New(tb testing.TB, opts ...Option) *ExecModuleTester {
 		cfg.FcuBackgroundPrune,
 		cfg.FcuBackgroundCommit,
 		onlySnapDownloadOnStart,
+		readAheader,
 		func() error { return nil },
 	)
 	mock.ForkValidator = mock.ExecModule.ForkValidator()

execution/stagedsync/exec3.go

@@ -583,7 +583,7 @@ func (te *txExecutor) executeBlocks(ctx context.Context, startBlockNum uint64, m
 			if err != nil {
 				return err
 			}
-			b, ok := exec.ReadBlockWithSendersFromGlobalReadAheader(canonicalHash)
+			b, ok := te.cfg.readAheader.ReadBlockWithSenders(canonicalHash)
 			if b == nil || !ok {
 				b, err = exec.BlockWithSenders(ctx, te.cfg.db, blockTx, te.cfg.blockReader, blockNum)
 			}

execution/stagedsync/exec3_serial.go

@@ -101,7 +101,7 @@ func (se *serialExecutor) exec(ctx context.Context, execStage *StageState, u Unw
 			return nil, rwTx, err
 		}
 		var ok bool
-		b, ok = exec.ReadBlockWithSendersFromGlobalReadAheader(canonicalHash)
+		b, ok = se.cfg.readAheader.ReadBlockWithSenders(canonicalHash)
 		if b == nil || !ok {
 			b, err = exec.BlockWithSenders(ctx, se.cfg.db, se.applyTx, se.cfg.blockReader, blockNum)
 			if err != nil {

execution/stagedsync/stage_execute.go

@@ -43,6 +43,7 @@ import (
 	"github.com/erigontech/erigon/db/state/execctx"
 	"github.com/erigontech/erigon/diagnostics/metrics"
 	"github.com/erigontech/erigon/execution/chain"
+	"github.com/erigontech/erigon/execution/exec"
 	"github.com/erigontech/erigon/execution/protocol/rules"
 	"github.com/erigontech/erigon/execution/stagedsync/stages"
 	"github.com/erigontech/erigon/execution/types"
@@ -85,6 +86,7 @@ type ExecuteBlockCfg struct {
 	genesis   *types.Genesis
 
 	experimentalBAL bool
+	readAheader     *exec.BlockReadAheader
 }
 
 func StageExecuteBlocksCfg(
@@ -104,6 +106,7 @@ func StageExecuteBlocksCfg(
 	genesis *types.Genesis,
 	syncCfg ethconfig.Sync,
 	experimentalBAL bool,
+	readAheader *exec.BlockReadAheader,
 ) ExecuteBlockCfg {
 	if dirs.SnapDomain == "" {
 		panic("empty `dirs` variable")
@@ -126,6 +129,7 @@ func StageExecuteBlocksCfg(
 		historyV3:       true,
 		syncCfg:         syncCfg,
 		experimentalBAL: experimentalBAL,
+		readAheader:     readAheader,
 	}
 }
 

execution/stagedsync/stage_senders.go

@@ -56,9 +56,10 @@ type SendersCfg struct {
 	chainConfig     *chain.Config
 	hd              *headerdownload.HeaderDownload
 	blockReader     services.FullBlockReader
+	readAheader     *exec.BlockReadAheader
 }
 
-func StageSendersCfg(chainCfg *chain.Config, syncCfg ethconfig.Sync, badBlockHalt bool, tmpdir string, prune prune.Mode, blockReader services.FullBlockReader, hd *headerdownload.HeaderDownload) SendersCfg {
+func StageSendersCfg(chainCfg *chain.Config, syncCfg ethconfig.Sync, badBlockHalt bool, tmpdir string, prune prune.Mode, blockReader services.FullBlockReader, hd *headerdownload.HeaderDownload, readAheader *exec.BlockReadAheader) SendersCfg {
 	const sendersBatchSize = 1000
 	return SendersCfg{
 		batchSize:       sendersBatchSize,
@@ -68,6 +69,7 @@ func StageSendersCfg(chainCfg *chain.Config, syncCfg ethconfig.Sync, badBlockHal
 		chainConfig:     chainCfg,
 		hd:              hd,
 		blockReader:     blockReader,
+		readAheader:     readAheader,
 	}
 }
 
@@ -167,7 +169,7 @@ func SpawnRecoverSendersStage(cfg SendersCfg, s *StageState, u Unwinder, tx kv.R
 
 				// Check if block is complete
 				if pb.received == pb.txCount {
-					exec.AddSendersToGlobalReadAheader(pb.senders, pb.hash)
+					cfg.readAheader.AddSenders(pb.senders, pb.hash)
 					if err := collectorSenders.Collect(dbutils.BlockBodyKey(s.BlockNumber+uint64(j.blockIndex)+1, pb.hash), pb.senders); err != nil {
 						pendingMu.Unlock()
 						errCh <- senderRecoveryError{err: err}
@@ -238,7 +240,7 @@ Loop:
 			continue
 		}
 
-		body, ok := exec.ReadBodyWithTransactionsFromGlobalReadAheader(blockHash)
+		body, ok := cfg.readAheader.ReadBodyWithTransactions(blockHash)
 		if body == nil || !ok {
 			if body, err = cfg.blockReader.BodyWithTransactions(ctx, tx, blockHash, blockNumber); err != nil {
 				return err

execution/stagedsync/stage_senders_test.go

@@ -30,6 +30,7 @@ import (
 	"github.com/erigontech/erigon/db/kv/prune"
 	"github.com/erigontech/erigon/db/rawdb"
 	"github.com/erigontech/erigon/execution/chain"
+	"github.com/erigontech/erigon/execution/exec"
 	"github.com/erigontech/erigon/execution/execmodule/execmoduletester"
 	"github.com/erigontech/erigon/execution/stagedsync"
 	"github.com/erigontech/erigon/execution/stagedsync/stages"
@@ -145,7 +146,7 @@ func TestSenders(t *testing.T) {
 
 	require.NoError(stages.SaveStageProgress(tx, stages.Bodies, 3))
 
-	cfg := stagedsync.StageSendersCfg(chain.TestChainBerlinConfig, ethconfig.Defaults.Sync, false, "", prune.Mode{}, br, nil)
+	cfg := stagedsync.StageSendersCfg(chain.TestChainBerlinConfig, ethconfig.Defaults.Sync, false, "", prune.Mode{}, br, nil, exec.NewBlockReadAheader())
 	err = stagedsync.SpawnRecoverSendersStage(cfg, &stagedsync.StageState{ID: stages.Senders}, nil, tx, 3, m.Ctx, log.New())
 	require.NoError(err)