From e14a084f0d380420fe4046e3348ef12c76cfe15e Mon Sep 17 00:00:00 2001 From: Will Miao Date: Thu, 17 Sep 2026 23:59:22 +0800 Subject: [PATCH] fix(cache): make shared cache state survive a second instance Installing a second LoRA Manager instance (standalone or a second ComfyUI install) that shares the settings directory puts two processes on the same cache databases. Three things made that unsafe. - The updater preserved cache/ and model_cache/ but not a legacy recipe_cache/ directory, so a portable install predating the cache/ move lost its recipe database on a git-based update. Add it to _PRESERVE_DIRS and to .gitignore. - Cache connections used the sqlite3 default 5s timeout, which a scanning instance can exceed, turning a concurrent write into "database is locked". Route every shared cache connection through connect_cache_db(), which raises the timeout to 30s and sets busy_timeout + synchronous=NORMAL to match the existing WAL mode. App-private databases (download queue, update history) are unchanged. - A full-table cache replace is a read-modify-write that SQLite cannot make atomic across processes, so two instances could interleave and one snapshot could overwrite the other. Guard the recipe and model save_cache paths with a cross-process advisory lock (flock on POSIX, msvcrt on Windows). Locking is best-effort: if it is unavailable the call proceeds and the SQLite busy timeout is the fallback. The lock file is a hidden sibling of the database and is deliberately never unlinked, so a second process cannot lock a fresh inode. --- .gitignore | 1 + py/routes/update_routes.py | 15 +- py/services/persistent_model_cache.py | 496 +++++++++++++------------ py/services/persistent_recipe_cache.py | 130 +++---- py/services/recipe_fts_index.py | 18 +- py/services/recipe_scanner.py | 4 + py/services/tag_fts_index.py | 18 +- py/utils/cache_db.py | 81 ++++ py/utils/file_lock.py | 146 ++++++++ tests/utils/test_cache_db.py | 107 ++++++ tests/utils/test_file_lock.py | 118 ++++++ 11 files changed, 804 insertions(+), 330 deletions(-) create mode 100644 py/utils/cache_db.py create mode 100644 py/utils/file_lock.py create mode 100644 tests/utils/test_cache_db.py create mode 100644 tests/utils/test_file_lock.py diff --git a/.gitignore b/.gitignore index 94a92695..beaa6cf9 100644 --- a/.gitignore +++ b/.gitignore @@ -15,6 +15,7 @@ node_modules/ coverage/ .coverage model_cache/ +recipe_cache/ # agent / dev tooling .opencode/ diff --git a/py/routes/update_routes.py b/py/routes/update_routes.py index 84853172..1a3064d2 100644 --- a/py/routes/update_routes.py +++ b/py/routes/update_routes.py @@ -21,7 +21,20 @@ NETWORK_EXCEPTIONS = (ClientError, OSError, asyncio.TimeoutError) # otherwise delete them because they are untracked and, in released tags, # not listed in ``.gitignore``. ``-e`` excludes a path from cleaning # regardless of whether it is ignored. -_PRESERVE_DIRS = ('settings.json', 'civitai', 'wildcards', 'backups', 'stats', 'logs', 'cache', 'model_cache') +# ``cache`` covers the resolved cache tree (cache/model, cache/recipe, +# cache/fts, ...); the legacy ``recipe_cache`` / ``model_cache`` directories +# are listed too because a portable install can predate the cache/ move. +_PRESERVE_DIRS = ( + 'settings.json', + 'civitai', + 'wildcards', + 'backups', + 'stats', + 'logs', + 'cache', + 'model_cache', + 'recipe_cache', +) def _clean_excludes() -> List[str]: diff --git a/py/services/persistent_model_cache.py b/py/services/persistent_model_cache.py index 7c8d89c0..144f5027 100644 --- a/py/services/persistent_model_cache.py +++ b/py/services/persistent_model_cache.py @@ -6,7 +6,9 @@ import threading from dataclasses import dataclass, field from typing import Any, Dict, List, Mapping, Optional, Sequence, Tuple +from ..utils.cache_db import connect_cache_db from ..utils.cache_paths import CacheType, resolve_cache_path_with_migration +from ..utils.file_lock import exclusive_lock from .model_sources import normalize_metadata_source logger = logging.getLogger(__name__) @@ -257,267 +259,271 @@ class PersistentModelCache: return try: with self._db_lock: - conn = self._connect() - try: - conn.execute("PRAGMA foreign_keys = ON") - conn.execute("BEGIN") + # Cross-process serialization: another LoRA Manager instance may + # share this settings directory, and the read-merge-write below + # spans several statements. + with exclusive_lock(self._db_path): + conn = self._connect() + try: + conn.execute("PRAGMA foreign_keys = ON") + conn.execute("BEGIN") - model_rows = [self._prepare_model_row(model_type, item) for item in raw_data] - model_map: Dict[str, Tuple[Any, ...]] = { - row[1]: row for row in model_rows if row[1] # row[1] is file_path - } + model_rows = [self._prepare_model_row(model_type, item) for item in raw_data] + model_map: Dict[str, Tuple[Any, ...]] = { + row[1]: row for row in model_rows if row[1] # row[1] is file_path + } - existing_models = conn.execute( - "SELECT " - + ", ".join(self._MODEL_COLUMNS[1:]) - + " FROM models WHERE model_type = ?", - (model_type,), - ).fetchall() - existing_model_map: Dict[str, sqlite3.Row] = { - row["file_path"]: row for row in existing_models - } - - to_remove_models = [ - (model_type, path) - for path in existing_model_map.keys() - if path not in model_map - ] - if to_remove_models: - conn.executemany( - "DELETE FROM models WHERE model_type = ? AND file_path = ?", - to_remove_models, - ) - conn.executemany( - "DELETE FROM model_tags WHERE model_type = ? AND file_path = ?", - to_remove_models, - ) - conn.executemany( - "DELETE FROM hash_index WHERE model_type = ? AND file_path = ?", - to_remove_models, - ) - conn.executemany( - "DELETE FROM autov3_index WHERE model_type = ? AND file_path = ?", - to_remove_models, - ) - conn.executemany( - "DELETE FROM excluded_models WHERE model_type = ? AND file_path = ?", - to_remove_models, - ) - - insert_rows: List[Tuple[Any, ...]] = [] - update_rows: List[Tuple[Any, ...]] = [] - - for file_path, row in model_map.items(): - existing = existing_model_map.get(file_path) - if existing is None: - insert_rows.append(row) - continue - - existing_values = tuple( - existing[column] for column in self._MODEL_COLUMNS[1:] - ) - current_values = row[1:] - if existing_values != current_values: - update_rows.append(row[2:] + (model_type, file_path)) - - if insert_rows: - conn.executemany(self._insert_model_sql(), insert_rows) - - if update_rows: - set_clause = ", ".join( - f"{column} = ?" - for column in self._MODEL_UPDATE_COLUMNS - ) - update_sql = ( - f"UPDATE models SET {set_clause} WHERE model_type = ? AND file_path = ?" - ) - conn.executemany(update_sql, update_rows) - - existing_tags_rows = conn.execute( - "SELECT file_path, tag FROM model_tags WHERE model_type = ?", - (model_type,), - ).fetchall() - existing_tags: Dict[str, set[str]] = {} - for row in existing_tags_rows: - existing_tags.setdefault(row["file_path"], set()).add(row["tag"]) - - new_tags: Dict[str, set[str]] = {} - for item in raw_data: - file_path = item.get("file_path") - if not file_path: - continue - tags = set(item.get("tags") or []) - if tags: - new_tags[file_path] = tags - - tag_inserts: List[Tuple[str, str, str]] = [] - tag_deletes: List[Tuple[str, str, str]] = [] - - all_tag_paths = set(existing_tags.keys()) | set(new_tags.keys()) - for path in all_tag_paths: - existing_set = existing_tags.get(path, set()) - new_set = new_tags.get(path, set()) - to_add = new_set - existing_set - to_remove = existing_set - new_set - - for tag in to_add: - tag_inserts.append((model_type, path, tag)) - for tag in to_remove: - tag_deletes.append((model_type, path, tag)) - - if tag_deletes: - conn.executemany( - "DELETE FROM model_tags WHERE model_type = ? AND file_path = ? AND tag = ?", - tag_deletes, - ) - if tag_inserts: - conn.executemany( - "INSERT INTO model_tags (model_type, file_path, tag) VALUES (?, ?, ?)", - tag_inserts, - ) - - existing_hash_rows = conn.execute( - "SELECT sha256, file_path FROM hash_index WHERE model_type = ?", - (model_type,), - ).fetchall() - existing_hash_map: Dict[str, set[str]] = {} - for row in existing_hash_rows: - sha_value = (row["sha256"] or "").lower() - if not sha_value: - continue - existing_hash_map.setdefault(sha_value, set()).add(row["file_path"]) - - new_hash_map: Dict[str, set[str]] = {} - for sha_value, paths in hash_index.items(): - normalized_sha = (sha_value or "").lower() - if not normalized_sha: - continue - bucket = new_hash_map.setdefault(normalized_sha, set()) - for path in paths: - if path: - bucket.add(path) - - hash_inserts: List[Tuple[str, str, str]] = [] - hash_deletes: List[Tuple[str, str, str]] = [] - - all_shas = set(existing_hash_map.keys()) | set(new_hash_map.keys()) - for sha_value in all_shas: - existing_paths = existing_hash_map.get(sha_value, set()) - new_paths = new_hash_map.get(sha_value, set()) - - for path in existing_paths - new_paths: - hash_deletes.append((model_type, sha_value, path)) - for path in new_paths - existing_paths: - hash_inserts.append((model_type, sha_value, path)) - - if hash_deletes: - conn.executemany( - "DELETE FROM hash_index WHERE model_type = ? AND sha256 = ? AND file_path = ?", - hash_deletes, - ) - if hash_inserts: - conn.executemany( - "INSERT OR IGNORE INTO hash_index (model_type, sha256, file_path) VALUES (?, ?, ?)", - hash_inserts, - ) - - if autov3_hash_index is not None: - existing_autov3_rows = conn.execute( - "SELECT autov3, file_path FROM autov3_index WHERE model_type = ?", + existing_models = conn.execute( + "SELECT " + + ", ".join(self._MODEL_COLUMNS[1:]) + + " FROM models WHERE model_type = ?", (model_type,), ).fetchall() - existing_autov3_map: Dict[str, set[str]] = {} - for row in existing_autov3_rows: - autov3_value = (row["autov3"] or "").lower() - if not autov3_value: - continue - existing_autov3_map.setdefault(autov3_value, set()).add(row["file_path"]) + existing_model_map: Dict[str, sqlite3.Row] = { + row["file_path"]: row for row in existing_models + } - new_autov3_map: Dict[str, set[str]] = {} - for autov3_value, paths in autov3_hash_index.items(): - normalized_autov3 = (autov3_value or "").lower() - if not normalized_autov3: + to_remove_models = [ + (model_type, path) + for path in existing_model_map.keys() + if path not in model_map + ] + if to_remove_models: + conn.executemany( + "DELETE FROM models WHERE model_type = ? AND file_path = ?", + to_remove_models, + ) + conn.executemany( + "DELETE FROM model_tags WHERE model_type = ? AND file_path = ?", + to_remove_models, + ) + conn.executemany( + "DELETE FROM hash_index WHERE model_type = ? AND file_path = ?", + to_remove_models, + ) + conn.executemany( + "DELETE FROM autov3_index WHERE model_type = ? AND file_path = ?", + to_remove_models, + ) + conn.executemany( + "DELETE FROM excluded_models WHERE model_type = ? AND file_path = ?", + to_remove_models, + ) + + insert_rows: List[Tuple[Any, ...]] = [] + update_rows: List[Tuple[Any, ...]] = [] + + for file_path, row in model_map.items(): + existing = existing_model_map.get(file_path) + if existing is None: + insert_rows.append(row) continue - bucket = new_autov3_map.setdefault(normalized_autov3, set()) + + existing_values = tuple( + existing[column] for column in self._MODEL_COLUMNS[1:] + ) + current_values = row[1:] + if existing_values != current_values: + update_rows.append(row[2:] + (model_type, file_path)) + + if insert_rows: + conn.executemany(self._insert_model_sql(), insert_rows) + + if update_rows: + set_clause = ", ".join( + f"{column} = ?" + for column in self._MODEL_UPDATE_COLUMNS + ) + update_sql = ( + f"UPDATE models SET {set_clause} WHERE model_type = ? AND file_path = ?" + ) + conn.executemany(update_sql, update_rows) + + existing_tags_rows = conn.execute( + "SELECT file_path, tag FROM model_tags WHERE model_type = ?", + (model_type,), + ).fetchall() + existing_tags: Dict[str, set[str]] = {} + for row in existing_tags_rows: + existing_tags.setdefault(row["file_path"], set()).add(row["tag"]) + + new_tags: Dict[str, set[str]] = {} + for item in raw_data: + file_path = item.get("file_path") + if not file_path: + continue + tags = set(item.get("tags") or []) + if tags: + new_tags[file_path] = tags + + tag_inserts: List[Tuple[str, str, str]] = [] + tag_deletes: List[Tuple[str, str, str]] = [] + + all_tag_paths = set(existing_tags.keys()) | set(new_tags.keys()) + for path in all_tag_paths: + existing_set = existing_tags.get(path, set()) + new_set = new_tags.get(path, set()) + to_add = new_set - existing_set + to_remove = existing_set - new_set + + for tag in to_add: + tag_inserts.append((model_type, path, tag)) + for tag in to_remove: + tag_deletes.append((model_type, path, tag)) + + if tag_deletes: + conn.executemany( + "DELETE FROM model_tags WHERE model_type = ? AND file_path = ? AND tag = ?", + tag_deletes, + ) + if tag_inserts: + conn.executemany( + "INSERT INTO model_tags (model_type, file_path, tag) VALUES (?, ?, ?)", + tag_inserts, + ) + + existing_hash_rows = conn.execute( + "SELECT sha256, file_path FROM hash_index WHERE model_type = ?", + (model_type,), + ).fetchall() + existing_hash_map: Dict[str, set[str]] = {} + for row in existing_hash_rows: + sha_value = (row["sha256"] or "").lower() + if not sha_value: + continue + existing_hash_map.setdefault(sha_value, set()).add(row["file_path"]) + + new_hash_map: Dict[str, set[str]] = {} + for sha_value, paths in hash_index.items(): + normalized_sha = (sha_value or "").lower() + if not normalized_sha: + continue + bucket = new_hash_map.setdefault(normalized_sha, set()) for path in paths: if path: bucket.add(path) - autov3_inserts: List[Tuple[str, str, str]] = [] - autov3_deletes: List[Tuple[str, str, str]] = [] + hash_inserts: List[Tuple[str, str, str]] = [] + hash_deletes: List[Tuple[str, str, str]] = [] - all_autov3 = set(existing_autov3_map.keys()) | set(new_autov3_map.keys()) - for autov3_value in all_autov3: - existing_paths = existing_autov3_map.get(autov3_value, set()) - new_paths = new_autov3_map.get(autov3_value, set()) + all_shas = set(existing_hash_map.keys()) | set(new_hash_map.keys()) + for sha_value in all_shas: + existing_paths = existing_hash_map.get(sha_value, set()) + new_paths = new_hash_map.get(sha_value, set()) for path in existing_paths - new_paths: - autov3_deletes.append((model_type, autov3_value, path)) + hash_deletes.append((model_type, sha_value, path)) for path in new_paths - existing_paths: - autov3_inserts.append((model_type, autov3_value, path)) + hash_inserts.append((model_type, sha_value, path)) - if autov3_deletes: + if hash_deletes: conn.executemany( - "DELETE FROM autov3_index WHERE model_type = ? AND autov3 = ? AND file_path = ?", - autov3_deletes, + "DELETE FROM hash_index WHERE model_type = ? AND sha256 = ? AND file_path = ?", + hash_deletes, ) - if autov3_inserts: + if hash_inserts: conn.executemany( - "INSERT OR IGNORE INTO autov3_index (model_type, autov3, file_path) VALUES (?, ?, ?)", - autov3_inserts, + "INSERT OR IGNORE INTO hash_index (model_type, sha256, file_path) VALUES (?, ?, ?)", + hash_inserts, ) - existing_excluded_rows = conn.execute( - "SELECT file_path FROM excluded_models WHERE model_type = ?", - (model_type,), - ).fetchall() - existing_excluded = {row["file_path"] for row in existing_excluded_rows} - new_excluded = {path for path in excluded_models if path} + if autov3_hash_index is not None: + existing_autov3_rows = conn.execute( + "SELECT autov3, file_path FROM autov3_index WHERE model_type = ?", + (model_type,), + ).fetchall() + existing_autov3_map: Dict[str, set[str]] = {} + for row in existing_autov3_rows: + autov3_value = (row["autov3"] or "").lower() + if not autov3_value: + continue + existing_autov3_map.setdefault(autov3_value, set()).add(row["file_path"]) - excluded_deletes = [ - (model_type, path) - for path in existing_excluded - new_excluded - ] - excluded_inserts = [ - (model_type, path) - for path in new_excluded - existing_excluded - ] + new_autov3_map: Dict[str, set[str]] = {} + for autov3_value, paths in autov3_hash_index.items(): + normalized_autov3 = (autov3_value or "").lower() + if not normalized_autov3: + continue + bucket = new_autov3_map.setdefault(normalized_autov3, set()) + for path in paths: + if path: + bucket.add(path) - if excluded_deletes: - conn.executemany( - "DELETE FROM excluded_models WHERE model_type = ? AND file_path = ?", - excluded_deletes, - ) - if excluded_inserts: - conn.executemany( - "INSERT OR IGNORE INTO excluded_models (model_type, file_path) VALUES (?, ?)", - excluded_inserts, - ) + autov3_inserts: List[Tuple[str, str, str]] = [] + autov3_deletes: List[Tuple[str, str, str]] = [] - if all_folders is not None: - conn.execute( - "DELETE FROM folders WHERE model_type = ?", + all_autov3 = set(existing_autov3_map.keys()) | set(new_autov3_map.keys()) + for autov3_value in all_autov3: + existing_paths = existing_autov3_map.get(autov3_value, set()) + new_paths = new_autov3_map.get(autov3_value, set()) + + for path in existing_paths - new_paths: + autov3_deletes.append((model_type, autov3_value, path)) + for path in new_paths - existing_paths: + autov3_inserts.append((model_type, autov3_value, path)) + + if autov3_deletes: + conn.executemany( + "DELETE FROM autov3_index WHERE model_type = ? AND autov3 = ? AND file_path = ?", + autov3_deletes, + ) + if autov3_inserts: + conn.executemany( + "INSERT OR IGNORE INTO autov3_index (model_type, autov3, file_path) VALUES (?, ?, ?)", + autov3_inserts, + ) + + existing_excluded_rows = conn.execute( + "SELECT file_path FROM excluded_models WHERE model_type = ?", (model_type,), - ) - folder_inserts = [ - (model_type, path) for path in all_folders if path - ] - if folder_inserts: - conn.executemany( - "INSERT OR IGNORE INTO folders (model_type, path) VALUES (?, ?)", - folder_inserts, - ) - # Mark the snapshot as having folder data even when the - # library has no subfolders, so an empty list is not - # mistaken for "never recorded" on load. - conn.execute( - "INSERT OR REPLACE INTO cache_meta (key, value) VALUES (?, ?)", - (f"folders_recorded:{model_type}", "1"), - ) + ).fetchall() + existing_excluded = {row["file_path"] for row in existing_excluded_rows} + new_excluded = {path for path in excluded_models if path} - conn.commit() - finally: - conn.close() + excluded_deletes = [ + (model_type, path) + for path in existing_excluded - new_excluded + ] + excluded_inserts = [ + (model_type, path) + for path in new_excluded - existing_excluded + ] + + if excluded_deletes: + conn.executemany( + "DELETE FROM excluded_models WHERE model_type = ? AND file_path = ?", + excluded_deletes, + ) + if excluded_inserts: + conn.executemany( + "INSERT OR IGNORE INTO excluded_models (model_type, file_path) VALUES (?, ?)", + excluded_inserts, + ) + + if all_folders is not None: + conn.execute( + "DELETE FROM folders WHERE model_type = ?", + (model_type,), + ) + folder_inserts = [ + (model_type, path) for path in all_folders if path + ] + if folder_inserts: + conn.executemany( + "INSERT OR IGNORE INTO folders (model_type, path) VALUES (?, ?)", + folder_inserts, + ) + # Mark the snapshot as having folder data even when the + # library has no subfolders, so an empty list is not + # mistaken for "never recorded" on load. + conn.execute( + "INSERT OR REPLACE INTO cache_meta (key, value) VALUES (?, ?)", + (f"folders_recorded:{model_type}", "1"), + ) + + conn.commit() + finally: + conn.close() except Exception as exc: logger.warning("Failed to persist cache for %s: %s", model_type, exc) @@ -650,16 +656,14 @@ class PersistentModelCache: conn.execute(f"ALTER TABLE models ADD COLUMN {column} {definition}") def _connect(self, readonly: bool = False) -> sqlite3.Connection: - uri = False - path = self._db_path - if readonly: - if not os.path.exists(path): - raise FileNotFoundError(path) - path = f"file:{path}?mode=ro" - uri = True - conn = sqlite3.connect(path, check_same_thread=False, uri=uri, detect_types=sqlite3.PARSE_DECLTYPES) - conn.row_factory = sqlite3.Row - return conn + if readonly and not os.path.exists(self._db_path): + raise FileNotFoundError(self._db_path) + return connect_cache_db( + self._db_path, + readonly=readonly, + detect_types=sqlite3.PARSE_DECLTYPES, + row_factory=sqlite3.Row, + ) def _prepare_model_row(self, model_type: str, item: Dict[str, Any]) -> Tuple[Any, ...]: # Keep `source_*` and the legacy `hf_url` alias consistent no matter diff --git a/py/services/persistent_recipe_cache.py b/py/services/persistent_recipe_cache.py index f0a03b98..8b023135 100644 --- a/py/services/persistent_recipe_cache.py +++ b/py/services/persistent_recipe_cache.py @@ -19,7 +19,9 @@ import threading from dataclasses import dataclass, field from typing import Any, Dict, List, Optional, Set, Tuple +from ..utils.cache_db import connect_cache_db from ..utils.cache_paths import CacheType, resolve_cache_path_with_migration +from ..utils.file_lock import exclusive_lock logger = logging.getLogger(__name__) @@ -197,64 +199,68 @@ class PersistentRecipeCache: try: with self._db_lock: - conn = self._connect() - try: - conn.execute("PRAGMA foreign_keys = ON") - conn.execute("BEGIN") + # Cross-process serialization: another LoRA Manager instance may + # share this settings directory, and a full-table replace is a + # read-modify-write that SQLite alone cannot make atomic. + with exclusive_lock(self._db_path): + conn = self._connect() + try: + conn.execute("PRAGMA foreign_keys = ON") + conn.execute("BEGIN") - if skip_if_empty and not recipes: - existing = conn.execute( - "SELECT COUNT(*) FROM recipes" - ).fetchone() - if existing and existing[0]: - conn.rollback() - logger.warning( - "Refusing to persist an empty recipe cache: the " - "stored cache still holds %d recipe(s). The scan " - "found nothing, which usually means the recipes " - "path was unavailable or resolved elsewhere; " - "keeping the stored cache so the data stays " - "recoverable.", - existing[0], + if skip_if_empty and not recipes: + existing = conn.execute( + "SELECT COUNT(*) FROM recipes" + ).fetchone() + if existing and existing[0]: + conn.rollback() + logger.warning( + "Refusing to persist an empty recipe cache: the " + "stored cache still holds %d recipe(s). The scan " + "found nothing, which usually means the recipes " + "path was unavailable or resolved elsewhere; " + "keeping the stored cache so the data stays " + "recoverable.", + existing[0], + ) + return False + + # Clear existing data + conn.execute("DELETE FROM recipes") + + # Prepare and insert all rows + recipe_rows = [] + for recipe in recipes: + recipe_id = str(recipe.get("id", "")) + if not recipe_id: + continue + + json_path = "" + if json_paths: + json_path = json_paths.get(recipe_id, "") + + row = self._prepare_recipe_row(recipe, json_path) + recipe_rows.append(row) + + if recipe_rows: + placeholders = ", ".join(["?"] * len(self._RECIPE_COLUMNS)) + columns = ", ".join(self._RECIPE_COLUMNS) + conn.executemany( + f"INSERT INTO recipes ({columns}) VALUES ({placeholders})", + recipe_rows, ) - return False - # Clear existing data - conn.execute("DELETE FROM recipes") - - # Prepare and insert all rows - recipe_rows = [] - for recipe in recipes: - recipe_id = str(recipe.get("id", "")) - if not recipe_id: - continue - - json_path = "" - if json_paths: - json_path = json_paths.get(recipe_id, "") - - row = self._prepare_recipe_row(recipe, json_path) - recipe_rows.append(row) - - if recipe_rows: - placeholders = ", ".join(["?"] * len(self._RECIPE_COLUMNS)) - columns = ", ".join(self._RECIPE_COLUMNS) - conn.executemany( - f"INSERT INTO recipes ({columns}) VALUES ({placeholders})", - recipe_rows, + # Persist image_id_map for O(1) lookups on cache load + conn.execute( + "INSERT OR REPLACE INTO cache_metadata (key, value) VALUES (?, ?)", + ("image_id_map", json.dumps(image_id_map or {})), ) - # Persist image_id_map for O(1) lookups on cache load - conn.execute( - "INSERT OR REPLACE INTO cache_metadata (key, value) VALUES (?, ?)", - ("image_id_map", json.dumps(image_id_map or {})), - ) - - conn.commit() - logger.debug("Persisted %d recipes to cache", len(recipe_rows)) - return True - finally: - conn.close() + conn.commit() + logger.debug("Persisted %d recipes to cache", len(recipe_rows)) + return True + finally: + conn.close() except Exception as exc: logger.warning("Failed to persist recipe cache: %s", exc) return False @@ -515,16 +521,14 @@ class PersistentRecipeCache: logger.warning("Failed to initialize persistent recipe cache schema: %s", exc) def _connect(self, readonly: bool = False) -> sqlite3.Connection: - uri = False - path = self._db_path - if readonly: - if not os.path.exists(path): - raise FileNotFoundError(path) - path = f"file:{path}?mode=ro" - uri = True - conn = sqlite3.connect(path, check_same_thread=False, uri=uri, detect_types=sqlite3.PARSE_DECLTYPES) - conn.row_factory = sqlite3.Row - return conn + if readonly and not os.path.exists(self._db_path): + raise FileNotFoundError(self._db_path) + return connect_cache_db( + self._db_path, + readonly=readonly, + detect_types=sqlite3.PARSE_DECLTYPES, + row_factory=sqlite3.Row, + ) def _prepare_recipe_row(self, recipe: Dict[str, Any], json_path: str) -> Tuple[Any, ...]: """Convert a recipe dict to a row tuple for SQLite insertion.""" diff --git a/py/services/recipe_fts_index.py b/py/services/recipe_fts_index.py index 2a74af7a..92737463 100644 --- a/py/services/recipe_fts_index.py +++ b/py/services/recipe_fts_index.py @@ -16,6 +16,7 @@ import threading import time from typing import Any, Dict, List, Optional, Set, Tuple +from ..utils.cache_db import connect_cache_db from ..utils.cache_paths import CacheType, resolve_cache_path_with_migration logger = logging.getLogger(__name__) @@ -633,16 +634,13 @@ class RecipeFTSIndex: def _connect(self, readonly: bool = False) -> sqlite3.Connection: """Create a database connection.""" - uri = False - path = self._db_path - if readonly: - if not os.path.exists(path): - raise FileNotFoundError(path) - path = f"file:{path}?mode=ro" - uri = True - conn = sqlite3.connect(path, check_same_thread=False, uri=uri) - conn.row_factory = sqlite3.Row - return conn + if readonly and not os.path.exists(self._db_path): + raise FileNotFoundError(self._db_path) + return connect_cache_db( + self._db_path, + readonly=readonly, + row_factory=sqlite3.Row, + ) def _remove_recipe_locked(self, conn: sqlite3.Connection, recipe_id: str) -> None: """Remove a recipe entry. Caller must hold the lock.""" diff --git a/py/services/recipe_scanner.py b/py/services/recipe_scanner.py index bdb3e036..13e3650e 100644 --- a/py/services/recipe_scanner.py +++ b/py/services/recipe_scanner.py @@ -2729,6 +2729,10 @@ class RecipeScanner: try: # Invalidate persistent cache so the sync path does a # full directory scan instead of reconciling stale data. + # This is the deliberate escape hatch from the + # all-missing prune guard: an explicit user rebuild is + # allowed to clear the stored cache, while an implicit + # startup scan is not. if self._persistent_cache: self._persistent_cache.save_cache([], {}) self._json_path_map = {} diff --git a/py/services/tag_fts_index.py b/py/services/tag_fts_index.py index a5f9cf1f..f9f75c3f 100644 --- a/py/services/tag_fts_index.py +++ b/py/services/tag_fts_index.py @@ -20,6 +20,7 @@ import time from pathlib import Path from typing import Any, Dict, List, Optional, Set +from ..utils.cache_db import connect_cache_db from ..utils.cache_paths import CacheType, resolve_cache_path_with_migration logger = logging.getLogger(__name__) @@ -677,16 +678,13 @@ class TagFTSIndex: def _connect(self, readonly: bool = False) -> sqlite3.Connection: """Create a database connection.""" - uri = False - path = self._db_path - if readonly: - if not os.path.exists(path): - raise FileNotFoundError(path) - path = f"file:{path}?mode=ro" - uri = True - conn = sqlite3.connect(path, check_same_thread=False, uri=uri) - conn.row_factory = sqlite3.Row - return conn + if readonly and not os.path.exists(self._db_path): + raise FileNotFoundError(self._db_path) + return connect_cache_db( + self._db_path, + readonly=readonly, + row_factory=sqlite3.Row, + ) def _build_fts_query(self, query: str) -> str: """Build an FTS5 query string with prefix matching. diff --git a/py/utils/cache_db.py b/py/utils/cache_db.py new file mode 100644 index 00000000..7afdb6a8 --- /dev/null +++ b/py/utils/cache_db.py @@ -0,0 +1,81 @@ +"""Shared SQLite connection setup for LoRA Manager cache databases. + +Cache databases live under the settings directory (``cache/model/.sqlite``, +``cache/recipe/.sqlite``, ``cache/fts/*.sqlite``). With portable mode or a +pinned ``LORA_MANAGER_SETTINGS_DIR`` off, that directory is shared by every ComfyUI +instance on the machine, so two processes can open the same cache file at once. + +SQLite serializes writers, but the default ``timeout`` is 5 seconds: a second +instance that writes while the first is mid-transaction fails with "database is +locked". These settings make concurrent access wait instead of failing, and keep +the write path in WAL so readers are never blocked by a writer. +""" + +from __future__ import annotations + +import sqlite3 +from typing import Any + +# How long a connection waits for a competing writer before raising. +CONCURRENT_TIMEOUT_SECONDS = 30.0 + +# PRAGMAs applied to every cache connection. +# +# ``busy_timeout`` mirrors the connection timeout so a busy database is retried +# inside SQLite rather than surfacing as an immediate error. ``synchronous=NORMAL`` +# is the documented companion of WAL: still crash-safe, far fewer fsyncs. +_TUNING_PRAGMAS = ( + "PRAGMA busy_timeout = 30000", + "PRAGMA synchronous = NORMAL", +) + + +def connect_cache_db( + path: str, + *, + readonly: bool = False, + uri: bool = False, + detect_types: int = 0, + row_factory: Any = None, +) -> sqlite3.Connection: + """Open a cache database with multi-instance-friendly settings. + + Args: + path: Database path, or a ``file:`` URI when *uri* is True. + readonly: Open through a read-only URI. Callers still pass the + plain path; the ``mode=ro`` suffix is added here. The + write-oriented tuning pragmas are skipped in that case so a + read-only connection never attempts to change the file. + uri: Treat *path* as a SQLite URI. + detect_types: Forwarded to :func:`sqlite3.connect`. + row_factory: Optional ``row_factory`` for the connection. + + Returns: + A configured :class:`sqlite3.Connection`. + """ + if readonly: + if not uri and not path.startswith("file:"): + path = f"file:{path}?mode=ro" + uri = True + + conn = sqlite3.connect( + path, + check_same_thread=False, + uri=uri, + detect_types=detect_types, + timeout=CONCURRENT_TIMEOUT_SECONDS, + ) + if row_factory is not None: + conn.row_factory = row_factory + + try: + for pragma in _TUNING_PRAGMAS: + # A read-only connection may reject write PRAGMAs; they are not + # needed there anyway. + conn.execute(pragma) + except sqlite3.Error: + # Tuning is best-effort: a connection that cannot set pragmas still + # works, just without the concurrency headroom. + pass + + return conn diff --git a/py/utils/file_lock.py b/py/utils/file_lock.py new file mode 100644 index 00000000..356e4b5c --- /dev/null +++ b/py/utils/file_lock.py @@ -0,0 +1,146 @@ +"""Cross-process advisory locking for shared LoRA Manager state. + +Two LoRA Manager processes (the ComfyUI plugin and a standalone server, or two +ComfyUI installs pointed at the same settings directory) can open the same cache +database. SQLite serializes individual statements, but it cannot make a +read-modify-write *sequence* atomic across processes: two full-table cache +replacements can interleave so that one process's snapshot overwrites the +other's. + +This module provides a small advisory file lock for those sequences. It is +deliberately non-fatal: if locking is unavailable or the wait times out, callers +keep working with SQLite's own ``busy_timeout`` as the fallback. +""" + +from __future__ import annotations + +import logging +import os +import time + +logger = logging.getLogger(__name__) + +# How long to wait for another process to release the lock before giving up. +DEFAULT_LOCK_TIMEOUT_SECONDS = 30.0 +_POLL_INTERVAL_SECONDS = 0.05 + +# Windows byte-range locks; fcntl.flock on POSIX. +try: # pragma: no cover - platform dependent + import fcntl +except ImportError: # pragma: no cover - Windows + fcntl = None # type: ignore[assignment] + +try: # pragma: no cover - Windows only + import msvcrt +except ImportError: # pragma: no cover - POSIX + msvcrt = None # type: ignore[assignment] + + +class FileLockUnavailable(RuntimeError): + """Raised when the lock could not be acquired within the timeout.""" + + +def lock_path_for(db_path: str) -> str: + """Return the sibling lock file path used for *db_path*.""" + absolute = os.path.abspath(db_path) + directory = os.path.dirname(absolute) + if not directory: + raise ValueError(f"Cannot derive a lock directory from {db_path!r}") + return os.path.join(directory, f".{os.path.basename(absolute)}.lock") + + +class CrossProcessLock: + """A best-effort advisory lock backed by a lock file. + + The lock file is a sibling of the guarded resource and is never deleted: + unlinking it would let a second process create a fresh inode and lock that + instead, defeating mutual exclusion. + """ + + def __init__(self, path: str, timeout: float = DEFAULT_LOCK_TIMEOUT_SECONDS): + self.path = path + self.timeout = timeout + self._handle = None + + def acquire(self) -> bool: + """Try to take the lock, waiting up to ``timeout`` seconds. + + Returns: + True when the lock is held (including when another lock is already + held by *this* process — the calls are not reentrant, so callers must + not nest them). False when locking is unsupported or timed out; the + caller should proceed and rely on the SQLite busy timeout instead. + """ + if fcntl is None and msvcrt is None: # pragma: no cover - exotic platform + return False + + os.makedirs(os.path.dirname(self.path), exist_ok=True) + try: + handle = open(self.path, "a+b") + except OSError as exc: + logger.debug("Could not open lock file %s: %s", self.path, exc) + return False + + deadline = time.monotonic() + max(0.0, self.timeout) + while True: + if self._try_lock(handle): + self._handle = handle + return True + if time.monotonic() >= deadline: + handle.close() + return False + time.sleep(_POLL_INTERVAL_SECONDS) + + def release(self) -> None: + """Release the lock if held. Safe to call more than once.""" + handle = self._handle + if handle is None: + return + self._handle = None + try: + self._unlock(handle) + except OSError as exc: # pragma: no cover - defensive + logger.debug("Failed to release lock %s: %s", self.path, exc) + finally: + try: + handle.close() + except OSError: # pragma: no cover - defensive + pass + + def __enter__(self) -> "CrossProcessLock": + self.acquire() + return self + + def __exit__(self, *_exc_info: object) -> None: + self.release() + + # -- platform primitives ------------------------------------------------- + + def _try_lock(self, handle) -> bool: + if fcntl is not None: + try: + fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + return True + except OSError: + return False + if msvcrt is not None: # pragma: no cover - Windows + try: + handle.seek(0) + msvcrt.locking(handle.fileno(), msvcrt.LK_NBLCK, 1) + return True + except OSError: + return False + return False + + def _unlock(self, handle) -> None: + if fcntl is not None: + fcntl.flock(handle.fileno(), fcntl.LOCK_UN) + return + if msvcrt is not None: # pragma: no cover - Windows + handle.seek(0) + msvcrt.locking(handle.fileno(), msvcrt.LK_UNLCK, 1) + + +def exclusive_lock(db_path: str, timeout: float = DEFAULT_LOCK_TIMEOUT_SECONDS): + """Return a :class:`CrossProcessLock` for the database at *db_path*.""" + return CrossProcessLock(lock_path_for(db_path), timeout=timeout) diff --git a/tests/utils/test_cache_db.py b/tests/utils/test_cache_db.py new file mode 100644 index 00000000..7b088ca8 --- /dev/null +++ b/tests/utils/test_cache_db.py @@ -0,0 +1,107 @@ +"""Tests for the shared cache SQLite connection settings (:mod:`py.utils.cache_db`). + +Two LoRA Manager processes can share one settings directory, so cache +connections must tolerate a competing writer instead of failing immediately +with "database is locked". +""" + +from __future__ import annotations + +import sqlite3 +import threading +import time + +from py.utils.cache_db import CONCURRENT_TIMEOUT_SECONDS, connect_cache_db + + +def test_busy_timeout_pragma_is_applied(tmp_path): + """The connection must retry inside SQLite, not just at connect() time.""" + conn = connect_cache_db(str(tmp_path / "cache.sqlite")) + try: + value = conn.execute("PRAGMA busy_timeout").fetchone()[0] + finally: + conn.close() + assert value == int(CONCURRENT_TIMEOUT_SECONDS * 1000) + + +def test_waiting_writer_succeeds_after_competing_writer_commits(tmp_path): + """A blocked writer waits for the lock instead of raising.""" + db_path = str(tmp_path / "cache.sqlite") + + holder = connect_cache_db(db_path) + holder.execute("CREATE TABLE t (v INTEGER)") + holder.commit() + holder.execute("BEGIN IMMEDIATE") + + def release_after_delay() -> None: + time.sleep(0.5) + holder.commit() + + releaser = threading.Thread(target=release_after_delay) + releaser.start() + try: + waiter = connect_cache_db(db_path) + try: + # Under the old 5s default this still worked, but an immediate + # failure is what low-timeout connections produced; assert the + # write lands rather than propagating "database is locked". + waiter.execute("INSERT INTO t VALUES (1)") + waiter.commit() + finally: + waiter.close() + finally: + releaser.join() + holder.close() + + check = connect_cache_db(db_path) + try: + assert check.execute("SELECT COUNT(*) FROM t").fetchone()[0] == 1 + finally: + check.close() + + +def test_readwrite_connection_uses_row_factory(tmp_path): + conn = connect_cache_db(str(tmp_path / "cache.sqlite"), row_factory=sqlite3.Row) + try: + conn.execute("CREATE TABLE t (v INTEGER)") + conn.execute("INSERT INTO t VALUES (7)") + conn.commit() + row = conn.execute("SELECT v FROM t").fetchone() + assert row["v"] == 7 + finally: + conn.close() + + +def test_readonly_connection_reads_without_writing(tmp_path): + db_path = str(tmp_path / "cache.sqlite") + writer = connect_cache_db(db_path) + writer.execute("CREATE TABLE t (v INTEGER)") + writer.execute("INSERT INTO t VALUES (1)") + writer.commit() + writer.close() + + conn = connect_cache_db(db_path, readonly=True) + try: + assert conn.execute("SELECT v FROM t").fetchone()[0] == 1 + finally: + conn.close() + + +def test_readonly_connection_rejects_writes(tmp_path): + db_path = str(tmp_path / "cache.sqlite") + writer = connect_cache_db(db_path) + writer.execute("CREATE TABLE t (v INTEGER)") + writer.commit() + writer.close() + + conn = connect_cache_db(db_path, readonly=True) + try: + try: + conn.execute("INSERT INTO t VALUES (1)") + conn.commit() + except sqlite3.OperationalError: + pass + else: # pragma: no cover - would mean mode=ro was not applied + raise AssertionError("read-only connection accepted a write") + finally: + conn.close() diff --git a/tests/utils/test_file_lock.py b/tests/utils/test_file_lock.py new file mode 100644 index 00000000..22ba1d44 --- /dev/null +++ b/tests/utils/test_file_lock.py @@ -0,0 +1,118 @@ +"""Tests for the cross-process advisory lock (:mod:`py.utils.file_lock`).""" + +from __future__ import annotations + +import os +import time + +import pytest + +from py.utils.file_lock import ( + CrossProcessLock, + FileLockUnavailable, + exclusive_lock, + lock_path_for, +) + + +def test_lock_path_is_a_sibling_of_the_resource(tmp_path): + db_path = str(tmp_path / "recipe" / "default.sqlite") + lock_path = lock_path_for(db_path) + + assert os.path.dirname(lock_path) == os.path.dirname(db_path) + assert os.path.basename(lock_path) == ".default.sqlite.lock" + + +def test_acquire_and_release_round_trip(tmp_path): + lock = exclusive_lock(str(tmp_path / "cache.sqlite")) + + assert lock.acquire() is True + lock.release() + # Releasing twice must be safe. + lock.release() + # ...and the lock is reusable afterwards. + assert lock.acquire() is True + lock.release() + + +def test_second_lock_holder_waits_until_release(tmp_path): + """A held lock blocks a competing holder for the same resource.""" + db_path = str(tmp_path / "cache.sqlite") + first = exclusive_lock(db_path) + second = CrossProcessLock(lock_path_for(db_path), timeout=0.2) + + assert first.acquire() is True + try: + started = time.monotonic() + assert second.acquire() is False + # It must have waited for the timeout rather than failing instantly. + assert time.monotonic() - started >= 0.15 + finally: + first.release() + + # Once released, the contender gets the lock. + assert second.acquire() is True + second.release() + + +def test_context_manager_releases_on_exception(tmp_path): + lock = exclusive_lock(str(tmp_path / "cache.sqlite")) + contender = CrossProcessLock(lock.path, timeout=0.2) + + with pytest.raises(RuntimeError): + with lock: + raise RuntimeError("boom") + + assert contender.acquire() is True + contender.release() + + +def test_lock_file_is_not_deleted(tmp_path): + """Deleting the lock file would let a second process lock a fresh inode.""" + lock = exclusive_lock(str(tmp_path / "cache.sqlite")) + assert lock.acquire() is True + lock.release() + + assert os.path.exists(lock.path) + + +def test_unsupported_platform_degrades_gracefully(tmp_path, monkeypatch): + """Without a platform primitive the lock reports failure instead of raising.""" + import py.utils.file_lock as file_lock_module + + monkeypatch.setattr(file_lock_module, "fcntl", None) + monkeypatch.setattr(file_lock_module, "msvcrt", None) + + lock = exclusive_lock(str(tmp_path / "cache.sqlite")) + assert lock.acquire() is False + # Callers use it as a context manager and continue without the lock. + with exclusive_lock(str(tmp_path / "cache.sqlite")): + pass + + +def test_file_lock_unavailable_is_exported(): + assert issubclass(FileLockUnavailable, RuntimeError) + + +def test_save_cache_creates_lock_next_to_database(tmp_path): + """The recipe cache write path actually takes the cross-process lock.""" + from py.services.persistent_recipe_cache import PersistentRecipeCache + + db_path = tmp_path / "recipe_cache.sqlite" + cache = PersistentRecipeCache(db_path=str(db_path)) + assert cache.save_cache([{"id": "r1", "title": "One"}], {"r1": "/tmp/r1.json"}) + + assert os.path.exists(lock_path_for(str(db_path))) + + +def test_save_cache_releases_lock_after_write(tmp_path): + """A second writer must not be blocked once the first has finished.""" + from py.services.persistent_recipe_cache import PersistentRecipeCache + + db_path = tmp_path / "recipe_cache.sqlite" + cache = PersistentRecipeCache(db_path=str(db_path)) + cache.save_cache([{"id": "r1", "title": "One"}], {"r1": "/tmp/r1.json"}) + + contender = CrossProcessLock(lock_path_for(str(db_path)), timeout=0.2) + assert contender.acquire() is True + contender.release()