diff --git a/docs/connections.md b/docs/connections.md index ba37a079eaa..b1cc561aa70 100644 --- a/docs/connections.md +++ b/docs/connections.md @@ -61,7 +61,7 @@ A pair encrypts when one side is `required`. A SyftBox has one policy, kept in `//private/syft_settings.json`. The first client to start on a new SyftBox writes it: the `encryption` it logged in with, or `optional`. A later client, the notebook or syft-bg, logs in without `encryption` and gets the policy on disk. A client that asks for another policy raises `EncryptionPolicyMismatchError`, and so does a running client at its next sync when the file changed under it. Files at rest are read with the current policy, so to change the policy, delete the SyftBox (`client.delete_syftbox()`) and log in again. `syft-bg init --encryption ...` sets the policy the same way as a login. -Nothing is sent to a peer until its policy is known, that is, until it accepts and shares its version file. A scientist's change for a peer whose policy or bundle is not known is dropped, and the send raises. An owner's send to that peer raises, and the change stays pending, so a later sync sends it. An owner keeps a message it cannot read yet in the inbox, with a warning. +Nothing is sent to a peer until its policy is known, that is, until it accepts and shares its version file. A scientist's change for a peer whose policy or bundle is not known is dropped, and the send raises. An owner's change for that peer waits, with a warning, and a later sync sends it. Other owner sends to that peer are skipped with a warning. An owner's change to a file that the peer may read is not lost: the owner keeps a record of what each peer holds, and that record changes only after a send, so a later sync sends it. Other owner sends, such as a private dataset share or an enclave result, are not kept. A failed send is tried up to three times in the same sync, then logged as an error. An owner keeps a message it cannot read yet in the inbox, with a warning. The format of a message must match the pair: plaintext in a pair that encrypts is refused, and so is an envelope in a pair that does not. The owner archives such a message unread, and a scientist skips it, each with a warning. diff --git a/packages/syft-enclave/src/syft_enclaves/client.py b/packages/syft-enclave/src/syft_enclaves/client.py index 8d7613a7fa0..99a1772f900 100644 --- a/packages/syft-enclave/src/syft_enclaves/client.py +++ b/packages/syft-enclave/src/syft_enclaves/client.py @@ -595,11 +595,10 @@ def _push_files_to_recipients( events_message = syncer.event_cache.create_events_for_files( files_by_datasite_path ) - syncer.queue_event_for_syftbox( + syncer.process_events_for_own_state_and_peers( recipients=recipients, file_change_events_message=events_message, ) - syncer.process_syftbox_events_queue() def _push_approval_file(self, approval_file: Path) -> None: """Push a changed approval file to the enclave.""" @@ -704,11 +703,10 @@ def _forward_job_to_dos(self, job_dir: Path, do_emails: list[str]): events_message = syncer.event_cache.create_events_for_files( files_by_datasite_path ) - syncer.queue_event_for_syftbox( + syncer.process_events_for_own_state_and_peers( recipients=do_emails, file_change_events_message=events_message, ) - syncer.process_syftbox_events_queue() def _forward_approval_files_to_dos(self, review_dir: Path, do_emails: list[str]): """Forward each DO's approval state file to them individually.""" @@ -722,11 +720,10 @@ def _forward_approval_files_to_dos(self, review_dir: Path, do_emails: list[str]) events_message = syncer.event_cache.create_events_for_files( files_by_datasite_path ) - syncer.queue_event_for_syftbox( + syncer.process_events_for_own_state_and_peers( recipients=[do_email], file_change_events_message=events_message, ) - syncer.process_syftbox_events_queue() def _get_files_in_dir(self, directory: Path) -> dict[Path, bytes]: """Read all files under directory, keyed by path relative to the datasite root.""" diff --git a/packages/syft-perms/src/syft_perms/syftperm_context.py b/packages/syft-perms/src/syft_perms/syftperm_context.py index 571ef691bbc..9b86a2b5595 100644 --- a/packages/syft-perms/src/syft_perms/syftperm_context.py +++ b/packages/syft-perms/src/syft_perms/syftperm_context.py @@ -1,6 +1,6 @@ from pathlib import Path -from syft_permissions import ACLService +from syft_permissions import ACLRequest, ACLService, AccessLevel, User from syft_perms.browser import FilesBrowser from syft_perms.file import SyftFile @@ -31,6 +31,15 @@ def open(self, path: str | Path) -> SyftFile | SyftFolder: return SyftFolder(abs_path, self) return SyftFile(abs_path, self) + def can_read(self, path: str, user: str) -> bool: + """Whether ``user`` may read the file at ``path``, relative to the datasite. + + Unlike ``open(path).has_read_access(user)``, this does not touch the + filesystem. + """ + request = ACLRequest(path=path, level=AccessLevel.READ, user=User(id=user)) + return self.service.can_access(request) + @property def files(self) -> FilesBrowser: return FilesBrowser(self, kind="files") diff --git a/packages/syft-perms/tests/test_syft_perm.py b/packages/syft-perms/tests/test_syft_perm.py index 0af81818e13..875ee80819b 100644 --- a/packages/syft-perms/tests/test_syft_perm.py +++ b/packages/syft-perms/tests/test_syft_perm.py @@ -161,6 +161,14 @@ def test_no_permission_denies(sp): assert not file.has_admin_access(USER) +def test_can_read_follows_grants(sp): + sp.open("shared/").grant_read_access(USER) + assert sp.can_read("shared/data.csv", USER) + assert not sp.can_read("shared/data.csv", USER2) + assert sp.can_read("other.csv", OWNER) + assert not sp.can_read("other.csv", USER) + + def test_wildcard_grant(sp, datasite): yaml_data = { "rules": [ diff --git a/packages/syft-rds/src/syft_rds/client.py b/packages/syft-rds/src/syft_rds/client.py index f994babcc4f..0066807958d 100644 --- a/packages/syft-rds/src/syft_rds/client.py +++ b/packages/syft-rds/src/syft_rds/client.py @@ -1249,11 +1249,10 @@ def share_private_dataset(self, tag: str, enclave_email: str): events_message = self.sync_engine.datasite_owner_syncer.event_cache.create_events_for_files( files ) - self.sync_engine.datasite_owner_syncer.queue_event_for_syftbox( + self.sync_engine.datasite_owner_syncer.process_events_for_own_state_and_peers( recipients=[enclave_email], file_change_events_message=events_message, ) - self.sync_engine.datasite_owner_syncer.process_syftbox_events_queue() def _private_share_protocol_version(self, tag: str, peer_email: str) -> str: """The protocol version of the copy to ship privately to this peer. diff --git a/syft/sync/connections/connection_router.py b/syft/sync/connections/connection_router.py index b3fee7d2b04..2ccf05718cd 100644 --- a/syft/sync/connections/connection_router.py +++ b/syft/sync/connections/connection_router.py @@ -29,7 +29,7 @@ from syft.sync.peers.peer_store import PeerStore from syft.sync.transport.naming import PEERS_META_KEY from syft.sync.transport.protocols import Transport -from syft.sync.transport.types import MessageKind +from syft.sync.transport.types import MessageKind, SendStatus from syft.sync.utils.print_utils import ( print_peer_added_to_platform, print_peer_adding_to_platform, @@ -189,12 +189,14 @@ def owner_get_next_proposed_filechange_message( def owner_write_event_messages_to_outbox( self, recipient_email: str, events_message: FileChangeEventsMessage - ): + ) -> SendStatus: events_message = self._downgrade_for_peer(events_message, recipient_email) data = events_message.as_compressed_data() data = self.peer_store.encrypt_if_needed(recipient_email, data) fname = events_message.message_filepath.as_string() - self.transport.p2p.send(recipient_email, fname, data, MessageKind.FILE_EVENTS) + return self.transport.p2p.send( + recipient_email, fname, data, MessageKind.FILE_EVENTS + ) def watcher_get_events_messages( self, peer_email: str, since_timestamp: float | None @@ -249,15 +251,6 @@ def owner_write_events_message_to_syftbox( filename = events_message.message_filepath.as_string() self.transport.owner_store.append_event(filename, data) - def owner_get_all_accepted_events_messages(self) -> List[FileChangeEventsMessage]: - raw_list = self.transport.owner_store.fetch_all_events() - return [ - FileChangeEventsMessage.from_compressed_data( - self.peer_store.decrypt_and_verify_for_self_if_needed(raw) - ) - for raw in raw_list - ] - def download_events_message_by_id( self, events_message_id: str ) -> FileChangeEventsMessage: diff --git a/syft/sync/events/file_change_event.py b/syft/sync/events/file_change_event.py index 0ce4151030e..13ee2644621 100644 --- a/syft/sync/events/file_change_event.py +++ b/syft/sync/events/file_change_event.py @@ -127,6 +127,7 @@ def from_proposed_filechange( submitted_timestamp=proposed_filechange.submitted_timestamp, timestamp=create_event_timestamp(), datasite_email=proposed_filechange.datasite_email, + is_deleted=proposed_filechange.is_deleted, ) def __hash__(self) -> int: diff --git a/syft/sync/syftbox_manager.py b/syft/sync/syftbox_manager.py index acb76484692..8235d03369e 100644 --- a/syft/sync/syftbox_manager.py +++ b/syft/sync/syftbox_manager.py @@ -967,12 +967,7 @@ def sync( ignore_peer_version=ignore_peer_version, ) ) - self.datasite_owner_syncer.sync( - compatible_emails, - deferred_peer_emails=[ - e for e in peer_emails if e not in compatible_emails - ], - ) + self.datasite_owner_syncer.sync(compatible_emails) if auto_compact: for peer_email in compatible_emails: self.datasite_owner_syncer.compact_outbox_if_needed( @@ -1023,8 +1018,17 @@ def _emit_peers_loaded(self) -> None: self._emit("peers_loaded") def reject_peer_request(self, email_or_peer: str | Peer): - """Reject a pending peer request. Delegates to PeerManager.""" + """Reject a pending peer request. Delegates to PeerManager. + + The DO resets the peer's delivery view, so a later approval sends the + peer every file it may read, and deletes the files it may no longer have. + """ self.peer_manager.reject_peer_request(email_or_peer) + if self.has_do_role: + peer_email = ( + email_or_peer if isinstance(email_or_peer, str) else email_or_peer.email + ) + self.datasite_owner_syncer.event_cache.delivery_tracker.reset(peer_email) def validate_peer(self, peer_email: str) -> Peer: """Load the peers, then return ``peer_email`` if it is approved. @@ -1098,8 +1102,10 @@ def _add_connection(self, connection: SyftboxPlatformConnection): def _send_file_change(self, path: str | Path, content: str): self.file_writer.write_file(path, content) - def _get_all_accepted_events_do(self) -> List[FileChangeEvent]: - return self.datasite_owner_syncer.connection_router.owner_get_all_accepted_events_messages() + def _get_all_accepted_events_do(self) -> List[FileChangeEventsMessage]: + """The events of the owner's state on the backend since the last full + checkpoint, as a restore reads them.""" + return self.datasite_owner_syncer.fetch_owner_state().events_messages() @property def _connection_router(self) -> ConnectionRouter: diff --git a/syft/sync/sync/caches/datasite_owner_cache.py b/syft/sync/sync/caches/datasite_owner_cache.py index c968abec178..cc06dc00345 100644 --- a/syft/sync/sync/caches/datasite_owner_cache.py +++ b/syft/sync/sync/caches/datasite_owner_cache.py @@ -21,7 +21,12 @@ from syft.sync.sync.constants import ( CACHE_DIR, OWNER_FILE_HASHES_FILENAME, - OWNER_PENDING_DELIVERIES_FILENAME, +) +from syft.sync.sync.delivery_tracker import ( + DELIVERY_VIEWS_FILENAME, + DeliveryTracker, + LocalDeliveryStateStore, + MemoryDeliveryStateStore, ) @@ -62,9 +67,10 @@ class DataSiteOwnerEventCache(BaseModelCallbackMixin): # Wired in pre-init validator: persisted-to-disk when syftbox_folder is set, # plain in-memory PersistedDict otherwise. file_hashes: PersistedDict = Field(default_factory=PersistedDict) - # peer email -> {path in datasite: name of the last message that holds it}. - # A path stays here until the upload of that message to the peer succeeds. - pending_deliveries: PersistedDict = Field(default_factory=PersistedDict) + # What each peer holds. Kept next to file_hashes when syftbox_folder is set. + delivery_tracker: DeliveryTracker = Field( + default_factory=lambda: DeliveryTracker(MemoryDeliveryStateStore()) + ) email: str # Full path to collections (datasets) folder collections_folder: Path | None = None @@ -85,11 +91,13 @@ def _build_file_hashes(cls, data): key_serializer=str, key_deserializer=Path, ) - if isinstance(data, dict) and "pending_deliveries" not in data: + if isinstance(data, dict) and "delivery_tracker" not in data: folder = data.get("syftbox_folder") if folder is not None: - data["pending_deliveries"] = PersistedDict( - path=Path(folder) / CACHE_DIR / OWNER_PENDING_DELIVERIES_FILENAME + data["delivery_tracker"] = DeliveryTracker( + LocalDeliveryStateStore( + Path(folder) / CACHE_DIR / DELIVERY_VIEWS_FILENAME + ) ) return data @@ -102,7 +110,7 @@ def from_config(cls, config: DataSiteOwnerEventCacheConfig): email=config.email, syftbox_folder=config.syftbox_folder, file_hashes=PersistedDict(), - pending_deliveries=PersistedDict(), + delivery_tracker=DeliveryTracker(MemoryDeliveryStateStore()), collections_folder=config.collections_folder, collection_variants=config.collection_variants, ) @@ -310,6 +318,7 @@ def clear_cache(self): self.events_messages_connection.clear_cache() self.file_connection.clear_cache() self.file_hashes.clear() + self.delivery_tracker.clear() def has_conflict(self, proposed_event: ProposedFileChange) -> bool: if proposed_event.path_in_datasite not in self.file_hashes: @@ -440,9 +449,10 @@ def apply_checkpoint(self, checkpoint: Checkpoint, write_files: bool = True): self.file_hashes.clear() # Restore from checkpoint - # Always write to file_connection to maintain consistent state for - # process_local_changes comparison. This is needed even when write_files=False - # because file_connection may be in-memory and serves as the source of truth. + # Always write to file_connection to maintain consistent state for the + # comparison in process_local_file_changes. This is needed even when + # write_files=False, because file_connection may be in-memory and serves + # as the source of truth. for file_entry in checkpoint.files: path = Path(file_entry.path) self.file_hashes[path] = file_entry.hash diff --git a/syft/sync/sync/caches/persisted_dict.py b/syft/sync/sync/caches/persisted_dict.py index af22ddf5735..a811bef0cdd 100644 --- a/syft/sync/sync/caches/persisted_dict.py +++ b/syft/sync/sync/caches/persisted_dict.py @@ -26,7 +26,7 @@ import os from contextlib import contextmanager from pathlib import Path -from typing import Any, Callable +from typing import Any, Callable, Iterator from uuid import uuid4 import portalocker @@ -37,6 +37,36 @@ PERSISTED_DICT_VERSION = 1 +@contextmanager +def locked_file(path: Path, exclusive: bool = True) -> Iterator[None]: + """Hold a cross-process lock on ``path``, exclusive or shared. + + The lock is on a sibling ``.lock`` file, so ``path`` itself can be + replaced while the lock is held. + """ + path.parent.mkdir(parents=True, exist_ok=True) + flag = portalocker.LOCK_EX if exclusive else portalocker.LOCK_SH + with open(path.with_suffix(f"{path.suffix}.lock"), "a") as f: + portalocker.lock(f, flag) + try: + yield + finally: + portalocker.unlock(f) + + +def write_text_atomic(path: Path, text: str) -> None: + """Replace the content of ``path`` with ``text`` in one step.""" + path.parent.mkdir(parents=True, exist_ok=True) + # Per-process unique tmp path: even with the file lock, this guards + # against any path where two writers share a tmp filename. + tmp = path.with_suffix(f".tmp.{os.getpid()}.{uuid4().hex}") + try: + tmp.write_text(text) + tmp.replace(path) + finally: + tmp.unlink(missing_ok=True) + + class PersistedDict(dict): """Dict that persists to a JSON file. With path=None it's a plain in-memory dict.""" @@ -55,22 +85,13 @@ def __init__( # --- locking primitives --- - def _lock_path(self) -> Path: - return self._path.with_suffix(self._path.suffix + ".lock") - @contextmanager def _locked(self, exclusive: bool): if self._path is None: yield return - self._path.parent.mkdir(parents=True, exist_ok=True) - flag = portalocker.LOCK_EX if exclusive else portalocker.LOCK_SH - with open(self._lock_path(), "a") as f: - portalocker.lock(f, flag) - try: - yield - finally: - portalocker.unlock(f) + with locked_file(self._path, exclusive): + yield def exclusive_lock(self): """Acquire a cross-process exclusive lock on this dict's file. @@ -125,19 +146,11 @@ def _entries_of(data: Any) -> dict: def _write_to_file(self) -> None: if self._path is None: return - self._path.parent.mkdir(parents=True, exist_ok=True) - # Per-process unique tmp path: even with the file lock, this guards - # against any path where two writers share a tmp filename. - tmp = self._path.with_suffix(f".tmp.{os.getpid()}.{uuid4().hex}") serialized = { "version": PERSISTED_DICT_VERSION, "entries": {self._key_serializer(k): v for k, v in super().items()}, } - try: - tmp.write_text(json.dumps(serialized)) - tmp.replace(self._path) - finally: - tmp.unlink(missing_ok=True) + write_text_atomic(self._path, json.dumps(serialized)) # --- I/O with locking --- diff --git a/syft/sync/sync/constants.py b/syft/sync/sync/constants.py index 8d195bfad55..2e8508efb4f 100644 --- a/syft/sync/sync/constants.py +++ b/syft/sync/sync/constants.py @@ -1,5 +1,4 @@ CACHE_DIR = ".cache" OWNER_FILE_HASHES_FILENAME = "owner_file_hashes.json" -OWNER_PENDING_DELIVERIES_FILENAME = "owner_pending_deliveries.json" ROLLING_STATE_FILENAME = "rolling_state.json" diff --git a/syft/sync/sync/datasite_owner_syncer.py b/syft/sync/sync/datasite_owner_syncer.py index 1a80444fff1..8e01312751b 100644 --- a/syft/sync/sync/datasite_owner_syncer.py +++ b/syft/sync/sync/datasite_owner_syncer.py @@ -1,45 +1,50 @@ import logging +from collections import deque import warnings +from concurrent.futures import ThreadPoolExecutor +from functools import partial from pathlib import Path +from queue import Queue +from typing import List, Mapping, NamedTuple from uuid import uuid4 -from functools import partial -from pydantic import ConfigDict, Field, BaseModel, PrivateAttr -from concurrent.futures import ThreadPoolExecutor -from queue import Queue -from typing import List, NamedTuple, Tuple -from syft.sync.events.file_change_event import ( - FileChangeEventsMessage, - FileChangeEventsMessageFileName, - FileChangeEvent, -) -from syft.sync.connections.base_connection import ( - ConnectionConfig, - FileCollection, -) -from syft.sync.sync.caches.datasite_owner_cache import ( - DataSiteOwnerEventCacheConfig, -) -from syft.sync.connections.connection_router import ConnectionRouter -from syft.sync.peers.exceptions import SyftPeerNotReadyError -from syft.sync.sync.caches.datasite_owner_cache import DataSiteOwnerEventCache -from syft.sync.sync.collection_spec import CollectionSyncSpec +from pydantic import BaseModel, ConfigDict, Field, PrivateAttr +from syft_perms import SyftPermContext + from syft.sync.callback_mixin import BaseModelCallbackMixin -from syft.sync.messages.proposed_filechange import ProposedFileChangesMessage -from syft.sync.utils.path_filters import is_normal_syncable_path from syft.sync.checkpoints.checkpoint import ( + DEFAULT_COMPACTING_THRESHOLD, Checkpoint, CheckpointFile, IncrementalCheckpoint, compact_incremental_checkpoints, - DEFAULT_COMPACTING_THRESHOLD, ) from syft.sync.checkpoints.rolling_state import ( RollingState, raise_for_later_version, ) -from syft_perms import SyftPermContext +from syft.sync.connections.base_connection import ( + ConnectionConfig, + FileCollection, +) +from syft.sync.connections.connection_router import ConnectionRouter +from syft.sync.events.file_change_event import ( + FileChangeEvent, + FileChangeEventsMessage, + FileChangeEventsMessageFileName, +) +from syft.sync.messages.proposed_filechange import ProposedFileChangesMessage +from syft.sync.peers.exceptions import SyftPeerNotReadyError +from syft.sync.sync.caches.datasite_owner_cache import ( + DataSiteOwnerEventCache, + DataSiteOwnerEventCacheConfig, +) +from syft.sync.sync.collection_spec import CollectionSyncSpec from syft.sync.sync.constants import CACHE_DIR, ROLLING_STATE_FILENAME +from syft.sync.sync.delivery_tracker import ( + DeliveryPlan, +) +from syft.sync.utils.path_filters import is_normal_syncable_path from syft.sync.utils.syftbox_utils import ( create_event_timestamp, get_event_hash_from_content, @@ -51,25 +56,66 @@ # Default threshold for creating incremental checkpoint from rolling state DEFAULT_CHECKPOINT_EVENT_THRESHOLD = 50 -# Default: upload rolling state to GDrive after this many events +# save_owner_state() uploads the rolling state once this many events were added +# since the last upload. At 1 the rolling state on Drive holds every change after +# each save_owner_state(), so the owner's event log is not written. DEFAULT_ROLLING_STATE_UPLOAD_THRESHOLD = 1 +# A delivery message holds about this much file content at most. A larger +# catch-up goes out in more than one message. +MAX_DELIVERY_MESSAGE_CONTENT = 50 * 1024 * 1024 + # Minimum number of message files in a peer's outbox before # compact_outbox_if_needed() will merge them into a single message. MIN_MESSAGES_COMPACT = 20 +# How often the owner tries an outbox message in one send pass before it gives +# up and logs an error. A file that a view tracks is planned again next sync. +MAX_SEND_ATTEMPTS = 3 + -class _CheckpointRestore(NamedTuple): - """What restoring from checkpoints recovered. +class OwnerStateOnBackend(NamedTuple): + """The owner's persisted state on the backend, in the order a restore + applies it. - ``found_checkpoint`` is not the same as ``events_since_timestamp is not - None``: a full checkpoint may carry no event cursor, and in that case we - must NOT fall back to downloading every event. + ``log_messages`` are the event-log messages newer than the checkpoints and + the rolling state. When there is no checkpoint and no rolling state, they + are the whole log, and ``log_is_whole`` is True. A full checkpoint that + carries no event cursor gets no log messages: it is the whole state. """ - events_since_timestamp: float | None - events: List[FileChangeEvent] - found_checkpoint: bool + full_checkpoint: Checkpoint | None + incremental_checkpoints: List[IncrementalCheckpoint] + # None when the backend has no rolling state, or an empty one. + rolling_state: RollingState | None + log_messages: List[FileChangeEventsMessage] + log_is_whole: bool + + def events_messages(self) -> List[FileChangeEventsMessage]: + """Every event since the full checkpoint, in order.""" + messages = [ + FileChangeEventsMessage(events=checkpoint.events) + for checkpoint in self.incremental_checkpoints + ] + if self.rolling_state is not None: + messages.append(FileChangeEventsMessage(events=self.rolling_state.events)) + return messages + self.log_messages + + +def _last_persisted_event_timestamp( + full_checkpoint: Checkpoint | None, + incrementals: List[IncrementalCheckpoint], + rolling_state: RollingState | None = None, +) -> float | None: + """The timestamp of the last event that the checkpoints and the rolling + state hold. The event log is read after it.""" + since = full_checkpoint.last_event_timestamp if full_checkpoint else None + for checkpoint in incrementals: + for event in checkpoint.events: + since = _later_timestamp(since, event.timestamp) + if rolling_state is not None: + since = _later_timestamp(since, rolling_state.last_event_timestamp) + return since def _later_timestamp(current: float | None, candidate: float | None) -> float | None: @@ -109,6 +155,32 @@ class DatasiteOwnerSyncerConfig(BaseModel): connection_configs: List[ConnectionConfig] = [] +class OutgoingMessage(NamedTuple): + recipient: str + message: FileChangeEventsMessage + # Failed sends of this message so far. + attempts: int = 0 + + +def _batches_by_content_size( + events: list[FileChangeEvent], limit: int +) -> list[list[FileChangeEvent]]: + """Split ``events`` in order into lists of at most ``limit`` content bytes. + + An event larger than ``limit`` gets a list of its own. + """ + batches: list[list[FileChangeEvent]] = [] + size = 0 + for event in events: + event_size = len(event.content or b"") + if not batches or size + event_size > limit: + batches.append([]) + size = 0 + batches[-1].append(event) + size += event_size + return batches + + class DatasiteOwnerSyncer(BaseModelCallbackMixin): """Responsible for downloading files and checking permissions""" @@ -129,20 +201,13 @@ class DatasiteOwnerSyncer(BaseModelCallbackMixin): collection_specs: List[CollectionSyncSpec] = [] syftbox_events_queue: Queue[FileChangeEventsMessage] = Field(default_factory=Queue) - outbox_queue: Queue[Tuple[str, FileChangeEventsMessage]] = Field( - default_factory=Queue - ) + outbox_queue: Queue[OutgoingMessage] = Field(default_factory=Queue) _executor: ThreadPoolExecutor = PrivateAttr( default_factory=lambda: ThreadPoolExecutor(max_workers=10) ) # Cache of collections shared with "any" - list of (tag, content_hash) tuples _any_shared_collections: List[tuple] = PrivateAttr(default_factory=list) - # Cache of read permissions per file path → frozenset of peer emails - _read_perm_cache: dict[str, frozenset[str]] = PrivateAttr(default_factory=dict) - # Approved peers that fail the version gate in this sync. Their messages - # stay pending and are not uploaded. - _deferred_peer_emails: frozenset[str] = PrivateAttr(default_factory=frozenset) # In-memory rolling state for tracking events since last checkpoint _rolling_state: RollingState | None = PrivateAttr(default=None) @@ -201,56 +266,57 @@ def _save_rolling_state(self) -> None: tmp.write_text(self._rolling_state.model_dump_json()) tmp.rename(path) - def sync( - self, - peer_emails: list[str], - recompute_hashes: bool = True, - deferred_peer_emails: tuple[str, ...] | list[str] = (), - ): - """Push local changes and pull proposed changes from ``peer_emails``. + def sync(self, peer_emails: list[str], recompute_hashes: bool = True): + """Record the owner's changes, pull the proposed changes of + ``peer_emails``, then send each of them what it does not hold yet, in + one pass. - ``deferred_peer_emails`` are approved peers that fail the version gate. - Changes they can read stay pending, and go out in a later sync where - the peer is in ``peer_emails``. + Accepted proposals and the owner's own changes go out the same way: to + every peer that may read the file. A proposal that fails is logged as + an error and does not hold back the send. """ if self.persist_owner_state: self._load_rolling_state() - self._deferred_peer_emails = frozenset(deferred_peer_emails) try: if not self.initial_sync_done: self.pull_initial_state() - - self._queue_pending_deliveries(peer_emails) + # The one permission load of this pass. Each step below reads it. + self.perm_context._reload() + # Before the local changes are recorded, so that a file changed + # while the owner was offline is not seeded as held. + self._seed_views_if_missing(peer_emails) + # Before the proposals, so that a proposal based on an older + # version of a file the owner changed is refused as outdated. if recompute_hashes: - self.process_local_changes( - recipients=[*peer_emails, *deferred_peer_emails] - ) - - # first, pull existing state + self._record_local_changes() for peer_email in peer_emails: - proposed_count = 0 - try: - while True: - msg = self.pull_and_process_next_proposed_filechange( - peer_email, raise_on_none=False - ) - if msg is None: - # no new message, we are done - break - proposed_count += len(msg.proposed_file_changes) - except SyftPeerNotReadyError as e: - warnings.warn(f"Messages from {peer_email} stay in the inbox: {e}") - if proposed_count: - print( - f"Received {proposed_count} proposed change(s) " - f"from {peer_email}" - ) - self.process_syftbox_events_queue() + self._pull_proposals(peer_email) + # The owner's log goes out before any peer gets a change. + self.save_owner_state() + self.deliver_file_changes(peer_emails) finally: - self._deferred_peer_emails = frozenset() if self.persist_owner_state: self._save_rolling_state() + def _pull_proposals(self, peer_email: str) -> None: + """Apply every proposed change waiting from ``peer_email``. + + A proposal that fails is logged as an error, so that it does not hold + back the other peers or the owner's changes. + """ + proposed_count = 0 + try: + while msg := self.pull_and_process_next_proposed_filechange( + peer_email, raise_on_none=False + ): + proposed_count += len(msg.proposed_file_changes) + except SyftPeerNotReadyError as e: + warnings.warn(f"Messages from {peer_email} stay in the inbox: {e}") + except Exception as e: + logger.error(f"Proposed changes from {peer_email} failed: {e!r}") + if proposed_count: + print(f"Received {proposed_count} proposed change(s) from {peer_email}") + def download_events_message_by_id_with_connection( self, events_message_id: str ) -> FileChangeEventsMessage: @@ -291,109 +357,102 @@ def pull_initial_state(self): def _restore_owner_state(self) -> None: """Rebuild the cache from the owner's persisted state on the backend.""" - restore = self._restore_from_checkpoints() - - if restore.events_since_timestamp is not None: - self._download_events_since(restore.events_since_timestamp) - elif not restore.found_checkpoint: - print("No checkpoints found, downloading all events...") - self._download_all_events() + state = self.fetch_owner_state( + whole_log_since=self.event_cache.latest_cached_timestamp + ) + self._apply_full_checkpoint(state.full_checkpoint) + events = self._apply_incremental_checkpoints(state.incremental_checkpoints) + events += self._apply_rolling_state(state) + self._apply_log_messages(state) # The restore above only populates file_hashes and file_connection, not # events_messages_connection, which get_cached_events() reads from. - if restore.events and not self.event_cache.get_cached_events(): - self._write_events_to_messages_cache(restore.events) + if events and not self.event_cache.get_cached_events(): + self._write_events_to_messages_cache(events) - def _restore_from_checkpoints(self) -> _CheckpointRestore: - """Apply the full checkpoint, then the incrementals, then the rolling state. + def fetch_owner_state( + self, whole_log_since: float | None = None + ) -> OwnerStateOnBackend: + """Download the owner's persisted state: the full checkpoint, the + incrementals, the rolling state, and the event-log messages after them. - Each step restores cache state and moves the event cursor forward. + ``whole_log_since`` limits the log when it is read whole, because + there is no checkpoint and no rolling state. """ - full_checkpoint = self.connection_router.get_latest_checkpoint() - since = self._apply_full_checkpoint(full_checkpoint) - - incremental_cps = self.connection_router.get_all_incremental_checkpoints() - since, events = self._apply_incremental_checkpoints(incremental_cps, since) - - since, rolling_events = self._apply_rolling_state(since) - return _CheckpointRestore( - events_since_timestamp=since, - events=events + rolling_events, - found_checkpoint=full_checkpoint is not None or bool(incremental_cps), + router = self.connection_router + full_checkpoint = router.get_latest_checkpoint() + incrementals = router.get_all_incremental_checkpoints() + rolling_state = router.get_rolling_state() + if rolling_state is not None and rolling_state.event_count == 0: + rolling_state = None + since = _last_persisted_event_timestamp( + full_checkpoint, incrementals, rolling_state + ) + log_is_whole = since is None and full_checkpoint is None and not incrementals + if since is not None: + log_messages = router.get_events_messages_since_timestamp(since) + elif log_is_whole: + log_messages = self.get_all_accepted_events_messages(whole_log_since) + else: + log_messages = [] + return OwnerStateOnBackend( + full_checkpoint, incrementals, rolling_state, log_messages, log_is_whole ) - def _apply_full_checkpoint(self, checkpoint: Checkpoint | None) -> float | None: - """Restore a full snapshot. Returns the event cursor it carries.""" + def _apply_full_checkpoint(self, checkpoint: Checkpoint | None) -> None: + """Restore a full snapshot.""" if checkpoint is None: - return None + return print(f"Found full checkpoint with {len(checkpoint.files)} files, restoring...") self.event_cache.apply_checkpoint(checkpoint, write_files=self.write_files) - return checkpoint.last_event_timestamp def _apply_incremental_checkpoints( - self, - checkpoints: List[IncrementalCheckpoint], - since: float | None, - ) -> Tuple[float | None, List[FileChangeEvent]]: - """Apply incrementals in order, advancing the cursor past their events.""" + self, checkpoints: List[IncrementalCheckpoint] + ) -> List[FileChangeEvent]: + """Apply incrementals in order. Returns their events.""" if not checkpoints: - return since, [] - + return [] print(f"Found {len(checkpoints)} incremental checkpoints, applying...") events: List[FileChangeEvent] = [] for inc_cp in checkpoints: self._apply_incremental_checkpoint_to_cache(inc_cp) events.extend(inc_cp.events) - for event in inc_cp.events: - since = _later_timestamp(since, event.timestamp) - return since, events - - def _apply_rolling_state( - self, since: float | None - ) -> Tuple[float | None, List[FileChangeEvent]]: - """Apply the rolling state, or start an empty one when there is none.""" - rolling_state = self.connection_router.get_rolling_state() - if rolling_state is None or rolling_state.event_count == 0: + return events + + def _apply_rolling_state(self, state: OwnerStateOnBackend) -> List[FileChangeEvent]: + """Apply the rolling state, or start an empty one when there is none. + Returns its events.""" + rolling_state = state.rolling_state + if rolling_state is None: + since = _last_persisted_event_timestamp( + state.full_checkpoint, state.incremental_checkpoints + ) self._rolling_state = RollingState( - email=self.email, - base_checkpoint_timestamp=since or 0.0, + email=self.email, base_checkpoint_timestamp=since or 0.0 ) - return since, [] - + return [] print( f"Found rolling state with {rolling_state.event_count} events, applying..." ) self._apply_rolling_state_to_cache(rolling_state) self._rolling_state = rolling_state - return ( - _later_timestamp(since, rolling_state.last_event_timestamp), - rolling_state.events, - ) + return rolling_state.events - def _download_events_since(self, timestamp: float) -> None: - """Download and cache every events message newer than `timestamp`.""" - events_messages = self.connection_router.get_events_messages_since_timestamp( - timestamp - ) - if not events_messages: - return - - print( - f"Downloading {len(events_messages)} events since " - "checkpoint/rolling state..." - ) - for events_message in events_messages: - self.event_cache.add_events_message_to_local_cache(events_message) - self._add_events_to_rolling_state(events_message) - - def _download_all_events(self) -> None: - """Download every accepted events message - the no-checkpoint path.""" - since_timestamp = self.event_cache.latest_cached_timestamp - events_messages_list: List[FileChangeEventsMessage] = ( - self.get_all_accepted_events_messages(since_timestamp=since_timestamp) - ) - for events_message in events_messages_list: + def _apply_log_messages(self, state: OwnerStateOnBackend) -> None: + """Cache the event-log messages after the checkpoints and the rolling + state. Only those that follow a checkpoint or a rolling state go into + the rolling state.""" + if state.log_is_whole: + print("No checkpoints found, downloading all events...") + elif state.log_messages: + print( + f"Downloading {len(state.log_messages)} events since " + "checkpoint/rolling state..." + ) + for events_message in state.log_messages: self.event_cache.add_events_message_to_local_cache(events_message) + if not state.log_is_whole: + self._add_events_to_rolling_state(events_message) def _apply_incremental_checkpoint_to_cache( self, checkpoint: IncrementalCheckpoint @@ -587,208 +646,119 @@ def _download_file_with_new_connection(self, file_id: str) -> bytes: connection = self.connection_router.connection_for_parallel_download() return connection.watcher_download_collection_file(file_id) - def _get_readers(self, path: str, recipients: list[str]) -> frozenset[str]: - """Return the set of recipients that have read access to the given path.""" - return frozenset(r for r in recipients if self.check_read_permissions(r, path)) + def _create_current_filestate_event(self, path: str) -> FileChangeEvent: + """An event that sets ``path`` on a peer to its current local state. - def _get_paths_under_perm_file(self, perm_path: str) -> list[str]: - """Find all files under the perm file's parent directory on the filesystem.""" - datasite = self.syftbox_folder / self.email - parent_dir = datasite / str(Path(perm_path).parent) - if not parent_dir.exists(): - return [] - collections_rel = self.event_cache.collections_relative_path() - paths = [] - for file_path in parent_dir.rglob("*"): - if file_path.is_file(): - rel = str(file_path.relative_to(datasite)) - if is_normal_syncable_path(rel, collections_path=collections_rel): - paths.append(rel) - return paths - - def _create_resend_event(self, path: str) -> "FileChangeEvent | None": - """Create a FileChangeEvent for an existing file to resend to new readers.""" + A file missing from the cache gives a deletion. A read error raises. + """ try: content = self.event_cache.file_connection.read_file(path) - except Exception: - return None - if content is None: - return None - from syft.sync.utils.syftbox_utils import ( - get_event_hash_from_content, - create_event_timestamp, - ) - + except (KeyError, FileNotFoundError): + # The in-memory cache raises KeyError, the filesystem cache + # FileNotFoundError. + return self._deletion_event(path) timestamp = create_event_timestamp() return FileChangeEvent( id=uuid4(), path_in_datasite=Path(path), content=content, - old_hash=None, new_hash=get_event_hash_from_content(content), submitted_timestamp=timestamp, timestamp=timestamp, datasite_email=self.email, - is_deleted=False, ) - def _create_current_state_event(self, path: str) -> FileChangeEvent: - """An event that sets ``path`` on a peer to its current local state. - - A missing file gives a deletion. A read error raises. - """ - content = None - if (self.syftbox_folder / self.email / path).exists(): - content = self.event_cache.file_connection.read_file(path) + def _deletion_event(self, path: str) -> FileChangeEvent: + """An event that removes ``path`` from a peer.""" timestamp = create_event_timestamp() return FileChangeEvent( id=uuid4(), path_in_datasite=Path(path), - content=content, - new_hash=None if content is None else get_event_hash_from_content(content), + content=None, + new_hash=None, submitted_timestamp=timestamp, timestamp=timestamp, datasite_email=self.email, - is_deleted=content is None, + is_deleted=True, ) - def _queue_pending_deliveries(self, peer_emails: list[str]) -> None: - """Queue the current state of each pending path for each peer.""" - pending = self.event_cache.pending_deliveries - for peer_email in peer_emails: - paths = [ - path - for path in pending.get(peer_email) or {} - if self.check_read_permissions(peer_email, path) - ] - if paths: - msg = FileChangeEventsMessage( - events=[self._create_current_state_event(p) for p in paths] - ) - self._queue_for_outbox(peer_email, msg) - name = msg.message_filepath.as_string() - else: - name = None - # Drop the paths that this resend does not hold. The peer can no - # longer read them. - pending.update_entry( - peer_email, - lambda old: {p: m for p, m in (old or {}).items() if m == name} or None, - ) + def _record_local_changes(self) -> None: + """Update the file hashes from the local files, and add the changes to + the owner's persisted state. The caller then calls ``save_owner_state``.""" + file_change_events_message = self.event_cache.process_local_file_changes() + if file_change_events_message is not None: + self._queue_for_owner_state(file_change_events_message) - def _queue_for_outbox(self, recipient: str, msg: FileChangeEventsMessage) -> None: - """Queue ``msg`` for ``recipient``. Its paths stay pending until the upload succeeds.""" - name = msg.message_filepath.as_string() - paths = {str(e.path_in_datasite): name for e in msg.events} - self.event_cache.pending_deliveries.update_entry( - recipient, lambda old: {**(old or {}), **paths} - ) - self.outbox_queue.put((recipient, msg)) - - def _clear_delivered(self, recipient: str, msg: FileChangeEventsMessage) -> None: - """Remove the paths that ``msg`` delivered, unless a later message holds them.""" - name = msg.message_filepath.as_string() - self.event_cache.pending_deliveries.update_entry( - recipient, - lambda old: {p: m for p, m in (old or {}).items() if m != name} or None, - ) + def _is_syncable(self, path: str) -> bool: + """Whether a peer can get ``path`` by a delivery plan.""" + collections_rel = self.event_cache.collections_relative_path() + return is_normal_syncable_path(path, collections_path=collections_rel) + + def _syncable_hashes(self) -> dict[str, str]: + """The current hash of every file that a peer can get, by path.""" + return { + str(path): file_hash + for path, file_hash in self.event_cache.file_hashes.items() + if self._is_syncable(str(path)) + } - def process_local_changes(self, recipients: list[str]): - file_change_events_message = self.event_cache.process_local_file_changes() - if file_change_events_message is None: - return + def _readable_files_by_peer( + self, peer_emails: list[str] + ) -> dict[str, dict[str, str]]: + """For each peer, the current hash of every syncable file it may read. - perm_events, data_events = self._split_events(file_change_events_message.events) - events_by_recipient: dict[str, list[FileChangeEvent]] = { - r: [] for r in recipients + Maps each peer email to ``{path: hash}``. The caller reloads the + permissions first. + """ + syncable = self._syncable_hashes() + # peer email -> {path: hash} + return { + peer: { + p: h for p, h in syncable.items() if self.perm_context.can_read(p, peer) + } + for peer in peer_emails } - data_event_sent_to: dict[str, set[str]] = {} - self._route_data_events( - data_events, recipients, events_by_recipient, data_event_sent_to - ) - self._route_perm_events( - perm_events, recipients, events_by_recipient, data_event_sent_to - ) - self._queue_events_for_outbox(events_by_recipient) - if self.persist_owner_state: - # Record local events in the owner's log and feed them into rolling - # state so they flow through incremental/full checkpoints. - self.syftbox_events_queue.put(file_change_events_message) - self._add_events_to_rolling_state(file_change_events_message) - self.process_syftbox_events_queue() - - def _split_events( - self, events: list[FileChangeEvent] - ) -> tuple[list[FileChangeEvent], list[FileChangeEvent]]: - """Split events into (permission_events, data_events).""" - from syft_permissions.spec.ruleset import PERMISSION_FILE_NAME - - perm_events = [] - data_events = [] - for event in events: - # Compare the last path part without case, as the ACL gate does. - if event.path_in_datasite.name.casefold() == PERMISSION_FILE_NAME: - perm_events.append(event) - else: - data_events.append(event) - return perm_events, data_events + def _may_hold(self, peer: str, path: str, syncable: Mapping[str, str]) -> bool: + """Whether a view of ``peer`` may hold ``path``: a file in ``syncable`` + that ``peer`` may read.""" + return path in syncable and self.perm_context.can_read(path, peer) - def _route_data_events( - self, - data_events: list[FileChangeEvent], - recipients: list[str], - events_by_recipient: dict[str, list[FileChangeEvent]], - data_event_sent_to: dict[str, set[str]], - ): - """Route data events to recipients who have read access.""" - for event in data_events: - path_str = str(event.path_in_datasite) - readers = self._get_readers(path_str, recipients) - self._read_perm_cache[path_str] = readers - data_event_sent_to[path_str] = set(readers) - for reader in readers: - events_by_recipient[reader].append(event) - - def _route_perm_events( - self, - perm_events: list[FileChangeEvent], - recipients: list[str], - events_by_recipient: dict[str, list[FileChangeEvent]], - data_event_sent_to: dict[str, set[str]], - ): - """Share perm files with readers and resend affected files to newly-permitted readers.""" - for event in perm_events: - path_str = str(event.path_in_datasite) - perm_readers = self._get_readers(path_str, recipients) - for reader in perm_readers: - events_by_recipient[reader].append(event) - - affected_paths = self._get_paths_under_perm_file(path_str) - for affected_path in affected_paths: - new_readers = self._get_readers(affected_path, recipients) - old_readers = self._read_perm_cache.get(affected_path, frozenset()) - newly_permitted = new_readers - old_readers - self._read_perm_cache[affected_path] = new_readers - - if not newly_permitted: - continue - already_sent = data_event_sent_to.get(affected_path, set()) - needs_resend = newly_permitted - already_sent - if needs_resend: - resend_event = self._create_resend_event(affected_path) - if resend_event is not None: - for reader in needs_resend: - events_by_recipient[reader].append(resend_event) - - def _queue_events_for_outbox( - self, events_by_recipient: dict[str, list[FileChangeEvent]] - ): - """Wrap events into messages per recipient and queue for outbox.""" - for recipient, events in events_by_recipient.items(): - if events: - msg = FileChangeEventsMessage(events=events) - self._queue_for_outbox(recipient, msg) + def _seed_views_if_missing(self, peer_emails: list[str]) -> None: + """Seed the views one time, for an owner that has file hashes but no views. + + Each approved peer, and each peer in ``peer_emails``, is taken to hold + every file it may read now. The caller reloads the permissions first. + """ + delivery_tracker = self.event_cache.delivery_tracker + if delivery_tracker.is_seeded(): + return + approved = [p.email for p in self.connection_router.peer_store.approved_peers] + peers_to_seed = sorted({*approved, *peer_emails}) + delivery_tracker.seed(self._readable_files_by_peer(peers_to_seed)) + + def deliver_file_changes(self, peer_emails: list[str]) -> None: + """Send each peer the files its view lacks, and deletes for the files it + may no longer have. The caller loads the permissions first.""" + delivery_tracker = self.event_cache.delivery_tracker + plans = delivery_tracker.plan(self._readable_files_by_peer(peer_emails)) + # Build every message before the first one is queued, so a read error + # leaves no message of this plan in the queue. + messages = [ + OutgoingMessage(peer, FileChangeEventsMessage(events=batch)) + for peer, plan in plans.items() + for batch in _batches_by_content_size( + self._plan_events(plan), MAX_DELIVERY_MESSAGE_CONTENT + ) + ] + for message in messages: + self.outbox_queue.put(message) + self._send_outbox_messages() + + def _plan_events(self, plan: DeliveryPlan) -> list[FileChangeEvent]: + """The events that carry out ``plan``: its sends, then its deletes.""" + events = [self._create_current_filestate_event(p) for p in plan.send] + return events + [self._deletion_event(p) for p in plan.delete] def pull_and_process_next_proposed_filechange( self, sender_email: str, raise_on_none=True @@ -825,11 +795,6 @@ def check_write_permission(self, sender_email: str, path: str) -> bool: self.perm_context._reload() return self.perm_context.open(path).has_write_access(sender_email) - def check_read_permissions(self, recipient_email: str, path: str) -> bool: - """Check if recipient has read access to the given path.""" - self.perm_context._reload() - return self.perm_context.open(path).has_read_access(recipient_email) - def handle_proposed_filechange_events_message( self, sender_email: str, proposed_events_message: ProposedFileChangesMessage ): @@ -854,43 +819,123 @@ def handle_proposed_filechange_events_message( accepted_events_message = self.event_cache.process_proposed_events_message( filtered_message ) + # The next plan sends the accepted change to every peer that may read + # the file, the submitter included. if accepted_events_message is not None: - self.queue_event_for_syftbox( - recipients=[sender_email], - file_change_events_message=accepted_events_message, - ) - if self.persist_owner_state: - self._add_events_to_rolling_state(accepted_events_message) + self._queue_for_owner_state(accepted_events_message) - def queue_event_for_syftbox( + def process_events_for_own_state_and_peers( self, recipients: list[str], file_change_events_message: FileChangeEventsMessage - ): - # The syftbox queue is the owner's own append-only log, read back only - # to restore this datasite. The outbox is how peers get the change, so - # it is queued either way. - if self.persist_owner_state: - self.syftbox_events_queue.put(file_change_events_message) - + ) -> None: + """Add ``file_change_events_message`` to the owner state, save that + state, then send the message to each of ``recipients``.""" + self._queue_for_owner_state(file_change_events_message) + self.save_owner_state() for recipient in recipients: - self._queue_for_outbox(recipient, file_change_events_message) + self.outbox_queue.put( + OutgoingMessage(recipient, file_change_events_message) + ) + self._send_outbox_messages() + + def _queue_for_owner_state(self, events_message: FileChangeEventsMessage) -> None: + """Add ``events_message`` to the owner's persisted state, read back only + to restore this datasite. + + It goes into the rolling state, and into the owner's event log unless + the rolling state holds every change after each ``save_owner_state``. A rolling state + that is not loaded yet is loaded from disk. - def process_syftbox_events_queue(self): + Raises: + RuntimeError: when there is no rolling state on disk either, because + this datasite never synced here. + """ + if not self.persist_owner_state: + return + if self._rolling_state is None: + self._load_rolling_state() + if self._rolling_state is None: + raise RuntimeError( + "Sync this datasite before changing it: its owner state is not loaded." + ) + if DEFAULT_ROLLING_STATE_UPLOAD_THRESHOLD > 1: + self.syftbox_events_queue.put(events_message) + self._add_events_to_rolling_state(events_message) + + def save_owner_state(self): + """Write the owner's event log, and upload and save its rolling state.""" # TODO: make this atomic while not self.syftbox_events_queue.empty(): file_change_events_message = self.syftbox_events_queue.get() self.connection_router.owner_write_events_message_to_syftbox( file_change_events_message ) + if self.persist_owner_state: + self._upload_rolling_state_if_due() + # Also outside sync(), whose start reloads the rolling state from disk. + self._save_rolling_state() + + def _send_outbox_messages(self) -> None: + """Send the queued messages, and record each sent one in its peer's view. + + Nothing raises, and a failed send does not stop the others. A failed + message goes to the back of this pass, and after ``MAX_SEND_ATTEMPTS`` + tries it is dropped with an error in the log. A peer that is not ready + gets none of its messages in this pass, with a warning. A file that a + view tracks is planned again at the next sync. + """ + pending = deque() while not self.outbox_queue.empty(): - recipient, message = self.outbox_queue.get() - if recipient in self._deferred_peer_emails: + pending.append(self.outbox_queue.get()) + not_ready = set() + while pending: + entry = pending.popleft() + if entry.recipient in not_ready: continue - # A send that cannot go raises before the clear, so the paths stay - # pending and a later sync sends them. - self.connection_router.owner_write_event_messages_to_outbox( - recipient, message + try: + self.connection_router.owner_write_event_messages_to_outbox( + entry.recipient, entry.message + ) + except SyftPeerNotReadyError as e: + not_ready.add(entry.recipient) + warnings.warn(f"Messages for {entry.recipient} are not sent: {e}") + except Exception as e: + pending.extend(self._retry_or_give_up(entry, e)) + else: + self._record_sent(entry.recipient, entry.message) + + def _retry_or_give_up( + self, entry: OutgoingMessage, error: Exception + ) -> list[OutgoingMessage]: + """The entry to try again for a failed send, if any try is left.""" + name = entry.message.message_filepath.as_string() + attempts = entry.attempts + 1 + if attempts >= MAX_SEND_ATTEMPTS: + logger.error( + f"Gave up sending {name} to {entry.recipient} after {attempts} tries: {error!r}" ) - self._clear_delivered(recipient, message) + return [] + logger.info( + f"Send of {name} to {entry.recipient} failed, trying again: {error!r}" + ) + return [entry._replace(attempts=attempts)] + + def _record_sent(self, recipient: str, message: FileChangeEventsMessage) -> None: + """Add the changes of a sent message to ``recipient``'s view. + + A delete always counts. A file counts only if it is syncable and + ``recipient`` may read it, so a later permission change sends a delete + for it. Other files, such as a private dataset sent to an enclave, are + not in the view. + """ + syncable = self._syncable_hashes() + changes = {} + for event in message.events: + path = str(event.path_in_datasite) + if event.is_deleted: + changes[path] = None + elif self._may_hold(recipient, path, syncable): + changes[path] = event.new_hash + self.event_cache.delivery_tracker.record_send(recipient, changes) def _download_outbox_message_with_new_connection( self, file_id: str @@ -905,10 +950,11 @@ def compact_outbox_if_needed( recipient_email: str, min_messages: int = MIN_MESSAGES_COMPACT, ) -> int: - """Merge accumulated event-message files in a recipient's outbox into one. + """Merge accumulated event-message files in a recipient's outbox into fewer. - Compaction downloads them, concatenates their events into one fresh message, and - deletes the originals. + Compaction downloads them, concatenates their events, writes them again + as messages of at most ``MAX_DELIVERY_MESSAGE_CONTENT`` of file content, + and deletes the originals. Returns the number of source messages compacted (0 if below threshold, if no outbox folder, or if peer encryption is on — DO can't read @@ -962,14 +1008,16 @@ def compact_outbox_if_needed( for msg in messages: merged_events.extend(msg.events) - merged = FileChangeEventsMessage(events=merged_events) - - # Write the merged message FIRST. A DS that pulls between this and + # Write the merged messages FIRST. A DS that pulls between this and # the delete sees `merged ∪ some originals`; its file_hashes cache - # makes duplicate events no-op writes. - self.connection_router.owner_write_event_messages_to_outbox( - recipient_email, merged - ) + # makes duplicate events no-op writes. The size cap of a delivery + # message holds here too. + for batch in _batches_by_content_size( + merged_events, MAX_DELIVERY_MESSAGE_CONTENT + ): + self.connection_router.owner_write_event_messages_to_outbox( + recipient_email, FileChangeEventsMessage(events=batch) + ) conn.delete_multiple_files_by_ids(file_ids) return len(parsed) @@ -1007,26 +1055,28 @@ def _apply_rolling_state_to_cache(self, rolling_state: RollingState) -> None: ) def _add_events_to_rolling_state( - self, - events_message: FileChangeEventsMessage, - upload_threshold: int = DEFAULT_ROLLING_STATE_UPLOAD_THRESHOLD, + self, events_message: FileChangeEventsMessage ) -> None: - """ - Add events to the in-memory rolling state and upload if threshold is reached. - - Args: - events_message: The events to add. - upload_threshold: Upload to GDrive after this many events added. - """ + """Add events to the in-memory rolling state. The next ``save_owner_state`` uploads it.""" if self._rolling_state is None: return - self._rolling_state.add_events_message(events_message) self._events_since_rolling_state_upload += len(events_message.events) - # Upload if threshold reached - if self._events_since_rolling_state_upload >= upload_threshold: + def _upload_rolling_state_if_due(self) -> None: + """Upload the rolling state once enough events were added since the last + upload. A failed upload is logged, and the next ``save_owner_state`` tries again.""" + if ( + self._events_since_rolling_state_upload + < DEFAULT_ROLLING_STATE_UPLOAD_THRESHOLD + ): + return + try: self._upload_rolling_state() + except Exception as e: + logger.error( + f"Rolling state upload failed, the next sync tries again: {e!r}" + ) def _upload_rolling_state(self) -> None: """Upload the in-memory rolling state to GDrive.""" diff --git a/syft/sync/sync/delivery_tracker.py b/syft/sync/sync/delivery_tracker.py index 963bdd8479a..0a14ff946b2 100644 --- a/syft/sync/sync/delivery_tracker.py +++ b/syft/sync/sync/delivery_tracker.py @@ -1,10 +1,8 @@ """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``. +works on paths and content hashes, is told of each send that returned, and +keeps its record through a ``DeliveryStateStore``. Views ----- @@ -16,76 +14,62 @@ 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. +A change that the owner accepts from a peer goes, like any change, to every +peer that may read the file, that peer included. -Recording a send ----------------- -``sent`` takes the ``SendStatus`` from ``P2PConnection.send``: +A new peer starts with an empty view, so it is sent every file it may read. +When the owner rejects a peer, ``reset`` sets each path in its view to +``UNSENT_HASH``. Thus a peer +that the owner rejects and then approves again is sent every file it may read, +and a delete for each file that it may no longer have. -- ``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. +An owner that lost its views file but kept or restored its file hashes, for +example one with ``persist_owner_state`` on a new machine, has file hashes but +no views. ``seed`` then starts the view of each approved peer with every file +it may read now, so the owner does not send every file again. A view that a +send already recorded is kept. -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. +Recording a send +---------------- +``record_send`` changes the view once a send returned. A change that is +not sent, because the upload failed, the peer was skipped, or a process +stopped before the send, stays in the plan of the next sync. 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. +The views are kept on the owner's local disk only, also with +``persist_owner_state``. An owner that loses its local state, file hashes +included, sends every peer every file it may read once again. With +``persist_owner_state``, the file hashes come back from Drive, so ``seed`` +starts the views from them. """ from abc import ABC, abstractmethod -from typing import Iterable, Mapping +from contextlib import nullcontext +from pathlib import Path +from typing import ContextManager, Mapping from pydantic import BaseModel, ConfigDict, Field -from syft.sync.transport.types import SendStatus +from syft.sync.sync.caches.persisted_dict import locked_file, write_text_atomic DELIVERY_VIEWS_FILENAME = "owner_delivery_views.json" # Format of DeliveryState. Raise it when the layout changes, and keep a read # path for every earlier version. DELIVERY_STATE_VERSION = 1 - - -class QueuedSend(BaseModel): - """Changes sent to ``peer`` that wait for confirmation. - - ``changes`` maps each path to its new hash, or None for a delete. - """ - - peer: str - changes: dict[str, str | None] +# A view hash that matches no file, for a path that the peer may hold in any +# version. The next plan sends the path, or deletes it. +UNSENT_HASH = "" class DeliveryState(BaseModel): - """Everything the tracker keeps: views, and sends awaiting confirmation.""" + """Everything the tracker keeps.""" version: int = DELIVERY_STATE_VERSION - # peer -> {path: hash} + # peer email -> {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) + # Whether ``DeliveryTracker.seed`` ran on this state. + seeded: bool = False class DeliveryPlan(BaseModel): @@ -98,56 +82,145 @@ class DeliveryPlan(BaseModel): class DeliveryStateStore(ABC): - """Where a ``DeliveryState`` is kept: a local file, Drive, or memory.""" + """Where a ``DeliveryState`` is kept: a local file or memory.""" @abstractmethod def load(self) -> DeliveryState: - """The stored state, or an empty one.""" + """The stored state, or an empty one. + + Raises: + ValueError: when a later client wrote the state. + """ @abstractmethod def save(self, state: DeliveryState) -> None: """Store ``state``, replacing what was there.""" + def locked(self) -> ContextManager[None]: + """Keep other writers out from a ``load`` to the ``save`` that follows.""" + return nullcontext() + + +class MemoryDeliveryStateStore(DeliveryStateStore): + """A state kept in this process only.""" + + def __init__(self): + self._state = DeliveryState() + + def load(self) -> DeliveryState: + return self._state.model_copy(deep=True) + + def save(self, state: DeliveryState) -> None: + self._state = state.model_copy(deep=True) + + +class LocalDeliveryStateStore(DeliveryStateStore): + """A state kept in a JSON file, shared by the processes of one datasite.""" + + def __init__(self, path: Path): + self.path = path + + def load(self) -> DeliveryState: + if not self.path.exists(): + return DeliveryState() + state = DeliveryState.model_validate_json(self.path.read_text()) + if state.version > DELIVERY_STATE_VERSION: + raise ValueError( + f"{self.path} has version {state.version}, and this client reads " + f"up to version {DELIVERY_STATE_VERSION}." + ) + return state + + def locked(self) -> ContextManager[None]: + return locked_file(self.path) + + def save(self, state: DeliveryState) -> None: + write_text_atomic(self.path, state.model_dump_json()) + + +def _apply_changes_to_view( + view: dict[str, str], changes: Mapping[str, str | None] +) -> None: + """Apply ``changes`` (path -> new hash, None for a delete) to ``view`` + (path -> hash).""" + for path, new_hash in changes.items(): + if new_hash is None: + view.pop(path, None) + else: + view[path] = new_hash + class DeliveryTracker: - """Per-peer views of confirmed file versions, for the data owner.""" + """Per-peer views of the file versions sent to each peer, for the data owner. + + Each call reads the state from the store. Each change reads and writes it + back under the store lock, so the processes of one datasite see the same + views and do not lose each other's changes. + """ 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 clear(self) -> None: + """Forget the whole state: views and the seeded flag.""" + with self.store.locked(): + self.store.save(DeliveryState()) def reset(self, peer: str) -> None: - """Start ``peer``'s view empty (a ``PeerReset``), so the next plan sends - it every file it may read.""" - raise NotImplementedError + """Set each path in ``peer``'s view to ``UNSENT_HASH``. - 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. + The next plan sends ``peer`` every file it may read, and deletes the + paths of its view that it may no longer have. """ - raise NotImplementedError - - 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``. + with self.store.locked(): + state = self.store.load() + view = state.views.get(peer, {}) + state.views[peer] = dict.fromkeys(view, UNSENT_HASH) + self.store.save(state) + + def is_seeded(self) -> bool: + """Whether ``seed`` ran on the stored state.""" + return self.store.load().seeded + + def seed(self, held_by_peer: Mapping[str, Mapping[str, str]]) -> None: + """Add the files in ``held_by_peer`` to each peer's view, one time. + + ``held_by_peer`` maps each peer email to ``{path: hash}``. A path that the view already holds keeps its hash. After the first + call, later calls do nothing. """ - raise NotImplementedError + with self.store.locked(): + state = self.store.load() + if state.seeded: + return + # held_by_peer: peer email -> {path: hash} + for peer, files in held_by_peer.items(): + state.views[peer] = {**files, **state.views.get(peer, {})} + state.seeded = True + self.store.save(state) - def confirm(self, msg_ids: Iterable[str]) -> None: - """Apply the queued sends that ``P2PConnection.delivered()`` reported.""" - raise NotImplementedError + def plan( + self, readable_by_peer: Mapping[str, Mapping[str, str]] + ) -> dict[str, DeliveryPlan]: + """What to send and delete for each peer in ``readable_by_peer``. + + ``readable_by_peer`` maps each peer email to ``{path: hash}``: every + path it may read now, with the path's current hash. + """ + views = self.store.load().views + plans = {} + # readable_by_peer: peer email -> {path: hash} + for peer, readable in readable_by_peer.items(): + view = views.get(peer, {}) + plans[peer] = DeliveryPlan( + send=sorted(p for p, h in readable.items() if view.get(p) != h), + delete=sorted(p for p in view if p not in readable), + ) + return plans + + def record_send(self, peer: str, changes: Mapping[str, str | None]) -> None: + """Record a send of ``changes`` (path -> new hash, None for a delete).""" + with self.store.locked(): + state = self.store.load() + peer_view = state.views.setdefault(peer, {}) + _apply_changes_to_view(peer_view, changes) + self.store.save(state) diff --git a/tests/migrations/unit/test_file_change_events_serialization.py b/tests/migrations/unit/test_file_change_events_serialization.py index fca62d4e252..751e6542a8f 100644 --- a/tests/migrations/unit/test_file_change_events_serialization.py +++ b/tests/migrations/unit/test_file_change_events_serialization.py @@ -95,3 +95,15 @@ def test_from_proposed_filechange_carries_identity_free_items(): assert event.new_hash == proposed.new_hash # Items carry no identity fields on the wire; only envelopes do. assert "canonical_name" not in json.loads(event.model_dump_json()) + + +def test_from_proposed_filechange_keeps_delete(): + proposed = ProposedFileChange( + path_in_datasite="data/file.txt", + old_hash="h1", + datasite_email="do@test.org", + is_deleted=True, + ) + event = FileChangeEvent.from_proposed_filechange(proposed) + assert event.is_deleted + assert event.content is None diff --git a/tests/unit/test_checkpoints.py b/tests/unit/test_checkpoints.py index ca561bc60a3..6f54a66cf50 100644 --- a/tests/unit/test_checkpoints.py +++ b/tests/unit/test_checkpoints.py @@ -12,6 +12,7 @@ import pytest +from syft.sync.sync import datasite_owner_syncer from syft.sync.checkpoints.checkpoint import ( Checkpoint, IncrementalCheckpoint, @@ -143,8 +144,12 @@ def test_checkpoint_restore_on_sync(): assert len(do_cache.file_hashes) == 2 -def test_checkpoint_events_since(): +def test_checkpoint_events_since(monkeypatch): """Test getting events since checkpoint timestamp.""" + # At 1 the event log is not written. + monkeypatch.setattr( + datasite_owner_syncer, "DEFAULT_ROLLING_STATE_UPLOAD_THRESHOLD", 2 + ) ds_manager, do_manager = SyftboxManager.pair_with_mock_drive_service_connection( use_in_memory_cache=True, ) @@ -732,6 +737,6 @@ def test_owner_state_is_written_by_default(): do_manager.sync(auto_checkpoint=False) router = do_manager._connection_router - assert router.owner_get_all_accepted_event_file_ids() != [] + assert do_manager._get_all_accepted_events_do() != [] assert router.get_rolling_state() is not None assert do_manager.datasite_owner_syncer._rolling_state is not None diff --git a/tests/unit/test_delivery_tracker.py b/tests/unit/test_delivery_tracker.py new file mode 100644 index 00000000000..273c08ccc4b --- /dev/null +++ b/tests/unit/test_delivery_tracker.py @@ -0,0 +1,522 @@ +"""The data owner sends each peer what it lacks, and deletes what it may no +longer have, from a per-peer record of what the peer holds.""" + +import threading +from pathlib import Path + +import pytest + +from syft.sync.connections.connection_router import ConnectionRouter +from syft.sync.peers.exceptions import SyftPeerNotReadyError +from syft.sync.peers.peer import PeerState +from syft.sync.sync import datasite_owner_syncer +from syft.sync.sync.constants import CACHE_DIR +from syft.sync.sync.caches.datasite_owner_cache import ( + ProposedEventFileOutdatedException, +) +from syft.sync.sync.delivery_tracker import ( + DELIVERY_STATE_VERSION, + DELIVERY_VIEWS_FILENAME, + DeliveryState, + DeliveryTracker, + LocalDeliveryStateStore, + MemoryDeliveryStateStore, +) +from syft.sync.syftbox_manager import SyftboxManager +from syft_perms import SyftPermContext + +PEER = "ds@example.com" + + +def _tracker_holding(view: dict[str, str]) -> DeliveryTracker: + tracker = DeliveryTracker(MemoryDeliveryStateStore()) + tracker.record_send(PEER, view) + return tracker + + +def test_plan_sends_changed_files_deletes_unreadable(): + tracker = _tracker_holding({"same": "h1", "old": "h1", "revoked": "h1"}) + plan = tracker.plan({PEER: {"same": "h1", "old": "h2", "new": "h3"}})[PEER] + assert plan.send == ["new", "old"] + assert plan.delete == ["revoked"] + + +def test_delivered_send_updates_view(): + tracker = _tracker_holding({"a": "h1", "b": "h1"}) + tracker.record_send(PEER, {"a": "h2", "b": None}) + assert tracker.plan({PEER: {"a": "h2"}})[PEER].send == [] + assert tracker.plan({PEER: {"a": "h2"}})[PEER].delete == [] + + +def test_seed_runs_once_and_keeps_recorded_views(): + tracker = _tracker_holding({"b": "h9"}) + tracker.seed({PEER: {"a": "h1", "b": "h1"}}) + tracker.seed({PEER: {}}) + assert tracker.plan({PEER: {"a": "h1", "b": "h9"}})[PEER].send == [] + + +def test_reset_resends_readable_deletes_gone(): + tracker = _tracker_holding({"a": "h1", "gone": "h1"}) + tracker.reset(PEER) + plan = tracker.plan({PEER: {"a": "h1"}})[PEER] + assert plan.send == ["a"] + assert plan.delete == ["gone"] + + +def test_local_store_persists_state_refuses_later_version(tmp_path): + path = tmp_path / DELIVERY_VIEWS_FILENAME + DeliveryTracker(LocalDeliveryStateStore(path)).record_send(PEER, {"a": "h1"}) + assert ( + DeliveryTracker(LocalDeliveryStateStore(path)) + .plan({PEER: {"a": "h1"}})[PEER] + .send + == [] + ) + + path.write_text(DeliveryState(version=DELIVERY_STATE_VERSION + 1).model_dump_json()) + with pytest.raises(ValueError, match="version"): + LocalDeliveryStateStore(path).load() + + +def test_overlapping_writers_keep_both_views(tmp_path, monkeypatch): + path = tmp_path / DELIVERY_VIEWS_FILENAME + first_loaded = threading.Event() + go = threading.Event() + load = LocalDeliveryStateStore.load + + def load_then_wait_in_first_writer(self): + state = load(self) + if threading.current_thread().name == "first": + first_loaded.set() + go.wait(timeout=5) + return state + + monkeypatch.setattr(LocalDeliveryStateStore, "load", load_then_wait_in_first_writer) + + def send(peer: str): + tracker = DeliveryTracker(LocalDeliveryStateStore(path)) + tracker.record_send(peer, {"a": "h1"}) + + first = threading.Thread(target=send, args=("one@x.com",), name="first") + second = threading.Thread(target=send, args=("two@x.com",), name="second") + first.start() + first_loaded.wait(timeout=5) + second.start() + # Without the lock, the second writer saves now, and the first writer then + # overwrites it with the state it read before. + second.join(timeout=0.5) + go.set() + first.join() + second.join() + + plans = DeliveryTracker(LocalDeliveryStateStore(path)).plan( + {"one@x.com": {"a": "h1"}, "two@x.com": {"a": "h1"}} + ) + assert plans["one@x.com"].send == plans["two@x.com"].send == [] + + +# --- The owner syncer --------------------------------------------------------- + + +def _pair(**kwargs): + """A DS and DO pair. The DO synced once, so its owner state is loaded.""" + ds_manager, do_manager = SyftboxManager.pair_with_mock_drive_service_connection( + use_in_memory_cache=False, **kwargs + ) + do_manager.sync(auto_checkpoint=False) + return ds_manager, do_manager + + +def _write(do_manager, path: str, content: str): + file = do_manager.syftbox_folder / do_manager.email / path + file.parent.mkdir(parents=True, exist_ok=True) + file.write_text(content) + + +def _received(ds_manager, do_manager, path: str): + return ds_manager.syftbox_folder / do_manager.email / path + + +def _outbox_paths(ds_manager, do_manager) -> list[str]: + messages = ds_manager._connection_router.watcher_get_events_messages( + do_manager.email, None + ) + return [str(e.path_in_datasite) for m in messages for e in m.events] + + +def _sent_events(ds_manager, do_manager, path: str) -> list[tuple[bool, object]]: + """(is_deleted, content) of each event for ``path`` in the DS's outbox.""" + messages = ds_manager._connection_router.watcher_get_events_messages( + do_manager.email, None + ) + return [ + (e.is_deleted, e.content) + for m in messages + for e in m.events + if str(e.path_in_datasite) == path + ] + + +def _lose_views(do_manager) -> None: + """Delete the DO's views file, keeping its file hashes.""" + (do_manager.syftbox_folder / CACHE_DIR / DELIVERY_VIEWS_FILENAME).unlink() + + +def _push_private_file(do_manager, ds_manager) -> None: + """Send a file outside the plan, as a private dataset share does.""" + syncer = do_manager.datasite_owner_syncer + _write(do_manager, "private/data.csv", "rows") + message = syncer.event_cache.create_events_for_files( + {Path("private/data.csv"): b"rows"} + ) + syncer.process_events_for_own_state_and_peers( + recipients=[ds_manager.email], file_change_events_message=message + ) + + +def test_peer_approved_later_gets_existing_files(): + ds_manager, do_manager = _pair(add_peers=False) + do_manager.datasite_owner_syncer.perm_context.open("shared/").grant_read_access( + ds_manager.email + ) + _write(do_manager, "shared/data.txt", "existing") + do_manager.sync(auto_checkpoint=False) + + ds_manager.add_peer(do_manager.email, sync=False) + do_manager.load_peers() + do_manager.approve_peer_request(ds_manager.email) + ds_manager.load_peers() + do_manager.sync(auto_checkpoint=False) + ds_manager.sync() + + assert ( + _received(ds_manager, do_manager, "shared/data.txt").read_text() == "existing" + ) + + +@pytest.mark.parametrize( + "use_in_memory_cache, persist_owner_state", + [(False, True), (False, False), (True, False)], +) +def test_revoked_peer_gets_content_then_delete( + use_in_memory_cache, persist_owner_state +): + ds_manager, do_manager = SyftboxManager.pair_with_mock_drive_service_connection( + use_in_memory_cache=use_in_memory_cache, persist_owner_state=persist_owner_state + ) + syncer = do_manager.datasite_owner_syncer + folder = syncer.perm_context.open("shared/") + folder.grant_read_access(ds_manager.email) + syncer.event_cache.file_connection.write_file("shared/data.txt", "secret") + + do_manager.sync(auto_checkpoint=False) + do_manager.sync(auto_checkpoint=False) + folder.revoke_read_access(ds_manager.email) + do_manager.sync(auto_checkpoint=False) + assert _sent_events(ds_manager, do_manager, "shared/data.txt") == [ + (False, "secret"), + (True, None), + ] + + +def test_rejected_then_approved_peer_gets_every_file_and_deletes(): + ds_manager, do_manager = _pair() + do_manager.datasite_owner_syncer.perm_context.open("shared/").grant_read_access( + ds_manager.email + ) + _write(do_manager, "shared/data.txt", "one") + _write(do_manager, "shared/gone.txt", "one") + do_manager.sync(auto_checkpoint=False) + + # The DS asks again. The DO rejects it, deletes a file, and then approves a + # later request. + peer = do_manager.peer_manager.get_cached_peer(ds_manager.email) + peer.state = PeerState.REQUESTED_BY_PEER + do_manager.reject_peer_request(ds_manager.email) + (do_manager.syftbox_folder / do_manager.email / "shared/gone.txt").unlink() + peer.state = PeerState.REQUESTED_BY_PEER + do_manager.approve_peer_request(ds_manager.email) + + do_manager.sync(auto_checkpoint=False) + assert _outbox_paths(ds_manager, do_manager).count("shared/data.txt") == 2 + assert _sent_events(ds_manager, do_manager, "shared/gone.txt")[-1] == (True, None) + + +def test_restarted_owner_sends_nothing_new(): + ds_manager, do_manager = _pair() + do_manager.datasite_owner_syncer.perm_context.open("shared/").grant_read_access( + ds_manager.email + ) + _write(do_manager, "shared/data.txt", "one") + do_manager.sync(auto_checkpoint=False) + sent_before = _outbox_paths(ds_manager, do_manager) + + do_manager._copy().sync(auto_checkpoint=False) + assert _outbox_paths(ds_manager, do_manager) == sent_before + + +def test_lost_views_sends_only_offline_changes(): + ds_manager, do_manager = _pair() + do_manager.datasite_owner_syncer.perm_context.open("shared/").grant_read_access( + ds_manager.email + ) + _write(do_manager, "shared/held.txt", "one") + do_manager.sync(auto_checkpoint=False) + + _lose_views(do_manager) + _write(do_manager, "shared/offline.txt", "new") + + do_manager.sync(auto_checkpoint=False) + sent = _outbox_paths(ds_manager, do_manager) + assert sent.count("shared/held.txt") == 1 + assert sent.count("shared/offline.txt") == 1 + + +def test_lost_views_seeded_after_send_outside_plan(): + ds_manager, do_manager = _pair() + do_manager.datasite_owner_syncer.perm_context.open("shared/").grant_read_access( + ds_manager.email + ) + _write(do_manager, "shared/held.txt", "one") + do_manager.sync(auto_checkpoint=False) + _lose_views(do_manager) + + _push_private_file(do_manager, ds_manager) + do_manager.sync(auto_checkpoint=False) + assert _outbox_paths(ds_manager, do_manager).count("shared/held.txt") == 1 + + +def test_lost_views_seeds_peer_skipped_in_first_sync(): + ds_manager, do_manager = _pair() + do_manager.datasite_owner_syncer.perm_context.open("shared/").grant_read_access( + ds_manager.email + ) + _write(do_manager, "shared/held.txt", "one") + do_manager.sync(auto_checkpoint=False) + _lose_views(do_manager) + + # The version gate skips the DS in the first sync after the views are lost. + do_manager.datasite_owner_syncer.sync([]) + do_manager.sync(auto_checkpoint=False) + assert _outbox_paths(ds_manager, do_manager).count("shared/held.txt") == 1 + + +def test_send_to_peer_not_ready_dropped(monkeypatch): + ds_manager, do_manager = _pair() + + def peer_not_ready(self, recipient, *args, **kwargs): + raise SyftPeerNotReadyError(recipient, "keys not known yet", "sync later") + + with monkeypatch.context() as m: + m.setattr( + ConnectionRouter, "owner_write_event_messages_to_outbox", peer_not_ready + ) + with pytest.warns(UserWarning, match="are not sent"): + _push_private_file(do_manager, ds_manager) + + do_manager.datasite_owner_syncer._send_outbox_messages() + assert _sent_events(ds_manager, do_manager, "private/data.csv") == [] + + +def test_peer_not_ready_does_not_block_other_peers(): + ds_manager, do_manager = _pair() + syncer = do_manager.datasite_owner_syncer + not_ready = "stranger@example.com" + shared = syncer.perm_context.open("shared/") + shared.grant_read_access(not_ready) + shared.grant_read_access(ds_manager.email) + _write(do_manager, "shared/data.txt", "one") + + syncer._record_local_changes() + with pytest.warns(UserWarning, match=not_ready): + syncer.deliver_file_changes([not_ready, ds_manager.email]) + assert _outbox_paths(ds_manager, do_manager) == ["shared/data.txt"] + # The skipped peer's view did not change, so its next plan sends the file. + readable = syncer._readable_files_by_peer([not_ready]) + assert syncer.event_cache.delivery_tracker.plan(readable)[not_ready].send == [ + "shared/data.txt" + ] + + +def test_failed_send_retried_then_logged_without_blocking_others(monkeypatch, caplog): + ds_manager, do_manager = _pair() + syncer = do_manager.datasite_owner_syncer + broken = "broken@example.com" + shared = syncer.perm_context.open("shared/") + shared.grant_read_access(broken) + shared.grant_read_access(ds_manager.email) + _write(do_manager, "shared/data.txt", "one") + syncer._record_local_changes() + + send = ConnectionRouter.owner_write_event_messages_to_outbox + + def fail_for_broken(self, recipient, message): + if recipient == broken: + raise ConnectionError("upload failed") + return send(self, recipient, message) + + monkeypatch.setattr( + ConnectionRouter, "owner_write_event_messages_to_outbox", fail_for_broken + ) + syncer.deliver_file_changes([broken, ds_manager.email]) + assert _outbox_paths(ds_manager, do_manager) == ["shared/data.txt"] + tries = [r for r in caplog.records if broken in r.getMessage()] + assert [r.levelname for r in tries] == ["INFO", "INFO", "ERROR"] + + +def test_send_to_stranger_does_not_block_later_syncs(): + ds_manager, do_manager = _pair() + syncer = do_manager.datasite_owner_syncer + _write(do_manager, "private/data.csv", "rows") + message = syncer.event_cache.create_events_for_files( + {Path("private/data.csv"): b"rows"} + ) + with pytest.warns(UserWarning, match="stranger@example.com"): + syncer.process_events_for_own_state_and_peers( + recipients=["stranger@example.com"], file_change_events_message=message + ) + + syncer.perm_context.open("shared/").grant_read_access(ds_manager.email) + _write(do_manager, "shared/data.txt", "one") + do_manager.sync(auto_checkpoint=False) + assert _outbox_paths(ds_manager, do_manager) == ["shared/data.txt"] + + +def test_private_dataset_sent_to_peer_never_revoked(): + ds_manager, do_manager = _pair() + _push_private_file(do_manager, ds_manager) + + do_manager.sync(auto_checkpoint=False) + do_manager.sync(auto_checkpoint=False) + assert _sent_events(ds_manager, do_manager, "private/data.csv") == [ + (False, b"rows") + ] + + +def test_sync_reloads_permissions_once(monkeypatch): + ds_manager, do_manager = _pair() + do_manager.datasite_owner_syncer.perm_context.open("shared/").grant_read_access( + ds_manager.email + ) + _write(do_manager, "shared/data.txt", "one") + reload = SyftPermContext._reload + reloads = [] + + def counting_reload(self): + reloads.append(self) + reload(self) + + monkeypatch.setattr(SyftPermContext, "_reload", counting_reload) + do_manager.sync(auto_checkpoint=False) + assert _outbox_paths(ds_manager, do_manager) == ["shared/data.txt"] + assert len(reloads) == 1 + + +def test_large_catch_up_split_into_messages(monkeypatch): + monkeypatch.setattr(datasite_owner_syncer, "MAX_DELIVERY_MESSAGE_CONTENT", 6) + ds_manager, do_manager = _pair() + do_manager.datasite_owner_syncer.perm_context.open("shared/").grant_read_access( + ds_manager.email + ) + for name in ("a", "b", "c"): + _write(do_manager, f"shared/{name}.txt", "abc") + + do_manager.sync(auto_checkpoint=False) + messages = ds_manager._connection_router.watcher_get_events_messages( + do_manager.email, None + ) + assert sorted(len(m.events) for m in messages) == [1, 2] + + +def test_outdated_proposal_logged_without_blocking_owner_changes(caplog): + ds_manager, do_manager = _pair() + shared = do_manager.datasite_owner_syncer.perm_context.open("shared/") + shared.grant_write_access(ds_manager.email) + _write(do_manager, "shared/a.txt", "v1") + do_manager.sync(auto_checkpoint=False) + ds_manager.sync() + + # The DS edits a.txt from v1 while the DO edits it too, so the + # proposal is outdated and its handling fails on every sync. + ds_manager._send_file_change(f"{do_manager.email}/shared/a.txt", "ds edit") + _write(do_manager, "shared/a.txt", "do edit") + _write(do_manager, "shared/b.txt", "new") + do_manager.sync(auto_checkpoint=False) + assert ProposedEventFileOutdatedException.__name__ in caplog.text + assert ( + do_manager.syftbox_folder / do_manager.email / "shared/a.txt" + ).read_text() == "do edit" + + ds_manager.sync() + assert _received(ds_manager, do_manager, "shared/b.txt").read_text() == "new" + + +def test_failed_send_not_retried_after_revoke(monkeypatch): + monkeypatch.setattr(datasite_owner_syncer, "MAX_DELIVERY_MESSAGE_CONTENT", 3) + ds_manager, do_manager = _pair() + folder = do_manager.datasite_owner_syncer.perm_context.open("shared/") + folder.grant_read_access(ds_manager.email) + _write(do_manager, "shared/a.txt", "abc") + _write(do_manager, "shared/b.txt", "abc") + + def fail_upload(*args, **kwargs): + raise ConnectionError("upload failed") + + with monkeypatch.context() as m: + m.setattr(ConnectionRouter, "owner_write_event_messages_to_outbox", fail_upload) + do_manager.sync(auto_checkpoint=False) + + # The failed messages must not go out after the revoke. + folder.revoke_read_access(ds_manager.email) + do_manager.sync(auto_checkpoint=False) + assert _outbox_paths(ds_manager, do_manager) == [] + + +def test_read_error_leaves_no_planned_message(monkeypatch): + ds_manager, do_manager = _pair() + syncer = do_manager.datasite_owner_syncer + shared = syncer.perm_context.open("shared/") + shared.grant_read_access(ds_manager.email) + shared.grant_read_access("peer_b@example.com") + _write(do_manager, "shared/data.txt", "one") + syncer._record_local_changes() + + # The plan reads the file once per peer. The read for the second peer fails. + reads = [] + create_event = ( + datasite_owner_syncer.DatasiteOwnerSyncer._create_current_filestate_event + ) + + def fail_second_read(self, path): + reads.append(path) + if len(reads) == 2: + raise PermissionError(path) + return create_event(self, path) + + monkeypatch.setattr( + datasite_owner_syncer.DatasiteOwnerSyncer, + "_create_current_filestate_event", + fail_second_read, + ) + with pytest.raises(PermissionError): + syncer.deliver_file_changes([ds_manager.email, "peer_b@example.com"]) + assert syncer.outbox_queue.empty() + + +def test_compaction_keeps_message_size_cap(monkeypatch): + monkeypatch.setattr(datasite_owner_syncer, "MAX_DELIVERY_MESSAGE_CONTENT", 3) + ds_manager, do_manager = _pair() + do_manager.datasite_owner_syncer.perm_context.open("shared/").grant_read_access( + ds_manager.email + ) + for name in ("a", "b", "c", "d"): + _write(do_manager, f"shared/{name}.txt", "abc") + do_manager.sync(auto_checkpoint=False) + + monkeypatch.setattr(datasite_owner_syncer, "MAX_DELIVERY_MESSAGE_CONTENT", 6) + syncer = do_manager.datasite_owner_syncer + assert syncer.compact_outbox_if_needed(ds_manager.email, min_messages=2) == 4 + messages = ds_manager._connection_router.watcher_get_events_messages( + do_manager.email, None + ) + assert [len(m.events) for m in messages] == [2, 2] diff --git a/tests/unit/test_drive_connections.py b/tests/unit/test_drive_connections.py index eabfbd7135c..bd735ed3898 100644 --- a/tests/unit/test_drive_connections.py +++ b/tests/unit/test_drive_connections.py @@ -11,6 +11,7 @@ import pytest import syft.sync.connections.connection_router as connection_router_module +from syft.sync.sync import datasite_owner_syncer from syft.sync.connections.connection_router import ConnectionRouter from syft.sync.connections.drive.gdrive_transport import CollectionFolder from syft.sync.connections.drive.mock_drive_service import pair_with_mock_service @@ -320,9 +321,13 @@ def test_the_router_holds_no_drive_specific_code(): # -- restoring the owner's state -------------------------------------------------- -def test_restoring_from_the_event_log_keeps_the_newest_version(): +def test_restoring_from_the_event_log_keeps_the_newest_version(monkeypatch): """A fresh client that restores from the event log alone replays the messages oldest first, so a file changed twice ends at its last version.""" + # At 1 the event log is not written. + monkeypatch.setattr( + datasite_owner_syncer, "DEFAULT_ROLLING_STATE_UPLOAD_THRESHOLD", 2 + ) _, do = SyftboxManager.pair_with_mock_drive_service_connection( use_in_memory_cache=True ) diff --git a/tests/unit/test_encryption.py b/tests/unit/test_encryption.py index 21db1ae4f89..51567c8ef2a 100644 --- a/tests/unit/test_encryption.py +++ b/tests/unit/test_encryption.py @@ -610,7 +610,7 @@ def test_verified_encrypted_message_flow(): assert len(matching) > 0 -def test_tampered_inbox_message_raises(): +def test_tampered_inbox_message_logs_error(caplog): """Tampered message in mock inbox fails signature verification on sync.""" ds_manager, do_manager = SyftboxManager.pair_with_mock_drive_service_connection( encryption=True, @@ -633,6 +633,9 @@ def test_tampered_inbox_message_raises(): tampered[-1] ^= 0xFF mock_file.content = bytes(tampered) - # DO sync should fail due to signature verification - with pytest.raises(Exception): - do_manager.sync() + # DO sync logs the failed signature verification + do_manager.sync() + assert any( + r.levelname == "ERROR" and ds_manager.email in r.getMessage() + for r in caplog.records + ) diff --git a/tests/unit/test_encryption_policy.py b/tests/unit/test_encryption_policy.py index 56cd8736859..dad4165ab32 100644 --- a/tests/unit/test_encryption_policy.py +++ b/tests/unit/test_encryption_policy.py @@ -27,6 +27,7 @@ resolve_encryption_policy, ) from syft.sync.syftbox_manager import SyftboxManager +from syft.sync.sync.datasite_owner_syncer import OutgoingMessage 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 @@ -427,15 +428,15 @@ def test_negotiated_encryption_flow(ds_policy, do_policy): assert len(_ds_cached_contents(ds_manager)) > 0 -def test_owner_send_to_non_peer_raises(): +def test_owner_send_to_non_peer_warns(): _, do_manager = SyftboxManager.pair_with_mock_drive_service_connection( do_encryption_policy=REQUIRED, ) syncer = do_manager.datasite_owner_syncer message = FileChangeEventsMessage(events=[]) - syncer.outbox_queue.put(("former@example.com", message)) - with pytest.raises(SyftPeerNotReadyError, match="not known"): - syncer.process_syftbox_events_queue() + syncer.outbox_queue.put(OutgoingMessage("former@example.com", message)) + with pytest.warns(UserWarning, match="not known"): + syncer._send_outbox_messages() def test_owner_leaves_envelope_from_peer_without_bundle_in_inbox(): @@ -548,7 +549,7 @@ def send_message(self, recipient, message): def test_unsent_owner_change_stays_pending(monkeypatch): - """The send raises, and the next sync sends the change.""" + """The send warns, and the next sync sends the change.""" ds_manager, do_manager = _pair_with_unsent_do_file() def fail(self, recipient, message): @@ -556,7 +557,7 @@ def fail(self, recipient, message): with monkeypatch.context() as m: m.setattr(ConnectionRouter, "owner_write_event_messages_to_outbox", fail) - with pytest.raises(SyftPeerNotReadyError): + with pytest.warns(UserWarning, match="are not sent"): do_manager.sync(auto_checkpoint=False) _assert_ds_receives_after_notebook_sync(ds_manager, do_manager) diff --git a/tests/unit/test_permission_file_case.py b/tests/unit/test_permission_file_case.py deleted file mode 100644 index 8cb8920074d..00000000000 --- a/tests/unit/test_permission_file_case.py +++ /dev/null @@ -1,47 +0,0 @@ -"""A permission file keeps its meaning whatever case its name has. - -macOS and Windows keep the case of a name but match it without case, so a -write to SYFT.PUB.YAML replaces the content of syft.pub.yaml. The event split -must classify that write as a permission event, or the DO sends it as ordinary -data and never recomputes who may read the files it governs. -""" - -import uuid - -from syft.sync.events.file_change_event import FileChangeEvent -from syft.sync.sync.datasite_owner_syncer import DatasiteOwnerSyncer - - -def _event(path: str) -> FileChangeEvent: - return FileChangeEvent( - id=uuid.uuid4(), - path_in_datasite=path, - datasite_email="owner@test.com", - content="", - submitted_timestamp=0.0, - timestamp=0.0, - ) - - -def _split(paths: list[str]) -> tuple[list[str], list[str]]: - """Split by path. _split_events reads no instance state.""" - perm, data = DatasiteOwnerSyncer._split_events(None, [_event(p) for p in paths]) - return ( - [str(e.path_in_datasite) for e in perm], - [str(e.path_in_datasite) for e in data], - ) - - -def test_permission_events_are_split_by_name_without_case(): - """A name that only ends with the constant stays a data event.""" - perm, data = _split( - [ - "SYFT.PUB.YAML", - "sub/Syft.Pub.Yaml", - "syft.pub.yaml", - "notsyft.pub.yaml", - "sub/data.csv", - ] - ) - assert perm == ["SYFT.PUB.YAML", "sub/Syft.Pub.Yaml", "syft.pub.yaml"] - assert data == ["notsyft.pub.yaml", "sub/data.csv"] diff --git a/tests/unit/test_persisted_cache.py b/tests/unit/test_persisted_cache.py index 7d9c0f49246..a6a462225c2 100644 --- a/tests/unit/test_persisted_cache.py +++ b/tests/unit/test_persisted_cache.py @@ -15,6 +15,11 @@ ) +def _owner_event_count(do_manager) -> int: + """The events in the owner's state on the backend.""" + return sum(len(m.events) for m in do_manager._get_all_accepted_events_do()) + + def test_no_duplicate_events_across_processes(): """Two managers sharing a syftbox_folder must not create duplicate events for the same file change. @@ -40,17 +45,13 @@ def test_no_duplicate_events_across_processes(): # Process A syncs — detects changes, uploads events, saves cache do_manager.sync(auto_checkpoint=False) - events_after_a = len( - do_manager.datasite_owner_syncer.connection_router.owner_get_all_accepted_event_file_ids() - ) + events_after_a = _owner_event_count(do_manager) assert events_after_a > 0 # Process B (syft-bg) syncs — must NOT re-create the same events do_manager_b = do_manager._copy() do_manager_b.sync(auto_checkpoint=False) - events_after_b = len( - do_manager_b.datasite_owner_syncer.connection_router.owner_get_all_accepted_event_file_ids() - ) + events_after_b = _owner_event_count(do_manager_b) # Verify cache files were created cache_dir = do_manager.syftbox_folder / CACHE_DIR @@ -171,8 +172,7 @@ def fail_upload(*args, **kwargs): # ConnectionRouter is a pydantic model, so patch the class for B's sync only. with monkeypatch.context() as m: m.setattr(ConnectionRouter, "owner_write_event_messages_to_outbox", fail_upload) - with pytest.raises(ConnectionError): - do_manager_b.sync(auto_checkpoint=False) + do_manager_b.sync(auto_checkpoint=False) _assert_ds_receives_after_notebook_sync(ds_manager, do_manager) @@ -201,8 +201,8 @@ def test_peer_loses_file_deleted_while_other_process_skipped_it(monkeypatch): assert not received.exists(), "The DS kept a file that the DO deleted." -def test_read_error_on_pending_file_raises_and_keeps_peer_copy(monkeypatch): - """A pending file that cannot be read must not reach the peer as a deletion.""" +def test_read_error_on_unsent_file_raises_and_keeps_peer_copy(monkeypatch): + """A changed file that cannot be read must not reach the peer as a deletion.""" ds_manager, do_manager = _pair_with_unsent_do_file() do_manager.sync(auto_checkpoint=False) ds_manager.sync() @@ -220,7 +220,7 @@ def test_read_error_on_pending_file_raises_and_keeps_peer_copy(monkeypatch): api_file.write_text("print('api v2')") do_manager_b.sync(auto_checkpoint=False) - # The first read is the resend of the pending path. + # The first read is the send of the changed file. read_file = FSFileConnection.read_file calls = [] diff --git a/tests/unit/test_sync_manager.py b/tests/unit/test_sync_manager.py index cfcd0036905..ff516214a4f 100644 --- a/tests/unit/test_sync_manager.py +++ b/tests/unit/test_sync_manager.py @@ -53,6 +53,7 @@ def test_valid_and_invalid_proposed_filechange_event(): do_manager.datasite_owner_syncer.perm_context.open(".").grant_write_access( ds_manager.email ) + do_manager.sync() ds_email = ds_manager.email do_email = do_manager.email @@ -461,6 +462,18 @@ def _add_known_peer(do_manager, email: str) -> None: do_manager.peer_manager.peer_store.add_peer(Peer(email=email, info=info)) +def _peer_router(ds_manager, do_manager, email: str) -> ConnectionRouter: + """Make ``email`` a second peer of the DO, and return its router.""" + conn = GDriveConnection.from_service( + email, ds_manager._connection_router.connections[0].drive_service + ) + conn.add_peer(do_manager.email) + _add_known_peer(do_manager, email) + return ConnectionRouter( + connections=[conn], peer_store=_store_knowing(email, do_manager.email) + ) + + def test_job_files_sync_to_submitter_only(): """Test that job files only sync to the peer who submitted the job. @@ -510,16 +523,8 @@ def test_job_files_sync_to_submitter_only(): # Process local changes with both recipients recipients = [submitter_email, non_submitter_email] - submitter_conn = GDriveConnection.from_service( - submitter_email, ds_manager._connection_router.connections[0].drive_service - ) - submitter_conn.add_peer(do_manager.email) - _add_known_peer(do_manager, submitter_email) - submitter_router = ConnectionRouter( - connections=[submitter_conn], - peer_store=_store_knowing(submitter_email, do_manager.email), - ) - do_manager.datasite_owner_syncer.process_local_changes(recipients) + submitter_router = _peer_router(ds_manager, do_manager, submitter_email) + do_manager.datasite_owner_syncer.sync(recipients) messages_for_non_submitter = ( ds_manager._connection_router.watcher_get_events_messages( @@ -599,7 +604,7 @@ def test_job_file_beside_outputs_does_not_sync_to_submitter(): assert (review_dir / "outputs" / "result.txt").exists() review_rel = review_dir.relative_to(do_manager.syftbox_folder / do_manager.email) - do_manager.datasite_owner_syncer.process_local_changes([ds_email]) + do_manager.datasite_owner_syncer.sync([ds_email]) paths = [ str(event.path_in_datasite) @@ -712,16 +717,7 @@ def test_permission_change_triggers_resend(): datasite_dir_do = do_manager.syftbox_folder / do_manager.email - # Set up a second peer connection router for peer B - peer_b_conn = GDriveConnection.from_service( - peer_b_email, ds_manager._connection_router.connections[0].drive_service - ) - peer_b_conn.add_peer(do_manager.email) - _add_known_peer(do_manager, peer_b_email) - peer_b_router = ConnectionRouter( - connections=[peer_b_conn], - peer_store=_store_knowing(peer_b_email, do_manager.email), - ) + peer_b_router = _peer_router(ds_manager, do_manager, peer_b_email) # Grant only peer A read access to project/ ctx = do_manager.datasite_owner_syncer.perm_context @@ -734,7 +730,7 @@ def test_permission_change_triggers_resend(): # First sync: only peer A should receive the file recipients = [peer_a_email, peer_b_email] - do_manager.datasite_owner_syncer.process_local_changes(recipients) + do_manager.datasite_owner_syncer.sync(recipients) messages_for_a = ds_manager._connection_router.watcher_get_events_messages( do_manager.email, None @@ -757,7 +753,7 @@ def test_permission_change_triggers_resend(): ctx.open("project/").grant_read_access(peer_b_email) # Second sync: peer B should receive data.txt via resend - do_manager.datasite_owner_syncer.process_local_changes(recipients) + do_manager.datasite_owner_syncer.sync(recipients) messages_for_b_after = peer_b_router.watcher_get_events_messages( do_manager.email, None @@ -771,6 +767,105 @@ def test_permission_change_triggers_resend(): ) +def _events_for(router: ConnectionRouter, do_manager) -> list[tuple[str, bool]]: + """(path, is_deleted) of each event in the DO's outbox for ``router``'s peer.""" + return [ + (str(e.path_in_datasite), e.is_deleted) + for m in router.watcher_get_events_messages(do_manager.email, None) + for e in m.events + ] + + +def _accept_setup(ds_manager, do_manager, peer_b_email: str): + """Peer A may write ``shared/``, peer B may read it, and both hold + ``shared/data.txt`` with "one". Returns the syncer, the folder and the peers.""" + syncer = do_manager.datasite_owner_syncer + shared = syncer.perm_context.open("shared/") + shared.grant_write_access(ds_manager.email) + shared.grant_read_access(peer_b_email) + peers = [ds_manager.email, peer_b_email] + data = do_manager.syftbox_folder / do_manager.email / "shared/data.txt" + data.parent.mkdir(parents=True, exist_ok=True) + data.write_text("one") + syncer.sync(peers) + ds_manager.sync() + return syncer, shared, peers + + +def test_accepted_change_reaches_every_reader(): + """A change that the DO accepts from peer A goes to peer A, and to peer B, who + may read the file.""" + ds_manager, do_manager = SyftboxManager.pair_with_mock_drive_service_connection( + use_in_memory_cache=False, + ) + peer_b_email = "peer_b@example.com" + peer_b_router = _peer_router(ds_manager, do_manager, peer_b_email) + syncer, _, peers = _accept_setup(ds_manager, do_manager, peer_b_email) + + ds_manager._send_file_change(f"{do_manager.email}/shared/proposal.txt", "from A") + syncer.sync(peers) + syncer.sync(peers) + assert ("shared/proposal.txt", False) in _events_for( + ds_manager._connection_router, do_manager + ) + assert ("shared/proposal.txt", False) in _events_for(peer_b_router, do_manager) + + +def test_revoke_after_accepted_change_sends_delete(): + """Peer B gets peer A's accepted change to a file, and a delete once it loses + read access.""" + ds_manager, do_manager = SyftboxManager.pair_with_mock_drive_service_connection( + use_in_memory_cache=False, + ) + peer_b_email = "peer_b@example.com" + peer_b_router = _peer_router(ds_manager, do_manager, peer_b_email) + syncer, shared, peers = _accept_setup(ds_manager, do_manager, peer_b_email) + + ds_manager._send_file_change(f"{do_manager.email}/shared/data.txt", "from A") + syncer.sync(peers) + syncer.sync(peers) + shared.revoke_read_access(peer_b_email) + syncer.sync(peers) + assert _events_for(peer_b_router, do_manager) == [ + ("shared/data.txt", False), + ("shared/data.txt", False), + ("shared/data.txt", True), + ] + + +def test_accepted_delete_reaches_every_reader(): + """A delete that the DO accepts from peer A goes to peer A, and to peer B, who + may read the file.""" + ds_manager, do_manager = SyftboxManager.pair_with_mock_drive_service_connection( + use_in_memory_cache=False, + ) + peer_b_email = "peer_b@example.com" + peer_b_router = _peer_router(ds_manager, do_manager, peer_b_email) + syncer, _, peers = _accept_setup(ds_manager, do_manager, peer_b_email) + + proposal = ProposedFileChangesMessage( + sender_email=ds_manager.email, + proposed_file_changes=[ + ProposedFileChange( + path_in_datasite="shared/data.txt", + old_hash=syncer.event_cache.file_hashes[Path("shared/data.txt")], + datasite_email=do_manager.email, + is_deleted=True, + ) + ], + ) + syncer.handle_proposed_filechange_events_message(ds_manager.email, proposal) + syncer.save_owner_state() + syncer.sync(peers) + assert ("shared/data.txt", True) in _events_for( + ds_manager._connection_router, do_manager + ) + assert _events_for(peer_b_router, do_manager) == [ + ("shared/data.txt", False), + ("shared/data.txt", True), + ] + + def test_default_collections_folder_picks_shareable_spec_regardless_of_order(): """The collections folder is chosen by owner_only, not by position.""" from syft.sync.syftbox_manager import default_collections_folder diff --git a/tests/unit/test_transport_contracts.py b/tests/unit/test_transport_contracts.py index dbb84239d10..467fb3aed46 100644 --- a/tests/unit/test_transport_contracts.py +++ b/tests/unit/test_transport_contracts.py @@ -1,6 +1,5 @@ """The transport contracts: the abstract jobs, their helper methods, the -message types, the naming rules and the DeliveryTracker stub. No backend -implements them yet.""" +message types, the naming rules and the DeliveryTracker state.""" from types import SimpleNamespace from unittest.mock import Mock @@ -13,7 +12,6 @@ DELIVERY_STATE_VERSION, DeliveryState, DeliveryStateStore, - DeliveryTracker, ) from syft.sync.transport import ( Collections, @@ -122,12 +120,6 @@ 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(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() @@ -137,4 +129,3 @@ def test_an_empty_delivery_state_is_versioned(): state = DeliveryState() assert state.version == DELIVERY_STATE_VERSION assert state.views == {} - assert state.queued == {}