graph/db: version channelCache by gossip version

Add a channelCacheKey struct keyed by {GossipVersion, chanID}, matching
the pattern already used by rejectCache. This prevents v1 and v2
channel data from colliding in the shared cache.

All callers in KVStore (always GossipVersion1) and SQLStore (version
from context) are updated to pass the version parameter.

This will be needed for later on when we update methods that use this
cache to be versioned (like ChannelUpdatesInHorizon).

graph/db/channel_cache.go

@@ -1,11 +1,20 @@
 package graphdb
 
+import "github.com/lightningnetwork/lnd/lnwire"
+
+// channelCacheKey uniquely identifies a channel entry in the channel cache by
+// gossip version and channel ID.
+type channelCacheKey struct {
+	version lnwire.GossipVersion
+	chanID  uint64
+}
+
 // channelCache is an in-memory cache used to improve the performance of
 // ChanUpdatesInHorizon. It caches the chan info and edge policies for a
 // particular channel.
 type channelCache struct {
 	n        int
-	channels map[uint64]ChannelEdge
+	channels map[channelCacheKey]ChannelEdge
 }
 
 // newChannelCache creates a new channelCache with maximum capacity of n
@@ -13,13 +22,18 @@ type channelCache struct {
 func newChannelCache(n int) *channelCache {
 	return &channelCache{
 		n:        n,
-		channels: make(map[uint64]ChannelEdge),
+		channels: make(map[channelCacheKey]ChannelEdge),
 	}
 }
 
 // get returns the channel from the cache, if it exists.
-func (c *channelCache) get(chanid uint64) (ChannelEdge, bool) {
-	channel, ok := c.channels[chanid]
+func (c *channelCache) get(version lnwire.GossipVersion,
+	chanid uint64) (ChannelEdge, bool) {
+
+	channel, ok := c.channels[channelCacheKey{
+		version: version,
+		chanID:  chanid,
+	}]
 	return channel, ok
 }
 
@@ -27,10 +41,17 @@ func (c *channelCache) get(chanid uint64) (ChannelEdge, bool) {
 // exists, it will be replaced with the new entry. If the entry doesn't exist,
 // it will be inserted to the cache, performing a random eviction if the cache
 // is at capacity.
-func (c *channelCache) insert(chanid uint64, channel ChannelEdge) {
+func (c *channelCache) insert(version lnwire.GossipVersion, chanid uint64,
+	channel ChannelEdge) {
+
+	key := channelCacheKey{
+		version: version,
+		chanID:  chanid,
+	}
+
 	// If entry exists, replace it.
-	if _, ok := c.channels[chanid]; ok {
-		c.channels[chanid] = channel
+	if _, ok := c.channels[key]; ok {
+		c.channels[key] = channel
 		return
 	}
 
@@ -41,10 +62,13 @@ func (c *channelCache) insert(chanid uint64, channel ChannelEdge) {
 			break
 		}
 	}
-	c.channels[chanid] = channel
+	c.channels[key] = channel
 }
 
 // remove deletes an edge for chanid from the cache, if it exists.
-func (c *channelCache) remove(chanid uint64) {
-	delete(c.channels, chanid)
+func (c *channelCache) remove(version lnwire.GossipVersion, chanid uint64) {
+	delete(c.channels, channelCacheKey{
+		version: version,
+		chanID:  chanid,
+	})
 }

graph/db/channel_cache_test.go

@@ -6,6 +6,7 @@ import (
 
 	"github.com/btcsuite/btcd/chaincfg/chainhash"
 	"github.com/lightningnetwork/lnd/graph/db/models"
+	"github.com/lightningnetwork/lnd/lnwire"
 	"github.com/lightningnetwork/lnd/routing/route"
 )
 
@@ -14,37 +15,39 @@ import (
 func TestChannelCache(t *testing.T) {
 	const cacheSize = 100
 
+	v := lnwire.GossipVersion1
+
 	// Create a new channel cache with the configured max size.
 	c := newChannelCache(cacheSize)
 
 	// As a sanity check, assert that querying the empty cache does not
 	// return an entry.
-	_, ok := c.get(0)
+	_, ok := c.get(v, 0)
 	if ok {
 		t.Fatalf("channel cache should be empty")
 	}
 
 	// Now, fill up the cache entirely.
 	for i := uint64(0); i < cacheSize; i++ {
-		c.insert(i, channelForInt(i))
+		c.insert(v, i, channelForInt(i))
 	}
 
 	// Assert that the cache has all of the entries just inserted, since no
 	// eviction should occur until we try to surpass the max size.
-	assertHasChanEntries(t, c, 0, cacheSize)
+	assertHasChanEntries(t, c, v, 0, cacheSize)
 
 	// Now, insert a new element that causes the cache to evict an element.
-	c.insert(cacheSize, channelForInt(cacheSize))
+	c.insert(v, cacheSize, channelForInt(cacheSize))
 
 	// Assert that the cache has this last entry, as the cache should evict
 	// some prior element and not the newly inserted one.
-	assertHasChanEntries(t, c, cacheSize, cacheSize)
+	assertHasChanEntries(t, c, v, cacheSize, cacheSize)
 
 	// Iterate over all inserted elements and construct a set of the evicted
 	// elements.
 	evicted := make(map[uint64]struct{})
 	for i := uint64(0); i < cacheSize+1; i++ {
-		_, ok := c.get(i)
+		_, ok := c.get(v, i)
 		if !ok {
 			evicted[i] = struct{}{}
 		}
@@ -58,35 +61,37 @@ func TestChannelCache(t *testing.T) {
 
 	// Remove the highest item which initially caused the eviction and
 	// reinsert the element that was evicted prior.
-	c.remove(cacheSize)
+	c.remove(v, cacheSize)
 	for i := range evicted {
-		c.insert(i, channelForInt(i))
+		c.insert(v, i, channelForInt(i))
 	}
 
 	// Since the removal created an extra slot, the last insertion should
 	// not have caused an eviction and the entries for all channels in the
 	// original set that filled the cache should be present.
-	assertHasChanEntries(t, c, 0, cacheSize)
+	assertHasChanEntries(t, c, v, 0, cacheSize)
 
 	// Finally, reinsert the existing set back into the cache and test that
 	// the cache still has all the entries. If the randomized eviction were
 	// happening on inserts for existing cache items, we expect this to fail
 	// with high probability.
 	for i := uint64(0); i < cacheSize; i++ {
-		c.insert(i, channelForInt(i))
+		c.insert(v, i, channelForInt(i))
 	}
-	assertHasChanEntries(t, c, 0, cacheSize)
+	assertHasChanEntries(t, c, v, 0, cacheSize)
 
 }
 
 // assertHasEntries queries the edge cache for all channels in the range [start,
 // end), asserting that they exist and their value matches the entry produced by
 // entryForInt.
-func assertHasChanEntries(t *testing.T, c *channelCache, start, end uint64) {
+func assertHasChanEntries(t *testing.T, c *channelCache,
+	v lnwire.GossipVersion, start, end uint64) {
+
 	t.Helper()
 
 	for i := start; i < end; i++ {
-		entry, ok := c.get(i)
+		entry, ok := c.get(v, i)
 		if !ok {
 			t.Fatalf("channel cache should contain chan %d", i)
 		}

graph/db/kv_store.go

@@ -1186,7 +1186,9 @@ func (c *KVStore) AddChannelEdge(ctx context.Context,
 				c.rejectCache.remove(
 					lnwire.GossipVersion1, edge.ChannelID,
 				)
-				c.chanCache.remove(edge.ChannelID)
+				c.chanCache.remove(
+					lnwire.GossipVersion1, edge.ChannelID,
+				)
 
 				return nil
 			}
@@ -1588,7 +1590,7 @@ func (c *KVStore) PruneGraph(spentOutputs []*wire.OutPoint,
 
 	for _, channel := range chansClosed {
 		c.rejectCache.remove(lnwire.GossipVersion1, channel.ChannelID)
-		c.chanCache.remove(channel.ChannelID)
+		c.chanCache.remove(lnwire.GossipVersion1, channel.ChannelID)
 	}
 
 	return chansClosed, prunedNodes, nil
@@ -1855,7 +1857,7 @@ func (c *KVStore) DisconnectBlockAtHeight(height uint32) (
 
 	for _, channel := range removedChans {
 		c.rejectCache.remove(lnwire.GossipVersion1, channel.ChannelID)
-		c.chanCache.remove(channel.ChannelID)
+		c.chanCache.remove(lnwire.GossipVersion1, channel.ChannelID)
 	}
 
 	return removedChans, nil
@@ -1974,7 +1976,7 @@ func (c *KVStore) DeleteChannelEdges(v lnwire.GossipVersion,
 
 	for _, chanID := range chanIDs {
 		c.rejectCache.remove(lnwire.GossipVersion1, chanID)
-		c.chanCache.remove(chanID)
+		c.chanCache.remove(lnwire.GossipVersion1, chanID)
 	}
 
 	return infos, nil
@@ -2104,7 +2106,7 @@ func (c *KVStore) updateChanCacheBatch(edgesToCache map[uint64]ChannelEdge) {
 	defer c.cacheMu.Unlock()
 
 	for cid, edge := range edgesToCache {
-		c.chanCache.insert(cid, edge)
+		c.chanCache.insert(lnwire.GossipVersion1, cid, edge)
 	}
 }
 
@@ -2256,7 +2258,10 @@ func (c *KVStore) fetchNextChanUpdateBatch(
 			}
 
 			// Check cache (we already hold shared read lock).
-			if channel, ok := c.chanCache.get(chanIDInt); ok {
+			channel, ok := c.chanCache.get(
+				lnwire.GossipVersion1, chanIDInt,
+			)
+			if ok {
 				state.edgesSeen[chanIDInt] = struct{}{}
 
 				batch = append(batch, channel)
@@ -2300,7 +2305,7 @@ func (c *KVStore) fetchNextChanUpdateBatch(
 
 			// Now we have all the information we need to build the
 			// channel edge.
-			channel := ChannelEdge{
+			channel = ChannelEdge{
 				Info:    edgeInfo,
 				Policy1: edge1,
 				Policy2: edge2,
@@ -3302,13 +3307,16 @@ func (c *KVStore) updateEdgeCache(e *models.ChannelEdgePolicy,
 	// the entry with the updated policy for the direction that was just
 	// written. If the edge doesn't exist, we'll defer loading the info and
 	// policies and lazily read from disk during the next query.
-	if channel, ok := c.chanCache.get(e.ChannelID); ok {
+	channel, ok := c.chanCache.get(
+		lnwire.GossipVersion1, e.ChannelID,
+	)
+	if ok {
 		if isUpdate1 {
 			channel.Policy1 = e
 		} else {
 			channel.Policy2 = e
 		}
-		c.chanCache.insert(e.ChannelID, channel)
+		c.chanCache.insert(lnwire.GossipVersion1, e.ChannelID, channel)
 	}
 }
 
@@ -4215,7 +4223,7 @@ func (c *KVStore) MarkEdgeZombie(chanID uint64,
 	}
 
 	c.rejectCache.remove(lnwire.GossipVersion1, chanID)
-	c.chanCache.remove(chanID)
+	c.chanCache.remove(lnwire.GossipVersion1, chanID)
 
 	return nil
 }
@@ -4284,7 +4292,7 @@ func (c *KVStore) markEdgeLiveUnsafe(tx kvdb.RwTx, chanID uint64) error {
 	}
 
 	c.rejectCache.remove(lnwire.GossipVersion1, chanID)
-	c.chanCache.remove(chanID)
+	c.chanCache.remove(lnwire.GossipVersion1, chanID)
 
 	return nil
 }

graph/db/sql_store.go

@@ -750,7 +750,9 @@ func (s *SQLStore) AddChannelEdge(ctx context.Context,
 				s.rejectCache.remove(
 					edge.Version, edge.ChannelID,
 				)
-				s.chanCache.remove(edge.ChannelID)
+				s.chanCache.remove(
+					edge.Version, edge.ChannelID,
+				)
 
 				return nil
 			}
@@ -883,13 +885,13 @@ func (s *SQLStore) updateEdgeCache(e *models.ChannelEdgePolicy,
 	// the entry with the updated policy for the direction that was just
 	// written. If the edge doesn't exist, we'll defer loading the info and
 	// policies and lazily read from disk during the next query.
-	if channel, ok := s.chanCache.get(e.ChannelID); ok {
+	if channel, ok := s.chanCache.get(e.Version, e.ChannelID); ok {
 		if isUpdate1 {
 			channel.Policy1 = e
 		} else {
 			channel.Policy2 = e
 		}
-		s.chanCache.insert(e.ChannelID, channel)
+		s.chanCache.insert(e.Version, e.ChannelID, channel)
 	}
 }
 
@@ -1120,7 +1122,9 @@ func (s *SQLStore) buildChannelFromRow(ctx context.Context, db SQLQueries,
 
 // updateChanCacheBatch updates the channel cache with multiple edges at once.
 // This method acquires the cache lock only once for the entire batch.
-func (s *SQLStore) updateChanCacheBatch(edgesToCache map[uint64]ChannelEdge) {
+func (s *SQLStore) updateChanCacheBatch(v lnwire.GossipVersion,
+	edgesToCache map[uint64]ChannelEdge) {
+
 	if len(edgesToCache) == 0 {
 		return
 	}
@@ -1129,7 +1133,7 @@ func (s *SQLStore) updateChanCacheBatch(edgesToCache map[uint64]ChannelEdge) {
 	defer s.cacheMu.Unlock()
 
 	for chanID, edge := range edgesToCache {
-		s.chanCache.insert(chanID, edge)
+		s.chanCache.insert(v, chanID, edge)
 	}
 }
 
@@ -1232,6 +1236,7 @@ func (s *SQLStore) ChanUpdatesInHorizon(startTime, endTime time.Time,
 						// Check cache (we already hold
 						// shared read lock).
 						channel, ok := s.chanCache.get(
+							lnwire.GossipVersion1,
 							chanIDInt,
 						)
 						if ok {
@@ -1287,7 +1292,9 @@ func (s *SQLStore) ChanUpdatesInHorizon(startTime, endTime time.Time,
 
 			// Update cache after successful batch yield, setting
 			// the cache lock only once for the entire batch.
-			s.updateChanCacheBatch(edgesToCache)
+			s.updateChanCacheBatch(
+				lnwire.GossipVersion1, edgesToCache,
+			)
 			edgesToCache = make(map[uint64]ChannelEdge)
 
 			// If the batch didn't yield anything, then we're done.
@@ -1779,7 +1786,7 @@ func (s *SQLStore) MarkEdgeZombie(chanID uint64,
 	}
 
 	s.rejectCache.remove(lnwire.GossipVersion1, chanID)
-	s.chanCache.remove(chanID)
+	s.chanCache.remove(lnwire.GossipVersion1, chanID)
 
 	return nil
 }
@@ -1828,7 +1835,7 @@ func (s *SQLStore) MarkEdgeLive(chanID uint64) error {
 	}
 
 	s.rejectCache.remove(lnwire.GossipVersion1, chanID)
-	s.chanCache.remove(chanID)
+	s.chanCache.remove(lnwire.GossipVersion1, chanID)
 
 	return err
 }
@@ -2010,7 +2017,7 @@ func (s *SQLStore) DeleteChannelEdges(v lnwire.GossipVersion,
 
 	for _, chanID := range chanIDs {
 		s.rejectCache.remove(v, chanID)
-		s.chanCache.remove(chanID)
+		s.chanCache.remove(v, chanID)
 	}
 
 	return edges, nil
@@ -2901,7 +2908,7 @@ func (s *SQLStore) PruneGraph(spentOutputs []*wire.OutPoint,
 
 	for _, channel := range closedChans {
 		s.rejectCache.remove(channel.Version, channel.ChannelID)
-		s.chanCache.remove(channel.ChannelID)
+		s.chanCache.remove(channel.Version, channel.ChannelID)
 	}
 
 	return closedChans, prunedNodes, nil
@@ -3170,7 +3177,7 @@ func (s *SQLStore) DisconnectBlockAtHeight(height uint32) (
 	s.cacheMu.Lock()
 	for _, channel := range removedChans {
 		s.rejectCache.remove(channel.Version, channel.ChannelID)
-		s.chanCache.remove(channel.ChannelID)
+		s.chanCache.remove(channel.Version, channel.ChannelID)
 	}
 	s.cacheMu.Unlock()