From a493111820039deeb26bbe4356aa02fd317e1e4d Mon Sep 17 00:00:00 2001 From: Abdessamad Derraz <3028866+Abdess@users.noreply.github.com> Date: Sat, 10 Oct 2026 04:48:30 +0200 Subject: [PATCH] fix: update the native source index under a lock --- scripts/artifacts.py | 38 ++++++++++++++++++++++++++++++++++ scripts/export_native.py | 22 +++++++++++++------- scripts/refresh_data_dirs.py | 23 +------------------- tests/test_atomic_artifacts.py | 36 ++++++++++++++++++++++++++++++++ 4 files changed, 90 insertions(+), 29 deletions(-) diff --git a/scripts/artifacts.py b/scripts/artifacts.py index 08ff5baa..8bf26f18 100644 --- a/scripts/artifacts.py +++ b/scripts/artifacts.py @@ -69,6 +69,44 @@ def write_text_atomic(path: str, content: str) -> None: raise +def write_bytes_atomic(path: str, content: bytes) -> None: + """Write a whole binary file or nothing, as write_text_atomic does.""" + directory = os.path.dirname(os.path.abspath(path)) + handle, scratch = tempfile.mkstemp( + dir=directory, prefix=f".{os.path.basename(path)}.", suffix=".tmp" + ) + try: + with os.fdopen(handle, "wb") as f: + f.write(content) + os.replace(scratch, path) + except BaseException: + with contextlib.suppress(OSError): + os.unlink(scratch) + raise + + +@contextlib.contextmanager +def file_lock(lock_path: str | os.PathLike, shared: bool = False): + """Hold a lock on lock_path, waiting for it if taken. + + Several sessions refresh the same caches: without it, two swaps of one + tree interleave, or two read-modify-writes of one index lose an entry. + On platforms without flock the lock is a no-op. + """ + try: + import fcntl + except ImportError: + yield + return + os.makedirs(os.path.dirname(os.path.abspath(lock_path)), exist_ok=True) + with open(lock_path, "a") as handle: + fcntl.flock(handle, fcntl.LOCK_SH if shared else fcntl.LOCK_EX) + try: + yield + finally: + fcntl.flock(handle, fcntl.LOCK_UN) + + def copy_file_atomic(source: str, path: str) -> None: """Copy a file into place whole or not at all, metadata included.""" import shutil diff --git a/scripts/export_native.py b/scripts/export_native.py index 97a3d8ff..770043d5 100644 --- a/scripts/export_native.py +++ b/scripts/export_native.py @@ -24,6 +24,7 @@ from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent)) +from artifacts import file_lock, write_bytes_atomic, write_text_atomic from common import list_registered_platforms, load_platform_config, yaml_load from exporter import discover_exporters from exporter.baseline import build_native_model @@ -80,9 +81,12 @@ def fetch( it is what a rescrape calls, so the original and its transcription describe the same moment. """ - recorded = _load_sources(index) key = source_key(destination, index) - if not refresh and destination.exists() and recorded.get(key, url) == url: + if ( + not refresh + and destination.exists() + and _load_sources(index).get(key, url) == url + ): return destination.read_bytes() request = urllib.request.Request(url, headers={"User-Agent": _USER_AGENT}) with urllib.request.urlopen(request, timeout=60) as response: @@ -90,13 +94,17 @@ def fetch( if len(payload) > _MAX_BYTES: raise ValueError(f"{url}: response larger than {_MAX_BYTES} bytes") destination.parent.mkdir(parents=True, exist_ok=True) - destination.write_bytes(payload) + write_bytes_atomic(str(destination), payload) if index is not None: - recorded[key] = url + # Platforms refresh in parallel: the index is read again under the + # lock, or the last writer erased what the others had recorded. index.parent.mkdir(parents=True, exist_ok=True) - index.write_text( - json.dumps(recorded, indent=2, sort_keys=True) + "\n", encoding="utf-8" - ) + with file_lock(index.with_name(f".{index.name}.lock")): + recorded = _load_sources(index) + recorded[key] = url + write_text_atomic( + str(index), json.dumps(recorded, indent=2, sort_keys=True) + "\n" + ) return payload diff --git a/scripts/refresh_data_dirs.py b/scripts/refresh_data_dirs.py index 50d2c783..a34b6d35 100644 --- a/scripts/refresh_data_dirs.py +++ b/scripts/refresh_data_dirs.py @@ -24,6 +24,7 @@ import urllib.error import urllib.request import zipfile from pathlib import Path +from artifacts import file_lock as _file_lock from common import yaml_load try: @@ -51,28 +52,6 @@ def load_registry(registry_path: str = DEFAULT_REGISTRY) -> dict[str, dict]: return data.get("data_directories", {}) -@contextlib.contextmanager -def _file_lock(lock_path: Path, shared: bool = False): - """Hold a lock on lock_path, waiting for it if taken. - - Several sessions refresh the same data directories: without it, two - swaps of one tree interleave and the second lands inside the first. - On platforms without flock the lock is a no-op. - """ - try: - import fcntl - except ImportError: - yield - return - lock_path.parent.mkdir(parents=True, exist_ok=True) - with open(lock_path, "a") as handle: - fcntl.flock(handle, fcntl.LOCK_SH if shared else fcntl.LOCK_EX) - try: - yield - finally: - fcntl.flock(handle, fcntl.LOCK_UN) - - def cache_lock(cache_dir: str | Path, shared: bool = False): """The lock a refresh holds while it swaps cache_dir. diff --git a/tests/test_atomic_artifacts.py b/tests/test_atomic_artifacts.py index 5e7748a7..497e6650 100644 --- a/tests/test_atomic_artifacts.py +++ b/tests/test_atomic_artifacts.py @@ -93,5 +93,41 @@ class EveryWriterGoesThroughIt(unittest.TestCase): target.read_bytes().decode("utf-8"), "name: \u00e9mulateur \u2014 \u30d5\n" ) + +class TheNativeIndexKeepsEveryPlatform(unittest.TestCase): + """Native originals refresh platform by platform, in parallel. Each run + rewrote the whole URL index it had read before downloading, and the last + writer erased the others' entries.""" + + def test_an_entry_recorded_during_a_download_survives(self): + import io + import json + + import export_native + + with tempfile.TemporaryDirectory(dir=REPO_ROOT / "tmp") as directory: + root = Path(directory) + index = root / export_native.SOURCES_INDEX + + class Response(io.BytesIO): + def __enter__(self): + return self + + def __exit__(self, *exc): + return False + + def other_run_records_meanwhile(request, timeout): + index.write_text(json.dumps({"retrobat/file.json": "https://b"})) + return Response(b"payload") + + with mock.patch.object( + export_native.urllib.request, "urlopen", other_run_records_meanwhile + ): + export_native.fetch("https://a", root / "batocera" / "file", index) + recorded = json.loads(index.read_text()) + self.assertEqual(recorded.get("retrobat/file.json"), "https://b") + self.assertEqual(recorded.get("batocera/file"), "https://a") + self.assertEqual((root / "batocera" / "file").read_bytes(), b"payload") + if __name__ == "__main__": unittest.main()