Replace lockfile with upload last_ping db column

This is a refactoring that replaces the filesystem-based Toucher/lockfile mechanism with a database-based heartbeat approach for tracking active uploads.
This eliminates the NFS-specific hack (os.access() + os.utime()) that was needed to bust NFS attribute caches in the old approach. The new approach is cleaner and more cloud-native.

Author: varmar05

server/mergin/sync/models.py

@@ -2,12 +2,14 @@
 #
 # SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-MerginMaps-Commercial
 from __future__ import annotations
+from contextlib import contextmanager
 import json
 import logging
 import os
+import threading
 import time
 import uuid
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 from enum import Enum
 from typing import Optional, List, Dict, Set, Tuple
 from dataclasses import dataclass, asdict
@@ -21,7 +23,7 @@
 from sqlalchemy.types import String
 from sqlalchemy.ext.hybrid import hybrid_property
 from pygeodiff.geodifflib import GeoDiffLibError, GeoDiffLibConflictError
-from flask import current_app
+from flask import Flask, current_app
 
 from .files import (
     DeltaChangeMerged,
@@ -44,7 +46,6 @@
     LOG_BASE,
     Checkpoint,
     generate_checksum,
-    Toucher,
     get_chunk_location,
     get_project_path,
     is_supported_type,
@@ -1805,6 +1806,8 @@ class Upload(db.Model):
         db.Integer, db.ForeignKey("user.id", ondelete="CASCADE"), nullable=True
     )
     created = db.Column(db.DateTime, default=datetime.utcnow)
