From 61877e5d9feace848e86e1523f0b3b0af5ae11c9 Mon Sep 17 00:00:00 2001 From: Vlasdislav Date: Tue, 7 Jul 2026 19:28:38 +0300 Subject: [PATCH 1/8] Feat: Delete data from trash in parallel --- config/instance.go | 4 ++ config/vacuum.go | 1 + pkg/message/delete_message.go | 14 ++--- pkg/proc/delete_handler.go | 112 ++++++++++++++++++++++++++-------- 4 files changed, 97 insertions(+), 34 deletions(-) diff --git a/config/instance.go b/config/instance.go index 57d2f78..96692f2 100644 --- a/config/instance.go +++ b/config/instance.go @@ -82,6 +82,7 @@ const ( DefaultFileChunkPerSec = 1000 DefaultTrashRetentionDays = 7 + DefaultTrashDeleteWorkers = 8 ) 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 } diff --git a/config/vacuum.go b/config/vacuum.go index fbf4f06..3ded4d7 100644 --- a/config/vacuum.go +++ b/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"` } diff --git a/pkg/message/delete_message.go b/pkg/message/delete_message.go index 07b41af..26cc6ee 100644 --- a/pkg/message/delete_message.go +++ b/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{} diff --git a/pkg/proc/delete_handler.go b/pkg/proc/delete_handler.go index 0552dde..918c183 100644 --- a/pkg/proc/delete_handler.go +++ b/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,7 +109,7 @@ 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 } @@ -122,18 +122,31 @@ func (dh *BasicGarbageMgr) DeleteGarbageInBucket(bucket string, msg message.Dele var failed []string for retryCount := 0; len(fileList) > 0 && retryCount < 10; retryCount++ { + if msg.CrazyDrop { + for _, file := range fileList { + /* Don't move too fast */ + if err := limiter.Wait(ctx); err != nil { + break + } + ylogger.Zero.Info().Str("bucket", bucket).Str("path", file).Msg("simply delete") + err = dh.StorageInterractor.DeleteObject(bucket, file) // Actual delete + if err != nil { + ylogger.Zero.Warn().AnErr("err", err).Str("bucket", bucket).Str("file", file).Msg("failed to obsolete file") + failed = append(failed, file) + } + } + fileList = failed + failed = make([]string, 0) + continue + } + for _, file := range fileList { /* Dont 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) - } + tp := TrashPathFromRegPath(file, int(msg.Segnum)) + err = dh.StorageInterractor.MoveObject(bucket, file, tp) if err != nil { ylogger.Zero.Warn().AnErr("err", err).Str("bucket", bucket).Str("file", file).Msg("failed to obsolete file") failed = append(failed, file) @@ -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 { From 98797e6f0b0accc5efbd4db25d70837fc46d139c Mon Sep 17 00:00:00 2001 From: Vlasdislav Date: Tue, 7 Jul 2026 22:29:03 +0300 Subject: [PATCH 2/8] Refactor: Code duplication with what we had and fix the default number of workers --- config/instance.go | 2 +- pkg/proc/delete_handler.go | 52 +++++++++++++++++++------------------- 2 files changed, 27 insertions(+), 27 deletions(-) diff --git a/config/instance.go b/config/instance.go index 96692f2..cb99946 100644 --- a/config/instance.go +++ b/config/instance.go @@ -82,7 +82,7 @@ const ( DefaultFileChunkPerSec = 1000 DefaultTrashRetentionDays = 7 - DefaultTrashDeleteWorkers = 8 + DefaultTrashDeleteWorkers = 1 ) func EmbedDefaults(cfgInstance *Instance) { diff --git a/pkg/proc/delete_handler.go b/pkg/proc/delete_handler.go index 918c183..5734078 100644 --- a/pkg/proc/delete_handler.go +++ b/pkg/proc/delete_handler.go @@ -115,40 +115,40 @@ func (dh *BasicGarbageMgr) DeleteGarbageInBucket(bucket string, msg message.Dele } /* 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 failed []string - for retryCount := 0; len(fileList) > 0 && retryCount < 10; retryCount++ { - if msg.CrazyDrop { - for _, file := range fileList { - /* Don't move too fast */ - if err := limiter.Wait(ctx); err != nil { - break - } - ylogger.Zero.Info().Str("bucket", bucket).Str("path", file).Msg("simply delete") - err = dh.StorageInterractor.DeleteObject(bucket, file) // Actual delete - if err != nil { - ylogger.Zero.Warn().AnErr("err", err).Str("bucket", bucket).Str("file", file).Msg("failed to obsolete file") - failed = append(failed, file) - } - } - fileList = failed - failed = make([]string, 0) - continue + 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 } - 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) } } @@ -157,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 { From c69fb962cdae29db30d5d1e0c6d080f3161b19e7 Mon Sep 17 00:00:00 2001 From: Vlasdislav Date: Wed, 8 Jul 2026 13:00:29 +0300 Subject: [PATCH 3/8] Fix: Return the conditions for deletion from trash --- pkg/proc/delete_handler.go | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/pkg/proc/delete_handler.go b/pkg/proc/delete_handler.go index 5734078..88d93b7 100644 --- a/pkg/proc/delete_handler.go +++ b/pkg/proc/delete_handler.go @@ -6,6 +6,7 @@ import ( "path" "strings" "sync" + "time" "github.com/pkg/errors" "github.com/yezzey-gp/yproxy/config" @@ -257,8 +258,11 @@ func (dh *BasicGarbageMgr) garbageTrashParallel(bucket string, fileList []string func (dh *BasicGarbageMgr) extractObjectPaths(fileList []*object.ObjectInfo) []string { paths := make([]string, 0, len(fileList)) + trashRetention := time.Hour * 24 * time.Duration(dh.Cnf.TrashRetentionDays) for _, file := range fileList { - paths = append(paths, file.Path) + if strings.Contains(file.Path, "trash") && file.LastMod.Add(trashRetention).Unix() < time.Now().Unix() { + paths = append(paths, file.Path) + } } return paths } From 548cfe8f25229f22ad1bf927125f9a7a7be98296 Mon Sep 17 00:00:00 2001 From: Vlasdislav Date: Wed, 8 Jul 2026 13:01:54 +0300 Subject: [PATCH 4/8] Test: Add any unit test for parallel delition from trash --- pkg/proc/delete_handler_test.go | 199 ++++++++++++++++++++++++++++++++ 1 file changed, 199 insertions(+) diff --git a/pkg/proc/delete_handler_test.go b/pkg/proc/delete_handler_test.go index fd4ccb7..a9fff80 100644 --- a/pkg/proc/delete_handler_test.go +++ b/pkg/proc/delete_handler_test.go @@ -1,8 +1,11 @@ package proc_test import ( + "fmt" + "sync" "testing" + "github.com/pkg/errors" "github.com/stretchr/testify/assert" "github.com/yezzey-gp/yproxy/config" "github.com/yezzey-gp/yproxy/pkg/message" @@ -118,3 +121,199 @@ func TestListDelete2Files(t *testing.T) { assert.Equal(t, len(filesInStorage), len(actualFilesToDelete)) assert.Equal(t, filesInStorage, actualFilesToDelete) } + +func TestDeletePrefixInBucketDeletesAllFilesOnceInParallelGarbagePass(t *testing.T) { + ctrl := gomock.NewController(t) + + msg := message.Delete2Message{ + Prefix: "trash", + Garbage: true, + Confirm: true, + } + + filesInStorage := []*object.ObjectInfo{ + {Path: "trash/a"}, + {Path: "trash/b"}, + {Path: "trash/c"}, + {Path: "trash/d"}, + } + + storage := mock.NewMockStorageInteractor(ctrl) + storage.EXPECT().ListBucketPath("bucket", msg.Prefix, true).Return(filesInStorage, nil) + storage.EXPECT().ListFailedMultipartUploads("bucket").Return(map[string]string{}, nil) + + var mu sync.Mutex + deleted := make(map[string]int) + storage.EXPECT().DeleteObject("bucket", gomock.Any()).DoAndReturn(func(bucket, key string) error { + mu.Lock() + defer mu.Unlock() + deleted[key]++ + return nil + }).Times(len(filesInStorage)) + + handler := proc.BasicGarbageMgr{ + StorageInterractor: storage, + Cnf: &config.Vacuum{TrashDeleteWorkers: 4}, + } + + err := handler.DeletePrefixInBucket("bucket", msg) + assert.NoError(t, err) + for _, file := range filesInStorage { + assert.Equal(t, 1, deleted[file.Path], fmt.Sprintf("unexpected delete count for %s", file.Path)) + } +} + +func TestDeletePrefixInBucketRetriesFailedGarbageDeletes(t *testing.T) { + ctrl := gomock.NewController(t) + + msg := message.Delete2Message{ + Prefix: "trash", + Garbage: true, + Confirm: true, + } + + filesInStorage := []*object.ObjectInfo{ + {Path: "trash/a"}, + {Path: "trash/b"}, + {Path: "trash/c"}, + } + + storage := mock.NewMockStorageInteractor(ctrl) + storage.EXPECT().ListBucketPath("bucket", msg.Prefix, true).Return(filesInStorage, nil) + storage.EXPECT().ListFailedMultipartUploads("bucket").Return(map[string]string{}, nil) + + var mu sync.Mutex + attempts := make(map[string]int) + storage.EXPECT().DeleteObject("bucket", gomock.Any()).DoAndReturn(func(bucket, key string) error { + mu.Lock() + defer mu.Unlock() + attempts[key]++ + if key == "trash/b" && attempts[key] == 1 { + return errors.New("transient delete failure") + } + return nil + }).Times(4) + + handler := proc.BasicGarbageMgr{ + StorageInterractor: storage, + Cnf: &config.Vacuum{TrashDeleteWorkers: 3}, + } + + err := handler.DeletePrefixInBucket("bucket", msg) + assert.NoError(t, err) + assert.Equal(t, 1, attempts["trash/a"]) + assert.Equal(t, 2, attempts["trash/b"]) + assert.Equal(t, 1, attempts["trash/c"]) +} + +func TestDeletePrefixInBucketReturnsFailedGarbageDeletesAfterRetries(t *testing.T) { + ctrl := gomock.NewController(t) + + msg := message.Delete2Message{ + Prefix: "trash", + Garbage: true, + Confirm: true, + } + + filesInStorage := []*object.ObjectInfo{ + {Path: "trash/a"}, + {Path: "trash/b"}, + } + + storage := mock.NewMockStorageInteractor(ctrl) + storage.EXPECT().ListBucketPath("bucket", msg.Prefix, true).Return(filesInStorage, nil) + storage.EXPECT().ListFailedMultipartUploads("bucket").Return(map[string]string{}, nil) + storage.EXPECT().DeleteObject("bucket", gomock.Any()).DoAndReturn(func(bucket, key string) error { + if key == "trash/b" { + return errors.New("persistent delete failure") + } + return nil + }).Times(11) + + handler := proc.BasicGarbageMgr{ + StorageInterractor: storage, + Cnf: &config.Vacuum{TrashDeleteWorkers: 2}, + } + + err := handler.DeletePrefixInBucket("bucket", msg) + assert.Error(t, err) + assert.Contains(t, err.Error(), "failed to delete some files") +} + +func TestDeletePrefixInBucketCapsWorkerCountToFileCount(t *testing.T) { + ctrl := gomock.NewController(t) + + msg := message.Delete2Message{ + Prefix: "trash", + Garbage: true, + Confirm: true, + } + + filesInStorage := []*object.ObjectInfo{ + {Path: "trash/a"}, + {Path: "trash/b"}, + } + + storage := mock.NewMockStorageInteractor(ctrl) + storage.EXPECT().ListBucketPath("bucket", msg.Prefix, true).Return(filesInStorage, nil) + storage.EXPECT().ListFailedMultipartUploads("bucket").Return(map[string]string{}, nil) + + var mu sync.Mutex + deleted := make(map[string]int) + storage.EXPECT().DeleteObject("bucket", gomock.Any()).DoAndReturn(func(bucket, key string) error { + mu.Lock() + defer mu.Unlock() + deleted[key]++ + return nil + }).Times(len(filesInStorage)) + + handler := proc.BasicGarbageMgr{ + StorageInterractor: storage, + Cnf: &config.Vacuum{TrashDeleteWorkers: 10}, + } + + err := handler.DeletePrefixInBucket("bucket", msg) + assert.NoError(t, err) + assert.Equal(t, 1, deleted["trash/a"]) + assert.Equal(t, 1, deleted["trash/b"]) +} + +func TestDeletePrefixInBucketUsesDefaultWorkerCountWhenConfiguredZero(t *testing.T) { + ctrl := gomock.NewController(t) + + msg := message.Delete2Message{ + Prefix: "trash", + Garbage: true, + Confirm: true, + } + + filesInStorage := []*object.ObjectInfo{ + {Path: "trash/a"}, + {Path: "trash/b"}, + {Path: "trash/c"}, + } + + storage := mock.NewMockStorageInteractor(ctrl) + storage.EXPECT().ListBucketPath("bucket", msg.Prefix, true).Return(filesInStorage, nil) + storage.EXPECT().ListFailedMultipartUploads("bucket").Return(map[string]string{}, nil) + + var mu sync.Mutex + deleted := make(map[string]int) + storage.EXPECT().DeleteObject("bucket", gomock.Any()).DoAndReturn(func(bucket, key string) error { + mu.Lock() + defer mu.Unlock() + deleted[key]++ + return nil + }).Times(len(filesInStorage)) + + handler := proc.BasicGarbageMgr{ + StorageInterractor: storage, + Cnf: &config.Vacuum{TrashDeleteWorkers: 0}, + } + + err := handler.DeletePrefixInBucket("bucket", msg) + assert.NoError(t, err) + assert.Equal(t, 1, deleted["trash/a"]) + assert.Equal(t, 1, deleted["trash/b"]) + assert.Equal(t, 1, deleted["trash/c"]) +} From 8488fdcac443674cfed01cf0d346c1a90292122f Mon Sep 17 00:00:00 2001 From: Vlasdislav Date: Thu, 9 Jul 2026 11:12:21 +0300 Subject: [PATCH 5/8] Fix: Garbage cleanup config add test and refactoring --- pkg/proc/delete_handler.go | 91 +++++++++++++++++---------------- pkg/proc/delete_handler_test.go | 43 +++++++++++++++- pkg/proc/interaction.go | 4 +- 3 files changed, 89 insertions(+), 49 deletions(-) diff --git a/pkg/proc/delete_handler.go b/pkg/proc/delete_handler.go index 88d93b7..1b04851 100644 --- a/pkg/proc/delete_handler.go +++ b/pkg/proc/delete_handler.go @@ -104,7 +104,7 @@ func (dh *BasicGarbageMgr) DeleteGarbageInBucket(bucket string, msg message.Dele ylogger.Zero.Info().Str("bucket", bucket).Int("amount", len(uploads)).Msg("multipart uploads will be aborted") for _, file := range fileList { - ylogger.Zero.Info().Str("bucket", bucket).Bool("crazy mode", msg.CrazyDrop).Str("file", file).Msg("file will be deleted") + ylogger.Zero.Info().Str("bucket", bucket).Bool("crazy mode", msg.CrazyDrop).Str("file", file.Path).Msg("file will be deleted") } for _, upload := range uploads { ylogger.Zero.Info().Str("bucket", bucket).Str("uploadId", upload).Msg("upload will be aborted") @@ -115,32 +115,37 @@ func (dh *BasicGarbageMgr) DeleteGarbageInBucket(bucket string, msg message.Dele return nil } - /* Burst at 20% of vacuum rate capacity. It is pretty arbitrary at this time, + /* + * Burst at 20% of vacuum rate capacity. It is pretty arbitrary at this time, * but its not like something we need config field for... */ - limRate := config.InstanceConfig().VacuumCnf.FileChunkPerSec + limRate := dh.Cnf.FileChunkPerSec + if limRate <= 0 { + limRate = config.DefaultFileChunkPerSec + } limiter := rate.NewLimiter(rate.Limit(limRate), limRate/5) ctx := context.Background() var failedActionMsg, failedFilesMsg string - var operate func(file string) error + var operate func(file *object.ObjectInfo) 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 + operate = func(file *object.ObjectInfo) error { + ylogger.Zero.Info().Str("bucket", bucket).Str("path", file.Path).Msg("immediately delete garbage file") + return dh.StorageInterractor.DeleteObject(bucket, file.Path) } } 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) + operate = func(file *object.ObjectInfo) error { + tp := TrashPathFromRegPath(file.Path, int(msg.Segnum)) + ylogger.Zero.Info().Str("bucket", bucket).Str("path", file.Path).Msg("move garbage file to trash") + return dh.StorageInterractor.MoveObject(bucket, file.Path, tp) } } - var failed []string + var failed []*object.ObjectInfo for retryCount := 0; len(fileList) > 0 && retryCount < 10; retryCount++ { for _, file := range fileList { /* Don't move too fast */ @@ -149,12 +154,12 @@ func (dh *BasicGarbageMgr) DeleteGarbageInBucket(bucket string, msg message.Dele } err = operate(file) if err != nil { - ylogger.Zero.Warn().AnErr("err", err).Str("bucket", bucket).Str("file", file).Msg(failedActionMsg) + ylogger.Zero.Warn().AnErr("err", err).Str("bucket", bucket).Str("file", file.Path).Msg(failedActionMsg) failed = append(failed, file) } } fileList = failed - failed = make([]string, 0) + failed = make([]*object.ObjectInfo, 0) } if len(fileList) > 0 { @@ -164,7 +169,7 @@ func (dh *BasicGarbageMgr) DeleteGarbageInBucket(bucket string, msg message.Dele } for key, uploadId := range uploads { - /* Dont move too fast */ + /* Don't move too fast */ if err := limiter.Wait(ctx); err != nil { break } @@ -185,10 +190,10 @@ func (dh *BasicGarbageMgr) HandleDeleteGarbage(msg message.DeleteMessage) error return nil } func (dh *BasicGarbageMgr) ListDelete2Files(bucket string, msg message.Delete2Message) ([]*object.ObjectInfo, error) { - //get first backup lsn + // Get first backup lsn var err error - //list files in storage + // List files in storage ylogger.Zero.Info().Str("path", msg.Prefix).Msg("listing prefix") objectMetas, err := dh.StorageInterractor.ListBucketPath(bucket, msg.Prefix, true) if err != nil { @@ -210,7 +215,7 @@ func (dh *BasicGarbageMgr) ListDelete2Files(bucket string, msg message.Delete2Me return filesToDelete, nil } -func (dh *BasicGarbageMgr) garbageTrashParallel(bucket string, fileList []string) ([]string, error) { +func (dh *BasicGarbageMgr) garbageTrashParallel(bucket string, fileList []*object.ObjectInfo) ([]*object.ObjectInfo, error) { workerCount := dh.Cnf.TrashDeleteWorkers if workerCount <= 0 { workerCount = config.DefaultTrashDeleteWorkers @@ -222,15 +227,15 @@ func (dh *BasicGarbageMgr) garbageTrashParallel(bucket string, fileList []string return nil, nil } - jobs := make(chan string) - failedCh := make(chan string, len(fileList)) + jobs := make(chan *object.ObjectInfo) + failedCh := make(chan *object.ObjectInfo, 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") + if err := dh.StorageInterractor.DeleteObject(bucket, file.Path); err != nil { + ylogger.Zero.Warn().AnErr("err", err).Str("bucket", bucket).Str("file", file.Path).Msg("failed to delete garbage file") failedCh <- file } } @@ -245,7 +250,7 @@ func (dh *BasicGarbageMgr) garbageTrashParallel(bucket string, fileList []string wg.Wait() close(failedCh) - failed := make([]string, 0, len(fileList)) + failed := make([]*object.ObjectInfo, 0, len(fileList)) for file := range failedCh { failed = append(failed, file) } @@ -256,17 +261,6 @@ func (dh *BasicGarbageMgr) garbageTrashParallel(bucket string, fileList []string return nil, nil } -func (dh *BasicGarbageMgr) extractObjectPaths(fileList []*object.ObjectInfo) []string { - paths := make([]string, 0, len(fileList)) - trashRetention := time.Hour * 24 * time.Duration(dh.Cnf.TrashRetentionDays) - for _, file := range fileList { - if strings.Contains(file.Path, "trash") && file.LastMod.Add(trashRetention).Unix() < time.Now().Unix() { - 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 { @@ -285,7 +279,7 @@ func (dh *BasicGarbageMgr) DeletePrefixInBucket(bucket string, msg message.Delet 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 } @@ -293,17 +287,24 @@ func (dh *BasicGarbageMgr) DeletePrefixInBucket(bucket string, msg message.Delet ylogger.Zero.Info().Msg("prefix doesn't contain trash aborted") return nil } - pathsToDelete := dh.extractObjectPaths(fileList) - for retryCount := 0; len(pathsToDelete) > 0 && retryCount < 10; retryCount++ { - pathsToDelete, err = dh.garbageTrashParallel(bucket, pathsToDelete) + trashRetention := time.Hour * 24 * time.Duration(dh.Cnf.TrashRetentionDays) + filtered := fileList[:0] + for _, file := range fileList { + if strings.Contains(file.Path, "trash") && file.LastMod.Add(trashRetention).Unix() < time.Now().Unix() { + filtered = append(filtered, file) + } + } + fileList = filtered + for retryCount := 0; len(fileList) > 0 && retryCount < 10; retryCount++ { + fileList, err = dh.garbageTrashParallel(bucket, fileList) if err == nil { break } } - 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") + if len(fileList) > 0 { + ylogger.Zero.Error().Str("bucket", bucket).Int("failed files count", len(fileList)).Msg("some files were not deleted") + ylogger.Zero.Error().Str("bucket", bucket).Any("failed files", fileList).Msg("failed to delete some files") return errors.Wrap(err, "failed to delete some files") } @@ -338,8 +339,8 @@ func (dh *BasicGarbageMgr) HandleDeleteFile(msg message.DeleteMessage) error { return nil } -func (dh *BasicGarbageMgr) ListGarbageFiles(bucket string, msg message.DeleteMessage) ([]string, error) { - //get first backup lsn +func (dh *BasicGarbageMgr) ListGarbageFiles(bucket string, msg message.DeleteMessage) ([]*object.ObjectInfo, error) { + // Get first backup lsn var firstBackupLSN uint64 var err error @@ -355,7 +356,7 @@ func (dh *BasicGarbageMgr) ListGarbageFiles(bucket string, msg message.DeleteMes ylogger.Zero.Info().Uint64("lsn", firstBackupLSN).Msg("omit first backup LSN") } - //list files in storage + // List files in storage ylogger.Zero.Info().Str("path", msg.Name).Msg("listing prefix") objectMetas, err := dh.StorageInterractor.ListBucketPath(bucket, msg.Name, true) if err != nil { @@ -370,7 +371,7 @@ func (dh *BasicGarbageMgr) ListGarbageFiles(bucket string, msg message.DeleteMes } ylogger.Zero.Info().Int("virtual", len(vi)).Int("expire", len(ei)).Msg("received virtual index and expire index") - filesToDelete := make([]string, 0) + filesToDelete := make([]*object.ObjectInfo, 0) for i := range objectMetas { reworkedName := objectMetas[i].Path ylogger.Zero.Debug().Str("reworked name", reworkedName).Msg("lookup chunk") @@ -386,7 +387,7 @@ func (dh *BasicGarbageMgr) ListGarbageFiles(bucket string, msg message.DeleteMes Bool("file in expire index", ok). Bool("lsn is less than in first backup", lsn < firstBackupLSN). Msg("file does not persist in virtual index, nor needed for PITR, so will be deleted") - filesToDelete = append(filesToDelete, objectMetas[i].Path) + filesToDelete = append(filesToDelete, objectMetas[i]) } } diff --git a/pkg/proc/delete_handler_test.go b/pkg/proc/delete_handler_test.go index a9fff80..d20c492 100644 --- a/pkg/proc/delete_handler_test.go +++ b/pkg/proc/delete_handler_test.go @@ -62,8 +62,8 @@ func TestFilesToDeletion(t *testing.T) { assert.NoError(t, err) assert.Equal(t, 2, len(list)) - assert.Equal(t, "1663_16530_deleted-before-backup_18002_", list[0]) - assert.Equal(t, "some_trash", list[1]) + assert.Equal(t, "1663_16530_deleted-before-backup_18002_", list[0].Path) + assert.Equal(t, "some_trash", list[1].Path) } func TestTrashPathConversion(t *testing.T) { @@ -317,3 +317,42 @@ func TestDeletePrefixInBucketUsesDefaultWorkerCountWhenConfiguredZero(t *testing assert.Equal(t, 1, deleted["trash/b"]) assert.Equal(t, 1, deleted["trash/c"]) } + +func TestDeleteGarbageInBucketMovesObjectsWhenCrazyDropDisabled(t *testing.T) { + ctrl := gomock.NewController(t) + + msg := message.DeleteMessage{ + Name: "path", + Port: 6000, + Segnum: 0, + Confirm: true, + CrazyDrop: false, + } + + filesInStorage := []*object.ObjectInfo{ + {Path: "segments_005/seg0/basebackups_005/yezzey/file1"}, + {Path: "segments_005/seg0/basebackups_005/yezzey/file2"}, + } + + storage := mock.NewMockStorageInteractor(ctrl) + storage.EXPECT().ListBucketPath("bucket", msg.Name, true).Return(filesInStorage, nil) + storage.EXPECT().ListFailedMultipartUploads("bucket").Return(map[string]string{}, nil) + storage.EXPECT().MoveObject("bucket", filesInStorage[0].Path, proc.TrashPathFromRegPath(filesInStorage[0].Path, int(msg.Segnum))).Return(nil) + storage.EXPECT().MoveObject("bucket", filesInStorage[1].Path, proc.TrashPathFromRegPath(filesInStorage[1].Path, int(msg.Segnum))).Return(nil) + + database := mock.NewMockDatabaseInterractor(ctrl) + database.EXPECT().GetVirtualExpireIndexes(msg.Port).Return(map[string]bool{}, map[string]uint64{ + filesInStorage[0].Path: 0, + filesInStorage[1].Path: 0, + }, nil) + + handler := proc.BasicGarbageMgr{ + StorageInterractor: storage, + DbInterractor: database, + BackupInterractor: nil, + Cnf: &config.Vacuum{CheckBackup: false, FileChunkPerSec: 1000}, + } + + err := handler.DeleteGarbageInBucket("bucket", msg) + assert.NoError(t, err) +} diff --git a/pkg/proc/interaction.go b/pkg/proc/interaction.go index 8779b10..5553a65 100644 --- a/pkg/proc/interaction.go +++ b/pkg/proc/interaction.go @@ -287,7 +287,7 @@ func ProcessCopyExtended( for len(objectMetas) > 0 && retryCount < 10 { retryCount++ - sem := semaphore.NewWeighted(config.InstanceConfig().StorageCnf.CopyStorageConcurrency) + sem := semaphore.NewWeighted(sourceInstanceCnf.StorageCnf.CopyStorageConcurrency) wg := sync.WaitGroup{} @@ -919,7 +919,7 @@ func ListFilesToCopy(prefix string, port uint64, cfg config.Storage, src storage path := strings.TrimPrefix(objectMetas[i].Path, cfg.StoragePrefix) reworked := path if _, ok := vi[reworked]; !ok { - skipCopy := config.InstanceConfig().StorageCnf.StorageOptimizeCopy + skipCopy := cfg.StorageOptimizeCopy skipped = append(skipped, objectMetas[i]) ylogger.Zero.Info().Int("index", i).Str("reworked name", reworked).Str("object path", objectMetas[i].Path).Bool("skipping", skipCopy).Msg("not in virtual index") From 6bb759125ae01d7b4ce15d71d1db36d794b405bc Mon Sep 17 00:00:00 2001 From: Vlasdislav Date: Thu, 9 Jul 2026 14:07:03 +0300 Subject: [PATCH 6/8] Fix: Unit-test for DeleteGarbageInBucket with awkward logic --- pkg/proc/delete_handler.go | 12 ++++++++---- pkg/proc/delete_handler_test.go | 8 +++++++- pkg/proc/interaction.go | 2 +- 3 files changed, 16 insertions(+), 6 deletions(-) diff --git a/pkg/proc/delete_handler.go b/pkg/proc/delete_handler.go index 1b04851..55ef694 100644 --- a/pkg/proc/delete_handler.go +++ b/pkg/proc/delete_handler.go @@ -92,6 +92,13 @@ func (dh *BasicGarbageMgr) HandleUntrashifyFile(msg message.UntrashifyMessage) e return nil } +/* + * The design looks an awkward, + * because the function depends on two different vacuum config: + * - the local dh.Cnf + * - the global config.InstanceConfig() + * Example: TestDeleteGarbageInBucketMovesObjectsWhenCrazyDropDisabled + */ func (dh *BasicGarbageMgr) DeleteGarbageInBucket(bucket string, msg message.DeleteMessage) error { fileList, err := dh.ListGarbageFiles(bucket, msg) if err != nil { @@ -119,10 +126,7 @@ func (dh *BasicGarbageMgr) DeleteGarbageInBucket(bucket string, msg message.Dele * Burst at 20% of vacuum rate capacity. It is pretty arbitrary at this time, * but its not like something we need config field for... */ - limRate := dh.Cnf.FileChunkPerSec - if limRate <= 0 { - limRate = config.DefaultFileChunkPerSec - } + limRate := config.InstanceConfig().VacuumCnf.FileChunkPerSec limiter := rate.NewLimiter(rate.Limit(limRate), limRate/5) ctx := context.Background() diff --git a/pkg/proc/delete_handler_test.go b/pkg/proc/delete_handler_test.go index d20c492..73939e1 100644 --- a/pkg/proc/delete_handler_test.go +++ b/pkg/proc/delete_handler_test.go @@ -350,9 +350,15 @@ func TestDeleteGarbageInBucketMovesObjectsWhenCrazyDropDisabled(t *testing.T) { StorageInterractor: storage, DbInterractor: database, BackupInterractor: nil, - Cnf: &config.Vacuum{CheckBackup: false, FileChunkPerSec: 1000}, + Cnf: &config.InstanceConfig().VacuumCnf, } + cnfBackup := *config.InstanceConfig() + defer func() { + *config.InstanceConfig() = cnfBackup + }() + config.EmbedDefaults(config.InstanceConfig()) + err := handler.DeleteGarbageInBucket("bucket", msg) assert.NoError(t, err) } diff --git a/pkg/proc/interaction.go b/pkg/proc/interaction.go index 5553a65..c9c7c64 100644 --- a/pkg/proc/interaction.go +++ b/pkg/proc/interaction.go @@ -919,7 +919,7 @@ func ListFilesToCopy(prefix string, port uint64, cfg config.Storage, src storage path := strings.TrimPrefix(objectMetas[i].Path, cfg.StoragePrefix) reworked := path if _, ok := vi[reworked]; !ok { - skipCopy := cfg.StorageOptimizeCopy + skipCopy := config.InstanceConfig().StorageCnf.StorageOptimizeCopy skipped = append(skipped, objectMetas[i]) ylogger.Zero.Info().Int("index", i).Str("reworked name", reworked).Str("object path", objectMetas[i].Path).Bool("skipping", skipCopy).Msg("not in virtual index") From 4fc5e85242e124e440e155464f3ca5bca4d64b8c Mon Sep 17 00:00:00 2001 From: Vlasdislav Date: Thu, 9 Jul 2026 14:09:17 +0300 Subject: [PATCH 7/8] Chore: Update comment style --- pkg/proc/yio/yrreader.go | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/pkg/proc/yio/yrreader.go b/pkg/proc/yio/yrreader.go index 653cf34..404fffc 100644 --- a/pkg/proc/yio/yrreader.go +++ b/pkg/proc/yio/yrreader.go @@ -39,7 +39,6 @@ func (y *YRestartReader) Close() error { // Read implements RestartReader. func (y *YRestartReader) Read(p []byte) (n int, err error) { /* read with rate limiter */ - return y.underlying.Read(p) } @@ -48,8 +47,10 @@ func NewRestartReader(s storage.StorageInteractor, l := limiter.GetLimiter() - /* due to storage config "enable limiter" can change on read-restart, allocate - * limiter unconditionally */ + /* + * Due to storage config "enable limiter" can change on read-restart, + * allocate limiter unconditionally + */ return &YRestartReader{ s: s, name: name, From 380824a3a04d14918e5c517484e35ad80e0aae79 Mon Sep 17 00:00:00 2001 From: Vlasdislav Date: Thu, 9 Jul 2026 14:25:08 +0300 Subject: [PATCH 8/8] Fix: Unit-test semantic and call make fmt --- pkg/proc/delete_handler_test.go | 10 +++++----- pkg/proc/yio/yrreader.go | 2 +- 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/pkg/proc/delete_handler_test.go b/pkg/proc/delete_handler_test.go index 73939e1..4443513 100644 --- a/pkg/proc/delete_handler_test.go +++ b/pkg/proc/delete_handler_test.go @@ -335,10 +335,10 @@ func TestDeleteGarbageInBucketMovesObjectsWhenCrazyDropDisabled(t *testing.T) { } storage := mock.NewMockStorageInteractor(ctrl) - storage.EXPECT().ListBucketPath("bucket", msg.Name, true).Return(filesInStorage, nil) - storage.EXPECT().ListFailedMultipartUploads("bucket").Return(map[string]string{}, nil) - storage.EXPECT().MoveObject("bucket", filesInStorage[0].Path, proc.TrashPathFromRegPath(filesInStorage[0].Path, int(msg.Segnum))).Return(nil) - storage.EXPECT().MoveObject("bucket", filesInStorage[1].Path, proc.TrashPathFromRegPath(filesInStorage[1].Path, int(msg.Segnum))).Return(nil) + storage.EXPECT().ListBucketPath("trash", msg.Name, true).Return(filesInStorage, nil) + storage.EXPECT().ListFailedMultipartUploads("trash").Return(map[string]string{}, nil) + storage.EXPECT().MoveObject("trash", filesInStorage[0].Path, proc.TrashPathFromRegPath(filesInStorage[0].Path, int(msg.Segnum))).Return(nil) + storage.EXPECT().MoveObject("trash", filesInStorage[1].Path, proc.TrashPathFromRegPath(filesInStorage[1].Path, int(msg.Segnum))).Return(nil) database := mock.NewMockDatabaseInterractor(ctrl) database.EXPECT().GetVirtualExpireIndexes(msg.Port).Return(map[string]bool{}, map[string]uint64{ @@ -359,6 +359,6 @@ func TestDeleteGarbageInBucketMovesObjectsWhenCrazyDropDisabled(t *testing.T) { }() config.EmbedDefaults(config.InstanceConfig()) - err := handler.DeleteGarbageInBucket("bucket", msg) + err := handler.DeleteGarbageInBucket("trash", msg) assert.NoError(t, err) } diff --git a/pkg/proc/yio/yrreader.go b/pkg/proc/yio/yrreader.go index 404fffc..f72e4db 100644 --- a/pkg/proc/yio/yrreader.go +++ b/pkg/proc/yio/yrreader.go @@ -47,7 +47,7 @@ func NewRestartReader(s storage.StorageInteractor, l := limiter.GetLimiter() - /* + /* * Due to storage config "enable limiter" can change on read-restart, * allocate limiter unconditionally */