diff --git a/config/instance.go b/config/instance.go index 57d2f78..cb99946 100644 --- a/config/instance.go +++ b/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 } 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..55ef694 100644 --- a/pkg/proc/delete_handler.go +++ b/pkg/proc/delete_handler.go @@ -5,6 +5,7 @@ import ( "fmt" "path" "strings" + "sync" "time" "github.com/pkg/errors" @@ -91,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 { @@ -103,54 +111,69 @@ 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") } - 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... */ + /* + * 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 limiter := rate.NewLimiter(rate.Limit(limRate), limRate/5) ctx := context.Background() - var failed []string + var failedActionMsg, failedFilesMsg string + 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 *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 *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 []*object.ObjectInfo 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.Path).Msg(failedActionMsg) failed = append(failed, file) } } fileList = failed - failed = make([]string, 0) + 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") + 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 { - /* Dont move too fast */ + /* Don't move too fast */ if err := limiter.Wait(ctx); err != nil { break } @@ -171,10 +194,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 { @@ -196,6 +219,52 @@ func (dh *BasicGarbageMgr) ListDelete2Files(bucket string, msg message.Delete2Me return filesToDelete, nil } +func (dh *BasicGarbageMgr) garbageTrashParallel(bucket string, fileList []*object.ObjectInfo) ([]*object.ObjectInfo, 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 *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.Path); err != nil { + ylogger.Zero.Warn().AnErr("err", err).Str("bucket", bucket).Str("file", file.Path).Msg("failed to delete garbage file") + failedCh <- file + } + } + }) + } + + for _, file := range fileList { + jobs <- file + } + close(jobs) + + wg.Wait() + close(failedCh) + + failed := make([]*object.ObjectInfo, 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) DeletePrefixInBucket(bucket string, msg message.Delete2Message) error { fileList, err := dh.ListDelete2Files(bucket, msg) // Return the list of files to be deleted if err != nil { @@ -214,7 +283,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 } @@ -223,26 +292,24 @@ func (dh *BasicGarbageMgr) DeletePrefixInBucket(bucket string, msg message.Delet return nil } trashRetention := time.Hour * 24 * time.Duration(dh.Cnf.TrashRetentionDays) - var failed []*object.ObjectInfo + 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++ { - 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) - } + fileList, err = dh.garbageTrashParallel(bucket, fileList) + 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") + 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") } for key, uploadId := range uploads { @@ -276,8 +343,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 @@ -293,7 +360,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 { @@ -308,7 +375,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") @@ -324,7 +391,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 fd4ccb7..4443513 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" @@ -59,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) { @@ -118,3 +121,244 @@ 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"]) +} + +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("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{ + filesInStorage[0].Path: 0, + filesInStorage[1].Path: 0, + }, nil) + + handler := proc.BasicGarbageMgr{ + StorageInterractor: storage, + DbInterractor: database, + BackupInterractor: nil, + Cnf: &config.InstanceConfig().VacuumCnf, + } + + cnfBackup := *config.InstanceConfig() + defer func() { + *config.InstanceConfig() = cnfBackup + }() + config.EmbedDefaults(config.InstanceConfig()) + + err := handler.DeleteGarbageInBucket("trash", msg) + assert.NoError(t, err) +} diff --git a/pkg/proc/interaction.go b/pkg/proc/interaction.go index 8779b10..c9c7c64 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{} diff --git a/pkg/proc/yio/yrreader.go b/pkg/proc/yio/yrreader.go index 653cf34..f72e4db 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,