feat: add a per-peer protocol switch and a cross-version test harness

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}"

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}"

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

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",
 ]

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.
 

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
 

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"

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):

syft/sync/connections/connection_router.py

@@ -4,26 +4,27 @@
 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,
 )
 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,
 )
 from syft.sync.messages.proposed_filechange import ProposedFileChangesMessage
 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,18 +100,15 @@ 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}. "
                 f"This client writes protocol {SYFT_CLIENT_PROTOCOL_VERSION}. A "
                 "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

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