execmodule: extract PipelineExecutor for unified hasMore loop

mh0lt · mh0lt · commit fc65e1855fd8 · 2026-03-16T23:22:36.000Z
Consolidate the hasMore pipeline loop pattern shared by
ProcessFrozenBlocks (startup catchup) and updateForkChoice (FCU
catchup) into a single PipelineExecutor.RunLoop method.

Callers provide their specific commit/break/setup logic as callbacks:
- CommitCycleFn: flush SD, commit, return fresh tx
- ShouldBreakFn: optional early exit (e.g. frozen blocks reached)
- BeforeIterationFn: optional per-iteration setup (e.g. state cache)

Key design decision: RunPrune and CommitCycle only execute when
hasMore=true, since the FCU overlay tx (MemoryMutation on RO tx)
cannot handle PruneSmallBatches type assertions.
diff --git a/execution/execmodule/exec_module.go b/execution/execmodule/exec_module.go
@@ -538,11 +538,11 @@ func (e *ExecModule) Start(ctx context.Context, hook *stageloop.Hook) {
 	}
 	defer e.semaphore.Release(1)
 
-	if err := stageloop.ProcessFrozenBlocks(ctx, e.db, e.blockReader, e.executionPipeline, hook, e.onlySnapDownloadOnStart, e.logger); err != nil {
+	if err := e.processFrozenBlocks(ctx, hook); err != nil {
 		if !errors.Is(err, context.Canceled) {
 			e.logger.Error("Could not start execution service", "err", err)
 		}
-		// During parallel execution, an invalid block in initial sync (ProcessFrozenBlocks)
+		// During parallel execution, an invalid block in initial sync (processFrozenBlocks)
 		// is unrecoverable: the parallel executor cannot unwind and retrying will hit the
 		// same block forever, pushing Caplin's backward target further back.
 		// Exit the process so the operator can investigate.
@@ -558,6 +558,111 @@ func (e *ExecModule) Start(ctx context.Context, hook *stageloop.Hook) {
 	}
 }
 
+// processFrozenBlocks runs the pipeline over snapshot blocks at startup,
+// using PipelineExecutor for the hasMore loop.
+func (e *ExecModule) processFrozenBlocks(ctx context.Context, hook *stageloop.Hook) error {
+	sawZeroBlocksTimes := 0
+	tx, err := e.db.BeginTemporalRw(ctx)
+	if err != nil {
+		return err
+	}
+	defer tx.Rollback()
+
+	// Run snapshots stage — downloads block files.
+	if err = e.executionPipeline.RunSnapshots(nil, tx); err != nil {
+		return err
+	}
+	if e.onlySnapDownloadOnStart {
+		return nil
+	}
+
+	// If domains are ahead of block files, nothing to execute.
+	if execctx.IsDomainAheadOfBlocks(ctx, tx, e.logger) {
+		return tx.Commit()
+	}
+
+	doms, err := execctx.NewSharedDomains(ctx, tx, e.logger)
+	if err != nil {
+		return err
+	}
+	defer doms.Close()
+
+	var finishStageBeforeSync uint64
+	if hook != nil {
+		finishStageBeforeSync, err = stages.GetStageProgress(tx, stages.Finish)
+		if err != nil {
+			return err
+		}
+		if err = hook.BeforeRun(tx, false); err != nil {
+			return err
+		}
+	}
+
+	pe := NewPipelineExecutor(e.executionPipeline, e.logger)
+	result, err := pe.RunLoop(ctx, RunLoopConfig{
+		SD:           doms,
+		Tx:           tx,
+		InitialCycle: true,
+		FirstCycle:   false,
+		PruneTimeout: 0,
+		CommitCycle: func(ctx context.Context, sd *execctx.SharedDomains) (kv.TemporalRwTx, error) {
+			if err := sd.Flush(ctx, tx); err != nil {
+				return nil, fmt.Errorf("processFrozenBlocks: flush: %w", err)
+			}
+			sd.ClearRam(true)
+			if err := tx.Commit(); err != nil {
+				return nil, err
+			}
+			newTx, err := e.db.BeginTemporalRw(ctx) //nolint:gocritic
+			if err != nil {
+				return nil, err
+			}
+			tx = newTx
+			return newTx, nil
+		},
+		ShouldBreak: func(curTx kv.TemporalRwTx) (bool, error) {
+			if e.blockReader.FrozenBlocks() > 0 {
+				p, err := stages.GetStageProgress(curTx, stages.Finish)
+				if err != nil {
+					return false, err
+				}
+				return p >= e.blockReader.FrozenBlocks(), nil
+			}
+			sawZeroBlocksTimes++
+			return sawZeroBlocksTimes > 2, nil
+		},
+	})
+	if err != nil {
+		return fmt.Errorf("processFrozenBlocks: %w", err)
+	}
+	tx = result.FinalTx
+
+	if err := doms.Flush(ctx, tx); err != nil {
+		return fmt.Errorf("processFrozenBlocks: final flush: %w", err)
+	}
+	doms.ClearRam(true)
+	if err := tx.Commit(); err != nil {
+		return err
+	}
+
+	if hook != nil {
+		if err := e.db.View(ctx, func(tx kv.Tx) error {
+			headersProgress, err := stages.GetStageProgress(tx, stages.Headers)
+			if err != nil {
+				return err
+			}
+			if err = hook.AfterRun(tx, finishStageBeforeSync, false); err != nil {
+				return err
+			}
+			hook.LastNewBlockSeen(headersProgress)
+			return nil
+		}); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
 func (e *ExecModule) Ready(ctx context.Context, _ *emptypb.Empty) (*executionproto.ReadyResponse, error) {
 
 	// setup a timeout for the context to avoid waiting indefinitely
diff --git a/execution/execmodule/executor.go b/execution/execmodule/executor.go
@@ -0,0 +1,120 @@
+// Copyright 2024 The Erigon Authors
+// This file is part of Erigon.
+//
+// Erigon is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// Erigon is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with Erigon. If not, see <http://www.gnu.org/licenses/>.
+
+package execmodule
+
+import (
+	"context"
+	"time"
+
+	"github.com/erigontech/erigon/common/log/v3"
+	"github.com/erigontech/erigon/db/kv"
+	"github.com/erigontech/erigon/db/state/execctx"
+	"github.com/erigontech/erigon/execution/stagedsync"
+)
+
+// PipelineExecutor runs the staged sync pipeline in a hasMore loop,
+// abstracting the common pattern shared by ProcessFrozenBlocks (startup
+// catchup) and updateForkChoice (FCU catchup). The executor is stateless
+// between calls — callers provide their own tx/SD lifecycle and commit
+// strategy via callbacks.
+type PipelineExecutor struct {
+	sync   *stagedsync.Sync
+	logger log.Logger
+}
+
+// NewPipelineExecutor creates a new executor wrapping the given pipeline.
+func NewPipelineExecutor(sync *stagedsync.Sync, logger log.Logger) *PipelineExecutor {
+	return &PipelineExecutor{sync: sync, logger: logger}
+}
+
+// CommitCycleFn is called between hasMore iterations to persist accumulated
+// state and obtain a fresh tx for the next iteration. Implementations must
+// flush the SharedDomains, commit, and return a fresh kv.TemporalRwTx.
+type CommitCycleFn func(ctx context.Context, sd *execctx.SharedDomains) (kv.TemporalRwTx, error)
+
+// ShouldBreakFn is called after each iteration (after prune, before commit)
+// to check whether the loop should stop early. Return true to break.
+type ShouldBreakFn func(tx kv.TemporalRwTx) (bool, error)
+
+// BeforeIterationFn is called before each pipeline Run (e.g. to set state cache).
+type BeforeIterationFn func(sd *execctx.SharedDomains)
+
+// RunLoopConfig configures a single RunLoop invocation.
+type RunLoopConfig struct {
+	SD              *execctx.SharedDomains
+	Tx              kv.TemporalRwTx
+	InitialCycle    bool
+	FirstCycle      bool
+	PruneTimeout    time.Duration
+	CommitCycle     CommitCycleFn     // required when hasMore
+	ShouldBreak     ShouldBreakFn     // optional early exit
+	BeforeIteration BeforeIterationFn // optional per-iteration setup
+}
+
+// RunLoopResult contains the state after RunLoop completes.
+type RunLoopResult struct {
+	// FinalTx is the tx after the last iteration. May differ from the
+	// input Tx if CommitCycle was called (which returns a fresh tx).
+	FinalTx kv.TemporalRwTx
+}
+
+// RunLoop executes the pipeline in a hasMore loop:
+//
+//	for hasMore {
+//	    [BeforeIteration] → sync.Run → if hasMore { RunPrune → [ShouldBreak] → CommitCycle }
+//	}
+//
+// The loop continues until Run returns hasMore=false, ShouldBreak returns
+// true, or an error occurs. On error, FinalTx is the tx from the last
+// iteration (may be used by the caller for error recovery reads).
+func (pe *PipelineExecutor) RunLoop(ctx context.Context, cfg RunLoopConfig) (RunLoopResult, error) {
+	tx := cfg.Tx
+	for hasMore := true; hasMore; {
+		if cfg.BeforeIteration != nil {
+			cfg.BeforeIteration(cfg.SD)
+		}
+
+		var err error
+		hasMore, err = pe.sync.Run(cfg.SD, tx, cfg.InitialCycle, cfg.FirstCycle)
+		if err != nil {
+			return RunLoopResult{FinalTx: tx}, err
+		}
+
+		if hasMore {
+			if err := pe.sync.RunPrune(ctx, tx, cfg.InitialCycle, cfg.PruneTimeout); err != nil {
+				return RunLoopResult{FinalTx: tx}, err
+			}
+
+			if cfg.ShouldBreak != nil {
+				stop, err := cfg.ShouldBreak(tx)
+				if err != nil {
+					return RunLoopResult{FinalTx: tx}, err
+				}
+				if stop {
+					break
+				}
+			}
+
+			newTx, err := cfg.CommitCycle(ctx, cfg.SD)
+			if err != nil {
+				return RunLoopResult{FinalTx: tx}, err
+			}
+			tx = newTx
+		}
+	}
+	return RunLoopResult{FinalTx: tx}, nil
+}
diff --git a/execution/execmodule/forkchoice.go b/execution/execmodule/forkchoice.go
@@ -490,71 +490,69 @@ func (e *ExecModule) updateForkChoice(ctx context.Context, originalBlockHash, sa
 	initialCycle := limitedBigJump
 	firstCycle := false
 
-	sendError := func(msg string, err error) error {
-		err = fmt.Errorf("%s: %w", msg, err)
-		e.logger.Warn("Cannot update chain head", "hash", blockHash, "err", err)
-		return sendForkchoiceErrorWithoutWaiting(e.logger, outcomeCh, err, stateFlushingInParallel)
-	}
-
-	for hasMore := true; hasMore; {
-		currentContext.SetStateCache(e.stateCache)
-		hasMore, err = e.executionPipeline.Run(currentContext, tx, initialCycle, firstCycle)
-		if err != nil {
-			err = fmt.Errorf("updateForkChoice: %w", err)
-			e.logger.Warn("Cannot update chain head", "hash", blockHash, "err", err)
-			if errors.Is(err, rules.ErrInvalidBlock) {
-				return sendForkchoiceReceiptWithoutWaiting(outcomeCh, &executionproto.ForkChoiceReceipt{
-					Status:          executionproto.ExecutionStatus_BadBlock,
-					ValidationError: err.Error(),
-					LatestValidHash: gointerfaces.ConvertHashToH256(rawdb.ReadHeadBlockHash(tx)),
-				}, stateFlushingInParallel)
-			}
-			return sendForkchoiceErrorWithoutWaiting(e.logger, outcomeCh, err, stateFlushingInParallel)
-		}
-		if hasMore {
-			err = e.executionPipeline.RunPrune(ctx, tx, initialCycle, 500*time.Millisecond)
-			if err != nil {
-				return sendError("updateForkChoice: RunPrune after hasMore", err)
-			}
+	pe := NewPipelineExecutor(e.executionPipeline, e.logger)
+	result, err := pe.RunLoop(ctx, RunLoopConfig{
+		SD:           currentContext,
+		Tx:           tx,
+		InitialCycle: initialCycle,
+		FirstCycle:   firstCycle,
+		PruneTimeout: 500 * time.Millisecond,
+		BeforeIteration: func(sd *execctx.SharedDomains) {
+			sd.SetStateCache(e.stateCache)
+		},
+		CommitCycle: func(ctx context.Context, sd *execctx.SharedDomains) (kv.TemporalRwTx, error) {
 			// Flush SD + overlay to a brief RwTx to relieve memory pressure.
 			commitRwTx, err := e.db.BeginTemporalRw(ctx) //nolint:gocritic
 			if err != nil {
-				return sendError("updateForkChoice: begin rw after hasMore", err)
+				return nil, err
 			}
-			if err := currentContext.Flush(ctx, commitRwTx); err != nil {
+			if err := sd.Flush(ctx, commitRwTx); err != nil {
 				commitRwTx.Rollback()
-				return sendError("updateForkChoice: flush sd after hasMore", err)
+				return nil, err
 			}
 			if err := fcuOverlay.Flush(ctx, commitRwTx); err != nil {
 				commitRwTx.Rollback()
-				return sendError("updateForkChoice: flush overlay after hasMore", err)
+				return nil, err
 			}
-			currentContext.ClearRam(true)
+			sd.ClearRam(true)
 			if err = commitRwTx.Commit(); err != nil {
-				return sendError("updateForkChoice: tx commit after hasMore", err)
+				return nil, err
 			}
 			// Recreate RO tx + overlay on the fresh committed state.
 			fcuOverlay.Close()
 			roTx.Rollback()
 			roTx, err = e.db.BeginTemporalRo(ctx) //nolint:gocritic
 			if err != nil {
-				return sendError("updateForkChoice: begin ro after hasMore", err)
+				return nil, err
 			}
 			fcuOverlay, err = membatchwithdb.NewMemoryBatch(roTx, roTx.Debug().Dirs().Tmp, e.logger)
 			if err != nil {
 				roTx.Rollback()
-				return sendError("updateForkChoice: new overlay after hasMore", err)
+				return nil, err
 			}
-			tx = fcuOverlay
 			// Re-flush InsertBlocks data into the fresh overlay.
 			if hasOverlay {
 				e.currentContext.BlockOverlay().UpdateTxn(roTx)
 				if err := e.currentContext.BlockOverlay().Flush(ctx, fcuOverlay); err != nil {
-					return sendError("updateForkChoice: re-flush overlay after hasMore", err)
+					return nil, err
 				}
 			}
+			return fcuOverlay, nil
+		},
+	})
+	if err != nil {
+		err = fmt.Errorf("updateForkChoice: %w", err)
+		e.logger.Warn("Cannot update chain head", "hash", blockHash, "err", err)
+		if errors.Is(err, rules.ErrInvalidBlock) {
+			return sendForkchoiceReceiptWithoutWaiting(outcomeCh, &executionproto.ForkChoiceReceipt{
+				Status:          executionproto.ExecutionStatus_BadBlock,
+				ValidationError: err.Error(),
+				LatestValidHash: gointerfaces.ConvertHashToH256(rawdb.ReadHeadBlockHash(result.FinalTx)),
+			}, stateFlushingInParallel)
 		}
+		return sendForkchoiceErrorWithoutWaiting(e.logger, outcomeCh, err, stateFlushingInParallel)
 	}
+	tx = result.FinalTx
 
 	// if head hash was set then success otherwise no
 	headHash := rawdb.ReadHeadBlockHash(tx)
