Compare commits

..

5 Commits

Author SHA1 Message Date
Will Miao 74f889f160 fix(recipes): validate FTS index from stored metadata instead of scanning 2026-08-25 17:38:31 +08:00
Will Miao c51090ab16 fix(recipes): make source_path backfill a one-shot migration 2026-08-25 17:38:17 +08:00
Will Miao cdb044cb45 fix(scanner): offload persisted-cache hydration from the event loop 2026-08-25 17:38:09 +08:00
Will Miao c83b26b556 fix(delete): run startup reconciliation walk off the event loop 2026-08-25 17:38:01 +08:00
Will Miao a202c666bc fix(download): return 200 for missing queue items and quiet download-progress 404s
The browser extension's apiFetch treats any 404 as a missing endpoint and
retries the legacy non-/api/lm URL, producing two spurious
'error_middleware - WARNING - API GET ... 404' log lines per occurrence.

- complete_download_in_queue / update_download_queue_status /
  retry_download_from_history: 'not found' is a normal business outcome,
  return 200 + success:false instead of 404 (extension behavior unchanged;
  apiGet ignores the HTTP status)
- error_middleware: downgrade /api/lm/download-progress/ 404s to debug like
  previews - the 404 status itself stays (extension uses it for failure
  detection), only the log level is lowered
