Share the locked atomic JSON store between registries

AliasRegistry and RemoteRegistry duplicated the lock-file, read-modify- write and atomic-replace logic almost line for line. Move it into a generic JsonRegistry base; each registry now only declares its error type, empty value, parser and serializer. Behaviour, file format, permissions and the corrupt-file guarantee are unchanged. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
2026-10-03 03:10:06 +00:00
parent 1a4cb1f2ca
commit 8cac4370bf
3 changed files with 132 additions and 137 deletions
@@ -1,16 +1,13 @@
 from __future__ import annotations

-import fcntl
-import json
 import os
 from pathlib import Path
-import tempfile
-from typing import Callable, TypeVar
+
+from .jsonstore import JsonRegistry


 ALIAS_REGISTRY_VERSION = 1
 MAX_ALIAS_LENGTH = 48
-UpdateResult = TypeVar("UpdateResult")


 class AliasRegistryError(RuntimeError):
@@ -34,23 +31,24 @@ def default_alias_registry_path() -> Path:
    return root / "herdr-hub" / "aliases.json"


-class AliasRegistry:
+class AliasRegistry(JsonRegistry[dict[str, str]]):
+    error_type = AliasRegistryError
+
    def __init__(self, path: Path | None = None) -> None:
-        self.path = path or default_alias_registry_path()
+        super().__init__(path or default_alias_registry_path())

-    @property
-    def lock_path(self) -> Path:
-        return self.path.with_name(f".{self.path.name}.lock")
+    def _empty(self) -> dict[str, str]:
+        return {}

-    def load(self) -> dict[str, str]:
-        if not self.path.exists():
-            return {}
-        try:
-            payload = json.loads(self.path.read_text(encoding="utf-8"))
-        except (OSError, json.JSONDecodeError) as error:
-            raise AliasRegistryError(f"cannot read {self.path}: {error}") from error
+    def _parse(self, payload: object) -> dict[str, str]:
        return _parse_aliases(payload, self.path)

+    def _serialize(self, aliases: dict[str, str]) -> object:
+        return {
+            "version": ALIAS_REGISTRY_VERSION,
+            "aliases": dict(sorted(aliases.items())),
+        }
+
    def set(self, endpoint_key: str, label: str) -> bool:
        _validate_endpoint_key(endpoint_key)
        normalized = normalize_alias(label)
@@ -67,57 +65,6 @@ class AliasRegistry:

        return self._update(update)

