Repository: lightningnetwork/lnd (language: go)

Issue:
[bug]: Adding future message got error: can't insert entry of size XXX into cache with capacity 100
Versions:

lnd v0.15.5-beta-8-gb09afd4dd
bitcoind v23.0

A few days (I noticed it late), only one of the nodes [lnd-10] (apparently a very rare bug - I couldn't find anything about it) lost synchronization with the blockchain:

```
lncli getinfo|grep sync
    "synced_to_chain": false,
    "synced_to_graph": true,
```

But the problem was an error about which I could not find anything anywhere (this is already pure `grep`):

```
2023-02-06 14:03:01.403 [ERR] DISC: Adding future message got error: can't insert entry of size 101 into cache with capacity 100
2023-02-06 14:03:01.646 [ERR] DISC: Adding future message got error: can't insert entry of size 102 into cache with capacity 100
2023-02-06 14:03:03.430 [ERR] DISC: Adding future message got error: can't insert entry of size 103 into cache with capacity 100
2023-02-06 14:03:03.431 [ERR] DISC: Adding future message got error: can't insert entry of size 104 into cache with capacity 100
2023-02-06 14:03:06.571 [ERR] DISC: Adding future message got error: can't insert entry of size 105 into cache with capacity 100
2023-02-06 14:03:10.018 [ERR] DISC: Adding future message got error: can't insert entry of size 106 into cache with capacity 100
2023-02-06 14:03:10.019 [ERR] DISC: Adding future message got error: can't insert entry of size 107 into cache with capacity 100
2023-02-06 14:03:10.324 [ERR] DISC: Adding future message got error: can't insert entry of size 108 into cache with capacity 100
2023-02-06 14:03:11.307 [ERR] DISC: Adding future message got error: can't insert entry of size 109 into cache with capacity 100
2023-02-06 14:03:11.767 [ERR] DISC: Adding future message got error: can't insert entry of size 110 into cache with capacity 100
2023-02-06 14:03:11.767 [ERR] DISC: Adding future message got error: can't insert entry of size 111 into cache with capacity 100
2023-02-06 14:03:11.818 [ERR] DISC: Adding future message got error: can't insert entry of size 112 into cache with capacity 100
2023-02-06 14:03:12.913 [ERR] DISC: Adding future message got error: can't insert entry of size 113 into cache with capacity 100
...
2023-02-10 12:50:46.563 [ERR] DISC: Adding future message got error: can't insert entry of size 634 into cache with capacity 100
2023-02-10 12:50:48.818 [ERR] DISC: Adding future message got error: can't insert entry of size 635 into cache with capacity 100
2023-02-10 12:50:51.144 [ERR] DISC: Adding future message got error: can't insert entry of size 298 into cache with capacity 100
2023-02-10 12:51:06.149 [ERR] DISC: Adding future message got error: can't insert entry of size 281 into cache with capacity 100
2023-02-10 12:51:06.526 [ERR] DISC: Adding future message got error: can't insert entry of size 282 into cache with capacity 100
2023-02-10 12:51:16.850 [ERR] DISC: Adding future message got error: can't insert entry of size 115 into cache with capacity 100
2023-02-10 12:51:26.156 [ERR] DISC: Adding future message got error: can't insert entry of size 188 into cache with capacity 100
2023-02-10 12:51:33.202 [ERR] DISC: Adding future message got error: can't insert entry of size 116 into cache with capacity 100
2023-02-10 12:52:06.089 [ERR] DISC: Adding future message got error: can't insert entry of size 636 into cache with capacity 100
2023-02-10 12:53:28.213 [ERR] DISC: Adding future message got error: can't insert entry of size 399 into cache with capacity 100
2023-02-10 12:54:38.969 [ERR] DISC: Adding future message got error: can't insert entry of size 152 into cache with capacity 100
2023-02-10 12:54:47.504 [ERR] DISC: Adding future message got error: can't insert entry of size 400 into cache with capacity 100
```

before the error occurred , there were such lines:

```
2023-02-06 14:01:15.435 [INF] CRTR: Processed channels=0 updates=108 nodes=5 in last 59.999988483s
2023-02-06 14:01:28.614 [WRN] DISC: Announcement for chan_id=(852433773016449025), is premature: advertises height 775284, only height 775282 is known
2023-02-06 14:01:31.658 [WRN] DISC: Announcement for chan_id=(852433773016449025), is premature: advertises height 775284, only height 775282 is known
2023-02-06 14:01:36.460 [WRN] DISC: Announcement for chan_id=(852433773016449025), is premature: advertises height 775284, only height 775282 is known
2023-02-06 14:01:40.325 [WRN] DISC: Announcement for chan_id=(852433773016449025), is premature: advertises height 775284, only height 775282 is known
2023-02-06 14:01:41.639 [WRN] DISC: Announcement for chan_id=(852433773016449025), is premature: advertises height 775284, only height 775282 is known
2023-02-06 14:01:41.817 [WRN] DISC: Announcement for chan_id=(852433773016449025), is premature: advertises height 775284, only height 775282 is known
2023-02-06 14:01:43.798 [WRN] DISC: Announcement for chan_id=(852433773016449025), is premature: advertises height 775284, only height 775282 is known
2023-02-06 14:01:45.481 [WRN] DISC: Announcement for chan_id=(852433773016449025), is premature: advertises height 775284, only height 775282 is known
2023-02-06 14:01:45.665 [WRN] DISC: Announcement for chan_id=(852433773016449025), is premature: advertises height 775284, only height 775282 is known
2023-02-06 14:01:45.936 [WRN] DISC: Announcement for chan_id=(852433773016449025), is premature: advertises height 775284, only height 775282 is known
2023-02-06 14:01:47.943 [WRN] DISC: Announcement for chan_id=(852433773016449025), is premature: advertises height 775284, only height 775282 is known
2023-02-06 14:01:48.983 [WRN] DISC: Announcement for chan_id=(852433773016449025), is premature: advertises height 775284, only height 775282 is known
2023-02-06 14:01:49.315 [WRN] DISC: Announcement for chan_id=(852433773016449025), is premature: advertises height 775284, only height 775282 is known
2023-02-06 14:01:51.776 [WRN] DISC: Announcement for chan_id=(852433773016449025), is premature: advertises height 775284, only height 775282 is known
2023-02-06 14:01:53.474 [WRN] DISC: Announcement for chan_id=(852433773016449025), is premature: advertises height 775284, only height 775282 is known
2023-02-06 14:01:57.735 [WRN] DISC: Announcement for chan_id=(852433773016449025), is premature: advertises height 775284, only height 775282 is known
2023-02-06 14:01:57.904 [WRN] DISC: Announcement for chan_id=(852433773016449025), is premature: advertises height 775284, only height 775282 is known
2023-02-06 14:01:59.622 [WRN] DISC: Announcement for chan_id=(852433773016449025), is premature: advertises height 775284, only height 775282 is known
..
2023-02-06 14:02:56.615 [WRN] DISC: Announcement for chan_id=(852433773016449025), is premature: advertises height 775284, only height 775282 is known
2023-02-06 14:02:57.792 [WRN] DISC: Announcement for chan_id=(852433773024116737), is premature: advertises height 775284, only height 775282 is known
2023-02-06 14:02:57.792 [WRN] DISC: Announcement for chan_id=(852433773016449025), is premature: advertises height 775284, only height 775282 is known
2023-02-06 14:02:58.605 [WRN] DISC: Announcement for chan_id=(852433773024116737), is premature: advertises height 775284, only height 775282 is known
2023-02-06 14:02:59.017 [WRN] DISC: Announcement for chan_id=(852433773016449025), is premature: advertises height 775284, only height 775282 is known
2023-02-06 14:03:00.190 [WRN] DISC: Announcement for chan_id=(852433773016449025), is premature: advertises height 775284, only height 775282 is known
2023-02-06 14:03:00.190 [WRN] DISC: Announcement for chan_id=(852433773024116737), is premature: advertises height 775284, only height 775282 is known
2023-02-06 14:03:01.402 [WRN] DISC: Announcement for chan_id=(852433773016449025), is premature: advertises height 775284, only height 775282 is known
2023-02-06 14:03:01.403 [ERR] DISC: Adding future message got error: can't insert entry of size 101 into cache with capacity 100
2023-02-06 14:03:01.403 [WRN] DISC: Announcement for chan_id=(852433773024116737), is premature: advertises height 775284, only height 775282 is known
2023-02-06 14:03:01.646 [ERR] DISC: Adding future message got error: can't insert entry of size 102 into cache with capacity 100
2023-02-06 14:03:01.646 [WRN] DISC: Announcement for chan_id=(852433773024116737), is premature: advertises height 775284, only height 775282 is known
2023-02-06 14:03:03.430 [ERR] DISC: Adding future message got error: can't insert entry of size 103 into cache with capacity 100
2023-02-06 14:03:03.430 [WRN] DISC: Announcement for chan_id=(852433773024116737), is premature: advertises height 775284, only height 775282 is known
2023-02-06 14:03:03.431 [ERR] DISC: Adding future message got error: can't insert entry of size 104 into cache with capacity 100
2023-02-06 14:03:03.431 [WRN] DISC: Announcement for chan_id=(852433773016449025), is premature: advertises height 775284, only height 775282 is known
```

I found that the error is generated here `discovery/gossiper.go:1806` and to be related with `github.com/lightninglabs/neutrino/cache/lru`



Discussion / hints:
When the `lnd` node is not synced, it cannot process gossip messages that have future block height(from its view), thus it puts the messages in a cache and will reprocess them once it's synced. The cache has a limited size to prevent DoS, but I guess we could make this cache size configurable.

```
2023-02-10 12:50:48.818 [ERR] DISC: Adding future message got error: can't insert entry of size 635 into cache with capacity 100
```

Looks like there are quite some.
I think this is the issue: https://github.com/lightningnetwork/lnd/blob/939614b70db8bea84e0e47543a17b1b700b3d0ab/discovery/gossiper.go#L331-L336

`Size()` should return the size of the element. If we want to limit by count, then we make that `1`. In this case, the "capacity" is 100, but the number of messages (`Size()`) here exceeds that. I think we want to instead "unroll" that, so `cachedNetworkMsg` is just a single message. Then things should work properly. 
This hasn't been fully resolved. 
I think there are multiple ways to fix this,
- make sure the `value` in `cache` is always one item, not a list. So we could construct a unique `key`, like the combination of `blockHeight+shortChannelID+msgType`.
- the other is to use another level of `cache`. So the `futureMsgs` will be `blockHeight->shortChanID->[msgs...]`. And the 100 limit applies to the inner cache, as we'd expect less than 100 future message sent by a particular channel in a specific height.

Fix patch:
diff --git a/discovery/gossiper.go b/discovery/gossiper.go
index 96f6b07554..5eac87d91c 100644
--- a/discovery/gossiper.go
+++ b/discovery/gossiper.go
@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/btcsuite/btcd/btcec/v2"
@@ -47,6 +48,10 @@ const (
 	// updates that we'll hold onto.
 	maxPrematureUpdates = 100
 
+	// maxFutureMessages tracks the max amount of future messages that
+	// we'll hold onto.
+	maxFutureMessages = 1000
+
 	// DefaultSubBatchDelay is the default delay we'll use when
 	// broadcasting the next announcement batch.
 	DefaultSubBatchDelay = 5 * time.Second
@@ -424,8 +429,8 @@ type AuthenticatedGossiper struct {
 	// height specified in the future. We will save them and resend it to
 	// the chan networkMsgs once the block height has reached. The cached
 	// map format is,
-	//   {blockHeight: [msg1, msg2, ...], ...}
-	futureMsgs *lru.Cache[uint32, *cachedNetworkMsg]
+	//   {msgID1: msg1, msgID2: msg2, ...}
+	futureMsgs *futureMsgCache
 
 	// chanPolicyUpdates is a channel that requests to update the
 	// forwarding policy of a set of channels is sent over.
@@ -478,13 +483,11 @@ type AuthenticatedGossiper struct {
 // passed configuration parameters.
 func New(cfg Config, selfKeyDesc *keychain.KeyDescriptor) *AuthenticatedGossiper {
 	gossiper := &AuthenticatedGossiper{
-		selfKey:     selfKeyDesc.PubKey,
-		selfKeyLoc:  selfKeyDesc.KeyLocator,
-		cfg:         &cfg,
-		networkMsgs: make(chan *networkMsg),
-		futureMsgs: lru.NewCache[uint32, *cachedNetworkMsg](
-			maxPrematureUpdates,
-		),
+		selfKey:           selfKeyDesc.PubKey,
+		selfKeyLoc:        selfKeyDesc.KeyLocator,
+		cfg:               &cfg,
+		networkMsgs:       make(chan *networkMsg),
+		futureMsgs:        newFutureMsgCache(maxFutureMessages),
 		quit:              make(chan struct{}),
 		chanPolicyUpdates: make(chan *chanPolicyUpdateRequest),
 		prematureChannelUpdates: lru.NewCache[uint64, *cachedNetworkMsg]( //nolint: lll
@@ -632,33 +635,89 @@ func (d *AuthenticatedGossiper) syncBlockHeight() {
 	}
 }
 
+// futureMsgCache embeds a `lru.Cache` with a message counter that's served as
+// the unique ID when saving the message.
+type futureMsgCache struct {
+	*lru.Cache[uint64, *cachedFutureMsg]
+
+	// msgID is a monotonically increased integer.
+	msgID atomic.Uint64
+}
+
+// nextMsgID returns a unique message ID.
+func (f *futureMsgCache) nextMsgID() uint64 {
+	return f.msgID.Add(1)
+}
+
+// newFutureMsgCache creates a new future message cache with the underlying lru
+// cache being initialized with the specified capacity.
+func newFutureMsgCache(capacity uint64) *futureMsgCache {
+	// Create a new cache.
+	cache := lru.NewCache[uint64, *cachedFutureMsg](capacity)
+
+	return &futureMsgCache{
+		Cache: cache,
+	}
+}
+
+// cachedFutureMsg is a future message that's saved to the `futureMsgCache`.
+type cachedFutureMsg struct {
+	// msg is the network message.
+	msg *networkMsg
+
+	// height is the block height.
+	height uint32
+}
+
+// Size returns the size of the message.
+func (c *cachedFutureMsg) Size() (uint64, error) {
+	// Return a constant 1.
+	return 1, nil
+}
+
 // resendFutureMessages takes a block height, resends all the future messages
-// found at that height and deletes those messages found in the gossiper's
-// futureMsgs.
+// found below and equal to that height and deletes those messages found in the
+// gossiper's futureMsgs.
 func (d *AuthenticatedGossiper) resendFutureMessages(height uint32) {
-	result, err := d.futureMsgs.Get(height)
+	var (
+		// msgs are the target messages.
+		msgs []*networkMsg
 
-	// Return early if no messages found.
-	if err == cache.ErrElementNotFound {
-		return
+		// keys are the target messages' caching keys.
+		keys []uint64
+	)
+
+	// filterMsgs is the visitor used when iterating the future cache.
+	filterMsgs := func(k uint64, cmsg *cachedFutureMsg) bool {
+		if cmsg.height <= height {
+			msgs = append(msgs, cmsg.msg)
+			keys = append(keys, k)
+		}
+
+		return true
 	}
 
-	// The error must nil, we will log an error and exit.
-	if err != nil {
-		log.Errorf("Reading future messages got error: %v", err)
+	// Filter out the target messages.
+	d.futureMsgs.Range(filterMsgs)
+
+	// Return early if no messages found.
+	if len(msgs) == 0 {
 		return
 	}
 
-	msgs := result.msgs
+	// Remove the filtered messages.
+	for _, key := range keys {
+		d.futureMsgs.Delete(key)
+	}
 
 	log.Debugf("Resending %d network messages at height %d",
 		len(msgs), height)
 
-	for _, pMsg := range msgs {
+	for _, msg := range msgs {
 		select {
-		case d.networkMsgs <- pMsg.msg:
+		case d.networkMsgs <- msg:
 		case <-d.quit:
-			pMsg.msg.err <- ErrGossiperShuttingDown
+			msg.err <- ErrGossiperShuttingDown
 		}
 	}
 }
@@ -1875,23 +1934,11 @@ func (d *AuthenticatedGossiper) isPremature(chanID lnwire.ShortChannelID,
 		return false
 	}
 
-	// Add the premature message to our future messages which will
-	// be resent once the block height has reached.
+	// Add the premature message to our future messages which will be
+	// resent once the block height has reached.
 	//
-	// Init an empty cached message and overwrite it if there are cached
-	// messages found.
-	cachedMsgs := &cachedNetworkMsg{
-		msgs: make([]*processedNetworkMsg, 0),
-	}
-
-	result, err := d.futureMsgs.Get(msgHeight)
-	// No error returned means we have old messages cached.
-	if err == nil {
-		cachedMsgs = result
-	}
-
-	// Copy the networkMsgs since the old message's err chan will
-	// be consumed.
+	// Copy the networkMsgs since the old message's err chan will be
+	// consumed.
 	copied := &networkMsg{
 		peer:              msg.peer,
 		source:            msg.source,
@@ -1901,15 +1948,15 @@ func (d *AuthenticatedGossiper) isPremature(chanID lnwire.ShortChannelID,
 		err:               make(chan error, 1),
 	}
 
-	// The processed boolean is unused in the futureMsgs case.
-	pMsg := &processedNetworkMsg{msg: copied}
+	// Create the cached message.
+	cachedMsg := &cachedFutureMsg{
+		msg:    copied,
+		height: msgHeight,
+	}
 
-	// Add the network message.
-	msgs := cachedMsgs.msgs
-	msgs = append(msgs, pMsg)
-	_, err = d.futureMsgs.Put(msgHeight, &cachedNetworkMsg{
-		msgs: msgs,
-	})
+	// Increment the msg ID and add it to the cache.
+	nextMsgID := d.futureMsgs.nextMsgID()
+	_, err := d.futureMsgs.Put(nextMsgID, cachedMsg)
 	if err != nil {
 		log.Errorf("Adding future message got error: %v", err)
 	}
diff --git a/go.mod b/go.mod
index 353aa55e73..5569832eee 100644
--- a/go.mod
+++ b/go.mod
@@ -31,7 +31,7 @@ require (
 	github.com/jrick/logrotate v1.0.0
 	github.com/kkdai/bstream v1.0.0
 	github.com/lightninglabs/neutrino v0.15.0
-	github.com/lightninglabs/neutrino/cache v1.1.0
+	github.com/lightninglabs/neutrino/cache v1.1.1
 	github.com/lightninglabs/protobuf-hex-display v1.4.3-hex-display
 	github.com/lightningnetwork/lightning-onion v1.2.1-0.20221202012345-ca23184850a1
 	github.com/lightningnetwork/lnd/cert v1.2.1
diff --git a/go.sum b/go.sum
index e4700756ab..76d2ebd0c4 100644
--- a/go.sum
+++ b/go.sum
@@ -393,8 +393,8 @@ github.com/lightninglabs/gozmq v0.0.0-20191113021534-d20a764486bf h1:HZKvJUHlcXI
 github.com/lightninglabs/gozmq v0.0.0-20191113021534-d20a764486bf/go.mod h1:vxmQPeIQxPf6Jf9rM8R+B4rKBqLA2AjttNxkFBL2Plk=
 github.com/lightninglabs/neutrino v0.15.0 h1:yr3uz36fLAq8hyM0TRUVlef1TRNoWAqpmmNlVtKUDtI=
 github.com/lightninglabs/neutrino v0.15.0/go.mod h1:pmjwElN/091TErtSE9Vd5W4hpxoG2/+xlb+HoPm9Gug=
-github.com/lightninglabs/neutrino/cache v1.1.0 h1:szZIhVabiQIsGzJjhvo76sj05Au+zVotj2M34EquGME=
-github.com/lightninglabs/neutrino/cache v1.1.0/go.mod h1:XJNcgdOw1LQnanGjw8Vj44CvguYA25IMKjWFZczwZuo=
+github.com/lightninglabs/neutrino/cache v1.1.1 h1:TllWOSlkABhpgbWJfzsrdUaDH2fBy/54VSIB4vVqV8M=
+github.com/lightninglabs/neutrino/cache v1.1.1/go.mod h1:XJNcgdOw1LQnanGjw8Vj44CvguYA25IMKjWFZczwZuo=
 github.com/lightninglabs/protobuf-hex-display v1.4.3-hex-display h1:RZJ8H4ueU/aQ9pFtx5wqsuD3B/DezrewJeVwDKKYY8E=
 github.com/lightninglabs/protobuf-hex-display v1.4.3-hex-display/go.mod h1:2oKOBU042GKFHrdbgGiKax4xVrFiZu51lhacUZQ9MnE=
 github.com/lightningnetwork/lightning-onion v1.2.1-0.20221202012345-ca23184850a1 h1:Wm0g70gkcAu2pGpNZwfWPSVOY21j8IyYsNewwK4OkT4=


Test patch:
diff --git a/discovery/gossiper_test.go b/discovery/gossiper_test.go
index 37823469c4..cc2fdef05d 100644
--- a/discovery/gossiper_test.go
+++ b/discovery/gossiper_test.go
@@ -20,6 +20,7 @@ import (
 	"github.com/btcsuite/btcd/wire"
 	"github.com/davecgh/go-spew/spew"
 	"github.com/go-errors/errors"
+	"github.com/lightninglabs/neutrino/cache"
 	"github.com/lightningnetwork/lnd/batch"
 	"github.com/lightningnetwork/lnd/chainntnfs"
 	"github.com/lightningnetwork/lnd/channeldb"
@@ -4098,3 +4099,38 @@ func TestRejectCacheChannelAnn(t *testing.T) {
 		t.Fatal("did not process remote announcement")
 	}
 }
+
+// TestFutureMsgCacheEviction checks that when the cache's capacity is reached,
+// saving one more item will evict the oldest item.
+func TestFutureMsgCacheEviction(t *testing.T) {
+	t.Parallel()
+
+	// Create a future message cache with size 1.
+	c := newFutureMsgCache(1)
+
+	// Send two messages to the cache, which ends in the first message
+	// being evicted.
+	//
+	// Put the first item.
+	id := c.nextMsgID()
+	evicted, err := c.Put(id, &cachedFutureMsg{height: uint32(id)})
+	require.NoError(t, err)
+	require.False(t, evicted, "should not be evicted")
+
+	// Put the second item.
+	id = c.nextMsgID()
+	evicted, err = c.Put(id, &cachedFutureMsg{height: uint32(id)})
+	require.NoError(t, err)
+	require.True(t, evicted, "should be evicted")
+
+	// The first item should have been evicted.
+	//
+	// NOTE: msg ID starts at 1, not 0.
+	_, err = c.Get(1)
+	require.ErrorIs(t, err, cache.ErrElementNotFound)
+
+	// The second item should be found.
+	item, err := c.Get(2)
+	require.NoError(t, err)
+	require.EqualValues(t, 2, item.height, "should be the second item")
+}