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/__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/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..2322130cf46 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_schema(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") @@ -34,34 +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_version="1") == "1" - assert reg.infer_protocol_version_to_use(peer_version="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_version="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_version="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_version="3", peer_min="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_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_schema("0", floor="0")) == "0" 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..851bf4ea828 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, 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, @@ -50,6 +51,37 @@ 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) + 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 + ``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. + + 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) + # ========================================================================= # MESSAGE SEND/RECEIVE (with encryption) # ========================================================================= @@ -68,7 +100,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 +108,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 = 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 0ccfab5c849..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,6 +34,9 @@ # 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. Raise it when the layout of the file +# changes. A file of a later version is refused. +DATASITE_SETTINGS_VERSION = 1 class _KeyFile(BaseModel): @@ -81,8 +85,26 @@ 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 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) + def resolve_encryption_policy( settings_path: Path | str, requested: EncryptionPolicy | None @@ -99,14 +121,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..6c6ddd4c2d5 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,75 @@ 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. + + 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] = {} + 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..f0c402f76f2 --- /dev/null +++ b/tests/migrations/p2p/test_protocol_skew_pair.py @@ -0,0 +1,72 @@ +"""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) + # 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): + 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..2d3aa2c4fdd 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,19 @@ def test_client_with_other_policy_than_disk_raises(tmp_path): resolve_encryption_policy(path, REQUIRED) +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()) + # 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}"): + 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 +526,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