feat: Add SetNewReadersBlocked for purger.purge

tsdb/engine/tsm1/file_store.go

@@ -1603,6 +1603,15 @@ func (p *purger) purge(fileNames []string) {
 		for {
 			p.mu.Lock()
 			for k, v := range p.files {
+				// https://github.com/influxdata/influxdb/issues/26110
+				// Blocks new readers so there is not a potential time of check/time of use
+				// bug below when tsmfile.InUse() is called -> tsmfile.Close() is called
+				//err := p.fileStore.SetNewReadersBlocked(true)
+				//if err != nil {
+				//	logger.Error("failed to block new readers", zap.Strings("files", fileNames), zap.Error(err))
+				//	continue
+				//}
+
 				// In order to ensure that there are no races with this (file held externally calls Ref
 				// after we check InUse), we need to maintain the invariant that every handle to a file
 				// is handed out in use (Ref'd), and handlers only ever relinquish the file once (call Unref
@@ -1611,17 +1620,33 @@ func (p *purger) purge(fileNames []string) {
 				if !v.InUse() {
 					if err := v.Close(); err != nil {
 						logger.Error("close file failed", zap.String("file", k), zap.Error(err))
+						//err = p.fileStore.SetNewReadersBlocked(false)
+						//if err != nil {
+						//	logger.Error("failed to block new readers", zap.Strings("file", fileNames), zap.Error(err))
+						//	continue
+						//}
 						continue
 					}
 
 					if err := v.Remove(); err != nil {
 						logger.Error("remove file failed", zap.String("file", k), zap.Error(err))
+						//err = p.fileStore.SetNewReadersBlocked(false)
+						//if err != nil {
+						//	logger.Error("failed to block new readers", zap.Strings("file", fileNames), zap.Error(err))
+						//	continue
+						//}
 						continue
 					}
 					logger.Debug("successfully removed", zap.String("file", k))
 					delete(p.files, k)
 					purgeCount++
 				}
+
+				//err = p.fileStore.SetNewReadersBlocked(false)
+				//if err != nil {
+				//	logger.Error("failed to block new readers", zap.Strings("file", fileNames), zap.Error(err))
+				//	continue
+				//}
 			}
 
 			if len(p.files) == 0 {

tsdb/engine/tsm1/file_store_test.go

@@ -3,6 +3,7 @@ package tsm1_test
 import (
 	"context"
 	"fmt"
+	"math/rand"
 	"os"
 	"path/filepath"
 	"reflect"
@@ -3043,6 +3044,58 @@ func TestFileStore_Observer(t *testing.T) {
 	unlinks, finishes = nil, nil
 }
 
+func TestFileStore_PurgerContention(t *testing.T) {
+	t.Helper()
+
+	dir := MustTempDir()
+
+	// Create some TSM files...
+	data := make([]keyValues, 0, 10)
+	for i := 0; i < 1000; i++ {
+		data = append(data, keyValues{"cpu", []tsm1.Value{tsm1.NewValue(0, 1.0)}})
+	}
+
+	regFiles, err := newFileDir(dir, data...)
+	require.NoError(t, err)
+	tmpFiles, err := newTmpFileDir(dir, data...)
+	require.NoError(t, err)
+
+	fs := tsm1.NewFileStore(dir)
+	err = fs.Open()
+	require.NoError(t, err)
+
+	done := make(chan struct{})
+
+	// Simulate a query that will call filestore.Apply to modify TSM files
+	go func() {
+		fs.Apply(func(r tsm1.TSMFile) error {
+			time.Sleep(time.Duration(rand.Intn(5)) * time.Millisecond)
+			return nil
+		})
+		done <- struct{}{}
+	}()
+	go func() {
+		err := fs.Replace(regFiles, tmpFiles)
+		require.NoError(t, err)
+		done <- struct{}{}
+	}()
+	// Simulate a query
+	go func() {
+		fs.Apply(func(r tsm1.TSMFile) error {
+			time.Sleep(time.Duration(rand.Intn(5)) * time.Millisecond)
+			return nil
+		})
+		done <- struct{}{}
+	}()
+
+	select {
+	case <-done:
+		<-done
+	case <-time.After(10 * time.Second):
+		t.Fatalf("timed out after 10 seconds")
+	}
+}
+
 func newFileDir(dir string, values ...keyValues) ([]string, error) {
 	var files []string
 
@@ -3074,7 +3127,39 @@ func newFileDir(dir string, values ...keyValues) ([]string, error) {
 		files = append(files, newName)
 	}
 	return files, nil
+}
 
+func newTmpFileDir(dir string, values ...keyValues) ([]string, error) {
+	var files []string
+
+	id := 1
+	for _, v := range values {
+		f := MustTempFile(dir)
+		w, err := tsm1.NewTSMWriter(f)
+		if err != nil {
+			return nil, err
+		}
+
+		if err := w.Write([]byte(v.key), v.values); err != nil {
+			return nil, err
+		}
+
+		if err := w.WriteIndex(); err != nil {
+			return nil, err
+		}
+
+		if err := w.Close(); err != nil {
+			return nil, err
+		}
+		newName := filepath.Join(filepath.Dir(f.Name()), tsm1.DefaultFormatFileName(id, 1)+".tsm.tmp")
+		if err := os.Rename(f.Name(), newName); err != nil {
+			return nil, err
+		}
+		id++
+
+		files = append(files, newName)
+	}
+	return files, nil
 }
 
 func newFiles(dir string, values ...keyValues) ([]string, error) {

tsdb/index/tsi1/log_file.go

@@ -179,17 +179,26 @@ func (f *LogFile) Close() error {
 	f.wg.Wait()
 
 	if f.w != nil {
-		f.w.Flush()
+		err := f.w.Flush()
+		if err != nil {
+			return err
+		}
 		f.w = nil
 	}
 
 	if f.file != nil {
-		f.file.Close()
+		err := f.file.Close()
+		if err != nil {
+			return err
+		}
 		f.file = nil
 	}
 
 	if f.data != nil {
-		mmap.Unmap(f.data)
+		err := mmap.Unmap(f.data)
+		if err != nil {
+			return err
+		}
 	}
 
 	f.mms = make(logMeasurements)

tsdb/index/tsi1/partition_test.go

@@ -3,13 +3,13 @@ package tsi1_test
 import (
 	"errors"
 	"fmt"
+	"github.com/influxdata/influxdb/tsdb"
+	"github.com/influxdata/influxdb/tsdb/index/tsi1"
+	"github.com/stretchr/testify/require"
 	"os"
 	"path/filepath"
 	"syscall"
 	"testing"
-
-	"github.com/influxdata/influxdb/tsdb"
-	"github.com/influxdata/influxdb/tsdb/index/tsi1"
 )
 
 func TestPartition_Open(t *testing.T) {
@@ -164,6 +164,26 @@ func TestPartition_Compact_Write_Fail(t *testing.T) {
 	})
 }
 
+func TestPartition_Compact_And_Close_Contention(t *testing.T) {
+	t.Helper()
+	const ROUNDS = 100
+	sfile := MustOpenSeriesFile()
+	t.Cleanup(func() {
+		err := sfile.Close()
+		require.NoError(t, err)
+	})
+
+	p := MustOpenPartition(sfile.SeriesFile)
+	p.Partition.SetMaxLogFileSize(-1)
+	go func() {
+		p.Compact()
+	}()
+	go func() {
+		err := p.Close()
+		require.NoError(t, err)
+	}()
+}
+
 // Partition is a test wrapper for tsi1.Partition.
 type Partition struct {
 	*tsi1.Partition