fix(querier): stop query cache growing past its limit

Drop cached buckets overlapping a new one instead of trimming them: the trimmed remainder is out-of-window data that would pile up until the TTL. Delete before set so ristretto evicts on updates, not only on new keys.

pkg/cache/memorycache/provider.go

@@ -120,7 +120,7 @@ func (provider *provider) Set(ctx context.Context, orgID valuer.UUID, cacheKey s
 		span.SetAttributes(attribute.Bool("memory.cloneable", true))
 		span.SetAttributes(attribute.Int64("memory.cost", cost))
 		toCache := cloneable.Clone()
-		if ok := provider.cc.SetWithTTL(strings.Join([]string{orgID.StringValue(), cacheKey}, "::"), toCache, cost, ttl); !ok {
+		if ok := provider.setWithEviction(strings.Join([]string{orgID.StringValue(), cacheKey}, "::"), toCache, cost, ttl); !ok {
 			return errors.New(errors.TypeInternal, errors.CodeInternal, "error writing to cache")
 		}
 
@@ -137,7 +137,7 @@ func (provider *provider) Set(ctx context.Context, orgID valuer.UUID, cacheKey s
 	span.SetAttributes(attribute.Bool("memory.cloneable", false))
 	span.SetAttributes(attribute.Int64("memory.cost", cost))
 
-	if ok := provider.cc.SetWithTTL(strings.Join([]string{orgID.StringValue(), cacheKey}, "::"), toCache, cost, ttl); !ok {
+	if ok := provider.setWithEviction(strings.Join([]string{orgID.StringValue(), cacheKey}, "::"), toCache, cost, ttl); !ok {
 		return errors.New(errors.TypeInternal, errors.CodeInternal, "error writing to cache")
 	}
 
@@ -228,3 +228,10 @@ func (provider *provider) unmarshalBinary(ctx context.Context, dest cachetypes.C
 	defer span.End()
 	return dest.UnmarshalBinary(fromCache)
 }
+
+// setWithEviction writes the key as a new item: ristretto evicts only on
+// insert, so updating an existing key in place can grow the cache past MaxCost.
+func (provider *provider) setWithEviction(key string, value any, cost int64, ttl time.Duration) bool {
+	provider.cc.Del(key)
+	return provider.cc.SetWithTTL(key, value, cost, ttl)
+}

pkg/cache/memorycache/provider_test.go

@@ -260,3 +260,46 @@ func TestCloneableConcurrentSetGet(t *testing.T) {
 		assert.NotSame(t, cachedCloneable, cloneables[i])
 	}
 }
+
+func TestSetGrowingExistingKeysStaysWithinMaxCost(t *testing.T) {
+	testCases := []struct {
+		name     string
+		newValue func(key string, size int) cachetypes.Cacheable
+	}{
+		{
+			name: "Cloneable",
+			newValue: func(key string, size int) cachetypes.Cacheable {
+				return &LargeCloneable{Key: key, CostHint: int64(size)}
+			},
+		},
+		{
+			name: "Cacheable",
+			newValue: func(key string, size int) cachetypes.Cacheable {
+				return &CacheableB{Key: strings.Repeat(key, size/len(key))}
+			},
+		},
+	}
+
+	for _, testCase := range testCases {
+		t.Run(testCase.name, func(t *testing.T) {
+			var maxCost int64 = 1 << 20
+			c, err := New(context.Background(), factorytest.NewSettings(), cache.Config{Provider: "memory", Memory: cache.Memory{
+				NumCounters: 10 * 1000,
+				MaxCost:     maxCost,
+			}})
+			require.NoError(t, err)
+
+			orgID := valuer.GenerateUUID()
+			// 20 keys rewritten up to 100 KB each: 2 MB in total if updates never evict.
+			for round := 1; round <= 50; round++ {
+				for k := 0; k < 20; k++ {
+					key := fmt.Sprintf("key-%02d", k)
+					_ = c.Set(context.Background(), orgID, key, testCase.newValue(key, round*2000), time.Hour)
+				}
+			}
+
+			metrics := c.(*provider).cc.Metrics
+			assert.LessOrEqual(t, int64(metrics.CostAdded())-int64(metrics.CostEvicted()), maxCost)
+		})
+	}
+}

pkg/querier/bucket_cache.go

@@ -676,27 +676,22 @@ func (bc *bucketCache) resultToBuckets(ctx context.Context, result *qbtypes.Resu
 }
 
 // mergeAndDeduplicateBuckets combines and deduplicates bucket lists.
+// Existing buckets overlapping a fresh one are dropped; the fresh bucket holds the merged result for its range.
 func (bc *bucketCache) mergeAndDeduplicateBuckets(existing, fresh []*qbtypes.CachedBucket) []*qbtypes.CachedBucket {
-	// Create a map to deduplicate by time range
-	bucketMap := make(map[string]*qbtypes.CachedBucket)
+	result := make([]*qbtypes.CachedBucket, 0, len(existing)+len(fresh))
 
 	// Add existing buckets
 	for _, bucket := range existing {
-		key := fmt.Sprintf("%d-%d", bucket.StartMs, bucket.EndMs)
-		bucketMap[key] = bucket
+		overlaps := slices.ContainsFunc(fresh, func(f *qbtypes.CachedBucket) bool {
+			return bucket.StartMs < f.EndMs && f.StartMs < bucket.EndMs
+		})
+		if !overlaps {
+			result = append(result, bucket)
+		}
 	}
 
 	// Add/update with fresh buckets
-	for _, bucket := range fresh {
-		key := fmt.Sprintf("%d-%d", bucket.StartMs, bucket.EndMs)
-		bucketMap[key] = bucket
-	}
-
-	// Convert back to slice with pre-allocated capacity
-	result := make([]*qbtypes.CachedBucket, 0, len(bucketMap))
-	for _, bucket := range bucketMap {
-		result = append(result, bucket)
-	}
+	result = append(result, fresh...)
 
 	// Sort by start time
 	slices.SortFunc(result, func(a, b *qbtypes.CachedBucket) int {

pkg/querier/bucket_cache_test.go

@@ -1401,3 +1401,61 @@ func TestBucketCache_NoCache(t *testing.T) {
 	// The actual NoCache logic is implemented in querier.run(), not in bucket cache
 	// This test verifies that the cache works normally and NoCache bypasses it at a higher level
 }
+
+func TestBucketCache_Put_OverlappingBuckets(t *testing.T) {
+	testCases := []struct {
+		name            string
+		windows         []qbtypes.TimeRange
+		expectedBuckets []qbtypes.TimeRange
+	}{
+		{
+			name:            "SlidingWindows_KeepNewest",
+			windows:         []qbtypes.TimeRange{{From: 1000, To: 5000}, {From: 2000, To: 6000}, {From: 3000, To: 7000}},
+			expectedBuckets: []qbtypes.TimeRange{{From: 3000, To: 7000}},
+		},
+		{
+			name:            "DisjointWindows_KeepAll",
+			windows:         []qbtypes.TimeRange{{From: 1000, To: 3000}, {From: 5000, To: 7000}},
+			expectedBuckets: []qbtypes.TimeRange{{From: 1000, To: 3000}, {From: 5000, To: 7000}},
+		},
+		{
+			name:            "PartialOverlap_DropOnlyOverlapping",
+			windows:         []qbtypes.TimeRange{{From: 1000, To: 3000}, {From: 5000, To: 7000}, {From: 2000, To: 4000}},
+			expectedBuckets: []qbtypes.TimeRange{{From: 2000, To: 4000}, {From: 5000, To: 7000}},
+		},
+	}
+
+	for _, testCase := range testCases {
+		t.Run(testCase.name, func(t *testing.T) {
+			memCache := createTestCache(t)
+			bc := NewBucketCache(instrumentationtest.New().ToProviderSettings(), memCache, cacheTTL, defaultFluxInterval)
+			step := qbtypes.Step{Duration: time.Second}
+
+			for _, window := range testCase.windows {
+				query := &mockQuery{fingerprint: testCase.name, startMs: window.From, endMs: window.To}
+				bc.Put(context.Background(), valuer.UUID{}, query, step, &qbtypes.Result{
+					Type:  qbtypes.RequestTypeTimeSeries,
+					Value: createTestTimeSeries("A", window.From, window.To, 1000),
+				})
+			}
+
+			var data qbtypes.CachedData
+			require.NoError(t, memCache.Get(context.Background(), valuer.UUID{}, "v5:query:"+testCase.name, &data))
+			buckets := make([]qbtypes.TimeRange, 0, len(data.Buckets))
+			for _, bucket := range data.Buckets {
+				buckets = append(buckets, qbtypes.TimeRange{From: bucket.StartMs, To: bucket.EndMs})
+			}
+			assert.Equal(t, testCase.expectedBuckets, buckets)
+
+			last := testCase.windows[len(testCase.windows)-1]
+			cached, missing := bc.GetMissRanges(context.Background(), valuer.UUID{}, &mockQuery{fingerprint: testCase.name, startMs: last.From, endMs: last.To}, step)
+			assert.Empty(t, missing)
+			require.NotNil(t, cached)
+			tsData, ok := cached.Value.(*qbtypes.TimeSeriesData)
+			require.True(t, ok)
+			require.Len(t, tsData.Aggregations, 1)
+			require.Len(t, tsData.Aggregations[0].Series, 1)
+			assert.Len(t, tsData.Aggregations[0].Series[0].Values, int((last.To-last.From)/1000))
+		})
+	}
+}