diff --git a/graph/db/channel_cache.go b/graph/db/channel_cache.go index b50bbf498..b65a5ab5e 100644 --- a/graph/db/channel_cache.go +++ b/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, + }) } diff --git a/graph/db/channel_cache_test.go b/graph/db/channel_cache_test.go index 27ff65432..6ecbed8f3 100644 --- a/graph/db/channel_cache_test.go +++ b/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) } diff --git a/graph/db/kv_store.go b/graph/db/kv_store.go index 22c87f0ee..dcb84d3f4 100644 --- a/graph/db/kv_store.go +++ b/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 } diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index 44e2fd75b..ce95c102f 100644 --- a/graph/db/sql_store.go +++ b/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()