From dcf76a59b256ebbc13346b9922c2dce2a955a0ab Mon Sep 17 00:00:00 2001
From: Conner Fromknecht <conner@mit.edu>
Date: Thu, 7 Jun 2018 20:05:32 -0700
Subject: [PATCH 1/8] lnpeer: birth of peer package!

This commit moves the generic interface for
Lightning Network peers so that it can be
imported directly into the switch and gossiper.
Eventually, the majority of the peer logic
would be moved into this package for testing
and modularization.
---
 lnpeer/peer.go | 26 ++++++++++++++++++++++++++
 1 file changed, 26 insertions(+)
 create mode 100644 lnpeer/peer.go

diff --git a/lnpeer/peer.go b/lnpeer/peer.go
new file mode 100644
index 000000000..2a9dea0d8
--- /dev/null
+++ b/lnpeer/peer.go
@@ -0,0 +1,26 @@
+package lnpeer
+
+import (
+	"github.com/lightningnetwork/lnd/lnwire"
+	"github.com/roasbeef/btcd/btcec"
+	"github.com/roasbeef/btcd/wire"
+)
+
+// Peer is an interface which represents the remote lightning node inside our
+// system.
+type Peer interface {
+	// SendMessage sends a variadic number of message to remote peer. The
+	// first argument denotes if the method should block until the message
+	// has been sent to the remote peer.
+	SendMessage(sync bool, msg ...lnwire.Message) error
+
+	// WipeChannel removes the channel uniquely identified by its channel
+	// point from all indexes associated with the peer.
+	WipeChannel(*wire.OutPoint) error
+
+	// PubKey returns the serialized public key of the remote peer.
+	PubKey() [33]byte
+
+	// IdentityKey returns the public key of the remote peer.
+	IdentityKey() *btcec.PublicKey
+}

From 769f0f0a946852439c17237c6a8e8ce991a299cf Mon Sep 17 00:00:00 2001
From: Conner Fromknecht <conner@mit.edu>
Date: Fri, 8 Jun 2018 01:34:48 -0700
Subject: [PATCH 2/8] discovery/gossiper_test: mock out new lnpeer.Peer

---
 discovery/gossiper_test.go | 167 +++++++++++++++++++++++++++++--------
 1 file changed, 132 insertions(+), 35 deletions(-)

