From 850833614dbbbadb2b378c0a032511fdc06ea5e3 Mon Sep 17 00:00:00 2001 From: Pedro Werneck Date: Tue, 3 Nov 2026 09:51:57 -0300 Subject: [PATCH 1/5] feat: save DS messages before upload and send them at the next sync - The DS saves each message to private/spool// before any network call, and removes it once the DO has it. The upload downgrades and encrypts the message, so it uses the DO's protocol and key at that time. - A push and sync() upload the saved messages oldest first, under a lock, so that the notebook and syft-bg do not upload a message twice. P2PConnection.is_delivered() finds a message that an upload sent before the process stopped, so it is not sent again. - The file hashes advance when a message is saved, not when it is sent, so a second edit made offline builds on the first. - A message to a peer that is not ready stays saved. The push raises and says so, and sync() warns for each peer that waits. The "not known" remedy no longer says "send again". - A spool entry carries a version, and an entry of a later version is refused. --- syft/sync/connections/connection_router.py | 4 + syft/sync/connections/drive/drive_p2p.py | 12 ++ syft/sync/connections/protocols.py | 9 ++ syft/sync/peers/peer_store.py | 3 +- syft/sync/syftbox_manager.py | 1 + syft/sync/sync/datasite_watcher_syncer.py | 116 +++++++++++------- syft/sync/sync/spool.py | 114 ++++++++++++++++++ tests/unit/test_encryption_policy.py | 12 +- tests/unit/test_spool.py | 129 +++++++++++++++++++++ 9 files changed, 351 insertions(+), 49 deletions(-) create mode 100644 syft/sync/sync/spool.py create mode 100644 tests/unit/test_spool.py diff --git a/syft/sync/connections/connection_router.py b/syft/sync/connections/connection_router.py index 851bf4ea828..006f0103ee8 100644 --- a/syft/sync/connections/connection_router.py +++ b/syft/sync/connections/connection_router.py @@ -176,6 +176,10 @@ def mark_all_processed(self, messages: list[RawMessage]) -> None: """Finish ``messages``, as received by either role.""" self.connections.p2p_con.mark_all_processed(messages) + def is_delivered(self, peer_email: str, msg_id: str) -> bool: + """Whether the message sent to ``peer_email`` under ``msg_id`` reached it.""" + return self.connections.p2p_con.is_delivered(peer_email, msg_id) + def owner_write_event_messages_to_outbox( self, recipient_email: str, events_message: FileChangeEventsMessage ) -> SendStatus: diff --git a/syft/sync/connections/drive/drive_p2p.py b/syft/sync/connections/drive/drive_p2p.py index 825d5e2d4ac..a7a0dfbd009 100644 --- a/syft/sync/connections/drive/drive_p2p.py +++ b/syft/sync/connections/drive/drive_p2p.py @@ -109,6 +109,18 @@ def delivered(self) -> list[str]: # Every Drive send is DELIVERED when it returns; none waits for a peer. return [] + @raise_unavailable("Could not check Drive for a sent message.") + def is_delivered(self, peer: str, msg_id: str) -> bool: + # The sender owns the file it uploads, also after the peer moves it to + # an archive, so a search of my own files by name finds it in any folder. + query = f"name='{msg_id}' and 'me' in owners and trashed=false" + results = execute_with_retries( + self._cached_client.client.drive_service.files().list( + q=query, fields="files(id)", pageSize=1 + ) + ) + return bool(results["files"]) + def receive( self, peer: str, diff --git a/syft/sync/connections/protocols.py b/syft/sync/connections/protocols.py index 6b744386c54..c3b3a1cfce6 100644 --- a/syft/sync/connections/protocols.py +++ b/syft/sync/connections/protocols.py @@ -75,6 +75,15 @@ def delivered(self) -> list[str]: """The ``msg_id`` of each queued message the peer stored since the last call. A connection that only returns ``DELIVERED`` returns nothing.""" + @abstractmethod + def is_delivered(self, peer: str, msg_id: str) -> bool: + """Whether the message sent to ``peer`` under ``msg_id`` is stored where + the peer finds it, or was processed there. + + A sender that stopped between a send and its record of the send asks + this before it sends the message again. + """ + @abstractmethod def receive( self, diff --git a/syft/sync/peers/peer_store.py b/syft/sync/peers/peer_store.py index 3ffac197b9d..034cf8a88e1 100644 --- a/syft/sync/peers/peer_store.py +++ b/syft/sync/peers/peer_store.py @@ -264,8 +264,7 @@ def _peer_for_message(self, email: str) -> Peer: email, cause=f"The encryption policy of {email} is not known yet.", remedy=( - "Peer with them and wait until they accept. Then run " - "client.sync() and send again." + "Peer with them and wait until they accept. Then run client.sync()." ), ) return peer diff --git a/syft/sync/syftbox_manager.py b/syft/sync/syftbox_manager.py index fbe7d2ce2a4..85d8dc47f15 100644 --- a/syft/sync/syftbox_manager.py +++ b/syft/sync/syftbox_manager.py @@ -1121,6 +1121,7 @@ def sync( self.try_create_checkpoint(checkpoint_threshold) if self.has_ds_role: + self.datasite_watcher_syncer.sync_up() peer_emails = [peer.email for peer in self.peer_manager.syncable_peers] self.peer_manager.warn_if_all_peers_incompatible(peer_emails) self.datasite_watcher_syncer.sync_down(peer_emails) diff --git a/syft/sync/sync/datasite_watcher_syncer.py b/syft/sync/sync/datasite_watcher_syncer.py index 5ff8df81435..db0af14bbed 100644 --- a/syft/sync/sync/datasite_watcher_syncer.py +++ b/syft/sync/sync/datasite_watcher_syncer.py @@ -1,26 +1,28 @@ -from pathlib import Path +import warnings from concurrent.futures import ThreadPoolExecutor +from pathlib import Path from queue import Queue -from typing import Dict, List +from typing import Dict, Iterable, List from pydantic import BaseModel, Field, PrivateAttr -from syft.sync.connections.connection_router import ConnectionRouter from syft.sync.callback_mixin import BaseModelCallbackMixin -from syft.sync.peers.exceptions import SyftPeerNotReadyError +from syft.sync.connections.connection_router import ConnectionRouter from syft.sync.messages.proposed_filechange import ( ProposedFileChange, ProposedFileChangesMessage, ) +from syft.sync.peers.exceptions import SyftPeerNotReadyError from syft.sync.sync.caches.datasite_watcher_cache import ( DataSiteWatcherCache, DataSiteWatcherCacheConfig, ) +from syft.sync.sync.spool import Spool, datasite_spool_path class DatasiteWatcherSyncerConfig(BaseModel): - syftbox_folder: Path | None = None - email: str | None = None + syftbox_folder: Path + email: str datasite_watcher_cache_config: DataSiteWatcherCacheConfig = Field( default_factory=DataSiteWatcherCacheConfig ) @@ -36,6 +38,7 @@ class Config: email: str connection_router: ConnectionRouter datasite_watcher_cache: DataSiteWatcherCache + spool: Spool queue: Queue = Field(default_factory=Queue) _executor: ThreadPoolExecutor = PrivateAttr( @@ -57,6 +60,7 @@ def from_config( datasite_watcher_cache=DataSiteWatcherCache.from_config( config.datasite_watcher_cache_config, connection_router ), + spool=Spool(datasite_spool_path(config.syftbox_folder, config.email)), ) # --- Pushing proposed file changes --- @@ -79,6 +83,14 @@ def get_proposed_file_change_object( ) def process_file_changes_queue(self): + """Save the queued changes as one message per datasite, then upload the + saved messages to those datasites. + + Raises: + SyftPeerNotReadyError: when a recipient cannot receive yet. Its + message stays saved, and ``sync_up`` sends it later. The other + recipients still get their messages. + """ # One message per datasite: each change goes to the owner of its path. changes_by_recipient: Dict[str, List[ProposedFileChange]] = {} while not self.queue.empty(): @@ -108,34 +120,24 @@ def process_file_changes_queue(self): path_in_datasite, content, datasite_email=recipient ) changes_by_recipient.setdefault(recipient, []).append(file_change) - self._send_by_recipient(changes_by_recipient) - - def _send_by_recipient( - self, changes_by_recipient: dict[str, list[ProposedFileChange]] - ) -> None: - """Send one message to each recipient. - - The changes for a recipient that cannot receive now are dropped. The - other recipients still get their changes, and then the first error is - raised. - - Raises: - SyftPeerNotReadyError: when the policy of a recipient is not known, - or the pair encrypts and the peer bundle is missing. - """ - error = None for recipient, changes in changes_by_recipient.items(): message = ProposedFileChangesMessage( sender_email=self.email, proposed_file_changes=changes ) - try: - self._send_message(recipient, message) - except SyftPeerNotReadyError as e: - error = error or e - if error is not None: - raise error - - def _record_sent_hashes(self, message: ProposedFileChangesMessage) -> None: + self.spool.add(recipient, message) + # The next change to a path builds on this one, sent or not. + self._record_proposed_hashes(message) + waiting = self._upload_spooled(changes_by_recipient) + if waiting: + error = next(iter(waiting.values())) + raise SyftPeerNotReadyError( + error.peer_email, + cause=error.cause, + remedy=f"{error.remedy} The change is saved, and client.sync() " + "sends it once the peer is ready.", + ) from error + + def _record_proposed_hashes(self, message: ProposedFileChangesMessage) -> None: hashes = self.datasite_watcher_cache.file_hashes for fc in message.proposed_file_changes: path_key = Path(fc.datasite_email) / fc.path_in_datasite @@ -144,19 +146,48 @@ def _record_sent_hashes(self, message: ProposedFileChangesMessage) -> None: else: hashes[path_key] = fc.new_hash - def _send_message( - self, recipient: str, message: ProposedFileChangesMessage - ) -> None: - """Send ``message`` and record the hashes it sent. + def _upload_spooled( + self, recipients: Iterable[str] + ) -> dict[str, SyftPeerNotReadyError]: + """Upload the saved messages to each of ``recipients``, oldest first, + and remove each message once its recipient has it. + + Returns why the messages to each recipient that cannot receive yet + wait, by recipient email. They stay saved, in order. Raises: - SyftPeerNotReadyError: when the policy of ``recipient`` is not - known, or the pair encrypts and the peer bundle is missing. + ConnectionUnavailableError: when the backend cannot be reached. An + upload that fails for another reason raises too. That message + and the later ones stay saved. """ - self.connection_router.watcher_send_proposed_file_changes_message( - recipient, message - ) - self._record_sent_hashes(message) + waiting = {} + with self.spool.locked(): + for recipient in recipients: + try: + self._upload_spooled_to(recipient) + except SyftPeerNotReadyError as e: + waiting[recipient] = e + return waiting + + def _upload_spooled_to(self, recipient: str) -> None: + for message in self.spool.pending(recipient): + msg_id = message.message_filename.as_string() + # An earlier upload may have stopped before it removed the message. + if not self.connection_router.is_delivered(recipient, msg_id): + self.connection_router.watcher_send_proposed_file_changes_message( + recipient, message + ) + self.spool.remove(recipient, msg_id) + + def sync_up(self) -> None: + """Upload every saved message. Warn for each recipient that cannot + receive yet; its messages wait for the next sync.""" + waiting = self._upload_spooled(self.spool.recipients()) + for recipient, error in waiting.items(): + warnings.warn( + f"Changes to {recipient} wait in " + f"{self.spool.recipient_folder(recipient)}: {error}" + ) def on_file_change( self, @@ -193,8 +224,7 @@ def sync_down(self, peer_emails: list[str]): self.datasite_watcher_cache.sync_down_collections_parallel( peer_email, self._executor, - lambda fid, - pe=peer_email: self.download_collection_file_with_new_connection( - fid, pe + lambda fid, pe=peer_email: ( + self.download_collection_file_with_new_connection(fid, pe) ), ) diff --git a/syft/sync/sync/spool.py b/syft/sync/sync/spool.py new file mode 100644 index 00000000000..4cd1f45efba --- /dev/null +++ b/syft/sync/sync/spool.py @@ -0,0 +1,114 @@ +"""The messages a DS saves before it sends them. + +A dropped connection or a stopped kernel must not lose a job submission. Each +message to a peer is saved to ``//private/spool//`` before +any network call, and it is removed when the peer has it. ``private/`` is never +synced. + +A saved message is the message before the downgrade and the encryption for its +peer. The upload does both, so it uses the protocol and the key that the peer +has at that time. An enclave gets a new key at every boot. +""" + +import json +from contextlib import AbstractContextManager +from pathlib import Path + +from pydantic import BaseModel, ConfigDict + +from syft.sync.messages.proposed_filechange import ProposedFileChangesMessage +from syft.sync.peers.peer_store import PRIVATE_DIR_NAME +from syft.sync.sync.caches.persisted_dict import locked_file, write_text_atomic + +SPOOL_DIR_NAME = "spool" +SPOOL_ENTRY_SUFFIX = ".json" +# Format of a spool entry. Raise it when the layout of an entry changes. An +# entry of a later version is refused. +SPOOL_ENTRY_VERSION = 1 + + +def datasite_spool_path(syftbox_folder: Path | str, email: str) -> Path: + """Per-datasite spool: ``//private/spool``.""" + return Path(syftbox_folder) / email / PRIVATE_DIR_NAME / SPOOL_DIR_NAME + + +class SpoolEntry(BaseModel): + """One saved message.""" + + model_config = ConfigDict( + frozen=True, + extra="forbid", + ser_json_bytes="base64", + val_json_bytes="base64", + ) + + version: int + # The compressed message, before the downgrade and the encryption for its peer. + message: bytes + + @classmethod + def read(cls, path: Path) -> "SpoolEntry": + """Read the entry at ``path``. + + Raises: + ValueError: if a later client wrote the entry, or it has another + shape. + """ + text = path.read_text() + data = json.loads(text) + # Checked before the fields, so a later layout is refused by its version. + if data["version"] > SPOOL_ENTRY_VERSION: + raise ValueError( + f"The spool entry at {path} has version {data['version']}, and " + f"this client reads up to version {SPOOL_ENTRY_VERSION}." + ) + return cls.model_validate_json(text) + + +class Spool: + """The saved messages of one DS, in one folder per recipient. An entry is + named after the ``msg_id`` its message gets.""" + + def __init__(self, folder: Path): + self.folder = folder + + def add(self, recipient: str, message: ProposedFileChangesMessage) -> None: + """Save ``message`` to ``recipient``.""" + entry = SpoolEntry( + version=SPOOL_ENTRY_VERSION, message=message.as_compressed_data() + ) + msg_id = message.message_filename.as_string() + write_text_atomic(self._entry_path(recipient, msg_id), entry.model_dump_json()) + + def recipients(self) -> list[str]: + """The recipients that have a folder in the spool.""" + if not self.folder.is_dir(): + return [] + return sorted(p.name for p in self.folder.iterdir() if p.is_dir()) + + def pending(self, recipient: str) -> list[ProposedFileChangesMessage]: + """The saved messages to ``recipient``, oldest first. + + Raises: + ValueError: if an entry is of a later version, or has another shape. + """ + paths = self.recipient_folder(recipient).glob(f"*{SPOOL_ENTRY_SUFFIX}") + messages = [ + ProposedFileChangesMessage.from_compressed_data(SpoolEntry.read(p).message) + for p in paths + ] + return sorted(messages, key=lambda m: m.message_filename.submitted_timestamp) + + def remove(self, recipient: str, msg_id: str) -> None: + """Remove the saved message ``msg_id`` to ``recipient``.""" + self._entry_path(recipient, msg_id).unlink() + + def locked(self) -> AbstractContextManager[None]: + """A lock that one process at a time holds while it uploads.""" + return locked_file(self.folder) + + def recipient_folder(self, recipient: str) -> Path: + return self.folder / recipient + + def _entry_path(self, recipient: str, msg_id: str) -> Path: + return self.recipient_folder(recipient) / f"{msg_id}{SPOOL_ENTRY_SUFFIX}" diff --git a/tests/unit/test_encryption_policy.py b/tests/unit/test_encryption_policy.py index 2d3aa2c4fdd..09a15a30e73 100644 --- a/tests/unit/test_encryption_policy.py +++ b/tests/unit/test_encryption_policy.py @@ -491,8 +491,9 @@ def _queued_changes(manager) -> int: @pytest.mark.parametrize("do_policy", [OPTIONAL, REQUIRED]) -def test_send_before_owner_accepts_raises(do_policy): - """Before the owner accepts, its policy is not known, so nothing is sent.""" +def test_send_before_owner_accepts_goes_once_accepted(do_policy): + """Before the owner accepts, its policy is not known. The change is saved, + and the first sync after the owner accepts sends it.""" ds_manager, do_manager = SyftboxManager.pair_with_mock_drive_service_connection( add_peers=False, do_encryption_policy=do_policy, @@ -500,15 +501,18 @@ def test_send_before_owner_accepts_raises(do_policy): ds_manager.add_peer(do_manager.email, sync=False) file_path = path_for_job(do_manager.email, ds_manager.email, "my.job") - with pytest.raises(SyftPeerNotReadyError, match="not known"): + with pytest.raises(SyftPeerNotReadyError, match="not known.*is saved"): ds_manager._send_file_change(file_path, "early content") assert _queued_changes(ds_manager) == 0 + with pytest.warns(UserWarning, match="wait in"): + ds_manager.sync() do_manager.load_peers() do_manager.approve_peer_request(ds_manager.email) grant_job_inbox_access(do_manager, ds_manager.email) + ds_manager.sync() do_manager.sync() - assert not any("early content" in c for c in _accepted_contents(do_manager)) + assert any("early content" in c for c in _accepted_contents(do_manager)) def test_send_to_non_peer_raises(): diff --git a/tests/unit/test_spool.py b/tests/unit/test_spool.py new file mode 100644 index 00000000000..ddc4136cef1 --- /dev/null +++ b/tests/unit/test_spool.py @@ -0,0 +1,129 @@ +"""The DS saves each message before it sends it, so a dropped connection or a +stopped kernel does not lose a submission.""" + +import json + +import pytest + +from syft.sync.connections.drive.drive_p2p import DriveP2PConnection +from syft.sync.connections.exceptions import ConnectionUnavailableError +from syft.sync.connections.types import MessageKind +from syft.sync.peers.peer import PeerState +from syft.sync.syftbox_manager import SyftboxManager, SyftboxManagerConfig +from syft.sync.sync.spool import SPOOL_ENTRY_VERSION, Spool +from tests.unit.test_sync_manager import path_for_job +from tests.unit.utils import grant_job_inbox_access, mock_message + + +class _Stopped(Exception): + """The process stopped.""" + + +def _drop_connection(monkeypatch) -> None: + def fail(self, recipient, filename, data): + raise ConnectionUnavailableError("Drive cannot be reached.") + + monkeypatch.setattr(DriveP2PConnection, "watcher_send_raw_bytes_to_inbox", fail) + + +def _restart(ds_manager) -> SyftboxManager: + """A new DS manager on the folder and the Drive of ``ds_manager``.""" + config = SyftboxManagerConfig._base_config_for_testing( + email=ds_manager.email, + syftbox_folder=ds_manager.syftbox_folder, + has_ds_role=True, + ) + return SyftboxManager.from_config(config, ds_manager._connection_router.connections) + + +def _accepted_contents(do_manager) -> list[str]: + return [ + str(e.content) + for msg in do_manager._get_all_accepted_events_do() + for e in msg.events + ] + + +def _pair_with_job_inbox(encryption: bool = False): + ds_manager, do_manager = SyftboxManager.pair_with_mock_drive_service_connection( + encryption=encryption + ) + grant_job_inbox_access(do_manager, ds_manager.email) + return ds_manager, do_manager, path_for_job(do_manager.email, ds_manager.email) + + +def test_failed_upload_sent_by_sync_after_restart(monkeypatch): + ds_manager, do_manager, file_path = _pair_with_job_inbox() + with monkeypatch.context() as m: + _drop_connection(m) + with pytest.raises(ConnectionUnavailableError): + ds_manager._send_file_change(file_path, "job") + + _restart(ds_manager).sync() + do_manager.sync() + assert "job" in _accepted_contents(do_manager) + + +def test_crash_after_upload_sends_once(monkeypatch): + ds_manager, do_manager, file_path = _pair_with_job_inbox() + + def stop(self, recipient, msg_id): + raise _Stopped() + + with monkeypatch.context() as m: + m.setattr(Spool, "remove", stop) + with pytest.raises(_Stopped): + ds_manager._send_file_change(file_path, "job") + + _restart(ds_manager).sync() + inbox = do_manager._connection_router.connections.p2p_con.receive( + ds_manager.email, MessageKind.PROPOSED_CHANGES + ) + assert len(inbox) == 1 + + +def test_edits_saved_offline_apply_in_order(monkeypatch): + ds_manager, do_manager, file_path = _pair_with_job_inbox() + with monkeypatch.context() as m: + _drop_connection(m) + for content in ["first", "second"]: + with pytest.raises(ConnectionUnavailableError): + ds_manager._send_file_change(file_path, content) + + ds_manager.sync() + do_manager.sync() + assert "second" in _accepted_contents(do_manager) + + +def test_saved_message_encrypted_with_peer_key_at_upload(monkeypatch): + ds_manager, do_manager, file_path = _pair_with_job_inbox(encryption=True) + ds_manager.load_peers() + with monkeypatch.context() as m: + _drop_connection(m) + with pytest.raises(ConnectionUnavailableError): + ds_manager._send_file_change(file_path, "job") + + # The DO boots with a new key, and the DS learns it, as for an enclave. + do_manager._peer_store.generate_keys() + new_bundle = do_manager._peer_store.get_public_bundle() + ds_manager._peer_store.set_peer_bundle(do_manager.email, new_bundle) + ds_manager._connection_router.update_peer_state( + do_manager.email, PeerState.ACCEPTED.value, public_encryption_bundle=new_bundle + ) + + ds_manager.sync() + do_manager.sync() + assert "job" in _accepted_contents(do_manager) + + +def test_spool_entry_of_later_version_refused(tmp_path): + spool = Spool(tmp_path) + spool.add("do@example.com", mock_message()) + entry_path = next(spool.recipient_folder("do@example.com").iterdir()) + later_version = SPOOL_ENTRY_VERSION + 1 + data = json.loads(entry_path.read_text()) + data.update(version=later_version, new_setting=True) + entry_path.write_text(json.dumps(data)) + + with pytest.raises(ValueError, match=f"version {later_version}"): + spool.pending("do@example.com") From 9dbfe8f7a2bd0604e1c78564e178798bd76cf4f5 Mon Sep 17 00:00:00 2001 From: Pedro Werneck Date: Tue, 3 Nov 2026 11:00:36 -0300 Subject: [PATCH 2/5] fix: keep saved DS messages safe after a restart, a stuck message or a lost connection - After a restart, the DS records the hashes of its saved messages again. The next edit of a path then builds on a saved edit that is not sent yet. The DO then does not refuse it as outdated. - sync() pulls before it uploads the saved messages, so a message that cannot go never holds back the results. - DriveP2PConnection.send raises ConnectionUnavailableError when Drive cannot be reached. A push then says that the change is saved and that the next sync sends it, so the user does not submit the job again. - The crash test also covers a message that the DO already archived. --- syft/sync/connections/drive/drive_p2p.py | 13 +++--- syft/sync/connections/protocols.py | 3 ++ syft/sync/syftbox_manager.py | 3 +- syft/sync/sync/datasite_watcher_syncer.py | 30 +++++++++--- tests/unit/test_spool.py | 56 +++++++++++++++++++---- 5 files changed, 83 insertions(+), 22 deletions(-) diff --git a/syft/sync/connections/drive/drive_p2p.py b/syft/sync/connections/drive/drive_p2p.py index a7a0dfbd009..2ea63774a3e 100644 --- a/syft/sync/connections/drive/drive_p2p.py +++ b/syft/sync/connections/drive/drive_p2p.py @@ -97,12 +97,13 @@ def reset_caches(self) -> None: def send( self, peer: str, msg_id: str, data: bytes, kind: MessageKind ) -> SendStatus: - if kind is MessageKind.PROPOSED_CHANGES: - self.watcher_send_raw_bytes_to_inbox(peer, msg_id, data) - elif kind is MessageKind.FILE_EVENTS: - self.owner_write_raw_bytes_to_outbox(peer, msg_id, data) - else: - raise ValueError(f"Drive has no folder for messages of kind {kind!r}") + with raise_unavailable(f"Could not send message {msg_id} to Drive."): + if kind is MessageKind.PROPOSED_CHANGES: + self.watcher_send_raw_bytes_to_inbox(peer, msg_id, data) + elif kind is MessageKind.FILE_EVENTS: + self.owner_write_raw_bytes_to_outbox(peer, msg_id, data) + else: + raise ValueError(f"Drive has no folder for messages of kind {kind!r}") return SendStatus.DELIVERED def delivered(self) -> list[str]: diff --git a/syft/sync/connections/protocols.py b/syft/sync/connections/protocols.py index c3b3a1cfce6..df952a55fc2 100644 --- a/syft/sync/connections/protocols.py +++ b/syft/sync/connections/protocols.py @@ -68,6 +68,9 @@ def send( ``msg_id`` is chosen by the sender and stays the same if the message is sent again. ``kind`` lets a connection keep the two kinds apart, as Drive does with an inbox and an outbox. + + Raises: + ConnectionUnavailableError: when the backend cannot be reached. """ @abstractmethod diff --git a/syft/sync/syftbox_manager.py b/syft/sync/syftbox_manager.py index 85d8dc47f15..9a36f0d6363 100644 --- a/syft/sync/syftbox_manager.py +++ b/syft/sync/syftbox_manager.py @@ -1121,10 +1121,11 @@ def sync( self.try_create_checkpoint(checkpoint_threshold) if self.has_ds_role: - self.datasite_watcher_syncer.sync_up() peer_emails = [peer.email for peer in self.peer_manager.syncable_peers] self.peer_manager.warn_if_all_peers_incompatible(peer_emails) self.datasite_watcher_syncer.sync_down(peer_emails) + # After the pull, so a saved message that cannot go never holds back results. + self.datasite_watcher_syncer.sync_up() def load_peers(self, force_download: bool = False): """Load peers from connection router. Delegates to PeerManager. diff --git a/syft/sync/sync/datasite_watcher_syncer.py b/syft/sync/sync/datasite_watcher_syncer.py index db0af14bbed..f5de83ccfeb 100644 --- a/syft/sync/sync/datasite_watcher_syncer.py +++ b/syft/sync/sync/datasite_watcher_syncer.py @@ -8,6 +8,7 @@ from syft.sync.callback_mixin import BaseModelCallbackMixin from syft.sync.connections.connection_router import ConnectionRouter +from syft.sync.connections.exceptions import ConnectionUnavailableError from syft.sync.messages.proposed_filechange import ( ProposedFileChange, ProposedFileChangesMessage, @@ -53,7 +54,7 @@ def from_config( shared with the other parts of the manager, so the caller builds it once and passes it in next to the config (see ``SyftboxManager.from_config``).""" - return cls( + syncer = cls( syftbox_folder=config.syftbox_folder, email=config.email, connection_router=connection_router, @@ -62,9 +63,18 @@ def from_config( ), spool=Spool(datasite_spool_path(config.syftbox_folder, config.email)), ) + syncer._record_spooled_hashes() + return syncer # --- Pushing proposed file changes --- + def _record_spooled_hashes(self) -> None: + """Record the hashes of the saved messages, so that a change after a + restart builds on a saved change that is not sent yet.""" + for recipient in self.spool.recipients(): + for message in self.spool.pending(recipient): + self._record_proposed_hashes(message) + def get_proposed_file_change_object( self, relative_path: Path, @@ -84,12 +94,15 @@ def get_proposed_file_change_object( def process_file_changes_queue(self): """Save the queued changes as one message per datasite, then upload the - saved messages to those datasites. + saved messages to those datasites. A message that an upload error stops + stays saved, and so do the messages after it. Raises: - SyftPeerNotReadyError: when a recipient cannot receive yet. Its - message stays saved, and ``sync_up`` sends it later. The other - recipients still get their messages. + SyftPeerNotReadyError: when a recipient cannot receive yet. + ``sync_up`` sends its message later. The other recipients still + get their messages. + ConnectionUnavailableError: when the backend cannot be reached. + ``sync_up`` sends the messages later. """ # One message per datasite: each change goes to the owner of its path. changes_by_recipient: Dict[str, List[ProposedFileChange]] = {} @@ -127,7 +140,12 @@ def process_file_changes_queue(self): self.spool.add(recipient, message) # The next change to a path builds on this one, sent or not. self._record_proposed_hashes(message) - waiting = self._upload_spooled(changes_by_recipient) + try: + waiting = self._upload_spooled(changes_by_recipient) + except ConnectionUnavailableError as e: + raise ConnectionUnavailableError( + f"{e} The change is saved, and client.sync() sends it." + ) from e if waiting: error = next(iter(waiting.values())) raise SyftPeerNotReadyError( diff --git a/tests/unit/test_spool.py b/tests/unit/test_spool.py index ddc4136cef1..87a76f929e0 100644 --- a/tests/unit/test_spool.py +++ b/tests/unit/test_spool.py @@ -3,6 +3,7 @@ import json +import httplib2 import pytest from syft.sync.connections.drive.drive_p2p import DriveP2PConnection @@ -11,6 +12,7 @@ from syft.sync.peers.peer import PeerState from syft.sync.syftbox_manager import SyftboxManager, SyftboxManagerConfig from syft.sync.sync.spool import SPOOL_ENTRY_VERSION, Spool +from tests.unit.test_persisted_cache import _pair_with_unsent_do_file from tests.unit.test_sync_manager import path_for_job from tests.unit.utils import grant_job_inbox_access, mock_message @@ -21,7 +23,7 @@ class _Stopped(Exception): def _drop_connection(monkeypatch) -> None: def fail(self, recipient, filename, data): - raise ConnectionUnavailableError("Drive cannot be reached.") + raise httplib2.ServerNotFoundError("Unable to find the server") monkeypatch.setattr(DriveP2PConnection, "watcher_send_raw_bytes_to_inbox", fail) @@ -52,11 +54,23 @@ def _pair_with_job_inbox(encryption: bool = False): return ds_manager, do_manager, path_for_job(do_manager.email, ds_manager.email) +def _save_later_version_entry(spool: Spool, recipient: str) -> int: + """Save a message to ``recipient`` as a later client writes it, and return + that later version.""" + spool.add(recipient, mock_message()) + entry_path = next(spool.recipient_folder(recipient).iterdir()) + later_version = SPOOL_ENTRY_VERSION + 1 + data = json.loads(entry_path.read_text()) + data.update(version=later_version, new_setting=True) + entry_path.write_text(json.dumps(data)) + return later_version + + def test_failed_upload_sent_by_sync_after_restart(monkeypatch): ds_manager, do_manager, file_path = _pair_with_job_inbox() with monkeypatch.context() as m: _drop_connection(m) - with pytest.raises(ConnectionUnavailableError): + with pytest.raises(ConnectionUnavailableError, match="is saved"): ds_manager._send_file_change(file_path, "job") _restart(ds_manager).sync() @@ -74,12 +88,15 @@ def stop(self, recipient, msg_id): m.setattr(Spool, "remove", stop) with pytest.raises(_Stopped): ds_manager._send_file_change(file_path, "job") + # The DO applies the message and moves it to its archive. + do_manager.sync() _restart(ds_manager).sync() inbox = do_manager._connection_router.connections.p2p_con.receive( ds_manager.email, MessageKind.PROPOSED_CHANGES ) - assert len(inbox) == 1 + assert inbox == [] + assert _accepted_contents(do_manager).count("job") == 1 def test_edits_saved_offline_apply_in_order(monkeypatch): @@ -95,6 +112,20 @@ def test_edits_saved_offline_apply_in_order(monkeypatch): assert "second" in _accepted_contents(do_manager) +def test_edit_after_restart_builds_on_saved_edit(monkeypatch): + ds_manager, do_manager, file_path = _pair_with_job_inbox() + with monkeypatch.context() as m: + _drop_connection(m) + with pytest.raises(ConnectionUnavailableError): + ds_manager._send_file_change(file_path, "first") + + restarted = _restart(ds_manager) + restarted.load_peers() + restarted.datasite_watcher_syncer.on_file_change(file_path, "second") + do_manager.sync() + assert "second" in _accepted_contents(do_manager) + + def test_saved_message_encrypted_with_peer_key_at_upload(monkeypatch): ds_manager, do_manager, file_path = _pair_with_job_inbox(encryption=True) ds_manager.load_peers() @@ -118,12 +149,19 @@ def test_saved_message_encrypted_with_peer_key_at_upload(monkeypatch): def test_spool_entry_of_later_version_refused(tmp_path): spool = Spool(tmp_path) - spool.add("do@example.com", mock_message()) - entry_path = next(spool.recipient_folder("do@example.com").iterdir()) - later_version = SPOOL_ENTRY_VERSION + 1 - data = json.loads(entry_path.read_text()) - data.update(version=later_version, new_setting=True) - entry_path.write_text(json.dumps(data)) + later_version = _save_later_version_entry(spool, "do@example.com") with pytest.raises(ValueError, match=f"version {later_version}"): spool.pending("do@example.com") + + +def test_stuck_message_does_not_block_pull(): + ds_manager, do_manager = _pair_with_unsent_do_file() + spool = ds_manager.datasite_watcher_syncer.spool + _save_later_version_entry(spool, do_manager.email) + do_manager.sync(auto_checkpoint=False) + + with pytest.raises(ValueError, match="version"): + ds_manager.sync() + received = ds_manager.syftbox_folder / do_manager.email / "app_data/apis/x/a.py" + assert received.exists() From 35aaa13c3123f41b103a0c7ff8c9c930110da33c Mon Sep 17 00:00:00 2001 From: Pedro Werneck Date: Tue, 3 Nov 2026 11:52:03 -0300 Subject: [PATCH 3/5] refactor: rename the spool's message bytes, pending messages and wait errors - SpoolEntry.message is now compressed_message, because it holds the compressed bytes of the message. - Spool.pending() is now pending_messages(). - In DatasiteWatcherSyncer, waiting is now not_ready_errors. It maps each recipient email to the error that says why its messages wait. --- syft/sync/sync/datasite_watcher_syncer.py | 20 ++++++++++---------- syft/sync/sync/spool.py | 10 ++++++---- tests/unit/test_spool.py | 2 +- 3 files changed, 17 insertions(+), 15 deletions(-) diff --git a/syft/sync/sync/datasite_watcher_syncer.py b/syft/sync/sync/datasite_watcher_syncer.py index f5de83ccfeb..5c3e90c2aa1 100644 --- a/syft/sync/sync/datasite_watcher_syncer.py +++ b/syft/sync/sync/datasite_watcher_syncer.py @@ -72,7 +72,7 @@ def _record_spooled_hashes(self) -> None: """Record the hashes of the saved messages, so that a change after a restart builds on a saved change that is not sent yet.""" for recipient in self.spool.recipients(): - for message in self.spool.pending(recipient): + for message in self.spool.pending_messages(recipient): self._record_proposed_hashes(message) def get_proposed_file_change_object( @@ -141,13 +141,13 @@ def process_file_changes_queue(self): # The next change to a path builds on this one, sent or not. self._record_proposed_hashes(message) try: - waiting = self._upload_spooled(changes_by_recipient) + not_ready_errors = self._upload_spooled(changes_by_recipient) except ConnectionUnavailableError as e: raise ConnectionUnavailableError( f"{e} The change is saved, and client.sync() sends it." ) from e - if waiting: - error = next(iter(waiting.values())) + if not_ready_errors: + error = next(iter(not_ready_errors.values())) raise SyftPeerNotReadyError( error.peer_email, cause=error.cause, @@ -178,17 +178,17 @@ def _upload_spooled( upload that fails for another reason raises too. That message and the later ones stay saved. """ - waiting = {} + not_ready_errors = {} with self.spool.locked(): for recipient in recipients: try: self._upload_spooled_to(recipient) except SyftPeerNotReadyError as e: - waiting[recipient] = e - return waiting + not_ready_errors[recipient] = e + return not_ready_errors def _upload_spooled_to(self, recipient: str) -> None: - for message in self.spool.pending(recipient): + for message in self.spool.pending_messages(recipient): msg_id = message.message_filename.as_string() # An earlier upload may have stopped before it removed the message. if not self.connection_router.is_delivered(recipient, msg_id): @@ -200,8 +200,8 @@ def _upload_spooled_to(self, recipient: str) -> None: def sync_up(self) -> None: """Upload every saved message. Warn for each recipient that cannot receive yet; its messages wait for the next sync.""" - waiting = self._upload_spooled(self.spool.recipients()) - for recipient, error in waiting.items(): + not_ready_errors = self._upload_spooled(self.spool.recipients()) + for recipient, error in not_ready_errors.items(): warnings.warn( f"Changes to {recipient} wait in " f"{self.spool.recipient_folder(recipient)}: {error}" diff --git a/syft/sync/sync/spool.py b/syft/sync/sync/spool.py index 4cd1f45efba..23aeb26bde3 100644 --- a/syft/sync/sync/spool.py +++ b/syft/sync/sync/spool.py @@ -44,7 +44,7 @@ class SpoolEntry(BaseModel): version: int # The compressed message, before the downgrade and the encryption for its peer. - message: bytes + compressed_message: bytes @classmethod def read(cls, path: Path) -> "SpoolEntry": @@ -75,7 +75,7 @@ def __init__(self, folder: Path): def add(self, recipient: str, message: ProposedFileChangesMessage) -> None: """Save ``message`` to ``recipient``.""" entry = SpoolEntry( - version=SPOOL_ENTRY_VERSION, message=message.as_compressed_data() + version=SPOOL_ENTRY_VERSION, compressed_message=message.as_compressed_data() ) msg_id = message.message_filename.as_string() write_text_atomic(self._entry_path(recipient, msg_id), entry.model_dump_json()) @@ -86,7 +86,7 @@ def recipients(self) -> list[str]: return [] return sorted(p.name for p in self.folder.iterdir() if p.is_dir()) - def pending(self, recipient: str) -> list[ProposedFileChangesMessage]: + def pending_messages(self, recipient: str) -> list[ProposedFileChangesMessage]: """The saved messages to ``recipient``, oldest first. Raises: @@ -94,7 +94,9 @@ def pending(self, recipient: str) -> list[ProposedFileChangesMessage]: """ paths = self.recipient_folder(recipient).glob(f"*{SPOOL_ENTRY_SUFFIX}") messages = [ - ProposedFileChangesMessage.from_compressed_data(SpoolEntry.read(p).message) + ProposedFileChangesMessage.from_compressed_data( + SpoolEntry.read(p).compressed_message + ) for p in paths ] return sorted(messages, key=lambda m: m.message_filename.submitted_timestamp) diff --git a/tests/unit/test_spool.py b/tests/unit/test_spool.py index 87a76f929e0..54d2e4be3ef 100644 --- a/tests/unit/test_spool.py +++ b/tests/unit/test_spool.py @@ -152,7 +152,7 @@ def test_spool_entry_of_later_version_refused(tmp_path): later_version = _save_later_version_entry(spool, "do@example.com") with pytest.raises(ValueError, match=f"version {later_version}"): - spool.pending("do@example.com") + spool.pending_messages("do@example.com") def test_stuck_message_does_not_block_pull(): From fdf59e25eb3b7ed08d62d3b193dd2aa08e4f715e Mon Sep 17 00:00:00 2001 From: Pedro Werneck Date: Tue, 3 Nov 2026 13:22:47 -0300 Subject: [PATCH 4/5] refactor: make send() safe to repeat instead of asking is_delivered() - P2PConnection.is_delivered() is gone. The WebRTC connection has no record of receipts after a restart, so it could not answer, and it could not be built without the method. - send() may be repeated with the same msg_id, and the peer applies the message once. Drive's send() skips a proposal that it already uploaded. The WebRTC queue already drops a repeated id while the message waits. - The spool sends each saved message and then removes it, with no separate check. --- syft/sync/connections/connection_router.py | 4 --- syft/sync/connections/drive/drive_p2p.py | 29 ++++++++++++---------- syft/sync/connections/protocols.py | 15 +++-------- syft/sync/sync/datasite_watcher_syncer.py | 9 +++---- 4 files changed, 24 insertions(+), 33 deletions(-) diff --git a/syft/sync/connections/connection_router.py b/syft/sync/connections/connection_router.py index 006f0103ee8..851bf4ea828 100644 --- a/syft/sync/connections/connection_router.py +++ b/syft/sync/connections/connection_router.py @@ -176,10 +176,6 @@ def mark_all_processed(self, messages: list[RawMessage]) -> None: """Finish ``messages``, as received by either role.""" self.connections.p2p_con.mark_all_processed(messages) - def is_delivered(self, peer_email: str, msg_id: str) -> bool: - """Whether the message sent to ``peer_email`` under ``msg_id`` reached it.""" - return self.connections.p2p_con.is_delivered(peer_email, msg_id) - def owner_write_event_messages_to_outbox( self, recipient_email: str, events_message: FileChangeEventsMessage ) -> SendStatus: diff --git a/syft/sync/connections/drive/drive_p2p.py b/syft/sync/connections/drive/drive_p2p.py index 2ea63774a3e..9055c4f2a78 100644 --- a/syft/sync/connections/drive/drive_p2p.py +++ b/syft/sync/connections/drive/drive_p2p.py @@ -97,9 +97,12 @@ def reset_caches(self) -> None: def send( self, peer: str, msg_id: str, data: bytes, kind: MessageKind ) -> SendStatus: + """A proposal that an earlier send uploaded is not uploaded again. File + events are not checked: the DS skips an outbox message it received.""" with raise_unavailable(f"Could not send message {msg_id} to Drive."): if kind is MessageKind.PROPOSED_CHANGES: - self.watcher_send_raw_bytes_to_inbox(peer, msg_id, data) + if not self._is_uploaded(msg_id): + self.watcher_send_raw_bytes_to_inbox(peer, msg_id, data) elif kind is MessageKind.FILE_EVENTS: self.owner_write_raw_bytes_to_outbox(peer, msg_id, data) else: @@ -110,18 +113,6 @@ def delivered(self) -> list[str]: # Every Drive send is DELIVERED when it returns; none waits for a peer. return [] - @raise_unavailable("Could not check Drive for a sent message.") - def is_delivered(self, peer: str, msg_id: str) -> bool: - # The sender owns the file it uploads, also after the peer moves it to - # an archive, so a search of my own files by name finds it in any folder. - query = f"name='{msg_id}' and 'me' in owners and trashed=false" - results = execute_with_retries( - self._cached_client.client.drive_service.files().list( - q=query, fields="files(id)", pageSize=1 - ) - ) - return bool(results["files"]) - def receive( self, peer: str, @@ -233,6 +224,18 @@ def delete_messages(self, handles: list[str]) -> None: # -- Drive ----------------------------------------------------------------- + def _is_uploaded(self, msg_id: str) -> bool: + """Whether I uploaded a message named ``msg_id``.""" + # I own the file I upload, also after the peer moves it to its archive, + # so a search of my own files by name finds it in any folder. + query = f"name='{msg_id}' and 'me' in owners and trashed=false" + results = execute_with_retries( + self._cached_client.client.drive_service.files().list( + q=query, fields="files(id)", pageSize=1 + ) + ) + return bool(results["files"]) + def _list_inbox(self, peer: str) -> list[dict]: """The messages in the inbox from ``peer``, oldest first, each with ``file_id`` and ``file_name``, without downloading them.""" diff --git a/syft/sync/connections/protocols.py b/syft/sync/connections/protocols.py index df952a55fc2..f833cbeda43 100644 --- a/syft/sync/connections/protocols.py +++ b/syft/sync/connections/protocols.py @@ -66,8 +66,10 @@ def send( """Send ``data`` to ``peer`` under ``msg_id``. ``msg_id`` is chosen by the sender and stays the same if the message is - sent again. ``kind`` lets a connection keep the two kinds apart, as - Drive does with an inbox and an outbox. + sent again, as after a stop. The peer then applies it once: the + connection drops the second copy, or the peer skips it. ``kind`` lets a + connection keep the two kinds apart, as Drive does with an inbox and an + outbox. Raises: ConnectionUnavailableError: when the backend cannot be reached. @@ -78,15 +80,6 @@ def delivered(self) -> list[str]: """The ``msg_id`` of each queued message the peer stored since the last call. A connection that only returns ``DELIVERED`` returns nothing.""" - @abstractmethod - def is_delivered(self, peer: str, msg_id: str) -> bool: - """Whether the message sent to ``peer`` under ``msg_id`` is stored where - the peer finds it, or was processed there. - - A sender that stopped between a send and its record of the send asks - this before it sends the message again. - """ - @abstractmethod def receive( self, diff --git a/syft/sync/sync/datasite_watcher_syncer.py b/syft/sync/sync/datasite_watcher_syncer.py index 5c3e90c2aa1..1d181e32942 100644 --- a/syft/sync/sync/datasite_watcher_syncer.py +++ b/syft/sync/sync/datasite_watcher_syncer.py @@ -190,11 +190,10 @@ def _upload_spooled( def _upload_spooled_to(self, recipient: str) -> None: for message in self.spool.pending_messages(recipient): msg_id = message.message_filename.as_string() - # An earlier upload may have stopped before it removed the message. - if not self.connection_router.is_delivered(recipient, msg_id): - self.connection_router.watcher_send_proposed_file_changes_message( - recipient, message - ) + # The send drops a message that an upload stored before a stop. + self.connection_router.watcher_send_proposed_file_changes_message( + recipient, message + ) self.spool.remove(recipient, msg_id) def sync_up(self) -> None: From b6af17d3a3881324b168e60b6fdd205da969ccf6 Mon Sep 17 00:00:00 2001 From: Pedro Werneck Date: Tue, 3 Nov 2026 14:06:45 -0300 Subject: [PATCH 5/5] refactor: read the spool entry and the settings file through one version check - read_versioned_json() in syft/sync/utils/versioned_file.py reads a JSON file and refuses a later version before it checks the fields. - SpoolEntry.read() and _DatasiteSettings.read() call it. Their error messages do not change. --- syft/sync/peers/peer_store.py | 13 ++++--------- syft/sync/sync/spool.py | 12 ++---------- syft/sync/utils/versioned_file.py | 30 ++++++++++++++++++++++++++++++ 3 files changed, 36 insertions(+), 19 deletions(-) create mode 100644 syft/sync/utils/versioned_file.py diff --git a/syft/sync/peers/peer_store.py b/syft/sync/peers/peer_store.py index 034cf8a88e1..411d081d413 100644 --- a/syft/sync/peers/peer_store.py +++ b/syft/sync/peers/peer_store.py @@ -4,7 +4,6 @@ Shared between PeerManager and all ConnectionRouter instances. """ -import json import os import tempfile from pathlib import Path @@ -21,6 +20,7 @@ ) from syft.sync.peers.exceptions import SyftPeerNotReadyError from syft.sync.peers.peer import Peer, PeerState +from syft.sync.utils.versioned_file import read_versioned_json # Encryption key bundles persist inside the participant's own SyftBox datasite # folder, under private/ (which is never synced to Drive). This scopes keys per @@ -96,14 +96,9 @@ def read(cls, path: Path) -> "_DatasiteSettings": ValueError: if a later client wrote the file, or it has another shape. """ - data = json.loads(path.read_text()) - # Checked before the fields, so a later layout is refused by its version. - if data["version"] > DATASITE_SETTINGS_VERSION: - raise ValueError( - f"The settings file at {path} has version {data['version']}, and " - f"this client reads up to version {DATASITE_SETTINGS_VERSION}." - ) - return cls.model_validate(data) + return read_versioned_json( + path, cls, DATASITE_SETTINGS_VERSION, "settings file" + ) def resolve_encryption_policy( diff --git a/syft/sync/sync/spool.py b/syft/sync/sync/spool.py index 23aeb26bde3..21c3ecd2438 100644 --- a/syft/sync/sync/spool.py +++ b/syft/sync/sync/spool.py @@ -10,7 +10,6 @@ has at that time. An enclave gets a new key at every boot. """ -import json from contextlib import AbstractContextManager from pathlib import Path @@ -19,6 +18,7 @@ from syft.sync.messages.proposed_filechange import ProposedFileChangesMessage from syft.sync.peers.peer_store import PRIVATE_DIR_NAME from syft.sync.sync.caches.persisted_dict import locked_file, write_text_atomic +from syft.sync.utils.versioned_file import read_versioned_json SPOOL_DIR_NAME = "spool" SPOOL_ENTRY_SUFFIX = ".json" @@ -54,15 +54,7 @@ def read(cls, path: Path) -> "SpoolEntry": ValueError: if a later client wrote the entry, or it has another shape. """ - text = path.read_text() - data = json.loads(text) - # Checked before the fields, so a later layout is refused by its version. - if data["version"] > SPOOL_ENTRY_VERSION: - raise ValueError( - f"The spool entry at {path} has version {data['version']}, and " - f"this client reads up to version {SPOOL_ENTRY_VERSION}." - ) - return cls.model_validate_json(text) + return read_versioned_json(path, cls, SPOOL_ENTRY_VERSION, "spool entry") class Spool: diff --git a/syft/sync/utils/versioned_file.py b/syft/sync/utils/versioned_file.py new file mode 100644 index 00000000000..65c803221f6 --- /dev/null +++ b/syft/sync/utils/versioned_file.py @@ -0,0 +1,30 @@ +"""Read a JSON file that carries a format version.""" + +import json +from pathlib import Path +from typing import TypeVar + +from pydantic import BaseModel + +ModelT = TypeVar("ModelT", bound=BaseModel) + + +def read_versioned_json( + path: Path, model: type[ModelT], max_version: int, kind: str +) -> ModelT: + """The file at ``path`` as ``model``. ``kind`` names the file in errors. + + The version is checked before the fields, so a file of a later layout is + refused by its version, not by a field error. + + Raises: + ValueError: if a later client wrote the file, or it has another shape. + """ + text = path.read_text() + version = json.loads(text)["version"] + if version > max_version: + raise ValueError( + f"The {kind} at {path} has version {version}, and this client " + f"reads up to version {max_version}." + ) + return model.model_validate_json(text)