fix: stop DO changes from being lost when a peer misses a sync

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(

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

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

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"

syft/sync/sync/datasite_owner_syncer.py

@@ -38,6 +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,
+    get_event_hash_from_content,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -133,6 +137,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 +198,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 +241,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 +631,67 @@ 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.
+
+        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=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=content is None,
+        )
+
+    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 +787,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 +871,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 +882,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

tests/unit/test_persisted_cache.py

@@ -3,6 +3,10 @@
 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.sync.caches.cache_file_writer_connection import FSFileConnection
 from syft.sync.syftbox_manager import SyftboxManager
 from syft.sync.sync.constants import (
     CACHE_DIR,
@@ -115,3 +119,120 @@ 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."
+
+
+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."