diff --git a/execution/execmodule/exec_module.go b/execution/execmodule/exec_module.go index 6d771fa4001..ae74f06120b 100644 --- a/execution/execmodule/exec_module.go +++ b/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) } diff --git a/execution/execmodule/exec_module_internal_test.go b/execution/execmodule/exec_module_internal_test.go index ce50ac66813..99cdb26b305 100644 --- a/execution/execmodule/exec_module_internal_test.go +++ b/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{ diff --git a/execution/execmodule/executor.go b/execution/execmodule/executor.go index c172ce51913..b76381b1668 100644 --- a/execution/execmodule/executor.go +++ b/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 },