-    def _update(
-        self,
-        callback: Callable[[dict[str, str]], tuple[dict[str, str], UpdateResult]],
-    ) -> UpdateResult:
-        try:
-            self.path.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
-            with self.lock_path.open("a+", encoding="utf-8") as lock:
-                fcntl.flock(lock.fileno(), fcntl.LOCK_EX)
-                aliases = self.load()
-                updated, outcome = callback(aliases)
-                if updated != aliases:
-                    self._write(updated)
-                return outcome
-        except AliasRegistryError:
-            raise
-        except OSError as error:
-            raise AliasRegistryError(f"cannot update {self.path}: {error}") from error
-
-    def _write(self, aliases: dict[str, str]) -> None:
-        payload = {
-            "version": ALIAS_REGISTRY_VERSION,
-            "aliases": dict(sorted(aliases.items())),
-        }
-        temporary_name: str | None = None
-        try:
-            with tempfile.NamedTemporaryFile(
-                mode="w",
-                encoding="utf-8",
-                dir=self.path.parent,
-                prefix=f".{self.path.name}.",
-                delete=False,
-            ) as temporary:
-                temporary_name = temporary.name
-                json.dump(payload, temporary, ensure_ascii=False, indent=2)
-                temporary.write("\n")
-                temporary.flush()
-                os.fsync(temporary.fileno())
-            os.chmod(temporary_name, 0o600)
-            os.replace(temporary_name, self.path)
-            directory_fd = os.open(self.path.parent, os.O_RDONLY | os.O_DIRECTORY)
-            try:
-                os.fsync(directory_fd)
-            finally:
-                os.close(directory_fd)
-        finally:
-            if temporary_name is not None:
-                try:
-                    Path(temporary_name).unlink(missing_ok=True)
-                except OSError:
-                    pass
-

 def _parse_aliases(payload: object, path: Path) -> dict[str, str]:
    if (
@@ -0,0 +1,96 @@
+"""Locked, atomically replaced JSON files for the small per-user registries."""
+from __future__ import annotations
+
+import fcntl
+import json
+import os
+from pathlib import Path
+import tempfile
+from typing import Callable, Generic, TypeVar
+
+
+Value = TypeVar("Value")
+Outcome = TypeVar("Outcome")
+
+
+class JsonRegistry(Generic[Value]):
+    """A private JSON file updated under an exclusive lock.
+
+    Subclasses provide the user-facing error type and the mapping between the
+    stored payload and a Python value. Every update reads, modifies and writes
+    under the lock so concurrent Hub processes cannot lose each other's
+    changes, the file is replaced atomically with mode 0600, and a corrupt
+    file is never overwritten.
+    """
+
+    error_type: type[RuntimeError] = RuntimeError
+
+    def __init__(self, path: Path) -> None:
+        self.path = path
+
+    @property
+    def lock_path(self) -> Path:
+        return self.path.with_name(f".{self.path.name}.lock")
+
+    def _empty(self) -> Value:
+        raise NotImplementedError
+
+    def _parse(self, payload: object) -> Value:
+        raise NotImplementedError
+
+    def _serialize(self, value: Value) -> object:
+        raise NotImplementedError
+
+    def load(self) -> Value:
+        if not self.path.exists():
+            return self._empty()
+        try:
+            payload = json.loads(self.path.read_text(encoding="utf-8"))
+        except (OSError, json.JSONDecodeError) as error:
+            raise self.error_type(f"cannot read {self.path}: {error}") from error
+        return self._parse(payload)
+
+    def _update(self, callback: Callable[[Value], tuple[Value, Outcome]]) -> Outcome:
+        try:
+            self.path.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
+            with self.lock_path.open("a+", encoding="utf-8") as lock:
+                fcntl.flock(lock.fileno(), fcntl.LOCK_EX)
+                current = self.load()
+                updated, outcome = callback(current)
+                if updated != current:
+                    self._write(updated)
+                return outcome
+        except self.error_type:
+            raise
+        except OSError as error:
+            raise self.error_type(f"cannot update {self.path}: {error}") from error
+
+    def _write(self, value: Value) -> None:
+        payload = self._serialize(value)
+        temporary_name: str | None = None
+        try:
+            with tempfile.NamedTemporaryFile(
+                mode="w",
+                encoding="utf-8",
+                dir=self.path.parent,
+                prefix=f".{self.path.name}.",
+                delete=False,
+            ) as temporary:
+                temporary_name = temporary.name
+                json.dump(payload, temporary, ensure_ascii=False, indent=2)
+                temporary.write("\n")
+                temporary.flush()
+                os.fsync(temporary.fileno())
+            os.chmod(temporary_name, 0o600)
+            os.replace(temporary_name, self.path)
+            directory_fd = os.open(self.path.parent, os.O_RDONLY | os.O_DIRECTORY)
+            try:
+                os.fsync(directory_fd)
+            finally:
+                os.close(directory_fd)
+        finally:
+            if temporary_name is not None:
+                try:
+                    Path(temporary_name).unlink(missing_ok=True)
+                except OSError:
+                    pass
@@ -4,19 +4,18 @@ import base64
 import codecs
 from collections import deque
 from dataclasses import dataclass
-import fcntl
 import json
 import os
 from pathlib import Path
 import re
 import shlex
 import subprocess
-import tempfile
 import time
 from typing import Callable, Iterable, Sequence
 from urllib.parse import SplitResult, unquote, urlsplit

 from .domain import is_valid_session_name
+from .jsonstore import JsonRegistry


 REGISTRY_VERSION = 1
@@ -143,23 +142,31 @@ def default_registry_path() -> Path:
    return root / "herdr-hub" / "remotes.json"


-class RemoteRegistry:
+class RemoteRegistry(JsonRegistry[list[RemoteEntry]]):
+    error_type = RegistryError
+
    def __init__(self, path: Path | None = None) -> None:
-        self.path = path or default_registry_path()
+        super().__init__(path or default_registry_path())

-    @property
-    def lock_path(self) -> Path:
-        return self.path.with_name(f".{self.path.name}.lock")
+    def _empty(self) -> list[RemoteEntry]:
+        return []

-    def load(self) -> list[RemoteEntry]:
-        if not self.path.exists():
-            return []
-        try:
-            payload = json.loads(self.path.read_text(encoding="utf-8"))
-        except (OSError, json.JSONDecodeError) as error:
-            raise RegistryError(f"cannot read {self.path}: {error}") from error
+    def _parse(self, payload: object) -> list[RemoteEntry]:
        return _parse_registry(payload, self.path)

+    def _serialize(self, entries: list[RemoteEntry]) -> object:
+        return {
+            "version": REGISTRY_VERSION,
+            "remotes": [
+                {
+                    "target": entry.target,
+                    "remote_keybindings": entry.remote_keybindings,
+                    "remote_transport": entry.remote_transport,
+                }
+                for entry in entries
+            ],
+        }
+
    def upsert(self, entry: RemoteEntry) -> bool:
        def update(entries: list[RemoteEntry]) -> tuple[list[RemoteEntry], bool]:
            by_target = {item.target: item for item in entries}
@@ -226,61 +233,6 @@ class RemoteRegistry:

        return self._update(update)

-    def _update(self, callback):  # type: ignore[no-untyped-def]
-        try:
-            self.path.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
-            with self.lock_path.open("a+", encoding="utf-8") as lock:
-                fcntl.flock(lock.fileno(), fcntl.LOCK_EX)
-                entries = self.load()
-                updated, outcome = callback(entries)
-                if updated != entries:
-                    self._write(updated)
-                return outcome
-        except RegistryError:
-            raise
-        except OSError as error:
-            raise RegistryError(f"cannot update {self.path}: {error}") from error
-
-    def _write(self, entries: Sequence[RemoteEntry]) -> None:
-        payload = {
-            "version": REGISTRY_VERSION,
-            "remotes": [
-                {
-                    "target": entry.target,
-                    "remote_keybindings": entry.remote_keybindings,
-                    "remote_transport": entry.remote_transport,
-                }
-                for entry in entries
-            ],
-        }
-        temporary_name: str | None = None
-        try:
-            with tempfile.NamedTemporaryFile(
-                mode="w",
-                encoding="utf-8",
-                dir=self.path.parent,
-                prefix=f".{self.path.name}.",
-                delete=False,
-            ) as temporary:
-                temporary_name = temporary.name
-                json.dump(payload, temporary, ensure_ascii=False, indent=2)
-                temporary.write("\n")
-                temporary.flush()
-                os.fsync(temporary.fileno())
-            os.chmod(temporary_name, 0o600)
-            os.replace(temporary_name, self.path)
-            directory_fd = os.open(self.path.parent, os.O_RDONLY | os.O_DIRECTORY)
-            try:
-                os.fsync(directory_fd)
-            finally:
-                os.close(directory_fd)
-        finally:
-            if temporary_name is not None:
-                try:
-                    Path(temporary_name).unlink(missing_ok=True)
-                except OSError:
-                    pass
-

 def _parse_registry(payload: object, path: Path) -> list[RemoteEntry]:
    if not isinstance(payload, dict) or payload.get("version") != REGISTRY_VERSION: