feat(meshchat): implement storage locking and improve database restore process with relaunch capability

meshchatx/meshchat.py

@@ -331,6 +331,14 @@ class ReticulumMeshChat:
             reticulum_config_dir,
         )
         self.storage_dir = storage_dir or os.path.join("storage")
+        from meshchatx.src.backend.storage_lock import StorageLock, StorageLockError
+
+        self._storage_lock = StorageLock(self.storage_dir)
+        try:
+            self._storage_lock.acquire()
+        except StorageLockError as exc:
+            print(str(exc))
+            raise SystemExit(1) from exc
         self.ssl_cert_path = ssl_cert_path
         self.ssl_key_path = ssl_key_path
         self.identity_file_path = identity_file_path
@@ -693,10 +701,42 @@ class ReticulumMeshChat:
             raise RuntimeError("Database not initialized")
         return self.database.backup_database(self.storage_dir, backup_path)
 
-    def restore_database(self, backup_path):
-        if not self.database:
-            raise RuntimeError("Database not initialized")
-        return self.database.restore_database(backup_path)
+    def prepare_for_database_restore(self) -> str | None:
+        db_path = self.database_path
+        self._teardown_all_contexts_for_reload()
+        from meshchatx.src.backend.database.provider import DatabaseProvider
+
+        if DatabaseProvider._instance is not None:
+            DatabaseProvider._instance.close_all()
+            DatabaseProvider._instance = None
+        return db_path
+
+    @staticmethod
+    def _schedule_process_restart(delay: float = 1.0) -> None:
+        def restart():
+            time.sleep(delay)
+            try:
+                os.execv(sys.executable, [sys.executable] + sys.argv)  # noqa: S606
+            except Exception as e:
+                print(f"Failed to restart: {e}")
+                os._exit(0)
+
+        threading.Thread(target=restart, daemon=True).start()
+
+    def restore_database(self, backup_path, *, relaunch: bool = False):
+        db_path = self.prepare_for_database_restore()
+        if not db_path:
+            raise RuntimeError("Database path is unknown")
+        from meshchatx.src.backend.database import Database
+
+        db = Database(db_path)
+        try:
+            result = db.restore_database(backup_path)
+        finally:
+            db.close_all()
+        if relaunch:
+            self._schedule_process_restart()
+        return result
 
     def reset_password(self):
         """Clear the stored password hash so a new password can be set via the web UI."""
@@ -1910,7 +1950,7 @@ class ReticulumMeshChat:
             print(f"Auto recovery completed: {actions}")
         finally:
             try:
-                self.database.close()
+                self.database.close_all()
             except Exception as e:
                 print(f"Failed to close database during recovery: {e}")
 
@@ -3792,9 +3832,14 @@ class ReticulumMeshChat:
                             status=404,
                         )
 
