[ENG-11727] Corrupted file versions are unpreviewable, undownloadable, and undeletable

addons/osfstorage/models.py

@@ -14,6 +14,7 @@
 from osf.models import AbstractNode
 from osf.models.files import File, FileVersion, Folder, TrashedFileNode, BaseFileNode, BaseFileNodeManager
 from osf.utils import permissions
+from osf.utils.requests import check_select_for_update
 from website.files import exceptions
 from website.files import utils as files_utils
 from website.util import api_url_for
@@ -327,6 +328,12 @@ def update_region_from_latest_version(self, destination_parent):
             most_recent_fileversion.save()
 
     def create_version(self, creator, location, metadata=None):
+        if check_select_for_update():
+            # Lock the file row for the duration of the request's transaction to avoid
+            # concurrent/retried requests for the same file to read the same version
+            # count and insert duplicate identifiers
+            self.__class__.objects.select_for_update().get(pk=self.pk)
+
         latest_version = self.get_version()
         version = FileVersion(identifier=self.versions.count() + 1, creator=creator, location=location)
 
@@ -354,12 +361,13 @@ def get_version(self, version=None, required=False):
                 return self.versions.first()
             return None
 
-        try:
-            return self.versions.get(identifier=version)
-        except FileVersion.DoesNotExist:
-            if required:
-                raise exceptions.VersionNotFoundError(version)
-            return None
+        # .filter().first() better than .get(): some files have more
+        # than one FileVersion sharing the same identifier, which would
+        # otherwise raise MultipleObjectsReturned here instead of retrieving a version
+        result = self.versions.filter(identifier=version).order_by('created').first()
+        if result is None and required:
+            raise exceptions.VersionNotFoundError(version)
+        return result
 
     def add_tag_log(self, action, tag, auth):
         if isinstance(self.target, Loggable):

addons/osfstorage/tests/test_models.py

@@ -3,6 +3,8 @@
 
 import pytest
 import pytz
+from django.db import connection, transaction
+from django.test.utils import CaptureQueriesContext
 from django.utils import timezone
 from importlib import import_module
 from django.conf import settings as django_conf_settings
@@ -195,9 +197,43 @@ def test_download_count_file(self):
         assert child.get_download_count(1) == 1
         assert child.get_download_count(2) == 1
 
-    @unittest.skip
-    def test_create_version(self):
-        pass
+    def test_create_version_locks_file_row(self):
+
+        file = self.node_settings.get_root().append_file('locked.txt')
+
+        with transaction.atomic(), CaptureQueriesContext(connection) as ctx:
+            file.create_version(
+                self.user,
+                {
+                    'service': 'cloud',
+                    settings.WATERBUTLER_RESOURCE: 'osf',
+                    'object': '06d80e',
+                }, {
+                    'size': 1234,
+                    'contentType': 'text/plain'
+                })
+
+        for_update_sql = connection.ops.for_update_sql()
+        assert any(for_update_sql in query['sql'] for query in ctx.captured_queries)
+
+    @mock.patch('osf.utils.requests.settings.SELECT_FOR_UPDATE_ENABLED', False)
+    def test_create_version_does_not_lock_file_row_when_disabled(self):
+        file = self.node_settings.get_root().append_file('unlocked.txt')
+
+        with transaction.atomic(), CaptureQueriesContext(connection) as ctx:
+            file.create_version(
+                self.user,
+                {
+                    'service': 'cloud',
+                    settings.WATERBUTLER_RESOURCE: 'osf',
+                    'object': '06d80e',
+                }, {
+                    'size': 1234,
+                    'contentType': 'text/plain'
+                })
+
+        for_update_sql = connection.ops.for_update_sql()
+        assert not any(for_update_sql in query['sql'] for query in ctx.captured_queries)
 
     def test_delete_folder(self):
         parent = self.node_settings.get_root().append_folder('Test')

osf/management/commands/dedupe_file_versions.py

