From 23d10336c2c8b3384dbce9ce40156c5709c72de2 Mon Sep 17 00:00:00 2001
From: Conner Fromknecht <conner@lightning.engineering>
Date: Fri, 26 Apr 2019 20:03:14 -0700
Subject: [PATCH 1/4] discovery/syncer: separate query + reply handlers

This commit creates a distinct replyHandler, completely isolating the
requesting state machine from the processing of queries from the remote
peer. Before the two were interlaced, and the syncer could only reply to
messages in certain states. Now the two will be complete separated,
which is preliminary step to make the replies synchronous (as otherwise
we would be blocking our own requesting state machine).

With this changes, the channelGraphSyncer of each peer will drive the
replyHanlder of the other. The two can now operate independently, or
even spun up conditionally depending on advertised support for gossip
queries, as shown below:

          A                                 B
 channelGraphSyncer ---control-msg--->
                                        replyHandler
 channelGraphSyncer <--control-msg----
           gossiper <--gossip-msgs----

                    <--control-msg---- channelGraphSyncer
       replyHandler
                    ---control-msg---> channelGraphSyncer
                    ---gossip-msgs---> gossiper
---
 discovery/syncer.go      | 108 +++++--
 discovery/syncer_test.go | 656 +++++++++++----------------------------
 2 files changed, 255 insertions(+), 509 deletions(-)

diff --git a/discovery/syncer.go b/discovery/syncer.go
index 09f3ee3f0..346215bfb 100644
--- a/discovery/syncer.go
+++ b/discovery/syncer.go
@@ -219,6 +219,16 @@ type gossipSyncerCfg struct {
 	// responding to gossip queries after replying to
 	// maxUndelayedQueryReplies queries.
 	delayedQueryReplyInterval time.Duration
+
+	// noSyncChannels will prevent the GossipSyncer from spawning a
+	// channelGraphSyncer, meaning we will not try to reconcile unknown
+	// channels with the remote peer.
+	noSyncChannels bool
+
+	// noReplyQueries will prevent the GossipSyncer from spawning a
+	// replyHandler, meaning we will not reply to queries from our remote
+	// peer.
+	noReplyQueries bool
 }
 
 // GossipSyncer is a struct that handles synchronizing the channel graph state
@@ -271,10 +281,15 @@ type GossipSyncer struct {
 	// PassiveSync to ActiveSync.
 	genHistoricalChanRangeQuery bool
 
-	// gossipMsgs is a channel that all messages from the target peer will
-	// be sent over.
+	// gossipMsgs is a channel that all responses to our queries from the
+	// target peer will be sent over, these will be read by the
+	// channelGraphSyncer.
 	gossipMsgs chan lnwire.Message
 
+	// queryMsgs is a channel that all queries from the remote peer will be
+	// received over, these will be read by the replyHandler.
+	queryMsgs chan lnwire.Message
+
 	// bufferedChanRangeReplies is used in the waitingQueryChanReply to
 	// buffer all the chunked response to our query.
 	bufferedChanRangeReplies []lnwire.ShortChannelID
@@ -332,6 +347,7 @@ func newGossipSyncer(cfg gossipSyncerCfg) *GossipSyncer {
 		syncTransitionReqs: make(chan *syncTransitionReq),
 		historicalSyncReqs: make(chan *historicalSyncReq),
 		gossipMsgs:         make(chan lnwire.Message, 100),
+		queryMsgs:          make(chan lnwire.Message, 100),
 		quit:               make(chan struct{}),
 	}
 }
@@ -342,8 +358,17 @@ func (g *GossipSyncer) Start() {
 	g.started.Do(func() {
 		log.Debugf("Starting GossipSyncer(%x)", g.cfg.peerPub[:])
 
-		g.wg.Add(1)
-		go g.channelGraphSyncer()
+		// TODO(conner): only spawn channelGraphSyncer if remote
+		// supports gossip queries, and only spawn replyHandler if we
+		// advertise support
+		if !g.cfg.noSyncChannels {
+			g.wg.Add(1)
+			go g.channelGraphSyncer()
+		}
+		if !g.cfg.noReplyQueries {
+			g.wg.Add(1)
+			go g.replyHandler()
+		}
 	})
 }
 
