Merge pull request #11019 from Roasbeef/coop-close-race-public

lnwallet/chancloser: fix data race in the legacy coop close state machine

docs/release-notes/release-notes-0.21.2.md

@@ -58,6 +58,16 @@
   and legacy payment paths, including keysend records and preimage-dependent
   settlement outcomes.
 
+* [Fixed a data race](https://github.com/lightningnetwork/lnd/pull/11019) in the
+  legacy cooperative close state machine, which was advanced from both the link
+  goroutine and the peer goroutine with nothing synchronizing the two. The link
+  now reports a flushed channel to the peer's channel manager instead of driving
+  the closer itself, so every step of a close runs on a single goroutine. The
+  same change has the RBF closer validate the remote party's delivery script in
+  all cases, rather than only when an upfront shutdown script was on record for
+  that peer, and rejects an absent script instead of treating it as nothing to
+  check.
+
 # New Features
 
 ## Functional Enhancements

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
@@ -592,10 +598,13 @@ func (c *ChanCloser) ReceiveShutdown(msg lnwire.Shutdown) (
 	noShutdown := fn.None[lnwire.Shutdown]()
 
 	// We'll track their remote close output, even if it's dust in BTC
-	// terms, it might still carry value in custom channel terms.
+	// terms, it might still carry value in custom channel terms. We only
+	// commit it to our state in the branches below that go on to accept the
+	// message: a Shutdown that shows up at a point where we can't act on it
+	// has no business overwriting an output we already settled on.
 	_, dustAmt := c.cfg.Channel.RemoteBalanceDust()
 	_, remoteBalance := c.cfg.Channel.CommitBalances()
-	c.remoteCloseOutput = fn.Some(types.CloseOutput{
+	remoteCloseOutput := fn.Some(types.CloseOutput{
 		Amt:             remoteBalance,
 		DustLimit:       dustAmt,
 		PkScript:        msg.Address,
@@ -642,6 +651,7 @@ func (c *ChanCloser) ReceiveShutdown(msg lnwire.Shutdown) (
 		// address. We'll use this when we craft the closure
 		// transaction.
 		c.remoteDeliveryScript = msg.Address
+		c.remoteCloseOutput = remoteCloseOutput
 
 		// We'll generate a shutdown message of our own to send across
 		// the wire.
@@ -691,6 +701,7 @@ func (c *ChanCloser) ReceiveShutdown(msg lnwire.Shutdown) (
 		// address, we'll record their preferred delivery closing
 		// script.
 		c.remoteDeliveryScript = msg.Address
+		c.remoteCloseOutput = remoteCloseOutput
 
 		// At this point, we can now start the fee negotiation state, by
 		// constructing and sending our initial signature for what we

lnwallet/chancloser/rbf_coop_test.go

@@ -1466,6 +1466,89 @@ func TestRbfChannelActiveTransitions(t *testing.T) {
 		)
 	})
 
+	// Even when the remote party never committed to an upfront shutdown
+	// script, we should still validate the delivery script they send, and
+	// reject one that isn't a well-formed delivery script.
+	name := "remote_initiated_bad_script_no_upfront_fail"
+	t.Run(name, func(t *testing.T) {
+		// The spec dropped p2pkh and p2sh for co-op closes to keep the
+		// dust calculations uniform, and a delivery script has to be
+		// something we can actually pay to, so none of these are
+		// acceptable even though some of them are perfectly valid
+		// scripts in their own right.
+		badScripts := []struct {
+			name   string
+			script lnwire.DeliveryAddress
+		}{
+			{
+				name:   "empty",
+				script: lnwire.DeliveryAddress{},
+			},
+			{
+				name: "garbage",
+				script: lnwire.DeliveryAddress(
+					bytes.Repeat([]byte{0xff}, 5),
+				),
+			},
+			{
+				// Provably unspendable: paying a close output
+				// here would burn the remote party's balance.
+				name: "op_return",
+				script: lnwire.DeliveryAddress(append(
+					[]byte{txscript.OP_RETURN, 32},
+					bytes.Repeat([]byte{0xAB}, 32)...,
+				)),
+			},
+			{
+				name: "bare_op_return",
+				script: lnwire.DeliveryAddress(
+					[]byte{txscript.OP_RETURN},
+				),
+			},
+			{
+				name: "p2pkh",
+				script: lnwire.DeliveryAddress(append(append(
+					[]byte{
+						txscript.OP_DUP,
+						txscript.OP_HASH160, 20,
+					},
+					bytes.Repeat([]byte{0xAB}, 20)...,
+				),
+					txscript.OP_EQUALVERIFY,
+					txscript.OP_CHECKSIG,
+				)),
+			},
+			{
+				name: "p2sh",
+				script: lnwire.DeliveryAddress(append(append(
+					[]byte{txscript.OP_HASH160, 20},
+					bytes.Repeat([]byte{0xAB}, 20)...,
+				), txscript.OP_EQUAL)),
+			},
+		}
+
+		for _, badScript := range badScripts {
+			t.Run(badScript.name, func(t *testing.T) {
+				// Note the config carries no remoteUpfrontAddr,
+				// so the only thing standing between the peer's
+				// script and the rest of the close flow is the
+				// delivery-script validation itself.
+				closeHarness := newCloser(t, &harnessCfg{
+					localUpfrontAddr: fn.Some(localAddr),
+				})
+				defer closeHarness.stopAndAssert()
+
+				event := &ShutdownReceived{
+					ShutdownScript: badScript.script,
+				}
+				closeHarness.sendEventAndExpectFailure(
+					ctx, event, ErrInvalidShutdownScript,
+				)
+				closeHarness.assertNoStateTransitions()
+			})
+		}
+	})
+
 	// When we receive a shutdown, we should transition to the shutdown
 	// pending state, with the local+remote shutdown addrs known.
 	t.Run("remote_initiated_close_ok", func(t *testing.T) {
@@ -1731,8 +1814,12 @@ func TestRbfShutdownPendingTransitions(t *testing.T) {
 		// This will cause a self transition back to ShutdownPending.
 		closeHarness.assertStateTransitions(&ShutdownPending{})
 
-		// Next, we'll send in a shutdown complete event.
-		closeHarness.chanCloser.SendEvent(ctx, &ShutdownReceived{})
+		// Next, we'll send in a shutdown complete event. The script is
+		// incidental to what this test exercises, but a shutdown always
+		// carries one, so we supply the remote party's.
+		closeHarness.chanCloser.SendEvent(ctx, &ShutdownReceived{
+			ShutdownScript: remoteAddr,
+		})
 
 		// We should transition to the channel flushing state, then the
 		// self event to have this state cache he early offer should
@@ -3114,7 +3201,8 @@ func TestNextCloseeNonceStorageFromClosingSig(t *testing.T) {
 	// updateAndValidateCloseTerms should only validate close terms, not
 	// update the nonce. The nonce rotation happens in
 	// LocalOfferSent.ProcessEvent.
-	err := negotiation.updateAndValidateCloseTerms(sigEvent, true)
+	env := &Environment{ChainParams: chaincfg.RegressionNetParams}
+	err := negotiation.updateAndValidateCloseTerms(sigEvent, env)
 	require.NoError(t, err)
 
 	// Verify the RemoteCloseeNonce was NOT modified — it should still

lnwallet/chancloser/rbf_coop_transitions.go

@@ -200,13 +200,32 @@ func validateShutdown(chanThawHeight fn.Option[uint32],
 		return ErrTaprootShutdownNonceMissing
 	}
 
-	// Next, we'll verify that the remote party is sending the expected
-	// shutdown script.
-	return fn.MapOption(func(addr lnwire.DeliveryAddress) error {
-		return validateShutdownScript(
-			addr, msg.ShutdownScript, &chainParams,
-		)
-	})(upfrontAddr).UnwrapOr(nil)
+	// Finally, verify the remote party's delivery script. We validate it in
+	// all cases (mirroring the negotiation closer), rather than only when
+	// an upfront shutdown script is on record: passing a nil upfront script
+	// still runs the well-formedness check on the peer's script, and a
+	// non-nil upfront script additionally enforces the exact match.
+	return validateRemoteDeliveryScript(
+		upfrontAddr, msg.ShutdownScript, chainParams,
+	)
+}
+
+// validateRemoteDeliveryScript checks a delivery script the remote party sent
+// us, against any upfront shutdown script we have on record for them. We end up
+// paying to this script, so it has to be present, and it has to be one of the
+// delivery forms we accept. An absent script is rejected here rather than
+// treated as nothing to check.
+func validateRemoteDeliveryScript(upfrontAddr fn.Option[lnwire.DeliveryAddress],
+	script lnwire.DeliveryAddress, chainParams chaincfg.Params) error {
+
+	if len(script) == 0 {
+		return fmt.Errorf("%w: no delivery script",
+			ErrInvalidShutdownScript)
+	}
+
+	return validateShutdownScript(
+		upfrontAddr.UnwrapOr(nil), script, &chainParams,
+	)
 }
 
 // ProcessEvent takes a protocol event, and implements a state transition for
@@ -902,7 +921,7 @@ func validateAndExtractSigAndNonce(
 // incoming event, and decide if we need to update the remote party's address,
 // or reject it if it doesn't include our latest address.
 func (c *ClosingNegotiation) updateAndValidateCloseTerms(event ProtocolEvent,
-	isTaproot bool) error {
+	env *Environment) error {
 
 	assertLocalScriptMatches := func(localScriptInMsg []byte) error {
 		if !bytes.Equal(
@@ -933,9 +952,19 @@ func (c *ClosingNegotiation) updateAndValidateCloseTerms(event ProtocolEvent,
 		oldRemoteAddr := c.RemoteDeliveryScript
 		newRemoteAddr := msg.SigMsg.CloserScript
 
-		// If they're sending a new script, then we'll update to the new
-		// one.
+		// If they're sending a new script, then we'll make sure it's
+		// well-formed (and matches any upfront script on record) before
+		// we update to the new one, just as we do for the initial
+		// shutdown script.
 		if !bytes.Equal(oldRemoteAddr, newRemoteAddr) {
+			err := validateRemoteDeliveryScript(
+				env.RemoteUpfrontShutdown, newRemoteAddr,
+				env.ChainParams,
+			)
+			if err != nil {
+				return err
+			}
+
 			c.RemoteDeliveryScript = newRemoteAddr
 		}
 
@@ -986,7 +1015,7 @@ func (c *ClosingNegotiation) ProcessEvent(event ProtocolEvent, env *Environment,
 	// At this point, we know its a new signature message. We'll validate,
 	// and maybe update the set of close terms based on what we receive. We
 	// might update the remote party's address for example.
-	err := c.updateAndValidateCloseTerms(event, env.IsTaproot())
+	err := c.updateAndValidateCloseTerms(event, env)
 	if err != nil {
 		return nil, fmt.Errorf("event violates close terms: %w", err)
 	}

lnwallet/parameters.go

@@ -41,8 +41,10 @@ func DefaultRoutingFeeLimitForAmount(a lnwire.MilliSatoshi) lnwire.MilliSatoshi
 
 // DustLimitForSize retrieves the dust limit for a given pkscript size. Given
 // the size, it automatically determines whether the script is a witness script
-// or not. It calls btcd's GetDustThreshold method under the hood. It must be
-// called with a proper size parameter or else a panic occurs.
+// or not. It calls btcd's GetDustThreshold method under the hood. Any size that
+// doesn't map to one of the well-known templates is treated as a generic
+// witness output, so the helper stays well-defined for arbitrary (including
+// future witness-version) script lengths.
 func DustLimitForSize(scriptSize int) btcutil.Amount {
 	var (
 		dustlimit btcutil.Amount
@@ -66,11 +68,11 @@ func DustLimitForSize(scriptSize int) btcutil.Amount {
 	case input.P2PKHSize:
 		pkscript, _ = input.GenerateP2PKH([]byte{})
 
-	case input.UnknownWitnessSize:
-		pkscript, _ = input.GenerateUnknownWitness()
-
+	// Any other length (the explicit UnknownWitnessSize, or an otherwise
+	// unrecognized size) is priced as a generic witness output rather than
+	// treated as a hard error.
 	default:
-		panic("invalid script size")
+		pkscript, _ = input.GenerateUnknownWitness()
 	}
 
 	// Call GetDustThreshold with a TxOut containing the generated

lnwallet/parameters_test.go

@@ -81,6 +81,21 @@ func TestDustLimitForSize(t *testing.T) {
 			size:          input.UnknownWitnessSize,
 			expectedLimit: btcutil.Amount(354),
 		},
+		{
+			// An arbitrary short length that matches no known
+			// template is priced as a generic witness output
+			// rather than treated as an error.
+			name:          "arbitrary small size",
+			size:          7,
+			expectedLimit: btcutil.Amount(354),
+		},
+		{
+			// The largest witness program length is also handled
+			// as a generic witness output.
+			name:          "arbitrary large witness size",
+			size:          42,
+			expectedLimit: btcutil.Amount(354),
+		},
 	}
 
 	for _, test := range tests {

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,31 +5362,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 {
@@ -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(