feat: delivery tracker - deliver owner changes from per-peer views

docs/connections.md

@@ -61,7 +61,7 @@ A pair encrypts when one side is `required`.
 
 A SyftBox has one policy, kept in `<syftbox>/<email>/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.
 

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."""

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

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": [

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.

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:

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:

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:

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

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

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"