From dead4a7aa37b45a83e79082f2c514ea0a95c4597 Mon Sep 17 00:00:00 2001 From: Vito Sansevero Date: Thu, 22 Oct 2026 20:04:09 -0700 Subject: [PATCH] fix(config): read under the commit lock and publish saves in write order - ConfigStore.read takes the commit lock, so no read handle is open while config.json is replaced (Windows refuses that replace). - os.replace is retried briefly on a Windows PermissionError (antivirus or indexer handles) before the save is reported as not written. - The temp file is written beside realpath(config.json), so a symlinked config keeps its link and a bind mount fails truthfully. - commit() takes a publish callback that runs under the lock after the write; settings and LoRA saves publish there, so two concurrent saves reach the runtime in write order and the runtime ends on the last one. - The concurrency tests now wait until the second caller is blocked on the lock (tests/lock_probe.py) before releasing the first. --- py/api/admin.py | 22 +++-- py/api/lora_integration.py | 15 ++-- py/config_store.py | 48 ++++++++-- tests/lock_probe.py | 35 ++++++++ tests/test_api_lora.py | 95 +++++++++++++++++++- tests/test_config_store.py | 175 ++++++++++++++++++++++++++++++++++--- 6 files changed, 356 insertions(+), 34 deletions(-) create mode 100644 tests/lock_probe.py diff --git a/py/api/admin.py b/py/api/admin.py index bb93b42..7c1ace6 100644 --- a/py/api/admin.py +++ b/py/api/admin.py @@ -1026,11 +1026,19 @@ class AdminRoutesMixin: if error: return web.json_response({"success": False, "error": error}, status=400) + published = {} + + def publish(_config): + # Under the commit lock: saves reach the runtime in the order + # they were written, so the runtime ends on the last one. + published.update(self._publish_settings(updates, new_roots)) + try: _saved, revision = await self._run_in_executor( get_config_store().commit, _settings_mutation(updates, new_roots), expected_revision=expected_revision, + publish=publish, ) except (ConfigUnreadable, RevisionConflict, OSError) as exc: return self._config_commit_failure(exc) @@ -1039,7 +1047,10 @@ class AdminRoutesMixin: { "success": True, "message": "Settings saved successfully", - **self._publish_settings(updates, new_roots, revision), + "restart_required": bool(published["restart_required_fields"]), + "persisted": True, + "revision": revision, + **published, } ) except Exception as e: @@ -1049,12 +1060,12 @@ class AdminRoutesMixin: status=500, ) - def _publish_settings(self, updates, new_roots, revision): - """Install committed settings at runtime; the ConfigCommit fields. + def _publish_settings(self, updates, new_roots): + """Install committed settings at runtime (inside the config commit). The values are already durable. A publication that fails is reported as ``runtime_applied: false`` with the fields that need a restart; - nothing is retried or rolled back. + nothing is retried or rolled back. Never raises. """ from ..config import PromptManagerConfig @@ -1073,9 +1084,6 @@ class AdminRoutesMixin: runtime_applied = False restart_fields.append("gallery_root_paths") return { - "restart_required": bool(restart_fields), - "persisted": True, - "revision": revision, "runtime_applied": runtime_applied, "restart_required_fields": restart_fields, } diff --git a/py/api/lora_integration.py b/py/api/lora_integration.py index c24b852..63a87c3 100644 --- a/py/api/lora_integration.py +++ b/py/api/lora_integration.py @@ -204,20 +204,25 @@ class LoraIntegrationMixin: "trigger_words_enabled": trigger_words, "civitai_api_key": civitai_key, } + + def publish(_config): + # Under the commit lock, so the last save written is the one + # the runtime keeps. + IntegrationConfig.LORA_MANAGER_ENABLED = enabled + IntegrationConfig.LORA_MANAGER_PATH = path + IntegrationConfig.LORA_TRIGGER_WORDS_ENABLED = trigger_words + IntegrationConfig.CIVITAI_API_KEY = civitai_key + try: _saved, revision = await self._run_in_executor( get_config_store().commit, _lora_settings_mutation(values), expected_revision=None, + publish=publish, ) except (ConfigUnreadable, RevisionConflict, OSError) as exc: return self._config_commit_failure(exc) - IntegrationConfig.LORA_MANAGER_ENABLED = enabled - IntegrationConfig.LORA_MANAGER_PATH = path - IntegrationConfig.LORA_TRIGGER_WORDS_ENABLED = trigger_words - IntegrationConfig.CIVITAI_API_KEY = civitai_key - cache = get_trigger_cache() if enabled and trigger_words and path: try: diff --git a/py/config_store.py b/py/config_store.py index b05bb0d..949423a 100644 --- a/py/config_store.py +++ b/py/config_store.py @@ -12,6 +12,11 @@ A revision names the file content a client last saw: ``"sha256:"`` of the file bytes, or ``"absent"`` when there is no file. A commit that names a revision refuses to write over different content (``RevisionConflict``). +Reads take the same lock, so a read handle is never open while the file is +replaced (Windows refuses that), and a caller's ``publish`` runs inside the +lock right after its write, so runtimes publish saves in the order they were +written. A symlinked config.json is written beside its real target. + config.json can hold the CivitAI key, so the file is written owner-only. """ @@ -21,12 +26,19 @@ import json import os import tempfile import threading +import time from typing import Callable, Optional, Tuple ABSENT_REVISION = "absent" -# One lock for the process: every ConfigStore writes the same file. -_COMMIT_LOCK = threading.Lock() +# One lock for the process: every ConfigStore writes the same file. It is +# re-entrant so a commit's ``publish`` callback may read the config again. +_COMMIT_LOCK = threading.RLock() + +# Windows: an antivirus, indexer or backup handle on config.json makes +# os.replace fail with PermissionError for a moment, so it is retried briefly. +_REPLACE_ATTEMPTS = 5 if os.name == "nt" else 1 +_REPLACE_BACKOFF_SECONDS = 0.05 class ConfigUnreadable(Exception): @@ -54,13 +66,29 @@ def _default_path() -> str: return PromptManagerConfig.get_config_path() +def _replace(source: str, target: str) -> None: + """``os.replace`` with a short retry on a transient Windows PermissionError.""" + for attempt in range(1, _REPLACE_ATTEMPTS + 1): + try: + os.replace(source, target) + return + except PermissionError: + if attempt == _REPLACE_ATTEMPTS: + raise + time.sleep(_REPLACE_BACKOFF_SECONDS) + + def _write_atomically(path: str, payload: bytes) -> None: """Replace ``path`` with ``payload``: temp file beside it, fsync, replace. The descriptor is closed before ``os.replace`` (Windows refuses to replace over an open handle), and on any failure the temp file is closed and removed, so a failed write leaves the old file and no stray temp file. + ``path`` is resolved first: a symlink keeps pointing at the updated + target. A single-file bind mount cannot be replaced (EBUSY) and fails as + an OSError, which callers report as not saved. """ + path = os.path.realpath(path) parent = os.path.dirname(path) or "." os.makedirs(parent, exist_ok=True) fd, temp_path = tempfile.mkstemp(dir=parent, prefix=".config-", suffix=".tmp") @@ -76,7 +104,7 @@ def _write_atomically(path: str, payload: bytes) -> None: os.fsync(handle.fileno()) if os.name == "posix": os.chmod(temp_path, 0o600) - os.replace(temp_path, path) + _replace(temp_path, path) except BaseException: try: os.unlink(temp_path) @@ -100,9 +128,11 @@ class ConfigStore: """``(config, revision)``; a missing file is ``({}, "absent")``. Raises ConfigUnreadable when the file cannot be read, is not JSON, or - is not a JSON object. + is not a JSON object. Waits for a running commit (see the module + docstring). """ - return self._read(self.path) + with _COMMIT_LOCK: + return self._read(self.path) @staticmethod def _read(path: str) -> Tuple[dict, str]: @@ -128,6 +158,7 @@ class ConfigStore: mutate: Callable[[dict], dict], *, expected_revision: Optional[str], + publish: Optional[Callable[[dict], None]] = None, ) -> Tuple[dict, str]: """Apply ``mutate`` to the current config and write the result. @@ -137,6 +168,11 @@ class ConfigStore: write over different content. Returns ``(config, revision)`` as written. Raises ConfigUnreadable, RevisionConflict, TypeError (a non-object result) or OSError (the write failed; the old file stays). + + ``publish(config)``, when given, runs under the lock once the file + holds the new config, so concurrent saves reach the runtime in the + order they were written. It must handle its own errors: the values + are already durable when it runs. """ with _COMMIT_LOCK: path = self.path @@ -148,6 +184,8 @@ class ConfigStore: raise TypeError("A config commit must produce a JSON object") payload = json.dumps(candidate, indent=2).encode("utf-8") _write_atomically(path, payload) + if publish is not None: + publish(candidate) return candidate, revision_of(payload) diff --git a/tests/lock_probe.py b/tests/lock_probe.py new file mode 100644 index 0000000..88146f2 --- /dev/null +++ b/tests/lock_probe.py @@ -0,0 +1,35 @@ +"""A lock stand-in that reports when a second caller has to wait for it. + +Concurrency tests patch it over a module's lock so they can wait until the +second caller is really blocked on the lock, instead of hoping a thread +start or one event-loop tick got it there. +""" + +import threading + + +class ProbeLock: + """A re-entrant lock whose ``contended`` event is set when a caller blocks.""" + + def __init__(self): + self._lock = threading.RLock() + self.contended = threading.Event() + + def acquire(self, blocking=True, timeout=-1): + if self._lock.acquire(blocking=False): + return True + self.contended.set() + if not blocking: + return False + return self._lock.acquire(timeout=timeout) + + def release(self): + self._lock.release() + + def __enter__(self): + self.acquire() + return self + + def __exit__(self, *exc): + self.release() + return False diff --git a/tests/test_api_lora.py b/tests/test_api_lora.py index 7279d88..743f1e3 100644 --- a/tests/test_api_lora.py +++ b/tests/test_api_lora.py @@ -36,15 +36,18 @@ from aiohttp.test_utils import AioHTTPTestCase from database.operations import PromptDatabase from py.api import PromptManagerAPI -from py import lora_utils +from py import config_store, lora_utils from py.config import IntegrationConfig, PromptManagerConfig from py.api import admin as admin_module +from py.api import lora_integration as lora_module from py.lora_utils import get_trigger_cache from utils.hashing import generate_prompt_hash try: + from tests.lock_probe import ProbeLock from tests.open_handles import assert_closed except ImportError: # discovered with tests/ as the top-level directory + from lock_probe import ProbeLock from open_handles import assert_closed SECRET_KEY = "sk-civitai-super-secret-0123456789" @@ -624,8 +627,27 @@ class TestConfigWritersShareOneStore(LoraAPITestCase): return gated + lora_calls = [] + real_lora_mutation = lora_module._lora_settings_mutation + + def recorded_lora_mutation(values): + mutate = real_lora_mutation(values) + + def recorded(config): + lora_calls.append(dict(config)) + return mutate(config) + + return recorded + + probe = ProbeLock() loop = asyncio.get_running_loop() - with patch.object(admin_module, "_settings_mutation", gated_mutation): + with ( + patch.object(admin_module, "_settings_mutation", gated_mutation), + patch.object( + lora_module, "_lora_settings_mutation", recorded_lora_mutation + ), + patch.object(config_store, "_COMMIT_LOCK", probe), + ): settings = asyncio.ensure_future( self._post_json("/prompt_manager/settings", {"result_timeout": 13}) ) @@ -636,9 +658,14 @@ class TestConfigWritersShareOneStore(LoraAPITestCase): {"enabled": False, "trigger_words_enabled": True}, ) ) - await asyncio.sleep(0) # let the LoRA request reach the lock + # Wait until the LoRA commit is blocked on the lock the settings + # commit holds; its mutation must not have run yet. + self.assertTrue(await loop.run_in_executor(None, probe.contended.wait, 10)) + self.assertEqual(lora_calls, []) release.set() settings_resp, lora_resp = await asyncio.gather(settings, lora) + self.assertEqual(len(lora_calls), 1) + self.assertEqual(lora_calls[0]["web_ui"]["result_timeout"], 13) self.assertEqual((settings_resp.status, lora_resp.status), (200, 200)) saved = self._saved() @@ -651,6 +678,68 @@ class TestConfigWritersShareOneStore(LoraAPITestCase): lora_body = await lora_resp.json() self.assertNotEqual(settings_body["revision"], lora_body["revision"]) + def _delay_first_commit_result(self): + """Hold back the first config commit's result from its handler. + + The commit itself has run (the file is written); only the handler's + continuation waits until the returned event is set. Before #220's fix + round the handler published after this point, so a later save could + publish first and the earlier one overwrite it at runtime. + """ + original = self.api._run_in_executor + written = asyncio.Event() + resume = asyncio.Event() + delayed = [] + + async def run(func, *args, **kwargs): + result = await original(func, *args, **kwargs) + if getattr(func, "__func__", None) is config_store.ConfigStore.commit: + if not delayed: + delayed.append(func) + written.set() + await resume.wait() + return result + + patcher = patch.object(self.api, "_run_in_executor", run) + patcher.start() + self.addCleanup(patcher.stop) + return written, resume + + async def test_concurrent_settings_saves_end_on_the_last_written(self): + written, resume = self._delay_first_commit_result() + first = asyncio.ensure_future( + self._post_json("/prompt_manager/settings", {"result_timeout": 11}) + ) + await asyncio.wait_for(written.wait(), 10) + second = await self._post_json( + "/prompt_manager/settings", {"result_timeout": 22} + ) + resume.set() + first = await first + self.assertEqual((first.status, second.status), (200, 200)) + self.assertEqual(self._saved()["web_ui"]["result_timeout"], 22) + self.assertEqual(PromptManagerConfig.RESULT_TIMEOUT, 22) + + async def test_concurrent_lora_saves_end_on_the_last_written(self): + written, resume = self._delay_first_commit_result() + first = asyncio.ensure_future( + self._post_json( + "/prompt_manager/lora/enable", + {"enabled": False, "trigger_words_enabled": True}, + ) + ) + await asyncio.wait_for(written.wait(), 10) + second = await self._post_json( + "/prompt_manager/lora/enable", + {"enabled": False, "trigger_words_enabled": False}, + ) + resume.set() + first = await first + self.assertEqual((first.status, second.status), (200, 200)) + lora = self._saved()["integrations"]["lora_manager"] + self.assertIs(lora["trigger_words_enabled"], False) + self.assertIs(IntegrationConfig.LORA_TRIGGER_WORDS_ENABLED, False) + class TestLoraScanRoute(LoraAPITestCase): diff --git a/tests/test_config_store.py b/tests/test_config_store.py index a9833cf..5099dbb 100644 --- a/tests/test_config_store.py +++ b/tests/test_config_store.py @@ -32,8 +32,10 @@ from py.config_store import ( # noqa: E402 ) try: + from tests.lock_probe import ProbeLock # noqa: E402 from tests.open_handles import assert_closed # noqa: E402 except ImportError: # discovered with tests/ as the top-level directory + from lock_probe import ProbeLock # noqa: E402 from open_handles import assert_closed # noqa: E402 @@ -210,42 +212,187 @@ class ConfigStoreTests(unittest.TestCase): self.store.commit(lambda c: c, expected_revision=None) self.assertEqual(stat.S_IMODE(os.stat(self.path).st_mode), 0o600) - def test_concurrent_commits_serialize(self): - self._write({"keep": True}) + def _probe_lock(self): + probe = ProbeLock() + patcher = patch.object(config_store, "_COMMIT_LOCK", probe) + patcher.start() + self.addCleanup(patcher.stop) + return probe + + def _hold_lock_in_a_commit(self, value="first"): + """Start a commit that holds the lock until the returned event is set.""" inside = threading.Event() release = threading.Event() errors = [] def slow(config): inside.set() - release.wait(10) # hold the lock while the second commit waits - config["first"] = 1 + release.wait(10) + config[value] = 1 return config + def run(): + try: + self.store.commit(slow, expected_revision=None) + except Exception as exc: # pragma: no cover - reported by the test + errors.append(exc) + + thread = threading.Thread(target=run) + thread.start() + self.assertTrue(inside.wait(10)) + return thread, release, errors + + def test_concurrent_commits_serialize(self): + # The second commit is observed blocked on the lock before the first + # one may finish, so this fails whenever commit stops taking the lock. + self._write({"keep": True}) + probe = self._probe_lock() + first, release, errors = self._hold_lock_in_a_commit() + calls = [] + def fast(config): + calls.append(dict(config)) config["second"] = 2 return config - def run(mutate): - try: - self.store.commit(mutate, expected_revision=None) - except Exception as exc: # pragma: no cover - reported below - errors.append(exc) - - first = threading.Thread(target=run, args=(slow,)) - first.start() - self.assertTrue(inside.wait(10)) - second = threading.Thread(target=run, args=(fast,)) + second = threading.Thread( + target=lambda: self.store.commit(fast, expected_revision=None) + ) second.start() + self.assertTrue(probe.contended.wait(10), "second commit never waited") + self.assertEqual(calls, []) # not called while the first holds the lock release.set() first.join(10) second.join(10) self.assertEqual(errors, []) + self.assertEqual(calls, [{"keep": True, "first": 1}]) # re-read inside self.assertEqual( json.loads(self._bytes()), {"keep": True, "first": 1, "second": 2} ) + def test_read_waits_for_a_running_commit(self): + # Windows cannot replace a file another handle has open, so a read + # runs under the commit lock and never overlaps a replace. + self._write({"keep": True}) + probe = self._probe_lock() + first, release, errors = self._hold_lock_in_a_commit() + seen = [] + reader = threading.Thread(target=lambda: seen.append(self.store.read())) + reader.start() + self.assertTrue(probe.contended.wait(10), "read did not take the lock") + self.assertEqual(seen, []) + release.set() + first.join(10) + reader.join(10) + self.assertEqual(errors, []) + self.assertEqual(seen[0][0], {"keep": True, "first": 1}) + self.assertEqual(seen[0][1], revision_of(self._bytes())) + + def test_replace_is_retried_after_a_transient_permission_error(self): + # Windows: an antivirus or indexer handle makes os.replace fail with + # PermissionError for a moment; the commit retries before failing. + self._write({"a": 1}) + real_replace = os.replace + attempts = [] + + def flaky(src, dst): + attempts.append(dst) + if len(attempts) == 1: + raise PermissionError("in use by another process") + return real_replace(src, dst) + + with ( + patch.object(config_store, "_REPLACE_ATTEMPTS", 5), + patch.object(config_store.os, "replace", side_effect=flaky), + patch.object(config_store.time, "sleep") as sleep, + ): + saved, _ = self.store.commit( + lambda c: {**c, "b": 2}, expected_revision=None + ) + self.assertEqual(saved, {"a": 1, "b": 2}) + self.assertEqual(json.loads(self._bytes()), {"a": 1, "b": 2}) + self.assertEqual(len(attempts), 2) + sleep.assert_called_once_with(config_store._REPLACE_BACKOFF_SECONDS) + self.assertEqual(self._leftovers(), []) + + def test_replace_gives_up_after_the_last_attempt(self): + self._write({"a": 1}) + before = self._bytes() + with ( + patch.object(config_store, "_REPLACE_ATTEMPTS", 3), + patch.object( + config_store.os, "replace", side_effect=PermissionError("locked") + ) as replace, + patch.object(config_store.time, "sleep") as sleep, + ): + with self.assertRaises(PermissionError): + self.store.commit(lambda c: {**c, "b": 2}, expected_revision=None) + self.assertEqual((replace.call_count, sleep.call_count), (3, 2)) + self.assertEqual(self._bytes(), before) + self.assertEqual(self._leftovers(), []) + + def test_replace_is_tried_once_outside_windows(self): + expected = 5 if os.name == "nt" else 1 + self.assertEqual(config_store._REPLACE_ATTEMPTS, expected) + + @unittest.skipUnless(hasattr(os, "symlink"), "needs symlinks") + def test_symlinked_config_is_written_through_the_link(self): + # A config.json that is a symlink (or a single-file bind mount) is + # written beside its real target, so the link stays a link. + real_dir = os.path.join(self.tmpdir, "real") + os.mkdir(real_dir) + target = os.path.join(real_dir, "settings.json") + Path(target).write_text(json.dumps({"a": 1}), encoding="utf-8") + try: + os.symlink(target, self.path) + except (OSError, NotImplementedError): # pragma: no cover - Windows + self.skipTest("symlinks are not permitted here") + + saved, revision = self.store.commit( + lambda c: {**c, "b": 2}, expected_revision=None + ) + + self.assertTrue(os.path.islink(self.path)) + self.assertEqual(json.loads(Path(target).read_text()), {"a": 1, "b": 2}) + self.assertEqual(self.store.read(), (saved, revision)) + self.assertEqual(os.listdir(real_dir), ["settings.json"]) + + def test_publish_runs_under_the_lock_after_the_write(self): + # Runtimes publish inside the commit, so two saves publish in the + # order they were written and the runtime ends on the last one. + self._write({"a": 1}) + seen = [] + + # The lock is re-entrant, so another thread probes whether it is held. + def probe_publish(config): + result = [] + thread = threading.Thread( + target=lambda: result.append( + config_store._COMMIT_LOCK.acquire(blocking=False) + ) + ) + thread.start() + thread.join(10) + seen.append((config, json.loads(self._bytes()), result[0])) + + saved, _ = self.store.commit( + lambda c: {**c, "b": 2}, expected_revision=None, publish=probe_publish + ) + self.assertEqual(seen, [(saved, {"a": 1, "b": 2}, False)]) + + def test_publish_is_skipped_when_the_write_fails(self): + self._write({"a": 1}) + published = [] + with patch.object(config_store.os, "fsync", side_effect=OSError("disk full")): + with self.assertRaises(OSError): + self.store.commit( + lambda c: {**c, "b": 2}, + expected_revision=None, + publish=published.append, + ) + self.assertEqual(published, []) + def test_handles_closed_after_commit(self): self._write({"a": 1}) self.store.read()