diff --git a/tsdb/head.go b/tsdb/head.go index 24efe56d5d..210becc71f 100644 --- a/tsdb/head.go +++ b/tsdb/head.go @@ -73,7 +73,7 @@ type Head struct { numSeries atomic.Uint64 numStaleSeries atomic.Uint64 minOOOTime, maxOOOTime atomic.Int64 // TODO(jesusvazquez) These should be updated after garbage collection. - minTime, maxTime atomic.Int64 // Current min and max of the samples included in the head. TODO(jesusvazquez) Ensure these are properly tracked. + minTime, maxTime atomic.Int64 // Current min and max of the samples included in the head. minTime != math.MaxInt64 is used to determine whether the head has been initialized, so care must be taken that the initialized state only changes after maxTime is already updated. minValidTime atomic.Int64 // Mint allowed to be added to the head. It shouldn't be lower than the maxt of the last persisted block. lastWALTruncationTime atomic.Int64 lastMemoryTruncationTime atomic.Int64 @@ -1143,6 +1143,8 @@ func (h *Head) PostingsCardinalityStats(statsByLabelName string, limit int) *ind return h.cardinalityCache } +// Expands the Head time bounds to include the provided in-order range. +// Concurrent updates retry with compare-and-swap so the bounds only move outward. func (h *Head) updateMinMaxTime(mint, maxt int64) { for { lt := h.MinTime() @@ -1817,7 +1819,8 @@ func (h *Head) Meta() BlockMeta { } } -// MinTime returns the lowest time bound on visible data in the head. +// MinTime returns the lowest time bound on visible data in the head. If this +// value is math.MaxInt64, we consider the head to be uninitialized. func (h *Head) MinTime() int64 { return h.minTime.Load() } diff --git a/tsdb/head_append.go b/tsdb/head_append.go index 3bb9de9584..4a6beab482 100644 --- a/tsdb/head_append.go +++ b/tsdb/head_append.go @@ -117,11 +117,17 @@ func (a *initAppender) AppendSTZeroSample(ref storage.SeriesRef, lset labels.Lab // initTime initializes a head with the first timestamp. This only needs to be called // for a completely fresh head with an empty WAL. func (h *Head) initTime(t int64) { - if !h.minTime.CompareAndSwap(math.MaxInt64, t) { - // Concurrent appends that are initializing. - // Wait until h.maxTime is swapped to avoid minTime/maxTime races. + // maxTime must be set before minTime, because initialized() keys off minTime. + // If minTime were set to t first, a concurrent Head.Appender call could + // observe initialized() == true while maxTime is still math.MinInt64; the + // resulting underflow in appendableMinValidTime would reject in-range samples + // with ErrOutOfBounds. + if !h.maxTime.CompareAndSwap(math.MinInt64, t) { + // Another goroutine already won the init race. Wait until it also sets + // minTime, so callers that next read initialized() can rely on both + // bounds being valid. antiDeadlockTimeout := time.After(500 * time.Millisecond) - for h.maxTime.Load() == math.MinInt64 { + for h.minTime.Load() == math.MaxInt64 { select { case <-antiDeadlockTimeout: return @@ -130,8 +136,7 @@ func (h *Head) initTime(t int64) { } return } - // Ensure that max time is initialized to at least the min time we just set. - h.maxTime.CompareAndSwap(math.MinInt64, t) + h.minTime.CompareAndSwap(math.MaxInt64, t) } func (a *initAppender) GetRef(lset labels.Labels, hash uint64) (storage.SeriesRef, labels.Labels) {