fix(querier): evict per-tenant metadata fetchers for inactive tenants (#7573)

BucketScanBlocksFinder lazily creates one block.MetadataFetcher per
tenant (each with its own Prometheus registry and on-disk meta cache)
and caches it in d.fetchers. These entries were never removed, so for a
process that has seen N tenants over its lifetime the fetcher map, the
aggregated per-tenant Prometheus registries, and the on-disk meta caches
all grew unbounded even after tenants were deleted from storage.

After every scan, reconcile d.fetchers against the active-user set
returned by ScanUsers: evict the fetcher, unregister its per-tenant
registry via RemoveUserRegistry, and delete the fetcher's on-disk
meta-syncer cache for tenants that are no longer active. The active set
comes from a successful ScanUsers call, so reconciliation runs even when
individual per-tenant scans fail, keeping the leak bounded under tenant
churn; it is skipped only when the context is cancelled.

Only the fetcher's own "meta-syncer" cache sub-directory is removed (not
the whole CacheDir/<userID> tree) because in single-binary mode CacheDir
is the store-gateway's SyncDir, whose block data lives under the same
per-tenant directory and must not be deleted here.

Fixes #7572

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

Author: sandy2008

CHANGELOG.md

@@ -54,6 +54,7 @@
 * [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
 * [BUGFIX] Query Frontend: Fix native histogram responses not being handled correctly in `minTime()` sort ordering for split_by_interval merge. #7555
+* [BUGFIX] Querier: Fix unbounded resource leak in the bucket-scan blocks finder (used when the bucket index is disabled). Per-tenant metadata fetchers, their Prometheus registries, and on-disk meta caches are now evicted once a tenant is no longer active, instead of being retained for the lifetime of the process. #7573
 
 ## 1.21.0 2026-04-24
 

pkg/querier/blocks_finder_bucket_scan.go

@@ -3,6 +3,7 @@ package querier
 import (
 	"context"
 	"maps"
+	"os"
 	"path"
 	"path/filepath"
 	"slices"
@@ -267,6 +268,17 @@ pushJobsLoop:
 	}
 	d.userMx.Unlock()
 
+	// Reconcile the cached metadata fetchers (and their per-tenant Prometheus registries and
+	// on-disk meta caches) against the set of currently active tenants, so these resources stay
+	// bounded as tenants are deleted from storage. userIDs comes from a successful ScanUsers call
+	// (we return early above if it failed), so it is the authoritative active set regardless of
+	// any per-tenant scan errors collected in resErrs; we therefore reconcile even on the
+	// partial-error path, so the leak stays bounded under tenant churn. We only skip when the
+	// context has been cancelled (i.e. the service is shutting down).
+	if ctx.Err() == nil {
+		d.evictInactiveUserFetchers(userIDs)
+	}
+
 	return resErrs.Err()
 }
 
@@ -417,6 +429,61 @@ func (d *BucketScanBlocksFinder) createMetaFetcher(userID string) (block.Metadat
 	return f, userBucket, deletionMarkFilter, nil
 }
 
+// metaSyncerCacheDirName is the sub-directory, under the per-tenant cache directory, where
+// block.NewMetaFetcher stores its cached meta.json files (see createMetaFetcher).
+const metaSyncerCacheDirName = "meta-syncer"
+
+// evictInactiveUserFetchers reconciles the per-tenant metadata fetchers against the set of
+// currently active tenants. For every tenant that is no longer active it removes the cached
+// fetcher, unregisters its per-tenant Prometheus registry, and deletes the fetcher's on-disk meta
+// cache. Without this, d.fetchers, d.fetchersMetrics and the on-disk cache would grow unbounded
+// for the lifetime of the process as tenants are deleted from storage.
+func (d *BucketScanBlocksFinder) evictInactiveUserFetchers(activeUserIDs []string) {
+	active := make(map[string]struct{}, len(activeUserIDs))
+	for _, userID := range activeUserIDs {
+		active[userID] = struct{}{}
+	}
+
+	// Evict the in-memory fetchers and their per-tenant Prometheus registries.
+	var evicted []string
+	d.fetchersMx.Lock()
+	for userID := range d.fetchers {
+		if _, ok := active[userID]; ok {
+			continue
+		}
+
+		d.fetchersMetrics.RemoveUserRegistry(userID)
+		delete(d.fetchers, userID)
+		evicted = append(evicted, userID)
+	}
+	d.fetchersMx.Unlock()
+
+	if len(evicted) == 0 {
+		return
+	}
+
+	level.Info(d.logger).Log("msg", "evicted metadata fetchers for inactive tenants", "count", len(evicted))
+
+	// Delete each evicted fetcher's on-disk meta cache, outside the lock to keep disk I/O off it.
+	// We remove only the fetcher's own "meta-syncer" sub-directory, not the whole CacheDir/<userID>
+	// tree: in single-binary mode CacheDir is the store-gateway's SyncDir, whose block data also
+	// lives under CacheDir/<userID>/ and must not be deleted here. We key this off the fetchers
+	// this process evicted rather than sweeping CacheDir, so we never reach into a co-located
+	// store-gateway's cache; stale directories left by a previous process are reaped by the
+	// store-gateway's own cleanup (single-binary) and are otherwise a negligible disk residual.
+	for _, userID := range evicted {
+		metaCacheDir := filepath.Join(d.cfg.CacheDir, userID, metaSyncerCacheDirName)
+		if err := os.RemoveAll(metaCacheDir); err != nil {
+			level.Warn(d.logger).Log("msg", "failed to delete cached metadata fetcher directory for inactive user", "user", userID, "dir", metaCacheDir, "err", err)
+			continue
+		}
+
+		// Best-effort removal of the now-empty per-tenant directory. os.Remove only succeeds on an
+		// empty directory, so a co-located store-gateway's data under the same path is preserved.
+		_ = os.Remove(filepath.Join(d.cfg.CacheDir, userID))
+	}
+}
+
 func (d *BucketScanBlocksFinder) getBlockMeta(userID string, blockID ulid.ULID) *bucketindex.Block {
 	d.userMx.RLock()
 	defer d.userMx.RUnlock()

pkg/querier/blocks_finder_bucket_scan_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"os"
+	"path/filepath"
 	"strings"
 	"testing"
 	"time"
@@ -416,6 +417,204 @@ func TestBucketScanBlocksFinder_PeriodicScanFindsUserWhichWasPreviouslyDeleted(t
 	assert.Empty(t, deletionMarks)
 }
 
+func TestBucketScanBlocksFinder_PeriodicScanEvictsDeletedUserFetcher(t *testing.T) {
+	t.Parallel()
+	ctx := context.Background()
+	s, bucket, _, reg := prepareBucketScanBlocksFinder(t, prepareBucketScanBlocksFinderConfig())
+
+	cortex_testutil.MockStorageBlock(t, bucket, "user-1", 10, 20)
+	cortex_testutil.MockStorageBlock(t, bucket, "user-1", 20, 30)
+
+	require.NoError(t, services.StartAndAwaitRunning(ctx, s))
+
+	// The initial scan must have created and cached a per-tenant metadata fetcher, registered a
+	// per-tenant Prometheus registry, and created an on-disk meta cache for the active tenant.
+	s.fetchersMx.Lock()
+	require.Equal(t, 1, len(s.fetchers))
+	s.fetchersMx.Unlock()
+
+	userCacheDir := filepath.Join(s.cfg.CacheDir, "user-1")
+	metaSyncerDir := filepath.Join(userCacheDir, metaSyncerCacheDirName)
+	require.DirExists(t, metaSyncerDir)
+
+	// The per-tenant registry contributes to the aggregated cortex_blocks_meta_synced gauge.
+	syncedBefore, err := testutil.GatherAndCount(reg, "cortex_blocks_meta_synced")
+	require.NoError(t, err)
+	require.Greater(t, syncedBefore, 0)
+
+	// Delete the user from the bucket so it is no longer active.
+	require.NoError(t, bucket.Delete(ctx, "user-1"))
+
+	// Trigger a periodic scan.
+	require.NoError(t, s.scan(ctx))
+
+	// Once the user is no longer active, its cached fetcher, per-tenant Prometheus registry and
+	// on-disk meta cache must all be released, otherwise they leak for the lifetime of the process.
+	s.fetchersMx.Lock()
+	assert.Equal(t, 0, len(s.fetchers))
+	s.fetchersMx.Unlock()
+
+	// The fetcher's own meta-syncer cache must be removed; the now-empty parent dir is then
+	// removed on a best-effort basis.
+	assert.NoDirExists(t, metaSyncerDir)
+	assert.NoDirExists(t, userCacheDir)
+
+	// The per-tenant registry was unregistered, so its gauge series no longer appear.
+	syncedAfter, err := testutil.GatherAndCount(reg, "cortex_blocks_meta_synced")
+	require.NoError(t, err)
+	assert.Equal(t, 0, syncedAfter)
+}
+
+func TestBucketScanBlocksFinder_PeriodicScanEvictsOnlyInactiveUserFetchers(t *testing.T) {
+	t.Parallel()
+	ctx := context.Background()
+	s, bucket, _, _ := prepareBucketScanBlocksFinder(t, prepareBucketScanBlocksFinderConfig())
+
+	cortex_testutil.MockStorageBlock(t, bucket, "user-1", 10, 20)
+	cortex_testutil.MockStorageBlock(t, bucket, "user-2", 10, 20)
+	cortex_testutil.MockStorageBlock(t, bucket, "user-3", 10, 20)
+
+	require.NoError(t, services.StartAndAwaitRunning(ctx, s))
+
+	s.fetchersMx.Lock()
+	require.Equal(t, 3, len(s.fetchers))
+	s.fetchersMx.Unlock()
+
+	// Delete only user-2.
+	require.NoError(t, bucket.Delete(ctx, "user-2"))
+	require.NoError(t, s.scan(ctx))
+
+	// Only the inactive tenant's fetcher is evicted; the active tenants' fetchers are retained.
+	s.fetchersMx.Lock()
+	_, has1 := s.fetchers["user-1"]
+	_, has2 := s.fetchers["user-2"]
+	_, has3 := s.fetchers["user-3"]
+	require.Equal(t, 2, len(s.fetchers))
+	s.fetchersMx.Unlock()
+	assert.True(t, has1)
+	assert.False(t, has2)
+	assert.True(t, has3)
+
+	// A returning tenant gets its fetcher re-created on the next scan.
+	cortex_testutil.MockStorageBlock(t, bucket, "user-2", 20, 30)
+	require.NoError(t, s.scan(ctx))
+
+	s.fetchersMx.Lock()
+	_, has2 = s.fetchers["user-2"]
+	require.Equal(t, 3, len(s.fetchers))
+	s.fetchersMx.Unlock()
+	assert.True(t, has2)
+}
+
+// failUserBucket makes the per-tenant block listing fail for a single user, so that scanning that
+// user returns an error (resErrs != 0) while the top-level user listing used by ScanUsers (Iter
+// with an empty prefix) still succeeds.
+type failUserBucket struct {
+	objstore.InstrumentedBucket
+	failUser string
+}
+
+func (b *failUserBucket) Iter(ctx context.Context, dir string, f func(string) error, options ...objstore.IterOption) error {
+	if strings.HasPrefix(dir, b.failUser) {
+		return errors.New("injected listing failure")
+	}
+	return b.InstrumentedBucket.Iter(ctx, dir, f, options...)
+}
+
+func (b *failUserBucket) IterWithAttributes(ctx context.Context, dir string, f func(objstore.IterObjectAttributes) error, options ...objstore.IterOption) error {
+	if strings.HasPrefix(dir, b.failUser) {
+		return errors.New("injected listing failure")
+	}
+	return b.InstrumentedBucket.IterWithAttributes(ctx, dir, f, options...)
+}
+
+func (b *failUserBucket) WithExpectedErrs(objstore.IsOpFailureExpectedFunc) objstore.Bucket {
+	return b
+}
+
+func (b *failUserBucket) ReaderWithExpectedErrs(objstore.IsOpFailureExpectedFunc) objstore.BucketReader {
+	return b
+}
+
+func TestBucketScanBlocksFinder_PeriodicScanEvictsInactiveUserDespiteOtherTenantScanError(t *testing.T) {
+	t.Parallel()
+	ctx := context.Background()
+
+	bkt, _ := cortex_testutil.PrepareFilesystemBucket(t)
+	wrapped := &failUserBucket{InstrumentedBucket: bkt, failUser: "user-2"}
+
+	cfg := prepareBucketScanBlocksFinderConfig()
+	cfg.CacheDir = t.TempDir()
+	usersScanner, err := users.NewScanner(users.UsersScannerConfig{
+		Strategy:       users.UserScanStrategyList,
+		MaxStalePeriod: time.Hour,
+		CacheTTL:       0,
+	}, wrapped, log.NewNopLogger(), nil)
+	require.NoError(t, err)
+
+	s := NewBucketScanBlocksFinder(cfg, usersScanner, wrapped, nil, log.NewNopLogger(), nil)
+	t.Cleanup(func() {
+		s.StopAsync()
+		require.NoError(t, s.AwaitTerminated(context.Background()))
+	})
+
+	// Only user-1 is active at startup, so the initial scan succeeds and caches its fetcher.
+	cortex_testutil.MockStorageBlock(t, bkt, "user-1", 10, 20)
+	require.NoError(t, services.StartAndAwaitRunning(ctx, s))
+
+	s.fetchersMx.Lock()
+	require.Equal(t, 1, len(s.fetchers))
+	s.fetchersMx.Unlock()
+
+	// Introduce an active-but-erroring tenant and delete user-1.
+	cortex_testutil.MockStorageBlock(t, bkt, "user-2", 10, 20)
+	require.NoError(t, bkt.Delete(ctx, "user-1"))
+
+	// This scan collects a per-tenant error for user-2 (resErrs != 0). The failing tenant is
+	// retried with the finder's hardcoded backoff, so this scan takes a few seconds; that cost is
+	// intrinsic and must not be "optimised" with a short-deadline context, which would cancel the
+	// context and make scanBucket skip eviction entirely (eviction is gated on ctx.Err() == nil).
+	require.Error(t, s.scanBucket(ctx))
+
+	// ...but eviction is decoupled from per-tenant scan errors, so the now-inactive user-1 is still
+	// evicted while the erroring (but still active) user-2 is retained.
+	s.fetchersMx.Lock()
+	_, has1 := s.fetchers["user-1"]
+	_, has2 := s.fetchers["user-2"]
+	s.fetchersMx.Unlock()
+	assert.False(t, has1)
+	assert.True(t, has2)
+}
+
+// TestBucketScanBlocksFinder_PeriodicScanPreservesNonMetaSyncerDataOnEviction guards the
+// single-binary safety property: CacheDir is the store-gateway's SyncDir, whose block data lives
+// under the same per-tenant directory, so evicting a tenant must remove only the fetcher's own
+// meta-syncer cache and never the sibling block data.
+func TestBucketScanBlocksFinder_PeriodicScanPreservesNonMetaSyncerDataOnEviction(t *testing.T) {
+	t.Parallel()
+	ctx := context.Background()
+	s, bucket, _, _ := prepareBucketScanBlocksFinder(t, prepareBucketScanBlocksFinderConfig())
+
+	cortex_testutil.MockStorageBlock(t, bucket, "user-1", 10, 20)
+	require.NoError(t, services.StartAndAwaitRunning(ctx, s))
+
+	metaSyncerDir := filepath.Join(s.cfg.CacheDir, "user-1", metaSyncerCacheDirName)
+	require.DirExists(t, metaSyncerDir)
+
+	// Simulate a co-located store-gateway's block data under the same per-tenant directory.
+	sgBlockFile := filepath.Join(s.cfg.CacheDir, "user-1", "01DTVP434PA9VFXSW2JKB3392D", "index")
+	require.NoError(t, os.MkdirAll(filepath.Dir(sgBlockFile), 0o755))
+	require.NoError(t, os.WriteFile(sgBlockFile, []byte("block-data"), 0o644))
+
+	require.NoError(t, bucket.Delete(ctx, "user-1"))
+	require.NoError(t, s.scan(ctx))
+
+	// The fetcher's own meta-syncer cache is deleted...
+	assert.NoDirExists(t, metaSyncerDir)
+	// ...but the co-located store-gateway block data under the same per-tenant dir must survive.
+	assert.FileExists(t, sgBlockFile)
+}
+
 func TestBucketScanBlocksFinder_GetBlocks(t *testing.T) {
 	//parallel testing causes data race
 	ctx := context.Background()