Gracefully skip compaction during shipping

CHANGELOG.md

@@ -120,6 +120,7 @@
 * [BUGFIX] Config: Fix CSV-list flags/YAML fields (e.g. `-compactor.enabled-tenants`) treating an explicitly empty string as a one-element list containing an empty tenant name instead of an empty list. #7714
 * [BUGFIX] Tenant Federation: Fix regex tenant federation dropping tenants when `-blocks-storage.users-scanner.cache-ttl` is set. The regex resolver sorted the user list returned by the users scanner in place, corrupting the scanner cache and progressively losing tenants on every sync until the cache expired. #7812
 * [BUGFIX] Tenant Federation: Fix regex tenant federation resolving to an empty user list right after startup. #7811
+* [BUGFIX] Ingester: Don't count a forced head compaction skipped because blocks shipping is in progress as a failure. Previously such skips incremented `cortex_ingester_tsdb_compactions_failed_total`, producing spurious alerts. #7842
 
 ## 1.21.1 2026-06-04
 

pkg/ingester/ingester.go

@@ -111,6 +111,7 @@ var (
 	errIngesterStopping = errors.New("ingester stopping")
 	errNoUserDb         = errors.New("no user db")
 	errLabelsOutOfOrder = errors.New("labels out of order")
+	errTsdbShipping     = errors.New("tsdb is in state activeShipping")
 
 	tsChunksPool zeropool.Pool[[]client.TimeSeriesChunk]
 
@@ -471,21 +472,24 @@ func (u *userTSDB) StartTime() (int64, error) {
 	return u.db.StartTime()
 }
 
-func (u *userTSDB) casState(from, to tsdbState) bool {
+func (u *userTSDB) casState(from, to tsdbState) (bool, tsdbState) {
 	u.stateMtx.Lock()
 	defer u.stateMtx.Unlock()
 
 	if u.state != from {
-		return false
+		return false, u.state
 	}
 	u.state = to
-	return true
+	return true, u.state
 }
 
 // compactHead compacts the Head block at specified block durations avoiding a single huge block.
 func (u *userTSDB) compactHead(ctx context.Context, blockDuration int64) error {
-	if !u.casState(active, forceCompacting) {
-		return errors.New("TSDB head cannot be compacted because it is not in active state (possibly being closed or blocks shipping in progress)")
+	if success, state := u.casState(active, forceCompacting); !success {
+		if state == activeShipping {
+			return errTsdbShipping
+		}
+		return fmt.Errorf("TSDB head cannot be compacted because it is not in active state (state: %d)", state)
 	}
 
 	defer u.casState(forceCompacting, active)
@@ -3409,7 +3413,7 @@ func (i *Ingester) shipBlocks(ctx context.Context, allowed *users.AllowedTenants
 
 		// Run the shipper's Sync() to upload unshipped blocks. Make sure the TSDB state is active, in order to
 		// avoid any race condition with closing idle TSDBs.
-		if !userDB.casState(active, activeShipping) {
+		if success, _ := userDB.casState(active, activeShipping); !success {
 			level.Info(logutil.WithContext(ctx, i.logger)).Log("msg", "shipper skipped because the TSDB is not active", "user", userID)
 			return nil
 		}
@@ -3523,8 +3527,13 @@ func (i *Ingester) compactBlocks(ctx context.Context, force bool, allowed *users
 		}
 
 		if err != nil {
-			i.TSDBState.compactionsFailed.Inc()
-			level.Warn(logutil.WithContext(ctx, i.logger)).Log("msg", "TSDB blocks compaction for user has failed", "user", userID, "err", err, "compactReason", reason)
+			// Don't treat blocks shipping as a failure
+			if errors.Is(err, errTsdbShipping) {
+				level.Info(logutil.WithContext(ctx, i.logger)).Log("msg", "TSDB blocks compaction for user was skipped", "user", userID, "err", err, "compactReason", reason)
+			} else {
+				i.TSDBState.compactionsFailed.Inc()
+				level.Warn(logutil.WithContext(ctx, i.logger)).Log("msg", "TSDB blocks compaction for user has failed", "user", userID, "err", err, "compactReason", reason)
+			}
 		} else {
 			level.Debug(logutil.WithContext(ctx, i.logger)).Log("msg", "TSDB blocks compaction completed successfully", "user", userID, "compactReason", reason)
 		}
@@ -3574,7 +3583,7 @@ func (i *Ingester) closeAndDeleteUserTSDBIfIdle(userID string) tsdbCloseCheckRes
 	}
 
 	// This disables pushes and force-compactions. Not allowed to close while shipping is in progress.
-	if !userDB.casState(active, closing) {
+	if success, _ := userDB.casState(active, closing); !success {
 		return tsdbNotActive
 	}
 

pkg/ingester/ingester_test.go

@@ -5388,8 +5388,8 @@ func TestIngester_ReadNotFailWhenTSDBIsBeingDeleted(t *testing.T) {
 			err = db.Close()
 			require.NoError(t, err)
 
-			b := db.casState(active, c.state)
-			require.True(t, b)
+			casSuccess, _ := db.casState(active, c.state)
+			require.True(t, casSuccess)
 
 			// Mock request
 			ctx = user.InjectOrgID(context.Background(), userID)
@@ -6395,7 +6395,8 @@ func TestIngesterPushErrorDuringForcedCompaction(t *testing.T) {
 	db, err := i.getTSDB(userID)
 	require.NoError(t, err)
 	require.NotNil(t, db)
-	require.True(t, db.casState(active, forceCompacting))
+	casSuccess, _ := db.casState(active, forceCompacting)
+	require.True(t, casSuccess)
 
 	// Ingestion should fail with a 503.
 	req, _ := mockWriteRequest(t, labels.FromStrings(labels.MetricName, "test"), 0, util.TimeToMillis(time.Now()))
@@ -6404,10 +6405,89 @@ func TestIngesterPushErrorDuringForcedCompaction(t *testing.T) {
 	require.Equal(t, httpgrpc.Errorf(http.StatusServiceUnavailable, "%s", wrapWithUser(errors.New("forced compaction in progress"), userID).Error()), err)
 
 	// Ingestion is successful after a flush.
-	require.True(t, db.casState(forceCompacting, active))
+	casSuccess, _ = db.casState(forceCompacting, active)
+	require.True(t, casSuccess)
 	pushSingleSampleWithMetadata(t, i)
 }
 
+func TestUserTSDB_compactHead_returnsShippingError(t *testing.T) {
+	i, err := prepareIngesterWithBlocksStorage(t, defaultIngesterTestConfig(t), prometheus.NewRegistry())
+	require.NoError(t, err)
+
+	require.NoError(t, services.StartAndAwaitRunning(context.Background(), i))
+	t.Cleanup(func() {
+		_ = services.StopAndAwaitTerminated(context.Background(), i)
+	})
+
+	// Wait until it's ACTIVE
+	test.Poll(t, 1*time.Second, ring.ACTIVE, func() any {
+		return i.lifecycler.GetState()
+	})
+
+	pushSingleSampleWithMetadata(t, i)
+
+	db, err := i.getTSDB(userID)
+	require.NoError(t, err)
+	require.NotNil(t, db)
+
+	blockDuration := i.cfg.BlocksStorageConfig.TSDB.BlockRanges[0].Milliseconds()
+
+	// When the TSDB is shipping blocks, compactHead should return the sentinel errTsdbShipping
+	// so callers can distinguish it from a genuine compaction failure.
+	casSuccess, _ := db.casState(active, activeShipping)
+	require.True(t, casSuccess)
+	err = db.compactHead(context.Background(), blockDuration)
+	require.ErrorIs(t, err, errTsdbShipping)
+	casSuccess, _ = db.casState(activeShipping, active)
+	require.True(t, casSuccess)
+
+	// For any other non-active state, compactHead returns a generic error (not errTsdbShipping)
+	// reporting the current state.
+	casSuccess, _ = db.casState(active, closing)
+	require.True(t, casSuccess)
+	err = db.compactHead(context.Background(), blockDuration)
+	require.Error(t, err)
+	require.NotErrorIs(t, err, errTsdbShipping)
+	require.Contains(t, err.Error(), "not in active state")
+	casSuccess, _ = db.casState(closing, active)
+	require.True(t, casSuccess)
+}
+
+func TestIngester_compactBlocks_shippingIsNotCountedAsFailure(t *testing.T) {
+	i, err := prepareIngesterWithBlocksStorage(t, defaultIngesterTestConfig(t), prometheus.NewRegistry())
+	require.NoError(t, err)
+
+	require.NoError(t, services.StartAndAwaitRunning(context.Background(), i))
+	t.Cleanup(func() {
+		_ = services.StopAndAwaitTerminated(context.Background(), i)
+	})
+
+	// Wait until it's ACTIVE
+	test.Poll(t, 1*time.Second, ring.ACTIVE, func() any {
+		return i.lifecycler.GetState()
+	})
+
+	pushSingleSampleWithMetadata(t, i)
+
+	db, err := i.getTSDB(userID)
+	require.NoError(t, err)
+	require.NotNil(t, db)
+
+	// Simulate blocks shipping being in progress. Forced compaction will fail the CAS to
+	// forceCompacting, but this must not be counted as a compaction failure.
+	casSuccess, _ := db.casState(active, activeShipping)
+	require.True(t, casSuccess)
+	t.Cleanup(func() { db.casState(activeShipping, active) })
+
+	require.Equal(t, float64(0), testutil.ToFloat64(i.TSDBState.compactionsFailed))
+
+	i.compactBlocks(context.Background(), true, nil)
+
+	// Compaction was triggered but the shipping skip must not increment the failure counter.
+	require.Equal(t, float64(1), testutil.ToFloat64(i.TSDBState.compactionsTriggered))
+	require.Equal(t, float64(0), testutil.ToFloat64(i.TSDBState.compactionsFailed))
+}
+
 func TestIngesterNoFlushWithInFlightRequest(t *testing.T) {
 	registry := prometheus.NewRegistry()
 	i, err := prepareIngesterWithBlocksStorage(t, defaultIngesterTestConfig(t), registry)