filter out blocks marked for deletion in the bucket index ids fetcher…

… too, so we are protected even when apply ignoreDeletionMarkerFilter the block has been deleted (cortexproject#5712)

pkg/compactor/compactor.go

@@ -821,14 +821,18 @@ func (c *Compactor) compactUser(ctx context.Context, userID string) error {
 	noCompactMarkerFilter := compact.NewGatherNoCompactionMarkFilter(ulogger, bucket, c.compactorCfg.MetaSyncConcurrency)
 
 	var blockIDsFetcher block.BlockIDsFetcher
+	var fetcherULogger log.Logger
 	if c.storageCfg.BucketStore.BucketIndex.Enabled {
-		blockIDsFetcher = bucketindex.NewBlockIDsFetcher(ulogger, c.bucketClient, userID, c.limits)
+		fetcherULogger = log.With(ulogger, "blockIdsFetcher", "BucketIndexBlockIDsFetcher")
+		blockIDsFetcher = bucketindex.NewBlockIDsFetcher(fetcherULogger, c.bucketClient, userID, c.limits)
+
 	} else {
-		blockIDsFetcher = block.NewBaseBlockIDsFetcher(ulogger, bucket)
+		fetcherULogger = log.With(ulogger, "blockIdsFetcher", "BaseBlockIDsFetcher")
+		blockIDsFetcher = block.NewBaseBlockIDsFetcher(fetcherULogger, bucket)
 	}
 
 	fetcher, err := block.NewMetaFetcher(
-		ulogger,
+		fetcherULogger,
 		c.compactorCfg.MetaSyncConcurrency,
 		bucket,
 		blockIDsFetcher,

pkg/storage/tsdb/bucketindex/block_ids_fetcher.go

@@ -60,8 +60,16 @@ func (f *BlockIDsFetcher) GetActiveAndPartialBlockIDs(ctx context.Context, ch ch
 		return nil, err
 	}
 
-	// Sent the active block ids
+	blocksMarkedForDeletion := idx.BlockDeletionMarks.GetULIDs()
+	blocksMarkedForDeletionMap := make(map[ulid.ULID]struct{})
+	for _, block := range blocksMarkedForDeletion {
+		blocksMarkedForDeletionMap[block] = struct{}{}
+	}
+	// Sent the ids of blocks not marked for deletion
 	for _, b := range idx.Blocks {
+		if _, ok := blocksMarkedForDeletionMap[b.ID]; ok {
+			continue
+		}
 		select {
 		case <-ctx.Done():
 			return nil, ctx.Err()

pkg/storage/tsdb/bucketindex/block_ids_fetcher_test.go

@@ -56,7 +56,7 @@ func TestBlockIDsFetcher_Fetch(t *testing.T) {
 	blockIdsFetcher.GetActiveAndPartialBlockIDs(ctx, ch)
 	close(ch)
 	wg.Wait()
-	require.Equal(t, []ulid.ULID{block1.ID, block2.ID, block3.ID}, blockIds)
+	require.Equal(t, []ulid.ULID{block3.ID}, blockIds)
 }
 
 func TestBlockIDsFetcherFetcher_Fetch_NoBucketIndex(t *testing.T) {