rpc, execution, db: FcuBackgroundCommit groundwork, coherent cache fixes

cmd/rpcdaemon/cli/config.go

@@ -138,7 +138,7 @@ func RootCommand() (*cobra.Command, *httpcfg.HttpCfg) {
 	rootCmd.PersistentFlags().BoolVar(&cfg.GethCompatibility, "rpc.gethcompat", false, "Enables Geth-compatible storage iteration order for debug_storageRangeAt (sorted by keccak256 hash). Disabled by default for performance.")
 	rootCmd.PersistentFlags().StringVar(&cfg.TxPoolApiAddr, "txpool.api.addr", "", "txpool api network address, for example: 127.0.0.1:9090 (default: use value of --private.api.addr)")
 
-	rootCmd.PersistentFlags().StringVar(&stateCacheStr, "state.cache", "0MB", "Amount of data to store in StateCache (enabled if no --datadir set). Set 0 to disable StateCache. Defaults to 0MB RAM")
+	rootCmd.PersistentFlags().StringVar(&stateCacheStr, "state.cache", "0MB", "Amount of data to store in the version-keyed StateCache (an equally-sized code cache is budgeted on top). Set 0 to disable entry retention while preserving snapshot-consistent reads")
 	rootCmd.PersistentFlags().BoolVar(&cfg.GRPCServerEnabled, "grpc", false, "Enable GRPC server")
 	rootCmd.PersistentFlags().StringVar(&cfg.GRPCListenAddress, "grpc.addr", nodecfg.DefaultGRPCHost, "GRPC server listening interface")
 	rootCmd.PersistentFlags().IntVar(&cfg.GRPCPort, "grpc.port", nodecfg.DefaultGRPCPort, "GRPC server listening port")
@@ -242,6 +242,13 @@ type StateChangesClient interface {
 	StateChanges(ctx context.Context, in *remoteproto.StateChangeRequest, opts ...grpc.CallOption) (remoteproto.KV_StateChangesClient, error)
 }
 
+func newRemoteStateCache(cfg kvcache.CoherentConfig) kvcache.Cache {
+	if cfg.CacheSize == 0 && cfg.CodeCacheSize == 0 {
+		cfg.WaitForNewBlock = false
+	}
+	return kvcache.New(cfg)
+}
+
 func subscribeToStateChangesLoop(ctx context.Context, client StateChangesClient, cache kvcache.Cache) {
 	go func() {
 		for {
@@ -334,11 +341,8 @@ func EmbeddedServices(ctx context.Context,
 		// the overlay is always current, has zero memory overhead, and
 		// doesn't need the StateChanges gRPC stream to stay coherent.
 		stateCache = stateCacheCfg.LocalCache
-	} else if stateCacheCfg.CacheSize > 0 {
-		// Remote RPCDaemon: use coherent cache fed by StateChanges stream.
-		stateCache = kvcache.New(stateCacheCfg)
 	} else {
-		stateCache = kvcache.NewSimple()
+		stateCache = newRemoteStateCache(stateCacheCfg)
 	}
 
 	subscribeToStateChangesLoop(ctx, stateDiffClient, stateCache)
@@ -530,22 +534,18 @@ func RemoteServices(ctx context.Context, cfg *httpcfg.HttpCfg, logger log.Logger
 		if err != nil {
 			return nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, err
 		}
-		stateCache = kvcache.NewSimple()
 	}
 	// If DB can't be configured - used PrivateApiAddr as remote DB
 	if db == nil {
 		db = remoteKv
 	}
 
 	if !cfg.WithDatadir {
-		if cfg.StateCache.CacheSize > 0 {
-			stateCache = kvcache.New(cfg.StateCache)
-		} else {
-			stateCache = kvcache.NewSimple()
-		}
 		logger.Info("if you run RPCDaemon on same machine with Erigon add --datadir option")
 	}
 
+	stateCache = newRemoteStateCache(cfg.StateCache)
+
 	subscribeToStateChangesLoop(ctx, remoteKvClient, stateCache)
 
 	txpoolConn := conn

cmd/rpcdaemon/cli/config_test.go

@@ -24,11 +24,20 @@ import (
 	"github.com/holiman/uint256"
 	"github.com/stretchr/testify/require"
 
+	"github.com/erigontech/erigon/common"
 	"github.com/erigontech/erigon/common/log/v3"
+	"github.com/erigontech/erigon/db/datadir"
+	"github.com/erigontech/erigon/db/kv"
+	"github.com/erigontech/erigon/db/kv/kvcache"
+	"github.com/erigontech/erigon/db/kv/temporal/temporaltest"
+	"github.com/erigontech/erigon/db/state/execctx"
 	"github.com/erigontech/erigon/execution/chain"
 	"github.com/erigontech/erigon/execution/protocol/rules/ethash"
 	"github.com/erigontech/erigon/execution/protocol/rules/merge"
 	"github.com/erigontech/erigon/execution/types"
+	"github.com/erigontech/erigon/execution/types/accounts"
+	"github.com/erigontech/erigon/node/gointerfaces"
+	"github.com/erigontech/erigon/node/gointerfaces/remoteproto"
 )
 
 // TestIsWebsocket tests if an incoming websocket upgrade request is detected properly.
@@ -70,3 +79,55 @@ func TestRemoteRulesEngineFinalizeDelegates(t *testing.T) {
 		require.NoError(t, err)
 	})
 }
+
+func TestZeroBudgetRemoteCachePinsCommittedState(t *testing.T) {
+	cfg := kvcache.DefaultCoherentConfig
+	cfg.CacheSize = 0
+	cfg.CodeCacheSize = 0
+	cache := newRemoteStateCache(cfg)
+
+	db := temporaltest.NewTestDB(t, datadir.New(t.TempDir()))
+	addr := common.Address{1}
+	committedAccount := accounts.Account{Nonce: 1, Balance: *uint256.NewInt(1), CodeHash: accounts.EmptyCodeHash}
+	announcedAccount := committedAccount
+	announcedAccount.Nonce = 2
+	committedData := accounts.SerialiseV3(&committedAccount)
+	announcedData := accounts.SerialiseV3(&announcedAccount)
+
+	require.NoError(t, db.UpdateTemporal(t.Context(), func(tx kv.TemporalRwTx) error {
+		domains, err := execctx.NewSharedDomains(t.Context(), tx, log.New())
+		if err != nil {
+			return err
+		}
+		defer domains.Close()
+		if err := domains.DomainPut(kv.AccountsDomain, tx, addr[:], committedData, 0, nil); err != nil {
+			return err
+		}
+		return domains.Flush(t.Context(), tx)
+	}))
+
+	tx, err := db.BeginTemporalRo(t.Context())
+	require.NoError(t, err)
+	defer tx.Rollback()
+	stateVersion, err := tx.ReadSequence(string(kv.PlainStateVersion))
+	require.NoError(t, err)
+
+	cache.OnNewBlock(&remoteproto.StateChangeBatch{
+		StateVersionId: stateVersion + 1,
+		ChangeBatch: []*remoteproto.StateChange{{
+			Direction: remoteproto.Direction_FORWARD,
+			Changes: []*remoteproto.AccountChange{{
+				Action:  remoteproto.Action_UPSERT,
+				Address: gointerfaces.ConvertAddressToH160(addr),
+				Data:    announcedData,
+			}},
+		}},
+	})
+	require.Zero(t, cache.Len())
+
+	view, err := cache.View(t.Context(), tx)
+	require.NoError(t, err)
+	data, err := view.Get(addr[:])
+	require.NoError(t, err)
+	require.Equal(t, committedData, data)
+}

cmd/utils/app/import_cmd.go

@@ -417,7 +417,9 @@ func InsertChain(ethereum *eth.Ethereum, chain *blockgen.ChainPack, setHead bool
 	}
 
 	// UpdateForkChoice has an async commit so we need to wait to make sure
-	// it is completed before assuming all state changes etc are inserted
+	// it is completed before assuming all state changes etc are inserted.
+	// State-change events are dispatched pre-commit, so waiting on the stream
+	// only ensures the dispatcher fired — not that MDBX is flushed.
 	var lastSeenBlock uint64
 	for len(insertedBlocks) > 0 {
 		req, err := stream.Recv()
@@ -445,6 +447,10 @@ func InsertChain(ethereum *eth.Ethereum, chain *blockgen.ChainPack, setHead bool
 		}
 	}
 
+	// Wait for the FCU background commit so HeadBlockHash can't land in MDBX
+	// ahead of the header it points to.
+	ethereum.ExecutionModule().WaitIdle(ethereum.SentryCtx())
+
 	return ethereum.ChainDB().Update(ethereum.SentryCtx(), func(tx kv.RwTx) error {
 		rawdb.WriteHeadBlockHash(tx, lvh)
 		return nil

cmd/utils/flags.go

@@ -1143,7 +1143,7 @@ var (
 	}
 	FcuBackgroundCommitFlag = cli.BoolFlag{
 		Name:  "fcu.background.commit",
-		Usage: "Enables background flush and commit",
+		Usage: "Return FCU response before MDBX flush+commit lands (commit runs in background; remote rpcdaemon 'latest' stays consistent but can lag for the commit duration)",
 		Value: ethconfig.Defaults.FcuBackgroundCommit,
 	}
 	MCPDisableFlag = cli.BoolFlag{