fix(reconciliation): serialize compute sync and enforce UUID identity

computes/tests.py

@@ -1,7 +1,7 @@
 import os
 from django.core.exceptions import ObjectDoesNotExist
 from django.shortcuts import reverse
-from django.test import TestCase
+from django.test import TestCase, TransactionTestCase
 
 from vrtManager.connection import CONN_SOCKET, connection_manager
 from .models import Compute
@@ -185,3 +185,133 @@ class ComputesTestCase(TestCase):
             )
         )
         self.assertEqual(response.status_code, 200)
+
+
+class ComputeConcurrencyTestCase(TransactionTestCase):
+    def test_concurrent_refresh_instance_database(self):
+        import threading
+        from django.db import connection
+        from unittest.mock import MagicMock
+        from computes.utils import refresh_instance_database
+        from instances.models import Instance
+
+        compute = Compute.objects.create(
+            name="concurrent-test",
+            hostname="127.0.0.1",
+            login="test",
+            password="pwd",
+            type=CONN_SOCKET,
+        )
+
+        mock_dom1 = MagicMock()
+        mock_dom1.UUIDString.return_value = "11111111-2222-3333-4444-555555555555"
+        mock_dom1.name.return_value = "vm-concurrent-1"
+
+        mock_dom2 = MagicMock()
+        mock_dom2.UUIDString.return_value = "22222222-3333-4444-5555-666666666666"
+        mock_dom2.name.return_value = "vm-concurrent-2"
+
+        mock_proxy = MagicMock()
+        mock_proxy.wvm.listAllDomains.return_value = [mock_dom1, mock_dom2]
+        compute.proxy = mock_proxy
+
+        exceptions = []
+
+        def worker():
+            try:
+                refresh_instance_database(compute)
+            except Exception as e:
+                exceptions.append(e)
+            finally:
+                connection.close()
+
+        connection.close()
+        threads = [threading.Thread(target=worker) for _ in range(5)]
+        for t in threads:
+            t.start()
+        for t in threads:
+            t.join()
+
+        self.assertEqual(exceptions, [])
+        self.assertEqual(Instance.objects.filter(compute=compute).count(), 2)
+
+    def test_refresh_fails_closed_on_lock_error(self):
+        from unittest.mock import patch, MagicMock
+        from computes.utils import refresh_instance_database
+        from instances.models import Instance
+
+        compute = Compute.objects.create(
+            name="lock-fail-test",
+            hostname="127.0.0.1",
+            login="test",
+            password="pwd",
+            type=CONN_SOCKET,
+        )
+        Instance.objects.create(compute=compute, name="existing-vm", uuid="existing-uuid-1234")
+
+        # Libvirt returns empty list - if it did not fail closed, existing-vm would be deleted!
+        mock_proxy = MagicMock()
+        mock_proxy.wvm.listAllDomains.return_value = []
+        compute.proxy = mock_proxy
+
+        # 1. When lock directory cannot be acquired
+        with patch("computes.utils._get_lock_directory", return_value=None):
+            refresh_instance_database(compute)
+            self.assertEqual(Instance.objects.filter(compute=compute).count(), 1)
+
+        # 2. When lock file is a symlink
+        with patch("os.path.islink", return_value=True):
+            refresh_instance_database(compute)
+            self.assertEqual(Instance.objects.filter(compute=compute).count(), 1)
+
+        # 3. When lock file cannot be opened
+        with patch("os.open", side_effect=OSError("Permission denied")):
+            refresh_instance_database(compute)
+            self.assertEqual(Instance.objects.filter(compute=compute).count(), 1)
+
+        # 4. When flock times out
+        curr_time = [0.0]
+        def advance_time():
+            curr_time[0] += 20.0
+            return curr_time[0]
+
+        with patch("fcntl.flock", side_effect=BlockingIOError("Resource temporarily unavailable")):
+            with patch("time.time", side_effect=advance_time):
+                refresh_instance_database(compute)
+                self.assertEqual(Instance.objects.filter(compute=compute).count(), 1)
+
+    def test_libvirt_compute_lock_thread_timeout(self):
+        from computes.utils import libvirt_compute_lock, _get_compute_thread_lock
+        compute = Compute.objects.create(
+            name="thread-lock-test",
+            hostname="127.0.0.1",
+            login="test",
+            password="pwd",
+            type=CONN_SOCKET,
+        )
+        tlock = _get_compute_thread_lock(compute.id)
+        tlock.acquire()
+        try:
+            with self.assertRaises(TimeoutError):
+                with libvirt_compute_lock(compute, timeout=0.05):
+                    pass
+        finally:
+            tlock.release()
+
+    def test_libvirt_compute_lock_reentrant(self):
+        from computes.utils import libvirt_compute_lock
+        compute = Compute.objects.create(
+            name="reentrant-lock-test",
+            hostname="127.0.0.1",
+            login="test",
+            password="pwd",
+            type=CONN_SOCKET,
+        )
+        executed = []
+        with libvirt_compute_lock(compute, timeout=1.0):
+            executed.append("outer")
+            with libvirt_compute_lock(compute, timeout=1.0):
+                executed.append("inner")
+                with libvirt_compute_lock(compute, timeout=1.0):
+                    executed.append("deep")
+        self.assertEqual(executed, ["outer", "inner", "deep"])

