From 1046930bd9771e3a50dbbf265c5c9c8c6452f803 Mon Sep 17 00:00:00 2001 From: Pedro Werneck Date: Mon, 2 Nov 2026 12:35:06 -0300 Subject: [PATCH 1/3] feat: add a per-peer protocol switch and a cross-version test harness - ConnectionRouter.pair_speaks_protocol() says whether this client and a peer both speak a syft protocol version. A behaviour that a later protocol adds can then stay off for a peer that does not speak it. A peer with an unknown schema gets False. - ConnectionRouter.negotiated_protocol_version_for_peer() returns the negotiated version. _downgrade_for_peer() uses the same negotiation. - syft-migration exports protocol_version_at_least(), which compares versions as integers. - The next_protocol and pin_syft_protocol test fixtures hold one client at the released protocol and run the other on a test-only next protocol. test_protocol_skew_pair.py runs a job round trip between them. - The datasite settings file carries a format version. A file of another version is refused, as the key file is. --- .../src/syft_migration/__init__.py | 3 +- .../src/syft_migration/identity.py | 9 +++ .../tests/test_version_ordering.py | 8 +- syft/sync/connections/connection_router.py | 56 ++++++++++--- syft/sync/peers/peer_store.py | 38 +++++++-- tests/conftest.py | 79 +++++++++++++++++++ .../migrations/p2p/test_protocol_skew_pair.py | 69 ++++++++++++++++ tests/unit/test_encryption_policy.py | 23 ++++-- 8 files changed, 261 insertions(+), 24 deletions(-) create mode 100644 tests/migrations/p2p/test_protocol_skew_pair.py diff --git a/packages/syft-migration/src/syft_migration/__init__.py b/packages/syft-migration/src/syft_migration/__init__.py index 7ef27add58f..da89eb373e6 100644 --- a/packages/syft-migration/src/syft_migration/__init__.py +++ b/packages/syft-migration/src/syft_migration/__init__.py @@ -6,7 +6,7 @@ unregistered_objects, versioned_objects, ) -from syft_migration.identity import MigrationError +from syft_migration.identity import MigrationError, protocol_version_at_least from syft_migration.registry import MigrationRegistry from syft_migration.schema import ( PackageInfo, @@ -31,6 +31,7 @@ "import_all_modules", "missing_downgrade_paths", "missing_upgrade_paths", + "protocol_version_at_least", "unregistered_objects", "versioned_objects", ] diff --git a/packages/syft-migration/src/syft_migration/identity.py b/packages/syft-migration/src/syft_migration/identity.py index 6ab95ab81d4..7379921dccf 100644 --- a/packages/syft-migration/src/syft_migration/identity.py +++ b/packages/syft-migration/src/syft_migration/identity.py @@ -38,6 +38,15 @@ def _version_order(version: str) -> int: ) from None +def protocol_version_at_least(version: str, minimum: str) -> bool: + """Whether protocol ``version`` is ``minimum`` or later. + + Protocol versions are incrementing integers held as strings, so ``"10"`` is + later than ``"2"``. + """ + return _version_order(version) >= _version_order(minimum) + + def _identity(cls: type[MigratableObject]) -> tuple[str, str]: """Return (canonical_name, version) for a concrete subclass. diff --git a/packages/syft-migration/tests/test_version_ordering.py b/packages/syft-migration/tests/test_version_ordering.py index c9f3937aafc..dd55cf772ef 100644 --- a/packages/syft-migration/tests/test_version_ordering.py +++ b/packages/syft-migration/tests/test_version_ordering.py @@ -1,12 +1,12 @@ """Object versions order by number, not as strings.""" import pytest - from syft_migration import ( MigratableObject, MigrationError, MigrationRegistry, ProtocolSchema, + protocol_version_at_least, ) @@ -94,6 +94,12 @@ class ThingV3(MigratableObject, registry=reg): ) +def test_protocol_version_at_least_orders_by_number(): + assert protocol_version_at_least("10", "2") + assert not protocol_version_at_least("2", "10") + assert protocol_version_at_least("2", "2") + + def test_non_numeric_object_version_is_rejected(): reg = _registry() with pytest.raises(MigrationError): diff --git a/syft/sync/connections/connection_router.py b/syft/sync/connections/connection_router.py index 38fe2da1feb..0a930a51a29 100644 --- a/syft/sync/connections/connection_router.py +++ b/syft/sync/connections/connection_router.py @@ -4,9 +4,10 @@ from typing import TYPE_CHECKING, List, Optional from pydantic import BaseModel -from syft_migration import MigratableObject +from syft_migration import MigratableObject, ProtocolSchema, protocol_version_at_least from syft.migrations import ( + PROTOCOL_NAME, SYFT_CLIENT_PROTOCOL_VERSION, client_migration_service, client_registry, @@ -14,6 +15,9 @@ from syft.sync.checkpoints.checkpoint import Checkpoint, IncrementalCheckpoint from syft.sync.checkpoints.rolling_state import RollingState from syft.sync.connections.base_connection import FileCollection +from syft.sync.connections.naming import PEERS_META_KEY +from syft.sync.connections.protocols import Connections +from syft.sync.connections.types import MessageKind, RawMessage, SendStatus from syft.sync.events.file_change_event import ( FileChangeEventsMessage, ) @@ -21,9 +25,6 @@ from syft.sync.peers.exceptions import SyftPeerNotReadyError from syft.sync.peers.peer import Peer, PeerState from syft.sync.peers.peer_store import PeerStore -from syft.sync.connections.naming import PEERS_META_KEY -from syft.sync.connections.protocols import Connections -from syft.sync.connections.types import MessageKind, RawMessage, SendStatus from syft.sync.utils.print_utils import ( print_peer_added_to_platform, print_peer_adding_to_platform, @@ -36,6 +37,20 @@ logger = logging.getLogger(__name__) +def _protocol_version_to_use(peer_schema: ProtocolSchema) -> str: + """The syft protocol version to speak with a peer that advertises + ``peer_schema``: the lower of the two current versions, at or above both + floors. + + Raises: + MigrationError: when no version satisfies both sides. + """ + return client_registry.infer_protocol_version_to_use( + peer_version=peer_schema.version, + peer_min=peer_schema.min_supported_version, + ) + + class ConnectionRouter(BaseModel): """The one router of a manager. ``SyftboxManager.from_config`` builds it and hands it to every part, so they share one set of connections and one @@ -50,6 +65,32 @@ class ConnectionRouter(BaseModel): def get_authenticated_email(self) -> str: return self.connections.peering_con.whoami() + # ========================================================================= + # PROTOCOL NEGOTIATION + # ========================================================================= + + def negotiated_protocol_version_for_peer(self, peer_email: str) -> str | None: + """The syft protocol version this client and ``peer_email`` both speak, + or None when the peer's schema is not known. + + Raises: + MigrationError: when no version satisfies both sides. + """ + schema = self.peer_store.protocol_schema(peer_email, PROTOCOL_NAME) + return None if schema is None else _protocol_version_to_use(schema) + + def pair_speaks_protocol(self, peer_email: str, version: str) -> bool: + """Whether this client and ``peer_email`` both speak syft protocol + ``version`` or later. False when the peer's schema is not known. + + Gate the behaviour that a protocol version adds on this check. Each side + decides from the version file it read last, so one side can switch + before the other. The new behaviour must still work while the peer uses + the old one. + """ + negotiated = self.negotiated_protocol_version_for_peer(peer_email) + return negotiated is not None and protocol_version_at_least(negotiated, version) + # ========================================================================= # MESSAGE SEND/RECEIVE (with encryption) # ========================================================================= @@ -68,7 +109,7 @@ def _downgrade_for_peer( A peer without a known schema is assumed to run the current protocol, the same policy as jobs; the assumption is logged. """ - schema = self.peer_store.protocol_schema(peer_email, "syft") + schema = self.peer_store.protocol_schema(peer_email, PROTOCOL_NAME) if schema is None: logger.warning( f"No syft protocol schema known for peer {peer_email!r}. " @@ -76,10 +117,7 @@ def _downgrade_for_peer( "peer that speaks an earlier protocol cannot read this message." ) return message - protocol_version = client_registry.infer_protocol_version_to_use( - peer_version=schema.version, - peer_min=schema.min_supported_version, - ) + protocol_version = _protocol_version_to_use(schema) if schema.version == protocol_version: # The peer speaks the negotiated version, so its advertised slim # schema is the target; computing our own full schema on every diff --git a/syft/sync/peers/peer_store.py b/syft/sync/peers/peer_store.py index 0ccfab5c849..c5afd6bff8f 100644 --- a/syft/sync/peers/peer_store.py +++ b/syft/sync/peers/peer_store.py @@ -33,6 +33,21 @@ # Format of the crypto key file. Raise it when the layout of the file changes. # A file of any other version is refused. CRYPTO_KEYS_VERSION = 2 +# Format of the datasite settings file, under the same rule. +DATASITE_SETTINGS_VERSION = 1 + + +def _require_file_version(path: Path, version: int, expected: int) -> None: + """Refuse the file at ``path`` unless its format ``version`` is ``expected``. + + Raises: + ValueError: if ``version`` is not ``expected``. + """ + if version != expected: + raise ValueError( + f"The file at {path} has version {version}, and this client reads " + f"only version {expected}." + ) class _KeyFile(BaseModel): @@ -53,11 +68,7 @@ def read(cls, path: Path, email: str) -> "_KeyFile": another identity. """ key_file = cls.model_validate_json(path.read_text()) - if key_file.version != CRYPTO_KEYS_VERSION: - raise ValueError( - f"The crypto key file at {path} has version {key_file.version}, " - f"and this client reads only version {CRYPTO_KEYS_VERSION}." - ) + _require_file_version(path, key_file.version, CRYPTO_KEYS_VERSION) if key_file.email != email: raise ValueError( f"Encryption key file {path} belongs to {key_file.email!r}, " @@ -81,8 +92,20 @@ class _DatasiteSettings(BaseModel): model_config = ConfigDict(frozen=True, extra="forbid") + version: int encryption_policy: EncryptionPolicy + @classmethod + def read(cls, path: Path) -> "_DatasiteSettings": + """Read the settings file at ``path``. + + Raises: + ValueError: if the file has another shape or version. + """ + settings = cls.model_validate_json(path.read_text()) + _require_file_version(path, settings.version, DATASITE_SETTINGS_VERSION) + return settings + def resolve_encryption_policy( settings_path: Path | str, requested: EncryptionPolicy | None @@ -99,14 +122,15 @@ def resolve_encryption_policy( path = Path(settings_path) if not path.exists(): settings = _DatasiteSettings( - encryption_policy=requested or EncryptionPolicy.OPTIONAL + version=DATASITE_SETTINGS_VERSION, + encryption_policy=requested or EncryptionPolicy.OPTIONAL, ) try: _create_private_file(path, settings.model_dump_json(indent=2)) return settings.encryption_policy except FileExistsError: pass # Another process started first; its policy applies. - on_disk = _DatasiteSettings.model_validate_json(path.read_text()).encryption_policy + on_disk = _DatasiteSettings.read(path).encryption_policy if requested is not None and requested != on_disk: raise EncryptionPolicyMismatchError(path, on_disk, requested) return on_disk diff --git a/tests/conftest.py b/tests/conftest.py index 08e93461760..9f75318ee23 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -5,10 +5,21 @@ """ import os +from typing import NamedTuple import pytest +from syft_migration import protocol_version_at_least +from syft.migrations import PROTOCOL_NAME, client_registry from syft.sync.peers.peer_store import PeerStore +from syft.sync.syftbox_manager import SyftboxManager + + +class ProtocolStep(NamedTuple): + """The syft protocol of the last release, and the next one.""" + + released: str + next: str @pytest.fixture(scope="session", autouse=True) @@ -53,3 +64,71 @@ def protocol_schema(self, email, protocol_name): monkeypatch.setattr(PeerStore, "protocol_schema", protocol_schema) return overrides + + +@pytest.fixture +def next_protocol(monkeypatch) -> ProtocolStep: + """A test-only next syft protocol, with the current one frozen as released. + + The registry stores its current schema as a released protocol, then speaks + one version higher. A client that ``pin_syft_protocol`` holds at the + released version then plays a client of the last release. Build the + clients after this fixture, so they publish the next version. + """ + released = client_registry.protocol_version + monkeypatch.setitem( + client_registry.protocol_version_history, + released, + client_registry.compute_protocol_schema(), + ) + next_version = str(int(released) + 1) + monkeypatch.setattr(client_registry, "protocol_version", next_version) + return ProtocolStep(released=released, next=next_version) + + +@pytest.fixture +def pin_syft_protocol(monkeypatch, advertised_schemas): + """Hold a client at an older syft protocol, as a client of an older release. + + ``pin_syft_protocol(manager, version)`` republishes the version file of + ``manager`` with the schema of ``version``, and caps every peer's syft + schema that ``manager`` reads at ``version``. Its own downgrade and switch + then negotiate at most ``version``. Call ``load_peers()`` on its peers + afterwards, so they read the new version file. + + It wraps ``advertised_schemas``, so an advertised schema is capped too. + """ + # email of a held client -> the syft protocol version it is held at + held_versions: dict[str, str] = {} + read_schema = PeerStore.protocol_schema + + def protocol_schema(self, email, protocol_name): + schema = read_schema(self, email, protocol_name) + held = held_versions.get(self.email) + if schema is None or held is None or protocol_name != PROTOCOL_NAME: + return schema + if protocol_version_at_least(held, schema.version): + return schema + held_schema = client_registry.schema_for_protocol_version(held) + return held_schema.model_copy( + update={"min_supported_version": schema.min_supported_version} + ) + + monkeypatch.setattr(PeerStore, "protocol_schema", protocol_schema) + + def pin(manager: SyftboxManager, version: str) -> None: + peer_manager = manager.peer_manager + own_version = peer_manager.get_own_version() + held_schema = client_registry.schema_for_protocol_version(version) + peer_manager._own_version = own_version.model_copy( + update={ + "protocol_schemas": { + **own_version.protocol_schemas, + PROTOCOL_NAME: held_schema, + } + } + ) + peer_manager.write_own_peer_info() + held_versions[manager.email] = version + + return pin diff --git a/tests/migrations/p2p/test_protocol_skew_pair.py b/tests/migrations/p2p/test_protocol_skew_pair.py new file mode 100644 index 00000000000..f85b872beab --- /dev/null +++ b/tests/migrations/p2p/test_protocol_skew_pair.py @@ -0,0 +1,69 @@ +"""A client of the last release and a client of the next one work together. + +Both clients run today's code. ``next_protocol`` makes the registry speak a +test-only next syft protocol, and ``pin_syft_protocol`` holds one client at the +released one. Each side then negotiates the released protocol, and a behaviour +that the next protocol adds stays off on both sides. +""" + +from pathlib import Path + +import pytest +from syft_rds import SyftRDSClient + +from syft.migrations import client_registry +from syft.sync.syftbox_manager import SyftboxManager + +RESULT_SCRIPT = "mkdir -p code/outputs && echo 42 > code/outputs/result.txt" + + +@pytest.mark.parametrize("held", ["do", "ds"]) +def test_held_client_exchanges_job_at_released_protocol( + held, next_protocol, pin_syft_protocol +): + ds, do = SyftRDSClient.pair_with_mock_drive_service_connection( + use_in_memory_cache=False + ) + pin_syft_protocol({"do": do, "ds": ds}[held].sync_engine, next_protocol.released) + ds.load_peers() + do.load_peers() + ds_router = ds.sync_engine._connection_router + do_router = do.sync_engine._connection_router + + released = next_protocol.released + assert ds_router.negotiated_protocol_version_for_peer(do.email) == released + assert do_router.negotiated_protocol_version_for_peer(ds.email) == released + assert not ds_router.pair_speaks_protocol(do.email, next_protocol.next) + assert not do_router.pair_speaks_protocol(ds.email, next_protocol.next) + + ds.submit_bash_job(do.email, RESULT_SCRIPT, "skew.job") + do.sync() + do.job_client.jobs[0].approve() + do.process_approved_jobs(stream_output=False, share_outputs_with_submitter=True) + do.sync() + ds.sync() + + [output_path] = ds.job_client.jobs[-1].output_paths + assert Path(output_path).read_text().strip() == "42" + + +def test_pair_on_next_protocol_speaks_it(next_protocol): + # The control: without a held client both sides speak the next protocol, so + # the test above measures the pin and not a broken switch. + ds_manager, do_manager = SyftboxManager.pair_with_mock_drive_service_connection() + ds_router = ds_manager._connection_router + do_router = do_manager._connection_router + + assert ds_router.pair_speaks_protocol(do_manager.email, next_protocol.next) + assert do_router.pair_speaks_protocol(ds_manager.email, next_protocol.next) + + +def test_unknown_peer_speaks_no_protocol(advertised_schemas): + ds_manager, do_manager = SyftboxManager.pair_with_mock_drive_service_connection() + advertised_schemas[ds_manager.email, "syft"] = None + do_router = do_manager._connection_router + + assert do_router.negotiated_protocol_version_for_peer(ds_manager.email) is None + assert not do_router.pair_speaks_protocol( + ds_manager.email, client_registry.protocol_version + ) diff --git a/tests/unit/test_encryption_policy.py b/tests/unit/test_encryption_policy.py index b79065e92ee..8d81b48f03b 100644 --- a/tests/unit/test_encryption_policy.py +++ b/tests/unit/test_encryption_policy.py @@ -9,9 +9,10 @@ import pytest from pydantic import ValidationError -from syft.sync.connections.drive.drive_peering import DrivePeeringConnection from syft.sync.connections.connection_router import ConnectionRouter from syft.sync.connections.drive import mock_drive_service +from syft.sync.connections.drive.drive_peering import DrivePeeringConnection +from syft.sync.connections.types import MessageKind from syft.sync.encryption_policy import ( EncryptionPolicy, EncryptionPolicyMismatchError, @@ -23,13 +24,13 @@ from syft.sync.peers.exceptions import SyftPeerNotReadyError from syft.sync.peers.peer import Peer from syft.sync.peers.peer_store import ( + DATASITE_SETTINGS_VERSION, PeerStore, datasite_settings_path, resolve_encryption_policy, ) from syft.sync.syftbox_manager import SyftboxManager from syft.sync.sync.datasite_owner_syncer import OutgoingMessage -from syft.sync.connections.types import MessageKind from syft.sync.version.peer_info import PeerInfo from syft.sync.version.peer_manager import PeerManager, PeerManagerConfig from syft.sync.version.version_info import VersionInfo @@ -280,6 +281,18 @@ def test_client_with_other_policy_than_disk_raises(tmp_path): resolve_encryption_policy(path, REQUIRED) +def test_settings_file_of_other_version_refused(tmp_path): + path = tmp_path / "syft_settings.json" + resolve_encryption_policy(path, OPTIONAL) + later_version = DATASITE_SETTINGS_VERSION + 1 + data = json.loads(path.read_text()) + data["version"] = later_version + path.write_text(json.dumps(data)) + + with pytest.raises(ValueError, match=f"version {later_version}"): + resolve_encryption_policy(path, None) + + def test_client_without_policy_takes_policy_of_syftbox(tmp_path): """syft-bg starts after a notebook that required encryption.""" @@ -512,10 +525,8 @@ def test_changes_for_two_datasites_go_to_each(monkeypatch): monkeypatch.setattr( ConnectionRouter, "watcher_send_proposed_file_changes_message", - lambda self, recipient, message: ( - sent.append( - (recipient, {fc.datasite_email for fc in message.proposed_file_changes}) - ) + lambda self, recipient, message: sent.append( + (recipient, {fc.datasite_email for fc in message.proposed_file_changes}) ), ) watcher = ds_manager.datasite_watcher_syncer From 8d4f8ad716b78c95758e488aae8d223d14874cca Mon Sep 17 00:00:00 2001 From: Pedro Werneck Date: Mon, 2 Nov 2026 14:46:20 -0300 Subject: [PATCH 2/3] fix: refuse a later settings file by its version, before its fields - The settings file's version is checked on the raw JSON before the fields are validated, so a later layout is refused by its version, not by a pydantic error. Only a later version is refused, as for checkpoints and rolling state. The key file keeps its own check. - MigrationRegistry.infer_protocol_version_to_use() takes the peer's ProtocolSchema. The router, JobStorage and DatasetStorage pass the schema, and the router's wrapper is gone. - The control test also checks that a pair speaks an earlier protocol, so an equality check fails it. - pair_speaks_protocol() documents MigrationError, the pin_syft_protocol docstring says what it does not hold back, and the syft-migration README lists protocol_version_at_least(). --- .../src/syft_datasets/dataset_storage.py | 5 +-- packages/syft-job/src/syft_job/job_storage.py | 5 +-- packages/syft-migration/README.md | 2 + .../src/syft_migration/registry.py | 18 ++++----- .../tests/test_protocol_floor.py | 24 ++++++------ syft/sync/connections/connection_router.py | 25 ++++--------- syft/sync/peers/peer_store.py | 37 +++++++++---------- tests/conftest.py | 4 ++ .../migrations/p2p/test_protocol_skew_pair.py | 3 ++ tests/unit/test_encryption_policy.py | 5 ++- 10 files changed, 62 insertions(+), 66 deletions(-) diff --git a/packages/syft-datasets/src/syft_datasets/dataset_storage.py b/packages/syft-datasets/src/syft_datasets/dataset_storage.py index a5531b6b3c6..20c4febf302 100644 --- a/packages/syft-datasets/src/syft_datasets/dataset_storage.py +++ b/packages/syft-datasets/src/syft_datasets/dataset_storage.py @@ -170,10 +170,7 @@ def negotiated_protocol_version_for_peer( """ schema = self.peer_schema(peer_email) if schema is not None: - return self.registry.infer_protocol_version_to_use( - peer_version=schema.version, - peer_min=schema.min_supported_version, - ) + return self.registry.infer_protocol_version_to_use(schema) if raise_on_unknown: raise MigrationError( f"No dataset protocol schema known for peer {peer_email!r}" diff --git a/packages/syft-job/src/syft_job/job_storage.py b/packages/syft-job/src/syft_job/job_storage.py index edc9ca4c26c..b79d3b4f190 100644 --- a/packages/syft-job/src/syft_job/job_storage.py +++ b/packages/syft-job/src/syft_job/job_storage.py @@ -90,10 +90,7 @@ def negotiated_protocol_version_for_peer( """ schema = self.peer_schema(peer_email) if schema is not None: - return self.registry.infer_protocol_version_to_use( - peer_version=schema.version, - peer_min=schema.min_supported_version, - ) + return self.registry.infer_protocol_version_to_use(schema) if raise_on_unknown: raise MigrationError( f"No job protocol schema known for peer {peer_email!r}" diff --git a/packages/syft-migration/README.md b/packages/syft-migration/README.md index 569bea45022..36ec9905dc6 100644 --- a/packages/syft-migration/README.md +++ b/packages/syft-migration/README.md @@ -15,6 +15,8 @@ serialized objects to a version the other side understands. the current + historical protocol schemas. - `MigrationService` — upgrades/downgrades objects, including to the version a peer's package version supports. +- `protocol_version_at_least` — compares two protocol versions as integers, so `"10"` is + later than `"2"`. Use it instead of a string or `int()` comparison. - `coverage` — checks a package runs against its own registry: `unregistered_objects` finds a versioned object filed into another package's registry, and `missing_upgrade_paths` / `missing_downgrade_paths` find a version that cannot reach diff --git a/packages/syft-migration/src/syft_migration/registry.py b/packages/syft-migration/src/syft_migration/registry.py index a699e8e3a3d..fe835a3b63d 100644 --- a/packages/syft-migration/src/syft_migration/registry.py +++ b/packages/syft-migration/src/syft_migration/registry.py @@ -229,28 +229,28 @@ def compute_protocol_schema(self) -> ProtocolSchema: }, ) - def infer_protocol_version_to_use( - self, peer_version: str, peer_min: str | None = None - ) -> str: - """The protocol version to speak with a peer. + def infer_protocol_version_to_use(self, peer_schema: ProtocolSchema) -> str: + """The protocol version to speak with a peer that advertises ``peer_schema``. Both sides speak the lower of the two current versions, because each side must read what the other writes. That version must also be at or above - both floors. A peer that publishes no floor is treated as ``"0"``, which - refuses nothing. + both floors. Raises MigrationError when no version satisfies both sides. """ - chosen = min(self.protocol_version, peer_version, key=_version_order) + chosen = min(self.protocol_version, peer_schema.version, key=_version_order) floor = max( - self.min_supported_protocol_version, peer_min or "0", key=_version_order + self.min_supported_protocol_version, + peer_schema.min_supported_version, + key=_version_order, ) if _version_order(chosen) < _version_order(floor): raise MigrationError( f"No usable {self.protocol_name} protocol version with this peer. " f"This client speaks {self.protocol_version} and reads down to " f"{self.min_supported_protocol_version}; the peer speaks " - f"{peer_version} and reads down to {peer_min or '0'}." + f"{peer_schema.version} and reads down to " + f"{peer_schema.min_supported_version}." ) return chosen diff --git a/packages/syft-migration/tests/test_protocol_floor.py b/packages/syft-migration/tests/test_protocol_floor.py index 78086ccf22b..a5c5a527b45 100644 --- a/packages/syft-migration/tests/test_protocol_floor.py +++ b/packages/syft-migration/tests/test_protocol_floor.py @@ -18,6 +18,13 @@ def _registry(protocol_version: str = "2", floor: str = "0") -> MigrationRegistr ) +def _peer(version: str, floor: str = "0") -> ProtocolSchema: + """The schema a peer advertises.""" + return ProtocolSchema( + protocol_name="p", version=version, min_supported_version=floor + ) + + def test_schema_floor_defaults_to_zero(): # A peer that predates the floor field says nothing, so it refuses nothing. schema = ProtocolSchema(protocol_name="p", version="1") @@ -36,32 +43,27 @@ def test_registry_floor_defaults_to_zero(): def test_the_lower_version_is_chosen(): reg = _registry(protocol_version="2") - assert reg.infer_protocol_version_to_use(peer_version="1") == "1" - assert reg.infer_protocol_version_to_use(peer_version="3") == "2" + assert reg.infer_protocol_version_to_use(_peer("1")) == "1" + assert reg.infer_protocol_version_to_use(_peer("3")) == "2" def test_versions_order_by_number(): reg = _registry(protocol_version="10") - assert reg.infer_protocol_version_to_use(peer_version="9") == "9" + assert reg.infer_protocol_version_to_use(_peer("9")) == "9" def test_our_floor_refuses_an_older_peer(): reg = _registry(protocol_version="2", floor="2") with pytest.raises(MigrationError, match="1"): - reg.infer_protocol_version_to_use(peer_version="1") + reg.infer_protocol_version_to_use(_peer("1")) def test_the_peer_floor_refuses_us(): reg = _registry(protocol_version="2", floor="0") with pytest.raises(MigrationError): - reg.infer_protocol_version_to_use(peer_version="3", peer_min="3") + reg.infer_protocol_version_to_use(_peer("3", floor="3")) def test_a_zero_floor_on_both_sides_refuses_nothing(): reg = _registry(protocol_version="5", floor="0") - assert reg.infer_protocol_version_to_use(peer_version="0", peer_min="0") == "0" - - -def test_an_unknown_peer_floor_is_treated_as_zero(): - reg = _registry(protocol_version="2", floor="0") - assert reg.infer_protocol_version_to_use(peer_version="1", peer_min=None) == "1" + assert reg.infer_protocol_version_to_use(_peer("0", floor="0")) == "0" diff --git a/syft/sync/connections/connection_router.py b/syft/sync/connections/connection_router.py index 0a930a51a29..851bf4ea828 100644 --- a/syft/sync/connections/connection_router.py +++ b/syft/sync/connections/connection_router.py @@ -4,7 +4,7 @@ from typing import TYPE_CHECKING, List, Optional from pydantic import BaseModel -from syft_migration import MigratableObject, ProtocolSchema, protocol_version_at_least +from syft_migration import MigratableObject, protocol_version_at_least from syft.migrations import ( PROTOCOL_NAME, @@ -37,20 +37,6 @@ logger = logging.getLogger(__name__) -def _protocol_version_to_use(peer_schema: ProtocolSchema) -> str: - """The syft protocol version to speak with a peer that advertises - ``peer_schema``: the lower of the two current versions, at or above both - floors. - - Raises: - MigrationError: when no version satisfies both sides. - """ - return client_registry.infer_protocol_version_to_use( - peer_version=peer_schema.version, - peer_min=peer_schema.min_supported_version, - ) - - class ConnectionRouter(BaseModel): """The one router of a manager. ``SyftboxManager.from_config`` builds it and hands it to every part, so they share one set of connections and one @@ -77,7 +63,9 @@ def negotiated_protocol_version_for_peer(self, peer_email: str) -> str | None: MigrationError: when no version satisfies both sides. """ schema = self.peer_store.protocol_schema(peer_email, PROTOCOL_NAME) - return None if schema is None else _protocol_version_to_use(schema) + if schema is None: + return None + return client_registry.infer_protocol_version_to_use(schema) def pair_speaks_protocol(self, peer_email: str, version: str) -> bool: """Whether this client and ``peer_email`` both speak syft protocol @@ -87,6 +75,9 @@ def pair_speaks_protocol(self, peer_email: str, version: str) -> bool: decides from the version file it read last, so one side can switch before the other. The new behaviour must still work while the peer uses the old one. + + Raises: + MigrationError: when no version satisfies both sides. """ negotiated = self.negotiated_protocol_version_for_peer(peer_email) return negotiated is not None and protocol_version_at_least(negotiated, version) @@ -117,7 +108,7 @@ def _downgrade_for_peer( "peer that speaks an earlier protocol cannot read this message." ) return message - protocol_version = _protocol_version_to_use(schema) + protocol_version = client_registry.infer_protocol_version_to_use(schema) if schema.version == protocol_version: # The peer speaks the negotiated version, so its advertised slim # schema is the target; computing our own full schema on every diff --git a/syft/sync/peers/peer_store.py b/syft/sync/peers/peer_store.py index c5afd6bff8f..3ffac197b9d 100644 --- a/syft/sync/peers/peer_store.py +++ b/syft/sync/peers/peer_store.py @@ -4,6 +4,7 @@ Shared between PeerManager and all ConnectionRouter instances. """ +import json import os import tempfile from pathlib import Path @@ -33,23 +34,11 @@ # Format of the crypto key file. Raise it when the layout of the file changes. # A file of any other version is refused. CRYPTO_KEYS_VERSION = 2 -# Format of the datasite settings file, under the same rule. +# Format of the datasite settings file. Raise it when the layout of the file +# changes. A file of a later version is refused. DATASITE_SETTINGS_VERSION = 1 -def _require_file_version(path: Path, version: int, expected: int) -> None: - """Refuse the file at ``path`` unless its format ``version`` is ``expected``. - - Raises: - ValueError: if ``version`` is not ``expected``. - """ - if version != expected: - raise ValueError( - f"The file at {path} has version {version}, and this client reads " - f"only version {expected}." - ) - - class _KeyFile(BaseModel): """The private key file of one identity. It holds the only copy of its keys.""" @@ -68,7 +57,11 @@ def read(cls, path: Path, email: str) -> "_KeyFile": another identity. """ key_file = cls.model_validate_json(path.read_text()) - _require_file_version(path, key_file.version, CRYPTO_KEYS_VERSION) + if key_file.version != CRYPTO_KEYS_VERSION: + raise ValueError( + f"The crypto key file at {path} has version {key_file.version}, " + f"and this client reads only version {CRYPTO_KEYS_VERSION}." + ) if key_file.email != email: raise ValueError( f"Encryption key file {path} belongs to {key_file.email!r}, " @@ -100,11 +93,17 @@ def read(cls, path: Path) -> "_DatasiteSettings": """Read the settings file at ``path``. Raises: - ValueError: if the file has another shape or version. + ValueError: if a later client wrote the file, or it has another + shape. """ - settings = cls.model_validate_json(path.read_text()) - _require_file_version(path, settings.version, DATASITE_SETTINGS_VERSION) - return settings + 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) def resolve_encryption_policy( diff --git a/tests/conftest.py b/tests/conftest.py index 9f75318ee23..6c6ddd4c2d5 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -97,6 +97,10 @@ def pin_syft_protocol(monkeypatch, advertised_schemas): afterwards, so they read the new version file. It wraps ``advertised_schemas``, so an advertised schema is capped too. + + Only the negotiation is held back. Code that reads + ``client_registry.protocol_version`` directly still acts at the next + version, so gate a new behaviour on ``ConnectionRouter.pair_speaks_protocol()``. """ # email of a held client -> the syft protocol version it is held at held_versions: dict[str, str] = {} diff --git a/tests/migrations/p2p/test_protocol_skew_pair.py b/tests/migrations/p2p/test_protocol_skew_pair.py index f85b872beab..f0c402f76f2 100644 --- a/tests/migrations/p2p/test_protocol_skew_pair.py +++ b/tests/migrations/p2p/test_protocol_skew_pair.py @@ -56,6 +56,9 @@ def test_pair_on_next_protocol_speaks_it(next_protocol): assert ds_router.pair_speaks_protocol(do_manager.email, next_protocol.next) assert do_router.pair_speaks_protocol(ds_manager.email, next_protocol.next) + # A pair on a protocol also speaks every earlier one. + assert ds_router.pair_speaks_protocol(do_manager.email, next_protocol.released) + assert do_router.pair_speaks_protocol(ds_manager.email, next_protocol.released) def test_unknown_peer_speaks_no_protocol(advertised_schemas): diff --git a/tests/unit/test_encryption_policy.py b/tests/unit/test_encryption_policy.py index 8d81b48f03b..2d3aa2c4fdd 100644 --- a/tests/unit/test_encryption_policy.py +++ b/tests/unit/test_encryption_policy.py @@ -281,12 +281,13 @@ def test_client_with_other_policy_than_disk_raises(tmp_path): resolve_encryption_policy(path, REQUIRED) -def test_settings_file_of_other_version_refused(tmp_path): +def test_settings_file_of_later_version_refused(tmp_path): path = tmp_path / "syft_settings.json" resolve_encryption_policy(path, OPTIONAL) later_version = DATASITE_SETTINGS_VERSION + 1 data = json.loads(path.read_text()) - data["version"] = later_version + # A later release that bumps the version also changes the fields. + data.update(version=later_version, new_setting=True) path.write_text(json.dumps(data)) with pytest.raises(ValueError, match=f"version {later_version}"): From 5212f115ffc988e8883c62fdce85e24a6d1f2a1b Mon Sep 17 00:00:00 2001 From: Pedro Werneck Date: Mon, 2 Nov 2026 15:04:03 -0300 Subject: [PATCH 3/3] test: drop filler from the protocol floor test names - _peer() is now _peer_schema(), because it returns the schema a peer advertises. - Four floor tests drop their articles, for example test_the_peer_floor_refuses_us() is now test_peer_floor_refuses_us(). --- .../tests/test_protocol_floor.py | 22 +++++++++---------- 1 file changed, 11 insertions(+), 11 deletions(-) diff --git a/packages/syft-migration/tests/test_protocol_floor.py b/packages/syft-migration/tests/test_protocol_floor.py index a5c5a527b45..2322130cf46 100644 --- a/packages/syft-migration/tests/test_protocol_floor.py +++ b/packages/syft-migration/tests/test_protocol_floor.py @@ -18,7 +18,7 @@ def _registry(protocol_version: str = "2", floor: str = "0") -> MigrationRegistr ) -def _peer(version: str, floor: str = "0") -> ProtocolSchema: +def _peer_schema(version: str, floor: str = "0") -> ProtocolSchema: """The schema a peer advertises.""" return ProtocolSchema( protocol_name="p", version=version, min_supported_version=floor @@ -41,29 +41,29 @@ def test_registry_floor_defaults_to_zero(): assert reg.min_supported_protocol_version == "0" -def test_the_lower_version_is_chosen(): +def test_lower_version_chosen(): reg = _registry(protocol_version="2") - assert reg.infer_protocol_version_to_use(_peer("1")) == "1" - assert reg.infer_protocol_version_to_use(_peer("3")) == "2" + assert reg.infer_protocol_version_to_use(_peer_schema("1")) == "1" + assert reg.infer_protocol_version_to_use(_peer_schema("3")) == "2" def test_versions_order_by_number(): reg = _registry(protocol_version="10") - assert reg.infer_protocol_version_to_use(_peer("9")) == "9" + assert reg.infer_protocol_version_to_use(_peer_schema("9")) == "9" -def test_our_floor_refuses_an_older_peer(): +def test_our_floor_refuses_older_peer(): reg = _registry(protocol_version="2", floor="2") with pytest.raises(MigrationError, match="1"): - reg.infer_protocol_version_to_use(_peer("1")) + reg.infer_protocol_version_to_use(_peer_schema("1")) -def test_the_peer_floor_refuses_us(): +def test_peer_floor_refuses_us(): reg = _registry(protocol_version="2", floor="0") with pytest.raises(MigrationError): - reg.infer_protocol_version_to_use(_peer("3", floor="3")) + reg.infer_protocol_version_to_use(_peer_schema("3", floor="3")) -def test_a_zero_floor_on_both_sides_refuses_nothing(): +def test_zero_floor_on_both_sides_refuses_nothing(): reg = _registry(protocol_version="5", floor="0") - assert reg.infer_protocol_version_to_use(_peer("0", floor="0")) == "0" + assert reg.infer_protocol_version_to_use(_peer_schema("0", floor="0")) == "0"