Add unit tests and tunning of retry mechanism

Author: MarcelGeo

mergin/client.py

@@ -25,7 +25,6 @@
     PUSH_ATTEMPTS,
     SYNC_CALLBACK_WAIT,
     ClientError,
-    ErrorCode,
     LoginError,
     WorkspaceRole,
     ProjectRole,
@@ -807,7 +806,7 @@ def download_project(self, project_path, directory, version=None):
     def user_info(self):
         server_type = self.server_type()
         if server_type == ServerType.OLD:
-            resp = self.get("/v1/user/" + self.username())
+            resp = self.get(f"/v1/user/{self.username()}")
         else:
             resp = self.get("/v1/user/profile")
         return json.load(resp)
@@ -1526,11 +1525,11 @@ def sync_project(self, project_directory, progress_callback=None):
                         sleep(SYNC_CALLBACK_WAIT)
                         current_size = job.transferred_size
                         progress_callback(current_size - last_size, job)  # call callback with transferred size increment
-                        last = current_size
+                        last_size = current_size
                 push_project_finalize(job)
                 _, has_changes = get_push_changes_batch(self, mp, job.server_resp)
             except ClientError as e:
-                if e.sync_retry and server_conflict_attempts <= PUSH_ATTEMPTS:
+                if e.is_retryable_sync() and server_conflict_attempts < PUSH_ATTEMPTS - 1:
                     # retry on conflict, e.g. when server has changes that we do not have yet
                     mp.log.info(
                         f"Restarting sync process (conflict on server) - {server_conflict_attempts + 1}/{PUSH_ATTEMPTS}"

mergin/client_push.py

@@ -24,7 +24,7 @@
 
 from .local_changes import LocalChange, LocalChanges
 
-from .common import UPLOAD_CHUNK_SIZE, ClientError, ErrorCode
+from .common import UPLOAD_CHUNK_ATTEMPT_WAIT, UPLOAD_CHUNK_ATTEMPTS, UPLOAD_CHUNK_SIZE, ClientError, ErrorCode
 from .merginproject import MerginProject
 from .editor import filter_changes
 from .utils import get_data_checksum
@@ -111,8 +111,7 @@ def upload_blocking(self):
             checksum_str = get_data_checksum(data)
 
             self.mp.log.debug(f"Uploading {self.file_path} part={self.chunk_index}")
-            attempts = 2
-            for attempt in range(attempts):
+            for attempt in range(UPLOAD_CHUNK_ATTEMPTS):
                 try:
                     if self.mc.server_features().get("v2_push_enabled"):
                         # use v2 API for uploading chunks
@@ -122,8 +121,8 @@ def upload_blocking(self):
                         self.upload_chunk(data, checksum_str)
                     break  # exit loop if upload was successful
                 except ClientError as e:
-                    if attempt < attempts - 1:
-                        time.sleep(5)
+                    if attempt < UPLOAD_CHUNK_ATTEMPTS - 1:
+                        time.sleep(UPLOAD_CHUNK_ATTEMPT_WAIT)
                         continue
                     raise
 
@@ -249,7 +248,7 @@ def create_upload_job(
             )
             push_start_resp = json.load(resp)
     except ClientError as err:
-        if err.server_code not in [ErrorCode.AnotherUploadRunning.value, ErrorCode.ProjectVersionExists.value]:
+        if not err.is_blocking_sync():
             mp.log.error("Error starting transaction: " + str(err))
             mp.log.info("--- push aborted")
             raise
@@ -415,10 +414,9 @@ def push_project_finalize(job: UploadJob):
             project_info = json.load(resp)
             job.server_resp = project_info
         except ClientError as err:
-            if err.server_code in [ErrorCode.AnotherUploadRunning.value, ErrorCode.ProjectVersionExists.value]:
-                err.sync_retry = True
-            else:
-                job.mc.upload_chunks_cache.clear()  # clear the upload chunks cache, as we are getting fatal from server
+            if not err.is_retryable_sync():
+                # clear the upload chunks cache, as we are getting fatal from server
+                job.mc.upload_chunks_cache.clear()
             job.mp.log.error("--- push finish failed! " + str(err))
             raise err
     elif with_upload_of_files:

mergin/common.py

@@ -6,16 +6,22 @@
 # there is an upper limit for chunk size on server, ideally should be requested from there once implemented
 UPLOAD_CHUNK_SIZE = 10 * 1024 * 1024
 
+# number of attempts to upload a chunk
+UPLOAD_CHUNK_ATTEMPTS = 2
+
+# seconds to wait between attempts of uploading a chunk
+UPLOAD_CHUNK_ATTEMPT_WAIT = 5
+
 # size of the log file part to send (if file is larger only this size from end will be sent)
 MAX_LOG_FILE_SIZE_TO_SEND = 5 * 1024 * 1024
 
 # number of attempts to push changes (in case of network issues etc)
-PUSH_ATTEMPTS = 12
+PUSH_ATTEMPTS = 10
 
-# seconds to wait between attempts
+# seconds to wait between attempts to push changes
 PUSH_ATTEMPT_WAIT = 5
 
- # seconds to wait between sync callback calls
+# seconds to wait between sync callback calls
 SYNC_CALLBACK_WAIT = 0.01
 
 # default URL for submitting logs
@@ -33,7 +39,7 @@ class ErrorCode(Enum):
 
 
 class ClientError(Exception):
-    def __init__(self, detail, url=None, server_code=None, server_response=None, http_error=None, http_method=None):
+    def __init__(self, detail: str, url=None, server_code=None, server_response=None, http_error=None, http_method=None):
         self.detail = detail
         self.url = url
         self.http_error = http_error
@@ -43,8 +49,6 @@ 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"
@@ -60,6 +64,35 @@ def __str__(self):
             string_res += f"{self.extra}\n"
         return string_res
 
+    def is_rate_limit(self) -> bool:
+        """Check if error is rate limit error based on server code"""
+        return self.http_error == 429
+
+    def is_blocking_sync(self) -> bool:
+        """
+        Check if error is blocking sync based on server code.
+        Blocking sync means, that the sync is blocked by another user in server.
+        """
+        return self.server_code in [
+            ErrorCode.AnotherUploadRunning.value,
+            ErrorCode.ProjectVersionExists.value,
+        ]
+
+    def is_retryable_sync(self) -> bool:
+        # Check if error is retryable based on server code
+        if self.is_blocking_sync() or self.is_rate_limit():
+            return True
+
+        if (
+            self.http_error
+            and self.detail
+            and self.http_error == 400
+            and ("Another process" in self.detail or "Version mismatch" in self.detail)
+        ):
+            return True
+
+        return False
+
 
 class LoginError(Exception):
     pass

mergin/test/test_client.py

@@ -11,6 +11,7 @@
 import pytz
 import sqlite3
 import glob
+from unittest.mock import patch, Mock
 
 from .. import InvalidProject
 from ..client import (
@@ -3157,3 +3158,28 @@ def test_client_project_sync(mc):
         os.path.join(project_dir_2, "base.gpkg"),
     )
     mc.sync_project(project_dir_2)
+
+
+def test_client_project_sync_retry(mc):
+    test_project = "test_client_project_sync_retry"
+    project = API_USER + "/" + test_project
+    project_dir = os.path.join(TMP_DIR, test_project)
+    cleanup(mc, project, [project_dir])
+    mc.create_project(test_project)
+    mc.download_project(project, project_dir)
+    shutil.copy(os.path.join(TEST_DATA_DIR, "test.txt"), project_dir)
+    with patch("mergin.client.push_project_finalize", autospec=True) as mock_push_project_finalize:
+        mock_push_project_finalize.side_effect = ClientError("test error")
+        with pytest.raises(ClientError, match="test error"):
+            mc.sync_project(project_dir)
+
+    with patch("mergin.client.push_project_finalize") as mock_push_project_finalize, patch(
+        "mergin.client.PUSH_ATTEMPTS", 2
+    ):
+        mock_push_project_finalize.side_effect = ClientError(
+            detail="",
+            server_code=ErrorCode.AnotherUploadRunning.value,
+        )
+        with pytest.raises(ClientError):
+            mc.sync_project(project_dir)
+    assert mock_push_project_finalize.call_count == 2

mergin/test/test_common.py

@@ -0,0 +1,46 @@
+from ..common import ClientError, ErrorCode
+
+def test_client_error_is_blocked_sync():
+    """Test the is_blocked_sync method of ClientError."""
+    error = ClientError(detail="", server_code=None)
+    assert error.is_blocking_sync() is False
+    error.server_code = ErrorCode.ProjectsLimitHit.value
+    assert error.is_blocking_sync() is False
+
+    error.server_code = ErrorCode.AnotherUploadRunning.value
+    assert error.is_blocking_sync() is True
+    error.server_code = ErrorCode.ProjectVersionExists.value
+    assert error.is_blocking_sync() is True
+
+def test_client_error_is_rate_limit():
+    """Test the is_rate_limit method of ClientError."""
+    error = ClientError(detail="", http_error=None)
+    assert error.is_rate_limit() is False
+    error.http_error = 500
+    assert error.is_rate_limit() is False
+    error.http_error = 429
+    assert error.is_rate_limit() is True
+
+def test_client_error_is_retryable_sync():
+    """Test the is_retryable_sync method of ClientError."""
+    error = ClientError(detail="", server_code=None, http_error=None)
+    assert error.is_retryable_sync() is False
+
+    error.server_code = ErrorCode.ProjectsLimitHit.value
+    assert error.is_retryable_sync() is False
+    error.server_code = ErrorCode.AnotherUploadRunning.value
+    assert error.is_retryable_sync() is True
+
+    error.server_code = None
+    error.http_error = 400
+    error.detail = "Some other error"
+    assert error.is_retryable_sync() is False
+    error.detail = "Another process"
+    assert error.is_retryable_sync() is True
+    error.detail = "Version mismatch"
+    assert error.is_retryable_sync() is True
+
+    error.http_error = 500
+    assert error.is_retryable_sync() is False
+    error.http_error = 429
+    assert error.is_retryable_sync() is True