Repository: lightningnetwork/lnd
Pull request #6246 (2022-02-09 17:05:30)

htlcswitch: only pipeline settle if add is locked-in
The counter-party shouldn't be doing this anyways as they would be giving away a preimage for free. Them doing this would bork their own channel due to open circuits not getting trimmed on startup. Removing this faulty behavior also makes it easier to reason about the circuit logic.


Patch:
diff --git a/chanbackup/pubsub.go b/chanbackup/pubsub.go
index 3e1580ddd..4eef45e1c 100644
--- a/chanbackup/pubsub.go
+++ b/chanbackup/pubsub.go
@@ -137,7 +137,7 @@ func NewSubSwapper(startingChans []Single, chanNotifier ChannelNotifier,
 func (s *SubSwapper) Start() error {
 	var startErr error
 	s.started.Do(func() {
-		log.Infof("chanbackup.SubSwapper starting")
+		log.Infof("Starting chanbackup.SubSwapper")
 
 		// Before we enter our main loop, we'll update the on-disk
 		// state with the latest Single state, as nodes may have new
diff --git a/chanfitness/chaneventstore.go b/chanfitness/chaneventstore.go
index 53a699932..d6a634793 100644
--- a/chanfitness/chaneventstore.go
+++ b/chanfitness/chaneventstore.go
@@ -142,8 +142,6 @@ func NewChannelEventStore(config *Config) *ChannelEventStore {
 // information from the store. If this function fails, it cancels its existing
 // subscriptions and returns an error.
 func (c *ChannelEventStore) Start() error {
-	log.Info("ChannelEventStore starting")
-
 	// Create a subscription to channel events.
 	channelClient, err := c.cfg.SubscribeChannelEvents()
 	if err != nil {
diff --git a/channelnotifier/channelnotifier.go b/channelnotifier/channelnotifier.go
index 7c6cc3095..2cf6015c4 100644
--- a/channelnotifier/channelnotifier.go
+++ b/channelnotifier/channelnotifier.go
@@ -87,7 +87,7 @@ func New(chanDB *channeldb.ChannelStateDB) *ChannelNotifier {
 func (c *ChannelNotifier) Start() error {
 	var err error
 	c.started.Do(func() {
-		log.Info("ChannelNotifier starting")
+		log.Trace("ChannelNotifier starting")
 		err = c.ntfnServer.Start()
 	})
 	return err
diff --git a/contractcourt/breacharbiter.go b/contractcourt/breacharbiter.go
index 479c0b8ae..6f160141b 100644
--- a/contractcourt/breacharbiter.go
+++ b/contractcourt/breacharbiter.go
@@ -206,13 +206,14 @@ func NewBreachArbiter(cfg *BreachConfig) *BreachArbiter {
 func (b *BreachArbiter) Start() error {
 	var err error
 	b.started.Do(func() {
-		brarLog.Info("Breach arbiter starting")
 		err = b.start()
 	})
 	return err
 }
 
 func (b *BreachArbiter) start() error {
+	brarLog.Tracef("Starting breach arbiter")
+
 	// Load all retributions currently persisted in the retribution store.
 	var breachRetInfos map[wire.OutPoint]retributionInfo
 	if err := b.cfg.Store.ForAll(func(ret *retributionInfo) error {
diff --git a/contractcourt/chain_arbitrator.go b/contractcourt/chain_arbitrator.go
index d88a52b64..ba007ea6b 100644
--- a/contractcourt/chain_arbitrator.go
+++ b/contractcourt/chain_arbitrator.go
@@ -487,7 +487,7 @@ func (c *ChainArbitrator) Start() error {
 		return nil
 	}
 
-	log.Info("ChainArbitrator starting")
+	log.Tracef("Starting ChainArbitrator")
 
 	// First, we'll fetch all the channels that are still open, in order to
 	// collect them within our set of active contracts.
@@ -964,9 +964,16 @@ type ContractUpdate struct {
 	Htlcs []channeldb.HTLC
 }
 
-// ContractSignals is used by outside subsystems to notify a channel arbitrator
-// of its ShortChannelID.
+// ContractSignals wraps the two signals that affect the state of a channel
+// being watched by an arbitrator. The two signals we care about are: the
+// channel has a new set of HTLC's, and the remote party has just broadcast
+// their version of the commitment transaction.
 type ContractSignals struct {
+	// HtlcUpdates is a channel that the link will use to update the
+	// designated channel arbitrator when the set of HTLCs on any valid
+	// commitment changes.
+	HtlcUpdates chan *ContractUpdate
+
 	// ShortChanID is the up to date short channel ID for a contract. This
 	// can change either if when the contract was added it didn't yet have
 	// a stable identifier, or in the case of a reorg.
@@ -994,25 +1001,6 @@ func (c *ChainArbitrator) UpdateContractSignals(chanPoint wire.OutPoint,
 	return nil
 }
 
-// NotifyContractUpdate lets a channel arbitrator know that a new
-// ContractUpdate is available. This calls the ChannelArbitrator's internal
-// method NotifyContractUpdate which waits for a response on a done chan before
-// returning. This method will return an error if the ChannelArbitrator is not
-// in the activeChannels map. However, this only happens if the arbitrator is
-// resolved and the related link would already be shut down.
-func (c *ChainArbitrator) NotifyContractUpdate(chanPoint wire.OutPoint,
-	update *ContractUpdate) error {
-
-	c.Lock()
-	arbitrator, ok := c.activeChannels[chanPoint]
-	c.Unlock()
-	if !ok {
-		return fmt.Errorf("can't find arbitrator for %v", chanPoint)
-	}
-
-	return arbitrator.notifyContractUpdate(update)
-}
-
 // GetChannelArbitrator safely returns the channel arbitrator for a given
 // channel outpoint.
 func (c *ChainArbitrator) GetChannelArbitrator(chanPoint wire.OutPoint) (
diff --git a/contractcourt/channel_arbitrator.go b/contractcourt/channel_arbitrator.go
index 1d2c0c859..a67c5e1e1 100644
--- a/contractcourt/channel_arbitrator.go
+++ b/contractcourt/channel_arbitrator.go
@@ -28,12 +28,6 @@ var (
 	// close a channel that's already in the process of doing so.
 	errAlreadyForceClosed = errors.New("channel is already in the " +
 		"process of being force closed")
-
-	// errChanArbShuttingDown is an error returned when the channel arb is
-	// shutting down during the hand-off in notifyContractUpdate. This is
-	// mainly used to be able to notify the original caller (the link) that
-	// an error occurred.
-	errChanArbShuttingDown = errors.New("channel arb shutting down")
 )
 
 const (
@@ -348,7 +342,7 @@ type ChannelArbitrator struct {
 	// htlcUpdates is a channel that is sent upon with new updates from the
 	// active channel. Each time a new commitment state is accepted, the
 	// set of HTLC's on the new state should be sent across this channel.
-	htlcUpdates chan *contractUpdateSignal
+	htlcUpdates <-chan *ContractUpdate
 
 	// activeResolvers is a slice of any active resolvers. This is used to
 	// be able to signal them for shutdown in the case that we shutdown.
@@ -384,7 +378,7 @@ func NewChannelArbitrator(cfg ChannelArbitratorConfig,
 		log:              log,
 		blocks:           make(chan int32, arbitratorBlockBufferSize),
 		signalUpdates:    make(chan *signalUpdateMsg),
-		htlcUpdates:      make(chan *contractUpdateSignal),
+		htlcUpdates:      make(<-chan *ContractUpdate),
 		resolutionSignal: make(chan struct{}),
 		forceCloseReqs:   make(chan *forceCloseReq),
 		activeHTLCs:      htlcSets,
@@ -2393,41 +2387,6 @@ func (c *ChannelArbitrator) UpdateContractSignals(newSignals *ContractSignals) {
 	}
 }
 
-// contractUpdateSignal is a struct that carries the latest set of
-// ContractUpdate for a particular key. It also carries a done chan that should
-// be closed by the recipient.
-type contractUpdateSignal struct {
-	// newUpdate contains the latest ContractUpdate for a key.
-	newUpdate *ContractUpdate
-
-	// doneChan is an acknowledgement channel.
-	doneChan chan struct{}
-}
-
-// notifyContractUpdate notifies the ChannelArbitrator that a new
-// ContractUpdate is available from the link. The link will be paused until
-// this function returns.
-func (c *ChannelArbitrator) notifyContractUpdate(upd *ContractUpdate) error {
-	done := make(chan struct{})
-
-	select {
-	case c.htlcUpdates <- &contractUpdateSignal{
-		newUpdate: upd,
-		doneChan:  done,
-	}:
-	case <-c.quit:
-		return errChanArbShuttingDown
-	}
-
-	select {
-	case <-done:
-	case <-c.quit:
-		return errChanArbShuttingDown
-	}
-
-	return nil
-}
-
 // channelAttendant is the primary goroutine that acts at the judicial
 // arbitrator between our channel state, the remote channel peer, and the
 // blockchain (Our judge). This goroutine will ensure that we faithfully execute
@@ -2489,12 +2448,13 @@ func (c *ChannelArbitrator) channelAttendant(bestHeight int32) {
 			log.Tracef("ChannelArbitrator(%v) got new signal "+
 				"update!", c.cfg.ChanPoint)
 
-			// We'll update the ShortChannelID.
+			// First, we'll update our set of signals.
+			c.htlcUpdates = signalUpdate.newSignals.HtlcUpdates
 			c.cfg.ShortChanID = signalUpdate.newSignals.ShortChanID
 
-			// Now that the signal has been updated, we'll now
+			// Now that the signals have been updated, we'll now
 			// close the done channel to signal to the caller we've
-			// registered the new ShortChannelID.
+			// registered the new contracts.
 			close(signalUpdate.doneChan)
 
 		// A new set of HTLC's has been added or removed from the
@@ -2505,19 +2465,14 @@ func (c *ChannelArbitrator) channelAttendant(bestHeight int32) {
 			// htlcSetKey type included in this update in order to
 			// only monitor the HTLCs that are still active on this
 			// target commitment.
-			htlcKey := htlcUpdate.newUpdate.HtlcKey
-			c.activeHTLCs[htlcKey] = newHtlcSet(
-				htlcUpdate.newUpdate.Htlcs,
+			c.activeHTLCs[htlcUpdate.HtlcKey] = newHtlcSet(
+				htlcUpdate.Htlcs,
 			)
 
-			// Now that the activeHTLCs have been updated, we'll
-			// close the done channel.
-			close(htlcUpdate.doneChan)
-
 			log.Tracef("ChannelArbitrator(%v): fresh set of htlcs=%v",
 				c.cfg.ChanPoint,
 				newLogClosure(func() string {
-					return spew.Sdump(htlcUpdate.newUpdate)
+					return spew.Sdump(htlcUpdate)
 				}),
 			)
 
diff --git a/contractcourt/utxonursery.go b/contractcourt/utxonursery.go
index 5a964c418..d5cc31a79 100644
--- a/contractcourt/utxonursery.go
+++ b/contractcourt/utxonursery.go
@@ -241,7 +241,7 @@ func (u *UtxoNursery) Start() error {
 		return nil
 	}
 
-	utxnLog.Info("UTXO nursery starting")
+	utxnLog.Tracef("Starting UTXO nursery")
 
 	// Retrieve the currently best known block. This is needed to have the
 	// state machine catch up with the blocks we missed when we were down.
diff --git a/discovery/gossiper.go b/discovery/gossiper.go
index f5deee56c..b463a0b24 100644
--- a/discovery/gossiper.go
+++ b/discovery/gossiper.go
@@ -498,13 +498,14 @@ func (d *AuthenticatedGossiper) PropagateChanPolicyUpdate(
 func (d *AuthenticatedGossiper) Start() error {
 	var err error
 	d.started.Do(func() {
-		log.Info("Authenticated Gossiper starting")
 		err = d.start()
 	})
 	return err
 }
 
 func (d *AuthenticatedGossiper) start() error {
+	log.Info("Authenticated Gossiper is starting")
+
 	// First we register for new notifications of newly discovered blocks.
 	// We do this immediately so we'll later be able to consume any/all
 	// blocks which were discovered.
diff --git a/docs/release-notes/release-notes-0.15.0.md b/docs/release-notes/release-notes-0.15.0.md
index 301f41e02..658f149c8 100644
--- a/docs/release-notes/release-notes-0.15.0.md
+++ b/docs/release-notes/release-notes-0.15.0.md
@@ -18,6 +18,8 @@
 
 ## Bug Fixes
 
+* [Pipelining an UpdateFulfillHTLC message now only happens when the related UpdateAddHTLC is locked-in.](https://github.com/lightningnetwork/lnd/pull/6246)
+
 * [Fixed an inactive invoice subscription not removed from invoice
   registry](https://github.com/lightningnetwork/lnd/pull/6053). When an invoice
   subscription is created and canceled immediately, it could be left uncleaned
@@ -36,10 +38,6 @@
   could result in an "invoice too large" error when creating invoices. Hints 
   are now properly limited to our maximum of 20.
 
-* [Fixed an edge case where the lnd might be stuck at starting due to channel
-  arbitrator relying on htlcswitch to be started
-  first](https://github.com/lightningnetwork/lnd/pull/6214).
-
 ## Misc
 
 * [An example systemd service file](https://github.com/lightningnetwork/lnd/pull/6033)
@@ -120,11 +118,6 @@ gRPC performance metrics (latency to process `GetInfo`, etc)](https://github.com
 * The [`whitespace` linter](https://github.com/lightningnetwork/lnd/pull/6270)
   was enabled to make sure multi-line `if` conditions and function/method
   declarations are followed by an empty line to improve readability.
-  **Note to developers**: please make sure you delete the old version of
-  `golangci-lint` in your `$GOPATH/bin` directory. `make lint` does not
-  automatically replace it with the new version if the binary already exists!
-  
-* [`ChannelLink` in the `htlcswitch` now performs a 2-way handoff instead of a 1-way handoff with its `ChannelArbitrator`.](https://github.com/lightningnetwork/lnd/pull/6221)
 
 # Contributors (Alphabetical Order)
 
@@ -137,7 +130,6 @@ gRPC performance metrics (latency to process `GetInfo`, etc)](https://github.com
 * Dan Bolser
 * Daniel McNally
 * ErikEk
-* Eugene Siegel
 * henta
 * Joost Jager
 * Jordi Montes
diff --git a/funding/manager.go b/funding/manager.go
index 28acd2f16..6306f7d91 100644
--- a/funding/manager.go
+++ b/funding/manager.go
@@ -588,13 +588,14 @@ func NewFundingManager(cfg Config) (*Manager, error) {
 func (f *Manager) Start() error {
 	var err error
 	f.started.Do(func() {
-		log.Info("Funding manager starting")
 		err = f.start()
 	})
 	return err
 }
 
 func (f *Manager) start() error {
+	log.Tracef("Funding manager running")
+
 	// Upon restart, the Funding Manager will check the database to load any
 	// channels that were  waiting for their funding transactions to be
 	// confirmed on the blockchain at the time when the daemon last went
@@ -1234,7 +1235,7 @@ func (f *Manager) handleFundingOpen(peer lnpeer.Peer,
 	if amt < f.cfg.MinChanSize {
 		f.failFundingFlow(
 			peer, msg.PendingChannelID,
-			lnwallet.ErrChanTooSmall(amt, f.cfg.MinChanSize),
+			lnwallet.ErrChanTooSmall(amt, btcutil.Amount(f.cfg.MinChanSize)),
 		)
 		return
 	}
diff --git a/htlcswitch/hop/iterator.go b/htlcswitch/hop/iterator.go
index 45a084755..20963bd7b 100644
--- a/htlcswitch/hop/iterator.go
+++ b/htlcswitch/hop/iterator.go
@@ -132,7 +132,6 @@ func NewOnionProcessor(router *sphinx.Router) *OnionProcessor {
 
 // Start spins up the onion processor's sphinx router.
 func (p *OnionProcessor) Start() error {
-	log.Info("Onion processor starting")
 	return p.router.Start()
 }
 
diff --git a/htlcswitch/htlcnotifier.go b/htlcswitch/htlcnotifier.go
index 00bba7113..fd067e2b4 100644
--- a/htlcswitch/htlcnotifier.go
+++ b/htlcswitch/htlcnotifier.go
@@ -81,7 +81,7 @@ func NewHtlcNotifier(now func() time.Time) *HtlcNotifier {
 func (h *HtlcNotifier) Start() error {
 	var err error
 	h.started.Do(func() {
-		log.Info("HtlcNotifier starting")
+		log.Trace("HtlcNotifier starting")
 		err = h.ntfnServer.Start()
 	})
 	return err
diff --git a/htlcswitch/link.go b/htlcswitch/link.go
index f4714803d..55df2d6cc 100644
--- a/htlcswitch/link.go
+++ b/htlcswitch/link.go
@@ -187,14 +187,11 @@ type ChannelLinkConfig struct {
 		LinkFailureError)
 
 	// UpdateContractSignals is a function closure that we'll use to update
-	// outside sub-systems with this channel's latest ShortChannelID.
+	// outside sub-systems with the latest signals for our inner Lightning
+	// channel. These signals will notify the caller when the channel has
+	// been closed, or when the set of active HTLC's is updated.
 	UpdateContractSignals func(*contractcourt.ContractSignals) error
 
-	// NotifyContractUpdate is a function closure that we'll use to update
-	// the contractcourt and more specifically the ChannelArbitrator of the
-	// latest channel state.
-	NotifyContractUpdate func(*contractcourt.ContractUpdate) error
-
 	// ChainEvents is an active subscription to the chain watcher for this
 	// channel to be notified of any on-chain activity related to this
 	// channel.
@@ -375,6 +372,10 @@ type channelLink struct {
 	// sent across.
 	localUpdateAdd chan *localUpdateAddMsg
 
+	// htlcUpdates is a channel that we'll use to update outside
+	// sub-systems with the latest set of active HTLC's on our channel.
+	htlcUpdates chan *contractcourt.ContractUpdate
+
 	// shutdownRequest is a channel that the channelLink will listen on to
 	// service shutdown requests from ShutdownIfChannelClean calls.
 	shutdownRequest chan *shutdownReq
@@ -420,9 +421,11 @@ func NewChannelLink(cfg ChannelLinkConfig,
 	logPrefix := fmt.Sprintf("ChannelLink(%v):", channel.ChannelPoint())
 
 	return &channelLink{
-		cfg:             cfg,
-		channel:         channel,
-		shortChanID:     channel.ShortChanID(),
+		cfg:         cfg,
+		channel:     channel,
+		shortChanID: channel.ShortChanID(),
+		// TODO(roasbeef): just do reserve here?
+		htlcUpdates:     make(chan *contractcourt.ContractUpdate),
 		shutdownRequest: make(chan *shutdownReq),
 		hodlMap:         make(map[channeldb.CircuitKey]hodlHtlc),
 		hodlQueue:       queue.NewConcurrentQueue(10),
@@ -493,6 +496,7 @@ func (l *channelLink) Start() error {
 		// TODO(roasbeef): split goroutines within channel arb to avoid
 		go func() {
 			signals := &contractcourt.ContractSignals{
+				HtlcUpdates: l.htlcUpdates,
 				ShortChanID: l.channel.ShortChanID(),
 			}
 
@@ -1672,6 +1676,28 @@ func (l *channelLink) handleUpstreamMsg(msg lnwire.Message) {
 	case *lnwire.UpdateFulfillHTLC:
 		pre := msg.PaymentPreimage
 		idx := msg.ID
+
+		// Before we pipeline the settle, we'll check the set of active
+		// htlc's to see if the related UpdateAddHTLC has been fully
+		// locked-in.
+		var lockedin bool
+		htlcs := l.channel.ActiveHtlcs()
+		for _, add := range htlcs {
+			// The HTLC will be outgoing and match idx.
+			if !add.Incoming && add.HtlcIndex == idx {
+				lockedin = true
+				break
+			}
+		}
+
+		if !lockedin {
+			l.fail(
+				LinkFailureError{code: ErrInvalidUpdate},
+				"unable to handle upstream settle",
+			)
+			return
+		}
+
 		if err := l.channel.ReceiveHTLCSettle(pre, idx); err != nil {
 			l.fail(
 				LinkFailureError{
@@ -1833,23 +1859,15 @@ func (l *channelLink) handleUpstreamMsg(msg lnwire.Message) {
 		l.cfg.Peer.SendMessage(false, nextRevocation)
 
 		// Since we just revoked our commitment, we may have a new set
-		// of HTLC's on our commitment, so we'll send them using our
-		// function closure NotifyContractUpdate.
-		newUpdate := &contractcourt.ContractUpdate{
+		// of HTLC's on our commitment, so we'll send them over our
+		// HTLC update channel so any callers can be notified.
+		select {
+		case l.htlcUpdates <- &contractcourt.ContractUpdate{
 			HtlcKey: contractcourt.LocalHtlcSet,
 			Htlcs:   currentHtlcs,
-		}
-		err = l.cfg.NotifyContractUpdate(newUpdate)
-		if err != nil {
-			l.log.Errorf("unable to notify contract update: %v",
-				err)
-			return
-		}
-
-		select {
+		}:
 		case <-l.quit:
 			return
-		default:
 		}
 
 		// If both commitment chains are fully synced from our PoV,
@@ -1883,21 +1901,13 @@ func (l *channelLink) handleUpstreamMsg(msg lnwire.Message) {
 		// The remote party now has a new primary commitment, so we'll
 		// update the contract court to be aware of this new set (the
 		// prior old remote pending).
-		newUpdate := &contractcourt.ContractUpdate{
+		select {
+		case l.htlcUpdates <- &contractcourt.ContractUpdate{
 			HtlcKey: contractcourt.RemoteHtlcSet,
 			Htlcs:   remoteHTLCs,
-		}
-		err = l.cfg.NotifyContractUpdate(newUpdate)
-		if err != nil {
-			l.log.Errorf("unable to notify contract update: %v",
-				err)
-			return
-		}
-
-		select {
+		}:
 		case <-l.quit:
 			return
-		default:
 		}
 
 		// If we have a tower client for this channel type, we'll
@@ -2105,20 +2115,13 @@ func (l *channelLink) updateCommitTx() error {
 	// The remote party now has a new pending commitment, so we'll update
 	// the contract court to be aware of this new set (the prior old remote
 	// pending).
-	newUpdate := &contractcourt.ContractUpdate{
+	select {
+	case l.htlcUpdates <- &contractcourt.ContractUpdate{
 		HtlcKey: contractcourt.RemotePendingHtlcSet,
 		Htlcs:   pendingHTLCs,
-	}
-	err = l.cfg.NotifyContractUpdate(newUpdate)
-	if err != nil {
-		l.log.Errorf("unable to notify contract update: %v", err)
-		return err
-	}
-
-	select {
+	}:
 	case <-l.quit:
 		return ErrLinkShuttingDown
-	default:
 	}
 
 	commitSig := &lnwire.CommitSig{
@@ -2186,6 +2189,7 @@ func (l *channelLink) UpdateShortChanID() (lnwire.ShortChannelID, error) {
 
 	go func() {
 		err := l.cfg.UpdateContractSignals(&contractcourt.ContractSignals{
+			HtlcUpdates: l.htlcUpdates,
 			ShortChanID: sid,
 		})
 		if err != nil {
diff --git a/htlcswitch/switch.go b/htlcswitch/switch.go
index a501abf40..860076d4d 100644
--- a/htlcswitch/switch.go
+++ b/htlcswitch/switch.go
@@ -1846,7 +1846,7 @@ func (s *Switch) Start() error {
 		return errors.New("htlc switch already started")
 	}
 
-	log.Infof("HTLC Switch starting")
+	log.Infof("Starting HTLC Switch")
 
 	blockEpochStream, err := s.cfg.Notifier.RegisterBlockEpochNtfn(nil)
 	if err != nil {
diff --git a/invoices/invoiceregistry.go b/invoices/invoiceregistry.go
index 11eca9dd3..6ba3b2386 100644
--- a/invoices/invoiceregistry.go
+++ b/invoices/invoiceregistry.go
@@ -238,8 +238,6 @@ func (i *InvoiceRegistry) Start() error {
 		return err
 	}
 
-	log.Info("InvoiceRegistry starting")
-
 	i.wg.Add(1)
 	go i.invoiceEventLoop()
 
diff --git a/lnwallet/sigpool.go b/lnwallet/sigpool.go
index cf174df14..ec1d4f346 100644
--- a/lnwallet/sigpool.go
+++ b/lnwallet/sigpool.go
@@ -155,7 +155,6 @@ func NewSigPool(numWorkers int, signer input.Signer) *SigPool {
 // carry out its duties.
 func (s *SigPool) Start() error {
 	s.started.Do(func() {
-		walletLog.Info("SigPool starting")
 		for i := 0; i < s.numWorkers; i++ {
 			s.wg.Add(1)
 			go s.poolWorker()
diff --git a/netann/chan_status_manager.go b/netann/chan_status_manager.go
index 4e05be7df..fc2b94c05 100644
--- a/netann/chan_status_manager.go
+++ b/netann/chan_status_manager.go
@@ -172,7 +172,6 @@ func NewChanStatusManager(cfg *ChanStatusConfig) (*ChanStatusManager, error) {
 func (m *ChanStatusManager) Start() error {
 	var err error
 	m.started.Do(func() {
-		log.Info("Channel Status Manager starting")
 		err = m.start()
 	})
 	return err
diff --git a/netann/host_ann.go b/netann/host_ann.go
index bc50289c1..c96b1af8e 100644
--- a/netann/host_ann.go
+++ b/netann/host_ann.go
@@ -58,7 +58,6 @@ func NewHostAnnouncer(cfg HostAnnouncerConfig) *HostAnnouncer {
 // Start starts the HostAnnouncer.
 func (h *HostAnnouncer) Start() error {
 	h.startOnce.Do(func() {
-		log.Info("HostAnnouncer starting")
 		h.wg.Add(1)
 		go h.hostWatcher()
 	})
diff --git a/peer/brontide.go b/peer/brontide.go
index b5571e596..576143824 100644
--- a/peer/brontide.go
+++ b/peer/brontide.go
@@ -815,10 +815,6 @@ func (p *Brontide) addLink(chanPoint *wire.OutPoint,
 		return p.cfg.ChainArb.UpdateContractSignals(*chanPoint, signals)
 	}
 
-	notifyContractUpdate := func(update *contractcourt.ContractUpdate) error {
-		return p.cfg.ChainArb.NotifyContractUpdate(*chanPoint, update)
-	}
-
 	chanType := lnChan.State().ChanType
 
 	// Select the appropriate tower client based on the channel type. It's
@@ -846,7 +842,6 @@ func (p *Brontide) addLink(chanPoint *wire.OutPoint,
 		PreimageCache:           p.cfg.WitnessBeacon,
 		ChainEvents:             chainEvents,
 		UpdateContractSignals:   updateContractSignals,
-		NotifyContractUpdate:    notifyContractUpdate,
 		OnChannelFailure:        onChannelFailure,
 		SyncStates:              syncStates,
 		BatchTicker:             ticker.New(p.cfg.ChannelCommitInterval),
diff --git a/routing/router.go b/routing/router.go
index 9b7a1f96d..27ae7bbd6 100644
--- a/routing/router.go
+++ b/routing/router.go
@@ -490,7 +490,7 @@ func (r *ChannelRouter) Start() error {
 		return nil
 	}
 
-	log.Info("Channel Router starting")
+	log.Tracef("Channel Router starting")
 
 	bestHash, bestHeight, err := r.cfg.Chain.GetBestBlock()
 	if err != nil {
diff --git a/server.go b/server.go
index d36e013eb..b729dc97c 100644
--- a/server.go
+++ b/server.go
@@ -1777,15 +1777,6 @@ func (s *server) Start() error {
 		}
 		cleanup = cleanup.add(s.fundingMgr.Stop)
 
-		// htlcSwitch must be started before chainArb since the latter
-		// relies on htlcSwitch to deliver resolution message upon
-		// start.
-		if err := s.htlcSwitch.Start(); err != nil {
-			startErr = err
-			return
-		}
-		cleanup = cleanup.add(s.htlcSwitch.Stop)
-
 		if err := s.chainArb.Start(); err != nil {
 			startErr = err
 			return
@@ -1816,6 +1807,12 @@ func (s *server) Start() error {
 		}
 		cleanup = cleanup.add(s.sphinx.Stop)
 
+		if err := s.htlcSwitch.Start(); err != nil {
+			startErr = err
+			return
+		}
+		cleanup = cleanup.add(s.htlcSwitch.Stop)
+
 		if err := s.chanStatusMgr.Start(); err != nil {
 			startErr = err
 			return
diff --git a/sweep/sweeper.go b/sweep/sweeper.go
index fa46a4dc1..af067445a 100644
--- a/sweep/sweeper.go
+++ b/sweep/sweeper.go
@@ -338,7 +338,7 @@ func (s *UtxoSweeper) Start() error {
 		return nil
 	}
 
-	log.Info("Sweeper starting")
+	log.Tracef("Sweeper starting")
 
 	// Retrieve last published tx from database.
 	lastTx, err := s.cfg.Store.GetLastPublishedTx()
diff --git a/watchtower/wtclient/client.go b/watchtower/wtclient/client.go
index eabf00bea..f81284d53 100644
--- a/watchtower/wtclient/client.go
+++ b/watchtower/wtclient/client.go
@@ -439,7 +439,7 @@ func (c *TowerClient) buildHighestCommitHeights() {
 func (c *TowerClient) Start() error {
 	var err error
 	c.started.Do(func() {
-		c.log.Infof("Watchtower client starting")
+		c.log.Infof("Starting watchtower client")
 
 		// First, restart a session queue for any sessions that have
 		// committed but unacked state updates. This ensures that these


Test patch:
diff --git a/contractcourt/channel_arbitrator_test.go b/contractcourt/channel_arbitrator_test.go
index 03141ee91..bd9461739 100644
--- a/contractcourt/channel_arbitrator_test.go
+++ b/contractcourt/channel_arbitrator_test.go
@@ -829,7 +829,11 @@ func TestChannelArbitratorLocalForceClosePendingHtlc(t *testing.T) {
 	}
 	defer chanArb.Stop()
 
+	// Create htlcUpdates channel.
+	htlcUpdates := make(chan *ContractUpdate)
+
 	signals := &ContractSignals{
+		HtlcUpdates: htlcUpdates,
 		ShortChanID: lnwire.ShortChannelID{},
 	}
 	chanArb.UpdateContractSignals(signals)
@@ -860,12 +864,10 @@ func TestChannelArbitratorLocalForceClosePendingHtlc(t *testing.T) {
 		htlc, outgoingDustHtlc, incomingDustHtlc,
 	}
 
-	newUpdate := &ContractUpdate{
+	htlcUpdates <- &ContractUpdate{
 		HtlcKey: LocalHtlcSet,
 		Htlcs:   htlcSet,
 	}
-	err = chanArb.notifyContractUpdate(newUpdate)
-	require.NoError(t, err)
 
 	errChan := make(chan error, 1)
 	respChan := make(chan *wire.MsgTx, 1)
@@ -1847,7 +1849,9 @@ func TestChannelArbitratorDanglingCommitForceClose(t *testing.T) {
 			// Now that our channel arb has started, we'll set up
 			// its contract signals channel so we can send it
 			// various HTLC updates for this test.
+			htlcUpdates := make(chan *ContractUpdate)
 			signals := &ContractSignals{
+				HtlcUpdates: htlcUpdates,
 				ShortChanID: lnwire.ShortChannelID{},
 			}
 			chanArb.UpdateContractSignals(signals)
@@ -1868,12 +1872,10 @@ func TestChannelArbitratorDanglingCommitForceClose(t *testing.T) {
 				HtlcIndex:     htlcIndex,
 				RefundTimeout: htlcExpiry,
 			}
-			newUpdate := &ContractUpdate{
+			htlcUpdates <- &ContractUpdate{
 				HtlcKey: htlcKey,
 				Htlcs:   []channeldb.HTLC{danglingHTLC},
 			}
-			err = chanArb.notifyContractUpdate(newUpdate)
-			require.NoError(t, err)
 
 			// At this point, we now have a split commitment state
 			// from the PoV of the channel arb. There's now an HTLC
@@ -2042,7 +2044,9 @@ func TestChannelArbitratorPendingExpiredHTLC(t *testing.T) {
 	// Now that our channel arb has started, we'll set up
 	// its contract signals channel so we can send it
 	// various HTLC updates for this test.
+	htlcUpdates := make(chan *ContractUpdate)
 	signals := &ContractSignals{
+		HtlcUpdates: htlcUpdates,
 		ShortChanID: lnwire.ShortChannelID{},
 	}
 	chanArb.UpdateContractSignals(signals)
@@ -2057,12 +2061,10 @@ func TestChannelArbitratorPendingExpiredHTLC(t *testing.T) {
 		HtlcIndex:     htlcIndex,
 		RefundTimeout: htlcExpiry,
 	}
-	newUpdate := &ContractUpdate{
+	htlcUpdates <- &ContractUpdate{
 		HtlcKey: RemoteHtlcSet,
 		Htlcs:   []channeldb.HTLC{pendingHTLC},
 	}
-	err = chanArb.notifyContractUpdate(newUpdate)
-	require.NoError(t, err)
 
 	// We will advance the uptime to 10 seconds which should be still within
 	// the grace period and should not trigger going to chain.
@@ -2529,7 +2531,11 @@ func TestChannelArbitratorAnchors(t *testing.T) {
 		}
 	}()
 
+	// Create htlcUpdates channel.
+	htlcUpdates := make(chan *ContractUpdate)
+
 	signals := &ContractSignals{
+		HtlcUpdates: htlcUpdates,
 		ShortChanID: lnwire.ShortChannelID{},
 	}
 	chanArb.UpdateContractSignals(signals)
@@ -2553,7 +2559,7 @@ func TestChannelArbitratorAnchors(t *testing.T) {
 
 	// We now send two HTLC updates, one for local HTLC set and the other
 	// for remote HTLC set.
-	newUpdate := &ContractUpdate{
+	htlcUpdates <- &ContractUpdate{
 		HtlcKey: LocalHtlcSet,
 		// This will make the deadline of the local anchor resolution
 		// to be htlcWithPreimage's CLTV minus heightHint since the
@@ -2561,18 +2567,13 @@ func TestChannelArbitratorAnchors(t *testing.T) {
 		// preimage available.
 		Htlcs: []channeldb.HTLC{htlc, htlcWithPreimage},
 	}
-	err = chanArb.notifyContractUpdate(newUpdate)
-	require.NoError(t, err)
-
-	newUpdate = &ContractUpdate{
+	htlcUpdates <- &ContractUpdate{
 		HtlcKey: RemoteHtlcSet,
 		// This will make the deadline of the remote anchor resolution
 		// to be htlcWithPreimage's CLTV minus heightHint because the
 		// incoming HTLC (toRemoteHTLCs) has a lower CLTV.
 		Htlcs: []channeldb.HTLC{htlc, htlcWithPreimage},
 	}
-	err = chanArb.notifyContractUpdate(newUpdate)
-	require.NoError(t, err)
 
 	errChan := make(chan error, 1)
 	respChan := make(chan *wire.MsgTx, 1)
diff --git a/htlcswitch/link_test.go b/htlcswitch/link_test.go
index 39d215005..6170d3622 100644
--- a/htlcswitch/link_test.go
+++ b/htlcswitch/link_test.go
@@ -1944,17 +1944,6 @@ func newSingleLinkTestHarness(chanAmt, chanReserve btcutil.Amount) (
 		return nil, nil, nil, nil, nil, nil, err
 	}
 
-	notifyUpdateChan := make(chan *contractcourt.ContractUpdate)
-	doneChan := make(chan struct{})
-	notifyContractUpdate := func(u *contractcourt.ContractUpdate) error {
-		select {
-		case notifyUpdateChan <- u:
-		case <-doneChan:
-		}
-
-		return nil
-	}
-
 	// Instantiate with a long interval, so that we can precisely control
 	// the firing via force feeding.
 	bticker := ticker.NewForce(time.Hour)
@@ -1978,13 +1967,12 @@ func newSingleLinkTestHarness(chanAmt, chanReserve btcutil.Amount) (
 		UpdateContractSignals: func(*contractcourt.ContractSignals) error {
 			return nil
 		},
-		NotifyContractUpdate: notifyContractUpdate,
-		Registry:             invoiceRegistry,
-		FeeEstimator:         newMockFeeEstimator(),
-		ChainEvents:          &contractcourt.ChainEventSubscription{},
-		BatchTicker:          bticker,
-		FwdPkgGCTicker:       ticker.NewForce(15 * time.Second),
-		PendingCommitTicker:  ticker.New(time.Minute),
+		Registry:            invoiceRegistry,
+		FeeEstimator:        newMockFeeEstimator(),
+		ChainEvents:         &contractcourt.ChainEventSubscription{},
+		BatchTicker:         bticker,
+		FwdPkgGCTicker:      ticker.NewForce(15 * time.Second),
+		PendingCommitTicker: ticker.New(time.Minute),
 		// Make the BatchSize and Min/MaxFeeUpdateTimeout large enough
 		// to not trigger commit updates automatically during tests.
 		BatchSize:             10000,
@@ -2005,9 +1993,8 @@ func newSingleLinkTestHarness(chanAmt, chanReserve btcutil.Amount) (
 	go func() {
 		for {
 			select {
-			case <-notifyUpdateChan:
+			case <-aliceLink.(*channelLink).htlcUpdates:
 			case <-aliceLink.(*channelLink).quit:
-				close(doneChan)
 				return
 			}
 		}
@@ -4495,17 +4482,6 @@ func (h *persistentLinkHarness) restartLink(
 		}
 	}
 
-	notifyUpdateChan := make(chan *contractcourt.ContractUpdate)
-	doneChan := make(chan struct{})
-	notifyContractUpdate := func(u *contractcourt.ContractUpdate) error {
-		select {
-		case notifyUpdateChan <- u:
-		case <-doneChan:
-		}
-
-		return nil
-	}
-
 	// Instantiate with a long interval, so that we can precisely control
 	// the firing via force feeding.
 	bticker := ticker.NewForce(time.Hour)
@@ -4529,13 +4505,12 @@ func (h *persistentLinkHarness) restartLink(
 		UpdateContractSignals: func(*contractcourt.ContractSignals) error {
 			return nil
 		},
-		NotifyContractUpdate: notifyContractUpdate,
-		Registry:             h.coreLink.cfg.Registry,
-		FeeEstimator:         newMockFeeEstimator(),
-		ChainEvents:          &contractcourt.ChainEventSubscription{},
-		BatchTicker:          bticker,
-		FwdPkgGCTicker:       ticker.New(5 * time.Second),
-		PendingCommitTicker:  ticker.New(time.Minute),
+		Registry:            h.coreLink.cfg.Registry,
+		FeeEstimator:        newMockFeeEstimator(),
+		ChainEvents:         &contractcourt.ChainEventSubscription{},
+		BatchTicker:         bticker,
+		FwdPkgGCTicker:      ticker.New(5 * time.Second),
+		PendingCommitTicker: ticker.New(time.Minute),
 		// Make the BatchSize and Min/MaxFeeUpdateTimeout large enough
 		// to not trigger commit updates automatically during tests.
 		BatchSize:           10000,
@@ -4559,9 +4534,8 @@ func (h *persistentLinkHarness) restartLink(
 	go func() {
 		for {
 			select {
-			case <-notifyUpdateChan:
+			case <-aliceLink.(*channelLink).htlcUpdates:
 			case <-aliceLink.(*channelLink).quit:
-				close(doneChan)
 				return
 			}
 		}
@@ -5500,8 +5474,8 @@ func TestChannelLinkFail(t *testing.T) {
 			false,
 		},
 		{
-			// Test that we force close the channel if we receive
-			// an invalid Settle message.
+			// Test that we don't force close the channel if we
+			// receive an invalid Settle message.
 			func(c *channelLink) {
 			},
 			func(t *testing.T, c *channelLink, _ *lnwallet.LightningChannel) {
@@ -5513,7 +5487,7 @@ func TestChannelLinkFail(t *testing.T) {
 				}
 				c.HandleChannelUpdate(htlcSettle)
 			},
-			true,
+			false,
 			false,
 		},
 		{
@@ -6684,6 +6658,179 @@ func TestShutdownIfChannelClean(t *testing.T) {
 	}
 }
 
+// TestPipelineSettle tests that a link should only pipeline a settle if the
+// related add is fully locked-in meaning it is on both sides' commitment txns.
+func TestPipelineSettle(t *testing.T) {
+	t.Parallel()
+
+	const chanAmt = btcutil.SatoshiPerBitcoin * 5
+	const chanReserve = btcutil.SatoshiPerBitcoin * 1
+	aliceLink, bobChannel, _, start, cleanUp, restore, err :=
+		newSingleLinkTestHarness(chanAmt, chanReserve)
+	require.NoError(t, err)
+	defer cleanUp()
+
+	alice := newPersistentLinkHarness(
+		t, aliceLink, nil, restore,
+	)
+
+	linkErrors := make(chan LinkFailureError, 1)
+
+	// Modify OnChannelFailure so we are notified when the link is failed.
+	alice.coreLink.cfg.OnChannelFailure = func(_ lnwire.ChannelID,
+		_ lnwire.ShortChannelID, linkErr LinkFailureError) {
+
+		linkErrors <- linkErr
+	}
+
+	// Modify ForwardPackets so we are notified if a settle packet is
+	// erroneously forwarded. If the forwardChan is closed before the last
+	// step, then the test will fail.
+	forwardChan := make(chan struct{})
+	fwdPkts := func(c chan struct{}, hp ...*htlcPacket) error {
+		close(forwardChan)
+		return nil
+	}
+	alice.coreLink.cfg.ForwardPackets = fwdPkts
+
+	// Put Alice in ExitSettle mode, so we can simulate a multi-hop route
+	// without actually doing so. This allows us to test the locked-in add
+	// logic without having the add being removed by Alice sending a
+	// settle.
+	alice.coreLink.cfg.HodlMask = hodl.Mask(hodl.ExitSettle)
+
+	err = start()
+	require.NoError(t, err)
+
+	ctx := linkTestContext{
+		t:          t,
+		aliceLink:  alice.link,
+		bobChannel: bobChannel,
+		aliceMsgs:  alice.msgs,
+	}
+
+	// First lock in an HTLC from Bob to Alice.
+	htlc1, invoice1 := generateHtlcAndInvoice(t, 0)
+	preimage1 := invoice1.Terms.PaymentPreimage
+
+	// Add the invoice to Alice's registry so she expects it.
+	aliceReg := alice.coreLink.cfg.Registry.(*mockInvoiceRegistry)
+	err = aliceReg.AddInvoice(*invoice1, htlc1.PaymentHash)
+	require.NoError(t, err)
+
+	// <---add-----
+	ctx.sendHtlcBobToAlice(htlc1)
+	// <---sig-----
+	ctx.sendCommitSigBobToAlice(1)
+	// ----rev---->
+	ctx.receiveRevAndAckAliceToBob()
+	// ----sig---->
+	ctx.receiveCommitSigAliceToBob(1)
+	// <---rev-----
+	ctx.sendRevAndAckBobToAlice()
+
+	// Bob will send the preimage for the HTLC he just sent. This will test
+	// the check that the HTLC is locked-in. The channel should not be
+	// force closed if everything is working correctly.
+	settle1 := &lnwire.UpdateFulfillHTLC{
+		ID:              0,
+		PaymentPreimage: *preimage1,
+	}
+	ctx.aliceLink.HandleChannelUpdate(settle1)
+
+	// ForceClose should be false.
+	select {
+	case linkErr := <-linkErrors:
+		require.False(t, linkErr.ForceClose)
+	case <-forwardChan:
+		t.Fatal("packet was erroneously forwarded")
+	}
+
+	// Restart Alice's link with the hodl.ExitSettle and hodl.Commit flags.
+	alice.restart(false, false, hodl.ExitSettle, hodl.Commit)
+	ctx.aliceLink = alice.link
+	ctx.aliceMsgs = alice.msgs
+
+	alice.coreLink.cfg.OnChannelFailure = func(_ lnwire.ChannelID,
+		_ lnwire.ShortChannelID, linkErr LinkFailureError) {
+
+		linkErrors <- linkErr
+	}
+	alice.coreLink.cfg.ForwardPackets = fwdPkts
+
+	// Alice will now send an HTLC to Bob, but won't sign a commitment for
+	// it. This HTLC will have the same payment hash as the one above.
+	htlc2 := htlc1
+
+	// ----add--->
+	ctx.sendHtlcAliceToBob(0, htlc2)
+	ctx.receiveHtlcAliceToBob()
+
+	// Now Bob will send a settle backwards before the HTLC is locked in
+	// and the link should be failed again.
+	settle2 := &lnwire.UpdateFulfillHTLC{
+		ID:              0,
+		PaymentPreimage: *preimage1,
+	}
+	ctx.aliceLink.HandleChannelUpdate(settle2)
+
+	// ForceClose should be false.
+	select {
+	case linkErr := <-linkErrors:
+		require.False(t, linkErr.ForceClose)
+	case <-forwardChan:
+		t.Fatal("packet was erroneously forwarded")
+	}
+
+	// Restart Alice's link without the hodl.Commit flag.
+	alice.restart(false, false, hodl.ExitSettle)
+	ctx.aliceLink = alice.link
+	ctx.aliceMsgs = alice.msgs
+
+	alice.coreLink.cfg.OnChannelFailure = func(_ lnwire.ChannelID,
+		_ lnwire.ShortChannelID, linkErr LinkFailureError) {
+
+		linkErrors <- linkErr
+	}
+	alice.coreLink.cfg.ForwardPackets = fwdPkts
+
+	// Alice's mailbox should give the link the HTLC to send again.
+	select {
+	case msg := <-ctx.aliceMsgs:
+		_, ok := msg.(*lnwire.UpdateAddHTLC)
+		require.True(t, ok)
+	case <-time.After(5 * time.Second):
+		t.Fatal("did not receive htlc from alice")
+	}
+
+	// Trigger the BatchTicker.
+	select {
+	case alice.batchTicker <- time.Now():
+	case <-time.After(5 * time.Second):
+		t.Fatalf("could not force commit sig")
+	}
+
+	// ----sig--->
+	ctx.receiveCommitSigAliceToBob(2)
+	// <---rev----
+	ctx.sendRevAndAckBobToAlice()
+	// <---sig----
+	ctx.sendCommitSigBobToAlice(2)
+	// ----rev--->
+	ctx.receiveRevAndAckAliceToBob()
+
+	// Bob should now be able to send the settle to Alice without making
+	// the link fail.
+	ctx.aliceLink.HandleChannelUpdate(settle2)
+
+	select {
+	case <-linkErrors:
+		t.Fatal("should not have received a link error")
+	case <-forwardChan:
+		// success
+	}
+}
+
 // assertFailureCode asserts that an error is of type ClearTextError and that
 // the failure code is as expected.
 func assertFailureCode(t *testing.T, err error, code lnwire.FailCode) {
diff --git a/htlcswitch/test_utils.go b/htlcswitch/test_utils.go
index 94c301be6..3fa57f601 100644
--- a/htlcswitch/test_utils.go
+++ b/htlcswitch/test_utils.go
@@ -1122,17 +1122,6 @@ func (h *hopNetwork) createChannelLink(server, peer *mockServer,
 		maxFeeUpdateTimeout = 40 * time.Minute
 	)
 
-	notifyUpdateChan := make(chan *contractcourt.ContractUpdate)
-	doneChan := make(chan struct{})
-	notifyContractUpdate := func(u *contractcourt.ContractUpdate) error {
-		select {
-		case notifyUpdateChan <- u:
-		case <-doneChan:
-		}
-
-		return nil
-	}
-
 	link := NewChannelLink(
 		ChannelLinkConfig{
 			Switch:             server.htlcSwitch,
@@ -1153,7 +1142,6 @@ func (h *hopNetwork) createChannelLink(server, peer *mockServer,
 			UpdateContractSignals: func(*contractcourt.ContractSignals) error {
 				return nil
 			},
-			NotifyContractUpdate:    notifyContractUpdate,
 			ChainEvents:             &contractcourt.ChainEventSubscription{},
 			SyncStates:              true,
 			BatchSize:               10,
@@ -1181,9 +1169,8 @@ func (h *hopNetwork) createChannelLink(server, peer *mockServer,
 	go func() {
 		for {
 			select {
-			case <-notifyUpdateChan:
+			case <-link.(*channelLink).htlcUpdates:
 			case <-link.(*channelLink).quit:
-				close(doneChan)
 				return
 			}
 		}