Retry sync job using

Introduce mapping based on chunk_id (generated by client) and server_chunk_id generated by server.

Author: MarcelGeo

mergin/client.py

@@ -22,6 +22,7 @@
 
 from .common import (
     ClientError,
+    ErrorCode,
     LoginError,
     WorkspaceRole,
     ProjectRole,
@@ -41,7 +42,14 @@
     download_diffs_finalize,
 )
 from .client_pull import pull_project_async, pull_project_wait, pull_project_finalize
-from .client_push import get_push_changes_batch, push_project_async, push_project_is_running, push_project_wait, push_project_finalize, UploadChunksCache
+from .client_push import (
+    get_push_changes_batch,
+    push_project_async,
+    push_project_is_running,
+    push_project_wait,
+    push_project_finalize,
+    UploadChunksCache,
+)
 from .utils import DateTimeEncoder, get_versions_with_file_changes, int_version, is_version_acceptable
 from .version import __version__
 
@@ -1511,9 +1519,14 @@ def sync_project(self, project_directory):
                 push_project_finalize(job)
                 _, has_changes = get_push_changes_batch(self, mp, job.server_resp)
             except ClientError as e:
-                if e.http_error == 409 and server_conflict_attempts < 2:
+                if (
+                    e.sync_retry
+                    and server_conflict_attempts < 2
+                ):
                     # retry on conflict, e.g. when server has changes that we do not have yet
-                    mp.log.info("Attempting sync process due to conflicts between server and local directory or another user is syncing.")
+                    mp.log.info(
+                        "Attempting sync process due to conflicts between server and local directory or another user is syncing."
+                    )
                     server_conflict_attempts += 1
                     sleep(5)
                     continue
