for gpkg files (added, updated) commit changes from wal file before starting upload transaction

Author: varmar05

mergin/client.py

@@ -134,6 +134,22 @@ def is_versioned_file(self, file):
         f_extension = os.path.splitext(file)[1]
         return f_extension in diff_extensions
 
+    def is_gpkg_open(self, path):
+        """
+        Check whether geopackage file is open (and wal file exists)
+
+        :param path: absolute path of file on disk
+        :type path: str
+        :returns: whether file is open
+        :rtype: bool
+        """
+        f_extension = os.path.splitext(path)[1]
+        if f_extension != '.gpkg':
+            return False
+        if os.path.exists(f'{path}-wal') and os.path.exists(f'{path}-shm'):
+            return True
+        return False
+
     def ignore_file(self, file):
         """
         Helper function for blacklisting certain types of files.
@@ -165,8 +181,6 @@ def inspect_files(self):
             for file in files:
                 if self.ignore_file(file):
                     continue
-                if "gpkg" in file:
-                    do_sqlite_checkpoint(os.path.join(root, file))
 
                 abs_path = os.path.abspath(os.path.join(root, file))
                 rel_path = os.path.relpath(abs_path, start=self.dir)
@@ -225,7 +239,8 @@ def compare_file_sets(self, origin, current):
             path = f["path"]
             if path not in origin_map:
                 continue
-            if f["checksum"] == origin_map[path]["checksum"]:
+            # with open WAL files we don't know yet, better to mark file as updated
+            if not self.is_gpkg_open(self.fpath(path)) and f["checksum"] == origin_map[path]["checksum"]:
                 continue
             f["origin_checksum"] = origin_map[path]["checksum"]
             updated.append(f)
@@ -301,13 +316,12 @@ def get_push_changes(self):
         :rtype: dict
         """
         changes = self.compare_file_sets(self.metadata['files'], self.inspect_files())
-        # do checkpoint to push changes from wal file to gpkg if new file
-        for file in changes['added']:
-            if ".gpkg" in file["path"]:
-                do_sqlite_checkpoint(self.fpath(file["path"]))
-                file["checksum"] = generate_checksum(self.fpath(file["path"]))
-
+        # do checkpoint to push changes from wal file to gpkg
         for file in changes['added'] + changes['updated']:
+            size, checksum = do_sqlite_checkpoint(self.fpath(file["path"]))
+            if size and checksum:
+                file["size"] = size
+                file["checksum"] = checksum
             file['chunks'] = [str(uuid.uuid4()) for i in range(math.ceil(file["size"] / UPLOAD_CHUNK_SIZE))]
 
         if not self.geodiff:
@@ -320,8 +334,9 @@ def get_push_changes(self):
             if not self.is_versioned_file(path):
                 continue
 
+            # we use geodiff to check if we can push only diff files
             current_file = self.fpath(path)
-            origin_file = self.fpath(path, self.meta_dir)
+            origin_file = self.fpath_meta(path)
             diff_id = str(uuid.uuid4())
             diff_name = path + '-diff-' + diff_id
             diff_file = self.fpath_meta(diff_name)
@@ -341,11 +356,8 @@ def get_push_changes(self):
                 else:
                     not_updated.append(file)
             except (pygeodiff.GeoDiffLibError, pygeodiff.GeoDiffLibConflictError) as e:
-                # do checkpoint to push changes from wal file to gpkg if create changeset failed
-                do_sqlite_checkpoint(self.fpath(file["path"]))
-                file["checksum"] = generate_checksum(self.fpath(file["path"]))
-                file["size"] = os.path.getsize(self.fpath(file["path"]))
-                file['chunks'] = [str(uuid.uuid4()) for i in range(math.ceil(file["size"] / UPLOAD_CHUNK_SIZE))]
+                # changes from wal file already committed
+                pass
 
         changes['updated'] = [f for f in changes['updated'] if f not in not_updated]
         return changes
@@ -490,15 +502,16 @@ def apply_push_changes(self, changes):
                 elif k == 'added':
                     shutil.copy(self.fpath(path), basefile)
                 elif k == 'updated':
-                    # in case for geopackage cannot be created diff
+                    # in case for geopackage cannot be created diff (e.g. forced update with committed changes from wal file)
                     if "diff" not in item:
-                        continue
-                    # better to apply diff to previous basefile to avoid issues with geodiff tmp files
-                    changeset = self.fpath_meta(item['diff']['path'])
-                    patch_error = self.apply_diffs(basefile, [changeset])
-                    if patch_error:
-                        # in case of local sync issues it is safier to remove basefile, next time it will be downloaded from server
-                        os.remove(basefile)
+                        shutil.copy(self.fpath(path), basefile)
+                    else:
+                        # better to apply diff to previous basefile to avoid issues with geodiff tmp files
+                        changeset = self.fpath_meta(item['diff']['path'])
+                        patch_error = self.apply_diffs(basefile, [changeset])
+                        if patch_error:
+                            # in case of local sync issues it is safier to remove basefile, next time it will be downloaded from server
+                            os.remove(basefile)
                 else:
                     pass
 
