Change content type to an array in info object schema (#7193, #7675)

Author: nadove-ucsc

docs/mirror.rst

@@ -67,8 +67,9 @@ hexadecimal form of a hash of the corresponding file object's content and
 algorithm used to derive that hash. The content of an ``info`` object is JSON of
 the form ``{"$schema":"…", "content-type":…}``.
 
-The ``content-type`` property contains the content type of the file, as defined
-for the HTTP response header of the same name [4]_.
+The ``content-type`` property contains a lost of all content types known to be
+associated with the file, as defined for the HTTP response header of the same
+name [4]_.
 
 .. [4] https://developer.mozilla.org/en-US/docs/Web/HTTP/Reference/Headers/Content-Type
 

src/azul/indexer/mirror_service.py

@@ -2,6 +2,7 @@
     ABCMeta,
     abstractmethod,
 )
+import bisect
 from functools import (
     singledispatchmethod,
 )
@@ -46,6 +47,9 @@
 from azul.auth import (
     Authentication,
 )
+from azul.collections import (
+    alist,
+)
 from azul.deployment import (
     aws,
 )
@@ -466,11 +470,9 @@ def _get_info(self, file: File) -> JSON | None:
             return None
         else:
             json_content = json.loads(content)
-            content_type = json_content['content-type']
-            if content_type != file.content_type:
-                # FIXME: Content type in mirror info objects inconsistent with index
-                #        https://github.com/DataBiosphere/azul/issues/7193
-                log.warning('Conflicting content type %r for file %r', content_type, file)
+            content_types = json_content['content-type']
+            if file.content_type not in content_types:
+                log.warning('Conflicting content type %r for file %r', content_types, file)
             return json_content
 
     info_prefix, file_prefix = 'info', 'file'
@@ -597,6 +599,7 @@ def _(self, a: MirrorFileAction) -> Iterator[MirrorAction]:
         assert a.file.size is not None, R('File size unknown', a.file)
         if self.info_exists(a.file):
             log.info('File is already mirrored, skipping upload: %r', a.file)
+            self.update_content_type(a.file)
         elif self.file_exists(a.file):
             assert False, R('File object is already present', a.file)
         else:
@@ -689,10 +692,25 @@ def _(self, a: FinalizeFileAction) -> Iterator[MirrorAction]:
 
     def _info(self, file: File) -> JSON:
         return {
-            'content-type': file.content_type,
+            'content-type': alist(file.content_type),
             '$schema': str(self._schema_url_func(schema_name='info', version=1))
         }
 
+    def update_content_type(self, file: File):
+        content_type = file.content_type
+
+        def update(data: bytes) -> bytes:
+            json_content = json.loads(data)
+            content_types = json_content['content-type']
+            if content_type is not None:
+                index = bisect.bisect_left(content_types, content_type)
+                if index == len(content_types) or content_types[index] != content_type:
+                    content_types.insert(index, content_type)
+            return json.dumps(json_content).encode()
+
+        key = self._info_object_key(file)
+        self._storage.update(key, update)
+
     def _put_info(self, file: File):
         object_key = self._info_object_key(file)
         info = self._info(file)

src/azul/service/storage_service.py

@@ -18,6 +18,7 @@
 )
 import time
 from typing import (
+    Callable,
     Collection,
     IO,
     TYPE_CHECKING,
@@ -115,6 +116,7 @@ def put_object(self,
                    data: bytes,
                    content_type: str | None = None,
                    tagging: Tagging | None = None,
+                   etag: str | None = None,
                    overwrite: bool = True):
         try:
             request: PutObjectRequestTypeDef
@@ -123,12 +125,55 @@ def put_object(self,
                 request['ContentType'] = content_type
             if tagging is not None:
                 request['Tagging'] = urlencode(tagging)
+            if etag is not None:
+                request['IfMatch'] = etag
             if overwrite is False:
                 request['IfNoneMatch'] = '*'
             self._s3.put_object(**request)
         except botocore.exceptions.ClientError as e:
             self._handle_overwrite(e, object_key)
 
+    def update(self,
+               object_key: str,
+               updater: Callable[[bytes], bytes],
+               max_attempts: int = 10
+               ) -> bool:
+        """
+        Attempt to update the contents of an object based on its existing
+        contents using conditional writes. Expects a callback that returns the
+        desired contents of the object given its current contents. If the
+        callback ever returns its argument unchanged, no further writes will be
+        attempted and this method will return False. It returns True if a write
+        succeeds, and raises StorageObjectNotFound if the object does not exist
+        at any point during the update.
+        """
+        for i in range(1, max_attempts + 1):
+            response = self._get(object_key)
+            etag = response['ETag']
+            data = response['Body'].read()
+            new_data = updater(data)
+            if new_data == data:
+                log.info('Update of %r canceled during attempt #%r/%r.',
+                         object_key, i, max_attempts)
+                return False
+            else:
+                try:
+                    self.put_object(object_key=object_key, data=new_data, etag=etag)
+                except botocore.exceptions.ClientError as e:
+                    error = e.response['Error']
+                    code, condition = error['Code'], error.get('Condition')
+                    if code == 'PreconditionFailed' and condition == 'If-Match':
+                        log.info('Conflict during attempt #%r/%r of updating %r from %r to %r',
+                                 i, max_attempts, object_key)
+                        if i == max_attempts:
+                            raise
+                    else:
+                        raise
+                else:
+                    log.info('Update of %r succeeded after %r attempts', object_key, i)
+                    return True
+        assert False
+
     def delete_objects(self,
                        object_keys: Collection[str],
                        batch_size: int = 1000

test/indexer/test_mirror_controller.py

@@ -115,6 +115,9 @@ def test_mirroring(self):
                     with self.subTest('mirror_file', corrupted=False, info_exists=False, file_exists=False):
                         self._test_mirror_file(file, file_message)
 
+                    with self.subTest('mirror_file', corrupted=False, info_exists=True, file_exists=True):
+                        self._test_content_type_update(file, file_message)
+
                     self._s3.delete_object(Bucket=self.mirror_bucket,
                                            Key=self.service._info_object_key(file))
 
@@ -204,6 +207,32 @@ def _test_reuploaded_file(self, file_message):
         self.assertTrue(R.caused(e.exception))
         self.assertEqual(e.exception.args[0].args[0], 'File object is already present')
 
+    def _test_content_type_update(self, file, file_message):
+        for content_type in [
+            'application/octet-stream',
+            'application/octet-stream',
+            'text/csv; charset="utf-8"',
+        ]:
+            changed_message = {
+                **file_message,
+                'file': attrs.evolve(file, content_type=content_type).to_json()
+            }
+            old_content_types = self._assertContentType(file)
+            event = self._mirror_event(changed_message)
+            self.mirror_controller.mirror(event)
+            new_content_types = self._assertContentType(file)
+            if content_type in old_content_types:
+                self.assertEqual(old_content_types, new_content_types)
+            else:
+                self.assertIn(content_type, new_content_types)
+
+    def _assertContentType(self, file) -> list[str]:
+        info = self.file_service._get_info(file)
+        content_types = info['content-type']
+        self.assertIsInstance(content_types, list)
+        self.assertEqual(sorted(set(content_types)), content_types)
+        return content_types
+
     def test_info_schema(self):
         client = http_client(log)
         file = MagicMock(content_type='text/plain')