execution: FCU background-commit worker + generation-chain reader consistency

cl/beacon/handler/block_production_test.go

@@ -517,7 +517,7 @@ func TestCaplinBlockProductionWithWithdrawalRequest(t *testing.T) {
 		)
 		require.NoError(t, err)
 		gen.AddTx(tx)
-	})
+	}, m.PublishedSD())
 	require.NoError(t, err)
 	err = m.InsertChain(chainPack)
 	require.NoError(t, err)
@@ -649,7 +649,7 @@ func TestCaplinBlockProductionGlamsterdamSlotNumber(t *testing.T) {
 		)
 		require.NoError(t, err)
 		gen.AddTx(tx)
-	})
+	}, m.PublishedSD())
 	require.NoError(t, err)
 	err = m.InsertChain(chainPack)
 	require.NoError(t, err)

cmd/rpcdaemon/rpcdaemontest/test_util.go

@@ -37,6 +37,7 @@ import (
 	"github.com/erigontech/erigon/common/log/v3"
 	"github.com/erigontech/erigon/common/u256"
 	"github.com/erigontech/erigon/db/kv"
+	"github.com/erigontech/erigon/db/state/execctx"
 	"github.com/erigontech/erigon/execution/abi/bind"
 	"github.com/erigontech/erigon/execution/abi/bind/backends"
 	"github.com/erigontech/erigon/execution/builder"
@@ -113,11 +114,11 @@ func genTestChainOnce(t *testing.T) {
 		defer contractBackend.Close()
 
 		var err error
-		testOrphanedChain, err = blockgen.GenerateChain(m.ChainConfig, m.Genesis, m.Engine, m.DB, 5, func(i int, block *blockgen.BlockGen) {})
+		testOrphanedChain, err = blockgen.GenerateChain(m.ChainConfig, m.Genesis, m.Engine, m.DB, 5, func(i int, block *blockgen.BlockGen) {}, m.PublishedSD())
 		if err != nil {
 			t.Fatalf("rpcdaemontest: failed to generate orphaned chain: %v", err)
 		}
-		testChain, err = generateChain(&addresses, m.ChainConfig, m.Genesis, m.Engine, m.DB, contractBackend)
+		testChain, err = generateChain(&addresses, m.ChainConfig, m.Genesis, m.Engine, m.DB, contractBackend, m.PublishedSD())
 		if err != nil {
 			t.Fatalf("rpcdaemontest: failed to generate chain: %v", err)
 		}
@@ -166,6 +167,7 @@ func generateChain(
 	engine rules.Engine,
 	db kv.TemporalRwDB,
 	contractBackend *backends.SimulatedBackend,
+	readSD ...*execctx.SharedDomains,
 ) (*blockgen.ChainPack, error) {
 	var (
 		key      = addresses.key
@@ -349,7 +351,7 @@ func generateChain(
 			block.AddTx(txn)
 		}
 		contractBackend.Commit()
-	})
+	}, readSD...)
 }
 
 func computeMappingStorageKey(addr common.Address, slot uint64) common.Hash {
@@ -569,7 +571,7 @@ func CreateTestExecModuleForTraces(t *testing.T) *execmoduletester.ExecModuleTes
 		tx, _ := types.SignTx(types.NewTransaction(0, a2,
 			&u256.Num0, 50000, &u256.Num1, []byte{0x01, 0x00, 0x01, 0x00}), *types.LatestSignerForChainID(nil), key)
 		b.AddTx(tx)
-	})
+	}, m.PublishedSD())
 	if err != nil {
 		t.Fatalf("generate blocks: %v", err)
 	}
@@ -678,7 +680,7 @@ func CreateTestExecModuleForTracesCollision(t *testing.T) *execmoduletester.Exec
 		tx, _ = types.SignTx(types.NewTransaction(2, bb,
 			&u256.Num0, 100000, &u256.Num1, nil), *types.LatestSignerForChainID(nil), key)
 		b.AddTx(tx)
-	})
+	}, m.PublishedSD())
 	if err != nil {
 		t.Fatalf("generate blocks: %v", err)
 	}

cmd/utils/app/import_cmd.go

