multi: link quit can interrupt commitment signing

In this commit, we make sig job handling when singing a next commitment
non-blocking by allowing the shutdown of a channel link to prevent
further waiting on sig jobs by the channel state machine. This addresses
possible cases where the aux signer may be shut down via a separate quit
signal, so the state machine could block indefinitely on receiving an
update on a sig job.

htlcswitch/link.go

@@ -387,8 +387,10 @@ type channelLink struct {
 	// our next CommitSig.
 	incomingCommitHooks hookMap
 
-	wg   sync.WaitGroup
-	quit chan struct{}
+	// ContextGuard is a helper that encapsulates a wait group and quit
+	// channel and allows contexts that either block or cancel on those
+	// depending on the use case.
+	*fn.ContextGuard
 }
 
 // hookMap is a data structure that is used to track the hooks that need to be
@@ -469,7 +471,7 @@ func NewChannelLink(cfg ChannelLinkConfig,
 		flushHooks:          newHookMap(),
 		outgoingCommitHooks: newHookMap(),
 		incomingCommitHooks: newHookMap(),
-		quit:                make(chan struct{}),
+		ContextGuard:        fn.NewContextGuard(),
 	}
 }
 
@@ -548,7 +550,7 @@ func (l *channelLink) Start() error {
 
 	l.updateFeeTimer = time.NewTimer(l.randomFeeUpdateTimeout())
 
-	l.wg.Add(1)
+	l.Wg.Add(1)
 	go l.htlcManager()
 
 	return nil
@@ -588,8 +590,8 @@ func (l *channelLink) Stop() {
 		l.hodlQueue.Stop()
 	}
 
-	close(l.quit)
-	l.wg.Wait()
+	close(l.Quit)
+	l.Wg.Wait()
 
 	// Now that the htlcManager has completely exited, reset the packet
 	// courier. This allows the mailbox to revaluate any lingering Adds that
@@ -614,7 +616,7 @@ func (l *channelLink) Stop() {
 // WaitForShutdown blocks until the link finishes shutting down, which includes
 // termination of all dependent goroutines.
 func (l *channelLink) WaitForShutdown() {
-	l.wg.Wait()
+	l.Wg.Wait()
 }
 
 // EligibleToForward returns a bool indicating if the channel is able to
@@ -675,7 +677,7 @@ func (l *channelLink) IsFlushing(linkDirection LinkDirection) bool {
 func (l *channelLink) OnFlushedOnce(hook func()) {
 	select {
 	case l.flushHooks.newTransients <- hook:
-	case <-l.quit:
+	case <-l.Quit:
 	}
 }
 
@@ -694,7 +696,7 @@ func (l *channelLink) OnCommitOnce(direction LinkDirection, hook func()) {
 
 	select {
 	case queue <- hook:
-	case <-l.quit:
+	case <-l.Quit:
 	}
 }
 
@@ -903,8 +905,10 @@ func (l *channelLink) syncChanStates() error {
 		// We've just received a ChanSync message from the remote
 		// party, so we'll process the message  in order to determine
 		// if we need to re-transmit any messages to the remote party.
+		ctx, cancel := l.WithCtxQuitNoTimeout()
+		defer cancel()
 		msgsToReSend, openedCircuits, closedCircuits, err =
-			l.channel.ProcessChanSyncMsg(remoteChanSyncMsg)
+			l.channel.ProcessChanSyncMsg(ctx, remoteChanSyncMsg)
 		if err != nil {
 			return err
 		}
@@ -933,7 +937,7 @@ func (l *channelLink) syncChanStates() error {
 			l.cfg.Peer.SendMessage(false, msg)
 		}
 
-	case <-l.quit:
+	case <-l.Quit:
 		return ErrLinkShuttingDown
 	}
 
@@ -1023,7 +1027,7 @@ func (l *channelLink) resolveFwdPkg(fwdPkg *channeldb.FwdPkg) error {
 //
 // NOTE: This MUST be run as a goroutine.
 func (l *channelLink) fwdPkgGarbager() {
-	defer l.wg.Done()
+	defer l.Wg.Done()
 
 	l.cfg.FwdPkgGCTicker.Resume()
 	defer l.cfg.FwdPkgGCTicker.Stop()
@@ -1040,7 +1044,7 @@ func (l *channelLink) fwdPkgGarbager() {
 					err)
 				continue
 			}
-		case <-l.quit:
+		case <-l.Quit:
 			return
 		}
 	}
@@ -1163,7 +1167,7 @@ func (l *channelLink) handleChanSyncErr(err error) {
 func (l *channelLink) htlcManager() {
 	defer func() {
 		l.cfg.BatchTicker.Stop()
-		l.wg.Done()
+		l.Wg.Done()
 		l.log.Infof("exited")
 	}()
 
@@ -1257,7 +1261,7 @@ func (l *channelLink) htlcManager() {
 		// With our link's in-memory state fully reconstructed, spawn a
 		// goroutine to manage the reclamation of disk space occupied by
 		// completed forwarding packages.
-		l.wg.Add(1)
+		l.Wg.Add(1)
 		go l.fwdPkgGarbager()
 	}
 
@@ -1441,7 +1445,7 @@ func (l *channelLink) htlcManager() {
 				)
 			}
 
-		case <-l.quit:
+		case <-l.Quit:
 			return
 		}
 	}
@@ -2299,7 +2303,7 @@ func (l *channelLink) handleUpstreamMsg(msg lnwire.Message) {
 		}
 
 		select {
-		case <-l.quit:
+		case <-l.Quit:
 			return
 		default:
 		}
@@ -2360,7 +2364,7 @@ func (l *channelLink) handleUpstreamMsg(msg lnwire.Message) {
 		}
 
 		select {
-		case <-l.quit:
+		case <-l.Quit:
 			return
 		default:
 		}
@@ -2590,7 +2594,10 @@ func (l *channelLink) updateCommitTx() error {
 		return nil
 	}
 
-	newCommit, err := l.channel.SignNextCommitment()
+	ctx, done := l.WithCtxQuitNoTimeout()
+	defer done()
+
+	newCommit, err := l.channel.SignNextCommitment(ctx)
 	if err == lnwallet.ErrNoWindow {
 		l.cfg.PendingCommitTicker.Resume()
 		l.log.Trace("PendingCommitTicker resumed")
@@ -2627,7 +2634,7 @@ func (l *channelLink) updateCommitTx() error {
 	}
 
 	select {
-	case <-l.quit:
+	case <-l.Quit:
 		return ErrLinkShuttingDown
 	default:
 	}
@@ -3233,7 +3240,7 @@ func (l *channelLink) handleSwitchPacket(pkt *htlcPacket) error {
 // NOTE: Part of the ChannelLink interface.
 func (l *channelLink) HandleChannelUpdate(message lnwire.Message) {
 	select {
-	case <-l.quit:
+	case <-l.Quit:
 		// Return early if the link is already in the process of
 		// quitting. It doesn't make sense to hand the message to the
 		// mailbox here.
@@ -3932,7 +3939,7 @@ func (l *channelLink) forwardBatch(replay bool, packets ...*htlcPacket) {
 		filteredPkts = append(filteredPkts, pkt)
 	}
 
-	err := l.cfg.ForwardPackets(l.quit, replay, filteredPkts...)
+	err := l.cfg.ForwardPackets(l.Quit, replay, filteredPkts...)
 	if err != nil {
 		log.Errorf("Unhandled error while reforwarding htlc "+
 			"settle/fail over htlcswitch: %v", err)

htlcswitch/link_isolated_test.go

@@ -1,6 +1,7 @@
 package htlcswitch
 
 import (
+	"context"
 	"crypto/sha256"
 	"testing"
 	"time"
@@ -94,7 +95,9 @@ func (l *linkTestContext) receiveHtlcAliceToBob() {
 func (l *linkTestContext) sendCommitSigBobToAlice(expHtlcs int) {
 	l.t.Helper()
 
-	sigs, err := l.bobChannel.SignNextCommitment()
+	testQuit, testQuitFunc := context.WithCancel(context.Background())
+	defer testQuitFunc()
+	sigs, err := l.bobChannel.SignNextCommitment(testQuit)
 	if err != nil {
 		l.t.Fatalf("error signing commitment: %v", err)
 	}

htlcswitch/link_test.go

@@ -2252,12 +2252,14 @@ func newSingleLinkTestHarness(t *testing.T, chanAmt,
 		return aliceSwitch.AddLink(aliceLink)
 	}
 	go func() {
-		for {
-			select {
-			case <-notifyUpdateChan:
-			case <-aliceLink.(*channelLink).quit:
-				close(doneChan)
-				return
+		if chanLink, ok := aliceLink.(*channelLink); ok {
+			for {
+				select {
+				case <-notifyUpdateChan:
+				case <-chanLink.Quit:
+					close(doneChan)
+					return
+				}
 			}
 		}
 	}()
@@ -2324,7 +2326,10 @@ func handleStateUpdate(link *channelLink,
 	}
 	link.HandleChannelUpdate(remoteRev)
 
-	remoteSigs, err := remoteChannel.SignNextCommitment()
+	ctx, done := link.WithCtxQuitNoTimeout()
+	defer done()
+
+	remoteSigs, err := remoteChannel.SignNextCommitment(ctx)
 	if err != nil {
 		return err
 	}
@@ -2367,7 +2372,7 @@ func updateState(batchTick chan time.Time, link *channelLink,
 		// Trigger update by ticking the batchTicker.
 		select {
 		case batchTick <- time.Now():
-		case <-link.quit:
+		case <-link.Quit:
 			return fmt.Errorf("link shutting down")
 		}
 		return handleStateUpdate(link, remoteChannel)
@@ -2375,7 +2380,10 @@ func updateState(batchTick chan time.Time, link *channelLink,
 
 	// The remote is triggering the state update, emulate this by
 	// signing and sending CommitSig to the link.
-	remoteSigs, err := remoteChannel.SignNextCommitment()
+	ctx, done := link.WithCtxQuitNoTimeout()
+	defer done()
+
+	remoteSigs, err := remoteChannel.SignNextCommitment(ctx)
 	if err != nil {
 		return err
 	}
@@ -4931,12 +4939,14 @@ func (h *persistentLinkHarness) restartLink(
 		return nil, nil, err
 	}
 	go func() {
-		for {
-			select {
-			case <-notifyUpdateChan:
-			case <-aliceLink.(*channelLink).quit:
-				close(doneChan)
-				return
+		if chanLink, ok := aliceLink.(*channelLink); ok {
+			for {
+				select {
+				case <-notifyUpdateChan:
+				case <-chanLink.Quit:
+					close(doneChan)
+					return
+				}
 			}
 		}
 	}()
@@ -5919,7 +5929,12 @@ func TestChannelLinkFail(t *testing.T) {
 
 				// Sign a commitment that will include
 				// signature for the HTLC just sent.
-				sigs, err := remoteChannel.SignNextCommitment()
+				quitCtx, done := c.WithCtxQuitNoTimeout()
+				defer done()
+
+				sigs, err := remoteChannel.SignNextCommitment(
+					quitCtx,
+				)
 				if err != nil {
 					t.Fatalf("error signing commitment: %v",
 						err)
@@ -5961,7 +5976,12 @@ func TestChannelLinkFail(t *testing.T) {
 
 				// Sign a commitment that will include
 				// signature for the HTLC just sent.
-				sigs, err := remoteChannel.SignNextCommitment()
+				quitCtx, done := c.WithCtxQuitNoTimeout()
+				defer done()
+
+				sigs, err := remoteChannel.SignNextCommitment(
+					quitCtx,
+				)
 				if err != nil {
 					t.Fatalf("error signing commitment: %v",
 						err)

htlcswitch/test_utils.go

@@ -1197,12 +1197,14 @@ func (h *hopNetwork) createChannelLink(server, peer *mockServer,
 	}
 
 	go func() {
-		for {
-			select {
-			case <-notifyUpdateChan:
-			case <-link.(*channelLink).quit:
-				close(doneChan)
-				return
+		if chanLink, ok := link.(*channelLink); ok {
+			for {
+				select {
+				case <-notifyUpdateChan:
+				case <-chanLink.Quit:
+					close(doneChan)
+					return
+				}
 			}
 		}
 	}()