fix: correct locking bug when deleting a series (#26649) (#27143)

Co-authored-by: Geoffrey Wossum <gwossum@influxdata.com>
fix: correct locking bug when deleting a series (#26649)
closes: #26648

Author: devanbenz

tsdb/engine/tsm1/engine.go

@@ -1623,6 +1623,9 @@ func (e *Engine) deleteSeriesRange(ctx context.Context, seriesKeys [][]byte, min
 	// would delete it from the index.
 	minKey := seriesKeys[0]
 
+	// Ensure seriesKeys slice is correctly read and written concurrently in the Apply func.
+	var seriesKeysLock sync.RWMutex
+
 	// Apply runs this func concurrently.  The seriesKeys slice is mutated concurrently
 	// by different goroutines setting positions to nil.
 	if err := e.FileStore.Apply(ctx, func(r TSMFile) error {
@@ -1639,18 +1642,28 @@ func (e *Engine) deleteSeriesRange(ctx context.Context, seriesKeys [][]byte, min
 			seriesKey, _ := SeriesAndFieldFromCompositeKey(indexKey)
 
 			// Skip over any deleted keys that are less than our tsm key
-			cmp := bytes.Compare(seriesKeys[j], seriesKey)
-			for j < len(seriesKeys) && cmp < 0 {
-				j++
-				if j >= len(seriesKeys) {
-					return nil
+			cmp, cont := func() (int, bool) {
+				seriesKeysLock.RLock()
+				defer seriesKeysLock.RUnlock()
+				cmp := bytes.Compare(seriesKeys[j], seriesKey)
+				for j < len(seriesKeys) && cmp < 0 {
+					j++
+					if j >= len(seriesKeys) {
+						return 0, false // don't continue processing seriesKeys.
+					}
+					cmp = bytes.Compare(seriesKeys[j], seriesKey)
 				}
-				cmp = bytes.Compare(seriesKeys[j], seriesKey)
+				return cmp, true // continue processing seriesKeys.
+			}()
+			if !cont {
+				return nil
 			}
 
 			// We've found a matching key, cross it out so we do not remove it from the index.
 			if j < len(seriesKeys) && cmp == 0 {
+				seriesKeysLock.Lock()
 				seriesKeys[j] = emptyBytes
+				seriesKeysLock.Unlock()
 				j++
 			}
 		}

tsdb/engine/tsm1/engine_test.go

@@ -1355,6 +1355,95 @@ func TestEngine_DeleteSeriesRange(t *testing.T) {
 	}
 }
 
+// TestEngine_DeleteSeriesRange_MultiTSMLocking checks for locking issues when deleting series
+// ranges from a large number of TSM files. This test is derived from TestEngine_DeleteSeriesRange.
+func TestEngine_DeleteSeriesRange_MultiTSMLocking(t *testing.T) {
+	for _, index := range tsdb.RegisteredIndexes() {
+		t.Run(index, func(t *testing.T) {
+			e, err := NewEngine(t, index)
+			require.NoError(t, err)
+
+			// mock the planner so compactions don't run during the test
+			e.CompactionPlan = &mockPlanner{}
+			require.NoError(t, e.Open(t.Context()))
+
+			// In order to trigger potentially locking issues within DeleteSeriesRange, we must create much more
+			// TSM files than the number of CPUs. This combined with how the test data interleaves deleted and not
+			// deleted series within a TSM file triggers locking issues if they are present.
+			numTSMFiles := runtime.NumCPU() * 20
+			var ts int64
+			for range numTSMFiles {
+				ts += 1000000000
+
+				// Create a few points.
+				p1 := MustParsePointString(fmt.Sprintf("cpu,host=0 value=1.1 %d", ts+5000000000)) // Should not be deleted
+				p2 := MustParsePointString(fmt.Sprintf("cpu,host=A value=1.2 %d", ts+1000000000))
+				p3 := MustParsePointString(fmt.Sprintf("cpu,host=A value=1.3 %d", ts+2000000000))
+				p4 := MustParsePointString(fmt.Sprintf("cpu,host=B value=1.3 %d", ts+3000000000)) // Should not be deleted
+				p5 := MustParsePointString(fmt.Sprintf("cpu,host=B value=1.3 %d", ts+4000000000)) // Should not be deleted
+				p6 := MustParsePointString(fmt.Sprintf("cpu,host=C value=1.3 %d", ts))
+				p7 := MustParsePointString(fmt.Sprintf("mem,host=C value=1.3 %d", ts))  // Should not be deleted
+				p8 := MustParsePointString(fmt.Sprintf("disk,host=C value=1.3 %d", ts)) // Should not be deleted
+
+				for _, p := range []models.Point{p1, p2, p3, p4, p5, p6, p7, p8} {
+					require.NoError(t, e.CreateSeriesIfNotExists(p.Key(), p.Name(), p.Tags()), "create series index error")
+				}
+
+				require.NoError(t, e.WritePoints(t.Context(), []models.Point{p1, p2, p3, p4, p5, p6, p7, p8}), "failed to write points")
+				require.NoError(t, e.WriteSnapshot(), "failed to snapshot")
+			}
+
+			keys := e.FileStore.Keys()
+			require.Len(t, keys, 6, "series count mismatch")
+
+			itr := &seriesIterator{keys: [][]byte{[]byte("cpu,host=0"), []byte("cpu,host=A"), []byte("cpu,host=B"), []byte("cpu,host=C")}}
+			require.NoError(t, e.DeleteSeriesRange(t.Context(), itr, 0, ts+2000000000), "failed to delete series")
+
+			keys = e.FileStore.Keys()
+			require.Len(t, keys, 4, "series count mismatch")
+
+			require.Contains(t, keys, "cpu,host=B#!~#value", "wrong series deleted")
+
+			// Check that the series still exists in the index
+			indexSet := tsdb.IndexSet{Indexes: []tsdb.Index{e.index}, SeriesFile: e.sfile}
+			iter, err := indexSet.MeasurementSeriesIDIterator([]byte("cpu"))
+			require.NoError(t, err, "iterator error")
+			defer func() {
+				if iter != nil {
+					require.NoError(t, iter.Close())
+				}
+			}()
+
+			elem, err := iter.Next()
+			require.NoError(t, err)
+			require.NotEqual(t, uint64(0), elem.SeriesID, "series index mismatch: EOF, exp 2 series")
+
+			// Lookup series.
+			name, tags := e.sfile.Series(elem.SeriesID)
+			require.Equal(t, []byte("cpu"), name, "series mismatch")
+
+			require.True(t, tags.Equal(models.NewTags(map[string]string{"host": "0"})) || tags.Equal(models.NewTags(map[string]string{"host": "B"})),
+				`series mismatch: got %s, exp either "host=0" or "host=B"`, tags)
+			require.NoError(t, iter.Close())
+
+			// Deleting remaining series should remove them from the series.
+			itr = &seriesIterator{keys: [][]byte{[]byte("cpu,host=0"), []byte("cpu,host=B")}}
+			require.NoError(t, e.DeleteSeriesRange(t.Context(), itr, 0, ts+9000000000), "failed to delete series")
+
+			indexSet = tsdb.IndexSet{Indexes: []tsdb.Index{e.index}, SeriesFile: e.sfile}
+			iter, err = indexSet.MeasurementSeriesIDIterator([]byte("cpu"))
+			require.NoError(t, err, "iterator error")
+			if iter == nil {
+				return
+			}
+
+			elem, err = iter.Next()
+			require.NoError(t, err)
+			require.Equal(t, uint64(0), elem.SeriesID, "got an undeleted series id, but series should be dropped from index")
+		})
+	}
+}
+
 func TestEngine_DeleteSeriesRangeWithPredicate(t *testing.T) {
 	for _, index := range tsdb.RegisteredIndexes() {
 		t.Run(index, func(t *testing.T) {