2026-08-25 09:59:33 +08:00
15 changed files with 906 additions and 93 deletions
+10
View File
@@ -46,6 +46,16 @@ async def api_json_error(
if request.path.startswith("/api/lm/previews") and exc.status == 404:
logger_method = logger.debug
# Download-progress 404 is routine too: in-memory tracking is removed
# once a download finishes/fails, so the extension's final polls 404.
# The extension relies on the 404 status itself (failure detection),
# so only the log level is lowered.
if (
request.path.startswith("/api/lm/download-progress/")
and exc.status == 404
):
logger_method = logger.debug
logger_method(
"API %s %s returned HTTP %d: %s",
request.method,
+12 -5
View File
@@ -1998,9 +1998,11 @@ class ModelDownloadHandler:
item_id=item_id, download_id=download_id
)
if item is None:
# Missing or non-retryable history entry is a business
# outcome, not a routing error: 200 lets the extension's
# apiFetch 404-fallback and error middleware stay quiet.
return web.json_response(
{"success": False, "error": "History item not found or not retryable"},
status=404,
{"success": False, "error": "History item not found or not retryable"}
)
return web.json_response({"success": True, "item": item})
except Exception as exc:
@@ -2051,8 +2053,12 @@ class ModelDownloadHandler:
completed_at=completed_at,
)
if item is None:
# A missing queue item (already completed, or never queued) is
# a normal business outcome, not a routing error. Return 200
# so the browser extension's apiFetch 404-fallback and the
# error middleware stay quiet.
return web.json_response(
{"success": False, "error": "Download not found in queue"}, status=404
{"success": False, "error": "Download not found in queue"}
)
return web.json_response({"success": True, "item": item})
except Exception as exc:
@@ -2094,9 +2100,10 @@ class ModelDownloadHandler:
service = await DownloadQueueService.get_instance()
updated = await service.update_status(download_id, status)
if not updated:
# Same rationale as complete_download_in_queue: a missing
# queue item is a business outcome, not a routing error.
return web.json_response(
{"success": False, "error": "Download not found in queue"},
status=404,
{"success": False, "error": "Download not found in queue"}
)
return web.json_response({"success": True})
except Exception as exc:
@@ -236,24 +236,33 @@ class DownloadedVersionHistoryService:
return
async with self._lock:
conn = self._get_conn()
conn.executemany(
"""
INSERT INTO downloaded_model_versions (
model_type, version_id, model_id, first_seen_at, last_seen_at,
source, last_file_path, last_library_name, is_deleted_override
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, 0)
ON CONFLICT(model_type, version_id) DO UPDATE SET
model_id = COALESCE(excluded.model_id, downloaded_model_versions.model_id),
last_seen_at = excluded.last_seen_at,
source = excluded.source,
last_file_path = COALESCE(excluded.last_file_path, downloaded_model_versions.last_file_path),
last_library_name = COALESCE(excluded.last_library_name, downloaded_model_versions.last_library_name),
is_deleted_override = 0
""",
payload,
)
conn.commit()
# The connection is created with check_same_thread=False and all
# access is serialized by self._lock, so the executemany upsert +
# commit can run in the default executor without blocking the
# event loop on large hydration payloads.
loop = asyncio.get_running_loop()
await loop.run_in_executor(None, self._mark_downloaded_bulk_sync, payload)
def _mark_downloaded_bulk_sync(self, payload: Sequence[tuple[object, ...]]) -> None:
"""Synchronous executemany upsert + commit; runs in a worker thread."""
conn = self._get_conn()
conn.executemany(
"""
INSERT INTO downloaded_model_versions (
model_type, version_id, model_id, first_seen_at, last_seen_at,
source, last_file_path, last_library_name, is_deleted_override
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, 0)
ON CONFLICT(model_type, version_id) DO UPDATE SET
model_id = COALESCE(excluded.model_id, downloaded_model_versions.model_id),
last_seen_at = excluded.last_seen_at,
source = excluded.source,
last_file_path = COALESCE(excluded.last_file_path, downloaded_model_versions.last_file_path),
last_library_name = COALESCE(excluded.last_library_name, downloaded_model_versions.last_library_name),
is_deleted_override = 0
""",
payload,
)
conn.commit()
async def mark_as_deleted(self, model_type: str, version_id: int) -> None:
normalized_type = _normalize_model_type(model_type)
+75 -55
View File
@@ -535,16 +535,21 @@ class ModelScanner:
self._is_initializing = False
async def _load_persisted_cache(self, page_type: str) -> bool:
"""Attempt to hydrate the in-memory cache from the SQLite snapshot."""
"""Attempt to hydrate the in-memory cache from the SQLite snapshot.
The SQLite read and the per-model rebuild (entry adjustment, tag
counting, validation/repair, hash index reconstruction) run in the
default executor so the event loop stays responsive; only applying
the result to shared cache state happens on the loop.
"""
if not getattr(self, '_persistent_cache', None):
return False
loop = asyncio.get_event_loop()
try:
persisted = await loop.run_in_executor(
rebuilt = await loop.run_in_executor(
None,
self._persistent_cache.load_cache,
self.model_type
self._rebuild_persisted_cache
)
except FileNotFoundError:
return False
@@ -552,47 +557,14 @@ class ModelScanner:
logger.debug("%s Scanner: Could not load persisted cache: %s", self.model_type.capitalize(), exc)
return False
if not persisted or not persisted.raw_data:
if rebuilt is None:
return False
hash_index = ModelHashIndex()
for sha_value, path in persisted.hash_rows:
if sha_value and path:
hash_index.add_entry(sha_value.lower(), path)
# Rebuild the AutoV3 index from the persisted autov3_index rows. These
# cover every known autov3 -> path mapping regardless of whether a
# sha256 row also exists for the same file.
for autov3_value, path in persisted.autov3_hash_rows:
if autov3_value and path:
hash_index.add_autov3(autov3_value.lower(), path)
tags_count: Dict[str, int] = {}
adjusted_raw_data: List[Dict[str, Any]] = []
for item in persisted.raw_data:
adjusted_item = self.adjust_cached_entry(dict(item))
adjusted_raw_data.append(adjusted_item)
for tag in adjusted_item.get('tags') or []:
tags_count[tag] = tags_count.get(tag, 0) + 1
# Validate cache entries and check health.
# Always use the validated/repaired entries — even when there are no
# invalid entries, auto_repair may have filled in missing optional
# fields (model_name, file_name, folder) with safe defaults on a copied
# working_entry. Without this unconditional replacement the repaired
# copies are discarded and None values propagate to format_response.
# See issue #730.
valid_entries, invalid_entries = CacheEntryValidator.validate_batch(
adjusted_raw_data, auto_repair=True
)
# Always use the validated entries (repaired copies)
adjusted_raw_data = valid_entries
scan_result, invalid_entries = rebuilt
if invalid_entries:
monitor = CacheHealthMonitor()
report = monitor.check_health(adjusted_raw_data, auto_repair=True)
report = monitor.check_health(scan_result.raw_data, auto_repair=True)
if report.status != CacheHealthStatus.HEALTHY:
# Broadcast health warning to frontend
@@ -602,31 +574,22 @@ class ModelScanner:
f"{report.invalid_entries} invalid entries, {report.repaired_entries} repaired"
)
# Use only valid entries
adjusted_raw_data = valid_entries
# Rebuild tags count from valid entries only
tags_count = {}
for item in adjusted_raw_data:
for item in scan_result.raw_data:
for tag in item.get('tags') or []:
tags_count[tag] = tags_count.get(tag, 0) + 1
scan_result.tags_count = tags_count
# Remove invalid entries from hash index
for invalid_entry in invalid_entries:
file_path = CacheEntryValidator.get_file_path_safe(invalid_entry)
sha256 = CacheEntryValidator.get_sha256_safe(invalid_entry)
if file_path:
hash_index.remove_by_path(file_path, sha256)
scan_result = CacheBuildResult(
raw_data=adjusted_raw_data,
hash_index=hash_index,
tags_count=tags_count,
excluded_models=list(persisted.excluded_models)
)
scan_result.hash_index.remove_by_path(file_path, sha256)
await self._apply_scan_result(scan_result)
await self._sync_download_history(adjusted_raw_data, source='scan')
await self._sync_download_history(scan_result.raw_data, source='scan')
await ws_manager.broadcast_init_progress({
'stage': 'loading_cache',
@@ -651,6 +614,63 @@ class ModelScanner:
return True
def _rebuild_persisted_cache(self) -> Optional[Tuple[CacheBuildResult, List[Dict[str, Any]]]]:
"""Load the SQLite snapshot and rebuild a ready-to-apply scan result.
Runs entirely in a worker thread: it must not touch ``self._cache``,
the websocket manager, or any asyncio primitives. Returns ``None``
when no usable snapshot exists, otherwise a tuple of the scan result
(built from validated/repaired entries) and the invalid entries.
"""
persisted = self._persistent_cache.load_cache(self.model_type)
if not persisted or not persisted.raw_data:
return None
hash_index = ModelHashIndex()
for sha_value, path in persisted.hash_rows:
if sha_value and path:
hash_index.add_entry(sha_value.lower(), path)
# Rebuild the AutoV3 index from the persisted autov3_index rows. These
# cover every known autov3 -> path mapping regardless of whether a
# sha256 row also exists for the same file.
for autov3_value, path in persisted.autov3_hash_rows:
if autov3_value and path:
hash_index.add_autov3(autov3_value.lower(), path)
tags_count: Dict[str, int] = {}
adjusted_raw_data: List[Dict[str, Any]] = []
for item in persisted.raw_data:
# load_cache builds a fresh dict per row, and validate_batch below
# works on its own per-entry copy when auto_repair=True, so no
# additional dict copy is needed here.
adjusted_item = self.adjust_cached_entry(item)
adjusted_raw_data.append(adjusted_item)
for tag in adjusted_item.get('tags') or []:
tags_count[tag] = tags_count.get(tag, 0) + 1
# Validate cache entries and check health.
# Always use the validated/repaired entries — even when there are no
# invalid entries, auto_repair may have filled in missing optional
# fields (model_name, file_name, folder) with safe defaults on a copied
# working_entry. Without this unconditional replacement the repaired
# copies are discarded and None values propagate to format_response.
# See issue #730.
valid_entries, invalid_entries = CacheEntryValidator.validate_batch(
adjusted_raw_data, auto_repair=True
)
# Always use the validated entries (repaired copies)
scan_result = CacheBuildResult(
raw_data=valid_entries,
hash_index=hash_index,
tags_count=tags_count,
excluded_models=list(persisted.excluded_models)
)
return scan_result, invalid_entries
async def _run_autov3_backfill(self) -> None:
"""Backfill autov3 for entries loaded from the persisted cache that lack it."""
try:
@@ -1392,8 +1412,8 @@ class ModelScanner:
else:
self._cache.raw_data = list(scan_result.raw_data)
self._cache.rebuild_version_index()
# resort() rebuilds folders and the version index on every path, so a
# separate rebuild_version_index() call here would be redundant.
await self._cache.resort()
self._log_duplicate_filename_summary()
+31 -6
View File
@@ -500,10 +500,36 @@ class PendingDeleteService:
QUARANTINE them (preserving the pre-registry sweep semantics). The
walk only descends into dirs literally named ``.lm-pending-delete``,
so false positives are structurally limited.
The filesystem walk itself runs in a worker thread so a large or slow
library cannot block the event loop at startup; only the (rare) batch
registration awaits run on the loop.
"""
roots = await self._get_all_model_roots()
loop = asyncio.get_event_loop()
staging_parents = await loop.run_in_executor(
None, # Use default thread pool
self._collect_staging_parents, # Run the tree walk off the loop
roots,
)
for staging_parent in staging_parents:
await self._register_batch_candidates(staging_parent)
def _collect_staging_parents(self, roots: Sequence[str]) -> List[str]:
"""Walk every model root and return its staging-parent dirs.
Pure synchronous filesystem discovery with no awaits: walks with
``followlinks=True, topdown=True``, prunes symlink cycles via a
per-root ``visited`` realpath set (realpath is used ONLY for this
dedup set - the returned paths are the unresolved business paths),
filters out :func:`_is_excluded_dir` dirs, and collects every dir
named ``.lm-pending-delete`` (including the case where a model root
itself is one). Results are returned in walk order.
"""
from .model_scanner import _is_excluded_dir
for root in await self._get_all_model_roots():
staging_parents: List[str] = []
for root in roots:
if not os.path.isdir(root):
continue
visited: Set[str] = set()
@@ -518,21 +544,20 @@ class PendingDeleteService:
visited.add(real_dir)
if os.path.basename(dirpath) == PENDING_DELETE_DIR_NAME:
# The current dir IS a staging parent (reachable only when
# a model root itself is one): register its batches.
await self._register_batch_candidates(dirpath)
# a model root itself is one): collect its batches.
staging_parents.append(dirpath)
dirnames[:] = []
continue
next_dirs: List[str] = []
for name in dirnames:
if name == PENDING_DELETE_DIR_NAME:
await self._register_batch_candidates(
os.path.join(dirpath, name)
)
staging_parents.append(os.path.join(dirpath, name))
elif _is_excluded_dir(name):
continue
else:
next_dirs.append(name)
dirnames[:] = next_dirs
return staging_parents
async def _register_batch_candidates(self, staging_parent: str) -> None:
"""Register every non-orphaned batch subdir of a staging parent."""
+38
View File
@@ -333,6 +333,44 @@ class PersistentRecipeCache:
except Exception as exc:
logger.debug("Failed to persist image_id_map: %s", exc)
def get_metadata_value(self, key: str) -> Optional[str]:
"""Return a value from cache_metadata, or None if missing."""
if not self.is_enabled() or not self._schema_initialized:
return None
try:
with self._db_lock:
conn = self._connect(readonly=True)
try:
row = conn.execute(
"SELECT value FROM cache_metadata WHERE key = ?",
(key,),
).fetchone()
return row["value"] if row else None
finally:
conn.close()
except Exception:
return None
def set_metadata_value(self, key: str, value: str) -> None:
"""Store a value in cache_metadata without rewriting the full cache."""
if not self.is_enabled() or not self._schema_initialized:
return
try:
with self._db_lock:
conn = self._connect()
try:
conn.execute(
"INSERT OR REPLACE INTO cache_metadata (key, value) VALUES (?, ?)",
(key, value),
)
conn.commit()
finally:
conn.close()
except Exception as exc:
logger.debug("Failed to persist cache metadata %s: %s", key, exc)
def get_indexed_recipe_ids(self) -> Set[str]:
"""Return all recipe IDs in the cache.
+167 -7
View File
@@ -7,13 +7,14 @@ enabling sub-100ms search times even with 20k+ recipes.
from __future__ import annotations
import asyncio
import hashlib
import logging
import os
import re
import sqlite3
import threading
import time
from typing import Any, Dict, List, Optional, Set
from typing import Any, Dict, List, Optional, Set, Tuple
from ..utils.cache_paths import CacheType, resolve_cache_path_with_migration
@@ -165,6 +166,7 @@ class RecipeFTSIndex:
batch_size = 500
total = len(recipes)
inserted = 0
indexed_ids: Set[str] = set()
for i in range(0, total, batch_size):
batch = recipes[i:i + batch_size]
@@ -179,6 +181,7 @@ class RecipeFTSIndex:
row = self._prepare_fts_row(recipe)
rows.append(row)
inserted += 1
indexed_ids.add(recipe_id)
if rows:
# Insert into FTS table
@@ -213,7 +216,11 @@ class RecipeFTSIndex:
)
conn.execute(
"INSERT OR REPLACE INTO fts_metadata (key, value) VALUES (?, ?)",
('recipe_count', str(inserted))
(self._COUNT_METADATA_KEY, str(inserted))
)
conn.execute(
"INSERT OR REPLACE INTO fts_metadata (key, value) VALUES (?, ?)",
(self._FINGERPRINT_METADATA_KEY, self._compute_ids_fingerprint(indexed_ids))
)
conn.commit()
@@ -288,6 +295,12 @@ class RecipeFTSIndex:
with self._lock:
conn = self._connect()
try:
# Check existence via the rowid mapping (fast PK lookup)
existed = conn.execute(
"SELECT 1 FROM recipe_rowid WHERE recipe_id = ?",
(recipe_id,)
).fetchone() is not None
# Remove existing entry if present
self._remove_recipe_locked(conn, recipe_id)
@@ -312,6 +325,10 @@ class RecipeFTSIndex:
(recipe_id, result[0])
)
# Keep validation metadata in sync (only a new id changes it)
if not existed:
self._update_mutation_metadata_locked(conn, recipe_id, delta=1)
conn.commit()
return True
finally:
@@ -339,7 +356,13 @@ class RecipeFTSIndex:
with self._lock:
conn = self._connect()
try:
existed = conn.execute(
"SELECT 1 FROM recipe_rowid WHERE recipe_id = ?",
(recipe_id,)
).fetchone() is not None
self._remove_recipe_locked(conn, recipe_id)
if existed:
self._update_mutation_metadata_locked(conn, recipe_id, delta=-1)
conn.commit()
return True
finally:
@@ -371,6 +394,15 @@ class RecipeFTSIndex:
try:
conn.execute("DELETE FROM recipe_fts")
conn.execute("DELETE FROM recipe_rowid")
# Reset validation metadata to the empty index state
conn.execute(
"INSERT OR REPLACE INTO fts_metadata (key, value) VALUES (?, ?)",
(self._COUNT_METADATA_KEY, '0')
)
conn.execute(
"INSERT OR REPLACE INTO fts_metadata (key, value) VALUES (?, ?)",
(self._FINGERPRINT_METADATA_KEY, self._compute_ids_fingerprint(set()))
)
conn.commit()
self._ready.clear()
return True
@@ -427,10 +459,12 @@ class RecipeFTSIndex:
"""Check if the FTS index matches the expected recipes.
This method validates whether the existing FTS index can be reused
without a full rebuild. It checks:
1. The index has been initialized
2. The count matches
3. The recipe IDs match
without a full rebuild. It compares the expected count and recipe ID
fingerprint against metadata recorded when the index was (re)built,
so it does not scan the FTS content table. Indexes built by older
versions lack this metadata; for those the validation falls back to
a one-time scan of the content table and records the metadata so
subsequent startups are cheap.
Args:
recipe_count: Expected number of recipes.
@@ -446,7 +480,28 @@ class RecipeFTSIndex:
return False
try:
metadata = self._read_validation_metadata()
if metadata is not None:
stored_count, stored_fingerprint = metadata
if stored_count != recipe_count:
logger.debug(
"FTS index count mismatch: indexed=%d, expected=%d",
stored_count, recipe_count
)
return False
if stored_fingerprint != self._compute_ids_fingerprint(recipe_ids):
logger.debug("FTS index recipe ID fingerprint mismatch")
return False
return True
# Legacy fallback: no stored metadata, scan the content table once
# and persist the metadata so later validations are cheap.
indexed_count = self.get_indexed_count()
indexed_ids = self.get_indexed_recipe_ids()
self._store_validation_metadata(indexed_count, indexed_ids)
if indexed_count != recipe_count:
logger.debug(
"FTS index count mismatch: indexed=%d, expected=%d",
@@ -454,7 +509,6 @@ class RecipeFTSIndex:
)
return False
indexed_ids = self.get_indexed_recipe_ids()
if indexed_ids != recipe_ids:
missing = recipe_ids - indexed_ids
extra = indexed_ids - recipe_ids
@@ -471,6 +525,112 @@ class RecipeFTSIndex:
# Internal helpers
_FINGERPRINT_METADATA_KEY = 'recipe_ids_fingerprint'
_COUNT_METADATA_KEY = 'recipe_count'
@staticmethod
def _fingerprint_recipe_id(recipe_id: str) -> int:
"""Return a stable 64-bit fingerprint contribution for a recipe ID."""
digest = hashlib.sha256(recipe_id.encode("utf-8")).digest()
return int.from_bytes(digest[:8], "big")
@classmethod
def _compute_ids_fingerprint(cls, recipe_ids: Set[str]) -> str:
"""Order-independent fingerprint of a recipe ID set (XOR of per-id hashes)."""
fingerprint = 0
for recipe_id in recipe_ids:
fingerprint ^= cls._fingerprint_recipe_id(str(recipe_id))
return f"{fingerprint:016x}"
def _read_validation_metadata(self) -> Optional[Tuple[int, str]]:
"""Return stored (recipe count, ID fingerprint), or None if absent."""
try:
with self._lock:
conn = self._connect(readonly=True)
try:
rows = conn.execute(
"SELECT key, value FROM fts_metadata WHERE key IN (?, ?)",
(self._COUNT_METADATA_KEY, self._FINGERPRINT_METADATA_KEY)
).fetchall()
values = {row[0]: row[1] for row in rows}
fingerprint = values.get(self._FINGERPRINT_METADATA_KEY)
if fingerprint is None:
return None
try:
count = int(values.get(self._COUNT_METADATA_KEY) or 0)
except (TypeError, ValueError):
return None
return count, fingerprint
finally:
conn.close()
except FileNotFoundError:
return None
except Exception as exc:
logger.debug("Failed to read FTS validation metadata: %s", exc)
return None
def _store_validation_metadata(self, recipe_count: int, recipe_ids: Set[str]) -> None:
"""Persist recipe count and ID fingerprint for cheap future validation."""
try:
with self._lock:
conn = self._connect()
try:
conn.execute(
"INSERT OR REPLACE INTO fts_metadata (key, value) VALUES (?, ?)",
(self._COUNT_METADATA_KEY, str(recipe_count))
)
conn.execute(
"INSERT OR REPLACE INTO fts_metadata (key, value) VALUES (?, ?)",
(self._FINGERPRINT_METADATA_KEY, self._compute_ids_fingerprint(recipe_ids))
)
conn.commit()
finally:
conn.close()
except Exception as exc:
logger.debug("Failed to store FTS validation metadata: %s", exc)
def _update_mutation_metadata_locked(
self,
conn: sqlite3.Connection,
recipe_id: str,
delta: int,
) -> None:
"""Incrementally maintain validation metadata after add/remove.
Caller must hold the lock. The fingerprint is only updated when it
already exists; without it, validation falls back to a one-time scan
that records fresh metadata.
"""
fingerprint_row = conn.execute(
"SELECT value FROM fts_metadata WHERE key = ?",
(self._FINGERPRINT_METADATA_KEY,)
).fetchone()
if fingerprint_row and fingerprint_row[0]:
try:
fingerprint = int(fingerprint_row[0], 16)
except ValueError:
fingerprint = None
if fingerprint is not None:
fingerprint ^= self._fingerprint_recipe_id(recipe_id)
conn.execute(
"INSERT OR REPLACE INTO fts_metadata (key, value) VALUES (?, ?)",
(self._FINGERPRINT_METADATA_KEY, f"{fingerprint & 0xFFFFFFFFFFFFFFFF:016x}")
)
count_row = conn.execute(
"SELECT value FROM fts_metadata WHERE key = ?",
(self._COUNT_METADATA_KEY,)
).fetchone()
if count_row:
try:
count = max(0, int(count_row[0] or 0) + delta)
except (TypeError, ValueError):
return
conn.execute(
"INSERT OR REPLACE INTO fts_metadata (key, value) VALUES (?, ?)",
(self._COUNT_METADATA_KEY, str(count))
)
def _connect(self, readonly: bool = False) -> sqlite3.Connection:
"""Create a database connection."""
uri = False
+20 -2
View File
@@ -1547,7 +1547,7 @@ class RecipeScanner:
self._cache.raw_data = recipes
self._update_folder_metadata(self._cache)
self._sort_cache_sync()
# Backfill source_path from JSON files if missing (schema migration)
# Backfill source_path from JSON files if missing (one-shot schema migration)
if self._backfill_source_path_if_needed(recipes, json_paths):
self._cache.image_id_map = self._build_image_id_map()
self._persistent_cache.save_cache(
@@ -1574,7 +1574,7 @@ class RecipeScanner:
self._cache.raw_data = recipes
self._update_folder_metadata(self._cache)
self._sort_cache_sync()
# Backfill source_path from JSON files if missing (schema migration)
# Backfill source_path from JSON files if missing (one-shot schema migration)
self._backfill_source_path_if_needed(recipes, json_paths)
self._cache.image_id_map = self._build_image_id_map()
# Persist updated cache
@@ -1711,6 +1711,9 @@ class RecipeScanner:
return recipes, changed, json_paths
# Metadata key recording that the one-shot source_path backfill has run.
_SOURCE_PATH_BACKFILL_MARKER = "source_path_backfilled"
def _backfill_source_path_if_needed(
self,
recipes: List[Dict[str, Any]],
@@ -1718,8 +1721,21 @@ class RecipeScanner:
) -> bool:
"""Backfill source_path from recipe JSON files if missing from cache.
This is a one-shot schema migration: once it has run, a completion
marker is stored in the persistent cache metadata and later startups
skip it entirely. Recipes without a source_path in their JSON file
would otherwise be re-read and re-parsed on every startup. New or
changed recipe files still get source_path from the normal parse path
during reconciliation.
Returns True if any recipes were updated (caller should persist cache).
"""
cache = self._persistent_cache
if (
cache is not None
and cache.get_metadata_value(self._SOURCE_PATH_BACKFILL_MARKER) == "1"
):
return False
updated = False
for recipe in recipes:
if recipe.get("source_path"):
@@ -1737,6 +1753,8 @@ class RecipeScanner:
updated = True
except Exception:
pass
if cache is not None:
cache.set_metadata_value(self._SOURCE_PATH_BACKFILL_MARKER, "1")
return updated
def _full_directory_scan_sync(
@@ -0,0 +1,186 @@
"""Handler-level tests for download queue terminal/status transitions.
Regression test: a "not found in queue" outcome must be returned as HTTP 200
with ``success: false``, not 404. The browser extension's apiFetch treats any
404 as a missing endpoint and retries a legacy URL, producing spurious
``/api/downloads/queue/complete ... 404`` warnings on every completion.
"""
import json
import logging
from pathlib import Path
import pytest
from aiohttp import web
from aiohttp.test_utils import make_mocked_request
from py.routes.handlers.model_handlers import ModelDownloadHandler
from py.services.download_queue_service import DownloadQueueService
def _make_handler() -> ModelDownloadHandler:
return ModelDownloadHandler(
ws_manager=None, # pyright: ignore[reportArgumentType] - unused by queue endpoints
logger=logging.getLogger("test-download-queue"),
download_use_case=None, # pyright: ignore[reportArgumentType] - unused by queue endpoints
download_coordinator=None, # pyright: ignore[reportArgumentType] - unused by queue endpoints
)
def _queue_request(path: str, query: dict[str, str]) -> web.Request:
query_string = "&".join(f"{key}={value}" for key, value in query.items())
return make_mocked_request("GET", f"{path}?{query_string}")
@pytest.fixture
def queue_service(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> DownloadQueueService:
"""Return a tmp-backed DownloadQueueService and stub the singleton."""
service = DownloadQueueService(db_path=str(tmp_path / "queue.sqlite"))
async def fake_get_instance(_cls: object = None) -> DownloadQueueService:
return service
monkeypatch.setattr(DownloadQueueService, "get_instance", fake_get_instance)
return service
@pytest.mark.asyncio
async def test_complete_missing_download_returns_200_not_404(
queue_service: DownloadQueueService,
) -> None:
"""Completing a download that is not queued is a business outcome."""
handler = _make_handler()
response = await handler.complete_download_in_queue(
_queue_request(
"/api/lm/downloads/queue/complete",
{"download_id": "dl-nope", "status": "completed"},
)
)
assert response.status == 200
text = response.text
assert text is not None
assert json.loads(text) == {
"success": False,
"error": "Download not found in queue",
}
# The failed completion must not have side effects on the queue.
assert await queue_service.get_queue() == []
@pytest.mark.asyncio
async def test_complete_queued_download_returns_success(
queue_service: DownloadQueueService,
) -> None:
"""The happy path still moves the item to history with HTTP 200."""
await queue_service.add_to_queue(download_id="dl-1", model_id=1)
handler = _make_handler()
response = await handler.complete_download_in_queue(
_queue_request(
"/api/lm/downloads/queue/complete",
{"download_id": "dl-1", "status": "completed"},
)
)
assert response.status == 200
text = response.text
assert text is not None
payload = json.loads(text)
assert payload["success"] is True
# The returned item reflects the pre-transition queue record; the
# terminal status lands in history.
history = await queue_service.get_history()
assert len(history["items"]) == 1
assert history["items"][0]["status"] == "completed"
@pytest.mark.asyncio
async def test_status_missing_download_returns_200_not_404(
queue_service: DownloadQueueService,
) -> None:
"""Status updates for unknown items also return 200 with success: false."""
handler = _make_handler()
response = await handler.update_download_queue_status(
_queue_request(
"/api/lm/downloads/queue/status",
{"download_id": "dl-nope", "status": "downloading"},
)
)
assert response.status == 200
text = response.text
assert text is not None
assert json.loads(text) == {
"success": False,
"error": "Download not found in queue",
}
assert await queue_service.get_queue() == []
@pytest.mark.asyncio
async def test_status_queued_download_returns_success(
queue_service: DownloadQueueService,
) -> None:
"""The happy path still updates the queue item with HTTP 200."""
await queue_service.add_to_queue(download_id="dl-2", model_id=2)
handler = _make_handler()
response = await handler.update_download_queue_status(
_queue_request(
"/api/lm/downloads/queue/status",
{"download_id": "dl-2", "status": "downloading"},
)
)
assert response.status == 200
text = response.text
assert text is not None
assert json.loads(text) == {"success": True}
@pytest.mark.asyncio
async def test_retry_missing_history_returns_200_not_404(
queue_service: DownloadQueueService,
) -> None:
"""Retrying a history entry that no longer exists is a business outcome."""
handler = _make_handler()
response = await handler.retry_download_from_history(
_queue_request(
"/api/lm/downloads/history/retry",
{"download_id": "dl-nope"},
)
)
assert response.status == 200
text = response.text
assert text is not None
assert json.loads(text) == {
"success": False,
"error": "History item not found or not retryable",
}
# No side effects: history and queue stay empty.
history = await queue_service.get_history()
assert history["items"] == []
assert await queue_service.get_queue() == []
@pytest.mark.asyncio
async def test_retry_failed_history_returns_success(
queue_service: DownloadQueueService,
) -> None:
"""The happy path still re-queues a retryable history entry with HTTP 200."""
await queue_service.add_to_history(
download_id="dl-fail", model_id=1, status="failed"
)
handler = _make_handler()
response = await handler.retry_download_from_history(
_queue_request(
"/api/lm/downloads/history/retry",
{"download_id": "dl-fail"},
)
)
assert response.status == 200
text = response.text
assert text is not None
payload = json.loads(text)
assert payload["success"] is True
# The retried item is re-queued under a fresh download_id.
queue = await queue_service.get_queue()
assert len(queue) == 1
assert queue[0]["status"] == "queued"
@@ -1,5 +1,7 @@
from pathlib import Path
import threading
import pytest
from py.services.downloaded_version_history_service import (
@@ -70,6 +72,37 @@ async def test_download_history_bulk_lookup(tmp_path: Path) -> None:
}
@pytest.mark.asyncio
async def test_mark_downloaded_bulk_writes_off_event_loop(
tmp_path: Path, monkeypatch
) -> None:
"""The executemany upsert + commit must not run on the event loop thread."""
db_path = tmp_path / "download-history.sqlite"
service = DownloadedVersionHistoryService(
str(db_path),
settings_manager=DummySettings(),
)
loop_thread = threading.get_ident()
write_threads: list[int] = []
original_write = service._mark_downloaded_bulk_sync
def tracking_write(payload):
write_threads.append(threading.get_ident())
return original_write(payload)
monkeypatch.setattr(service, "_mark_downloaded_bulk_sync", tracking_write)
await service.mark_downloaded_bulk(
"lora",
[{"model_id": 7, "version_id": 701, "file_path": "/m/x.safetensors"}],
source="scan",
)
assert write_threads and write_threads[0] != loop_thread
assert await service.has_been_downloaded("lora", 701) is True
@pytest.mark.asyncio
async def test_per_file_history_tracking(tmp_path: Path) -> None:
"""Per-file records coexist with the version-level row (#1058)."""
+69
View File
@@ -4,6 +4,7 @@ import asyncio
import json
import os
import sqlite3
import threading
import time
from collections.abc import Iterator
from pathlib import Path
@@ -391,6 +392,74 @@ async def test_load_persisted_cache_populates_cache(tmp_path: Path, monkeypatch)
assert ws_stub.payloads[-1]['progress'] == 1
@pytest.mark.asyncio
async def test_load_persisted_cache_rebuilds_off_event_loop(tmp_path: Path, monkeypatch):
"""The SQLite read and per-model rebuild must not run on the event loop."""
monkeypatch.setenv('LORA_MANAGER_DISABLE_PERSISTENT_CACHE', '0')
db_path = tmp_path / 'cache.sqlite'
store = PersistentModelCache(db_path=str(db_path))
file_path = tmp_path / 'one.txt'
file_path.write_text('one', encoding='utf-8')
normalized = _normalize_path(file_path)
raw_model = {
'file_path': normalized,
'file_name': 'one',
'model_name': 'one',
'folder': '',
'size': 3,
'modified': 123.0,
'sha256': 'hash-one',
'base_model': 'test',
'preview_url': '',
'preview_nsfw_level': 0,
'from_civitai': True,
'favorite': False,
'notes': '',
'usage_tips': '',
'exclude': False,
'db_checked': False,
'last_checked_at': 0.0,
'tags': ['alpha'],
'civitai': {'id': 11, 'modelId': 22, 'name': 'ver', 'trainedWords': ['abc']},
}
store.save_cache('dummy', [raw_model], {'hash-one': [normalized]}, [])
monkeypatch.setattr(model_scanner, 'get_persistent_cache', lambda: store)
scanner = DummyScanner(tmp_path)
ws_stub = RecordingWebSocketManager()
monkeypatch.setattr(model_scanner, 'ws_manager', ws_stub)
loop_thread = threading.get_ident()
worker_threads: List[int] = []
original_load_cache = store.load_cache
def tracking_load_cache(model_type):
worker_threads.append(threading.get_ident())
return original_load_cache(model_type)
monkeypatch.setattr(store, 'load_cache', tracking_load_cache)
original_adjust = scanner.adjust_cached_entry
def tracking_adjust(entry):
worker_threads.append(threading.get_ident())
return original_adjust(entry)
monkeypatch.setattr(scanner, 'adjust_cached_entry', tracking_adjust)
loaded = await scanner._load_persisted_cache('dummy')
assert loaded is True
# Both the SQLite read and the per-entry adjustment ran off the loop
assert len(worker_threads) == 2
assert all(tid != loop_thread for tid in worker_threads)
cache = await scanner.get_cached_data()
assert len(cache.raw_data) == 1
assert cache.raw_data[0]['file_path'] == normalized
@pytest.mark.asyncio
async def test_update_single_model_cache_persists_changes(tmp_path: Path, monkeypatch):
monkeypatch.setenv('LORA_MANAGER_DISABLE_PERSISTENT_CACHE', '0')
@@ -16,6 +16,7 @@ import errno
import json
import os
import shutil
import threading
import time
from collections.abc import Iterator
from pathlib import Path
@@ -1683,6 +1684,33 @@ async def test_reg_g_reconciliation_finds_external_batches(
assert service._known_batch_dirs.get("ext-fresh") == str(fresh_dir)
# (g2) reconciliation runs the filesystem walk off the event loop so a large
# or slow library cannot block startup
async def test_reg_g2_reconciliation_walk_runs_in_worker_thread(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
root = tmp_path / "loras"
root.mkdir()
(root / PENDING_DELETE_DIR_NAME).mkdir()
await _register_model_root(monkeypatch, lora_roots=[root])
loop_thread = threading.get_ident()
walk_threads: List[int] = []
real_walk = os.walk
def _recording_walk(*args: Any, **kwargs: Any):
walk_threads.append(threading.get_ident())
return real_walk(*args, **kwargs)
monkeypatch.setattr(os, "walk", _recording_walk)
service = await PendingDeleteService.get_instance()
await service._reconcile_scan_roots()
assert walk_threads, "reconciliation never walked the model roots"
assert all(thread_id != loop_thread for thread_id in walk_threads)
# (h) _find_batch_dir with cleared registry locates + registers (restart sim)
async def test_reg_h_find_batch_dir_restart_simulation(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
@@ -0,0 +1,104 @@
"""Tests for the one-shot source_path backfill in RecipeScanner."""
import json
from pathlib import Path
import pytest
from py.services.persistent_recipe_cache import PersistentRecipeCache
from py.services.recipe_scanner import RecipeScanner
@pytest.fixture
def scanner_with_cache(tmp_path: Path):
"""RecipeScanner instance backed by a real persistent cache in tmp_path."""
cache = PersistentRecipeCache(db_path=str(tmp_path / "recipe_cache.sqlite"))
scanner = RecipeScanner.__new__(RecipeScanner)
scanner._persistent_cache = cache
return scanner, cache
def _write_recipe_json(path: Path, recipe_id: str, source_path: str | None = None) -> None:
data = {"id": recipe_id}
if source_path is not None:
data["source_path"] = source_path
path.write_text(json.dumps(data), encoding="utf-8")
def test_backfill_processes_recipes_when_marker_absent(scanner_with_cache, tmp_path: Path):
"""Without the completion marker, missing source_path values are backfilled."""
scanner, cache = scanner_with_cache
json_with_source = tmp_path / "r1.recipe.json"
_write_recipe_json(json_with_source, "r1", source_path="https://civitai.com/images/1")
json_without_source = tmp_path / "r2.recipe.json"
_write_recipe_json(json_without_source, "r2")
recipes = [
{"id": "r1", "source_path": ""},
{"id": "r2", "source_path": ""},
{"id": "r3", "source_path": "https://civitai.com/images/3"},
]
json_paths = {"r1": str(json_with_source), "r2": str(json_without_source)}
updated = scanner._backfill_source_path_if_needed(recipes, json_paths)
assert updated is True
assert recipes[0]["source_path"] == "https://civitai.com/images/1"
# JSON legitimately has no source_path: stays empty, no error
assert recipes[1]["source_path"] == ""
assert recipes[2]["source_path"] == "https://civitai.com/images/3"
# The run records the completion marker
assert (
cache.get_metadata_value(RecipeScanner._SOURCE_PATH_BACKFILL_MARKER) == "1"
)
def test_backfill_is_skipped_once_marker_is_set(scanner_with_cache, tmp_path: Path, monkeypatch):
"""The second initialization must not re-read recipe JSON files."""
scanner, cache = scanner_with_cache
json_path = tmp_path / "r1.recipe.json"
_write_recipe_json(json_path, "r1", source_path="https://civitai.com/images/1")
recipes = [{"id": "r1", "source_path": ""}]
json_paths = {"r1": str(json_path)}
# First run: backfills and records the marker
assert scanner._backfill_source_path_if_needed(recipes, json_paths) is True
assert recipes[0]["source_path"] == "https://civitai.com/images/1"
# Second run: simulate a fresh startup where the cache still lacks
# source_path. Under the old behavior the file would be re-read and
# re-parsed; now the marker must suppress any file access.
recipes = [{"id": "r1", "source_path": ""}]
import os
real_exists = os.path.exists
def _failing_exists(path):
if str(path).endswith(".recipe.json"):
raise AssertionError("backfill touched the filesystem despite marker")
return real_exists(path)
monkeypatch.setattr(os.path, "exists", _failing_exists)
updated = scanner._backfill_source_path_if_needed(recipes, json_paths)
assert updated is False
assert recipes[0]["source_path"] == ""
def test_backfill_marker_does_not_suppress_reconcile_parsed_source_path(scanner_with_cache):
"""The marker only gates the backfill; parsed recipes keep their source_path."""
scanner, cache = scanner_with_cache
cache.set_metadata_value(RecipeScanner._SOURCE_PATH_BACKFILL_MARKER, "1")
# A recipe that arrived from the normal parse path with a source_path is
# left untouched by the (skipped) backfill.
recipes = [{"id": "r1", "source_path": "https://civitai.com/images/9"}]
updated = scanner._backfill_source_path_if_needed(recipes, {})
assert updated is False
assert recipes[0]["source_path"] == "https://civitai.com/images/9"
+21
View File
@@ -552,6 +552,27 @@ class TestPersistentRecipeCache:
assert loaded is not None
assert loaded.image_id_map == {"222": "new-only"}
def test_metadata_value_roundtrip(self, temp_db_path):
"""set_metadata_value/get_metadata_value store and replace values."""
cache = PersistentRecipeCache(db_path=temp_db_path)
assert cache.get_metadata_value("source_path_backfilled") is None
cache.set_metadata_value("source_path_backfilled", "1")
assert cache.get_metadata_value("source_path_backfilled") == "1"
cache.set_metadata_value("source_path_backfilled", "2")
assert cache.get_metadata_value("source_path_backfilled") == "2"
def test_metadata_value_survives_save_cache(self, temp_db_path, sample_recipes):
"""A full save_cache must not drop unrelated cache_metadata entries."""
cache = PersistentRecipeCache(db_path=temp_db_path)
cache.set_metadata_value("source_path_backfilled", "1")
cache.save_cache(sample_recipes)
assert cache.get_metadata_value("source_path_backfilled") == "1"
class TestHasWorkflowColumn:
"""has_workflow column persistence (plan 3.1)."""
+85
View File
@@ -181,3 +181,88 @@ class TestFTSIndexValidation:
# Search should still work
results = fts.search("anime")
assert "recipe-001" in results
@staticmethod
def _forbid_content_scan(monkeypatch):
"""Fail the test if validation falls back to scanning the FTS content table."""
def _raise(*args, **kwargs):
raise AssertionError("validate_index scanned the FTS content table")
monkeypatch.setattr(RecipeFTSIndex, "get_indexed_count", _raise)
monkeypatch.setattr(RecipeFTSIndex, "get_indexed_recipe_ids", _raise)
def test_validate_index_with_metadata_does_not_scan(
self, temp_db_path, sample_recipes, monkeypatch
):
"""Validation must rely on stored metadata, not content-table scans."""
fts = RecipeFTSIndex(db_path=temp_db_path)
fts.build_index(sample_recipes)
self._forbid_content_scan(monkeypatch)
result = fts.validate_index(3, {"recipe-001", "recipe-002", "recipe-003"})
assert result is True
# Mismatches are also detected from metadata alone
result = fts.validate_index(4, {"recipe-001", "recipe-002", "recipe-003"})
assert result is False
result = fts.validate_index(3, {"recipe-001", "recipe-002", "recipe-999"})
assert result is False
def test_validate_index_legacy_fallback_writes_metadata(
self, temp_db_path, sample_recipes, monkeypatch
):
"""Indexes built by older versions validate via a one-time scan, then metadata."""
import sqlite3
fts = RecipeFTSIndex(db_path=temp_db_path)
fts.build_index(sample_recipes)
# Simulate a legacy index: drop the fingerprint metadata
conn = sqlite3.connect(temp_db_path)
try:
conn.execute(
"DELETE FROM fts_metadata WHERE key = ?",
(RecipeFTSIndex._FINGERPRINT_METADATA_KEY,)
)
conn.commit()
finally:
conn.close()
# Fallback validation scans once and succeeds
result = fts.validate_index(3, {"recipe-001", "recipe-002", "recipe-003"})
assert result is True
# Metadata was written, so the next validation needs no scan
self._forbid_content_scan(monkeypatch)
result = fts.validate_index(3, {"recipe-001", "recipe-002", "recipe-003"})
assert result is True
def test_validate_index_metadata_tracks_incremental_mutations(
self, temp_db_path, sample_recipes, monkeypatch
):
"""add_recipe/remove_recipe keep the validation metadata in sync."""
fts = RecipeFTSIndex(db_path=temp_db_path)
fts.build_index(sample_recipes[:2])
fts.add_recipe(sample_recipes[2])
fts.remove_recipe("recipe-001")
self._forbid_content_scan(monkeypatch)
result = fts.validate_index(2, {"recipe-002", "recipe-003"})
assert result is True
def test_validate_index_after_clear_uses_metadata(
self, temp_db_path, sample_recipes, monkeypatch
):
"""clear() resets the validation metadata to the empty index state."""
fts = RecipeFTSIndex(db_path=temp_db_path)
fts.build_index(sample_recipes)
fts.clear()
self._forbid_content_scan(monkeypatch)
assert fts.validate_index(0, set()) is True
assert fts.validate_index(3, {"recipe-001", "recipe-002", "recipe-003"}) is False