diff --git a/crdt-sync/crdt_sync/__init__.py b/crdt-sync/crdt_sync/__init__.py index cda73bf..4b94eb3 100644 --- a/crdt-sync/crdt_sync/__init__.py +++ b/crdt-sync/crdt_sync/__init__.py @@ -35,9 +35,12 @@ from crdt_sync._remote import RemoteNotFoundError, RemoteStore, RemoteSyncError from crdt_sync._store import dump_log, load_log, read_log, write_log from crdt_sync._sync import ( FileSyncStateStore, + LogCodec, MemorySyncStateStore, + RevisionTracking, SyncState, SyncStateStore, + SyncTarget, default_revs_path, revision_of, sync_log, @@ -64,6 +67,7 @@ __all__ = [ "GitHubSyncError", "Hlc", "Log", + "LogCodec", "MemoryCredentialStore", "MemorySyncStateStore", "MirrorSyncClient", @@ -72,8 +76,10 @@ __all__ = [ "RemoteStore", "RemoteSyncError", "RepoNotFoundError", + "RevisionTracking", "SyncState", "SyncStateStore", + "SyncTarget", "credential_store_for", "default_revs_path", "dump_log", diff --git a/crdt-sync/crdt_sync/_sync.py b/crdt-sync/crdt_sync/_sync.py index 0ae1307..06d6af1 100644 --- a/crdt-sync/crdt_sync/_sync.py +++ b/crdt-sync/crdt_sync/_sync.py @@ -202,15 +202,27 @@ def _remote_revs(client: RemoteStore, revs_path: str) -> dict[str, str]: return get_string_map(revs_path) +@dataclass(frozen=True) +class _PullContext: + """Everything :func:`_pull_remote_logs` needs for one tick's pull. + + A bundle rather than eight positional parameters: they are always passed + together, from one call site, and several share a type -- exactly the + shape where a positional mix-up is silent. + """ + + client: RemoteStore + own_ids: frozenset[str] + path_prefix: str + filename: str + decode: Callable[[str], Log] + remote_revs: dict[str, str] + state: SyncState + + def _pull_remote_logs( - client: RemoteStore, - own_ids: frozenset[str], - path_prefix: str, - filename: str, - decode: Callable[[str], Log], - remote_revs: dict[str, str], + ctx: _PullContext, seen_revs: dict[str, str], - state: SyncState, ) -> list[Log]: """Return every other device's last-pushed log, skipping this one. @@ -229,11 +241,11 @@ def _pull_remote_logs( tick can skip them. """ remote_logs: list[Log] = [] - for other_device_id in client.list_directory(path_prefix): - if other_device_id in own_ids: + for other_device_id in ctx.client.list_directory(ctx.path_prefix): + if other_device_id in ctx.own_ids: continue - remote_rev = remote_revs.get(other_device_id) - if remote_rev is not None and remote_rev == state.peer_revs.get( + remote_rev = ctx.remote_revs.get(other_device_id) + if remote_rev is not None and remote_rev == ctx.state.peer_revs.get( other_device_id ): # Unchanged since we last merged it, and that merge is already @@ -242,11 +254,13 @@ def _pull_remote_logs( # skipped next tick. seen_revs[other_device_id] = remote_rev continue - text = client.get_file_text(f"{path_prefix}/{other_device_id}/{filename}") + text = ctx.client.get_file_text( + f"{ctx.path_prefix}/{other_device_id}/{ctx.filename}" + ) if text is None: continue try: - remote_logs.append(decode(text)) + remote_logs.append(ctx.decode(text)) except (ValueError, KeyError, TypeError): _logger.warning( "Unparsable log pushed by device %r, skipping", @@ -259,26 +273,9 @@ def _pull_remote_logs( return remote_logs -def sync_log( - *, - client: RemoteStore, - 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", - state_store: SyncStateStore | None = None, - revs_path: str | None = None, - legacy_device_id: str | None = None, -) -> Log: - """Run one full sync tick: pull every other device's log, merge, push. - - Pulls from ``//`` for every - device directory the remote reports under ``path_prefix``, merges each - into ``local_log`` with :func:`crdt_sync.merge_logs`, then pushes this - device's own merged result to ``//``. +@dataclass(frozen=True) +class SyncTarget: + """Which remote, which device, and where its logs live. Args: client: An authenticated :class:`RemoteStore` -- a @@ -288,7 +285,31 @@ def sync_log( log is pushed under. path_prefix: The directory holding one subdirectory per device (e.g. ``"devices"``). - local_log: This device's current full log (including tombstones). + legacy_device_id: The id this device pushed under before migrating to + a persisted uuid. Treated as this device's own for skip-own + purposes, so its pre-migration log is not pulled back and + re-merged as a peer's. Pass ``None`` once the old path has been + reclaimed. + """ + + client: RemoteStore + device_id: str + path_prefix: str + legacy_device_id: str | None = None + + @property + def own_ids(self) -> frozenset[str]: + """Every id whose pushed log belongs to this device.""" + if self.legacy_device_id is None: + return frozenset({self.device_id}) + return frozenset({self.device_id, self.legacy_device_id}) + + +@dataclass(frozen=True) +class LogCodec: + """How a log is turned into pushed text and back. + + Args: 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 @@ -296,17 +317,51 @@ def sync_log( 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. - state_store: Pass one to enable revision tracking, which skips - downloading peers that have not changed and skips pushing a log - that has not changed. Without it the tick behaves exactly as it - always did: fetch everything, push unconditionally. + """ + + encode: Callable[[Log], str] + decode: Callable[[str], Log] + filename: str = _DEFAULT_FILENAME + commit_message: str = "crdt_sync: update log" + + +@dataclass(frozen=True) +class RevisionTracking: + """Optional skip-unchanged behaviour for a tick. + + Omit it entirely and the tick behaves exactly as it always did: fetch + everything, push unconditionally. + + Args: + state_store: Persists what this device last pushed and last merged. revs_path: Where revisions live; defaults to :func:`default_revs_path`. - legacy_device_id: The id this device pushed under before migrating to - a persisted uuid. Treated as this device's own for skip-own - purposes, so its pre-migration log is not pulled back and - re-merged as a peer's. Pass ``None`` once the old path has been - reclaimed. + """ + + state_store: SyncStateStore | None = None + revs_path: str | None = None + + +def sync_log( + target: SyncTarget, + local_log: Log, + codec: LogCodec, + revisions: RevisionTracking | None = None, +) -> Log: + """Run one full sync tick: pull every other device's log, merge, push. + + Pulls from ``//`` for every + device directory the remote 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: + target: Which remote and device this tick is for. + local_log: This device's current full log (including tombstones). + codec: How to serialize and parse a log. + revisions: Pass one to skip downloading peers that have not changed + and skip pushing a log that has not changed. ``None`` fetches + everything and pushes unconditionally. Returns: The merged log, as pushed. @@ -319,45 +374,83 @@ def sync_log( unchanged and never be fetched again. Per-device keys make that failure impossible rather than merely avoided. """ - revs = revs_path if revs_path is not None else default_revs_path(path_prefix) - state = state_store.load() if state_store is not None else SyncState() - remote_revs = _remote_revs(client, revs) if state_store is not None else {} - - own_ids = frozenset( - {device_id} if legacy_device_id is None else {device_id, legacy_device_id} + tracking = revisions if revisions is not None else RevisionTracking() + state_store = tracking.state_store + revs = ( + tracking.revs_path + if tracking.revs_path is not None + else default_revs_path(target.path_prefix) ) - merged = dict(local_log) - seen_revs: dict[str, str] = {} - for remote_log in _pull_remote_logs( - client, - own_ids, - path_prefix, - filename, - decode, - remote_revs, - seen_revs, - state, - ): - merged = merge_logs(merged, remote_log) + state = state_store.load() if state_store is not None else SyncState() + remote_revs = _remote_revs(target.client, revs) if state_store is not None else {} - encoded = encode(merged) + seen_revs: dict[str, str] = {} + merged = _merge_peers( + _PullContext( + client=target.client, + own_ids=target.own_ids, + path_prefix=target.path_prefix, + filename=codec.filename, + decode=codec.decode, + remote_revs=remote_revs, + state=state, + ), + local_log, + seen_revs, + ) + + encoded = codec.encode(merged) revision = revision_of(encoded) - unchanged = state_store is not None and revision == state.pushed_rev - if not unchanged: - client.put_file_text( - f"{path_prefix}/{device_id}/{filename}", + if not (state_store is not None and revision == state.pushed_rev): + _push( + target, + codec, encoded, - message=commit_message, + _RevisionMarker(revs, revision) if state_store is not None else None, ) - if state_store is not None: - # Published after the log, never before: a peer that cached - # "seen rev X" against a log it never received would skip it - # forever. - client.put_file_text( - f"{revs}/{device_id}", - revision, - message=commit_message, - ) if state_store is not None: state_store.save(SyncState(pushed_rev=revision, peer_revs=seen_revs)) return merged + + +def _merge_peers( + ctx: _PullContext, + local_log: Log, + seen_revs: dict[str, str], +) -> Log: + """Return ``local_log`` with every peer's pushed log merged in.""" + merged = dict(local_log) + for remote_log in _pull_remote_logs(ctx, seen_revs): + merged = merge_logs(merged, remote_log) + return merged + + +@dataclass(frozen=True) +class _RevisionMarker: + """The revision node a tracked push publishes after its log.""" + + revs_path: str + revision: str + + +def _push( + target: SyncTarget, + codec: LogCodec, + encoded: str, + marker: _RevisionMarker | None, +) -> None: + """Push this device's merged log, then its revision marker.""" + target.client.put_file_text( + f"{target.path_prefix}/{target.device_id}/{codec.filename}", + encoded, + message=codec.commit_message, + ) + if marker is not None: + # Published after the log, never before: a peer that cached + # "seen rev X" against a log it never received would skip it + # forever. + target.client.put_file_text( + f"{marker.revs_path}/{target.device_id}", + marker.revision, + message=codec.commit_message, + ) diff --git a/crdt-sync/crdt_sync/tests/conftest.py b/crdt-sync/crdt_sync/tests/conftest.py index b930d35..efc6dc3 100644 --- a/crdt-sync/crdt_sync/tests/conftest.py +++ b/crdt-sync/crdt_sync/tests/conftest.py @@ -7,6 +7,7 @@ from typing import TYPE_CHECKING import pytest from crdt_sync import Hlc +from crdt_sync._remote import RemoteSyncError if TYPE_CHECKING: from collections.abc import Callable @@ -20,3 +21,124 @@ def make_hlc() -> Callable[..., Hlc]: return Hlc(wall_time_ms=wall_time_ms, counter=counter, node_id=node_id) return _make + + +class FakeStore: + """An in-memory remote: records its traffic, and can be told to fail. + + One class rather than a near-copy per test module. It is the union of + what those copies needed -- ``reads``/``writes`` for asserting on traffic + rather than only on results, and ``failing`` for driving the error paths + -- because the shared half was the fiddly part (the prefix slicing in + :meth:`list_directory` and :meth:`get_string_map`) and two copies of that + is where they drift. + """ + + def __init__( + self, + files: dict[str, str] | None = None, + *, + failing: bool = False, + ) -> None: + """Start holding ``files``, failing every call if ``failing``.""" + self.files = dict(files or {}) + self.failing = failing + self.reads: list[str] = [] + self.writes: list[str] = [] + + def _guard(self, what: str) -> None: + if self.failing: + msg = f"{what} failed" + raise RemoteSyncError(msg) + + def list_directory(self, path: str) -> list[str]: + """Return the distinct first segments under ``path``.""" + self._guard("list") + prefix = f"{path}/" + return sorted( + { + key[len(prefix) :].split("/")[0] + for key in self.files + if key.startswith(prefix) + } + ) + + def get_file_text(self, path: str) -> str | None: + """Return the stored text, recording the read.""" + self._guard("read") + self.reads.append(path) + return self.files.get(path) + + def put_file_text(self, path: str, text: str, *, message: str) -> None: + """Store ``text``, recording the write.""" + del message + self._guard("write") + self.writes.append(path) + self.files[path] = text + + def get_string_map(self, path: str) -> dict[str, str]: + """Return the direct children of ``path`` as a flat map.""" + self._guard("map read") + prefix = f"{path}/" + return { + key[len(prefix) :]: value + for key, value in self.files.items() + if key.startswith(prefix) + } + + def delete_file(self, path: str, *, message: str = "") -> None: + """Remove ``path`` if present.""" + del message + self._guard("delete") + self.files.pop(path, None) + + def can_access_remote(self) -> bool: + """Report health, which ``failing`` inverts.""" + return not self.failing + + +class FakeStoreWithoutBulkRead: + """A store with no bulk-map read, standing in for GitHub. + + Composes rather than subclasses :class:`FakeStore`, because a subclass + would inherit ``get_string_map`` -- the capability this fake must lack. + """ + + def __init__(self, files: dict[str, str] | None = None) -> None: + """Wrap a plain :class:`FakeStore` holding ``files``.""" + self._inner = FakeStore(files) + + @property + def files(self) -> dict[str, str]: + """The wrapped store's contents.""" + return self._inner.files + + @property + def reads(self) -> list[str]: + """The wrapped store's recorded reads.""" + return self._inner.reads + + @property + def writes(self) -> list[str]: + """The wrapped store's recorded writes.""" + return self._inner.writes + + def list_directory(self, path: str) -> list[str]: + """Delegate to the wrapped store.""" + return self._inner.list_directory(path) + + def get_file_text(self, path: str) -> str | None: + """Delegate to the wrapped store.""" + return self._inner.get_file_text(path) + + def put_file_text(self, path: str, text: str, *, message: str) -> None: + """Delegate to the wrapped store.""" + self._inner.put_file_text(path, text, message=message) + + def delete_file(self, path: str, *, message: str = "") -> None: + """Delegate to the wrapped store.""" + self._inner.delete_file(path, message=message) + + def can_access_remote(self) -> bool: + """Delegate to the wrapped store.""" + return self._inner.can_access_remote() diff --git a/crdt-sync/crdt_sync/tests/test_config.py b/crdt-sync/crdt_sync/tests/test_config.py index a441c08..dc42ce9 100644 --- a/crdt-sync/crdt_sync/tests/test_config.py +++ b/crdt-sync/crdt_sync/tests/test_config.py @@ -6,6 +6,11 @@ newline a text editor appended), so each is asserted to name the field at fault rather than merely to raise. """ +# These tests assert on private attributes on purpose: `_timeout_seconds` +# and `_path` are exactly what the fixes under test set, and there is no +# public accessor for either. +# pylint: disable=protected-access + from __future__ import annotations import json diff --git a/crdt-sync/crdt_sync/tests/test_mirror.py b/crdt-sync/crdt_sync/tests/test_mirror.py index 7b8402e..98bf643 100644 --- a/crdt-sync/crdt_sync/tests/test_mirror.py +++ b/crdt-sync/crdt_sync/tests/test_mirror.py @@ -14,115 +14,17 @@ import pytest from crdt_sync import ( Hlc, Log, + LogCodec, MirrorSyncClient, Record, RemoteStore, RemoteSyncError, + SyncTarget, dump_log, load_log, sync_log, ) - - -class FakeStore: - """A scriptable in-memory store that can be told to fail.""" - - def __init__( - self, - files: dict[str, str] | None = None, - *, - failing: bool = False, - ) -> None: - """Start holding ``files``, failing every call if ``failing``.""" - self.files = dict(files or {}) - self.failing = failing - self.writes: list[str] = [] - - def _guard(self, what: str) -> None: - if self.failing: - msg = f"{what} failed" - raise RemoteSyncError(msg) - - def list_directory(self, path: str) -> list[str]: - """Return the distinct first segments under ``path``.""" - self._guard("list") - prefix = f"{path}/" - return sorted( - { - key[len(prefix) :].split("/")[0] - for key in self.files - if key.startswith(prefix) - } - ) - - def get_file_text(self, path: str) -> str | None: - """Return the stored text, if any.""" - self._guard("read") - return self.files.get(path) - - def put_file_text(self, path: str, text: str, *, message: str) -> None: - """Store ``text`` at ``path``.""" - del message - self._guard("write") - self.writes.append(path) - self.files[path] = text - - def get_string_map(self, path: str) -> dict[str, str]: - """Return the direct children of ``path`` as a flat map.""" - self._guard("map read") - prefix = f"{path}/" - return { - key[len(prefix) :]: value - for key, value in self.files.items() - if key.startswith(prefix) - } - - def delete_file(self, path: str, *, message: str = "") -> None: - """Remove ``path`` if present.""" - del message - self._guard("delete") - self.files.pop(path, None) - - def can_access_remote(self) -> bool: - """Return whether this backend is currently healthy.""" - return not self.failing - - -class FakeStoreWithoutBulkRead: - """A store with no bulk-map read, standing in for GitHub. - - Composes rather than subclasses :class:`FakeStore`, because a subclass - would inherit ``get_string_map`` -- the capability this fake must lack. - """ - - def __init__(self, files: dict[str, str] | None = None) -> None: - """Wrap a plain :class:`FakeStore` holding ``files``.""" - self._inner = FakeStore(files) - - @property - def files(self) -> dict[str, str]: - """The wrapped store's contents.""" - return self._inner.files - - def list_directory(self, path: str) -> list[str]: - """Delegate to the wrapped store.""" - return self._inner.list_directory(path) - - def get_file_text(self, path: str) -> str | None: - """Delegate to the wrapped store.""" - return self._inner.get_file_text(path) - - def put_file_text(self, path: str, text: str, *, message: str) -> None: - """Delegate to the wrapped store.""" - self._inner.put_file_text(path, text, message=message) - - def delete_file(self, path: str, *, message: str = "") -> None: - """Delegate to the wrapped store.""" - self._inner.delete_file(path, message=message) - - def can_access_remote(self) -> bool: - """Delegate to the wrapped store.""" - return self._inner.can_access_remote() +from crdt_sync.tests.conftest import FakeStore, FakeStoreWithoutBulkRead def _mirror( @@ -433,20 +335,28 @@ class TestEndToEndThroughSyncLog: } sync_log( - client=github, - device_id="phone", - path_prefix="ns/devices", - local_log=log("from-phone", "node-phone"), - encode=dump_log, - decode=load_log, + SyncTarget( + client=github, + device_id="phone", + path_prefix="ns/devices", + ), + log("from-phone", "node-phone"), + LogCodec( + decode=load_log, + encode=dump_log, + ), ) merged = sync_log( - client=MirrorSyncClient(firebase, github), - device_id="pc", - path_prefix="ns/devices", - local_log=log("from-pc", "node-pc"), - encode=dump_log, - decode=load_log, + SyncTarget( + client=MirrorSyncClient(firebase, github), + device_id="pc", + path_prefix="ns/devices", + ), + log("from-pc", "node-pc"), + LogCodec( + decode=load_log, + encode=dump_log, + ), ) assert {"from-phone", "from-pc"} <= set(merged) # The pc's merged result is mirrored back to GitHub, so the diff --git a/crdt-sync/crdt_sync/tests/test_record.py b/crdt-sync/crdt_sync/tests/test_record.py index ad352be..7b3e9df 100644 --- a/crdt-sync/crdt_sync/tests/test_record.py +++ b/crdt-sync/crdt_sync/tests/test_record.py @@ -94,6 +94,9 @@ class TestMergeRecord: 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) + # Swapping the operands is the property under test: merge_record + # must give the same answer either way round. + # pylint: disable-next=arguments-out-of-order assert merge_record(b, a).deleted_hlc == make_hlc(200) def test_is_commutative(self, make_hlc: Callable[..., Hlc]) -> None: diff --git a/crdt-sync/crdt_sync/tests/test_sync.py b/crdt-sync/crdt_sync/tests/test_sync.py index 6bfe908..2ca9bc6 100644 --- a/crdt-sync/crdt_sync/tests/test_sync.py +++ b/crdt-sync/crdt_sync/tests/test_sync.py @@ -7,7 +7,13 @@ import logging from typing import TYPE_CHECKING from unittest.mock import MagicMock -from crdt_sync import GitHubSyncClient, Record, sync_log +from crdt_sync import ( + GitHubSyncClient, + LogCodec, + Record, + SyncTarget, + sync_log, +) if TYPE_CHECKING: from collections.abc import Callable @@ -54,12 +60,16 @@ class TestSyncLog: client = _mock_client() merged = sync_log( - client=client, - device_id="pc", - path_prefix="devices", - local_log=local_log, - encode=_encode, - decode=_decode, + SyncTarget( + client=client, + device_id="pc", + path_prefix="devices", + ), + local_log, + LogCodec( + decode=_decode, + encode=_encode, + ), ) assert merged == local_log @@ -73,12 +83,16 @@ class TestSyncLog: ) sync_log( - client=client, - device_id="pc", - path_prefix="devices", - local_log={}, - encode=_encode, - decode=_decode, + SyncTarget( + client=client, + device_id="pc", + path_prefix="devices", + ), + {}, + LogCodec( + decode=_decode, + encode=_encode, + ), ) client.get_file_text.assert_called_once_with("devices/phone/log.json") @@ -95,13 +109,17 @@ class TestSyncLog: ) sync_log( - client=client, - device_id="new-uuid", - path_prefix="devices", - local_log={}, - encode=_encode, - decode=_decode, - legacy_device_id="pc", + SyncTarget( + client=client, + device_id="new-uuid", + legacy_device_id="pc", + path_prefix="devices", + ), + {}, + LogCodec( + decode=_decode, + encode=_encode, + ), ) client.get_file_text.assert_called_once_with("devices/phone/log.json") @@ -113,12 +131,16 @@ class TestSyncLog: ) sync_log( - client=client, - device_id="new-uuid", - path_prefix="devices", - local_log={}, - encode=_encode, - decode=_decode, + SyncTarget( + client=client, + device_id="new-uuid", + path_prefix="devices", + ), + {}, + LogCodec( + decode=_decode, + encode=_encode, + ), ) client.get_file_text.assert_called_once_with("devices/pc/log.json") @@ -128,12 +150,16 @@ class TestSyncLog: client = _mock_client(devices=("phone",), files={}) merged = sync_log( - client=client, - device_id="pc", - path_prefix="devices", - local_log={}, - encode=_encode, - decode=_decode, + SyncTarget( + client=client, + device_id="pc", + path_prefix="devices", + ), + {}, + LogCodec( + decode=_decode, + encode=_encode, + ), ) assert not merged @@ -151,12 +177,16 @@ class TestSyncLog: with caplog.at_level(logging.WARNING): merged = sync_log( - client=client, - device_id="pc", - path_prefix="devices", - local_log={}, - encode=_encode, - decode=_decode, + SyncTarget( + client=client, + device_id="pc", + path_prefix="devices", + ), + {}, + LogCodec( + decode=_decode, + encode=_encode, + ), ) assert not merged @@ -174,12 +204,16 @@ class TestSyncLog: ) merged = sync_log( - client=client, - device_id="pc", - path_prefix="devices", - local_log={}, - encode=_encode, - decode=_decode, + SyncTarget( + client=client, + device_id="pc", + path_prefix="devices", + ), + {}, + LogCodec( + decode=_decode, + encode=_encode, + ), ) assert not merged @@ -197,12 +231,16 @@ class TestSyncLog: ) merged = sync_log( - client=client, - device_id="pc", - path_prefix="devices", - local_log={}, - encode=_encode, - decode=_decode, + SyncTarget( + client=client, + device_id="pc", + path_prefix="devices", + ), + {}, + LogCodec( + decode=_decode, + encode=_encode, + ), ) assert merged == remote_log @@ -212,14 +250,18 @@ class TestSyncLog: 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", + SyncTarget( + client=client, + device_id="pc", + path_prefix="devices", + ), + {}, + LogCodec( + commit_message="custom message", + decode=_decode, + encode=_encode, + filename="notes.json", + ), ) client.put_file_text.assert_called_once_with( diff --git a/crdt-sync/crdt_sync/tests/test_sync_revisions.py b/crdt-sync/crdt_sync/tests/test_sync_revisions.py index b7d0ccf..f1521f2 100644 --- a/crdt-sync/crdt_sync/tests/test_sync_revisions.py +++ b/crdt-sync/crdt_sync/tests/test_sync_revisions.py @@ -19,15 +19,19 @@ from crdt_sync import ( FileSyncStateStore, Hlc, Log, + LogCodec, MemorySyncStateStore, Record, + RevisionTracking, SyncState, + SyncTarget, default_revs_path, dump_log, load_log, revision_of, sync_log, ) +from crdt_sync.tests.conftest import FakeStore, FakeStoreWithoutBulkRead if TYPE_CHECKING: from pathlib import Path @@ -50,107 +54,8 @@ _encode = dump_log _decode = load_log -class FakeRemote: - """An in-memory store that also serves revision maps in one read. - - Stands in for :class:`crdt_sync.FirebaseSyncClient`, counting every call - so tests can assert on traffic rather than only on results. - """ - - def __init__(self, files: dict[str, str] | None = None) -> None: - """Start holding ``files``, with empty call logs.""" - self.files = dict(files or {}) - self.reads: list[str] = [] - self.writes: list[str] = [] - - def list_directory(self, path: str) -> list[str]: - """Return the distinct first segments under ``path``.""" - prefix = f"{path}/" - return sorted( - { - key[len(prefix) :].split("/")[0] - for key in self.files - if key.startswith(prefix) - } - ) - - def get_file_text(self, path: str) -> str | None: - """Return the stored text, recording the read.""" - self.reads.append(path) - return self.files.get(path) - - def put_file_text(self, path: str, text: str, *, message: str) -> None: - """Store ``text``, recording the write.""" - del message - self.writes.append(path) - self.files[path] = text - - def get_string_map(self, path: str) -> dict[str, str]: - """Return the direct children of ``path`` as a flat map.""" - prefix = f"{path}/" - return { - key[len(prefix) :]: value - for key, value in self.files.items() - if key.startswith(prefix) - } - - def delete_file(self, path: str, *, message: str = "") -> None: - """Remove ``path`` if present.""" - del message - self.files.pop(path, None) - - def can_access_remote(self) -> bool: - """Always reachable.""" - return True - - -class FakeRemoteWithoutBulkRead: - """A store with no bulk-map read, standing in for GitHub. - - Revision tracking must degrade to "fetch everything", not break. - Composes rather than subclasses :class:`FakeRemote`, because a subclass - would inherit ``get_string_map`` -- the very capability this fake exists - to lack. - """ - - def __init__(self, files: dict[str, str] | None = None) -> None: - """Wrap a plain :class:`FakeRemote` holding ``files``.""" - self._inner = FakeRemote(files) - - @property - def files(self) -> dict[str, str]: - """The wrapped store's contents.""" - return self._inner.files - - @property - def reads(self) -> list[str]: - """The wrapped store's read log.""" - return self._inner.reads - - @property - def writes(self) -> list[str]: - """The wrapped store's write log.""" - return self._inner.writes - - def list_directory(self, path: str) -> list[str]: - """Delegate to the wrapped store.""" - return self._inner.list_directory(path) - - def get_file_text(self, path: str) -> str | None: - """Delegate to the wrapped store.""" - return self._inner.get_file_text(path) - - def put_file_text(self, path: str, text: str, *, message: str) -> None: - """Delegate to the wrapped store.""" - self._inner.put_file_text(path, text, message=message) - - def delete_file(self, path: str, *, message: str = "") -> None: - """Delegate to the wrapped store.""" - self._inner.delete_file(path, message=message) - - def can_access_remote(self) -> bool: - """Delegate to the wrapped store.""" - return self._inner.can_access_remote() +FakeRemote = FakeStore +FakeRemoteWithoutBulkRead = FakeStoreWithoutBulkRead def _tick( @@ -160,13 +65,19 @@ def _tick( device_id: str = "pc", ) -> Log: return sync_log( - client=remote, - device_id=device_id, - path_prefix="ns/devices", - local_log=local or {}, - encode=_encode, - decode=_decode, - state_store=store, + SyncTarget( + client=remote, + device_id=device_id, + path_prefix="ns/devices", + ), + local or {}, + LogCodec( + decode=_decode, + encode=_encode, + ), + RevisionTracking( + state_store=store, + ), ) @@ -398,14 +309,20 @@ class TestRevisionPublishingOrder: """An explicit revs path overrides the default.""" remote = FakeRemote() sync_log( - client=remote, - device_id="pc", - path_prefix="ns/devices", - local_log=_log("a", "1"), - encode=_encode, - decode=_decode, - state_store=MemorySyncStateStore(), - revs_path="custom/place", + SyncTarget( + client=remote, + device_id="pc", + path_prefix="ns/devices", + ), + _log("a", "1"), + LogCodec( + decode=_decode, + encode=_encode, + ), + RevisionTracking( + revs_path="custom/place", + state_store=MemorySyncStateStore(), + ), ) assert "custom/place/pc" in remote.writes @@ -419,13 +336,19 @@ class TestTwoDevicesConverge: pc_store, phone_store = MemorySyncStateStore(), MemorySyncStateStore() pc = _tick(remote, pc_store, _log("a", "from-pc", "node-pc")) phone = sync_log( - client=remote, - device_id="phone", - path_prefix="ns/devices", - local_log=_log("b", "from-phone", "node-phone"), - encode=_encode, - decode=_decode, - state_store=phone_store, + SyncTarget( + client=remote, + device_id="phone", + path_prefix="ns/devices", + ), + _log("b", "from-phone", "node-phone"), + LogCodec( + decode=_decode, + encode=_encode, + ), + RevisionTracking( + state_store=phone_store, + ), ) assert {"a", "b"} <= set(phone) pc_after = _tick(remote, pc_store, pc) diff --git a/crdt-sync/pyproject.toml b/crdt-sync/pyproject.toml index 4b1f93d..4cbecf2 100644 --- a/crdt-sync/pyproject.toml +++ b/crdt-sync/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "crdt-sync" -version = "0.8.1" +version = "0.9.0" description = "Shared CRDT merge scheme + GitHub-Contents-API sync transport for cross-device apps" requires-python = ">=3.10" dependencies = ["requests"] diff --git a/crdt-sync/tool/interop_seed.py b/crdt-sync/tool/interop_seed.py index c84fe5d..2fa986f 100644 --- a/crdt-sync/tool/interop_seed.py +++ b/crdt-sync/tool/interop_seed.py @@ -27,8 +27,11 @@ import sys from crdt_sync import ( FirebaseConfig, Hlc, + LogCodec, MemorySyncStateStore, Record, + RevisionTracking, + SyncTarget, dump_log, firebase_client_for, load_log, @@ -52,10 +55,12 @@ def main() -> int: client = firebase_client_for("interop", config=FirebaseConfig.load()) merged = sync_log( - client=client, - device_id=_DEVICE_ID, - path_prefix=_PATH_PREFIX, - local_log={ + SyncTarget( + client=client, + device_id=_DEVICE_ID, + path_prefix=_PATH_PREFIX, + ), + { "py-rec": Record( id="py-rec", fields={ @@ -66,9 +71,13 @@ def main() -> int: }, ) }, - encode=dump_log, - decode=load_log, - state_store=MemorySyncStateStore(), + LogCodec( + decode=load_log, + encode=dump_log, + ), + RevisionTracking( + state_store=MemorySyncStateStore(), + ), ) _logger.info("python sees records: %s", sorted(merged))