tapchannel: improve quit handling for aux signer

In this commit, we add checks of the aux signer cancel and quit signals
at all points during aux sig batch processing when a response may be
sent. This mirrors the signal handling used in the lnwallet sigpool
worker goroutine. We also add an extra check for the job cancel signal,
since it could have been sent by LND before it could be sent from here.

tapchannel/aux_leaf_signer.go

@@ -28,6 +28,9 @@ import (
 	"github.com/lightningnetwork/lnd/tlv"
 )
 
+// shutdownErr is used in multiple spots when exiting the sig batch processor.
+var shutdownErr = fmt.Errorf("tapd is shutting down")
+
 // VirtualPacketSigner is an interface that can be used to sign virtual packets.
 type VirtualPacketSigner interface {
 	// SignVirtualPacket signs the virtual transaction of the given packet
@@ -241,35 +244,50 @@ func (s *AuxLeafSigner) processAuxSigBatch(chanState *channeldb.OpenChannel,
 	defer s.Wg.Done()
 
 	log.Tracef("Processing %d aux sig jobs", len(sigJobs))
-
 	for idx := range sigJobs {
 		sigJob := sigJobs[idx]
 		cancelAndErr := func(err error) {
 			log.Errorf("Error processing aux sig job: %v", err)
 
-			close(sigJob.Cancel)
+			// Check that the cancel signal was not already sent
+			// before cancelling all jobs. The cancel signal must
+			// only be sent exactly once.
+			select {
+			case <-sigJob.Cancel:
+			default:
+				close(sigJob.Cancel)
+			}
+
 			sigJob.Resp <- lnwallet.AuxSigJobResp{
 				Err: err,
 			}
 		}
 
-		// If we're shutting down, we cancel the job and return.
+		// Check for cancel or quit signals before beginning the job.
 		select {
+		case <-sigJob.Cancel:
+			continue
 		case <-s.Quit:
-			cancelAndErr(fmt.Errorf("tapd is shutting down"))
+			cancelAndErr(shutdownErr)
 			return
-
 		default:
 		}
 
 		// If there is no commit blob, this isn't a custom channel. We
 		// still need to signal the job as done though, even if we don't
 		// have a signature to return.
 		if sigJob.CommitBlob.IsNone() {
-			sigJob.Resp <- lnwallet.AuxSigJobResp{
+			select {
+			case sigJob.Resp <- lnwallet.AuxSigJobResp{
 				HtlcIndex: sigJob.HTLC.HtlcIndex,
+			}:
+				continue
+			case <-sigJob.Cancel:
+				continue
+			case <-s.Quit:
+				cancelAndErr(shutdownErr)
+				return
 			}
-			continue
 		}
 
 		com, err := cmsg.DecodeCommitment(
@@ -299,10 +317,17 @@ func (s *AuxLeafSigner) processAuxSigBatch(chanState *channeldb.OpenChannel,
 		// If the HTLC doesn't have any asset outputs, it's not an
 		// asset HTLC, so we can skip it.
 		if len(htlcOutputs) == 0 {
-			sigJob.Resp <- lnwallet.AuxSigJobResp{
+			select {
+			case sigJob.Resp <- lnwallet.AuxSigJobResp{
 				HtlcIndex: sigJob.HTLC.HtlcIndex,
+			}:
+				continue
+			case <-sigJob.Cancel:
+				continue
+			case <-s.Quit:
+				cancelAndErr(shutdownErr)
+				return
 			}
-			continue
 		}
 
 		resp, err := s.generateHtlcSignature(
@@ -318,7 +343,15 @@ func (s *AuxLeafSigner) processAuxSigBatch(chanState *channeldb.OpenChannel,
 		// Success!
 		log.Tracef("Generated HTLC signature for HTLC with index %d",
 			sigJob.HTLC.HtlcIndex)
-		sigJob.Resp <- resp
+
+		select {
+		case sigJob.Resp <- resp:
+		case <-sigJob.Cancel:
+			continue
+		case <-s.Quit:
+			cancelAndErr(shutdownErr)
+			return
+		}
 	}
 }