Fix push finish on racing conditions

server/mergin/sync/files.py

@@ -80,17 +80,18 @@ class ProjectFileChange(ProjectFile):
     change: PushChangeType
 
 
-def files_changes_from_upload(changes: dict, version: int) -> List["ProjectFileChange"]:
+def files_changes_from_upload(
+    changes: dict, location_dir: str
+) -> List["ProjectFileChange"]:
     """Create a list of version file changes from upload changes dictionary used by public API.
 
     It flattens changes dict and adds change type to each item. Also generates location for each file.
     """
     secure_filenames = []
     version_changes = []
-    version = "v" + str(version)
     for key in ("added", "updated", "removed"):
         for item in changes.get(key, []):
-            location = os.path.join(version, mergin_secure_filename(item["path"]))
+            location = os.path.join(location_dir, mergin_secure_filename(item["path"]))
             diff = None
 
             # make sure we have unique location for each file
@@ -110,7 +111,7 @@ def files_changes_from_upload(changes: dict, version: int) -> List["ProjectFileC
                 if item.get("diff"):
                     change = PushChangeType.UPDATE_DIFF
                     diff_location = os.path.join(
-                        version, mergin_secure_filename(item["diff"]["path"])
+                        location_dir, mergin_secure_filename(item["diff"]["path"])
                     )
                     if diff_location in secure_filenames:
                         filename, file_extension = os.path.splitext(diff_location)

server/mergin/sync/public_api_controller.py

@@ -11,6 +11,7 @@
 from typing import Dict
 from urllib.parse import quote
 from datetime import datetime
+import uuid
 from marshmallow import ValidationError
 
 import gevent
@@ -52,7 +53,6 @@
 from .files import (
     ChangesSchema,
     ProjectDiffFile,
-    ProjectFile,
     ProjectFileChange,
     ProjectFileSchema,
     files_changes_from_upload,
@@ -67,7 +67,7 @@
     FileHistorySchema,
     ProjectVersionListSchema,
 )
-from .storages.storage import DataSyncError, InitializationError
+from .storages.storage import InitializationError
 from .storages.disk import save_to_file, move_to_tmp
 from .permissions import (
     require_project,
@@ -880,7 +880,9 @@ def project_push(namespace, project_name):
         next_version = version + 1
         user_agent = get_user_agent(request)
         device_id = get_device_id(request)
-        file_changes = files_changes_from_upload(upload.changes, version=next_version)
+        file_changes = files_changes_from_upload(
+            upload.changes, ProjectVersion.to_v_name(next_version)
+        )
         try:
             pv = ProjectVersion(
                 project,
@@ -981,23 +983,25 @@ def push_finish(transaction_id):
         abort(make_response(jsonify(ProjectLocked().to_dict()), 422))
 
     project_path = get_project_path(project)
-    next_version = project.latest_version + 1
-    v_next_version = ProjectVersion.to_v_name(next_version)
+    tmp_location_dir = "tmp"
     try:
         upload_changes = ChangesSchema().load(upload.changes)
     except ValidationError as err:
         msg = err.messages[0] if type(err.messages) == list else "Invalid input data"
         abort(422, msg)
 
-    file_changes = files_changes_from_upload(upload_changes, next_version)
+    file_changes = files_changes_from_upload(upload_changes, tmp_location_dir)
     chunks_map = {
         f["path"]: f["chunks"]
         for f in upload_changes["added"] + upload_changes["updated"]
     }
 
     corrupted_files = []
+    sync_errors = {}
+    to_remove = [i.path for i in file_changes if i.change == PushChangeType.DELETE]
+    current_files = [f for f in project.files if f.path not in to_remove]
 
-    # Concatenate chunks into single file
+    # Concatenate chunks into single file and apply gpkg updates if needed
     for f in file_changes:
         if f.change == PushChangeType.DELETE:
             continue
@@ -1048,109 +1052,138 @@ def push_finish(transaction_id):
             )
             corrupted_files.append(f.path)
 
-    if corrupted_files:
-        move_to_tmp(upload_dir)
-        abort(422, {"corrupted_files": corrupted_files})
+        # for updates try to apply diff to create a full updated gpkg file or from full .gpkg try to create corresponding diff
+        if f.change in (
+            PushChangeType.UPDATE,
+            PushChangeType.UPDATE_DIFF,
+        ) and is_versioned_file(f.path):
+            current_file = next((i for i in current_files if i.path == f.path), None)
+            if not current_file:
+                sync_errors[f.path] = "file not found on server "
+                continue
+            # yield to gevent hub since geodiff action can take some time to prevent worker timeout
+            sleep(0)
 
-    files_dir = os.path.join(upload_dir, "files", v_next_version)
-    target_dir = os.path.join(project.storage.project_dir, v_next_version)
+            if f.diff:
+                changeset = temporary_location
+                patched_file = os.path.join(upload_dir, "files", f.location)
 
-    # double check someone else has not created the same version meanwhile
-    if ProjectVersion.query.filter_by(
-        project_id=project.id, name=next_version
-    ).count() or os.path.exists(target_dir):
-        abort(
-            409,
-            f"There is already version with this name {v_next_version}",
+                result = project.storage.apply_diff(
+                    current_file, changeset, patched_file
+                )
+                if result.ok():
+                    checksum, size = result.value
+                    f.checksum = checksum
+                    f.size = size
+                else:
+                    sync_errors[f.path] = (
+                        f"project: {project.workspace.name}/{project.name}, {result.value}"
+                    )
+            else:
+                diff_name = mergin_secure_filename(
+                    f.path + "-diff-" + str(uuid.uuid4())
+                )
+                changeset = os.path.join(
+                    upload_dir, "files", tmp_location_dir, diff_name
+                )
+                patched_file = temporary_location
+                result = project.storage.construct_diff(
+                    current_file, changeset, patched_file
+                )
+                if result.ok():
+                    checksum, size = result.value
+                    f.diff = ProjectDiffFile(
+                        checksum=checksum,
+                        size=size,
+                        path=diff_name,
+                        location=os.path.join(tmp_location_dir, diff_name),
+                    )
+                    f.change = PushChangeType.UPDATE_DIFF
+                else:
+                    # if diff cannot be constructed it would be force update
+                    logging.warning(f"Geodiff: create changeset error {result.value}")
+
+    # geodiff related issues, project push would never succeed, whole upload is aborted
+    if sync_errors:
+        msg = ""
+        for key, value in sync_errors.items():
+            msg += key + " error=" + value + "\n"
+        logging.error(
+            f"Failed to finish push for project: {project.id}, project version: {project.next_version()}, {msg}"
+            f"transaction id: {transaction_id}.: {msg}"
         )
+        upload.clear()
+        abort(422, f"Failed to create new version: {msg}")
 
-    try:
-        # let's move uploaded files where they are expected to be
-        os.renames(files_dir, target_dir)
-        # apply gpkg updates
-        sync_errors = {}
-        to_remove = [i.path for i in file_changes if i.change == PushChangeType.DELETE]
-        current_files = [f for f in project.files if f.path not in to_remove]
-        for file in file_changes:
-            # for updates try to apply diff to create a full updated gpkg file or from full .gpkg try to create corresponding diff
-            if file.change in (
-                PushChangeType.UPDATE,
-                PushChangeType.UPDATE_DIFF,
-            ) and is_versioned_file(file.path):
-                current_file = next(
-                    (i for i in current_files if i.path == file.path), None
-                )
-                if not current_file:
-                    sync_errors[file.path] = "file not found on server "
-                    continue
+    if corrupted_files:
+        move_to_tmp(upload_dir)
+        abort(422, {"corrupted_files": corrupted_files})
 
-                # yield to gevent hub since geodiff action can take some time to prevent worker timeout
-                sleep(0)
+    # let's move files to the target directory, we need to determine what will be the next version (head could have changed meanwhile)
+    temp_files_dir = os.path.join(upload_dir, "files", tmp_location_dir)
+    project = Project.query.get(project.id)
+    next_version = project.next_version()
+    v_next_version = ProjectVersion.to_v_name(next_version)
+    version_dir = os.path.join(project.storage.project_dir, v_next_version)
+    # update files metadata to correspond to the new version
+    for f in file_changes:
+        if not f.location:
+            continue
+        f.location = os.path.join(
+            v_next_version, f.location.removeprefix(tmp_location_dir + os.path.sep)
+        )
 
-                if file.diff:
-                    result = project.storage.apply_diff(
-                        current_file, file, next_version
-                    )
-                    if result.ok():
-                        checksum, size = result.value
-                        file.checksum = checksum
-                        file.size = size
-                    else:
-                        sync_errors[file.path] = (
-                            f"project: {project.workspace.name}/{project.name}, {result.value}"
-                        )
-                else:
-                    result = project.storage.construct_diff(
-                        current_file, file, next_version
-                    )
-                    if result.ok():
-                        file.diff = result.value
-                        file.change = PushChangeType.UPDATE_DIFF
-                    else:
-                        # if diff cannot be constructed it would be force update
-                        logging.warning(
-                            f"Geodiff: create changeset error {result.value}"
-                        )
-
-        if sync_errors:
-            msg = ""
-            for key, value in sync_errors.items():
-                msg += key + " error=" + value + "\n"
-            raise DataSyncError(msg)
+        if f.diff:
+            f.diff.location = os.path.join(
+                v_next_version,
+                f.diff.location.removeprefix(tmp_location_dir + os.path.sep),
+            )
 
-        user_agent = get_user_agent(request)
-        device_id = get_device_id(request)
+    try:
         pv = ProjectVersion(
             project,
             next_version,
             current_user.id,
             file_changes,
             get_ip(request),
-            user_agent,
-            device_id,
+            get_user_agent(request),
+            get_device_id(request),
         )
         db.session.add(pv)
         db.session.add(project)
         db.session.commit()
+        # let's move uploaded files where they are expected to be
+        os.renames(temp_files_dir, version_dir)
+
         logging.info(
             f"Push finished for project: {project.id}, project version: {v_next_version}, transaction id: {transaction_id}."
         )
         project_version_created.send(pv)
         push_finished.send(pv)
-    except (psycopg2.Error, FileNotFoundError, DataSyncError, IntegrityError) as err:
+    except (psycopg2.Error, FileNotFoundError, IntegrityError) as err:
         db.session.rollback()
         logging.exception(
             f"Failed to finish push for project: {project.id}, project version: {v_next_version}, "
             f"transaction id: {transaction_id}.: {str(err)}"
         )
-        if os.path.exists(target_dir):
-            move_to_tmp(target_dir)
+        if (
+            os.path.exists(version_dir)
+            and not ProjectVersion.query.filter_by(
+                project_id=project.id, name=next_version
+            ).count()
+        ):
+            move_to_tmp(version_dir)
         abort(422, "Failed to create new version: {}".format(str(err)))
     # catch exception during pg transaction so we can rollback and prevent PendingRollbackError during upload clean up
     except gevent.timeout.Timeout:
         db.session.rollback()
-        if os.path.exists(target_dir):
-            move_to_tmp(target_dir)
+        if (
+            os.path.exists(version_dir)
+            and not ProjectVersion.query.filter_by(
+                project_id=project.id, name=next_version
+            ).count()
+        ):
+            move_to_tmp(version_dir)
         raise
     finally:
         # remove artifacts

server/mergin/sync/storages/disk.py

@@ -245,17 +245,15 @@ def _generator():
         return _generator()
 
     def apply_diff(
-        self, current_file: ProjectFile, upload_file: ProjectFile, version: int
+        self, current_file: ProjectFile, changeset: str, patchedfile: str
     ) -> Result:
         """Apply geodiff diff file on current gpkg basefile. Creates GeodiffActionHistory record of the action.
         Returns checksum and size of generated file. If action fails it returns geodiff error message.
         """
         from ..models import GeodiffActionHistory, ProjectVersion
 
-        v_name = ProjectVersion.to_v_name(version)
+        v_name = ProjectVersion.to_v_name(self.project.next_version())
         basefile = os.path.join(self.project_dir, current_file.location)
-        changeset = os.path.join(self.project_dir, upload_file.diff.location)
-        patchedfile = os.path.join(self.project_dir, upload_file.location)
         # create local copy of basefile which will be updated in next version and changeset needed
         # TODO this can potentially fail for large files
         logging.info(f"Apply changes: copying {basefile} to {patchedfile}")
@@ -313,28 +311,23 @@ def apply_diff(
                 return Err(self.gediff_log.getvalue())
 
     def construct_diff(
-        self, current_file: ProjectFile, upload_file: ProjectFile, version: int
+        self,
+        current_file: ProjectFile,
+        changeset: str,
+        uploaded_file: str,
     ) -> Result:
         """Construct geodiff diff file from uploaded gpkg and current basefile. Returns diff metadata as a result.
         If action fails it returns geodiff error message.
         """
-        from ..models import ProjectVersion
-
-        v_name = ProjectVersion.to_v_name(version)
         basefile = os.path.join(self.project_dir, current_file.location)
-        uploaded_file = os.path.join(self.project_dir, upload_file.location)
-        diff_name = upload_file.path + "-diff-" + str(uuid.uuid4())
-        changeset = os.path.join(self.project_dir, v_name, diff_name)
+        diff_name = os.path.basename(changeset)
         with self.geodiff_copy(basefile) as basefile_tmp, self.geodiff_copy(
             uploaded_file
         ) as uploaded_file_tmp:
             try:
                 # create changeset next to uploaded file copy
                 changeset_tmp = os.path.join(
-                    uploaded_file_tmp.replace(upload_file.location, "").rstrip(
-                        os.path.sep
-                    ),
-                    v_name,
+                    os.path.dirname(uploaded_file_tmp),
                     diff_name,
                 )
                 self.flush_geodiff_logger()
@@ -344,15 +337,13 @@ def construct_diff(
                 self.geodiff.create_changeset(
                     basefile_tmp, uploaded_file_tmp, changeset_tmp
                 )
-                # create diff metadata as it would be created by other clients
-                diff_file = ProjectDiffFile(
-                    path=diff_name,
-                    checksum=generate_checksum(changeset_tmp),
-                    size=os.path.getsize(changeset_tmp),
-                    location=os.path.join(v_name, mergin_secure_filename(diff_name)),
-                )
                 copy_file(changeset_tmp, changeset)
-                return Ok(diff_file)
+                return Ok(
+                    (
+                        generate_checksum(changeset_tmp),
+                        os.path.getsize(changeset_tmp),
+                    )
+                )
             except (GeoDiffLibError, GeoDiffLibConflictError) as e:
                 # diff is not possible to create - file will be overwritten
                 move_to_tmp(changeset)

server/mergin/tests/fixtures.py

@@ -214,7 +214,7 @@ def diff_project(app):
                 # no files uploaded, hence no action needed
                 pass
 
-            file_changes = files_changes_from_upload(change, version=i + 2)
+            file_changes = files_changes_from_upload(change, location_dir=f"v{i + 2}")
             pv = ProjectVersion(
                 project,
                 i + 2,