From 9045fe5acd4d12cc6ff4173087363c2a142a7c98 Mon Sep 17 00:00:00 2001 From: rasswanth-s <43314053+rasswanth-s@users.noreply.github.com> Date: Tue, 13 Oct 2026 22:00:37 +0530 Subject: [PATCH 1/4] Added initial stub contracts --- syft/sync/sync/delivery_tracker.py | 103 +++++++++++ syft/sync/transport/__init__.py | 23 +++ syft/sync/transport/naming.py | 35 ++++ syft/sync/transport/protocols.py | 286 +++++++++++++++++++++++++++++ syft/sync/transport/storage.py | 70 +++++++ syft/sync/transport/types.py | 33 ++++ 6 files changed, 550 insertions(+) create mode 100644 syft/sync/sync/delivery_tracker.py create mode 100644 syft/sync/transport/__init__.py create mode 100644 syft/sync/transport/naming.py create mode 100644 syft/sync/transport/protocols.py create mode 100644 syft/sync/transport/storage.py create mode 100644 syft/sync/transport/types.py diff --git a/syft/sync/sync/delivery_tracker.py b/syft/sync/sync/delivery_tracker.py new file mode 100644 index 00000000000..fede3bb1d68 --- /dev/null +++ b/syft/sync/sync/delivery_tracker.py @@ -0,0 +1,103 @@ +"""What each peer was sent, so the data owner can resend, delete and catch up. + +STUB: the names and contracts are agreed; the bodies are to be written (PR 2). +A draft implementation exists from the planning work. + +For every peer the tracker keeps a view: datasite path -> hash of the version +that peer holds. Each sync compares what a peer may read with its view: + + the peer holds an old version -> send the new one + the peer holds nothing (new peer) -> send every file it may read + the peer may no longer read a file -> send a delete + the owner deleted the file -> send a delete + +The view changes only after the message carrying a version was written to the +peer's outbox, so a crash before the write leaves it unchanged and the +delivery is retried. Paths in flight are written down before routing and +cleared after the last outbox write, so a crash in between is finished by the +next sync. + +Open decisions (see research/merged_transport_plan.md): +- First run with no views: assume current peers hold what they may read + (proposed), or resend everything. +- A rejected peer approved again: reset its view and resend (proposed). +- Enclaves lose the views at every boot until they are also kept on Drive. +""" + +from pathlib import Path +from typing import Iterable + +from syft.sync.events.file_change_event import FileChangeEvent + +DELIVERY_VIEWS_FILENAME = "owner_delivery_views.json" +# Format of the views file. Raise it when the layout changes, and keep a read +# path for every earlier version. +DELIVERY_VIEWS_VERSION = 1 + + +class DeliveryTracker: + """Per-peer views of delivered file versions, persisted next to the owner cache. + + ``path`` None keeps the views in memory only, as the in-memory owner cache + does in tests. + """ + + def __init__(self, path: Path | None = None): + self.path = path + + # -- reading --------------------------------------------------------------- + def reload(self) -> None: + """Read the views from disk once; lookups use them until the next reload.""" + raise NotImplementedError + + @property + def initialized(self) -> bool: + """Whether views were ever recorded for this datasite.""" + raise NotImplementedError + + def has_view(self, peer: str) -> bool: + """Whether ``peer`` was ever delivered to; False means a new peer.""" + raise NotImplementedError + + def delivered_hash(self, peer: str, path: str) -> str | None: + """The hash of the version of ``path`` that ``peer`` holds, or None.""" + raise NotImplementedError + + def holders(self, path: str, peers: Iterable[str]) -> frozenset[str]: + """The peers among ``peers`` that hold some version of ``path``.""" + raise NotImplementedError + + # -- writing --------------------------------------------------------------- + def mark_initialized(self) -> None: + """Record that views exist, with DELIVERY_VIEWS_VERSION.""" + raise NotImplementedError + + def set_view(self, peer: str, view: dict[str, str]) -> None: + """Replace ``peer``'s whole view, e.g. to seed it on the first run.""" + raise NotImplementedError + + def record(self, peer: str, events: Iterable[FileChangeEvent]) -> None: + """Record that ``events`` reached ``peer``'s outbox. + + A delete removes the path from the view; any other event sets its hash. + Call only after the outbox write succeeded. + """ + raise NotImplementedError + + def reset(self, peer: str) -> None: + """Forget ``peer``'s view, so the next sync sends it everything again.""" + raise NotImplementedError + + # -- paths in flight ------------------------------------------------------- + @property + def pending(self) -> list[str]: + """Paths whose routing started but did not finish.""" + raise NotImplementedError + + def mark_pending(self, paths: Iterable[str]) -> None: + """Write ``paths`` down before routing them.""" + raise NotImplementedError + + def clear_pending(self) -> None: + """Clear the paths in flight once every outbox write is done.""" + raise NotImplementedError diff --git a/syft/sync/transport/__init__.py b/syft/sync/transport/__init__.py new file mode 100644 index 00000000000..f94132df8a4 --- /dev/null +++ b/syft/sync/transport/__init__.py @@ -0,0 +1,23 @@ +"""Backend-neutral transport contracts for the syft sync engine.""" + +from syft.sync.transport.protocols import ( + Collections, + Mailbox, + PeerState, + PrivateStore, + Transport, +) +from syft.sync.transport.storage import OwnerStorage, WatcherStorage +from syft.sync.transport.types import Entry, Lane + +__all__ = [ + "Collections", + "Entry", + "Lane", + "Mailbox", + "OwnerStorage", + "PeerState", + "PrivateStore", + "Transport", + "WatcherStorage", +] diff --git a/syft/sync/transport/naming.py b/syft/sync/transport/naming.py new file mode 100644 index 00000000000..87877b0efa0 --- /dev/null +++ b/syft/sync/transport/naming.py @@ -0,0 +1,35 @@ +"""Syft's naming rules for stored files and documents. + +These rules belong to syft, not to any backend: a backend stores names, and the +sync layer gives them meaning. They move here from the Drive transport one at a +time. +""" + +EVENTS_MESSAGE_PREFIX = "syfteventsmessagev3_" +PROPOSALS_MESSAGE_PREFIX = "msgv2_" + +# SYFT_peers.json is a flat map of peer email to entry, so a version at the top +# level would look like a peer email. The version goes under this reserved key. +# A client written before the key reads a peer state from that entry and fails. +# The key therefore never appears as a peer. +PEERS_META_KEY = "_meta" +# Shape of one entry in SYFT_peers.json. Raise this when an entry changes. A file +# with no reserved entry was written before the version, and is version 0. +SYFT_PEERS_VERSION = 1 + + +def timestamp_from_message_name(name: str) -> float | None: + """The send time a message file carries in its name, or None. + + Both message formats put the time second: ``syfteventsmessagev3__`` + for events and ``msgv2__`` for proposals. + """ + if not name.startswith((EVENTS_MESSAGE_PREFIX, PROPOSALS_MESSAGE_PREFIX)): + return None + parts = name.split("_") + if len(parts) < 2: + return None + try: + return float(parts[1]) + except ValueError: + return None diff --git a/syft/sync/transport/protocols.py b/syft/sync/transport/protocols.py new file mode 100644 index 00000000000..fdf6c758830 --- /dev/null +++ b/syft/sync/transport/protocols.py @@ -0,0 +1,286 @@ +"""The four jobs a transport backend does, one abstract class each. + +A backend implements the classes it can. The sync layer above never sees +backend ids: it passes back the ``Entry`` objects and handles a backend +returns. + +These are the version-1 contracts. They name today's operations in +backend-neutral terms so the sync layer can depend on them while the Drive +code moves behind them. Later versions narrow them to generic storage +primitives once syft's naming rules have moved into ``naming.py``. +""" + +from __future__ import annotations + +from abc import ABC, abstractmethod +from typing import TYPE_CHECKING + +from pydantic import BaseModel, ConfigDict + +from syft.sync.platforms.base_platform import BasePlatform +from syft.sync.transport.types import Entry, Lane + +if TYPE_CHECKING: + from syft.sync.connections.base_connection import FileCollection + from syft.sync.version.version_info import VersionInfo + + +class Mailbox(ABC): + """Messages between two peers, kept until the recipient processes them.""" + + @abstractmethod + def send(self, lane: Lane, peer: str, name: str, data: bytes) -> None: + """Store ``data`` as ``name`` in the ``lane`` channel towards ``peer``. + + Raises when the channel does not exist, so a send never vanishes. + """ + + @abstractmethod + def list( + self, lane: Lane, peer: str, since_timestamp: float | None = None + ) -> list[Entry]: + """The messages from ``peer`` in ``lane``, oldest first. + + PROPOSALS lists every unacknowledged message, and raises when the + channel does not exist. EVENTS lists the messages sent after + ``since_timestamp``, and returns nothing when the channel does not + exist yet. + """ + + @abstractmethod + def fetch(self, entry: Entry) -> bytes: + """The bytes of one listed message.""" + + @abstractmethod + def ack(self, lane: Lane, peer: str, entry: Entry) -> None: + """Record that ``entry`` from ``peer`` was processed. + + An acknowledged PROPOSALS message never appears in ``list`` again. For + EVENTS this does nothing yet: the reader tracks its own position. + """ + + +class PrivateStore(ABC): + """My own storage: only I write it, only I read it, it is never shared.""" + + # -- event log ------------------------------------------------------------- + @abstractmethod + def append_event(self, name: str, data: bytes) -> str: + """Append one events message to my event log; return its handle.""" + + @abstractmethod + def fetch(self, handle: str) -> bytes: + """The bytes stored under ``handle``.""" + + @abstractmethod + def list_event_handles(self, since_timestamp: float | None = None) -> list[str]: + """Handles of my events messages sent after ``since_timestamp``.""" + + @abstractmethod + def fetch_all_events(self) -> list[bytes]: + """Every events message in my event log.""" + + @abstractmethod + def fetch_events_since(self, since_timestamp: float) -> list[bytes]: + """My events messages sent after ``since_timestamp``.""" + + @abstractmethod + def count_events_since(self, since_timestamp: float | None) -> int: + """How many events messages were sent after ``since_timestamp``.""" + + # -- checkpoints ----------------------------------------------------------- + @abstractmethod + def put_checkpoint(self, name: str, data: bytes) -> str: + """Store a full checkpoint; it replaces every earlier one.""" + + @abstractmethod + def latest_checkpoint(self) -> bytes | None: + """The newest full checkpoint, or None.""" + + @abstractmethod + def put_incremental(self, name: str, data: bytes) -> str: + """Store an incremental checkpoint.""" + + @abstractmethod + def all_incrementals(self) -> list[bytes]: + """Every incremental checkpoint since the last full one.""" + + @abstractmethod + def incremental_count(self) -> int: + """How many incremental checkpoints are stored.""" + + @abstractmethod + def next_incremental_sequence(self) -> int: + """The sequence number the next incremental checkpoint takes.""" + + @abstractmethod + def delete_incrementals(self) -> None: + """Remove every incremental checkpoint.""" + + # -- rolling state --------------------------------------------------------- + @abstractmethod + def put_rolling_state(self, name: str, data: bytes) -> str: + """Store the rolling state, overwriting the previous one.""" + + @abstractmethod + def get_rolling_state(self) -> bytes | None: + """The rolling state, or None.""" + + @abstractmethod + def delete_rolling_state(self) -> None: + """Remove the rolling state.""" + + # -- peer registry (SYFT_peers.json) --------------------------------------- + @abstractmethod + def read_peer_registry(self, force_download: bool = False) -> dict: + """My record of every peer and its state, keyed by email.""" + + @abstractmethod + def update_peer_registry( + self, peer: str, state: str, public_encryption_bundle: dict | None = None + ) -> None: + """Record ``peer``'s state, and its key bundle when given.""" + + # -- watcher area: the data scientist's memory, one area per peer ------------ + # Placeholders until watcher storage lands (PR 3). A backend that does not + # override them has no watcher storage yet. + + def put_watcher_checkpoint(self, peer: str, name: str, data: bytes) -> str: + """Store a snapshot of every file received from ``peer``; it replaces + the previous one.""" + raise NotImplementedError("Watcher storage is not implemented yet") + + def latest_watcher_checkpoint(self, peer: str) -> bytes | None: + """The newest snapshot of files received from ``peer``, or None.""" + raise NotImplementedError("Watcher storage is not implemented yet") + + def put_watcher_rolling_state(self, peer: str, name: str, data: bytes) -> str: + """Store the changes received from ``peer`` since the snapshot, and the + ids of messages already processed.""" + raise NotImplementedError("Watcher storage is not implemented yet") + + def get_watcher_rolling_state(self, peer: str) -> bytes | None: + """The rolling state for ``peer``, or None.""" + raise NotImplementedError("Watcher storage is not implemented yet") + + def delete_watcher_state(self, peer: str) -> None: + """Forget everything received from ``peer``, e.g. before a full resend.""" + raise NotImplementedError("Watcher storage is not implemented yet") + + +class Collections(ABC): + """Sets of files a data owner shares with named readers (datasets).""" + + @abstractmethod + def create(self, prefix: str, tag: str, content_hash: str, owner_email: str) -> str: + """Create the collection if it does not exist; return its handle.""" + + @abstractmethod + def mark_shared_with_any(self, prefix: str, tag: str, content_hash: str) -> None: + """Flag the collection as meant for every approved peer.""" + + @abstractmethod + def grant_read( + self, prefix: str, tag: str, content_hash: str, users: list[str] + ) -> None: + """Let ``users`` read the collection. Granting twice is harmless.""" + + @abstractmethod + def upload( + self, prefix: str, tag: str, content_hash: str, files: dict[str, bytes] + ) -> None: + """Store ``files`` in the collection.""" + + @abstractmethod + def list_mine(self, prefix: str) -> list[str]: + """Names of the collections I published under ``prefix``.""" + + @abstractmethod + def list_mine_with_permissions(self, prefix: str) -> list[FileCollection]: + """The collections I published under ``prefix``, with their flags.""" + + @abstractmethod + def delete(self, prefix: str, tag: str) -> None: + """Remove every layout of ``tag`` published under ``prefix``.""" + + @abstractmethod + def list_shared_with_me(self, prefix: str) -> list[dict]: + """Collections peers shared with me, each with owner_email, tag, + content_hash and variant.""" + + @abstractmethod + def download( + self, prefix: str, tag: str, content_hash: str, owner_email: str + ) -> dict[str, bytes]: + """Every file of one collection a peer shared with me.""" + + @abstractmethod + def list_files( + self, prefix: str, tag: str, content_hash: str, owner_email: str + ) -> list[dict]: + """The files of one shared collection, each with file_id and file_name.""" + + @abstractmethod + def fetch_file(self, handle: str) -> bytes: + """The bytes of one collection file.""" + + +class PeerState(ABC): + """Finding and trusting a peer: identity, channels, published documents. + + Not to be confused with ``syft.sync.peers.peer.PeerState``, the enum of + handshake states. + """ + + @abstractmethod + def whoami(self) -> str: + """The email of the account behind this backend's credentials.""" + + @abstractmethod + def open_channels(self, peer: str) -> None: + """Create both mailbox channels with ``peer`` and let them write.""" + + @abstractmethod + def incoming_requests(self) -> list[str]: + """Emails of peers who opened channels towards me.""" + + @abstractmethod + def publish_key_bundle(self, peer: str, bundle_json: str) -> None: + """Publish my public key bundle for ``peer``.""" + + @abstractmethod + def grant_key_bundle_read(self, peer: str) -> None: + """Let ``peer`` read the key bundles I publish.""" + + @abstractmethod + def read_peer_key_bundle(self, peer: str) -> str | None: + """The key bundle ``peer`` published for me, or None.""" + + @abstractmethod + def publish_version(self, version_info: VersionInfo) -> None: + """Publish my version document.""" + + @abstractmethod + def read_own_version(self) -> VersionInfo | None: + """My published version document, or None.""" + + @abstractmethod + def read_peer_version(self, peer: str) -> VersionInfo | None: + """``peer``'s published version document, or None.""" + + @abstractmethod + def grant_version_read(self, peer: str) -> None: + """Let ``peer`` read my version document.""" + + +class Transport(BaseModel): + """One backend's implementation of the four jobs.""" + + model_config = ConfigDict(frozen=True, arbitrary_types_allowed=True) + + mailbox: Mailbox + private_store: PrivateStore + collections: Collections + peer_state: PeerState + # Recorded on each Peer so a peer list says which backend reaches it. + platform: BasePlatform diff --git a/syft/sync/transport/storage.py b/syft/sync/transport/storage.py new file mode 100644 index 00000000000..9d6e01e9787 --- /dev/null +++ b/syft/sync/transport/storage.py @@ -0,0 +1,70 @@ +"""Readable views over ``PrivateStore``: one for each role. + +A backend implements ``PrivateStore`` once. Owner code holds an +``OwnerStorage`` and data scientist code holds a ``WatcherStorage``, so each +sees only its own methods and cannot write into the other's area. +""" + +from syft.sync.transport.protocols import PrivateStore + + +class OwnerStorage: + """The data owner's own storage: event log, checkpoints, rolling state.""" + + def __init__(self, store: PrivateStore): + self._store = store + + def append_event(self, name: str, data: bytes) -> str: + return self._store.append_event(name, data) + + def fetch(self, handle: str) -> bytes: + return self._store.fetch(handle) + + def fetch_events_since(self, since_timestamp: float) -> list[bytes]: + return self._store.fetch_events_since(since_timestamp) + + def put_checkpoint(self, name: str, data: bytes) -> str: + return self._store.put_checkpoint(name, data) + + def latest_checkpoint(self) -> bytes | None: + return self._store.latest_checkpoint() + + def put_incremental(self, name: str, data: bytes) -> str: + return self._store.put_incremental(name, data) + + def all_incrementals(self) -> list[bytes]: + return self._store.all_incrementals() + + def put_rolling_state(self, name: str, data: bytes) -> str: + return self._store.put_rolling_state(name, data) + + def get_rolling_state(self) -> bytes | None: + return self._store.get_rolling_state() + + +class WatcherStorage: + """The data scientist's memory of what one owner sent it. + + Stored in the scientist's own private storage, never shared, so it + survives a lost local disk such as a Colab runtime reset. + """ + + def __init__(self, store: PrivateStore, peer: str): + self._store = store + self.peer = peer + + def put_checkpoint(self, name: str, data: bytes) -> str: + return self._store.put_watcher_checkpoint(self.peer, name, data) + + def latest_checkpoint(self) -> bytes | None: + return self._store.latest_watcher_checkpoint(self.peer) + + def put_rolling_state(self, name: str, data: bytes) -> str: + return self._store.put_watcher_rolling_state(self.peer, name, data) + + def get_rolling_state(self) -> bytes | None: + return self._store.get_watcher_rolling_state(self.peer) + + def reset(self) -> None: + """Forget everything from this owner, before asking for a full resend.""" + self._store.delete_watcher_state(self.peer) diff --git a/syft/sync/transport/types.py b/syft/sync/transport/types.py new file mode 100644 index 00000000000..3f22d6295f9 --- /dev/null +++ b/syft/sync/transport/types.py @@ -0,0 +1,33 @@ +"""Types shared by every transport backend.""" + +from enum import Enum + +from pydantic import BaseModel, ConfigDict + + +class Lane(str, Enum): + """The two message streams between a pair of peers. + + The two streams run in opposite directions and live in different places, + and a dual-role pair runs both at once, so a mailbox call always names its + lane. + """ + + # Data scientist -> data owner: proposed file changes (job submissions). + PROPOSALS = "proposals" + # Data owner -> data scientist: accepted file change events (results). + EVENTS = "events" + + +class Entry(BaseModel): + """One stored message or file, as a backend lists it. + + ``name`` is the syft file name. ``handle`` is whatever the backend needs to + fetch or acknowledge the entry again; callers only pass it back. It is None + when the entry is known by name only. + """ + + model_config = ConfigDict(frozen=True) + + name: str + handle: str | None = None From ee354b53599eb7d6f965873cf5a8b7b8d3a34c3f Mon Sep 17 00:00:00 2001 From: rasswanth-s <43314053+rasswanth-s@users.noreply.github.com> Date: Fri, 16 Oct 2026 10:15:17 +0530 Subject: [PATCH 2/4] refactor(transport): use the V2 P2PConnection message contract --- syft/sync/messages/raw_message.py | 14 +--- syft/sync/transport/__init__.py | 11 +-- syft/sync/transport/protocols.py | 58 +++++++------- syft/sync/transport/types.py | 37 +++++---- tests/unit/test_transport_contracts.py | 104 +++++++++++++++++++++++++ 5 files changed, 167 insertions(+), 57 deletions(-) create mode 100644 tests/unit/test_transport_contracts.py diff --git a/syft/sync/messages/raw_message.py b/syft/sync/messages/raw_message.py index ad5077f589a..6ce9a329196 100644 --- a/syft/sync/messages/raw_message.py +++ b/syft/sync/messages/raw_message.py @@ -1,13 +1,5 @@ -from pydantic import BaseModel +"""``RawMessage`` lives with the other transport types; this path keeps working.""" +from syft.sync.transport.types import RawMessage -class RawMessage(BaseModel): - """Generic container for downloaded bytes + metadata. - - Transport returns this — no deserialization, no encryption awareness. - """ - - data: bytes - sender_email: str | None = None - platform_id: str | None = None - filename: str | None = None +__all__ = ["RawMessage"] diff --git a/syft/sync/transport/__init__.py b/syft/sync/transport/__init__.py index f94132df8a4..0f849585161 100644 --- a/syft/sync/transport/__init__.py +++ b/syft/sync/transport/__init__.py @@ -2,22 +2,23 @@ from syft.sync.transport.protocols import ( Collections, - Mailbox, + P2PConnection, PeerState, PrivateStore, Transport, ) from syft.sync.transport.storage import OwnerStorage, WatcherStorage -from syft.sync.transport.types import Entry, Lane +from syft.sync.transport.types import MessageKind, RawMessage, SendStatus __all__ = [ "Collections", - "Entry", - "Lane", - "Mailbox", + "MessageKind", "OwnerStorage", + "P2PConnection", "PeerState", "PrivateStore", + "RawMessage", + "SendStatus", "Transport", "WatcherStorage", ] diff --git a/syft/sync/transport/protocols.py b/syft/sync/transport/protocols.py index fdf6c758830..0d3dab275f0 100644 --- a/syft/sync/transport/protocols.py +++ b/syft/sync/transport/protocols.py @@ -1,8 +1,7 @@ """The four jobs a transport backend does, one abstract class each. A backend implements the classes it can. The sync layer above never sees -backend ids: it passes back the ``Entry`` objects and handles a backend -returns. +backend ids: it passes back the messages and handles a backend returns. These are the version-1 contracts. They name today's operations in backend-neutral terms so the sync layer can depend on them while the Drive @@ -18,47 +17,54 @@ from pydantic import BaseModel, ConfigDict from syft.sync.platforms.base_platform import BasePlatform -from syft.sync.transport.types import Entry, Lane +from syft.sync.transport.types import MessageKind, RawMessage, SendStatus if TYPE_CHECKING: from syft.sync.connections.base_connection import FileCollection from syft.sync.version.version_info import VersionInfo -class Mailbox(ABC): - """Messages between two peers, kept until the recipient processes them.""" +class P2PConnection(ABC): + """Messages between peers, kept until the recipient acknowledges them. - @abstractmethod - def send(self, lane: Lane, peer: str, name: str, data: bytes) -> None: - """Store ``data`` as ``name`` in the ``lane`` channel towards ``peer``. - - Raises when the channel does not exist, so a send never vanishes. - """ + One rule for every message, whatever its kind: ``receive`` returns the + messages not acknowledged yet, and ``ack`` finishes one. On Drive a written + message is ``DELIVERED`` and ``ack`` removes it. On a live link such as + WebRTC a message is ``QUEUED`` until the peer acknowledges it, which + ``delivered()`` then reports. + """ @abstractmethod - def list( - self, lane: Lane, peer: str, since_timestamp: float | None = None - ) -> list[Entry]: - """The messages from ``peer`` in ``lane``, oldest first. + def send( + self, peer: str, msg_id: str, data: bytes, kind: MessageKind + ) -> SendStatus: + """Send ``data`` to ``peer`` under ``msg_id``. - PROPOSALS lists every unacknowledged message, and raises when the - channel does not exist. EVENTS lists the messages sent after - ``since_timestamp``, and returns nothing when the channel does not - exist yet. + ``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. """ @abstractmethod - def fetch(self, entry: Entry) -> bytes: - """The bytes of one listed message.""" + def delivered(self) -> list[str]: + """The ``msg_id`` of each queued message the peer acknowledged since + the last call. A connection that only returns ``DELIVERED`` returns + nothing.""" @abstractmethod - def ack(self, lane: Lane, peer: str, entry: Entry) -> None: - """Record that ``entry`` from ``peer`` was processed. + def receive(self, max_messages: int | None = None) -> list[RawMessage]: + """Messages from every peer that are not acknowledged yet, oldest first + per sender, at most ``max_messages`` of them. - An acknowledged PROPOSALS message never appears in ``list`` again. For - EVENTS this does nothing yet: the reader tracks its own position. + A peer whose channel does not exist yet has no messages. The connection + sets each ``RawMessage.sender`` from where the message came from. """ + @abstractmethod + def ack(self, message: RawMessage) -> None: + """Record that ``message`` was processed; ``receive`` never returns it + again.""" + class PrivateStore(ABC): """My own storage: only I write it, only I read it, it is never shared.""" @@ -278,7 +284,7 @@ class Transport(BaseModel): model_config = ConfigDict(frozen=True, arbitrary_types_allowed=True) - mailbox: Mailbox + p2p: P2PConnection private_store: PrivateStore collections: Collections peer_state: PeerState diff --git a/syft/sync/transport/types.py b/syft/sync/transport/types.py index 3f22d6295f9..1f9cffa0d36 100644 --- a/syft/sync/transport/types.py +++ b/syft/sync/transport/types.py @@ -5,29 +5,36 @@ from pydantic import BaseModel, ConfigDict -class Lane(str, Enum): - """The two message streams between a pair of peers. - - The two streams run in opposite directions and live in different places, - and a dual-role pair runs both at once, so a mailbox call always names its - lane. - """ +class MessageKind(str, Enum): + """What a message carries. Both kinds can travel between the same two peers.""" # Data scientist -> data owner: proposed file changes (job submissions). - PROPOSALS = "proposals" + PROPOSED_CHANGES = "proposed_changes" # Data owner -> data scientist: accepted file change events (results). - EVENTS = "events" + FILE_EVENTS = "file_events" + + +class SendStatus(str, Enum): + """What ``P2PConnection.send`` achieved.""" + + # Stored where the peer will find it, even while it is offline. + DELIVERED = "delivered" + # Not stored yet; confirmed later through ``P2PConnection.delivered()``. + QUEUED = "queued" -class Entry(BaseModel): - """One stored message or file, as a backend lists it. +class RawMessage(BaseModel): + """One received message, as bytes, before decoding or decryption. - ``name`` is the syft file name. ``handle`` is whatever the backend needs to - fetch or acknowledge the entry again; callers only pass it back. It is None - when the entry is known by name only. + ``sender`` is set by the connection from where the message came from, + never from its contents. ``handle`` is opaque: only the connection that + returned the message reads it, to acknowledge it. """ model_config = ConfigDict(frozen=True) - name: str + sender: str + msg_id: str + kind: MessageKind + data: bytes handle: str | None = None diff --git a/tests/unit/test_transport_contracts.py b/tests/unit/test_transport_contracts.py new file mode 100644 index 00000000000..c8e1b0c30cc --- /dev/null +++ b/tests/unit/test_transport_contracts.py @@ -0,0 +1,104 @@ +"""The transport contracts: the abstract jobs, the role wrappers, the naming +rules and the DeliveryTracker stub. No backend implements them yet.""" + +from unittest.mock import Mock + +import pytest +from pydantic import ValidationError + +from syft.sync.platforms.gdrive_files_platform import GdriveFilesPlatform +from syft.sync.sync.delivery_tracker import DeliveryTracker +from syft.sync.transport import ( + Collections, + MessageKind, + P2PConnection, + RawMessage, + OwnerStorage, + PeerState, + PrivateStore, + Transport, + WatcherStorage, +) +from syft.sync.transport.naming import timestamp_from_message_name + + +@pytest.mark.parametrize("job", [P2PConnection, PrivateStore, Collections, PeerState]) +def test_a_job_cannot_be_used_without_a_backend(job): + with pytest.raises(TypeError): + job() + + +def test_owner_storage_forwards_to_the_private_store(): + store = Mock(spec=PrivateStore) + store.get_rolling_state.return_value = b"state" + owner = OwnerStorage(store) + owner.put_rolling_state("rolling_state_1.0.tar.gz", b"state") + store.put_rolling_state.assert_called_once_with( + "rolling_state_1.0.tar.gz", b"state" + ) + assert owner.get_rolling_state() == b"state" + + +def test_watcher_storage_names_its_peer_on_every_call(): + store = Mock(spec=PrivateStore) + watcher = WatcherStorage(store, "do@example.com") + watcher.put_checkpoint("checkpoint_1.0.tar.gz", b"snapshot") + watcher.reset() + store.put_watcher_checkpoint.assert_called_once_with( + "do@example.com", "checkpoint_1.0.tar.gz", b"snapshot" + ) + store.delete_watcher_state.assert_called_once_with("do@example.com") + + +def test_watcher_area_is_not_implemented_by_default(): + with pytest.raises(NotImplementedError): + PrivateStore.latest_watcher_checkpoint(object(), "do@example.com") + + +@pytest.mark.parametrize( + "name, expected", + [ + ("syfteventsmessagev3_1759800000.95_abc.tar.gz", 1759800000.95), + ("msgv2_12.5_abc.tar.gz", 12.5), + ("checkpoint_12.5.tar.gz", None), + ("msgv2_notanumber_abc.tar.gz", None), + ], +) +def test_message_names_carry_their_send_time(name, expected): + assert timestamp_from_message_name(name) == expected + + +def test_delivery_tracker_is_a_stub_until_implemented(): + tracker = DeliveryTracker(path=None) + with pytest.raises(NotImplementedError): + tracker.has_view("ds@example.com") + + +def test_a_transport_bundles_the_four_jobs_and_cannot_be_changed(): + transport = Transport( + p2p=Mock(spec=P2PConnection), + private_store=Mock(spec=PrivateStore), + collections=Mock(spec=Collections), + peer_state=Mock(spec=PeerState), + platform=GdriveFilesPlatform(), + ) + with pytest.raises(ValidationError): + transport.p2p = Mock(spec=P2PConnection) + + +def test_a_received_message_cannot_be_changed(): + message = RawMessage( + sender="ds@example.com", + msg_id="msgv2_1.0_abc", + kind=MessageKind.PROPOSED_CHANGES, + data=b"job", + handle="file-id", + ) + with pytest.raises(ValidationError): + message.sender = "someone-else@example.com" + + +def test_the_old_raw_message_import_path_still_works(): + from syft.sync.messages.raw_message import RawMessage as OldPathRawMessage + + assert OldPathRawMessage is RawMessage From 0d0de89076bf96c68699a6269edf949c28f6be97 Mon Sep 17 00:00:00 2001 From: rasswanth-s <43314053+rasswanth-s@users.noreply.github.com> Date: Fri, 16 Oct 2026 11:01:57 +0530 Subject: [PATCH 3/4] refactor(transport): split storage by role and rename peering --- syft/sync/transport/__init__.py | 13 +-- syft/sync/transport/protocols.py | 146 ++++++++++++++----------- syft/sync/transport/storage.py | 70 ------------ tests/unit/test_transport_contracts.py | 111 +++++++++++-------- 4 files changed, 153 insertions(+), 187 deletions(-) delete mode 100644 syft/sync/transport/storage.py diff --git a/syft/sync/transport/__init__.py b/syft/sync/transport/__init__.py index 0f849585161..f7460ff2e7e 100644 --- a/syft/sync/transport/__init__.py +++ b/syft/sync/transport/__init__.py @@ -2,23 +2,22 @@ from syft.sync.transport.protocols import ( Collections, + OwnerStore, P2PConnection, - PeerState, - PrivateStore, + Peering, Transport, + WatcherStore, ) -from syft.sync.transport.storage import OwnerStorage, WatcherStorage from syft.sync.transport.types import MessageKind, RawMessage, SendStatus __all__ = [ "Collections", "MessageKind", - "OwnerStorage", + "OwnerStore", "P2PConnection", - "PeerState", - "PrivateStore", + "Peering", "RawMessage", "SendStatus", "Transport", - "WatcherStorage", + "WatcherStore", ] diff --git a/syft/sync/transport/protocols.py b/syft/sync/transport/protocols.py index 0d3dab275f0..f8b4ec2e62e 100644 --- a/syft/sync/transport/protocols.py +++ b/syft/sync/transport/protocols.py @@ -1,12 +1,14 @@ -"""The four jobs a transport backend does, one abstract class each. +"""The jobs a transport backend does, one abstract class each. -A backend implements the classes it can. The sync layer above never sees -backend ids: it passes back the messages and handles a backend returns. +``P2PConnection`` carries messages between peers, ``OwnerStore`` and +``WatcherStore`` hold each role's own storage, ``Collections`` shares datasets, +and ``Peering`` finds and trusts peers. A backend implements the classes it +can. The sync layer above never sees backend ids: it passes back the messages +and handles a backend returns. -These are the version-1 contracts. They name today's operations in -backend-neutral terms so the sync layer can depend on them while the Drive -code moves behind them. Later versions narrow them to generic storage -primitives once syft's naming rules have moved into ``naming.py``. +Each class keeps its abstract methods to the few a backend must write. Methods +that only combine those are ordinary methods on the base class, which a +backend may override when it can do them faster. """ from __future__ import annotations @@ -66,33 +68,37 @@ def ack(self, message: RawMessage) -> None: again.""" -class PrivateStore(ABC): - """My own storage: only I write it, only I read it, it is never shared.""" +class OwnerStore(ABC): + """The data owner's own storage: event log, checkpoints, rolling state. + + Only the owner writes and reads it, and it is never shared. + """ # -- event log ------------------------------------------------------------- @abstractmethod def append_event(self, name: str, data: bytes) -> str: - """Append one events message to my event log; return its handle.""" + """Append one events message to the event log; return its handle.""" + + @abstractmethod + def list_event_handles(self, since_timestamp: float | None = None) -> list[str]: + """Handles of the events messages sent after ``since_timestamp``, oldest + first; every message when it is None.""" @abstractmethod def fetch(self, handle: str) -> bytes: """The bytes stored under ``handle``.""" - @abstractmethod - def list_event_handles(self, since_timestamp: float | None = None) -> list[str]: - """Handles of my events messages sent after ``since_timestamp``.""" + def fetch_events_since(self, since_timestamp: float | None) -> list[bytes]: + """The events messages sent after ``since_timestamp``, oldest first.""" + return [self.fetch(h) for h in self.list_event_handles(since_timestamp)] - @abstractmethod def fetch_all_events(self) -> list[bytes]: - """Every events message in my event log.""" - - @abstractmethod - def fetch_events_since(self, since_timestamp: float) -> list[bytes]: - """My events messages sent after ``since_timestamp``.""" + """Every events message in the event log, oldest first.""" + return self.fetch_events_since(None) - @abstractmethod def count_events_since(self, since_timestamp: float | None) -> int: """How many events messages were sent after ``since_timestamp``.""" + return len(self.list_event_handles(since_timestamp)) # -- checkpoints ----------------------------------------------------------- @abstractmethod @@ -113,7 +119,7 @@ def all_incrementals(self) -> list[bytes]: @abstractmethod def incremental_count(self) -> int: - """How many incremental checkpoints are stored.""" + """How many incremental checkpoints are stored, without fetching them.""" @abstractmethod def next_incremental_sequence(self) -> int: @@ -136,42 +142,36 @@ def get_rolling_state(self) -> bytes | None: def delete_rolling_state(self) -> None: """Remove the rolling state.""" - # -- peer registry (SYFT_peers.json) --------------------------------------- - @abstractmethod - def read_peer_registry(self, force_download: bool = False) -> dict: - """My record of every peer and its state, keyed by email.""" - @abstractmethod - def update_peer_registry( - self, peer: str, state: str, public_encryption_bundle: dict | None = None - ) -> None: - """Record ``peer``'s state, and its key bundle when given.""" +class WatcherStore(ABC): + """The data scientist's memory of what each owner sent it. - # -- watcher area: the data scientist's memory, one area per peer ------------ - # Placeholders until watcher storage lands (PR 3). A backend that does not - # override them has no watcher storage yet. + One area per owner, in the scientist's own storage, never shared, so it + survives a lost local disk such as a Colab runtime reset. + """ - def put_watcher_checkpoint(self, peer: str, name: str, data: bytes) -> str: + @abstractmethod + def put_checkpoint(self, peer: str, name: str, data: bytes) -> str: """Store a snapshot of every file received from ``peer``; it replaces the previous one.""" - raise NotImplementedError("Watcher storage is not implemented yet") - def latest_watcher_checkpoint(self, peer: str) -> bytes | None: + @abstractmethod + def latest_checkpoint(self, peer: str) -> bytes | None: """The newest snapshot of files received from ``peer``, or None.""" - raise NotImplementedError("Watcher storage is not implemented yet") - def put_watcher_rolling_state(self, peer: str, name: str, data: bytes) -> str: + @abstractmethod + def put_rolling_state(self, peer: str, name: str, data: bytes) -> str: """Store the changes received from ``peer`` since the snapshot, and the - ids of messages already processed.""" - raise NotImplementedError("Watcher storage is not implemented yet") + ids of the messages already processed.""" - def get_watcher_rolling_state(self, peer: str) -> bytes | None: + @abstractmethod + def get_rolling_state(self, peer: str) -> bytes | None: """The rolling state for ``peer``, or None.""" - raise NotImplementedError("Watcher storage is not implemented yet") - def delete_watcher_state(self, peer: str) -> None: - """Forget everything received from ``peer``, e.g. before a full resend.""" - raise NotImplementedError("Watcher storage is not implemented yet") + @abstractmethod + def reset(self, peer: str) -> None: + """Forget everything received from ``peer``, before asking it to send + everything again.""" class Collections(ABC): @@ -197,14 +197,15 @@ def upload( ) -> None: """Store ``files`` in the collection.""" - @abstractmethod - def list_mine(self, prefix: str) -> list[str]: - """Names of the collections I published under ``prefix``.""" - @abstractmethod def list_mine_with_permissions(self, prefix: str) -> list[FileCollection]: """The collections I published under ``prefix``, with their flags.""" + def list_mine(self, prefix: str) -> list[str]: + """The tags of the collections I published under ``prefix``, each once.""" + tags = (c.tag for c in self.list_mine_with_permissions(prefix)) + return list(dict.fromkeys(tags)) + @abstractmethod def delete(self, prefix: str, tag: str) -> None: """Remove every layout of ``tag`` published under ``prefix``.""" @@ -214,12 +215,6 @@ def list_shared_with_me(self, prefix: str) -> list[dict]: """Collections peers shared with me, each with owner_email, tag, content_hash and variant.""" - @abstractmethod - def download( - self, prefix: str, tag: str, content_hash: str, owner_email: str - ) -> dict[str, bytes]: - """Every file of one collection a peer shared with me.""" - @abstractmethod def list_files( self, prefix: str, tag: str, content_hash: str, owner_email: str @@ -230,13 +225,17 @@ def list_files( def fetch_file(self, handle: str) -> bytes: """The bytes of one collection file.""" + def download( + self, prefix: str, tag: str, content_hash: str, owner_email: str + ) -> dict[str, bytes]: + """Every file of one collection a peer shared with me, by file name.""" + files = self.list_files(prefix, tag, content_hash, owner_email) + return {f["file_name"]: self.fetch_file(f["file_id"]) for f in files} -class PeerState(ABC): - """Finding and trusting a peer: identity, channels, published documents. - Not to be confused with ``syft.sync.peers.peer.PeerState``, the enum of - handshake states. - """ +class Peering(ABC): + """Finding and trusting a peer: identity, channels, peer list, published + documents.""" @abstractmethod def whoami(self) -> str: @@ -244,12 +243,24 @@ def whoami(self) -> str: @abstractmethod def open_channels(self, peer: str) -> None: - """Create both mailbox channels with ``peer`` and let them write.""" + """Create the message channels with ``peer`` and let it write to them.""" @abstractmethod def incoming_requests(self) -> list[str]: """Emails of peers who opened channels towards me.""" + # -- peer list (SYFT_peers.json) ------------------------------------------- + @abstractmethod + def read_peer_registry(self, force_download: bool = False) -> dict: + """My record of every peer and its state, keyed by email.""" + + @abstractmethod + def update_peer_registry( + self, peer: str, state: str, public_encryption_bundle: dict | None = None + ) -> None: + """Record ``peer``'s state, and its key bundle when given.""" + + # -- key bundles ----------------------------------------------------------- @abstractmethod def publish_key_bundle(self, peer: str, bundle_json: str) -> None: """Publish my public key bundle for ``peer``.""" @@ -258,10 +269,15 @@ def publish_key_bundle(self, peer: str, bundle_json: str) -> None: def grant_key_bundle_read(self, peer: str) -> None: """Let ``peer`` read the key bundles I publish.""" + @abstractmethod + def read_own_key_bundle(self, peer: str) -> str | None: + """The key bundle I published for ``peer``, or None.""" + @abstractmethod def read_peer_key_bundle(self, peer: str) -> str | None: """The key bundle ``peer`` published for me, or None.""" + # -- version documents ----------------------------------------------------- @abstractmethod def publish_version(self, version_info: VersionInfo) -> None: """Publish my version document.""" @@ -280,13 +296,15 @@ def grant_version_read(self, peer: str) -> None: class Transport(BaseModel): - """One backend's implementation of the four jobs.""" + """One implementation of each job. The parts may come from different + backends, though a peer's ``p2p`` and ``peering`` go together.""" model_config = ConfigDict(frozen=True, arbitrary_types_allowed=True) p2p: P2PConnection - private_store: PrivateStore + owner_store: OwnerStore + watcher_store: WatcherStore collections: Collections - peer_state: PeerState + peering: Peering # Recorded on each Peer so a peer list says which backend reaches it. platform: BasePlatform diff --git a/syft/sync/transport/storage.py b/syft/sync/transport/storage.py deleted file mode 100644 index 9d6e01e9787..00000000000 --- a/syft/sync/transport/storage.py +++ /dev/null @@ -1,70 +0,0 @@ -"""Readable views over ``PrivateStore``: one for each role. - -A backend implements ``PrivateStore`` once. Owner code holds an -``OwnerStorage`` and data scientist code holds a ``WatcherStorage``, so each -sees only its own methods and cannot write into the other's area. -""" - -from syft.sync.transport.protocols import PrivateStore - - -class OwnerStorage: - """The data owner's own storage: event log, checkpoints, rolling state.""" - - def __init__(self, store: PrivateStore): - self._store = store - - def append_event(self, name: str, data: bytes) -> str: - return self._store.append_event(name, data) - - def fetch(self, handle: str) -> bytes: - return self._store.fetch(handle) - - def fetch_events_since(self, since_timestamp: float) -> list[bytes]: - return self._store.fetch_events_since(since_timestamp) - - def put_checkpoint(self, name: str, data: bytes) -> str: - return self._store.put_checkpoint(name, data) - - def latest_checkpoint(self) -> bytes | None: - return self._store.latest_checkpoint() - - def put_incremental(self, name: str, data: bytes) -> str: - return self._store.put_incremental(name, data) - - def all_incrementals(self) -> list[bytes]: - return self._store.all_incrementals() - - def put_rolling_state(self, name: str, data: bytes) -> str: - return self._store.put_rolling_state(name, data) - - def get_rolling_state(self) -> bytes | None: - return self._store.get_rolling_state() - - -class WatcherStorage: - """The data scientist's memory of what one owner sent it. - - Stored in the scientist's own private storage, never shared, so it - survives a lost local disk such as a Colab runtime reset. - """ - - def __init__(self, store: PrivateStore, peer: str): - self._store = store - self.peer = peer - - def put_checkpoint(self, name: str, data: bytes) -> str: - return self._store.put_watcher_checkpoint(self.peer, name, data) - - def latest_checkpoint(self) -> bytes | None: - return self._store.latest_watcher_checkpoint(self.peer) - - def put_rolling_state(self, name: str, data: bytes) -> str: - return self._store.put_watcher_rolling_state(self.peer, name, data) - - def get_rolling_state(self) -> bytes | None: - return self._store.get_watcher_rolling_state(self.peer) - - def reset(self) -> None: - """Forget everything from this owner, before asking for a full resend.""" - self._store.delete_watcher_state(self.peer) diff --git a/tests/unit/test_transport_contracts.py b/tests/unit/test_transport_contracts.py index c8e1b0c30cc..4d377951abd 100644 --- a/tests/unit/test_transport_contracts.py +++ b/tests/unit/test_transport_contracts.py @@ -1,6 +1,8 @@ -"""The transport contracts: the abstract jobs, the role wrappers, the naming -rules and the DeliveryTracker stub. No backend implements them yet.""" +"""The transport contracts: the abstract jobs, their helper methods, the +message types, the naming rules and the DeliveryTracker stub. No backend +implements them yet.""" +from types import SimpleNamespace from unittest.mock import Mock import pytest @@ -11,75 +13,73 @@ from syft.sync.transport import ( Collections, MessageKind, + OwnerStore, P2PConnection, + Peering, RawMessage, - OwnerStorage, - PeerState, - PrivateStore, Transport, - WatcherStorage, + WatcherStore, ) from syft.sync.transport.naming import timestamp_from_message_name +JOBS = [P2PConnection, OwnerStore, WatcherStore, Collections, Peering] -@pytest.mark.parametrize("job", [P2PConnection, PrivateStore, Collections, PeerState]) + +@pytest.mark.parametrize("job", JOBS) def test_a_job_cannot_be_used_without_a_backend(job): with pytest.raises(TypeError): job() -def test_owner_storage_forwards_to_the_private_store(): - store = Mock(spec=PrivateStore) - store.get_rolling_state.return_value = b"state" - owner = OwnerStorage(store) - owner.put_rolling_state("rolling_state_1.0.tar.gz", b"state") - store.put_rolling_state.assert_called_once_with( - "rolling_state_1.0.tar.gz", b"state" - ) - assert owner.get_rolling_state() == b"state" +def test_owner_store_helpers_are_built_from_list_and_fetch(): + store = Mock(spec=OwnerStore) + store.list_event_handles.return_value = ["h1", "h2"] + store.fetch.side_effect = lambda handle: f"bytes of {handle}".encode() + assert OwnerStore.fetch_events_since(store, 5.0) == [b"bytes of h1", b"bytes of h2"] + store.list_event_handles.assert_called_with(5.0) + assert OwnerStore.count_events_since(store, 5.0) == 2 -def test_watcher_storage_names_its_peer_on_every_call(): - store = Mock(spec=PrivateStore) - watcher = WatcherStorage(store, "do@example.com") - watcher.put_checkpoint("checkpoint_1.0.tar.gz", b"snapshot") - watcher.reset() - store.put_watcher_checkpoint.assert_called_once_with( - "do@example.com", "checkpoint_1.0.tar.gz", b"snapshot" - ) - store.delete_watcher_state.assert_called_once_with("do@example.com") +def test_fetching_all_events_lists_from_the_beginning(): + store = Mock(spec=OwnerStore) + store.fetch_events_since.return_value = [b"all"] + assert OwnerStore.fetch_all_events(store) == [b"all"] + store.fetch_events_since.assert_called_once_with(None) -def test_watcher_area_is_not_implemented_by_default(): - with pytest.raises(NotImplementedError): - PrivateStore.latest_watcher_checkpoint(object(), "do@example.com") +def test_list_mine_returns_each_tag_once_in_order(): + collections = Mock(spec=Collections) + collections.list_mine_with_permissions.return_value = [ + SimpleNamespace(tag="census"), + SimpleNamespace(tag="census"), # the same dataset in a second layout + SimpleNamespace(tag="sales"), + ] + assert Collections.list_mine(collections, "syft_datasetcollection") == [ + "census", + "sales", + ] -@pytest.mark.parametrize( - "name, expected", - [ - ("syfteventsmessagev3_1759800000.95_abc.tar.gz", 1759800000.95), - ("msgv2_12.5_abc.tar.gz", 12.5), - ("checkpoint_12.5.tar.gz", None), - ("msgv2_notanumber_abc.tar.gz", None), - ], -) -def test_message_names_carry_their_send_time(name, expected): - assert timestamp_from_message_name(name) == expected +def test_download_fetches_every_listed_file_by_name(): + collections = Mock(spec=Collections) + collections.list_files.return_value = [ + {"file_id": "id-1", "file_name": "mock.csv"}, + {"file_id": "id-2", "file_name": "readme.md"}, + ] + collections.fetch_file.side_effect = lambda handle: handle.encode() -def test_delivery_tracker_is_a_stub_until_implemented(): - tracker = DeliveryTracker(path=None) - with pytest.raises(NotImplementedError): - tracker.has_view("ds@example.com") + files = Collections.download(collections, "prefix", "census", "abc", "do@x.org") + assert files == {"mock.csv": b"id-1", "readme.md": b"id-2"} -def test_a_transport_bundles_the_four_jobs_and_cannot_be_changed(): +def test_a_transport_bundles_one_of_each_job_and_cannot_be_changed(): transport = Transport( p2p=Mock(spec=P2PConnection), - private_store=Mock(spec=PrivateStore), + owner_store=Mock(spec=OwnerStore), + watcher_store=Mock(spec=WatcherStore), collections=Mock(spec=Collections), - peer_state=Mock(spec=PeerState), + peering=Mock(spec=Peering), platform=GdriveFilesPlatform(), ) with pytest.raises(ValidationError): @@ -102,3 +102,22 @@ def test_the_old_raw_message_import_path_still_works(): from syft.sync.messages.raw_message import RawMessage as OldPathRawMessage assert OldPathRawMessage is RawMessage + + +@pytest.mark.parametrize( + "name, expected", + [ + ("syfteventsmessagev3_1759800000.95_abc.tar.gz", 1759800000.95), + ("msgv2_12.5_abc.tar.gz", 12.5), + ("checkpoint_12.5.tar.gz", None), + ("msgv2_notanumber_abc.tar.gz", None), + ], +) +def test_message_names_carry_their_send_time(name, expected): + assert timestamp_from_message_name(name) == expected + + +def test_delivery_tracker_is_a_stub_until_implemented(): + tracker = DeliveryTracker(path=None) + with pytest.raises(NotImplementedError): + tracker.has_view("ds@example.com") From 594bf2a793e221da9dd1d62f69994d7ef2e034a1 Mon Sep 17 00:00:00 2001 From: rasswanth-s <43314053+rasswanth-s@users.noreply.github.com> Date: Fri, 16 Oct 2026 11:41:21 +0530 Subject: [PATCH 4/4] refactor(transport): add the connection contracts and a DeliveryTracker stub --- syft/sync/sync/delivery_tracker.py | 196 ++++++++++++++++--------- tests/unit/test_transport_contracts.py | 21 ++- 2 files changed, 142 insertions(+), 75 deletions(-) diff --git a/syft/sync/sync/delivery_tracker.py b/syft/sync/sync/delivery_tracker.py index fede3bb1d68..963bdd8479a 100644 --- a/syft/sync/sync/delivery_tracker.py +++ b/syft/sync/sync/delivery_tracker.py @@ -1,103 +1,153 @@ -"""What each peer was sent, so the data owner can resend, delete and catch up. - -STUB: the names and contracts are agreed; the bodies are to be written (PR 2). -A draft implementation exists from the planning work. - -For every peer the tracker keeps a view: datasite path -> hash of the version -that peer holds. Each sync compares what a peer may read with its view: - - the peer holds an old version -> send the new one - the peer holds nothing (new peer) -> send every file it may read - the peer may no longer read a file -> send a delete - the owner deleted the file -> send a delete - -The view changes only after the message carrying a version was written to the -peer's outbox, so a crash before the write leaves it unchanged and the -delivery is retried. Paths in flight are written down before routing and -cleared after the last outbox write, so a crash in between is finished by the -next sync. - -Open decisions (see research/merged_transport_plan.md): -- First run with no views: assume current peers hold what they may read - (proposed), or resend everything. -- A rejected peer approved again: reset its view and resend (proposed). -- Enclaves lose the views at every boot until they are also kept on Drive. +"""What each peer holds, so the data owner can resend, delete and catch up. + +STUB: the names and contracts are agreed; the bodies are to be written. + +The tracker knows nothing about the transport or where its record is kept. It +works on paths and content hashes, takes the outcome of each send as a +``SendStatus``, and keeps its record through a ``DeliveryStateStore``. + +Views +----- +For every peer the tracker keeps a view, ``{path: hash}``: the version of each +file that peer is confirmed to hold. ``plan`` compares a view with what the +peer may read now: + + readable, and the hash differs from the view -> send + in the view, but no longer readable -> delete + in the view, but the file is gone -> delete + +A peer with no view, because it is new or its record was lost, gets a +``PeerReset``: its view starts empty, so it is sent every file it may read. + +Recording a send +---------------- +``sent`` takes the ``SendStatus`` from ``P2PConnection.send``: + +- ``DELIVERED``: the view changes at once. +- ``QUEUED``: the changes are held under their ``msg_id``, and the view changes + only when ``P2PConnection.delivered()`` reports that ``msg_id`` and + ``confirm`` is called. A receipt is not enough; only the peer's ack counts. + +A view therefore never changes before a delivery is confirmed. + +Pending deliveries +------------------ +Changes queued for a peer but not yet sent, for example because the upload +failed or the peer was skipped by the version gate, are kept per peer by the +owner cache's pending deliveries (PR #9563). The tracker does not duplicate +them. + +Storage +------- +``DeliveryState`` is the whole record, versioned. A ``DeliveryStateStore`` +loads and saves it: a local file next to the owner's file hashes, and with +``persist_owner_state`` also Drive, inside the rolling state and checkpoints, +so an enclave that wipes its local state at every boot does not reset every +peer. + +Sends not routed by permission, such as the echo of a submitter's own change +or a private dataset shared with an enclave, are recorded only where the +recipient may read the file, so a later permission change never deletes them. + +Open decisions: +- A peer rejected, then approved again: reset its view and resend (proposed). +- Accepted job submissions: echo only to the submitter, as today (proposed), or + also send to every peer who may read the file. """ -from pathlib import Path -from typing import Iterable +from abc import ABC, abstractmethod +from typing import Iterable, Mapping -from syft.sync.events.file_change_event import FileChangeEvent +from pydantic import BaseModel, ConfigDict, Field + +from syft.sync.transport.types import SendStatus DELIVERY_VIEWS_FILENAME = "owner_delivery_views.json" -# Format of the views file. Raise it when the layout changes, and keep a read +# Format of DeliveryState. Raise it when the layout changes, and keep a read # path for every earlier version. -DELIVERY_VIEWS_VERSION = 1 +DELIVERY_STATE_VERSION = 1 -class DeliveryTracker: - """Per-peer views of delivered file versions, persisted next to the owner cache. +class QueuedSend(BaseModel): + """Changes sent to ``peer`` that wait for confirmation. - ``path`` None keeps the views in memory only, as the in-memory owner cache - does in tests. + ``changes`` maps each path to its new hash, or None for a delete. """ - def __init__(self, path: Path | None = None): - self.path = path + peer: str + changes: dict[str, str | None] - # -- reading --------------------------------------------------------------- - def reload(self) -> None: - """Read the views from disk once; lookups use them until the next reload.""" - raise NotImplementedError - @property - def initialized(self) -> bool: - """Whether views were ever recorded for this datasite.""" - raise NotImplementedError +class DeliveryState(BaseModel): + """Everything the tracker keeps: views, and sends awaiting confirmation.""" - def has_view(self, peer: str) -> bool: - """Whether ``peer`` was ever delivered to; False means a new peer.""" - raise NotImplementedError + version: int = DELIVERY_STATE_VERSION + # peer -> {path: hash} + views: dict[str, dict[str, str]] = Field(default_factory=dict) + # msg_id -> the send it confirms + queued: dict[str, QueuedSend] = Field(default_factory=dict) - def delivered_hash(self, peer: str, path: str) -> str | None: - """The hash of the version of ``path`` that ``peer`` holds, or None.""" - raise NotImplementedError - def holders(self, path: str, peers: Iterable[str]) -> frozenset[str]: - """The peers among ``peers`` that hold some version of ``path``.""" - raise NotImplementedError +class DeliveryPlan(BaseModel): + """What ``plan`` decided for one peer.""" - # -- writing --------------------------------------------------------------- - def mark_initialized(self) -> None: - """Record that views exist, with DELIVERY_VIEWS_VERSION.""" - raise NotImplementedError + model_config = ConfigDict(frozen=True) - def set_view(self, peer: str, view: dict[str, str]) -> None: - """Replace ``peer``'s whole view, e.g. to seed it on the first run.""" - raise NotImplementedError + send: list[str] = Field(default_factory=list) + delete: list[str] = Field(default_factory=list) - def record(self, peer: str, events: Iterable[FileChangeEvent]) -> None: - """Record that ``events`` reached ``peer``'s outbox. - A delete removes the path from the view; any other event sets its hash. - Call only after the outbox write succeeded. - """ +class DeliveryStateStore(ABC): + """Where a ``DeliveryState`` is kept: a local file, Drive, or memory.""" + + @abstractmethod + def load(self) -> DeliveryState: + """The stored state, or an empty one.""" + + @abstractmethod + def save(self, state: DeliveryState) -> None: + """Store ``state``, replacing what was there.""" + + +class DeliveryTracker: + """Per-peer views of confirmed file versions, for the data owner.""" + + def __init__(self, store: DeliveryStateStore): + self.store = store + + def has_view(self, peer: str) -> bool: + """Whether ``peer`` has a view. False means a ``PeerReset`` is due.""" raise NotImplementedError def reset(self, peer: str) -> None: - """Forget ``peer``'s view, so the next sync sends it everything again.""" + """Start ``peer``'s view empty (a ``PeerReset``), so the next plan sends + it every file it may read.""" raise NotImplementedError - # -- paths in flight ------------------------------------------------------- - @property - def pending(self) -> list[str]: - """Paths whose routing started but did not finish.""" + def plan( + self, peer: str, readable: Mapping[str, str], paths: Iterable[str] + ) -> DeliveryPlan: + """What to send and delete for ``peer`` among ``paths``. + + ``readable`` maps every path ``peer`` may read now to its current hash. + ``paths`` are the dirty paths; for a peer with no view, every readable + path. + """ raise NotImplementedError - def mark_pending(self, paths: Iterable[str]) -> None: - """Write ``paths`` down before routing them.""" + def sent( + self, + peer: str, + msg_id: str, + status: SendStatus, + changes: Mapping[str, str | None], + ) -> None: + """Record a send of ``changes`` (path -> new hash, None for a delete). + + ``DELIVERED`` updates the view now; ``QUEUED`` waits for ``confirm``. + """ raise NotImplementedError - def clear_pending(self) -> None: - """Clear the paths in flight once every outbox write is done.""" + def confirm(self, msg_ids: Iterable[str]) -> None: + """Apply the queued sends that ``P2PConnection.delivered()`` reported.""" raise NotImplementedError diff --git a/tests/unit/test_transport_contracts.py b/tests/unit/test_transport_contracts.py index 4d377951abd..dbb84239d10 100644 --- a/tests/unit/test_transport_contracts.py +++ b/tests/unit/test_transport_contracts.py @@ -9,7 +9,12 @@ from pydantic import ValidationError from syft.sync.platforms.gdrive_files_platform import GdriveFilesPlatform -from syft.sync.sync.delivery_tracker import DeliveryTracker +from syft.sync.sync.delivery_tracker import ( + DELIVERY_STATE_VERSION, + DeliveryState, + DeliveryStateStore, + DeliveryTracker, +) from syft.sync.transport import ( Collections, MessageKind, @@ -118,6 +123,18 @@ def test_message_names_carry_their_send_time(name, expected): def test_delivery_tracker_is_a_stub_until_implemented(): - tracker = DeliveryTracker(path=None) + tracker = DeliveryTracker(store=Mock(spec=DeliveryStateStore)) with pytest.raises(NotImplementedError): tracker.has_view("ds@example.com") + + +def test_a_delivery_state_store_needs_a_backend(): + with pytest.raises(TypeError): + DeliveryStateStore() + + +def test_an_empty_delivery_state_is_versioned(): + state = DeliveryState() + assert state.version == DELIVERY_STATE_VERSION + assert state.views == {} + assert state.queued == {}