feat: add flags to disable broadcast and receiving of tx gossip (#1194)

* add flags to disable broadcast and receiving of tx gossip respectively

* chore: auto version bump [bot]

---------

Co-authored-by: Morty <70688412+yiweichi@users.noreply.github.com>

cmd/geth/main.go

@@ -176,6 +176,8 @@ var (
 		utils.CircuitCapacityCheckWorkersFlag,
 		utils.RollupVerifyEnabledFlag,
 		utils.ShadowforkPeersFlag,
+		utils.TxGossipBroadcastDisabledFlag,
+		utils.TxGossipReceivingDisabledFlag,
 		utils.DASyncEnabledFlag,
 		utils.DABlockNativeAPIEndpointFlag,
 		utils.DABlobScanAPIEndpointFlag,

cmd/geth/usage.go

@@ -248,6 +248,8 @@ var AppHelpFlagGroups = []flags.FlagGroup{
 			utils.DARecoveryProduceBlocksFlag,
 			utils.CircuitCapacityCheckEnabledFlag,
 			utils.CircuitCapacityCheckWorkersFlag,
+			utils.TxGossipBroadcastDisabledFlag,
+			utils.TxGossipReceivingDisabledFlag,
 		},
 	},
 	{

cmd/utils/flags.go

@@ -893,6 +893,16 @@ var (
 		Usage: "peer ids of shadow fork peers",
 	}
 
+	// Tx gossip settings
+	TxGossipBroadcastDisabledFlag = cli.BoolFlag{
+		Name:  "txgossip.disablebroadcast",
+		Usage: "Disable gossip broadcast transactions to other peers",
+	}
+	TxGossipReceivingDisabledFlag = cli.BoolFlag{
+		Name:  "txgossip.disablereceiving",
+		Usage: "Disable gossip receiving transactions from other peers",
+	}
+
 	// DA syncing settings
 	DASyncEnabledFlag = cli.BoolFlag{
 		Name:  "da.sync",
@@ -1790,6 +1800,14 @@ func SetEthConfig(ctx *cli.Context, stack *node.Node, cfg *ethconfig.Config) {
 		cfg.ShadowForkPeerIDs = ctx.GlobalStringSlice(ShadowforkPeersFlag.Name)
 		log.Info("Shadow fork peers", "ids", cfg.ShadowForkPeerIDs)
 	}
+	if ctx.GlobalIsSet(TxGossipBroadcastDisabledFlag.Name) {
+		cfg.TxGossipBroadcastDisabled = ctx.GlobalBool(TxGossipBroadcastDisabledFlag.Name)
+		log.Info("Transaction gossip broadcast disabled", "disabled", cfg.TxGossipBroadcastDisabled)
+	}
+	if ctx.GlobalIsSet(TxGossipReceivingDisabledFlag.Name) {
+		cfg.TxGossipReceivingDisabled = ctx.GlobalBool(TxGossipReceivingDisabledFlag.Name)
+		log.Info("Transaction gossip receiving disabled", "disabled", cfg.TxGossipReceivingDisabled)
+	}
 
 	// Cap the cache allowance and tune the garbage collector
 	mem, err := gopsutil.VirtualMemory()

eth/backend.go

@@ -273,16 +273,18 @@ func New(stack *node.Node, config *ethconfig.Config, l1Client l1.Client) (*Ether
 		checkpoint = params.TrustedCheckpoints[genesisHash]
 	}
 	if eth.handler, err = newHandler(&handlerConfig{
-		Database:          chainDb,
+		Database:           chainDb,
-		Chain:             eth.blockchain,
+		Chain:              eth.blockchain,
-		TxPool:            eth.txPool,
+		TxPool:             eth.txPool,
-		Network:           config.NetworkId,
+		Network:            config.NetworkId,
-		Sync:              config.SyncMode,
+		Sync:               config.SyncMode,
-		BloomCache:        uint64(cacheLimit),
+		BloomCache:         uint64(cacheLimit),
-		EventMux:          eth.eventMux,
+		EventMux:           eth.eventMux,
-		Checkpoint:        checkpoint,
+		Checkpoint:         checkpoint,
-		Whitelist:         config.Whitelist,
+		Whitelist:          config.Whitelist,
-		ShadowForkPeerIDs: config.ShadowForkPeerIDs,
+		ShadowForkPeerIDs:  config.ShadowForkPeerIDs,
+		DisableTxBroadcast: config.TxGossipBroadcastDisabled,
+		DisableTxReceiving: config.TxGossipReceivingDisabled,
 	}); err != nil {
 		return nil, err
 	}

eth/ethconfig/config.go

@@ -230,6 +230,9 @@ type Config struct {
 
 	// DA syncer options
 	DA da_syncer.Config
+
+	TxGossipBroadcastDisabled bool
+	TxGossipReceivingDisabled bool
 }
 
 // CreateConsensusEngine creates a consensus engine for the given chain configuration.

eth/handler.go

@@ -93,6 +93,9 @@ type handlerConfig struct {
 	Checkpoint        *params.TrustedCheckpoint // Hard coded checkpoint for sync challenges
 	Whitelist         map[uint64]common.Hash    // Hard coded whitelist for sync challenged
 	ShadowForkPeerIDs []string                  // List of peer ids that take part in the shadow-fork
+
+	DisableTxBroadcast bool
+	DisableTxReceiving bool
 }
 
 type handler struct {
@@ -131,7 +134,9 @@ type handler struct {
 	wg        sync.WaitGroup
 	peerWG    sync.WaitGroup
 
-	shadowForkPeerIDs []string
+	shadowForkPeerIDs  []string
+	disableTxBroadcast bool
+	disableTxReceiving bool
 }
 
 // newHandler returns a handler for all Ethereum chain management protocol.
@@ -141,16 +146,18 @@ func newHandler(config *handlerConfig) (*handler, error) {
 		config.EventMux = new(event.TypeMux) // Nicety initialization for tests
 	}
 	h := &handler{
-		networkID:         config.Network,
+		networkID:          config.Network,
-		forkFilter:        forkid.NewFilter(config.Chain),
+		forkFilter:         forkid.NewFilter(config.Chain),
-		eventMux:          config.EventMux,
+		eventMux:           config.EventMux,
-		database:          config.Database,
+		database:           config.Database,
-		txpool:            config.TxPool,
+		txpool:             config.TxPool,
-		chain:             config.Chain,
+		chain:              config.Chain,
-		peers:             newPeerSet(),
+		peers:              newPeerSet(),
-		whitelist:         config.Whitelist,
+		whitelist:          config.Whitelist,
-		quitSync:          make(chan struct{}),
+		quitSync:           make(chan struct{}),
-		shadowForkPeerIDs: config.ShadowForkPeerIDs,
+		shadowForkPeerIDs:  config.ShadowForkPeerIDs,
+		disableTxBroadcast: config.DisableTxBroadcast,
+		disableTxReceiving: config.DisableTxReceiving,
 	}
 	if config.Sync == downloader.FullSync {
 		// The database seems empty as the current block is the genesis. Yet the fast
@@ -415,10 +422,12 @@ func (h *handler) Start(maxPeers int) {
 	h.maxPeers = maxPeers
 
 	// broadcast transactions
-	h.wg.Add(1)
+	if !h.disableTxBroadcast {
-	h.txsCh = make(chan core.NewTxsEvent, txChanSize)
+		h.wg.Add(1)
-	h.txsSub = h.txpool.SubscribeNewTxsEvent(h.txsCh)
+		h.txsCh = make(chan core.NewTxsEvent, txChanSize)
-	go h.txBroadcastLoop()
+		h.txsSub = h.txpool.SubscribeNewTxsEvent(h.txsCh)
+		go h.txBroadcastLoop()
+	}
 
 	// broadcast mined blocks
 	h.wg.Add(1)

eth/handler_eth.go

@@ -56,6 +56,9 @@ func (h *ethHandler) PeerInfo(id enode.ID) interface{} {
 // AcceptTxs retrieves whether transaction processing is enabled on the node
 // or if inbound transactions should simply be dropped.
 func (h *ethHandler) AcceptTxs() bool {
+	if h.disableTxReceiving {
+		return false
+	}
 	return atomic.LoadUint32(&h.acceptTxs) == 1
 }
 

params/version.go

@@ -24,7 +24,7 @@ import (
 const (
 	VersionMajor = 5         // Major version component of the current release
 	VersionMinor = 8         // Minor version component of the current release
-	VersionPatch = 56        // Patch version component of the current release
+	VersionPatch = 57        // Patch version component of the current release
 	VersionMeta  = "mainnet" // Version metadata to append to the version string
 )