From 3e81e89062efed9f784bb854c624dfa498763481 Mon Sep 17 00:00:00 2001 From: Wilmer Paulino Date: Wed, 13 Mar 2019 13:35:31 -0700 Subject: [PATCH 1/5] discovery: add log when attempting to send msg reliably and peer is offline --- discovery/reliable_sender.go | 12 +++++++++++- 1 file changed, 11 insertions(+), 1 deletion(-) diff --git a/discovery/reliable_sender.go b/discovery/reliable_sender.go index c336b01de..1887ff1ca 100644 --- a/discovery/reliable_sender.go +++ b/discovery/reliable_sender.go @@ -181,9 +181,19 @@ out: // ignored for now since the peer is currently offline. Once // they reconnect, the messages will be sent since they should // have been persisted to disk. - case <-peerMgr.msgs: + case msg := <-peerMgr.msgs: + // Retrieve the short channel ID for which this message + // applies for logging purposes. The error can be + // ignored as the store can only contain messages which + // have a ShortChannelID field. + shortChanID, _ := msgShortChanID(msg) + log.Debugf("Received request to send %v message for "+ + "channel=%v while peer=%x is offline", + msg.MsgType(), shortChanID, peerPubKey) + case peer = <-peerChan: break out + case <-s.quit: return } From 3a2c4ec5941715be511ce56dddb56e2d9d91dadf Mon Sep 17 00:00:00 2001 From: Wilmer Paulino Date: Wed, 13 Mar 2019 13:35:44 -0700 Subject: [PATCH 2/5] discovery: add missing offline peer check before sending message reliably --- discovery/reliable_sender.go | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/discovery/reliable_sender.go b/discovery/reliable_sender.go index 1887ff1ca..6b8dc6d0f 100644 --- a/discovery/reliable_sender.go +++ b/discovery/reliable_sender.go @@ -225,6 +225,14 @@ out: // can only contain messages which have a ShortChannelID field. shortChanID, _ := msgShortChanID(msg) + // Ensure the peer is still online right before sending the + // message. + select { + case <-offlineChan: + goto waitUntilOnline + default: + } + if err := peer.SendMessage(false, msg); err != nil { log.Errorf("Unable to send %v message for channel=%v "+ "to %x: %v", msg.MsgType(), shortChanID, From 95ed11b01b42cd1a15f9f88f213d702bcd288ed5 Mon Sep 17 00:00:00 2001 From: Wilmer Paulino Date: Wed, 13 Mar 2019 13:35:55 -0700 Subject: [PATCH 3/5] discovery: properly store and retrieve edge policies within mockGraphSource In this commit, we address an issue with our router mock in which it was not properly storing and retrieving edge policies. Previously, they were being appended to a slice of policies, but this doesn't always work like when you attempt to update the same edge twice. Instead, the slice can only contain up to two entries, each one being the latest version of each direction. --- discovery/gossiper_test.go | 37 +++++++++++++++++++++++++++---------- 1 file changed, 27 insertions(+), 10 deletions(-) diff --git a/discovery/gossiper_test.go b/discovery/gossiper_test.go index 8b8a3917e..212246ccf 100644 --- a/discovery/gossiper_test.go +++ b/discovery/gossiper_test.go @@ -155,7 +155,16 @@ func (r *mockGraphSource) UpdateEdge(edge *channeldb.ChannelEdgePolicy) error { r.mu.Lock() defer r.mu.Unlock() - r.edges[edge.ChannelID] = append(r.edges[edge.ChannelID], *edge) + if len(r.edges[edge.ChannelID]) == 0 { + r.edges[edge.ChannelID] = make([]channeldb.ChannelEdgePolicy, 2) + } + + if edge.ChannelFlags&lnwire.ChanUpdateDirection == 0 { + r.edges[edge.ChannelID][0] = *edge + } else { + r.edges[edge.ChannelID][1] = *edge + } + return nil } @@ -226,13 +235,17 @@ func (r *mockGraphSource) GetChannelByID(chanID lnwire.ShortChannelID) ( return &chanInfo, nil, nil, nil } - if len(edges) == 1 { - edge1 := edges[0] - return &chanInfo, &edge1, nil, nil + var edge1 *channeldb.ChannelEdgePolicy + if !reflect.DeepEqual(edges[0], channeldb.ChannelEdgePolicy{}) { + edge1 = &edges[0] } - edge1, edge2 := edges[0], edges[1] - return &chanInfo, &edge1, &edge2, nil + var edge2 *channeldb.ChannelEdgePolicy + if !reflect.DeepEqual(edges[1], channeldb.ChannelEdgePolicy{}) { + edge2 = &edges[1] + } + + return &chanInfo, edge1, edge2, nil } func (r *mockGraphSource) FetchLightningNode( @@ -327,11 +340,15 @@ func (r *mockGraphSource) IsStaleEdgePolicy(chanID lnwire.ShortChannelID, } switch { - case len(edges) >= 1 && edges[0].ChannelFlags == flags: - return !edges[0].LastUpdate.Before(timestamp) + case flags&lnwire.ChanUpdateDirection == 0 && + !reflect.DeepEqual(edges[0], channeldb.ChannelEdgePolicy{}): - case len(edges) >= 2 && edges[1].ChannelFlags == flags: - return !edges[1].LastUpdate.Before(timestamp) + return !timestamp.After(edges[0].LastUpdate) + + case flags&lnwire.ChanUpdateDirection == 1 && + !reflect.DeepEqual(edges[1], channeldb.ChannelEdgePolicy{}): + + return !timestamp.After(edges[1].LastUpdate) default: return false From 93414bd27ac9a2ed80d6eba369892fa0ba62f084 Mon Sep 17 00:00:00 2001 From: Wilmer Paulino Date: Wed, 13 Mar 2019 13:36:07 -0700 Subject: [PATCH 4/5] discovery: clean up TestSignatureAnnouncementRetryAtStartup In this commit, we remove code from TestSignatureAnnouncementRetryAtStartup that is not crucial to the assumptions the test is exercising. --- discovery/gossiper_test.go | 131 ++++--------------------------------- 1 file changed, 14 insertions(+), 117 deletions(-) diff --git a/discovery/gossiper_test.go b/discovery/gossiper_test.go index 212246ccf..447ce1c94 100644 --- a/discovery/gossiper_test.go +++ b/discovery/gossiper_test.go @@ -1362,25 +1362,14 @@ func TestSignatureAnnouncementRetryAtStartup(t *testing.T) { sentToPeer := make(chan lnwire.Message, 1) remotePeer := &mockPeer{remoteKey, sentToPeer, ctx.gossiper.quit} - // Override NotifyWhenOnline to return the remote peer which we expect - // meesages to be sent to. + // Since the reliable send to the remote peer of the local channel proof + // requires a notification when the peer comes online, we'll capture the + // channel through which it gets sent to control exactly when to + // dispatch it. + notifyPeers := make(chan chan<- lnpeer.Peer, 1) ctx.gossiper.reliableSender.cfg.NotifyWhenOnline = func(peer *btcec.PublicKey, - peerChan chan<- lnpeer.Peer) { - - peerChan <- remotePeer - } - - // Override NotifyWhenOffline to return the channel which will notify - // the gossiper that the peer is offline. We'll use this to signal that - // the peer is offline so that the gossiper requests a notification when - // it comes back online. - notifyOffline := make(chan chan struct{}, 1) - ctx.gossiper.reliableSender.cfg.NotifyWhenOffline = func( - _ [33]byte) <-chan struct{} { - - c := make(chan struct{}) - notifyOffline <- c - return c + connectedChan chan<- lnpeer.Peer) { + notifyPeers <- connectedChan } // Recreate lightning network topology. Initialize router with channel @@ -1401,102 +1390,12 @@ func TestSignatureAnnouncementRetryAtStartup(t *testing.T) { case <-time.After(2 * trickleDelay): } - select { - case err = <-ctx.gossiper.ProcessLocalAnnouncement( - batch.chanUpdAnn1, localKey, - ): - case <-time.After(2 * time.Second): - t.Fatal("did not process local announcement") - } - if err != nil { - t.Fatalf("unable to process channel update: %v", err) - } - select { - case <-ctx.broadcastedMessage: - t.Fatal("channel update announcement was broadcast") - case <-time.After(2 * trickleDelay): - } - select { - case msg := <-sentToPeer: - assertMessage(t, batch.chanUpdAnn1, msg) - case <-time.After(1 * time.Second): - t.Fatal("gossiper did not send channel update to peer") - } - - select { - case err = <-ctx.gossiper.ProcessLocalAnnouncement( - batch.nodeAnn1, localKey, - ): - case <-time.After(2 * time.Second): - t.Fatal("did not process local announcement") - } - if err != nil { - t.Fatalf("unable to process node ann: %v", err) - } - select { - case <-ctx.broadcastedMessage: - t.Fatal("node announcement was broadcast") - case <-time.After(2 * trickleDelay): - } - - select { - case err = <-ctx.gossiper.ProcessRemoteAnnouncement( - batch.chanUpdAnn2, remotePeer, - ): - case <-time.After(2 * time.Second): - t.Fatal("did not process remote announcement") - } - if err != nil { - t.Fatalf("unable to process channel update: %v", err) - } - select { - case <-ctx.broadcastedMessage: - t.Fatal("channel update announcement was broadcast") - case <-time.After(2 * trickleDelay): - } - - select { - case err = <-ctx.gossiper.ProcessRemoteAnnouncement( - batch.nodeAnn2, remotePeer, - ): - case <-time.After(2 * time.Second): - t.Fatal("did not process local announcement") - } - if err != nil { - t.Fatalf("unable to process node ann: %v", err) - } - select { - case <-ctx.broadcastedMessage: - t.Fatal("node announcement was broadcast") - case <-time.After(2 * trickleDelay): - } - - // Since the reliable send to the remote peer of the local channel proof - // requires a notification when the peer comes online, we'll capture the - // channel through which it gets sent to control exactly when to - // dispatch it. - notifyPeers := make(chan chan<- lnpeer.Peer, 1) - ctx.gossiper.reliableSender.cfg.NotifyWhenOnline = func(peer *btcec.PublicKey, - connectedChan chan<- lnpeer.Peer) { - notifyPeers <- connectedChan - } - - // Before sending the local channel proof, we'll notify that the peer is - // offline, so that it's not sent to the peer. - var peerOffline chan struct{} - select { - case peerOffline = <-notifyOffline: - case <-time.After(2 * time.Second): - t.Fatalf("gossiper did not request notification for when " + - "peer disconnects") - } - close(peerOffline) - // Pretending that we receive local channel announcement from funding // manager, thereby kick off the announcement exchange process. select { - case err = <-ctx.gossiper.ProcessLocalAnnouncement(batch.localProofAnn, - localKey): + case err = <-ctx.gossiper.ProcessLocalAnnouncement( + batch.localProofAnn, localKey, + ): case <-time.After(2 * time.Second): t.Fatal("did not process remote announcement") } @@ -1615,12 +1514,10 @@ out: t.Fatalf("unable to process :%v", err) } - for i := 0; i < 5; i++ { - select { - case <-ctx.broadcastedMessage: - case <-time.After(time.Second): - t.Fatal("announcement wasn't broadcast") - } + select { + case <-ctx.broadcastedMessage: + case <-time.After(time.Second): + t.Fatal("announcement wasn't broadcast") } number = 0 From 0ab97957ea731c87f9e6c0858b1429dbe6ce8c32 Mon Sep 17 00:00:00 2001 From: Wilmer Paulino Date: Wed, 13 Mar 2019 13:36:21 -0700 Subject: [PATCH 5/5] discovery: check if stale within isMsgStale for ChannelUpdate messages In this commit, we address an assumption of the gossiper's recently introduce reliable sender. The reliable sender is currently only used for messages of unannounced channels. This makes sense as peers should be able to retrieve messages from the network if they've previously announced. However, within isMsgStale, we assumed that the reliable sender would be used for every ChannelUpdate being sent, even if the channel is already announced. Due to this, checking if the policy is stale was unnecessary. But since this isn't the case, we should actually be checking whether it is stale to prevent sending it later on. --- discovery/gossiper.go | 39 ++++++--- discovery/gossiper_test.go | 171 +++++++++++++++++++++++++++++++++---- 2 files changed, 184 insertions(+), 26 deletions(-) diff --git a/discovery/gossiper.go b/discovery/gossiper.go index 171b04aba..fb6cd7d29 100644 --- a/discovery/gossiper.go +++ b/discovery/gossiper.go @@ -221,7 +221,10 @@ type AuthenticatedGossiper struct { peerSyncers map[routing.Vertex]*gossipSyncer // reliableSender is a subsystem responsible for handling reliable - // message send requests to peers. + // message send requests to peers. This should only be used for channels + // that are unadvertised at the time of handling the message since if it + // is advertised, then peers should be able to get the message from the + // network. reliableSender *reliableSender sync.Mutex @@ -2364,16 +2367,32 @@ func (d *AuthenticatedGossiper) isMsgStale(msg lnwire.Message) bool { return chanInfo.AuthProof != nil case *lnwire.ChannelUpdate: - // The MessageStore will always store the latest ChannelUpdate - // as it is not aware of its timestamp (by design), so it will - // never be stale. We should still however check if the channel - // is part of our graph. If it's not, we can mark it as stale. - _, _, _, err := d.cfg.Router.GetChannelByID(msg.ShortChannelID) - if err != nil && err != channeldb.ErrEdgeNotFound { - log.Debugf("Unable to retrieve channel=%v from graph: "+ - "%v", err) + _, p1, p2, err := d.cfg.Router.GetChannelByID(msg.ShortChannelID) + + // If the channel cannot be found, it is most likely a leftover + // message for a channel that was closed, so we can consider it + // stale. + if err == channeldb.ErrEdgeNotFound { + return true } - return err == channeldb.ErrEdgeNotFound + if err != nil { + log.Debugf("Unable to retrieve channel=%v from graph: "+ + "%v", msg.ShortChannelID, err) + return false + } + + // Otherwise, we'll retrieve the correct policy that we + // currently have stored within our graph to check if this + // message is stale by comparing its timestamp. + var p *channeldb.ChannelEdgePolicy + if msg.ChannelFlags&lnwire.ChanUpdateDirection == 0 { + p = p1 + } else { + p = p2 + } + + timestamp := time.Unix(int64(msg.Timestamp), 0) + return p.LastUpdate.After(timestamp) default: // We'll make sure to not mark any unsupported messages as stale diff --git a/discovery/gossiper_test.go b/discovery/gossiper_test.go index 447ce1c94..2e19f9fe2 100644 --- a/discovery/gossiper_test.go +++ b/discovery/gossiper_test.go @@ -24,6 +24,7 @@ import ( "github.com/lightningnetwork/lnd/chainntnfs" "github.com/lightningnetwork/lnd/channeldb" "github.com/lightningnetwork/lnd/lnpeer" + "github.com/lightningnetwork/lnd/lntest" "github.com/lightningnetwork/lnd/lnwire" "github.com/lightningnetwork/lnd/routing" ) @@ -2956,16 +2957,17 @@ func TestSendChannelUpdateReliably(t *testing.T) { return c } - // assertReceivedChannelUpdate is a helper closure we'll use to - // determine if the correct channel update was received. - assertReceivedChannelUpdate := func(channelUpdate *lnwire.ChannelUpdate) { + // assertMsgSent is a helper closure we'll use to determine if the + // correct gossip message was sent. + assertMsgSent := func(msg lnwire.Message) { t.Helper() select { - case msg := <-sentToPeer: - assertMessage(t, batch.chanUpdAnn1, msg) + case msgSent := <-sentToPeer: + assertMessage(t, msg, msgSent) case <-time.After(2 * time.Second): - t.Fatal("did not send local channel update to peer") + t.Fatalf("did not send %v message to peer", + msg.MsgType()) } } @@ -3022,7 +3024,7 @@ func TestSendChannelUpdateReliably(t *testing.T) { // We can go ahead and notify the peer, which should trigger the message // to be sent. peerChan <- remotePeer - assertReceivedChannelUpdate(batch.chanUpdAnn1) + assertMsgSent(batch.chanUpdAnn1) // The gossiper should now request a notification for when the peer // disconnects. We'll also trigger this now. @@ -3046,12 +3048,9 @@ func TestSendChannelUpdateReliably(t *testing.T) { } // Now that the remote peer is offline, we'll send a new channel update. - prevTimestamp := batch.chanUpdAnn1.Timestamp - newChanUpdate, err := createUpdateAnnouncement( - 0, 0, nodeKeyPriv1, prevTimestamp+1, - ) - if err != nil { - t.Fatalf("unable to create new channel update: %v", err) + batch.chanUpdAnn1.Timestamp++ + if err := signUpdate(nodeKeyPriv1, batch.chanUpdAnn1); err != nil { + t.Fatalf("unable to sign new channel update: %v", err) } // With the new update created, we'll go ahead and process it. @@ -3081,10 +3080,150 @@ func TestSendChannelUpdateReliably(t *testing.T) { case <-time.After(time.Second): } - // Finally, we'll notify the peer is online and ensure the new channel - // update is received. + // Once again, we'll notify the peer is online and ensure the new + // channel update is received. This will also cause an offline + // notification to be requested again. peerChan <- remotePeer - assertReceivedChannelUpdate(newChanUpdate) + assertMsgSent(batch.chanUpdAnn1) + + select { + case offlineChan = <-notifyOffline: + case <-time.After(2 * time.Second): + t.Fatal("gossiper did not request notification upon peer " + + "disconnection") + } + + // We'll then exchange proofs with the remote peer in order to announce + // the channel. + select { + case err = <-ctx.gossiper.ProcessLocalAnnouncement( + batch.localProofAnn, localKey, + ): + case <-time.After(2 * time.Second): + t.Fatal("did not process local channel proof") + } + if err != nil { + t.Fatalf("unable to process local channel proof: %v", err) + } + + // No messages should be broadcast as we don't have the full proof yet. + select { + case <-ctx.broadcastedMessage: + t.Fatal("channel announcement was broadcast") + case <-time.After(2 * trickleDelay): + } + + // Our proof should be sent to the remote peer however. + assertMsgSent(batch.localProofAnn) + + select { + case err = <-ctx.gossiper.ProcessRemoteAnnouncement( + batch.remoteProofAnn, remotePeer, + ): + case <-time.After(2 * time.Second): + t.Fatal("did not process remote channel proof") + } + if err != nil { + t.Fatalf("unable to process remote channel proof: %v", err) + } + + // Now that we've constructed our full proof, we can assert that the + // channel has been announced. + for i := 0; i < 2; i++ { + select { + case <-ctx.broadcastedMessage: + case <-time.After(2 * trickleDelay): + t.Fatal("expected channel to be announced") + } + } + + // With the channel announced, we'll generate a new channel update. This + // one won't take the path of the reliable sender, as the channel has + // already been announced. We'll keep track of the old message that is + // now stale to use later on. + staleChannelUpdate := batch.chanUpdAnn1 + newChannelUpdate := &lnwire.ChannelUpdate{} + *newChannelUpdate = *staleChannelUpdate + newChannelUpdate.Timestamp++ + if err := signUpdate(nodeKeyPriv1, newChannelUpdate); err != nil { + t.Fatalf("unable to sign new channel update: %v", err) + } + + // Process the new channel update. It should not be sent to the peer + // directly since the reliable sender only applies when the channel is + // not announced. + select { + case err = <-ctx.gossiper.ProcessLocalAnnouncement( + newChannelUpdate, localKey, + ): + case <-time.After(2 * time.Second): + t.Fatal("did not process local channel update") + } + if err != nil { + t.Fatalf("unable to process local channel update: %v", err) + } + select { + case <-ctx.broadcastedMessage: + case <-time.After(2 * trickleDelay): + t.Fatal("channel update was not broadcast") + } + select { + case msg := <-sentToPeer: + t.Fatalf("received unexpected message: %v", spew.Sdump(msg)) + case <-time.After(time.Second): + } + + // Then, we'll trigger the reliable sender to send its pending messages + // by triggering an offline notification for the peer, followed by an + // online one. + close(offlineChan) + + select { + case peerChan = <-notifyOnline: + case <-time.After(2 * time.Second): + t.Fatal("gossiper did not request notification upon peer " + + "connection") + } + + peerChan <- remotePeer + + // At this point, we should have sent both the AnnounceSignatures and + // stale ChannelUpdate. + for i := 0; i < 2; i++ { + var msg lnwire.Message + select { + case msg = <-sentToPeer: + case <-time.After(time.Second): + t.Fatal("expected to send message") + } + + switch msg := msg.(type) { + case *lnwire.ChannelUpdate: + assertMessage(t, staleChannelUpdate, msg) + case *lnwire.AnnounceSignatures: + assertMessage(t, batch.localProofAnn, msg) + default: + t.Fatalf("send unexpected %v message", msg.MsgType()) + } + } + + // Since the messages above are now deemed as stale, they should be + // removed from the message store. + err = lntest.WaitNoError(func() error { + msgs, err := ctx.gossiper.cfg.MessageStore.Messages() + if err != nil { + return fmt.Errorf("unable to retrieve pending "+ + "messages: %v", err) + } + if len(msgs) != 0 { + return fmt.Errorf("expected no messages left, found %d", + len(msgs)) + } + return nil + }, time.Second) + if err != nil { + t.Fatal(err) + } } func assertMessage(t *testing.T, expected, got lnwire.Message) {