diff --git a/.github/workflows/hive-versions.json b/.github/workflows/hive-versions.json
index e05ef90be23..2a3f1fdb1de 100644
--- a/.github/workflows/hive-versions.json
+++ b/.github/workflows/hive-versions.json
@@ -1,4 +1,5 @@
{
- "hive_ref": "3184ac37d3dfbb9b8b0348ae70ea98bb61fa6f13",
+ "hive_repository": "taratorio/hive",
+ "hive_ref": "fix/withdrawals-clmock-config",
"execution_apis_ref": "f74de4b86e3b011384808c294c3d71f2854729a2"
}
diff --git a/.github/workflows/test-hive-eest.yml b/.github/workflows/test-hive-eest.yml
index e1e8ef7d670..9ec50fda2f5 100644
--- a/.github/workflows/test-hive-eest.yml
+++ b/.github/workflows/test-hive-eest.yml
@@ -221,14 +221,19 @@ jobs:
path: erigon-full
persist-credentials: false
- - name: Read pinned Hive ref
+ - name: Read Hive checkout
id: hive-version
- run: echo "ref=$(jq -r .hive_ref erigon-full/.github/workflows/hive-versions.json)" >> "$GITHUB_OUTPUT"
+ run: |
+ versions=erigon-full/.github/workflows/hive-versions.json
+ {
+ echo "repository=$(jq -r .hive_repository "$versions")"
+ echo "ref=$(jq -r .hive_ref "$versions")"
+ } >> "$GITHUB_OUTPUT"
- name: Checkout Hive
uses: actions/checkout@v7
with:
- repository: ethereum/hive
+ repository: ${{ steps.hive-version.outputs.repository }}
ref: ${{ steps.hive-version.outputs.ref }}
path: hive
persist-credentials: false
diff --git a/.github/workflows/test-hive.yml b/.github/workflows/test-hive.yml
index 7f32d638bd8..6c7e1763bd2 100644
--- a/.github/workflows/test-hive.yml
+++ b/.github/workflows/test-hive.yml
@@ -97,17 +97,20 @@ jobs:
path: erigon-full
persist-credentials: false
- - name: Read pinned versions
+ - name: Read Hive checkout
id: hive-version
run: |
- echo "ref=$(jq -r .hive_ref erigon-full/.github/workflows/hive-versions.json)" >> "$GITHUB_OUTPUT"
- echo "execution_apis_ref=$(jq -r '.execution_apis_ref // empty' erigon-full/.github/workflows/hive-versions.json)" >> "$GITHUB_OUTPUT"
+ versions=erigon-full/.github/workflows/hive-versions.json
+ {
+ echo "repository=$(jq -r .hive_repository "$versions")"
+ echo "ref=$(jq -r .hive_ref "$versions")"
+ echo "execution_apis_ref=$(jq -r '.execution_apis_ref // empty' "$versions")"
+ } >> "$GITHUB_OUTPUT"
- name: Checkout Hive
uses: actions/checkout@v7
with:
- repository: ethereum/hive
- # version hive and update periodically/on-demand to prevent upstream changes in Hive affecting us with red CI
+ repository: ${{ steps.hive-version.outputs.repository }}
ref: ${{ steps.hive-version.outputs.ref }}
path: hive
persist-credentials: false
diff --git a/Makefile b/Makefile
index 5563a628878..61b9977e11b 100644
--- a/Makefile
+++ b/Makefile
@@ -385,6 +385,7 @@ EEST_DEVNET_URL = $(shell jq -r '."eest_devnet".url' test-fixtures.json)
EEST_DEVNET_BRANCH = $(shell jq -r '."eest_devnet".branch' test-fixtures.json)
EEST_STABLE_ERIGON_FLAGS = --fcu.background.prune=false --fcu.timeout=0
EEST_GLAMSTERDAM_ERIGON_FLAGS = $(EEST_STABLE_ERIGON_FLAGS) --experimental.bal
+EEST_HIVE_REPOSITORY = $(shell jq -r '.hive_repository' .github/workflows/hive-versions.json)
EEST_HIVE_REF = $(shell jq -r '.hive_ref' .github/workflows/hive-versions.json)
HIVE_SIM_PARALLELISM ?= 8
@@ -394,7 +395,7 @@ eest-devnet:
@if [ ! -d "temp" ]; then mkdir temp; fi
docker build -t "test/erigon:$(SHORT_COMMIT)" .
rm -rf "temp/eest-hive-$(SHORT_COMMIT)" && mkdir "temp/eest-hive-$(SHORT_COMMIT)"
- cd "temp/eest-hive-$(SHORT_COMMIT)" && git clone https://github.com/ethereum/hive
+ cd "temp/eest-hive-$(SHORT_COMMIT)" && git clone "https://github.com/$(EEST_HIVE_REPOSITORY)"
cd "temp/eest-hive-$(SHORT_COMMIT)/hive" && git checkout --detach "$(EEST_HIVE_REF)"
cd "temp/eest-hive-$(SHORT_COMMIT)/hive" && \
sed -i'' -e "s/^ARG baseimage=erigontech\/erigon$$/ARG baseimage=test\/erigon/" clients/erigon/Dockerfile && \
@@ -456,7 +457,7 @@ eest-hive:
@if [ ! -d "temp" ]; then mkdir temp; fi
docker build -t "test/erigon:$(SHORT_COMMIT)" .
rm -rf "temp/eest-hive-$(SHORT_COMMIT)" && mkdir "temp/eest-hive-$(SHORT_COMMIT)"
- cd "temp/eest-hive-$(SHORT_COMMIT)" && git clone https://github.com/ethereum/hive
+ cd "temp/eest-hive-$(SHORT_COMMIT)" && git clone "https://github.com/$(EEST_HIVE_REPOSITORY)"
cd "temp/eest-hive-$(SHORT_COMMIT)/hive" && git checkout --detach "$(EEST_HIVE_REF)"
cd "temp/eest-hive-$(SHORT_COMMIT)/hive" && \
sed -i'' -e "s/^ARG baseimage=erigontech\/erigon$$/ARG baseimage=test\/erigon/" clients/erigon/Dockerfile && \
diff --git a/cmd/integration/commands/dump_state_test.go b/cmd/integration/commands/dump_state_test.go
index a39601ce31d..e7120530f43 100644
--- a/cmd/integration/commands/dump_state_test.go
+++ b/cmd/integration/commands/dump_state_test.go
@@ -158,7 +158,7 @@ func seedTestAccounts(t *testing.T) kv.TemporalTx {
t.Helper()
dirs := datadir.New(t.TempDir())
- db := temporaltest.NewTestDBWithStepSize(t, dirs, 16)
+ db := temporaltest.NewTestDB(t, dirs, temporaltest.WithStepSize(16))
t.Cleanup(db.Close)
tx, err := db.BeginTemporalRw(context.Background())
require.NoError(t, err)
@@ -209,7 +209,7 @@ func seedManyAccounts(t testing.TB, n int) kv.TemporalTx {
t.Helper()
dirs := datadir.New(t.TempDir())
- db := temporaltest.NewTestDBWithStepSize(t, dirs, 16)
+ db := temporaltest.NewTestDB(t, dirs, temporaltest.WithStepSize(16))
t.Cleanup(db.Close)
tx, err := db.BeginTemporalRw(context.Background())
require.NoError(t, err)
diff --git a/cmd/integration/commands/stages.go b/cmd/integration/commands/stages.go
index ed127dad301..bf8223902c1 100644
--- a/cmd/integration/commands/stages.go
+++ b/cmd/integration/commands/stages.go
@@ -830,13 +830,14 @@ func stageExec(db kv.TemporalRwDB, ctx context.Context, logger log.Logger) error
}
collateAndPrune := func() error {
- return agg.CollateAndPrune(ctx, db, func(tx kv.TemporalRwTx) error {
+ _, _, err := agg.CollateAndPrune(ctx, db, func(tx kv.TemporalRwTx) error {
pruneStage, err := sync.PruneStageState(stages.Execution, s.BlockNumber, tx, s.CurrentSyncCycle.IsInitialCycle)
if err != nil {
return err
}
return stagedsync.PruneExecutionStage(ctx, pruneStage, tx, cfg, 0, logger)
}, logger)
+ return err
}
if chainTipMode {
diff --git a/cmd/utils/app/import_cmd.go b/cmd/utils/app/import_cmd.go
index 3442606c9d9..4f48d60be14 100644
--- a/cmd/utils/app/import_cmd.go
+++ b/cmd/utils/app/import_cmd.go
@@ -17,6 +17,7 @@
package app
import (
+ "bufio"
"compress/gzip"
"context"
"errors"
@@ -133,8 +134,12 @@ func importChain(ctx context.Context, cliCtx *cli.Command) error {
return err
}
- return importFiles(cliCtx.Args().Slice(), logger, func(fn string) error {
- return ImportChain(ethereum, ethereum.ChainDB(), fn, logger)
+ files := cliCtx.Args().Slice()
+ fileIndex := 0
+ return importFiles(files, logger, func(fn string) error {
+ lastFile := fileIndex == len(files)-1
+ fileIndex++
+ return importFile(ethereum, ethereum.ChainDB(), fn, lastFile, logger)
})
}
@@ -157,7 +162,7 @@ func importFiles(files []string, logger log.Logger, importOne func(fn string) er
return importErr
}
-func ImportChain(ethereum *eth.Ethereum, chainDB kv.RwDB, fn string, logger log.Logger) error {
+func importFile(ethereum *eth.Ethereum, chainDB kv.RwDB, fn string, lastFile bool, logger log.Logger) error {
// Watch for Ctrl-C while the import is running.
// If a signal is received, the import will stop at the next batch.
interrupt := make(chan os.Signal, 1)
@@ -195,7 +200,8 @@ func ImportChain(ethereum *eth.Ethereum, chainDB kv.RwDB, fn string, logger log.
return err
}
}
- stream := rlp.NewStream(reader, 0)
+ bufferedReader := bufio.NewReader(reader)
+ stream := rlp.NewStream(bufferedReader, 0)
// Run actual the import.
blocks := make(types.Blocks, importBatchSize)
@@ -224,6 +230,11 @@ func ImportChain(ethereum *eth.Ethereum, chainDB kv.RwDB, fn string, logger log.
if i == 0 {
break
}
+ lastBatch := i < importBatchSize
+ if !lastBatch {
+ _, err := bufferedReader.Peek(1)
+ lastBatch = errors.Is(err, io.EOF)
+ }
// Import the batch.
if checkInterrupt() {
return errInterrupted
@@ -242,7 +253,7 @@ func ImportChain(ethereum *eth.Ethereum, chainDB kv.RwDB, fn string, logger log.
TopBlock: missing[len(missing)-1],
}
- if err := InsertChain(ethereum, missingChain, true); err != nil {
+ if err := insertChain(ethereum, missingChain, true, lastFile && lastBatch); err != nil {
return err
}
}
@@ -284,7 +295,7 @@ func missingBlocks(chainDB kv.RwDB, blocks []*types.Block, blockReader dbservice
return nil
}
-func InsertChain(ethereum *eth.Ethereum, chain *blockgen.ChainPack, setHead bool) error {
+func insertChain(ethereum *eth.Ethereum, chain *blockgen.ChainPack, setHead, finalize bool) error {
if len(chain.Blocks) == 0 {
return nil
}
@@ -309,9 +320,15 @@ func InsertChain(ethereum *eth.Ethereum, chain *blockgen.ChainPack, setHead bool
firstBlock := chain.Blocks[0]
tipBlock := chain.TopBlock
var parentTd, currentHeadTd *uint256.Int
+ var genesisHash common.Hash
var currentHeadHash common.Hash
var currentHeadNumber uint64
if err := ethereum.ChainDB().View(ctx, func(tx kv.Tx) error {
+ var err error
+ genesisHash, err = rawdb.ReadCanonicalHash(tx, 0)
+ if err != nil {
+ return fmt.Errorf("read genesis hash: %w", err)
+ }
if firstBlock.NumberU64() > 0 {
td, readErr := rawdb.ReadTd(tx, firstBlock.ParentHash(), firstBlock.NumberU64()-1)
if readErr != nil {
@@ -398,7 +415,11 @@ func InsertChain(ethereum *eth.Ethereum, chain *blockgen.ChainPack, setHead bool
}
tipHash := chain.TopBlock.Hash()
- status, validationErr, lvh, err := chainRW.UpdateForkChoice(ctx, tipHash, tipHash, tipHash)
+ safeHash, finalizedHash := genesisHash, genesisHash
+ if finalize {
+ safeHash, finalizedHash = tipHash, tipHash
+ }
+ status, validationErr, lvh, err := chainRW.UpdateForkChoice(ctx, tipHash, safeHash, finalizedHash)
if err != nil {
return err
}
diff --git a/cmd/utils/app/import_reorg_test.go b/cmd/utils/app/import_reorg_test.go
index 40fd6179d35..fc2af01a3ed 100644
--- a/cmd/utils/app/import_reorg_test.go
+++ b/cmd/utils/app/import_reorg_test.go
@@ -85,7 +85,7 @@ func TestImportReorgUnwindToGenesis(t *testing.T) {
require.NoError(t, err)
defer db.Close()
- var storedGenesis, head common.Hash
+ var storedGenesis, head, safe, finalized common.Hash
var headNumber *uint64
require.NoError(t, db.View(ctx, func(tx kv.Tx) error {
var err error
@@ -94,6 +94,8 @@ func TestImportReorgUnwindToGenesis(t *testing.T) {
}
head = rawdb.ReadHeadBlockHash(tx)
headNumber = rawdb.ReadHeaderNumber(tx, head)
+ safe = rawdb.ReadForkchoiceSafe(tx)
+ finalized = rawdb.ReadForkchoiceFinalized(tx)
return nil
}))
require.Equal(t, genesisHash, storedGenesis.Hex(), "genesis hash mismatch — chain config drift?")
@@ -103,6 +105,8 @@ func TestImportReorgUnwindToGenesis(t *testing.T) {
"head did not advance to the heavier side chain (block 4); import err: %v", importErr)
require.Equalf(t, tc.LastBlockHash, head.Hex(),
"final head mismatch (import err: %v)", importErr)
+ require.Equal(t, head, safe, "safe block is not the final imported head")
+ require.Equal(t, head, finalized, "finalized block is not the final imported head")
}
// TestImportClosesChaindataOnInitError makes ethereum.Init fail after eth.New
diff --git a/db/integrity/commitment_state_verify_test.go b/db/integrity/commitment_state_verify_test.go
index 7bf07a126fd..43db7d859a3 100644
--- a/db/integrity/commitment_state_verify_test.go
+++ b/db/integrity/commitment_state_verify_test.go
@@ -48,7 +48,7 @@ func TestCheckStateVerify(t *testing.T) {
stepSize := uint64(100)
dirs := datadir.New(t.TempDir())
- db := temporaltest.NewTestDBWithStepSize(t, dirs, stepSize)
+ db := temporaltest.NewTestDB(t, dirs, temporaltest.WithStepSize(stepSize))
agg := db.(state.HasAgg).Agg().(*state.Aggregator)
tx, err := db.BeginTemporalRw(ctx)
@@ -126,7 +126,7 @@ func TestCheckStateVerify_NoopWrite(t *testing.T) {
stepSize := uint64(100)
dirs := datadir.New(t.TempDir())
- db := temporaltest.NewTestDBWithStepSize(t, dirs, stepSize)
+ db := temporaltest.NewTestDB(t, dirs, temporaltest.WithStepSize(stepSize))
agg := db.(state.HasAgg).Agg().(*state.Aggregator)
tx, err := db.BeginTemporalRw(ctx)
@@ -242,7 +242,7 @@ func TestVerifyBranchHashesFromDB(t *testing.T) {
stepSize := uint64(100)
dirs := datadir.New(t.TempDir())
- db := temporaltest.NewTestDBWithStepSize(t, dirs, stepSize)
+ db := temporaltest.NewTestDB(t, dirs, temporaltest.WithStepSize(stepSize))
tx, err := db.BeginTemporalRw(ctx)
require.NoError(t, err)
diff --git a/db/integrity/commitment_version_integration_test.go b/db/integrity/commitment_version_integration_test.go
index 14876105741..5134493529c 100644
--- a/db/integrity/commitment_version_integration_test.go
+++ b/db/integrity/commitment_version_integration_test.go
@@ -63,7 +63,7 @@ func runVersionRegimeCheck(t *testing.T, referencesInCommitmentBranches bool) {
const txs = 80 // 8 steps -> merge produces a >= threshold commitment file
dirs := datadir.New(t.TempDir())
- db := temporaltest.NewTestDBWithStepSize(t, dirs, stepSize)
+ db := temporaltest.NewTestDB(t, dirs, temporaltest.WithStepSize(stepSize))
agg := db.(state.HasAgg).Agg().(*state.Aggregator)
agg.ForTestReferencesInCommitmentBranches(kv.CommitmentDomain, referencesInCommitmentBranches)
diff --git a/db/kv/membatchwithdb/memory_mutation_test.go b/db/kv/membatchwithdb/memory_mutation_test.go
index 9bcfb912c68..4c0a3e10806 100644
--- a/db/kv/membatchwithdb/memory_mutation_test.go
+++ b/db/kv/membatchwithdb/memory_mutation_test.go
@@ -335,7 +335,7 @@ func newTestTx(tb testing.TB) (kv.TemporalRwDB, kv.TemporalRwTx) {
tb.Helper()
dirs := datadir.New(tb.TempDir())
stepSize := uint64(16)
- db := temporaltest.NewTestDBWithStepSize(tb, dirs, stepSize)
+ db := temporaltest.NewTestDB(tb, dirs, temporaltest.WithStepSize(stepSize))
tx, err := db.BeginTemporalRw(tb.Context()) //nolint:gocritic
if err != nil {
tb.Fatal(err)
diff --git a/db/kv/temporal/temporaltest/kv_temporal_testdb.go b/db/kv/temporal/temporaltest/kv_temporal_testdb.go
index 51e8180ed9e..cff519bd690 100644
--- a/db/kv/temporal/temporaltest/kv_temporal_testdb.go
+++ b/db/kv/temporal/temporaltest/kv_temporal_testdb.go
@@ -48,17 +48,38 @@ func NewTestTx(tb testing.TB) (kv.TemporalRwDB, kv.TemporalRwTx) {
return db, tx
}
-// nolint:thelper
-func NewTestDB(tb testing.TB, dirs datadir.Dirs) kv.TemporalRwDB {
- return newTestDB(tb, dirs, config3.DefaultStepSize)
+type Option func(*options)
+
+type options struct {
+ stepSize uint64
+ reorgBlockDepth uint64
+}
+
+func WithStepSize(stepSize uint64) Option {
+ return func(opts *options) {
+ opts.stepSize = stepSize
+ }
}
-func NewTestDBWithStepSize(tb testing.TB, dirs datadir.Dirs, stepSize uint64) kv.TemporalRwDB {
- return newTestDB(tb, dirs, stepSize)
+func WithReorgBlockDepth(reorgBlockDepth uint64) Option {
+ return func(opts *options) {
+ opts.reorgBlockDepth = reorgBlockDepth
+ }
+}
+
+// nolint:thelper
+func NewTestDB(tb testing.TB, dirs datadir.Dirs, opts ...Option) kv.TemporalRwDB {
+ config := options{
+ stepSize: config3.DefaultStepSize,
+ }
+ for _, opt := range opts {
+ opt(&config)
+ }
+ return newTestDB(tb, dirs, config.stepSize, config.reorgBlockDepth)
}
// nolint:thelper
-func newTestDB(tb testing.TB, dirs datadir.Dirs, stepSize uint64) kv.TemporalRwDB {
+func newTestDB(tb testing.TB, dirs datadir.Dirs, stepSize, reorgBlockDepth uint64) kv.TemporalRwDB {
if tb != nil {
tb.Helper()
}
@@ -84,7 +105,7 @@ func newTestDB(tb testing.TB, dirs datadir.Dirs, stepSize uint64) kv.TemporalRwD
panic(err)
}
- stateSnapshots := state.NewTest(dirs).StepSize(stepSize).MustOpen(ctx, rawDB)
+ stateSnapshots := state.NewTest(dirs).StepSize(stepSize).ReorgBlockDepth(reorgBlockDepth).MustOpen(ctx, rawDB)
if tb != nil {
tb.Cleanup(stateSnapshots.Close)
}
diff --git a/db/rawdb/accessors_chain.go b/db/rawdb/accessors_chain.go
index 37adcc6bfa3..3a86820acad 100644
--- a/db/rawdb/accessors_chain.go
+++ b/db/rawdb/accessors_chain.go
@@ -285,6 +285,18 @@ func ReadForkchoiceFinalized(db kv.Getter) common.Hash {
return common.BytesToHash(data)
}
+func ReadForkchoiceFinalizedNum(db kv.Getter) uint64 {
+ h := ReadForkchoiceFinalized(db)
+ if h == (common.Hash{}) {
+ return 0
+ }
+ n := ReadHeaderNumber(db, h)
+ if n == nil {
+ return 0
+ }
+ return *n
+}
+
// WriteForkchoiceFinalized stores finalizedBlockHash from the last Engine API forkChoiceUpdated.
func WriteForkchoiceFinalized(db kv.Putter, hash common.Hash) {
if err := db.Put(kv.LastForkchoice, []byte("finalizedBlockHash"), hash[:]); err != nil {
diff --git a/db/snapshotsync/freezeblocks/block_snapshots.go b/db/snapshotsync/freezeblocks/block_snapshots.go
index f6cc9608536..3f9b8e856ed 100644
--- a/db/snapshotsync/freezeblocks/block_snapshots.go
+++ b/db/snapshotsync/freezeblocks/block_snapshots.go
@@ -173,16 +173,27 @@ func (br *BlockRetire) snapshots() *blocksnapshots.RoSnapshots {
return br.blockReader.Snapshots().(*blocksnapshots.RoSnapshots)
}
-func (br *BlockRetire) canRetire(curBlockNum uint64, blocksInSnapshots uint64, snapType snaptype.Enum) (blockFrom, blockTo uint64, can bool) {
- //
- // TODO(milen): finalisedHash check
- //
- keep := br.config.MaxReorgDepth
- if curBlockNum <= keep {
- return
+func (br *BlockRetire) canRetire(ctx context.Context, curBlockNum uint64, blocksInSnapshots uint64, snapType snaptype.Enum) (blockFrom, blockTo uint64, can bool, err error) {
+ var finalisedBlockNum uint64
+ err = br.db.View(ctx, func(tx kv.Tx) error {
+ finalisedBlockNum = rawdb.ReadForkchoiceFinalizedNum(tx)
+ return nil
+ })
+ if err != nil {
+ return 0, 0, false, err
+ }
+ if finalisedBlockNum > 0 {
+ blockTo = finalisedBlockNum
+ } else {
+ keep := br.config.MaxReorgDepth
+ if curBlockNum <= keep {
+ return
+ }
+ blockTo = curBlockNum - keep
}
blockFrom = blocksInSnapshots + 1
- return snapshotsync.CanRetire(blockFrom, curBlockNum-keep, snapType, br.snCfg, br.config.Snapshot.E2RetireStep)
+ blockFrom, blockTo, can = snapshotsync.CanRetire(blockFrom, blockTo, snapType, br.snCfg, br.config.Snapshot.E2RetireStep)
+ return blockFrom, blockTo, can, nil
}
func CanDeleteTo(curBlockNum uint64, blocksInSnapshots uint64) (blockTo uint64) {
@@ -248,7 +259,10 @@ func (br *BlockRetire) buildFiles(
notifier, logger, blockReader, tmpDir, db, workers := br.notifier, br.logger, br.blockReader, br.tmpDir, br.db, br.workers.Load()
snapshots := br.snapshots()
- blockFrom, blockTo, ok := br.canRetire(maxBlockNum, minBlockNum, snaptype.Unknown)
+ blockFrom, blockTo, ok, err := br.canRetire(ctx, maxBlockNum, minBlockNum, snaptype.Unknown)
+ if err != nil {
+ return false, err
+ }
if ok {
if has, err := br.dbHasEnoughDataForBlocksRetire(ctx); err != nil {
return false, err
diff --git a/db/state/aggregator.go b/db/state/aggregator.go
index 03db19fef9e..0eb05fa0085 100644
--- a/db/state/aggregator.go
+++ b/db/state/aggregator.go
@@ -34,6 +34,7 @@ import (
"time"
"github.com/erigontech/erigon/db/kv/prune"
+ "github.com/erigontech/erigon/db/rawdb"
"golang.org/x/sync/errgroup"
"golang.org/x/sync/semaphore"
@@ -1019,7 +1020,7 @@ func (a *Aggregator) buildFiles(ctx context.Context, step kv.Step) error {
a.LockWorkersEditing()
defer a.UnlockWorkersEditing()
- lastBlockInStep, lastBlockInDB, lastTxInDB, ok, err := a.readyForCollation(ctx, step)
+ finalisedBlockNum, lastBlockInStep, lastBlockInDB, lastTxInDB, ok, err := a.readyForCollation(ctx, step)
if err != nil {
return err
}
@@ -1029,13 +1030,16 @@ func (a *Aggregator) buildFiles(ctx context.Context, step kv.Step) error {
if lastStepInDB > 0 {
lastCollatableStepInDB = lastStepInDB - 1
}
- a.logger.Debug("[snapshots] holding state collation at reorg depth",
+ a.logger.Debug(
+ "[snapshots] holding state collation at reorg depth",
"step", step,
+ "finalisedBlockNum", finalisedBlockNum,
"lastBlockInStep", lastBlockInStep,
"lastBlockInDB", lastBlockInDB,
"lastTxInDB", lastTxInDB,
"reorgBlockDepth", a.reorgBlockDepth,
- "lastCollatableStepInDB", lastCollatableStepInDB)
+ "lastCollatableStepInDB", lastCollatableStepInDB,
+ )
return errStepNotReady
}
var (
@@ -1144,13 +1148,14 @@ func (a *Aggregator) buildFiles(ctx context.Context, step kv.Step) error {
return nil
}
-func (a *Aggregator) readyForCollation(ctx context.Context, step kv.Step) (lastBlockInStep, lastBlockInDB, lastTxInDB uint64, ok bool, err error) {
+func (a *Aggregator) readyForCollation(ctx context.Context, step kv.Step) (finalisedBlockNum, lastBlockInStep, lastBlockInDB, lastTxInDB uint64, ok bool, err error) {
if a.reorgBlockDepth == 0 {
- return 0, 0, 0, true, nil
+ return 0, 0, 0, 0, true, nil
}
a.commitGate.RLock()
defer a.commitGate.RUnlock()
err = a.db.View(ctx, func(tx kv.Tx) error {
+ finalisedBlockNum = rawdb.ReadForkchoiceFinalizedNum(tx)
lastBlockInStep, ok, err = rawdbv3.TxNums.FindBlockNum(ctx, tx, step.LastTxNum(a.stepSize.Load()))
if err != nil {
return err
@@ -1161,7 +1166,13 @@ func (a *Aggregator) readyForCollation(ctx context.Context, step kv.Step) (lastB
lastBlockInDB, lastTxInDB, err = rawdbv3.TxNums.Last(tx)
return err
})
- ok = err == nil && lastBlockInDB > lastBlockInStep+a.reorgBlockDepth
+ var ready bool
+ if finalisedBlockNum > 0 {
+ ready = lastBlockInStep <= finalisedBlockNum
+ } else {
+ ready = lastBlockInDB > lastBlockInStep+a.reorgBlockDepth
+ }
+ ok = err == nil && ready
return
}
@@ -1197,7 +1208,7 @@ func (a *Aggregator) reorgSafeBlockAndStep(ctx context.Context) (reorgSafeBlock
}
func (a *Aggregator) BuildFiles(toTxNum uint64) (err error) {
- finished := a.buildFilesInBackground(toTxNum, true)
+ finished, _ := a.buildFilesInBackground(toTxNum, true)
if !(a.buildingFiles.Load() || a.mergingFiles.Load()) {
return nil
}
@@ -1781,16 +1792,16 @@ func (a *Aggregator) CommitGate() *sync.RWMutex { return &a.commitGate }
// CollateAndPrune runs a single prune pass and kicks background file
// building. The block-snapshot-boundary gate inside readyForCollation
// keeps state files from extending past block files, so no external cap
-// is needed.
-func (a *Aggregator) CollateAndPrune(ctx context.Context, db kv.TemporalRwDB, pruneFn func(tx kv.TemporalRwTx) error, logger log.Logger) error {
+// is needed. It returns whether file building started and its completion channel.
+func (a *Aggregator) CollateAndPrune(ctx context.Context, db kv.TemporalRwDB, pruneFn func(tx kv.TemporalRwTx) error, logger log.Logger) (bool, <-chan struct{}, error) {
a.commitGate.Lock()
err := db.UpdateTemporal(ctx, pruneFn)
a.commitGate.Unlock()
if err != nil {
- return err
+ return false, nil, err
}
- a.BuildFilesInBackground(a.EndTxNumMinimax() + a.StepSize())
- return nil
+ finished, started := a.buildFilesInBackground(a.EndTxNumMinimax()+a.StepSize(), true)
+ return started, finished, nil
}
func (a *Aggregator) FilesAmount() (res []int) {
a.dirtyFilesLock.Lock()
@@ -2223,35 +2234,36 @@ func (a *Aggregator) SetProduceMod(produce bool) {
}
func (a *Aggregator) BuildFilesInBackground(txNum uint64) chan struct{} {
- return a.buildFilesInBackground(txNum, true)
+ finished, _ := a.buildFilesInBackground(txNum, true)
+ return finished
}
-// Returns channel which is closed when aggregation is done
-func (a *Aggregator) buildFilesInBackground(txNum uint64, doMerge bool) chan struct{} {
+// Returns a channel which is closed when aggregation is done and whether it started.
+func (a *Aggregator) buildFilesInBackground(txNum uint64, doMerge bool) (chan struct{}, bool) {
fin := make(chan struct{})
if dbg.NoBackgroundMaintenance() {
close(fin)
- return fin
+ return fin, false
}
if !a.produce {
a.logger.Debug("[snapshots] buildFiles: produce=false")
close(fin)
- return fin
+ return fin, false
}
visMin := a.visible.Load().minimaxTxNum
if (txNum + 1) <= visMin+a.stepSize.Load() {
a.logger.Debug("[snapshots] buildFiles: not enough data", "txNum", txNum, "visibleMin", visMin, "stepSize", a.stepSize.Load())
close(fin)
- return fin
+ return fin, false
}
if ok := a.buildingFiles.CompareAndSwap(false, true); !ok {
a.logger.Debug("[snapshots] buildFiles: already building")
close(fin)
- return fin
+ return fin, false
}
step := kv.Step(a.EndTxNumMinimax() / a.StepSize())
@@ -2391,7 +2403,7 @@ func (a *Aggregator) buildFilesInBackground(txNum uint64, doMerge bool) chan str
a.buildingFiles.Store(false)
close(fin)
}
- return fin
+ return fin, started
}
// Returns the first known txNum found in history files of a given domain
diff --git a/db/state/aggregator_bench_test.go b/db/state/aggregator_bench_test.go
index bdfce567b63..fee03e414da 100644
--- a/db/state/aggregator_bench_test.go
+++ b/db/state/aggregator_bench_test.go
@@ -50,7 +50,7 @@ import (
func testDbAndAggregatorBench(b *testing.B, aggStep uint64) (kv.TemporalRwDB, *state.Aggregator) {
b.Helper()
dirs := datadir.New(b.TempDir())
- db := temporaltest.NewTestDBWithStepSize(b, dirs, aggStep)
+ db := temporaltest.NewTestDB(b, dirs, temporaltest.WithStepSize(aggStep))
return db, db.(state.HasAgg).Agg().(*state.Aggregator)
}
diff --git a/db/test/domain_shared_bench_test.go b/db/test/domain_shared_bench_test.go
index 7b6940e34fc..ed2f5872318 100644
--- a/db/test/domain_shared_bench_test.go
+++ b/db/test/domain_shared_bench_test.go
@@ -56,7 +56,7 @@ func (r *rndGen) Read(p []byte) (n int, err error) { return r.oldGen.Read(p) } /
func testDbAndAggregatorBench(b *testing.B, aggStep uint64) (kv.TemporalRwDB, *state.Aggregator) {
b.Helper()
dirs := datadir.New(b.TempDir())
- db := temporaltest.NewTestDBWithStepSize(b, dirs, aggStep)
+ db := temporaltest.NewTestDB(b, dirs, temporaltest.WithStepSize(aggStep))
return db, db.(state.HasAgg).Agg().(*state.Aggregator)
}
diff --git a/db/test/domains_restart_test.go b/db/test/domains_restart_test.go
index fa738ffffa2..a8be22d26b3 100644
--- a/db/test/domains_restart_test.go
+++ b/db/test/domains_restart_test.go
@@ -59,7 +59,7 @@ func testDbAndAggregatorv3(t *testing.T, fpath string, stepSize uint64) (kv.Temp
path = fpath
}
dirs := datadir.New(path)
- db := temporaltest.NewTestDBWithStepSize(t, dirs, stepSize)
+ db := temporaltest.NewTestDB(t, dirs, temporaltest.WithStepSize(stepSize))
return db, db.(state.HasAgg).Agg().(*state.Aggregator), path
}
diff --git a/execution/engineapi/engine_api_state_churn_files_test.go b/execution/engineapi/engine_api_state_churn_files_test.go
index 2ebc0340e37..b39362f4ba9 100644
--- a/execution/engineapi/engine_api_state_churn_files_test.go
+++ b/execution/engineapi/engine_api_state_churn_files_test.go
@@ -167,7 +167,6 @@ func TestEngineApiUnwindToSnapshotBoundaryPreservesDeletedSlots(t *testing.T) {
EthConfigTweaker: func(c *ethconfig.Config) {
c.Snapshot.ProduceE3 = true
c.AlwaysGenerateChangesets = true
- c.MaxReorgDepth = 400 // the boundary sits deep below the tip
},
})
require.NoError(t, err)
diff --git a/execution/engineapi/engine_api_state_churn_prune_test.go b/execution/engineapi/engine_api_state_churn_prune_test.go
index 1573b25188b..17a637a4cc1 100644
--- a/execution/engineapi/engine_api_state_churn_prune_test.go
+++ b/execution/engineapi/engine_api_state_churn_prune_test.go
@@ -126,65 +126,3 @@ func TestEngineApiReorgWithPruningInterference(t *testing.T) {
churnAndAssert(ctx, t, eat, churn, 4, func(k int) int64 { return int64(4_000 + k) })
})
}
-
-// TestEngineApiUnwindBeyondRetainedChangesetsRejectedCleanly runs without
-// AlwaysGenerateChangesets, so once churn history is collated into snapshot
-// files the changesets an unwind would need are pruned away. A forkchoice to
-// a block below what is still unwindable must then be rejected loudly — a
-// silently partial unwind would leave phantom state — while a shallow unwind
-// and continued churn keep working on the same node.
-func TestEngineApiUnwindBeyondRetainedChangesetsRejectedCleanly(t *testing.T) {
- ctx := t.Context()
- logger := testlog.Logger(t, log.LvlError)
- dataDir := newSmallStepDataDir(t)
-
- genesis, coinbaseKey, err := engineapitester.DefaultEngineApiTesterGenesis()
- require.NoError(t, err)
- eat, err := engineapitester.InitialiseEngineApiTester(ctx, engineapitester.EngineApiTesterInitArgs{
- Logger: logger,
- DataDir: dataDir,
- Genesis: genesis,
- CoinbaseKey: coinbaseKey,
- EthConfigTweaker: func(c *ethconfig.Config) {
- c.Snapshot.ProduceE3 = true
- c.MaxReorgDepth = 400 // deeper than the rejection target: the changeset check must fire, not the depth cap
- },
- })
- require.NoError(t, err)
- t.Cleanup(func() { require.NoError(t, eat.Close()) })
-
- eat.Run(t, func(ctx context.Context, t *testing.T, eat engineapitester.EngineApiTester) {
- const pokes = 300
- payloads, _, churn, sums := buildChurnChain(ctx, t, eat, pokes, func(k int) int64 { return int64(k) })
- tip := uint64(2 + pokes)
-
- waitForDomainFilesSettled(ctx, t, eat.StateAgg)
- t.Logf("domain files settled: %v", eat.StateAgg.FilesAmount())
-
- // A shallow unwind at the tip must still work.
- shallow := tip - 8
- require.NoError(t, eat.MockCl.UpdateForkChoice(ctx, payloads[shallow-2]))
- assertChurnState(ctx, t, eat, churn, payloads[shallow-2], sums[shallow-2])
- for h := shallow + 1; h <= tip; h++ {
- status, err := eat.MockCl.InsertNewPayload(ctx, payloads[h-2])
- require.NoError(t, err)
- require.Equalf(t, enginetypes.ValidStatus, status.Status, "re-insert of block %d while redoing", h)
- }
- require.NoError(t, eat.MockCl.UpdateForkChoice(ctx, payloads[tip-2]))
- assertChurnState(ctx, t, eat, churn, payloads[tip-2], sums[tip-2])
-
- // Deep below what remains unwindable: must fail loudly, not silently
- // no-op or partially apply.
- deep := uint64(20)
- deepErr := eat.MockCl.UpdateForkChoice(ctx, payloads[deep-2])
- require.Errorf(t, deepErr, "unwind to block %d must be rejected once its history is gone", deep)
- t.Logf("deep unwind to %d rejected: %v", deep, deepErr)
-
- // The head must remain restorable and the state readable and correct.
- // Block production after this rejection is still broken (the
- // SeekCommitment wedge, https://github.com/erigontech/erigon/issues/22301),
- // so this test stops at the read-side contract.
- require.NoError(t, eat.MockCl.UpdateForkChoice(ctx, payloads[tip-2]))
- assertChurnState(ctx, t, eat, churn, payloads[tip-2], sums[tip-2])
- })
-}
diff --git a/execution/exec/txtask_test.go b/execution/exec/txtask_test.go
index 56e8156acc5..d1bd310de5e 100644
--- a/execution/exec/txtask_test.go
+++ b/execution/exec/txtask_test.go
@@ -213,7 +213,7 @@ func TestHistoricalBlockEndLogs(t *testing.T) {
} {
t.Run(tc.name, func(t *testing.T) {
logger := log.New()
- db := temporaltest.NewTestDBWithStepSize(t, datadir.New(t.TempDir()), 16)
+ db := temporaltest.NewTestDB(t, datadir.New(t.TempDir()), temporaltest.WithStepSize(16))
require.NoError(t, db.UpdateTemporal(t.Context(), func(rwTx kv.TemporalRwTx) error {
domains, err := execctx.NewSharedDomains(t.Context(), rwTx, logger)
if err != nil {
diff --git a/execution/execmodule/exec_module_non_finality_test.go b/execution/execmodule/exec_module_non_finality_test.go
new file mode 100644
index 00000000000..4118d329a92
--- /dev/null
+++ b/execution/execmodule/exec_module_non_finality_test.go
@@ -0,0 +1,256 @@
+// Copyright 2026 The Erigon Authors
+// This file is part of Erigon.
+//
+// Erigon 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.
+//
+// Erigon 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 Erigon. If not, see .
+
+package execmodule_test
+
+import (
+ "context"
+ "testing"
+ "time"
+
+ "github.com/holiman/uint256"
+ "github.com/stretchr/testify/require"
+
+ "github.com/erigontech/erigon/common"
+ "github.com/erigontech/erigon/db/kv"
+ "github.com/erigontech/erigon/execution/chain"
+ "github.com/erigontech/erigon/execution/execmodule"
+ "github.com/erigontech/erigon/execution/execmodule/execmoduletester"
+ "github.com/erigontech/erigon/execution/protocol/params"
+ "github.com/erigontech/erigon/execution/tests/blockgen"
+ "github.com/erigontech/erigon/execution/types"
+)
+
+func TestExecModule_GivenReorgPastFinalised_WhenFinality_ThenInvalidFCU(t *testing.T) {
+ // in normal circumstances our MAX_REORG_DEPTH aligns with the depth of the finalised hash
+ // (i.e. on ethereum we have T-96 finalised block in 99.999999% of the time and our MAX_REORG_DEPTH=96)
+ // this test check that when in that we're in that scenario, and we get a FCU for a fork that goes beyond
+ // the finalised number we return invalid fcu
+ ctx, cancel := context.WithTimeout(t.Context(), time.Minute)
+ defer cancel()
+ const (
+ e2RetireStepSize = 10
+ e3RetireStepSize = 3 // 3 txns per block (1 + 2 system txns)
+ maxReorgDepth = 2
+ chainLen = e2RetireStepSize + maxReorgDepth + 1
+ finalisedBlockNum = chainLen - maxReorgDepth // when normal finality matches our maxReorgDepth
+ reorgBlock = finalisedBlockNum - maxReorgDepth
+ )
+ emt := execmoduletester.New(
+ t,
+ execmoduletester.WithChainConfig(chain.AllProtocolChanges),
+ execmoduletester.WithMaxReorgDepth(maxReorgDepth),
+ execmoduletester.WithE2RetireStep(e2RetireStepSize),
+ execmoduletester.WithStepSize(e3RetireStepSize),
+ )
+ require.NoError(t, emt.WaitForBlockRetirement(ctx))
+ require.NoError(t, emt.WaitForStateRetirement(ctx))
+ cp1, err := emt.GenerateChain(chainLen, func(i int, gen *blockgen.BlockGen) {
+ tx, err := types.SignTx(
+ types.NewTransaction(
+ gen.TxNonce(emt.Address),
+ common.Address{1},
+ uint256.NewInt(10_000),
+ params.TxGas,
+ uint256.NewInt(emt.Genesis.BaseFee().Uint64()),
+ nil,
+ ),
+ *types.LatestSignerForChainID(emt.ChainConfig.ChainID),
+ emt.Key,
+ )
+ require.NoError(t, err)
+ gen.AddTx(tx)
+ })
+ require.NoError(t, err)
+ cp2, err := emt.GenerateChain(chainLen, func(i int, gen *blockgen.BlockGen) {
+ var to common.Address
+ if i < reorgBlock {
+ to = common.Address{1}
+ } else {
+ to = common.Address{2}
+ }
+ tx, err := types.SignTx(
+ types.NewTransaction(
+ gen.TxNonce(emt.Address),
+ to,
+ uint256.NewInt(10_000),
+ params.TxGas,
+ uint256.NewInt(emt.Genesis.BaseFee().Uint64()),
+ nil,
+ ),
+ *types.LatestSignerForChainID(emt.ChainConfig.ChainID),
+ emt.Key,
+ )
+ require.NoError(t, err)
+ gen.AddTx(tx)
+ })
+ require.NoError(t, err)
+ safeHash := cp1.Blocks[finalisedBlockNum-1].Hash()
+ finalisedHash := safeHash
+ fcuOptSeq := make([][]execmoduletester.UFCOpt, chainLen)
+ for i, h := range cp1.Headers {
+ if h.Number.Uint64() <= uint64(maxReorgDepth) {
+ fcuOptSeq[i] = []execmoduletester.UFCOpt{}
+ } else {
+ idx := min(i-maxReorgDepth, finalisedBlockNum-1)
+ fcuOptSeq[i] = []execmoduletester.UFCOpt{
+ execmoduletester.WithSafeHash(cp1.Headers[idx].Hash()),
+ execmoduletester.WithFinalisedHash(cp1.Headers[idx].Hash()),
+ }
+ }
+ }
+ // chain 1 block insert + fcu with head=T', safe=T'-2, finalised=T'-2
+ err = emt.InsertValidateAndUfc1By1(
+ ctx,
+ cp1.Blocks,
+ execmoduletester.WithFcuOptSeq(fcuOptSeq),
+ execmoduletester.WithWaitForBlockRetirement(),
+ execmoduletester.WithWaitForStateFiles(),
+ )
+ require.NoError(t, err)
+ // chain 2 block insert + fcu with head=T'', safe=T'-2, finalised=T'-2, reorgPoint=T'-4, maxReorgDepth=2
+ status, err := emt.InsertBlocks(ctx, cp2.Blocks[reorgBlock-1:])
+ require.NoError(t, err)
+ require.Equal(t, execmodule.ExecutionStatusSuccess, status)
+ result, err := emt.UpdateForkChoice(
+ ctx,
+ cp2.TopBlock.Header(),
+ execmoduletester.WithSafeHash(safeHash),
+ execmoduletester.WithFinalisedHash(finalisedHash),
+ )
+ require.NoError(t, err)
+ require.Equal(t, execmodule.ExecutionStatusInvalidForkchoice, result.Status)
+ // also check that we didnt create any snapshot files for non-finalised blocks
+ err = emt.BlockSnapshots.OpenFolder()
+ require.NoError(t, err)
+ require.Equal(t, uint64(9), emt.BlockSnapshots.BlocksAvailable())
+ tx, err := emt.DB.BeginTemporalRo(ctx)
+ require.NoError(t, err)
+ defer tx.Rollback()
+ require.Equal(t, uint64(33), tx.Debug().TxNumsInFiles(kv.CommitmentDomain))
+}
+
+func TestExecModule_GivenReorgPastMaxReorgDepth_WhenNonFinality_ThenReorg(t *testing.T) {
+ // in normal circumstances our MAX_REORG_DEPTH aligns with the depth of the finalised hash
+ // (i.e. on ethereum we have T-96 finalised block in 99.999999% of the time and our MAX_REORG_DEPTH=96)
+ // this test check that when in the highly unlikely scenario of non-finality we support long reorgs
+ // longer than MAX_REORG_DEPTH but not beyond the last finalised hash.
+ ctx, cancel := context.WithTimeout(t.Context(), time.Minute)
+ defer cancel()
+ const (
+ e2RetireStepSize = 10
+ e3RetireStepSize = 3 // 3 txns per block (1 + 2 system txns)
+ maxReorgDepth = 2
+ chainLen = e2RetireStepSize + maxReorgDepth + 1
+ finalisedBlockNum = chainLen - 3*maxReorgDepth // when non-finality exceeds our MAX_REORG_DEPTH
+ reorgBlock = finalisedBlockNum + 2 // and reorg block is still a descendant of the finalised block
+ )
+ emt := execmoduletester.New(
+ t,
+ execmoduletester.WithChainConfig(chain.AllProtocolChanges),
+ execmoduletester.WithMaxReorgDepth(maxReorgDepth),
+ execmoduletester.WithE2RetireStep(e2RetireStepSize),
+ execmoduletester.WithStepSize(e3RetireStepSize),
+ )
+ require.NoError(t, emt.WaitForBlockRetirement(ctx))
+ require.NoError(t, emt.WaitForStateRetirement(ctx))
+ cp1, err := emt.GenerateChain(chainLen, func(i int, gen *blockgen.BlockGen) {
+ tx, err := types.SignTx(
+ types.NewTransaction(
+ gen.TxNonce(emt.Address),
+ common.Address{1},
+ uint256.NewInt(10_000),
+ params.TxGas,
+ uint256.NewInt(emt.Genesis.BaseFee().Uint64()),
+ nil,
+ ),
+ *types.LatestSignerForChainID(emt.ChainConfig.ChainID),
+ emt.Key,
+ )
+ require.NoError(t, err)
+ gen.AddTx(tx)
+ })
+ require.NoError(t, err)
+ cp2, err := emt.GenerateChain(chainLen, func(i int, gen *blockgen.BlockGen) {
+ var to common.Address
+ if i+1 < reorgBlock {
+ to = common.Address{1}
+ } else {
+ to = common.Address{2}
+ }
+ nonce := gen.TxNonce(emt.Address)
+ tx, err := types.SignTx(
+ types.NewTransaction(
+ nonce,
+ to,
+ uint256.NewInt(10_000),
+ params.TxGas,
+ uint256.NewInt(emt.Genesis.BaseFee().Uint64()),
+ nil,
+ ),
+ *types.LatestSignerForChainID(emt.ChainConfig.ChainID),
+ emt.Key,
+ )
+ require.NoError(t, err)
+ gen.AddTx(tx)
+ })
+ require.NoError(t, err)
+ safeHash := cp1.Blocks[finalisedBlockNum-1].Hash()
+ finalisedHash := safeHash
+ fcuOptSeq := make([][]execmoduletester.UFCOpt, len(cp1.Blocks))
+ for i, h := range cp1.Headers {
+ if h.Number.Uint64() <= uint64(maxReorgDepth) {
+ fcuOptSeq[i] = []execmoduletester.UFCOpt{}
+ } else {
+ idx := min(i-maxReorgDepth, finalisedBlockNum-1)
+ fcuOptSeq[i] = []execmoduletester.UFCOpt{
+ execmoduletester.WithSafeHash(cp1.Headers[idx].Hash()),
+ execmoduletester.WithFinalisedHash(cp1.Headers[idx].Hash()),
+ }
+ }
+ }
+ // chain 1 block insert + fcu with head=T', safe=T'-6, finalised=T'-6
+ err = emt.InsertValidateAndUfc1By1(
+ ctx,
+ cp1.Blocks,
+ execmoduletester.WithFcuOptSeq(fcuOptSeq),
+ execmoduletester.WithWaitForBlockRetirement(),
+ execmoduletester.WithWaitForStateFiles(),
+ )
+ require.NoError(t, err)
+ // chain 2 block insert + fcu with head=T'', safe=T'-6, finalised=T'-6, reorgPoint=T'-4, maxReorgDepth=2
+ status, err := emt.InsertBlocks(ctx, cp2.Blocks[reorgBlock-1:])
+ require.NoError(t, err)
+ require.Equal(t, execmodule.ExecutionStatusSuccess, status)
+ result, err := emt.UpdateForkChoice(
+ ctx,
+ cp2.TopBlock.Header(),
+ execmoduletester.WithSafeHash(safeHash),
+ execmoduletester.WithFinalisedHash(finalisedHash),
+ )
+ require.NoError(t, err)
+ require.Equal(t, execmodule.ExecutionStatusSuccess, result.Status)
+ // also check that we didnt create any snapshot files for non-finalised blocks
+ require.NoError(t, emt.WaitForBlockRetirement(ctx))
+ require.NoError(t, emt.WaitForStateRetirement(ctx))
+ err = emt.BlockSnapshots.OpenFolder()
+ require.NoError(t, err)
+ require.Equal(t, uint64(0), emt.BlockSnapshots.BlocksAvailable())
+ tx, err := emt.DB.BeginTemporalRo(ctx)
+ require.NoError(t, err)
+ defer tx.Rollback()
+ require.Equal(t, uint64(21), tx.Debug().TxNumsInFiles(kv.CommitmentDomain))
+}
diff --git a/execution/execmodule/exec_module_test.go b/execution/execmodule/exec_module_test.go
index 68c203a457c..810dd3a4591 100644
--- a/execution/execmodule/exec_module_test.go
+++ b/execution/execmodule/exec_module_test.go
@@ -2610,10 +2610,8 @@ func TestInsertBlocksWithBatchedFCU_BadBlockRecovery(t *testing.T) {
}))
}
-// transferGen returns a deterministic per-block tx generator: identical
-// inputs produce identical blocks, which lets tests build forks that share
-// a prefix with the canonical chain (requires a pre-Cancun config — Cancun+
-// blocks get a random ParentBeaconBlockRoot in blockgen).
+// transferGen returns a deterministic per-block tx generator so tests can
+// build forks that share a prefix with the canonical chain.
func transferGen(t *testing.T, key *ecdsa.PrivateKey, to common.Address, amount uint64) func(int, *blockgen.BlockGen) {
return func(i int, b *blockgen.BlockGen) {
tx, err := types.SignTx(
@@ -2737,10 +2735,6 @@ func TestUpdateForkChoiceShallowReorgAfterLargeBatchExec(t *testing.T) {
// could not see that. It also asserts the compute-ahead path actually engaged (a
// real count) so a silent degrade-to-incremental can't make the differential pass
// trivially.
-//
-// (An end-to-end divergent-fork reorg would be a stronger check, but blockgen
-// randomises ParentBeaconBlockRoot per block, so under Amsterdam — required for
-// BALs — two chains can't share a prefix and a mid-chain parent has no state.)
func TestBALDrivenComputeAheadChangesetIntegrity(t *testing.T) {
off := runBALComputeAheadChangeset(t, false, false)
on := runBALComputeAheadChangeset(t, true, false)
@@ -2788,11 +2782,13 @@ func runBALComputeAheadChangeset(t *testing.T, computeAhead, shadow bool) balCom
dbg.BALShadowCompute = shadow
stagedsync.ResetComputedAheadForTest()
- const chainLen = 12
- // maxReorgDepth places the changeset window at maxBlock-depth = 12-4 = 8, so
- // blocks 1..7 are pre-window (compute-ahead candidates) and 8..12 own changesets.
- const maxReorgDepth = 4
- const windowStart = chainLen - maxReorgDepth
+ // Blocks 1..7 are compute-ahead candidates; blocks 8..12 own changesets.
+ const (
+ chainLen = 12
+ maxReorgDepth = 4
+ windowStart = chainLen - maxReorgDepth
+ reorgBackTo = chainLen - 2
+ )
ctx := t.Context()
// Deterministic key: compute-ahead-off and compute-ahead-on must execute the identical chain
@@ -2816,16 +2812,30 @@ func runBALComputeAheadChangeset(t *testing.T, computeAhead, shadow bool) balCom
// AllProtocolChanges is post-London, so txs need a fee cap above the base fee
// (transferGen's 1-wei price is only valid on the pre-London default).
- canonical, err := m.GenerateChain(chainLen,
- func(i int, b *blockgen.BlockGen) {
+ forkRecipient := common.Address{0x42}
+ generate := func(divergent bool) *blockgen.ChainPack {
+ chainPack, err := m.GenerateChain(chainLen, func(i int, b *blockgen.BlockGen) {
+ to := senderAddr
+ amount := uint64(1_000)
+ if divergent && i >= reorgBackTo {
+ to = forkRecipient
+ amount = 2_000
+ }
tx, txErr := types.SignTx(
- types.NewTransaction(uint64(i), senderAddr, uint256.NewInt(1_000), 50000, uint256.NewInt(10_000_000_000), nil),
+ types.NewTransaction(uint64(i), to, uint256.NewInt(amount), 50000, uint256.NewInt(10_000_000_000), nil),
*types.LatestSignerForChainID(nil), privKey,
)
require.NoError(t, txErr)
b.AddTx(tx)
})
- require.NoError(t, err)
+ require.NoError(t, err)
+ return chainPack
+ }
+ canonical := generate(false)
+ fork := generate(true)
+ require.Equal(t, canonical.Blocks[reorgBackTo-1].Hash(), fork.Blocks[reorgBackTo-1].Hash())
+ require.NotEqual(t, canonical.Blocks[reorgBackTo].Hash(), fork.Blocks[reorgBackTo].Hash())
+ require.NotEqual(t, canonical.TopBlock.Root(), fork.TopBlock.Root())
insRes, err := m.InsertBlocks(ctx, canonical.Blocks)
require.NoError(t, err)
@@ -2872,32 +2882,22 @@ func runBALComputeAheadChangeset(t *testing.T, computeAhead, shadow bool) balCom
return nil
}))
- // End-to-end: FCU back into the window unwinds using those changesets, then
- // FCU forward re-executes. If the compute-ahead-built state were wrong, the
- // unwind restores a bad root and the forward re-exec fails.
- const reorgBackTo = chainLen - 2 // within the window (>= windowStart)
- back, err := m.UpdateForkChoice(ctx, canonical.Blocks[reorgBackTo-1].Header())
+ // The alternate suffix changes state, so success requires restoring the
+ // branch-point state before executing the fork.
+ insRes, err = m.InsertBlocks(ctx, fork.Blocks[reorgBackTo:])
require.NoError(t, err)
- require.Equal(t, execmodule.ExecutionStatusSuccess, back.Status, "reorg back must succeed")
- m.ExecModule.WaitIdle(ctx)
- require.NoError(t, m.DB.ViewTemporal(ctx, func(tx kv.TemporalTx) error {
- execProg, err := stages.GetStageProgress(tx, stages.Execution)
- require.NoError(t, err)
- require.Equal(t, uint64(reorgBackTo), execProg, "FCU back must unwind execution (consuming the window changesets)")
- return nil
- }))
-
- fwd, err := m.UpdateForkChoice(ctx, canonical.TopBlock.Header())
+ require.Equal(t, execmodule.ExecutionStatusSuccess, insRes)
+ reorg, err := m.UpdateForkChoice(ctx, fork.TopBlock.Header())
require.NoError(t, err)
- require.Equal(t, execmodule.ExecutionStatusSuccess, fwd.Status,
- "forward re-exec after unwind must reach the correct root (compute-ahead=%v); validationError=%q",
- computeAhead, fwd.ValidationError)
+ require.Equal(t, execmodule.ExecutionStatusSuccess, reorg.Status,
+ "divergent reorg must reach the correct root (compute-ahead=%v); validationError=%q",
+ computeAhead, reorg.ValidationError)
m.ExecModule.WaitIdle(ctx)
require.NoError(t, m.DB.ViewTemporal(ctx, func(tx kv.TemporalTx) error {
execProg, err := stages.GetStageProgress(tx, stages.Execution)
require.NoError(t, err)
require.Equal(t, uint64(chainLen), execProg)
- require.Equal(t, canonical.TopBlock.Hash(), rawdb.ReadHeadBlockHash(tx))
+ require.Equal(t, fork.TopBlock.Hash(), rawdb.ReadHeadBlockHash(tx))
return nil
}))
return res
diff --git a/execution/execmodule/execmoduletester/exec_module_tester.go b/execution/execmodule/execmoduletester/exec_module_tester.go
index c095bd97ae7..3c54d04d99e 100644
--- a/execution/execmodule/execmoduletester/exec_module_tester.go
+++ b/execution/execmodule/execmoduletester/exec_module_tester.go
@@ -102,35 +102,37 @@ type StateChangesClient interface {
// ExecModuleTester aims to construct all parts necessary to test the PoS GRPC API of our EthereumExecModule.
type ExecModuleTester struct {
sentryproto.UnimplementedSentryServer
- Ctx context.Context
- Log log.Logger
- tb testing.TB
- cancel context.CancelFunc
- DB kv.TemporalRwDB
- Dirs datadir.Dirs
- Engine rules.Engine
- ChainConfig *chain.Config
- Sync *stagedsync.Sync
- MiningSync *stagedsync.Sync
- PendingBlocks chan *types.Block
- MinedBlocks chan *types.BlockWithReceipts
- sentriesClient *sentry_multi_client.MultiClient
- Key *ecdsa.PrivateKey
- Genesis *types.Block
- SentryClient direct.SentryClient
- PeerId *typesproto.H512
- streams map[sentryproto.MessageId][]sentryproto.Sentry_MessagesServer
- sentMessagesMu sync.Mutex
- sentMessages []*sentryproto.OutboundMessageData
- StreamWg sync.WaitGroup
- ReceiveWg sync.WaitGroup
- Address common.Address
- ForkValidator *execmodule.ForkValidator
- ExecModule *execmodule.ExecModule
- BlockBuilder *builder.Builder
- StateCache *execmodule.Cache
- retirementStart chan bool
- retirementDone chan struct{}
+ Ctx context.Context
+ Log log.Logger
+ tb testing.TB
+ cancel context.CancelFunc
+ DB kv.TemporalRwDB
+ Dirs datadir.Dirs
+ Engine rules.Engine
+ ChainConfig *chain.Config
+ Sync *stagedsync.Sync
+ MiningSync *stagedsync.Sync
+ PendingBlocks chan *types.Block
+ MinedBlocks chan *types.BlockWithReceipts
+ sentriesClient *sentry_multi_client.MultiClient
+ Key *ecdsa.PrivateKey
+ Genesis *types.Block
+ SentryClient direct.SentryClient
+ PeerId *typesproto.H512
+ streams map[sentryproto.MessageId][]sentryproto.Sentry_MessagesServer
+ sentMessagesMu sync.Mutex
+ sentMessages []*sentryproto.OutboundMessageData
+ StreamWg sync.WaitGroup
+ ReceiveWg sync.WaitGroup
+ Address common.Address
+ ForkValidator *execmodule.ForkValidator
+ ExecModule *execmodule.ExecModule
+ BlockBuilder *builder.Builder
+ StateCache *execmodule.Cache
+ retirementStart chan bool
+ retirementDone chan struct{}
+ stateRetirementStart chan bool
+ stateRetirementDone chan struct{}
Notifications *shards.Notifications
stateChangesClient StateChangesClient
@@ -490,6 +492,7 @@ func New(tb testing.TB, opts ...Option) *ExecModuleTester {
cfg.StateStream = true
cfg.BatchSize = 5 * datasize.MB
cfg.Sync.BodyDownloadTimeoutSeconds = 10
+ cfg.Sync.ParallelStateFlushing = false
cfg.TxPool.Disable = !withTxPool
cfg.Dirs = dirs
if opt.alwaysGenerateChangesets != nil {
@@ -521,12 +524,11 @@ func New(tb testing.TB, opts ...Option) *ExecModuleTester {
logger.SetHandler(log.LvlFilterHandler(logLvl, log.StderrHandler))
ctx, ctxCancel := context.WithCancel(context.Background())
- var db kv.TemporalRwDB
+ dbOpts := []temporaltest.Option{temporaltest.WithReorgBlockDepth(cfg.Sync.MaxReorgDepth)}
if opt.stepSize != nil {
- db = temporaltest.NewTestDBWithStepSize(tb, dirs, *opt.stepSize)
- } else {
- db = temporaltest.NewTestDB(tb, dirs)
+ dbOpts = append(dbOpts, temporaltest.WithStepSize(*opt.stepSize))
}
+ db := temporaltest.NewTestDB(tb, dirs, dbOpts...)
// Enable domains before any background goroutines start (e.g. InsertChain
// spawns a pipeline that calls agg.OpenFolder concurrently).
@@ -564,6 +566,8 @@ func New(tb testing.TB, opts ...Option) *ExecModuleTester {
}
mock.retirementStart, _ = mock.Notifications.Events.AddRetirementStartSubscription()
mock.retirementDone, _ = mock.Notifications.Events.AddRetirementDoneSubscription()
+ mock.stateRetirementStart, _ = mock.Notifications.Events.AddStateRetirementStartSubscription()
+ mock.stateRetirementDone, _ = mock.Notifications.Events.AddStateRetirementDoneSubscription()
if tb != nil {
tb.Cleanup(mock.Close)
@@ -872,9 +876,10 @@ func (emt *ExecModuleTester) ValidateChain(ctx context.Context, header *types.He
})
}
-func (emt *ExecModuleTester) UpdateForkChoice(ctx context.Context, header *types.Header) (execmodule.ForkChoiceResult, error) {
+func (emt *ExecModuleTester) UpdateForkChoice(ctx context.Context, header *types.Header, opts ...UFCOpt) (execmodule.ForkChoiceResult, error) {
+ ufcOpts := applyUfcOpts(opts...)
return retryBusy(ctx, func() (execmodule.ForkChoiceResult, bool, error) {
- result, err := emt.ExecModule.UpdateForkChoice(ctx, header.Hash(), common.Hash{}, common.Hash{})
+ result, err := emt.ExecModule.UpdateForkChoice(ctx, header.Hash(), ufcOpts.safeHash, ufcOpts.finalisedHash)
if err != nil {
return execmodule.ForkChoiceResult{}, false, err
}
@@ -900,7 +905,29 @@ func (emt *ExecModuleTester) WaitForBlockRetirement(ctx context.Context) error {
}
}
-func (emt *ExecModuleTester) InsertValidateAndUfc1By1(ctx context.Context, blocks []*types.Block) error {
+func (emt *ExecModuleTester) WaitForStateRetirement(ctx context.Context) error {
+ select {
+ case started := <-emt.stateRetirementStart:
+ if !started {
+ return nil
+ }
+ case <-ctx.Done():
+ return fmt.Errorf("waiting for state retirement start: %w", ctx.Err())
+ }
+
+ select {
+ case <-emt.stateRetirementDone:
+ return nil
+ case <-ctx.Done():
+ return fmt.Errorf("waiting for state retirement completion: %w", ctx.Err())
+ }
+}
+
+func (emt *ExecModuleTester) InsertValidateAndUfc1By1(ctx context.Context, blocks []*types.Block, opt ...IVUOpt) error {
+ ivuOpts := applyIVUOpts(opt...)
+ if len(ivuOpts.fcuOptSeq) > 0 && len(ivuOpts.fcuOptSeq) != len(blocks) {
+ panic(fmt.Errorf("length of fcuOptSeq %d must equal length of blocks %d", len(ivuOpts.fcuOptSeq), len(blocks)))
+ }
insertStatus, err := emt.InsertBlocks(ctx, blocks)
if err != nil {
return err
@@ -908,7 +935,7 @@ func (emt *ExecModuleTester) InsertValidateAndUfc1By1(ctx context.Context, block
if insertStatus != execmodule.ExecutionStatusSuccess {
return fmt.Errorf("unexpected insertBlocks status: %s", insertStatus)
}
- for _, block := range blocks {
+ for i, block := range blocks {
header := block.Header()
validationResult, err := emt.ValidateChain(ctx, header)
if err != nil {
@@ -918,18 +945,31 @@ func (emt *ExecModuleTester) InsertValidateAndUfc1By1(ctx context.Context, block
return fmt.Errorf("unexpected validateChain status: %s (block %d, validation error: %q)",
validationResult.ValidationStatus, header.Number.Uint64(), validationResult.ValidationError)
}
- forkChoiceResult, err := emt.UpdateForkChoice(ctx, header)
+ var ufcOpt []UFCOpt
+ if len(ivuOpts.fcuOptSeq) > 0 {
+ ufcOpt = ivuOpts.fcuOptSeq[i]
+ }
+ forkChoiceResult, err := emt.UpdateForkChoice(ctx, header, ufcOpt...)
if err != nil {
return err
}
if forkChoiceResult.Status != execmodule.ExecutionStatusSuccess {
return fmt.Errorf("unexpected updateForkChoice status: %s", forkChoiceResult.Status)
}
+ if ivuOpts.waitForBlockRetirement {
+ err := emt.WaitForBlockRetirement(ctx)
+ if err != nil {
+ return err
+ }
+ }
+ if ivuOpts.waitForStateFiles {
+ err := emt.WaitForStateRetirement(ctx)
+ if err != nil {
+ return err
+ }
+ }
}
- if len(blocks) > 0 {
- _, err = emt.UpdateForkChoice(ctx, blocks[len(blocks)-1].Header())
- }
- return err
+ return nil
}
func (emt *ExecModuleTester) AssembleBlock(ctx context.Context, params *builder.Parameters) (uint64, error) {
@@ -1004,7 +1044,7 @@ func (emt *ExecModuleTester) insertChain(chain *blockgen.ChainPack) error {
tipHash := chain.TopBlock.Hash()
- status, verr, _, err := wr.UpdateForkChoice(emt.Ctx, tipHash, tipHash, tipHash)
+ status, verr, _, err := wr.UpdateForkChoice(emt.Ctx, tipHash, emt.Genesis.Hash(), emt.Genesis.Hash())
if err != nil {
return err
}
diff --git a/execution/execmodule/execmoduletester/options.go b/execution/execmodule/execmoduletester/options.go
new file mode 100644
index 00000000000..c82529fdaf6
--- /dev/null
+++ b/execution/execmodule/execmoduletester/options.go
@@ -0,0 +1,80 @@
+// Copyright 2026 The Erigon Authors
+// This file is part of Erigon.
+//
+// Erigon 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.
+//
+// Erigon 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 Erigon. If not, see .
+
+package execmoduletester
+
+import "github.com/erigontech/erigon/common"
+
+type UFCOpt func(o *ufcOpt)
+
+func WithSafeHash(h common.Hash) UFCOpt {
+ return func(o *ufcOpt) {
+ o.safeHash = h
+ }
+}
+
+func WithFinalisedHash(h common.Hash) UFCOpt {
+ return func(o *ufcOpt) {
+ o.finalisedHash = h
+ }
+}
+
+type ufcOpt struct {
+ safeHash common.Hash
+ finalisedHash common.Hash
+}
+
+func applyUfcOpts(opts ...UFCOpt) ufcOpt {
+ var o ufcOpt
+ for _, opt := range opts {
+ opt(&o)
+ }
+ return o
+}
+
+type IVUOpt func(o *ivuOpt)
+
+func WithFcuOptSeq(seq [][]UFCOpt) IVUOpt {
+ return func(o *ivuOpt) {
+ o.fcuOptSeq = seq
+ }
+}
+
+func WithWaitForBlockRetirement() IVUOpt {
+ return func(o *ivuOpt) {
+ o.waitForBlockRetirement = true
+ }
+}
+
+func WithWaitForStateFiles() IVUOpt {
+ return func(o *ivuOpt) {
+ o.waitForStateFiles = true
+ }
+}
+
+type ivuOpt struct {
+ fcuOptSeq [][]UFCOpt
+ waitForBlockRetirement bool
+ waitForStateFiles bool
+}
+
+func applyIVUOpts(opts ...IVUOpt) ivuOpt {
+ var o ivuOpt
+ for _, opt := range opts {
+ opt(&o)
+ }
+ return o
+}
diff --git a/execution/execmodule/forkchoice.go b/execution/execmodule/forkchoice.go
index f5c4ed47245..4bc3c4935da 100644
--- a/execution/execmodule/forkchoice.go
+++ b/execution/execmodule/forkchoice.go
@@ -230,6 +230,12 @@ func (e *ExecModule) unwindIfNeeded(
return nil, err
}
for !isCanonicalHash {
+ if currentParentNumber < finalisedBlockNum {
+ return &ForkChoiceResult{
+ LatestValidHash: common.Hash{},
+ Status: ExecutionStatusInvalidForkchoice,
+ }, nil
+ }
newCanonicals = append(newCanonicals, &canonicalEntry{
hash: currentParentHash,
number: currentParentNumber,
@@ -879,16 +885,24 @@ func (e *ExecModule) runForkchoicePrune(initialCycle bool) ([]any, error) {
baseTimeout := time.Duration(e.config.SecondsPerSlot()*1000/3) * time.Millisecond
maxTimeout := time.Duration(e.config.SecondsPerSlot()*2000/3) * time.Millisecond
pruneTimeout := min(baseTimeout+time.Duration(agg.MaxPrunableStepsBacklog()/100)*200*time.Millisecond, maxTimeout)
- if err := agg.CollateAndPrune(e.backgroundCtx, e.db, func(tx kv.TemporalRwTx) error {
+ started, finished, err := agg.CollateAndPrune(e.backgroundCtx, e.db, func(tx kv.TemporalRwTx) error {
if e.codeStore != nil {
if err := e.codeStore.Evict(tx); err != nil {
return err
}
}
return e.pipelineExecutor.RunPrune(e.backgroundCtx, tx, initialCycle, pruneTimeout)
- }, e.logger); err != nil {
+ }, e.logger)
+ if err != nil {
return nil, err
}
+ e.hook.NotifyStateRetirementStart(started)
+ if started {
+ go func() {
+ <-finished
+ e.hook.NotifyStateRetirementDone()
+ }()
+ }
}
}
diff --git a/execution/stagedsync/committer_step_boundary_test.go b/execution/stagedsync/committer_step_boundary_test.go
index a9e4ca6dfd8..1fa4d2dcd12 100644
--- a/execution/stagedsync/committer_step_boundary_test.go
+++ b/execution/stagedsync/committer_step_boundary_test.go
@@ -564,7 +564,7 @@ func setupStepTest(t *testing.T) (kv.TemporalRwDB, kv.TemporalRwTx, *execctx.Sha
ctx := context.Background()
logger := log.New()
dirs := datadir.New(t.TempDir())
- db := temporaltest.NewTestDBWithStepSize(t, dirs, 16)
+ db := temporaltest.NewTestDB(t, dirs, temporaltest.WithStepSize(16))
tx, err := db.BeginTemporalRw(ctx) //nolint:gocritic
require.NoError(t, err)
diff --git a/execution/stagedsync/exec3_parallel_test.go b/execution/stagedsync/exec3_parallel_test.go
index fc7689c6f96..ac2bd3a02b9 100644
--- a/execution/stagedsync/exec3_parallel_test.go
+++ b/execution/stagedsync/exec3_parallel_test.go
@@ -1375,7 +1375,7 @@ func newResumeTestDB(t *testing.T) kv.TemporalRwDB {
t.Skip("mdbx InMem test databases are not supported on windows")
}
dirs := datadir.New(t.TempDir())
- db := temporaltest.NewTestDBWithStepSize(t, dirs, 16)
+ db := temporaltest.NewTestDB(t, dirs, temporaltest.WithStepSize(16))
return db
}
diff --git a/execution/stagedsync/stage_execute.go b/execution/stagedsync/stage_execute.go
index 96b5658284b..3f10de9fe1c 100644
--- a/execution/stagedsync/stage_execute.go
+++ b/execution/stagedsync/stage_execute.go
@@ -642,12 +642,19 @@ func PruneExecutionStage(ctx context.Context, s *PruneState, tx kv.TemporalRwTx,
return remaining
}
+ var blockPruneTo uint64
+ finalisedBlockNum := rawdb.ReadForkchoiceFinalizedNum(tx)
+ if finalisedBlockNum > 0 {
+ blockPruneTo = finalisedBlockNum
+ } else if s.ForwardProgress > cfg.syncCfg.MaxReorgDepth {
+ blockPruneTo = s.ForwardProgress - cfg.syncCfg.MaxReorgDepth
+ }
// AlwaysGenerateChangesets disables this prune so the node retains
// changesets for unwinds deeper than MaxReorgDepth (debug / integration
// tool / explicit --experimental.always-generate-changesets flag).
// Without the guard, the flag still controls *generation* but every
// generated changeset is pruned 96 blocks later, defeating the point.
- if s.ForwardProgress > cfg.syncCfg.MaxReorgDepth && !cfg.syncCfg.AlwaysGenerateChangesets {
+ if !cfg.syncCfg.AlwaysGenerateChangesets {
// (chunkLen is 8Kb) * (1_000 chunks) = 8mb
// Some chains produce blocks with 400 chunks of diff = 3mb
if pruneChangeSetsTimeout := remainingPruneTimeout(); pruneChangeSetsTimeout > 0 {
@@ -655,7 +662,7 @@ func PruneExecutionStage(ctx context.Context, s *PruneState, tx kv.TemporalRwTx,
if err := rawdb.PruneTable(
tx,
kv.ChangeSets3,
- s.ForwardProgress-cfg.syncCfg.MaxReorgDepth,
+ blockPruneTo,
ctx,
pruneDiffsLimit,
pruneChangeSetsTimeout,
@@ -675,20 +682,18 @@ func PruneExecutionStage(ctx context.Context, s *PruneState, tx kv.TemporalRwTx,
}
}
- if s.ForwardProgress > cfg.syncCfg.MaxReorgDepth {
- if pruneTimeout := remainingPruneTimeout(); pruneTimeout > 0 {
- if err := rawdb.PruneTable(
- tx,
- kv.BlockAccessList,
- s.ForwardProgress-cfg.syncCfg.MaxReorgDepth,
- ctx,
- pruneBalLimit,
- pruneTimeout,
- logger,
- s.LogPrefix(),
- ); err != nil {
- return err
- }
+ if pruneTimeout := remainingPruneTimeout(); pruneTimeout > 0 {
+ if err := rawdb.PruneTable(
+ tx,
+ kv.BlockAccessList,
+ blockPruneTo,
+ ctx,
+ pruneBalLimit,
+ pruneTimeout,
+ logger,
+ s.LogPrefix(),
+ ); err != nil {
+ return err
}
}
diff --git a/execution/stagedsync/stage_execute_resume_test.go b/execution/stagedsync/stage_execute_resume_test.go
index 0dacc732b86..73f72a7bd92 100644
--- a/execution/stagedsync/stage_execute_resume_test.go
+++ b/execution/stagedsync/stage_execute_resume_test.go
@@ -39,7 +39,7 @@ func TestResolveExecResumePoint(t *testing.T) {
ctx := context.Background()
dirs := datadir.New(t.TempDir())
- db := temporaltest.NewTestDBWithStepSize(t, dirs, 10_000)
+ db := temporaltest.NewTestDB(t, dirs, temporaltest.WithStepSize(10_000))
tx, err := db.BeginTemporalRw(ctx)
require.NoError(t, err)
diff --git a/execution/stagedsync/stage_execute_unwind_routing_test.go b/execution/stagedsync/stage_execute_unwind_routing_test.go
index b0e4c626333..e30988da62a 100644
--- a/execution/stagedsync/stage_execute_unwind_routing_test.go
+++ b/execution/stagedsync/stage_execute_unwind_routing_test.go
@@ -152,7 +152,7 @@ func TestUnwindOnExecError(t *testing.T) {
// temporal tx: one 40-byte ChangeSets3 key pins the lowest unwindable block
// to 4, so CanUnwindToBlockNum = 3 <= 10 and the target is not clamped.
dirs := datadir.New(t.TempDir())
- db := temporaltest.NewTestDBWithStepSize(t, dirs, 10_000)
+ db := temporaltest.NewTestDB(t, dirs, temporaltest.WithStepSize(10_000))
tx, err := db.BeginTemporalRw(context.Background())
require.NoError(t, err)
defer tx.Rollback()
diff --git a/execution/stagedsync/stage_execute_unwind_test.go b/execution/stagedsync/stage_execute_unwind_test.go
index 1e14ff7c7d2..a1112e7deb2 100644
--- a/execution/stagedsync/stage_execute_unwind_test.go
+++ b/execution/stagedsync/stage_execute_unwind_test.go
@@ -48,7 +48,7 @@ func TestUnwindExecutionStage_PrunesUncommittedOverlayWrite(t *testing.T) {
logger := log.New()
dirs := datadir.New(t.TempDir())
- db := temporaltest.NewTestDBWithStepSize(t, dirs, 10_000)
+ db := temporaltest.NewTestDB(t, dirs, temporaltest.WithStepSize(10_000))
snaps := db.(freezeblocks.HasBlockFiles).DebugBlockFiles()
br := freezeblocks.NewBlockReader(snaps)
@@ -189,7 +189,7 @@ func TestFindExecutedDiffsetAtHeight_FallsBackAfterCanonicalReorg(t *testing.T)
logger := log.New()
dirs := datadir.New(t.TempDir())
- db := temporaltest.NewTestDBWithStepSize(t, dirs, 16)
+ db := temporaltest.NewTestDB(t, dirs, temporaltest.WithStepSize(16))
// Block reader backed only by MDBX — the unwind range is at the tip, above any
// frozen snapshot boundary, so no snapshots are needed.
diff --git a/execution/stagedsync/stageloop/stageloop.go b/execution/stagedsync/stageloop/stageloop.go
index 03ad33f0cdb..322ba28db57 100644
--- a/execution/stagedsync/stageloop/stageloop.go
+++ b/execution/stagedsync/stageloop/stageloop.go
@@ -194,6 +194,20 @@ func (h *Hook) NotifySyncState(tx kv.Tx) {
}
}
+func (h *Hook) NotifyStateRetirementStart(started bool) {
+ if h == nil || h.notifications == nil || h.notifications.Events == nil {
+ return
+ }
+ h.notifications.Events.OnStateRetirementStart(started)
+}
+
+func (h *Hook) NotifyStateRetirementDone() {
+ if h == nil || h.notifications == nil || h.notifications.Events == nil {
+ return
+ }
+ h.notifications.Events.OnStateRetirementDone()
+}
+
func (h *Hook) maybeAnnounceBlockRange(finishStageBeforeSync, finishStageAfterSync uint64, isSynced bool) {
if h.blockRangePublisher == nil || h.statusDataGetter == nil {
return
diff --git a/execution/state/state_test.go b/execution/state/state_test.go
index 36fffd904e3..51ad1da50e8 100644
--- a/execution/state/state_test.go
+++ b/execution/state/state_test.go
@@ -519,7 +519,7 @@ func NewTestRwTx(tb testing.TB) (kv.TemporalRwDB, kv.TemporalRwTx, *execctx.Shar
dirs := datadir.New(tb.TempDir())
stepSize := uint64(16)
- db := temporaltest.NewTestDBWithStepSize(tb, dirs, stepSize)
+ db := temporaltest.NewTestDB(tb, dirs, temporaltest.WithStepSize(stepSize))
tb.Cleanup(db.Close)
tx, err := db.BeginTemporalRw(context.Background()) //nolint:gocritic
require.NoError(tb, err)
diff --git a/execution/tests/blockgen/chain_makers.go b/execution/tests/blockgen/chain_makers.go
index ab62182059a..c925cea4c08 100644
--- a/execution/tests/blockgen/chain_makers.go
+++ b/execution/tests/blockgen/chain_makers.go
@@ -22,7 +22,6 @@ package blockgen
import (
"bytes"
"context"
- "crypto/rand"
"errors"
"fmt"
@@ -496,11 +495,10 @@ func GenerateChain(config *chain.Config, parent *types.Block, engine rules.Engin
// Set ParentBeaconBlockRoot for Cancun+ blocks before InitializeBlockExecution
// so that EIP-4788 can store it during initialization.
if config.IsCancun(b.header.Time) {
- var beaconBlockRoot common.Hash
- if _, err := rand.Read(beaconBlockRoot[:]); err != nil {
- return nil, nil, fmt.Errorf("can't create beacon block root: %w", err)
- }
- b.header.ParentBeaconBlockRoot = &beaconBlockRoot
+ beaconBlockRoot := b.header.Hash().U256()
+ beaconBlockRoot.AddUint64(&beaconBlockRoot, 1)
+ parentBeaconBlockRoot := common.U256ToHash(beaconBlockRoot)
+ b.header.ParentBeaconBlockRoot = &parentBeaconBlockRoot
}
if b.engine != nil {
// Set tx context for system init call (txIndex -1)
diff --git a/execution/tests/state_database_test.go b/execution/tests/state_database_test.go
index e07be890578..cacd7f0a503 100644
--- a/execution/tests/state_database_test.go
+++ b/execution/tests/state_database_test.go
@@ -1612,7 +1612,7 @@ func TestTxLookupUnwind(t *testing.T) {
func newTestRwTx(tb testing.TB) (kv.TemporalRwDB, kv.TemporalRwTx, *execctx.SharedDomains) {
tb.Helper()
dirs := datadir.New(tb.TempDir())
- db := temporaltest.NewTestDBWithStepSize(tb, dirs, 16)
+ db := temporaltest.NewTestDB(tb, dirs, temporaltest.WithStepSize(16))
tb.Cleanup(db.Close)
tx, err := db.BeginTemporalRw(context.Background()) //nolint:gocritic
require.NoError(tb, err)
diff --git a/node/eth/backend.go b/node/eth/backend.go
index 42a9320be06..41538f3c349 100644
--- a/node/eth/backend.go
+++ b/node/eth/backend.go
@@ -1241,7 +1241,12 @@ func SetUpBlockReader(ctx context.Context, db kv.RwDB, dirs datadir.Dirs, snConf
if settingsErr != nil {
return nil, nil, nil, nil, settingsErr
}
- aggOpts := state.New(dirs).Logger(logger).SanityOldNaming().GenSaltIfNeed(createNewSaltFileIfNeeded).WithErigonDBSettings(erigonDBSettings)
+ aggOpts := state.New(dirs).
+ Logger(logger).
+ SanityOldNaming().
+ GenSaltIfNeed(createNewSaltFileIfNeeded).
+ WithErigonDBSettings(erigonDBSettings).
+ ReorgBlockDepth(snConfig.MaxReorgDepth)
if snConfig.ErigondbDomainStepsInFrozenFile != nil {
v := *snConfig.ErigondbDomainStepsInFrozenFile
stepsStr := "Inf"
diff --git a/node/shards/events.go b/node/shards/events.go
index c2f3329bb40..16edabe8b4c 100644
--- a/node/shards/events.go
+++ b/node/shards/events.go
@@ -49,6 +49,8 @@ type Events struct {
syncStateSubscriptions map[int]chan *remoteproto.SyncingReply
retirementStartSubscription map[int]chan bool
retirementDoneSubscription map[int]chan struct{}
+ stateRetirementStartSubs map[int]chan bool
+ stateRetirementDoneSubs map[int]chan struct{}
pendingLogsSubscriptions map[int]PendingLogsSubscription
pendingBlockSubscriptions map[int]PendingBlockSubscription
pendingTxsSubscriptions map[int]PendingTxsSubscription
@@ -76,6 +78,8 @@ func NewEvents() *Events {
syncStateSubscriptions: map[int]chan *remoteproto.SyncingReply{},
retirementStartSubscription: map[int]chan bool{},
retirementDoneSubscription: map[int]chan struct{}{},
+ stateRetirementStartSubs: map[int]chan bool{},
+ stateRetirementDoneSubs: map[int]chan struct{}{},
}
}
@@ -189,6 +193,36 @@ func (e *Events) AddRetirementDoneSubscription() (chan struct{}, func()) {
}
}
+func (e *Events) AddStateRetirementStartSubscription() (chan bool, func()) {
+ e.lock.Lock()
+ defer e.lock.Unlock()
+ ch := make(chan bool, 8)
+ e.id++
+ id := e.id
+ e.stateRetirementStartSubs[id] = ch
+ return ch, func() {
+ e.lock.Lock()
+ defer e.lock.Unlock()
+ delete(e.stateRetirementStartSubs, id)
+ close(ch)
+ }
+}
+
+func (e *Events) AddStateRetirementDoneSubscription() (chan struct{}, func()) {
+ e.lock.Lock()
+ defer e.lock.Unlock()
+ ch := make(chan struct{}, 8)
+ e.id++
+ id := e.id
+ e.stateRetirementDoneSubs[id] = ch
+ return ch, func() {
+ e.lock.Lock()
+ defer e.lock.Unlock()
+ delete(e.stateRetirementDoneSubs, id)
+ close(ch)
+ }
+}
+
func (e *Events) AddLogsSubscription() (chan []*notifications.LogNotification, func()) {
e.lock.Lock()
defer e.lock.Unlock()
@@ -321,6 +355,22 @@ func (e *Events) OnRetirementDone() {
}
}
+func (e *Events) OnStateRetirementStart(started bool) {
+ e.lock.Lock()
+ defer e.lock.Unlock()
+ for _, ch := range e.stateRetirementStartSubs {
+ common.PrioritizedSend(ch, started)
+ }
+}
+
+func (e *Events) OnStateRetirementDone() {
+ e.lock.Lock()
+ defer e.lock.Unlock()
+ for _, ch := range e.stateRetirementDoneSubs {
+ common.PrioritizedSend(ch, struct{}{})
+ }
+}
+
type Notifications struct {
Events *Events
Accumulator *Accumulator // StateAccumulator
diff --git a/node/shards/events_test.go b/node/shards/events_test.go
index 21adb96926d..cb95958811e 100644
--- a/node/shards/events_test.go
+++ b/node/shards/events_test.go
@@ -51,3 +51,26 @@ func TestSyncStateSubscriptionUnsubscribeStopsDelivery(t *testing.T) {
t.Fatal("expected closed channel with no pending notifications after unsubscribe")
}
}
+
+func TestStateRetirementSubscriptionsReceiveNotifications(t *testing.T) {
+ events := NewEvents()
+ start, unsubscribeStart := events.AddStateRetirementStartSubscription()
+ defer unsubscribeStart()
+ done, unsubscribeDone := events.AddStateRetirementDoneSubscription()
+ defer unsubscribeDone()
+
+ events.OnStateRetirementStart(true)
+ events.OnStateRetirementDone()
+
+ select {
+ case started := <-start:
+ require.True(t, started)
+ default:
+ t.Fatal("expected a state retirement start notification")
+ }
+ select {
+ case <-done:
+ default:
+ t.Fatal("expected a state retirement completion notification")
+ }
+}