freezeblocks, eth1: fix block retirement timing and prune backlog detection (#92)

Author: HugoDutchie

db/snapshotsync/freezeblocks/block_snapshots.go

@@ -455,7 +455,7 @@ func (br *BlockRetire) RetireBlocks(ctx context.Context, requestedMinBlockNum ui
 	}
 	includeBor := br.chainConfig.Bor != nil
 
-	if includeBor && time.Now().After(br.borDataNotReadyBefore) {
+	if includeBor && time.Now().Before(br.borDataNotReadyBefore) {
 		return nil
 	}
 

execution/eth1/forkchoice.go

@@ -46,6 +46,9 @@ import (
 
 const startPruneFrom = 1024
 
+// Track prune mode for logging only on transitions
+var lastPruneModeAggressive bool
+
 type forkchoiceOutcome struct {
 	receipt *executionproto.ForkChoiceReceipt
 	err     error
@@ -446,10 +449,48 @@ func (e *EthereumExecutionModule) updateForkChoice(ctx context.Context, original
 		}
 	}
 	// Run the forkchoice
+	// initialCycle enables aggressive prune rate (10K blocks vs 10 blocks per cycle).
+	// Enable if: big jump OR too many blocks in mdbx need retirement/pruning.
 	initialCycle := limitedBigJump
+	initialCycleReason := ""
+	if limitedBigJump {
+		initialCycleReason = "bigJump"
+	}
+
+	// Always calculate blocksInDB for logging
+	chainTip := fcuHeader.Number.Uint64()
+	frozenBlocks := e.blockReader.FrozenBlocks()
+	firstBlockInDB, ok, _ := rawdb.ReadFirstNonGenesisHeaderNumber(tx)
+	var blocksInDB uint64
+	if ok && firstBlockInDB > 0 && chainTip > firstBlockInDB {
+		blocksInDB = chainTip - firstBlockInDB
+	}
+
+	if !initialCycle && blocksInDB > 5_000 {
+		initialCycle = true
+		initialCycleReason = "backlog"
+	}
+
+	// Log only on mode transitions
+	if initialCycle != lastPruneModeAggressive {
+		lastPruneModeAggressive = initialCycle
+		if initialCycle {
+			e.logger.Info("[forkchoice] backlog prune: on",
+				"reason", initialCycleReason,
+				"blocksInDB", blocksInDB,
+				"frozenBlocks", frozenBlocks)
+		} else {
+			e.logger.Info("[forkchoice] backlog prune: off",
+				"blocksInDB", blocksInDB,
+				"frozenBlocks", frozenBlocks)
+		}
+	}
+
 	firstCycle := false
+	loopIter := 0
 	for {
 		hasMore, err := e.executionPipeline.Run(e.db, wrap.NewTxContainer(tx, nil), initialCycle, firstCycle)
+		loopIter++
 		if err != nil {
 			err = fmt.Errorf("updateForkChoice: %w", err)
 			e.logger.Warn("Cannot update chain head", "hash", blockHash, "err", err)
@@ -465,8 +506,10 @@ func (e *EthereumExecutionModule) updateForkChoice(ctx context.Context, original
 			return
 		}
 		if !hasMore {
+			e.logger.Debug("[forkchoice] execution loop done", "iterations", loopIter)
 			break
 		}
+		e.logger.Debug("[forkchoice] RunPrune in loop", "iteration", loopIter, "initialCycle", initialCycle)
 		err = e.executionPipeline.RunPrune(e.db, tx, initialCycle)
 		if err != nil {
 			err = fmt.Errorf("updateForkChoice: RunPrune after hasMore: %w", err)

execution/stagedsync/stage_snapshots.go

@@ -38,6 +38,7 @@ import (
 	"github.com/erigontech/erigon/db/kv/prune"
 	"github.com/erigontech/erigon/db/kv/rawdbv3"
 	"github.com/erigontech/erigon/db/kv/temporal"
+	"github.com/erigontech/erigon/db/rawdb"
 	"github.com/erigontech/erigon/db/snapshotsync"
 	"github.com/erigontech/erigon/db/snapshotsync/freezeblocks"
 	"github.com/erigontech/erigon/db/snaptype"
@@ -54,6 +55,10 @@ import (
 	"github.com/erigontech/erigon/turbo/silkworm"
 )
 
+// Track mode for logging only on transitions
+var lastRetireMode string
+var lastPruneMode string
+
 type SnapshotsCfg struct {
 	db          kv.TemporalRwDB
 	chainConfig *chain.Config
@@ -393,10 +398,42 @@ func SnapshotsPrune(s *PruneState, cfg SnapshotsCfg, ctx context.Context, tx kv.
 
 		var minBlockNumber uint64
 
+		// Calculate blocksInDB to detect backlog needing aggressive retirement
+		var blocksInDB uint64
+		chainTip, err := stages.GetStageProgress(tx, stages.Execution)
+		if err == nil && chainTip > 0 {
+			firstBlockInDB, ok, _ := rawdb.ReadFirstNonGenesisHeaderNumber(tx)
+			if ok && firstBlockInDB > 0 && chainTip > firstBlockInDB {
+				blocksInDB = chainTip - firstBlockInDB
+			}
+		}
+		frozenBlocks := cfg.blockReader.FrozenBlocks()
+
+		// Use aggressive workers when backlog exists, not just during initial sync
+		retireWorkers := 1
+		retireMode := "normal"
 		if s.CurrentSyncCycle.IsInitialCycle {
-			cfg.blockRetire.SetWorkers(estimate.CompressSnapshot.Workers())
-		} else {
-			cfg.blockRetire.SetWorkers(1)
+			retireWorkers = estimate.CompressSnapshot.Workers()
+			retireMode = "initialCycle"
+		} else if blocksInDB > 5_000 {
+			retireWorkers = estimate.CompressSnapshot.Workers()
+			retireMode = "backlog"
+		}
+		cfg.blockRetire.SetWorkers(retireWorkers)
+
+		// Log only on mode transitions
+		if retireMode != lastRetireMode {
+			lastRetireMode = retireMode
+			if retireMode == "backlog" {
+				logger.Info("[OtterSync] retire backlog: on",
+					"workers", retireWorkers,
+					"blocksInDB", blocksInDB,
+					"frozenBlocks", frozenBlocks)
+			} else if retireMode == "normal" {
+				logger.Info("[OtterSync] retire backlog: off",
+					"blocksInDB", blocksInDB,
+					"frozenBlocks", frozenBlocks)
+			}
 		}
 
 		noDl := cfg.snapshotDownloader == nil || reflect.ValueOf(cfg.snapshotDownloader).IsNil()
@@ -438,10 +475,44 @@ func SnapshotsPrune(s *PruneState, cfg SnapshotsCfg, ctx context.Context, tx kv.
 
 	pruneLimit := 10
 	pruneTimeout := 125 * time.Millisecond
+	pruneMode := "normal"
+	var pruneBlocksInDB uint64
+
 	if s.CurrentSyncCycle.IsInitialCycle {
 		pruneLimit = 10_000
 		pruneTimeout = time.Hour
+		pruneMode = "initialCycle"
+	} else {
+		// Check for backlog even when not initial cycle
+		chainTip, _ := stages.GetStageProgress(tx, stages.Execution)
+		firstBlockInDB, ok, _ := rawdb.ReadFirstNonGenesisHeaderNumber(tx)
+		if ok && firstBlockInDB > 0 && chainTip > firstBlockInDB {
+			pruneBlocksInDB = chainTip - firstBlockInDB
+			if pruneBlocksInDB > 5_000 {
+				pruneLimit = 10_000
+				pruneTimeout = time.Hour
+				pruneMode = "aggressive"
+			} else if pruneBlocksInDB > 2_500 {
+				pruneLimit = 1_000
+				pruneTimeout = 10 * time.Minute
+				pruneMode = "medium"
+			}
+		}
+	}
+
+	// Log only on mode transitions
+	if pruneMode != lastPruneMode {
+		prevMode := lastPruneMode
+		lastPruneMode = pruneMode
+		if pruneMode == "aggressive" {
+			logger.Info("[OtterSync] prune backlog: aggressive", "limit", pruneLimit, "blocksInDB", pruneBlocksInDB)
+		} else if pruneMode == "medium" {
+			logger.Info("[OtterSync] prune backlog: medium", "limit", pruneLimit, "blocksInDB", pruneBlocksInDB)
+		} else if pruneMode == "normal" && (prevMode == "aggressive" || prevMode == "medium") {
+			logger.Info("[OtterSync] prune backlog: off", "blocksInDB", pruneBlocksInDB)
+		}
 	}
+
 	if _, err := cfg.blockRetire.PruneAncientBlocks(tx, pruneLimit, pruneTimeout); err != nil {
 		return err
 	}