param: txs announced only (#1467)

docs/cli/example_config.toml

@@ -42,6 +42,7 @@ devfakeauthor = false           # Run miner without validator set authorization
   nodekey = ""                # P2P node key file
   nodekeyhex = ""             # P2P node key as hex
   txarrivalwait = "500ms"     # Maximum duration to wait before requesting an announced transaction
+  txannouncementonly = false  # Whether to only announce transactions to peers
   [p2p.discovery]
     v4disc = true       # Enables the V4 discovery mechanism
     v5disc = false      # Enables the experimental RLPx V5 (Topic Discovery) mechanism

docs/cli/server.md

@@ -242,6 +242,8 @@ The ```bor server``` command runs the Bor client.
 
 - ```txarrivalwait```: Maximum duration to wait for a transaction before explicitly requesting it (default: 500ms)
 
+- ```txannouncementonly```: Whether to only announce transactions to peers (default: false)
+
 - ```v4disc```: Enables the V4 discovery mechanism (default: true)
 
 - ```v5disc```: Enables the experimental RLPx V5 (Topic Discovery) mechanism (default: false)

eth/backend.go

@@ -306,6 +306,7 @@ func New(stack *node.Node, config *ethconfig.Config) (*Ethereum, error) {
 		EthAPI:              blockChainAPI,
 		checker:             checker,
 		enableBlockTracking: eth.config.EnableBlockTracking,
+		txAnnouncementOnly:  eth.p2pServer.TxAnnouncementOnly,
 	}); err != nil {
 		return nil, err
 	}

eth/handler.go