@@ -369,7 +394,7 @@ func (g *GossipSyncer) channelGraphSyncer() {
 		log.Debugf("GossipSyncer(%x): state=%v, type=%v",
 			g.cfg.peerPub[:], state, syncType)
 
-		switch syncerState(state) {
+		switch state {
 		// When we're in this state, we're trying to synchronize our
 		// view of the network with the remote peer. We'll kick off
 		// this sync by asking them for the set of channels they
@@ -382,14 +407,14 @@ func (g *GossipSyncer) channelGraphSyncer() {
 				g.genHistoricalChanRangeQuery,
 			)
 			if err != nil {
-				log.Errorf("unable to gen chan range "+
+				log.Errorf("Unable to gen chan range "+
 					"query: %v", err)
 				return
 			}
 
 			err = g.cfg.sendToPeer(queryRangeMsg)
 			if err != nil {
-				log.Errorf("unable to send chan range "+
+				log.Errorf("Unable to send chan range "+
 					"query: %v", err)
 				return
 			}
@@ -417,22 +442,16 @@ func (g *GossipSyncer) channelGraphSyncer() {
 				if ok {
 					err := g.processChanRangeReply(queryReply)
 					if err != nil {
-						log.Errorf("unable to "+
+						log.Errorf("Unable to "+
 							"process chan range "+
 							"query: %v", err)
 						return
 					}
-
 					continue
 				}
 
-				// Otherwise, it's the remote peer performing a
-				// query, which we'll attempt to reply to.
-				err := g.replyPeerQueries(msg)
-				if err != nil && err != ErrGossipSyncerExiting {
-					log.Errorf("unable to reply to peer "+
-						"query: %v", err)
-				}
+				log.Warnf("Unexpected message: %T in state=%v",
+					msg, state)
 
 			case <-g.quit:
 				return
@@ -447,7 +466,7 @@ func (g *GossipSyncer) channelGraphSyncer() {
 			// query chunk.
 			done, err := g.synchronizeChanIDs()
 			if err != nil {
-				log.Errorf("unable to sync chan IDs: %v", err)
+				log.Errorf("Unable to sync chan IDs: %v", err)
 			}
 
 			// If this wasn't our last query, then we'll need to
@@ -480,13 +499,8 @@ func (g *GossipSyncer) channelGraphSyncer() {
 					continue
 				}
 
-				// Otherwise, it's the remote peer performing a
-				// query, which we'll attempt to deploy to.
-				err := g.replyPeerQueries(msg)
-				if err != nil && err != ErrGossipSyncerExiting {
-					log.Errorf("unable to reply to peer "+
-						"query: %v", err)
-				}
+				log.Warnf("Unexpected message: %T in state=%v",
+					msg, state)
 
 			case <-g.quit:
 				return
@@ -520,13 +534,6 @@ func (g *GossipSyncer) channelGraphSyncer() {
 			// messages or process any state transitions and exit if
 			// needed.
 			select {
-			case msg := <-g.gossipMsgs:
-				err := g.replyPeerQueries(msg)
-				if err != nil && err != ErrGossipSyncerExiting {
-					log.Errorf("unable to reply to peer "+
-						"query: %v", err)
-				}
-
 			case req := <-g.syncTransitionReqs:
 				req.errChan <- g.handleSyncTransition(req)
 
@@ -540,6 +547,35 @@ func (g *GossipSyncer) channelGraphSyncer() {
 	}
 }
 
+// replyHandler is an event loop whose sole purpose is to reply to the remote
+// peers queries. Our replyHandler will respond to messages generated by their
+// channelGraphSyncer, and vice versa. Each party's channelGraphSyncer drives
+// the other's replyHandler, allowing the replyHandler to operate independently
+// from the state machine maintained on the same node.
+//
+// NOTE: This method MUST be run as a goroutine.
+func (g *GossipSyncer) replyHandler() {
+	defer g.wg.Done()
+
+	for {
+		select {
+		case msg := <-g.queryMsgs:
+			err := g.replyPeerQueries(msg)
+			switch {
+			case err == ErrGossipSyncerExiting:
+				return
+
+			case err != nil:
+				log.Errorf("Unable to reply to peer "+
+					"query: %v", err)
+			}
+
+		case <-g.quit:
+			return
+		}
+	}
+}
+
 // sendGossipTimestampRange constructs and sets a GossipTimestampRange for the
 // syncer and sends it to the remote peer.
 func (g *GossipSyncer) sendGossipTimestampRange(firstTimestamp time.Time,
@@ -1065,8 +1101,16 @@ 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, peerQuit <-chan struct{}) {
+	var msgChan chan lnwire.Message
+	switch msg.(type) {
+	case *lnwire.QueryChannelRange, *lnwire.QueryShortChanIDs:
+		msgChan = g.queryMsgs
+	default:
+		msgChan = g.gossipMsgs
+	}
+
 	select {
-	case g.gossipMsgs <- msg:
+	case msgChan <- msg:
 	case <-peerQuit:
 	case <-g.quit:
 	}
diff --git a/discovery/syncer_test.go b/discovery/syncer_test.go
index 5a5eb725b..b9ee24c7e 100644
--- a/discovery/syncer_test.go
+++ b/discovery/syncer_test.go
@@ -116,16 +116,37 @@ func (m *mockChannelGraphTimeSeries) FetchChanUpdates(chain chainhash.Hash,
 
 var _ ChannelGraphTimeSeries = (*mockChannelGraphTimeSeries)(nil)
 
+// newTestSyncer creates a new test instance of a GossipSyncer. A buffered
+// message channel is returned for intercepting messages sent from the syncer,
+// in addition to a mock channel series which allows the test to control which
+// messages the syncer knows of or wishes to filter out. The variadic flags are
+// treated as positional arguments where the first index signals that the syncer
+// should spawn a channelGraphSyncer and second index signals that the syncer
+// should spawn a replyHandler. Any flags beyond the first two are currently
+// ignored. If no flags are provided, both a channelGraphSyncer and replyHandler
+// will be spawned by default.
 func newTestSyncer(hID lnwire.ShortChannelID,
 	encodingType lnwire.ShortChanIDEncoding, chunkSize int32,
-) (chan []lnwire.Message, *GossipSyncer, *mockChannelGraphTimeSeries) {
+	flags ...bool) (chan []lnwire.Message,
+	*GossipSyncer, *mockChannelGraphTimeSeries) {
+
+	syncChannels := true
+	replyQueries := true
+	if len(flags) > 0 {
+		syncChannels = flags[0]
+	}
+	if len(flags) > 1 {
+		replyQueries = flags[1]
+	}
 
 	msgChan := make(chan []lnwire.Message, 20)
 	cfg := gossipSyncerCfg{
-		channelSeries: newMockChannelGraphTimeSeries(hID),
-		encodingType:  encodingType,
-		chunkSize:     chunkSize,
-		batchSize:     chunkSize,
+		channelSeries:  newMockChannelGraphTimeSeries(hID),
+		encodingType:   encodingType,
+		chunkSize:      chunkSize,
+		batchSize:      chunkSize,
+		noSyncChannels: !syncChannels,
+		noReplyQueries: !replyQueries,
 		sendToPeer: func(msgs ...lnwire.Message) error {
 			msgChan <- msgs
 			return nil
@@ -1020,13 +1041,13 @@ func TestGossipSyncerDelayDOS(t *testing.T) {
 		BlockHeight: 1144,
 	}
 	msgChan1, syncer1, chanSeries1 := newTestSyncer(
-		startHeight, defaultEncoding, chunkSize,
+		startHeight, defaultEncoding, chunkSize, true, false,
 	)
 	syncer1.Start()
 	defer syncer1.Stop()
 
 	msgChan2, syncer2, chanSeries2 := newTestSyncer(
-		startHeight, defaultEncoding, chunkSize,
+		startHeight, defaultEncoding, chunkSize, false, true,
 	)
 	syncer2.Start()
 	defer syncer2.Stop()
@@ -1042,7 +1063,7 @@ func TestGossipSyncerDelayDOS(t *testing.T) {
 	numQueryResponses := numUndelayedQueries + numDelayedQueries
 
 	// The total number of responses must include the initial reply each
-	// syner will make to QueryChannelRange.
+	// syncer will make to QueryChannelRange.
 	numTotalQueries := 1 + numQueryResponses
 
 	// The total number of channels each syncer needs to request must be
@@ -1051,12 +1072,6 @@ func TestGossipSyncerDelayDOS(t *testing.T) {
 
 	// Although both nodes are at the same height, they'll have a
 	// completely disjoint set of chan ID's that they know of.
-	var syncer1Chans []lnwire.ShortChannelID
-	for i := 0; i < numTotalChans; i++ {
-		syncer1Chans = append(
-			syncer1Chans, lnwire.NewShortChanIDFromInt(uint64(i)),
-		)
-	}
 	var syncer2Chans []lnwire.ShortChannelID
 	for i := numTotalChans; i < numTotalChans+numTotalChans; i++ {
 		syncer2Chans = append(
@@ -1064,8 +1079,8 @@ func TestGossipSyncerDelayDOS(t *testing.T) {
 		)
 	}
 
-	// We'll kick off the test by passing over the QueryChannelRange
-	// messages from one node to the other.
+	// We'll kick off the test by asserting syncer1 sends over the
+	// QueryChannelRange message the other node.
 	select {
 	case <-time.After(time.Second * 2):
 		t.Fatalf("didn't get msg from syncer1")
@@ -1083,46 +1098,16 @@ func TestGossipSyncerDelayDOS(t *testing.T) {
 			case <-time.After(time.Second * 2):
 				t.Fatalf("node 2 didn't read msg")
 
-			case syncer2.gossipMsgs <- msg:
-
-			}
-		}
-	}
-	select {
-	case <-time.After(time.Second * 2):
-		t.Fatalf("didn't get msg from syncer2")
-
-	case msgs := <-msgChan2:
-		for _, msg := range msgs {
-			// The message MUST be a QueryChannelRange message.
-			_, ok := msg.(*lnwire.QueryChannelRange)
-			if !ok {
-				t.Fatalf("wrong message: expected "+
-					"QueryChannelRange for %T", msg)
-			}
-
-			select {
-			case <-time.After(time.Second * 2):
-				t.Fatalf("node 2 didn't read msg")
-
-			case syncer1.gossipMsgs <- msg:
+			case syncer2.queryMsgs <- msg:
 
 			}
 		}
 	}
 
-	// At this point, we'll need to send responses to both nodes from their
-	// respective channel series. Both nodes will simply request the entire
-	// set of channels from the other. This will count as the first
-	// undelayed response for each syncer.
-	select {
-	case <-time.After(time.Second * 2):
-		t.Fatalf("no query recvd")
-
-	case <-chanSeries1.filterRangeReqs:
-		// We'll send all the channels that it should know of.
-		chanSeries1.filterRangeResp <- syncer1Chans
-	}
+	// At this point, we'll need to a response from syncer2's channel
+	// series. This will cause syncer1 to simply request the entire set of
+	// channels from syncer2. This will count as the first undelayed
+	// response for sycner2.
 	select {
 	case <-time.After(time.Second * 2):
 		t.Fatalf("no query recvd")
@@ -1132,31 +1117,9 @@ func TestGossipSyncerDelayDOS(t *testing.T) {
 		chanSeries2.filterRangeResp <- syncer2Chans
 	}
 
-	// At this point, we'll forward the ReplyChannelRange messages to both
-	// parties. After receiving the set of channels known to the remote peer
+	// At this point, we'll assert that the ReplyChannelRange message is
+	// sent by sycner2.
 	for i := 0; i < numQueryResponses; i++ {
-		select {
-		case <-time.After(time.Second * 2):
-			t.Fatalf("didn't get msg from syncer1")
-
-		case msgs := <-msgChan1:
-			for _, msg := range msgs {
-				// The message MUST be a ReplyChannelRange message.
-				_, ok := msg.(*lnwire.ReplyChannelRange)
-				if !ok {
-					t.Fatalf("wrong message: expected "+
-						"QueryChannelRange for %T", msg)
-				}
-
-				select {
-				case <-time.After(time.Second * 2):
-					t.Fatalf("node 2 didn't read msg")
-
-				case syncer2.gossipMsgs <- msg:
-				}
-			}
-		}
-
 		select {
 		case <-time.After(time.Second * 2):
 			t.Fatalf("didn't get msg from syncer2")
@@ -1180,8 +1143,7 @@ func TestGossipSyncerDelayDOS(t *testing.T) {
 		}
 	}
 
-	// We'll now send back a chunked response for both parties of the known
-	// short chan ID's.
+	// We'll now have syncer1 process the received sids from syncer2.
 	select {
 	case <-time.After(time.Second * 2):
 		t.Fatalf("no query recvd")
@@ -1189,207 +1151,140 @@ func TestGossipSyncerDelayDOS(t *testing.T) {
 	case <-chanSeries1.filterReq:
 		chanSeries1.filterResp <- syncer2Chans
 	}
+
+	// At this point, syncer1 should start to send out initial requests to
+	// query the chan IDs of the remote party. We'll keep track of the
+	// number of queries made using the iterated value, which starts at one
+	// due the initial contribution of the QueryChannelRange msgs.
+	for i := 1; i < numTotalQueries; i++ {
+		expDelayResponse := i >= numUndelayedQueries
+		queryBatch(t,
+			msgChan1, msgChan2,
+			syncer1, syncer2,
+			chanSeries2,
+			expDelayResponse,
+			delayedQueryInterval,
+			delayTolerance,
+		)
+	}
+}
+
+// queryBatch is a helper method that will query for a single batch of channels
+// from a peer and assert the responses. The method can also be used to assert
+// the same transition happens, but is delayed by the remote peer's DOS
+// rate-limiting. The provided chanSeries should belong to syncer2.
+//
+// The state transition performed is the following:
+//   syncer1  -- QueryShortChanIDs -->   syncer2
+//                                       chanSeries.FetchChanAnns()
+//   syncer1 <-- ReplyShortChanIDsEnd -- syncer2
+//
+// If expDelayResponse is true, this method will assert that the call the
+// FetchChanAnns happens between:
+//   [delayedQueryInterval-delayTolerance, delayedQueryInterval+delayTolerance].
+func queryBatch(t *testing.T,
+	msgChan1, msgChan2 chan []lnwire.Message,
+	syncer1, syncer2 *GossipSyncer,
+	chanSeries *mockChannelGraphTimeSeries,
+	expDelayResponse bool,
+	delayedQueryInterval, delayTolerance time.Duration) {
+
+	t.Helper()
+
+	// First, we'll assert that syncer1 sends a QueryShortChanIDs message to
+	// the remote peer.
 	select {
 	case <-time.After(time.Second * 2):
-		t.Fatalf("no query recvd")
+		t.Fatalf("didn't get msg from syncer2")
 
-	case <-chanSeries2.filterReq:
-		chanSeries2.filterResp <- syncer1Chans
+	case msgs := <-msgChan1:
+		for _, msg := range msgs {
+			// The message MUST be a QueryShortChanIDs message.
+			_, ok := msg.(*lnwire.QueryShortChanIDs)
+			if !ok {
+				t.Fatalf("wrong message: expected "+
+					"QueryShortChanIDs for %T", msg)
+			}
+
+			select {
+			case <-time.After(time.Second * 2):
+				t.Fatalf("node 2 didn't read msg")
+
+			case syncer2.queryMsgs <- msg:
+			}
+		}
 	}
 
-	// At this point, both parties should start to send out initial
-	// requests to query the chan IDs of the remote party. We'll keep track
-	// of the number of queries made using the iterated value, which starts
-	// at one due the initial contribution of the QueryChannelRange msgs.
-	for i := 1; i < numTotalQueries; i++ {
-		// Both parties should now have sent out the initial requests
-		// to query the chan IDs of the other party.
+	// We'll then respond to with an empty set of replies (as it doesn't
+	// affect the test).
+	switch {
+
+	// If this query has surpassed the undelayed query threshold, we will
+	// impose stricter timing constraints on the response times. We'll first
+	// test that syncer2's chanSeries doesn't immediately receive a query,
+	// and then check that the query hasn't gone unanswered entirely.
+	case expDelayResponse:
+		// Create a before and after timeout to test, our test
+		// will ensure the messages are delivered to the peer
+		// in this timeframe.
+		before := time.After(
+			delayedQueryInterval - delayTolerance,
+		)
+		after := time.After(
+			delayedQueryInterval + delayTolerance,
+		)
+
+		// First, ensure syncer2 doesn't try to respond up until the
+		// before time fires.
 		select {
-		case <-time.After(time.Second * 2):
-			t.Fatalf("didn't get msg from syncer1")
+		case <-before:
+			// Query is delayed, proceed.
 
-		case msgs := <-msgChan1:
-			for _, msg := range msgs {
-				// The message MUST be a QueryShortChanIDs message.
-				_, ok := msg.(*lnwire.QueryShortChanIDs)
-				if !ok {
-					t.Fatalf("wrong message: expected "+
-						"QueryShortChanIDs for %T", msg)
-				}
-
-				select {
-				case <-time.After(time.Second * 2):
-					t.Fatalf("node 2 didn't read msg")
-
-				case syncer2.gossipMsgs <- msg:
-
-				}
-			}
+		case <-chanSeries.annReq:
+			t.Fatalf("DOSy query was not delayed")
 		}
+
+		// If syncer2 doesn't attempt a response within the allowed
+		// interval, then the messages are probably lost.
 		select {
-		case <-time.After(time.Second * 2):
-			t.Fatalf("didn't get msg from syncer2")
+		case <-after:
+			t.Fatalf("no delayed query received")
 
-		case msgs := <-msgChan2:
-			for _, msg := range msgs {
-				// The message MUST be a QueryShortChanIDs message.
-				_, ok := msg.(*lnwire.QueryShortChanIDs)
-				if !ok {
-					t.Fatalf("wrong message: expected "+
-						"QueryShortChanIDs for %T", msg)
-				}
-
-				select {
-				case <-time.After(time.Second * 2):
-					t.Fatalf("node 2 didn't read msg")
-
-				case syncer1.gossipMsgs <- msg:
-
-				}
-			}
+		case <-chanSeries.annReq:
+			chanSeries.annResp <- []lnwire.Message{}
 		}
 
-		// We'll then respond to both parties with an empty set of
-		// replies (as it doesn't affect the test).
-		switch {
-
-		// If this query has surpassed the undelayed query threshold, we
-		// will impose stricter timing constraints on the response
-		// times. We'll first test that the peers don't immediately
-		// receive a query, and then check that both queries haven't
-		// gone unanswered entirely.
-		case i >= numUndelayedQueries:
-			// Create a before and after timeout to test, our test
-			// will ensure the messages are delivered to the peers
-			// in this timeframe.
-			before := time.After(
-				delayedQueryInterval - delayTolerance,
-			)
-			after := time.After(
-				delayedQueryInterval + delayTolerance,
-			)
-
-			// First, ensure neither peer tries to respond up until
-			// the before time fires.
-			select {
-			case <-before:
-				// Queries are delayed, proceed.
-
-			case <-chanSeries1.annReq:
-				t.Fatalf("DOSy query was not delayed")
-
-			case <-chanSeries2.annReq:
-				t.Fatalf("DOSy query was not delayed")
-			}
-
-			// Next, we'll need to test that both queries are
-			// received before the after timer expires. To account
-			// for ordering, we will try to pull a message from both
-			// peers, and then test that the opposite peer also
-			// receives the message promptly.
-			var (
-				firstChanSeries *mockChannelGraphTimeSeries
-				laterChanSeries *mockChannelGraphTimeSeries
-			)
-
-			// If neither peer attempts a response within the
-			// allowed interval, then the messages are probably
-			// lost. Otherwise, process the message and record the
-			// induced ordering.
-			select {
-			case <-after:
-				t.Fatalf("no delayed query received")
-
-			case <-chanSeries1.annReq:
-				chanSeries1.annResp <- []lnwire.Message{}
-				firstChanSeries = chanSeries1
-				laterChanSeries = chanSeries2
-
-			case <-chanSeries2.annReq:
-				chanSeries2.annResp <- []lnwire.Message{}
-				firstChanSeries = chanSeries2
-				laterChanSeries = chanSeries1
-			}
-
-			// Finally, using the same interval timeout as before,
-			// ensure the later peer also responds promptly. We also
-			// assert that the first peer doesn't attempt another
-			// response.
-			select {
-			case <-after:
-				t.Fatalf("no delayed query received")
-
-			case <-firstChanSeries.annReq:
-				t.Fatalf("spurious undelayed response")
-
-			case <-laterChanSeries.annReq:
-				laterChanSeries.annResp <- []lnwire.Message{}
-			}
-
-		// Otherwise, we still haven't exceeded our undelayed query
-		// limit. Assert that both peers promptly attempt a response to
-		// the queries.
-		default:
-			select {
-			case <-time.After(50 * time.Millisecond):
-				t.Fatalf("no query recvd")
-
-			case <-chanSeries1.annReq:
-				chanSeries1.annResp <- []lnwire.Message{}
-			}
-			select {
-			case <-time.After(50 * time.Millisecond):
-				t.Fatalf("no query recvd")
-
-			case <-chanSeries2.annReq:
-				chanSeries2.annResp <- []lnwire.Message{}
-			}
-		}
-
-		// Finally, both sides should then receive a
-		// ReplyShortChanIDsEnd as the first chunk has been replied to.
+	// Otherwise, syncer2 should query its chanSeries promtly.
+	default:
 		select {
 		case <-time.After(50 * time.Millisecond):
-			t.Fatalf("didn't get msg from syncer1")
+			t.Fatalf("no query recvd")
 
-		case msgs := <-msgChan1:
-			for _, msg := range msgs {
-				// The message MUST be a ReplyShortChanIDsEnd message.
-				_, ok := msg.(*lnwire.ReplyShortChanIDsEnd)
-				if !ok {
-					t.Fatalf("wrong message: expected "+
-						"QueryChannelRange for %T", msg)
-				}
-
-				select {
-				case <-time.After(time.Second * 2):
-					t.Fatalf("node 2 didn't read msg")
-
-				case syncer2.gossipMsgs <- msg:
-
-				}
-			}
+		case <-chanSeries.annReq:
+			chanSeries.annResp <- []lnwire.Message{}
 		}
-		select {
-		case <-time.After(50 * time.Millisecond):
-			t.Fatalf("didn't get msg from syncer2")
+	}
 
-		case msgs := <-msgChan2:
-			for _, msg := range msgs {
-				// The message MUST be a ReplyShortChanIDsEnd message.
-				_, ok := msg.(*lnwire.ReplyShortChanIDsEnd)
-				if !ok {
-					t.Fatalf("wrong message: expected "+
-						"ReplyShortChanIDsEnd for %T", msg)
-				}
+	// Finally, assert that syncer2 replies to syncer1 with a
+	// ReplyShortChanIDsEnd.
+	select {
+	case <-time.After(50 * time.Millisecond):
+		t.Fatalf("didn't get msg from syncer2")
 
-				select {
-				case <-time.After(time.Second * 2):
-					t.Fatalf("node 2 didn't read msg")
+	case msgs := <-msgChan2:
+		for _, msg := range msgs {
+			// The message MUST be a ReplyShortChanIDsEnd message.
+			_, ok := msg.(*lnwire.ReplyShortChanIDsEnd)
+			if !ok {
+				t.Fatalf("wrong message: expected "+
+					"ReplyShortChanIDsEnd for %T", msg)
+			}
 
-				case syncer1.gossipMsgs <- msg:
+			select {
+			case <-time.After(time.Second * 2):
+				t.Fatalf("node 2 didn't read msg")
 
-				}
+			case syncer1.gossipMsgs <- msg:
 			}
 		}
 	}
@@ -1413,24 +1308,19 @@ func TestGossipSyncerRoutineSync(t *testing.T) {
 		BlockHeight: 1144,
 	}
 	msgChan1, syncer1, chanSeries1 := newTestSyncer(
-		startHeight, defaultEncoding, chunkSize,
+		startHeight, defaultEncoding, chunkSize, true, false,
 	)
 	syncer1.Start()
 	defer syncer1.Stop()
 
 	msgChan2, syncer2, chanSeries2 := newTestSyncer(
-		startHeight, defaultEncoding, chunkSize,
+		startHeight, defaultEncoding, chunkSize, false, true,
 	)
 	syncer2.Start()
 	defer syncer2.Stop()
 
-	// Although both nodes are at the same height, they'll have a
-	// completely disjoint set of 3 chan ID's that they know of.
-	syncer1Chans := []lnwire.ShortChannelID{
-		lnwire.NewShortChanIDFromInt(1),
-		lnwire.NewShortChanIDFromInt(2),
-		lnwire.NewShortChanIDFromInt(3),
-	}
+	// Although both nodes are at the same height, syncer will have 3 chan
+	// ID's that syncer1 doesn't know of.
 	syncer2Chans := []lnwire.ShortChannelID{
 		lnwire.NewShortChanIDFromInt(4),
 		lnwire.NewShortChanIDFromInt(5),
@@ -1438,7 +1328,7 @@ func TestGossipSyncerRoutineSync(t *testing.T) {
 	}
 
 	// We'll kick off the test by passing over the QueryChannelRange
-	// messages from one node to the other.
+	// messages from syncer1 to syncer2.
 	select {
 	case <-time.After(time.Second * 2):
 		t.Fatalf("didn't get msg from syncer1")
@@ -1456,45 +1346,15 @@ func TestGossipSyncerRoutineSync(t *testing.T) {
 			case <-time.After(time.Second * 2):
 				t.Fatalf("node 2 didn't read msg")
 
-			case syncer2.gossipMsgs <- msg:
-
-			}
-		}
-	}
-	select {
-	case <-time.After(time.Second * 2):
-		t.Fatalf("didn't get msg from syncer2")
-
-	case msgs := <-msgChan2:
-		for _, msg := range msgs {
-			// The message MUST be a QueryChannelRange message.
-			_, ok := msg.(*lnwire.QueryChannelRange)
-			if !ok {
-				t.Fatalf("wrong message: expected "+
-					"QueryChannelRange for %T", msg)
-			}
-
-			select {
-			case <-time.After(time.Second * 2):
-				t.Fatalf("node 2 didn't read msg")
-
-			case syncer1.gossipMsgs <- msg:
+			case syncer2.queryMsgs <- msg:
 
 			}
 		}
 	}
 
-	// At this point, we'll need to send responses to both nodes from their
-	// respective channel series. Both nodes will simply request the entire
-	// set of channels from the other.
-	select {
-	case <-time.After(time.Second * 2):
-		t.Fatalf("no query recvd")
-
-	case <-chanSeries1.filterRangeReqs:
-		// We'll send all the channels that it should know of.
-		chanSeries1.filterRangeResp <- syncer1Chans
-	}
+	// At this point, we'll need to send a response from syncer2 to syncer1
+	// using syncer2's channels This will cause syncer1 to simply request
+	// the entire set of channels from the other.
 	select {
 	case <-time.After(time.Second * 2):
 		t.Fatalf("no query recvd")
@@ -1504,32 +1364,9 @@ func TestGossipSyncerRoutineSync(t *testing.T) {
 		chanSeries2.filterRangeResp <- syncer2Chans
 	}
 
-	// At this point, we'll forward the ReplyChannelRange messages to both
-	// parties. Two replies are expected since the chunk size is 2, and we
-	// need to query for 3 channels.
-	for i := 0; i < chunkSize; i++ {
-		select {
-		case <-time.After(time.Second * 2):
-			t.Fatalf("didn't get msg from syncer1")
-
-		case msgs := <-msgChan1:
-			for _, msg := range msgs {
-				// The message MUST be a ReplyChannelRange message.
-				_, ok := msg.(*lnwire.ReplyChannelRange)
-				if !ok {
-					t.Fatalf("wrong message: expected "+
-						"QueryChannelRange for %T", msg)
-				}
-
-				select {
-				case <-time.After(time.Second * 2):
-					t.Fatalf("node 2 didn't read msg")
-
-				case syncer2.gossipMsgs <- msg:
-				}
-			}
-		}
-	}
+	// At this point, we'll assert that syncer2 replies with the
+	// ReplyChannelRange messages. Two replies are expected since the chunk
+	// size is 2, and we need to query for 3 channels.
 	for i := 0; i < chunkSize; i++ {
 		select {
 		case <-time.After(time.Second * 2):
@@ -1554,8 +1391,7 @@ func TestGossipSyncerRoutineSync(t *testing.T) {
 		}
 	}
 
-	// We'll now send back a chunked response for both parties of the known
-	// short chan ID's.
+	// We'll now send back a chunked response from syncer2 back to sycner1.
 	select {
 	case <-time.After(time.Second * 2):
 		t.Fatalf("no query recvd")
@@ -1563,133 +1399,21 @@ func TestGossipSyncerRoutineSync(t *testing.T) {
 	case <-chanSeries1.filterReq:
 		chanSeries1.filterResp <- syncer2Chans
 	}
-	select {
-	case <-time.After(time.Second * 2):
-		t.Fatalf("no query recvd")
 
-	case <-chanSeries2.filterReq:
-		chanSeries2.filterResp <- syncer1Chans
-	}
-
-	// At this point, both parties should start to send out initial
-	// requests to query the chan IDs of the remote party. As the chunk
-	// size is 2, they'll need 2 rounds in order to fully reconcile the
-	// state.
+	// At this point, syncer1 should start to send out initial requests to
+	// query the chan IDs of the remote party. As the chunk size is 2,
+	// they'll need 2 rounds in order to fully reconcile the state.
 	for i := 0; i < chunkSize; i++ {
-		// Both parties should now have sent out the initial requests
-		// to query the chan IDs of the other party.
-		select {
-		case <-time.After(time.Second * 2):
-			t.Fatalf("didn't get msg from syncer1")
-
-		case msgs := <-msgChan1:
-			for _, msg := range msgs {
-				// The message MUST be a QueryShortChanIDs message.
-				_, ok := msg.(*lnwire.QueryShortChanIDs)
-				if !ok {
-					t.Fatalf("wrong message: expected "+
-						"QueryShortChanIDs for %T", msg)
-				}
-
-				select {
-				case <-time.After(time.Second * 2):
-					t.Fatalf("node 2 didn't read msg")
-
-				case syncer2.gossipMsgs <- msg:
-
-				}
-			}
-		}
-		select {
-		case <-time.After(time.Second * 2):
-			t.Fatalf("didn't get msg from syncer2")
-
-		case msgs := <-msgChan2:
-			for _, msg := range msgs {
-				// The message MUST be a QueryShortChanIDs message.
-				_, ok := msg.(*lnwire.QueryShortChanIDs)
-				if !ok {
-					t.Fatalf("wrong message: expected "+
-						"QueryShortChanIDs for %T", msg)
-				}
-
-				select {
-				case <-time.After(time.Second * 2):
-					t.Fatalf("node 2 didn't read msg")
-
-				case syncer1.gossipMsgs <- msg:
-
-				}
-			}
-		}
-
-		// We'll then respond to both parties with an empty set of replies (as
-		// it doesn't affect the test).
-		select {
-		case <-time.After(time.Second * 2):
-			t.Fatalf("no query recvd")
-
-		case <-chanSeries1.annReq:
-			chanSeries1.annResp <- []lnwire.Message{}
-		}
-		select {
-		case <-time.After(time.Second * 2):
-			t.Fatalf("no query recvd")
-
-		case <-chanSeries2.annReq:
-			chanSeries2.annResp <- []lnwire.Message{}
-		}
-
-		// Both sides should then receive a ReplyShortChanIDsEnd as the first
-		// chunk has been replied to.
-		select {
-		case <-time.After(time.Second * 2):
-			t.Fatalf("didn't get msg from syncer1")
-
-		case msgs := <-msgChan1:
-			for _, msg := range msgs {
-				// The message MUST be a ReplyShortChanIDsEnd message.
-				_, ok := msg.(*lnwire.ReplyShortChanIDsEnd)
-				if !ok {
-					t.Fatalf("wrong message: expected "+
-						"QueryChannelRange for %T", msg)
-				}
-
-				select {
-				case <-time.After(time.Second * 2):
-					t.Fatalf("node 2 didn't read msg")
-
-				case syncer2.gossipMsgs <- msg:
-
-				}
-			}
-		}
-		select {
-		case <-time.After(time.Second * 2):
-			t.Fatalf("didn't get msg from syncer1")
-
-		case msgs := <-msgChan2:
-			for _, msg := range msgs {
-				// The message MUST be a ReplyShortChanIDsEnd message.
-				_, ok := msg.(*lnwire.ReplyShortChanIDsEnd)
-				if !ok {
-					t.Fatalf("wrong message: expected "+
-						"ReplyShortChanIDsEnd for %T", msg)
-				}
-
-				select {
-				case <-time.After(time.Second * 2):
-					t.Fatalf("node 2 didn't read msg")
-
-				case syncer1.gossipMsgs <- msg:
-
-				}
-			}
-		}
+		queryBatch(t,
+			msgChan1, msgChan2,
+			syncer1, syncer2,
+			chanSeries2,
+			false, 0, 0,
+		)
 	}
 
-	// At this stage both parties should now be sending over their initial
-	// GossipTimestampRange messages as they should both be fully synced.
+	// At this stage syncer1 should now be sending over its initial
+	// GossipTimestampRange messages as it should be fully synced.
 	select {
 	case <-time.After(time.Second * 2):
 		t.Fatalf("didn't get msg from syncer1")
@@ -1709,28 +1433,6 @@ func TestGossipSyncerRoutineSync(t *testing.T) {
 
 			case syncer2.gossipMsgs <- msg:
 
-			}
-		}
-	}
-	select {
-	case <-time.After(time.Second * 2):
-		t.Fatalf("didn't get msg from syncer1")
-
-	case msgs := <-msgChan2:
-		for _, msg := range msgs {
-			// The message MUST be a GossipTimestampRange message.
-			_, ok := msg.(*lnwire.GossipTimestampRange)
-			if !ok {
-				t.Fatalf("wrong message: expected "+
-					"QueryChannelRange for %T", msg)
-			}
-
-			select {
-			case <-time.After(time.Second * 2):
-				t.Fatalf("node 2 didn't read msg")
-
-			case syncer1.gossipMsgs <- msg:
-
 			}
 		}
 	}
@@ -1796,7 +1498,7 @@ func TestGossipSyncerAlreadySynced(t *testing.T) {
 			case <-time.After(time.Second * 2):
 				t.Fatalf("node 2 didn't read msg")
 
-			case syncer2.gossipMsgs <- msg:
+			case syncer2.queryMsgs <- msg:
 
 			}
 		}
@@ -1818,7 +1520,7 @@ func TestGossipSyncerAlreadySynced(t *testing.T) {
 			case <-time.After(time.Second * 2):
 				t.Fatalf("node 2 didn't read msg")
 
-			case syncer1.gossipMsgs <- msg:
+			case syncer1.queryMsgs <- msg:
 
 			}
 		}

From ca358e9673b86ca04b53433e69d63305b0bb85d4 Mon Sep 17 00:00:00 2001
From: Conner Fromknecht <conner@lightning.engineering>
Date: Fri, 26 Apr 2019 17:31:27 -0700
Subject: [PATCH 2/4] lnpeer+peer: use importable lnpeer.ErrPeerExiting

As a prepatory step to making gossip replies synchronous, we will move
the ErrPeerExiting error into the lnpeer package so that it can be
imported by the discovery package. With synchronous sends, this error
can now be detected and handled by goroutines in the syncer, and cause
them to exit instead of continue to sending backlogs.
---
 lnpeer/errors.go |  8 ++++++++
 peer.go          | 15 ++++++---------
 2 files changed, 14 insertions(+), 9 deletions(-)
 create mode 100644 lnpeer/errors.go

diff --git a/lnpeer/errors.go b/lnpeer/errors.go
new file mode 100644
index 000000000..3787afe36
--- /dev/null
+++ b/lnpeer/errors.go
@@ -0,0 +1,8 @@
+package lnpeer
+
+import "fmt"
+
+var (
+	// ErrPeerExiting signals that the peer received a disconnect request.
+	ErrPeerExiting = fmt.Errorf("peer exiting")
+)
diff --git a/peer.go b/peer.go
index 0e79088f8..71b06fc4c 100644
--- a/peer.go
+++ b/peer.go
@@ -32,9 +32,6 @@ import (
 
 var (
 	numNodes int32
-
-	// ErrPeerExiting signals that the peer received a disconnect request.
-	ErrPeerExiting = fmt.Errorf("peer exiting")
 )
 
 const (
@@ -1411,7 +1408,7 @@ func (p *peer) logWireMessage(msg lnwire.Message, read bool) {
 func (p *peer) writeMessage(msg lnwire.Message) error {
 	// Simply exit if we're shutting down.
 	if atomic.LoadInt32(&p.disconnect) != 0 {
-		return ErrPeerExiting
+		return lnpeer.ErrPeerExiting
 	}
 
 	// Only log the message on the first attempt.
@@ -1559,7 +1556,7 @@ out:
 			}
 
 		case <-p.quit:
-			exitErr = ErrPeerExiting
+			exitErr = lnpeer.ErrPeerExiting
 			break out
 		}
 	}
@@ -1691,7 +1688,7 @@ func (p *peer) queue(priority bool, msg lnwire.Message, errChan chan error) {
 	case <-p.quit:
 		peerLog.Tracef("Peer shutting down, could not enqueue msg.")
 		if errChan != nil {
-			errChan <- ErrPeerExiting
+			errChan <- lnpeer.ErrPeerExiting
 		}
 	}
 }
@@ -2504,7 +2501,7 @@ func (p *peer) sendMessage(sync, priority bool, msgs ...lnwire.Message) error {
 		case err := <-errChan:
 			return err
 		case <-p.quit:
-			return ErrPeerExiting
+			return lnpeer.ErrPeerExiting
 		}
 	}
 
@@ -2550,7 +2547,7 @@ func (p *peer) AddNewChannel(channel *channeldb.OpenChannel,
 	case <-cancel:
 		return errors.New("canceled adding new channel")
 	case <-p.quit:
-		return ErrPeerExiting
+		return lnpeer.ErrPeerExiting
 	}
 
 	// We pause here to wait for the peer to recognize the new channel
@@ -2559,7 +2556,7 @@ func (p *peer) AddNewChannel(channel *channeldb.OpenChannel,
 	case err := <-errChan:
 		return err
 	case <-p.quit:
-		return ErrPeerExiting
+		return lnpeer.ErrPeerExiting
 	}
 }
 

From bf4543e2bd48081358a1559d86bf0ea1d1232cf4 Mon Sep 17 00:00:00 2001
From: Conner Fromknecht <conner@lightning.engineering>
Date: Fri, 26 Apr 2019 20:05:10 -0700
Subject: [PATCH 3/4] discovery/syncer: make gossip sends synchronous

This commit makes all replies in the gossip syncer synchronous, meaning
that they will wait for each message to be successfully written to the
remote peer before attempting to send the next. This helps throttle
messages the remote peer has requested, preventing unintended
disconnects when the remote peer is slow to process messages. This
changes also helps out congestion in the peer by forcing the syncer to
buffer the messages instead of dumping them into the peer's queue.
---
 discovery/sync_manager.go |  3 ++
 discovery/syncer.go       | 58 +++++++++++++++++++++++++--------------
 discovery/syncer_test.go  | 53 +++++++++++++++++++++--------------
 3 files changed, 73 insertions(+), 41 deletions(-)

diff --git a/discovery/sync_manager.go b/discovery/sync_manager.go
index 060225184..c1bb72c31 100644
--- a/discovery/sync_manager.go
+++ b/discovery/sync_manager.go
@@ -386,6 +386,9 @@ func (m *SyncManager) createGossipSyncer(peer lnpeer.Peer) *GossipSyncer {
 		sendToPeer: func(msgs ...lnwire.Message) error {
 			return peer.SendMessageLazy(false, msgs...)
 		},
+		sendToPeerSync: func(msgs ...lnwire.Message) error {
+			return peer.SendMessageLazy(true, msgs...)
+		},
 	})
 
 	// Gossip syncers are initialized by default in a PassiveSync type
diff --git a/discovery/syncer.go b/discovery/syncer.go
index 346215bfb..02f90b79f 100644
--- a/discovery/syncer.go
+++ b/discovery/syncer.go
@@ -9,6 +9,7 @@ import (
 	"time"
 
 	"github.com/btcsuite/btcd/chaincfg/chainhash"
+	"github.com/lightningnetwork/lnd/lnpeer"
 	"github.com/lightningnetwork/lnd/lnwire"
 	"golang.org/x/time/rate"
 )
@@ -206,11 +207,16 @@ type gossipSyncerCfg struct {
 	// the remote node in a single QueryShortChanIDs request.
 	batchSize int32
 
-	// sendToPeer is a function closure that should send the set of
-	// targeted messages to the peer we've been assigned to sync the graph
-	// state from.
+	// sendToPeer sends a variadic number of messages to the remote peer.
+	// This method should not block while waiting for sends to be written
+	// to the wire.
 	sendToPeer func(...lnwire.Message) error
 
+	// sendToPeerSync sends a variadic number of messages to the remote
+	// peer, blocking until all messages have been sent successfully or a
+	// write error is encountered.
+	sendToPeerSync func(...lnwire.Message) error
+
 	// maxUndelayedQueryReplies specifies how many gossip queries we will
 	// respond to immediately before starting to delay responses.
 	maxUndelayedQueryReplies int
@@ -565,6 +571,9 @@ func (g *GossipSyncer) replyHandler() {
 			case err == ErrGossipSyncerExiting:
 				return
 
+			case err == lnpeer.ErrPeerExiting:
+				return
+
 			case err != nil:
 				log.Errorf("Unable to reply to peer "+
 					"query: %v", err)
@@ -854,7 +863,7 @@ func (g *GossipSyncer) replyChanRangeQuery(query *lnwire.QueryChannelRange) erro
 		if isFinalChunk {
 			replyChunk.Complete = 1
 		}
-		if err := g.cfg.sendToPeer(&replyChunk); err != nil {
+		if err := g.cfg.sendToPeerSync(&replyChunk); err != nil {
 			return err
 		}
 
@@ -882,7 +891,7 @@ func (g *GossipSyncer) replyShortChanIDs(query *lnwire.QueryShortChanIDs) error
 			"chain=%v, we're on chain=%v", g.cfg.chainHash,
 			query.ChainHash)
 
-		return g.cfg.sendToPeer(&lnwire.ReplyShortChanIDsEnd{
+		return g.cfg.sendToPeerSync(&lnwire.ReplyShortChanIDsEnd{
 			ChainHash: query.ChainHash,
 			Complete:  0,
 		})
@@ -909,23 +918,22 @@ func (g *GossipSyncer) replyShortChanIDs(query *lnwire.QueryShortChanIDs) error
 			query.ShortChanIDs[0].ToUint64(), err)
 	}
 
-	// If we didn't find any messages related to those channel ID's, then
-	// we'll send over a reply marking the end of our response, and exit
-	// early.
-	if len(replyMsgs) == 0 {
-		return g.cfg.sendToPeer(&lnwire.ReplyShortChanIDsEnd{
-			ChainHash: query.ChainHash,
-			Complete:  1,
-		})
+	// Reply with any messages related to those channel ID's, we'll write
+	// each one individually and synchronously to throttle the sends and
+	// perform buffering of responses in the syncer as opposed to the peer.
+	for _, msg := range replyMsgs {
+		err := g.cfg.sendToPeerSync(msg)
+		if err != nil {
+			return err
+		}
 	}
 
-	// Otherwise, we'll send over our set of messages responding to the
-	// query, with the ending message appended to it.
-	replyMsgs = append(replyMsgs, &lnwire.ReplyShortChanIDsEnd{
+	// Regardless of whether we had any messages to reply with, send over
+	// the sentinel message to signal that the stream has terminated.
+	return g.cfg.sendToPeerSync(&lnwire.ReplyShortChanIDsEnd{
 		ChainHash: query.ChainHash,
 		Complete:  1,
 	})
-	return g.cfg.sendToPeer(replyMsgs...)
 }
 
 // ApplyGossipFilter applies a gossiper filter sent by the remote node to the
@@ -966,9 +974,19 @@ func (g *GossipSyncer) ApplyGossipFilter(filter *lnwire.GossipTimestampRange) er
 	go func() {
 		defer g.wg.Done()
 
-		if err := g.cfg.sendToPeer(newUpdatestoSend...); err != nil {
-			log.Errorf("unable to send messages for peer catch "+
-				"up: %v", err)
+		for _, msg := range newUpdatestoSend {
+			err := g.cfg.sendToPeerSync(msg)
+			switch {
+			case err == ErrGossipSyncerExiting:
+				return
+
+			case err == lnpeer.ErrPeerExiting:
+				return
+
+			case err != nil:
+				log.Errorf("Unable to send message for "+
+					"peer catch up: %v", err)
+			}
 		}
 	}()
 
diff --git a/discovery/syncer_test.go b/discovery/syncer_test.go
index b9ee24c7e..fd6577d17 100644
--- a/discovery/syncer_test.go
+++ b/discovery/syncer_test.go
@@ -151,6 +151,10 @@ func newTestSyncer(hID lnwire.ShortChannelID,
 			msgChan <- msgs
 			return nil
 		},
+		sendToPeerSync: func(msgs ...lnwire.Message) error {
+			msgChan <- msgs
+			return nil
+		},
 		delayedQueryReplyInterval: 2 * time.Second,
 	}
 	syncer := newGossipSyncer(cfg)
@@ -540,30 +544,37 @@ func TestGossipSyncerReplyShortChanIDs(t *testing.T) {
 		t.Fatalf("unable to query for chan IDs: %v", err)
 	}
 
-	select {
-	case <-time.After(time.Second * 15):
-		t.Fatalf("no msgs received")
+	for i := 0; i < len(queryReply)+1; i++ {
+		select {
+		case <-time.After(time.Second * 15):
+			t.Fatalf("no msgs received")
 
-		// We should get back exactly 4 messages. The first 3 are the same
-		// messages we sent above, and the query end message.
-	case msgs := <-msgChan:
-		if len(msgs) != 4 {
-			t.Fatalf("wrong messages: expected %v, got %v",
-				4, len(msgs))
-		}
+		// We should get back exactly 4 messages. The first 3 are the
+		// same messages we sent above, and the query end message.
+		case msgs := <-msgChan:
+			if len(msgs) != 1 {
+				t.Fatalf("wrong number of messages: "+
+					"expected %v, got %v", 1, len(msgs))
+			}
 
-		if !reflect.DeepEqual(queryReply, msgs[:3]) {
-			t.Fatalf("wrong set of messages: expected %v, got %v",
-				spew.Sdump(queryReply), spew.Sdump(msgs[:3]))
-		}
+			isQueryReply := i < len(queryReply)
+			finalMsg, ok := msgs[0].(*lnwire.ReplyShortChanIDsEnd)
 
-		finalMsg, ok := msgs[3].(*lnwire.ReplyShortChanIDsEnd)
-		if !ok {
-			t.Fatalf("expected lnwire.ReplyShortChanIDsEnd "+
-				"instead got %T", msgs[3])
-		}
-		if finalMsg.Complete != 1 {
-			t.Fatalf("complete wasn't set")
+			switch {
+			case isQueryReply &&
+				!reflect.DeepEqual(queryReply[i], msgs[0]):
+
+				t.Fatalf("wrong message: expected %v, got %v",
+					spew.Sdump(queryReply[i]),
+					spew.Sdump(msgs[0]))
+
+			case !isQueryReply && !ok:
+				t.Fatalf("expected lnwire.ReplyShortChanIDsEnd"+
+					" instead got %T", msgs[3])
+
+			case !isQueryReply && finalMsg.Complete != 1:
+				t.Fatalf("complete wasn't set")
+			}
 		}
 	}
 }

From 5251ebe1173582c71ac3e592caf942e010830ccb Mon Sep 17 00:00:00 2001
From: Conner Fromknecht <conner@lightning.engineering>
Date: Fri, 26 Apr 2019 20:05:23 -0700
Subject: [PATCH 4/4] discovery/syncer_test: fix benign off-by-one in DelayDOS
 test

Prior to this change, the numQueryResponses that we calculated would be
one more than what we actually wanted since it didn't account for the
initial QueryChannelRange msg. This resulted in the test sending one
extra delayed query than was configured. This doesn't fundamentally
impact the test, but does make what happens in the test more reflective
of the configuration.
---
 discovery/syncer_test.go | 13 ++++++++-----
 1 file changed, 8 insertions(+), 5 deletions(-)

diff --git a/discovery/syncer_test.go b/discovery/syncer_test.go
index fd6577d17..68c4a8c92 100644
--- a/discovery/syncer_test.go
+++ b/discovery/syncer_test.go
@@ -1070,8 +1070,10 @@ func TestGossipSyncerDelayDOS(t *testing.T) {
 	numUndelayedQueries := syncer1.cfg.maxUndelayedQueryReplies
 
 	// We will send enough queries to exhaust the undelayed responses, and
-	// then send two more queries which should be delayed.
-	numQueryResponses := numUndelayedQueries + numDelayedQueries
+	// then send two more queries which should be delayed. An additional one
+	// is subtracted from the total since undelayed message will be consumed
+	// by the initial QueryChannelRange.
+	numQueryResponses := numUndelayedQueries + numDelayedQueries - 1
 
 	// The total number of responses must include the initial reply each
 	// syncer will make to QueryChannelRange.
@@ -1081,10 +1083,11 @@ func TestGossipSyncerDelayDOS(t *testing.T) {
 	// scaled by the chunk size being used.
 	numTotalChans := numQueryResponses * chunkSize
 
-	// Although both nodes are at the same height, they'll have a
-	// completely disjoint set of chan ID's that they know of.
+	// Construct enough channels so that all of the queries will have enough
+	// channels. Since syncer1 won't know of any channels, their sets are
+	// inherently disjoint.
 	var syncer2Chans []lnwire.ShortChannelID
-	for i := numTotalChans; i < numTotalChans+numTotalChans; i++ {
+	for i := 0; i < numTotalChans; i++ {
 		syncer2Chans = append(
 			syncer2Chans, lnwire.NewShortChanIDFromInt(uint64(i)),
 		)