From 8cac4370bf151f9f8d03e0f76534808b7f000785 Mon Sep 17 00:00:00 2001 From: Donald Date: Sat, 3 Oct 2026 03:10:06 +0000 Subject: [PATCH] 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 --- herdr_hub/aliases.py | 83 +++++++----------------------------- herdr_hub/jsonstore.py | 96 ++++++++++++++++++++++++++++++++++++++++++ herdr_hub/remotes.py | 90 +++++++++------------------------------ 3 files changed, 132 insertions(+), 137 deletions(-) create mode 100644 herdr_hub/jsonstore.py diff --git a/herdr_hub/aliases.py b/herdr_hub/aliases.py index b9a4e21..5e5c13f 100644 --- a/herdr_hub/aliases.py +++ b/herdr_hub/aliases.py @@ -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 ( diff --git a/herdr_hub/jsonstore.py b/herdr_hub/jsonstore.py new file mode 100644 index 0000000..fd97dfe --- /dev/null +++ b/herdr_hub/jsonstore.py @@ -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 diff --git a/herdr_hub/remotes.py b/herdr_hub/remotes.py index 66dc174..586e0f2 100644 --- a/herdr_hub/remotes.py +++ b/herdr_hub/remotes.py @@ -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: