fix(blooms): Fix eviction of multiple blockcache items (#13573)

This commit fixes a bug in the eviction of least recently used items. When an element from the doubly-linked-list of the cache is removed, then the element's prev and next items become nil. Therefore the call elem.Prev() always broke the loop's condition for checking entries to be evicted.

Signed-off-by: Christian Haudum <[email protected]>

pkg/storage/stores/shipper/bloomshipper/blockscache.go

@@ -272,6 +272,10 @@ func (c *BlocksCache) put(key string, value BlockDirectory) (*Entry, error) {
 
 func (c *BlocksCache) evict(key string, element *list.Element, reason string) {
 	entry := element.Value.(*Entry)
+	if key != entry.Key {
+		level.Error(c.logger).Log("msg", "failed to remove entry: entry key and map key do not match", "map_key", key, "entry_key", entry.Key)
+		return
+	}
 	err := c.remove(entry)
 	if err != nil {
 		level.Error(c.logger).Log("msg", "failed to remove entry from disk", "err", err)
@@ -400,6 +404,7 @@ func (c *BlocksCache) evictLeastRecentlyUsedItems() {
 	)
 	elem := c.lru.Back()
 	for c.currSizeBytes >= int64(c.cfg.SoftLimit) && elem != nil {
+		nextElem := elem.Prev()
 		entry := elem.Value.(*Entry)
 		if entry.refCount.Load() == 0 {
 			level.Debug(c.logger).Log(
@@ -408,7 +413,7 @@ func (c *BlocksCache) evictLeastRecentlyUsedItems() {
 			)
 			c.evict(entry.Key, elem, reasonFull)
 		}
-		elem = elem.Prev()
+		elem = nextElem
 	}
 }
 

pkg/storage/stores/shipper/bloomshipper/blockscache_test.go

@@ -197,8 +197,8 @@ func TestBlocksCache_TTLEviction(t *testing.T) {
 func TestBlocksCache_LRUEviction(t *testing.T) {
 	cfg := config.BlocksCacheConfig{
 		TTL:       time.Hour,
-		SoftLimit: flagext.Bytes(15),
-		HardLimit: flagext.Bytes(20),
+		SoftLimit: flagext.Bytes(25),
+		HardLimit: flagext.Bytes(50),
 		// no need for TTL evictions
 		PurgeInterval: time.Minute,
 	}
@@ -216,27 +216,32 @@ func TestBlocksCache_LRUEviction(t *testing.T) {
 	// oldest without refcount - will be evicted
 	err = cache.Put(ctx, "c", CacheValue("c", 4))
 	require.NoError(t, err)
+	err = cache.Put(ctx, "d", CacheValue("d", 4))
+	require.NoError(t, err)
+	err = cache.Put(ctx, "e", CacheValue("e", 4))
+	require.NoError(t, err)
+	err = cache.Put(ctx, "f", CacheValue("f", 4))
+	require.NoError(t, err)
 
 	// increase ref counter on "b"
 	_, found := cache.Get(ctx, "b")
 	require.True(t, found)
 
 	// exceed soft limit
-	err = cache.Put(ctx, "d", CacheValue("d", 4))
+	err = cache.Put(ctx, "g", CacheValue("g", 16))
 	require.NoError(t, err)
 
 	time.Sleep(time.Second)
 
 	l, ok := cache.len()
 	require.True(t, ok)
-	require.Equal(t, 3, l)
 
-	// key "b" was evicted because it was the oldest
-	// and it had no ref counts
-	_, found = cache.Get(ctx, "c")
-	require.False(t, found)
-
-	require.Equal(t, int64(12), cache.currSizeBytes)
+	// expect 3 items in cache:
+	// * item a with size 4
+	// * item b with size 4
+	// * item g with size 16
+	require.Equal(t, 3, l)
+	require.Equal(t, int64(24), cache.currSizeBytes)
 }
 
 func TestBlocksCache_RefCounter(t *testing.T) {