Make mirror_file fail if file object already exists (#7134)

Author: nadove-ucsc

src/azul/indexer/mirror_controller.py

@@ -152,8 +152,13 @@ def mirror_file(self,
         deployment_is_stable = (config.deployment.is_stable
                                 and not config.deployment.is_unit_test
                                 and catalog not in config.integration_test_catalogs)
+
         if file_is_large and not deployment_is_stable:
             log.info('Not mirroring file to save cost: %r', file)
+        elif self.service.info_exists(catalog, file):
+            log.info('File is already mirrored, skipping upload: %r', file)
+        elif self.service.file_exists(catalog, file):
+            assert False, R('File object is already present', file)
         else:
             # Ensure we test with multiple parts on lower deployments
             part_size = FilePart.default_size if deployment_is_stable else FilePart.min_size

src/azul/indexer/mirror_service.py

@@ -194,6 +194,14 @@ def info_object_key(self, file: File) -> str:
     def info_exists(self, catalog: CatalogName, file: File) -> bool:
         return self._get_info(catalog, file) is not None
 
+    def file_exists(self, catalog: CatalogName, file: File) -> bool:
+        try:
+            self._storage(catalog).head(self.mirror_object_key(file))
+        except StorageObjectNotFound:
+            return False
+        else:
+            return True
+
     def _file_key(self, prefix: str, file: File, *, extension: str = '') -> str:
         digest = file.digest
         assert all(c in string.hexdigits for c in digest.value), R(
@@ -214,17 +222,15 @@ def mirror_file(self, catalog: CatalogName, file: File):
         Upload the file in a single request. For larger files, use
         :meth:`begin_mirroring_file` instead.
         """
-        if self.info_exists(catalog, file):
-            log.info('File is already mirrored, skipping upload: %r', file)
-        else:
-            file_content = self._download(catalog, file)
-            self._storage(catalog).put(object_key=self.mirror_object_key(file),
-                                       data=file_content,
-                                       content_type=file.content_type)
-            hasher = get_resumable_hasher(file.digest.type)
-            hasher.update(file_content)
-            self._verify_digest(file, hasher)
-            self._put_info(catalog, file)
+        file_content = self._download(catalog, file)
+        self._storage(catalog).put(object_key=self.mirror_object_key(file),
+                                   data=file_content,
+                                   content_type=file.content_type,
+                                   overwrite=False)
+        hasher = get_resumable_hasher(file.digest.type)
+        hasher.update(file_content)
+        self._verify_digest(file, hasher)
+        self._put_info(catalog, file)
 
     def begin_mirroring_file(self, catalog: CatalogName, file: File) -> str:
         """
@@ -268,7 +274,9 @@ def finish_mirroring_file(self,
         Complete a multipart upload begun with :meth:`begin_mirroring_file`.
         """
         upload = self._get_upload(catalog, file, upload_id)
-        self._storage(catalog).complete_multipart_upload(upload, etags)
+        self._storage(catalog).complete_multipart_upload(upload,
+                                                         etags,
+                                                         overwrite=False)
         self._verify_digest(file, hasher)
         self._get_info(catalog, file)
         self._put_info(catalog, file)

test/indexer/test_mirror_controller.py

@@ -75,7 +75,7 @@ def test_mirroring(self):
                 with self.subTest('mirror_partition'):
                     file, file_message = self._test_mirror_partition(partition_message)
 
-                    with self.subTest('mirror_file', corrupted=False):
+                    with self.subTest('mirror_file', corrupted=False, exists=False):
                         self._test_mirror_file(file, file_message)
 
                     self._s3.delete_object(Bucket=self.mirror_bucket,
@@ -84,6 +84,9 @@ def test_mirroring(self):
                     with self.subTest('mirror_file', corrupted=True):
                         self._test_corrupted_download(file_message)
 
+                    with self.subTest('mirror_file', corrupted=False, exists=True):
+                        self._test_reuploaded_file(file_message)
+
     _file_contents = b'lorem ipsum dolor sit\n'
 
     @property
@@ -151,6 +154,14 @@ def _test_corrupted_download(self, file_message):
                 self.mirror_controller.mirror(event)
             self.assertTrue(R.caused(e.exception))
 
+    def _test_reuploaded_file(self, file_message):
+        event = [self._mock_sqs_record(file_message)]
+        with patch.object(MirrorService, '_download', return_value=self._file_contents):
+            with self.assertRaises(AssertionError) as e:
+                self.mirror_controller.mirror(event)
+        self.assertTrue(R.caused(e.exception))
+        self.assertEqual(e.exception.args[0].args[0], 'File object is already present')
+
     def test_info_schema(self):
         client = http_client(log)
         file = MagicMock(content_type='text/plain')