Merge pull request #2135 from cfromknecht/isolate-gossip-rate-limiting

[discovery] Isolate gossip rate limiting

discovery/gossiper.go

@@ -456,12 +456,67 @@ func (d *AuthenticatedGossiper) Stop() {
 func (d *AuthenticatedGossiper) ProcessRemoteAnnouncement(msg lnwire.Message,
 	peer lnpeer.Peer) chan error {
 
+	errChan := make(chan error, 1)
+
+	// For messages in the known set of channel series queries, we'll
+	// dispatch the message directly to the gossipSyncer, and skip the main
+	// processing loop.
+	switch m := msg.(type) {
+	case *lnwire.QueryShortChanIDs,
+		*lnwire.QueryChannelRange,
+		*lnwire.ReplyChannelRange,
+		*lnwire.ReplyShortChanIDsEnd:
+
+		syncer, err := d.findGossipSyncer(peer.IdentityKey())
+		if err != nil {
+			log.Warnf("Unable to find gossip syncer for "+
+				"peer=%x: %v", peer.PubKey(), err)
+
+			errChan <- err
+			return errChan
+		}
+
+		// If we've found the message target, then we'll dispatch the
+		// message directly to it.
+		syncer.ProcessQueryMsg(m, peer.QuitSignal())
+
+		errChan <- nil
+		return errChan
+
+	// If a peer is updating its current update horizon, then we'll dispatch
+	// that directly to the proper gossipSyncer.
+	case *lnwire.GossipTimestampRange:
+		syncer, err := d.findGossipSyncer(peer.IdentityKey())
+		if err != nil {
+			log.Warnf("Unable to find gossip syncer for "+
+				"peer=%x: %v", peer.PubKey(), err)
+
+			errChan <- err
+			return errChan
+		}
+
+		// If we've found the message target, then we'll dispatch the
+		// message directly to it.
+		err = syncer.ApplyGossipFilter(m)
+		if err != nil {
+			log.Warnf("unable to apply gossip "+
+				"filter for peer=%x: %v",
+				peer.PubKey(), err)
+
+			errChan <- err
+			return errChan
+		}
+
+		errChan <- nil
+		return errChan
+	}
+
 	nMsg := &networkMsg{
 		msg:      msg,
 		isRemote: true,
 		peer:     peer,
 		source:   peer.IdentityKey(),
-		err:      make(chan error, 1),
+		err:      errChan,
 	}
 
 	select {
@@ -965,7 +1020,7 @@ func (d *AuthenticatedGossiper) networkHandler() {
 			policyUpdate.errResp <- nil
 
 		case announcement := <-d.networkMsgs:
-			switch msg := announcement.msg.(type) {
+			switch announcement.msg.(type) {
 			// Channel announcement signatures are amongst the only
 			// messages that we'll process serially.
 			case *lnwire.AnnounceSignatures:
@@ -978,51 +1033,6 @@ func (d *AuthenticatedGossiper) networkHandler() {
 					)
 				}
 				continue
-
-			// If a peer is updating its current update horizon,
-			// then we'll dispatch that directly to the proper
-			// gossipSyncer.
-			case *lnwire.GossipTimestampRange:
-				syncer, err := d.findGossipSyncer(
-					announcement.source,
-				)
-				if err != nil {
-					log.Warnf("Unable to find gossip "+
-						"syncer for peer=%x: %v",
-						announcement.peer.PubKey(), err)
-					continue
-				}
-
-				// If we've found the message target, then
-				// we'll dispatch the message directly to it.
-				err = syncer.ApplyGossipFilter(msg)
-				if err != nil {
-					log.Warnf("unable to apply gossip "+
-						"filter for peer=%x: %v",
-						announcement.peer.PubKey(), err)
-				}
-				continue
-
-			// For messages in the known set of channel series
-			// queries, we'll dispatch the message directly to the
-			// peer, and skip the main processing loop.
-			case *lnwire.QueryShortChanIDs,
-				*lnwire.QueryChannelRange,
-				*lnwire.ReplyChannelRange,
-				*lnwire.ReplyShortChanIDsEnd:
-
-				syncer, err := d.findGossipSyncer(
-					announcement.source,
-				)
-				if err != nil {
-					log.Warnf("Unable to find gossip "+
-						"syncer for peer=%x: %v",
-						announcement.source, err)
-					continue
-				}
-
-				syncer.ProcessQueryMsg(announcement.msg)
-				continue
 			}
 
 			// If this message was recently rejected, then we won't

discovery/syncer.go

@@ -982,12 +982,11 @@ func (g *gossipSyncer) FilterGossipMsgs(msgs ...msgWithSenders) {
 
 // ProcessQueryMsg is used by outside callers to pass new channel time series
 // queries to the internal processing goroutine.
-func (g *gossipSyncer) ProcessQueryMsg(msg lnwire.Message) {
+func (g *gossipSyncer) ProcessQueryMsg(msg lnwire.Message, peerQuit <-chan struct{}) {
 	select {
 	case g.gossipMsgs <- msg:
-		return
+	case <-peerQuit:
 	case <-g.quit:
-		return
 	}
 }
 

peer.go

@@ -793,7 +793,7 @@ func (ms *msgStream) msgConsumer() {
 
 // AddMsg adds a new message to the msgStream. This function is safe for
 // concurrent access.
-func (ms *msgStream) AddMsg(msg lnwire.Message, quit chan struct{}) {
+func (ms *msgStream) AddMsg(msg lnwire.Message) {
 	// First, we'll attempt to receive from the producerSema struct. This
 	// acts as a sempahore to prevent us from indefinitely buffering
 	// incoming items from the wire. Either the msg queue isn't full, and
@@ -801,7 +801,7 @@ func (ms *msgStream) AddMsg(msg lnwire.Message, quit chan struct{}) {
 	// we're signalled to quit, or a slot is freed up.
 	select {
 	case <-ms.producerSema:
-	case <-quit:
+	case <-ms.peer.quit:
 		return
 	case <-ms.quit:
 		return
@@ -1019,7 +1019,7 @@ out:
 			// forward the error to all channels with this peer.
 			case msg.ChanID == lnwire.ConnectionWideID:
 				for chanID, chanStream := range chanMsgStreams {
-					chanStream.AddMsg(nextMsg, p.quit)
+					chanStream.AddMsg(nextMsg)
 
 					// Also marked this channel as failed,
 					// so we won't try to restart it on
@@ -1081,7 +1081,7 @@ out:
 			*lnwire.ReplyChannelRange,
 			*lnwire.ReplyShortChanIDsEnd:
 
-			discStream.AddMsg(msg, p.quit)
+			discStream.AddMsg(msg)
 
 		default:
 			peerLog.Errorf("unknown message %v received from peer "+
@@ -1104,7 +1104,7 @@ out:
 
 			// With the stream obtained, add the message to the
 			// stream so we can continue processing message.
-			chanStream.AddMsg(nextMsg, p.quit)
+			chanStream.AddMsg(nextMsg)
 		}
 
 		idleTimer.Reset(idleTimeout)