mirror of
https://github.com/willmiao/ComfyUI-Lora-Manager.git
synced 2026-08-07 14:30:15 -03:00
feat(cache): opportunistic cache sync on metadata read with in-place update
- Add PersistentModelCache.update_single_model() for lightweight targeted SQL update (single row + incremental tag/hash deltas, no full table scan) - Add ModelScanner.sync_cache_from_metadata() with compare-first logic: skips entirely when cache is already in sync; when stale, updates the entry in-place (O(1) instead of O(n) remove+append), incrementally adjusts tag counts/hash index/version index, and resorts only when sort-relevant fields changed - Wire sync_cache_from_metadata() into BaseModelService.get_model_metadata() via fire-and-forget asyncio.create_task — disk I/O is already paid for - Include identity re-validation guard against concurrent cache replacement - Add 16 tests covering _cache_entries_differ, sync_cache_from_metadata (no-change, in-place, fallback, conditional resort), and update_single_model (insert, tag delta, hash delta)
This commit is contained in:
Will Miao
@@ -1092,6 +1092,11 @@ class BaseModelService(ABC):
|
||||
|
||||
Listing/search endpoints return lightweight cache entries; this method performs
|
||||
a lazy read of the on-disk metadata snapshot when callers need full detail.
|
||||
|
||||
As a beneficial side effect, the in-memory and persistent caches are
|
||||
opportunistically synchronised with the on-disk metadata — this keeps the
|
||||
caches fresh even when a ``.metadata.json`` file was edited outside of the
|
||||
normal save path (e.g. manually or by an external script).
|
||||
"""
|
||||
metadata, should_skip = await MetadataManager.load_metadata(
|
||||
file_path, self.metadata_class
|
||||
@@ -1109,6 +1114,19 @@ class BaseModelService(ABC):
|
||||
MetadataManager.save_metadata(file_path, metadata)
|
||||
)
|
||||
|
||||
# Opportunistically sync the in-memory + persistent caches.
|
||||
# The .metadata.json disk read is already paid for; the sync only
|
||||
# performs work when the cache is actually stale, and uses targeted,
|
||||
# in-place operations to minimise overhead even with large model sets.
|
||||
#
|
||||
# Fire-and-forget by design: the task is intentionally untracked.
|
||||
# sync_cache_from_metadata handles its own errors internally.
|
||||
asyncio.create_task(
|
||||
self.scanner.sync_cache_from_metadata(
|
||||
file_path, metadata.to_dict()
|
||||
)
|
||||
)
|
||||
|
||||
return self.filter_civitai_data(metadata.to_dict().get("civitai", {}))
|
||||
|
||||
async def get_model_description(self, file_path: str) -> Optional[str]:
|
||||
|
||||
@@ -1566,6 +1566,218 @@ class ModelScanner:
|
||||
|
||||
return cache_entry if metadata else True
|
||||
|
||||
async def sync_cache_from_metadata(
|
||||
self, file_path: str, metadata_dict: Dict[str, Any]
|
||||
) -> bool:
|
||||
"""Opportunistically sync in-memory and persistent caches from metadata.
|
||||
|
||||
Builds a prospective cache entry from *metadata_dict* (deserialized
|
||||
``.metadata.json`` content) and compares it against the current cache
|
||||
entry. When the two are already identical this method returns
|
||||
``False`` without touching anything — avoiding the overhead of
|
||||
``update_single_model_cache``, which always removes and re-inserts
|
||||
the entry, triggers a full resort, and persists via the heavyweight
|
||||
``save_cache()``.
|
||||
|
||||
When differences are detected the update is applied **in-place** with
|
||||
targeted operations:
|
||||
|
||||
* The existing ``raw_data`` entry is modified rather than removed and
|
||||
re-appended (O(1) instead of O(n)).
|
||||
* Tag counts and the hash index are updated incrementally.
|
||||
* The version index is rebuilt only for the affected entry.
|
||||
* ``resort()`` is called **only** when a sort-relevant field changed
|
||||
(``model_name`` / ``file_name`` for name-sort, ``modified`` for
|
||||
date-sort, ``size`` for size-sort).
|
||||
* The persistent (SQLite) cache receives a targeted single-row update
|
||||
via :meth:`PersistentModelCache.update_single_model` rather than a
|
||||
full-table ``save_cache()``.
|
||||
|
||||
Returns:
|
||||
``True`` if any cache update was performed, ``False`` if the
|
||||
caches were already in sync.
|
||||
|
||||
.. note::
|
||||
|
||||
This is a **best-effort** operation. Failures are logged but
|
||||
never propagated — callers should fire-and-forget via
|
||||
:func:`asyncio.create_task`.
|
||||
"""
|
||||
try:
|
||||
return await self._sync_cache_from_metadata_impl(
|
||||
file_path, metadata_dict
|
||||
)
|
||||
except Exception:
|
||||
logger.warning(
|
||||
"sync_cache_from_metadata failed for %s",
|
||||
file_path,
|
||||
exc_info=True,
|
||||
)
|
||||
return False
|
||||
|
||||
async def _sync_cache_from_metadata_impl(
|
||||
self, file_path: str, metadata_dict: Dict[str, Any]
|
||||
) -> bool:
|
||||
cache = await self.get_cached_data()
|
||||
|
||||
# Locate the existing cache entry -----------------------------------
|
||||
existing_idx: Optional[int] = None
|
||||
existing_entry: Optional[Dict[str, Any]] = None
|
||||
for i, item in enumerate(cache.raw_data):
|
||||
if item.get("file_path") == file_path:
|
||||
existing_entry = item
|
||||
existing_idx = i
|
||||
break
|
||||
|
||||
# Build the desired entry from metadata ------------------------------
|
||||
folder_value = (
|
||||
existing_entry.get("folder", "")
|
||||
if existing_entry
|
||||
else self._calculate_folder(file_path)
|
||||
)
|
||||
desired_entry = self._build_cache_entry(
|
||||
metadata_dict,
|
||||
folder=folder_value,
|
||||
file_path_override=file_path,
|
||||
)
|
||||
|
||||
# Ensure sha256 is populated (defensive — metadata should have it)
|
||||
if (
|
||||
not desired_entry.get("sha256")
|
||||
and file_path
|
||||
and os.path.exists(file_path)
|
||||
):
|
||||
try:
|
||||
sha256 = await calculate_sha256(file_path)
|
||||
if sha256:
|
||||
desired_entry["sha256"] = sha256.lower()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# Not in cache at all — delegate to the full update path ------------
|
||||
if existing_entry is None:
|
||||
result = await self.update_single_model_cache(
|
||||
file_path, file_path, metadata_dict
|
||||
)
|
||||
return bool(result)
|
||||
|
||||
# Compare — skip everything if already in sync -----------------------
|
||||
if not self._cache_entries_differ(existing_entry, desired_entry):
|
||||
return False
|
||||
|
||||
# Re-validate: the cache may have been replaced concurrently
|
||||
# (e.g. by _apply_scan_result). Use identity check, not equality,
|
||||
# so we detect when the raw_data list was swapped out from under us.
|
||||
if self._cache is None or not any(
|
||||
item is existing_entry for item in self._cache.raw_data
|
||||
):
|
||||
return False
|
||||
|
||||
# ---- Differences detected: apply targeted, in-place updates --------
|
||||
|
||||
# Snapshot old values for delta computations
|
||||
old_tags = list(existing_entry.get("tags") or [])
|
||||
old_sha256: str = existing_entry.get("sha256", "") or ""
|
||||
old_model_name: str = existing_entry.get("model_name", "") or ""
|
||||
old_file_name: str = existing_entry.get("file_name", "") or ""
|
||||
old_modified: float = float(existing_entry.get("modified", 0.0) or 0.0)
|
||||
old_size: int = int(existing_entry.get("size", 0) or 0)
|
||||
old_civitai = existing_entry.get("civitai")
|
||||
|
||||
# ---- In-place update of the cache entry ----
|
||||
existing_entry.clear()
|
||||
existing_entry.update(desired_entry)
|
||||
|
||||
# ---- Incremental tag count update ----
|
||||
new_tags: set = set(desired_entry.get("tags") or [])
|
||||
old_tag_set: set = set(old_tags)
|
||||
for tag in old_tag_set - new_tags:
|
||||
current = self._tags_count.get(tag, 0)
|
||||
if current <= 1:
|
||||
self._tags_count.pop(tag, None)
|
||||
else:
|
||||
self._tags_count[tag] = current - 1
|
||||
for tag in new_tags - old_tag_set:
|
||||
self._tags_count[tag] = self._tags_count.get(tag, 0) + 1
|
||||
|
||||
# ---- Incremental hash index update ----
|
||||
new_sha = (desired_entry.get("sha256", "") or "").lower()
|
||||
old_sha = (old_sha256 or "").lower()
|
||||
if new_sha != old_sha:
|
||||
if old_sha:
|
||||
self._hash_index.remove_by_path(file_path)
|
||||
if new_sha:
|
||||
self._hash_index.add_entry(new_sha, file_path)
|
||||
|
||||
# ---- Incremental version index update ----
|
||||
new_civitai = desired_entry.get("civitai")
|
||||
if old_civitai != new_civitai:
|
||||
temp_old = {
|
||||
"file_path": file_path,
|
||||
"file_name": old_file_name,
|
||||
"civitai": old_civitai,
|
||||
}
|
||||
cache.remove_from_version_index(temp_old)
|
||||
cache.add_to_version_index(existing_entry)
|
||||
|
||||
# ---- Conditional resort (only when sort-key fields changed) ----
|
||||
need_resort = False
|
||||
_last = cache._last_sort
|
||||
sort_key: Optional[str] = _last[0] if _last != (None, None) else None
|
||||
if sort_key == "name":
|
||||
if (
|
||||
old_model_name != desired_entry.get("model_name", "")
|
||||
or old_file_name != desired_entry.get("file_name", "")
|
||||
):
|
||||
need_resort = True
|
||||
elif sort_key == "date":
|
||||
if old_modified != float(desired_entry.get("modified", 0.0) or 0.0):
|
||||
need_resort = True
|
||||
elif sort_key == "size":
|
||||
if old_size != int(desired_entry.get("size", 0) or 0):
|
||||
need_resort = True
|
||||
|
||||
if need_resort:
|
||||
await cache.resort()
|
||||
|
||||
# ---- Targeted SQL update (single row, not full save_cache) ----
|
||||
persistent = getattr(self, "_persistent_cache", None)
|
||||
if persistent is not None:
|
||||
old_item_for_sql: Dict[str, Any] = {
|
||||
"file_path": file_path,
|
||||
"tags": old_tags,
|
||||
"sha256": old_sha256,
|
||||
}
|
||||
await asyncio.get_event_loop().run_in_executor(
|
||||
None,
|
||||
persistent.update_single_model,
|
||||
self.model_type,
|
||||
desired_entry,
|
||||
old_item_for_sql,
|
||||
)
|
||||
|
||||
return True
|
||||
|
||||
@staticmethod
|
||||
def _cache_entries_differ(a: Dict[str, Any], b: Dict[str, Any]) -> bool:
|
||||
"""Return ``True`` when two cache-entry dicts differ in any field.
|
||||
|
||||
Tag lists are compared order-insensitively; all other keys use
|
||||
standard equality.
|
||||
"""
|
||||
a_tags = sorted(a.get("tags") or [])
|
||||
b_tags = sorted(b.get("tags") or [])
|
||||
if a_tags != b_tags:
|
||||
return True
|
||||
|
||||
all_keys = set(a.keys()) | set(b.keys())
|
||||
for key in all_keys:
|
||||
if key == "tags":
|
||||
continue
|
||||
if a.get(key) != b.get(key):
|
||||
return True
|
||||
return False
|
||||
|
||||
def has_hash(self, sha256: str) -> bool:
|
||||
"""Check if a model with given hash exists"""
|
||||
return self._hash_index.has_hash(sha256.lower())
|
||||
|
||||
@@ -587,6 +587,95 @@ class PersistentModelCache:
|
||||
placeholders = ", ".join(["?"] * len(self._MODEL_COLUMNS))
|
||||
return f"INSERT INTO models ({columns}) VALUES ({placeholders})"
|
||||
|
||||
def update_single_model(
|
||||
self,
|
||||
model_type: str,
|
||||
new_item: Dict,
|
||||
old_item: Optional[Dict] = None,
|
||||
) -> None:
|
||||
"""Update a single model row in the persistent cache.
|
||||
|
||||
A lightweight alternative to :meth:`save_cache` that performs a targeted
|
||||
DELETE + INSERT for the model row and computes incremental tag / hash-index
|
||||
deltas from *old_item*. When *old_item* is omitted the previous tags and
|
||||
hash are not cleaned up (callers should only omit it for brand-new entries).
|
||||
|
||||
All operations run inside a single transaction so readers see a consistent
|
||||
view.
|
||||
"""
|
||||
if not self.is_enabled():
|
||||
return
|
||||
if not self._schema_initialized:
|
||||
self._initialize_schema()
|
||||
if not self._schema_initialized:
|
||||
return
|
||||
|
||||
file_path: Optional[str] = new_item.get("file_path")
|
||||
if not file_path:
|
||||
return
|
||||
|
||||
try:
|
||||
with self._db_lock:
|
||||
conn = self._connect()
|
||||
try:
|
||||
conn.execute("PRAGMA foreign_keys = ON")
|
||||
conn.execute("BEGIN")
|
||||
|
||||
# --- model row (DELETE + INSERT = upsert) ---
|
||||
conn.execute(
|
||||
"DELETE FROM models WHERE model_type = ? AND file_path = ?",
|
||||
(model_type, file_path),
|
||||
)
|
||||
row = self._prepare_model_row(model_type, new_item)
|
||||
conn.execute(self._insert_model_sql(), row)
|
||||
|
||||
# --- tags ---
|
||||
new_tags: set = set(new_item.get("tags") or [])
|
||||
old_tags: set = set(old_item.get("tags") or []) if old_item else set()
|
||||
tags_to_delete = old_tags - new_tags
|
||||
tags_to_insert = new_tags - old_tags
|
||||
|
||||
if tags_to_delete:
|
||||
conn.executemany(
|
||||
"DELETE FROM model_tags WHERE model_type = ? AND file_path = ? AND tag = ?",
|
||||
[(model_type, file_path, t) for t in tags_to_delete],
|
||||
)
|
||||
if tags_to_insert:
|
||||
conn.executemany(
|
||||
"INSERT INTO model_tags (model_type, file_path, tag) VALUES (?, ?, ?)",
|
||||
[(model_type, file_path, t) for t in tags_to_insert],
|
||||
)
|
||||
|
||||
# --- hash_index ---
|
||||
new_sha: Optional[str] = (new_item.get("sha256") or "").lower() or None
|
||||
old_sha: Optional[str] = (
|
||||
(old_item.get("sha256") or "").lower() or None
|
||||
) if old_item else None
|
||||
if new_sha != old_sha:
|
||||
if old_sha:
|
||||
conn.execute(
|
||||
"DELETE FROM hash_index WHERE model_type = ? AND sha256 = ? AND file_path = ?",
|
||||
(model_type, old_sha, file_path),
|
||||
)
|
||||
if new_sha:
|
||||
conn.execute(
|
||||
"INSERT OR IGNORE INTO hash_index (model_type, sha256, file_path) VALUES (?, ?, ?)",
|
||||
(model_type, new_sha, file_path),
|
||||
)
|
||||
|
||||
conn.execute("COMMIT")
|
||||
except Exception:
|
||||
conn.execute("ROLLBACK")
|
||||
raise
|
||||
finally:
|
||||
conn.close()
|
||||
except Exception as exc:
|
||||
logger.warning(
|
||||
"Failed to update single model in persistent cache (%s): %s",
|
||||
file_path,
|
||||
exc,
|
||||
)
|
||||
|
||||
def _load_tags(self, conn: sqlite3.Connection, model_type: str) -> Dict[str, List[str]]:
|
||||
tag_rows = conn.execute(
|
||||
"SELECT file_path, tag FROM model_tags WHERE model_type = ?",
|
||||
|
||||
Reference in New Issue
Block a user