vacuum garbage: abort dangling multipart uploads (#70)

* abort dangling multipart uploads

* check running multipart uploads in yproxy and s3

pkg/proc/delete_handler.go

@@ -33,6 +33,11 @@ func (dh *BasicDeleteHandler) HandleDeleteGarbage(msg message.DeleteMessage) err
 	if err != nil {
 		return errors.Wrap(err, "failed to delete file")
 	}
+	uploads, err := dh.StorageInterractor.ListFailedMultipartUploads()
+	if err != nil {
+		return err
+	}
+	ylogger.Zero.Info().Int("amount", len(uploads)).Msg("multipart uploads will be aborted")
 
 	if !msg.Confirm { //do not delete files if no confirmation flag provided
 		return nil
@@ -76,6 +81,12 @@ func (dh *BasicDeleteHandler) HandleDeleteGarbage(msg message.DeleteMessage) err
 		return errors.Wrap(err, "failed to move some files")
 	}
 
+	for key, uploadId := range uploads {
+		if err := dh.StorageInterractor.AbortMultipartUpload(key, uploadId); err != nil {
+			return err
+		}
+	}
+
 	return nil
 }
 

pkg/storage/filestorage.go

@@ -85,3 +85,7 @@ func (s *FileStorageInteractor) MoveObject(from string, to string) error {
 func (s *FileStorageInteractor) DeleteObject(key string) error {
 	return os.Remove(path.Join(s.cnf.StoragePrefix, key))
 }
+
+func (s *FileStorageInteractor) AbortMultipartUploads() error {
+	return nil
+}

pkg/storage/s3storage.go

@@ -8,6 +8,7 @@ import (
 	"path"
 	"strconv"
 	"strings"
+	"sync"
 
 	"github.com/yezzey-gp/aws-sdk-go/aws"
 	"github.com/yezzey-gp/aws-sdk-go/service/s3"
@@ -27,7 +28,8 @@ type S3StorageInteractor struct {
 
 	cnf *config.Storage
 
-	bucketMap map[string]string
+	bucketMap        map[string]string
+	multipartUploads sync.Map
 }
 
 func (s *S3StorageInteractor) CatFileFromStorage(name string, offset int64, setts []settings.StorageSettings) (io.ReadCloser, error) {
@@ -96,6 +98,7 @@ func (s *S3StorageInteractor) PutFileToDest(name string, r io.Reader, settings [
 	}
 
 	if multipartUpload {
+		s.multipartUploads.Store(objectPath, true)
 		_, err = up.Upload(
 			&s3manager.UploadInput{
 				Bucket:       aws.String(bucket),
@@ -104,6 +107,7 @@ func (s *S3StorageInteractor) PutFileToDest(name string, r io.Reader, settings [
 				StorageClass: aws.String(storageClass),
 			},
 		)
+		s.multipartUploads.Delete(objectPath)
 	} else {
 		var body []byte
 		body, err = io.ReadAll(r)
@@ -249,3 +253,52 @@ func (s *S3StorageInteractor) MoveObject(from string, to string) error {
 	}
 	return s.DeleteObject(from)
 }
+
+func (s *S3StorageInteractor) AbortMultipartUpload(key, uploadId string) error {
+	sess, err := s.pool.GetSession(context.TODO())
+	if err != nil {
+		return err
+	}
+
+	_, err = sess.AbortMultipartUpload(&s3.AbortMultipartUploadInput{
+		Bucket:   aws.String(s.cnf.StorageBucket),
+		UploadId: aws.String(uploadId),
+		Key:      aws.String(key),
+	})
+	return err
+}
+
+func (s *S3StorageInteractor) ListFailedMultipartUploads() (map[string]string, error) {
+	sess, err := s.pool.GetSession(context.TODO())
+	if err != nil {
+		return nil, err
+	}
+
+	uploads := make([]*s3.MultipartUpload, 0)
+	var keyMarker *string
+	for {
+		out, err := sess.ListMultipartUploads(&s3.ListMultipartUploadsInput{
+			Bucket:    aws.String(s.cnf.StorageBucket),
+			KeyMarker: keyMarker,
+		})
+		if err != nil {
+			return nil, err
+		}
+
+		uploads = append(uploads, out.Uploads...)
+
+		if !*out.IsTruncated {
+			break
+		}
+
+		keyMarker = out.NextKeyMarker
+	}
+
+	out := make(map[string]string)
+	for _, upload := range uploads {
+		if _, ok := s.multipartUploads.Load(*upload.Key); !ok {
+			out[*upload.Key] = *upload.UploadId
+		}
+	}
+	return out, nil
+}

pkg/storage/storage.go

@@ -21,11 +21,13 @@ type StorageWriter interface {
 
 type StorageLister interface {
 	ListPath(prefix string) ([]*object.ObjectInfo, error)
+	ListFailedMultipartUploads() (map[string]string, error)
 }
 
 type StorageMover interface {
 	MoveObject(from string, to string) error
 	DeleteObject(key string) error
+	AbortMultipartUpload(key, uploadId string) error
 }
 
 //go:generate mockgen -destination=pkg/mock/storage.go -package=mock