Speed up the deletion of objects in trash

config/instance.go

@@ -82,6 +82,7 @@ const (
 
 	DefaultFileChunkPerSec    = 1000
 	DefaultTrashRetentionDays = 7
+	DefaultTrashDeleteWorkers = 1
 )
 
 func EmbedDefaults(cfgInstance *Instance) {
@@ -122,6 +123,9 @@ func EmbedDefaults(cfgInstance *Instance) {
 	if cfgInstance.VacuumCnf.TrashRetentionDays == 0 {
 		cfgInstance.VacuumCnf.TrashRetentionDays = DefaultTrashRetentionDays
 	}
+	if cfgInstance.VacuumCnf.TrashDeleteWorkers == 0 {
+		cfgInstance.VacuumCnf.TrashDeleteWorkers = DefaultTrashDeleteWorkers
+	}
 	cfgInstance.YezzeyRestoreParanoid = false
 }
 

config/vacuum.go

@@ -4,4 +4,5 @@ type Vacuum struct {
 	CheckBackup        bool `json:"check_backup" toml:"check_backup" yaml:"check_backup"`
 	FileChunkPerSec    int  `json:"file_chunk_per_sec" toml:"file_chunk_per_sec" yaml:"file_chunk_per_sec"`
 	TrashRetentionDays int  `json:"trash_retention_days" toml:"trash_retention_days" yaml:"trash_retention_days"`
+	TrashDeleteWorkers int  `json:"trash_delete_workers" toml:"trash_delete_workers" yaml:"trash_delete_workers"`
 }

pkg/message/delete_message.go

@@ -4,13 +4,13 @@ import (
 	"encoding/binary"
 )
 
-type DeleteMessage struct { //seg port
-	Name      string
-	Port      uint64
-	Segnum    uint64
-	Confirm   bool
-	Garbage   bool
-	CrazyDrop bool
+type DeleteMessage struct { // Seg port
+	Name      string // File path
+	Port      uint64 // Port segment/instance DB
+	Segnum    uint64 // Segment number
+	Confirm   bool   // Execute or Dry-run
+	Garbage   bool   // Run vacuum-style garbage deletion instead of deleting a single file
+	CrazyDrop bool   // For garbage mode: delete immediately instead of moving to trash
 }
 
 var _ ProtoMessage = &DeleteMessage{}

pkg/proc/delete_handler.go

@@ -5,7 +5,7 @@ import (
 	"fmt"
 	"path"
 	"strings"
-	"time"
+	"sync"
 
 	"github.com/pkg/errors"
 	"github.com/yezzey-gp/yproxy/config"
@@ -109,33 +109,46 @@ func (dh *BasicGarbageMgr) DeleteGarbageInBucket(bucket string, msg message.Dele
 		ylogger.Zero.Info().Str("bucket", bucket).Str("uploadId", upload).Msg("upload will be aborted")
 	}
 
-	if !msg.Confirm { //do not delete files if no confirmation flag provided
+	if !msg.Confirm { // Do not delete files if no confirmation flag provided
 		ylogger.Zero.Info().Msg("do not perform actual delete files as no confirmation flag provided")
 		return nil
 	}
 
 	/* Burst at 20% of vacuum rate capacity. It is pretty arbitrary at this time,
-	* but its not like something we need config field for... */
+	 * but its not like something we need config field for...
+	 */
 	limRate := config.InstanceConfig().VacuumCnf.FileChunkPerSec
 	limiter := rate.NewLimiter(rate.Limit(limRate), limRate/5)
 	ctx := context.Background()
 
+	var failedActionMsg, failedFilesMsg string
+	var operate func(file string) error
+	if msg.CrazyDrop {
+		failedActionMsg = "failed to delete some files"
+		failedFilesMsg = "some files were not deleted"
+		operate = func(file string) error {
+			ylogger.Zero.Info().Str("bucket", bucket).Str("path", file).Msg("simply delete")
+			return dh.StorageInterractor.DeleteObject(bucket, file) // Actual delete
+		}
+	} else {
+		failedActionMsg = "failed to move some files"
+		failedFilesMsg = "some files were not moved"
+		operate = func(file string) error {
+			tp := TrashPathFromRegPath(file, int(msg.Segnum))
+			return dh.StorageInterractor.MoveObject(bucket, file, tp)
+		}
+	}
+
 	var failed []string
 	for retryCount := 0; len(fileList) > 0 && retryCount < 10; retryCount++ {
 		for _, file := range fileList {
-			/* Dont move too fast */
+			/* Don't move too fast */
 			if err := limiter.Wait(ctx); err != nil {
 				break
 			}
-			if msg.CrazyDrop {
-				ylogger.Zero.Info().Str("bucket", bucket).Str("path", file).Msg("simply delete")
-				err = dh.StorageInterractor.DeleteObject(bucket, file)
-			} else {
-				tp := TrashPathFromRegPath(file, int(msg.Segnum))
-				err = dh.StorageInterractor.MoveObject(bucket, file, tp)
-			}
+			err = operate(file)
 			if err != nil {
-				ylogger.Zero.Warn().AnErr("err", err).Str("bucket", bucket).Str("file", file).Msg("failed to obsolete file")
+				ylogger.Zero.Warn().AnErr("err", err).Str("bucket", bucket).Str("file", file).Msg(failedActionMsg)
 				failed = append(failed, file)
 			}
 		}
@@ -144,9 +157,9 @@ func (dh *BasicGarbageMgr) DeleteGarbageInBucket(bucket string, msg message.Dele
 	}
 
 	if len(fileList) > 0 {
-		ylogger.Zero.Error().Str("bucket", bucket).Int("failed files count", len(fileList)).Msg("some files were not moved")
-		ylogger.Zero.Error().Str("bucket", bucket).Any("failed files", fileList).Msg("failed to move some files")
-		return errors.Wrap(err, "failed to move some files")
+		ylogger.Zero.Error().Str("bucket", bucket).Int("failed files count", len(fileList)).Msg(failedFilesMsg)
+		ylogger.Zero.Error().Str("bucket", bucket).Any("failed files", fileList).Msg(failedActionMsg)
+		return errors.Wrap(err, failedActionMsg)
 	}
 
 	for key, uploadId := range uploads {
@@ -196,6 +209,60 @@ func (dh *BasicGarbageMgr) ListDelete2Files(bucket string, msg message.Delete2Me
 	return filesToDelete, nil
 }
 
+func (dh *BasicGarbageMgr) garbageTrashParallel(bucket string, fileList []string) ([]string, error) {
+	workerCount := dh.Cnf.TrashDeleteWorkers
+	if workerCount <= 0 {
+		workerCount = config.DefaultTrashDeleteWorkers
+	}
+	if workerCount > len(fileList) {
+		workerCount = len(fileList)
+	}
+	if workerCount == 0 {
+		return nil, nil
+	}
+
+	jobs := make(chan string)
+	failedCh := make(chan string, len(fileList))
+	var wg sync.WaitGroup
+
+	for i := 0; i < workerCount; i++ {
+		wg.Go(func() {
+			for file := range jobs {
+				if err := dh.StorageInterractor.DeleteObject(bucket, file); err != nil {
+					ylogger.Zero.Warn().AnErr("err", err).Str("bucket", bucket).Str("file", file).Msg("failed to delete garbage file")
+					failedCh <- file
+				}
+			}
+		})
+	}
+
+	for _, file := range fileList {
+		jobs <- file
+	}
+	close(jobs)
+
+	wg.Wait()
+	close(failedCh)
+
+	failed := make([]string, 0, len(fileList))
+	for file := range failedCh {
+		failed = append(failed, file)
+	}
+	if len(failed) > 0 {
+		return failed, errors.New("failed to delete some garbage files")
+	}
+
+	return nil, nil
+}
+
+func (dh *BasicGarbageMgr) extractObjectPaths(fileList []*object.ObjectInfo) []string {
+	paths := make([]string, 0, len(fileList))
+	for _, file := range fileList {
+		paths = append(paths, file.Path)
+	}
+	return paths
+}
+
 func (dh *BasicGarbageMgr) DeletePrefixInBucket(bucket string, msg message.Delete2Message) error {
 	fileList, err := dh.ListDelete2Files(bucket, msg) // Return the list of files to be deleted
 	if err != nil {
@@ -222,27 +289,18 @@ func (dh *BasicGarbageMgr) DeletePrefixInBucket(bucket string, msg message.Delet
 		ylogger.Zero.Info().Msg("prefix doesn't contain trash aborted")
 		return nil
 	}
-	trashRetention := time.Hour * 24 * time.Duration(dh.Cnf.TrashRetentionDays)
-	var failed []*object.ObjectInfo
-	for retryCount := 0; len(fileList) > 0 && retryCount < 10; retryCount++ {
-		for _, file := range fileList {
-			if strings.Contains(file.Path, "trash") && file.LastMod.Add(trashRetention).Unix() < time.Now().Unix() {
-				ylogger.Zero.Debug().Str("bucket", bucket).Str("path", file.Path).Msg("simply delete")
-				err = dh.StorageInterractor.DeleteObject(bucket, file.Path) // Actual deletion
-			}
-			if err != nil {
-				ylogger.Zero.Warn().AnErr("err", err).Str("bucket", bucket).Str("file", file.Path).Msg("failed to delete file")
-				failed = append(failed, file)
-			}
+	pathsToDelete := dh.extractObjectPaths(fileList)
+	for retryCount := 0; len(pathsToDelete) > 0 && retryCount < 10; retryCount++ {
+		pathsToDelete, err = dh.garbageTrashParallel(bucket, pathsToDelete)
+		if err == nil {
+			break
 		}
-		fileList = failed
-		failed = make([]*object.ObjectInfo, 0)
 	}
 
-	if len(fileList) > 0 {
-		ylogger.Zero.Error().Str("bucket", bucket).Int("failed files count", len(fileList)).Msg("some files were not moved")
-		ylogger.Zero.Error().Str("bucket", bucket).Any("failed files", fileList).Msg("failed to move some files")
-		return errors.Wrap(err, "failed to move some files")
+	if len(pathsToDelete) > 0 {
+		ylogger.Zero.Error().Str("bucket", bucket).Int("failed files count", len(pathsToDelete)).Msg("some files were not deleted")
+		ylogger.Zero.Error().Str("bucket", bucket).Any("failed files", pathsToDelete).Msg("failed to delete some files")
+		return errors.Wrap(err, "failed to delete some files")
 	}
 
 	for key, uploadId := range uploads {