diff --git a/server/mergin/sync/files.py b/server/mergin/sync/files.py index 6073b0fb..83ca3381 100644 --- a/server/mergin/sync/files.py +++ b/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) diff --git a/server/mergin/sync/public_api_controller.py b/server/mergin/sync/public_api_controller.py index e9fb4a76..54f226bc 100644 --- a/server/mergin/sync/public_api_controller.py +++ b/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 diff --git a/server/mergin/sync/storages/disk.py b/server/mergin/sync/storages/disk.py index 459d704d..210065db 100644 --- a/server/mergin/sync/storages/disk.py +++ b/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) diff --git a/server/mergin/tests/fixtures.py b/server/mergin/tests/fixtures.py index 9f39909d..2013e54b 100644 --- a/server/mergin/tests/fixtures.py +++ b/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, diff --git a/server/mergin/tests/test_project_controller.py b/server/mergin/tests/test_project_controller.py index 989133e6..e56a9d2d 100644 --- a/server/mergin/tests/test_project_controller.py +++ b/server/mergin/tests/test_project_controller.py @@ -38,7 +38,7 @@ PushChangeType, ProjectFilePath, ) -from ..sync.files import ChangesSchema, files_changes_from_upload +from ..sync.files import files_changes_from_upload from ..sync.schemas import ProjectListSchema from ..sync.utils import generate_checksum, is_versioned_file, get_project_path from ..auth.models import User, UserProfile @@ -1499,38 +1499,26 @@ def test_push_finish(client): with open(os.path.join(upload_dir, "chunks", chunk), "wb") as out_file: out_file.write(in_file.read(CHUNK_SIZE)) - # test finish upload, pretend another upload is already being processed - os.makedirs( - os.path.join( - upload.project.storage.project_dir, f"v{upload.project.latest_version + 1}" - ), - exist_ok=True, - ) - resp = client.post( - url, - headers=json_headers, - ) - assert resp.status_code == 409 - # bump to fake version to make upload finish pass - upload.project.latest_version += 1 - db.session.add(upload.project) - pv = ProjectVersion( - upload.project, - upload.project.latest_version, - upload.project.creator.id, - [], - "127.0.0.1", - ) - pv.project = upload.project - db.session.add(pv) - db.session.commit() + # test finish upload when another upload was already processed + original_version = upload.project.latest_version + with patch("mergin.sync.models.Project.next_version") as mock_next_version: + mock_next_version.return_value = original_version + 2 + + resp = client.post( + url, + headers=json_headers, + ) - resp2 = client.post(url, headers={**json_headers, "User-Agent": "Werkzeug"}) - assert resp2.status_code == 200 - assert not os.path.exists(upload_dir) - version = upload.project.get_latest_version() - assert version.user_agent - assert version.device_id == json_headers["X-Device-Id"] + assert resp.status_code == 200 + project = Project.query.get(upload.project.id) + version = project.get_latest_version() + # finish created higher version than origially expected + assert version.name == mock_next_version.return_value + assert os.path.exists( + os.path.join(upload.project.storage.project_dir, f"v{version.name}") + ) + assert version.user_agent + assert version.device_id == json_headers["X-Device-Id"] # tests basic failures resp3 = client.post("/v1/project/push/finish/not-existing") @@ -1557,7 +1545,7 @@ def test_push_finish(client): assert resp4.status_code == 403 # other failures with error code 403, 404 does to count to failures history - assert SyncFailuresHistory.query.count() == 2 + assert SyncFailuresHistory.query.count() == 1 def test_push_close(client): @@ -2429,7 +2417,7 @@ def add_project_version(project, changes, version=None): else User.query.filter_by(username=DEFAULT_USER[0]).first() ) next_version = version or project.next_version() - file_changes = files_changes_from_upload(changes, version=next_version) + file_changes = files_changes_from_upload(changes, location_dir="v{next_version}") pv = ProjectVersion( project, next_version, diff --git a/server/mergin/tests/utils.py b/server/mergin/tests/utils.py index 0c4448a6..692a901c 100644 --- a/server/mergin/tests/utils.py +++ b/server/mergin/tests/utils.py @@ -349,7 +349,9 @@ def push_change(project, action, path, src_dir): else: return - file_changes = files_changes_from_upload(changes, version=project.next_version()) + file_changes = files_changes_from_upload( + changes, location_dir=f"v{project.next_version()}" + ) pv = ProjectVersion( project, project.next_version(),