peer+lnwallet/chancloser: advance the legacy closer from one goroutine

In this commit, we give the legacy ChanCloser a single owner, rather than
letting two goroutines advance it. The peer's channelManager drives the state
machine for the Shutdown and ClosingSigned messages that come off the wire, and
for local close requests. The link drives it as well: while we wait for the
channel to drain we register a flush hook, and the link invokes that hook from
its own goroutine, where it called BeginNegotiation directly. Nothing kept the
two apart, so the state field, the priorFeeOffers map, and the signing step
could all be touched at once. Under `go test -race` this shows up as a data race
on the state field.

Rather than reach for a lock, we route the flush through the channelManager. The
hook now only reports the channel ID over a new chanCloseFlushed channel, and
handleChanFlushed picks it up next to the close messages. Every transition, the
cached offer processing, the fee map, and the signing then happen on the one
goroutine, so the closer needs no synchronization of its own. We spell that out
on the type, since it's an invariant a new caller can break from the outside.

The report goes out from a fresh goroutine, which matters more than it looks.
The link may well be holding its own lock while it invokes the hook, and
channelManager reaches for that same lock in DisableAdds, so blocking on the
handoff would trade the race for a deadlock. The `go` in front of RemoveLink
just above it is there for the same reason.

We look the closer up with a plain map load rather than through
fetchActiveChanCloser, as that one builds a fresh closer when it doesn't find
an existing one, and a flush that lands after the negotiation was torn down has
no business starting a new negotiation.

One behavior change falls out of the move: the flush path now runs the same
finalization tail as the message path. It skipped that before, so a responder
that drained a cached offer would reach closeFinished and broadcast, but nothing
ran finalizeChanClosure until the next close message showed up, and having
already sent its final signature, there may not be one. The link == nil path
already ran the tail, so this makes all three paths agree.

The new test drives a close with a link that hands us the flush hook instead of
running it inline, so we can check that negotiation waits on the report, and
that a report for a channel we have no closer for is dropped.

(cherry picked from commit e5e134ddac)

lnwallet/chancloser/chancloser.go

@@ -163,6 +163,12 @@ type ChanCloseCfg struct {
 // procedure. This includes shutting down a channel, marking it ineligible for
 // routing HTLC's, negotiating fees with the remote party, and finally
 // broadcasting the fully signed closure transaction to the network.
+//
+// NOTE: The state machine takes no locks of its own. Nearly every method reads
+// and writes the same fields, so all of them MUST be driven from a single
+// goroutine. In production that's the peer's channelManager, which is the one
+// place the close messages from the wire, the local close requests, and the
+// link's flush notification all meet.
 type ChanCloser struct {
 	// state is the current state of the state machine.
 	state closeState

peer/brontide.go

@@ -674,6 +674,14 @@ type Brontide struct {
 	// well as lnwire.ClosingSigned messages.
 	chanCloseMsgs chan *closeMsg
 
+	// chanCloseFlushed carries the ID of a channel whose link has finished
+	// draining its HTLCs, which is the point a legacy cooperative close can
+	// move on to fee negotiation. The link notices this from its own
+	// goroutine, so it hands the channel over here rather than advance the
+	// closer itself, which keeps every step of the negotiation on the
+	// channelManager goroutine.
+	chanCloseFlushed chan lnwire.ChannelID
+
 	// remoteFeatures is the feature vector received from the peer during
 	// the connection handshake.
 	remoteFeatures *lnwire.FeatureVector
@@ -752,6 +760,7 @@ func NewBrontide(cfg Config) *Brontide {
 		localCloseChanReqs: make(chan *htlcswitch.ChanClose),
 		linkFailures:       make(chan linkFailureReport),
 		chanCloseMsgs:      make(chan *closeMsg),
+		chanCloseFlushed:   make(chan lnwire.ChannelID),
 		resentChanSyncMsg:  make(map[lnwire.ChannelID]struct{}),
 		startReady:         make(chan struct{}),
 		log:                peerLog.WithPrefix(logPrefix),
@@ -3275,6 +3284,11 @@ out:
 		case closeMsg := <-p.chanCloseMsgs:
 			p.handleCloseMsg(closeMsg)
 
+		// A link has finished draining the HTLCs from a channel we're
+		// cooperatively closing, so we can now start fee negotiation.
+		case cid := <-p.chanCloseFlushed:
+			p.handleChanFlushed(cid)
+
 		// The channel reannounce delay has elapsed, broadcast the
 		// reenabled channel updates to the network. This should only
 		// fire once, so we set the reenableTimeout channel to nil to
@@ -5302,23 +5316,7 @@ func (p *Brontide) handleCloseMsg(msg *closeMsg) {
 		chanCloser = c
 	})
 
-	handleErr := func(err error) {
-		err = fmt.Errorf("unable to process close msg: %w", err)
-		p.log.Error(err)
-
-		// As the negotiations failed, we'll reset the channel state
-		// machine to ensure we act to on-chain events as normal.
-		chanCloser.Channel().ResetState()
-		if chanCloser.CloseRequest() != nil {
-			chanCloser.CloseRequest().Err <- err
-		}
-
-		p.deleteActiveChanCloser(
-			msg.cid, chanCloser.Channel().ChannelPoint(),
-		)
-
-		p.Disconnect(err)
-	}
+	handleErr := p.negotiateCloseErrHandler(msg.cid, chanCloser)
 
 	// Next, we'll process the next message using the target state machine.
 	// We'll either continue negotiation, or halt.
@@ -5360,31 +5358,35 @@ func (p *Brontide) handleCloseMsg(msg *closeMsg) {
 			})
 		})
 