computes/utils.py

@@ -1,9 +1,156 @@
+from contextlib import contextmanager
+import fcntl
 import logging
+import os
+import stat
+import tempfile
+import threading
+import time
 from django.db import transaction
+from computes.models import Compute
 from instances.models import Instance
 
 logger = logging.getLogger(__name__)
 
+_COMPUTE_THREAD_LOCKS = {}
+_COMPUTE_THREAD_LOCKS_GUARD = threading.Lock()
+
+
+def _get_compute_thread_lock(compute_pk):
+    with _COMPUTE_THREAD_LOCKS_GUARD:
+        if compute_pk not in _COMPUTE_THREAD_LOCKS:
+            _COMPUTE_THREAD_LOCKS[compute_pk] = threading.Lock()
+        return _COMPUTE_THREAD_LOCKS[compute_pk]
+
+
+def _get_lock_directory():
+    candidates = [
+        "/srv/webvirtcloud/data/locks",
+        os.path.join(tempfile.gettempdir(), f"webvirtcloud_locks_{os.geteuid()}"),
+    ]
+    for candidate in candidates:
+        try:
+            if os.path.islink(candidate):
+                continue
+            if not os.path.exists(candidate):
+                os.makedirs(candidate, mode=0o700, exist_ok=True)
+            if not os.path.isdir(candidate) or os.path.islink(candidate):
+                continue
+
+            stat_info = os.stat(candidate)
+            # Ensure candidate directory is owned by the current process user
+            if stat_info.st_uid != os.geteuid():
+                continue
+            # Ensure candidate directory has private permissions (mode 0700)
+            if (stat.S_IMODE(stat_info.st_mode) & 0o077) != 0:
+                try:
+                    os.chmod(candidate, 0o700)
+                except OSError:
+                    continue
+                stat_info = os.stat(candidate)
+                if (stat.S_IMODE(stat_info.st_mode) & 0o077) != 0:
+                    continue
+
+            test_file = os.path.join(candidate, f".test_write_{os.getpid()}")
+            with open(test_file, "w") as tf:
+                tf.write("1")
+            os.unlink(test_file)
+            return candidate
+        except (OSError, PermissionError):
+            continue
+    return None
+
+
+_thread_local = threading.local()
+
+
+def _get_thread_held_locks():
+    if not hasattr(_thread_local, "compute_locks"):
+        _thread_local.compute_locks = {}
+    return _thread_local.compute_locks
+
+
+@contextmanager
+def libvirt_compute_lock(compute, timeout=15.0):
+    """
+    Context manager providing multi-thread and multi-process synchronization
+    for libvirt compute operations and instance database synchronization.
+    Supports re-entrant locking within the same thread.
+    Fails closed on lock directory or permission errors.
+    """
+    compute_pk = getattr(compute, "pk", getattr(compute, "id", None))
+    if not compute_pk:
+        yield
+        return
+
+    held_locks = _get_thread_held_locks()
+    if compute_pk in held_locks:
+        held_locks[compute_pk] += 1
+        try:
+            yield
+        finally:
+            held_locks[compute_pk] -= 1
+            if held_locks[compute_pk] <= 0:
+                held_locks.pop(compute_pk, None)
+        return
+
+    thread_lock = _get_compute_thread_lock(compute_pk)
+    start_time = time.time()
+    acquired_thread = thread_lock.acquire(timeout=timeout)
+    if not acquired_thread:
+        logger.error("Timeout waiting for in-process thread lock on compute %s; aborting to fail closed", compute_pk)
+        raise TimeoutError(f"Timeout waiting for lock on compute {compute_pk}")
+
+    try:
+        lock_dir = _get_lock_directory()
+        if not lock_dir:
+            logger.error("Failed to acquire application lock directory for compute %s; aborting to fail closed", compute_pk)
+            raise OSError(f"Could not acquire secure lock directory for compute {compute_pk}")
+
+        lock_file_path = os.path.join(lock_dir, f"sync_compute_{compute_pk}.lock")
+        if os.path.islink(lock_file_path):
+            logger.error("Lock file %s is a symlink! Aborting to fail closed", lock_file_path)
+            raise PermissionError(f"Compute lock file {lock_file_path} is a symlink")
+
+        flags = os.O_CREAT | os.O_RDWR | getattr(os, "O_NOFOLLOW", 0)
+        try:
+            lock_fd = os.open(lock_file_path, flags, 0o600)
+        except OSError as e:
+            logger.error("Could not open synchronization lock file for compute %s: %s; aborting to fail closed", compute_pk, e)
+            raise
+
+        lock_acquired = False
+        try:
+            while time.time() - start_time < timeout:
+                try:
+                    fcntl.flock(lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
+                    lock_acquired = True
+                    break
+                except (BlockingIOError, OSError):
+                    time.sleep(0.05)
+
+            if not lock_acquired:
+                logger.error("Timeout waiting for cross-process synchronization lock on compute %s; aborting to fail closed", compute_pk)
+                raise TimeoutError(f"Timeout waiting for lock on compute {compute_pk}")
+
+            held_locks[compute_pk] = 1
+            try:
+                yield
+            finally:
+                held_locks.pop(compute_pk, None)
+        finally:
+            if lock_acquired:
+                try:
+                    fcntl.flock(lock_fd, fcntl.LOCK_UN)
+                except OSError:
+                    pass
+            try:
+                os.close(lock_fd)
+            except OSError:
+                pass
+    finally:
+        thread_lock.release()
+
 
 def refresh_instance_database(compute):
     """
@@ -16,49 +163,66 @@ def refresh_instance_database(compute):
       - If libvirt is unreachable (connection/network error), the database is left
         untouched to avoid false deletions.
     """
-    try:
-        if hasattr(compute, "status") and compute.status is False:
-            return
-        domains = compute.proxy.wvm.listAllDomains()
-    except Exception as e:
-        logger.warning(
-            "Failed to retrieve domains from compute %s (%s): %s",
-            getattr(compute, "name", compute),
-            getattr(compute, "hostname", ""),
-            e,
-        )
+    compute_pk = getattr(compute, "pk", getattr(compute, "id", None))
+    if not compute_pk:
         return
 
-    # Map libvirt UUIDs to current domain names
-    host_domains = {}
-    for d in domains:
-        try:
-            host_domains[d.UUIDString()] = d.name()
-        except Exception as e:
-            logger.warning(
-                "Error reading domain metadata from compute %s: %s",
-                getattr(compute, "name", compute),
-                e,
-            )
-            # If we fail to read a domain's metadata, abort sync to prevent accidental deletion
-            return
+    try:
+        with libvirt_compute_lock(compute, timeout=15.0):
+            with transaction.atomic():
+                comp_locked = Compute.objects.select_for_update().filter(pk=compute_pk).first()
+                if not comp_locked:
+                    return
 
-    db_instances = Instance.objects.filter(compute=compute)
-    db_uuids = set(db_instances.values_list("uuid", flat=True))
+                if hasattr(compute, "status") and compute.status is False:
+                    return
 
-    with transaction.atomic():
-        # 1. Update name for any VM whose name changed out-of-band in virsh/virt-manager
-        for inst in db_instances:
-            if inst.uuid in host_domains:
-                new_name = host_domains[inst.uuid]
-                if inst.name != new_name:
-                    inst.name = new_name
-                    inst.save(update_fields=["name"])
+                try:
+                    domains = compute.proxy.wvm.listAllDomains()
+                except Exception as e:
+                    logger.warning(
+                        "Failed to retrieve domains from compute %s (%s): %s",
+                        getattr(compute, "name", compute),
+                        getattr(compute, "hostname", ""),
+                        e,
+                    )
+                    return
 
-        # 2. Delete instances that were genuinely undefined/deleted in virsh/virt-manager
-        db_instances.exclude(uuid__in=host_domains.keys()).delete()
+                # Map libvirt UUIDs to current domain names
+                host_domains = {}
+                for d in domains:
+                    try:
+                        host_domains[d.UUIDString()] = d.name()
+                    except Exception as e:
+                        logger.warning(
+                            "Error reading domain metadata from compute %s: %s",
+                            getattr(compute, "name", compute),
+                            e,
+                        )
+                        return
 
-        # 3. Create new instances that were created out-of-band in virsh/virt-manager
-        for uuid, name in host_domains.items():
-            if uuid not in db_uuids:
-                Instance.objects.create(compute=compute, name=name, uuid=uuid)
+                db_instances = list(Instance.objects.filter(compute=compute).select_for_update())
+                db_uuids = set(inst.uuid for inst in db_instances)
+
+                # 1. Update name for any VM whose name changed out-of-band in virsh/virt-manager
+                for inst in db_instances:
+                    if inst.uuid in host_domains:
+                        new_name = host_domains[inst.uuid]
+                        if inst.name != new_name:
+                            inst.name = new_name
+                            inst.save(update_fields=["name"])
+
+                # 2. Delete instances that were genuinely undefined/deleted in virsh/virt-manager
+                Instance.objects.filter(compute=compute).exclude(uuid__in=host_domains.keys()).delete()
+
+                # 3. Create or update new instances that were created out-of-band in virsh/virt-manager
+                for uuid, name in host_domains.items():
+                    if uuid not in db_uuids:
+                        Instance.objects.get_or_create(
+                            compute=compute,
+                            uuid=uuid,
+                            defaults={"name": name},
+                        )
+    except (OSError, PermissionError, TimeoutError) as e:
+        logger.error("Could not synchronize instances for compute %s: %s; failing closed", compute_pk, e)
+        return