terminal+rpcmiddleware: add main err queue

Author: guggero

rpcmiddleware/manager.go

@@ -14,23 +14,23 @@ type Manager struct {
 	lndClient        lndclient.LightningClient
 	interceptors     []RequestInterceptor
 
-	errChan  chan error
-	wg       sync.WaitGroup
-	cancel   context.CancelFunc
-	quit     chan struct{}
-	stopOnce sync.Once
+	mainErrChan chan<- error
+	wg          sync.WaitGroup
+	cancel      context.CancelFunc
+	quit        chan struct{}
+	stopOnce    sync.Once
 }
 
 // NewManager returns a new middleware manager.
 func NewManager(interceptTimeout time.Duration,
-	lndClient lndclient.LightningClient,
+	lndClient lndclient.LightningClient, errChan chan<- error,
 	interceptors ...RequestInterceptor) *Manager {
 
 	return &Manager{
 		interceptTimeout: interceptTimeout,
 		lndClient:        lndClient,
 		interceptors:     interceptors,
-		errChan:          make(chan error),
+		mainErrChan:      errChan,
 		quit:             make(chan struct{}),
 	}
 }
@@ -74,7 +74,7 @@ func (f *Manager) Start() error {
 						err)
 
 					select {
-					case f.errChan <- err:
+					case f.mainErrChan <- err:
 					case <-f.quit:
 					case <-ctxc.Done():
 					}
@@ -88,12 +88,6 @@ func (f *Manager) Start() error {
 	return nil
 }
 
-// Errors returns a channel on which any errors the manager encounters will be
-// delivered.
-func (f *Manager) Errors() chan error {
-	return f.errChan
-}
-
 // Stop shuts down the middleware manager.
 func (f *Manager) Stop() {
 	f.stopOnce.Do(func() {

terminal.go

@@ -142,8 +142,8 @@ type LightningTerminal struct {
 	// guards all incoming calls. This is only set in integrated mode!
 	lndInterceptorChain *rpcperms.InterceptorChain
 
-	wg         sync.WaitGroup
-	lndErrChan chan error
+	wg       sync.WaitGroup
+	errQueue *queue.ConcurrentQueue[error]
 
 	lndClient   *lndclient.GrpcLndServices
 	basicClient lnrpc.LightningClient
@@ -175,9 +175,7 @@ type LightningTerminal struct {
 
 // New creates a new instance of the lightning-terminal daemon.
 func New() *LightningTerminal {
-	return &LightningTerminal{
-		lndErrChan: make(chan error, 1),
-	}
+	return &LightningTerminal{}
 }
 
 // Run starts everything and then blocks until either the application is shut
@@ -202,9 +200,9 @@ func (g *LightningTerminal) Run() error {
 	// This concurrent error queue can be used by every component that can
 	// raise runtime errors. Using a queue will prevent us from blocking on
 	// sending errors to it, as long as the queue is running.
-	errQueue := queue.NewConcurrentQueue[error](queue.DefaultQueueSize)
-	errQueue.Start()
-	defer errQueue.Stop()
+	g.errQueue = queue.NewConcurrentQueue[error](queue.DefaultQueueSize)
+	g.errQueue.Start()
+	defer g.errQueue.Stop()
 
 	// Construct a new PermissionsManager.
 	g.permsMgr, err = NewPermissionsManager()
@@ -268,6 +266,7 @@ func (g *LightningTerminal) Run() error {
 	readyChan := make(chan struct{})
 	bufReadyChan := make(chan struct{})
 	unlockChan := make(chan struct{})
+	lndQuit := make(chan struct{})
 	macChan := make(chan []byte, 1)
 
 	if g.cfg.LndMode == ModeIntegrated {
@@ -304,11 +303,11 @@ func (g *LightningTerminal) Run() error {
 				(!ok || e.Type != flags.ErrHelp) {
 
 				log.Errorf("Error running main lnd: %v", err)
-				g.lndErrChan <- err
+				g.errQueue.ChanIn() <- err
 				return
 			}
 
-			close(g.lndErrChan)
+			close(lndQuit)
 		}()
 	} else {
 		close(unlockChan)
@@ -337,9 +336,12 @@ func (g *LightningTerminal) Run() error {
 	// get the ready signal immediately.
 	case <-readyChan:
 
-	case err := <-g.lndErrChan:
+	case err := <-g.errQueue.ChanOut():
 		return err
 
+	case <-lndQuit:
+		return nil
+
 	case <-shutdownInterceptor.ShutdownChannel():
 		return errors.New("shutting down")
 	}
@@ -375,9 +377,12 @@ func (g *LightningTerminal) Run() error {
 	select {
 	case <-readyChan:
 
-	case err := <-g.lndErrChan:
+	case err := <-g.errQueue.ChanOut():
 		return err
 
+	case <-lndQuit:
+		return nil
+
 	case <-shutdownInterceptor.ShutdownChannel():
 		return errors.New("shutting down")
 	}
@@ -404,12 +409,15 @@ func (g *LightningTerminal) Run() error {
 		log.Errorf("Received critical error from loop, shutting down: "+
 			"%v", err)
 
-	case err := <-g.lndErrChan:
+	case err := <-g.errQueue.ChanOut():
 		if err != nil {
-			log.Errorf("Received critical error from lnd, "+
+			log.Errorf("Received critical error from subsystem, "+
 				"shutting down: %v", err)
 		}
 
+	case <-lndQuit:
+		return nil
+
 	case <-shutdownInterceptor.ShutdownChannel():
 		log.Infof("Shutdown signal received")
 	}
@@ -607,7 +615,7 @@ func (g *LightningTerminal) startSubservers() error {
 		// Start the middleware manager.
 		g.middleware = mid.NewManager(
 			g.cfg.RPCMiddleware.InterceptTimeout,
-			g.lndClient.Client,
+			g.lndClient.Client, g.errQueue.ChanIn(),
 		)
 
 		if err = g.middleware.Start(); err != nil {
@@ -935,15 +943,25 @@ func (g *LightningTerminal) shutdown() error {
 
 	g.wg.Wait()
 
-	// The lnd error channel is only used if we are actually running lnd in
-	// the same process.
-	if g.cfg.LndMode == ModeIntegrated {
-		err := <-g.lndErrChan
-		if err != nil {
-			log.Errorf("Error stopping lnd: %v", err)
-			returnErr = err
+	// Do we have any last errors to display? We use an anonymous function,
+	// so we can use return instead of breaking to a label in the default
+	// case.
+	func() {
+		for {
+			select {
+			case err := <-g.errQueue.ChanOut():
+				if err != nil {
+					log.Errorf("Error while stopping "+
+						"litd: %v", err)
+
+					returnErr = err
+				}
+
+			default:
+				return
+			}
 		}
-	}
+	}()
 
 	return returnErr
 }