From dba6c595f786c6b9afecacd3496a2186f56b37db Mon Sep 17 00:00:00 2001 From: Pedro Werneck Date: Thu, 15 Oct 2026 10:33:28 -0300 Subject: [PATCH 1/2] fix: keep DO changes pending until each peer receives them --- syft/sync/syftbox_manager.py | 7 +- syft/sync/sync/caches/datasite_owner_cache.py | 16 +++- syft/sync/sync/caches/persisted_dict.py | 14 +++ syft/sync/sync/constants.py | 1 + syft/sync/sync/datasite_owner_syncer.py | 88 ++++++++++++++++++- tests/unit/test_persisted_cache.py | 83 +++++++++++++++++ 6 files changed, 203 insertions(+), 6 deletions(-) diff --git a/syft/sync/syftbox_manager.py b/syft/sync/syftbox_manager.py index 8daacb29c31..34c32551271 100644 --- a/syft/sync/syftbox_manager.py +++ b/syft/sync/syftbox_manager.py @@ -992,7 +992,12 @@ def sync( ignore_peer_version=ignore_peer_version, ) ) - self.datasite_owner_syncer.sync(compatible_emails) + self.datasite_owner_syncer.sync( + compatible_emails, + deferred_peer_emails=[ + e for e in peer_emails if e not in compatible_emails + ], + ) if auto_compact: for peer_email in compatible_emails: self.datasite_owner_syncer.compact_outbox_if_needed( diff --git a/syft/sync/sync/caches/datasite_owner_cache.py b/syft/sync/sync/caches/datasite_owner_cache.py index 648e99bb4ec..c968abec178 100644 --- a/syft/sync/sync/caches/datasite_owner_cache.py +++ b/syft/sync/sync/caches/datasite_owner_cache.py @@ -18,7 +18,11 @@ ) from syft.sync.utils.syftbox_utils import get_event_hash_from_content from syft.sync.checkpoints.checkpoint import Checkpoint -from syft.sync.sync.constants import CACHE_DIR, OWNER_FILE_HASHES_FILENAME +from syft.sync.sync.constants import ( + CACHE_DIR, + OWNER_FILE_HASHES_FILENAME, + OWNER_PENDING_DELIVERIES_FILENAME, +) class ProposedEventFileOutdatedException(Exception): @@ -58,6 +62,9 @@ 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) email: str # Full path to collections (datasets) folder collections_folder: Path | None = None @@ -78,6 +85,12 @@ def _build_file_hashes(cls, data): key_serializer=str, key_deserializer=Path, ) + if isinstance(data, dict) and "pending_deliveries" 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 + ) return data @classmethod @@ -89,6 +102,7 @@ def from_config(cls, config: DataSiteOwnerEventCacheConfig): email=config.email, syftbox_folder=config.syftbox_folder, file_hashes=PersistedDict(), + pending_deliveries=PersistedDict(), collections_folder=config.collections_folder, collection_variants=config.collection_variants, ) diff --git a/syft/sync/sync/caches/persisted_dict.py b/syft/sync/sync/caches/persisted_dict.py index 5e43ee0fe9c..af22ddf5735 100644 --- a/syft/sync/sync/caches/persisted_dict.py +++ b/syft/sync/sync/caches/persisted_dict.py @@ -252,3 +252,17 @@ def clear(self): with self.exclusive_lock(): super().clear() self._write_to_file() + + def update_entry(self, key: Any, update: Callable[[Any], Any]) -> None: + """Set ``key`` to ``update(old value or None)`` under one exclusive lock. + + A None result deletes the key. + """ + with self.exclusive_lock(): + self._read_from_file() + value = update(super().get(key)) + if value is None: + super().pop(key, None) + else: + super().__setitem__(key, value) + self._write_to_file() diff --git a/syft/sync/sync/constants.py b/syft/sync/sync/constants.py index 2e8508efb4f..8d195bfad55 100644 --- a/syft/sync/sync/constants.py +++ b/syft/sync/sync/constants.py @@ -1,4 +1,5 @@ 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 4ca8204a878..9196bdebd32 100644 --- a/syft/sync/sync/datasite_owner_syncer.py +++ b/syft/sync/sync/datasite_owner_syncer.py @@ -38,6 +38,7 @@ ) from syft_perms import SyftPermContext from syft.sync.sync.constants import CACHE_DIR, ROLLING_STATE_FILENAME +from syft.sync.utils.syftbox_utils import create_event_timestamp logger = logging.getLogger(__name__) @@ -133,6 +134,9 @@ class DatasiteOwnerSyncer(BaseModelCallbackMixin): _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) @@ -191,15 +195,30 @@ 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): + 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``. + + ``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``. + """ 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) if recompute_hashes: - self.process_local_changes(recipients=peer_emails) + self.process_local_changes( + recipients=[*peer_emails, *deferred_peer_emails] + ) # first, pull existing state for peer_email in peer_emails: @@ -219,6 +238,7 @@ def sync(self, peer_emails: list[str], recompute_hashes: bool = True): ) self.process_syftbox_events_queue() finally: + self._deferred_peer_emails = frozenset() if self.persist_owner_state: self._save_rolling_state() @@ -608,6 +628,63 @@ def _create_resend_event(self, path: str) -> "FileChangeEvent | None": 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.""" + event = self._create_resend_event(path) + if event is not None: + return event + timestamp = create_event_timestamp() + return FileChangeEvent( + id=uuid4(), + path_in_datasite=Path(path), + content=None, + submitted_timestamp=timestamp, + timestamp=timestamp, + datasite_email=self.email, + 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 _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 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: @@ -703,7 +780,7 @@ def _queue_events_for_outbox( for recipient, events in events_by_recipient.items(): if events: msg = FileChangeEventsMessage(events=events) - self.outbox_queue.put((recipient, msg)) + self._queue_for_outbox(recipient, msg) def pull_and_process_next_proposed_filechange( self, sender_email: str, raise_on_none=True @@ -787,7 +864,7 @@ def queue_event_for_syftbox( self.syftbox_events_queue.put(file_change_events_message) for recipient in recipients: - self.outbox_queue.put((recipient, file_change_events_message)) + self._queue_for_outbox(recipient, file_change_events_message) def process_syftbox_events_queue(self): # TODO: make this atomic @@ -798,9 +875,12 @@ def process_syftbox_events_queue(self): ) while not self.outbox_queue.empty(): recipient, file_change_events_message = self.outbox_queue.get() + if recipient in self._deferred_peer_emails: + continue self.connection_router.owner_write_event_messages_to_outbox( recipient, file_change_events_message ) + self._clear_delivered(recipient, file_change_events_message) def _download_outbox_message_with_new_connection( self, file_id: str diff --git a/tests/unit/test_persisted_cache.py b/tests/unit/test_persisted_cache.py index 1d6c0dd0764..033e8fd170b 100644 --- a/tests/unit/test_persisted_cache.py +++ b/tests/unit/test_persisted_cache.py @@ -3,6 +3,9 @@ If the cache persistence feature is removed or broken, these tests fail. """ +import pytest + +from syft.sync.connections.connection_router import ConnectionRouter from syft.sync.syftbox_manager import SyftboxManager from syft.sync.sync.constants import ( CACHE_DIR, @@ -115,3 +118,83 @@ def test_rolling_state_not_overwritten_across_processes(): "file2.txt lost! B overwrote A's rolling state instead of building on it." ) assert "file3.txt" in paths_in_rs, "file3.txt missing from rolling state." + + +def _pair_with_unsent_do_file(): + """Pair a DO and a DS that share a disk cache, and write one unsent DO file.""" + ds_manager, do_manager = SyftboxManager.pair_with_mock_drive_service_connection( + use_in_memory_cache=False, check_versions=True + ) + do_manager.datasite_owner_syncer.perm_context.open(".").grant_read_access( + ds_manager.email + ) + do_manager.sync(auto_checkpoint=False) + api_file = do_manager.syftbox_folder / do_manager.email / "app_data/apis/x/a.py" + api_file.parent.mkdir(parents=True, exist_ok=True) + api_file.write_text("print('api')") + return ds_manager, do_manager + + +def _assert_ds_receives_after_notebook_sync(ds_manager, do_manager): + do_manager.sync(auto_checkpoint=False) + ds_manager.sync() + received = ds_manager.syftbox_folder / do_manager.email / "app_data/apis/x/a.py" + assert received.exists(), "The DS never received the DO file." + + +def test_peer_gets_file_when_other_process_cannot_read_peer_version(monkeypatch): + """A process that skips a peer must not mark the file as sent to that peer.""" + ds_manager, do_manager = _pair_with_unsent_do_file() + + # Process B (syft-bg) cannot read the peer version file, so it skips the DS. + do_manager_b = do_manager._copy() + monkeypatch.setattr( + do_manager_b.peer_manager, + "_load_single_peer_version", + lambda email: (email, None), + ) + do_manager_b.sync(auto_checkpoint=False) + + _assert_ds_receives_after_notebook_sync(ds_manager, do_manager) + + +def test_peer_gets_file_when_other_process_fails_outbox_upload(monkeypatch): + """A failed outbox upload must not mark the file as sent.""" + ds_manager, do_manager = _pair_with_unsent_do_file() + + do_manager_b = do_manager._copy() + + def fail_upload(*args, **kwargs): + raise ConnectionError("upload failed") + + # 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) + + _assert_ds_receives_after_notebook_sync(ds_manager, do_manager) + + +def test_peer_loses_file_deleted_while_other_process_skipped_it(monkeypatch): + """A deletion that a process defers for a skipped peer reaches the peer later.""" + ds_manager, do_manager = _pair_with_unsent_do_file() + do_manager.sync(auto_checkpoint=False) + ds_manager.sync() + received = ds_manager.syftbox_folder / do_manager.email / "app_data/apis/x/a.py" + assert received.exists() + + # Process B (syft-bg) is already running when the DO deletes the file. + do_manager_b = do_manager._copy() + monkeypatch.setattr( + do_manager_b.peer_manager, + "_load_single_peer_version", + lambda email: (email, None), + ) + do_manager_b.sync(auto_checkpoint=False) + (do_manager.syftbox_folder / do_manager.email / "app_data/apis/x/a.py").unlink() + do_manager_b.sync(auto_checkpoint=False) + + do_manager.sync(auto_checkpoint=False) + ds_manager.sync() + assert not received.exists(), "The DS kept a file that the DO deleted." From eac180f10d82f824a238e4ca28d384ec843de2aa Mon Sep 17 00:00:00 2001 From: Pedro Werneck Date: Thu, 15 Oct 2026 10:41:53 -0300 Subject: [PATCH 2/2] fix: raise on a read error instead of sending a deletion --- syft/sync/sync/datasite_owner_syncer.py | 21 +++++++++----- tests/unit/test_persisted_cache.py | 38 +++++++++++++++++++++++++ 2 files changed, 52 insertions(+), 7 deletions(-) diff --git a/syft/sync/sync/datasite_owner_syncer.py b/syft/sync/sync/datasite_owner_syncer.py index 9196bdebd32..f3cc88eef4d 100644 --- a/syft/sync/sync/datasite_owner_syncer.py +++ b/syft/sync/sync/datasite_owner_syncer.py @@ -38,7 +38,10 @@ ) from syft_perms import SyftPermContext from syft.sync.sync.constants import CACHE_DIR, ROLLING_STATE_FILENAME -from syft.sync.utils.syftbox_utils import create_event_timestamp +from syft.sync.utils.syftbox_utils import ( + create_event_timestamp, + get_event_hash_from_content, +) logger = logging.getLogger(__name__) @@ -629,19 +632,23 @@ def _create_resend_event(self, path: str) -> "FileChangeEvent | None": ) def _create_current_state_event(self, path: str) -> FileChangeEvent: - """An event that sets ``path`` on a peer to its current local state.""" - event = self._create_resend_event(path) - if event is not None: - return event + """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) timestamp = create_event_timestamp() return FileChangeEvent( id=uuid4(), path_in_datasite=Path(path), - content=None, + content=content, + new_hash=None if content is None else get_event_hash_from_content(content), submitted_timestamp=timestamp, timestamp=timestamp, datasite_email=self.email, - is_deleted=True, + is_deleted=content is None, ) def _queue_pending_deliveries(self, peer_emails: list[str]) -> None: diff --git a/tests/unit/test_persisted_cache.py b/tests/unit/test_persisted_cache.py index 033e8fd170b..ab1b31658b6 100644 --- a/tests/unit/test_persisted_cache.py +++ b/tests/unit/test_persisted_cache.py @@ -6,6 +6,7 @@ import pytest from syft.sync.connections.connection_router import ConnectionRouter +from syft.sync.sync.caches.cache_file_writer_connection import FSFileConnection from syft.sync.syftbox_manager import SyftboxManager from syft.sync.sync.constants import ( CACHE_DIR, @@ -198,3 +199,40 @@ def test_peer_loses_file_deleted_while_other_process_skipped_it(monkeypatch): do_manager.sync(auto_checkpoint=False) ds_manager.sync() 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.""" + ds_manager, do_manager = _pair_with_unsent_do_file() + do_manager.sync(auto_checkpoint=False) + ds_manager.sync() + received = ds_manager.syftbox_folder / do_manager.email / "app_data/apis/x/a.py" + assert received.exists() + + do_manager_b = do_manager._copy() + monkeypatch.setattr( + do_manager_b.peer_manager, + "_load_single_peer_version", + lambda email: (email, None), + ) + do_manager_b.sync(auto_checkpoint=False) + api_file = do_manager.syftbox_folder / do_manager.email / "app_data/apis/x/a.py" + api_file.write_text("print('api v2')") + do_manager_b.sync(auto_checkpoint=False) + + # The first read is the resend of the pending path. + read_file = FSFileConnection.read_file + calls = [] + + def fail_first_read(self, path): + calls.append(path) + if len(calls) == 1: + raise PermissionError(path) + return read_file(self, path) + + with monkeypatch.context() as m: + m.setattr(FSFileConnection, "read_file", fail_first_read) + with pytest.raises(PermissionError): + do_manager.sync(auto_checkpoint=False) + ds_manager.sync() + assert received.exists(), "A read error reached the DS as a deletion."