fix: do not leak file handles from Compactor.write

tsdb/engine/tsm1/compact.go

@@ -1064,11 +1064,11 @@ func (c *Compactor) writeNewFiles(generation, sequence int, src []string, iter K
 
 		// We've hit the max file limit and there is more to write.  Create a new file
 		// and continue.
-		if err == errMaxFileExceeded || err == ErrMaxBlocksExceeded {
+		if errors.Is(err, errMaxFileExceeded) || errors.Is(err, ErrMaxBlocksExceeded) {
 			files = append(files, fileName)
 			logger.Debug("file size or block count exceeded, opening another output file", zap.String("output_file", fileName))
 			continue
-		} else if err == ErrNoValues {
+		} else if errors.Is(err, ErrNoValues) {
 			logger.Debug("Dropping empty file", zap.String("output_file", fileName))
 			// If the file only contained tombstoned entries, then it would be a 0 length
 			// file that we can drop.
@@ -1128,33 +1128,34 @@ func (c *Compactor) write(path string, iter KeyIterator, throttle bool, logger *
 	// in memory.
 	if iter.EstimatedIndexSize() > 64*1024*1024 {
 		w, err = NewTSMWriterWithDiskBuffer(limitWriter)
-		if err != nil {
-			return err
-		}
 	} else {
 		w, err = NewTSMWriter(limitWriter)
-		if err != nil {
-			return err
-		}
 	}
-
+	if err != nil {
+		// Close the file and return if we can't create the TSMWriter
+		return errors.Join(err, fd.Close())
+	}
 	defer func() {
+		errs := make([]error, 0, 2)
+		errs = append(errs, err)
 		closeErr := w.Close()
 		if err == nil {
+			// Save closeErr as err for later checks
 			err = closeErr
 		}
+		errs = append(errs, closeErr)
 
 		// Check for errors where we should not remove the file
 		_, inProgress := err.(errCompactionInProgress)
-		maxBlocks := err == ErrMaxBlocksExceeded
-		maxFileSize := err == errMaxFileExceeded
+		maxBlocks := errors.Is(err, ErrMaxBlocksExceeded)
+		maxFileSize := errors.Is(err, errMaxFileExceeded)
 		if inProgress || maxBlocks || maxFileSize {
+			err = errors.Join(errs...)
 			return
+		} else if err != nil {
+			errs = append(errs, w.Remove())
 		}
-
-		if err != nil {
-			_ = w.Remove()
-		}
+		err = errors.Join(errs...)
 	}()
 
 	lastLogSize := w.Size()
@@ -1179,7 +1180,7 @@ func (c *Compactor) write(path string, iter KeyIterator, throttle bool, logger *
 		}
 
 		// Write the key and value
-		if err := w.WriteBlock(key, minTime, maxTime, block); err == ErrMaxBlocksExceeded {
+		if err := w.WriteBlock(key, minTime, maxTime, block); errors.Is(err, ErrMaxBlocksExceeded) {
 			if err := w.WriteIndex(); err != nil {
 				return err
 			}
@@ -1188,7 +1189,7 @@ func (c *Compactor) write(path string, iter KeyIterator, throttle bool, logger *
 			return err
 		}
 
-		// If we have a max file size configured and we're over it, close out the file
+		// If we're over maxTSMFileSize, close out the file
 		// and return the error.
 		if w.Size() > maxTSMFileSize {
 			if err := w.WriteIndex(); err != nil {

tsdb/engine/tsm1/writer.go

@@ -66,6 +66,7 @@ import (
 	"bufio"
 	"bytes"
 	"encoding/binary"
+	"errors"
 	"fmt"
 	"hash/crc32"
 	"io"
@@ -511,19 +512,15 @@ func (d *directIndex) Size() uint32 {
 }
 
 func (d *directIndex) Close() error {
+	errs := make([]error, 0, 3)
 	// Flush anything remaining in the index
-	if err := d.w.Flush(); err != nil {
-		return err
-	}
-
-	if d.fd == nil {
-		return nil
+	errs = append(errs, d.w.Flush())
+	if d.fd != nil {
+		// Close and remove the temporary index file
+		errs = append(errs, d.fd.Close())
+		errs = append(errs, os.Remove(d.fd.Name()))
 	}
-
-	if err := d.fd.Close(); err != nil {
-		return err
-	}
-	return os.Remove(d.fd.Name())
+	return errors.Join(errs...)
 }
 
 // Remove removes the index from any tempory storage
@@ -532,11 +529,14 @@ func (d *directIndex) Remove() error {
 		return nil
 	}
 
-	// Close the file handle to prevent leaking.  We ignore the error because
-	// we just want to cleanup and remove the file.
-	_ = d.fd.Close()
-
-	return os.Remove(d.fd.Name())
+	errs := make([]error, 0, 2)
+	// Close the file handle to prevent leaking.
+	// We don't let an error stop the removal.
+	if err := d.fd.Close(); err != nil && !errors.Is(err, os.ErrClosed) {
+		errs = append(errs, err)
+	}
+	errs = append(errs, os.Remove(d.fd.Name()))
+	return errors.Join(errs...)
 }
 
 // tsmWriter writes keys and values in the TSM format
@@ -756,25 +756,19 @@ func (t *tsmWriter) sync() error {
 }
 
 func (t *tsmWriter) Close() error {
-	if err := t.Flush(); err != nil {
-		return err
-	}
-
-	if err := t.index.Close(); err != nil {
-		return err
-	}
-
+	errs := make([]error, 0, 3)
+	errs = append(errs, t.Flush())
+	errs = append(errs, t.index.Close())
 	if c, ok := t.wrapped.(io.Closer); ok {
-		return c.Close()
+		errs = append(errs, c.Close())
 	}
-	return nil
+	return errors.Join(errs...)
 }
 
 // Remove removes any temporary storage used by the writer.
 func (t *tsmWriter) Remove() error {
-	if err := t.index.Remove(); err != nil {
-		return err
-	}
+	errs := make([]error, 0, 3)
+	errs = append(errs, t.index.Remove())
 
 	// nameCloser is the most permissive interface we can close the wrapped
 	// value with.
@@ -783,14 +777,16 @@ func (t *tsmWriter) Remove() error {
 		Name() string
 	}
 
+	// If the writer is not a memory buffer, we can remove the file.
 	if f, ok := t.wrapped.(nameCloser); ok {
-		// Close the file handle to prevent leaking.  We ignore the error because
-		// we just want to cleanup and remove the file.
-		_ = f.Close()
-
-		return os.Remove(f.Name())
+		// Close the file handle to prevent leaking.
+		if err := f.Close(); err != nil && !errors.Is(err, os.ErrClosed) {
+			errs = append(errs, err)
+		}
+		// Remove the file
+		errs = append(errs, os.Remove(f.Name()))
 	}
-	return nil
+	return errors.Join(errs...)
 }
 
 func (t *tsmWriter) Size() uint32 {