@@ -1090,6 +1103,7 @@ def _download_file(self, project_path, file, directory, parallel=True, diff_only
         }
         file_dir = os.path.dirname(os.path.normpath(os.path.join(directory, file['path'])))
         basename = os.path.basename(file['diff']['path']) if diff_only else os.path.basename(file['path'])
+        expected_size = file['diff']['size'] if diff_only else file['size']
 
         if file['size'] == 0:
             os.makedirs(file_dir, exist_ok=True)
@@ -1130,7 +1144,7 @@ def download_file_part(part):
                     shutil.copyfileobj(chunk, final)
                 os.remove(file_part)
 
-        if os.path.getsize(os.path.join(file_dir, basename)) != file['size']:
+        if os.path.getsize(os.path.join(file_dir, basename)) != expected_size:
             os.remove(os.path.join(file_dir, basename))
             raise ClientError(f'Download of file {basename} failed. Please try it again.')
 

mergin/test/modified_schema/base.gpkg-wal

@@ -136,11 +136,6 @@ def test_push_pull_changes(mc, parallel):
     f_updated = 'test3.txt'
     with open(os.path.join(project_dir, f_updated), 'w') as f:
         f.write('Modified')
-    src_files = os.listdir(CHANGED_SCHEMA_DIR)
-    for file_name in src_files:
-        full_file_name = os.path.join(CHANGED_SCHEMA_DIR, file_name)
-        if os.path.isfile(full_file_name):
-            shutil.copy(full_file_name, project_dir)
 
     # check changes before applied
     pull_changes, push_changes, _ = mc.project_status(project_dir)
@@ -342,3 +337,34 @@ def test_list_of_push_changes(mc):
         mc.project_status(project_dir)
 
 
+def test_force_gpkg_update(mc):
+    test_project = 'test_force_update'
+    project = API_USER + '/' + test_project
+    project_dir = os.path.join(TMP_DIR, test_project)  # primary project dir for updates
+
+    cleanup(mc, project, [project_dir])
+    # create remote project
+    shutil.copytree(TEST_DATA_DIR, project_dir)
+    mc.create_project(test_project, project_dir)
+
+    # test push changes with force gpkg file upload:
+    mp = MerginProject(project_dir)
+    f_updated = 'base.gpkg'
+    checksum = generate_checksum(mp.fpath(f_updated))
+
+    # base.gpkg updated to modified_schema (inserted new column)
+    shutil.move(mp.fpath(f_updated), mp.fpath_meta(f_updated))  # make local copy for changeset calculation (which will fail)
+    shutil.copy(os.path.join(CHANGED_SCHEMA_DIR, 'modified_schema.gpkg'), mp.fpath(f_updated))
+    shutil.copy(os.path.join(CHANGED_SCHEMA_DIR, 'modified_schema.gpkg-wal'), mp.fpath(f_updated + '-wal'))
+    shutil.copy(os.path.join(CHANGED_SCHEMA_DIR, 'modified_schema.gpkg-shm'), mp.fpath(f_updated + '-shm'))
+    mc.push_project(project_dir)
+    # by this point local file has been updated (changes committed from wal)
+    updated_checksum = generate_checksum(mp.fpath(f_updated))
+    assert checksum != updated_checksum
+
+    # check project after push
+    project_info = mc.project_info(project)
+    assert project_info['version'] == 'v2'
+    f_remote = next((f for f in project_info['files'] if f['path'] == f_updated), None)
+    assert f_remote['checksum'] == updated_checksum
+    assert 'diff' not in f_remote

mergin/test/modified_schema/base.gpkg …est/modified_schema/modified_schema.gpkgmergin/test/modified_schema/base.gpkg renamed to mergin/test/modified_schema/modified_schema.gpkg mergin/test/modified_schema/base.gpkg renamed to mergin/test/modified_schema/modified_schema.gpkg

@@ -71,12 +71,23 @@ def int_version(version):
 
 def do_sqlite_checkpoint(path):
     """
-    function to do checkpoint over the geopackage file which was not able to do diff file
+    Function to do checkpoint over the geopackage file which was not able to do diff file.
+
+    :param path: file's absolute path on disk
+    :type path: str
+    :returns: new size and checksum of file after checkpoint
+    :rtype: int, str
     """
+    new_size = None
+    new_checksum = None
     if ".gpkg" in path and os.path.exists(f'{path}-wal') and os.path.exists(f'{path}-shm'):
         conn = sqlite3.connect(path)
         cursor = conn.cursor()
         cursor.execute("PRAGMA wal_checkpoint=FULL")
         cursor.execute("VACUUM")
         conn.commit()
         conn.close()
+        new_size = os.path.getsize(path)
+        new_checksum = generate_checksum(path)
+
+    return new_size, new_checksum