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

### lnwallet/chancloser/chancloser.go
@@ -164,6 +164,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
@@ -675,6 +675,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
@@ -753,6 +761,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),
@@ -3278,6 +3287,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
@@ -5306,23 +5320,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.
@@ -5364,30 +5362,34 @@ func (p *Brontide) handleCloseMsg(msg *closeMsg) {
 			})
 		})
 
-		beginNegotiation := func() {
-			oClosingSigned, err := chanCloser.BeginNegotiation()
-			if err != nil {
-				handleErr(err)
-				return
-			}
+		// 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 {
+			p.beginNegotiation(chanCloser, handleErr)
 
-			oClosingSigned.WhenSome(func(msg lnwire.ClosingSigned) {
-				p.queueMsg(&msg, nil)
-			})
+			return
 		}
 
-		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()
-			})
-		}
+		// 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)
@@ -5404,6 +5406,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 {
@@ -5416,6 +5485,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
@@ -180,6 +180,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
@@ -44,6 +44,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
@@ -387,6 +391,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.
@@ -396,6 +406,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) {}
 
@@ -464,6 +486,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(