Fix: StorageObjectExists error when mirroring anvil14 (#8142)

.github/workflows/ci.yml

@@ -57,6 +57,7 @@ jobs:
           export azul_docker_registry=""
 
           make format
+          make isort
           make -C lambdas openapi
           make -C .github
           make anvil_schema

.gitlab-ci.yml

@@ -77,6 +77,7 @@ test:
   timeout: 1h 30m
   script:
     - make format
+    - make isort
     - python -m azul -t config.deployment.is_lower_sandbox && make requirements_update
     - make -C lambdas openapi
     - make environment.boot

.isort.cfg

@@ -5,4 +5,4 @@ include_trailing_comma=1
 force_sort_within_sections=1
 case_sensitive=True
 combine_as_imports=1
-known_first_party=azul_test_case,opensearch_test_case,indexer,health_check_test_case,app_test_case,sqs_test_case,service,docker_container_test_case,dynamodb_test_case
+known_first_party=azul_test_case,opensearch_test_case,indexer,health_check_test_case,app_test_case,sqs_test_case,service,docker_container_test_case,dynamodb_test_case,urllib3_mock

scripts/pull_request.py

@@ -20,9 +20,6 @@
 import re
 import subprocess
 import sys
-from typing import (
-    Literal,
-)
 
 import attrs
 from furl import (
@@ -34,9 +31,7 @@
 )
 from azul.lib.strings import (
     format_and_dedent as fd,
-)
-from azul.lib.types import (
-    check_type,
+    join_grammatically,
 )
 from azul.logging import (
     configure_script_logging,
@@ -58,15 +53,23 @@ def main(argv):
                         choices=['upgrade', 'promotion'],
                         help='Type of PR to create. '
                              'If omitted, a regular PR is created.')
-    parser.add_argument('--no-partial',
-                        action='store_true', default=False,
-                        help='Remove partial label and check partiality tasks.')
-    parser.add_argument('--no-mirror',
+    section_flags = {
+        'partial': 'Remove partial label and check partiality tasks.',
+        'mirror': 'Remove mirror labels and check mirror tasks.',
+        'reindex': 'Remove reindex labels and check reindex tasks.',
+        'api': 'Remove API label and check API tasks.',
+        'upgrade': 'Remove upgrade label and check upgrade tasks.',
+        'deploy': 'Remove deploy labels and check deploy tasks.',
+        'hotfix': 'Check hotfix tasks.',
+    }
+    for flag, help_text in section_flags.items():
+        parser.add_argument(f'--no-{flag}',
+                            action='store_true', default=False,
+                            help=help_text)
+    all_no_flags = join_grammatically([f'--no-{f}' for f in section_flags])
+    parser.add_argument('--vanilla',
                         action='store_true', default=False,
-                        help='Remove mirror labels and check mirror tasks.')
-    parser.add_argument('--no-reindex',
-                        action='store_true', default=False,
-                        help='Remove reindex labels and check reindex tasks.')
+                        help=f'Equivalent to {all_no_flags}.')
     fix_group = parser.add_mutually_exclusive_group()
     fix_group.add_argument('--fix',
                            action='store_true', default=None,
@@ -75,14 +78,15 @@ def main(argv):
                            action='store_false', dest='fix',
                            help='Do not prefix the PR title with "Fix: ".')
     args = parser.parse_args(argv)
+    if args.vanilla:
+        for flag in section_flags:
+            setattr(args, f'no_{flag}', True)
     if args.type is not None and args.fix is not None:
         parser.error('--fix/--no-fix cannot be used with --type')
-    if args.type is not None and args.no_partial:
-        parser.error('--no-partial cannot be used with --type')
-    if args.type is not None and args.no_mirror:
-        parser.error('--no-mirror cannot be used with --type')
-    if args.type is not None and args.no_reindex:
-        parser.error('--no-reindex cannot be used with --type')
+    if args.type is not None:
+        for flag in section_flags:
+            if getattr(args, f'no_{flag}'):
+                parser.error(f'--no-{flag} cannot be used with --type')
 
     branch = _current_branch()
     _check_working_copy()
@@ -160,18 +164,60 @@ def main(argv):
     body = _check_task(body, r'Added `u` tag to commit title.*', checked=has_u_tag)
     body = _check_task(body, r'This PR is labeled `upgrade`.*', checked=has_u_tag)
 
+    if args.no_upgrade:
+        assert not has_u_tag, R(
+            '--no-upgrade cannot be used when a commit is tagged `u`')
+        body = _check_task(body, r'Ran `make docker_images\.json`.*')
+        body = _check_task(body, r'Documented upgrading of deployments in UPGRADING\.rst.*')
+        body = _check_task(body, r'Added `u` tag to commit title.*')
+        body = _check_task(body, r'This PR is labeled `upgrade`.*')
+
+    labels_to_add = []
+    labels_to_remove = []
+    if has_u_tag:
+        labels_to_add.append('upgrade')
+    else:
+        labels_to_remove.append('upgrade')
+    if args.no_partial:
+        labels_to_remove.append('partial')
+    if args.no_api:
+        labels_to_remove.append('API')
+
+    if args.no_deploy:
+        deploy_labels = _deploy_labels(body)
+        for label in deploy_labels:
+            body = _check_task(body, r'This PR is labeled `' + re.escape(label) + '`.*')
+        labels_to_remove.extend(deploy_labels)
+
     if args.no_reindex:
         assert not _has_commit_tag(target_branch, 'r'), R(
             '--no-reindex cannot be used when a commit is tagged `r`')
         body = _check_task(body, r'Added `r` tag to commit title.*')
         reindex_labels = _reindex_labels(body)
         for label in reindex_labels:
             body = _check_task(body, r'This PR is labeled `' + re.escape(label) + '`.*')
+        labels_to_remove.extend(reindex_labels)
 
     if args.no_mirror:
         mirror_labels = _mirror_labels(body)
         for label in mirror_labels:
             body = _check_task(body, r'This PR is labeled `' + re.escape(label) + '`.*')
+        labels_to_remove.extend(mirror_labels)
+
+    if args.no_api:
+        assert not _has_commit_tag(target_branch, 'a'), R(
+            '--no-api cannot be used when a commit is tagged `a`')
+        assert not _has_commit_tag(target_branch, 'A'), R(
+            '--no-api cannot be used when a commit is tagged `A`')
+        body = _check_task(body, r'This PR and its linked issues are labeled `API`.*')
+        body = _check_task(body, r'Added `a` \(`A`\) tag to commit title.*')
+        body = _check_task(body, r'Updated REST API version number in `app\.py`.*')
+
+    if args.no_hotfix:
+        assert not _has_commit_tag(target_branch, 'F'), R(
+            '--no-hotfix cannot be used when a commit is tagged `F`')
+        body = _check_task(body, r'Added `F` tag to main commit title.*')
+        body = _check_task(body, r'Reverted the temporary hotfixes for any linked issues.*')
 
     body = _check_task(body, 'PR is assigned to the author')
     body = _check_task(body, r'Status of PR is \*In progress\*')
@@ -188,16 +234,16 @@ def main(argv):
 
     if existing_pr is None:
         log.info('Creating PR …')
-        result = subprocess.run(
-            [
-                'gh', 'pr', 'create',
-                '--base', target_branch,
-                '--title', title,
-                '--body', body,
-                '--assignee', '@me',
-            ],
-            capture_output=True, text=True
-        )
+        cmd = [
+            'gh', 'pr', 'create',
+            '--base', target_branch,
+            '--title', title,
+            '--body', body,
+            '--assignee', '@me',
+        ]
+        if labels_to_add:
+            cmd.extend(['--label', ','.join(labels_to_add)])
+        result = subprocess.run(cmd, capture_output=True, text=True)
         sys.stdout.write(result.stdout)
         sys.stderr.write(result.stderr)
         if result.returncode != 0:
@@ -206,30 +252,19 @@ def main(argv):
     else:
         pr_url = existing_pr['url']
         log.info('Updating PR …')
-        subprocess.run(
-            [
-                'gh', 'pr', 'edit', pr_url,
-                '--title', title,
-                '--body', body,
-                '--add-assignee', '@me',
-            ],
-            capture_output=True, text=True, check=True
-        )
+        cmd = [
+            'gh', 'pr', 'edit', pr_url,
+            '--title', title,
+            '--body', body,
+            '--add-assignee', '@me',
+        ]
+        if labels_to_add:
+            cmd.extend(['--add-label', ','.join(labels_to_add)])
+        if labels_to_remove:
+            cmd.extend(['--remove-label', ','.join(labels_to_remove)])
+        subprocess.run(cmd, capture_output=True, text=True, check=True)
         log.info('PR URL is %r', pr_url)
 
-    _label(pr_url, 'upgrade', mode='add' if has_u_tag else 'remove')
-
-    if args.no_partial:
-        _label(pr_url, 'partial', mode='remove')
-
-    if args.no_reindex:
-        for label in reindex_labels:
-            _label(pr_url, label, mode='remove')
-
-    if args.no_mirror:
-        for label in mirror_labels:
-            _label(pr_url, label, mode='remove')
-
     log.info('Setting PR status …')
     pr_node_id = _node_id(pr_url)
     _set_status(pr_node_id, 'In Progress')
@@ -450,6 +485,11 @@ def _mirror_labels(body: str) -> list[str]:
                       body, flags=re.MULTILINE)
 
 