-                result = self.database.restore_database(path)
+                result = self.restore_database(path, relaunch=True)
                 return web.json_response(
-                    {"status": "success", "result": result, "requires_relaunch": True},
+                    {
+                        "status": "success",
+                        "result": result,
+                        "requires_relaunch": True,
+                        "message": "Database restored. Application will restart.",
+                    },
                 )
             except Exception as e:
                 return web.json_response(
@@ -6443,13 +6488,14 @@ class ReticulumMeshChat:
                         tmp.write(chunk)
                     temp_path = tmp.name
 
-                result = self.database.restore_database(temp_path)
+                result = self.restore_database(temp_path, relaunch=True)
                 os.remove(temp_path)
 
                 return web.json_response(
                     {
-                        "message": "Database restored successfully",
+                        "message": "Database restored successfully. Application will restart.",
                         "database": result,
+                        "requires_relaunch": True,
                     },
                 )
             except Exception as e:
@@ -18249,6 +18295,7 @@ def main():
             print(
                 f"Snapshot restoration complete. Integrity check: {result['integrity_check']}",
             )
+            reticulum_meshchat.setup_identity(identity)
         else:
             print(f"Error: Snapshot not found at {snapshot_path}")
 

meshchatx/src/backend/database/__init__.py

@@ -32,6 +32,11 @@ MIN_SIZE_RATIO = 0.2
 
 _log = logging.getLogger("meshchatx.database")
 
+
+class DatabaseRestoreError(RuntimeError):
+    pass
+
+
 _PRAGMA_READ_NAME_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*\Z")
 
 _ALLOWED_WAL_CHECKPOINT_MODES = frozenset({"PASSIVE", "FULL", "RESTART", "TRUNCATE"})
@@ -373,13 +378,12 @@ class Database:
         except Exception as e:
             print(f"Failed to checkpoint WAL: {e}")
         try:
-            self.close()
+            self.close_all()
         except Exception as e:
             print(f"Failed to close database: {e}")
 
     def close(self):
-        if hasattr(self, "provider"):
-            self.provider.close()
+        self.close_all()
 
     def close_all(self):
         if hasattr(self, "provider"):
@@ -538,6 +542,14 @@ class Database:
             return True
         return False
 
+    @staticmethod
+    def _looks_like_sqlite(path: str) -> bool:
+        try:
+            with open(path, "rb") as handle:
+                return handle.read(16) == b"SQLite format 3\x00"
+        except OSError:
+            return False
+
     def restore_database(self, backup_path: str):
         if not os.path.exists(backup_path):
             msg = f"Backup not found at {backup_path}"
@@ -546,7 +558,6 @@ class Database:
         paths = self._database_paths()
         self._checkpoint_and_close()
 
-        # clean existing files
         for p in paths.values():
             if os.path.exists(p):
                 os.remove(p)
@@ -557,12 +568,30 @@ class Database:
         else:
             shutil.copy2(backup_path, paths["main"])
 
-        # reopen and retune
-        self.initialize()
+        if not self._looks_like_sqlite(paths["main"]):
+            raise DatabaseRestoreError("Restored file is not a valid SQLite database")
+
+        try:
+            self.initialize()
+        except Exception as exc:
+            raise DatabaseRestoreError(
+                f"Restored files from backup but database failed to open: {exc!s}",
+            ) from exc
         self._tune_sqlite_pragmas()
-        integrity = self.provider.integrity_check()
+        integrity_rows = self.provider.integrity_check()
+        integrity = []
+        for row in integrity_rows or []:
+            if isinstance(row, dict):
+                integrity.append(next(iter(row.values())))
+            else:
+                integrity.append(row[0])
+        if integrity and integrity[0] != "ok":
+            raise DatabaseRestoreError(
+                f"Restored backup failed integrity check: {integrity[0]!s}",
+            )
 
         return {
             "restored_from": backup_path,
-            "integrity_check": integrity,
+            "integrity_check": integrity_rows,
+            "health": self.get_database_health_snapshot(),
         }

meshchatx/src/backend/database/provider.py

@@ -12,7 +12,7 @@ if sys.version_info >= (3, 14):
 
 class DatabaseProvider:
     _instance = None
-    _lock = threading.Lock()
+    _lock = threading.RLock()
     _all_locals = weakref.WeakSet()
 
     def __init__(self, db_path=None):
@@ -30,8 +30,7 @@ class DatabaseProvider:
                     raise ValueError(msg)
                 cls._instance = cls(db_path)
             elif db_path is not None and cls._instance.db_path != db_path:
-                # If a different path is provided, close the old one and create new
-                cls._instance.close()
+                cls._instance.close_all()
                 cls._instance = cls(db_path)
             return cls._instance
 

meshchatx/src/backend/database/schema.py

@@ -7,6 +7,10 @@ from .provider import DatabaseProvider
 _IDENTIFIER_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
 
 
+class DatabaseMigrationError(RuntimeError):
+    pass
+
+
 def _validate_identifier(name: str, label: str = "identifier") -> str:
     if not _IDENTIFIER_RE.match(name):
         msg = f"Invalid SQL {label}: {name!r}"
@@ -19,15 +23,18 @@ class DatabaseSchema:
 
     def __init__(self, provider: DatabaseProvider):
         self.provider = provider
+        self._strict_migrations = False
+        self._migration_errors: list[str] = []
 
     def _safe_execute(self, query, params=None):
         try:
             return self.provider.execute(query, params)
         except Exception as e:
-            # Silence expected errors during migrations (e.g. duplicate columns/indexes)
             err_msg = str(e).lower()
             if "duplicate column name" in err_msg or "already exists" in err_msg:
                 return None
+            if self._strict_migrations:
+                self._migration_errors.append(str(e))
             print(f"Database operation failed: {query[:100]}... Error: {e}")
             return None
 
@@ -565,7 +572,33 @@ class DatabaseSchema:
                     "CREATE INDEX IF NOT EXISTS idx_debug_logs_anomaly ON debug_logs(is_anomaly)",
                 )
 
