Fix/bal selfdestruct netzero (#19528)

The PR reinstates the BlockAssembler and the move of BAL processing to
the versionedio code rather than as an auxiliary set of functions in the
sync process.

The main advantage of this change is that it consolidates behaviour
which makes agent understanding more focussed and makes the bal code
re-usable in tests. This makes the codebase more streamlined avoids
breaks and confusion in test analysis.

It contains fixes merged by the claude agent from
#19434 and
#19525, along with fixes that
get to a running bal-devnet-2 node.

This is ready for merging.

---------

Co-authored-by: taratorio <94537774+taratorio@users.noreply.github.com>
Co-authored-by: yperbasis <andrey.ashikhmin@gmail.com>
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
Co-authored-by: Andrew Ashikhmin <34320705+yperbasis@users.noreply.github.com>

Author: 4 people

.gitignore

@@ -116,6 +116,8 @@ mdbx.lck
 *.prof
 
 .claude/settings.local.json
+.claude/projects/
+.claude/skills/**/skill.md
 
 # Prevent accidental commit of locally-built binaries
 /erigon

execution/builder/builderstages/exec.go

@@ -18,7 +18,6 @@ package builderstages
 
 import (
 	context0 "context"
-	"errors"
 	"fmt"
 	"sync/atomic"
 	"time"
@@ -28,7 +27,6 @@ import (
 
 	"github.com/erigontech/erigon/common"
 	"github.com/erigontech/erigon/common/dbg"
-	"github.com/erigontech/erigon/common/empty"
 	"github.com/erigontech/erigon/common/log/v3"
 	"github.com/erigontech/erigon/db/kv"
 	"github.com/erigontech/erigon/db/kv/membatchwithdb"
@@ -40,7 +38,6 @@ import (
 	"github.com/erigontech/erigon/execution/exec"
 	"github.com/erigontech/erigon/execution/metrics"
 	"github.com/erigontech/erigon/execution/protocol"
-	"github.com/erigontech/erigon/execution/protocol/aa"
 	"github.com/erigontech/erigon/execution/protocol/params"
 	"github.com/erigontech/erigon/execution/protocol/rules"
 	"github.com/erigontech/erigon/execution/stagedsync"
@@ -49,7 +46,6 @@ import (
 	"github.com/erigontech/erigon/execution/types"
 	"github.com/erigontech/erigon/execution/types/accounts"
 	"github.com/erigontech/erigon/execution/vm"
-	"github.com/erigontech/erigon/execution/vm/evmtypes"
 	"github.com/erigontech/erigon/txnprovider"
 )
 
@@ -100,24 +96,23 @@ func SpawnBuilderExecStage(ctx context0.Context, s *stagedsync.StageState, sd *e
 	chainID, _ := uint256.FromBig(cfg.chainConfig.ChainID)
 	logPrefix := s.LogPrefix()
 	current := cfg.builderState.BuiltBlock
-	needBAL := execCfg.ChainConfig().IsAmsterdam(current.Header.Time) || execCfg.IsExperimentalBAL()
 
 	stateReader := state.NewReaderV3(sd.AsGetter(tx))
 	ibs := state.New(stateReader)
 	defer ibs.Release(false)
 	ibs.SetTxContext(current.Header.Number.Uint64(), -1)
-	var balIO *state.VersionedIO
-	var systemReads state.ReadSet
-	var systemWrites state.VersionedWrites
-	var systemAccess state.AccessSet
-	if needBAL {
+
+	ba := exec.NewBlockAssembler(exec.AssemblerCfg{
+		ChainConfig:     cfg.chainConfig,
+		Engine:          cfg.engine,
+		BlockReader:     cfg.blockReader,
+		ExperimentalBAL: execCfg.IsExperimentalBAL(),
+	}, cfg.payloadId, current.ParentHeaderTime, current.Header, current.Uncles, current.Withdrawals)
+
+	if ba.HasBAL() {
 		ibs.SetVersionMap(state.NewVersionMap(nil))
-		balIO = &state.VersionedIO{}
 	}
-	// Clique consensus needs forced author in the evm context
-	//if cfg.chainConfig.Consensus == chain.CliqueConsensus {
-	//	execCfg.author = &cfg.builderState.BuilderConfig.Etherbase
-	//}
+
 	execCfg = execCfg.WithAuthor(accounts.InternAddress(cfg.builderState.BuilderConfig.Etherbase))
 
 	getHeader := func(hash common.Hash, number uint64) (*types.Header, error) {
@@ -135,19 +130,13 @@ func SpawnBuilderExecStage(ctx context0.Context, s *stagedsync.StageState, sd *e
 	}
 	defer simSd.Close()
 
-	chainReader := exec.NewChainReader(cfg.chainConfig, tx, cfg.blockReader, logger)
-
 	txNum, _, err := sd.SeekCommitment(ctx, tx)
 	if err != nil {
 		return err
 	}
 
-	protocol.InitializeBlockExecution(cfg.engine, chainReader, current.Header, cfg.chainConfig, ibs, &state.NoopWriter{}, logger, nil)
-	if needBAL {
-		systemReads = stagedsync.MergeReadSets(systemReads, ibs.VersionedReads())
-		systemWrites = stagedsync.MergeVersionedWrites(systemWrites, ibs.VersionedWrites(false))
-		systemAccess = systemAccess.Merge(ibs.AccessedAddresses())
-		ibs.ResetVersionedIO()
+	if err := ba.Initialize(ibs, tx, logger); err != nil {
+		return err
 	}
 
 	coinbase := accounts.InternAddress(cfg.builderState.BuilderConfig.Etherbase)
@@ -170,7 +159,7 @@ func SpawnBuilderExecStage(ctx context0.Context, s *stagedsync.StageState, sd *e
 		}
 
 		if len(txns) > 0 {
-			logs, stop, err := addTransactionsToBlock(ctx, logPrefix, current, cfg.chainConfig, cfg.vmConfig, getHeader, cfg.engine, txns, coinbase, ibs, balIO, interrupt, cfg.payloadId, logger)
+			logs, stop, err := ba.AddTransactions(ctx, getHeader, txns, coinbase, cfg.vmConfig, ibs, interrupt, logPrefix, logger)
 			if err != nil {
 				return err
 			}
@@ -194,29 +183,27 @@ func SpawnBuilderExecStage(ctx context0.Context, s *stagedsync.StageState, sd *e
 
 	metrics.UpdateBlockProducerProductionDelay(current.ParentHeaderTime, current.Header.Number.Uint64(), logger)
 
-	logger.Debug("SpawnBuilderExecStage", "block", current.Header.Number, "txn", current.Txns.Len(), "payload", cfg.payloadId)
-	if current.Uncles == nil {
-		current.Uncles = []*types.Header{}
+	logger.Debug("SpawnBuilderExecStage", "block", current.Header.Number, "txn", ba.Txns.Len(), "payload", cfg.payloadId)
+	if ba.Uncles == nil {
+		ba.Uncles = []*types.Header{}
 	}
-	if current.Txns == nil {
-		current.Txns = []types.Transaction{}
+	if ba.Txns == nil {
+		ba.Txns = []types.Transaction{}
 	}
-	if current.Receipts == nil {
-		current.Receipts = types.Receipts{}
+	if ba.Receipts == nil {
+		ba.Receipts = types.Receipts{}
 	}
 
-	if err := cfg.engine.Prepare(chainReader, current.Header, ibs); err != nil {
+	block, err := ba.AssembleBlock(stateReader, ibs, tx, logger)
+	if err != nil {
 		return err
 	}
 
-	var block *types.Block
-	if needBAL {
-		ibs.ResetVersionedIO()
-	}
-	block, current.Requests, err = protocol.FinalizeBlockExecution(cfg.engine, stateReader, current.Header, current.Txns, current.Uncles, &state.NoopWriter{}, cfg.chainConfig, ibs, current.Receipts, current.Withdrawals, chainReader, true, logger, nil)
-	if err != nil {
-		return fmt.Errorf("cannot finalize block execution: %s", err)
-	}
+	// Copy results back to BuiltBlock
+	current.Txns = ba.Txns
+	current.Receipts = ba.Receipts
+	current.Requests = ba.Requests
+	current.BlockAccessList = ba.BlockAccessList
 
 	// Note: This gets reset in BuilderFinish - but we need it here to
 	// process execv3 - when we remove that this becomes redundant
@@ -231,33 +218,15 @@ func SpawnBuilderExecStage(ctx context0.Context, s *stagedsync.StageState, sd *e
 	}
 
 	blockHeight := block.NumberU64()
-	if needBAL {
-		systemReads = stagedsync.MergeReadSets(systemReads, ibs.VersionedReads())
-		systemWrites = stagedsync.MergeVersionedWrites(systemWrites, ibs.VersionedWrites(false))
-		systemAccess = systemAccess.Merge(ibs.AccessedAddresses())
-		ibs.ResetVersionedIO()
-
-		systemVersion := state.Version{BlockNum: blockHeight, TxIndex: -1}
-		balIO.RecordReads(systemVersion, systemReads)
-		balIO.RecordWrites(systemVersion, systemWrites)
-		balIO.RecordAccesses(systemVersion, systemAccess)
-		current.BlockAccessList = stagedsync.CreateBAL(blockHeight, balIO, execCfg.DirsDataDir())
-		// Note: This gets reset in BuilderFinish - but we need it here to
-		// process execv3 - when we remove that this becomes redundant
-		hash := current.BlockAccessList.Hash()
-		header.BlockAccessListHash = &hash
-	} else {
-		// Note: This gets reset in BuilderFinish - but we need it here to
-		// process execv3 - when we remove that this becomes redundant
-		if execCfg.ChainConfig().IsAmsterdam(current.Header.Time) {
-			header.BlockAccessListHash = &empty.BlockAccessListHash
-		}
-	}
 
 	writeBlockForExecution := func(rwTx kv.TemporalRwTx) error {
 		if err = rawdb.WriteHeader(rwTx, block.Header()); err != nil {
 			return fmt.Errorf("cannot write header: %s", err)
 		}
+		// Verify header round-trips correctly through RLP marshaling
+		if readBack := rawdb.ReadHeader(rwTx, block.Hash(), blockHeight); readBack == nil {
+			return fmt.Errorf("header round-trip failed: written header for block %d (hash %x) not readable", blockHeight, block.Hash())
+		}
 		if err = rawdb.WriteCanonicalHash(rwTx, block.Hash(), blockHeight); err != nil {
 			return fmt.Errorf("cannot write canonical hash: %s", err)
 		}
@@ -290,7 +259,7 @@ func SpawnBuilderExecStage(ctx context0.Context, s *stagedsync.StageState, sd *e
 
 	// This flag will skip checking the state root
 	execS := &stagedsync.StageState{State: s.State, ID: stages.Execution, BlockNumber: blockHeight - 1}
-	forceParallel := dbg.Exec3Parallel /*|| cfg.chainConfig.IsAmsterdam(current.Header.Time)*/ // TODO Re-enable after bals testing
+	forceParallel := dbg.Exec3Parallel || cfg.chainConfig.IsAmsterdam(current.Header.Time)
 	execTx := tx
 	execSd := sd
 	var execCleanup func()
@@ -547,199 +516,6 @@ func filterBadTransactions(transactions []types.Transaction, chainID *uint256.In
 	return filtered, nil
 }
 
-func addTransactionsToBlock(
-	ctx context0.Context,
-	logPrefix string,
-	current *BuiltBlock,
-	chainConfig *chain.Config,
-	vmConfig *vm.Config,
-	getHeader func(hash common.Hash, number uint64) (*types.Header, error),
-	engine rules.Engine,
-	txns types.Transactions,
-	coinbase accounts.Address,
-	ibs *state.IntraBlockState,
-	balIO *state.VersionedIO,
-	interrupt *atomic.Bool,
-	payloadId uint64,
-	logger log.Logger,
-) (types.Logs, bool, error) {
-	header := current.Header
-	txnIdx := ibs.TxnIndex() + 1
-	gasPool := new(protocol.GasPool).AddGas(header.GasLimit - header.GasUsed)
-	if header.BlobGasUsed != nil {
-		gasPool.AddBlobGas(chainConfig.GetMaxBlobGasPerBlock(header.Time) - *header.BlobGasUsed)
-	}
-	signer := types.MakeSigner(chainConfig, header.Number.Uint64(), header.Time)
-
-	var coalescedLogs types.Logs
-	noop := state.NewNoopWriter()
-	recordTxIO := func() {
-		if balIO == nil {
-			return
-		}
-		version := ibs.Version()
-		balIO.RecordReads(version, ibs.VersionedReads())
-		balIO.RecordWrites(version, ibs.VersionedWrites(false))
-		balIO.RecordAccesses(version, ibs.AccessedAddresses())
-		ibs.ResetVersionedIO()
-	}
-	clearTxIO := func() {
-		if balIO == nil {
-			return
-		}
-		ibs.AccessedAddresses()
-		ibs.ResetVersionedIO()
-	}
-
-	var builderCommitTx = func(txn types.Transaction, coinbase accounts.Address, vmConfig *vm.Config, chainConfig *chain.Config, ibs *state.IntraBlockState, current *BuiltBlock) ([]*types.Log, error) {
-		ibs.SetTxContext(current.Header.Number.Uint64(), txnIdx)
-		gasSnap := gasPool.Gas()
-		blobGasSnap := gasPool.BlobGas()
-		snap := ibs.PushSnapshot()
-		defer ibs.PopSnapshot(snap)
-
-		if txn.Type() == types.AccountAbstractionTxType {
-			aaTxn := txn.(*types.AccountAbstractionTransaction)
-			blockContext := protocol.NewEVMBlockContext(header, protocol.GetHashFn(header, getHeader), engine, coinbase, chainConfig)
-			evm := vm.NewEVM(blockContext, evmtypes.TxContext{}, ibs, chainConfig, *vmConfig)
-			paymasterContext, validationGasUsed, err := aa.ValidateAATransaction(aaTxn, ibs, gasPool, header, evm, chainConfig)
-			if err != nil {
-				ibs.RevertToSnapshot(snap, err)
-				gasPool = new(protocol.GasPool).AddGas(gasSnap).AddBlobGas(blobGasSnap) // restore gasPool as well as ibs
-				return nil, err
-			}
-
-			status, gasUsed, err := aa.ExecuteAATransaction(aaTxn, paymasterContext, validationGasUsed, gasPool, evm, header, ibs)
-			if err != nil {
-				ibs.RevertToSnapshot(snap, err)
-				gasPool = new(protocol.GasPool).AddGas(gasSnap).AddBlobGas(blobGasSnap) // restore gasPool as well as ibs
-				return nil, err
-			}
-
-			header.GasUsed += gasUsed
-			logs := ibs.GetLogs(ibs.TxnIndex(), txn.Hash(), header.Number.Uint64(), header.Hash())
-			receipt := aa.CreateAAReceipt(txn.Hash(), status, gasUsed, header.GasUsed, header.Number.Uint64(), uint64(ibs.TxnIndex()), logs)
-
-			current.AddTxn(txn)
-			current.Receipts = append(current.Receipts, receipt)
-			return receipt.Logs, nil
-		}
-
-		gasUsed := protocol.NewGasUsed(header, current.Receipts.CumulativeGasUsed())
-		receipt, err := protocol.ApplyTransaction(chainConfig, protocol.GetHashFn(header, getHeader), engine, coinbase, gasPool, ibs, noop, header, txn, gasUsed, *vmConfig)
-		if err != nil {
-			ibs.RevertToSnapshot(snap, err)
-			gasPool = new(protocol.GasPool).AddGas(gasSnap).AddBlobGas(blobGasSnap) // restore gasPool as well as ibs
-			return nil, err
-		}
-		protocol.SetGasUsed(header, gasUsed)
-		current.AddTxn(txn)
-		current.Receipts = append(current.Receipts, receipt)
-		return receipt.Logs, nil
-	}
-
-	var stopped *time.Ticker
-	defer func() {
-		if stopped != nil {
-			stopped.Stop()
-		}
-	}()
-
-	done := false
-
-LOOP:
-	for _, txn := range txns {
-		// see if we need to stop now
-		if stopped != nil {
-			select {
-			case <-stopped.C:
-				done = true
-				break LOOP
-			default:
-			}
-		}
-
-		if err := common.Stopped(ctx.Done()); err != nil {
-			return nil, true, err
-		}
-
-		if interrupt != nil && interrupt.Load() && stopped == nil {
-			logger.Debug("Transaction adding was requested to stop", "payload", payloadId)
-			// ensure we run for at least 500ms after the request to stop comes in from GetPayload
-			stopped = time.NewTicker(500 * time.Millisecond)
-		}
-		// If we don't have enough gas for any further transactions then we're done
-		if gasPool.Gas() < params.TxGas {
-			logger.Debug(fmt.Sprintf("[%s] Not enough gas for further transactions", logPrefix), "have", gasPool, "want", params.TxGas)
-			done = true
-			break
-		}
-
-		rlpSpacePostTxn := current.AvailableRlpSpace(chainConfig, txn)
-		if rlpSpacePostTxn < 0 {
-			rlpSpacePreTxn := current.AvailableRlpSpace(chainConfig)
-			logger.Debug(
-				fmt.Sprintf("[%s] Skipping transaction since it does not fit in available rlp space", logPrefix),
-				"hash", txn.Hash(),
-				"pre", rlpSpacePreTxn,
-				"post", rlpSpacePostTxn,
-			)
-			continue
-		}
-
-		// We use the eip155 signer regardless of the env hf.
-		from, err := txn.Sender(*signer)
-		if err != nil {
-			logger.Warn(fmt.Sprintf("[%s] Could not recover transaction sender", logPrefix), "hash", txn.Hash(), "err", err)
-			continue
-		}
-
-		// Check whether the txn is replay protected. If we're not in the EIP155 (Spurious Dragon) hf
-		// phase, start ignoring the sender until we do.
-		if txn.Protected() && !chainConfig.IsSpuriousDragon(header.Number.Uint64()) {
-			logger.Debug(fmt.Sprintf("[%s] Ignoring replay protected transaction", logPrefix), "hash", txn.Hash(), "eip155", chainConfig.SpuriousDragonBlock)
-			continue
-		}
-
-		// Start executing the transaction
-		logs, err := builderCommitTx(txn, coinbase, vmConfig, chainConfig, ibs, current)
-		if err == nil {
-			recordTxIO()
-		} else {
-			clearTxIO()
-		}
-		if errors.Is(err, protocol.ErrGasLimitReached) {
-			// Skip the env out-of-gas transaction
-			logger.Debug(fmt.Sprintf("[%s] Gas limit exceeded for env block", logPrefix), "hash", txn.Hash(), "sender", from)
-		} else if errors.Is(err, protocol.ErrNonceTooLow) {
-			// New head notification data race between the transaction pool and builder, skip
-			logger.Debug(fmt.Sprintf("[%s] Skipping transaction with low nonce", logPrefix), "hash", txn.Hash(), "sender", from, "nonce", txn.GetNonce(), "err", err)
-		} else if errors.Is(err, protocol.ErrNonceTooHigh) {
-			// Reorg notification data race between the transaction pool and builder, skip
-			logger.Debug(fmt.Sprintf("[%s] Skipping transaction with high nonce", logPrefix), "hash", txn.Hash(), "sender", from, "nonce", txn.GetNonce())
-		} else if err == nil {
-			// Everything ok, collect the logs and proceed to the next transaction
-			logger.Trace(fmt.Sprintf("[%s] Added transaction", logPrefix), "hash", txn.Hash(), "sender", from, "nonce", txn.GetNonce(), "payload", payloadId)
-			coalescedLogs = append(coalescedLogs, logs...)
-			txnIdx++
-		} else {
-			// Strange error, discard the transaction and get the next in line (note, the
-			// nonce-too-high clause will prevent us from executing in vain).
-			logger.Debug(fmt.Sprintf("[%s] Skipping transaction", logPrefix), "hash", txn.Hash(), "sender", from, "err", err)
-		}
-	}
-
-	/*
-		// Notify resubmit loop to decrease resubmitting interval if env interval is larger
-		// than the user-specified one.
-		if interrupt != nil {
-			w.resubmitAdjustCh <- &intervalAdjust{inc: false}
-		}
-	*/
-	return coalescedLogs, done, nil
-
-}
-
 func NotifyPendingLogs(logPrefix string, notifier stagedsync.ChainEventNotifier, logs types.Logs, logger log.Logger) {
 	if len(logs) == 0 {
 		return