From 74a9c6e8f06f13429a0fa8de6ac8d109e407e8ee Mon Sep 17 00:00:00 2001 From: Sandy Chen Date: Mon, 3 Aug 2026 11:48:07 +0900 Subject: [PATCH 1/3] Querier: prune bucket-scan finder per-tenant metadata for departed tenants on partial scan errors The bucket-scan blocks finder pruned its per-tenant maps (userMetas, userMetasLookup, userDeletionMarks) only when a scan completed with no errors, by replacing the maps wholesale. On the partial-error path the maps were merged via maps.Copy and entries were never deleted, so a single permanently-failing tenant meant no scan ever fully succeeded and entries for every tenant the process had ever seen were retained forever: unbounded memory growth, and GetBlocks kept resolving stale blocks for long-deleted tenants (failing the store-gateway consistency check instead of returning empty results). Since #7573 the same partial path already reconciled the metadata fetchers against the active set, so fetcher eviction and metadata pruning diverged. The fetcher eviction is now generalized into reconcileActiveUsers, which reconciles all per-tenant state (the three metadata maps, the fetchers, their Prometheus registries and on-disk caches) against the authoritative active set from ScanUsers in one place, on every scan regardless of per-tenant errors. Reconciling against the active set makes the partial-error path converge on what the fully-successful path already does: scan jobs are only ever scheduled for active tenants, so a wholesale replacement never contains entries for tenants in the deleting state either. Nothing is pruned when ScanUsers itself fails (no authoritative set) or when the context is cancelled during shutdown. Fixes #7728 Signed-off-by: Sandy Chen --- pkg/querier/blocks_finder_bucket_scan.go | 46 +++- pkg/querier/blocks_finder_bucket_scan_test.go | 255 +++++++++++++++++- 2 files changed, 277 insertions(+), 24 deletions(-) diff --git a/pkg/querier/blocks_finder_bucket_scan.go b/pkg/querier/blocks_finder_bucket_scan.go index aae57708108..ecc579faeeb 100644 --- a/pkg/querier/blocks_finder_bucket_scan.go +++ b/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() diff --git a/pkg/querier/blocks_finder_bucket_scan_test.go b/pkg/querier/blocks_finder_bucket_scan_test.go index 515e090bf0e..79328fca8c7 100644 --- a/pkg/querier/blocks_finder_bucket_scan_test.go +++ b/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) +} From 36d2077992b097a5dd654f8add8bf79c7d82b9d8 Mon Sep 17 00:00:00 2001 From: Sandy Chen Date: Mon, 3 Aug 2026 12:19:55 +0900 Subject: [PATCH 2/3] Add CHANGELOG entry for #7747 Signed-off-by: Sandy Chen --- CHANGELOG.md | 1 + 1 file changed, 1 insertion(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6cc35782015..3d882219b48 100644 --- a/CHANGELOG.md +++ b/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 From c9f2dd0a3dbe3f6a15d8c3fd356e21d47d8ee166 Mon Sep 17 00:00:00 2001 From: Sandy Chen Date: Mon, 3 Aug 2026 13:55:17 +0900 Subject: [PATCH 3/3] Retrigger CI (flaky TestExpandedPostingsCacheFuzz) Signed-off-by: Sandy Chen