Compactor: delete orphaned global block deletion marks

CHANGELOG.md

@@ -14,7 +14,7 @@
 * [BUGFIX] Ruler: Stop retrying staleness markers that can never be accepted. A rule moved between rule groups left the original group re-appending staleness markers for its series on every evaluation until the ruler restarted. #7888
 * [BUGFIX] Store Gateway: Fix the Parquet bucket store ignoring query sharding, which made every shard return all series. #7892
 * [BUGFIX] Querier: Fix the querier crashing instead of reporting the error when a blocks store set subservice fails. `blocksStoreReplicationSet` never initialised its `FailureWatcher`, so the failure callback dereferenced a nil pointer on the goroutine the service manager runs listeners on, and the resulting panic's stack trace buried the error logged for the actual cause. #7896
-
+* [BUGFIX] Compactor: Delete block deletion markers left in the global `markers/` location of a tenant after the block they refer to was removed. They were skipped at every bucket index update and kept the tenant location in the bucket forever. #7890
 
 ## 1.22.0 in progress
 * [CHANGE] Ruler: Remove the deprecated `-ruler.evaluation-delay-duration` flag and its `ruler_evaluation_delay_duration` per-tenant limit. Use `-ruler.query-offset` / `ruler_query_offset`, which no longer takes the higher of the two values. Cortex decodes the runtime config strictly, so a leftover `ruler_evaluation_delay_duration` override makes the runtime config fail to load: Cortex **exits at startup** (`module failed`, `module=runtime-config`), and on an already-running process every reload fails, pinning the last good overrides and dropping `cortex_runtime_config_last_reload_successful` to 0. Run `grep -r ruler_evaluation_delay_duration` over your runtime configs before upgrading. #7792

pkg/storage/tsdb/bucketindex/updater.go

@@ -254,6 +254,7 @@ func (w *Updater) updateBlockMarks(ctx context.Context, old []*BlockDeletionMark
 		if errors.Is(err, ErrBlockDeletionMarkNotFound) {
 			// This could happen if the block is permanently deleted between the "list objects" and now.
 			level.Warn(w.logger).Log("msg", "skipped missing block deletion mark when updating bucket index", "block", id.String())
+			w.deleteOrphanedGlobalDeletionMark(ctx, id)
 			continue
 		}
 		if errors.Is(err, ErrBlockDeletionMarkCorrupted) {
@@ -270,6 +271,35 @@ func (w *Updater) updateBlockMarks(ctx context.Context, old []*BlockDeletionMark
 	return out, deletedBlocks, totalBlocksBlocksMarkedForNoCompaction, nil
 }
 
+// deleteOrphanedGlobalDeletionMark removes the block deletion mark from the global markers location
+// when the block it refers to no longer exists in the storage. Without this, the stale marker would
+// be listed (and skipped) at every update of the bucket index, and would keep the tenant's location in
+// the storage forever. The marker is removed only if the block location is empty, so that a block
+// which is being marked for deletion is never affected. This is best effort: failures are only logged.
+func (w *Updater) deleteOrphanedGlobalDeletionMark(ctx context.Context, id ulid.ULID) {
+	// The listing is not stopped at the first object: returning an error from the callback makes some
+	// clients (e.g. S3) leave the listing undrained, and is counted and retried as a failed operation.
+	// A block location holds only a handful of objects.
+	hasObjects := false
+	err := w.bkt.Iter(ctx, id.String()+objstore.DirDelim, func(string) error {
+		hasObjects = true
+		return nil
+	}, objstore.WithRecursiveIter())
+	if err != nil {
+		level.Warn(w.logger).Log("msg", "failed to check whether the block of an orphaned deletion mark exists", "block", id.String(), "err", err)
+		return
+	}
+	if hasObjects {
+		return
+	}
+
+	if err := w.bkt.Delete(ctx, BlockDeletionMarkFilepath(id)); err != nil && !w.bkt.IsObjNotFoundErr(err) {
+		level.Warn(w.logger).Log("msg", "failed to delete orphaned block deletion mark", "block", id.String(), "err", err)
+		return
+	}
+	level.Info(w.logger).Log("msg", "deleted orphaned block deletion mark of a block which no longer exists", "block", id.String())
+}
+
 func (w *Updater) updateBlockDeletionMarkIndexEntry(ctx context.Context, id ulid.ULID) (*BlockDeletionMark, error) {
 	m := metadata.DeletionMark{}
 

pkg/storage/tsdb/bucketindex/updater_test.go

@@ -318,6 +318,48 @@ func TestUpdater_UpdateIndex_ShouldSkipBlockMarkedForDeletionWithMissingGlobalMa
 	assert.Empty(t, nonCompactBlocks)
 }
 
+func TestUpdater_UpdateIndex_ShouldDeleteOrphanedGlobalDeletionMarks(t *testing.T) {
+	const userID = "user-1"
+
+	bkt, _ := testutil.PrepareFilesystemBucket(t)
+
+	ctx := context.Background()
+	logger := log.NewNopLogger()
+
+	bkt = BucketWithGlobalMarkers(bkt)
+	block1 := testutil.MockStorageBlock(t, bkt, userID, 10, 20)
+	block2 := testutil.MockStorageBlock(t, bkt, userID, 20, 30)
+	testutil.MockStorageDeletionMark(t, bkt, userID, block2)
+
+	// Orphaned marker: the block does not exist at all.
+	orphanID := ulid.MustNew(ulid.Now()+1, nil)
+	mark, err := json.Marshal(metadata.DeletionMark{ID: orphanID, Version: metadata.DeletionMarkVersion1, DeletionTime: time.Now().Add(-24 * time.Hour).Unix()})
+	require.NoError(t, err)
+	orphanPath := path.Join(userID, BlockDeletionMarkFilepath(orphanID))
+	require.NoError(t, bkt.Upload(ctx, orphanPath, bytes.NewReader(mark)))
+
+	// Global marker of a block which still has files but no local marker is left untouched.
+	partialID := ulid.MustNew(ulid.Now()+2, nil)
+	partialPath := path.Join(userID, BlockDeletionMarkFilepath(partialID))
+	require.NoError(t, bkt.Upload(ctx, partialPath, bytes.NewReader(mark)))
+	require.NoError(t, bkt.Upload(ctx, path.Join(userID, partialID.String(), "index"), strings.NewReader("x")))
+
+	w := NewUpdater(bkt, userID, nil, logger)
+	idx, _, _, err := w.UpdateIndex(ctx, nil)
+	require.NoError(t, err)
+	assertBucketIndexEqual(t, idx, bkt, userID,
+		[]tsdb.BlockMeta{block1, block2},
+		[]*metadata.DeletionMark{{ID: block2.ULID, Version: metadata.DeletionMarkVersion1, DeletionTime: idx.BlockDeletionMarks[0].DeletionTime}})
+
+	exists, err := bkt.Exists(ctx, orphanPath)
+	require.NoError(t, err)
+	assert.False(t, exists)
+
+	exists, err = bkt.Exists(ctx, partialPath)
+	require.NoError(t, err)
+	assert.True(t, exists)
+}
+
 func TestUpdater_UpdateIndex_NoTenantInTheBucket(t *testing.T) {
 	const userID = "user-1"