Just dont shut down the queues on Message sender errors

Author: hsanjuan

bitswap/client/internal/messagequeue/messagequeue.go

@@ -562,7 +562,7 @@ func (mq *MessageQueue) sendMessage() {
 		// If we fail to initialize the sender, the networking layer will
 		// emit a Disconnect event and the MessageQueue will get cleaned up
 		log.Infof("Could not open message sender to peer %s: %s", mq.p, err)
-		mq.Shutdown()
+		// do not shudown the queue here, wait for Disconnect to arrive.
 		return
 	}
 
@@ -593,7 +593,7 @@ func (mq *MessageQueue) sendMessage() {
 			// If the message couldn't be sent, the networking layer will
 			// emit a Disconnect event and the MessageQueue will get cleaned up
 			log.Infof("Could not send message to peer %s: %s", mq.p, err)
-			mq.Shutdown()
+			// do not shudown the queue here, wait for Disconnect to arrive.
 			return
 		}
 

bitswap/network/bsnet/ipfs_impl.go

@@ -244,6 +244,7 @@ func (s *streamMessageSender) send(ctx context.Context, msg bsmsg.BitSwapMessage
 	stream, err := s.Connect(ctx)
 	if err != nil {
 		log.Infof("failed to open stream to %s: %s", s.to, err)
+		s.bsnet.connectEvtMgr.MarkUnresponsive(s.to)
 		return err
 	}
 
@@ -253,6 +254,7 @@ func (s *streamMessageSender) send(ctx context.Context, msg bsmsg.BitSwapMessage
 	timeout := s.opts.SendTimeout - time.Since(start)
 	if err = s.bsnet.msgToStream(ctx, stream, msg, timeout); err != nil {
 		log.Infof("failed to send message to %s: %s", s.to, err)
+		s.bsnet.connectEvtMgr.MarkUnresponsive(s.to)
 		return err
 	}