fix: update the native source index under a lock

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

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,12 +94,16 @@ 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
 

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.
 

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()