+    def _update_database_version(self):
+        self.provider.execute(
+            """
+            INSERT INTO config (key, value, created_at, updated_at)
+            VALUES (?, ?, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
+            ON CONFLICT(key) DO UPDATE SET
+                value = EXCLUDED.value,
+                updated_at = EXCLUDED.updated_at
+            """,
+            ("database_version", str(self.LATEST_VERSION)),
+        )
+
     def migrate(self, current_version):
+        self._strict_migrations = True
+        self._migration_errors = []
+        try:
+            self._run_migrations(current_version)
+        finally:
+            self._strict_migrations = False
+        if self._migration_errors:
+            first = self._migration_errors[0]
+            raise DatabaseMigrationError(
+                f"{len(self._migration_errors)} migration step(s) failed: {first}",
+            )
+        self._update_database_version()
+
+    def _run_migrations(self, current_version):
         if current_version < 7:
             self._safe_execute("""
                 CREATE TABLE IF NOT EXISTS archived_pages (
@@ -1261,15 +1294,3 @@ class DatabaseSchema:
         if current_version < 48:
             self._ensure_column("lxmf_messages", "path_finding_measure", "TEXT")
             self._ensure_column("lxmf_messages", "path_row_hash_hex", "TEXT")
-
-        # Update version in config
-        self._safe_execute(
-            """
-            INSERT INTO config (key, value, created_at, updated_at) 
-            VALUES (?, ?, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
-            ON CONFLICT(key) DO UPDATE SET 
-                value = EXCLUDED.value,
-                updated_at = EXCLUDED.updated_at
-            """,
-            ("database_version", str(self.LATEST_VERSION)),
-        )

meshchatx/src/backend/identity_context.py

@@ -185,8 +185,6 @@ class IdentityContext:
             self.database,
         )
 
-        # Vacuum and mark stuck messages
-        self.database.provider.vacuum()
         self.database.messages.mark_stuck_messages_as_failed()
 
         if not getattr(self.app, "emergency", False):
@@ -654,7 +652,6 @@ class IdentityContext:
                         print(
                             f"Database health at close for {self.identity_hash}: {', '.join(close_issues)}",
                         )
-                # 1. Checkpoint WAL and close database cleanly to ensure file is stable for hashing
                 self.database._checkpoint_and_close()
             except Exception as e:
                 print(

meshchatx/src/backend/identity_manager.py

@@ -122,7 +122,7 @@ class IdentityManager:
                 )
                 lxmf_address = temp_config_dao.get("lxmf_address_hash")
                 lxst_address = temp_config_dao.get("lxst_address_hash")
-                temp_provider.close()
+                temp_provider.close_all()
 
                 # Save metadata for next time
                 metadata = {
@@ -179,7 +179,7 @@ class IdentityManager:
             new_config_dao = ConfigDAO(new_provider)
             new_config_dao.set("display_name", display_name)
 
-        new_provider.close()
+        new_provider.close_all()
 
         # Save metadata
         metadata = {

meshchatx/src/backend/integrity_manager.py

@@ -148,15 +148,12 @@ class IntegrityManager:
                 actual_db_hash = self._hash_file(self.database_path)
 
                 if actual_db_hash != manifest_files.get(db_rel):
-                    # Check internal SQL integrity to see if it's just a dirty shutdown or actual tampering
                     is_db_ok, db_msg = self._check_db_integrity(self.database_path)
                     if not is_db_ok:
                         issues.append(f"Database structural issue: {db_msg}")
                     else:
-                        # Check entropy stability to see if content type shifted significantly
                         actual_entropy = self._calculate_entropy(self.database_path)
                         saved_entropy = manifest_metadata.get(db_rel, {}).get("entropy")
-
                         if (
                             saved_entropy is not None
                             and abs(actual_entropy - saved_entropy) > 1.0
@@ -164,10 +161,6 @@ class IntegrityManager:
                             issues.append(
                                 f"Database structural anomaly (Entropy Δ: {abs(actual_entropy - saved_entropy):.2f})",
                             )
-                        else:
-                            issues.append(
-                                f"Database binary signature mismatch: {db_rel}",
-                            )
 
             # Check other critical files in storage_dir
             for root, _, files_in_dir in os.walk(self.storage_dir):

meshchatx/src/backend/storage_lock.py

@@ -0,0 +1,62 @@
+# SPDX-License-Identifier: 0BSD
+
+import atexit
+import os
+import sys
+
+
+class StorageLockError(OSError):
+    pass
+
+
+class StorageLock:
+    def __init__(self, storage_dir: str):
+        self.storage_dir = os.path.abspath(storage_dir)
+        self.lock_path = os.path.join(self.storage_dir, ".meshchatx.lock")
+        self._handle = None
+
+    def acquire(self) -> None:
+        os.makedirs(self.storage_dir, exist_ok=True)
+        self._handle = open(self.lock_path, "a+b")
+        try:
+            if sys.platform == "win32":
+                import msvcrt
+
+                self._handle.seek(0)
+                msvcrt.locking(self._handle.fileno(), msvcrt.LK_NBLCK, 1)
+            else:
+                import fcntl
+
+                fcntl.flock(self._handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
+        except OSError as exc:
+            self._handle.close()
+            self._handle = None
+            raise StorageLockError(
+                f"Another MeshChatX instance is already using storage at {self.storage_dir}",
+            ) from exc
+        self._handle.seek(0)
+        self._handle.truncate()
+        self._handle.write(str(os.getpid()).encode())
+        self._handle.flush()
+        atexit.register(self.release)
+
+    def release(self) -> None:
+        if self._handle is None:
+            return
+        try:
+            if sys.platform == "win32":
+                import msvcrt
+
+                self._handle.seek(0)
+                msvcrt.locking(self._handle.fileno(), msvcrt.LK_UNLCK, 1)
+            else:
+                import fcntl
+
+                fcntl.flock(self._handle.fileno(), fcntl.LOCK_UN)
+        except OSError:
+            pass
+        try:
+            self._handle.close()
+        except OSError:
+            pass
+        self._handle = None

tests/backend/test_database_lifecycle_safety.py

@@ -0,0 +1,232 @@
+# SPDX-License-Identifier: 0BSD
+
+import os
+import shutil
+import sqlite3
+import tempfile
+import unittest
+from unittest.mock import patch
+
+import pytest
+
+from meshchatx.src.backend.database import Database, DatabaseRestoreError
+from meshchatx.src.backend.database.provider import DatabaseProvider
+from meshchatx.src.backend.database.schema import DatabaseMigrationError, DatabaseSchema
+from meshchatx.src.backend.integrity_manager import IntegrityManager
+from meshchatx.src.backend.storage_lock import StorageLock, StorageLockError
+
+
+@pytest.fixture
+def temp_dir():
+    path = tempfile.mkdtemp()
+    yield path
+    shutil.rmtree(path, ignore_errors=True)
+
+
+@pytest.fixture(autouse=True)
+def reset_provider():
+    DatabaseProvider._instance = None
+    yield
+    if DatabaseProvider._instance is not None:
+        DatabaseProvider._instance.close_all()
+    DatabaseProvider._instance = None
+
+
+def test_provider_path_switch_does_not_deadlock(temp_dir):
+    db_path_a = os.path.join(temp_dir, "a.db")
+    db_path_b = os.path.join(temp_dir, "b.db")
+    DatabaseProvider.get_instance(db_path_a)
+    provider_b = DatabaseProvider.get_instance(db_path_b)
+    assert provider_b.db_path == db_path_b
+    DatabaseProvider._instance.close_all()
+
+
+def test_provider_path_switch_calls_close_all(temp_dir):
+    db_path_a = os.path.join(temp_dir, "a.db")
+    db_path_b = os.path.join(temp_dir, "b.db")
+    provider_a = DatabaseProvider.get_instance(db_path_a)
+    with patch.object(provider_a, "close_all") as mock_close:
+        DatabaseProvider.get_instance(db_path_b)
+        mock_close.assert_called_once()
+    DatabaseProvider._instance.close_all()
+
+
+def test_restore_invokes_close_all_before_replace(temp_dir):
+    db_path = os.path.join(temp_dir, "live.db")
+    db = Database(db_path)
+    db.initialize()
+    db.execute_sql("INSERT INTO config (key, value) VALUES (?, ?)", ("k", "v1"))
+    backup_path = os.path.join(temp_dir, "backup.zip")
+    db.backup_database(temp_dir, backup_path=backup_path)
+    with patch.object(
+        db.provider, "close_all", wraps=db.provider.close_all
+    ) as mock_close:
+        db.restore_database(backup_path)
+        assert mock_close.call_count >= 1
+    row = db.provider.fetchone("SELECT value FROM config WHERE key = ?", ("k",))
+    assert row["value"] == "v1"
+    db.close_all()
+
+
+def test_migration_failure_does_not_bump_version(temp_dir):
+    db_path = os.path.join(temp_dir, "broken_migrate.db")
+    provider = DatabaseProvider.get_instance(db_path)
+    schema = DatabaseSchema(provider)
+    provider.execute(
+        """
+        CREATE TABLE config (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            key TEXT UNIQUE,
+            value TEXT
+        )
+        """,
+    )
+    provider.execute(
+        "INSERT INTO config (key, value) VALUES (?, ?)",
+        ("database_version", "47"),
+    )
+
+    def fail_run(_current_version):
+        schema._migration_errors.append("simulated migration failure")
+
+    schema._run_migrations = fail_run
+    with pytest.raises(DatabaseMigrationError):
+        schema.migrate(47)
+
+    row = provider.fetchone(
+        "SELECT value FROM config WHERE key = 'database_version'",
+    )
+    assert int(row["value"]) == 47
+    provider.close_all()
+
+
+def test_integrity_allows_hash_change_when_sqlite_ok(temp_dir):
+    db_path = os.path.join(temp_dir, "database.db")
+    conn = sqlite3.connect(db_path)
+    conn.execute("CREATE TABLE data (id INTEGER PRIMARY KEY, val TEXT)")
+    conn.execute("INSERT INTO data (val) VALUES ('x')")
+    conn.commit()
+    conn.close()
+
+    manager = IntegrityManager(temp_dir, db_path)
+    manager.save_manifest()
+
+    conn = sqlite3.connect(db_path)
+    conn.execute("INSERT INTO data (val) VALUES ('y')")
+    conn.commit()
+    conn.close()
+
+    is_ok, issues = manager.check_integrity()
+    assert is_ok, issues
+
+
+def test_integrity_flags_structural_damage(temp_dir):
+    db_path = os.path.join(temp_dir, "database.db")
+    conn = sqlite3.connect(db_path)
+    conn.execute("CREATE TABLE data (id INTEGER PRIMARY KEY)")
+    conn.commit()
+    conn.close()
+
+    manager = IntegrityManager(temp_dir, db_path)
+    manager.save_manifest()
+
+    with open(db_path, "r+b") as handle:
+        handle.seek(0)
+        handle.write(b"NOTASQLITEFILE")
+
+    is_ok, issues = manager.check_integrity()
+    assert not is_ok
+    assert any("Database structural issue" in i for i in issues)
+
+
+def test_storage_lock_rejects_second_instance(temp_dir):
+    lock_a = StorageLock(temp_dir)
+    lock_a.acquire()
+    lock_b = StorageLock(temp_dir)
+    with pytest.raises(StorageLockError):
+        lock_b.acquire()
+    lock_a.release()
+
+
+def test_restore_rejects_non_sqlite_backup(temp_dir):
+    db_path = os.path.join(temp_dir, "main.db")
+    db = Database(db_path)
+    db.initialize()
+
+    bad_backup = os.path.join(temp_dir, "bad.db")
+    with open(bad_backup, "wb") as handle:
+        handle.write(b"not a sqlite database")
+
+    with pytest.raises(DatabaseRestoreError, match="not a valid SQLite"):
+        db.restore_database(bad_backup)
+    db.close_all()
+
+
+def test_looks_like_sqlite_header():
+    path = tempfile.NamedTemporaryFile(delete=False).name
+    try:
+        conn = sqlite3.connect(path)
+        conn.execute("CREATE TABLE t (id INTEGER)")
+        conn.close()
+        assert Database._looks_like_sqlite(path)
+        with open(path, "r+b") as handle:
+            handle.seek(0)
+            handle.write(b"garbage")
+        assert not Database._looks_like_sqlite(path)
+    finally:
+        os.remove(path)
+
+
+class TestRestoreDatabaseMethod(unittest.TestCase):
+    def setUp(self):
+        self.test_dir = tempfile.mkdtemp()
+        self.db_path = os.path.join(self.test_dir, "test.db")
+
+    def tearDown(self):
+        if DatabaseProvider._instance is not None:
+            DatabaseProvider._instance.close_all()
+        DatabaseProvider._instance = None
+        shutil.rmtree(self.test_dir)
+
+    def test_checkpoint_and_close_uses_close_all(self):
+        db = Database(self.db_path)
+        db.initialize()
+        with patch.object(db.provider, "close_all") as mock_close:
+            db._checkpoint_and_close()
+            mock_close.assert_called_once()
+
+
+class TestMeshchatRestoreFlow(unittest.TestCase):
+    @patch("meshchatx.meshchat.ReticulumMeshChat._schedule_process_restart")
+    def test_restore_database_prepares_and_schedules_restart(self, mock_restart):
+        from meshchatx.meshchat import ReticulumMeshChat
+
+        temp = tempfile.mkdtemp()
+        try:
+            db_path = os.path.join(temp, "identities", "abc", "database.db")
+            os.makedirs(os.path.dirname(db_path), exist_ok=True)
+            app = object.__new__(ReticulumMeshChat)
+            app.contexts = {}
+            app.current_context = None
+            app._teardown_all_contexts_for_reload = unittest.mock.Mock()
+
+            db = Database(db_path)
+            db.initialize()
+            backup_path = os.path.join(temp, "b.zip")
+            db.backup_database(temp, backup_path=backup_path)
+            db.close_all()
+
+            with patch.object(
+                ReticulumMeshChat,
+                "prepare_for_database_restore",
+                return_value=db_path,
+            ):
+                result = ReticulumMeshChat.restore_database(
+                    app,
+                    backup_path,
+                    relaunch=True,
+                )
+            assert result["restored_from"] == backup_path
+            mock_restart.assert_called_once()
+        finally:
+            shutil.rmtree(temp)

tests/backend/test_database_snapshots.py

@@ -8,6 +8,16 @@ import tempfile
 import pytest
 
 from meshchatx.src.backend.database import Database
+from meshchatx.src.backend.database.provider import DatabaseProvider
+
+
+@pytest.fixture(autouse=True)
+def reset_database_provider():
+    DatabaseProvider._instance = None
+    yield
+    if DatabaseProvider._instance is not None:
+        DatabaseProvider._instance.close_all()
+    DatabaseProvider._instance = None
 
 
 @pytest.fixture

tests/backend/test_integrity.py

@@ -49,13 +49,13 @@ class TestIntegrityManager(unittest.TestCase):
         """Test detection of database modification."""
         self.manager.save_manifest()
 
-        # Modify DB in a way that breaks SQLite integrity or at least changes hash
-        with open(self.db_path, "a") as f:
-            f.write("tampered")
+        with open(self.db_path, "r+b") as f:
+            f.seek(0)
+            f.write(b"NOTASQLITEFILE")
 
         is_ok, issues = self.manager.check_integrity()
         self.assertFalse(is_ok)
-        self.assertTrue(any("Database" in i for i in issues))
+        self.assertTrue(any("Database structural issue" in i for i in issues))
         self.assertTrue(any("Last integrity snapshot" in i for i in issues))
 
     def test_identity_mismatch(self):