s3: batch published pool cleanup

AUTHORS

@@ -84,3 +84,4 @@ List of contributors, in chronological order:
 * Zhang Xiao (https://github.com/xzhang1)
 * Tom Nguyen (https://github.com/lecafard)
 * Philip Cramer (https://github.com/PhilipCramer)
+* Chris J Arges (https://github.com/arges)

aptly/interfaces.go

@@ -87,6 +87,12 @@ type PublishedStorage interface {
 	ReadLink(path string) (string, error)
 }
 
+// PublishedStorageBulkRemover optionally removes multiple files efficiently.
+type PublishedStorageBulkRemover interface {
+	// RemoveFiles removes multiple files under public path.
+	RemoveFiles(paths []string) error
+}
+
 // FileSystemPublishedStorage is published storage on filesystem
 type FileSystemPublishedStorage interface {
 	// PublicPath returns root of public part

deb/publish.go

@@ -1769,12 +1769,20 @@ func (collection *PublishedRepoCollection) CleanupPrefixComponentFiles(published
 		sort.Strings(existingFiles)
 
 		orphanedFiles := utils.StrSlicesSubstract(existingFiles, referencedFiles[component])
+		for i := range orphanedFiles {
+			orphanedFiles[i] = filepath.Join(path, orphanedFiles[i])
+		}
 
-		for _, file := range orphanedFiles {
-			err = publishedStorage.Remove(filepath.Join(path, file))
-			if err != nil {
+		if bulkRemover, ok := publishedStorage.(aptly.PublishedStorageBulkRemover); ok {
+			if err = bulkRemover.RemoveFiles(orphanedFiles); err != nil {
 				return err
 			}
+		} else {
+			for _, file := range orphanedFiles {
+				if err = publishedStorage.Remove(file); err != nil {
+					return err
+				}
+			}
 		}
 	}
 

deb/publish_test.go

@@ -70,6 +70,26 @@ func (p *FakeStorageProvider) GetPublishedStorage(name string) (aptly.PublishedS
 	return storage, nil
 }
 
+type recordingBulkPublishedStorage struct {
+	aptly.PublishedStorage
+	removed []string
+	err     error
+}
+
+func (s *recordingBulkPublishedStorage) RemoveFiles(paths []string) error {
+	s.removed = append(s.removed, paths...)
+	return s.err
+}
+
+type failingRemovePublishedStorage struct {
+	aptly.PublishedStorage
+	err error
+}
+
+func (s *failingRemovePublishedStorage) Remove(string) error {
+	return s.err
+}
+
 type PublishedRepoSuite struct {
 	PackageListMixinSuite
 	repo, repo2, repo3, repo4, repo5    *PublishedRepo
@@ -1028,6 +1048,35 @@ func (s *PublishedRepoRemoveSuite) TestRemoveFilesWithPrefixRoot(c *C) {
 	c.Check(filepath.Join(s.publishedStorage2.PublicPath(), "ppa/pool/contrib"), PathExists)
 }
 
+func (s *PublishedRepoRemoveSuite) TestCleanupPrefixComponentFilesRemovalRouting(c *C) {
+	orphan := filepath.Join("ppa", "pool", "main", "orphan.deb")
+	orphanPath := filepath.Join(s.publishedStorage.PublicPath(), orphan)
+	c.Assert(os.WriteFile(orphanPath, []byte("orphan"), 0644), IsNil)
+
+	err := s.collection.CleanupPrefixComponentFiles(s.provider, s.repo1, []string{"main"}, s.factory, nil)
+	c.Check(err, IsNil)
+	c.Check(orphanPath, Not(PathExists))
+
+	c.Assert(os.WriteFile(orphanPath, []byte("orphan"), 0644), IsNil)
+	bulkStorage := &recordingBulkPublishedStorage{PublishedStorage: s.publishedStorage}
+	s.provider.storages[""] = bulkStorage
+
+	err = s.collection.CleanupPrefixComponentFiles(s.provider, s.repo1, []string{"main"}, s.factory, nil)
+	c.Check(err, IsNil)
+	c.Check(bulkStorage.removed, DeepEquals, []string{orphan})
+
+	bulkStorage.err = errors.New("bulk removal failed")
+	err = s.collection.CleanupPrefixComponentFiles(s.provider, s.repo1, []string{"main"}, s.factory, nil)
+	c.Check(err, ErrorMatches, "bulk removal failed")
+
+	s.provider.storages[""] = &failingRemovePublishedStorage{
+		PublishedStorage: s.publishedStorage,
+		err:              errors.New("single removal failed"),
+	}
+	err = s.collection.CleanupPrefixComponentFiles(s.provider, s.repo1, []string{"main"}, s.factory, nil)
+	c.Check(err, ErrorMatches, "single removal failed")
+}
+
 func (s *PublishedRepoRemoveSuite) TestRemoveRepo1and2(c *C) {
 	err := s.collection.Remove(s.provider, "", "ppa", "anaconda", s.factory, nil, false, false)
 	c.Check(err, IsNil)

s3/public.go

@@ -58,9 +58,13 @@ type PublishedStorage struct {
 	encryptByDefault bool
 }
 
+// Amazon S3 accepts at most 1,000 keys in a DeleteObjects request.
+const maxDeleteObjectsPerRequest = 1000
+
 // Check interface
 var (
-	_ aptly.PublishedStorage = (*PublishedStorage)(nil)
+	_ aptly.PublishedStorage            = (*PublishedStorage)(nil)
+	_ aptly.PublishedStorageBulkRemover = (*PublishedStorage)(nil)
 )
 
 // NewPublishedStorageRaw creates published storage from raw aws credentials
@@ -262,7 +266,7 @@ func (storage *PublishedStorage) Remove(path string) error {
 
 // RemoveDirs removes directory structure under public path
 func (storage *PublishedStorage) RemoveDirs(path string, _ aptly.Progress) error {
-	const page = 1000
+	const page = maxDeleteObjectsPerRequest
 
 	filelist, _, err := storage.internalFilelist(path, false)
 	if err != nil {
@@ -330,6 +334,76 @@ func (storage *PublishedStorage) RemoveDirs(path string, _ aptly.Progress) error
 	return nil
 }
 
+// RemoveFiles removes multiple files under public path.
+func (storage *PublishedStorage) RemoveFiles(paths []string) error {
+	if storage.disableMultiDel {
+		for _, path := range paths {
+			if err := storage.Remove(path); err != nil {
+				return err
+			}
+		}
+		return nil
+	}
+
+	if storage.plusWorkaround {
+		expanded := make([]string, 0, len(paths))
+		for _, path := range paths {
+			expanded = append(expanded, path)
+			if strings.Contains(path, "+") {
+				expanded = append(expanded, strings.ReplaceAll(path, "+", " "))
+			}
+		}
+		paths = expanded
+	}
+
+	var failures []string
+	for offset := 0; offset < len(paths); offset += maxDeleteObjectsPerRequest {
+		part := paths[offset:min(offset+maxDeleteObjectsPerRequest, len(paths))]
+		objects := make([]types.ObjectIdentifier, len(part))
+		for i, path := range part {
+			objects[i] = types.ObjectIdentifier{
+				Key: aws.String(filepath.Join(storage.prefix, path)),
+			}
+		}
+
+		quiet := true
+		output, err := storage.s3.DeleteObjects(context.TODO(), &s3.DeleteObjectsInput{
+			Bucket: aws.String(storage.bucket),
+			Delete: &types.Delete{
+				Objects: objects,
+				Quiet:   &quiet,
+			},
+		})
+		if err != nil {
+			var notFoundErr *smithy.GenericAPIError
+			if errors.As(err, &notFoundErr) && notFoundErr.Code == "NoSuchBucket" {
+				return nil
+			}
+			return fmt.Errorf("error deleting multiple paths from %s: %s", storage, err)
+		}
+		failed := make(map[string]struct{}, len(output.Errors))
+		for _, failure := range output.Errors {
+			key := aws.ToString(failure.Key)
+			failed[key] = struct{}{}
+			failures = append(failures, fmt.Sprintf("%s: %s: %s", key,
+				aws.ToString(failure.Code), aws.ToString(failure.Message)))
+		}
+		storage.pathCacheMutex.Lock()
+		for _, path := range part {
+			if _, failed := failed[filepath.Join(storage.prefix, path)]; !failed {
+				delete(storage.pathCache, path)
+			}
+		}
+		storage.pathCacheMutex.Unlock()
+
+	}
+	if len(failures) != 0 {
+		return fmt.Errorf("errors deleting multiple paths from %s: %s", storage, strings.Join(failures, "; "))
+	}
+
+	return nil
+}
+
 // LinkFromPool links package file from pool to dist's pool location
 //
 // publishedPrefix is desired prefix for the location in the pool.

s3/public_test.go

@@ -3,6 +3,7 @@ package s3
 import (
 	"bytes"
 	"context"
+	"fmt"
 	"io"
 	"os"
 	"path/filepath"
@@ -250,6 +251,155 @@ func (s *PublishedStorageSuite) TestRemoveDirsNoSuchBucket(c *C) {
 	c.Check(err, ErrorMatches, ".*StatusCode: 404.*")
 }
 
+func (s *PublishedStorageSuite) countRequests(method, uriSubstring string) int {
+	count := 0
+	for _, r := range s.srv.Requests {
+		if r.Method == method && strings.Contains(r.RequestURI, uriSubstring) {
+			count++
+		}
+	}
+
+	return count
+}
+
+func (s *PublishedStorageSuite) TestRemoveFilesPrefixed(c *C) {
+	s.prefixedStorage.disableMultiDel = false
+
+	s.PutFile(c, "lala/xyz", []byte("test"))
+	s.PutFile(c, "lala/abc", []byte("test"))
+
+	err := s.prefixedStorage.RemoveFiles([]string{"xyz"})
+	c.Check(err, IsNil)
+
+	s.AssertNoFile(c, "lala/xyz")
+
+	list, err := s.storage.Filelist("")
+	c.Check(err, IsNil)
+	c.Check(list, DeepEquals, []string{"lala/abc"})
+	c.Check(s.countRequests("POST", "delete"), Equals, 1)
+	c.Check(s.countRequests("DELETE", ""), Equals, 0)
+}
+
+func (s *PublishedStorageSuite) TestRemoveFilesBatchesAtThousand(c *C) {
+	s.storage.disableMultiDel = false
+
+	paths := make([]string, maxDeleteObjectsPerRequest+1)
+	for i := range paths {
+		paths[i] = fmt.Sprintf("pool/main/p/pkg/file-%d.deb", i)
+	}
+
+	err := s.storage.RemoveFiles(paths)
+	c.Check(err, IsNil)
+
+	c.Check(s.countRequests("POST", "delete"), Equals, 2)
+	c.Check(s.countRequests("DELETE", ""), Equals, 0)
+}
+
+func (s *PublishedStorageSuite) TestRemoveFilesEmpty(c *C) {
+	s.storage.disableMultiDel = false
+
+	err := s.storage.RemoveFiles(nil)
+	c.Check(err, IsNil)
+	c.Check(s.countRequests("POST", "delete"), Equals, 0)
+}
+
+func (s *PublishedStorageSuite) TestRemoveFilesNoSuchBucket(c *C) {
+	s.noSuchBucketStorage.disableMultiDel = false
+
+	err := s.noSuchBucketStorage.RemoveFiles([]string{"a"})
+	c.Check(err, IsNil)
+}
+
+func (s *PublishedStorageSuite) TestRemoveFilesRequestError(c *C) {
+	s.storage.disableMultiDel = false
+	s.srv.config.DeleteObjectsError = "multi-delete failed"
+
+	err := s.storage.RemoveFiles([]string{"a"})
+	c.Check(err, ErrorMatches, "error deleting multiple paths.*AccessDenied.*multi-delete failed.*")
+}
+
+func (s *PublishedStorageSuite) TestRemoveFilesReportsAllFailuresAndInvalidatesSuccesses(c *C) {
+	s.prefixedStorage.disableMultiDel = false
+	s.srv.config.DeleteErrors = map[string]string{
+		"lala/a": "a failed",
+		"lala/c": "c failed",
+	}
+	s.prefixedStorage.pathCache = map[string]string{"a": "", "b": "", "c": ""}
+	for _, path := range []string{"a", "b", "c"} {
+		s.PutFile(c, filepath.Join("lala", path), []byte("test"))
+	}
+
+	err := s.prefixedStorage.RemoveFiles([]string{"a", "b", "c"})
+	c.Check(err, ErrorMatches, "errors deleting multiple paths.*lala/a: AccessDenied: a failed; lala/c: AccessDenied: c failed")
+
+	s.AssertNoFile(c, "lala/b")
+	c.Check(s.prefixedStorage.pathCache, DeepEquals, map[string]string{"a": "", "c": ""})
+}
+
+func (s *PublishedStorageSuite) TestRemoveFilesReportsFailuresAcrossBatches(c *C) {
+	s.storage.disableMultiDel = false
+	paths := make([]string, maxDeleteObjectsPerRequest+1)
+	for i := range paths {
+		paths[i] = fmt.Sprintf("file-%d", i)
+	}
+	s.srv.config.DeleteErrors = map[string]string{
+		paths[0]:                          "first batch failed",
+		paths[maxDeleteObjectsPerRequest]: "second batch failed",
+	}
+
+	err := s.storage.RemoveFiles(paths)
+	c.Check(err, ErrorMatches, "errors deleting multiple paths.*file-0: AccessDenied: first batch failed; file-1000: AccessDenied: second batch failed")
+	c.Check(s.countRequests("POST", "delete"), Equals, 2)
+}
+
+func (s *PublishedStorageSuite) TestRemoveFilesDisableMultiDel(c *C) {
+	s.storage.disableMultiDel = true
+
+	paths := []string{"a", "b", "c"}
+	for _, path := range paths {
+		s.PutFile(c, path, []byte("test"))
+	}
+
+	err := s.storage.RemoveFiles([]string{"a", "b"})
+	c.Check(err, IsNil)
+
+	list, err := s.storage.Filelist("")
+	c.Check(err, IsNil)
+	c.Check(list, DeepEquals, []string{"c"})
+
+	// Endpoints that cannot multi-delete fall back to one request per file.
+	c.Check(s.countRequests("POST", "delete"), Equals, 0)
+	c.Check(s.countRequests("DELETE", ""), Equals, 2)
+}
+
+func (s *PublishedStorageSuite) TestRemoveFilesDisableMultiDelError(c *C) {
+	s.storage.disableMultiDel = true
+	s.srv.config.DeleteObjectErrors = map[string]string{"a": "delete failed"}
+
+	err := s.storage.RemoveFiles([]string{"a"})
+	c.Check(err, ErrorMatches, "error deleting a.*AccessDenied.*delete failed.*")
+}
+
+func (s *PublishedStorageSuite) TestRemoveFilesPlusWorkaround(c *C) {
+	s.storage.disableMultiDel = false
+	s.storage.plusWorkaround = true
+
+	s.PutFile(c, "a/b+c", []byte("test"))
+	s.PutFile(c, "a/b", []byte("test"))
+
+	// Filelist hides the space-substituted duplicate, so RemoveFiles has to
+	// expand it the way Remove does or it would be orphaned forever.
+	err := s.storage.RemoveFiles([]string{"a/b+c"})
+	c.Check(err, IsNil)
+
+	s.AssertNoFile(c, "a/b+c")
+	s.AssertNoFile(c, "a/b c")
+
+	list, err := s.storage.Filelist("")
+	c.Check(err, IsNil)
+	c.Check(list, DeepEquals, []string{"a/b"})
+}
+
 func (s *PublishedStorageSuite) TestRenameFile(c *C) {
 	c.Skip("copy not available in s3test")
 }