feat: add flags to enable broadcast blocks and transactions to all peers (#1219)

* feat: add flag to enable broadcast blocks and transactions to all peers

* fix: typo

* feat: add broadcast to all cap

* chore: auto version bump [bot]

* fix: ci

---------

Co-authored-by: Jonas Theis <4181434+jonastheis@users.noreply.github.com>

cmd/geth/main.go

@@ -176,9 +176,11 @@ var (
 		utils.CircuitCapacityCheckWorkersFlag,
 		utils.RollupVerifyEnabledFlag,
 		utils.ShadowforkPeersFlag,
-		utils.TxGossipBroadcastDisabledFlag,
+		utils.GossipTxBroadcastDisabledFlag,
-		utils.TxGossipReceivingDisabledFlag,
+		utils.GossipTxReceivingDisabledFlag,
-		utils.TxGossipSequencerHTTPFlag,
+		utils.GossipSequencerHTTPFlag,
+		utils.GossipBroadcastToAllEnabledFlag,
+		utils.GossipBroadcastToAllCapFlag,
 		utils.DASyncEnabledFlag,
 		utils.DAMissingHeaderFieldsBaseURLFlag,
 		utils.DABlockNativeAPIEndpointFlag,

cmd/geth/usage.go

@@ -250,9 +250,11 @@ var AppHelpFlagGroups = []flags.FlagGroup{
 			utils.DARecoveryProduceBlocksFlag,
 			utils.CircuitCapacityCheckEnabledFlag,
 			utils.CircuitCapacityCheckWorkersFlag,
-			utils.TxGossipBroadcastDisabledFlag,
+			utils.GossipTxBroadcastDisabledFlag,
-			utils.TxGossipReceivingDisabledFlag,
+			utils.GossipTxReceivingDisabledFlag,
-			utils.TxGossipSequencerHTTPFlag,
+			utils.GossipSequencerHTTPFlag,
+			utils.GossipBroadcastToAllEnabledFlag,
+			utils.GossipBroadcastToAllCapFlag,
 		},
 	},
 	{

cmd/utils/flags.go

@@ -893,19 +893,28 @@ var (
 		Usage: "peer ids of shadow fork peers",
 	}
 
-	// Tx gossip settings
+	// Gossip settings
-	TxGossipBroadcastDisabledFlag = cli.BoolFlag{
+	GossipTxBroadcastDisabledFlag = cli.BoolFlag{
-		Name:  "gossip.disablebroadcast",
+		Name:  "gossip.disabletxbroadcast",
 		Usage: "Disable gossip broadcast transactions to other peers",
 	}
-	TxGossipReceivingDisabledFlag = cli.BoolFlag{
+	GossipTxReceivingDisabledFlag = cli.BoolFlag{
-		Name:  "gossip.disablereceiving",
+		Name:  "gossip.disabletxreceiving",
 		Usage: "Disable gossip receiving transactions from other peers",
 	}
-	TxGossipSequencerHTTPFlag = &cli.StringFlag{
+	GossipSequencerHTTPFlag = &cli.StringFlag{
 		Name:  "gossip.sequencerhttp",
 		Usage: "Sequencer mempool HTTP endpoint",
 	}
+	GossipBroadcastToAllEnabledFlag = cli.BoolFlag{
+		Name:  "gossip.enablebroadcasttoall",
+		Usage: "Enable gossip broadcast blocks and transactions to all peers",
+	}
+	GossipBroadcastToAllCapFlag = cli.IntFlag{
+		Name:  "gossip.broadcasttoallcap",
+		Usage: "Maximum number of peers for broadcasting blocks and transactions (effective only when gossip.enablebroadcasttoall is enabled)",
+		Value: 30,
+	}
 
 	// DA syncing settings
 	DASyncEnabledFlag = cli.BoolFlag{
@@ -1819,17 +1828,22 @@ 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) {
+	if ctx.GlobalIsSet(GossipTxBroadcastDisabledFlag.Name) {
-		cfg.TxGossipBroadcastDisabled = ctx.GlobalBool(TxGossipBroadcastDisabledFlag.Name)
+		cfg.GossipTxBroadcastDisabled = ctx.GlobalBool(GossipTxBroadcastDisabledFlag.Name)
-		log.Info("Transaction gossip broadcast disabled", "disabled", cfg.TxGossipBroadcastDisabled)
+		log.Info("Gossip transaction broadcast disabled", "disabled", cfg.GossipTxBroadcastDisabled)
 	}
-	if ctx.GlobalIsSet(TxGossipReceivingDisabledFlag.Name) {
+	if ctx.GlobalIsSet(GossipTxReceivingDisabledFlag.Name) {
-		cfg.TxGossipReceivingDisabled = ctx.GlobalBool(TxGossipReceivingDisabledFlag.Name)
+		cfg.GossipTxReceivingDisabled = ctx.GlobalBool(GossipTxReceivingDisabledFlag.Name)
-		log.Info("Transaction gossip receiving disabled", "disabled", cfg.TxGossipReceivingDisabled)
+		log.Info("Gossip transaction receiving disabled", "disabled", cfg.GossipTxReceivingDisabled)
+	}
+	if ctx.GlobalIsSet(GossipBroadcastToAllEnabledFlag.Name) {
+		cfg.GossipBroadcastToAllEnabled = ctx.GlobalBool(GossipBroadcastToAllEnabledFlag.Name)
+		cfg.GossipBroadcastToAllCap = ctx.GlobalInt(GossipBroadcastToAllCapFlag.Name)
+		log.Info("Gossip broadcast to all enabled", "enabled", cfg.GossipBroadcastToAllEnabled, "cap", cfg.GossipBroadcastToAllCap)
 	}
 	// Only configure sequencer http flag if we're running in verifier mode i.e. --mine is disabled.
-	if ctx.IsSet(TxGossipSequencerHTTPFlag.Name) && !ctx.IsSet(MiningEnabledFlag.Name) {
+	if ctx.IsSet(GossipSequencerHTTPFlag.Name) && !ctx.IsSet(MiningEnabledFlag.Name) {
-		cfg.TxGossipSequencerHTTP = ctx.String(TxGossipSequencerHTTPFlag.Name)
+		cfg.GossipSequencerHTTP = ctx.String(GossipSequencerHTTPFlag.Name)
 	}
 
 	// Cap the cache allowance and tune the garbage collector

eth/backend.go

@@ -285,18 +285,20 @@ 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,
+		DisableTxBroadcast:   config.GossipTxBroadcastDisabled,
-		DisableTxReceiving: config.TxGossipReceivingDisabled,
+		DisableTxReceiving:   config.GossipTxReceivingDisabled,
+		EnableBroadcastToAll: config.GossipBroadcastToAllEnabled,
+		BroadcastToAllCap:    config.GossipBroadcastToAllCap,
 	}); err != nil {
 		return nil, err
 	}
@@ -306,7 +308,7 @@ func New(stack *node.Node, config *ethconfig.Config, l1Client l1.Client) (*Ether
 	// Some of the extraData is used with Clique consensus (before EuclidV2). After EuclidV2 we use SystemContract consensus where this is overridden when creating a block.
 	eth.miner.SetExtra(makeExtraData(config.Miner.ExtraData))
 
-	eth.APIBackend = &EthAPIBackend{stack.Config().ExtRPCEnabled(), stack.Config().AllowUnprotectedTxs, config.TxGossipReceivingDisabled, eth, nil}
+	eth.APIBackend = &EthAPIBackend{stack.Config().ExtRPCEnabled(), stack.Config().AllowUnprotectedTxs, config.GossipTxReceivingDisabled, eth, nil}
 	if eth.APIBackend.allowUnprotectedTxs {
 		log.Info("Unprotected transactions allowed")
 	}
@@ -317,9 +319,9 @@ func New(stack *node.Node, config *ethconfig.Config, l1Client l1.Client) (*Ether
 	gpoParams.DefaultBasePrice = new(big.Int).SetUint64(config.TxPool.PriceLimit)
 	eth.APIBackend.gpo = gasprice.NewOracle(eth.APIBackend, gpoParams)
 
-	if config.TxGossipSequencerHTTP != "" {
+	if config.GossipSequencerHTTP != "" {
 		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
-		client, err := rpc.DialContext(ctx, config.TxGossipSequencerHTTP)
+		client, err := rpc.DialContext(ctx, config.GossipSequencerHTTP)
 		cancel()
 		if err != nil {
 			return nil, fmt.Errorf("cannot initialize rollup sequencer client: %w", err)

eth/ethconfig/config.go

@@ -231,9 +231,11 @@ type Config struct {
 	// DA syncer options
 	DA da_syncer.Config
 
-	TxGossipBroadcastDisabled bool
+	GossipTxBroadcastDisabled   bool
-	TxGossipReceivingDisabled bool
+	GossipTxReceivingDisabled   bool
-	TxGossipSequencerHTTP     string
+	GossipSequencerHTTP         string
+	GossipBroadcastToAllEnabled bool
+	GossipBroadcastToAllCap     int
 }
 
 // CreateConsensusEngine creates a consensus engine for the given chain configuration.

eth/handler.go

@@ -94,8 +94,11 @@ type handlerConfig struct {
 	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
+	// Gossip configs
-	DisableTxReceiving bool
+	DisableTxBroadcast   bool
+	DisableTxReceiving   bool
+	EnableBroadcastToAll bool
+	BroadcastToAllCap    int
 }
 
 type handler struct {
@@ -134,9 +137,11 @@ type handler struct {
 	wg        sync.WaitGroup
 	peerWG    sync.WaitGroup
 
-	shadowForkPeerIDs  []string
+	shadowForkPeerIDs    []string
-	disableTxBroadcast bool
+	disableTxBroadcast   bool
-	disableTxReceiving bool
+	disableTxReceiving   bool
+	enableBroadcastToAll bool
+	broadcastToAllCap    int
 }
 
 // newHandler returns a handler for all Ethereum chain management protocol.
@@ -146,18 +151,20 @@ 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,
+		disableTxBroadcast:   config.DisableTxBroadcast,
-		disableTxReceiving: config.DisableTxReceiving,
+		disableTxReceiving:   config.DisableTxReceiving,
+		enableBroadcastToAll: config.EnableBroadcastToAll,
+		broadcastToAllCap:    config.BroadcastToAllCap,
 	}
 	if config.Sync == downloader.FullSync {
 		// The database seems empty as the current block is the genesis. Yet the fast
@@ -477,7 +484,12 @@ func (h *handler) BroadcastBlock(block *types.Block, propagate bool) {
 			return
 		}
 		// Send the block to a subset of our peers
-		transfer := peers[:int(math.Sqrt(float64(len(peers))))]
+		numDirect := int(math.Sqrt(float64(len(peers))))
+		// If enableBroadcastToAll is true, broadcast blocks directly to all peers (capped at broadcastToAllCap).
+		if h.enableBroadcastToAll {
+			numDirect = min(h.broadcastToAllCap, len(peers))
+		}
+		transfer := peers[:numDirect]
 		for _, peer := range transfer {
 			peer.AsyncSendNewBlock(block, td)
 		}
@@ -518,6 +530,10 @@ func (h *handler) BroadcastTransactions(txs types.Transactions) {
 		peers := onlyShadowForkPeers(h.shadowForkPeerIDs, h.peers.peersWithoutTransaction(tx.Hash()))
 		// Send the tx unconditionally to a subset of our peers
 		numDirect := int(math.Sqrt(float64(len(peers))))
+		// If enableBroadcastToAll is true, broadcast transactions directly to all peers (capped at broadcastToAllCap).
+		if h.enableBroadcastToAll {
+			numDirect = min(h.broadcastToAllCap, len(peers))
+		}
 		for _, peer := range peers[:numDirect] {
 			txset[peer] = append(txset[peer], tx.Hash())
 		}

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 = 67        // Patch version component of the current release
+	VersionPatch = 68        // Patch version component of the current release
 	VersionMeta  = "mainnet" // Version metadata to append to the version string
 )