execution/execmodule: publish frozen-block commits to StateCache

execution/execmodule/exec_module.go

@@ -703,7 +703,7 @@ func (e *ExecModule) Start(ctx context.Context, hook *stageloop.Hook) {
 	}
 	defer e.semaphore.Release(1)
 
-	if err := e.pipelineExecutor.ProcessFrozenBlocks(ctx, hook, e.onlySnapDownloadOnStart); err != nil {
+	if err := e.pipelineExecutor.ProcessFrozenBlocks(ctx, hook, e.onlySnapDownloadOnStart, e.stateCache); err != nil {
 		if !errors.Is(err, context.Canceled) {
 			e.logger.Error("Could not start execution service", "err", err)
 		}

execution/execmodule/exec_module_internal_test.go

@@ -28,8 +28,11 @@ import (
 	"github.com/erigontech/erigon/common"
 	"github.com/erigontech/erigon/common/dbg"
 	"github.com/erigontech/erigon/common/log/v3"
+	"github.com/erigontech/erigon/db/datadir"
 	"github.com/erigontech/erigon/db/dbservices"
 	"github.com/erigontech/erigon/db/kv"
+	"github.com/erigontech/erigon/db/kv/temporal/temporaltest"
+	"github.com/erigontech/erigon/execution/cache"
 	"github.com/erigontech/erigon/execution/types"
 )
 
@@ -95,6 +98,49 @@ func TestNewDomainStateCacheRespectsUseStateCache(t *testing.T) {
 	scDefault.Close()
 }
 
+// Every SharedDomains used for frozen-block processing must publish catch-up
+// writes and revoke older views' fill authority through the normal commit path.
+func TestNewFrozenBlocksSDWiresStateCache(t *testing.T) {
+	t.Parallel()
+
+	ctx := t.Context()
+	db := temporaltest.NewTestDB(t, datadir.New(t.TempDir()))
+	sc := cache.NewStateCache(1<<20, 1<<20, 1<<20, 1<<20)
+	t.Cleanup(sc.Close)
+	sc.Applier().Initialize(0)
+
+	addr := make([]byte, 20)
+	addr[0] = 1
+	stale := []byte{1}
+	frontier := cache.FrontierWithStateVersion(cache.FrontierFunc(func(kv.Domain) (uint64, bool) {
+		return 10, true
+	}), 0)
+	preCatchup := sc.View(frontier)
+	preCatchup.Fill(kv.AccountsDomain, addr, stale, 5)
+
+	tx, err := db.BeginTemporalRw(ctx)
+	require.NoError(t, err)
+	defer tx.Rollback()
+	pe := &PipelineExecutor{logger: log.New()}
+	sd, err := pe.newFrozenBlocksSD(ctx, tx, sc)
+	require.NoError(t, err)
+	defer sd.Close()
+
+	fresh := []byte{2}
+	sd.SetTxNum(20)
+	require.NoError(t, sd.DomainPut(kv.AccountsDomain, tx, addr, fresh, 20, nil))
+	require.NoError(t, sd.Commit(ctx, tx))
+
+	got, ok := sc.View(nil).Get(kv.AccountsDomain, addr)
+	require.True(t, ok)
+	require.Equal(t, fresh, got, "catch-up commits must reach the cache")
+
+	preCatchup.Fill(kv.AccountsDomain, addr, stale, 5)
+	got, ok = sc.View(nil).Get(kv.AccountsDomain, addr)
+	require.True(t, ok)
+	require.Equal(t, fresh, got, "a pre-catch-up read view must not refill stale state")
+}
+
 func TestUnwindToCommonCanonicalReturnsCanonicalityError(t *testing.T) {
 	expectedErr := errors.New("canonicality read failed")
 	e := &ExecModule{

execution/execmodule/executor.go

@@ -28,6 +28,7 @@ import (
 	"github.com/erigontech/erigon/db/kv"
 	dbstate "github.com/erigontech/erigon/db/state"
 	"github.com/erigontech/erigon/db/state/execctx"
+	"github.com/erigontech/erigon/execution/cache"
 	"github.com/erigontech/erigon/execution/chain"
 	"github.com/erigontech/erigon/execution/protocol/rules"
 	"github.com/erigontech/erigon/execution/stagedsync"
@@ -177,10 +178,20 @@ func (pe *PipelineExecutor) RunLoop(ctx context.Context, sd *execctx.SharedDomai
 	return tx, sd, nil
 }
 
+func (pe *PipelineExecutor) newFrozenBlocksSD(ctx context.Context, tx kv.TemporalRwTx, stateCache *cache.StateCache) (*execctx.SharedDomains, error) {
+	sd, err := execctx.NewSharedDomains(ctx, tx, pe.logger)
+	if err != nil {
+		return nil, err
+	}
+	sd.SetInMemHistoryReads(inMemHistoryReads)
+	sd.SetStateCache(stateCache)
+	return sd, nil
+}
+
 // ProcessFrozenBlocks runs the pipeline over snapshot blocks at startup.
 // It downloads block files, then executes them in a hasMore loop until
 // all frozen blocks are processed.
-func (pe *PipelineExecutor) ProcessFrozenBlocks(ctx context.Context, hook *stageloop.Hook, onlySnapDownload bool) error {
+func (pe *PipelineExecutor) ProcessFrozenBlocks(ctx context.Context, hook *stageloop.Hook, onlySnapDownload bool, stateCache *cache.StateCache) error {
 	sawZeroBlocksTimes := 0
 	tx, err := pe.db.BeginTemporalRw(ctx)
 	if err != nil {
@@ -203,12 +214,11 @@ func (pe *PipelineExecutor) ProcessFrozenBlocks(ctx context.Context, hook *stage
 		return tx.Commit()
 	}
 
-	doms, err := execctx.NewSharedDomains(ctx, tx, pe.logger)
+	doms, err := pe.newFrozenBlocksSD(ctx, tx, stateCache)
 	if err != nil {
 		return err
 	}
 	defer func() { doms.Close() }() // RunLoop rotates doms; close whichever is current at exit
-	doms.SetInMemHistoryReads(inMemHistoryReads)
 
 	var finishStageBeforeSync uint64
 	if hook != nil {
@@ -247,11 +257,10 @@ func (pe *PipelineExecutor) ProcessFrozenBlocks(ctx context.Context, hook *stage
 				return nil, nil, err
 			}
 			tx = newTx
-			newSD, err := execctx.NewSharedDomains(ctx, newTx, pe.logger)
+			newSD, err := pe.newFrozenBlocksSD(ctx, newTx, stateCache)
 			if err != nil {
 				return nil, nil, err
 			}
-			newSD.SetInMemHistoryReads(inMemHistoryReads)
 			hook.NotifySyncState(newTx)
 			return newTx, newSD, nil
 		},