+def _deploy_labels(body: str) -> list[str]:
+    return re.findall(r'^- \[[ x]] This PR is labeled `(deploy:\w+)`',
+                      body, flags=re.MULTILINE)
+
+
 def _check_task(body: str, task: str, checked: bool = True) -> str:
     mark = 'x' if checked else ' '
     body, n = re.subn(r'^- \[[ x]] (' + task + ')$',
@@ -490,21 +530,6 @@ def _node_id(url: str) -> str:
     return result.stdout.strip()
 
 
-type LabelMode = Literal['add', 'remove']
-
-
-def _label(item_url: str, label: str, *, mode: LabelMode) -> None:
-    assert check_type(LabelMode, mode)
-    item_type = _gh_item_type(item_url)
-    verb = {'add': 'Adding', 'remove': 'Removing'}[mode]
-    preposition = {'add': 'to', 'remove': 'from'}[mode]
-    log.info('%s label %r %s %r …', verb, label, preposition, item_url)
-    subprocess.run(
-        ['gh', item_type, 'edit', item_url, f'--{mode}-label', label],
-        capture_output=True, text=True
-    )
-
-
 def _set_status(node_id: str, status: str) -> None:
     project_id = _project_id()
 

src/azul/indexer/mirror_service.py

@@ -814,6 +814,14 @@ def _(self, a: MirrorPartitionAction) -> Iterator[MirrorAction]:
     @_mirror.register
     def _(self, a: MirrorFileAction) -> Iterator[MirrorAction]:
         if self._info_exists(a.file):
+            # Note that this check is a best effort only and does not completely
+            # prevent the creation of multiple uploads to the same file object.
+            # In fact, such a race is very likely if the info object is absent
+            # while the queue contains multiple MirrorFileActions for files of
+            # the same digest but different uuids. The larger the file, the more
+            # likely the race, but the race can occur for small files, too, if
+            # the actions are close to each other in the queue, increasing the
+            # probability that they are handled concurrently.
             log.info('File is already mirrored, skipping upload: %r', a.file)
             self._update_info(a.file, mark=a.mark)
         elif self._file_exists(a.file):
@@ -844,7 +852,7 @@ def _mirror_file(self, file: File):
                            data=file_content,
                            content_type=self._file_object_content_type,
                            overwrite=False)
