storage: Test for errors returned by MaybePopulateLastTime

storage/local/heads.go

@@ -188,7 +188,9 @@ func (hs *headsScanner) scan() bool {
 				// This is NOT the head chunk. So it's a chunk
 				// to be persisted, and we need to populate lastTime.
 				hs.chunksToPersistTotal++
-				cd.MaybePopulateLastTime()
+				if hs.err = cd.MaybePopulateLastTime(); hs.err != nil {
+					return false
+				}
 			}
 			chunkDescs[i] = cd
 		}

storage/local/persistence.go

@@ -693,7 +693,7 @@ func (p *persistence) checkpointSeriesMapAndHeads(fingerprintToSeries *seriesMap
 			}
 			// persistWatermark. We only checkpoint chunks that need persisting, so
 			// this is always 0.
-			if _, err = codable.EncodeVarint(w, int64(0)); err != nil {
+			if _, err = codable.EncodeVarint(w, 0); err != nil {
 				return
 			}
 			if m.series.modTime.IsZero() {

storage/local/series.go

@@ -247,7 +247,9 @@ func (s *memorySeries) add(v model.SamplePair) (int, error) {
 
 	// Populate lastTime of now-closed chunks.
 	for _, cd := range s.chunkDescs[len(s.chunkDescs)-len(chunks) : len(s.chunkDescs)-1] {
-		cd.MaybePopulateLastTime()
+		if err := cd.MaybePopulateLastTime(); err != nil {
+			return 0, err
+		}
 	}
 
 	s.lastTime = v.Timestamp
@@ -261,19 +263,18 @@ func (s *memorySeries) add(v model.SamplePair) (int, error) {
 // If the head chunk is already closed, the method is a no-op and returns false.
 //
 // The caller must have locked the fingerprint of the series.
-func (s *memorySeries) maybeCloseHeadChunk() bool {
+func (s *memorySeries) maybeCloseHeadChunk() (bool, error) {
 	if s.headChunkClosed {
-		return false
+		return false, nil
 	}
 	if time.Now().Sub(s.lastTime.Time()) > headChunkTimeout {
 		s.headChunkClosed = true
 		// Since we cannot modify the head chunk from now on, we
 		// don't need to bother with cloning anymore.
 		s.headChunkUsedByIterator = false
-		s.head().MaybePopulateLastTime()
-		return true
+		return true, s.head().MaybePopulateLastTime()
 	}
-	return false
+	return false, nil
 }
 
 // evictChunkDescs evicts chunkDescs if the chunk is evicted.

storage/local/storage.go

@@ -1376,7 +1376,12 @@ func (s *MemorySeriesStorage) maintainMemorySeries(
 
 	defer s.seriesOps.WithLabelValues(memoryMaintenance).Inc()
 
-	if series.maybeCloseHeadChunk() {
+	closed, err := series.maybeCloseHeadChunk()
+	if err != nil {
+		s.quarantineSeries(fp, series.metric, err)
+		s.persistErrors.Inc()
+	}
+	if closed {
 		s.incNumChunksToPersist(1)
 	}