call remove unused chunks async job

Author: MarcelGeo

server/mergin/sync/public_api_v2_controller.py

@@ -14,6 +14,8 @@
 from marshmallow import ValidationError
 from sqlalchemy.exc import IntegrityError
 
+from mergin.sync.tasks import remove_transaction_chunks
+
 from .schemas_v2 import ProjectSchema as ProjectSchemaV2
 from ..app import db
 from ..auth import auth_required
@@ -318,6 +320,14 @@ def create_project_version(id):
             temp_files_dir = os.path.join(upload.upload_dir, "files", v_next_version)
             os.renames(temp_files_dir, version_dir)
 
+            # remove used chunks
+            # get chunks from added and updated files
+            chunks_ids = []
+            for file in to_be_added_files + to_be_updated_files:
+                file_chunks = file.get("chunks", [])
+                chunks_ids.extend(file_chunks)
+            remove_transaction_chunks.delay(chunks_ids)
+
         logging.info(
             f"Push finished for project: {project.id}, project version: {v_next_version}, upload id: {upload.id}."
         )

server/mergin/sync/tasks.py

@@ -13,7 +13,7 @@
 from .models import Project, ProjectVersion, FileHistory
 from .storages.disk import move_to_tmp
 from .config import Configuration
-from .utils import remove_outdated_files
+from .utils import get_chunk_location, remove_outdated_files
 from ..celery import celery
 from ..app import db
 
@@ -169,3 +169,12 @@ def remove_unused_chunks():
         if not os.path.isdir(dir):
             continue
         remove_outdated_files(dir, time_delta)
+
+
+@celery.task
+def remove_transaction_chunks(chunks=[]):
+    """Remove chunks related to a specific sync transaction"""
+    for chunk in chunks:
+        chunk_path = get_chunk_location(chunk)
+        if os.path.exists(chunk_path):
+            os.remove(chunk_path)

server/mergin/tests/test_public_api_v2.py

@@ -2,7 +2,7 @@
 #
 # SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-MerginMaps-Commercial
 
-from mergin.sync.tasks import remove_unused_chunks
+from mergin.sync.tasks import remove_transaction_chunks, remove_unused_chunks
 from . import DEFAULT_USER
 from .utils import (
     add_user,
@@ -358,6 +358,7 @@ def test_create_version(client, data, expected, err_code):
     assert project.latest_version == 1
 
     chunks = []
+    chunk_ids = []
     if expected == 201:
         # mimic chunks were uploaded
         for f in data["changes"]["added"] + data["changes"]["updated"]:
@@ -374,17 +375,21 @@ def test_create_version(client, data, expected, err_code):
                         out_file.write(in_file.read(CHUNK_SIZE))
 
                     chunks.append(chunk_location)
+                    chunk_ids.append(chunk)
 
-    response = client.post(f"v2/projects/{project.id}/versions", json=data)
+    with patch(
+        "mergin.sync.public_api_v2_controller.remove_transaction_chunks.delay"
+    ) as mock_remove:
+        response = client.post(f"v2/projects/{project.id}/versions", json=data)
     assert response.status_code == expected
-    # mock chunks expiration to check if removed
     if expected == 201:
         assert response.json["version"] == "v2"
         assert project.latest_version == 2
         # chunks exists after upload, cleanup job did not remove them
         assert all(os.path.exists(chunk) for chunk in chunks)
-        with patch.object(SyncConfiguration, "UPLOAD_CHUNKS_EXPIRATION", 0):
-            remove_unused_chunks()
+        if chunk_ids:
+            assert mock_remove.called_once_with(chunk_ids)
+        remove_transaction_chunks(chunk_ids)
         assert all(not os.path.exists(chunk) for chunk in chunks)
     else:
         assert project.latest_version == 1