From cdc76c25c608a5fcd159d1a885d882aafe52dd5f Mon Sep 17 00:00:00 2001 From: CalamitousFelicitousness Date: Sun, 13 Sep 2026 18:41:33 +0100 Subject: [PATCH] fix(json_helpers): make atomic saves work on windows, keep lock file writefile(atomic=True) renamed the temp file while it was still open, which Windows refuses, so every atomic save failed there and left a temp file behind. The temp file is now created with mkstemp, closed before os.replace and removed when the save fails; the replace retries briefly when another handle holds the target, which Windows reports as a permission error. Both helpers removed the .lock file after releasing it. That removal raced other holders on both platforms, and any failure switched locking off for the whole process without a log line. The lock file now stays in place; a failed release or a lock timeout is logged instead. writefile deep-copied and validated the live object in Python, so a concurrent insert into a shared dict raised "dictionary changed size during iteration" and dropped the save. dict.copy and list.copy run under the GIL, so the deep copy and the per-key validation now walk that snapshot; nested containers remain the caller's responsibility. test/test-json-helpers.py covers all three on Linux and Windows. --- modules/json_helpers.py | 54 +++++++---- test/test-json-helpers.py | 199 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 236 insertions(+), 17 deletions(-) create mode 100644 test/test-json-helpers.py diff --git a/modules/json_helpers.py b/modules/json_helpers.py index 4e6c5551a..5d9a6d707 100644 --- a/modules/json_helpers.py +++ b/modules/json_helpers.py @@ -1,5 +1,6 @@ import os import sys +import contextlib import time import json from typing import overload, Literal @@ -10,6 +11,18 @@ from modules.logger import log locking_available = True # used by file read/write locking +def replace_file(source: str, target: str, attempts: int = 10, delay: float = 0.05): + """os.replace that waits out a target another handle holds open, which Windows reports as a permission error.""" + for attempt in range(attempts): + try: + os.replace(source, target) + return + except PermissionError: + if attempt == attempts - 1: + raise + time.sleep(delay) + + @overload def readfile(filename: str | os.PathLike[str], silent: bool = False, lock: bool = False, *, as_type: Literal["dict"]) -> dict: ... @overload @@ -27,6 +40,8 @@ def readfile(filename: str | os.PathLike[str], silent: bool = False, lock: bool lock_file = fasteners.InterProcessReaderWriterLock(f"{filename}.lock") lock_file.logger.disabled = True # type: ignore - False positive. Bad typing in Fasteners. locked = lock_file.acquire_read_lock(blocking=True, timeout=3) + if not locked: + log.warning(f'File read lock: file="{filename}" timeout') except Exception as err: lock_file = None locking_available = False @@ -57,11 +72,9 @@ def readfile(filename: str | os.PathLike[str], silent: bool = False, lock: bool try: if locking_available and lock_file is not None: - lock_file.release_read_lock() - if locked and os.path.exists(f"{filename}.lock"): - os.remove(f"{filename}.lock") - except Exception: - locking_available = False + lock_file.release_read_lock() # the lock file stays: removing it after release races other holders + except Exception as err: + log.error(f'File read lock release: file="{filename}" {err}') if isinstance(data, list) and as_type == "dict": if not data: @@ -93,8 +106,9 @@ def writefile(obj: dict | list, filename: str | os.PathLike[str], mode="w", sile try: t0 = time.time() - data = copy.deepcopy(obj) # ensure keys/items aren't added/deleted during json.dumps - for k, v in obj.items() if isinstance(obj, dict) else []: # validate each key-by-key to avoid global exceptions + snapshot = obj.copy() if isinstance(obj, (dict, list)) else obj # dict.copy and list.copy run under the GIL, so a concurrent insert cannot interrupt them + data = copy.deepcopy(snapshot) + for k, v in list(data.items()) if isinstance(data, dict) else []: # validate each key-by-key to avoid global exceptions try: _tmp = json.dumps(v, indent=2, default=default, allow_nan=False, ensure_ascii=False) except Exception as err: @@ -111,6 +125,8 @@ def writefile(obj: dict | list, filename: str | os.PathLike[str], mode="w", sile lock_file = fasteners.InterProcessReaderWriterLock(f"{filename}.lock") if locking_available else None lock_file.logger.disabled = True # type: ignore - False positive. Bad typing in Fasteners. locked = lock_file.acquire_write_lock(blocking=True, timeout=3) if lock_file is not None else False + if not locked: + log.warning(f'File write lock: file="{filename}" timeout') except Exception as err: locking_available = False lock_file = None @@ -119,11 +135,17 @@ def writefile(obj: dict | list, filename: str | os.PathLike[str], mode="w", sile try: if atomic: - with tempfile.NamedTemporaryFile(mode=mode, encoding="utf8", delete=False, dir=os.path.dirname(filename)) as f: - f.write(output) - f.flush() - os.fsync(f.fileno()) - os.replace(f.name, filename) + fd, temp_name = tempfile.mkstemp(dir=os.path.dirname(os.path.abspath(filename)), prefix=f"{os.path.basename(filename)}.", suffix=".tmp") + try: + with os.fdopen(fd, mode, encoding="utf8") as f: + f.write(output) + f.flush() + os.fsync(f.fileno()) + replace_file(temp_name, filename) + except BaseException: + with contextlib.suppress(OSError): + os.remove(temp_name) + raise else: with open(filename, mode=mode, encoding="utf8") as file: file.write(output) @@ -136,8 +158,6 @@ def writefile(obj: dict | list, filename: str | os.PathLike[str], mode="w", sile try: if locking_available and lock_file is not None: - lock_file.release_write_lock() - if locked and os.path.exists(f"{filename}.lock"): - os.remove(f"{filename}.lock") - except Exception: - locking_available = False + lock_file.release_write_lock() # the lock file stays: removing it after release races other holders + except Exception as err: + log.error(f'File write lock release: file="{filename}" {err}') diff --git a/test/test-json-helpers.py b/test/test-json-helpers.py new file mode 100644 index 000000000..95e91ce31 --- /dev/null +++ b/test/test-json-helpers.py @@ -0,0 +1,199 @@ +#!/usr/bin/env python +""" +Offline tests for modules.json_helpers on Linux and Windows. + +Covers: + +- atomic saves replace a fresh or existing target and leave no temp file, including when the save fails +- an atomic save waits out a target another thread briefly holds open, which Windows otherwise refuses +- locking stays available under overlapping locked reads and writes and the lock file is left in place +- concurrent inserts into a shared dict never drop a save +- readfile returns what writefile wrote, as dict and as list + +No running server required. + +Usage: + python test/test-json-helpers.py +""" + +import importlib +import json +import os +import shutil +import sys +import tempfile +import threading + +script_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +sys.path.insert(0, script_dir) +os.chdir(script_dir) +sys.argv = [sys.argv[0]] + +from modules.logger import log # pylint: disable=wrong-import-position +from modules import json_helpers # pylint: disable=wrong-import-position + + +class ErrorLog: + """Counts error lines from json_helpers while passing everything else to the real logger.""" + + def __init__(self, inner): + self.inner = inner + self.errors = [] + self.lock = threading.Lock() + + def error(self, message, *args, **kwargs): + with self.lock: + self.errors.append(str(message)) + + def count(self, needle): + return sum(1 for m in self.errors if needle in m) + + def __getattr__(self, name): + return getattr(self.inner, name) + + +def fresh_helper(): + importlib.reload(json_helpers) + json_helpers.log = ErrorLog(log) + return json_helpers + + +def run_threads(worker, count): + pool = [threading.Thread(target=worker, args=(n,)) for n in range(count)] + for t in pool: + t.start() + for t in pool: + t.join() + + +def leftovers(folder, target): + keep = {os.path.basename(target), os.path.basename(target) + '.lock'} + return sorted(set(os.listdir(folder)) - keep) + + +def test_atomic_save_replaces_target(folder): + jh = fresh_helper() + target = os.path.join(folder, 'atomic.json') + jh.writefile({'first': 1}, target, silent=True, atomic=True) + assert jh.readfile(target, silent=True, as_type='dict') == {'first': 1}, 'fresh target not written' + jh.writefile({'second': 2}, target, silent=True, atomic=True) + assert jh.readfile(target, silent=True, as_type='dict') == {'second': 2}, 'existing target not replaced' + assert leftovers(folder, target) == [], f'temp files left: {leftovers(folder, target)}' + assert jh.log.errors == [], jh.log.errors + + +def test_failed_atomic_save_leaves_no_temp(folder): + jh = fresh_helper() + target = os.path.join(folder, 'blocked.json') + os.mkdir(target) # a directory cannot be replaced by a file + jh.writefile({'x': 1}, target, silent=True, atomic=True) + assert jh.log.count('Save failed') == 1, jh.log.errors + assert os.path.isdir(target), 'target directory was replaced' + assert leftovers(folder, target) == [], f'temp files left: {leftovers(folder, target)}' + + +def test_atomic_save_waits_for_open_target(folder): + jh = fresh_helper() + target = os.path.join(folder, 'open.json') + jh.writefile({'v': 0}, target, silent=True) + opened = threading.Event() + release = threading.Event() + + def holder(_n): + with open(target, 'rb') as f: + f.read(1) + opened.set() + release.wait(5) + + thread = threading.Thread(target=holder, args=(0,)) + thread.start() + opened.wait(5) + threading.Timer(0.2, release.set).start() + jh.writefile({'v': 1}, target, silent=True, atomic=True) + thread.join() + assert jh.readfile(target, silent=True, as_type='dict') == {'v': 1}, 'save did not wait for the open target' + assert jh.log.errors == [], jh.log.errors + assert leftovers(folder, target) == [], f'temp files left: {leftovers(folder, target)}' + + +def test_locking_survives_overlap(folder, threads=8, iterations=25): + jh = fresh_helper() + target = os.path.join(folder, 'locked.json') + jh.writefile({'seed': 0}, target, silent=True) + + def worker(n): + for i in range(iterations): + if (n + i) % 2: + jh.writefile({'n': n, 'i': i}, target, silent=True, atomic=True) + else: + jh.readfile(target, silent=True, lock=True, as_type='dict') + + run_threads(worker, threads) + assert jh.locking_available, 'locking switched itself off' + assert jh.log.errors == [], jh.log.errors + assert os.path.exists(target + '.lock'), 'lock file was removed' + assert leftovers(folder, target) == [], f'temp files left: {leftovers(folder, target)}' + + +def test_concurrent_inserts_keep_every_save(folder, threads=4, per_thread=40): + jh = fresh_helper() + target = os.path.join(folder, 'cache.json') + cache = {} + payload = {'metadata': {f'k{i}': 'x' * 64 for i in range(200)}, 'tensors': list(range(500))} + + def worker(n): + for i in range(per_thread): + cache[f'{n}-{i}'] = payload + jh.writefile(cache, target, silent=True, atomic=True) + + run_threads(worker, threads) + assert jh.log.count('changed size') == 0, f'{jh.log.count("changed size")} saves dropped' + assert jh.log.errors == [], jh.log.errors + with open(target, encoding='utf8') as f: + assert len(json.load(f)) == threads * per_thread, 'last save is incomplete' + + +def test_roundtrip(folder): + jh = fresh_helper() + target = os.path.join(folder, 'roundtrip.json') + data = {'a': 1, 'b': [1, 2, 3], 'c': {'d': 'e'}, 'f': None} + jh.writefile(data, target, silent=True) + assert jh.readfile(target, silent=True, as_type='dict') == data + jh.writefile([1, 'two', {'three': 3}], target, silent=True, atomic=True) + assert jh.readfile(target, silent=True, as_type='list') == [1, 'two', {'three': 3}] + assert jh.readfile(os.path.join(folder, 'missing.json'), silent=True, as_type='dict') == {} + assert jh.log.errors == [], jh.log.errors + + +def run_all(): + tests = [ + test_atomic_save_replaces_target, + test_failed_atomic_save_leaves_no_temp, + test_atomic_save_waits_for_open_target, + test_locking_survives_overlap, + test_concurrent_inserts_keep_every_save, + test_roundtrip, + ] + passed = 0 + failed = 0 + for fn in tests: + folder = tempfile.mkdtemp(prefix='sdnext-json-') + try: + fn(folder) + log.info(f' PASS {fn.__name__}') + passed += 1 + except Exception as e: + log.error(f' FAIL {fn.__name__}: {type(e).__name__}: {e}') + failed += 1 + finally: + shutil.rmtree(folder, ignore_errors=True) + log.warning(f'Total: {passed} passed, {failed} failed') + return failed == 0 + + +if __name__ == '__main__': + import time + t0 = time.time() + ok = run_all() + log.warning(f'Total time: {time.time() - t0:.2f}s') + sys.exit(0 if ok else 1)