Conditionally queue block bundles in expected responses and dequeue them in _handleInExpectedResponse. don't queue msg in _cleanupDonePeerState if there are no transactions. also exit early if peer can't accept inv messages

lib/peer.go

@@ -844,13 +844,24 @@ func (pp *Peer) _handleOutExpectedResponse(msg DeSoMessage) {
 	switch msg.GetMsgType() {
 	case MsgTypeGetBlocks:
 		getBlocks := msg.(*MsgDeSoGetBlocks)
-		// We have one block expected for each entry in the message.
-		for ii := range getBlocks.HashList {
+		if pp.NegotiatedProtocolVersion >= ProtocolVersion2 {
+			// TODO: What if the amount of blocks exceeds a single MsgTypeBlockBundle response.
+			// Note there isn't really a way in the current code for us to request more
+			// blocks in a single bundle that we should expect in a single response, so
+			// we should be ok.
 			pp._addExpectedResponse(&ExpectedResponse{
-				TimeExpected: time.Now().Add(
-					stallTimeout + time.Duration(int64(ii)*int64(stallTimeout))),
-				MessageType: MsgTypeBlock,
+				TimeExpected: time.Now().Add(stallTimeout),
+				MessageType:  MsgTypeBlockBundle,
 			})
+		} else {
+			// We have one block expected for each entry in the message.
+			for ii := range getBlocks.HashList {
+				pp._addExpectedResponse(&ExpectedResponse{
+					TimeExpected: time.Now().Add(
+						stallTimeout + time.Duration(int64(ii)*int64(stallTimeout))),
+					MessageType: MsgTypeBlock,
+				})
+			}
 		}
 	case MsgTypeGetHeaders:
 		// If we're sending a GetHeaders message, the Peer should respond within
@@ -1121,6 +1132,7 @@ func (pp *Peer) _handleInExpectedResponse(rmsg DeSoMessage) error {
 	// Do this in a separate switch to keep things clean.
 	msgType := rmsg.GetMsgType()
 	if msgType == MsgTypeBlock ||
+		msgType == MsgTypeBlockBundle ||
 		msgType == MsgTypeHeaderBundle ||
 		msgType == MsgTypeTransactionBundle ||
 		msgType == MsgTypeTransactionBundleV2 ||

lib/server.go

@@ -1852,10 +1852,10 @@ func (srv *Server) _cleanupDonePeerState(pp *Peer) {
 	// TODO: Sending a sync/mempool message to a random Peer periodically seems like it would
 	// be a good way to fill any gaps.
 	newPeer := srv.cmgr.RandomPeer()
-	if newPeer == nil {
-		// If we don't have a new Peer, remove everything that was destined for
-		// this Peer. Note we don't need to copy the iterator because everything
-		// below doesn't take a reference to it.
+	if newPeer == nil || !newPeer.canReceiveInvMessages {
+		// If we don't have a new Peer or the new peer can't receive INV messages,
+		// remove everything that was destined for this Peer. Note we don't need to
+		// copy the iterator because everything below doesn't take a reference to it.
 		for hashIter, requestInfo := range srv.requestedTransactionsMap {
 			hash := hashIter
 			if requestInfo.PeerWhoSentInv.ID == pp.ID {
@@ -1888,11 +1888,13 @@ func (srv *Server) _cleanupDonePeerState(pp *Peer) {
 		requestInfo.TimeRequested = time.Now()
 		txnHashesReassigned = append(txnHashesReassigned, hashCopy)
 	}
-	// Request any hashes we might have reassigned in a goroutine to keep things
-	// moving.
-	newPeer.AddDeSoMessage(&MsgDeSoGetTransactions{
-		HashList: txnHashesReassigned,
-	}, false)
+	if len(txnHashesReassigned) > 0 {
+		// Request any hashes we might have reassigned in a goroutine to keep things
+		// moving.
+		newPeer.AddDeSoMessage(&MsgDeSoGetTransactions{
+			HashList: txnHashesReassigned,
+		}, false)
+	}
 }
 
 func (srv *Server) _handleDisconnectedPeerMessage(pp *Peer) {