@@ -1548,9 +1561,14 @@ def sync_project_with_callback(self, project_directory, progress_callback=None,
                 push_project_finalize(job)
                 _, has_changes = get_push_changes_batch(self, mp, job.server_resp)
             except ClientError as e:
-                if e.http_error == 409 and server_conflict_attempts < 2:
+                if (
+                    e.sync_retry
+                    and server_conflict_attempts < 2
+                ):
                     # retry on conflict, e.g. when server has changes that we do not have yet
-                    mp.log.info("Attempting sync process due to conflicts between server and local directory or another user is syncing.")
+                    mp.log.info(
+                        "Attempting sync process due to conflicts between server and local directory or another user is syncing."
+                    )
                     server_conflict_attempts += 1
                     sleep(5)
                     continue

mergin/client_push.py

@@ -21,7 +21,7 @@
 
 from .local_changes import LocalChange, LocalChanges
 
-from .common import UPLOAD_CHUNK_SIZE, ClientError
+from .common import UPLOAD_CHUNK_SIZE, ClientError, ErrorCode
 from .merginproject import MerginProject
 from .editor import filter_changes
 
@@ -136,7 +136,7 @@ def upload_blocking(self):
                         continue
                     raise
 
-            self.mp.log.debug(f"Upload chunk {self.chunk_id} finished: {self.file_path}")
+            self.mp.log.debug(f"Upload chunk {self.server_chunk_id or self.chunk_id} finished: {self.file_path}")
 
 
 class UploadJob:
@@ -188,7 +188,7 @@ def add_items(self, items: List[UploadQueueItem], total_size: int):
     def update_chunks_from_items(self):
         """Update chunks in LocalChanges from the upload queue items."""
         self.changes.update_chunks(
-            [(item.file_checksum, item.server_chunk_id) for item in self.upload_queue_items]
+            [(item.chunk_id, item.server_chunk_id) for item in self.upload_queue_items]
         )
 
 
@@ -285,9 +285,9 @@ def create_upload_job_v2_api(
             {"Content-Type": "application/json"},
         )
     except ClientError as err:
-        # ignore 409 error, it means that the project is already in a transaction
+        # ignore 409 error codes, it means that the project is already in a transaction
         # but we still want to continue with the upload
-        if err.http_error not in [409]:
+        if err.server_code not in [ErrorCode.AnotherUploadRunning.value, ErrorCode.ProjectVersionExists]:
             mp.log.error("Error doing push check compatibility: " + str(err))
             mp.log.error("--- push aborted")
             raise
@@ -439,11 +439,10 @@ def push_project_finalize(job: UploadJob):
         job.mp.log.error("--- push finish failed! " + error_msg)
         raise ClientError("Upload error: " + error_msg)
 
-    job.update_chunks_from_items()
-
     if server_features.get("v2_push_enabled"):
         # v2 push uses a different endpoint
         try:
+            job.update_chunks_from_items()
             job.mp.log.info(f"Finishing transaction for project {job.mp.project_full_name()}")
             job.mc.post(
                 f"/v2/projects/{job.mp.project_id()}/versions",
@@ -456,6 +455,10 @@ def push_project_finalize(job: UploadJob):
             project_info = job.mc.project_info(job.mp.project_id())
             job.server_resp = project_info
         except ClientError as err:
+            if err.server_code in [ErrorCode.AnotherUploadRunning.value, ErrorCode.ProjectVersionExists]:
+                err.sync_retry = True
+            else:
+                job.mc.upload_chunks_cache.clear()  # clear the upload chunks cache, as we are getting fatal from server
             job.mp.log.error("--- push finish failed! " + str(err))
             raise err
     elif with_upload_of_files:

mergin/common.py

@@ -19,6 +19,8 @@
 class ErrorCode(Enum):
     ProjectsLimitHit = "ProjectsLimitHit"
     StorageLimitHit = "StorageLimitHit"
+    ProjectVersionExists = "ProjectVersionExists"
+    AnotherUploadRunning = "AnotherUploadRunning"
 
 
 class ClientError(Exception):
@@ -32,6 +34,8 @@ def __init__(self, detail, url=None, server_code=None, server_response=None, htt
         self.server_response = server_response
 
         self.extra = None
+        # Param to mark error as candidate for retry sync process
+        self.sync_retry = False
 
     def __str__(self):
         string_res = f"Detail: {self.detail}\n"

mergin/local_changes.py

@@ -66,24 +66,31 @@ def get_upload_changes(self) -> List[LocalChange]:
         This includes added and updated files.
         """
         return self.added + self.updated
+    
+    def _map_unique_chunks(self, change_chunks: List[str], server_chunks: List[Tuple[str, str]]) -> List[str]:
+        """
+        Helper function to map and deduplicate chunk ids for a single change.
+        """
+        mapped = []
+        seen = set()
+        for chunk in change_chunks:
+            for server_chunk in server_chunks:
+                chunk_id = server_chunk[0]
+                server_chunk_id = server_chunk[1]
+                if chunk_id == chunk and server_chunk_id not in seen:
+                    mapped.append(server_chunk_id)
+                    seen.add(server_chunk_id)
+        return mapped
 
-    def update_chunks(self, chunks: List[Tuple[str, str]]) -> None:
+    def update_chunks(self, server_chunks: List[Tuple[str, str]]) -> None:
         """
-        Map chunk ids to file checksums.
+        Map chunk ids to chunks returned from server (server_chunk_id).
 
         This method updates the `chunks` attribute of each change in `added` and `updated`
-        lists based on the provided `chunks` list, which contains tuples of (checksum, chunk_id).
+        lists based on the provided `server_chunks` list, which contains tuples of (chunk_id, server_chunk_id).
         """
         for change in self.added:
-            change.chunks = list({
-                chunk[1]
-                for chunk in chunks
-                if chunk[0] == change.checksum
-            })
+            change.chunks = self._map_unique_chunks(change.chunks, server_chunks)
 
         for change in self.updated:
-            change.chunks = list({
-                chunk[1]
-                for chunk in chunks
-                if chunk[0] == change.checksum
-            })
+            change.chunks = self._map_unique_chunks(change.chunks, server_chunks)

mergin/test/test_local_changes.py

@@ -94,11 +94,11 @@ def test_local_changes_get_server_request():
 def test_local_changes_update_chunks():
     """Test the update_chunks method of LocalChanges."""
     added = [
-        LocalChange(path="file1.txt", checksum="abc123", size=1024, mtime=datetime.now()),
-        LocalChange(path="file2.txt", checksum="abc123", size=1024, mtime=datetime.now())
+        LocalChange(path="file1.txt", checksum="abc123", size=1024, mtime=datetime.now(),chunks=["abc123"]),
+        LocalChange(path="file2.txt", checksum="abc123", size=1024, mtime=datetime.now(),chunks=["abc123"])
     ]
     updated = [
-        LocalChange(path="file2.txt", checksum="xyz789", size=2048, mtime=datetime.now())
+        LocalChange(path="file2.txt", checksum="xyz789", size=2048, mtime=datetime.now(), chunks=["xyz789"])
     ]
 
     local_changes = LocalChanges(added=added, updated=updated)