-		beginNegotiation := func() {
-			oClosingSigned, err := chanCloser.BeginNegotiation()
-			if err != nil {
-				handleErr(err)
-				return
-			}
-
-			oClosingSigned.WhenSome(func(msg lnwire.ClosingSigned) {
-				p.queueMsg(&msg, nil)
-			})
-		}
-
+		// Without a link there's no commitment traffic left to drain,
+		// so the channel is already flushed as far as we're concerned.
 		if link == nil {
-			beginNegotiation()
-		} else {
-			// Now we register a flush hook to advance the
-			// ChanCloser and possibly send out a ClosingSigned
-			// when the link finishes draining.
-			link.OnFlushedOnce(func() {
-				// Remove link in goroutine to prevent deadlock.
-				go p.cfg.Switch.RemoveLink(msg.cid)
-				beginNegotiation()
-			})
+			p.beginNegotiation(chanCloser, handleErr)
+
+			return
 		}
 
+		// Otherwise, we register a flush hook so we hear about it once
+		// the link finishes draining.
+		link.OnFlushedOnce(func() {
+			// Remove link in goroutine to prevent deadlock.
+			go p.cfg.Switch.RemoveLink(msg.cid)
+
+			// The link runs this hook on its own goroutine, and may
+			// well hold its lock while it does, so we hand the
+			// channel to the channelManager instead of advancing
+			// the closer from here. That keeps the state machine
+			// owned by a single goroutine, and it means we can't
+			// block the link on work the channelManager is doing,
+			// which may itself be waiting on the link's lock.
+			go func() {
+				select {
+				case p.chanCloseFlushed <- msg.cid:
+				case <-p.cg.Done():
+				}
+			}()
+		})
+
 	case *lnwire.ClosingSigned:
 		oClosingSigned, err := chanCloser.ReceiveClosingSigned(*typed)
 		if err != nil {
@@ -5400,6 +5402,73 @@ func (p *Brontide) handleCloseMsg(msg *closeMsg) {
 		panic("impossible closeMsg type")
 	}
 
+	p.maybeFinalizeChanClosure(chanCloser)
+}
+
+// handleChanFlushed is called once a link has drained the HTLCs from a channel
+// we're cooperatively closing, which is our cue to move the negotiation along.
+// The link notices the flush from its own goroutine and hands the channel to us
+// over chanCloseFlushed, so that the closer only ever advances here.
+//
+// NOTE: MUST be called from the channelManager goroutine.
+func (p *Brontide) handleChanFlushed(cid lnwire.ChannelID) {
+	// We deliberately don't go through fetchActiveChanCloser here, as that
+	// would build a fresh closer if the negotiation has already been torn
+	// down while we were waiting on the link.
+	chanCloserE, found := p.activeChanCloses.Load(cid)
+	if !found {
+		p.log.Debugf("ChannelID(%v) flushed, but no chan closer is "+
+			"active", cid)
+
+		return
+	}
+
+	// The RBF closer drives its own flush handling, so there's nothing for
+	// us to do if that's the one closing this channel.
+	if chanCloserE.IsRight() {
+		return
+	}
+
+	var chanCloser *chancloser.ChanCloser
+	chanCloserE.WhenLeft(func(c *chancloser.ChanCloser) {
+		chanCloser = c
+	})
+
+	p.beginNegotiation(
+		chanCloser, p.negotiateCloseErrHandler(cid, chanCloser),
+	)
+}
+
+// beginNegotiation starts the fee negotiation phase of a legacy cooperative
+// close, sending out our opening offer if it falls to us to make one, and wraps
+// the closure up if the negotiation ran all the way through to a broadcast
+// transaction.
+//
+// NOTE: MUST be called from the channelManager goroutine.
+func (p *Brontide) beginNegotiation(chanCloser *chancloser.ChanCloser,
+	handleErr func(error)) {
+
+	oClosingSigned, err := chanCloser.BeginNegotiation()
+	if err != nil {
+		handleErr(err)
+
+		return
+	}
+
+	oClosingSigned.WhenSome(func(msg lnwire.ClosingSigned) {
+		p.queueMsg(&msg, nil)
+	})
+
+	p.maybeFinalizeChanClosure(chanCloser)
+}
+
+// maybeFinalizeChanClosure wraps up a cooperative closure if the negotiation
+// has run to completion, and does nothing if it hasn't.
+//
+// NOTE: MUST be called from the channelManager goroutine.
+func (p *Brontide) maybeFinalizeChanClosure(
+	chanCloser *chancloser.ChanCloser) {
+
 	// If we haven't finished close negotiations, then we'll continue as we
 	// can't yet finalize the closure.
 	if _, err := chanCloser.ClosingTx(); err != nil {
@@ -5412,6 +5481,32 @@ func (p *Brontide) handleCloseMsg(msg *closeMsg) {
 	p.finalizeChanClosure(chanCloser)
 }
 
+// negotiateCloseErrHandler returns the function used to tear down a legacy
+// close negotiation once one of the steps we drive it through has failed.
+//
+// NOTE: MUST be called from the channelManager goroutine.
+func (p *Brontide) negotiateCloseErrHandler(cid lnwire.ChannelID,
+	chanCloser *chancloser.ChanCloser) func(error) {
+
+	return func(err error) {
+		err = fmt.Errorf("unable to process close msg: %w", err)
+		p.log.Error(err)
+
+		// As the negotiations failed, we'll reset the channel state
+		// machine to ensure we act to on-chain events as normal.
+		chanCloser.Channel().ResetState()
+		if chanCloser.CloseRequest() != nil {
+			chanCloser.CloseRequest().Err <- err
+		}
+
+		p.deleteActiveChanCloser(
+			cid, chanCloser.Channel().ChannelPoint(),
+		)
+
+		p.Disconnect(err)
+	}
+}
+
 // HandleLocalCloseChanReqs accepts a *htlcswitch.ChanClose and passes it onto
 // the channelManager goroutine, which will shut down the link and possibly
 // close the channel.

peer/brontide_test.go

@@ -179,6 +179,131 @@ func TestPeerChannelClosureAcceptFeeResponder(t *testing.T) {
 	notifier.ConfChan <- &chainntnfs.TxConfirmation{}
 }
 
+// TestPeerChannelClosureFlushDrivesNegotiation checks that a legacy cooperative
+// close holds off on fee negotiation until the link reports that the channel
+// has drained, and that the report is what carries the negotiation forward. The
+// link notices the flush on its own goroutine, so it hands the channel to the
+// channelManager rather than advancing the closer itself.
+func TestPeerChannelClosureFlushDrivesNegotiation(t *testing.T) {
+	t.Parallel()
+
+	harness, err := createTestPeerWithChannel(t, noUpdate)
+	require.NoError(t, err, "unable to create test channels")
+
+	var (
+		alicePeer       = harness.peer
+		bobChan         = harness.channel
+		mockSwitch      = harness.mockSwitch
+		broadcastTxChan = harness.publishTx
+		notifier        = harness.notifier
+	)
+
+	chanPoint := bobChan.ChannelPoint()
+	chanID := lnwire.NewChanIDFromOutPoint(chanPoint)
+
+	// The link holds on to the flush hook rather than running it inline, so
+	// we get to say when the channel looks drained.
+	mockLink := newDeferredFlushUpdateHandler(chanID)
+	mockSwitch.links = append(mockSwitch.links, mockLink)
+
+	dummyDeliveryScript := genScript(t, p2wshAddress)
+
+	// We send a shutdown request to Alice, and expect her own Shutdown in
+	// response.
+	alicePeer.chanCloseMsgs <- &closeMsg{
+		cid: chanID,
+		msg: lnwire.NewShutdown(chanID, dummyDeliveryScript),
+	}
+
+	var msg lnwire.Message
+	select {
+	case outMsg := <-alicePeer.outgoingQueue:
+		msg = outMsg.msg
+	case <-time.After(timeout):
+		t.Fatalf("did not receive shutdown message")
+	}
+
+	shutdownMsg, ok := msg.(*lnwire.Shutdown)
+	require.True(t, ok, "expected Shutdown message, got %T", msg)
+
+	respDeliveryScript := shutdownMsg.Address
+
+	// The channel hasn't drained yet, so Alice shouldn't have opened fee
+	// negotiation, even though she's the one that funded the channel.
+	select {
+	case outMsg := <-alicePeer.outgoingQueue:
+		t.Fatalf("negotiation started before the channel flushed: %T",
+			outMsg.msg)
+
+	case <-time.After(shortTimeout):
+	}
+
+	// A flush report for a channel we have no closer for should be dropped
+	// on the floor rather than start anything.
+	var unknownChanID lnwire.ChannelID
+	select {
+	case alicePeer.chanCloseFlushed <- unknownChanID:
+	case <-time.After(timeout):
+		t.Fatalf("channelManager not reading flush reports")
+	}
+
+	// Now we let the link report the flush, which is what should carry the
+	// negotiation into its fee phase.
+	select {
+	case hook := <-mockLink.flushHooks:
+		go hook()
+	case <-time.After(timeout):
+		t.Fatalf("no flush hook was registered")
+	}
+
+	select {
+	case outMsg := <-alicePeer.outgoingQueue:
+		msg = outMsg.msg
+	case <-time.After(timeout):
+		t.Fatalf("did not receive ClosingSigned message")
+	}
+
+	respClosingSigned, ok := msg.(*lnwire.ClosingSigned)
+	require.True(t, ok, "expected ClosingSigned message, got %T", msg)
+
+	// We accept the fee, and send a ClosingSigned with the same fee back so
+	// she knows we agreed.
+	aliceFee := respClosingSigned.FeeSatoshis
+	bobSig, _, _, err := bobChan.CreateCloseProposal(
+		aliceFee, dummyDeliveryScript, respDeliveryScript,
+	)
+	require.NoError(t, err, "error creating close proposal")
+
+	parsedSig, err := lnwire.NewSigFromSignature(bobSig)
+	require.NoError(t, err, "error parsing signature")
+
+	alicePeer.chanCloseMsgs <- &closeMsg{
+		cid: chanID,
+		msg: lnwire.NewClosingSigned(chanID, aliceFee, parsedSig),
+	}
+
+	// Alice should now see that we agreed on the fee, and broadcast the
+	// closing transaction.
+	select {
+	case <-broadcastTxChan:
+	case <-time.After(timeout):
+		t.Fatalf("closing tx not broadcast")
+	}
+
+	// Need to pull the remaining message off of Alice's outgoing queue.
+	select {
+	case outMsg := <-alicePeer.outgoingQueue:
+		msg = outMsg.msg
+	case <-time.After(timeout):
+		t.Fatalf("did not receive ClosingSigned message")
+	}
+	_, ok = msg.(*lnwire.ClosingSigned)
+	require.True(t, ok, "expected ClosingSigned message, got %T", msg)
+
+	// Alice should be waiting in a goroutine for a confirmation.
+	notifier.ConfChan <- &chainntnfs.TxConfirmation{}
+}
+
 // TestPeerChannelClosureAcceptFeeInitiator tests the shutdown initiator's
 // behavior if we can agree on the fee immediately.
 func TestPeerChannelClosureAcceptFeeInitiator(t *testing.T) {

peer/test_utils.go

@@ -43,6 +43,10 @@ const (
 	// a return value on a channel.
 	timeout = time.Second * 5
 
+	// shortTimeout is the window a test waits for when it expects nothing
+	// to show up on a channel.
+	shortTimeout = time.Millisecond * 250
+
 	// testCltvRejectDelta is the minimum delta between expiry and current
 	// height below which htlcs are rejected.
 	testCltvRejectDelta = 13
@@ -388,6 +392,12 @@ type mockUpdateHandler struct {
 	cid                  lnwire.ChannelID
 	isOutgoingAddBlocked atomic.Bool
 	isIncomingAddBlocked atomic.Bool
+
+	// flushHooks receives the hooks registered through OnFlushedOnce when
+	// the handler was built with deferFlush set. Tests that want to control
+	// when the channel looks flushed read the hook from here and call it
+	// themselves, standing in for the link's own goroutine.
+	flushHooks chan func()
 }
 
 // newMockUpdateHandler creates a new mockUpdateHandler.
@@ -397,6 +407,18 @@ func newMockUpdateHandler(cid lnwire.ChannelID) *mockUpdateHandler {
 	}
 }
 
+// newDeferredFlushUpdateHandler creates a mock link that holds on to the hooks
+// registered through OnFlushedOnce instead of running them inline, so a test
+// can decide when the channel becomes flushed.
+func newDeferredFlushUpdateHandler(
+	cid lnwire.ChannelID) *mockUpdateHandler {
+
+	return &mockUpdateHandler{
+		cid:        cid,
+		flushHooks: make(chan func(), 1),
+	}
+}
+
 // HandleChannelUpdate currently does nothing.
 func (m *mockUpdateHandler) HandleChannelUpdate(msg lnwire.Message) {}
 
@@ -465,6 +487,12 @@ func (m *mockUpdateHandler) IsFlushing(dir htlcswitch.LinkDirection) bool {
 }
 
 func (m *mockUpdateHandler) OnFlushedOnce(hook func()) {
+	if m.flushHooks != nil {
+		m.flushHooks <- hook
+
+		return
+	}
+
 	hook()
 }
 func (m *mockUpdateHandler) OnCommitOnce(