execution/execmodule: reserve execution during startup

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)

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)

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{

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)

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 {