@@ -0,0 +1,123 @@
+import logging
+
+from django.core.management.base import BaseCommand
+from django.db import transaction
+from django.db.models import Count
+
+from osf.models.files import BaseFileNode, BaseFileVersionsThrough
+from osf.utils.requests import check_select_for_update
+
+logger = logging.getLogger(__name__)
+
+
+def find_duplicate_groups():
+    """
+    Yields {'basefilenode_id', 'fileversion__identifier', 'count'} for every
+    (file, identifier) pair that has more than one linked FileVersion.
+    """
+    return (
+        BaseFileVersionsThrough.objects
+        .values('basefilenode_id', 'fileversion__identifier')
+        .annotate(count=Count('id'))
+        .filter(count__gt=1)
+        .order_by('basefilenode_id')
+    )
+
+
+def fetch_group_rows(basefilenode_id, identifier):
+    # `fileversion__id` is a secondary sort key so the choice of keeper is fully
+    # deterministic even if two duplicates share the exact same `created` timestamp.
+    return list(
+        BaseFileVersionsThrough.objects
+        .filter(basefilenode_id=basefilenode_id, fileversion__identifier=identifier)
+        .select_related('fileversion')
+        .order_by('fileversion__created', 'fileversion__id')
+    )
+
+
+def log_group(file_label, identifier, row_to_keep, rows_to_delete, dry_run):
+    logger.info(
+        f'{"[DRY-RUN] " if dry_run else ""}file={file_label} identifier={identifier} '
+        f'keeping {row_to_keep.fileversion._id} (location={row_to_keep.fileversion.location}) '
+        f'discarding={[(row.fileversion._id, row.fileversion.location) for row in rows_to_delete]}'
+    )
+
+
+def resolve_group(basefilenode_id, identifier, file_label, dry_run):
+    """
+    Fetches the current rows for one duplicate group, logs the keep/discard
+    decision, unless dry_run - deletes the discarded duplicate(s.
+    """
+    through_rows = fetch_group_rows(basefilenode_id, identifier)
+    if len(through_rows) < 2:
+        return False
+
+    row_to_keep, rows_to_delete = through_rows[0], through_rows[1:]
+    log_group(file_label, identifier, row_to_keep, rows_to_delete, dry_run=dry_run)
+
+    if not dry_run:
+        for row in rows_to_delete:
+            row.delete()
+
+    return True
+
+
+def dedupe_file_versions(dry_run=True):
+    """
+    Finds FileVersions that share the same `identifier` for the same file because of
+    race condition in OsfStorageFile.create_version() and delete duplicate
+    """
+    if dry_run:
+        logger.info('[DRY-RUN] Data will not be modified.')
+
+    groups = list(find_duplicate_groups())
+    file_labels = dict(
+        BaseFileNode.objects
+        .filter(id__in={group['basefilenode_id'] for group in groups})
+        .values_list('id', '_id')
+    )
+
+    fixed = 0
+
+    for group in groups:
+        basefilenode_id = group['basefilenode_id']
+        identifier = group['fileversion__identifier']
+        file_label = file_labels.get(basefilenode_id, basefilenode_id)
+
+        if dry_run:
+            resolved = resolve_group(basefilenode_id, identifier, file_label, dry_run=True)
+        else:
+            with transaction.atomic():
+                if check_select_for_update():
+                    # Lock the file row for the duration of this group's cleanup so
+                    # a concurrent create_version() call for the same file can't
+                    # interleave with the read-then-delete in resolve_group().
+                    BaseFileNode.objects.select_for_update().get(pk=basefilenode_id)
+                resolved = resolve_group(basefilenode_id, identifier, file_label, dry_run=False)
+
+        if resolved:
+            fixed += 1
+
+    logger.info(f'{fixed} duplicate group(s) resolved.')
+    return fixed
+
+
+class Command(BaseCommand):
+    help = """
+    Finds FileVersions that share the same `identifier` on the same file because of
+    a race condition in OsfStorageFile.create_version() and delete duplicate
+    """
+
+    def add_arguments(self, parser):
+        parser.add_argument(
+            '--apply',
+            action='store_false',
+            dest='dry_run',
+            default=True,
+            help='Actually unlink duplicate versions. Without this flag, only reports what would change.',
+        )
+
+    # Management command handler
+    def handle(self, *args, **options):
+        dry_run = options.get('dry_run', True)
+        dedupe_file_versions(dry_run=dry_run)

osf/models/files.py

@@ -6,7 +6,6 @@
 from django.apps import apps
 from django.db import models, IntegrityError
 from django.db.models import Manager
-from django.core.exceptions import ObjectDoesNotExist
 from django.utils import timezone
 from django.contrib.contenttypes.models import ContentType
 from django.contrib.contenttypes.fields import GenericForeignKey
