execution: support long reorgs in non-finality

cmd/integration/commands/dump_state_test.go

@@ -155,7 +155,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)
@@ -206,7 +206,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)

cmd/integration/commands/stages.go

@@ -744,13 +744,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 {

cmd/utils/app/import_cmd.go

@@ -133,9 +133,11 @@ 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)
-	})
+	importSession, err := newChainImportSession(ethereum, ethereum.ChainDB(), logger)
+	if err != nil {
+		return err
+	}
+	return importSession.finish(importFiles(cliCtx.Args().Slice(), logger, importSession.importFile))
 }
 
 // importFiles imports each file in order; with more than one file, per-file
@@ -157,7 +159,57 @@ func importFiles(files []string, logger log.Logger, importOne func(fn string) er
 	return importErr
 }
 
+type chainImportSession struct {
+	ethereum      *eth.Ethereum
+	chainDB       kv.RwDB
+	logger        log.Logger
+	safeHash      common.Hash
+	finalizedHash common.Hash
+}
+
+func newChainImportSession(ethereum *eth.Ethereum, chainDB kv.RwDB, logger log.Logger) (*chainImportSession, error) {
+	session := &chainImportSession{ethereum: ethereum, chainDB: chainDB, logger: logger}
+	err := chainDB.View(context.Background(), func(tx kv.Tx) error {
+		session.safeHash = rawdb.ReadForkchoiceSafe(tx)
+		session.finalizedHash = rawdb.ReadForkchoiceFinalized(tx)
+		return nil
+	})
+	if err != nil {
+		return nil, err
+	}
+	return session, nil
+}
+
+func (s *chainImportSession) finish(importErr error) error {
+	if errors.Is(importErr, errInterrupted) {
+		return importErr
+	}
+	return errors.Join(importErr, s.finalizeHead())
+}
+
+func (s *chainImportSession) finalizeHead() error {
+	return s.chainDB.Update(context.Background(), func(tx kv.RwTx) error {
+		headHash := rawdb.ReadHeadBlockHash(tx)
+		if headHash == (common.Hash{}) {
+			return nil
+		}
+		rawdb.WriteForkchoiceHead(tx, headHash)
+		rawdb.WriteForkchoiceSafe(tx, headHash)
+		rawdb.WriteForkchoiceFinalized(tx, headHash)
+		return nil
+	})
+}
+
 func ImportChain(ethereum *eth.Ethereum, chainDB kv.RwDB, fn string, logger log.Logger) error {
+	importSession, err := newChainImportSession(ethereum, chainDB, logger)
+	if err != nil {
+		return err
+	}
+	return importSession.finish(importSession.importFile(fn))
+}
+
+func (s *chainImportSession) importFile(fn string) error {
+	ethereum, chainDB, logger := s.ethereum, s.chainDB, s.logger
 	// 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)
@@ -242,7 +294,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, s.safeHash, s.finalizedHash); err != nil {
 			return err
 		}
 	}
@@ -285,6 +337,14 @@ func missingBlocks(chainDB kv.RwDB, blocks []*types.Block, blockReader dbservice
 }
 
 func InsertChain(ethereum *eth.Ethereum, chain *blockgen.ChainPack, setHead bool) error {
+	if len(chain.Blocks) == 0 {
+		return nil
+	}
+	tipHash := chain.TopBlock.Hash()
+	return insertChain(ethereum, chain, setHead, tipHash, tipHash)
+}
+
+func insertChain(ethereum *eth.Ethereum, chain *blockgen.ChainPack, setHead bool, safeHash, finalizedHash common.Hash) error {
 	if len(chain.Blocks) == 0 {
 		return nil
 	}
@@ -398,7 +458,7 @@ func InsertChain(ethereum *eth.Ethereum, chain *blockgen.ChainPack, setHead bool
 	}
 
 	tipHash := chain.TopBlock.Hash()
-	status, validationErr, lvh, err := chainRW.UpdateForkChoice(ctx, tipHash, tipHash, tipHash)
+	status, validationErr, lvh, err := chainRW.UpdateForkChoice(ctx, tipHash, safeHash, finalizedHash)
 	if err != nil {
 		return err
 	}

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

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)

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)
 

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)

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

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 {

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