@@ -99,6 +99,7 @@ type handlerConfig struct {
 	RequiredBlocks      map[uint64]common.Hash // Hard coded map of required block hashes for sync challenges
 	EthAPI              *ethapi.BlockChainAPI  // EthAPI to interact
 	enableBlockTracking bool                   // Whether to log information collected while tracking block lifecycle
+	txAnnouncementOnly  bool                   // Whether to only announce txs to peers
 }
 
 type handler struct {
@@ -129,6 +130,7 @@ type handler struct {
 	requiredBlocks map[uint64]common.Hash
 
 	enableBlockTracking bool
+	txAnnouncementOnly  bool
 
 	// channels for fetcher, syncer, txsyncLoop
 	quitSync chan struct{}
@@ -159,6 +161,7 @@ func newHandler(config *handlerConfig) (*handler, error) {
 		ethAPI:              config.EthAPI,
 		requiredBlocks:      config.RequiredBlocks,
 		enableBlockTracking: config.enableBlockTracking,
+		txAnnouncementOnly:  config.txAnnouncementOnly,
 		quitSync:            make(chan struct{}),
 		handlerDoneCh:       make(chan struct{}),
 		handlerStartCh:      make(chan struct{}),
@@ -633,7 +636,7 @@ func (h *handler) BroadcastTransactions(txs types.Transactions) {
 		case tx.Size() > txMaxBroadcastSize:
 			largeTxs++
 		default:
-			maybeDirect = true
+			maybeDirect = !h.txAnnouncementOnly
 		}
 		// Send the transaction (if it's small enough) directly to a subset of
 		// the peers that have not received it yet, ensuring that the flow of

eth/handler_eth_test.go

@@ -454,6 +454,102 @@ func testTransactionPropagation(t *testing.T, protocol uint) {
 	}
 }
 
+// This test checks that transactions are only announced when txannouncementonly is enabled
+func TestSendTransactionAnnouncementsOnly67(t *testing.T) {
+	testSendTransactionAnnouncementsOnly(t, eth.ETH67)
+}
+func TestSendTransactionAnnouncementsOnly68(t *testing.T) {
+	testSendTransactionAnnouncementsOnly(t, eth.ETH68)
+}
+
+func testSendTransactionAnnouncementsOnly(t *testing.T, protocol uint) {
+	t.Parallel()
+
+	// Create a source handler that has txannouncementonly enabled
+	source := newTestHandler()
+	source.handler.txAnnouncementOnly = true
+	defer source.close()
+
+	sink := newTestHandler()
+	defer sink.close()
+
+	sourcePipe, sinkPipe := p2p.MsgPipe()
+	defer sourcePipe.Close()
+	defer sinkPipe.Close()
+
+	sourcePeer := eth.NewPeer(protocol, p2p.NewPeerPipe(enode.ID{1}, "", nil, sourcePipe), sourcePipe, source.txpool)
+	sinkPeer := eth.NewPeer(protocol, p2p.NewPeerPipe(enode.ID{2}, "", nil, sinkPipe), sinkPipe, sink.txpool)
+	defer sourcePeer.Close()
+	defer sinkPeer.Close()
+
+	go source.handler.runEthPeer(sourcePeer, func(peer *eth.Peer) error {
+		return eth.Handle((*ethHandler)(source.handler), peer)
+	})
+
+	// Run the handshake locally
+	var (
+		genesis = source.chain.Genesis()
+		head    = source.chain.CurrentBlock()
+		td      = source.chain.GetTd(head.Hash(), head.Number.Uint64())
+	)
+	if err := sinkPeer.Handshake(1, td, head.Hash(), genesis.Hash(), forkid.NewIDWithChain(source.chain), forkid.NewFilter(source.chain)); err != nil {
+		t.Fatalf("failed to run protocol handshake: %v", err)
+	}
+
+	// Subscribe to all inbound network events on the sink peer
+	backend := new(testEthHandler)
+
+	anns := make(chan []common.Hash)
+	annSub := backend.txAnnounces.Subscribe(anns)
+	defer annSub.Unsubscribe()
+
+	bcasts := make(chan []*types.Transaction)
+	bcastSub := backend.txBroadcasts.Subscribe(bcasts)
+	defer bcastSub.Unsubscribe()
+
+	go eth.Handle(backend, sinkPeer)
+
+	// Fill the source pool with transactions and wait for them at the sink
+	txs := make([]*types.Transaction, 1024)
+	for nonce := range txs {
+		tx := types.NewTransaction(uint64(nonce), common.Address{}, big.NewInt(0), 100000, big.NewInt(0), nil)
+		tx, _ = types.SignTx(tx, types.HomesteadSigner{}, testKey)
+		txs[nonce] = tx
+	}
+	source.txpool.Add(txs, false, false)
+
+	// Make sure we get all the transactions as announcements
+	seen := make(map[common.Hash]struct{})
+	timeout := false
+	for len(seen) < len(txs) && !timeout {
+		switch protocol {
+		case 67, 68:
+			select {
+			case hashes := <-anns:
+				for _, hash := range hashes {
+					if _, ok := seen[hash]; ok {
+						t.Errorf("duplicate transaction announced: %x", hash)
+					}
+					seen[hash] = struct{}{}
+				}
+			case <-bcasts:
+				t.Errorf("received tx broadcast when txannouncementonly is true")
+			case <-time.After(5 * time.Second):
+				t.Errorf("transaction propagation timed out")
+				timeout = true
+			}
+
+		default:
+			panic("unsupported protocol, please extend test")
+		}
+	}
+	for _, tx := range txs {
+		if _, ok := seen[tx.Hash()]; !ok {
+			t.Errorf("missing transaction: %x", tx.Hash())
+		}
+	}
+}
+
 // Tests that blocks are broadcast to a sqrt number of peers only.
 func TestBroadcastBlock1Peer(t *testing.T)    { testBroadcastBlock(t, 1, 1) }
 func TestBroadcastBlock2Peers(t *testing.T)   { testBroadcastBlock(t, 2, 1) }

internal/cli/server/config.go

@@ -222,6 +222,9 @@ type P2PConfig struct {
 	// an announced transaction to arrive before explicitly requesting it
 	TxArrivalWait    time.Duration `hcl:"-,optional" toml:"-"`
 	TxArrivalWaitRaw string        `hcl:"txarrivalwait,optional" toml:"txarrivalwait,optional"`
+
+	// TxAnnouncementOnly is used to only announce transactions to peers
+	TxAnnouncementOnly bool `hcl:"txannouncementonly,optional" toml:"txannouncementonly,optional"`
 }
 
 type P2PDiscovery struct {
@@ -633,6 +636,7 @@ func DefaultConfig() *Config {
 			NAT:                "any",
 			NetRestrict:        "",
 			TxArrivalWait:      500 * time.Millisecond,
+			TxAnnouncementOnly: false,
 			Discovery: &P2PDiscovery{
 				DiscoveryV4:  true,
 				V5Enabled:    false,
@@ -1349,6 +1353,7 @@ func (c *Config) buildNode() (*node.Config, error) {
 			DiscoveryV4:        c.P2P.Discovery.DiscoveryV4,
 			DiscoveryV5:        c.P2P.Discovery.V5Enabled,
 			TxArrivalWait:      c.P2P.TxArrivalWait,
+			TxAnnouncementOnly: c.P2P.TxAnnouncementOnly,
 		},
 		HTTPModules:         c.JsonRPC.Http.API,
 		HTTPCors:            c.JsonRPC.Http.Cors,

internal/cli/server/flags.go

@@ -812,6 +812,13 @@ func (c *Command) Flags(config *Config) *flagset.Flagset {
 		Default: c.cliConfig.P2P.TxArrivalWait,
 		Group:   "P2P",
 	})
+	f.BoolFlag(&flagset.BoolFlag{
+		Name:    "txannouncementonly",
+		Usage:   "Whether to only announce transactions to peers",
+		Value:   &c.cliConfig.P2P.TxAnnouncementOnly,
+		Default: c.cliConfig.P2P.TxAnnouncementOnly,
+		Group:   "P2P",
+	})
 	f.SliceStringFlag(&flagset.SliceStringFlag{
 		Name:    "discovery.dns",
 		Usage:   "Comma separated list of enrtree:// URLs which will be queried for nodes to connect to",

internal/cli/server/testdata/default.toml

@@ -36,6 +36,7 @@ devfakeauthor = false
   nodekey = ""
   nodekeyhex = ""
   txarrivalwait = "500ms"
+  txannouncementonly = false
   [p2p.discovery]
     v4disc = true
     v5disc = false

p2p/server.go

@@ -172,6 +172,9 @@ type Config struct {
 	// TxArrivalWait is the duration (ms) that the node will wait after seeing
 	// an announced transaction before explicitly requesting it
 	TxArrivalWait time.Duration
+
+	// TxAnnouncementOnly is used to only announce transactions to peers
+	TxAnnouncementOnly bool
 }
 
 // Server manages all peer connections.

packaging/templates/mainnet-v1/archive/config.toml

@@ -35,6 +35,7 @@ gcmode = "archive"
     # nodekey = ""
     # nodekeyhex = ""
     # txarrivalwait = "500ms"
+    # txannouncementonly = false
     # [p2p.discovery]
         # v4disc = true
         # v5disc = false

packaging/templates/mainnet-v1/sentry/sentry/bor/config.toml

@@ -34,6 +34,7 @@ syncmode = "full"
     # nodekey = ""
     # nodekeyhex = ""
     # txarrivalwait = "500ms"
+    # txannouncementonly = false
     [p2p.discovery]
         # v4disc = true
         # v5disc = false

packaging/templates/mainnet-v1/sentry/sentry/bor/pbss_config.toml

@@ -36,6 +36,7 @@ syncmode = "full"
     # nodekey = ""
     # nodekeyhex = ""
     # txarrivalwait = "500ms"
+    # txannouncementonly = false
     [p2p.discovery]
         # v4disc = true
         # v5disc = false

packaging/templates/mainnet-v1/sentry/validator/bor/config.toml

@@ -36,6 +36,7 @@ syncmode = "full"
     # nodekey = ""
     # nodekeyhex = ""
     # txarrivalwait = "500ms"
+    # txannouncementonly = false
     [p2p.discovery]
         # v4disc = true
         # v5disc = false

packaging/templates/mainnet-v1/sentry/validator/bor/pbss_config.toml

@@ -38,6 +38,7 @@ syncmode = "full"
     # nodekey = ""
     # nodekeyhex = ""
     # txarrivalwait = "500ms"
+    # txannouncementonly = false
     [p2p.discovery]
         # v4disc = true
         # v5disc = false

packaging/templates/mainnet-v1/without-sentry/bor/config.toml

@@ -36,6 +36,7 @@ syncmode = "full"
     # nodekey = ""
     # nodekeyhex = ""
     # txarrivalwait = "500ms"
+    # txannouncementonly = false
     [p2p.discovery]
         # v4disc = true
         # v5disc = false

packaging/templates/mainnet-v1/without-sentry/bor/pbss_config.toml

@@ -37,6 +37,7 @@ syncmode = "full"
     # nodekey = ""
     # nodekeyhex = ""
     # txarrivalwait = "500ms"
+    # txannouncementonly = false
     [p2p.discovery]
         # v4disc = true
         # v5disc = false

packaging/templates/testnet-amoy/archive/config.toml

@@ -34,6 +34,7 @@ gcmode = "archive"
     # nodekey = ""
     # nodekeyhex = ""
     # txarrivalwait = "500ms"
+    # txannouncementonly = false
     [p2p.discovery]
         # v5disc = false
         # bootnodesv4 = []

packaging/templates/testnet-amoy/sentry/sentry/bor/config.toml

@@ -33,6 +33,7 @@ syncmode = "full"
     # nodekey = ""
     # nodekeyhex = ""
     # txarrivalwait = "500ms"
+    # txannouncementonly = false
     [p2p.discovery]
         # v5disc = false
         # bootnodesv4 = []

packaging/templates/testnet-amoy/sentry/sentry/bor/pbss_config.toml

@@ -34,6 +34,7 @@ syncmode = "full"
     # nodekey = ""
     # nodekeyhex = ""
     # txarrivalwait = "500ms"
+    # txannouncementonly = false
     [p2p.discovery]
         # v5disc = false
         # bootnodesv4 = []

packaging/templates/testnet-amoy/sentry/validator/bor/config.toml

@@ -35,6 +35,7 @@ syncmode = "full"
     # nodekey = ""
     # nodekeyhex = ""
     # txarrivalwait = "500ms"
+    # txannouncementonly = false
     [p2p.discovery]
         # v5disc = false
         # bootnodesv4 = []

packaging/templates/testnet-amoy/sentry/validator/bor/pbss_config.toml

@@ -36,6 +36,7 @@ syncmode = "full"
     # nodekey = ""
     # nodekeyhex = ""
     # txarrivalwait = "500ms"
+    # txannouncementonly = false
     [p2p.discovery]
         # v5disc = false
         # bootnodesv4 = []

packaging/templates/testnet-amoy/without-sentry/bor/config.toml

@@ -35,6 +35,7 @@ syncmode = "full"
     # nodekey = ""
     # nodekeyhex = ""
     # txarrivalwait = "500ms"
+    # txannouncementonly = false
     [p2p.discovery]
         # v5disc = false
         # bootnodesv4 = []

packaging/templates/testnet-amoy/without-sentry/bor/pbss_config.toml

@@ -36,6 +36,7 @@ syncmode = "full"
     # nodekey = ""
     # nodekeyhex = ""
     # txarrivalwait = "500ms"
+    # txannouncementonly = false
     [p2p.discovery]
         # v5disc = false
         # bootnodesv5 = []

scripts/getconfig.go

@@ -172,6 +172,7 @@ var nameTagMap = map[string]string{
 	"maxpeers":                "maxpeers",
 	"maxpendpeers":            "maxpendpeers",
 	"txarrivalwait":           "txarrivalwait",
+	"txannouncementonly":      "txannouncementonly",
 	"nat":                     "nat",
 	"nodiscover":              "nodiscover",
 	"v5disc":                  "v5disc",