Merge pull request #1652 from lightninglabs/msg-router-non-blocking

server+tapchannel: fix deadlock potential, queue msgs

Author: guggero

server.go

@@ -862,12 +862,16 @@ func (s *Server) Name() msgmux.EndpointName {
 //
 // NOTE: This method is part of the msgmux.MsgEndpoint interface.
 func (s *Server) CanHandle(msg msgmux.PeerMsg) bool {
-	err := s.waitForReady()
-	if err != nil {
-		srvrLog.Debugf("Can't handle PeerMsg, server not ready %v",
-			err)
+	// We can't wait for ready here, as this method is potentially called
+	// during startup. The `CanHandle` method is stateless, so we can call
+	// it if the funding controller has been created (but potentially has
+	// not yet been started).
+	if s == nil || s.cfg == nil || s.cfg.AuxFundingController == nil {
+		// This shouldn't happen, the server and funding controller
+		// should always be initialized before the msgmux is started.
 		return false
 	}
+
 	return s.cfg.AuxFundingController.CanHandle(msg)
 }
 
@@ -876,12 +880,18 @@ func (s *Server) CanHandle(msg msgmux.PeerMsg) bool {
 //
 // NOTE: This method is part of the msgmux.MsgEndpoint interface.
 func (s *Server) SendMessage(ctx context.Context, msg msgmux.PeerMsg) bool {
-	err := s.waitForReady()
-	if err != nil {
-		srvrLog.Debugf("Failed to send PeerMsg, server not ready %v",
-			err)
+	// We can't wait for ready here, as this method is potentially called
+	// during startup. The `SendMessage` method will buffer messages that
+	// come in between the funding controller being created and it being
+	// started (which only happens after waitForReady fires). So it's safe
+	// to call it here, as long as the funding controller has been created.
+	if s == nil || s.cfg == nil || s.cfg.AuxFundingController == nil {
+		// This shouldn't happen, the CanHandle method is always called
+		// first, and that should've already returned false in this
+		// case.
 		return false
 	}
+
 	return s.cfg.AuxFundingController.SendMessage(ctx, msg)
 }
 

tapchannel/aux_funding_controller.go

@@ -301,7 +301,7 @@ type FundingController struct {
 
 	cfg FundingControllerCfg
 
-	msgs chan msgmux.PeerMsg
+	msgQueue *fn.ConcurrentQueue[msgmux.PeerMsg]
 
 	bindFundingReqs chan *bindFundingReq
 
@@ -318,9 +318,11 @@ type FundingController struct {
 
 // NewFundingController creates a new instance of the FundingController.
 func NewFundingController(cfg FundingControllerCfg) *FundingController {
+	msgs := fn.NewConcurrentQueue[msgmux.PeerMsg](fn.DefaultQueueSize)
+	msgs.Start()
 	return &FundingController{
 		cfg:             cfg,
-		msgs:            make(chan msgmux.PeerMsg, 10),
+		msgQueue:        msgs,
 		bindFundingReqs: make(chan *bindFundingReq, 10),
 		newFundingReqs:  make(chan *FundReq, 10),
 		rootReqs:        make(chan *assetRootReq, 10),
@@ -387,6 +389,8 @@ func (f *FundingController) Stop() error {
 	close(f.Quit)
 	f.Wg.Wait()
 
+	f.msgQueue.Stop()
+
 	return nil
 }
 
@@ -1858,7 +1862,7 @@ func (f *FundingController) chanFunder() {
 		// The remote party has sent us some upfront proof for channel
 		// asset inputs. We'll log this pending chan ID, then validate
 		// the proofs included.
-		case msg := <-f.msgs:
+		case msg := <-f.msgQueue.ChanOut():
 			tempFundingID, err := f.processFundingMsg(
 				ctxc, fundingFlows, msg,
 			)
@@ -2438,7 +2442,7 @@ func (f *FundingController) CanHandle(msg msgmux.PeerMsg) bool {
 func (f *FundingController) SendMessage(_ context.Context,
 	msg msgmux.PeerMsg) bool {
 
-	return fn.SendOrQuit(f.msgs, msg, f.Quit)
+	return fn.SendOrQuit(f.msgQueue.ChanIn(), msg, f.Quit)
 }
 
 // TODO(roasbeef): try to protofsm it?