fix(lora): warn about refused default LoRA folders and log what failed (#362, phase 4)

ComfyUI's default LoRA folders already pass the folder guard; a refused one
that exists is now listed in the layout's refused_defaults, and the layout
cache warns once per newly refused folder. The cache writes one INFO line
when a stored layout differs from the last one logged. BAD_SIDECAR_ROOT and
the module docstring name a folder that contains the home folder. A relative
preview_url writes a DEBUG line before the suffix search; a non-OSError
failure while loading trigger words logs its traceback. Review follow-ups
harden two serve tests.
This commit is contained in:
Vito Sansevero
2026-10-26 22:59:07 -07:00
parent 5b0468eaee
commit f020f1d751
7 changed files with 270 additions and 15 deletions
+29
View File
@@ -88,6 +88,10 @@ class LayoutCache:
self._entry: Optional[_Entry] = None
self._generation = 0
self._lock = threading.Lock()
# What was last logged; survives clear() so a reload of the same
# layout stays quiet.
self._logged_layout: Optional[LoraManagerLayout] = None
self._logged_refused: Tuple[str, ...] = ()
def get(self, configured: str) -> Tuple[Optional[str], LoraManagerLayout]:
"""``(install, layout)`` for the configured path ("" auto-detects)."""
@@ -101,6 +105,7 @@ class LayoutCache:
generation = self._generation
install = self._resolve(key)
layout = self._load(install)
self._log_refused(layout)
try:
signatures = tuple(
(path, self._signature(path)) for path in self._watched(install, layout)
@@ -110,9 +115,18 @@ class LayoutCache:
if layout.error == UNREADABLE or self._racy(signatures, self._wall_ns()):
return install, layout
stored = _Entry(key, install, layout, signatures, self._monotonic())
changed = False
with self._lock:
if self._generation == generation:
self._entry = stored
changed = layout != self._logged_layout
if changed:
self._logged_layout = layout
if changed:
logger.info(
f"LoraManager layout changed: {layout.source or 'no'} settings, "
f"{layout.mode} storage, {len(layout.media_roots)} media folders"
)
return install, layout
def clear(self) -> None:
@@ -121,6 +135,21 @@ class LayoutCache:
self._entry = None
self._generation += 1
def _log_refused(self, layout: LoraManagerLayout) -> None:
"""One warning per default LoRA folder newly refused by the guard."""
refused = tuple(layout.refused_defaults)
with self._lock:
if refused == self._logged_refused:
return
new = [folder for folder in refused if folder not in self._logged_refused]
self._logged_refused = refused
for folder in new:
logger.warning(
f"ComfyUI LoRA folder '{folder}' is neither scanned nor served: "
"it is a filesystem root, the home folder or a folder that "
"contains it"
)
# ── validation ──────────────────────────────────────────────────
def _valid(self, entry: _Entry, key: str) -> bool:
+51 -8
View File
@@ -12,8 +12,10 @@ server share one definition:
plugin file, or the user config folder, in LoraManager's order), keeps only
the seven keys the import needs, and resolves the LoRA roots, the sidecar
root (centralized storage), the example-images root and the active
LoraManager library. Every folder from the settings passes the folder
guard (exists, not a filesystem root, not the home folder itself).
LoraManager library. Every folder from the settings, and each of
ComfyUI's default LoRA folders, passes the folder guard: it must exist and
must not be a filesystem root, the home folder or a folder that contains
it. Default folders the guard refuses are listed in ``refused_defaults``.
- ``discover_sidecars`` finds every sidecar whose model file exists inside a
LoRA root: beside the models in ``alongside`` mode, under the sidecar root
(matched to its model by ``file_path``) in ``centralized`` mode.
@@ -25,6 +27,7 @@ it is testable outside ComfyUI.
"""
import json
import logging
import os
import re
from dataclasses import dataclass, replace
@@ -98,7 +101,12 @@ NOT_OBJECT = "the settings file is not a JSON object"
TOO_LARGE = "the settings file is larger than 5 MB"
NO_PLATFORMDIRS = "platformdirs is not available"
NO_SIDECAR_ROOT = "the sidecar folder does not exist"
BAD_SIDECAR_ROOT = "the sidecar folder is a filesystem root or the home folder"
BAD_SIDECAR_ROOT = (
"the sidecar folder is a filesystem root, the home folder or a folder "
"that contains it"
)
logger = logging.getLogger("prompt_manager.lora_manager_layout")
@dataclass(frozen=True)
@@ -114,6 +122,9 @@ class LoraManagerLayout:
example_images_root: Optional[str]
library: str
media_roots: Tuple[str, ...]
# ComfyUI's default LoRA folders that exist but that the folder guard
# refused (canonical): the layout cache logs them (#362).
refused_defaults: Tuple[str, ...] = ()
@dataclass(frozen=True)
@@ -322,6 +333,23 @@ def _guarded(paths, home: str) -> Tuple[str, ...]:
return tuple(kept)
def _refused(paths, home: str) -> Tuple[str, ...]:
"""The folders in ``paths`` that exist but that the guard refuses."""
refused = []
for path in paths:
absolute = _absolute(path)
if not absolute or not os.path.isdir(absolute):
continue
if _guard_folder(absolute, home) is None:
try:
canonical = _canonical(absolute)
except (OSError, ValueError):
continue
if canonical not in refused:
refused.append(canonical)
return tuple(refused)
def _example_images_root(
settings: Mapping, library: str, libraries: int, home: str
) -> Optional[str]:
@@ -358,7 +386,9 @@ def load_layout(
default LoRA roots, so serving and trigger words keep working.
"""
home = str(Path.home())
defaults = _guarded(default_roots(), home)
default_list = list(default_roots())
defaults = _guarded(default_list, home)
refused_defaults = _refused(default_list, home)
fallback = LoraManagerLayout(
settings_path=None,
source=None,
@@ -369,6 +399,7 @@ def load_layout(
example_images_root=None,
library="default",
media_roots=defaults,
refused_defaults=refused_defaults,
)
if not lora_manager_path:
return fallback
@@ -384,10 +415,14 @@ def load_layout(
except _SettingsError as error:
return replace(fallback, error=str(error))
settings = _settings_subset(raw or {})
return _layout_from_settings(settings, settings_path, source, defaults, home)
return _layout_from_settings(
settings, settings_path, source, defaults, home, refused_defaults
)
def _layout_from_settings(settings, settings_path, source, defaults, home):
def _layout_from_settings(
settings, settings_path, source, defaults, home, refused_defaults=()
):
library, entry, libraries = _active_library(settings)
lora_roots = _media_roots(
defaults,
@@ -407,6 +442,7 @@ def _layout_from_settings(settings, settings_path, source, defaults, home):
example_images_root=example_root,
library=library,
media_roots=_media_roots(lora_roots, [example_root]),
refused_defaults=refused_defaults,
)
if mode != MODE_CENTRALIZED:
return layout
@@ -534,8 +570,15 @@ def _example_files(folder: str) -> List[str]:
def _preview_candidates(sidecar: Sidecar, metadata: Dict) -> Iterator[str]:
"""The paths a preview is looked for at, in LoraManager's order."""
url = metadata.get("preview_url")
if isinstance(url, str) and url.strip() and os.path.isabs(url.strip()):
yield url.strip()
if isinstance(url, str) and url.strip():
if os.path.isabs(url.strip()):
yield url.strip()
else:
# LoraManager writes absolute paths; it would resolve a relative
# one against its working directory, not the sidecar folder.
logger.debug(
f"Relative preview_url skipped in {sidecar.metadata_path.name}"
)
base = sidecar.metadata_path.name[: -len(METADATA_SUFFIX)]
for suffix in PREVIEW_EXTENSIONS:
yield str(sidecar.metadata_path.parent / f"{base}{suffix}")
+12 -4
View File
@@ -883,11 +883,13 @@ class TriggerWordCache:
"""
try:
new_cache = self._build(lora_manager_path)
except Exception as error:
except OSError as error:
logger.warning(f"Trigger word cache not refreshed: {error}")
with self._lock:
self._failed_at = time.monotonic()
return len(self._cache)
return self._failed()
except Exception:
# Not a disk error: keep the traceback (#362).
logger.exception("Trigger word cache not refreshed")
return self._failed()
with self._lock:
self._cache = new_cache
@@ -899,6 +901,12 @@ class TriggerWordCache:
)
return len(new_cache)
def _failed(self) -> int:
"""Keep the previous map and note when this load failed."""
with self._lock:
self._failed_at = time.monotonic()
return len(self._cache)
@staticmethod
def _build(lora_manager_path: str) -> Dict[str, List[str]]:
layout = lora_manager_layout(lora_manager_path)
+5
View File
@@ -1524,6 +1524,7 @@ class TestLoraManagerMediaServing(ImageRouteCoverageCase):
with patch.object(layout_module.Path, "home", return_value=outer / "home"):
self.assertEqual(await self._status(image_id), 200)
self.assertIsNotNone(lora_utils._layout_cache._entry)
# A skip after the rmtree is safe: setUp builds a fresh tree.
shutil.rmtree(self.examples)
self._make_symlink(outer, self.examples, target_is_directory=True)
self.assertEqual(await self._status(image_id), 403)
@@ -1533,8 +1534,12 @@ class TestLoraManagerMediaServing(ImageRouteCoverageCase):
self.assertEqual(await self._status(self._link_image(self._example())), 200)
self.assertIsNotNone(lora_utils._layout_cache._entry)
stored = lora_utils._layout_cache._entry
with patch.object(IntegrationConfig, "LORA_MANAGER_PATH", str(self.sidecars)):
self.assertEqual(self.api._lora_image_dirs(), [])
# The warm entry for the real install was never handed out for it.
self.assertIsNone(lora_utils.cached_lora_manager_layout(str(self.sidecars))[0])
self.assertIsNotNone(stored)
async def test_serve_image_builds_the_allow_list_in_the_executor(self):
calls = self._spy_executor()
+56
View File
@@ -16,6 +16,7 @@ import threading
import time
import unittest
from pathlib import Path, PureWindowsPath
from dataclasses import replace
from types import SimpleNamespace
from unittest.mock import patch
@@ -513,6 +514,61 @@ class ConcurrencyTests(CacheTestCase):
self.assertEqual(len(self.loaded), 2)
class LoggingTests(CacheTestCase):
LOGGER = "prompt_manager.lora_layout_cache"
def get(self, cache):
with self.assertLogs(self.LOGGER, "DEBUG") as logs:
cache.get("")
# assertLogs needs one record; this marker is not counted.
cache_module.logger.debug("marker")
return [r for r in logs.records if r.getMessage() != "marker"]
def test_a_new_layout_logs_one_info_line(self):
cache = self.cache()
records = self.get(cache)
self.assertEqual([r.levelname for r in records], ["INFO"])
self.assertEqual(
records[0].getMessage(),
"LoraManager layout changed: user config settings, alongside storage, "
"1 media folders",
)
self.mono += MAX_AGE_SECONDS
self.assertEqual(self.get(cache), []) # an equal layout, reloaded
cache.clear()
self.assertEqual(self.get(cache), [])
self.layout = replace(self.layout, media_roots=(), lora_roots=())
self.mono += MAX_AGE_SECONDS
self.assertEqual(len(self.get(cache)), 1)
def test_no_settings_source_prints_no(self):
self.layout = replace(self.layout, source=None)
records = self.get(self.cache())
self.assertIn("layout changed: no settings,", records[0].getMessage())
def test_an_unstored_load_logs_no_info_line(self):
self.write(self.settings, "{}", mtime_ns=self.now_ns)
self.assertEqual(
[r for r in self.get(self.cache()) if r.levelname == "INFO"], []
)
def test_a_newly_refused_default_folder_logs_one_warning(self):
self.layout = replace(self.layout, refused_defaults=("/home",))
cache = self.cache()
records = self.get(cache)
warnings = [r for r in records if r.levelname == "WARNING"]
self.assertEqual(len(warnings), 1)
self.assertIn("/home", warnings[0].getMessage())
self.mono += MAX_AGE_SECONDS
self.assertEqual(self.get(cache), [])
cache.clear()
self.assertEqual(self.get(cache), [])
self.layout = replace(self.layout, refused_defaults=("/home", "/"))
self.mono += MAX_AGE_SECONDS
warnings = [r for r in self.get(cache) if r.levelname == "WARNING"]
self.assertEqual([w.getMessage().count("'/'") for w in warnings], [1])
class ModuleTests(unittest.TestCase):
def test_constants(self):
self.assertEqual(cache_module.RACY_SECONDS, 3)
+89 -3
View File
@@ -33,6 +33,10 @@ MEDIA = frozenset(
{".png", ".jpg", ".jpeg", ".webp", ".gif", ".mp4", ".webm", ".mov", ".bmp"}
)
MARKER = "marker-from-unkept-settings-0123"
BAD_SIDECAR_REASON = (
"the sidecar folder is a filesystem root, the home folder or a folder "
"that contains it"
)
class LayoutTestCase(unittest.TestCase):
@@ -261,11 +265,26 @@ class UnusableLayoutTests(LayoutTestCase):
{"sidecar_storage_mode": "centralized", "sidecar_storage_path": "/"},
)
layout = self.load()
self.assertEqual(
layout.error, "the sidecar folder is a filesystem root or the home folder"
)
self.assertEqual(layout.error, BAD_SIDECAR_REASON)
self.assertNotIn(os.path.normcase(os.path.realpath("/")), layout.media_roots)
def test_a_centralized_sidecar_root_that_contains_home(self):
home = self.tmp / "outer" / "me"
home.mkdir(parents=True)
self.write_settings(
self.user_config,
{
"sidecar_storage_mode": "centralized",
"sidecar_storage_path": str(self.tmp / "outer"),
},
)
with patch.object(layout_module.Path, "home", return_value=home):
layout = self.load()
self.assertEqual(layout.error, BAD_SIDECAR_REASON)
self.assertEqual(
layout.media_roots, (os.path.normcase(str(self.default_loras)),)
)
class SecretTests(LayoutTestCase):
@@ -636,6 +655,16 @@ class LocalMediaTests(LayoutTestCase):
self.touch(self.default_loras / "alpha.webp")
self.assertEqual(self.media({"preview_url": preview}), [preview])
def test_a_relative_preview_url_is_logged_and_skipped(self):
fallback = self.touch(self.default_loras / "alpha.webp")
self.touch(self.default_loras / "relative.png")
with self.assertLogs("prompt_manager.lora_manager_layout", "DEBUG") as logs:
media = self.media({"preview_url": "relative.png"})
self.assertEqual(media, [fallback])
self.assertEqual(len(logs.records), 1)
self.assertEqual(logs.records[0].levelname, "DEBUG")
self.assertIn(self.meta.name, logs.output[0])
def test_preview_url_outside_media_roots_or_not_servable_is_ignored(self):
outside = self.touch(self.tmp / "elsewhere" / "p.png")
avif = self.touch(self.default_loras / "p.avif")
@@ -835,6 +864,63 @@ class FolderGuardTests(unittest.TestCase):
self.assertFalse(layout_module._inside(str(self.tmp / "other.png"), (root,)))
class RefusedDefaultsTests(LayoutTestCase):
"""#362: ComfyUI's default LoRA folders pass the same guard; the refused
ones that exist are listed so the layout cache can warn about them."""
def setUp(self):
super().setUp()
self.outer = self.tmp / "outer"
self.home = self.outer / "me"
self.home.mkdir(parents=True)
self.fs_root = os.path.abspath(os.sep)
self.defaults = [
str(self.home),
str(self.outer),
self.fs_root,
str(self.tmp / "missing"),
str(self.default_loras),
]
self.expected = (
layout_module._canonical(str(self.home)),
layout_module._canonical(str(self.outer)),
layout_module._canonical(self.fs_root),
)
home_patch = patch.object(layout_module.Path, "home", return_value=self.home)
home_patch.start()
self.addCleanup(home_patch.stop)
def load_defaults(self, path="__plugin__"):
return self.load(path, default_roots=lambda: list(self.defaults))
def test_refused_existing_defaults_are_listed_on_a_usable_layout(self):
self.write_settings(self.user_config, {})
layout = self.load_defaults()
self.assertIsNone(layout.error)
self.assertEqual(layout.refused_defaults, self.expected)
self.assertEqual(
layout.lora_roots, (os.path.normcase(str(self.default_loras)),)
)
def test_they_are_listed_on_an_unusable_centralized_layout(self):
self.write_settings(
self.user_config,
{"sidecar_storage_mode": "centralized", "sidecar_storage_path": "/"},
)
layout = self.load_defaults()
self.assertEqual(layout.error, BAD_SIDECAR_REASON)
self.assertEqual(layout.refused_defaults, self.expected)
def test_they_are_listed_without_an_install(self):
layout = self.load_defaults(None)
self.assertEqual(layout.error, "no LoraManager install found")
self.assertEqual(layout.refused_defaults, self.expected)
def test_nothing_is_listed_when_every_default_is_accepted(self):
self.write_settings(self.user_config, {})
self.assertEqual(self.load().refused_defaults, ())
class SummaryTests(LayoutTestCase):
def test_summary_names_what_the_import_read(self):
+28
View File
@@ -1791,6 +1791,34 @@ class TestTriggerWordCacheLayout(unittest.TestCase):
self.assertTrue(cache.is_loaded)
self.assertTrue(cache.retry_due())
def test_an_os_error_is_a_warning_without_a_traceback(self):
cache = TriggerWordCache()
cache._cache = {"old": ["word"]}
with (
patch.object(TriggerWordCache, "_build", side_effect=OSError("denied")),
self.assertLogs("prompt_manager.lora_utils", level="WARNING") as logs,
):
self.assertEqual(cache.load(str(self.lm)), 1)
self.assertEqual([r.levelname for r in logs.records], ["WARNING"])
self.assertIsNone(logs.records[0].exc_info)
self.assertEqual(cache.get_trigger_words("old"), ["word"])
self.assertFalse(cache.retry_due())
def test_an_unexpected_failure_logs_its_traceback(self):
# #362: a programming error must not look like a disk error.
cache = TriggerWordCache()
cache._cache = {"old": ["word"]}
with (
patch.object(TriggerWordCache, "_build", side_effect=ValueError("bug")),
self.assertLogs("prompt_manager.lora_utils", level="WARNING") as logs,
):
self.assertEqual(cache.load(str(self.lm)), 1)
self.assertEqual(len(logs.records), 1)
self.assertIsNotNone(logs.records[0].exc_info)
self.assertIs(logs.records[0].exc_info[0], ValueError)
self.assertEqual(cache.get_trigger_words("old"), ["word"])
self.assertFalse(cache.retry_due())
def test_any_other_failure_never_raises(self):
cache = TriggerWordCache()
with (