From 1c49abe8c2c9a975a7450665787847427864f244 Mon Sep 17 00:00:00 2001 From: Krzysztof Rudnicki Date: Sun, 5 Jul 2026 20:40:09 +0200 Subject: [PATCH 1/2] Initial implementation: unified CRDT merge scheme + GitHub sync transport Extracted and generalized from diet_guard's existing tombstone-merge scheme and GitHub Contents API client, extended with per-field HLC LWW so the same library can also cover todo's mutable notes and wake_alarm's single-value sync (Workstream D scoping decision, 2026-07-04). 100% branch coverage, clean ruff/mypy/pylint/bandit. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_017McuA1nAi6jBT1EaS5BrVf --- crdt-sync/.github/workflows/pre-commit.yml | 19 ++ crdt-sync/.github/workflows/python-tests.yml | 28 +++ crdt-sync/.gitignore | 12 ++ crdt-sync/.pre-commit-config.yaml | 125 ++++++++++++ crdt-sync/LICENSE | 21 ++ crdt-sync/README.md | 100 ++++++++++ crdt-sync/crdt_sync/__init__.py | 23 +++ crdt-sync/crdt_sync/_github.py | 194 +++++++++++++++++++ crdt-sync/crdt_sync/_hlc.py | 97 ++++++++++ crdt-sync/crdt_sync/_log.py | 32 +++ crdt-sync/crdt_sync/_record.py | 128 ++++++++++++ crdt-sync/crdt_sync/_sync.py | 106 ++++++++++ crdt-sync/crdt_sync/tests/__init__.py | 1 + crdt-sync/crdt_sync/tests/conftest.py | 22 +++ crdt-sync/crdt_sync/tests/test_github.py | 170 ++++++++++++++++ crdt-sync/crdt_sync/tests/test_hlc.py | 100 ++++++++++ crdt-sync/crdt_sync/tests/test_log.py | 80 ++++++++ crdt-sync/crdt_sync/tests/test_record.py | 111 +++++++++++ crdt-sync/crdt_sync/tests/test_sync.py | 160 +++++++++++++++ crdt-sync/pyproject.toml | 150 ++++++++++++++ crdt-sync/requirements.txt | 9 + 21 files changed, 1688 insertions(+) create mode 100644 crdt-sync/.github/workflows/pre-commit.yml create mode 100644 crdt-sync/.github/workflows/python-tests.yml create mode 100644 crdt-sync/.gitignore create mode 100644 crdt-sync/.pre-commit-config.yaml create mode 100644 crdt-sync/LICENSE create mode 100644 crdt-sync/README.md create mode 100644 crdt-sync/crdt_sync/__init__.py create mode 100644 crdt-sync/crdt_sync/_github.py create mode 100644 crdt-sync/crdt_sync/_hlc.py create mode 100644 crdt-sync/crdt_sync/_log.py create mode 100644 crdt-sync/crdt_sync/_record.py create mode 100644 crdt-sync/crdt_sync/_sync.py create mode 100644 crdt-sync/crdt_sync/tests/__init__.py create mode 100644 crdt-sync/crdt_sync/tests/conftest.py create mode 100644 crdt-sync/crdt_sync/tests/test_github.py create mode 100644 crdt-sync/crdt_sync/tests/test_hlc.py create mode 100644 crdt-sync/crdt_sync/tests/test_log.py create mode 100644 crdt-sync/crdt_sync/tests/test_record.py create mode 100644 crdt-sync/crdt_sync/tests/test_sync.py create mode 100644 crdt-sync/pyproject.toml create mode 100644 crdt-sync/requirements.txt diff --git a/crdt-sync/.github/workflows/pre-commit.yml b/crdt-sync/.github/workflows/pre-commit.yml new file mode 100644 index 0000000..98911e8 --- /dev/null +++ b/crdt-sync/.github/workflows/pre-commit.yml @@ -0,0 +1,19 @@ +name: pre-commit + +on: + push: + branches: [main] + pull_request: + branches: [main] + +jobs: + pre-commit: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: "3.12" + - name: Install dependencies + run: pip install -r requirements.txt + - uses: pre-commit/action@v3.0.1 diff --git a/crdt-sync/.github/workflows/python-tests.yml b/crdt-sync/.github/workflows/python-tests.yml new file mode 100644 index 0000000..66e5931 --- /dev/null +++ b/crdt-sync/.github/workflows/python-tests.yml @@ -0,0 +1,28 @@ +name: Tests + +on: + push: + branches: [main] + pull_request: + branches: [main] + +jobs: + test: + runs-on: ubuntu-latest + strategy: + matrix: + python-version: ["3.10", "3.11", "3.12"] + + steps: + - uses: actions/checkout@v4 + + - name: Set up Python ${{ matrix.python-version }} + uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python-version }} + + - name: Install dependencies + run: pip install -r requirements.txt + + - name: Run tests with coverage + run: python -m pytest -v diff --git a/crdt-sync/.gitignore b/crdt-sync/.gitignore new file mode 100644 index 0000000..eec3a7f --- /dev/null +++ b/crdt-sync/.gitignore @@ -0,0 +1,12 @@ +__pycache__/ +*.pyc +.venv/ +.mypy_cache/ +.ruff_cache/ +.pytest_cache/ +.coverage +coverage.lcov +htmlcov/ +dist/ +build/ +*.egg-info/ diff --git a/crdt-sync/.pre-commit-config.yaml b/crdt-sync/.pre-commit-config.yaml new file mode 100644 index 0000000..3b6f069 --- /dev/null +++ b/crdt-sync/.pre-commit-config.yaml @@ -0,0 +1,125 @@ +# Pre-commit Configuration for crdt-sync +# Install: pre-commit install && pre-commit install --hook-type pre-push +# Run: pre-commit run --all-files +# Update: pre-commit autoupdate + +default_language_version: + python: python3 +default_stages: [pre-commit] +fail_fast: false + +repos: + - repo: https://github.com/pre-commit/pre-commit-hooks + rev: v4.6.0 + hooks: + - id: trailing-whitespace + args: [--markdown-linebreak-ext=md] + - id: end-of-file-fixer + - id: check-yaml + - id: check-json + - id: check-toml + - id: check-added-large-files + args: [--maxkb=2000] + - id: check-merge-conflict + - id: detect-private-key + - id: debug-statements + - id: name-tests-test + args: [--pytest-test-first] + - id: check-ast + - id: mixed-line-ending + args: [--fix=lf] + - id: requirements-txt-fixer + + - repo: local + hooks: + - id: no-noqa + name: Block noqa comments + entry: '(?i)#\s*(noqa|type:\s*ignore)' + language: pygrep + types: [python] + - id: no-ruff-noqa + name: Block ruff noqa file-level comments + entry: '(?i)#\s*ruff:\s*noqa' + language: pygrep + types: [python] + + - repo: https://github.com/astral-sh/ruff-pre-commit + rev: v0.15.2 + hooks: + - id: ruff + args: [--fix, --unsafe-fixes, --exit-non-zero-on-fix, --show-fixes] + types_or: [python, pyi] + - id: ruff-format + types_or: [python, pyi] + + - repo: https://github.com/pre-commit/mirrors-mypy + rev: v1.13.0 + hooks: + - id: mypy + additional_dependencies: + - types-requests + args: + - --ignore-missing-imports + - --no-error-summary + - --disable-error-code=no-untyped-def + - --disable-error-code=no-untyped-call + - --disable-error-code=var-annotated + - --disable-error-code=no-any-unimported + - --disable-error-code=type-arg + - --disable-error-code=no-any-return + - --disable-error-code=misc + - --disable-error-code=unused-ignore + - --disable-error-code=unreachable + - --disable-error-code=assignment + - --disable-error-code=no-redef + - --disable-error-code=attr-defined + - --disable-error-code=arg-type + - --disable-error-code=union-attr + - --disable-error-code=call-overload + - --disable-error-code=return-value + - --disable-error-code=redundant-cast + - --disable-error-code=empty-body + - --disable-error-code=list-item + + - repo: https://github.com/pylint-dev/pylint + rev: v3.3.2 + hooks: + - id: pylint + args: + - --rcfile=pyproject.toml + - --fail-under=8.0 + - --jobs=4 + additional_dependencies: + - pytest + - requests + + - repo: https://github.com/PyCQA/bandit + rev: 1.7.10 + hooks: + - id: bandit + args: + - -c + - pyproject.toml + - --severity-level=high + - --confidence-level=medium + - --skip=B113 + additional_dependencies: ["bandit[toml]"] + exclude: ^(tests/|.*test.*\.py$) + + - repo: local + hooks: + - id: pytest-coverage + name: pytest with coverage enforcement + entry: python -m pytest + language: system + types: [python] + pass_filenames: false + require_serial: true + stages: [pre-push] + + - repo: https://github.com/codespell-project/codespell + rev: v2.3.0 + hooks: + - id: codespell + args: + - --skip=*.json,*.lock,.git,__pycache__,.venv diff --git a/crdt-sync/LICENSE b/crdt-sync/LICENSE new file mode 100644 index 0000000..d1aa637 --- /dev/null +++ b/crdt-sync/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2026 Krzysztof Rudnicki + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/crdt-sync/README.md b/crdt-sync/README.md new file mode 100644 index 0000000..9dab170 --- /dev/null +++ b/crdt-sync/README.md @@ -0,0 +1,100 @@ +# crdt-sync + +Shared CRDT merge scheme + GitHub-Contents-API sync transport, extracted for +reuse across four personal apps that each need cross-device sync: +[todo](https://github.com/kuhyx/todo-app), +[diet_guard](https://github.com/kuhyx/testsAndMisc/tree/main/python_pkg/diet_guard), +[wake_alarm](https://github.com/kuhyx/testsAndMisc/tree/main/python_pkg/wake_alarm), +and [screen-locker](https://github.com/kuhyx/screen-locker). + +## Why + +Two of those apps (diet_guard, and independently todo) had already converged +on the same shape of solution: treat a private GitHub repo as dumb file +storage via the REST Contents API (not a git clone), have each device push +its own file, and merge on read. diet_guard's merge logic +(entry-id-keyed, tombstone-wins) and todo's merge logic (`sqlite_crdt`'s +per-field HLC last-writer-wins) solve the same underlying problem — +convergent merge without a central server — for two different data shapes +(append-only logs vs. mutable per-field records). `crdt-sync` generalizes +both into one scheme so all four apps can share it, at the cost of todo's +notes and wake_alarm's single wake-time value both running through +CRDT machinery their own data doesn't strictly need — an explicit +consistency-over-leanness tradeoff. + +## The merge scheme + +A generic **LWW-map-with-sticky-remove**, built from three pieces: + +- **`Hlc`** — a Hybrid Logical Clock: `(wall_time_ms, counter, node_id)`, + totally ordered by tuple comparison (`node_id` breaks ties between two + clocks with identical wall time and counter, so no two distinct devices' + clocks ever compare equal). `Hlc.new_tick(node_id, previous)` advances off + wall-clock time, or bumps `counter` when the wall clock hasn't moved past + `previous` — the standard HLC construction, guaranteeing monotonicity even + across clock skew or two ticks in the same millisecond. +- **`Record`** — `id`, a `fields` map of `name -> (value, Hlc)`, and a sticky + `deleted` flag. `merge_field` keeps whichever side's `Hlc` is greater, + per field — this is what lets two devices concurrently edit *different* + fields of the same record (or the same field) and converge deterministically + without a central authority. `merge_record`'s `deleted` is **not** itself + per-field LWW: it's `a.deleted or b.deleted`, monotonic and one-directional. + This preserves a guarantee diet_guard's original tombstone scheme already + relied on — a delete can never be silently undone by merging in an older + non-deleted copy pulled from a device that hasn't seen the delete yet. +- **`Log`** = `dict[id, Record]`. `merge_logs` takes the union of ids and + `merge_record`s the ones present on both sides. + +All three merges are commutative and idempotent (provable directly from +`max`-by-`Hlc` and boolean-`or`), so pull order between devices never matters +and a repeated sync tick is a no-op — see `crdt_sync/tests/test_log.py`'s +property-style cases for the concrete proof. + +**Per-app adapters are out of scope for this library.** Translating a `Note`, +a `FoodEntry`, a workout entry, or wake_alarm's `{hour, minute}` value to and +from `Record` is each app's own job — keeping this library domain-agnostic is +what makes it actually shared, rather than shaped around one app's schema. + +## Install + +```bash +pip install "crdt-sync @ git+https://github.com/kuhyx/crdt-sync@v0.1.0" +``` + +## Usage + +```python +from crdt_sync import GitHubSyncClient, Hlc, Record, sync_log + +node_id = "pc" +clock = Hlc.new_tick(node_id) +record = Record(id="abc123", fields={"text": ("buy milk", clock)}, deleted=False) + +client = GitHubSyncClient(owner="kuhyx", repo="my-app-sync", token=token) +result = sync_log( + client=client, + device_id=node_id, + path_prefix="devices", + local_log={record.id: record}, + encode=lambda log: json.dumps({k: r.to_dict() for k, r in log.items()}), + decode=lambda text: { + k: Record.from_dict(v) for k, v in json.loads(text).items() + }, +) +``` + +`sync_log` pulls every other device's last-pushed log from +`//...`, merges each into the local log with +`merge_logs`, then pushes this device's own merged result back up — the same +pull-all/merge/push-own pattern diet_guard's `_sync.py` already used, made +domain-agnostic via the `encode`/`decode` callbacks so callers keep their own +on-disk JSON shape instead of `crdt-sync` dictating one. + +## Development + +```bash +python -m venv .venv && source .venv/bin/activate +pip install -r requirements.txt +pytest +pre-commit install && pre-commit install --hook-type pre-push +``` diff --git a/crdt-sync/crdt_sync/__init__.py b/crdt-sync/crdt_sync/__init__.py new file mode 100644 index 0000000..e6d96d2 --- /dev/null +++ b/crdt-sync/crdt_sync/__init__.py @@ -0,0 +1,23 @@ +"""Shared CRDT merge scheme + GitHub-Contents-API sync transport.""" + +from __future__ import annotations + +from crdt_sync._github import GitHubSyncClient, GitHubSyncError, RepoNotFoundError +from crdt_sync._hlc import Hlc +from crdt_sync._log import Log, merge_logs +from crdt_sync._record import Field, Record, merge_field, merge_record +from crdt_sync._sync import sync_log + +__all__ = [ + "Field", + "GitHubSyncClient", + "GitHubSyncError", + "Hlc", + "Log", + "Record", + "RepoNotFoundError", + "merge_field", + "merge_logs", + "merge_record", + "sync_log", +] diff --git a/crdt-sync/crdt_sync/_github.py b/crdt-sync/crdt_sync/_github.py new file mode 100644 index 0000000..a1a0ade --- /dev/null +++ b/crdt-sync/crdt_sync/_github.py @@ -0,0 +1,194 @@ +"""Minimal GitHub Contents API client used as dumb file storage. + +GitHub is used purely as file storage via the REST Contents API, not a git +clone -- there is no working tree and no git-level merge; the only merge is +the domain-level one in :mod:`crdt_sync._log`. Extracted from diet_guard's +``_sync_github.py`` (itself ported in spirit from ``todo``'s sync +transport), generalized to be domain-agnostic. +""" + +from __future__ import annotations + +import base64 + +import requests + +_API_BASE = "https://api.github.com" +_HTTP_NOT_FOUND = 404 +_DEFAULT_TIMEOUT_SECONDS = 15 + + +class GitHubSyncError(Exception): + """Raised for a GitHub API failure the caller must not silently ignore.""" + + +class RepoNotFoundError(GitHubSyncError): + """Raised when the configured repo itself is unreachable. + + Distinguished from a path-404 (nothing pushed to that path yet, which is + benign -- it just means no other device has synced before) so the caller + can tell "the repo name is wrong or the token isn't scoped to it" apart + from "no other device has synced yet". + """ + + +class GitHubSyncClient: + """Thin wrapper around the subset of the Contents API sync needs.""" + + def __init__( + self, + owner: str, + repo: str, + token: str, + *, + timeout_seconds: float = _DEFAULT_TIMEOUT_SECONDS, + ) -> None: + """Create a client scoped to one repo, authenticated with ``token``. + + Args: + owner: The repo owner/org (e.g. ``"kuhyx"``). + repo: The repo name (e.g. ``"my-app-sync"``). + token: A GitHub PAT with contents read/write on that repo. + timeout_seconds: Per-request network timeout. + """ + self._owner = owner + self._repo = repo + self._timeout_seconds = timeout_seconds + self._headers = { + "Authorization": f"Bearer {token}", + "Accept": "application/vnd.github+json", + } + + def _contents_url(self, path: str) -> str: + return f"{_API_BASE}/repos/{self._owner}/{self._repo}/contents/{path}" + + def _get(self, path: str) -> requests.Response: + try: + return requests.get( + self._contents_url(path), + headers=self._headers, + timeout=self._timeout_seconds, + ) + except requests.RequestException as exc: + msg = f"network error reading {path}" + raise GitHubSyncError(msg) from exc + + def _repo_exists(self) -> bool: + try: + response = requests.get( + f"{_API_BASE}/repos/{self._owner}/{self._repo}", + headers=self._headers, + timeout=self._timeout_seconds, + ) + except requests.RequestException: + return False + return response.ok + + def _raise_for_missing_path(self, path: str) -> None: + """Raise :class:`RepoNotFoundError` only if the repo is unreachable. + + A 404 on a path within a reachable repo just means nothing has been + pushed there yet, which is not an error worth raising on. + """ + if not self._repo_exists(): + msg = ( + f"{self._owner}/{self._repo} not found, private without " + f"access, or the token lacks contents permission " + f"(while reading {path})" + ) + raise RepoNotFoundError(msg) + + def get_file_text(self, path: str) -> str | None: + """Return the decoded text content at ``path``, or None if unused. + + Args: + path: A repo-relative file path, e.g. ``"devices/pc/log.json"``. + + Returns: + The file's text content, or None if nothing has been pushed + there yet (but the repo itself is reachable). + + Raises: + RepoNotFoundError: If the repo itself is unreachable. + GitHubSyncError: For any other non-2xx response or network error. + """ + response = self._get(path) + if response.status_code == _HTTP_NOT_FOUND: + self._raise_for_missing_path(path) + return None + if not response.ok: + msg = f"GET {path} failed: {response.status_code}" + raise GitHubSyncError(msg) + data = response.json() + content = data.get("content", "") if isinstance(data, dict) else "" + return base64.b64decode(content).decode("utf-8") + + def _existing_sha(self, path: str) -> str | None: + response = self._get(path) + if response.status_code == _HTTP_NOT_FOUND: + self._raise_for_missing_path(path) + return None + if not response.ok: + msg = f"GET {path} (for sha) failed: {response.status_code}" + raise GitHubSyncError(msg) + data = response.json() + sha = data.get("sha") if isinstance(data, dict) else None + return sha if isinstance(sha, str) else None + + def put_file_text(self, path: str, text: str, *, message: str) -> None: + """Create or update the file at ``path`` with ``text``. + + Args: + path: A repo-relative file path. + text: The full new content (this device's complete merged log). + message: The commit message for this push. + + Raises: + GitHubSyncError: On any non-2xx response or network error. + """ + sha = self._existing_sha(path) + payload: dict[str, object] = { + "message": message, + "content": base64.b64encode(text.encode("utf-8")).decode("ascii"), + } + if sha is not None: + payload["sha"] = sha + try: + response = requests.put( + self._contents_url(path), + headers=self._headers, + json=payload, + timeout=self._timeout_seconds, + ) + except requests.RequestException as exc: + msg = f"network error pushing {path}" + raise GitHubSyncError(msg) from exc + if not response.ok: + msg = f"PUT {path} failed: {response.status_code}" + raise GitHubSyncError(msg) + + def list_directory(self, path: str) -> list[str]: + """Return the entry names directly under ``path`` (empty if unused). + + Args: + path: A repo-relative directory path, e.g. ``"devices"``. + + Raises: + RepoNotFoundError: If the repo itself is unreachable. + GitHubSyncError: For any other non-2xx response or network error. + """ + response = self._get(path) + if response.status_code == _HTTP_NOT_FOUND: + self._raise_for_missing_path(path) + return [] + if not response.ok: + msg = f"GET {path} (list) failed: {response.status_code}" + raise GitHubSyncError(msg) + data = response.json() + if not isinstance(data, list): + return [] + return [ + item["name"] + for item in data + if isinstance(item, dict) and isinstance(item.get("name"), str) + ] diff --git a/crdt-sync/crdt_sync/_hlc.py b/crdt-sync/crdt_sync/_hlc.py new file mode 100644 index 0000000..1a6bbe6 --- /dev/null +++ b/crdt-sync/crdt_sync/_hlc.py @@ -0,0 +1,97 @@ +"""Hybrid Logical Clock: a totally-ordered, monotonic per-node timestamp. + +An HLC combines wall-clock time with a logical counter so that two ticks +issued by the same node in the same millisecond still get a strict order, +and two ticks from different nodes are always comparable (the node id breaks +ties), without requiring synchronized clocks across devices. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime, timezone +import time + +_ISO_PREFIX_LEN = 23 # len("YYYY-MM-DDTHH:MM:SS.mmm") +_EPOCH = datetime(1970, 1, 1, tzinfo=timezone.utc) + + +@dataclass(frozen=True, order=True) +class Hlc: + """A Hybrid Logical Clock value. + + Ordered by ``(wall_time_ms, counter, node_id)`` tuple comparison, which + is also the order ``@dataclass(order=True)`` derives from the field + declaration order below -- two clocks from different nodes never compare + equal, since ``node_id`` is the final, always-distinguishing tiebreaker. + """ + + wall_time_ms: int + counter: int + node_id: str + + @staticmethod + def new_tick( + node_id: str, + previous: Hlc | None = None, + wall_time_ms: int | None = None, + ) -> Hlc: + """Return the next clock value for ``node_id``. + + Args: + node_id: The issuing node's identifier. + previous: This node's last-issued clock value, if any. Passing + the previous value is what makes the clock monotonic even + when the wall clock hasn't advanced (or has gone backwards). + wall_time_ms: Override for the current wall-clock time in + milliseconds since the Unix epoch; defaults to the real + clock. Exposed for deterministic tests. + + Returns: + A new ``Hlc`` strictly greater than ``previous`` (if given). + """ + now = wall_time_ms if wall_time_ms is not None else int(time.time() * 1000) + if previous is None: + return Hlc(wall_time_ms=now, counter=0, node_id=node_id) + new_wall = max(now, previous.wall_time_ms) + counter = previous.counter + 1 if new_wall == previous.wall_time_ms else 0 + return Hlc(wall_time_ms=new_wall, counter=counter, node_id=node_id) + + def to_str(self) -> str: + """Serialize to a human-readable, lexicographically sortable string. + + Format: ``Z--``. The iso8601 + portion is fixed-width (always exactly ``YYYY-MM-DDTHH:MM:SS.mmm``), + which is what makes plain string comparison agree with ``Hlc`` + comparison for the wall-clock component. + """ + seconds, millis = divmod(self.wall_time_ms, 1000) + dt = datetime.fromtimestamp(seconds, tz=timezone.utc) + iso = dt.strftime("%Y-%m-%dT%H:%M:%S") + return f"{iso}.{millis:03d}Z-{self.counter:04x}-{self.node_id}" + + @classmethod + def from_str(cls, text: str) -> Hlc: + """Parse the format produced by :meth:`to_str`. + + Raises: + ValueError: If ``text`` is not in the expected format. + """ + iso_part, sep, rest = text.partition("Z-") + if not sep or len(iso_part) != _ISO_PREFIX_LEN: + msg = f"not a valid Hlc string: {text!r}" + raise ValueError(msg) + counter_hex, sep, node_id = rest.partition("-") + if not sep: + msg = f"not a valid Hlc string: {text!r}" + raise ValueError(msg) + dt = datetime.strptime(iso_part, "%Y-%m-%dT%H:%M:%S.%f").replace( + tzinfo=timezone.utc + ) + delta = dt - _EPOCH + wall_time_ms = ( + delta.days * 86_400_000 + delta.seconds * 1000 + delta.microseconds // 1000 + ) + return cls( + wall_time_ms=wall_time_ms, counter=int(counter_hex, 16), node_id=node_id + ) diff --git a/crdt-sync/crdt_sync/_log.py b/crdt-sync/crdt_sync/_log.py new file mode 100644 index 0000000..57710b6 --- /dev/null +++ b/crdt-sync/crdt_sync/_log.py @@ -0,0 +1,32 @@ +"""Log: a collection of Records keyed by id, with a union-merge.""" + +from __future__ import annotations + +from crdt_sync._record import Record, merge_record + +Log = dict[str, Record] + + +def merge_logs(local: Log, remote: Log) -> Log: + """Return the union of ``local`` and ``remote``, merging shared ids. + + Commutative and idempotent: ``merge_logs(a, b) == merge_logs(b, a)`` and + ``merge_logs(x, x) == x``, so pull order between devices never matters + and a repeated sync tick is a no-op. + + Args: + local: This device's current full log (including tombstones). + remote: Another device's last-pushed full log. + + Returns: + The merged log, containing every id present in either input. + """ + merged: Log = dict(local) + for record_id, remote_record in remote.items(): + local_record = merged.get(record_id) + merged[record_id] = ( + remote_record + if local_record is None + else merge_record(local_record, remote_record) + ) + return merged diff --git a/crdt-sync/crdt_sync/_record.py b/crdt-sync/crdt_sync/_record.py new file mode 100644 index 0000000..3fa02c8 --- /dev/null +++ b/crdt-sync/crdt_sync/_record.py @@ -0,0 +1,128 @@ +"""Record: a per-field LWW map with a sticky (monotonic) delete flag. + +This is the unit CRDT merges operate on. It's deliberately domain-agnostic -- +translating an app's own model (a note, a food-log entry, a workout entry, +a wake-time value) to and from ``Record`` is each app's job, not this +library's. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any + +from crdt_sync._hlc import Hlc + +Field = tuple[Any, Hlc] + + +@dataclass(frozen=True) +class Record: + """One CRDT record: an id, per-field LWW values, and a sticky tombstone. + + Attributes: + id: Stable identifier, unique within one ``Log``. + fields: Field name -> ``(value, Hlc)``. Concurrent edits to the same + field converge to whichever side has the greater ``Hlc``. + deleted: Whether this record is tombstoned. Deletion is monotonic: + once true on either side of a merge, it stays true. + deleted_hlc: The clock value of the delete, if any -- kept for + informational/ordering purposes only; it does not participate in + the merge decision (that's a plain boolean OR, see + :func:`merge_record`). + """ + + id: str + fields: dict[str, Field] + deleted: bool = False + deleted_hlc: Hlc | None = None + + def to_dict(self) -> dict[str, Any]: + """Serialize to a JSON-compatible dict.""" + return { + "id": self.id, + "fields": { + name: [value, hlc.to_str()] + for name, (value, hlc) in self.fields.items() + }, + "deleted": self.deleted, + "deleted_hlc": self.deleted_hlc.to_str() + if self.deleted_hlc is not None + else None, + } + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> Record: + """Parse the format produced by :meth:`to_dict`.""" + fields = { + name: (value, Hlc.from_str(hlc_str)) + for name, (value, hlc_str) in data["fields"].items() + } + deleted_hlc_str = data.get("deleted_hlc") + deleted_hlc = ( + Hlc.from_str(deleted_hlc_str) if deleted_hlc_str is not None else None + ) + return cls( + id=data["id"], + fields=fields, + deleted=bool(data["deleted"]), + deleted_hlc=deleted_hlc, + ) + + +def merge_field(a: Field, b: Field) -> Field: + """Return whichever of ``a``, ``b`` has the greater ``Hlc``. + + Ties (equal ``Hlc``) keep ``a``; two ticks never compare equal across + different nodes (see :class:`crdt_sync._hlc.Hlc`), so a tie only happens + when both sides are literally the same tick re-merged, in which case the + value is expected to be identical anyway. + """ + return a if a[1] >= b[1] else b + + +def _merge_deleted_hlc(a: Record, b: Record) -> Hlc | None: + """Return the delete-clock to keep, informational only (see class docs).""" + if a.deleted and b.deleted: + if a.deleted_hlc is None: + return b.deleted_hlc + if b.deleted_hlc is None: + return a.deleted_hlc + return max(a.deleted_hlc, b.deleted_hlc) + if a.deleted: + return a.deleted_hlc + if b.deleted: + return b.deleted_hlc + return None + + +def merge_record(a: Record, b: Record) -> Record: + """Merge two versions of the same record. + + Per-field last-writer-wins over the union of field names, plus a sticky + delete: ``deleted = a.deleted or b.deleted``, never the other way around. + A tombstone can never be resurrected by merging in an older, non-deleted + copy pulled from a device that hasn't seen the delete yet. + + Commutative and idempotent: ``merge_record(a, b) == merge_record(b, a)`` + and ``merge_record(a, a) == a``. + + Raises: + ValueError: If ``a.id != b.id`` -- merging two different records is + a caller bug, not a case to silently paper over. + """ + if a.id != b.id: + msg = f"cannot merge records with different ids: {a.id!r} != {b.id!r}" + raise ValueError(msg) + merged_fields = dict(a.fields) + for name, b_field in b.fields.items(): + a_field = merged_fields.get(name) + merged_fields[name] = ( + b_field if a_field is None else merge_field(a_field, b_field) + ) + return Record( + id=a.id, + fields=merged_fields, + deleted=a.deleted or b.deleted, + deleted_hlc=_merge_deleted_hlc(a, b), + ) diff --git a/crdt-sync/crdt_sync/_sync.py b/crdt-sync/crdt_sync/_sync.py new file mode 100644 index 0000000..b1edb4e --- /dev/null +++ b/crdt-sync/crdt_sync/_sync.py @@ -0,0 +1,106 @@ +"""Domain-agnostic pull/merge/push sync orchestration. + +Generalizes diet_guard's original ``_sync.py`` (pull every other device's +pushed log, merge with the local one, push this device's own merged result +back up) so any app can reuse the loop while keeping its own on-disk JSON +shape via the ``encode``/``decode`` callbacks -- this module has no opinion +on what a ``Record``'s fields actually mean. +""" + +from __future__ import annotations + +import logging +from typing import TYPE_CHECKING + +from crdt_sync._log import merge_logs + +if TYPE_CHECKING: + from collections.abc import Callable + + from crdt_sync._github import GitHubSyncClient + from crdt_sync._log import Log + +_logger = logging.getLogger(__name__) + +_DEFAULT_FILENAME = "log.json" + + +def _pull_remote_logs( + client: GitHubSyncClient, + device_id: str, + path_prefix: str, + filename: str, + decode: Callable[[str], Log], +) -> list[Log]: + """Return every other device's last-pushed log, skipping this one. + + A device whose pushed file is corrupt or unparsable (e.g. an interrupted + push) is logged and skipped, same as one that has never pushed at all -- + GitHub is an external system boundary, and one bad device's file must not + stall merging in every other device's. + """ + remote_logs: list[Log] = [] + for other_device_id in client.list_directory(path_prefix): + if other_device_id == device_id: + continue + text = client.get_file_text(f"{path_prefix}/{other_device_id}/{filename}") + if text is None: + continue + try: + remote_logs.append(decode(text)) + except (ValueError, KeyError, TypeError): + _logger.warning( + "Unparsable log pushed by device %r, skipping", + other_device_id, + ) + return remote_logs + + +def sync_log( + *, + client: GitHubSyncClient, + device_id: str, + path_prefix: str, + local_log: Log, + encode: Callable[[Log], str], + decode: Callable[[str], Log], + filename: str = _DEFAULT_FILENAME, + commit_message: str = "crdt_sync: update log", +) -> Log: + """Run one full sync tick: pull every other device's log, merge, push. + + Pulls from ``//`` for every + device directory GitHub reports under ``path_prefix``, merges each into + ``local_log`` with :func:`crdt_sync.merge_logs`, then pushes this + device's own merged result to ``//``. + + Args: + client: An authenticated :class:`GitHubSyncClient`. + device_id: This device's identifier; also the directory name its own + log is pushed under. + path_prefix: The repo-relative directory holding one subdirectory + per device (e.g. ``"devices"``). + local_log: This device's current full log (including tombstones). + encode: Serializes a merged log for pushing. + decode: Parses a remote device's pushed text back into a log. + Raising ``ValueError``, ``KeyError``, or ``TypeError`` is treated + as a corrupt/unparsable push, and that device is skipped for + this tick rather than aborting the whole sync. + filename: The file name each device pushes its log as. + commit_message: The commit message used for this device's push. + + Returns: + The merged log, as pushed. + """ + merged = dict(local_log) + for remote_log in _pull_remote_logs( + client, device_id, path_prefix, filename, decode + ): + merged = merge_logs(merged, remote_log) + + client.put_file_text( + f"{path_prefix}/{device_id}/{filename}", + encode(merged), + message=commit_message, + ) + return merged diff --git a/crdt-sync/crdt_sync/tests/__init__.py b/crdt-sync/crdt_sync/tests/__init__.py new file mode 100644 index 0000000..72c1224 --- /dev/null +++ b/crdt-sync/crdt_sync/tests/__init__.py @@ -0,0 +1 @@ +"""Tests for the crdt_sync package.""" diff --git a/crdt-sync/crdt_sync/tests/conftest.py b/crdt-sync/crdt_sync/tests/conftest.py new file mode 100644 index 0000000..b930d35 --- /dev/null +++ b/crdt-sync/crdt_sync/tests/conftest.py @@ -0,0 +1,22 @@ +"""Shared pytest fixtures for crdt_sync's test suite.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +import pytest + +from crdt_sync import Hlc + +if TYPE_CHECKING: + from collections.abc import Callable + + +@pytest.fixture +def make_hlc() -> Callable[..., Hlc]: + """Return a factory for deterministic Hlc values.""" + + def _make(wall_time_ms: int, counter: int = 0, node_id: str = "node-a") -> Hlc: + return Hlc(wall_time_ms=wall_time_ms, counter=counter, node_id=node_id) + + return _make diff --git a/crdt-sync/crdt_sync/tests/test_github.py b/crdt-sync/crdt_sync/tests/test_github.py new file mode 100644 index 0000000..439b49b --- /dev/null +++ b/crdt-sync/crdt_sync/tests/test_github.py @@ -0,0 +1,170 @@ +"""Tests for the GitHub Contents API sync client. + +The HTTP layer is fully mocked (``requests.get``/``requests.put``), so every +branch -- success, path-404-but-repo-ok, repo-404, non-2xx, and network +exceptions -- is exercised without any network access. +""" + +from __future__ import annotations + +import base64 +from unittest.mock import MagicMock, patch + +import pytest +import requests + +from crdt_sync import GitHubSyncClient, GitHubSyncError, RepoNotFoundError, _github + + +def _response(status_code: int = 200, json_data: object = None) -> MagicMock: + """Build a fake ``requests.Response`` with a fixed status and JSON body.""" + response = MagicMock() + response.status_code = status_code + response.ok = 200 <= status_code < 300 + response.json = MagicMock(return_value=json_data if json_data is not None else {}) + return response + + +def _client() -> GitHubSyncClient: + return GitHubSyncClient("kuhyx", "crdt-sync-demo", "fake-token") + + +def _patch_get(*responses: MagicMock) -> object: + """Patch ``requests.get`` to return each of ``responses`` in order.""" + return patch.object(_github.requests, "get", side_effect=list(responses)) + + +def _patch_get_raises() -> object: + return patch.object( + _github.requests, + "get", + side_effect=requests.ConnectionError("offline"), + ) + + +class TestGetFileText: + def test_returns_decoded_content_on_success(self) -> None: + encoded = base64.b64encode(b"hello world").decode("ascii") + with _patch_get(_response(200, {"content": encoded})): + assert _client().get_file_text("devices/pc/log.json") == "hello world" + + def test_returns_none_for_an_unused_path_on_a_real_repo(self) -> None: + with _patch_get(_response(404), _response(200)): + assert _client().get_file_text("devices/phone/log.json") is None + + def test_raises_repo_not_found_when_the_repo_itself_is_missing(self) -> None: + with ( + _patch_get(_response(404), _response(404)), + pytest.raises(RepoNotFoundError), + ): + _client().get_file_text("devices/pc/log.json") + + def test_raises_sync_error_on_a_non_2xx_non_404(self) -> None: + with _patch_get(_response(500)), pytest.raises(GitHubSyncError): + _client().get_file_text("devices/pc/log.json") + + def test_raises_sync_error_on_a_network_exception(self) -> None: + with _patch_get_raises(), pytest.raises(GitHubSyncError): + _client().get_file_text("devices/pc/log.json") + + def test_treats_a_network_error_during_the_repo_check_as_repo_missing( + self, + ) -> None: + with ( + patch.object( + _github.requests, + "get", + side_effect=[_response(404), requests.ConnectionError("offline")], + ), + pytest.raises(RepoNotFoundError), + ): + _client().get_file_text("devices/pc/log.json") + + +class TestListDirectory: + def test_returns_entry_names(self) -> None: + payload = [{"name": "pc"}, {"name": "phone"}, {"not_a_name": "x"}] + with _patch_get(_response(200, payload)): + assert _client().list_directory("devices") == ["pc", "phone"] + + def test_returns_empty_list_when_response_is_not_a_list(self) -> None: + with _patch_get(_response(200, {"unexpected": "shape"})): + assert _client().list_directory("devices") == [] + + def test_returns_empty_list_for_an_unused_path_on_a_real_repo(self) -> None: + with _patch_get(_response(404), _response(200)): + assert _client().list_directory("devices") == [] + + def test_raises_repo_not_found_when_the_repo_itself_is_missing(self) -> None: + with ( + _patch_get(_response(404), _response(404)), + pytest.raises(RepoNotFoundError), + ): + _client().list_directory("devices") + + def test_raises_sync_error_on_a_non_2xx_non_404(self) -> None: + with _patch_get(_response(500)), pytest.raises(GitHubSyncError): + _client().list_directory("devices") + + +class TestPutFileText: + def test_creates_a_new_file_with_no_sha_when_none_existed(self) -> None: + with ( + _patch_get(_response(404), _response(200)), + patch.object( + _github.requests, "put", return_value=_response(201) + ) as put_mock, + ): + _client().put_file_text("devices/pc/log.json", "{}", message="m") + assert "sha" not in put_mock.call_args.kwargs["json"] + + def test_updates_an_existing_file_by_including_its_sha(self) -> None: + with ( + _patch_get(_response(200, {"sha": "abc123"})), + patch.object( + _github.requests, "put", return_value=_response(200) + ) as put_mock, + ): + _client().put_file_text("devices/pc/log.json", "{}", message="m") + assert put_mock.call_args.kwargs["json"]["sha"] == "abc123" + + def test_treats_a_non_string_sha_field_as_absent(self) -> None: + with ( + _patch_get(_response(200, {"sha": 12345})), + patch.object( + _github.requests, "put", return_value=_response(200) + ) as put_mock, + ): + _client().put_file_text("devices/pc/log.json", "{}", message="m") + assert "sha" not in put_mock.call_args.kwargs["json"] + + def test_raises_repo_not_found_when_checking_sha_on_a_missing_repo(self) -> None: + with ( + _patch_get(_response(404), _response(404)), + pytest.raises(RepoNotFoundError), + ): + _client().put_file_text("devices/pc/log.json", "{}", message="m") + + def test_raises_sync_error_when_the_sha_check_itself_fails(self) -> None: + with _patch_get(_response(500)), pytest.raises(GitHubSyncError): + _client().put_file_text("devices/pc/log.json", "{}", message="m") + + def test_raises_sync_error_on_a_put_network_exception(self) -> None: + with ( + _patch_get(_response(404), _response(200)), + patch.object( + _github.requests, + "put", + side_effect=requests.ConnectionError("offline"), + ), + pytest.raises(GitHubSyncError), + ): + _client().put_file_text("devices/pc/log.json", "{}", message="m") + + def test_raises_sync_error_on_a_put_non_2xx_response(self) -> None: + with ( + _patch_get(_response(404), _response(200)), + patch.object(_github.requests, "put", return_value=_response(422)), + pytest.raises(GitHubSyncError), + ): + _client().put_file_text("devices/pc/log.json", "{}", message="m") diff --git a/crdt-sync/crdt_sync/tests/test_hlc.py b/crdt-sync/crdt_sync/tests/test_hlc.py new file mode 100644 index 0000000..381c7ee --- /dev/null +++ b/crdt-sync/crdt_sync/tests/test_hlc.py @@ -0,0 +1,100 @@ +"""Tests for the Hybrid Logical Clock.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +import pytest + +from crdt_sync import Hlc + +if TYPE_CHECKING: + from collections.abc import Callable + + +class TestOrdering: + def test_greater_wall_time_wins(self, make_hlc: Callable[..., Hlc]) -> None: + assert make_hlc(200) > make_hlc(100) + + def test_equal_wall_time_greater_counter_wins( + self, make_hlc: Callable[..., Hlc] + ) -> None: + assert make_hlc(100, counter=2) > make_hlc(100, counter=1) + + def test_equal_wall_time_and_counter_breaks_tie_on_node_id( + self, make_hlc: Callable[..., Hlc] + ) -> None: + assert make_hlc(100, counter=1, node_id="b") > make_hlc( + 100, counter=1, node_id="a" + ) + + def test_identical_clocks_compare_equal(self, make_hlc: Callable[..., Hlc]) -> None: + assert make_hlc(100, counter=1, node_id="a") == make_hlc( + 100, counter=1, node_id="a" + ) + + +class TestNewTick: + def test_first_tick_has_counter_zero(self) -> None: + tick = Hlc.new_tick("node-a", wall_time_ms=1000) + assert tick == Hlc(wall_time_ms=1000, counter=0, node_id="node-a") + + def test_advancing_wall_clock_resets_counter( + self, make_hlc: Callable[..., Hlc] + ) -> None: + previous = make_hlc(1000, counter=5) + tick = Hlc.new_tick("node-a", previous=previous, wall_time_ms=2000) + assert tick == Hlc(wall_time_ms=2000, counter=0, node_id="node-a") + + def test_stalled_wall_clock_increments_counter( + self, make_hlc: Callable[..., Hlc] + ) -> None: + previous = make_hlc(1000, counter=5) + tick = Hlc.new_tick("node-a", previous=previous, wall_time_ms=1000) + assert tick == Hlc(wall_time_ms=1000, counter=6, node_id="node-a") + + def test_regressed_wall_clock_still_advances_monotonically( + self, make_hlc: Callable[..., Hlc] + ) -> None: + previous = make_hlc(1000, counter=5) + tick = Hlc.new_tick("node-a", previous=previous, wall_time_ms=500) + assert tick == Hlc(wall_time_ms=1000, counter=6, node_id="node-a") + + def test_defaults_to_the_real_clock_when_unset(self) -> None: + tick = Hlc.new_tick("node-a") + assert tick.node_id == "node-a" + assert tick.wall_time_ms > 0 + + +class TestStringRoundTrip: + def test_round_trips_through_to_str_and_from_str( + self, make_hlc: Callable[..., Hlc] + ) -> None: + original = make_hlc(1_751_000_123_456, counter=7, node_id="phone") + assert Hlc.from_str(original.to_str()) == original + + def test_to_str_is_lexicographically_sortable_by_wall_time( + self, make_hlc: Callable[..., Hlc] + ) -> None: + earlier = make_hlc(1000) + later = make_hlc(2000) + assert earlier.to_str() < later.to_str() + + def test_to_str_is_lexicographically_sortable_by_counter( + self, make_hlc: Callable[..., Hlc] + ) -> None: + lower = make_hlc(1000, counter=1) + higher = make_hlc(1000, counter=200) + assert lower.to_str() < higher.to_str() + + def test_from_str_rejects_a_missing_z_separator(self) -> None: + with pytest.raises(ValueError, match="not a valid Hlc string"): + Hlc.from_str("not-a-valid-clock-string-at-all") + + def test_from_str_rejects_a_wrong_length_iso_prefix(self) -> None: + with pytest.raises(ValueError, match="not a valid Hlc string"): + Hlc.from_str("2026-07-05T12:00:00Z-0000-node-a") + + def test_from_str_rejects_a_missing_node_id_separator(self) -> None: + with pytest.raises(ValueError, match="not a valid Hlc string"): + Hlc.from_str("2026-07-05T12:00:00.000Z-0000") diff --git a/crdt-sync/crdt_sync/tests/test_log.py b/crdt-sync/crdt_sync/tests/test_log.py new file mode 100644 index 0000000..dc9e7da --- /dev/null +++ b/crdt-sync/crdt_sync/tests/test_log.py @@ -0,0 +1,80 @@ +"""Tests for merge_logs, including its CRDT convergence properties.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +from crdt_sync import Hlc, Record, merge_logs + +if TYPE_CHECKING: + from collections.abc import Callable + + from crdt_sync import Log + + +class TestMergeLogs: + def test_returns_the_union_of_disjoint_ids( + self, make_hlc: Callable[..., Hlc] + ) -> None: + local = {"a": Record(id="a", fields={"text": ("a", make_hlc(100))})} + remote = {"b": Record(id="b", fields={"text": ("b", make_hlc(100))})} + assert merge_logs(local, remote) == {**local, **remote} + + def test_merges_a_shared_id_field_by_field( + self, make_hlc: Callable[..., Hlc] + ) -> None: + local = {"a": Record(id="a", fields={"text": ("old", make_hlc(100))})} + remote = {"a": Record(id="a", fields={"text": ("new", make_hlc(200))})} + merged = merge_logs(local, remote) + assert merged["a"].fields["text"] == ("new", make_hlc(200)) + + def test_a_delete_on_one_side_survives_the_merge( + self, make_hlc: Callable[..., Hlc] + ) -> None: + local = { + "a": Record(id="a", fields={}, deleted=True, deleted_hlc=make_hlc(200)) + } + remote = {"a": Record(id="a", fields={"text": ("still here", make_hlc(100))})} + assert merge_logs(local, remote)["a"].deleted is True + + def test_does_not_mutate_either_input(self, make_hlc: Callable[..., Hlc]) -> None: + local = {"a": Record(id="a", fields={"text": ("a", make_hlc(100))})} + remote = {"b": Record(id="b", fields={"text": ("b", make_hlc(100))})} + local_before = dict(local) + remote_before = dict(remote) + merge_logs(local, remote) + assert local == local_before + assert remote == remote_before + + +class TestConvergenceProperties: + def _sample_logs(self, make_hlc: Callable[..., Hlc]) -> tuple[Log, Log]: + a = { + "shared": Record( + id="shared", fields={"text": ("a-version", make_hlc(100))} + ), + "only-in-a": Record(id="only-in-a", fields={"text": ("a", make_hlc(50))}), + } + b = { + "shared": Record( + id="shared", fields={"text": ("b-version", make_hlc(200))} + ), + "only-in-b": Record(id="only-in-b", fields={"text": ("b", make_hlc(50))}), + } + return a, b + + def test_is_commutative(self, make_hlc: Callable[..., Hlc]) -> None: + a, b = self._sample_logs(make_hlc) + assert merge_logs(a, b) == merge_logs(b, a) + + def test_is_idempotent(self, make_hlc: Callable[..., Hlc]) -> None: + a, _ = self._sample_logs(make_hlc) + assert merge_logs(a, a) == a + + def test_repeated_merge_of_the_same_remote_is_a_no_op( + self, make_hlc: Callable[..., Hlc] + ) -> None: + a, b = self._sample_logs(make_hlc) + once = merge_logs(a, b) + twice = merge_logs(once, b) + assert once == twice diff --git a/crdt-sync/crdt_sync/tests/test_record.py b/crdt-sync/crdt_sync/tests/test_record.py new file mode 100644 index 0000000..c372963 --- /dev/null +++ b/crdt-sync/crdt_sync/tests/test_record.py @@ -0,0 +1,111 @@ +"""Tests for Record, merge_field, and merge_record.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +import pytest + +from crdt_sync import Hlc, Record, merge_field, merge_record + +if TYPE_CHECKING: + from collections.abc import Callable + + +class TestMergeField: + def test_greater_hlc_wins(self, make_hlc: Callable[..., Hlc]) -> None: + older = ("a", make_hlc(100)) + newer = ("b", make_hlc(200)) + assert merge_field(older, newer) == newer + assert merge_field(newer, older) == newer + + def test_equal_hlc_keeps_the_first_argument( + self, make_hlc: Callable[..., Hlc] + ) -> None: + clock = make_hlc(100) + a = ("a", clock) + b = ("a", clock) + assert merge_field(a, b) == a + + +class TestMergeRecord: + def test_raises_when_ids_differ(self) -> None: + a = Record(id="a", fields={}) + b = Record(id="b", fields={}) + with pytest.raises(ValueError, match="different ids"): + merge_record(a, b) + + def test_merges_disjoint_fields(self, make_hlc: Callable[..., Hlc]) -> None: + a = Record(id="x", fields={"text": ("hello", make_hlc(100))}) + b = Record(id="x", fields={"priority": ("high", make_hlc(50))}) + merged = merge_record(a, b) + assert merged.fields == { + "text": ("hello", make_hlc(100)), + "priority": ("high", make_hlc(50)), + } + + def test_shared_field_keeps_the_later_write( + self, make_hlc: Callable[..., Hlc] + ) -> None: + a = Record(id="x", fields={"text": ("old", make_hlc(100))}) + b = Record(id="x", fields={"text": ("new", make_hlc(200))}) + merged = merge_record(a, b) + assert merged.fields["text"] == ("new", make_hlc(200)) + + def test_delete_is_sticky_against_an_older_non_deleted_copy( + self, make_hlc: Callable[..., Hlc] + ) -> None: + deleted = Record(id="x", fields={}, deleted=True, deleted_hlc=make_hlc(200)) + stale = Record(id="x", fields={"text": ("resurrected", make_hlc(100))}) + assert merge_record(deleted, stale).deleted is True + assert merge_record(stale, deleted).deleted is True + + def test_neither_side_deleted_stays_not_deleted(self) -> None: + a = Record(id="x", fields={}) + b = Record(id="x", fields={}) + merged = merge_record(a, b) + assert merged.deleted is False + assert merged.deleted_hlc is None + + def test_both_sides_deleted_keeps_the_later_delete_clock( + self, make_hlc: Callable[..., Hlc] + ) -> None: + a = Record(id="x", fields={}, deleted=True, deleted_hlc=make_hlc(100)) + b = Record(id="x", fields={}, deleted=True, deleted_hlc=make_hlc(200)) + merged = merge_record(a, b) + assert merged.deleted_hlc == make_hlc(200) + + def test_both_deleted_but_one_side_missing_a_clock_keeps_the_other( + self, make_hlc: Callable[..., Hlc] + ) -> None: + a = Record(id="x", fields={}, deleted=True, deleted_hlc=None) + b = Record(id="x", fields={}, deleted=True, deleted_hlc=make_hlc(200)) + assert merge_record(a, b).deleted_hlc == make_hlc(200) + assert merge_record(b, a).deleted_hlc == make_hlc(200) + + def test_is_commutative(self, make_hlc: Callable[..., Hlc]) -> None: + record_a = Record(id="x", fields={"text": ("a", make_hlc(100))}) + record_b = Record( + id="x", fields={"text": ("b", make_hlc(200)), "extra": ("e", make_hlc(1))} + ) + assert merge_record(record_a, record_b) == merge_record(record_b, record_a) + + def test_is_idempotent(self, make_hlc: Callable[..., Hlc]) -> None: + record = Record(id="x", fields={"text": ("a", make_hlc(100))}) + assert merge_record(record, record) == record + + +class TestRecordDictRoundTrip: + def test_round_trips_a_record_with_fields( + self, make_hlc: Callable[..., Hlc] + ) -> None: + record = Record(id="x", fields={"text": ("hello", make_hlc(100, node_id="pc"))}) + assert Record.from_dict(record.to_dict()) == record + + def test_round_trips_a_deleted_record(self, make_hlc: Callable[..., Hlc]) -> None: + record = Record(id="x", fields={}, deleted=True, deleted_hlc=make_hlc(200)) + assert Record.from_dict(record.to_dict()) == record + + def test_round_trips_a_record_with_no_delete_clock(self) -> None: + record = Record(id="x", fields={}, deleted=False, deleted_hlc=None) + assert Record.from_dict(record.to_dict()) == record diff --git a/crdt-sync/crdt_sync/tests/test_sync.py b/crdt-sync/crdt_sync/tests/test_sync.py new file mode 100644 index 0000000..593ad18 --- /dev/null +++ b/crdt-sync/crdt_sync/tests/test_sync.py @@ -0,0 +1,160 @@ +"""Tests for the pull/merge/push sync orchestration.""" + +from __future__ import annotations + +import json +import logging +from typing import TYPE_CHECKING +from unittest.mock import MagicMock + +from crdt_sync import GitHubSyncClient, Record, sync_log + +if TYPE_CHECKING: + from collections.abc import Callable + + import pytest + + from crdt_sync import Hlc, Log + + +def _encode(log: Log) -> str: + return json.dumps( + {record_id: record.to_dict() for record_id, record in log.items()} + ) + + +def _decode(text: str) -> Log: + return { + record_id: Record.from_dict(data) + for record_id, data in json.loads(text).items() + } + + +def _mock_client( + *, + devices: tuple[str, ...] = (), + files: dict[str, str] | None = None, +) -> MagicMock: + """Build a mock ``GitHubSyncClient`` covering the methods sync_log calls.""" + client = MagicMock(spec=GitHubSyncClient) + client.list_directory.return_value = list(devices) + resolved_files = files or {} + client.get_file_text.side_effect = resolved_files.get + return client + + +class TestSyncLog: + def test_pushes_local_log_when_no_other_devices_have_synced( + self, make_hlc: Callable[..., Hlc] + ) -> None: + local_log = {"a": Record(id="a", fields={"text": ("hello", make_hlc(100))})} + client = _mock_client() + + merged = sync_log( + client=client, + device_id="pc", + path_prefix="devices", + local_log=local_log, + encode=_encode, + decode=_decode, + ) + + assert merged == local_log + client.put_file_text.assert_called_once() + assert client.put_file_text.call_args.args[0] == "devices/pc/log.json" + + def test_skips_its_own_device_id_when_listing(self) -> None: + client = _mock_client( + devices=("pc", "phone"), files={"devices/phone/log.json": "{}"} + ) + + sync_log( + client=client, + device_id="pc", + path_prefix="devices", + local_log={}, + encode=_encode, + decode=_decode, + ) + + client.get_file_text.assert_called_once_with("devices/phone/log.json") + + def test_skips_a_device_with_no_pushed_file_yet(self) -> None: + client = _mock_client(devices=("phone",), files={}) + + merged = sync_log( + client=client, + device_id="pc", + path_prefix="devices", + local_log={}, + encode=_encode, + decode=_decode, + ) + + assert merged == {} + + def test_skips_a_device_whose_pushed_file_is_corrupt( + self, caplog: pytest.LogCaptureFixture + ) -> None: + """An interrupted/truncated push must not crash every other device's + merge -- it is treated the same as a device that hasn't pushed yet. + """ + client = _mock_client( + devices=("phone",), + files={"devices/phone/log.json": "{not valid json"}, + ) + + with caplog.at_level(logging.WARNING): + merged = sync_log( + client=client, + device_id="pc", + path_prefix="devices", + local_log={}, + encode=_encode, + decode=_decode, + ) + + assert merged == {} + assert "Unparsable log" in caplog.text + + def test_merges_in_a_remote_devices_entries( + self, make_hlc: Callable[..., Hlc] + ) -> None: + remote_log = { + "b": Record(id="b", fields={"text": ("from phone", make_hlc(100))}) + } + client = _mock_client( + devices=("phone",), + files={"devices/phone/log.json": _encode(remote_log)}, + ) + + merged = sync_log( + client=client, + device_id="pc", + path_prefix="devices", + local_log={}, + encode=_encode, + decode=_decode, + ) + + assert merged == remote_log + + def test_uses_a_custom_filename_and_commit_message(self) -> None: + client = _mock_client() + + sync_log( + client=client, + device_id="pc", + path_prefix="devices", + local_log={}, + encode=_encode, + decode=_decode, + filename="notes.json", + commit_message="custom message", + ) + + client.put_file_text.assert_called_once_with( + "devices/pc/notes.json", + "{}", + message="custom message", + ) diff --git a/crdt-sync/pyproject.toml b/crdt-sync/pyproject.toml new file mode 100644 index 0000000..056469e --- /dev/null +++ b/crdt-sync/pyproject.toml @@ -0,0 +1,150 @@ +[project] +name = "crdt-sync" +version = "0.1.0" +description = "Shared CRDT merge scheme + GitHub-Contents-API sync transport for cross-device apps" +requires-python = ">=3.10" +dependencies = ["requests"] + +[tool.ruff] +target-version = "py310" +include = ["*.py", "**/*.py"] +exclude = [".git", ".venv", "__pycache__", "build", "dist", ".eggs"] + +[tool.ruff.lint] +select = ["ALL"] +ignore = [ + "D203", + "D213", + "COM812", + "ISC001", + "S603", +] +fixable = ["ALL"] +unfixable = [] + +[tool.ruff.lint.per-file-ignores] +"**/tests/**/*.py" = ["ARG", "D", "PLC0415", "PLR2004", "S101", "SLF001"] +"**/test_*.py" = ["ARG", "D", "PLC0415", "PLR2004", "S101", "SLF001"] +"crdt_sync/_sync.py" = ["PLR0913"] + +[tool.ruff.lint.pydocstyle] +convention = "google" + +[tool.ruff.lint.isort] +force-single-line = false +force-sort-within-sections = true +known-first-party = ["crdt_sync"] + +[tool.ruff.lint.flake8-quotes] +docstring-quotes = "double" +inline-quotes = "double" + +[tool.ruff.lint.flake8-tidy-imports] +ban-relative-imports = "all" + +[tool.ruff.format] +quote-style = "double" +indent-style = "space" +skip-magic-trailing-comma = false +line-ending = "auto" +docstring-code-format = true + +[tool.mypy] +python_version = "3.10" +strict = true +warn_return_any = true +warn_unused_configs = true +disallow_untyped_defs = true +disallow_incomplete_defs = true +check_untyped_defs = true +disallow_untyped_decorators = true +no_implicit_optional = true +warn_redundant_casts = true +warn_unused_ignores = true +warn_no_return = true +warn_unreachable = true +disallow_any_unimported = true +disallow_any_explicit = false +disallow_any_generics = true +disallow_subclassing_any = true +strict_equality = true +extra_checks = true +ignore_missing_imports = true +show_error_codes = true +color_output = true +exclude = [".venv/"] + +[tool.pylint.main] +analyse-fallback-blocks = true +persistent = true +jobs = 0 +py-version = "3.10" +ignore = [".venv", "__pycache__"] +ignore-patterns = [".*\\.pyi$"] + +[tool.pylint.messages_control] +enable = "all" +disable = [] + +[tool.pylint.design] +min-public-methods = 0 +max-module-lines = 1000 +max-attributes = 10 + +[tool.pylint.typecheck] +generated-members = [ + ".*\\.assert_called_once_with", + ".*\\.assert_called_once", + ".*\\.assert_called", + ".*\\.assert_not_called", + ".*\\.assert_any_call", + ".*\\.call_args", + ".*\\.call_args_list", + ".*\\.call_count", +] + +[tool.bandit] +exclude_dirs = ["tests", ".venv"] + +[tool.pytest.ini_options] +testpaths = ["crdt_sync/tests"] +python_files = ["test_*.py", "*_test.py"] +python_classes = ["Test*"] +python_functions = ["test_*"] +addopts = [ + "-v", + "--strict-markers", + "--strict-config", + "-ra", + "--cov=crdt_sync", + "--cov-branch", + "--cov-report=term-missing", + "--cov-report=lcov", +] +filterwarnings = [ + "error", + "ignore::DeprecationWarning", + "default::pytest.PytestUnraisableExceptionWarning", +] + +[tool.coverage.run] +source = ["crdt_sync"] +branch = true +omit = [ + "*/__pycache__/*", + "*/tests/*", + "*/.venv/*", +] + +[tool.coverage.report] +fail_under = 100 +show_missing = true +skip_covered = false +exclude_lines = [ + "pragma: no cover", + "raise NotImplementedError", + "raise AssertionError", + "if TYPE_CHECKING:", + 'if __name__ == "__main__":', +] +partial_branches = ["pragma: no branch"] diff --git a/crdt-sync/requirements.txt b/crdt-sync/requirements.txt new file mode 100644 index 0000000..2e554ba --- /dev/null +++ b/crdt-sync/requirements.txt @@ -0,0 +1,9 @@ +bandit[toml]>=1.7 +mypy>=1.13 +pre-commit>=4.0 +pylint>=3.3 +pytest>=8.0 +pytest-cov>=5.0 +requests>=2.31 +ruff>=0.15 +types-requests>=2.31 From 9a7fd2daae51cbc8ff6d9118f38672f270b3b45d Mon Sep 17 00:00:00 2001 From: Krzysztof Rudnicki Date: Sun, 5 Jul 2026 20:55:03 +0200 Subject: [PATCH 2/2] Add cross-language wire-format fixture test and wrong-shape sync coverage The Python and Dart test suites each only verified round-tripping through their own encoder/decoder, so a Record.to_dict()/toJson() key mismatch between the two (deleted_hlc vs deletedHlc) would have gone undetected despite being the exact thing this shared library exists to prevent. Pin one canonical fixture in both suites so a future divergence fails loudly on at least one side. Also add a test for a valid-JSON-but-wrong-shape remote push, exercising the full (ValueError, KeyError, TypeError) skip path, not just JSON syntax errors. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_017McuA1nAi6jBT1EaS5BrVf --- crdt-sync/crdt_sync/tests/test_record.py | 29 ++++++++++++++++++++++++ crdt-sync/crdt_sync/tests/test_sync.py | 22 ++++++++++++++++++ 2 files changed, 51 insertions(+) diff --git a/crdt-sync/crdt_sync/tests/test_record.py b/crdt-sync/crdt_sync/tests/test_record.py index c372963..787c950 100644 --- a/crdt-sync/crdt_sync/tests/test_record.py +++ b/crdt-sync/crdt_sync/tests/test_record.py @@ -109,3 +109,32 @@ class TestRecordDictRoundTrip: def test_round_trips_a_record_with_no_delete_clock(self) -> None: record = Record(id="x", fields={}, deleted=False, deleted_hlc=None) assert Record.from_dict(record.to_dict()) == record + + +class TestCrossLanguageWireFormat: + """Pins the exact wire shape shared with crdt_sync_dart's Record. + + Values chosen here are duplicated verbatim in crdt_sync_dart's + `test/record_test.dart` (see its ``TestCrossLanguageWireFormat`` + group). If this test and that one both pass, the two languages agree + on the wire format; if only one changes, the two suites diverge and + at least one of them fails -- catching exactly the kind of + key-naming mismatch (``deleted_hlc`` vs ``deletedHlc``) that neither + language's own round-trip test can see on its own. + """ + + def test_matches_the_fixture_shared_with_the_dart_package(self) -> None: + record = Record( + id="abc123", + fields={"text": ("hello", Hlc(wall_time_ms=100, counter=0, node_id="pc"))}, + deleted=True, + deleted_hlc=Hlc(wall_time_ms=1000, counter=0, node_id="node-a"), + ) + expected = { + "id": "abc123", + "fields": {"text": ["hello", "1970-01-01T00:00:00.100Z-0000-pc"]}, + "deleted": True, + "deleted_hlc": "1970-01-01T00:00:01.000Z-0000-node-a", + } + assert record.to_dict() == expected + assert Record.from_dict(expected) == record diff --git a/crdt-sync/crdt_sync/tests/test_sync.py b/crdt-sync/crdt_sync/tests/test_sync.py index 593ad18..45fdbb4 100644 --- a/crdt-sync/crdt_sync/tests/test_sync.py +++ b/crdt-sync/crdt_sync/tests/test_sync.py @@ -117,6 +117,28 @@ class TestSyncLog: assert merged == {} assert "Unparsable log" in caplog.text + def test_skips_a_device_whose_pushed_json_has_the_wrong_shape(self) -> None: + """Valid JSON that isn't a record map (e.g. from an incompatible + writer) must be skipped like corrupt JSON, not crash the whole + sync -- this is what the broad ``(ValueError, KeyError, TypeError)`` + catch in ``_pull_remote_logs`` is for, not just JSON syntax errors. + """ + client = _mock_client( + devices=("phone",), + files={"devices/phone/log.json": '{"a": 5}'}, + ) + + merged = sync_log( + client=client, + device_id="pc", + path_prefix="devices", + local_log={}, + encode=_encode, + decode=_decode, + ) + + assert merged == {} + def test_merges_in_a_remote_devices_entries( self, make_hlc: Callable[..., Hlc] ) -> None: