Querier: prune bucket-scan finder per-tenant metadata for departed tenants on partial scan errors

CHANGELOG.md

@@ -84,6 +84,7 @@
 * [BUGFIX] Ring: Fix DynamoDB KV CAS not retrying on transactional conditional check failures. `TransactWriteItems` reports condition failures as `TransactionCanceledException` with a `ConditionalCheckFailed` cancellation reason, which was not recognized as retryable, so any concurrent ring update conflict (e.g. many ingesters joining during a rolling update) failed immediately instead of re-reading and retrying. `TransactionConflict` cancellation reasons are also treated as retryable. #7706
 * [BUGFIX] Distributor: Return HTTP 499 (Client Closed Request) instead of 500 when a remote-write or OTLP push is canceled by the client, so client-side cancellations are no longer counted as server-side errors. #7717
 * [BUGFIX] Querier: Fix gRPC `codes.Canceled` errors being mapped to HTTP 500 instead of 499 when a client cancels a query. #7738
+* [BUGFIX] Querier: Fix unbounded growth of the bucket-scan blocks finder's per-tenant metadata on the partial-error scan path (used when the bucket index is disabled). Metadata for departed tenants is now pruned even while other tenants' scans keep failing, so deleted tenants no longer serve stale block references or retain memory forever. #7747
 
 ## 1.21.1 2026-06-04
 

pkg/querier/blocks_finder_bucket_scan.go

@@ -271,15 +271,14 @@ 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
+	// Reconcile the cached metadata and fetchers 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), and is valid 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)
+		d.reconcileActiveUsers(userIDs)
 	}
 
 	return resErrs.Err()
@@ -436,17 +435,42 @@ func (d *BucketScanBlocksFinder) createMetaFetcher(userID string) (block.Metadat
 // 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) {
+// reconcileActiveUsers reconciles all per-tenant state against the set of currently active
+// tenants: the metadata maps (which otherwise grow unbounded on the partial-error scan path,
+// where the wholesale replacement of a fully-successful scan never runs) and the fetchers, their
+// Prometheus registries and on-disk meta caches (which would grow unbounded on every path).
+// Everything is deliberately reconciled against the same active set in one place, so the
+// individual pieces of per-tenant state cannot diverge again; if some state ever needs a
+// different liveness set, split this function deliberately rather than special-casing inside it.
+//
+// The maps are pruned in their own userMx critical section, after scanBucket's merge section has
+// released the lock: a concurrent GetBlocks may briefly observe a departed tenant between the
+// merge and the prune. That is indistinguishable from reading while a scan is still in progress,
+// and strictly better than the unbounded retention it replaces.
+func (d *BucketScanBlocksFinder) reconcileActiveUsers(activeUserIDs []string) {
 	active := make(map[string]struct{}, len(activeUserIDs))
 	for _, userID := range activeUserIDs {
 		active[userID] = struct{}{}
 	}
 
+	d.userMx.Lock()
+	for userID := range d.userMetas {
+		if _, ok := active[userID]; !ok {
+			delete(d.userMetas, userID)
+		}
+	}
+	for userID := range d.userMetasLookup {
+		if _, ok := active[userID]; !ok {
+			delete(d.userMetasLookup, userID)
+		}
+	}
+	for userID := range d.userDeletionMarks {
+		if _, ok := active[userID]; !ok {
+			delete(d.userDeletionMarks, userID)
+		}
+	}
+	d.userMx.Unlock()
+
 	// Evict the in-memory fetchers and their per-tenant Prometheus registries.
 	var evicted []string
 	d.fetchersMx.Lock()

pkg/querier/blocks_finder_bucket_scan_test.go

@@ -5,7 +5,9 @@ import (
 	"fmt"
 	"os"
 	"path/filepath"
+	"reflect"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
@@ -511,18 +513,19 @@ func TestBucketScanBlocksFinder_PeriodicScanEvictsOnlyInactiveUserFetchers(t *te
 // with an empty prefix) still succeeds.
 type failUserBucket struct {
 	objstore.InstrumentedBucket
-	failUser string
+	failUser      string
+	failScanUsers bool
 }
 
 func (b *failUserBucket) Iter(ctx context.Context, dir string, f func(string) error, options ...objstore.IterOption) error {
-	if strings.HasPrefix(dir, b.failUser) {
+	if (b.failScanUsers && dir == "") || (b.failUser != "" && 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) {
+	if (b.failScanUsers && dir == "") || (b.failUser != "" && strings.HasPrefix(dir, b.failUser)) {
 		return errors.New("injected listing failure")
 	}
 	return b.InstrumentedBucket.IterWithAttributes(ctx, dir, f, options...)
@@ -536,12 +539,12 @@ func (b *failUserBucket) ReaderWithExpectedErrs(objstore.IsOpFailureExpectedFunc
 	return b
 }
 
-func TestBucketScanBlocksFinder_PeriodicScanEvictsInactiveUserDespiteOtherTenantScanError(t *testing.T) {
+func TestBucketScanBlocksFinder_PeriodicScanReconcilesInactiveUserDespiteOtherTenantScanError(t *testing.T) {
 	t.Parallel()
 	ctx := context.Background()
 
 	bkt, _ := cortex_testutil.PrepareFilesystemBucket(t)
-	wrapped := &failUserBucket{InstrumentedBucket: bkt, failUser: "user-2"}
+	wrapped := &failUserBucket{InstrumentedBucket: bkt}
 
 	cfg := prepareBucketScanBlocksFinderConfig()
 	cfg.CacheDir = t.TempDir()
@@ -558,26 +561,57 @@ func TestBucketScanBlocksFinder_PeriodicScanEvictsInactiveUserDespiteOtherTenant
 		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)
+	// Both users initially scan successfully, so they have known-good metas and cached fetchers.
+	user1Block := cortex_testutil.MockStorageBlock(t, bkt, "user-1", 10, 20)
+	user2Block := cortex_testutil.MockStorageBlock(t, bkt, "user-2", 10, 20)
+	cortex_testutil.MockStorageDeletionMark(t, bkt, "user-1", user1Block)
+	user2Mark := bucketindex.BlockDeletionMarkFromThanosMarker(cortex_testutil.MockStorageDeletionMark(t, bkt, "user-2", user2Block))
 	require.NoError(t, services.StartAndAwaitRunning(ctx, s))
 
 	s.fetchersMx.Lock()
-	require.Equal(t, 1, len(s.fetchers))
+	require.Equal(t, 2, 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)
+	// Make user-2's own scan fail while it remains active, and delete user-1.
+	wrapped.failUser = "user-2"
 	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).
+	// context and make scanBucket skip reconciliation entirely (it 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.
+	// The departed user is removed from every metas map even though another user's scan failed.
+	s.userMx.RLock()
+	_, hasUser1Metas := s.userMetas["user-1"]
+	_, hasUser1Lookup := s.userMetasLookup["user-1"]
+	_, hasUser1DeletionMarks := s.userDeletionMarks["user-1"]
+	_, hasUser2Metas := s.userMetas["user-2"]
+	_, hasUser2Lookup := s.userMetasLookup["user-2"]
+	_, hasUser2DeletionMarks := s.userDeletionMarks["user-2"]
+	s.userMx.RUnlock()
+	assert.False(t, hasUser1Metas)
+	assert.False(t, hasUser1Lookup)
+	assert.False(t, hasUser1DeletionMarks)
+	assert.True(t, hasUser2Metas)
+	assert.True(t, hasUser2Lookup)
+	assert.True(t, hasUser2DeletionMarks)
+
+	// GetBlocks no longer serves stale blocks for the departed user.
+	blocks, deletionMarks, err := s.GetBlocks(ctx, "user-1", 0, 30, nil)
+	require.NoError(t, err)
+	assert.Equal(t, 0, len(blocks))
+	assert.Equal(t, 0, len(deletionMarks))
+
+	// The active user whose own scan failed keeps its previously-good metas and deletion marks.
+	blocks, deletionMarks, err = s.GetBlocks(ctx, "user-2", 0, 30, nil)
+	require.NoError(t, err)
+	require.Len(t, blocks, 1)
+	assert.Equal(t, user2Block.ULID, blocks[0].ID)
+	assert.Equal(t, map[ulid.ULID]*bucketindex.BlockDeletionMark{user2Block.ULID: user2Mark}, deletionMarks)
+
+	// Fetchers are reconciled against the same active set: user-1 is evicted and user-2 retained.
 	s.fetchersMx.Lock()
 	_, has1 := s.fetchers["user-1"]
 	_, has2 := s.fetchers["user-2"]
@@ -586,6 +620,53 @@ func TestBucketScanBlocksFinder_PeriodicScanEvictsInactiveUserDespiteOtherTenant
 	assert.True(t, has2)
 }
 
+func TestBucketScanBlocksFinder_PeriodicScanPreservesMetasWhenUserScanFails(t *testing.T) {
+	t.Parallel()
+	ctx := context.Background()
+
+	bkt, _ := cortex_testutil.PrepareFilesystemBucket(t)
+	wrapped := &failUserBucket{InstrumentedBucket: bkt}
+
+	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()))
+	})
+
+	block := cortex_testutil.MockStorageBlock(t, bkt, "user-1", 10, 20)
+	mark := bucketindex.BlockDeletionMarkFromThanosMarker(cortex_testutil.MockStorageDeletionMark(t, bkt, "user-1", block))
+	require.NoError(t, services.StartAndAwaitRunning(ctx, s))
+
+	// A ScanUsers failure returns before reconciliation and must not treat an unavailable listing
+	// as an authoritative empty active set.
+	wrapped.failScanUsers = true
+	require.Error(t, s.scanBucket(ctx))
+
+	s.userMx.RLock()
+	_, hasMetas := s.userMetas["user-1"]
+	_, hasLookup := s.userMetasLookup["user-1"]
+	_, hasDeletionMarks := s.userDeletionMarks["user-1"]
+	s.userMx.RUnlock()
+	assert.True(t, hasMetas)
+	assert.True(t, hasLookup)
+	assert.True(t, hasDeletionMarks)
+
+	blocks, deletionMarks, err := s.GetBlocks(ctx, "user-1", 0, 30, nil)
+	require.NoError(t, err)
+	require.Len(t, blocks, 1)
+	assert.Equal(t, block.ULID, blocks[0].ID)
+	assert.Equal(t, map[ulid.ULID]*bucketindex.BlockDeletionMark{block.ULID: mark}, deletionMarks)
+}
+
 // 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
@@ -752,3 +833,151 @@ func prepareBucketScanBlocksFinderConfig() BucketScanBlocksFinderConfig {
 		BlockDiscoveryStrategy:   string(cortex_tsdb.RecursiveDiscovery),
 	}
 }
+
+func TestBucketScanBlocksFinder_FullySuccessfulScanStillReplacesMapWholesale(t *testing.T) {
+	t.Parallel()
+	ctx := context.Background()
+	s, bkt, _, _ := prepareBucketScanBlocksFinder(t, prepareBucketScanBlocksFinderConfig())
+
+	cortex_testutil.MockStorageBlock(t, bkt, "user-1", 10, 20)
+	require.NoError(t, services.StartAndAwaitRunning(ctx, s))
+
+	// Inject a stale entry for a tenant with no bucket presence at all (neither active, deleting
+	// nor deleted), simulating a leftover from the past, and alias the maps: a fully successful
+	// scan must REPLACE the maps wholesale — which also reclaims the maps' internal bucket
+	// capacity, something merge+prune never does — not merely prune the stale entry out of the
+	// old maps. The pointer-identity assertion pins the replacement itself, so the partial-error
+	// reconciliation can never silently become the only cleanup mechanism.
+	s.userMx.Lock()
+	s.userMetas["ghost"] = bucketindex.Blocks{&bucketindex.Block{ID: ulid.MustNew(1, nil)}}
+	s.userMetasLookup["ghost"] = map[ulid.ULID]*bucketindex.Block{}
+	s.userDeletionMarks["ghost"] = map[ulid.ULID]*bucketindex.BlockDeletionMark{}
+	oldMetas := s.userMetas
+	s.userMx.Unlock()
+
+	require.NoError(t, s.scanBucket(ctx))
+
+	s.userMx.RLock()
+	replaced := reflect.ValueOf(s.userMetas).Pointer() != reflect.ValueOf(oldMetas).Pointer()
+	_, hasGhostMetas := s.userMetas["ghost"]
+	_, hasGhostLookup := s.userMetasLookup["ghost"]
+	_, hasGhostMarks := s.userDeletionMarks["ghost"]
+	s.userMx.RUnlock()
+	assert.True(t, replaced, "a fully successful scan must replace the metas maps wholesale, not merge into them")
+	assert.False(t, hasGhostMetas)
+	assert.False(t, hasGhostLookup)
+	assert.False(t, hasGhostMarks)
+}
+
+// TestBucketScanBlocksFinder_PruningDoesNotRaceWithConcurrentGetBlocks verifies the lock
+// discipline between the pruning writer in reconcileActiveUsers and concurrent GetBlocks readers
+// under -race. Note the race detector can only prove the absence of unsynchronized access; the
+// transient visibility of a departed tenant between the merge and prune critical sections is
+// documented at reconcileActiveUsers and is not what this test checks. The final assertion pins
+// that pruning actually happened.
+func TestBucketScanBlocksFinder_PruningDoesNotRaceWithConcurrentGetBlocks(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()))
+	})
+
+	cortex_testutil.MockStorageBlock(t, bkt, "user-1", 10, 20)
+	require.NoError(t, services.StartAndAwaitRunning(ctx, s))
+
+	require.NoError(t, bkt.Delete(ctx, "user-1"))
+	cortex_testutil.MockStorageBlock(t, bkt, "user-2", 10, 20)
+
+	// Hammer GetBlocks concurrently with a scan that prunes user-1 on the partial-error path, so
+	// the race detector can catch any locking gap between the pruning writer and readers.
+	stop := make(chan struct{})
+	var wg sync.WaitGroup
+	wg.Go(func() {
+		for {
+			select {
+			case <-stop:
+				return
+			default:
+				_, _, _ = s.GetBlocks(ctx, "user-1", 0, 30, nil)
+				_, _, _ = s.GetBlocks(ctx, "user-2", 0, 30, nil)
+			}
+		}
+	})
+
+	require.Error(t, s.scanBucket(ctx))
+	close(stop)
+	wg.Wait()
+
+	s.userMx.RLock()
+	_, hasUser1 := s.userMetas["user-1"]
+	s.userMx.RUnlock()
+	assert.False(t, hasUser1)
+}
+
+// TestBucketScanBlocksFinder_FetcherCreatedDuringFailingScanIsRetained preserves the original
+// #7573 regression scenario: a tenant whose scan errors from the moment it appears still gets a
+// metadata fetcher created during the (partially failing) scan, and reconciliation must retain
+// that fetcher because the tenant is active — while a departed tenant is still evicted by the
+// same reconciliation.
+func TestBucketScanBlocksFinder_FetcherCreatedDuringFailingScanIsRetained(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.
+	require.Error(t, s.scanBucket(ctx))
+
+	// Reconciliation is decoupled from per-tenant scan errors: the now-inactive user-1 is evicted
+	// while user-2's fetcher — created during the failing scan — 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)
+}