+    # last ping time to determine if upload is still active
+    last_ping = db.Column(db.DateTime, nullable=False, default=datetime.utcnow)
 
     user = db.relationship("User")
     project = db.relationship(
@@ -1827,17 +1830,67 @@ def __init__(self, project: Project, version: int, changes: dict, user_id: int):
     def upload_dir(self):
         return os.path.join(self.project.storage.project_dir, "tmp", self.id)
 
-    @property
-    def lockfile(self):
-        return os.path.join(self.upload_dir, "lockfile")
-
     def is_active(self):
-        """Check if upload is still active because there was a ping (lockfile update) from underlying process"""
-        return os.path.exists(self.lockfile) and (
-            time.time() - os.path.getmtime(self.lockfile)
-            < current_app.config["LOCKFILE_EXPIRATION"]
+        """Check if upload is still active because there was a ping from underlying process"""
+        return datetime.now(tz=timezone.utc) < self.last_ping.replace(
+            tzinfo=timezone.utc
+        ) + timedelta(seconds=current_app.config["LOCKFILE_EXPIRATION"])
+
+    def _heartbeat_task(self, app: Flask, stop_event: threading.Event, timeout: int):
+        """
+        Background task: Runs as a Thread (Sync) or Greenlet (Gevent) based on worker type.
+        Uses a fresh engine connection to stay pool-efficient.
+        """
+        # manual context push is required for background execution
+        with app.app_context():
+            while not stop_event.is_set():
+                try:
+                    # db.engine.begin() is efficient and isolated, it immediately returns a connection to the pool
+                    with db.engine.begin() as conn:
+                        conn.execute(
+                            db.text(
+                                "UPDATE upload SET last_ping = NOW() WHERE id = :id"
+                            ),
+                            {"id": self.id},
+                        )
+                except Exception as e:
+                    logging.exception(
+                        f"Upload heartbeat failed for ID {self.project_id} and version {self.version}: {e}"
+                    )
+
+                # wait for x seconds, but wake up immediately if stop_event is set
+                stop_event.wait(timeout)
+
+    @contextmanager
+    def heartbeat(self, timeout: int = 5):
+        """
+        Context manager to be used inside a Flask route.
+
+        Example of usage:
+        -----------------
+        with upload.heartbeat(interval):
+            do_something_slow
+        """
+        # we need to pass a real Flask app object to the thread
+        app = current_app._get_current_object()
+        stop_event = threading.Event()
+
+        bg = threading.Thread(
+            target=self._heartbeat_task, args=(app, stop_event, timeout), daemon=True
         )
 
+        bg.start()
+        try:
+            yield
+        finally:
+            # signal the loop to stop
+            stop_event.set()
+
+            # wait for the task to finish its last SQL call.
+            # in Gevent, this yields to other requests (non-blocking), while in Sync, this blocks the current thread for up to 2s
+            # this is to protect main thread / greenlet from zombie bg processes
+            bg.join(timeout=2)
+
     def clear(self):
         """Clean up pending upload.
         Uploaded files and table records are removed, and another upload can start.
@@ -1864,7 +1917,7 @@ def process_chunks(
         to_remove = [i.path for i in file_changes if i.change == PushChangeType.DELETE]
         current_files = [f for f in self.project.files if f.path not in to_remove]
 
-        with Toucher(self.lockfile, 5):
+        with self.heartbeat(5):
             for f in file_changes:
                 if f.change == PushChangeType.DELETE:
                     continue

server/mergin/sync/public_api.yaml

@@ -699,7 +699,7 @@ paths:
        - do integrity check comparing uploaded file sizes with what was expected
        - move uploaded files to new version dir and applying sync changes (e.g. geodiff apply_changeset)
        - bump up version in database
-       - remove artifacts (chunks, lockfile) by moving them to tmp directory"
+       - remove artifacts (chunks) by moving them to tmp directory"
       operationId: push_finish
       parameters:
         - name: transaction_id

server/mergin/sync/public_api_controller.py

@@ -75,7 +75,6 @@
 )
 from .utils import (
     generate_checksum,
-    Toucher,
     get_ip,
     get_user_agent,
     generate_location,
@@ -849,11 +848,10 @@ def project_push(namespace, project_name):
             logging.error(f"Failed to create upload session: {str(err)}")
             abort(422, "Failed to create upload session. Please try later.")
 
-    # Create transaction folder and lockfile
+    # Create transaction folder
     os.makedirs(upload.upload_dir)
-    open(upload.lockfile, "w").close()
 
-    # Update immediately without uploading of new/modified files and remove transaction/lockfile after successful commit
+    # Update immediately without uploading of new/modified files and remove transaction after successful commit
     if not (changes["added"] or changes["updated"]):
         next_version = version + 1
         file_changes = files_changes_from_upload(
@@ -920,7 +918,7 @@ def chunk_upload(transaction_id, chunk_id):
         abort(404)
 
     dest = os.path.join(upload_dir, "chunks", chunk_id)
-    with Toucher(upload.lockfile, 30):
+    with upload.heartbeat(30):
         try:
             # we could have used request.data here, but it could eventually cause OOM issue
             save_to_file(request.stream, dest, current_app.config["MAX_CHUNK_SIZE"])
@@ -945,7 +943,7 @@ def push_finish(transaction_id):
      - do integrity check comparing uploaded file sizes with what was expected
      - move uploaded files to new version dir and applying sync changes (e.g. geodiff apply_changeset)
      - bump up version in database
-     - remove artifacts (chunks, lockfile) by moving them to tmp directory
+     - remove artifacts (chunks) by moving them to tmp directory
 
     :param transaction_id: Transaction id.
     :type transaction_id: str

server/mergin/sync/public_api_v2_controller.py

@@ -327,9 +327,8 @@ def create_project_version(id):
             logging.error(f"Failed to create upload session: {str(err)}")
             return AnotherUploadRunning().response(409)
 
-    # Create transaction folder and lockfile
+    # Create transaction folder
     os.makedirs(upload.upload_dir)
-    open(upload.lockfile, "w").close()
 
     file_changes, errors = upload.process_chunks(use_shared_chunk_dir=True)
     # files consistency or geodiff related issues, project push would never succeed, whole upload is aborted

server/mergin/sync/utils.py

@@ -57,53 +57,6 @@ def generate_checksum(file, chunk_size=4096):
             checksum.update(chunk)
 
 
-class Toucher:
-    """
-    Helper class to periodically update modification time of file during
-    execution of longer lasting task.
-
-    Example of usage:
-    -----------------
-    with Toucher(file, interval):
-        do_something_slow
-
-    """
-
-    def __init__(self, lockfile, interval):
-        self.lockfile = lockfile
-        self.interval = interval
-        self.running = False
-        self.timer = None
-
-    def __enter__(self):
-        self.acquire()
-
-    def __exit__(self, type, value, tb):  # pylint: disable=W0612,W0622
-        self.release()
-
-    def release(self):
-        self.running = False
-        if self.timer:
-            self.timer.cancel()
-            self.timer = None
-
-    def acquire(self):
-        self.running = True
-        self.touch_lockfile()
-
-    def touch_lockfile(self):
-        # do an NFS ACCESS procedure request to clear the attribute cache (for various pods to actually see the file)
-        # https://docs.aws.amazon.com/efs/latest/ug/troubleshooting-efs-general.html#custom-nfs-settings-write-delays
-        os.access(self.lockfile, os.W_OK)
-        with open(self.lockfile, "a"):
-            os.utime(self.lockfile, None)
-
-        sleep(0)  # to unblock greenlet
-        if self.running:
-            self.timer = Timer(self.interval, self.touch_lockfile)
-            self.timer.start()
-
-
 def is_qgis(path: str) -> bool:
     """
     Check if file is a QGIS project file.

server/mergin/tests/test_project_controller.py

@@ -1294,7 +1294,6 @@ def create_transaction(username, changes, version=1):
     db.session.commit()
     upload_dir = os.path.join(upload.project.storage.project_dir, "tmp", upload.id)
     os.makedirs(upload_dir)
-    open(os.path.join(upload_dir, "lockfile"), "w").close()
     return upload, upload_dir
 
 
@@ -1320,6 +1319,7 @@ def test_chunk_upload(client, app):
     resp = client.post(url, data=data, headers=headers)
     assert resp.status_code == 200
     assert resp.json["checksum"] == checksum.hexdigest()
+    assert os.path.exists(os.path.join(upload_dir, "chunks", chunk_id))
 
     # tests to send bigger chunk than allowed
     app.config["MAX_CHUNK_SIZE"] = 10 * CHUNK_SIZE
@@ -1332,6 +1332,8 @@ def test_chunk_upload(client, app):
     failure = SyncFailuresHistory.query.filter_by(project_id=upload.project.id).first()
     assert failure.error_type == "chunk_upload"
     assert failure.error_details == "Too big chunk"
+    # residual after upload was removed
+    assert not os.path.exists(os.path.join(upload_dir, "chunks", chunk_id))
 
     # tests with transaction with no uploads expected
     changes = _get_changes(test_project_dir)
@@ -1342,9 +1344,8 @@ def test_chunk_upload(client, app):
     resp2 = client.post(url, data=data, headers=headers)
     assert resp2.status_code == 404
     assert SyncFailuresHistory.query.count() == 1
-
-    # cleanup
-    shutil.rmtree(upload_dir)
+    # we do not have any chunks, so parent dir was removed as well
+    assert not os.path.exists(os.path.join(upload_dir))
 
 
 def upload_chunks(upload_dir, changes, src_dir=test_project_dir):

server/mergin/tests/test_public_api_v2.py

@@ -1035,7 +1035,6 @@ def test_create_version_failures(client):
     db.session.add(upload)
     db.session.commit()
     os.makedirs(upload.upload_dir)
-    open(upload.lockfile, "w").close()
 
     response = client.post(f"v2/projects/{project.id}/versions", json=data)
     assert response.status_code == 409

server/migrations/community/e3a7f2b1c94d_add_upload_last_ping.py

@@ -0,0 +1,29 @@
+"""Add last_ping to upload
+
+Revision ID: e3a7f2b1c94d
+Revises: 4b4648483770
+Create Date: 2026-04-14 00:00:00.000000
+
+"""
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = "e3a7f2b1c94d"
+down_revision = "4b4648483770"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.add_column("upload", sa.Column("last_ping", sa.DateTime(), nullable=True))
+    # backfill existing rows before adding NOT NULL constraint
+    op.execute("UPDATE upload SET last_ping = NOW() WHERE last_ping IS NULL")
+    op.alter_column("upload", "last_ping", nullable=False)
+
+
+def downgrade():
+    # drop the column but required lockfiles will be missing - make sure all uploads are gone
+    op.drop_column("upload", "last_ping")