@@ -289,12 +288,13 @@ def get_version(self, revision, required=False):
         :returns: FileVersion or None
         :raises: VersionNotFoundError if required is True
         """
-        try:
-            return self.versions.get(identifier=revision)
-        except ObjectDoesNotExist:
-            if required:
-                raise VersionNotFoundError(revision)
-            return None
+        # .filter().first() better than .get(): some files have more
+        # than one FileVersion sharing the same identifier, which would otherwise raise
+        # MultipleObjectsReturned here instead of retrieving a version
+        version = self.versions.filter(identifier=revision).order_by('created').first()
+        if version is None and required:
+            raise VersionNotFoundError(revision)
+        return version
 
     def generate_waterbutler_url(self, **kwargs):
         base_url = None

osf_tests/management_commands/test_dedupe_file_versions.py

@@ -0,0 +1,73 @@
+import pytest
+from django.core.management import call_command
+
+from addons.osfstorage import settings as osfstorage_settings
+from addons.osfstorage.tests.factories import FileVersionFactory
+from osf.models import BaseFileVersionsThrough, FileVersion
+from osf_tests.factories import ProjectFactory
+
+
+def make_location(obj):
+    return {
+        'service': 'cloud',
+        osfstorage_settings.WATERBUTLER_RESOURCE: 'osf',
+        'object': obj,
+    }
+
+
+@pytest.mark.django_db
+class TestDedupeFileVersions:
+
+    @pytest.fixture()
+    def file_node(self):
+        project = ProjectFactory()
+        return project.get_addon('osfstorage').get_root().append_file('dupes.txt')
+
+    @pytest.fixture()
+    def add_duplicate_version(self, file_node):
+        def _add_duplicate_version(identifier, location):
+            version = FileVersionFactory(identifier=identifier, location=location)
+            file_node.add_version(version)
+            return version
+        return _add_duplicate_version
+
+    def test_dry_run_leaves_duplicates_untouched(self, file_node, add_duplicate_version):
+        add_duplicate_version('2', make_location('object-a'))
+        add_duplicate_version('2', make_location('object-b'))
+
+        call_command('dedupe_file_versions')
+
+        assert file_node.versions.filter(identifier='2').count() == 2
+
+    def test_apply_unlinks_duplicate_keeping_earliest(self, file_node, add_duplicate_version):
+        version_to_keep = add_duplicate_version('2', make_location('object-a'))
+        extra = add_duplicate_version('2', make_location('object-b'))
+
+        call_command('dedupe_file_versions', dry_run=False)
+
+        remaining = list(file_node.versions.filter(identifier='2'))
+        assert remaining == [version_to_keep]
+        assert FileVersion.objects.filter(id=extra.id).exists()
+        assert not BaseFileVersionsThrough.objects.filter(basefilenode=file_node, fileversion=extra).exists()
+
+    def test_leaves_non_duplicate_versions_alone(self, file_node, add_duplicate_version):
+        add_duplicate_version('1', make_location('object-1'))
+        add_duplicate_version('2', make_location('object-2'))
+
+        call_command('dedupe_file_versions', dry_run=False)
+
+        assert file_node.versions.count() == 2
+
+    def test_apply_deletes_every_duplicate_but_the_keeper(self, file_node, add_duplicate_version):
+        version_to_keep = add_duplicate_version('2', make_location('object-a'))
+        extra_1 = add_duplicate_version('2', make_location('object-b'))
+        extra_2 = add_duplicate_version('2', make_location('object-c'))
+
+        assert BaseFileVersionsThrough.objects.filter(basefilenode=file_node).count() == 3
+
+        call_command('dedupe_file_versions', dry_run=False)
+
+        assert BaseFileVersionsThrough.objects.filter(basefilenode=file_node).count() == 1
+        assert list(file_node.versions.all()) == [version_to_keep]
+        for extra in (extra_1, extra_2):
+            assert not BaseFileVersionsThrough.objects.filter(basefilenode=file_node, fileversion=extra).exists()

osf_tests/test_files.py

@@ -198,3 +198,24 @@ def test_file_shared_purged(project, create_test_file):
     assert freed == sum(list(test_file_1.versions.values_list('size', flat=True)))
     assert test_file_1.purged is not None
     assert version_0.purged is not None
+
+
+def test_get_version_resolves_duplicate_identifiers(project, create_test_file):
+    # Simulates the historical create_version race that could attach two
+    # FileVersions with the same identifier to one file. get_version() must
+    # resolve this instead of raising MultipleObjectsReturned.
+    test_file = create_test_file(target=project)
+    first_version = test_file.versions.first()
+
+    duplicate_version = FileVersion(
+        creator=first_version.creator,
+        identifier=first_version.identifier,
+        location=dict(first_version.location, object='deadbeef'),
+    )
+    duplicate_version.save()
+    test_file.add_version(duplicate_version)
+
+    assert test_file.versions.filter(identifier=first_version.identifier).count() == 2
+
+    resolved = test_file.get_version(first_version.identifier, required=True)
+    assert resolved == first_version