diff --git a/syft/sync/connections/drive/drive_p2p.py b/syft/sync/connections/drive/drive_p2p.py index 825d5e2d4ac..9055c4f2a78 100644 --- a/syft/sync/connections/drive/drive_p2p.py +++ b/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.""" diff --git a/syft/sync/connections/protocols.py b/syft/sync/connections/protocols.py index 6b744386c54..f833cbeda43 100644 --- a/syft/sync/connections/protocols.py +++ b/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 diff --git a/syft/sync/peers/peer_store.py b/syft/sync/peers/peer_store.py index 3ffac197b9d..411d081d413 100644 --- a/syft/sync/peers/peer_store.py +++ b/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 diff --git a/syft/sync/syftbox_manager.py b/syft/sync/syftbox_manager.py index fbe7d2ce2a4..9a36f0d6363 100644 --- a/syft/sync/syftbox_manager.py +++ b/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. diff --git a/syft/sync/sync/datasite_watcher_syncer.py b/syft/sync/sync/datasite_watcher_syncer.py index 5ff8df81435..1d181e32942 100644 --- a/syft/sync/sync/datasite_watcher_syncer.py +++ b/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) ), ) diff --git a/syft/sync/sync/spool.py b/syft/sync/sync/spool.py new file mode 100644 index 00000000000..21c3ecd2438 --- /dev/null +++ b/syft/sync/sync/spool.py @@ -0,0 +1,108 @@ +"""The messages a DS saves before it sends them. + +A dropped connection or a stopped kernel must not lose a job submission. Each +message to a peer is saved to ``//private/spool//`` before +any network call, and it is removed when the peer has it. ``private/`` is never +synced. + +A saved message is the message before the downgrade and the encryption for its +peer. The upload does both, so it uses the protocol and the key that the peer +has at that time. An enclave gets a new key at every boot. +""" + +from contextlib import AbstractContextManager +from pathlib import Path + +from pydantic import BaseModel, ConfigDict + +from syft.sync.messages.proposed_filechange import ProposedFileChangesMessage +from syft.sync.peers.peer_store import PRIVATE_DIR_NAME +from syft.sync.sync.caches.persisted_dict import locked_file, write_text_atomic +from syft.sync.utils.versioned_file import read_versioned_json + +SPOOL_DIR_NAME = "spool" +SPOOL_ENTRY_SUFFIX = ".json" +# Format of a spool entry. Raise it when the layout of an entry changes. An +# entry of a later version is refused. +SPOOL_ENTRY_VERSION = 1 + + +def datasite_spool_path(syftbox_folder: Path | str, email: str) -> Path: + """Per-datasite spool: ``//private/spool``.""" + return Path(syftbox_folder) / email / PRIVATE_DIR_NAME / SPOOL_DIR_NAME + + +class SpoolEntry(BaseModel): + """One saved message.""" + + model_config = ConfigDict( + frozen=True, + extra="forbid", + ser_json_bytes="base64", + val_json_bytes="base64", + ) + + version: int + # The compressed message, before the downgrade and the encryption for its peer. + compressed_message: bytes + + @classmethod + def read(cls, path: Path) -> "SpoolEntry": + """Read the entry at ``path``. + + Raises: + ValueError: if a later client wrote the entry, or it has another + shape. + """ + return read_versioned_json(path, cls, SPOOL_ENTRY_VERSION, "spool entry") + + +class Spool: + """The saved messages of one DS, in one folder per recipient. An entry is + named after the ``msg_id`` its message gets.""" + + def __init__(self, folder: Path): + self.folder = folder + + def add(self, recipient: str, message: ProposedFileChangesMessage) -> None: + """Save ``message`` to ``recipient``.""" + entry = SpoolEntry( + version=SPOOL_ENTRY_VERSION, compressed_message=message.as_compressed_data() + ) + msg_id = message.message_filename.as_string() + write_text_atomic(self._entry_path(recipient, msg_id), entry.model_dump_json()) + + def recipients(self) -> list[str]: + """The recipients that have a folder in the spool.""" + if not self.folder.is_dir(): + return [] + return sorted(p.name for p in self.folder.iterdir() if p.is_dir()) + + def pending_messages(self, recipient: str) -> list[ProposedFileChangesMessage]: + """The saved messages to ``recipient``, oldest first. + + Raises: + ValueError: if an entry is of a later version, or has another shape. + """ + paths = self.recipient_folder(recipient).glob(f"*{SPOOL_ENTRY_SUFFIX}") + messages = [ + ProposedFileChangesMessage.from_compressed_data( + SpoolEntry.read(p).compressed_message + ) + for p in paths + ] + return sorted(messages, key=lambda m: m.message_filename.submitted_timestamp) + + def remove(self, recipient: str, msg_id: str) -> None: + """Remove the saved message ``msg_id`` to ``recipient``.""" + self._entry_path(recipient, msg_id).unlink() + + def locked(self) -> AbstractContextManager[None]: + """A lock that one process at a time holds while it uploads.""" + return locked_file(self.folder) + + def recipient_folder(self, recipient: str) -> Path: + return self.folder / recipient + + def _entry_path(self, recipient: str, msg_id: str) -> Path: + return self.recipient_folder(recipient) / f"{msg_id}{SPOOL_ENTRY_SUFFIX}" diff --git a/syft/sync/utils/versioned_file.py b/syft/sync/utils/versioned_file.py new file mode 100644 index 00000000000..65c803221f6 --- /dev/null +++ b/syft/sync/utils/versioned_file.py @@ -0,0 +1,30 @@ +"""Read a JSON file that carries a format version.""" + +import json +from pathlib import Path +from typing import TypeVar + +from pydantic import BaseModel + +ModelT = TypeVar("ModelT", bound=BaseModel) + + +def read_versioned_json( + path: Path, model: type[ModelT], max_version: int, kind: str +) -> ModelT: + """The file at ``path`` as ``model``. ``kind`` names the file in errors. + + The version is checked before the fields, so a file of a later layout is + refused by its version, not by a field error. + + Raises: + ValueError: if a later client wrote the file, or it has another shape. + """ + text = path.read_text() + version = json.loads(text)["version"] + if version > max_version: + raise ValueError( + f"The {kind} at {path} has version {version}, and this client " + f"reads up to version {max_version}." + ) + return model.model_validate_json(text) diff --git a/tests/unit/test_encryption_policy.py b/tests/unit/test_encryption_policy.py index 2d3aa2c4fdd..09a15a30e73 100644 --- a/tests/unit/test_encryption_policy.py +++ b/tests/unit/test_encryption_policy.py @@ -491,8 +491,9 @@ def _queued_changes(manager) -> int: @pytest.mark.parametrize("do_policy", [OPTIONAL, REQUIRED]) -def test_send_before_owner_accepts_raises(do_policy): - """Before the owner accepts, its policy is not known, so nothing is sent.""" +def test_send_before_owner_accepts_goes_once_accepted(do_policy): + """Before the owner accepts, its policy is not known. The change is saved, + and the first sync after the owner accepts sends it.""" ds_manager, do_manager = SyftboxManager.pair_with_mock_drive_service_connection( add_peers=False, do_encryption_policy=do_policy, @@ -500,15 +501,18 @@ def test_send_before_owner_accepts_raises(do_policy): ds_manager.add_peer(do_manager.email, sync=False) file_path = path_for_job(do_manager.email, ds_manager.email, "my.job") - with pytest.raises(SyftPeerNotReadyError, match="not known"): + with pytest.raises(SyftPeerNotReadyError, match="not known.*is saved"): ds_manager._send_file_change(file_path, "early content") assert _queued_changes(ds_manager) == 0 + with pytest.warns(UserWarning, match="wait in"): + ds_manager.sync() do_manager.load_peers() do_manager.approve_peer_request(ds_manager.email) grant_job_inbox_access(do_manager, ds_manager.email) + ds_manager.sync() do_manager.sync() - assert not any("early content" in c for c in _accepted_contents(do_manager)) + assert any("early content" in c for c in _accepted_contents(do_manager)) def test_send_to_non_peer_raises(): diff --git a/tests/unit/test_spool.py b/tests/unit/test_spool.py new file mode 100644 index 00000000000..54d2e4be3ef --- /dev/null +++ b/tests/unit/test_spool.py @@ -0,0 +1,167 @@ +"""The DS saves each message before it sends it, so a dropped connection or a +stopped kernel does not lose a submission.""" + +import json + +import httplib2 +import pytest + +from syft.sync.connections.drive.drive_p2p import DriveP2PConnection +from syft.sync.connections.exceptions import ConnectionUnavailableError +from syft.sync.connections.types import MessageKind +from syft.sync.peers.peer import PeerState +from syft.sync.syftbox_manager import SyftboxManager, SyftboxManagerConfig +from syft.sync.sync.spool import SPOOL_ENTRY_VERSION, Spool +from tests.unit.test_persisted_cache import _pair_with_unsent_do_file +from tests.unit.test_sync_manager import path_for_job +from tests.unit.utils import grant_job_inbox_access, mock_message + + +class _Stopped(Exception): + """The process stopped.""" + + +def _drop_connection(monkeypatch) -> None: + def fail(self, recipient, filename, data): + raise httplib2.ServerNotFoundError("Unable to find the server") + + monkeypatch.setattr(DriveP2PConnection, "watcher_send_raw_bytes_to_inbox", fail) + + +def _restart(ds_manager) -> SyftboxManager: + """A new DS manager on the folder and the Drive of ``ds_manager``.""" + config = SyftboxManagerConfig._base_config_for_testing( + email=ds_manager.email, + syftbox_folder=ds_manager.syftbox_folder, + has_ds_role=True, + ) + return SyftboxManager.from_config(config, ds_manager._connection_router.connections) + + +def _accepted_contents(do_manager) -> list[str]: + return [ + str(e.content) + for msg in do_manager._get_all_accepted_events_do() + for e in msg.events + ] + + +def _pair_with_job_inbox(encryption: bool = False): + ds_manager, do_manager = SyftboxManager.pair_with_mock_drive_service_connection( + encryption=encryption + ) + grant_job_inbox_access(do_manager, ds_manager.email) + return ds_manager, do_manager, path_for_job(do_manager.email, ds_manager.email) + + +def _save_later_version_entry(spool: Spool, recipient: str) -> int: + """Save a message to ``recipient`` as a later client writes it, and return + that later version.""" + spool.add(recipient, mock_message()) + entry_path = next(spool.recipient_folder(recipient).iterdir()) + later_version = SPOOL_ENTRY_VERSION + 1 + data = json.loads(entry_path.read_text()) + data.update(version=later_version, new_setting=True) + entry_path.write_text(json.dumps(data)) + return later_version + + +def test_failed_upload_sent_by_sync_after_restart(monkeypatch): + ds_manager, do_manager, file_path = _pair_with_job_inbox() + with monkeypatch.context() as m: + _drop_connection(m) + with pytest.raises(ConnectionUnavailableError, match="is saved"): + ds_manager._send_file_change(file_path, "job") + + _restart(ds_manager).sync() + do_manager.sync() + assert "job" in _accepted_contents(do_manager) + + +def test_crash_after_upload_sends_once(monkeypatch): + ds_manager, do_manager, file_path = _pair_with_job_inbox() + + def stop(self, recipient, msg_id): + raise _Stopped() + + with monkeypatch.context() as m: + m.setattr(Spool, "remove", stop) + with pytest.raises(_Stopped): + ds_manager._send_file_change(file_path, "job") + # The DO applies the message and moves it to its archive. + do_manager.sync() + + _restart(ds_manager).sync() + inbox = do_manager._connection_router.connections.p2p_con.receive( + ds_manager.email, MessageKind.PROPOSED_CHANGES + ) + assert inbox == [] + assert _accepted_contents(do_manager).count("job") == 1 + + +def test_edits_saved_offline_apply_in_order(monkeypatch): + ds_manager, do_manager, file_path = _pair_with_job_inbox() + with monkeypatch.context() as m: + _drop_connection(m) + for content in ["first", "second"]: + with pytest.raises(ConnectionUnavailableError): + ds_manager._send_file_change(file_path, content) + + ds_manager.sync() + do_manager.sync() + assert "second" in _accepted_contents(do_manager) + + +def test_edit_after_restart_builds_on_saved_edit(monkeypatch): + ds_manager, do_manager, file_path = _pair_with_job_inbox() + with monkeypatch.context() as m: + _drop_connection(m) + with pytest.raises(ConnectionUnavailableError): + ds_manager._send_file_change(file_path, "first") + + restarted = _restart(ds_manager) + restarted.load_peers() + restarted.datasite_watcher_syncer.on_file_change(file_path, "second") + do_manager.sync() + assert "second" in _accepted_contents(do_manager) + + +def test_saved_message_encrypted_with_peer_key_at_upload(monkeypatch): + ds_manager, do_manager, file_path = _pair_with_job_inbox(encryption=True) + ds_manager.load_peers() + with monkeypatch.context() as m: + _drop_connection(m) + with pytest.raises(ConnectionUnavailableError): + ds_manager._send_file_change(file_path, "job") + + # The DO boots with a new key, and the DS learns it, as for an enclave. + do_manager._peer_store.generate_keys() + new_bundle = do_manager._peer_store.get_public_bundle() + ds_manager._peer_store.set_peer_bundle(do_manager.email, new_bundle) + ds_manager._connection_router.update_peer_state( + do_manager.email, PeerState.ACCEPTED.value, public_encryption_bundle=new_bundle + ) + + ds_manager.sync() + do_manager.sync() + assert "job" in _accepted_contents(do_manager) + + +def test_spool_entry_of_later_version_refused(tmp_path): + spool = Spool(tmp_path) + later_version = _save_later_version_entry(spool, "do@example.com") + + with pytest.raises(ValueError, match=f"version {later_version}"): + spool.pending_messages("do@example.com") + + +def test_stuck_message_does_not_block_pull(): + ds_manager, do_manager = _pair_with_unsent_do_file() + spool = ds_manager.datasite_watcher_syncer.spool + _save_later_version_entry(spool, do_manager.email) + do_manager.sync(auto_checkpoint=False) + + with pytest.raises(ValueError, match="version"): + ds_manager.sync() + received = ds_manager.syftbox_folder / do_manager.email / "app_data/apis/x/a.py" + assert received.exists()