-        self._create_info(file)
+        self._create_or_update_info(file)
 
     def _create_upload(self, file: File) -> FileUpload:
         object_key = self._file_object_key(file)
@@ -882,14 +890,29 @@ def _mirror_part(self,
 
     @_mirror.register
     def _(self, a: MirrorPartAction) -> Iterator[MirrorAction]:
-        # Some upload field values are mutable so we should make a copy
-        upload = a.upload.copy()
-        next_part = self._mirror_part(a.file, upload, a.part)
-        if next_part is None:
-            log.info('Uploaded all %d parts for file %r', len(upload.etags), a.file)
-            yield devolve(FinalizeFileAction, a, upload=upload)
+        if self._info_exists(a.file):
+            # This short-circuit is not as effective as one might think.
+            # Redundant MirrorFileActions for the same digest tend to be
+            # clustered together in the queue, so the concurrent uploads
+            # progress roughly in lockstep and all finish at about the same
+            # time. The info object only appears after one of them completes,
+            # by which point the others are nearly done, too. See #8162.
+            log.info('File was already mirrored, aborting upload %r of %r',
+                     a.upload.upload_id, a.file)
+            object_key = self._file_object_key(a.file)
+            storage = self._storage_for_file(a.file)
+            storage.abort_multipart_upload(object_key=object_key,
+                                           upload_id=a.upload.upload_id)
+            self._update_info(a.file, mark=True)
         else:
-            yield devolve(MirrorPartAction, a, upload=upload, part=next_part)
+            # Some upload field values are mutable so we should make a copy
+            upload = a.upload.copy()
+            next_part = self._mirror_part(a.file, upload, a.part)
+            if next_part is None:
+                log.info('Uploaded all %d parts for file %r', len(upload.etags), a.file)
+                yield devolve(FinalizeFileAction, a, upload=upload)
+            else:
+                yield devolve(MirrorPartAction, a, upload=upload, part=next_part)
 
     @_mirror.register
     def _(self, a: FinalizeFileAction) -> Iterator[MirrorAction]:
@@ -906,7 +929,7 @@ def _(self, a: FinalizeFileAction) -> Iterator[MirrorAction]:
             log.info('Discarding redundant upload %r of %r', a.upload.upload_id, a.file)
             storage.abort_multipart_upload(object_key=object_key,
                                            upload_id=a.upload.upload_id)
-        self._create_info(a.file)
+        self._create_or_update_info(a.file)
         log.info('Successfully mirrored file via multi-part upload: %r', a.file)
         return iter(())
 
@@ -936,6 +959,15 @@ def _info(self, file: File, old_info: JSON | None = None) -> JSON:
         }
 
     def _update_info(self, file: File, *, mark: bool):
