diff --git a/cmd/utils/app/import_cmd.go b/cmd/utils/app/import_cmd.go index 3442606c9d9..7e64933a315 100644 --- a/cmd/utils/app/import_cmd.go +++ b/cmd/utils/app/import_cmd.go @@ -132,6 +132,8 @@ func importChain(ctx context.Context, cliCtx *cli.Command) error { if err != nil { return err } + // No lifecycle Start will consume the execution startup reservation. + ethereum.ExecutionModule().FinishStartup() return importFiles(cliCtx.Args().Slice(), logger, func(fn string) error { return ImportChain(ethereum, ethereum.ChainDB(), fn, logger) diff --git a/execution/execmodule/exec_module.go b/execution/execmodule/exec_module.go index 6d771fa4001..3f09b706359 100644 --- a/execution/execmodule/exec_module.go +++ b/execution/execmodule/exec_module.go @@ -184,11 +184,11 @@ type ExecModule struct { // MDBX database db kv.TemporalRwDB // main database - // semaphore is the module's single mutual-exclusion domain: it guards the - // pipeline Sync and all FCU state. Ops either TryAcquire and report Busy - // (retried by the CL) or block, and the background FCU commit/prune - // goroutines inherit the semaphore, releasing it only when their work is done. + // semaphore guards pipeline Sync and all FCU state. NewExecModule reserves + // its permit until startup completes; normal ops report Busy or block, and + // background FCU work retains the permit until cleanup completes. semaphore *semaphore.Weighted + startupOnce sync.Once forkValidator *ForkValidator pipelineExecutor *PipelineExecutor @@ -259,6 +259,10 @@ func NewExecModule( codeStore = cache.NewCodeStore(cache.DefaultCodeStoreMemBytes, cache.DefaultCodeStoreTableBytes) } forkValidator := newForkValidator(ctx, currentBlockNumber, pipelineExecutor, blockReader, syncCfg.MaxReorgDepth) + executionSemaphore := semaphore.NewWeighted(1) + if !executionSemaphore.TryAcquire(1) { + panic("assert: new execution semaphore rejected its startup reservation") + } em := &ExecModule{ blockReader: blockReader, @@ -269,7 +273,7 @@ func NewExecModule( builders: make(map[uint64]*builder.BlockBuilder), builderFunc: builderFunc, config: config, - semaphore: semaphore.NewWeighted(1), + semaphore: executionSemaphore, hook: hook, accum: accum, engine: engine, @@ -695,12 +699,10 @@ func (e *ExecModule) purgeBadChain(ctx context.Context, tx kv.RwTx, latestValidH } func (e *ExecModule) Start(ctx context.Context, hook *stageloop.Hook) { - if err := e.semaphore.Acquire(ctx, 1); err != nil { - if !errors.Is(err, context.Canceled) { - e.logger.Error("Could not start execution service", "err", err) - } - return - } + e.startupOnce.Do(func() { e.start(ctx, hook) }) +} + +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 { @@ -734,6 +736,11 @@ func (e *ExecModule) Start(ctx context.Context, hook *stageloop.Hook) { } } +// FinishStartup releases the reservation when this module does not run initial sync. +func (e *ExecModule) FinishStartup() { + e.startupOnce.Do(func() { e.semaphore.Release(1) }) +} + func (e *ExecModule) Ready(ctx context.Context) (bool, error) { // setup a timeout for the context to avoid waiting indefinitely ctxWithTimeout, cancel := context.WithTimeout(ctx, time.Second) diff --git a/execution/execmodule/exec_module_internal_test.go b/execution/execmodule/exec_module_internal_test.go index ce50ac66813..6b49a9bd15e 100644 --- a/execution/execmodule/exec_module_internal_test.go +++ b/execution/execmodule/exec_module_internal_test.go @@ -30,7 +30,9 @@ import ( "github.com/erigontech/erigon/common/log/v3" "github.com/erigontech/erigon/db/dbservices" "github.com/erigontech/erigon/db/kv" + "github.com/erigontech/erigon/db/kv/rawdbv3" "github.com/erigontech/erigon/execution/types" + "github.com/erigontech/erigon/node/ethconfig" ) type headerNumberErrorReader struct { @@ -57,6 +59,20 @@ type sideForkReader struct { forkBody *types.Body } +type readyBlockReader struct { + dbservices.FullBlockReader +} + +func (readyBlockReader) Ready(context.Context) <-chan error { + ready := make(chan error, 1) + ready <- nil + return ready +} + +func (readyBlockReader) TxnumReader() rawdbv3.TxNumsReader { + return rawdbv3.TxNumsReader{} +} + func (r sideForkReader) IsCanonical(_ context.Context, _ kv.Getter, hash common.Hash, _ uint64) (bool, error) { return hash == r.canonicalHash, nil } @@ -95,6 +111,57 @@ func TestNewDomainStateCacheRespectsUseStateCache(t *testing.T) { scDefault.Close() } +func newStartupTestExecModule(t *testing.T) *ExecModule { + t.Helper() + previous := dbg.UseStateCache + t.Cleanup(func() { dbg.SetUseStateCache(previous) }) + dbg.SetUseStateCache(false) + + em := NewExecModule( + t.Context(), + readyBlockReader{}, + nil, + &PipelineExecutor{}, + 0, + nil, + nil, + nil, + nil, + nil, + 0, + log.New(), + nil, + ethconfig.Sync{MaxReorgDepth: 1}, + false, + false, + nil, + func() error { return nil }, + ) + t.Cleanup(em.FinishStartup) + return em +} + +func TestExecModuleReadinessFollowsStartup(t *testing.T) { + em := newStartupTestExecModule(t) + + ready, err := em.Ready(t.Context()) + require.NoError(t, err) + require.False(t, ready) + + em.FinishStartup() + ready, err = em.Ready(t.Context()) + require.NoError(t, err) + require.True(t, ready) +} + +func TestExecModuleRejectsValidationBeforeStartup(t *testing.T) { + em := newStartupTestExecModule(t) + + result, err := em.ValidateChain(t.Context(), common.Hash{}, 0) + require.NoError(t, err) + require.Equal(t, ExecutionStatusBusy, result.ValidationStatus) +} + func TestUnwindToCommonCanonicalReturnsCanonicalityError(t *testing.T) { expectedErr := errors.New("canonicality read failed") e := &ExecModule{ diff --git a/execution/execmodule/execmoduletester/exec_module_tester.go b/execution/execmodule/execmoduletester/exec_module_tester.go index 2ec702c496e..aae456bf790 100644 --- a/execution/execmodule/execmoduletester/exec_module_tester.go +++ b/execution/execmodule/execmoduletester/exec_module_tester.go @@ -800,6 +800,7 @@ func New(tb testing.TB, opts ...Option) *ExecModuleTester { readAheader, func() error { return nil }, ) + mock.ExecModule.FinishStartup() mock.ForkValidator = mock.ExecModule.ForkValidator() mock.StreamWg.Add(1) diff --git a/node/eth/backend.go b/node/eth/backend.go index 449485bd0db..99dfcbae51a 100644 --- a/node/eth/backend.go +++ b/node/eth/backend.go @@ -1473,6 +1473,7 @@ func (s *Ethereum) Start() error { // execution-P2P layer (message listener, peer tracker, publisher), // and the peer-count logger. See node/components/sentry/provider.go. if err := s.sentryProvider.Start(s.sentryCtx); err != nil { + s.execModule.FinishStartup() return err } @@ -1487,10 +1488,11 @@ func (s *Ethereum) Start() error { return currentTD } - if chainspec.IsChainPoS(s.chainConfig, currentTDProvider) { + switch { + case chainspec.IsChainPoS(s.chainConfig, currentTDProvider): diaglib.Send(diaglib.SyncStageList{StagesList: diaglib.InitStagesFromList(s.pipelineStagedSync.StagesIdsList())}) go s.execModule.Start(s.sentryCtx, hook) - } else if s.chainConfig.Bor != nil { + case s.chainConfig.Bor != nil: diaglib.Send(diaglib.SyncStageList{StagesList: diaglib.InitStagesFromList(s.stagedSync.StagesIdsList())}) s.bgComponentsEg.Go(func() error { defer s.logger.Info("[polygon.sync] exeuction server start goroutine completed") @@ -1513,6 +1515,8 @@ func (s *Ethereum) Start() error { }() return err }) + default: + s.execModule.FinishStartup() } if s.txPool != nil {