From 8eb71421b3ef23c94cea6ad5f6bab30f8420f854 Mon Sep 17 00:00:00 2001 From: Dimitry Linkov Date: Fri, 10 Jul 2026 14:01:06 -0700 Subject: [PATCH] fix(querier): serve sharded windows whose shards are at mixed compaction levels pruneIncompleteShardedBlocks grouped blocks by {level, minTime} and dropped any group missing a shard. But the N shards of a window can legitimately sit at different compaction levels: when late-arriving source blocks re-merge only the shards their data touches (e.g. with stacktracePartition splitting, where a small increment hashes to a subset of shards), those shards advance a level while the rest lag. Keying completeness on level then split one intact window into several sub-groups, each missing shards, and dropped all of them - silently returning zero results for a fully-intact window. Check shard completeness per sharding (shard count) and per time instant instead, ignoring compaction level: - For each distinct window start, require every shard to have a trusted block covering that instant. A window whose shards span L3/L4 is complete; pruneSupersededBlocks then collapses the per-shard level duplicates. - Coverage (minTime <= t < maxTime), not an exact minTime match, so a shard merged to a wider span still counts for the later windows it overlaps and does not orphan sibling shards' blocks there (which would under-count). - Different shard counts (e.g. a split-and-merge-shards reconfiguration) are distinct shardings, checked independently. Two subtleties that would otherwise reintroduce a silent under-count: - Only blocks at/above the deduplication level count. Intermediate L2 split blocks are never served (pruneSupersededBlocks always drops them for the L1 ancestor), so letting one satisfy a shard would make an incomplete set look whole and under-count once the L2 block is dropped. - pruneIncompleteShardedBlocks still runs BEFORE pruneSupersededBlocks, so an incomplete sharded set is dropped while its lower-level ancestors still exist and can serve as a query-time fallback. If more than one sharding (shard count) survives into the plan, they partition the same series differently, so force query-time deduplication rather than serving both directly and double-counting. Also log a WARN when the guard prunes a genuinely incomplete window, so this can no longer silently hide data. Co-Authored-By: Claude Opus 4.8 --- pkg/querier/replication.go | 189 +++++++++++------- pkg/querier/replication_test.go | 334 ++++++++++++++++++++++++++++++++ 2 files changed, 453 insertions(+), 70 deletions(-) diff --git a/pkg/querier/replication.go b/pkg/querier/replication.go index ad77882560..41a7689d2d 100644 --- a/pkg/querier/replication.go +++ b/pkg/querier/replication.go @@ -11,6 +11,7 @@ import ( "github.com/go-kit/log/level" "github.com/grafana/dskit/ring" "github.com/grafana/dskit/tracing" + "github.com/prometheus/common/model" "github.com/samber/lo" "golang.org/x/sync/errgroup" @@ -202,92 +203,121 @@ func (r *replicasPerBlockID) removeBlock(ulid string) { delete(r.meta, ulid) } -// this step removes sharded blocks that don't have all the shards present for a time window -func (r *replicasPerBlockID) pruneIncompleteShardedBlocks() (bool, error) { - type compactionKey struct { - level int32 - minTime int64 +// hasShardedBlocks reports whether any block carries a compactor shard label. +func (r *replicasPerBlockID) hasShardedBlocks() bool { + for blockID := range r.m { + if meta, ok := r.meta[blockID]; ok { + if _, _, ok := shardFromBlock(meta); ok { + return true + } + } } - compactions := make(map[compactionKey][]string) + return false +} - // group blocks by compaction level +// pruneIncompleteShardedBlocks drops sharded blocks for any window instant that +// is missing a shard. It must run before pruneSupersededBlocks, so an incomplete +// set's lower-level ancestors survive as a fallback. deduplicationLevel is the +// level at/above which blocks are deduplicated (3 when sharded). +func (r *replicasPerBlockID) pruneIncompleteShardedBlocks(deduplicationLevel int32) error { + // Completeness is checked per sharding (shard count) and per time instant, not + // by compaction level: a window's shards can legitimately sit at different + // levels (a partial late re-merge advances only some), so keying on level would + // falsely split a complete window and prune it. Different shard counts (e.g. a + // shard-count reconfiguration) are separate shardings and checked independently. + // + // For each distinct window start we require every shard to have a trusted block + // covering that instant. Coverage (minTime <= t < maxTime), rather than an exact + // minTime match, keeps a wider block (a shard merged to a longer span) counted + // for the later windows it overlaps, so sibling shards' blocks there are not + // orphaned and silently pruned. Only blocks >= deduplicationLevel count; + // intermediate lower blocks are never served (pruneSupersededBlocks drops them) + // so must not satisfy a shard. + type shardedBlock struct { + id string + shard uint64 + minTime int64 + maxTime int64 + } + byShardCount := make(map[uint64][]shardedBlock) for blockID := range r.m { meta, ok := r.meta[blockID] if !ok { - return false, fmt.Errorf("meta missing for block id %s", blockID) + return fmt.Errorf("meta missing for block id %s", blockID) } - - key := compactionKey{ - level: 0, - minTime: meta.MinTime, + shard, shardCount, ok := shardFromBlock(meta) + if !ok { + continue } - - if meta.Compaction != nil { - key.level = meta.Compaction.Level + if meta.Compaction == nil || meta.Compaction.Level < deduplicationLevel { + continue } - compactions[key] = append(compactions[key], blockID) + byShardCount[shardCount] = append(byShardCount[shardCount], + shardedBlock{id: blockID, shard: shard, minTime: meta.MinTime, maxTime: meta.MaxTime}) } - // now we go through every group and check if we see at least a block for each shard - var ( - shardsSeen []bool - shardedBlocks []string - hasShardedBlocks bool - ) - for _, blocks := range compactions { - shardsSeen = shardsSeen[:0] - shardedBlocks = shardedBlocks[:0] - for _, block := range blocks { - meta, ok := r.meta[block] - if !ok { - return false, fmt.Errorf("meta missing for block id %s", block) + for shardCount, blocks := range byShardCount { + // Distinct window starts, ascending: pruning an earlier incomplete window + // first removes wide blocks that would otherwise mask a gap in a later one. + starts := make([]int64, 0, len(blocks)) + seen := make(map[int64]struct{}, len(blocks)) + for _, b := range blocks { + if _, ok := seen[b.minTime]; !ok { + seen[b.minTime] = struct{}{} + starts = append(starts, b.minTime) } + } + sort.Slice(starts, func(i, j int) bool { return starts[i] < starts[j] }) - shardIdx, shards, ok := shardFromBlock(meta) - if !ok { - // not a sharded block continue - continue + shardsSeen := make([]bool, shardCount) + for _, t := range starts { + for i := range shardsSeen { + shardsSeen[i] = false } - hasShardedBlocks = true - shardedBlocks = append(shardedBlocks, block) - - if len(shardsSeen) == 0 { - if cap(shardsSeen) < int(shards) { - shardsSeen = make([]bool, shards) - } else { - shardsSeen = shardsSeen[:shards] - for idx := range shardsSeen { - shardsSeen[idx] = false - } + for _, b := range blocks { + if _, live := r.m[b.id]; !live { + continue // already pruned by an earlier incomplete window + } + if b.shard < shardCount && b.minTime <= t && t < b.maxTime { + shardsSeen[b.shard] = true } } - - if len(shardsSeen) != int(shards) { - return false, fmt.Errorf("shard length mismatch, shards seen: %d, shards as per label: %d", len(shardsSeen), shards) + complete := true + for _, s := range shardsSeen { + if !s { + complete = false + break + } } - - shardsSeen[shardIdx] = true - } - // check if all shards are present - allShardsPresent := true - for _, shardSeen := range shardsSeen { - if !shardSeen { - allShardsPresent = false - break + if complete { + continue } - } - - if allShardsPresent { - continue - } - // now remove all blocks that are shareded but not complete - for _, block := range shardedBlocks { - r.removeBlock(block) + // A shard is missing at this instant; dropping these can hide data (there + // may be no lower-level fallback), so log rather than prune silently. We + // prune the blocks that start here and fall back to the lower levels. + var pruned int + for _, b := range blocks { + if b.minTime != t { + continue + } + if _, live := r.m[b.id]; live { + r.removeBlock(b.id) + pruned++ + } + } + if pruned > 0 { + level.Warn(r.logger).Log( + "msg", "pruning incomplete sharded blocks from query plan; a shard is missing for this time window and its data will not be queried", + "min_time", model.Time(t).Time().String(), + "shards_expected", shardCount, + "blocks_pruned", pruned, + ) + } } } - return hasShardedBlocks, nil + return nil } // prunes blocks that are contained by a higher compaction level block @@ -355,13 +385,10 @@ func (r *replicasPerBlockID) blockPlan(ctx context.Context) map[string]*blockPla hash = xxhash.New() plan = make(map[string]*blockPlanEntry) smallestCompactionLevel = int32(0) + shardCounts = make(map[uint64]struct{}) ) - sharded, err := r.pruneIncompleteShardedBlocks() - if err != nil { - level.Warn(r.logger).Log("msg", "block planning failed to prune incomplete sharded blocks", "err", err) - return nil - } + sharded := r.hasShardedBlocks() // Depending on whether split sharding is used, the compaction level at // which the data gets deduplicated differs: if split sharding is enabled, @@ -371,6 +398,13 @@ func (r *replicasPerBlockID) blockPlan(ctx context.Context) map[string]*blockPla deduplicationLevel = 3 } + // Prune incomplete sharded sets first, so that any lower-level ancestors + // survive as a fallback, then collapse the surviving blocks per shard. + if err := r.pruneIncompleteShardedBlocks(deduplicationLevel); err != nil { + level.Warn(r.logger).Log("msg", "block planning failed to prune incomplete sharded blocks", "err", err) + return nil + } + if err := r.pruneSupersededBlocks(sharded); err != nil { level.Warn(r.logger).Log("msg", "block planning failed to prune superseded blocks", "err", err) return nil @@ -393,6 +427,11 @@ func (r *replicasPerBlockID) blockPlan(ctx context.Context) map[string]*blockPla deduplicate = true } + // track the distinct shardings present in the plan (see below). + if _, shardCount, ok := shardFromBlock(meta); ok { + shardCounts[shardCount] = struct{}{} + } + // record the lowest compaction level if meta.Compaction != nil && (smallestCompactionLevel == 0 || meta.Compaction.Level < smallestCompactionLevel) { smallestCompactionLevel = meta.Compaction.Level @@ -442,6 +481,16 @@ func (r *replicasPerBlockID) blockPlan(ctx context.Context) map[string]*blockPla r.m[blockID] = []string{selectedReplica} } + // If more than one sharding scheme survived into the plan (e.g. an _of_2 and + // an _of_4 set for overlapping data), they partition the same series + // differently, so a series can appear in both. The sharded fast path assumes + // each series is served by exactly one block, which only holds for a single + // complete sharding - so force query-time deduplication to reconcile the + // overlap instead of double-counting. + if len(shardCounts) > 1 { + deduplicate = true + } + // adapt the plan to make sure all replicas will deduplicate if deduplicate { for _, hints := range plan { diff --git a/pkg/querier/replication_test.go b/pkg/querier/replication_test.go index c834b423e0..43c256a4e9 100644 --- a/pkg/querier/replication_test.go +++ b/pkg/querier/replication_test.go @@ -1,6 +1,7 @@ package querier import ( + "bytes" "context" "fmt" "sort" @@ -93,6 +94,16 @@ func validatePlanBlocksOnReplica(replica string, blocks ...string) validatorFunc } } +// validatePlanDeduplication asserts the deduplication hint on every planned replica. +func validatePlanDeduplication(expected bool) validatorFunc { + return func(t *testing.T, plan map[string]*blockPlanEntry) { + require.NotEmpty(t, plan, "expected a non-empty plan to assert deduplication on") + for replica, planEntry := range plan { + require.Equal(t, expected, planEntry.Deduplication, "unexpected deduplication hint for replica %s", replica) + } + } +} + func Test_replicasPerBlockID_blockPlan(t *testing.T) { for _, tc := range []struct { name string @@ -222,6 +233,284 @@ func Test_replicasPerBlockID_blockPlan(t *testing.T) { validatePlanBlocksOnReplica("ingester-0", "a"), }, }, + { + // A window whose shards sit at different levels (shard 2 at L3, the rest + // at L4, same minTime) is complete and must be queried, not pruned. + name: "keep sharded window with shards at mixed compaction levels", + inputs: func(r *replicasPerBlockID) { + t1, _ := time.Parse(time.RFC3339, "2021-01-01T00:00:00Z") + r.add([]ResponseFromReplica[[]*typesv1.BlockInfo]{ + { + addr: "store-gateway-0", + response: []*typesv1.BlockInfo{ + newBlockInfo("s0"). + withCompactionLevel(4). + withCompactorShard(0, 4). + withMinTime(t1, 2*time.Hour). + info(), + newBlockInfo("s1"). + withCompactionLevel(4). + withCompactorShard(1, 4). + withMinTime(t1, 2*time.Hour). + info(), + // shard 2 lagged behind at level 3 (no late data touched it). + newBlockInfo("s2"). + withCompactionLevel(3). + withCompactorShard(2, 4). + withMinTime(t1, time.Hour). // different maxTime, same minTime + info(), + newBlockInfo("s3"). + withCompactionLevel(4). + withCompactorShard(3, 4). + withMinTime(t1, 2*time.Hour). + info(), + }, + }, + }, storeGatewayInstance) + }, + validators: []validatorFunc{ + // Pre-fix, (level, minTime) grouping split this into incomplete L3 + // and L4 groups and emptied the plan. + validatePlanBlockIDs("s0", "s1", "s2", "s3"), + // Complete deduplicated set -> served without query-time dedup. + validatePlanDeduplication(false), + }, + }, + { + // Shard 0 has both its L3 and derived L4 present; shard 1 lags at L3. + // Window is complete, and superseding collapses shard 0 to just its L4. + name: "collapse L3+L4 duplicate of a shard within a mixed-level window", + inputs: func(r *replicasPerBlockID) { + t1, _ := time.Parse(time.RFC3339, "2021-01-01T00:00:00Z") + r.add([]ResponseFromReplica[[]*typesv1.BlockInfo]{ + { + addr: "store-gateway-0", + response: []*typesv1.BlockInfo{ + // shard 0: L4 derived from its L3 parent; both still present. + newBlockInfo("s0-l4"). + withCompactionLevel(4). + withCompactionSources("s0-l3"). + withCompactionParents("s0-l3"). + withCompactorShard(0, 2). + withMinTime(t1, 2*time.Hour). + info(), + newBlockInfo("s0-l3"). + withCompactionLevel(3). + withCompactorShard(0, 2). + withMinTime(t1, time.Hour). + info(), + // shard 1: lagged at L3, no L4 derived yet. + newBlockInfo("s1-l3"). + withCompactionLevel(3). + withCompactorShard(1, 2). + withMinTime(t1, time.Hour). + info(), + }, + }, + }, storeGatewayInstance) + }, + validators: []validatorFunc{ + // One block per shard: shard 0's L3 superseded by its L4. + validatePlanBlockIDs("s0-l4", "s1-l3"), + }, + }, + { + // Mixed spans: shard 0 merged to a wide [t1,t1+4h) block, while shard 1 + // still has two narrower blocks [t1,t1+2h) and [t1+2h,t1+4h). Completeness + // must credit the wide block for the later window it covers, so shard 1's + // [t1+2h,t1+4h) block is NOT orphaned and pruned. All three are kept. + name: "keep mixed-span shards when a wide block covers a later window", + inputs: func(r *replicasPerBlockID) { + t1, _ := time.Parse(time.RFC3339, "2021-01-01T00:00:00Z") + t3 := t1.Add(2 * time.Hour) + r.add([]ResponseFromReplica[[]*typesv1.BlockInfo]{ + { + addr: "store-gateway-0", + response: []*typesv1.BlockInfo{ + newBlockInfo("s0-wide").withCompactionLevel(4).withCompactorShard(0, 2).withMinTime(t1, 4*time.Hour).info(), + newBlockInfo("s1-a").withCompactionLevel(3).withCompactorShard(1, 2).withMinTime(t1, 2*time.Hour).info(), + newBlockInfo("s1-b").withCompactionLevel(3).withCompactorShard(1, 2).withMinTime(t3, 2*time.Hour).info(), + }, + }, + }, storeGatewayInstance) + }, + validators: []validatorFunc{ + // Pre-fix, the (t1+2h) group saw only shard 1 and pruned s1-b, losing + // shard 1's data for [t1+2h,t1+4h). + validatePlanBlockIDs("s0-wide", "s1-a", "s1-b"), + validatePlanDeduplication(false), + }, + }, + { + // A window with a genuinely missing shard (absent at every level) must + // still be pruned: shard 2 of 4 has no block at any level. + name: "prune sharded window with a shard missing at all levels", + inputs: func(r *replicasPerBlockID) { + t1, _ := time.Parse(time.RFC3339, "2021-01-01T00:00:00Z") + r.add([]ResponseFromReplica[[]*typesv1.BlockInfo]{ + { + addr: "ingester-0", + response: []*typesv1.BlockInfo{ + newBlockInfo("fallback").withMinTime(t1, 2*time.Hour).info(), + }, + }, + }, ingesterInstance) + r.add([]ResponseFromReplica[[]*typesv1.BlockInfo]{ + { + addr: "store-gateway-0", + response: []*typesv1.BlockInfo{ + newBlockInfo("s0"). + withCompactionLevel(4). + withCompactorShard(0, 4). + withMinTime(t1, 2*time.Hour). + info(), + newBlockInfo("s1"). + withCompactionLevel(3). + withCompactorShard(1, 4). + withMinTime(t1, time.Hour). + info(), + // shard 2 is missing entirely; shard 3 too. + newBlockInfo("s3"). + withCompactionLevel(4). + withCompactorShard(3, 4). + withMinTime(t1, 2*time.Hour). + info(), + }, + }, + }, storeGatewayInstance) + }, + validators: []validatorFunc{ + // Incomplete across all levels -> sharded blocks pruned, fallback kept. + validatePlanBlockIDs("fallback"), + // Low-level fallback -> query-time dedup enabled. + validatePlanDeduplication(true), + }, + }, + { + // Shard-count change (2 -> 4): a complete _of_2 and a partial _of_4 + // scheme share a level and minTime, so only shardCount separates them. + // The _of_4 is pruned, the complete _of_2 served. Pre-fix, pooling by + // {level, minTime} tripped the shard-length-mismatch guard and emptied + // the plan. + name: "serve complete old sharding when a shard-count change leaves a partial new one", + inputs: func(r *replicasPerBlockID) { + t1, _ := time.Parse(time.RFC3339, "2021-01-01T00:00:00Z") + r.add([]ResponseFromReplica[[]*typesv1.BlockInfo]{ + { + addr: "store-gateway-0", + response: []*typesv1.BlockInfo{ + // complete old 2-shard scheme + newBlockInfo("old0").withCompactionLevel(4).withCompactorShard(0, 2).withMinTime(t1, 2*time.Hour).info(), + newBlockInfo("old1").withCompactionLevel(4).withCompactorShard(1, 2).withMinTime(t1, 2*time.Hour).info(), + // partial new 4-shard scheme (shards 0 and 3 not yet produced) + newBlockInfo("new1").withCompactionLevel(4).withCompactorShard(1, 4).withMinTime(t1, 2*time.Hour).info(), + newBlockInfo("new2").withCompactionLevel(4).withCompactorShard(2, 4).withMinTime(t1, 2*time.Hour).info(), + }, + }, + }, storeGatewayInstance) + }, + validators: []validatorFunc{ + validatePlanBlockIDs("old0", "old1"), + validatePlanDeduplication(false), + }, + }, + { + // Two complete, unrelated shardings (_of_2 and _of_4) for the same + // window: each is complete on its own so neither is pruned, and neither + // supersedes the other. They partition the same series differently, so a + // series can appear in both - query-time deduplication must be forced to + // avoid double-counting. + name: "deduplicate when two complete shardings overlap", + inputs: func(r *replicasPerBlockID) { + t1, _ := time.Parse(time.RFC3339, "2021-01-01T00:00:00Z") + r.add([]ResponseFromReplica[[]*typesv1.BlockInfo]{ + { + addr: "store-gateway-0", + response: []*typesv1.BlockInfo{ + newBlockInfo("of2-0").withCompactionLevel(4).withCompactorShard(0, 2).withMinTime(t1, 2*time.Hour).info(), + newBlockInfo("of2-1").withCompactionLevel(4).withCompactorShard(1, 2).withMinTime(t1, 2*time.Hour).info(), + newBlockInfo("of4-0").withCompactionLevel(4).withCompactorShard(0, 4).withMinTime(t1, 2*time.Hour).info(), + newBlockInfo("of4-1").withCompactionLevel(4).withCompactorShard(1, 4).withMinTime(t1, 2*time.Hour).info(), + newBlockInfo("of4-2").withCompactionLevel(4).withCompactorShard(2, 4).withMinTime(t1, 2*time.Hour).info(), + newBlockInfo("of4-3").withCompactionLevel(4).withCompactorShard(3, 4).withMinTime(t1, 2*time.Hour).info(), + }, + }, + }, storeGatewayInstance) + }, + validators: []validatorFunc{ + // Both complete shardings survive; nothing is dropped. + validatePlanBlockIDs("of2-0", "of2-1", "of4-0", "of4-1", "of4-2", "of4-3"), + // Overlapping schemes -> query-time dedup forced. + validatePlanDeduplication(true), + }, + }, + { + // Counterpart to the previous case: when the new _of_4 sharding is + // derived from the old _of_2 one (lists it as sources), pruneSupersededBlocks + // removes the superseded _of_2 blocks, leaving a single sharding. The + // multi-sharding dedup guard must NOT fire here - the fast path + // (Deduplication=false) is preserved. + name: "single sharding after superseding keeps the fast path", + inputs: func(r *replicasPerBlockID) { + t1, _ := time.Parse(time.RFC3339, "2021-01-01T00:00:00Z") + r.add([]ResponseFromReplica[[]*typesv1.BlockInfo]{ + { + addr: "store-gateway-0", + response: []*typesv1.BlockInfo{ + newBlockInfo("of2-0").withCompactionLevel(3).withCompactorShard(0, 2).withMinTime(t1, 2*time.Hour).info(), + newBlockInfo("of2-1").withCompactionLevel(3).withCompactorShard(1, 2).withMinTime(t1, 2*time.Hour).info(), + // _of_4, one level up, derived from the _of_2 blocks. + newBlockInfo("of4-0").withCompactionLevel(4).withCompactionSources("of2-0", "of2-1").withCompactorShard(0, 4).withMinTime(t1, 2*time.Hour).info(), + newBlockInfo("of4-1").withCompactionLevel(4).withCompactionSources("of2-0", "of2-1").withCompactorShard(1, 4).withMinTime(t1, 2*time.Hour).info(), + newBlockInfo("of4-2").withCompactionLevel(4).withCompactionSources("of2-0", "of2-1").withCompactorShard(2, 4).withMinTime(t1, 2*time.Hour).info(), + newBlockInfo("of4-3").withCompactionLevel(4).withCompactionSources("of2-0", "of2-1").withCompactorShard(3, 4).withMinTime(t1, 2*time.Hour).info(), + }, + }, + }, storeGatewayInstance) + }, + validators: []validatorFunc{ + // _of_2 superseded by the derived _of_4; only one sharding remains. + validatePlanBlockIDs("of4-0", "of4-1", "of4-2", "of4-3"), + validatePlanDeduplication(false), + }, + }, + { + // Mid-merge: shard 0 at L3, shard 1 only in an intermediate L2 block, no + // L1. The L2 block gets dropped by superseding, so it must not count shard + // 1 as present - otherwise we'd serve shard 0 alone (half the data). The + // set is incomplete -> everything pruned (transient empty, not undercount). + name: "do not let an intermediate L2 block satisfy shard completeness", + inputs: func(r *replicasPerBlockID) { + t1, _ := time.Parse(time.RFC3339, "2021-01-01T00:00:00Z") + r.add([]ResponseFromReplica[[]*typesv1.BlockInfo]{ + { + addr: "store-gateway-0", + response: []*typesv1.BlockInfo{ + newBlockInfo("s0-l3"). + withCompactionLevel(3). + withCompactionSources("s0-l2"). + withCompactorShard(0, 2). + withMinTime(t1, time.Hour). + info(), + newBlockInfo("s0-l2"). + withCompactionLevel(2). + withCompactorShard(0, 2). + withMinTime(t1, time.Hour). + info(), + // shard 1 only exists as an intermediate L2 block. + newBlockInfo("s1-l2"). + withCompactionLevel(2). + withCompactorShard(1, 2). + withMinTime(t1, time.Hour). + info(), + }, + }, + }, storeGatewayInstance) + }, + validators: []validatorFunc{ + validatePlanBlockIDs(), + }, + }, { // Using a split-and-merge compactor, deduplication happens at level 3, // level 2 is intermediate step, where series distributed among shards @@ -283,3 +572,48 @@ func Test_replicasPerBlockID_blockPlan(t *testing.T) { }) } } + +// Pruning an incomplete window must emit a WARN (it was previously silent); a +// healthy mixed-level window must not. +func Test_pruneIncompleteShardedBlocks_logging(t *testing.T) { + t1, _ := time.Parse(time.RFC3339, "2021-01-01T00:00:00Z") + + const warnMsg = "a shard is missing for this time window" + + t.Run("warns when pruning an incomplete sharded window", func(t *testing.T) { + var buf bytes.Buffer + r := newReplicasPerBlockID(log.NewLogfmtLogger(&buf)) + // shard 0 of 2 present, shard 1 missing at every level, no fallback. + r.add([]ResponseFromReplica[[]*typesv1.BlockInfo]{ + { + addr: "store-gateway-0", + response: []*typesv1.BlockInfo{ + newBlockInfo("s0").withCompactionLevel(3).withCompactorShard(0, 2).withMinTime(t1, 2*time.Hour).info(), + }, + }, + }, storeGatewayInstance) + + plan := r.blockPlan(context.TODO()) + require.Empty(t, plan) + require.Contains(t, buf.String(), warnMsg, "expected a WARN when an incomplete window is pruned") + }) + + t.Run("does not warn for a complete mixed-level window", func(t *testing.T) { + var buf bytes.Buffer + r := newReplicasPerBlockID(log.NewLogfmtLogger(&buf)) + // both shards present, at different levels. + r.add([]ResponseFromReplica[[]*typesv1.BlockInfo]{ + { + addr: "store-gateway-0", + response: []*typesv1.BlockInfo{ + newBlockInfo("s0").withCompactionLevel(4).withCompactorShard(0, 2).withMinTime(t1, 2*time.Hour).info(), + newBlockInfo("s1").withCompactionLevel(3).withCompactorShard(1, 2).withMinTime(t1, time.Hour).info(), + }, + }, + }, storeGatewayInstance) + + plan := r.blockPlan(context.TODO()) + require.NotEmpty(t, plan) + require.NotContains(t, buf.String(), warnMsg, "must not warn for a complete window") + }) +}