feat: save DS messages before upload and send them at the next sync

syft/sync/connections/drive/drive_p2p.py

@@ -97,12 +97,16 @@ def reset_caches(self) -> None:
     def send(
         self, peer: str, msg_id: str, data: bytes, kind: MessageKind
     ) -> SendStatus:
-        if kind is MessageKind.PROPOSED_CHANGES:
-            self.watcher_send_raw_bytes_to_inbox(peer, msg_id, data)
-        elif kind is MessageKind.FILE_EVENTS:
-            self.owner_write_raw_bytes_to_outbox(peer, msg_id, data)
-        else:
-            raise ValueError(f"Drive has no folder for messages of kind {kind!r}")
+        """A proposal that an earlier send uploaded is not uploaded again. File
+        events are not checked: the DS skips an outbox message it received."""
+        with raise_unavailable(f"Could not send message {msg_id} to Drive."):
+            if kind is MessageKind.PROPOSED_CHANGES:
+                if not self._is_uploaded(msg_id):
+                    self.watcher_send_raw_bytes_to_inbox(peer, msg_id, data)
+            elif kind is MessageKind.FILE_EVENTS:
+                self.owner_write_raw_bytes_to_outbox(peer, msg_id, data)
+            else:
+                raise ValueError(f"Drive has no folder for messages of kind {kind!r}")
         return SendStatus.DELIVERED
 
     def delivered(self) -> list[str]:
@@ -220,6 +224,18 @@ def delete_messages(self, handles: list[str]) -> None:
 
     # -- Drive -----------------------------------------------------------------
 
+    def _is_uploaded(self, msg_id: str) -> bool:
+        """Whether I uploaded a message named ``msg_id``."""
+        # I own the file I upload, also after the peer moves it to its archive,
+        # so a search of my own files by name finds it in any folder.
+        query = f"name='{msg_id}' and 'me' in owners and trashed=false"
+        results = execute_with_retries(
+            self._cached_client.client.drive_service.files().list(
+                q=query, fields="files(id)", pageSize=1
+            )
+        )
+        return bool(results["files"])
+
     def _list_inbox(self, peer: str) -> list[dict]:
         """The messages in the inbox from ``peer``, oldest first, each with
         ``file_id`` and ``file_name``, without downloading them."""

syft/sync/connections/protocols.py

