server+tapchannel: fix deadlock potential, queue msgs

This commit fixes a potential deadlock when running in conjunction with
lnd/litd. On startup, if a custom message needs to be sent by an lnd
component as part of its initialization, it would previously run into
the waitForReady method in tapd and wait. But that would block lnd from
fully starting up, meaning that waitForReady would never return.

We fix this by queuing up incoming messages so we can handle them
asynchronously once we've fully started up.
This is safe as long as the server and funding controller have been
created, even if not fully started. The queued messages will be
processed as soon as the funding controller's Start method is called.

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?