[BUGFIX] Ingester: fix inflight query counter leak on resource rejection (#7539)

* Fix ingester inflight query counter leak

Signed-off-by: Sandy Chen <Yuxuan.Chen@morganstanley.com>

* Address review feedback on inflight query leak fix

- Add clarifying comment before the post-rejection QueryStream block,
  per inline review nit.
- Replace `i.resourceBasedLimiter = nil` with mutating the underlying
  mock monitor's heap utilization, so the same live limiter admits the
  second query. The regression test now exercises the limiter machinery
  on the retry instead of bypassing it.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Signed-off-by: Sandy Chen <Yuxuan.Chen@morganstanley.com>

---------

Signed-off-by: Sandy Chen <Yuxuan.Chen@morganstanley.com>
Signed-off-by: Friedrich Gonzalez <1517449+friedrichg@users.noreply.github.com>
Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Co-authored-by: Friedrich Gonzalez <1517449+friedrichg@users.noreply.github.com>

Author: 3 people

CHANGELOG.md

@@ -48,6 +48,7 @@
 * [BUGFIX] Ingester: Close TSDB when compaction fails during `createTSDB`, preventing resource leaks (file descriptors, mmap handles) that could lead to ingester instability. #7560
 * [BUGFIX] Tenant Federation: Fix result cache returning stale data after a new tenant is added when `-tenant-federation.regex-matcher-enabled=true`. The resolved tenant set is now hashed and included in the cache key so that any change to the matched tenant list automatically invalidates cached entries. Non-regex users are unaffected. #7562
 * [BUGFIX] Tenant Federation: Fix regex resolver clearing known users list when user scan fails. #7534
+* [BUGFIX] Ingester: Fix inflight query counter leak when resource-based query protection rejects a request. #7539
 * [BUGFIX] Ingester: Release the TSDB appender on every early-return path in `Push` (e.g. out-of-order label set) by deferring `Rollback`. Previously such requests leaked TSDB head series references, mmap'd chunks and pending state per request, causing the `cortex_ingester_tsdb_head_active_appenders` gauge to grow unbounded. #7528
 * [BUGFIX] Ring: Fix ring token conflict resolution only applied to updated instance and make constantly token conflict check during instance observe period.
 * [BUGFIX] Distributor: Fix a panic (`slice bounds out of range`) in the stream push path when the context deadline expires while the worker goroutine is still marshalling a `WriteRequest`. #7541

pkg/ingester/ingester.go

@@ -2578,15 +2578,15 @@ func (i *Ingester) trackInflightQueryRequest() (func(), error) {
 		}
 	}
 
-	i.maxInflightQueryRequests.Track(i.inflightQueryRequests.Inc())
-
 	if i.resourceBasedLimiter != nil {
 		if err := i.resourceBasedLimiter.AcceptNewRequest(); err != nil {
 			level.Warn(i.logger).Log("msg", "failed to accept request", "err", err)
 			return nil, limiter.ErrResourceLimitReached
 		}
 	}
 
+	i.maxInflightQueryRequests.Track(i.inflightQueryRequests.Inc())
+
 	return func() {
 		i.inflightQueryRequests.Dec()
 	}, nil

pkg/ingester/ingester_test.go

@@ -3533,7 +3533,9 @@ func Test_Ingester_Query_ResourceThresholdBreached(t *testing.T) {
 		{labels.FromStrings("__name__", "test_1", "route", "get_user", "status", "200"), 1, 100000},
 	}
 
-	i, err := prepareIngesterWithBlocksStorage(t, defaultIngesterTestConfig(t), prometheus.NewRegistry())
+	cfg := defaultIngesterTestConfig(t)
+	cfg.DefaultLimits.MaxInflightQueryRequests = 1
+	i, err := prepareIngesterWithBlocksStorage(t, cfg, prometheus.NewRegistry())
 	require.NoError(t, err)
 	require.NoError(t, services.StartAndAwaitRunning(context.Background(), i))
 	defer services.StopAndAwaitTerminated(context.Background(), i) //nolint:errcheck
@@ -3542,7 +3544,8 @@ func Test_Ingester_Query_ResourceThresholdBreached(t *testing.T) {
 		resource.CPU:  0.5,
 		resource.Heap: 0.5,
 	}
-	i.resourceBasedLimiter, err = limiter.NewResourceBasedLimiter(&mockResourceMonitor{cpu: 0.4, heap: 0.6}, limits, nil, "ingester")
+	monitor := &mockResourceMonitor{cpu: 0.4, heap: 0.6}
+	i.resourceBasedLimiter, err = limiter.NewResourceBasedLimiter(monitor, limits, nil, "ingester")
 	require.NoError(t, err)
 
 	// Wait until it's ACTIVE
@@ -3566,6 +3569,12 @@ func Test_Ingester_Query_ResourceThresholdBreached(t *testing.T) {
 
 	// Expected error from isRetryableError in blocks_store_queryable.go
 	require.ErrorIs(t, err, limiter.ErrResourceLimitReached)
+	require.Equal(t, int64(0), i.inflightQueryRequests.Load())
+
+	// Verify that a query not blocked by the limiter still succeeds after the rejected request.
+	monitor.heap = 0.4
+	s = &mockQueryStreamServer{ctx: ctx}
+	require.NoError(t, i.QueryStream(rreq, s))
 }
 
 func TestIngester_LabelValues_ShouldNotCreateTSDBIfDoesNotExists(t *testing.T) {