Merge pull request #10211 from yyforyongyu/remove-flap-count

chanfitness: track flap counts for peers with channels

chanfitness/chaneventstore.go

@@ -245,7 +245,7 @@ func (c *ChannelEventStore) Stop() error {
 func (c *ChannelEventStore) addChannel(channelPoint wire.OutPoint,
 	peer route.Vertex) {
 
-	peerMonitor, err := c.getPeerMonitor(peer)
+	peerMonitor, err := c.getOrCreatePeerMonitor(peer)
 	if err != nil {
 		log.Error("could not create monitor: %v", err)
 		return
@@ -256,10 +256,12 @@ func (c *ChannelEventStore) addChannel(channelPoint wire.OutPoint,
 	}
 }
 
-// getPeerMonitor tries to get an existing peer monitor from our in memory list,
+// getOrCreatePeerMonitor tries to get an existing peer monitor from our in
-// and falls back to creating a new monitor if it is not currently known.
+// memory list. If the peer is not yet known to us, it will create a new
-func (c *ChannelEventStore) getPeerMonitor(peer route.Vertex) (peerMonitor,
+// monitor and add it to the list. When a new monitor is created, we also send
-	error) {
+// an initial online event for the peer.
+func (c *ChannelEventStore) getOrCreatePeerMonitor(
+	peer route.Vertex) (peerMonitor, error) {
 
 	peerMonitor, ok := c.peers[peer]
 	if ok {
@@ -289,6 +291,9 @@ func (c *ChannelEventStore) getPeerMonitor(peer route.Vertex) (peerMonitor,
 	peerMonitor = newPeerLog(c.cfg.Clock, flapCount, lastFlap)
 	c.peers[peer] = peerMonitor
 
+	// Send an online event given it's the first time we see this peer.
+	peerMonitor.onlineEvent(true)
+
 	return peerMonitor, nil
 }
 
@@ -313,12 +318,15 @@ func (c *ChannelEventStore) closeChannel(channelPoint wire.OutPoint,
 	}
 }
 
-// peerEvent creates a peer monitor for a peer if we do not currently have
+// peerEvent adds an online event to a peer's monitor. If the peer is not
-// one, and adds an online event to it.
+// yet known to the event store, the event is ignored. A peer is only known to
+// the event store if we have an open channel with them.
 func (c *ChannelEventStore) peerEvent(peer route.Vertex, online bool) {
-	peerMonitor, err := c.getPeerMonitor(peer)
+	peerMonitor, ok := c.peers[peer]
-	if err != nil {
+	if !ok {
-		log.Error("could not create monitor: %v", err)
+		log.Tracef("Ignore peer event (online=%v) from non-channel "+
+			"peer: %v", online, peer)
+
 		return
 	}
 
@@ -371,9 +379,12 @@ func (c *ChannelEventStore) consume(subscriptions *subscriptions) {
 						"from: %v", compressed)
 				}
 
-				c.addChannel(
+				op := event.Channel.FundingOutpoint
-					event.Channel.FundingOutpoint, peerKey,
+
-				)
+				log.Tracef("Received OpenChannelEvent(%v) "+
+					"from %v", op, peerKey)
+
+				c.addChannel(op, peerKey)
 
 			// A channel has been closed, we must remove the channel
 			// from the store and record a channel closed event.
@@ -487,7 +498,7 @@ func (c *ChannelEventStore) getChanInfo(req channelInfoRequest) (*ChannelInfo,
 
 	peerMonitor, ok := c.peers[req.peer]
 	if !ok {
-		return nil, ErrPeerNotFound
+		return nil, fmt.Errorf("%w: %v", ErrPeerNotFound, req.peer)
 	}
 
 	lifetime, uptime, err := peerMonitor.channelUptime(req.channelPoint)

chanfitness/chaneventstore_test.go

@@ -189,11 +189,11 @@ func TestStoreFlapCount(t *testing.T) {
 	// flush our tick count to disk.
 	testCtx.tickFlapCount()
 
-	// Since we just tracked a offline event, we expect a single flap for
+	// Since we just tracked an offline event, we expect two flaps for our
-	// our peer.
+	// peer because the channel opening also records an online event.
 	expectedUpdate := peerFlapCountMap{
 		pubkey: {
-			Count:    1,
+			Count:    2,
 			LastFlap: testCtx.clock.Now(),
 		},
 	}
@@ -210,9 +210,9 @@ func TestStoreFlapCount(t *testing.T) {
 	testCtx.tickFlapCount()
 
 	// Since we have processed 3 more events for our peer, we update our
-	// expected online map to have a flap count of 4 for this peer.
+	// expected online map to have a flap count of 5 for this peer.
 	expectedUpdate[pubkey] = &channeldb.FlapCount{
-		Count:    4,
+		Count:    5,
 		LastFlap: testCtx.clock.Now(),
 	}
 	testCtx.assertFlapCountUpdated()
@@ -234,13 +234,13 @@ func TestGetChanInfo(t *testing.T) {
 	peer, pk, channel := ctx.newChannel()
 
 	// Send an online event for our peer, although we do not yet have an
-	// open channel.
+	// open channel - this event will be ignored.
 	ctx.peerEvent(peer, true)
 
 	// Try to get info for a channel that has not been opened yet, we
 	// expect to get an error.
 	_, err := ctx.store.GetChanInfo(channel, peer)
-	require.Equal(t, ErrChannelNotFound, err)
+	require.ErrorIs(t, err, ErrPeerNotFound)
 
 	// Now we send our store a notification that a channel has been opened.
 	ctx.sendChannelOpenedUpdate(pk, channel)
@@ -310,9 +310,13 @@ func TestFlapCount(t *testing.T) {
 
 	ctx.start()
 
-	// Create test variables for a peer and channel, but do not add it to
+	// Create a valid public key for our peer.
-	// our store yet.
+	priv, err := btcec.NewPrivateKey()
-	peer1 := route.Vertex{1, 2, 3}
+	require.NoError(t, err)
+	pubkeyBytes := priv.PubKey().SerializeCompressed()
+
+	var peer1 route.Vertex
+	copy(peer1[:], pubkeyBytes)
 
 	// First, query for a peer that we have no record of in memory or on
 	// disk and confirm that we indicate that the peer was not found.
@@ -320,10 +324,23 @@ func TestFlapCount(t *testing.T) {
 	require.NoError(t, err)
 	require.Nil(t, ts)
 
-	// Send an online event for our peer.
+	// Send an online event for our peer - given this peer doesn't have a
+	// channel with, this event will be ignored.
 	ctx.peerEvent(peer1, true)
 
-	// Assert that we now find a record of the peer with flap count = 1.
+	// Assert the event is ignored.
+	_, ts, err = ctx.store.FlapCount(peer1)
+	require.NoError(t, err)
+	require.Nil(t, ts)
+
+	// Create a channel for the peer so that it will be tracked by the
+	// store.
+	pk, err := btcec.ParsePubKey(peer1[:])
+	require.NoError(t, err)
+	ctx.sendChannelOpenedUpdate(pk, wire.OutPoint{Index: 9})
+
+	// Assert that we now find a record of the peer with flap count = 1,
+	// because the channel open fires an online event.
 	count, ts, err := ctx.store.FlapCount(peer1)
 	require.NoError(t, err)
 	require.Equal(t, lastFlap, *ts)

docs/release-notes/release-notes-0.20.0.md

@@ -186,6 +186,10 @@ circuit. The indices are only available for forwarding events saved after v0.20.
   for the case where the DB is kvdb backed and no invoices have yet been added
   to the database.
 
+* The `FlapCount` and `LastFlapNs` have been
+  [changed](https://github.com/lightningnetwork/lnd/pull/10211) to track
+  exclusively for peers that have channels with us.
+
 ## lncli Updates
 * Previously, users could only specify one `outgoing_chan_id` when calling the 
   `lncli queryroutes` or the QueryRoutes RPC. With this change, multiple 

itest/lnd_misc_test.go

@@ -257,6 +257,17 @@ func testListChannels(ht *lntest.HarnessTest) {
 	assertChannelConstraintsEqual(
 		ht, aliceChannel.RemoteConstraints, bobChannel.LocalConstraints,
 	)
+
+	// Finally we assert that the flap count is updated as expected.
+	resp := alice.RPC.ListPeers()
+
+	// Assert Alice only have one peer.
+	require.Len(ht, resp.Peers, 1)
+	for _, p := range resp.Peers {
+		// The channel open event resulted in an online event, so we
+		// expect the flap count to be 1.
+		require.EqualValues(ht, 1, p.FlapCount)
+	}
 }
 
 // testMaxPendingChannels checks that error is returned from remote peer if

lnrpc/lightning.pb.go

@@ -5816,11 +5816,13 @@ type Peer struct {
 	// stored for peers that we have channels open with, to prevent peers from
 	// spamming us with errors at no cost.
 	Errors []*TimestampedError `protobuf:"bytes,12,rep,name=errors,proto3" json:"errors,omitempty"`
+	// This field is populated when the peer has at least one channel with us.
 	// The number of times we have recorded this peer going offline or coming
 	// online, recorded across restarts. Note that this value is decreased over
 	// time if the peer has not recently flapped, so that we can forgive peers
 	// with historically high flap counts.
 	FlapCount int32 `protobuf:"varint,13,opt,name=flap_count,json=flapCount,proto3" json:"flap_count,omitempty"`
+	// This field is populated when the peer has at least one channel with us.
 	// The timestamp of the last flap we observed for this peer. If this value is
 	// zero, we have not observed any flaps for this peer.
 	LastFlapNs int64 `protobuf:"varint,14,opt,name=last_flap_ns,json=lastFlapNs,proto3" json:"last_flap_ns,omitempty"`

lnrpc/lightning.proto

@@ -1919,6 +1919,7 @@ message Peer {
     repeated TimestampedError errors = 12;
 
     /*
+    This field is populated when the peer has at least one channel with us.
     The number of times we have recorded this peer going offline or coming
     online, recorded across restarts. Note that this value is decreased over
     time if the peer has not recently flapped, so that we can forgive peers
@@ -1927,6 +1928,7 @@ message Peer {
     int32 flap_count = 13;
 
     /*
+    This field is populated when the peer has at least one channel with us.
     The timestamp of the last flap we observed for this peer. If this value is
     zero, we have not observed any flaps for this peer.
     */

lnrpc/lightning.swagger.json

@@ -6926,12 +6926,12 @@
         "flap_count": {
           "type": "integer",
           "format": "int32",
-          "description": "The number of times we have recorded this peer going offline or coming\nonline, recorded across restarts. Note that this value is decreased over\ntime if the peer has not recently flapped, so that we can forgive peers\nwith historically high flap counts."
+          "description": "This field is populated when the peer has at least one channel with us.\nThe number of times we have recorded this peer going offline or coming\nonline, recorded across restarts. Note that this value is decreased over\ntime if the peer has not recently flapped, so that we can forgive peers\nwith historically high flap counts."
         },
         "last_flap_ns": {
           "type": "string",
           "format": "int64",
-          "description": "The timestamp of the last flap we observed for this peer. If this value is\nzero, we have not observed any flaps for this peer."
+          "description": "This field is populated when the peer has at least one channel with us.\nThe timestamp of the last flap we observed for this peer. If this value is\nzero, we have not observed any flaps for this peer."
         },
         "last_ping_payload": {
           "type": "string",

lntest/harness.go

@@ -2242,7 +2242,7 @@ func (h *HarnessTest) ReceiveChannelEvent(
 
 	select {
 	case <-time.After(DefaultTimeout):
-		require.Fail(h, "timeout", "timeout intercepting htlc")
+		require.Fail(h, "timeout", "timeout receiving channel events")
 
 	case err := <-errChan:
 		require.Failf(h, "err from stream",

rpcserver.go

@@ -3561,8 +3561,12 @@ func (r *rpcServer) ListPeers(ctx context.Context,
 			flap, ts, err := r.server.chanEventStore.FlapCount(
 				vertex,
 			)
+
+			// Log the error if we cannot get the flap count instead
+			// of failing this RPC call.
 			if err != nil {
-				return nil, err
+				rpcsLog.Debugf("Failed to get flap count for "+
+					"peer %v", vertex)
 			}
 
 			// If our timestamp is non-nil, we have values for our
@@ -5080,14 +5084,19 @@ func createRPCOpenChannel(ctx context.Context, r *rpcServer,
 	// being notified of it.
 	outpoint := dbChannel.FundingOutpoint
 	info, err := r.server.chanEventStore.GetChanInfo(outpoint, peer)
-	switch err {
+	switch {
+	// If the store does not know about the peer, we just log it.
+	case errors.Is(err, chanfitness.ErrPeerNotFound):
+		rpcsLog.Warnf("peer: %v not found by channel event store",
+			peer)
+
 	// If the store does not know about the channel, we just log it.
-	case chanfitness.ErrChannelNotFound:
+	case errors.Is(err, chanfitness.ErrChannelNotFound):
-		rpcsLog.Infof("channel: %v not found by channel event store",
+		rpcsLog.Warnf("channel: %v not found by channel event store",
 			outpoint)
 
 	// If we got our channel info, we further populate the channel.
-	case nil:
+	case err == nil:
 		channel.Uptime = int64(info.Uptime.Seconds())
 		channel.Lifetime = int64(info.Lifetime.Seconds())