+        """
+        Update the info object for the given file.
+
+        :param mark: Write the file even if it is already up-to-date, ensuring
+                     that its LastModified attribute is set to the current time
+
+        :raise StorageObjectNotFound: if the info object is absent
+        """
+
         def update(data: bytes) -> bytes:
             return json.dumps(self._info(file, json.loads(data))).encode()
 
@@ -944,6 +976,11 @@ def update(data: bytes) -> bytes:
         storage.update_object(key, update, content_type='application/json', touch=mark)
 
     def _create_info(self, file: File):
+        """
+        Create the info object for the given file.
+
+        :raise StorageObjectExists: if the info object already exists
+        """
         object_key = self._info_object_key(file)
         info = self._info(file)
         storage = self._storage_for_file(file)
@@ -952,6 +989,20 @@ def _create_info(self, file: File):
                            content_type='application/json',
                            overwrite=False)
 
+    def _create_or_update_info(self, file: File):
+        """
+        Ensure that an up-to-date info object exists for the given file,
+        creating it if necessary. Upon return, the info object's LastModified
+        time will be current.
+
+        Use this method in cases where multiple mirror threads may be uploading
+        duplicates, i.e. files with the same digest but a different uuid.
+        """
+        try:
+            self._create_info(file)
+        except StorageObjectExists:
+            self._update_info(file, mark=True)
+
     def _repository_url(self, file: File) -> str:
         plugin = self.repository_plugin
         assert isinstance(plugin, TDRPlugin), R(

test/app_test_case.py

@@ -19,6 +19,7 @@
     furl,
 )
 import urllib3
+
 from azul import (
     config,
 )

test/health_check_test_case.py

@@ -18,6 +18,7 @@
     MagicMock,
     patch,
 )
+
 from furl import (
     furl,
 )