[DO-NOT-MERGE] txpool: streamline initialisation

cmd/txpool/main.go

@@ -28,6 +28,7 @@ import (
 
 	libcommon "github.com/erigontech/erigon-lib/common"
 	"github.com/erigontech/erigon-lib/common/datadir"
+	"github.com/erigontech/erigon-lib/common/paths"
 	"github.com/erigontech/erigon-lib/direct"
 	"github.com/erigontech/erigon-lib/gointerfaces"
 	"github.com/erigontech/erigon-lib/gointerfaces/grpcutil"
@@ -38,16 +39,12 @@ import (
 	"github.com/erigontech/erigon-lib/kv/remotedbserver"
 	"github.com/erigontech/erigon-lib/log/v3"
 	"github.com/erigontech/erigon/cmd/rpcdaemon/rpcdaemontest"
-	"github.com/erigontech/erigon/consensus/misc"
-	"github.com/erigontech/erigon/ethdb/privateapi"
-	"github.com/erigontech/erigon/txnprovider/txpool"
-	"github.com/erigontech/erigon/txnprovider/txpool/txpoolcfg"
-	"github.com/erigontech/erigon/txnprovider/txpool/txpoolutil"
-
-	"github.com/erigontech/erigon-lib/common/paths"
 	"github.com/erigontech/erigon/cmd/utils"
+	"github.com/erigontech/erigon/ethdb/privateapi"
 	"github.com/erigontech/erigon/turbo/debug"
 	"github.com/erigontech/erigon/turbo/logging"
+	"github.com/erigontech/erigon/txnprovider/txpool"
+	"github.com/erigontech/erigon/txnprovider/txpool/txpoolcfg"
 )
 
 var (
@@ -183,26 +180,31 @@ func doTxpool(ctx context.Context, logger log.Logger) error {
 		cfg.TracedSenders[i] = string(sender[:])
 	}
 
-	newTxs := make(chan txpool.Announcements, 1024)
-	defer close(newTxs)
-	txPoolDB, txPool, fetch, send, txpoolGrpcServer, err := txpoolutil.AllComponents(ctx, cfg,
-		kvcache.New(cacheConfig), newTxs, coreDB, sentryClients, kvClient, misc.Eip1559FeeCalculator, logger)
+	notifyMiner := func() {}
+	txPool, txpoolGrpcServer, err := txpool.Assemble(
+		ctx,
+		cfg,
+		coreDB,
+		kvcache.New(cacheConfig),
+		sentryClients,
+		kvClient,
+		notifyMiner,
+		logger,
+	)
 	if err != nil {
 		return err
 	}
-	defer txPoolDB.Close()
-	fetch.ConnectCore()
-	fetch.ConnectSentries()
 
 	miningGrpcServer := privateapi.NewMiningServer(ctx, &rpcdaemontest.IsMiningMock{}, nil, logger)
-
 	grpcServer, err := txpool.StartGrpc(txpoolGrpcServer, miningGrpcServer, txpoolApiAddr, nil, logger)
 	if err != nil {
 		return err
 	}
 
-	notifyMiner := func() {}
-	txpool.MainLoop(ctx, txPool, newTxs, send, txpoolGrpcServer.NewSlotsStreams, notifyMiner)
+	err = txPool.Run(ctx)
+	if err != nil {
+		return err
+	}
 
 	grpcServer.GracefulStop()
 	return nil

cmd/utils/flags.go

@@ -1562,6 +1562,9 @@ func setTxPool(ctx *cli.Context, dbDir string, fullCfg *ethconfig.Config) {
 	if ctx.IsSet(DbWriteMapFlag.Name) {
 		cfg.MdbxWriteMap = ctx.Bool(DbWriteMapFlag.Name)
 	}
+	if ctx.IsSet(TxPoolGossipDisableFlag.Name) {
+		cfg.NoGossip = ctx.Bool(TxPoolGossipDisableFlag.Name)
+	}
 	cfg.LogEvery = 3 * time.Minute
 	cfg.CommitEvery = libcommon.RandomizeDuration(ctx.Duration(TxPoolCommitEveryFlag.Name))
 	cfg.DBDir = dbDir
@@ -1985,10 +1988,6 @@ func SetEthConfig(ctx *cli.Context, nodeConfig *nodecfg.Config, cfg *ethconfig.C
 	if ctx.IsSet(TrustedSetupFile.Name) {
 		libkzg.SetTrustedSetupFilePath(ctx.String(TrustedSetupFile.Name))
 	}
-
-	if ctx.IsSet(TxPoolGossipDisableFlag.Name) {
-		cfg.DisableTxPoolGossip = ctx.Bool(TxPoolGossipDisableFlag.Name)
-	}
 }
 
 // SetDNSDiscoveryDefaults configures DNS discovery with the given URL if

eth/backend.go

@@ -87,7 +87,6 @@ import (
 	"github.com/erigontech/erigon/consensus/clique"
 	"github.com/erigontech/erigon/consensus/ethash"
 	"github.com/erigontech/erigon/consensus/merge"
-	"github.com/erigontech/erigon/consensus/misc"
 	"github.com/erigontech/erigon/core"
 	"github.com/erigontech/erigon/core/rawdb"
 	"github.com/erigontech/erigon/core/rawdb/blockio"
@@ -134,7 +133,6 @@ import (
 	"github.com/erigontech/erigon/txnprovider/shutter"
 	"github.com/erigontech/erigon/txnprovider/txpool"
 	"github.com/erigontech/erigon/txnprovider/txpool/txpoolcfg"
-	"github.com/erigontech/erigon/txnprovider/txpool/txpoolutil"
 )
 
 // Config contains the configuration options of the ETH protocol.
@@ -193,16 +191,12 @@ type Ethereum struct {
 	waitForStageLoopStop chan struct{}
 	waitForMiningStop    chan struct{}
 
-	txPoolDB                kv.RwDB
-	txPool                  *txpool.TxPool
-	newTxs                  chan txpool.Announcements
-	txPoolFetch             *txpool.Fetch
-	txPoolSend              *txpool.Send
-	txPoolGrpcServer        txpoolproto.TxpoolServer
-	shutterPool             *shutter.Pool
-	notifyMiningAboutNewTxs chan struct{}
-	forkValidator           *engine_helpers.ForkValidator
-	downloader              *downloader.Downloader
+	txPool                    *txpool.TxPool
+	txPoolGrpcServer          txpoolproto.TxpoolServer
+	shutterPool               *shutter.Pool
+	blockBuilderNotifyNewTxns chan struct{}
+	forkValidator             *engine_helpers.ForkValidator
+	downloader                *downloader.Downloader
 
 	agg            *libstate.Aggregator
 	blockSnapshots *freezeblocks.RoSnapshots
@@ -644,31 +638,44 @@ func New(ctx context.Context, stack *node.Node, config *ethconfig.Config, logger
 		return nil, err
 	}
 
-	var txnProvider txnprovider.TxnProvider
-	config.TxPool.NoGossip = config.DisableTxPoolGossip
-	var miningRPC txpoolproto.MiningServer
 	stateDiffClient := direct.NewStateDiffClientDirect(kvRPC)
+	var txnProvider txnprovider.TxnProvider
 	if config.TxPool.Disable {
 		backend.txPoolGrpcServer = &txpool.GrpcDisabled{}
 	} else {
-		backend.newTxs = make(chan txpool.Announcements, 1024)
-		backend.txPoolDB, backend.txPool, backend.txPoolFetch, backend.txPoolSend, backend.txPoolGrpcServer, err = txpoolutil.AllComponents(
-			ctx, config.TxPool, kvcache.NewDummy(), backend.newTxs, backend.chainDB, backend.sentriesClient.Sentries(), stateDiffClient, misc.Eip1559FeeCalculator, logger,
+		sentries := backend.sentriesClient.Sentries()
+		blockBuilderNotifyNewTxns := func() {
+			select {
+			case backend.blockBuilderNotifyNewTxns <- struct{}{}:
+			default:
+			}
+		}
+		backend.txPool, backend.txPoolGrpcServer, err = txpool.Assemble(
+			ctx,
+			config.TxPool,
+			backend.chainDB,
+			kvcache.NewDummy(),
+			sentries,
+			stateDiffClient,
+			blockBuilderNotifyNewTxns,
+			logger,
 		)
 		if err != nil {
 			return nil, err
 		}
+
 		txnProvider = backend.txPool
 	}
 	if config.Shutter.Enabled {
 		if config.TxPool.Disable {
 			panic("can't enable shutter pool when devp2p txpool is disabled")
 		}
-		backend.shutterPool = shutter.NewPool(logger, config.Shutter, txnProvider)
+
+		backend.shutterPool = shutter.NewPool(logger, config.Shutter, backend.txPool)
 		txnProvider = backend.shutterPool
 	}
 
-	backend.notifyMiningAboutNewTxs = make(chan struct{}, 1)
+	backend.blockBuilderNotifyNewTxns = make(chan struct{}, 1)
 	backend.miningSealingQuit = make(chan struct{})
 	backend.pendingBlocks = make(chan *types.Block, 1)
 	backend.minedBlocks = make(chan *types.Block, 1)
@@ -794,7 +801,7 @@ func New(ctx context.Context, stack *node.Node, config *ethconfig.Config, logger
 	agg.SetSnapshotBuildSema(blockSnapBuildSema)
 	blockRetire := freezeblocks.NewBlockRetire(1, dirs, blockReader, blockWriter, backend.chainDB, heimdallStore, bridgeStore, backend.chainConfig, config, backend.notifications.Events, blockSnapBuildSema, logger)
 
-	miningRPC = privateapi.NewMiningServer(ctx, backend, ethashApi, logger)
+	var miningRPC txpoolproto.MiningServer = privateapi.NewMiningServer(ctx, backend, ethashApi, logger)
 
 	var creds credentials.TransportCredentials
 	if stack.Config().PrivateApiAddr != "" {
@@ -824,26 +831,6 @@ func New(ctx context.Context, stack *node.Node, config *ethconfig.Config, logger
 	if currentBlock == nil {
 		currentBlock = genesis
 	}
-	// We start the transaction pool on startup, for a couple of reasons:
-	// 1) Hive tests requires us to do so and starting it from eth_sendRawTransaction is not viable as we have not enough data
-	// to initialize it properly.
-	// 2) we cannot propose for block 1 regardless.
-
-	if !config.TxPool.Disable {
-		backend.txPoolFetch.ConnectCore()
-		backend.txPoolFetch.ConnectSentries()
-		var newTxsBroadcaster *txpool.NewSlotsStreams
-		if casted, ok := backend.txPoolGrpcServer.(*txpool.GrpcServer); ok {
-			newTxsBroadcaster = casted.NewSlotsStreams
-		}
-		go txpool.MainLoop(backend.sentryCtx, backend.txPool, backend.newTxs, backend.txPoolSend, newTxsBroadcaster,
-			func() {
-				select {
-				case backend.notifyMiningAboutNewTxs <- struct{}{}:
-				default:
-				}
-			})
-	}
 
 	go func() {
 		defer debug.LogPanic()
@@ -1283,8 +1270,8 @@ func (s *Ethereum) StartMining(ctx context.Context, db kv.RwDB, stateDiffClient
 					// block info in the state channel
 					hasWork = true
 
-				case <-s.notifyMiningAboutNewTxs:
-					//log.Warn("[dbg] notifyMiningAboutNewTxs")
+				case <-s.blockBuilderNotifyNewTxns:
+					//log.Warn("[dbg] blockBuilderNotifyNewTxns")
 
 					// Skip mining based on new txn notif for bor consensus
 					hasWork = s.chainConfig.Bor == nil
@@ -1601,6 +1588,14 @@ func (s *Ethereum) Start() error {
 		}
 	}
 
+	if s.txPool != nil {
+		// We start the transaction pool on startup, for a couple of reasons:
+		// 1) Hive tests requires us to do so and starting it from eth_sendRawTransaction is not viable as we have not enough data
+		// to initialize it properly.
+		// 2) we cannot propose for block 1 regardless.
+		s.bgComponentsEg.Go(func() error { return s.txPool.Run(s.sentryCtx) })
+	}
+
 	if s.shutterPool != nil {
 		s.bgComponentsEg.Go(func() error { return s.shutterPool.Run(s.sentryCtx) })
 	}
@@ -1642,9 +1637,6 @@ func (s *Ethereum) Stop() error {
 	for _, sentryServer := range s.sentryServers {
 		sentryServer.Close()
 	}
-	if s.txPoolDB != nil {
-		s.txPoolDB.Close()
-	}
 	if s.agg != nil {
 		s.agg.Close()
 	}

eth/ethconfig/config.go

@@ -257,8 +257,6 @@ type Config struct {
 	SilkwormRpcLogDumpResponse   bool
 	SilkwormRpcNumWorkers        uint32
 	SilkwormRpcJsonCompatibility bool
-
-	DisableTxPoolGossip bool
 }
 
 type Sync struct {