network: non-participating nodes request TX gossip only if ForceFetchTransactions: true

components/mocks/mockParticipationRegistry.go

@@ -69,6 +69,11 @@ func (m *MockParticipationRegistry) GetStateProofForRound(id account.Participati
 	return account.StateProofRecordForRound{}, nil
 }
 
+// HasLiveKeys quickly tests to see if there is a valid participation key over some range of rounds
+func (m *MockParticipationRegistry) HasLiveKeys(from, to basics.Round) bool {
+	return false
+}
+
 // Register updates the EffectiveFirst and EffectiveLast fields. If there are multiple records for the account
 // then it is possible for multiple records to be updated.
 func (m *MockParticipationRegistry) Register(id account.ParticipationID, on basics.Round) error {

data/account/participationRegistry.go

@@ -250,6 +250,9 @@ type ParticipationRegistry interface {
 	// GetStateProofForRound fetches a record with stateproof secrets for a particular round.
 	GetStateProofForRound(id ParticipationID, round basics.Round) (StateProofRecordForRound, error)
 
+	// HasLiveKeys quickly tests to see if there is a valid participation key over some range of rounds
+	HasLiveKeys(from, to basics.Round) bool
+
 	// Register updates the EffectiveFirst and EffectiveLast fields. If there are multiple records for the account
 	// then it is possible for multiple records to be updated.
 	Register(id ParticipationID, on basics.Round) error
@@ -726,6 +729,18 @@ func (db *participationDB) GetAll() []ParticipationRecord {
 	return results
 }
 
+func (db *participationDB) HasLiveKeys(from, to basics.Round) bool {
+	db.mutex.RLock()
+	defer db.mutex.RUnlock()
+
+	for _, record := range db.cache {
+		if record.OverlapsInterval(from, to) {
+			return true
+		}
+	}
+	return false
+}
+
 // GetStateProofForRound returns the state proof data required to sign the compact certificate for this round
 func (db *participationDB) GetStateProofForRound(id ParticipationID, round basics.Round) (StateProofRecordForRound, error) {
 	partRecord, err := db.GetForRound(id, round)

data/accountManager.go

@@ -92,12 +92,7 @@ func (manager *AccountManager) HasLiveKeys(from, to basics.Round) bool {
 	manager.mu.Lock()
 	defer manager.mu.Unlock()
 
-	for _, part := range manager.registry.GetAll() {
-		if part.OverlapsInterval(from, to) {
-			return true
-		}
-	}
-	return false
+	return manager.registry.HasLiveKeys(from, to)
 }
 
 // AddParticipation adds a new account.Participation to be managed.

network/wsNetwork.go

@@ -301,6 +301,19 @@ func Propagate(msg IncomingMessage) OutgoingMessage {
 // Contains {genesisID} param to be handled by gorilla/mux
 const GossipNetworkPath = "/v1/{genesisID}/gossip"
 
+// NodeInfo helps the network get information about the node it is running on
+type NodeInfo interface {
+	// IsParticipating returns true if this node has stake and may vote on blocks or propose blocks.
+	IsParticipating() bool
+}
+
+type nopeNodeInfo struct {
+}
+
+func (nnni *nopeNodeInfo) IsParticipating() bool {
+	return false
+}
+
 // WebsocketNetwork implements GossipNode
 type WebsocketNetwork struct {
 	listener net.Listener
@@ -397,21 +410,34 @@ type WebsocketNetwork struct {
 	// to be sent to new peers.  This is filled in at network start,
 	// at which point messagesOfInterestEncoded is set to prevent
 	// further changes.
-	messagesOfInterestEnc     []byte
-	messagesOfInterestEncoded bool
+	messagesOfInterestEnc        []byte
+	messagesOfInterestEncoded    bool
+	messagesOfInterestGeneration uint32
 
 	// messagesOfInterestMu protects messagesOfInterest and ensures
 	// that messagesOfInterestEnc does not change once it is set during
 	// network start.
-	messagesOfInterestMu deadlock.Mutex
+	messagesOfInterestMu   deadlock.Mutex
+	messagesOfInterestCond *sync.Cond
 
 	// peersConnectivityCheckTicker is the timer for testing that all the connected peers
 	// are still transmitting or receiving information. The channel produced by this ticker
 	// is consumed by any of the messageHandlerThread(s). The ticker itself is created during
 	// Start(), and being shut down when Stop() is called.
 	peersConnectivityCheckTicker *time.Ticker
+
+	nodeInfo NodeInfo
+
+	// atomic {0:unknown, 1:yes, 2:no}
+	wantTXGossip uint32
 }
 
+const (
+	wantTXGossipUnk = 0
+	wantTXGossipYes = 1
+	wantTXGossipNo  = 2
+)
+
 type broadcastRequest struct {
 	tags        []Tag
 	data        [][]byte
@@ -661,6 +687,9 @@ func (wn *WebsocketNetwork) setup() {
 	if wn.config.DNSSecurityRelayAddrEnforced() {
 		preferredResolver = dnssec.MakeDefaultDnssecResolver(wn.config.FallbackDNSResolverAddress, wn.log)
 	}
+	if wn.nodeInfo == nil {
+		wn.nodeInfo = &nopeNodeInfo{}
+	}
 	maxIdleConnsPerHost := int(wn.config.ConnectionsRateLimitingCount)
 	wn.dialer = makeRateLimitingDialer(wn.phonebook, preferredResolver)
 	wn.transport = makeRateLimitingTransport(wn.phonebook, 10*time.Second, &wn.dialer, maxIdleConnsPerHost)
@@ -684,6 +713,9 @@ func (wn *WebsocketNetwork) setup() {
 	wn.server.MaxHeaderBytes = httpServerMaxHeaderBytes
 	wn.ctx, wn.ctxCancel = context.WithCancel(context.Background())
 	wn.relayMessages = wn.config.NetAddress != "" || wn.config.ForceRelayMessages
+	if wn.relayMessages || wn.config.ForceFetchTransactions {
+		wn.wantTXGossip = wantTXGossipYes
+	}
 	// roughly estimate the number of messages that could be seen at any given moment.
 	// For the late/redo/down committee, which happen in parallel, we need to allocate
 	// extra space there.
@@ -732,6 +764,8 @@ func (wn *WebsocketNetwork) setup() {
 		SupportedProtocolVersions = []string{wn.config.NetworkProtocolVersion}
 	}
 
+	wn.messagesOfInterestCond = sync.NewCond(&wn.messagesOfInterestMu)
+	wn.messagesOfInterestGeneration = 1 // something nonzero so that any new wsPeer needs updating
 	if wn.relayMessages {
 		wn.RegisterMessageInterest(protocol.CompactCertSigTag)
 	}
@@ -798,6 +832,9 @@ func (wn *WebsocketNetwork) Start() {
 		wn.wg.Add(1)
 		go wn.prioWeightRefresh()
 	}
+
+	go wn.postMessagesOfInterestThread()
+
 	wn.log.Infof("serving genesisID=%s on %#v with RandomID=%s", wn.GenesisID, wn.PublicAddress(), wn.RandomID)
 }
 
@@ -1123,18 +1160,28 @@ func (wn *WebsocketNetwork) ServeHTTP(response http.ResponseWriter, request *htt
 			InstanceName: trackedRequest.otherInstanceName,
 		})
 
-	// We are careful to encode this prior to starting the server to avoid needing 'messagesOfInterestMu' here.
-	if wn.messagesOfInterestEnc != nil {
-		err = peer.Unicast(wn.ctx, wn.messagesOfInterestEnc, protocol.MsgOfInterestTag)
-		if err != nil {
-			wn.log.Infof("ws send msgOfInterest: %v", err)
-		}
-	}
+	wn.maybeSendMessagesOfInterest(peer, nil)
 
 	peers.Set(float64(wn.NumPeers()), nil)
 	incomingPeers.Set(float64(wn.numIncomingPeers()), nil)
 }
 
+func (wn *WebsocketNetwork) maybeSendMessagesOfInterest(peer *wsPeer, messagesOfInterestEnc []byte) {
+	messagesOfInterestGeneration := atomic.LoadUint32(&wn.messagesOfInterestGeneration)
+	if peer.messagesOfInterestGeneration != messagesOfInterestGeneration {
+		if messagesOfInterestEnc == nil {
+			wn.messagesOfInterestMu.Lock()
+			messagesOfInterestEnc = wn.messagesOfInterestEnc
+			wn.messagesOfInterestMu.Unlock()
+		}
+		if messagesOfInterestEnc != nil {
+			peer.sendMessagesOfInterest(messagesOfInterestGeneration, messagesOfInterestEnc)
+		} else {
+			wn.log.Infof("msgOfInterest Enc=nil")
+		}
+	}
+}
+
 func (wn *WebsocketNetwork) messageHandlerThread(peersConnectivityCheckCh <-chan time.Time) {
 	defer wn.wg.Done()
 
@@ -1677,6 +1724,17 @@ func (wn *WebsocketNetwork) OnNetworkAdvance() {
 	wn.lastNetworkAdvanceMu.Lock()
 	defer wn.lastNetworkAdvanceMu.Unlock()
 	wn.lastNetworkAdvance = time.Now().UTC()
+	if !wn.relayMessages && !wn.config.ForceFetchTransactions {
+		// if we're not a relay, and not participating, we don't need txn pool
+		wantTXGossip := wn.nodeInfo.IsParticipating()
+		if wantTXGossip && (wn.wantTXGossip != wantTXGossipYes) {
+			wn.RegisterMessageInterest(protocol.TxnTag)
+			wn.wantTXGossip = wantTXGossipYes
+		} else if !wantTXGossip && (wn.wantTXGossip != wantTXGossipNo) {
+			wn.DeregisterMessageInterest(protocol.TxnTag)
+			wn.wantTXGossip = wantTXGossipNo
+		}
+	}
 }
 
 // sendPeerConnectionsTelemetryStatus sends a snapshot of the currently connected peers
@@ -2046,6 +2104,8 @@ func (wn *WebsocketNetwork) tryConnect(addr, gossipAddr string) {
 			Endpoint:     peer.GetAddress(),
 		})
 
+	wn.maybeSendMessagesOfInterest(peer, nil)
+
 	peers.Set(float64(wn.NumPeers()), nil)
 	outgoingPeers.Set(float64(wn.numOutgoingPeers()), nil)
 
@@ -2085,7 +2145,7 @@ func (wn *WebsocketNetwork) SetPeerData(peer Peer, key string, value interface{}
 }
 
 // NewWebsocketNetwork constructor for websockets based gossip network
-func NewWebsocketNetwork(log logging.Logger, config config.Local, phonebookAddresses []string, genesisID string, networkID protocol.NetworkID) (wn *WebsocketNetwork, err error) {
+func NewWebsocketNetwork(log logging.Logger, config config.Local, phonebookAddresses []string, genesisID string, networkID protocol.NetworkID, node NodeInfo) (wn *WebsocketNetwork, err error) {
 	phonebook := MakePhonebook(config.ConnectionsRateLimitingCount,
 		time.Duration(config.ConnectionsRateLimitingWindowSeconds)*time.Second)
 	phonebook.ReplacePeerList(phonebookAddresses, config.DNSBootstrapID, PhoneBookEntryRelayRole)
@@ -2095,6 +2155,7 @@ func NewWebsocketNetwork(log logging.Logger, config config.Local, phonebookAddre
 		phonebook: phonebook,
 		GenesisID: genesisID,
 		NetworkID: networkID,
+		nodeInfo:  node,
 	}
 
 	wn.setup()
@@ -2103,7 +2164,7 @@ func NewWebsocketNetwork(log logging.Logger, config config.Local, phonebookAddre
 
 // NewWebsocketGossipNode constructs a websocket network node and returns it as a GossipNode interface implementation
 func NewWebsocketGossipNode(log logging.Logger, config config.Local, phonebookAddresses []string, genesisID string, networkID protocol.NetworkID) (gn GossipNode, err error) {
-	return NewWebsocketNetwork(log, config, phonebookAddresses, genesisID, networkID)
+	return NewWebsocketNetwork(log, config, phonebookAddresses, genesisID, networkID, nil)
 }
 
 // SetPrioScheme specifies the network priority scheme for a network node
@@ -2254,21 +2315,56 @@ func (wn *WebsocketNetwork) RegisterMessageInterest(t protocol.Tag) error {
 	wn.messagesOfInterestMu.Lock()
 	defer wn.messagesOfInterestMu.Unlock()
 
-	if wn.messagesOfInterestEncoded {
-		return fmt.Errorf("network already started")
+	if wn.messagesOfInterest == nil {
+		wn.messagesOfInterest = make(map[protocol.Tag]bool)
+		for tag, flag := range defaultSendMessageTags {
+			wn.messagesOfInterest[tag] = flag
+		}
 	}
 
+	wn.messagesOfInterest[t] = true
+	wn.updateMessagesOfInterestEnc()
+	return nil
+}
+
+// DeregisterMessageInterest will tell peers to no longer send us traffic with a protocol Tag
+func (wn *WebsocketNetwork) DeregisterMessageInterest(t protocol.Tag) error {
+	wn.messagesOfInterestMu.Lock()
+	defer wn.messagesOfInterestMu.Unlock()
+
 	if wn.messagesOfInterest == nil {
 		wn.messagesOfInterest = make(map[protocol.Tag]bool)
 		for tag, flag := range defaultSendMessageTags {
 			wn.messagesOfInterest[tag] = flag
 		}
 	}
 
-	wn.messagesOfInterest[t] = true
+	delete(wn.messagesOfInterest, t)
+	wn.updateMessagesOfInterestEnc()
 	return nil
 }
 
+func (wn *WebsocketNetwork) updateMessagesOfInterestEnc() {
+	// must run inside wn.messagesOfInterestMu.Lock
+	wn.messagesOfInterestEnc = MarshallMessageOfInterestMap(wn.messagesOfInterest)
+	wn.messagesOfInterestEncoded = true
+	atomic.AddUint32(&wn.messagesOfInterestGeneration, 1)
+	wn.messagesOfInterestCond.Broadcast()
+}
+
+func (wn *WebsocketNetwork) postMessagesOfInterestThread() {
+	var peers []*wsPeer
+	wn.messagesOfInterestMu.Lock()
+	defer wn.messagesOfInterestMu.Unlock()
+	for {
+		wn.messagesOfInterestCond.Wait()
+		peers, _ = wn.peerSnapshot(peers)
+		for _, peer := range peers {
+			wn.maybeSendMessagesOfInterest(peer, wn.messagesOfInterestEnc)
+		}
+	}
+}
+
 // SubstituteGenesisID substitutes the "{genesisID}" with their network-specific genesisID.
 func (wn *WebsocketNetwork) SubstituteGenesisID(rawURL string) string {
 	return strings.Replace(rawURL, "{genesisID}", wn.GenesisID, -1)