abort dangling multipart uploads

pkg/proc/delete_handler.go

@@ -76,6 +76,10 @@ func (dh *BasicDeleteHandler) HandleDeleteGarbage(msg message.DeleteMessage) err
 		return errors.Wrap(err, "failed to move some files")
 	}
 
+	if err := dh.StorageInterractor.AbortMultipartUploads(); 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

@@ -249,3 +249,43 @@ func (s *S3StorageInteractor) MoveObject(from string, to string) error {
 	}
 	return s.DeleteObject(from)
 }
+
+func (s *S3StorageInteractor) AbortMultipartUploads() error {
+	sess, err := s.pool.GetSession(context.TODO())
+	if err != nil {
+		return 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 err
+		}
+
+		uploads = append(uploads, out.Uploads...)
+
+		if !*out.IsTruncated {
+			break
+		}
+
+		keyMarker = out.NextKeyMarker
+	}
+
+	for i := 0; i < len(uploads); i++ {
+		_, err := sess.AbortMultipartUpload(&s3.AbortMultipartUploadInput{
+			Bucket:   aws.String(s.cnf.StorageBucket),
+			UploadId: uploads[i].UploadId,
+			Key:      uploads[i].Key,
+		})
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}

pkg/storage/storage.go

@@ -26,6 +26,7 @@ type StorageLister interface {
 type StorageMover interface {
 	MoveObject(from string, to string) error
 	DeleteObject(key string) error
+	AbortMultipartUploads() error
 }
 
 //go:generate mockgen -destination=pkg/mock/storage.go -package=mock