Fix apply retention issue (#5342)

* call mark phase started/finished only when applying retention

* correctly pass userID to IntervalMayHaveExpiredChunks call on retention and dletion expiry check

pkg/storage/stores/shipper/compactor/compactor.go

@@ -408,7 +408,7 @@ func (c *Compactor) CompactTable(ctx context.Context, tableName string, applyRet
 
 	interval := retention.ExtractIntervalFromTableName(tableName)
 	intervalMayHaveExpiredChunks := false
-	if c.cfg.RetentionEnabled && applyRetention {
+	if applyRetention {
 		intervalMayHaveExpiredChunks = c.expirationChecker.IntervalMayHaveExpiredChunks(interval, "")
 	}
 
@@ -424,7 +424,7 @@ func (c *Compactor) RunCompaction(ctx context.Context, applyRetention bool) erro
 	status := statusSuccess
 	start := time.Now()
 
-	if c.cfg.RetentionEnabled {
+	if applyRetention {
 		c.expirationChecker.MarkPhaseStarted()
 	}
 
@@ -439,7 +439,7 @@ func (c *Compactor) RunCompaction(ctx context.Context, applyRetention bool) erro
 			}
 		}
 
-		if c.cfg.RetentionEnabled {
+		if applyRetention {
 			if status == statusSuccess {
 				c.expirationChecker.MarkPhaseFinished()
 			} else {
@@ -550,7 +550,7 @@ func (e *expirationChecker) MarkPhaseFinished() {
 }
 
 func (e *expirationChecker) IntervalMayHaveExpiredChunks(interval model.Interval, userID string) bool {
-	return e.retentionExpiryChecker.IntervalMayHaveExpiredChunks(interval, "") || e.deletionExpiryChecker.IntervalMayHaveExpiredChunks(interval, "")
+	return e.retentionExpiryChecker.IntervalMayHaveExpiredChunks(interval, userID) || e.deletionExpiryChecker.IntervalMayHaveExpiredChunks(interval, userID)
 }
 
 func (e *expirationChecker) DropFromIndex(ref retention.ChunkEntry, tableEndTime model.Time, now model.Time) bool {

pkg/storage/stores/shipper/compactor/compactor_test.go

@@ -107,7 +107,7 @@ func TestCompactor_RunCompaction(t *testing.T) {
 	cm := storage.NewClientMetrics()
 	defer cm.Unregister()
 	compactor := setupTestCompactor(t, tempDir, cm)
-	err := compactor.RunCompaction(context.Background(), true)
+	err := compactor.RunCompaction(context.Background(), false)
 	require.NoError(t, err)
 
 	for name := range tables {