@@ -268,8 +268,10 @@ func missingBlocks(chainDB kv.RwDB, blocks []*types.Block, blockReader dbservice
 	})
 
 	for i, block := range blocks {
-		// If we're behind the chain head, only check block, state is available at head
-		if headBlock.NumberU64() > block.NumberU64() {
+		// No durable head yet (e.g. only genesis, or the head's commit is still
+		// in flight under background commit): nothing is behind us, so fall
+		// through to the per-block presence check.
+		if headBlock != nil && headBlock.NumberU64() > block.NumberU64() {
 			if !ChainHasBlock(chainDB, block) {
 				return blocks[i:]
 			}
@@ -445,6 +447,11 @@ func InsertChain(ethereum *eth.Ethereum, chain *blockgen.ChainPack, setHead bool
 		}
 	}
 
+	// UpdateForkChoice commits in the background; drain it so this file's
+	// header/TD/canonical state is durable in the raw DB before the next file's
+	// side-chain TD/head reads (and the final head write) observe it.
+	ethereum.ExecutionModule().WaitCommitsDrained()
+
 	return ethereum.ChainDB().Update(ethereum.SentryCtx(), func(tx kv.RwTx) error {
 		rawdb.WriteHeadBlockHash(tx, lvh)
 		return nil

db/kv/kv_interface.go

@@ -587,6 +587,10 @@ type TemporalMemBatch interface {
 	ClearRam()
 	IndexAdd(table InvertedIdx, key []byte, txNum uint64) (err error)
 	IteratePrefix(domain Domain, prefix []byte, roTx Tx, it func(k []byte, v []byte) (cont bool, err error)) error
+	RangeAsOf(ctx context.Context, domain Domain, fromKey, toKey []byte, ts uint64, asc order.By, limit int, roTx Tx) (stream.KV, error)
+	HistoryRange(ctx context.Context, domain Domain, fromTs, toTs int, asc order.By, limit int, roTx Tx) (stream.KV, error)
+	IndexRange(name InvertedIdx, k []byte, fromTs, toTs int, asc order.By, limit int, roTx Tx) (stream.U64, error)
+	HistorySeek(domain Domain, key []byte, ts uint64) (v []byte, ok bool, err error)
 	HasPrefix(domain Domain, prefix []byte, roTx Tx) ([]byte, []byte, bool, error)
 	HasPrefixInRAM(domain Domain, prefix []byte) bool
 	SizeEstimate() uint64
@@ -595,7 +599,6 @@ type TemporalMemBatch interface {
 	DiscardWrites(domain Domain)
 	Unwind(txNumUnwindTo uint64, changeset *[DomainLen][]DomainEntryDiff)
 	GetAsOf(domain Domain, key []byte, ts uint64) (v []byte, ok bool, err error)
-	HistorySeek(domain Domain, key []byte, ts uint64) (v []byte, ok bool, err error)
 	SetInMemHistoryReads(v bool)
 	InMemHistoryReads() bool
 }

db/kv/membatchwithdb/memory_mutation.go

@@ -42,6 +42,13 @@ type DomainReader interface {
 	HistorySeek(name kv.Domain, k []byte, ts uint64) ([]byte, bool, error)
 }
 
+// domainLatestReader is the subset of SharedDomains needed for GetLatest/HasPrefix
+// reads from an OverlayTemporalReadView. Implemented by SharedDomains.
+type domainLatestReader interface {
+	GetLatest(domain kv.Domain, tx kv.TemporalTx, k []byte) ([]byte, kv.Step, error)
+	HasPrefix(domain kv.Domain, prefix []byte, roTx kv.Tx) ([]byte, []byte, bool, error)
+}
+
 type MemoryMutation struct {
 	// mu protects concurrent access to the mutation's maps and backing tx.
 	// Read methods (GetOne, Has) acquire RLock; write methods (Put, Delete,
@@ -1141,13 +1148,25 @@ func (v *OverlayTemporalReadView) Apply(_ context.Context, f func(tx kv.Tx) erro
 	return f(v)
 }
 
-// Temporal methods — delegate to the independent temporal tx.
+// Temporal methods — check the installed SD chain before the committed tx.
 
 func (v *OverlayTemporalReadView) GetLatest(name kv.Domain, k []byte) ([]byte, kv.Step, error) {
+	// The installed reader walks the whole generation chain and falls through to
+	// v.temporalTx itself, so its value and error are authoritative — a nil here
+	// is a genuine tombstone, not a cue to re-read the committed tx.
+	if dr, ok := v.MemoryMutation.DomainReader.(domainLatestReader); ok {
+		return dr.GetLatest(name, v.temporalTx, k)
+	}
 	return v.temporalTx.GetLatest(name, k)
 }
 
 func (v *OverlayTemporalReadView) HasPrefix(name kv.Domain, prefix []byte) ([]byte, []byte, bool, error) {
+	// Authoritative for the same reason as GetLatest: the reader already honours
+	// parent-generation tombstones and the committed tx, so a false result must
+	// not be overridden by re-checking committed storage the overlay cleared.
+	if dr, ok := v.MemoryMutation.DomainReader.(domainLatestReader); ok {
+		return dr.HasPrefix(name, prefix, v.temporalTx)
+	}
 	return v.temporalTx.HasPrefix(name, prefix)
 }
 

db/snapshotsync/freezeblocks/block_reader.go

@@ -1436,6 +1436,9 @@ func (r *BlockReader) CurrentBlock(db kv.Tx) (*types.Block, error) {
 	if err != nil {
 		return nil, fmt.Errorf("failed HeaderNumber: %w", err)
 	}
+	if headNumber == nil {
+		return nil, nil
+	}
 	block, _, err := r.blockWithSenders(context.Background(), db, headHash, *headNumber, true)
 	return block, err
 }

db/snapshotsync/freezeblocks/dump_test.go

@@ -291,7 +291,7 @@ func createDumpTestKV(t *testing.T, chainConfig *chain.Config, chainSize int) *e
 			t.Fatalf("failed to create tx: %v", txErr)
 		}
 		b.AddTx(tx)
-	})
+	}, m.PublishedSD())
 	if err != nil {
 		t.Fatal(err)
 	}