Merge pull request #5603 from Crypt-iQ/peer_switch_refactor_08022021

htlcswitch+peer: introduce packetHandler, ChannelUpdateHandler interfaces for cleaner separation

docs/release-notes/release-notes-0.14.0.md

@@ -85,6 +85,8 @@ you.
 
 ## Code cleanup, refactor, typo fixes
 
+* [Refactor the interaction between the `htlcswitch` and `peer` packages for cleaner separation.](https://github.com/lightningnetwork/lnd/pull/5603)
+
 * [Unused error check 
   removed](https://github.com/lightningnetwork/lnd/pull/5537).
 
@@ -147,6 +149,7 @@ change](https://github.com/lightningnetwork/lnd/pull/5613).
 # Contributors (Alphabetical Order)
 * Andras Banki-Horvath
 * ErikEk
+* Eugene Siegel
 * Martin Habovstiak
 * Zero-1729
 * Oliver Gugger

htlcswitch/interfaces.go

@@ -41,6 +41,55 @@ type InvoiceDatabase interface {
 	HodlUnsubscribeAll(subscriber chan<- interface{})
 }
 
+// packetHandler is an interface used exclusively by the Switch to handle
+// htlcPacket and pass them to the link implementation.
+type packetHandler interface {
+	// handleSwitchPacket handles the switch packets. These packets might
+	// be forwarded to us from another channel link in case the htlc
+	// update came from another peer or if the update was created by user
+	// initially.
+	//
+	// NOTE: This function should block as little as possible.
+	handleSwitchPacket(*htlcPacket) error
+
+	// handleLocalAddPacket handles a locally-initiated UpdateAddHTLC
+	// packet. It will be processed synchronously.
+	handleLocalAddPacket(*htlcPacket) error
+}
+
+// ChannelUpdateHandler is an interface that provides methods that allow
+// sending lnwire.Message to the underlying link as well as querying state.
+type ChannelUpdateHandler interface {
+	// HandleChannelUpdate handles the htlc requests as settle/add/fail
+	// which sent to us from remote peer we have a channel with.
+	//
+	// NOTE: This function MUST be non-blocking (or block as little as
+	// possible).
+	HandleChannelUpdate(lnwire.Message)
+
+	// ChanID returns the channel ID for the channel link. The channel ID
+	// is a more compact representation of a channel's full outpoint.
+	ChanID() lnwire.ChannelID
+
+	// Bandwidth returns the amount of milli-satoshis which current link
+	// might pass through channel link. The value returned from this method
+	// represents the up to date available flow through the channel. This
+	// takes into account any forwarded but un-cleared HTLC's, and any
+	// HTLC's which have been set to the over flow queue.
+	Bandwidth() lnwire.MilliSatoshi
+
+	// EligibleToForward returns a bool indicating if the channel is able
+	// to actively accept requests to forward HTLC's. A channel may be
+	// active, but not able to forward HTLC's if it hasn't yet finalized
+	// the pre-channel operation protocol with the remote peer. The switch
+	// will use this function in forwarding decisions accordingly.
+	EligibleToForward() bool
+
+	// MayAddOutgoingHtlc returns an error if we may not add an outgoing
+	// htlc to the channel.
+	MayAddOutgoingHtlc() error
+}
+
 // ChannelLink is an interface which represents the subsystem for managing the
 // incoming htlc requests, applying the changes to the channel, and also
 // propagating/forwarding it to htlc switch.
@@ -62,33 +111,15 @@ type InvoiceDatabase interface {
 type ChannelLink interface {
 	// TODO(roasbeef): modify interface to embed mail boxes?
 
-	// HandleSwitchPacket handles the switch packets. This packets might be
-	// forwarded to us from another channel link in case the htlc update
-	// came from another peer or if the update was created by user
-	// initially.
-	//
-	// NOTE: This function MUST be non-blocking (or block as little as
-	// possible).
-	HandleSwitchPacket(*htlcPacket) error
+	// Embed the packetHandler interface.
+	packetHandler
 
-	// HandleLocalAddPacket handles a locally-initiated UpdateAddHTLC
-	// packet. It will be processed synchronously.
-	HandleLocalAddPacket(*htlcPacket) error
-
-	// HandleChannelUpdate handles the htlc requests as settle/add/fail
-	// which sent to us from remote peer we have a channel with.
-	//
-	// NOTE: This function MUST be non-blocking (or block as little as
-	// possible).
-	HandleChannelUpdate(lnwire.Message)
+	// Embed the ChannelUpdateHandler interface.
+	ChannelUpdateHandler
 
 	// ChannelPoint returns the channel outpoint for the channel link.
 	ChannelPoint() *wire.OutPoint
 
-	// ChanID returns the channel ID for the channel link. The channel ID
-	// is a more compact representation of a channel's full outpoint.
-	ChanID() lnwire.ChannelID
-
 	// ShortChanID returns the short channel ID for the channel link. The
 	// short channel ID encodes the exact location in the main chain that
 	// the original funding output can be found.
@@ -123,13 +154,6 @@ type ChannelLink interface {
 	CheckHtlcTransit(payHash [32]byte, amt lnwire.MilliSatoshi,
 		timeout uint32, heightNow uint32) *LinkError
 
-	// Bandwidth returns the amount of milli-satoshis which current link
-	// might pass through channel link. The value returned from this method
-	// represents the up to date available flow through the channel. This
-	// takes into account any forwarded but un-cleared HTLC's, and any
-	// HTLC's which have been set to the over flow queue.
-	Bandwidth() lnwire.MilliSatoshi
-
 	// Stats return the statistics of channel link. Number of updates,
 	// total sent/received milli-satoshis.
 	Stats() (uint64, lnwire.MilliSatoshi, lnwire.MilliSatoshi)
@@ -138,17 +162,6 @@ type ChannelLink interface {
 	// the channel link opened.
 	Peer() lnpeer.Peer
 
-	// EligibleToForward returns a bool indicating if the channel is able
-	// to actively accept requests to forward HTLC's. A channel may be
-	// active, but not able to forward HTLC's if it hasn't yet finalized
-	// the pre-channel operation protocol with the remote peer. The switch
-	// will use this function in forwarding decisions accordingly.
-	EligibleToForward() bool
-
-	// MayAddOutgoingHtlc returns an error if we may not add an outgoing
-	// htlc to the channel.
-	MayAddOutgoingHtlc() error
-
 	// AttachMailBox delivers an active MailBox to the link. The MailBox may
 	// have buffered messages.
 	AttachMailBox(MailBox)

htlcswitch/link.go

@@ -134,6 +134,9 @@ type ChannelLinkConfig struct {
 	// TODO(conner): remove after refactoring htlcswitch testing framework.
 	Switch *Switch
 
+	// BestHeight returns the best known height.
+	BestHeight func() uint32
+
 	// ForwardPackets attempts to forward the batch of htlcs through the
 	// switch. The function returns and error in case it fails to send one or
 	// more packets. The link's quit signal should be provided to allow
@@ -2384,23 +2387,23 @@ func (l *channelLink) String() string {
 	return l.channel.ChannelPoint().String()
 }
 
-// HandleSwitchPacket handles the switch packets. This packets which might be
+// handleSwitchPacket handles the switch packets. This packets which might be
 // forwarded to us from another channel link in case the htlc update came from
 // another peer or if the update was created by user
 //
-// NOTE: Part of the ChannelLink interface.
-func (l *channelLink) HandleSwitchPacket(pkt *htlcPacket) error {
+// NOTE: Part of the packetHandler interface.
+func (l *channelLink) handleSwitchPacket(pkt *htlcPacket) error {
 	l.log.Tracef("received switch packet inkey=%v, outkey=%v",
 		pkt.inKey(), pkt.outKey())
 
 	return l.mailBox.AddPacket(pkt)
 }
 
-// HandleLocalAddPacket handles a locally-initiated UpdateAddHTLC packet. It
+// handleLocalAddPacket handles a locally-initiated UpdateAddHTLC packet. It
 // will be processed synchronously.
 //
-// NOTE: Part of the ChannelLink interface.
-func (l *channelLink) HandleLocalAddPacket(pkt *htlcPacket) error {
+// NOTE: Part of the packetHandler interface.
+func (l *channelLink) handleLocalAddPacket(pkt *htlcPacket) error {
 	l.log.Tracef("received switch packet outkey=%v", pkt.outKey())
 
 	// Create a buffered result channel to prevent the link from blocking.
@@ -2677,7 +2680,7 @@ func (l *channelLink) processRemoteAdds(fwdPkg *channeldb.FwdPkg,
 			continue
 		}
 
-		heightNow := l.cfg.Switch.BestHeight()
+		heightNow := l.cfg.BestHeight()
 
 		pld, err := chanIterator.HopPayload()
 		if err != nil {

htlcswitch/link_isolated_test.go

@@ -56,7 +56,7 @@ func (l *linkTestContext) sendHtlcAliceToBob(htlcID int,
 		l.t.Fatalf("expected 1 adds, found %d", len(fwdActions.Adds))
 	}
 
-	err = l.aliceLink.HandleSwitchPacket(&htlcPacket{
+	err = l.aliceLink.handleSwitchPacket(&htlcPacket{
 		incomingHTLCID: uint64(htlcID),
 		htlc:           htlc,
 	})

htlcswitch/link_test.go

@@ -1951,6 +1951,7 @@ func newSingleLinkTestHarness(chanAmt, chanReserve btcutil.Amount) (
 		FwrdingPolicy:      globalPolicy,
 		Peer:               alicePeer,
 		Switch:             aliceSwitch,
+		BestHeight:         aliceSwitch.BestHeight,
 		Circuits:           aliceSwitch.CircuitModifier(),
 		ForwardPackets:     aliceSwitch.ForwardPackets,
 		DecodeHopIterators: decoder.DecodeHopIterators,
@@ -2246,7 +2247,7 @@ func TestChannelLinkBandwidthConsistency(t *testing.T) {
 	}
 
 	addPkt.circuit = &circuit
-	if err := aliceLink.HandleSwitchPacket(&addPkt); err != nil {
+	if err := aliceLink.handleSwitchPacket(&addPkt); err != nil {
 		t.Fatalf("unable to handle switch packet: %v", err)
 	}
 	time.Sleep(time.Millisecond * 500)
@@ -2326,7 +2327,7 @@ func TestChannelLinkBandwidthConsistency(t *testing.T) {
 	}
 
 	addPkt.circuit = &circuit
-	if err := aliceLink.HandleSwitchPacket(&addPkt); err != nil {
+	if err := aliceLink.handleSwitchPacket(&addPkt); err != nil {
 		t.Fatalf("unable to handle switch packet: %v", err)
 	}
 	time.Sleep(time.Millisecond * 500)
@@ -2466,7 +2467,7 @@ func TestChannelLinkBandwidthConsistency(t *testing.T) {
 		obfuscator: NewMockObfuscator(),
 	}
 
-	if err := aliceLink.HandleSwitchPacket(&settlePkt); err != nil {
+	if err := aliceLink.handleSwitchPacket(&settlePkt); err != nil {
 		t.Fatalf("unable to handle switch packet: %v", err)
 	}
 	time.Sleep(time.Millisecond * 500)
@@ -2570,7 +2571,7 @@ func TestChannelLinkBandwidthConsistency(t *testing.T) {
 		obfuscator: NewMockObfuscator(),
 	}
 
-	if err := aliceLink.HandleSwitchPacket(&failPkt); err != nil {
+	if err := aliceLink.handleSwitchPacket(&failPkt); err != nil {
 		t.Fatalf("unable to handle switch packet: %v", err)
 	}
 	time.Sleep(time.Millisecond * 500)
@@ -2708,7 +2709,7 @@ func TestChannelLinkTrimCircuitsPending(t *testing.T) {
 	// Since both were committed successfully, we will now deliver them to
 	// Alice's link.
 	for _, addPkt := range addPkts[:halfHtlcs] {
-		if err := alice.link.HandleSwitchPacket(addPkt); err != nil {
+		if err := alice.link.handleSwitchPacket(addPkt); err != nil {
 			t.Fatalf("unable to handle switch packet: %v", err)
 		}
 	}
@@ -2793,7 +2794,7 @@ func TestChannelLinkTrimCircuitsPending(t *testing.T) {
 	// Deliver the latter two HTLCs to Alice's links so that they can be
 	// processed and added to the in-memory commitment state.
 	for _, addPkt := range addPkts[halfHtlcs:] {
-		if err := alice.link.HandleSwitchPacket(addPkt); err != nil {
+		if err := alice.link.handleSwitchPacket(addPkt); err != nil {
 			t.Fatalf("unable to handle switch packet: %v", err)
 		}
 	}
@@ -2988,7 +2989,7 @@ func TestChannelLinkTrimCircuitsNoCommit(t *testing.T) {
 	// Since both were committed successfully, we will now deliver them to
 	// Alice's link.
 	for _, addPkt := range addPkts[:halfHtlcs] {
-		if err := alice.link.HandleSwitchPacket(addPkt); err != nil {
+		if err := alice.link.handleSwitchPacket(addPkt); err != nil {
 			t.Fatalf("unable to handle switch packet: %v", err)
 		}
 	}
@@ -3081,7 +3082,7 @@ func TestChannelLinkTrimCircuitsNoCommit(t *testing.T) {
 
 	// Deliver the last two HTLCs to the link via Alice's mailbox.
 	for _, addPkt := range addPkts[halfHtlcs:] {
-		if err := alice.link.HandleSwitchPacket(addPkt); err != nil {
+		if err := alice.link.handleSwitchPacket(addPkt); err != nil {
 			t.Fatalf("unable to handle switch packet: %v", err)
 		}
 	}
@@ -3248,7 +3249,7 @@ func TestChannelLinkTrimCircuitsRemoteCommit(t *testing.T) {
 	// Since both were committed successfully, we will now deliver them to
 	// Alice's link.
 	for _, addPkt := range addPkts {
-		if err := alice.link.HandleSwitchPacket(addPkt); err != nil {
+		if err := alice.link.handleSwitchPacket(addPkt); err != nil {
 			t.Fatalf("unable to handle switch packet: %v", err)
 		}
 	}
@@ -3389,7 +3390,7 @@ func TestChannelLinkBandwidthChanReserve(t *testing.T) {
 		t.Fatalf("unable to commit circuit: %v", err)
 	}
 
-	aliceLink.HandleSwitchPacket(addPkt)
+	_ = aliceLink.handleSwitchPacket(addPkt)
 	time.Sleep(time.Millisecond * 100)
 	assertLinkBandwidth(t, aliceLink, aliceStartingBandwidth-htlcAmt-htlcFee)
 
@@ -4454,6 +4455,7 @@ func (h *persistentLinkHarness) restartLink(
 		FwrdingPolicy:      globalPolicy,
 		Peer:               alicePeer,
 		Switch:             aliceSwitch,
+		BestHeight:         aliceSwitch.BestHeight,
 		Circuits:           aliceSwitch.CircuitModifier(),
 		ForwardPackets:     aliceSwitch.ForwardPackets,
 		DecodeHopIterators: decoder.DecodeHopIterators,
@@ -5137,7 +5139,7 @@ func TestChannelLinkCleanupSpuriousResponses(t *testing.T) {
 		obfuscator:     NewMockObfuscator(),
 		htlc:           &lnwire.UpdateFailHTLC{},
 	}
-	aliceLink.HandleSwitchPacket(fail0)
+	_ = aliceLink.handleSwitchPacket(fail0)
 
 	//  Bob               Alice
 	//   |<----- fal-1 ------|
@@ -5197,7 +5199,7 @@ func TestChannelLinkCleanupSpuriousResponses(t *testing.T) {
 		obfuscator:     NewMockObfuscator(),
 		htlc:           &lnwire.UpdateFailHTLC{},
 	}
-	aliceLink.HandleSwitchPacket(fail1)
+	_ = aliceLink.handleSwitchPacket(fail1)
 
 	//  Bob               Alice
 	//   |<----- fal-1 ------|
@@ -5254,7 +5256,7 @@ func TestChannelLinkCleanupSpuriousResponses(t *testing.T) {
 	// this should trigger an attempt to cleanup the spurious response.
 	// However, we expect it to result in a NOP since it is still missing
 	// its sourceRef.
-	aliceLink.HandleSwitchPacket(fail0)
+	_ = aliceLink.handleSwitchPacket(fail0)
 
 	// Allow the link enough time to process and reject the duplicate
 	// packet, we'll also check that this doesn't trigger Alice to send the
@@ -5309,7 +5311,7 @@ func TestChannelLinkCleanupSpuriousResponses(t *testing.T) {
 		obfuscator:     NewMockObfuscator(),
 		htlc:           &lnwire.UpdateFailHTLC{},
 	}
-	aliceLink.HandleSwitchPacket(fail0)
+	_ = aliceLink.handleSwitchPacket(fail0)
 
 	// Allow the link enough time to process and reject the duplicate
 	// packet, we'll also check that this doesn't trigger Alice to send the

htlcswitch/mock.go

@@ -706,12 +706,12 @@ func newMockChannelLink(htlcSwitch *Switch, chanID lnwire.ChannelID,
 	}
 }
 
-func (f *mockChannelLink) HandleSwitchPacket(pkt *htlcPacket) error {
+func (f *mockChannelLink) handleSwitchPacket(pkt *htlcPacket) error {
 	f.mailBox.AddPacket(pkt)
 	return nil
 }
 
-func (f *mockChannelLink) HandleLocalAddPacket(pkt *htlcPacket) error {
+func (f *mockChannelLink) handleLocalAddPacket(pkt *htlcPacket) error {
 	_ = f.mailBox.AddPacket(pkt)
 	return nil
 }

htlcswitch/switch.go

@@ -509,7 +509,7 @@ func (s *Switch) SendHTLC(firstHop lnwire.ShortChannelID, attemptID uint64,
 		return linkErr
 	}
 
-	return link.HandleLocalAddPacket(packet)
+	return link.handleLocalAddPacket(packet)
 }
 
 // UpdateForwardingPolicies sends a message to the switch to update the
@@ -1101,7 +1101,7 @@ func (s *Switch) handlePacketForward(packet *htlcPacket) error {
 		// Send the packet to the destination channel link which
 		// manages the channel.
 		packet.outgoingChanID = destination.ShortChanID()
-		return destination.HandleSwitchPacket(packet)
+		return destination.handleSwitchPacket(packet)
 
 	case *lnwire.UpdateFailHTLC, *lnwire.UpdateFulfillHTLC:
 		// If the source of this packet has not been set, use the
@@ -1956,6 +1956,15 @@ func (s *Switch) Stop() error {
 	return nil
 }
 
+// CreateAndAddLink will create a link and then add it to the internal maps
+// when given a ChannelLinkConfig and LightningChannel.
+func (s *Switch) CreateAndAddLink(linkCfg ChannelLinkConfig,
+	lnChan *lnwallet.LightningChannel) error {
+
+	link := NewChannelLink(linkCfg, lnChan)
+	return s.AddLink(link)
+}
+
 // AddLink is used to initiate the handling of the add link command. The
 // request will be propagated and handled in the main goroutine.
 func (s *Switch) AddLink(link ChannelLink) error {
@@ -2022,7 +2031,9 @@ func (s *Switch) addLiveLink(link ChannelLink) {
 
 // GetLink is used to initiate the handling of the get link command. The
 // request will be propagated/handled to/in the main goroutine.
-func (s *Switch) GetLink(chanID lnwire.ChannelID) (ChannelLink, error) {
+func (s *Switch) GetLink(chanID lnwire.ChannelID) (ChannelUpdateHandler,
+	error) {
+
 	s.indexMtx.RLock()
 	defer s.indexMtx.RUnlock()
 
@@ -2164,11 +2175,26 @@ func (s *Switch) UpdateShortChanID(chanID lnwire.ChannelID) error {
 
 // GetLinksByInterface fetches all the links connected to a particular node
 // identified by the serialized compressed form of its public key.
-func (s *Switch) GetLinksByInterface(hop [33]byte) ([]ChannelLink, error) {
+func (s *Switch) GetLinksByInterface(hop [33]byte) ([]ChannelUpdateHandler,
+	error) {
+
 	s.indexMtx.RLock()
 	defer s.indexMtx.RUnlock()
 
-	return s.getLinks(hop)
+	var handlers []ChannelUpdateHandler
+
+	links, err := s.getLinks(hop)
+	if err != nil {
+		return nil, err
+	}
+
+	// Range over the returned []ChannelLink to convert them into
+	// []ChannelUpdateHandler.
+	for _, link := range links {
+		handlers = append(handlers, link)
+	}
+
+	return handlers, nil
 }
 
 // getLinks is function which returns the channel links of the peer by hop

htlcswitch/test_utils.go

@@ -1119,6 +1119,7 @@ func (h *hopNetwork) createChannelLink(server, peer *mockServer,
 	link := NewChannelLink(
 		ChannelLinkConfig{
 			Switch:             server.htlcSwitch,
+			BestHeight:         server.htlcSwitch.BestHeight,
 			FwrdingPolicy:      h.globalPolicy,
 			Peer:               peer,
 			Circuits:           server.htlcSwitch.CircuitModifier(),

peer/brontide.go

@@ -177,7 +177,7 @@ type Config struct {
 
 	// Switch is a pointer to the htlcswitch. It is used to setup, get, and
 	// tear-down ChannelLinks.
-	Switch *htlcswitch.Switch
+	Switch messageSwitch
 
 	// InterceptSwitch is a pointer to the InterceptableSwitch, a wrapper around
 	// the regular Switch. We only export it here to pass ForwardPackets to the
@@ -813,7 +813,7 @@ func (p *Brontide) addLink(chanPoint *wire.OutPoint,
 		FetchLastChannelUpdate:  p.cfg.FetchLastChanUpdate,
 		HodlMask:                p.cfg.Hodl.Mask(),
 		Registry:                p.cfg.Invoices,
-		Switch:                  p.cfg.Switch,
+		BestHeight:              p.cfg.Switch.BestHeight,
 		Circuits:                p.cfg.Switch.CircuitModifier(),
 		ForwardPackets:          p.cfg.InterceptSwitch.ForwardPackets,
 		FwrdingPolicy:           *forwardingPolicy,
@@ -841,18 +841,17 @@ func (p *Brontide) addLink(chanPoint *wire.OutPoint,
 		HtlcNotifier:            p.cfg.HtlcNotifier,
 	}
 
-	link := htlcswitch.NewChannelLink(linkCfg, lnChan)
-
 	// Before adding our new link, purge the switch of any pending or live
 	// links going by the same channel id. If one is found, we'll shut it
 	// down to ensure that the mailboxes are only ever under the control of
 	// one link.
-	p.cfg.Switch.RemoveLink(link.ChanID())
+	chanID := lnwire.NewChanIDFromOutPoint(chanPoint)
+	p.cfg.Switch.RemoveLink(chanID)
 
 	// With the channel link created, we'll now notify the htlc switch so
 	// this channel can be used to dispatch local payments and also
 	// passively forward payments.
-	return p.cfg.Switch.AddLink(link)
+	return p.cfg.Switch.CreateAndAddLink(linkCfg, lnChan)
 }
 
 // maybeSendNodeAnn sends our node announcement to the remote peer if at least
@@ -1142,7 +1141,7 @@ func (ms *msgStream) AddMsg(msg lnwire.Message) {
 // ChannelLink to pass messages to. It accomplishes this by subscribing to
 // an ActiveLinkEvent which is emitted by the link when it first starts up.
 func waitUntilLinkActive(p *Brontide,
-	cid lnwire.ChannelID) htlcswitch.ChannelLink {
+	cid lnwire.ChannelID) htlcswitch.ChannelUpdateHandler {
 
 	// Subscribe to receive channel events.
 	//
@@ -1164,9 +1163,11 @@ func waitUntilLinkActive(p *Brontide,
 
 	// The link may already be active by this point, and we may have missed the
 	// ActiveLinkEvent. Check if the link exists.
-	link, _ := p.cfg.Switch.GetLink(cid)
-	if link != nil {
-		return link
+	links, _ := p.cfg.Switch.GetLinksByInterface(p.cfg.PubKeyBytes)
+	for _, link := range links {
+		if link.ChanID() == cid {
+			return link
+		}
 	}
 
 	// If the link is nil, we must wait for it to be active.
@@ -1194,8 +1195,16 @@ func waitUntilLinkActive(p *Brontide,
 			// The link shouldn't be nil as we received an
 			// ActiveLinkEvent. If it is nil, we return nil and the
 			// calling function should catch it.
-			link, _ = p.cfg.Switch.GetLink(cid)
-			return link
+			links, _ = p.cfg.Switch.GetLinksByInterface(
+				p.cfg.PubKeyBytes,
+			)
+			for _, link := range links {
+				if link.ChanID() == cid {
+					return link
+				}
+			}
+
+			return nil
 
 		case <-p.quit:
 			return nil
@@ -1211,7 +1220,7 @@ func waitUntilLinkActive(p *Brontide,
 // lookups.
 func newChanMsgStream(p *Brontide, cid lnwire.ChannelID) *msgStream {
 
-	var chanLink htlcswitch.ChannelLink
+	var chanLink htlcswitch.ChannelUpdateHandler
 
 	apply := func(msg lnwire.Message) {
 		// This check is fine because if the link no longer exists, it will

peer/interfaces.go

@@ -4,9 +4,36 @@ import (
 	"net"
 	"time"
 
+	"github.com/lightningnetwork/lnd/htlcswitch"
+	"github.com/lightningnetwork/lnd/lnwallet"
 	"github.com/lightningnetwork/lnd/lnwire"
 )
 
+// messageSwitch is an interface that abstracts managing the lifecycle of
+// abstract links. This is intended for peer package usage only.
+type messageSwitch interface {
+	// BestHeight returns the best height known to the messageSwitch.
+	BestHeight() uint32
+
+	// CircuitModifier returns a reference to the messageSwitch's internal
+	// CircuitModifier which abstracts the paths payments take and allows
+	// modifying them.
+	CircuitModifier() htlcswitch.CircuitModifier
+
+	// RemoveLink removes an abstract link given a ChannelID.
+	RemoveLink(cid lnwire.ChannelID)
+
+	// CreateAndAddLink creates an abstract link in the messageSwitch given
+	// a ChannelLinkConfig and raw LightningChannel pointer.
+	CreateAndAddLink(cfg htlcswitch.ChannelLinkConfig,
+		lnChan *lnwallet.LightningChannel) error
+
+	// GetLinksByInterface retrieves abstract links (represented by the
+	// ChannelUpdateHandler interface) based on the provided public key.
+	GetLinksByInterface(pub [33]byte) ([]htlcswitch.ChannelUpdateHandler,
+		error)
+}
+
 // LinkUpdater is an interface implemented by most messages in BOLT 2 that are
 // allowed to update the channel state.
 type LinkUpdater interface {

peer/test_utils.go

@@ -29,7 +29,6 @@ import (
 	"github.com/lightningnetwork/lnd/netann"
 	"github.com/lightningnetwork/lnd/queue"
 	"github.com/lightningnetwork/lnd/shachain"
-	"github.com/lightningnetwork/lnd/ticker"
 	"github.com/stretchr/testify/require"
 )
 
@@ -307,28 +306,7 @@ func createTestPeer(notifier chainntnfs.ChainNotifier,
 		},
 	}
 
-	_, currentHeight, err := chainIO.GetBestBlock()
-	if err != nil {
-		return nil, nil, nil, err
-	}
-
-	htlcSwitch, err := htlcswitch.New(htlcswitch.Config{
-		DB:             dbAlice,
-		SwitchPackager: channeldb.NewSwitchPackager(),
-		Notifier:       notifier,
-		FwdEventTicker: ticker.New(
-			htlcswitch.DefaultFwdEventInterval),
-		LogEventTicker: ticker.New(
-			htlcswitch.DefaultLogInterval),
-		AckEventTicker: ticker.New(
-			htlcswitch.DefaultAckInterval),
-	}, uint32(currentHeight))
-	if err != nil {
-		return nil, nil, nil, err
-	}
-	if err = htlcSwitch.Start(); err != nil {
-		return nil, nil, nil, err
-	}
+	htlcSwitch := &mockMessageSwitch{}
 
 	nodeSignerAlice := netann.NewNodeSigner(aliceKeySigner)
 
@@ -342,7 +320,7 @@ func createTestPeer(notifier chainntnfs.ChainNotifier,
 		Graph:                    dbAlice.ChannelGraph(),
 		MessageSigner:            nodeSignerAlice,
 		OurPubKey:                aliceKeyPub,
-		IsChannelActive:          htlcSwitch.HasActiveLink,
+		IsChannelActive:          func(lnwire.ChannelID) bool { return true },
 		ApplyChannelUpdate:       func(*lnwire.ChannelUpdate) error { return nil },
 	})
 	if err != nil {
@@ -374,7 +352,7 @@ func createTestPeer(notifier chainntnfs.ChainNotifier,
 		Switch:      htlcSwitch,
 
 		ChanActiveTimeout: chanActiveTimeout,
-		InterceptSwitch:   htlcswitch.NewInterceptableSwitch(htlcSwitch),
+		InterceptSwitch:   htlcswitch.NewInterceptableSwitch(nil),
 
 		ChannelDB:      dbAlice,
 		FeeEstimator:   estimator,
@@ -395,6 +373,37 @@ func createTestPeer(notifier chainntnfs.ChainNotifier,
 	return alicePeer, channelBob, cleanUpFunc, nil
 }
 
+// mockMessageSwitch is a mock implementation of the messageSwitch interface
+// used for testing without relying on a *htlcswitch.Switch in unit tests.
+type mockMessageSwitch struct{}
+
+// BestHeight currently returns a dummy value.
+func (m *mockMessageSwitch) BestHeight() uint32 {
+	return 0
+}
+
+// CircuitModifier currently returns a dummy value.
+func (m *mockMessageSwitch) CircuitModifier() htlcswitch.CircuitModifier {
+	return nil
+}
+
+// RemoveLink currently does nothing.
+func (m *mockMessageSwitch) RemoveLink(cid lnwire.ChannelID) {}
+
+// CreateAndAddLink currently returns a dummy value.
+func (m *mockMessageSwitch) CreateAndAddLink(cfg htlcswitch.ChannelLinkConfig,
+	lnChan *lnwallet.LightningChannel) error {
+
+	return nil
+}
+
+// GetLinksByInterface currently returns dummy values.
+func (m *mockMessageSwitch) GetLinksByInterface(pub [33]byte) (
+	[]htlcswitch.ChannelUpdateHandler, error) {
+
+	return nil, nil
+}
+
 type mockMessageConn struct {
 	t *testing.T