eth/stagedsync: implement stage polygon sync forward

eth/backend.go

@@ -34,10 +34,6 @@ import (
 	"sync/atomic"
 	"time"
 
-	"github.com/ledgerwatch/erigon-lib/common/dir"
-	"github.com/ledgerwatch/erigon-lib/common/disk"
-	"github.com/ledgerwatch/erigon-lib/common/mem"
-
 	"github.com/erigontech/mdbx-go/mdbx"
 	lru "github.com/hashicorp/golang-lru/arc/v2"
 	"github.com/holiman/uint256"
@@ -54,6 +50,9 @@ import (
 	libcommon "github.com/ledgerwatch/erigon-lib/common"
 	"github.com/ledgerwatch/erigon-lib/common/datadir"
 	"github.com/ledgerwatch/erigon-lib/common/dbg"
+	"github.com/ledgerwatch/erigon-lib/common/dir"
+	"github.com/ledgerwatch/erigon-lib/common/disk"
+	"github.com/ledgerwatch/erigon-lib/common/mem"
 	"github.com/ledgerwatch/erigon-lib/config3"
 	"github.com/ledgerwatch/erigon-lib/direct"
 	"github.com/ledgerwatch/erigon-lib/downloader"
@@ -609,7 +608,7 @@ func New(ctx context.Context, stack *node.Node, config *ethconfig.Config, logger
 		}
 	}
 
-	sentryMcDisableBlockDownload := config.PolygonSync
+	sentryMcDisableBlockDownload := config.PolygonSync || config.PolygonSyncStage
 	backend.sentriesClient, err = sentry_multi_client.NewMultiClient(
 		backend.chainDB,
 		chainConfig,
@@ -847,6 +846,7 @@ func New(ctx context.Context, stack *node.Node, config *ethconfig.Config, logger
 	if config.PolygonSyncStage {
 		backend.syncStages = stages2.NewPolygonSyncStages(
 			backend.sentryCtx,
+			logger,
 			backend.chainDB,
 			config,
 			backend.chainConfig,
@@ -859,6 +859,10 @@ func New(ctx context.Context, stack *node.Node, config *ethconfig.Config, logger
 			backend.silkworm,
 			backend.forkValidator,
 			heimdallClient,
+			polygonSyncSentry(sentries),
+			p2pConfig.MaxPeers,
+			statusDataProvider,
+			backend.stopNode,
 		)
 		backend.syncUnwindOrder = stagedsync.PolygonSyncUnwindOrder
 		backend.syncPruneOrder = stagedsync.PolygonSyncPruneOrder
@@ -966,27 +970,12 @@ func New(ctx context.Context, stack *node.Node, config *ethconfig.Config, logger
 	}
 
 	if config.PolygonSync {
-		// TODO - pending sentry multi client refactor
-		//      - sentry multi client should conform to the SentryClient interface and internally
-		//        multiplex
-		//      - for now we just use 1 sentry
-		var sentryClient direct.SentryClient
-		for _, client := range sentries {
-			if client.Protocol() == direct.ETH68 {
-				sentryClient = client
-				break
-			}
-		}
-		if sentryClient == nil {
-			return nil, errors.New("nil sentryClient for polygon sync")
-		}
-
 		backend.polygonSyncService = polygonsync.NewService(
 			logger,
 			chainConfig,
 			dirs.DataDir,
 			tmpdir,
-			sentryClient,
+			polygonSyncSentry(sentries),
 			p2pConfig.MaxPeers,
 			statusDataProvider,
 			config.HeimdallURL,
@@ -1741,3 +1730,23 @@ func setBorDefaultTxPoolPriceLimit(chainConfig *chain.Config, config txpoolcfg.C
 		config.MinFeeCap = txpoolcfg.BorDefaultTxPoolPriceLimit
 	}
 }
+
+func polygonSyncSentry(sentries []direct.SentryClient) direct.SentryClient {
+	// TODO - pending sentry multi client refactor
+	//      - sentry multi client should conform to the SentryClient interface and internally
+	//        multiplex
+	//      - for now we just use 1 sentry
+	var sentryClient direct.SentryClient
+	for _, client := range sentries {
+		if client.Protocol() == direct.ETH68 {
+			sentryClient = client
+			break
+		}
+	}
+
+	if sentryClient == nil {
+		panic("nil sentryClient for polygon sync")
+	}
+
+	return sentryClient
+}

eth/stagedsync/bor_heimdall_shared.go

@@ -13,7 +13,9 @@ import (
 	"github.com/ledgerwatch/log/v3"
 
 	"github.com/ledgerwatch/erigon-lib/kv"
+	"github.com/ledgerwatch/erigon/accounts/abi"
 	"github.com/ledgerwatch/erigon/core/types"
+	"github.com/ledgerwatch/erigon/polygon/bor/borcfg"
 	"github.com/ledgerwatch/erigon/polygon/heimdall"
 	"github.com/ledgerwatch/erigon/rlp"
 	"github.com/ledgerwatch/erigon/turbo/services"
@@ -299,7 +301,7 @@ func fetchAndWriteHeimdallMilestonesIfNeeded(
 	}
 
 	for milestoneId := lastId + 1; milestoneId <= uint64(count) && (lastMilestone == nil || lastMilestone.EndBlock().Uint64() < toBlockNum); milestoneId++ {
-		if _, lastMilestone, err = fetchAndWriteHeimdallMilestone(ctx, milestoneId, uint64(count), tx, cfg.heimdallClient, logPrefix, logger); err != nil {
+		if _, lastMilestone, err = fetchAndWriteHeimdallMilestone(ctx, milestoneId, tx, cfg.heimdallClient, logPrefix, logger); err != nil {
 			if !errors.Is(err, heimdall.ErrNotInMilestoneList) {
 				return 0, err
 			}
@@ -318,7 +320,6 @@ var activeMilestones uint64 = 100
 func fetchAndWriteHeimdallMilestone(
 	ctx context.Context,
 	milestoneId uint64,
-	count uint64,
 	tx kv.RwTx,
 	heimdallClient heimdall.HeimdallClient,
 	logPrefix string,
@@ -368,24 +369,35 @@ func fetchRequiredHeimdallStateSyncEventsIfNeeded(
 		return lastStateSyncEventID, 0, 0, nil
 	}
 
-	return fetchAndWriteHeimdallStateSyncEvents(ctx, header, lastStateSyncEventID, tx, cfg, logPrefix, logger)
+	return fetchAndWriteHeimdallStateSyncEvents(
+		ctx,
+		header,
+		lastStateSyncEventID,
+		tx,
+		cfg.borConfig,
+		cfg.blockReader,
+		cfg.heimdallClient,
+		cfg.chainConfig.ChainID.String(),
+		cfg.stateReceiverABI,
+		logPrefix,
+		logger,
+	)
 }
 
 func fetchAndWriteHeimdallStateSyncEvents(
 	ctx context.Context,
 	header *types.Header,
 	lastStateSyncEventID uint64,
 	tx kv.RwTx,
-	cfg BorHeimdallCfg,
+	config *borcfg.BorConfig,
+	blockReader services.FullBlockReader,
+	heimdallClient heimdall.HeimdallClient,
+	chainID string,
+	stateReceiverABI abi.ABI,
 	logPrefix string,
 	logger log.Logger,
 ) (uint64, int, time.Duration, error) {
 	fetchStart := time.Now()
-	config := cfg.borConfig
-	blockReader := cfg.blockReader
-	heimdallClient := cfg.heimdallClient
-	chainID := cfg.chainConfig.ChainID.String()
-	stateReceiverABI := cfg.stateReceiverABI
 	// Find out the latest eventId
 	var (
 		from uint64

eth/stagedsync/default_stages.go

@@ -739,6 +739,7 @@ func StateStages(ctx context.Context, headers HeadersCfg, bodies BodiesCfg, bloc
 func PolygonSyncStages(
 	ctx context.Context,
 	snapshots SnapshotsCfg,
+	polygonSyncStageCfg PolygonSyncStageCfg,
 	blockHashCfg BlockHashesCfg,
 	senders SendersCfg,
 	exec ExecuteBlockCfg,
@@ -766,7 +767,7 @@ func PolygonSyncStages(
 			ID:          stages.PolygonSync,
 			Description: "Use polygon sync component to sync headers, bodies and heimdall data",
 			Forward: func(firstCycle bool, badBlockUnwind bool, s *StageState, unwinder Unwinder, txc wrap.TxContainer, logger log.Logger) error {
-				return SpawnPolygonSyncStage()
+				return SpawnPolygonSyncStage(ctx, txc.Tx, s, polygonSyncStageCfg)
 			},
 			Unwind: func(firstCycle bool, u *UnwindState, s *StageState, txc wrap.TxContainer, logger log.Logger) error {
 				return UnwindPolygonSyncStage()

eth/stagedsync/stage_bor_heimdall.go

@@ -32,12 +32,12 @@ import (
 	"github.com/ledgerwatch/erigon/polygon/bor/finality/whitelist"
 	"github.com/ledgerwatch/erigon/polygon/bor/valset"
 	"github.com/ledgerwatch/erigon/polygon/heimdall"
+	"github.com/ledgerwatch/erigon/polygon/sync"
 	"github.com/ledgerwatch/erigon/turbo/services"
 	"github.com/ledgerwatch/erigon/turbo/stages/headerdownload"
 )
 
 const (
-	InMemorySignatures      = 4096 // Number of recent block signatures to keep in memory
 	inmemorySnapshots       = 128  // Number of recent vote snapshots to keep in memory
 	snapshotPersistInterval = 1024 // Number of blocks after which to persist the vote snapshot to the database
 )
@@ -172,7 +172,7 @@ func BorHeimdallForward(
 		return err
 	}
 
-	signatures, err := lru.NewARC[libcommon.Hash, libcommon.Address](InMemorySignatures)
+	signatures, err := lru.NewARC[libcommon.Hash, libcommon.Address](sync.InMemorySignatures)
 	if err != nil {
 		return err
 	}