diff --git a/meshchatx/meshchat.py b/meshchatx/meshchat.py index e5c15999..c151ee20 100644 --- a/meshchatx/meshchat.py +++ b/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}") diff --git a/meshchatx/src/backend/database/__init__.py b/meshchatx/src/backend/database/__init__.py index 8241946e..2b976601 100644 --- a/meshchatx/src/backend/database/__init__.py +++ b/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(), } diff --git a/meshchatx/src/backend/database/provider.py b/meshchatx/src/backend/database/provider.py index 124fccae..f5a58bcb 100644 --- a/meshchatx/src/backend/database/provider.py +++ b/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 diff --git a/meshchatx/src/backend/database/schema.py b/meshchatx/src/backend/database/schema.py index 296e951b..2d4033fe 100644 --- a/meshchatx/src/backend/database/schema.py +++ b/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)), - ) diff --git a/meshchatx/src/backend/identity_context.py b/meshchatx/src/backend/identity_context.py index 331d2a7b..60385ce8 100644 --- a/meshchatx/src/backend/identity_context.py +++ b/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( diff --git a/meshchatx/src/backend/identity_manager.py b/meshchatx/src/backend/identity_manager.py index 3998d05d..78a736e5 100644 --- a/meshchatx/src/backend/identity_manager.py +++ b/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 = { diff --git a/meshchatx/src/backend/integrity_manager.py b/meshchatx/src/backend/integrity_manager.py index a9f4bee8..a2a550de 100644 --- a/meshchatx/src/backend/integrity_manager.py +++ b/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): diff --git a/meshchatx/src/backend/storage_lock.py b/meshchatx/src/backend/storage_lock.py new file mode 100644 index 00000000..70a891e0 --- /dev/null +++ b/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 diff --git a/tests/backend/test_database_lifecycle_safety.py b/tests/backend/test_database_lifecycle_safety.py new file mode 100644 index 00000000..4d32cbdf --- /dev/null +++ b/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) diff --git a/tests/backend/test_database_snapshots.py b/tests/backend/test_database_snapshots.py index 8767e8f5..a9ef2671 100644 --- a/tests/backend/test_database_snapshots.py +++ b/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 diff --git a/tests/backend/test_integrity.py b/tests/backend/test_integrity.py index 07b69786..a5485937 100644 --- a/tests/backend/test_integrity.py +++ b/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):