diff --git a/discovery/gossiper_test.go b/discovery/gossiper_test.go
index 004e6a97f..2360012fc 100644
--- a/discovery/gossiper_test.go
+++ b/discovery/gossiper_test.go
@@ -22,6 +22,7 @@ import (
 	"github.com/go-errors/errors"
 	"github.com/lightningnetwork/lnd/chainntnfs"
 	"github.com/lightningnetwork/lnd/channeldb"
+	"github.com/lightningnetwork/lnd/lnpeer"
 	"github.com/lightningnetwork/lnd/lnwire"
 	"github.com/lightningnetwork/lnd/routing"
 	"github.com/roasbeef/btcd/btcec"
@@ -535,6 +536,9 @@ func createTestCtx(startHeight uint32) (*testCtx, func(), error) {
 		SendToPeer: func(target *btcec.PublicKey, msg ...lnwire.Message) error {
 			return nil
 		},
+		FindPeer: func(target *btcec.PublicKey) (lnpeer.Peer, error) {
+			return &mockPeer{target, nil, nil}, nil
+		},
 		Router:           router,
 		TrickleDelay:     trickleDelay,
 		RetransmitDelay:  retransmitDelay,
@@ -583,7 +587,7 @@ func TestProcessAnnouncement(t *testing.T) {
 		}
 	}
 
-	// Create node valid, signed announcement, process it with 
+	// Create node valid, signed announcement, process it with
 	// gossiper service, check that valid announcement have been
 	// propagated farther into the lightning network, and check that we
 	// added new node into router.
@@ -592,10 +596,10 @@ func TestProcessAnnouncement(t *testing.T) {
 		t.Fatalf("can't create node announcement: %v", err)
 	}
 
-	nodePub := nodeKeyPriv1.PubKey()
+	nodePeer := &mockPeer{nodeKeyPriv1.PubKey(), nil, nil}
 
 	select {
-	case err = <-ctx.gossiper.ProcessRemoteAnnouncement(na, nodePub):
+	case err = <-ctx.gossiper.ProcessRemoteAnnouncement(na, nodePeer):
 	case <-time.After(2 * time.Second):
 		t.Fatal("remote announcement not processed")
 	}
@@ -605,7 +609,7 @@ func TestProcessAnnouncement(t *testing.T) {
 
 	select {
 	case msg := <-ctx.broadcastedMessage:
-		assertSenderExistence(nodePub, msg)
+		assertSenderExistence(nodePeer.IdentityKey(), msg)
 	case <-time.After(2 * trickleDelay):
 		t.Fatal("announcement wasn't proceeded")
 	}
@@ -623,7 +627,7 @@ func TestProcessAnnouncement(t *testing.T) {
 	}
 
 	select {
-	case err = <-ctx.gossiper.ProcessRemoteAnnouncement(ca, nodePub):
+	case err = <-ctx.gossiper.ProcessRemoteAnnouncement(ca, nodePeer):
 	case <-time.After(2 * time.Second):
 		t.Fatal("remote announcement not processed")
 	}
@@ -633,7 +637,7 @@ func TestProcessAnnouncement(t *testing.T) {
 
 	select {
 	case msg := <-ctx.broadcastedMessage:
-		assertSenderExistence(nodePub, msg)
+		assertSenderExistence(nodePeer.IdentityKey(), msg)
 	case <-time.After(2 * trickleDelay):
 		t.Fatal("announcement wasn't proceeded")
 	}
@@ -651,7 +655,7 @@ func TestProcessAnnouncement(t *testing.T) {
 	}
 
 	select {
-	case err = <-ctx.gossiper.ProcessRemoteAnnouncement(ua, nodePub):
+	case err = <-ctx.gossiper.ProcessRemoteAnnouncement(ua, nodePeer):
 	case <-time.After(2 * time.Second):
 		t.Fatal("remote announcement not processed")
 	}
@@ -661,7 +665,7 @@ func TestProcessAnnouncement(t *testing.T) {
 
 	select {
 	case msg := <-ctx.broadcastedMessage:
-		assertSenderExistence(nodePub, msg)
+		assertSenderExistence(nodePeer.IdentityKey(), msg)
 	case <-time.After(2 * trickleDelay):
 		t.Fatal("announcement wasn't proceeded")
 	}
@@ -690,7 +694,7 @@ func TestPrematureAnnouncement(t *testing.T) {
 		t.Fatalf("can't create node announcement: %v", err)
 	}
 
-	nodePub := nodeKeyPriv1.PubKey()
+	nodePeer := &mockPeer{nodeKeyPriv1.PubKey(), nil, nil}
 
 	// Pretending that we receive the valid channel announcement from
 	// remote side, but block height of this announcement is greater than
@@ -702,7 +706,7 @@ func TestPrematureAnnouncement(t *testing.T) {
 	}
 
 	select {
-	case <-ctx.gossiper.ProcessRemoteAnnouncement(ca, nodePub):
+	case <-ctx.gossiper.ProcessRemoteAnnouncement(ca, nodePeer):
 		t.Fatal("announcement was proceeded")
 	case <-time.After(100 * time.Millisecond):
 	}
@@ -721,7 +725,7 @@ func TestPrematureAnnouncement(t *testing.T) {
 	}
 
 	select {
-	case <-ctx.gossiper.ProcessRemoteAnnouncement(ua, nodePub):
+	case <-ctx.gossiper.ProcessRemoteAnnouncement(ua, nodePeer):
 		t.Fatal("announcement was proceeded")
 	case <-time.After(100 * time.Millisecond):
 	}
@@ -770,6 +774,9 @@ func TestSignatureAnnouncementLocalFirst(t *testing.T) {
 	// Set up a channel that we can use to inspect the messages
 	// sent directly fromn the gossiper.
 	sentMsgs := make(chan lnwire.Message, 10)
+	ctx.gossiper.cfg.FindPeer = func(target *btcec.PublicKey) (lnpeer.Peer, error) {
+		return &mockPeer{target, sentMsgs, ctx.gossiper.quit}, nil
+	}
 	ctx.gossiper.cfg.SendToPeer = func(target *btcec.PublicKey, msg ...lnwire.Message) error {
 		select {
 		case sentMsgs <- msg[0]:
@@ -792,6 +799,7 @@ func TestSignatureAnnouncementLocalFirst(t *testing.T) {
 	if err != nil {
 		t.Fatalf("unable to parse pubkey: %v", err)
 	}
+	remotePeer := &mockPeer{remoteKey, sentMsgs, ctx.gossiper.quit}
 
 	// Recreate lightning network topology. Initialize router with channel
 	// between two nodes.
@@ -840,7 +848,7 @@ func TestSignatureAnnouncementLocalFirst(t *testing.T) {
 
 	select {
 	case err = <-ctx.gossiper.ProcessRemoteAnnouncement(batch.chanUpdAnn2,
-		remoteKey):
+		remotePeer):
 	case <-time.After(2 * time.Second):
 		t.Fatal("did not process remote announcement")
 	}
@@ -887,7 +895,7 @@ func TestSignatureAnnouncementLocalFirst(t *testing.T) {
 
 	select {
 	case err = <-ctx.gossiper.ProcessRemoteAnnouncement(batch.remoteProofAnn,
-		remoteKey):
+		remotePeer):
 	case <-time.After(2 * time.Second):
 		t.Fatal("did not process remote announcement")
 	}
@@ -932,6 +940,9 @@ func TestOrphanSignatureAnnouncement(t *testing.T) {
 	// Set up a channel that we can use to inspect the messages
 	// sent directly from the gossiper.
 	sentMsgs := make(chan lnwire.Message, 10)
+	ctx.gossiper.cfg.FindPeer = func(target *btcec.PublicKey) (lnpeer.Peer, error) {
+		return &mockPeer{target, sentMsgs, ctx.gossiper.quit}, nil
+	}
 	ctx.gossiper.cfg.SendToPeer = func(target *btcec.PublicKey, msg ...lnwire.Message) error {
 		select {
 		case sentMsgs <- msg[0]:
@@ -954,6 +965,7 @@ func TestOrphanSignatureAnnouncement(t *testing.T) {
 	if err != nil {
 		t.Fatalf("unable to parse pubkey: %v", err)
 	}
+	remotePeer := &mockPeer{remoteKey, sentMsgs, ctx.gossiper.quit}
 
 	// Pretending that we receive local channel announcement from funding
 	// manager, thereby kick off the announcement exchange process, in
@@ -961,7 +973,7 @@ func TestOrphanSignatureAnnouncement(t *testing.T) {
 	// because we haven't announce the channel yet.
 	select {
 	case err = <-ctx.gossiper.ProcessRemoteAnnouncement(batch.remoteProofAnn,
-		remoteKey):
+		remotePeer):
 	case <-time.After(2 * time.Second):
 		t.Fatal("did not process remote announcement")
 	}
@@ -1032,7 +1044,7 @@ func TestOrphanSignatureAnnouncement(t *testing.T) {
 
 	select {
 	case err = <-ctx.gossiper.ProcessRemoteAnnouncement(batch.chanUpdAnn2,
-		remoteKey):
+		remotePeer):
 	case <-time.After(2 * time.Second):
 		t.Fatal("did not process remote announcement")
 	}
@@ -1116,6 +1128,7 @@ func TestSignatureAnnouncementRetry(t *testing.T) {
 	if err != nil {
 		t.Fatalf("unable to parse pubkey: %v", err)
 	}
+	remotePeer := &mockPeer{remoteKey, nil, nil}
 
 	// Recreate lightning network topology. Initialize router with channel
 	// between two nodes.
@@ -1151,7 +1164,7 @@ func TestSignatureAnnouncementRetry(t *testing.T) {
 
 	select {
 	case err = <-ctx.gossiper.ProcessRemoteAnnouncement(batch.chanUpdAnn2,
-		remoteKey):
+		remotePeer):
 	case <-time.After(2 * time.Second):
 		t.Fatal("did not process remote announcement")
 	}
@@ -1248,7 +1261,7 @@ func TestSignatureAnnouncementRetry(t *testing.T) {
 	// broadcast of 3 messages (ChannelAnnouncement + 2 ChannelUpdate).
 	select {
 	case err = <-ctx.gossiper.ProcessRemoteAnnouncement(batch.remoteProofAnn,
-		remoteKey):
+		remotePeer):
 	case <-time.After(2 * time.Second):
 		t.Fatal("did not process local announcement")
 	}
@@ -1304,6 +1317,7 @@ func TestSignatureAnnouncementRetryAtStartup(t *testing.T) {
 	if err != nil {
 		t.Fatalf("unable to parse pubkey: %v", err)
 	}
+	remotePeer := &mockPeer{remoteKey, nil, nil}
 
 	// Recreate lightning network topology. Initialize router with channel
 	// between two nodes.
@@ -1339,7 +1353,7 @@ func TestSignatureAnnouncementRetryAtStartup(t *testing.T) {
 
 	select {
 	case err = <-ctx.gossiper.ProcessRemoteAnnouncement(batch.chanUpdAnn2,
-		remoteKey):
+		remotePeer):
 	case <-time.After(2 * time.Second):
 		t.Fatal("did not process remote announcement")
 	}
@@ -1472,7 +1486,7 @@ func TestSignatureAnnouncementRetryAtStartup(t *testing.T) {
 	// broadcast should continue as normal.
 	select {
 	case err = <-ctx.gossiper.ProcessRemoteAnnouncement(batch.remoteProofAnn,
-		remoteKey):
+		remotePeer):
 	case <-time.After(2 * time.Second):
 		t.Fatal("did not process remote announcement")
 	}
@@ -1529,6 +1543,7 @@ func TestSignatureAnnouncementFullProofWhenRemoteProof(t *testing.T) {
 	if err != nil {
 		t.Fatalf("unable to parse pubkey: %v", err)
 	}
+	remotePeer := &mockPeer{remoteKey, nil, nil}
 
 	// Recreate lightning network topology. Initialize router with channel
 	// between two nodes.
@@ -1564,7 +1579,7 @@ func TestSignatureAnnouncementFullProofWhenRemoteProof(t *testing.T) {
 
 	select {
 	case err = <-ctx.gossiper.ProcessRemoteAnnouncement(batch.chanUpdAnn2,
-		remoteKey):
+		remotePeer):
 	case <-time.After(2 * time.Second):
 		t.Fatal("did not process remote announcement")
 	}
@@ -1579,6 +1594,8 @@ func TestSignatureAnnouncementFullProofWhenRemoteProof(t *testing.T) {
 	// Set up a channel we can use to inspect messages sent by the
 	// gossiper to the remote peer.
 	sentToPeer := make(chan lnwire.Message, 1)
+	remotePeer.sentMsgs = sentToPeer
+	remotePeer.quit = ctx.gossiper.quit
 	ctx.gossiper.cfg.SendToPeer = func(target *btcec.PublicKey,
 		msg ...lnwire.Message) error {
 		select {
@@ -1609,7 +1626,7 @@ func TestSignatureAnnouncementFullProofWhenRemoteProof(t *testing.T) {
 
 	select {
 	case err = <-ctx.gossiper.ProcessRemoteAnnouncement(batch.remoteProofAnn,
-		remoteKey):
+		remotePeer):
 	case <-time.After(2 * time.Second):
 		t.Fatal("did not process local announcement")
 	}
@@ -1654,7 +1671,7 @@ func TestSignatureAnnouncementFullProofWhenRemoteProof(t *testing.T) {
 	// trigger a send of the full ChannelAnnouncement.
 	select {
 	case err = <-ctx.gossiper.ProcessRemoteAnnouncement(batch.remoteProofAnn,
-		remoteKey):
+		remotePeer):
 	case <-time.After(2 * time.Second):
 		t.Fatal("did not process local announcement")
 	}
@@ -1702,7 +1719,13 @@ func TestDeDuplicatedAnnouncements(t *testing.T) {
 	if err != nil {
 		t.Fatalf("can't create remote channel announcement: %v", err)
 	}
-	announcements.AddMsgs(networkMsg{msg: ca, peer: bitcoinKeyPub2})
+
+	nodePeer := &mockPeer{bitcoinKeyPub2, nil, nil}
+	announcements.AddMsgs(networkMsg{
+		msg:    ca,
+		peer:   nodePeer,
+		source: nodePeer.IdentityKey(),
+	})
 	if len(announcements.channelAnnouncements) != 1 {
 		t.Fatal("new channel announcement not stored in batch")
 	}
@@ -1715,7 +1738,11 @@ func TestDeDuplicatedAnnouncements(t *testing.T) {
 	if err != nil {
 		t.Fatalf("can't create remote channel announcement: %v", err)
 	}
-	announcements.AddMsgs(networkMsg{msg: ca2, peer: bitcoinKeyPub2})
+	announcements.AddMsgs(networkMsg{
+		msg:    ca2,
+		peer:   nodePeer,
+		source: nodePeer.IdentityKey(),
+	})
 	if len(announcements.channelAnnouncements) != 1 {
 		t.Fatal("channel announcement not replaced in batch")
 	}
@@ -1727,7 +1754,11 @@ func TestDeDuplicatedAnnouncements(t *testing.T) {
 	if err != nil {
 		t.Fatalf("can't create update announcement: %v", err)
 	}
-	announcements.AddMsgs(networkMsg{msg: ua, peer: bitcoinKeyPub2})
+	announcements.AddMsgs(networkMsg{
+		msg:    ua,
+		peer:   nodePeer,
+		source: nodePeer.IdentityKey(),
+	})
 	if len(announcements.channelUpdates) != 1 {
 		t.Fatal("new channel update not stored in batch")
 	}
@@ -1738,7 +1769,11 @@ func TestDeDuplicatedAnnouncements(t *testing.T) {
 	if err != nil {
 		t.Fatalf("can't create update announcement: %v", err)
 	}
-	announcements.AddMsgs(networkMsg{msg: ua2, peer: bitcoinKeyPub2})
+	announcements.AddMsgs(networkMsg{
+		msg:    ua2,
+		peer:   nodePeer,
+		source: nodePeer.IdentityKey(),
+	})
 	if len(announcements.channelUpdates) != 1 {
 		t.Fatal("channel update not replaced in batch")
 	}
@@ -1749,7 +1784,11 @@ func TestDeDuplicatedAnnouncements(t *testing.T) {
 	if err != nil {
 		t.Fatalf("can't create update announcement: %v", err)
 	}
-	announcements.AddMsgs(networkMsg{msg: ua3, peer: bitcoinKeyPub2})
+	announcements.AddMsgs(networkMsg{
+		msg:    ua3,
+		peer:   nodePeer,
+		source: nodePeer.IdentityKey(),
+	})
 	if len(announcements.channelUpdates) != 1 {
 		t.Fatal("channel update not replaced in batch")
 	}
@@ -1779,7 +1818,11 @@ func TestDeDuplicatedAnnouncements(t *testing.T) {
 	if err != nil {
 		t.Fatalf("can't create update announcement: %v", err)
 	}
-	announcements.AddMsgs(networkMsg{msg: ua4, peer: bitcoinKeyPub2})
+	announcements.AddMsgs(networkMsg{
+		msg:    ua4,
+		peer:   nodePeer,
+		source: nodePeer.IdentityKey(),
+	})
 	if len(announcements.channelUpdates) != 1 {
 		t.Fatal("channel update not in batch")
 	}
@@ -1791,7 +1834,11 @@ func TestDeDuplicatedAnnouncements(t *testing.T) {
 	if err != nil {
 		t.Fatalf("can't create node announcement: %v", err)
 	}
-	announcements.AddMsgs(networkMsg{msg: na, peer: bitcoinKeyPub2})
+	announcements.AddMsgs(networkMsg{
+		msg:    na,
+		peer:   nodePeer,
+		source: nodePeer.IdentityKey(),
+	})
 	if len(announcements.nodeAnnouncements) != 1 {
 		t.Fatal("new node announcement not stored in batch")
 	}
@@ -1801,7 +1848,11 @@ func TestDeDuplicatedAnnouncements(t *testing.T) {
 	if err != nil {
 		t.Fatalf("can't create node announcement: %v", err)
 	}
-	announcements.AddMsgs(networkMsg{msg: na2, peer: bitcoinKeyPub2})
+	announcements.AddMsgs(networkMsg{
+		msg:    na2,
+		peer:   nodePeer,
+		source: nodePeer.IdentityKey(),
+	})
 	if len(announcements.nodeAnnouncements) != 2 {
 		t.Fatal("second node announcement not stored in batch")
 	}
@@ -1812,7 +1863,11 @@ func TestDeDuplicatedAnnouncements(t *testing.T) {
 	if err != nil {
 		t.Fatalf("can't create node announcement: %v", err)
 	}
-	announcements.AddMsgs(networkMsg{msg: na3, peer: bitcoinKeyPub2})
+	announcements.AddMsgs(networkMsg{
+		msg:    na3,
+		peer:   nodePeer,
+		source: nodePeer.IdentityKey(),
+	})
 	if len(announcements.nodeAnnouncements) != 2 {
 		t.Fatal("second node announcement not replaced in batch")
 	}
@@ -1824,7 +1879,11 @@ func TestDeDuplicatedAnnouncements(t *testing.T) {
 	if err != nil {
 		t.Fatalf("can't create node announcement: %v", err)
 	}
-	announcements.AddMsgs(networkMsg{msg: na4, peer: bitcoinKeyPub2})
+	announcements.AddMsgs(networkMsg{
+		msg:    na4,
+		peer:   nodePeer,
+		source: nodePeer.IdentityKey(),
+	})
 	if len(announcements.nodeAnnouncements) != 2 {
 		t.Fatal("second node announcement not replaced again in batch")
 	}
@@ -1835,7 +1894,11 @@ func TestDeDuplicatedAnnouncements(t *testing.T) {
 	if err != nil {
 		t.Fatalf("can't create node announcement: %v", err)
 	}
-	announcements.AddMsgs(networkMsg{msg: na5, peer: bitcoinKeyPub2})
+	announcements.AddMsgs(networkMsg{
+		msg:    na5,
+		peer:   nodePeer,
+		source: nodePeer.IdentityKey(),
+	})
 	if len(announcements.nodeAnnouncements) != 2 {
 		t.Fatal("node announcement not replaced in batch")
 	}
@@ -1910,6 +1973,9 @@ func TestReceiveRemoteChannelUpdateFirst(t *testing.T) {
 	// Set up a channel that we can use to inspect the messages
 	// sent directly fromn the gossiper.
 	sentMsgs := make(chan lnwire.Message, 10)
+	ctx.gossiper.cfg.FindPeer = func(target *btcec.PublicKey) (lnpeer.Peer, error) {
+		return &mockPeer{target, sentMsgs, ctx.gossiper.quit}, nil
+	}
 	ctx.gossiper.cfg.SendToPeer = func(target *btcec.PublicKey, msg ...lnwire.Message) error {
 		select {
 		case sentMsgs <- msg[0]:
@@ -1932,11 +1998,12 @@ func TestReceiveRemoteChannelUpdateFirst(t *testing.T) {
 	if err != nil {
 		t.Fatalf("unable to parse pubkey: %v", err)
 	}
+	remotePeer := &mockPeer{remoteKey, nil, nil}
 
 	// Recreate the case where the remote node is sending us its ChannelUpdate
 	// before we have been able to process our own ChannelAnnouncement and
 	// ChannelUpdate.
-	err = <-ctx.gossiper.ProcessRemoteAnnouncement(batch.chanUpdAnn2, remoteKey)
+	err = <-ctx.gossiper.ProcessRemoteAnnouncement(batch.chanUpdAnn2, remotePeer)
 	if err != nil {
 		t.Fatalf("unable to process :%v", err)
 	}
@@ -2042,7 +2109,7 @@ func TestReceiveRemoteChannelUpdateFirst(t *testing.T) {
 		t.Fatal("wrong number of objects in storage")
 	}
 
-	err = <-ctx.gossiper.ProcessRemoteAnnouncement(batch.remoteProofAnn, remoteKey)
+	err = <-ctx.gossiper.ProcessRemoteAnnouncement(batch.remoteProofAnn, remotePeer)
 	if err != nil {
 		t.Fatalf("unable to process :%v", err)
 	}
@@ -2069,3 +2136,33 @@ func TestReceiveRemoteChannelUpdateFirst(t *testing.T) {
 		t.Fatal("waiting proof should be removed from storage")
 	}
 }
+
+// mockPeer implements the lnpeer.Peer interface and is used to test the
+// gossiper's interaction with peers.
+type mockPeer struct {
+	pk       *btcec.PublicKey
+	sentMsgs chan lnwire.Message
+	quit     chan struct{}
+}
+
+func (p *mockPeer) SendMessage(_ bool, msgs ...lnwire.Message) error {
+	if p.sentMsgs == nil && p.quit == nil {
+		return nil
+	}
+
+	for _, msg := range msgs {
+		select {
+		case p.sentMsgs <- msg:
+		case <-p.quit:
+		}
+	}
+
+	return nil
+}
+func (p *mockPeer) WipeChannel(_ *wire.OutPoint) error { return nil }
+func (p *mockPeer) IdentityKey() *btcec.PublicKey      { return p.pk }
+func (p *mockPeer) PubKey() [33]byte {
+	var pubkey [33]byte
+	copy(pubkey[:], p.pk.SerializeCompressed())
+	return pubkey
+}

From edf08458c1d6f346c5cb99d1d3990fd05910c744 Mon Sep 17 00:00:00 2001
From: Conner Fromknecht <conner@mit.edu>
Date: Thu, 7 Jun 2018 20:07:30 -0700
Subject: [PATCH 3/8] peer: changes to satisfy lnpeer.Peer

This commit adds:
 - variadic SendMessage method (w/ sync bool reorder)
 - passes peer object directly to ProcessRemoteAnnouncment
 - adds IdentityKey() method for lnpeer.Peer
---
 peer.go | 57 ++++++++++++++++++++++++++++++++++++++-------------------
 1 file changed, 38 insertions(+), 19 deletions(-)

diff --git a/peer.go b/peer.go
index e4c860736..ba9642eaa 100644
--- a/peer.go
+++ b/peer.go
@@ -21,6 +21,7 @@ import (
 	"github.com/lightningnetwork/lnd/lnrpc"
 	"github.com/lightningnetwork/lnd/lnwallet"
 	"github.com/lightningnetwork/lnd/lnwire"
+	"github.com/roasbeef/btcd/btcec"
 	"github.com/roasbeef/btcd/chaincfg/chainhash"
 	"github.com/roasbeef/btcd/connmgr"
 	"github.com/roasbeef/btcd/txscript"
@@ -92,7 +93,6 @@ type peer struct {
 	bytesReceived uint64
 	bytesSent     uint64
 
-
 	// pingTime is a rough estimate of the RTT (round-trip-time) between us
 	// and the connected peer. This time is expressed in micro seconds.
 	// To be used atomically.
@@ -502,10 +502,10 @@ func (p *peer) addLink(chanPoint *wire.OutPoint,
 				if linkErr.SendData != nil {
 					data = linkErr.SendData
 				}
-				err := p.SendMessage(&lnwire.Error{
+				err := p.SendMessage(true, &lnwire.Error{
 					ChanID: chanID,
 					Data:   data,
-				}, true)
+				})
 				if err != nil {
 					peerLog.Errorf("unable to send msg to "+
 						"remote peer: %v", err)
@@ -833,8 +833,7 @@ func newDiscMsgStream(p *peer) *msgStream {
 		"Update stream for gossiper exited",
 		1000,
 		func(msg lnwire.Message) {
-			p.server.authGossiper.ProcessRemoteAnnouncement(msg,
-				p.addr.IdentityKey)
+			p.server.authGossiper.ProcessRemoteAnnouncement(msg, p)
 		},
 	)
 }
@@ -1958,23 +1957,38 @@ func (p *peer) sendInitMsg() error {
 	return p.writeMessage(msg)
 }
 
-// SendMessage sends message to remote peer. The second argument denotes if the
-// method should block until the message has been sent to the remote peer.
-func (p *peer) SendMessage(msg lnwire.Message, sync bool) error {
-	if !sync {
-		p.queueMsg(msg, nil)
-		return nil
+// SendMessage sends a variadic number of message to remote peer. The first
+// argument denotes if the method should block until the message has been sent
+// to the remote peer.
+func (p *peer) SendMessage(sync bool, msgs ...lnwire.Message) error {
+	// Add all incoming messages to the outgoing queue. A list of error
+	// chans is populated for each message if the caller requested a sync
+	// send.
+	var errChans []chan error
+	for _, msg := range msgs {
+		// If a sync send was requested, create an error chan to listen
+		// for an ack from the writeHandler.
+		var errChan chan error
+		if sync {
+			errChan = make(chan error, 1)
+			errChans = append(errChans, errChan)
+		}
+
+		p.queueMsg(msg, errChan)
 	}
 
-	errChan := make(chan error, 1)
-	p.queueMsg(msg, errChan)
-
-	select {
-	case err := <-errChan:
-		return err
-	case <-p.quit:
-		return fmt.Errorf("peer shutting down")
+	// Wait for all replies from the writeHandler. For async sends, this
+	// will be a NOP as the list of error chans is nil.
+	for _, errChan := range errChans {
+		select {
+		case err := <-errChan:
+			return err
+		case <-p.quit:
+			return ErrPeerExiting
+		}
 	}
+
+	return nil
 }
 
 // PubKey returns the pubkey of the peer in compressed serialized format.
@@ -1982,6 +1996,11 @@ func (p *peer) PubKey() [33]byte {
 	return p.pubKeyBytes
 }
 
+// IdentityKey returns the public key of the remote peer.
+func (p *peer) IdentityKey() *btcec.PublicKey {
+	return p.addr.IdentityKey
+}
+
 // TODO(roasbeef): make all start/stop mutexes a CAS
 
 // fetchLastChanUpdate returns a function which is able to retrieve the last

From a670537f4c03f7c03f049c31e3d72e7fdb9b8062 Mon Sep 17 00:00:00 2001
From: Conner Fromknecht <conner@mit.edu>
Date: Thu, 7 Jun 2018 20:09:40 -0700
Subject: [PATCH 4/8] server: expose FindPeer for lnpeer.Peer to gossiper

---
 server.go | 20 ++++++++++++--------
 1 file changed, 12 insertions(+), 8 deletions(-)

diff --git a/server.go b/server.go
index 6722c1756..def6c6a76 100644
--- a/server.go
+++ b/server.go
@@ -24,6 +24,7 @@ import (
 	"github.com/lightningnetwork/lnd/contractcourt"
 	"github.com/lightningnetwork/lnd/discovery"
 	"github.com/lightningnetwork/lnd/htlcswitch"
+	"github.com/lightningnetwork/lnd/lnpeer"
 	"github.com/lightningnetwork/lnd/lnrpc"
 	"github.com/lightningnetwork/lnd/lnwallet"
 	"github.com/lightningnetwork/lnd/lnwire"
@@ -435,12 +436,15 @@ func newServer(listenAddrs []string, chanDB *channeldb.DB, cc *chainControl,
 	}
 
 	s.authGossiper, err = discovery.New(discovery.Config{
-		Router:           s.chanRouter,
-		Notifier:         s.cc.chainNotifier,
-		ChainHash:        *activeNetParams.GenesisHash,
-		Broadcast:        s.BroadcastMessage,
-		ChanSeries:       &chanSeries{s.chanDB.ChannelGraph()},
-		SendToPeer:       s.SendToPeer,
+		Router:     s.chanRouter,
+		Notifier:   s.cc.chainNotifier,
+		ChainHash:  *activeNetParams.GenesisHash,
+		Broadcast:  s.BroadcastMessage,
+		ChanSeries: &chanSeries{s.chanDB.ChannelGraph()},
+		SendToPeer: s.SendToPeer,
+		FindPeer: func(pub *btcec.PublicKey) (lnpeer.Peer, error) {
+			return s.FindPeer(pub)
+		},
 		NotifyWhenOnline: s.NotifyWhenOnline,
 		ProofMatureDelta: 0,
 		TrickleDelay:     time.Millisecond * time.Duration(cfg.TrickleDelay),
@@ -1882,7 +1886,7 @@ func (s *server) addPeer(p *peer) {
 		// TODO(roasbeef): craft s.t. we only get updates from a few
 		// peers
 		recvUpdates := !cfg.NoChanUpdates
-		go s.authGossiper.InitSyncState(p.addr.IdentityKey, recvUpdates)
+		go s.authGossiper.InitSyncState(p, recvUpdates)
 
 	// If the remote peer has the initial sync feature bit set, then we'll
 	// being the synchronization protocol to exchange authenticated channel
@@ -1892,7 +1896,7 @@ func (s *server) addPeer(p *peer) {
 		srvrLog.Infof("Requesting full table sync with %x",
 			p.pubKeyBytes[:])
 
-		go s.authGossiper.SynchronizeNode(p.addr.IdentityKey)
+		go s.authGossiper.SynchronizeNode(p)
 	}
 
 	// Check if there are listeners waiting for this peer to come online.

From bf515a234138dd4614852b4ce424586ff23f7fb2 Mon Sep 17 00:00:00 2001
From: Conner Fromknecht <conner@mit.edu>
Date: Thu, 7 Jun 2018 20:11:27 -0700
Subject: [PATCH 5/8] discovery/gossiper: mostly deprecate SendToPeer

This commit changes the gossiper to direct messages to
peer objects, instead of sending them through the
server every time. The primary motivation is to reduce
contention on the server's mutex and, more importantly,
avoid deadlocks in the Triangle of Death.
---
 discovery/gossiper.go | 181 +++++++++++++++++++++++++-----------------
 1 file changed, 109 insertions(+), 72 deletions(-)

diff --git a/discovery/gossiper.go b/discovery/gossiper.go
index 64302de94..56bd79294 100644
--- a/discovery/gossiper.go
+++ b/discovery/gossiper.go
@@ -14,6 +14,7 @@ import (
 	"github.com/go-errors/errors"
 	"github.com/lightningnetwork/lnd/chainntnfs"
 	"github.com/lightningnetwork/lnd/channeldb"
+	"github.com/lightningnetwork/lnd/lnpeer"
 	"github.com/lightningnetwork/lnd/lnwallet"
 	"github.com/lightningnetwork/lnd/lnwire"
 	"github.com/lightningnetwork/lnd/multimutex"
@@ -35,8 +36,9 @@ var (
 // networkMsg couples a routing related wire message with the peer that
 // originally sent it.
 type networkMsg struct {
-	peer *btcec.PublicKey
-	msg  lnwire.Message
+	peer   lnpeer.Peer
+	source *btcec.PublicKey
+	msg    lnwire.Message
 
 	isRemote bool
 
@@ -97,6 +99,11 @@ type Config struct {
 	// messages to a particular peer identified by the target public key.
 	SendToPeer func(target *btcec.PublicKey, msg ...lnwire.Message) error
 
+	// FindPeer returns the actively registered peer for a given remote
+	// public key. An error is returned if the peer was not found or a
+	// shutdown has been requested.
+	FindPeer func(identityKey *btcec.PublicKey) (lnpeer.Peer, error)
+
 	// NotifyWhenOnline is a function that allows the gossiper to be
 	// notified when a certain peer comes online, allowing it to
 	// retry sending a peer message.
@@ -147,8 +154,8 @@ type AuthenticatedGossiper struct {
 	// as we know it. To be used atomically.
 	bestHeight uint32
 
-	quit    chan struct{}
-	wg      sync.WaitGroup
+	quit chan struct{}
+	wg   sync.WaitGroup
 
 	// cfg is a copy of the configuration struct that the gossiper service
 	// was initialized with.
@@ -244,7 +251,7 @@ func New(cfg Config, selfKey *btcec.PublicKey) (*AuthenticatedGossiper, error) {
 // the entire network graph is read from disk, then serialized to the format
 // defined within the current wire protocol. This cache of graph data is then
 // sent directly to the target node.
-func (d *AuthenticatedGossiper) SynchronizeNode(pub *btcec.PublicKey) error {
+func (d *AuthenticatedGossiper) SynchronizeNode(syncPeer lnpeer.Peer) error {
 	// TODO(roasbeef): need to also store sig data in db
 	//  * will be nice when we switch to pairing sigs would only need one ^_^
 
@@ -355,12 +362,12 @@ func (d *AuthenticatedGossiper) SynchronizeNode(pub *btcec.PublicKey) error {
 	}
 
 	log.Infof("Syncing channel graph state with %x, sending %v "+
-		"vertexes and %v edges", pub.SerializeCompressed(),
+		"vertexes and %v edges", syncPeer.PubKey(),
 		numNodes, numEdges)
 
 	// With all the announcement messages gathered, send them all in a
 	// single batch to the target peer.
-	return d.cfg.SendToPeer(pub, announceMessages...)
+	return syncPeer.SendMessage(false, announceMessages...)
 }
 
 // PropagateChanPolicyUpdate signals the AuthenticatedGossiper to update the
@@ -452,12 +459,13 @@ func (d *AuthenticatedGossiper) Stop() {
 // peers.  Remote channel announcements should contain the announcement proof
 // and be fully validated.
 func (d *AuthenticatedGossiper) ProcessRemoteAnnouncement(msg lnwire.Message,
-	src *btcec.PublicKey) chan error {
+	peer lnpeer.Peer) chan error {
 
 	nMsg := &networkMsg{
 		msg:      msg,
 		isRemote: true,
-		peer:     src,
+		peer:     peer,
+		source:   peer.IdentityKey(),
 		err:      make(chan error, 1),
 	}
 
@@ -478,12 +486,12 @@ func (d *AuthenticatedGossiper) ProcessRemoteAnnouncement(msg lnwire.Message,
 // entire channel announcement and update messages will be re-constructed and
 // broadcast to the rest of the network.
 func (d *AuthenticatedGossiper) ProcessLocalAnnouncement(msg lnwire.Message,
-	src *btcec.PublicKey) chan error {
+	source *btcec.PublicKey) chan error {
 
 	nMsg := &networkMsg{
 		msg:      msg,
 		isRemote: false,
-		peer:     src,
+		source:   source,
 		err:      make(chan error, 1),
 	}
 
@@ -587,7 +595,7 @@ func (d *deDupedAnnouncements) addMsg(message networkMsg) {
 	// Channel announcements are identified by the short channel id field.
 	case *lnwire.ChannelAnnouncement:
 		deDupKey := msg.ShortChannelID
-		sender := routing.NewVertex(message.peer)
+		sender := routing.NewVertex(message.source)
 
 		mws, ok := d.channelAnnouncements[deDupKey]
 		if !ok {
@@ -609,7 +617,7 @@ func (d *deDupedAnnouncements) addMsg(message networkMsg) {
 	// Channel updates are identified by the (short channel id, flags)
 	// tuple.
 	case *lnwire.ChannelUpdate:
-		sender := routing.NewVertex(message.peer)
+		sender := routing.NewVertex(message.source)
 		deDupKey := channelUpdateID{
 			msg.ShortChannelID,
 			msg.Flags,
@@ -658,7 +666,7 @@ func (d *deDupedAnnouncements) addMsg(message networkMsg) {
 	// Node announcements are identified by the Vertex field.  Use the
 	// NodeID to create the corresponding Vertex.
 	case *lnwire.NodeAnnouncement:
-		sender := routing.NewVertex(message.peer)
+		sender := routing.NewVertex(message.source)
 		deDupKey := routing.Vertex(msg.NodeID)
 
 		// We do the same for node announcements as we did for channel
@@ -875,7 +883,7 @@ func (d *AuthenticatedGossiper) resendAnnounceSignatures() error {
 // gossip syncer for an inbound message so we can properly dispatch the
 // incoming message. If a gossip syncer isn't found, then one will be created
 // for the target peer.
-func (d *AuthenticatedGossiper) findGossipSyncer(pub *btcec.PublicKey) *gossipSyncer {
+func (d *AuthenticatedGossiper) findGossipSyncer(pub *btcec.PublicKey) (*gossipSyncer, error) {
 	target := routing.NewVertex(pub)
 
 	// First, we'll try to find an existing gossiper for this peer.
@@ -885,16 +893,26 @@ func (d *AuthenticatedGossiper) findGossipSyncer(pub *btcec.PublicKey) *gossipSy
 
 	// If one exists, then we'll return it directly.
 	if ok {
-		return syncer
+		return syncer, nil
 	}
 
-	// Otherwise, we'll obtain the mutex, then check again if a gossiper
-	// was added after we dropped the read mutex.
+	// A known gossip syncer doesn't exist, so we may have to create one
+	// from scratch. To do so, we'll query for a reference directly to the
+	// active peer.
+	syncPeer, err := d.cfg.FindPeer(pub)
+	if err != nil {
+		log.Debugf("unable to find gossip peer %v: %v",
+			pub.SerializeCompressed(), err)
+		return nil, err
+	}
+
+	// Finally, we'll obtain the exclusive mutex, then check again if a
+	// gossiper was added after we dropped the read mutex.
 	d.syncerMtx.Lock()
 	syncer, ok = d.peerSyncers[target]
 	if ok {
 		d.syncerMtx.Unlock()
-		return syncer
+		return syncer, nil
 	}
 
 	// At this point, a syncer doesn't yet exist, so we'll create a new one
@@ -905,7 +923,7 @@ func (d *AuthenticatedGossiper) findGossipSyncer(pub *btcec.PublicKey) *gossipSy
 		channelSeries:   d.cfg.ChanSeries,
 		encodingType:    lnwire.EncodingSortedPlain,
 		sendToPeer: func(msgs ...lnwire.Message) error {
-			return d.cfg.SendToPeer(pub, msgs...)
+			return syncPeer.SendMessage(false, msgs...)
 		},
 	})
 	copy(syncer.peerPub[:], pub.SerializeCompressed())
@@ -914,7 +932,7 @@ func (d *AuthenticatedGossiper) findGossipSyncer(pub *btcec.PublicKey) *gossipSy
 
 	d.syncerMtx.Unlock()
 
-	return syncer
+	return syncer, nil
 }
 
 // networkHandler is the primary goroutine that drives this service. The roles
@@ -992,15 +1010,20 @@ func (d *AuthenticatedGossiper) networkHandler() {
 			// then we'll dispatch that directly to the proper
 			// gossipSyncer.
 			case *lnwire.GossipTimestampRange:
-				syncer := d.findGossipSyncer(announcement.peer)
+				syncer, err := d.findGossipSyncer(
+					announcement.source,
+				)
+				if err != nil {
+					continue
+				}
 
 				// If we've found the message target, then
 				// we'll dispatch the message directly to it.
-				err := syncer.ApplyGossipFilter(msg)
+				err = syncer.ApplyGossipFilter(msg)
 				if err != nil {
 					log.Warnf("unable to apply gossip "+
 						"filter for peer=%x: %v",
-						announcement.peer.SerializeCompressed(), err)
+						announcement.peer.PubKey(), err)
 				}
 				continue
 
@@ -1012,7 +1035,12 @@ func (d *AuthenticatedGossiper) networkHandler() {
 				*lnwire.ReplyChannelRange,
 				*lnwire.ReplyShortChanIDsEnd:
 
-				syncer := d.findGossipSyncer(announcement.peer)
+				syncer, err := d.findGossipSyncer(
+					announcement.source,
+				)
+				if err != nil {
+					continue
+				}
 
 				syncer.ProcessQueryMsg(announcement.msg)
 				continue
@@ -1193,19 +1221,19 @@ func (d *AuthenticatedGossiper) networkHandler() {
 // needed to handle new queries. The recvUpdates bool indicates if we should
 // continue to receive real-time updates from the remote peer once we've synced
 // channel state.
-func (d *AuthenticatedGossiper) InitSyncState(peer *btcec.PublicKey, recvUpdates bool) {
+func (d *AuthenticatedGossiper) InitSyncState(syncPeer lnpeer.Peer, recvUpdates bool) {
 	d.syncerMtx.Lock()
 	defer d.syncerMtx.Unlock()
 
 	// If we already have a syncer, then we'll exit early as we don't want
 	// to override it.
-	nodeID := routing.NewVertex(peer)
+	nodeID := routing.Vertex(syncPeer.PubKey())
 	if _, ok := d.peerSyncers[nodeID]; ok {
 		return
 	}
 
 	log.Infof("Creating new gossipSyncer for peer=%x",
-		peer.SerializeCompressed())
+		nodeID)
 
 	syncer := newGossiperSyncer(gossipSyncerCfg{
 		chainHash:       d.cfg.ChainHash,
@@ -1213,10 +1241,10 @@ func (d *AuthenticatedGossiper) InitSyncState(peer *btcec.PublicKey, recvUpdates
 		channelSeries:   d.cfg.ChanSeries,
 		encodingType:    lnwire.EncodingSortedPlain,
 		sendToPeer: func(msgs ...lnwire.Message) error {
-			return d.cfg.SendToPeer(peer, msgs...)
+			return syncPeer.SendMessage(false, msgs...)
 		},
 	})
-	copy(syncer.peerPub[:], peer.SerializeCompressed())
+	copy(syncer.peerPub[:], nodeID[:])
 	d.peerSyncers[nodeID] = syncer
 
 	syncer.Start()
@@ -1429,8 +1457,8 @@ func (d *AuthenticatedGossiper) processChanPolicyUpdate(
 		// We set ourselves as the source of this message to indicate
 		// that we shouldn't skip any peers when sending this message.
 		chanUpdates = append(chanUpdates, networkMsg{
-			peer: d.selfKey,
-			msg:  chanUpdate,
+			source: d.selfKey,
+			msg:    chanUpdate,
 		})
 	}
 
@@ -1502,19 +1530,19 @@ func (d *AuthenticatedGossiper) processRejectedEdge(chanAnnMsg *lnwire.ChannelAn
 	// our peers.
 	announcements := make([]networkMsg, 0, 3)
 	announcements = append(announcements, networkMsg{
-		msg:  chanAnn,
-		peer: d.selfKey,
+		source: d.selfKey,
+		msg:    chanAnn,
 	})
 	if e1Ann != nil {
 		announcements = append(announcements, networkMsg{
-			msg:  e1Ann,
-			peer: d.selfKey,
+			source: d.selfKey,
+			msg:    e1Ann,
 		})
 	}
 	if e2Ann != nil {
 		announcements = append(announcements, networkMsg{
-			msg:  e2Ann,
-			peer: d.selfKey,
+			source: d.selfKey,
+			msg:    e2Ann,
 		})
 
 	}
@@ -1591,8 +1619,9 @@ func (d *AuthenticatedGossiper) processNetworkAnnouncement(nMsg *networkMsg) []n
 		// Node announcement was successfully proceeded and know it
 		// might be broadcast to other connected nodes.
 		announcements = append(announcements, networkMsg{
-			msg:  msg,
-			peer: nMsg.peer,
+			peer:   nMsg.peer,
+			source: nMsg.source,
+			msg:    msg,
 		})
 
 		nMsg.err <- nil
@@ -1776,7 +1805,7 @@ func (d *AuthenticatedGossiper) processNetworkAnnouncement(nMsg *networkMsg) []n
 							msg, nMsg.peer)
 					} else {
 						err = <-d.ProcessLocalAnnouncement(
-							msg, nMsg.peer)
+							msg, nMsg.source)
 					}
 					if err != nil {
 						log.Errorf("Failed reprocessing"+
@@ -1801,8 +1830,9 @@ func (d *AuthenticatedGossiper) processNetworkAnnouncement(nMsg *networkMsg) []n
 		// announcement with proof (remote).
 		if proof != nil {
 			announcements = append(announcements, networkMsg{
-				msg:  msg,
-				peer: nMsg.peer,
+				peer:   nMsg.peer,
+				source: nMsg.source,
+				msg:    msg,
 			})
 		}
 
@@ -1987,20 +2017,29 @@ func (d *AuthenticatedGossiper) processNetworkAnnouncement(nMsg *networkMsg) []n
 		// so we'll try sending the update directly to the remote peer.
 		if !nMsg.isRemote && chanInfo.AuthProof == nil {
 			// Get our peer's public key.
-			var remotePeer *btcec.PublicKey
+			var remotePub *btcec.PublicKey
 			switch {
 			case msg.Flags&lnwire.ChanUpdateDirection == 0:
-				remotePeer, _ = chanInfo.NodeKey2()
+				remotePub, _ = chanInfo.NodeKey2()
 			case msg.Flags&lnwire.ChanUpdateDirection == 1:
-				remotePeer, _ = chanInfo.NodeKey1()
+				remotePub, _ = chanInfo.NodeKey1()
 			}
 
-			// Send ChannelUpdate directly to remotePeer.
-			// TODO(halseth): make reliable send?
-			if err = d.cfg.SendToPeer(remotePeer, msg); err != nil {
-				log.Errorf("unable to send channel update "+
-					"message to peer %x: %v",
-					remotePeer.SerializeCompressed(), err)
+			sPeer, err := d.cfg.FindPeer(remotePub)
+			if err != nil {
+				log.Errorf("unable to send channel update -- "+
+					"could not find peer %x: %v",
+					remotePub, err)
+			} else {
+				// Send ChannelUpdate directly to remotePeer.
+				// TODO(halseth): make reliable send?
+				err = sPeer.SendMessage(false, msg)
+				if err != nil {
+					log.Errorf("unable to send channel "+
+						"update message to peer %x: %v",
+						remotePub.SerializeCompressed(),
+						err)
+				}
 			}
 		}
 
@@ -2010,8 +2049,9 @@ func (d *AuthenticatedGossiper) processNetworkAnnouncement(nMsg *networkMsg) []n
 		// has an attached authentication proof.
 		if chanInfo.AuthProof != nil {
 			announcements = append(announcements, networkMsg{
-				msg:  msg,
-				peer: nMsg.peer,
+				peer:   nMsg.peer,
+				source: nMsg.source,
+				msg:    msg,
 			})
 		}
 
@@ -2083,10 +2123,9 @@ func (d *AuthenticatedGossiper) processNetworkAnnouncement(nMsg *networkMsg) []n
 			return nil
 		}
 
-		isFirstNode := bytes.Equal(nMsg.peer.SerializeCompressed(),
-			chanInfo.NodeKey1Bytes[:])
-		isSecondNode := bytes.Equal(nMsg.peer.SerializeCompressed(),
-			chanInfo.NodeKey2Bytes[:])
+		nodeID := nMsg.source.SerializeCompressed()
+		isFirstNode := bytes.Equal(nodeID, chanInfo.NodeKey1Bytes[:])
+		isSecondNode := bytes.Equal(nodeID, chanInfo.NodeKey2Bytes[:])
 
 		// Ensure that channel that was retrieved belongs to the peer
 		// which sent the proof announcement.
@@ -2130,7 +2169,7 @@ func (d *AuthenticatedGossiper) processNetworkAnnouncement(nMsg *networkMsg) []n
 			// received our local proof yet. So be kind and send
 			// them the full proof.
 			if nMsg.isRemote {
-				peerID := nMsg.peer.SerializeCompressed()
+				peerID := nMsg.source.SerializeCompressed()
 				log.Debugf("Got AnnounceSignatures for " +
 					"channel with full proof.")
 
@@ -2151,7 +2190,7 @@ func (d *AuthenticatedGossiper) processNetworkAnnouncement(nMsg *networkMsg) []n
 						log.Errorf("unable to gen ann: %v", err)
 						return
 					}
-					err = d.cfg.SendToPeer(nMsg.peer, chanAnn)
+					err = nMsg.peer.SendMessage(false, chanAnn)
 					if err != nil {
 						log.Errorf("Failed sending "+
 							"full proof to "+
@@ -2271,19 +2310,22 @@ func (d *AuthenticatedGossiper) processNetworkAnnouncement(nMsg *networkMsg) []n
 		// Assemble the necessary announcements to add to the next
 		// broadcasting batch.
 		announcements = append(announcements, networkMsg{
-			msg:  chanAnn,
-			peer: nMsg.peer,
+			peer:   nMsg.peer,
+			source: nMsg.source,
+			msg:    chanAnn,
 		})
 		if e1Ann != nil {
 			announcements = append(announcements, networkMsg{
-				msg:  e1Ann,
-				peer: nMsg.peer,
+				peer:   nMsg.peer,
+				source: nMsg.source,
+				msg:    e1Ann,
 			})
 		}
 		if e2Ann != nil {
 			announcements = append(announcements, networkMsg{
-				msg:  e2Ann,
-				peer: nMsg.peer,
+				peer:   nMsg.peer,
+				source: nMsg.source,
+				msg:    e2Ann,
 			})
 		}
 
@@ -2493,15 +2535,10 @@ func (d *AuthenticatedGossiper) maybeRequestChanAnn(cid lnwire.ShortChannelID) e
 		// If this syncer is already at the terminal state, then we'll
 		// chose it to request the fully channel update.
 		if syncer.SyncState() == chansSynced {
-			pub, err := btcec.ParsePubKey(nodeID[:], btcec.S256())
-			if err != nil {
-				return err
-			}
-
 			log.Debugf("attempting to request chan ann for "+
 				"chan_id=%v from node=%x", cid, nodeID[:])
 
-			return d.cfg.SendToPeer(pub, &lnwire.QueryShortChanIDs{
+			return syncer.cfg.sendToPeer(&lnwire.QueryShortChanIDs{
 				ChainHash:    d.cfg.ChainHash,
 				EncodingType: lnwire.EncodingSortedPlain,
 				ShortChanIDs: []lnwire.ShortChannelID{cid},

From 3046ea6c5b8688b3d244394a975187deb3217839 Mon Sep 17 00:00:00 2001
From: Conner Fromknecht <conner@mit.edu>
Date: Thu, 7 Jun 2018 20:10:26 -0700
Subject: [PATCH 6/8] htlcswitch/interfaces: removes Peer iface, use new
 lnpeer.Peer

---
 htlcswitch/interfaces.go | 20 ++------------------
 1 file changed, 2 insertions(+), 18 deletions(-)

diff --git a/htlcswitch/interfaces.go b/htlcswitch/interfaces.go
index 1cd24f525..d1afa5683 100644
--- a/htlcswitch/interfaces.go
+++ b/htlcswitch/interfaces.go
@@ -2,9 +2,9 @@ package htlcswitch
 
 import (
 	"github.com/lightningnetwork/lnd/channeldb"
+	"github.com/lightningnetwork/lnd/lnpeer"
 	"github.com/lightningnetwork/lnd/lnwire"
 	"github.com/roasbeef/btcd/chaincfg/chainhash"
-	"github.com/roasbeef/btcd/wire"
 )
 
 // InvoiceDatabase is an interface which represents the persistent subsystem
@@ -97,7 +97,7 @@ type ChannelLink interface {
 
 	// Peer returns the representation of remote peer with which we have
 	// the channel link opened.
-	Peer() Peer
+	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
@@ -116,22 +116,6 @@ type ChannelLink interface {
 	Stop()
 }
 
-// Peer is an interface which represents the remote lightning node inside our
-// system.
-type Peer interface {
-	// SendMessage sends message to remote peer. The second argument
-	// denotes if the method should block until the message has been sent
-	// to the remote peer.
-	SendMessage(msg lnwire.Message, sync bool) error
-
-	// WipeChannel removes the channel uniquely identified by its channel
-	// point from all indexes associated with the peer.
-	WipeChannel(*wire.OutPoint) error
-
-	// PubKey returns the serialize public key of the source peer.
-	PubKey() [33]byte
-}
-
 // ForwardingLog is an interface that represents a time series database which
 // keep track of all successfully completed payment circuits. Every few
 // seconds, the switch will collate and flush out all the successful payment

From 4380c67124c0ba112845c5c4806384ad291d7849 Mon Sep 17 00:00:00 2001
From: Conner Fromknecht <conner@mit.edu>
Date: Thu, 7 Jun 2018 20:17:15 -0700
Subject: [PATCH 7/8] htlcswitch/link: upgrade to use lnpeer.Peer.SendMessage

---
 htlcswitch/link.go | 35 ++++++++++++++++++-----------------
 1 file changed, 18 insertions(+), 17 deletions(-)

diff --git a/htlcswitch/link.go b/htlcswitch/link.go
index 297578c29..c674a9524 100644
--- a/htlcswitch/link.go
+++ b/htlcswitch/link.go
@@ -15,6 +15,7 @@ import (
 	"github.com/lightningnetwork/lnd/channeldb"
 	"github.com/lightningnetwork/lnd/contractcourt"
 	"github.com/lightningnetwork/lnd/htlcswitch/hodl"
+	"github.com/lightningnetwork/lnd/lnpeer"
 	"github.com/lightningnetwork/lnd/lnwallet"
 	"github.com/lightningnetwork/lnd/lnwire"
 	"github.com/roasbeef/btcd/chaincfg/chainhash"
@@ -163,7 +164,7 @@ type ChannelLinkConfig struct {
 
 	// Peer is a lightning network node with which we have the channel link
 	// opened.
-	Peer Peer
+	Peer lnpeer.Peer
 
 	// Registry is a sub-system which responsible for managing the invoices
 	// in thread-safe manner.
@@ -534,7 +535,7 @@ func (l *channelLink) syncChanStates() error {
 		return fmt.Errorf("unable to generate chan sync message for "+
 			"ChannelPoint(%v)", l.channel.ChannelPoint())
 	}
-	if err := l.cfg.Peer.SendMessage(localChanSyncMsg, false); err != nil {
+	if err := l.cfg.Peer.SendMessage(false, localChanSyncMsg); err != nil {
 		return fmt.Errorf("Unable to send chan sync message for "+
 			"ChannelPoint(%v)", l.channel.ChannelPoint())
 	}
@@ -576,7 +577,7 @@ func (l *channelLink) syncChanStates() error {
 			fundingLockedMsg := lnwire.NewFundingLocked(
 				l.ChanID(), nextRevocation,
 			)
-			err = l.cfg.Peer.SendMessage(fundingLockedMsg, false)
+			err = l.cfg.Peer.SendMessage(false, fundingLockedMsg)
 			if err != nil {
 				return fmt.Errorf("unable to re-send "+
 					"FundingLocked: %v", err)
@@ -626,7 +627,7 @@ func (l *channelLink) syncChanStates() error {
 		// immediately so we return to a synchronized state as soon as
 		// possible.
 		for _, msg := range msgsToReSend {
-			l.cfg.Peer.SendMessage(msg, false)
+			l.cfg.Peer.SendMessage(false, msg)
 		}
 
 	case <-l.quit:
@@ -1107,7 +1108,7 @@ func (l *channelLink) handleDownStreamPkt(pkt *htlcPacket, isReProcess bool) {
 		l.openedCircuits = append(l.openedCircuits, pkt.inKey())
 		l.keystoneBatch = append(l.keystoneBatch, pkt.keystone())
 
-		l.cfg.Peer.SendMessage(htlc, false)
+		l.cfg.Peer.SendMessage(false, htlc)
 
 	case *lnwire.UpdateFulfillHTLC:
 		// If hodl.SettleOutgoing mode is active, we exit early to
@@ -1148,7 +1149,7 @@ func (l *channelLink) handleDownStreamPkt(pkt *htlcPacket, isReProcess bool) {
 
 		// Then we send the HTLC settle message to the connected peer
 		// so we can continue the propagation of the settle message.
-		l.cfg.Peer.SendMessage(htlc, false)
+		l.cfg.Peer.SendMessage(false, htlc)
 		isSettle = true
 
 	case *lnwire.UpdateFailHTLC:
@@ -1189,7 +1190,7 @@ func (l *channelLink) handleDownStreamPkt(pkt *htlcPacket, isReProcess bool) {
 
 		// Finally, we send the HTLC message to the peer which
 		// initially created the HTLC.
-		l.cfg.Peer.SendMessage(htlc, false)
+		l.cfg.Peer.SendMessage(false, htlc)
 		isSettle = true
 	}
 
@@ -1342,7 +1343,7 @@ func (l *channelLink) handleUpstreamMsg(msg lnwire.Message) {
 			log.Errorf("unable to revoke commitment: %v", err)
 			return
 		}
-		l.cfg.Peer.SendMessage(nextRevocation, false)
+		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 over our
@@ -1561,7 +1562,7 @@ func (l *channelLink) updateCommitTx() error {
 		CommitSig: theirCommitSig,
 		HtlcSigs:  htlcSigs,
 	}
-	l.cfg.Peer.SendMessage(commitSig, false)
+	l.cfg.Peer.SendMessage(false, commitSig)
 
 	// We've just initiated a state transition, attempt to stop the
 	// logCommitTimer. If the timer already ticked, then we'll consume the
@@ -1585,7 +1586,7 @@ func (l *channelLink) updateCommitTx() error {
 // channel link opened.
 //
 // NOTE: Part of the ChannelLink interface.
-func (l *channelLink) Peer() Peer {
+func (l *channelLink) Peer() lnpeer.Peer {
 	return l.cfg.Peer
 }
 
@@ -1852,7 +1853,7 @@ func (l *channelLink) updateChannelFee(feePerKw lnwallet.SatPerKWeight) error {
 	// We'll then attempt to send a new UpdateFee message, and also lock it
 	// in immediately by triggering a commitment update.
 	msg := lnwire.NewUpdateFee(l.ChanID(), uint32(feePerKw))
-	if err := l.cfg.Peer.SendMessage(msg, false); err != nil {
+	if err := l.cfg.Peer.SendMessage(false, msg); err != nil {
 		return err
 	}
 	return l.updateCommitTx()
@@ -2260,11 +2261,11 @@ func (l *channelLink) processRemoteAdds(fwdPkg *channeldb.FwdPkg,
 
 			// HTLC was successfully settled locally send
 			// notification about it remote peer.
-			l.cfg.Peer.SendMessage(&lnwire.UpdateFulfillHTLC{
+			l.cfg.Peer.SendMessage(false, &lnwire.UpdateFulfillHTLC{
 				ChanID:          l.ChanID(),
 				ID:              pd.HtlcIndex,
 				PaymentPreimage: preimage,
-			}, false)
+			})
 			needUpdate = true
 
 		// There are additional channels left within this route. So
@@ -2550,11 +2551,11 @@ func (l *channelLink) sendHTLCError(htlcIndex uint64, failure lnwire.FailureMess
 		return
 	}
 
-	l.cfg.Peer.SendMessage(&lnwire.UpdateFailHTLC{
+	l.cfg.Peer.SendMessage(false, &lnwire.UpdateFailHTLC{
 		ChanID: l.ChanID(),
 		ID:     htlcIndex,
 		Reason: reason,
-	}, false)
+	})
 }
 
 // sendMalformedHTLCError helper function which sends the malformed HTLC update
@@ -2569,12 +2570,12 @@ func (l *channelLink) sendMalformedHTLCError(htlcIndex uint64,
 		return
 	}
 
-	l.cfg.Peer.SendMessage(&lnwire.UpdateFailMalformedHTLC{
+	l.cfg.Peer.SendMessage(false, &lnwire.UpdateFailMalformedHTLC{
 		ChanID:       l.ChanID(),
 		ID:           htlcIndex,
 		ShaOnionBlob: shaOnionBlob,
 		FailureCode:  code,
-	}, false)
+	})
 }
 
 // fail is a function which is used to encapsulate the action necessary for

From 4dace6c941e9e18aa31ca1a3cd723421d9ebcfc6 Mon Sep 17 00:00:00 2001
From: Conner Fromknecht <conner@mit.edu>
Date: Thu, 7 Jun 2018 20:18:10 -0700
Subject: [PATCH 8/8] htlcswitch/multi: update tests for lnpeer.Peer iface

---
 htlcswitch/link_test.go  | 12 ++++++++----
 htlcswitch/mock.go       | 26 +++++++++++++++++---------
 htlcswitch/test_utils.go |  3 ++-
 3 files changed, 27 insertions(+), 14 deletions(-)

diff --git a/htlcswitch/link_test.go b/htlcswitch/link_test.go
index e0f3c07e8..ea4ba96ba 100644
--- a/htlcswitch/link_test.go
+++ b/htlcswitch/link_test.go
@@ -22,6 +22,7 @@ import (
 	"github.com/lightningnetwork/lnd/channeldb"
 	"github.com/lightningnetwork/lnd/contractcourt"
 	"github.com/lightningnetwork/lnd/htlcswitch/hodl"
+	"github.com/lightningnetwork/lnd/lnpeer"
 	"github.com/lightningnetwork/lnd/lnwallet"
 	"github.com/lightningnetwork/lnd/lnwire"
 	"github.com/roasbeef/btcd/btcec"
@@ -1396,14 +1397,14 @@ type mockPeer struct {
 	quit         chan struct{}
 }
 
-var _ Peer = (*mockPeer)(nil)
+var _ lnpeer.Peer = (*mockPeer)(nil)
 
-func (m *mockPeer) SendMessage(msg lnwire.Message, sync bool) error {
+func (m *mockPeer) SendMessage(sync bool, msgs ...lnwire.Message) error {
 	if m.disconnected {
 		return fmt.Errorf("disconnected")
 	}
 	select {
-	case m.sentMsgs <- msg:
+	case m.sentMsgs <- msgs[0]:
 	case <-m.quit:
 		return fmt.Errorf("mockPeer shutting down")
 	}
@@ -1415,8 +1416,11 @@ func (m *mockPeer) WipeChannel(*wire.OutPoint) error {
 func (m *mockPeer) PubKey() [33]byte {
 	return [33]byte{}
 }
+func (m *mockPeer) IdentityKey() *btcec.PublicKey {
+	return nil
+}
 
-var _ Peer = (*mockPeer)(nil)
+var _ lnpeer.Peer = (*mockPeer)(nil)
 
 func newSingleLinkTestHarness(chanAmt, chanReserve btcutil.Amount) (
 	ChannelLink, *lnwallet.LightningChannel, chan time.Time, func() error,
diff --git a/htlcswitch/mock.go b/htlcswitch/mock.go
index 0ed96408a..db1dcc655 100644
--- a/htlcswitch/mock.go
+++ b/htlcswitch/mock.go
@@ -20,6 +20,7 @@ import (
 	"github.com/lightningnetwork/lnd/chainntnfs"
 	"github.com/lightningnetwork/lnd/channeldb"
 	"github.com/lightningnetwork/lnd/contractcourt"
+	"github.com/lightningnetwork/lnd/lnpeer"
 	"github.com/lightningnetwork/lnd/lnwallet"
 	"github.com/lightningnetwork/lnd/lnwire"
 	"github.com/roasbeef/btcd/btcec"
@@ -119,7 +120,7 @@ type mockServer struct {
 	interceptorFuncs []messageInterceptor
 }
 
-var _ Peer = (*mockServer)(nil)
+var _ lnpeer.Peer = (*mockServer)(nil)
 
 func initSwitchWithDB(db *channeldb.DB) (*Switch, error) {
 	if db == nil {
@@ -450,12 +451,14 @@ func (s *mockServer) intersect(f messageInterceptor) {
 	s.interceptorFuncs = append(s.interceptorFuncs, f)
 }
 
-func (s *mockServer) SendMessage(message lnwire.Message, sync bool) error {
+func (s *mockServer) SendMessage(sync bool, msgs ...lnwire.Message) error {
 
-	select {
-	case s.messages <- message:
-	case <-s.quit:
-		return errors.New("server is stopped")
+	for _, msg := range msgs {
+		select {
+		case s.messages <- msg:
+		case <-s.quit:
+			return errors.New("server is stopped")
+		}
 	}
 
 	return nil
@@ -506,6 +509,11 @@ func (s *mockServer) PubKey() [33]byte {
 	return s.id
 }
 
+func (s *mockServer) IdentityKey() *btcec.PublicKey {
+	pubkey, _ := btcec.ParsePubKey(s.id[:], btcec.S256())
+	return pubkey
+}
+
 func (s *mockServer) WipeChannel(*wire.OutPoint) error {
 	return nil
 }
@@ -532,7 +540,7 @@ type mockChannelLink struct {
 
 	chanID lnwire.ChannelID
 
-	peer Peer
+	peer lnpeer.Peer
 
 	startMailBox bool
 
@@ -579,7 +587,7 @@ func (f *mockChannelLink) deleteCircuit(pkt *htlcPacket) error {
 }
 
 func newMockChannelLink(htlcSwitch *Switch, chanID lnwire.ChannelID,
-	shortChanID lnwire.ShortChannelID, peer Peer, eligible bool,
+	shortChanID lnwire.ShortChannelID, peer lnpeer.Peer, eligible bool,
 ) *mockChannelLink {
 
 	return &mockChannelLink{
@@ -624,7 +632,7 @@ func (f *mockChannelLink) Start() error {
 func (f *mockChannelLink) ChanID() lnwire.ChannelID                     { return f.chanID }
 func (f *mockChannelLink) ShortChanID() lnwire.ShortChannelID           { return f.shortChanID }
 func (f *mockChannelLink) Bandwidth() lnwire.MilliSatoshi               { return 99999999 }
-func (f *mockChannelLink) Peer() Peer                                   { return f.peer }
+func (f *mockChannelLink) Peer() lnpeer.Peer                            { return f.peer }
 func (f *mockChannelLink) Stop()                                        {}
 func (f *mockChannelLink) EligibleToForward() bool                      { return f.eligible }
 func (f *mockChannelLink) setLiveShortChanID(sid lnwire.ShortChannelID) { f.shortChanID = sid }
diff --git a/htlcswitch/test_utils.go b/htlcswitch/test_utils.go
index 5ff5a1d1e..da48a4126 100644
--- a/htlcswitch/test_utils.go
+++ b/htlcswitch/test_utils.go
@@ -21,6 +21,7 @@ import (
 	"github.com/lightningnetwork/lnd/channeldb"
 	"github.com/lightningnetwork/lnd/contractcourt"
 	"github.com/lightningnetwork/lnd/keychain"
+	"github.com/lightningnetwork/lnd/lnpeer"
 	"github.com/lightningnetwork/lnd/lnwallet"
 	"github.com/lightningnetwork/lnd/lnwire"
 	"github.com/lightningnetwork/lnd/shachain"
@@ -672,7 +673,7 @@ func (r *paymentResponse) Wait(d time.Duration) (chainhash.Hash, error) {
 // * from Alice to Bob
 // * from Alice to Carol through the Bob
 // * from Alice to some another peer through the Bob
-func (n *threeHopNetwork) makePayment(sendingPeer, receivingPeer Peer,
+func (n *threeHopNetwork) makePayment(sendingPeer, receivingPeer lnpeer.Peer,
 	firstHopPub [33]byte, hops []ForwardingInfo,
 	invoiceAmt, htlcAmt lnwire.MilliSatoshi,
 	timelock uint32) *paymentResponse {