receive: Active tenants not compacting due to wall clock check

pkg/receive/multitsdb.go

@@ -419,53 +419,54 @@ func (t *tenant) generateCompactionDelay() time.Duration {
 	return time.Duration(rand.Int63n((t.maxBlockDuration*compactionDelayPercentBlockLength)/100)) * time.Millisecond
 }
 
-func (t *tenant) startPeriodicHeadCompaction() {
-	var interval = time.Duration(t.maxBlockDuration) * time.Millisecond
+// tryCompactHead attempts to compact the head if it has accumulated enough data.
+func (t *tenant) tryCompactHead() error {
+	db := t.readyS.Get()
+	if db == nil {
+		return fmt.Errorf("no DB found")
+	}
+	head := db.Head()
+	if head.MinTime() < 0 {
+		return nil
+	}
 
-	doIter := func() error {
-		db := t.readyS.Get()
-		if db == nil {
-			return fmt.Errorf("no DB found")
-		}
-		head := db.Head()
-		if head.MinTime() < 0 {
-			return nil
-		}
+	// Wall-clock time determines whether the head is old enough to compact,
+	// ensuring tenants that stopped receiving samples still get flushed.
+	// The head's data span (MaxTime - MinTime) determines how many blocks
+	// to produce, compacting until the span drops below the threshold.
+	compactionThreshold := int64(1.5 * float64(t.maxBlockDuration))
+	sinceOldestSampleMs := time.Since(time.UnixMilli(head.MinTime())).Milliseconds()
+	if sinceOldestSampleMs <= compactionThreshold {
+		return nil
+	}
 
-		// Wall-clock time determines whether the head is old enough to compact,
-		// ensuring tenants that stopped receiving samples still get flushed.
-		// The head's data span (MaxTime - MinTime) determines how many blocks
-		// to produce, compacting until the span drops below the threshold.
-		compactionThreshold := int64(1.5 * float64(t.maxBlockDuration))
-		sinceOldestSampleMs := time.Since(time.UnixMilli(head.MinTime())).Milliseconds()
-		if sinceOldestSampleMs <= compactionThreshold {
+	for {
+		select {
+		case <-t.doneC:
 			return nil
+		default:
 		}
 
-		for {
-			select {
-			case <-t.doneC:
-				return nil
-			default:
-			}
-
-			if err := t.compactHead(db); err != nil {
-				return fmt.Errorf("compact head: %w", err)
-			}
-			t.lastSuccessfulHeadCompaction.Store(time.Now().UnixNano())
-
-			head = db.Head()
-			if head.MaxTime()-head.MinTime() <= compactionThreshold {
-				break
-			}
+		if err := t.compactHead(db); err != nil {
+			return fmt.Errorf("compact head: %w", err)
 		}
+		t.lastSuccessfulHeadCompaction.Store(time.Now().UnixNano())
 
-		if err := db.CompactOOOHead(context.Background()); err != nil {
-			return fmt.Errorf("compact ooo head: %w", err)
+		head = db.Head()
+		if head.MaxTime()-head.MinTime() <= compactionThreshold {
+			break
 		}
-		return nil
 	}
 
+	if err := db.CompactOOOHead(context.Background()); err != nil {
+		return fmt.Errorf("compact ooo head: %w", err)
+	}
+	return nil
+}
+
+func (t *tenant) startPeriodicHeadCompaction() {
+	var interval = time.Duration(t.maxBlockDuration) * time.Millisecond
+
 	compactionDelay := t.generateCompactionDelay()
 	go func() {
 		level.Info(t.logger).Log("msg", "starting periodic head compaction", "initial_delay", compactionDelay.String(), "interval", interval.String())
@@ -482,7 +483,7 @@ func (t *tenant) startPeriodicHeadCompaction() {
 			select {
 			case <-ticker.C:
 				level.Info(t.logger).Log("msg", "running periodic head compaction")
-				if err := doIter(); err != nil {
+				if err := t.tryCompactHead(); err != nil {
 					level.Error(t.logger).Log("msg", "periodic head compaction failed", "err", err)
 				}
 

pkg/receive/multitsdb_test.go

@@ -631,6 +631,90 @@
 	})
 }
 
+// TestWallClockCompactionBug reproduces the bug where wall-clock check prevents
+// compaction for active tenants after restart.
+//
+// - Tenant restarts/initializes at T0
+// - Samples arrive continuously from T0 to T0+2.5h (2.5 hours of data)
+// - Data span is 2.5h > 1.5h threshold (should compact!)
+// - But wall-clock time since MinTime is only 2.5h < 3h threshold
+// - With wall-clock check, compaction is blocked incorrectly
+func TestWallClockCompactionBug(t *testing.T) {
+	t.Parallel()
+
+	dir := t.TempDir()
+	maxBlockDuration := (2 * time.Hour).Milliseconds()
+	compactionThreshold := int64(1.5 * float64(maxBlockDuration)) // 3 hours
+
+	m := NewMultiTSDB(openTestRoot(t, dir), log.NewLogfmtLogger(os.Stderr), prometheus.NewRegistry(),
+		&tsdb.Options{
+			MinBlockDuration:  maxBlockDuration,
+			MaxBlockDuration:  maxBlockDuration,
+			RetentionDuration: (24 * time.Hour).Milliseconds(),
+		},
+		labels.FromStrings("replica", "test"),
+		"tenant_id",
+		nil,
+		false,
+		false,
+		metadata.NoneFunc,
+	)
+	defer m.Close()
+
+	// first sample arrives at "now"
+	now := time.Now()
+
+	// This creates 3.5 hours of data span, which exceeds the 3h threshold
+	for step := time.Duration(0); step <= 210*time.Minute; step += time.Minute {
+		testutil.Ok(t, appendSample(m, "test-tenant", now.Add(step)))
+	}
+
+	tenant := m.testGetTenant("test-tenant")
+	db := tenant.readyStorage().Get()
+	testutil.Assert(t, db != nil, "TSDB should be initialized")
+
+	head := db.Head()
+
+	dataSpanMs := head.MaxTime() - head.MinTime()
+	testutil.Assert(t, dataSpanMs > compactionThreshold,
+		"data span should exceed threshold: got %dms, threshold %dms", dataSpanMs, compactionThreshold)
+
+	// We have 3.5 hours of DATA (data span > 3h threshold) so it should compact
+	// But wall-clock time since MinTime is ~0 seconds (just appended now) < 3h threshold
+	sinceOldestSampleMs := time.Since(time.UnixMilli(head.MinTime())).Milliseconds()
+	testutil.Assert(t, sinceOldestSampleMs < compactionThreshold,
+		"wall-clock time should be less than threshold: got %dms, threshold %dms",
+		sinceOldestSampleMs, compactionThreshold)
+
+	t.Logf("Data span: %dms (%.1fh), threshold: %dms (%.1fh) should compact",
+		dataSpanMs, float64(dataSpanMs)/3600000, compactionThreshold, float64(compactionThreshold)/3600000)
+	t.Logf("Wall-clock since MinTime: %dms, which means we just skip compaction",
+		sinceOldestSampleMs)
+
+	initialBlocks := len(db.Blocks())
+	initialHeadMinTime := head.MinTime()
+
+	err := tenant.tryCompactHead()
+	testutil.Ok(t, err)
+
+	// Blocks were created
+	// Head MinTime advanced
+	head = db.Head()
+	testutil.Assert(t, len(db.Blocks()) > initialBlocks,
+		"compaction should have created blocks: before=%d, after=%d",
+		initialBlocks, len(db.Blocks()))
+
+	testutil.Assert(t, head.MinTime() > initialHeadMinTime,
+		"head MinTime should have advanced after truncation: before=%d, after=%d",
+		initialHeadMinTime, head.MinTime())
+
+	// Verify that data span is now below threshold (compaction worked)
+	newDataSpan := head.MaxTime() - head.MinTime()
+	testutil.Assert(t, newDataSpan <= compactionThreshold,
+		"data span should be below threshold after compaction: got %dms, threshold %dms",
+		newDataSpan, compactionThreshold)
+}
+
 func TestMultiTSDBAddNewTenant(t *testing.T) {
 	if testing.
 		Short() {