@@ -66,8 +66,13 @@ def send(
         """Send ``data`` to ``peer`` under ``msg_id``.
 
         ``msg_id`` is chosen by the sender and stays the same if the message is
-        sent again. ``kind`` lets a connection keep the two kinds apart, as
-        Drive does with an inbox and an outbox.
+        sent again, as after a stop. The peer then applies it once: the
+        connection drops the second copy, or the peer skips it. ``kind`` lets a
+        connection keep the two kinds apart, as Drive does with an inbox and an
+        outbox.
+
+        Raises:
+            ConnectionUnavailableError: when the backend cannot be reached.
         """
 
     @abstractmethod

syft/sync/peers/peer_store.py

@@ -4,7 +4,6 @@
 Shared between PeerManager and all ConnectionRouter instances.
 """
 
-import json
 import os
 import tempfile
 from pathlib import Path
@@ -21,6 +20,7 @@
 )
 from syft.sync.peers.exceptions import SyftPeerNotReadyError
 from syft.sync.peers.peer import Peer, PeerState
+from syft.sync.utils.versioned_file import read_versioned_json
 
 # Encryption key bundles persist inside the participant's own SyftBox datasite
 # folder, under private/ (which is never synced to Drive). This scopes keys per
@@ -96,14 +96,9 @@ def read(cls, path: Path) -> "_DatasiteSettings":
             ValueError: if a later client wrote the file, or it has another
                 shape.
         """
-        data = json.loads(path.read_text())
-        # Checked before the fields, so a later layout is refused by its version.
-        if data["version"] > DATASITE_SETTINGS_VERSION:
-            raise ValueError(
-                f"The settings file at {path} has version {data['version']}, and "
-                f"this client reads up to version {DATASITE_SETTINGS_VERSION}."
-            )
-        return cls.model_validate(data)
+        return read_versioned_json(
+            path, cls, DATASITE_SETTINGS_VERSION, "settings file"
+        )
 
 
 def resolve_encryption_policy(
@@ -264,8 +259,7 @@ def _peer_for_message(self, email: str) -> Peer:
                 email,
                 cause=f"The encryption policy of {email} is not known yet.",
                 remedy=(
-                    "Peer with them and wait until they accept. Then run "
-                    "client.sync() and send again."
+                    "Peer with them and wait until they accept. Then run client.sync()."
                 ),
             )
         return peer

syft/sync/syftbox_manager.py

@@ -1124,6 +1124,8 @@ def sync(
                 peer_emails = [peer.email for peer in self.peer_manager.syncable_peers]
                 self.peer_manager.warn_if_all_peers_incompatible(peer_emails)
                 self.datasite_watcher_syncer.sync_down(peer_emails)
+                # After the pull, so a saved message that cannot go never holds back results.
+                self.datasite_watcher_syncer.sync_up()
 
     def load_peers(self, force_download: bool = False):
         """Load peers from connection router. Delegates to PeerManager.

syft/sync/sync/datasite_watcher_syncer.py

@@ -1,26 +1,29 @@
-from pathlib import Path
+import warnings
 from concurrent.futures import ThreadPoolExecutor
+from pathlib import Path
 from queue import Queue
-from typing import Dict, List
+from typing import Dict, Iterable, List
 
 from pydantic import BaseModel, Field, PrivateAttr
 
-from syft.sync.connections.connection_router import ConnectionRouter
 from syft.sync.callback_mixin import BaseModelCallbackMixin
-from syft.sync.peers.exceptions import SyftPeerNotReadyError
+from syft.sync.connections.connection_router import ConnectionRouter
+from syft.sync.connections.exceptions import ConnectionUnavailableError
 from syft.sync.messages.proposed_filechange import (
     ProposedFileChange,
     ProposedFileChangesMessage,
 )
+from syft.sync.peers.exceptions import SyftPeerNotReadyError
 from syft.sync.sync.caches.datasite_watcher_cache import (
     DataSiteWatcherCache,
     DataSiteWatcherCacheConfig,
 )
+from syft.sync.sync.spool import Spool, datasite_spool_path
 
 
 class DatasiteWatcherSyncerConfig(BaseModel):
-    syftbox_folder: Path | None = None
-    email: str | None = None
+    syftbox_folder: Path
+    email: str
     datasite_watcher_cache_config: DataSiteWatcherCacheConfig = Field(
         default_factory=DataSiteWatcherCacheConfig
     )
@@ -36,6 +39,7 @@ class Config:
     email: str
     connection_router: ConnectionRouter
     datasite_watcher_cache: DataSiteWatcherCache
+    spool: Spool
     queue: Queue = Field(default_factory=Queue)
 
     _executor: ThreadPoolExecutor = PrivateAttr(
@@ -50,17 +54,27 @@ def from_config(
         shared with the other parts of the manager, so the caller builds it
         once and passes it in next to the config (see
         ``SyftboxManager.from_config``)."""
-        return cls(
+        syncer = cls(
             syftbox_folder=config.syftbox_folder,
             email=config.email,
             connection_router=connection_router,
             datasite_watcher_cache=DataSiteWatcherCache.from_config(
                 config.datasite_watcher_cache_config, connection_router
             ),
+            spool=Spool(datasite_spool_path(config.syftbox_folder, config.email)),
         )
+        syncer._record_spooled_hashes()
+        return syncer
 
     # --- Pushing proposed file changes ---
 
+    def _record_spooled_hashes(self) -> None:
+        """Record the hashes of the saved messages, so that a change after a
+        restart builds on a saved change that is not sent yet."""
+        for recipient in self.spool.recipients():
+            for message in self.spool.pending_messages(recipient):
+                self._record_proposed_hashes(message)
+
     def get_proposed_file_change_object(
         self,
         relative_path: Path,
@@ -79,6 +93,17 @@ def get_proposed_file_change_object(
         )
 
     def process_file_changes_queue(self):
+        """Save the queued changes as one message per datasite, then upload the
+        saved messages to those datasites. A message that an upload error stops
+        stays saved, and so do the messages after it.
+
+        Raises:
+            SyftPeerNotReadyError: when a recipient cannot receive yet.
+                ``sync_up`` sends its message later. The other recipients still
+                get their messages.
+            ConnectionUnavailableError: when the backend cannot be reached.
+                ``sync_up`` sends the messages later.
+        """
         # One message per datasite: each change goes to the owner of its path.
         changes_by_recipient: Dict[str, List[ProposedFileChange]] = {}
         while not self.queue.empty():
@@ -108,34 +133,29 @@ def process_file_changes_queue(self):
                 path_in_datasite, content, datasite_email=recipient
             )
             changes_by_recipient.setdefault(recipient, []).append(file_change)
-        self._send_by_recipient(changes_by_recipient)
-
-    def _send_by_recipient(
-        self, changes_by_recipient: dict[str, list[ProposedFileChange]]
-    ) -> None:
-        """Send one message to each recipient.
-
-        The changes for a recipient that cannot receive now are dropped. The
-        other recipients still get their changes, and then the first error is
-        raised.
-
-        Raises:
-            SyftPeerNotReadyError: when the policy of a recipient is not known,
-                or the pair encrypts and the peer bundle is missing.
-        """
-        error = None
         for recipient, changes in changes_by_recipient.items():
             message = ProposedFileChangesMessage(
                 sender_email=self.email, proposed_file_changes=changes
             )
-            try:
-                self._send_message(recipient, message)
-            except SyftPeerNotReadyError as e:
-                error = error or e
-        if error is not None:
-            raise error
-
-    def _record_sent_hashes(self, message: ProposedFileChangesMessage) -> None:
+            self.spool.add(recipient, message)
+            # The next change to a path builds on this one, sent or not.
+            self._record_proposed_hashes(message)
+        try:
+            not_ready_errors = self._upload_spooled(changes_by_recipient)
+        except ConnectionUnavailableError as e:
+            raise ConnectionUnavailableError(
+                f"{e} The change is saved, and client.sync() sends it."
+            ) from e
+        if not_ready_errors:
+            error = next(iter(not_ready_errors.values()))
+            raise SyftPeerNotReadyError(
+                error.peer_email,
+                cause=error.cause,
+                remedy=f"{error.remedy} The change is saved, and client.sync() "
+                "sends it once the peer is ready.",
+            ) from error
+
+    def _record_proposed_hashes(self, message: ProposedFileChangesMessage) -> None:
         hashes = self.datasite_watcher_cache.file_hashes
         for fc in message.proposed_file_changes:
             path_key = Path(fc.datasite_email) / fc.path_in_datasite
@@ -144,19 +164,47 @@ def _record_sent_hashes(self, message: ProposedFileChangesMessage) -> None:
             else:
                 hashes[path_key] = fc.new_hash
 
-    def _send_message(
-        self, recipient: str, message: ProposedFileChangesMessage
-    ) -> None:
-        """Send ``message`` and record the hashes it sent.
+    def _upload_spooled(
+        self, recipients: Iterable[str]
+    ) -> dict[str, SyftPeerNotReadyError]:
+        """Upload the saved messages to each of ``recipients``, oldest first,
+        and remove each message once its recipient has it.
+
+        Returns why the messages to each recipient that cannot receive yet
+        wait, by recipient email. They stay saved, in order.
 
         Raises:
-            SyftPeerNotReadyError: when the policy of ``recipient`` is not
-                known, or the pair encrypts and the peer bundle is missing.
+            ConnectionUnavailableError: when the backend cannot be reached. An
+                upload that fails for another reason raises too. That message
+                and the later ones stay saved.
         """
-        self.connection_router.watcher_send_proposed_file_changes_message(
-            recipient, message
-        )
-        self._record_sent_hashes(message)
+        not_ready_errors = {}
+        with self.spool.locked():
+            for recipient in recipients:
+                try:
+                    self._upload_spooled_to(recipient)
+                except SyftPeerNotReadyError as e:
+                    not_ready_errors[recipient] = e
+        return not_ready_errors
+
+    def _upload_spooled_to(self, recipient: str) -> None:
+        for message in self.spool.pending_messages(recipient):
+            msg_id = message.message_filename.as_string()
+            # The send drops a message that an upload stored before a stop.
+            self.connection_router.watcher_send_proposed_file_changes_message(
+                recipient, message
+            )
+            self.spool.remove(recipient, msg_id)
+
+    def sync_up(self) -> None:
+        """Upload every saved message. Warn for each recipient that cannot
+        receive yet; its messages wait for the next sync."""
+        not_ready_errors = self._upload_spooled(self.spool.recipients())
+        for recipient, error in not_ready_errors.items():
+            warnings.warn(
+                f"Changes to {recipient} wait in "
+                f"{self.spool.recipient_folder(recipient)}: {error}"
+            )
 
     def on_file_change(
         self,
@@ -193,8 +241,7 @@ def sync_down(self, peer_emails: list[str]):
             self.datasite_watcher_cache.sync_down_collections_parallel(
                 peer_email,
                 self._executor,
-                lambda fid,
-                pe=peer_email: self.download_collection_file_with_new_connection(
-                    fid, pe
+                lambda fid, pe=peer_email: (
+                    self.download_collection_file_with_new_connection(fid, pe)
                 ),
             )