GC: Collect resources in versioned bucket (#271)

Author: psrok1

karton/core/backend.py

@@ -986,6 +986,21 @@ def list_objects(self, bucket: str) -> List[str]:
                 objs.append(obj["Key"])
         return objs
 
+    def list_object_versions(self, bucket: str) -> Dict[str, List[str]]:
+        """
+        List version identifiers of stored resource objects
+        :param bucket: Bucket name
+        :return: Dictionary of object version identifiers {key: [version_ids, ...]}
+        """
+        objs = defaultdict(list)
+        paginator = self.s3.get_paginator("list_object_versions")
+        for page in paginator.paginate(Bucket=bucket):
+            for obj in page.get("Versions", list()):
+                objs[obj["Key"]].append(obj["VersionId"])
+            for obj in page.get("DeleteMarkers", list()):
+                objs[obj["Key"]].append(obj["VersionId"])
+        return dict(objs)
+
     def remove_object(self, bucket: str, object_uid: str) -> None:
         """
         Remove resource object from object storage
@@ -1002,7 +1017,28 @@ def remove_objects(self, bucket: str, object_uids: Iterable[str]) -> None:
         :param bucket: Bucket name
         :param object_uids: Object identifiers
         """
-        for delete_objects in chunks([{"Key": uid} for uid in object_uids], 1000):
+        for delete_objects in chunks([{"Key": uid} for uid in object_uids], 100):
+            self.s3.delete_objects(Bucket=bucket, Delete={"Objects": delete_objects})
+
+    def remove_object_versions(
+        self, bucket: str, object_versions: Dict[str, List[str]]
+    ) -> None:
+        """
+        Bulk remove resource object versions from object storage
+
+        :param bucket: Bucket name
+        :param object_versions: Object version identifiers
+        """
+        versions = iter(
+            (uid, version_id)
+            for uid, versions in object_versions.items()
+            for version_id in versions
+        )
+        deletion_chunks = chunks(
+            [{"Key": uid, "VersionId": version_id} for uid, version_id in versions],
+            100,
+        )
+        for delete_objects in deletion_chunks:
             self.s3.delete_objects(Bucket=bucket, Delete={"Objects": delete_objects})
 
     def check_bucket_exists(self, bucket: str, create: bool = False) -> bool:

karton/system/system.py

@@ -74,7 +74,7 @@ def _log_config(self):
     def gc_collect_resources(self) -> None:
         # Collects unreferenced resources left in object storage
         karton_bucket = self.backend.default_bucket_name
-        resources_to_remove = set(self.backend.list_objects(karton_bucket))
+        resources_to_remove = self.backend.list_object_versions(karton_bucket)
         # Note: it is important to get list of resources before getting list of tasks!
         # Task is created before resource upload to lock the reference to the resource.
         tasks = self.backend.iter_all_tasks()
@@ -85,10 +85,10 @@ def gc_collect_resources(self) -> None:
                     resource.bucket == karton_bucket
                     and resource.uid in resources_to_remove
                 ):
-                    resources_to_remove.remove(resource.uid)
+                    del resources_to_remove[resource.uid]
         # Remove unreferenced resources
         if resources_to_remove:
-            self.backend.remove_objects(karton_bucket, resources_to_remove)
+            self.backend.remove_object_versions(karton_bucket, resources_to_remove)
 
     def gc_collect_tasks(self) -> None:
         self.log.debug("GC: gc_collect_tasks started")