diff --git a/syft/sync/connections/connection_router.py b/syft/sync/connections/connection_router.py
index a9fafbc0383..38fe2da1feb 100644
--- a/syft/sync/connections/connection_router.py
+++ b/syft/sync/connections/connection_router.py
@@ -1,5 +1,6 @@
 import logging
 import warnings
+from collections.abc import Container
 from typing import TYPE_CHECKING, List, Optional
 
 from pydantic import BaseModel
@@ -13,9 +14,6 @@
 from syft.sync.checkpoints.checkpoint import Checkpoint, IncrementalCheckpoint
 from syft.sync.checkpoints.rolling_state import RollingState
 from syft.sync.connections.base_connection import FileCollection
-from syft.sync.encryption_policy import (
-    MessageFormatError,
-)
 from syft.sync.events.file_change_event import (
     FileChangeEventsMessage,
 )
@@ -109,7 +107,7 @@ def owner_receive_proposed_messages(
     ) -> list[tuple[RawMessage, ProposedFileChangesMessage]]:
         """Every readable proposal waiting from ``sender_email``, oldest first,
         as received and as read. Pass the received ones to
-        ``owner_mark_all_processed`` once they are handled.
+        ``mark_all_processed`` once they are handled.
 
         A message that cannot be read (wrong format for the pair, bad
         signature, cannot be decrypted or decoded) is marked processed unread,
@@ -145,7 +143,8 @@ def _read_proposed_message(
             self.connections.p2p_con.mark_processed(raw)
             return None
 
-    def owner_mark_all_processed(self, messages: list[RawMessage]) -> None:
+    def mark_all_processed(self, messages: list[RawMessage]) -> None:
+        """Finish ``messages``, as received by either role."""
         self.connections.p2p_con.mark_all_processed(messages)
 
     def owner_write_event_messages_to_outbox(
@@ -159,43 +158,56 @@ def owner_write_event_messages_to_outbox(
             recipient_email, fname, data, MessageKind.FILE_EVENTS
         )
 
-    def watcher_get_events_messages(
-        self, peer_email: str, since_timestamp: float | None
-    ) -> List[FileChangeEventsMessage]:
-        """The outbox messages of ``peer_email``, without the skipped ones."""
-        raw_list = self.connections.p2p_con.outbox_messages(peer_email, since_timestamp)
-        messages = [
-            self.watcher_read_events_message(peer_email, raw) for raw in raw_list
-        ]
-        return [m for m in messages if m is not None]
+    def watcher_receive_events_messages(
+        self,
+        sender_email: str,
+        after: str | None = None,
+        already_received: Container[str] = frozenset(),
+    ) -> list[tuple[RawMessage, FileChangeEventsMessage | None]]:
+        """The events messages from ``sender_email`` after the message
+        ``after``, and any that landed after it, without those in
+        ``already_received``, oldest first, as received and as read. Pass the
+        received ones to ``mark_all_processed`` once they are applied.
+
+        A message that can never be read (wrong format for the pair, bad
+        signature, cannot be decrypted or decoded) comes with None and a
+        warning, so the caller moves past it. The list ends, with a warning,
+        at a message that cannot be read yet because the peer is not set up;
+        it comes again at the next receive.
+        """
+        received = self.connections.p2p_con.receive(
+            sender_email,
+            MessageKind.FILE_EVENTS,
+            after=after,
+            already_received=already_received,
+        )
+        read = []
+        for raw in received:
+            try:
+                read.append((raw, self._read_events_message(raw)))
+            except SyftPeerNotReadyError as e:
+                warnings.warn(f"Outbox messages of {sender_email} wait: {e}")
+                break
+        return read
 
-    def watcher_read_events_message(
-        self, peer_email: str, raw: bytes
-    ) -> FileChangeEventsMessage | None:
-        """Verify, decrypt and parse one outbox message of ``peer_email``.
+    def _read_events_message(self, raw: RawMessage) -> FileChangeEventsMessage | None:
+        """``raw`` verified, decrypted and decoded, or None with a warning when
+        it can never be read.
 
-        Returns None when the message is skipped, with a warning: its format
-        does not match the pair, or it cannot be read now.
+        Raises:
+            SyftPeerNotReadyError: when the peer is not set up yet.
         """
         try:
-            data = self.peer_store.decrypt_and_verify_if_needed(peer_email, raw)
-        except (
-            MessageFormatError,
-            SyftPeerNotReadyError,
-        ) as e:
-            warnings.warn(f"Skipped an outbox message of {peer_email}: {e}")
+            data = self.peer_store.decrypt_and_verify_if_needed(raw.sender, raw.data)
+            return FileChangeEventsMessage.from_compressed_data(data)
+        except SyftPeerNotReadyError:
+            raise
+        except Exception as e:
+            warnings.warn(f"Skipped unreadable outbox message {raw.msg_id}: {e!r}")
             return None
-        return FileChangeEventsMessage.from_compressed_data(data)
-
-    def watcher_get_outbox_file_metadatas(
-        self, peer_email: str, since_timestamp: float | None
-    ) -> List[dict]:
-        return self.connections.p2p_con.outbox_message_metadatas(
-            peer_email, since_timestamp
-        )
 
     def download_outbox_message(self, file_id: str) -> bytes:
-        """Download one outbox message by its handle (thread-safe)."""
+        """Download one message of my own outbox by its handle (thread-safe)."""
         return self.connections.p2p_con.clone_for_thread().fetch_message(file_id)
 
     def owner_get_outbox_file_metadatas(self, recipient_email: str) -> List[dict]:
diff --git a/syft/sync/connections/drive/drive_p2p.py b/syft/sync/connections/drive/drive_p2p.py
index a62c8c1da61..50c3418bc99 100644
--- a/syft/sync/connections/drive/drive_p2p.py
+++ b/syft/sync/connections/drive/drive_p2p.py
@@ -2,6 +2,7 @@
 
 import threading
 import warnings
+from collections.abc import Container
 from concurrent.futures import ThreadPoolExecutor
 
 from syft.sync.connections.drive.gdrive_client import DRIVE_BATCH_SIZE, GDriveClient
@@ -14,6 +15,7 @@
     execute_with_retries,
     raise_unavailable,
 )
+from syft.sync.events.file_change_event import FileChangeEventsMessageFileName
 from syft.sync.messages.proposed_filechange import FileNameParseError, MessageFileName
 from syft.sync.connections.exceptions import ConnectionUnavailableError
 from syft.sync.connections.protocols import P2PConnection
@@ -22,9 +24,14 @@
 # Threads that download the messages of one receive at the same time.
 DOWNLOAD_WORKERS = 10
 
+# How far before ``after`` the outbox is listed again, for messages that land
+# after newer ones (a retried send, or notebook and syft-bg both writing).
+# The ones in that window the DS already has are in ``already_received``.
+OUTBOX_LOOKBACK_SECONDS = 60
+
 
 def _download_or_skip(client: GDriveClient, listed: dict) -> bytes | None:
-    """The bytes of one listed inbox file, or None with a warning when this
+    """The bytes of one listed message file, or None with a warning when this
     file cannot be downloaded, for example because it was deleted.
 
     Raises:
@@ -32,22 +39,21 @@ def _download_or_skip(client: GDriveClient, listed: dict) -> bytes | None:
     """
     name = listed["file_name"]
     try:
-        with raise_unavailable(f"Drive is unreachable; inbox message {name} waits"):
+        with raise_unavailable(f"Drive is unreachable; message {name} waits"):
             return client.download_file(listed["file_id"])
     except ConnectionUnavailableError:
         raise
     except Exception as e:
-        warnings.warn(f"Skipped inbox message {name}: download failed: {e!r}")
+        warnings.warn(f"Skipped message {name}: download failed: {e!r}")
         return None
 
 
-def _require_inbox_kind(kind: MessageKind) -> None:
-    """Only the inbox is read through ``receive`` so far."""
-    if kind is not MessageKind.PROPOSED_CHANGES:
-        raise NotImplementedError(
-            f"Messages of kind {kind!r} are still read through the outbox "
-            "methods; they move to receive with send-and-forget."
-        )
+def _before_first_failure(
+    listed: list[dict], contents: list[bytes | None]
+) -> tuple[list[dict], list[bytes | None]]:
+    """``listed`` and ``contents`` up to the first file that failed to download."""
+    end = contents.index(None) if None in contents else len(contents)
+    return listed[:end], contents[:end]
 
 
 def _valid_message_names(file_metadatas: list[dict]) -> list[MessageFileName]:
@@ -67,11 +73,13 @@ class DriveP2PConnection(P2PConnection):
     outbox. ``msg_id`` is the file name. A written file is ``DELIVERED``: it
     waits in the folder until the peer reads it.
 
-    The inbox is read through ``receive`` and a processed message is archived.
-    The outbox is still read through the Drive-only methods below, from a
-    timestamp cursor, and kept: the DS rebuilds a lost cache by reading it
-    again. It moves to ``receive`` with send-and-forget, once the DS keeps what
-    it received in its own storage.
+    Both are read through ``receive``. A processed inbox message is archived.
+    The outbox keeps every message, so the DS rebuilds a lost cache by reading
+    it again, and ``receive`` lists only the messages from shortly before
+    ``after`` on, leaving out the ones already received. That
+    ends with send-and-forget, once the DS keeps what it received in its own
+    storage. The DO still compacts its own outbox through the Drive-only
+    methods below.
     """
 
     def __init__(self, cached_client: DriveCachedClient):
@@ -100,20 +108,38 @@ def delivered(self) -> list[str]:
         return []
 
     def receive(
-        self, peer: str, kind: MessageKind, max_messages: int | None = None
+        self,
+        peer: str,
+        kind: MessageKind,
+        after: str | None = None,
+        max_messages: int | None = None,
+        already_received: Container[str] = frozenset(),
     ) -> list[RawMessage]:
-        """One listing for the whole inbox, then the files download in parallel.
+        """One listing of the folder, then the files download in parallel.
 
-        A file that cannot be downloaded is left out with a warning, so it does
-        not hold back the others.
+        Proposed changes come from my inbox, which archives processed messages,
+        so ``after`` is not needed there. File events come from ``peer``'s
+        outbox, which keeps them. Those from ``OUTBOX_LOOKBACK_SECONDS`` before
+        ``after`` on are listed, and the ones in ``already_received`` are left
+        out before downloading.
+
+        A file that cannot be downloaded is left out with a warning. In the
+        inbox it does not hold back the others. In the outbox it ends the list,
+        because the reader passes the last message it got as ``after`` and
+        would never get the missed one.
 
         Raises:
             ConnectionUnavailableError: when Drive cannot be reached. Every
-                message stays in the inbox.
+                message stays in its folder.
         """
-        _require_inbox_kind(kind)
-        listed = self._list_inbox(peer)[:max_messages]
+        if kind is MessageKind.PROPOSED_CHANGES:
+            listed = self._list_inbox(peer)
+        else:
+            listed = self._list_outbox(peer, after, already_received)
+        listed = listed[:max_messages]
         contents = self._download_all(listed)
+        if kind is MessageKind.FILE_EVENTS:
+            listed, contents = _before_first_failure(listed, contents)
         return [
             RawMessage(
                 sender=peer,
@@ -144,18 +170,20 @@ def download(m: dict) -> bytes | None:
             return list(pool.map(download, listed))
 
     def mark_processed(self, message: RawMessage) -> None:
+        # The outbox keeps file events; the DS passes ``after`` instead.
+        if message.kind is MessageKind.FILE_EVENTS:
+            return
         # The archive folder is named after ``message.sender``, which receive
         # set from the inbox folder, never from the message body.
-        _require_inbox_kind(message.kind)
         execute_with_retries(self._archive_request(message.handle, message.sender))
 
     def mark_all_processed(self, messages: list[RawMessage]) -> None:
-        """Archive every message, up to ``DRIVE_BATCH_SIZE`` moves per request.
+        """Archive every inbox message, up to ``DRIVE_BATCH_SIZE`` moves per
+        request. File events stay in the outbox.
 
         A move that fails is warned about and its file stays in the inbox.
         """
-        for message in messages:
-            _require_inbox_kind(message.kind)
+        messages = [m for m in messages if m.kind is MessageKind.PROPOSED_CHANGES]
 
         def callback(request_id, response, exception):
             if exception is not None:
@@ -172,42 +200,6 @@ def callback(request_id, response, exception):
                 )
             batch_execute_with_retries(batch)
 
-    def outbox_messages(self, peer: str, since_timestamp: float | None) -> list[bytes]:
-        """The messages in ``peer``'s outbox to me, newer than ``since_timestamp``."""
-        return [
-            self._cached_client.client.download_file(m["file_id"])
-            for m in self.outbox_message_metadatas(peer, since_timestamp)
-        ]
-
-    def outbox_message_metadatas(
-        self, peer: str, since_timestamp: float | None
-    ) -> list[dict]:
-        """Like ``outbox_messages``, without downloading: each entry's
-        ``file_id`` is the handle to ``fetch_message`` it with."""
-        folder_id = self._cached_client.get_peer_datasite_outbox_id(peer)
-        if folder_id is None:
-            return []
-
-        file_metadatas = self._cached_client.get_file_metadatas_from_folder(
-            folder_id, since_timestamp=since_timestamp
-        )
-        valid_fname_objs = valid_events_from_file_metadatas(file_metadatas)
-        name_to_id = {f["name"]: f["id"] for f in file_metadatas}
-
-        result = []
-        for fname_obj in sorted(valid_fname_objs, key=lambda x: x.timestamp):
-            if since_timestamp is None or fname_obj.timestamp > since_timestamp:
-                file_name = fname_obj.as_string()
-                if file_name in name_to_id:
-                    result.append(
-                        {
-                            "file_id": name_to_id[file_name],
-                            "file_name": file_name,
-                            "timestamp": fname_obj.timestamp,
-                        }
-                    )
-        return result
-
     def fetch_message(self, handle: str) -> bytes:
         return self._cached_client.client.download_file(handle)
 
@@ -244,6 +236,37 @@ def _list_inbox(self, peer: str) -> list[dict]:
             )
         ]
 
+    def _list_outbox(
+        self, peer: str, after: str | None, already_received: Container[str]
+    ) -> list[dict]:
+        """The messages in ``peer``'s outbox to me from ``OUTBOX_LOOKBACK_SECONDS``
+        before the message ``after`` on, without those in ``already_received``,
+        oldest first, each with ``file_id`` and ``file_name``, without
+        downloading them. An outbox whose folder is not found holds none.
+
+        The look-back lists ``after`` and the messages just before it again,
+        which the DS received in earlier syncs; ``already_received`` skips them."""
+        folder_id = self._cached_client.get_peer_datasite_outbox_id(peer)
+        if folder_id is None:
+            return []
+        since = None
+        if after is not None:
+            sent_at = FileChangeEventsMessageFileName.from_string(after).timestamp
+            since = sent_at - OUTBOX_LOOKBACK_SECONDS
+        file_metadatas = self._cached_client.get_file_metadatas_from_folder(
+            folder_id, since_timestamp=since
+        )
+        name_to_id = {f["name"]: f["id"] for f in file_metadatas}
+        names = sorted(
+            valid_events_from_file_metadatas(file_metadatas), key=lambda x: x.timestamp
+        )
+        return [
+            {"file_id": name_to_id[name.as_string()], "file_name": name.as_string()}
+            for name in names
+            if name.as_string() in name_to_id
+            and name.as_string() not in already_received
+        ]
+
     def _archive_request(self, file_id: str, sender: str):
         """The request that moves an inbox file to its archive folder. The inbox
         folder is the file's only parent and both folder ids are cached, so the
diff --git a/syft/sync/connections/protocols.py b/syft/sync/connections/protocols.py
index 1de8510c994..2df88176a38 100644
--- a/syft/sync/connections/protocols.py
+++ b/syft/sync/connections/protocols.py
@@ -15,6 +15,7 @@
 from __future__ import annotations
 
 from abc import ABC, abstractmethod
+from collections.abc import Container
 from typing import TYPE_CHECKING, Self
 
 from pydantic import BaseModel, ConfigDict
@@ -72,11 +73,23 @@ def delivered(self) -> list[str]:
 
     @abstractmethod
     def receive(
-        self, peer: str, kind: MessageKind, max_messages: int | None = None
+        self,
+        peer: str,
+        kind: MessageKind,
+        after: str | None = None,
+        max_messages: int | None = None,
+        already_received: Container[str] = frozenset(),
     ) -> list[RawMessage]:
         """Messages of ``kind`` from ``peer`` that are not processed yet, oldest
         first, at most ``max_messages`` of them.
 
+        ``after`` is the ``msg_id`` of the newest message the receiver
+        processed. A connection that keeps processed messages, as Drive keeps
+        the outbox, returns the later ones with no gap: a message it cannot
+        return now ends the list. It may also list back a bit before ``after``
+        for late messages, skipping the ones in ``already_received``. A
+        connection that drops processed messages ignores both.
+
         A peer whose channel does not exist yet has no messages. The connection
         sets each ``RawMessage.sender`` from where the message came from.
         """
@@ -84,7 +97,8 @@ def receive(
     @abstractmethod
     def mark_processed(self, message: RawMessage) -> None:
         """Record that ``message`` was processed; ``receive`` never returns it
-        again."""
+        again, or, on a connection that keeps processed messages, not once
+        ``after`` is at or past it."""
 
     def mark_all_processed(self, messages: list[RawMessage]) -> None:
         """``mark_processed`` for each of ``messages``."""
diff --git a/syft/sync/sync/caches/datasite_watcher_cache.py b/syft/sync/sync/caches/datasite_watcher_cache.py
index fb18ebb64c4..9255d7abf91 100644
--- a/syft/sync/sync/caches/datasite_watcher_cache.py
+++ b/syft/sync/sync/caches/datasite_watcher_cache.py
@@ -1,6 +1,6 @@
 import logging
 from concurrent.futures import ThreadPoolExecutor
-from typing import Callable, Dict, List
+from typing import Callable, Dict, List, Set
 from syft.sync.sync.caches.cache_file_writer_connection import FSFileConnection
 from pathlib import Path
 from pydantic import BaseModel, Field
@@ -8,6 +8,7 @@
 from syft.sync.events.file_change_event import (
     FileChangeEvent,
     FileChangeEventsMessage,
+    FileChangeEventsMessageFileName,
 )
 from syft.sync.connections.connection_router import ConnectionRouter
 from syft.sync.sync.caches.cache_file_writer_connection import (
@@ -32,6 +33,16 @@ def _is_within_peer_datasite(event: FileChangeEvent, peer_email: str) -> bool:
     return not path.is_absolute() and ".." not in path.parts
 
 
+def _peer_key(email: str) -> str:
+    """``email`` without case, as peers' emails are compared elsewhere."""
+    return email.casefold()
+
+
+def _sent_at(msg_id: str) -> float:
+    """The send time in the name of an events message."""
+    return FileChangeEventsMessageFileName.from_string(msg_id).timestamp
+
+
 def _events_within_peer_datasite(
     events: List[FileChangeEvent], peer_email: str
 ) -> List[FileChangeEvent]:
@@ -76,7 +87,12 @@ class DataSiteWatcherCache(BaseModel):
     last_sync: datetime | None = None
     seconds_before_syncing_down: int = SECONDS_BEFORE_SYNCING_DOWN
     peers: List[str] = []
-    last_event_timestamp_per_peer: Dict[str, float] = {}
+    # The msg_id of the newest events message received from each peer, keyed
+    # by ``_peer_key``.
+    last_msg_id_per_peer: Dict[str, str] = {}
+    # The msg_id of each events message received. The connection lists back a
+    # bit before the last one for late messages, and skips these.
+    received_msg_ids: Set[str] = set()
     # Base syftbox folder
     syftbox_folder: Path | None = None
     # Collections to sync down (prefix + local subpath per collection type)
@@ -125,39 +141,36 @@ def from_config(
             return cache
 
     def _load_cached_state(self):
-        """Load cached state from disk: file hashes, timestamps, and collection hashes."""
-        self._load_file_hashes_from_events()
+        """Load cached state from disk: file hashes, last messages, and collection hashes."""
+        self._load_saved_messages()
         self._load_collection_hashes_from_disk()
 
-    def _load_file_hashes_from_events(self):
-        """Load file hashes and timestamps from cached events."""
+    def _load_saved_messages(self):
+        """Restore the received messages, last messages and file hashes from
+        the saved events messages, oldest first."""
         try:
-            cached_messages = self.events_connection.get_all()
+            saved_messages = self.events_connection.get_all()
         except Exception:
-            cached_messages = []
-
-        if not cached_messages:
-            return
-
-        sorted_messages = sorted(cached_messages, key=lambda m: m.timestamp)
-
-        for events_message in sorted_messages:
-            for event in events_message.events:
-                # Update last_event_timestamp_per_peer
-                peer_email = event.datasite_email
-                current_ts = self.last_event_timestamp_per_peer.get(peer_email)
-                if current_ts is None or events_message.timestamp > current_ts:
-                    self.last_event_timestamp_per_peer[peer_email] = (
-                        events_message.timestamp
-                    )
-
-                # Update file_hashes
-                path_key = Path(event.path_in_syftbox)
-                if event.is_deleted:
-                    if path_key in self.file_hashes:
-                        del self.file_hashes[path_key]
-                else:
-                    self.file_hashes[path_key] = event.new_hash
+            saved_messages = []
+        for events_message in sorted(saved_messages, key=lambda m: m.timestamp):
+            self._restore_received(events_message)
+            self._restore_file_hashes(events_message)
+
+    def _restore_received(self, events_message: FileChangeEventsMessage) -> None:
+        """Record a saved message as received. It does not store its sender, but
+        its events name it: only events for the sender's datasite are saved."""
+        msg_id = events_message.message_filepath.as_string()
+        self.received_msg_ids.add(msg_id)
+        if events_message.events:
+            self._record_received(events_message.events[0].datasite_email, msg_id)
+
+    def _restore_file_hashes(self, events_message: FileChangeEventsMessage) -> None:
+        for event in events_message.events:
+            path_key = Path(event.path_in_syftbox)
+            if event.is_deleted:
+                self.file_hashes.pop(path_key, None)
+            else:
+                self.file_hashes[path_key] = event.new_hash
 
     def _load_collection_hashes_from_disk(self):
         """Scan local collection directories and compute hashes to populate collection_hashes."""
@@ -212,7 +225,8 @@ def clear_cache(self):
         self.last_sync = None
         self.peers = []
         self.current_check_point = None
-        self.last_event_timestamp_per_peer = {}
+        self.last_msg_id_per_peer = {}
+        self.received_msg_ids = set()
         self.collection_hashes = {}
 
     @property
@@ -221,60 +235,36 @@ def last_event_timestamp(self) -> float | None:
             return None
         return self.events_connection.get_latest().timestamp
 
-    def sync_down(self, peer_email: str):
-        # Use per-peer timestamp to avoid filtering out events from other peers
-        peer_timestamp = self.last_event_timestamp_per_peer.get(peer_email)
-
-        new_event_messages = self.connection_router.watcher_get_events_messages(
-            peer_email=peer_email,
-            since_timestamp=peer_timestamp,
+    def sync_down(self, peer_email: str) -> int:
+        """Apply the events messages of ``peer_email`` not received yet: those
+        after the last one, and any that landed late. Returns the number of
+        events applied. A message that can never be read is passed over, so it
+        does not hold back the later ones."""
+        received = self.connection_router.watcher_receive_events_messages(
+            peer_email,
+            after=self.last_msg_id_per_peer.get(_peer_key(peer_email)),
+            already_received=self.received_msg_ids,
         )
-        for event_message in sorted(new_event_messages, key=lambda x: x.timestamp):
-            self.apply_event_message(event_message, peer_email)
-            self.last_event_timestamp_per_peer[peer_email] = event_message.timestamp
-
-        self.last_sync = datetime.now()
-
-    def sync_down_parallel(
-        self,
-        peer_email: str,
-        executor: ThreadPoolExecutor,
-        download_fn: Callable[[str], FileChangeEventsMessage | None],
-    ) -> int:
-        """Sync with parallel file downloads. Returns the number of events applied.
-
-        ``download_fn`` gets the id of one file. It returns None for a message
-        that it skips.
-        """
-        peer_timestamp = self.last_event_timestamp_per_peer.get(peer_email)
-
-        # Get file metadata (no download yet)
-        file_metadatas = self.connection_router.watcher_get_outbox_file_metadatas(
-            peer_email=peer_email,
-            since_timestamp=peer_timestamp,
-        )
-
-        if not file_metadatas:
-            # No new messages to download
-            self.last_sync = datetime.now()
-            return 0
-
-        # Download all files in parallel
-        file_ids = [m["file_id"] for m in file_metadatas]
-        downloaded_messages = [
-            m for m in executor.map(download_fn, file_ids) if m is not None
-        ]
-
-        # Apply in timestamp order
         event_count = 0
-        for event_message in sorted(downloaded_messages, key=lambda x: x.timestamp):
-            self.apply_event_message(event_message, peer_email)
-            self.last_event_timestamp_per_peer[peer_email] = event_message.timestamp
-            event_count += len(event_message.events)
+        for raw, event_message in received:
+            if event_message is not None:
+                self.apply_event_message(event_message, peer_email)
+                event_count += len(event_message.events)
+            self._record_received(peer_email, raw.msg_id)
+        self.connection_router.mark_all_processed([raw for raw, _ in received])
 
         self.last_sync = datetime.now()
         return event_count
 
+    def _record_received(self, peer_email: str, msg_id: str) -> None:
+        """Remember ``msg_id``, and make it ``peer_email``'s last message when
+        it is newer. A message that landed late is older than the last one."""
+        self.received_msg_ids.add(msg_id)
+        key = _peer_key(peer_email)
+        last = self.last_msg_id_per_peer.get(key)
+        if last is None or _sent_at(msg_id) > _sent_at(last):
+            self.last_msg_id_per_peer[key] = msg_id
+
     def apply_event_message(
         self, event_message: FileChangeEventsMessage, peer_email: str
     ) -> None:
@@ -287,7 +277,7 @@ def apply_event_message(
         before it reaches the local file connection.
 
         The saved copy of the message holds only the accepted events.
-        ``_load_file_hashes_from_events`` reads saved messages back on start,
+        ``_load_saved_messages`` reads saved messages back on start,
         and a saved message does not record its sender, so a dropped event
         cannot be checked again there.
         """
diff --git a/syft/sync/sync/datasite_owner_syncer.py b/syft/sync/sync/datasite_owner_syncer.py
index 9e4df6bfa91..e33f6b869f9 100644
--- a/syft/sync/sync/datasite_owner_syncer.py
+++ b/syft/sync/sync/datasite_owner_syncer.py
@@ -330,7 +330,7 @@ def _pull_proposals(self, peer_email: str) -> None:
             for raw, message in received:
                 if self._apply_proposed_message(raw, message):
                     proposed_count += len(message.proposed_file_changes)
-            self.connection_router.owner_mark_all_processed([r for r, _ in received])
+            self.connection_router.mark_all_processed([r for r, _ in received])
         except SyftPeerNotReadyError as e:
             warnings.warn(f"Messages from {peer_email} stay in the inbox: {e}")
         except Exception as e:
diff --git a/syft/sync/sync/datasite_watcher_syncer.py b/syft/sync/sync/datasite_watcher_syncer.py
index 7771c88a1ee..5ff8df81435 100644
--- a/syft/sync/sync/datasite_watcher_syncer.py
+++ b/syft/sync/sync/datasite_watcher_syncer.py
@@ -7,7 +7,6 @@
 
 from syft.sync.connections.connection_router import ConnectionRouter
 from syft.sync.callback_mixin import BaseModelCallbackMixin
-from syft.sync.events.file_change_event import FileChangeEventsMessage
 from syft.sync.peers.exceptions import SyftPeerNotReadyError
 from syft.sync.messages.proposed_filechange import (
     ProposedFileChange,
@@ -172,17 +171,6 @@ def on_file_change(
 
     # --- Pulling from datasite outboxes ---
 
-    def download_events_message_with_new_connection(
-        self, file_id: str, peer_email: str
-    ) -> FileChangeEventsMessage | None:
-        """Download from outbox using a new connection (thread-safe).
-
-        Returns None when the message is skipped (see
-        ``ConnectionRouter.watcher_read_events_message``).
-        """
-        raw = self.connection_router.download_outbox_message(file_id)
-        return self.connection_router.watcher_read_events_message(peer_email, raw)
-
     def download_collection_file_with_new_connection(
         self, file_id: str, owner_email: str
     ) -> bytes:
@@ -198,15 +186,7 @@ def download_collection_file_with_new_connection(
 
     def sync_down(self, peer_emails: list[str]):
         for peer_email in peer_emails:
-            # Sync messages with parallel download
-            event_count = self.datasite_watcher_cache.sync_down_parallel(
-                peer_email,
-                self._executor,
-                lambda fid,
-                pe=peer_email: self.download_events_message_with_new_connection(
-                    fid, pe
-                ),
-            )
+            event_count = self.datasite_watcher_cache.sync_down(peer_email)
             if event_count:
                 print(f"Pulled {event_count} inbound event(s) from {peer_email}")
             # Sync collections with parallel download
diff --git a/tests/integration/with_unit_coverage/test_cache_aware_sync.py b/tests/integration/with_unit_coverage/test_cache_aware_sync.py
index e718ac2367f..3f1b703f273 100644
--- a/tests/integration/with_unit_coverage/test_cache_aware_sync.py
+++ b/tests/integration/with_unit_coverage/test_cache_aware_sync.py
@@ -129,7 +129,7 @@ def test_ds_cache_aware_sync_gdrive():
     verify cache has 3 events, sync and end up with 5.
 
     This test verifies that when a DS manager is restarted with a persisted filesystem cache,
-    it uses the cached last_event_timestamp_per_peer to only download new events.
+    it uses the cached last_msg_id_per_peer to only download new events.
     """
     # Create initial pair with filesystem cache
     ds_manager1, do_manager1 = (
@@ -171,8 +171,8 @@ def test_ds_cache_aware_sync_gdrive():
     assert initial_hash_count >= 3, (
         f"Should have at least 3 file hashes cached, got {initial_hash_count}"
     )
-    assert len(ds_cache1.last_event_timestamp_per_peer) > 0, (
-        "Should have timestamps for peers"
+    assert len(ds_cache1.last_msg_id_per_peer) > 0, (
+        "Should have the last message of each peer"
     )
 
     # Save the syftbox_folder for creating a new manager
@@ -203,8 +203,8 @@ def test_ds_cache_aware_sync_gdrive():
         f"New manager cache should have loaded {initial_cache_count} events from disk, "
         f"got {len(new_cache.events_connection)}"
     )
-    assert len(new_cache.last_event_timestamp_per_peer) > 0, (
-        "last_event_timestamp_per_peer should be pre-populated from disk cache"
+    assert len(new_cache.last_msg_id_per_peer) > 0, (
+        "last_msg_id_per_peer should be pre-populated from disk cache"
     )
     assert len(new_cache.file_hashes) >= 3, (
         f"New manager should have at least 3 file hashes pre-populated, got {len(new_cache.file_hashes)}"
diff --git a/tests/integration/with_unit_coverage/test_gdrive_pagination.py b/tests/integration/with_unit_coverage/test_gdrive_pagination.py
index a4f2b6cfb4f..7290eb9295f 100644
--- a/tests/integration/with_unit_coverage/test_gdrive_pagination.py
+++ b/tests/integration/with_unit_coverage/test_gdrive_pagination.py
@@ -8,6 +8,7 @@
     token_path_do,
     token_path_ds,
 )
+from syft.sync.events.file_change_event import FileChangeEventsMessageFileName
 from syft.sync.syftbox_manager import SyftboxManager
 from tests.integration.with_unit_coverage.utils import is_mock_mode
 
@@ -71,9 +72,13 @@ def get_files_page_size_2(self, folder_id, since_timestamp=None, page_size=100):
         manager_ds.datasite_watcher_syncer.datasite_watcher_cache.connection_router
     )
 
-    new_events = connection_router.watcher_get_events_messages(
-        EMAIL_DO, since_timestamp=checkpoint_timestamp
-    )
+    after = FileChangeEventsMessageFileName(timestamp=checkpoint_timestamp)
+    new_events = [
+        m
+        for _, m in connection_router.watcher_receive_events_messages(
+            EMAIL_DO, after=after.as_string()
+        )
+    ]
 
     new_event_timestamps = [e.timestamp for e in new_events]
     assert all(t > checkpoint_timestamp for t in new_event_timestamps)
diff --git a/tests/migrations/p2p/test_message_protocol_downgrade.py b/tests/migrations/p2p/test_message_protocol_downgrade.py
index 91457516c47..1ded76e66f6 100644
--- a/tests/migrations/p2p/test_message_protocol_downgrade.py
+++ b/tests/migrations/p2p/test_message_protocol_downgrade.py
@@ -129,8 +129,8 @@ def _raw_proposal_version(do_manager, ds_email: str) -> str:
 def _raw_outbox_versions(ds_manager, do_email: str) -> list[str]:
     """The version fields of the DO's outbox blobs for us, unparsed."""
     p2p = ds_manager._connection_router.connections.p2p_con
-    raw_list = p2p.outbox_messages(do_email, None)
-    return [_wire_version(ds_manager, do_email, raw) for raw in raw_list]
+    received = p2p.receive(do_email, MessageKind.FILE_EVENTS)
+    return [_wire_version(ds_manager, do_email, m.data) for m in received]
 
 
 def test_a_v2_proposal_for_a_protocol0_peer_downgrades_on_the_wire(
diff --git a/tests/migrations/p2p/test_private_dataset_protocol_skew.py b/tests/migrations/p2p/test_private_dataset_protocol_skew.py
index 7fbcdd057a1..e4474bd4af6 100644
--- a/tests/migrations/p2p/test_private_dataset_protocol_skew.py
+++ b/tests/migrations/p2p/test_private_dataset_protocol_skew.py
@@ -57,9 +57,12 @@ def _create(
 
 def _shipped_private_paths(ds_manager, do_manager) -> list[str]:
     """The private-file paths of the events the DO put in our outbox."""
-    messages = ds_manager.sync_engine._connection_router.watcher_get_events_messages(
-        do_manager.email, None
-    )
+    messages = [
+        m
+        for _, m in ds_manager.sync_engine._connection_router.watcher_receive_events_messages(
+            do_manager.email
+        )
+    ]
     return [
         str(event.path_in_datasite)
         for message in messages
diff --git a/tests/unit/test_cache_aware_sync.py b/tests/unit/test_cache_aware_sync.py
index 7cf609cc58b..f9991f9c427 100644
--- a/tests/unit/test_cache_aware_sync.py
+++ b/tests/unit/test_cache_aware_sync.py
@@ -4,8 +4,17 @@
 events they don't already have locally, avoiding redundant re-downloads on restart.
 """
 
+import warnings
 from pathlib import Path
 
+import pytest
+
+from syft.sync.connections.drive.gdrive_client import GDriveClient
+from syft.sync.connections.types import MessageKind
+from syft.sync.events.file_change_event import (
+    FileChangeEventsMessage,
+    FileChangeEventsMessageFileName,
+)
 from syft.sync.syftbox_manager import SyftboxManager
 from syft.sync.sync.collection_spec import CollectionSyncSpec
 
@@ -84,7 +93,7 @@ def counted_download(events_message_id):
     )
 
 
-def test_ds_incremental_sync_downloads_only_new_events():
+def test_ds_incremental_sync_downloads_only_new_events(monkeypatch):
     """Pre-populate cache with first n events, create new manager with same cache dir,
     sync with n+m in outbox, verify only m new events are downloaded."""
     import time
@@ -108,10 +117,7 @@ def test_ds_incremental_sync_downloads_only_new_events():
     initial_cache_count = len(ds_cache.events_connection)
     assert initial_cache_count == 3
 
-    # Verify last_event_timestamp_per_peer is populated
-    assert len(ds_cache.last_event_timestamp_per_peer) > 0, (
-        "last_event_timestamp_per_peer should be populated"
-    )
+    assert do_manager.email in ds_cache.last_msg_id_per_peer
 
     # Add 2 more events to outbox
     time.sleep(0.01)  # Ensure new events have later timestamps
@@ -129,15 +135,29 @@ def test_ds_incremental_sync_downloads_only_new_events():
     assert len(new_cache.events_connection) == initial_cache_count, (
         f"New manager cache should have loaded {initial_cache_count} events from disk"
     )
-    assert len(new_cache.last_event_timestamp_per_peer) > 0, (
-        "last_event_timestamp_per_peer should be pre-populated from disk cache"
-    )
+    assert (
+        new_cache.last_msg_id_per_peer[do_manager.email]
+        == ds_cache.last_msg_id_per_peer[do_manager.email]
+    ), "last_msg_id_per_peer should be pre-populated from disk cache"
     assert len(new_cache.file_hashes) == 3, (
         "file_hashes should be pre-populated from disk cache"
     )
 
     # Sync the new manager - should only download the 2 new events
+    p2p = new_ds_manager._connection_router.connections.p2p_con
+    original_receive = p2p.receive
+    received = []
+
+    def counted_receive(*args, **kwargs):
+        messages = original_receive(*args, **kwargs)
+        received.extend(m for m in messages if m.kind is MessageKind.FILE_EVENTS)
+        return messages
+
+    monkeypatch.setattr(p2p, "receive", counted_receive)
     new_ds_manager.sync()
+    assert len(received) == 2, (
+        f"Should receive only 2 new messages, got {len(received)}"
+    )
 
     # Verify total events in cache
     assert len(new_cache.events_connection) == 5, (
@@ -254,3 +274,117 @@ def test_ds_cache_handles_deletions_correctly():
     assert expected_path not in new_cache.file_hashes, (
         "Deleted file should not appear in file_hashes after cache reload"
     )
+
+
+# -- reading the DO's outbox -------------------------------------------------------
+
+
+def _pair():
+    return SyftboxManager.pair_with_mock_drive_service_connection(
+        use_in_memory_cache=True
+    )
+
+
+def _msg_id(message: FileChangeEventsMessage) -> str:
+    return message.message_filepath.as_string()
+
+
+def _sent_at(message: FileChangeEventsMessage, timestamp: float):
+    """``message`` under a name that carries ``timestamp``."""
+    name = FileChangeEventsMessageFileName(timestamp=timestamp)
+    return message.model_copy(update={"message_filepath": name})
+
+
+def _applied_msg_ids(ds_manager) -> set[str]:
+    cache = ds_manager.datasite_watcher_syncer.datasite_watcher_cache
+    return {_msg_id(m) for m in cache.events_connection.get_all()}
+
+
+def test_ds_receives_message_landing_after_newer_one():
+    ds_manager, do_manager = _pair()
+    late, newer = get_mock_events_messages(2, datasite_email=do_manager.email)
+    late = _sent_at(late, newer.timestamp - 10)
+    router = do_manager._connection_router
+
+    router.owner_write_event_messages_to_outbox(ds_manager.email, newer)
+    ds_manager.sync()
+    router.owner_write_event_messages_to_outbox(ds_manager.email, late)
+    ds_manager.sync()
+
+    cache = ds_manager.datasite_watcher_syncer.datasite_watcher_cache
+    assert _applied_msg_ids(ds_manager) == {_msg_id(late), _msg_id(newer)}
+    assert cache.last_msg_id_per_peer[do_manager.email] == _msg_id(newer)
+
+
+def test_ds_sync_waits_for_message_that_failed_to_download(monkeypatch):
+    ds_manager, do_manager = _pair()
+    first, second = get_mock_events_messages(2, datasite_email=do_manager.email)
+    second = _sent_at(second, first.timestamp + 1)
+    router = do_manager._connection_router
+    for message in (first, second):
+        router.owner_write_event_messages_to_outbox(ds_manager.email, message)
+    [first_id] = [
+        f["id"]
+        for f in router.owner_get_outbox_file_metadatas(ds_manager.email)
+        if f["name"] == _msg_id(first)
+    ]
+    original_download = GDriveClient.download_file
+
+    def fail_first(self, file_id: str) -> bytes:
+        if file_id == first_id:
+            raise ValueError("download cut off")
+        return original_download(self, file_id)
+
+    with monkeypatch.context() as m:
+        m.setattr(GDriveClient, "download_file", fail_first)
+        with pytest.warns(UserWarning, match="download failed"):
+            ds_manager.sync()
+    assert _applied_msg_ids(ds_manager) == set()
+
+    ds_manager.sync()
+    assert _applied_msg_ids(ds_manager) == {_msg_id(first), _msg_id(second)}
+
+
+def test_ds_sync_passes_over_unreadable_message():
+    ds_manager, do_manager = _pair()
+    [readable] = get_mock_events_messages(1, datasite_email=do_manager.email)
+    unreadable = FileChangeEventsMessageFileName(timestamp=readable.timestamp - 1)
+    router = do_manager._connection_router
+    router.connections.p2p_con.send(
+        ds_manager.email, unreadable.as_string(), b"garbage", MessageKind.FILE_EVENTS
+    )
+    router.owner_write_event_messages_to_outbox(ds_manager.email, readable)
+
+    with pytest.warns(UserWarning, match="Skipped unreadable outbox message"):
+        ds_manager.sync()
+    assert _applied_msg_ids(ds_manager) == {_msg_id(readable)}
+
+    # Received once: the next sync does not download it again.
+    with warnings.catch_warnings(record=True) as caught:
+        warnings.simplefilter("always")
+        ds_manager.sync()
+    assert not [w for w in caught if unreadable.as_string() in str(w.message)]
+
+
+def test_restored_bookmark_found_for_peer_email_in_other_case(monkeypatch):
+    ds_manager, do_manager = SyftboxManager.pair_with_mock_drive_service_connection(
+        use_in_memory_cache=False
+    )
+    [message] = get_mock_events_messages(1, datasite_email=do_manager.email)
+    do_manager._connection_router.owner_write_event_messages_to_outbox(
+        ds_manager.email, message
+    )
+    ds_manager.sync()
+
+    cache = ds_manager._copy().datasite_watcher_syncer.datasite_watcher_cache
+    p2p = cache.connection_router.connections.p2p_con
+    original_receive = p2p.receive
+    afters = []
+
+    def capture_after(*args, **kwargs):
+        afters.append(kwargs.get("after"))
+        return original_receive(*args, **kwargs)
+
+    monkeypatch.setattr(p2p, "receive", capture_after)
+    cache.sync_down(do_manager.email.upper())
+    assert afters == [_msg_id(message)]
diff --git a/tests/unit/test_delivery_tracker.py b/tests/unit/test_delivery_tracker.py
index 9623e9ce288..1968775afbc 100644
--- a/tests/unit/test_delivery_tracker.py
+++ b/tests/unit/test_delivery_tracker.py
@@ -138,17 +138,23 @@ def _received(ds_manager, do_manager, path: str):
 
 
 def _outbox_paths(ds_manager, do_manager) -> list[str]:
-    messages = ds_manager._connection_router.watcher_get_events_messages(
-        do_manager.email, None
-    )
+    messages = [
+        m
+        for _, m in ds_manager._connection_router.watcher_receive_events_messages(
+            do_manager.email
+        )
+    ]
     return [str(e.path_in_datasite) for m in messages for e in m.events]
 
 
 def _sent_events(ds_manager, do_manager, path: str) -> list[tuple[bool, object]]:
     """(is_deleted, content) of each event for ``path`` in the DS's outbox."""
-    messages = ds_manager._connection_router.watcher_get_events_messages(
-        do_manager.email, None
-    )
+    messages = [
+        m
+        for _, m in ds_manager._connection_router.watcher_receive_events_messages(
+            do_manager.email
+        )
+    ]
     return [
         (e.is_deleted, e.content)
         for m in messages
@@ -422,9 +428,12 @@ def test_large_catch_up_split_into_messages(monkeypatch):
         _write(do_manager, f"shared/{name}.txt", "abc")
 
     do_manager.sync(auto_checkpoint=False)
-    messages = ds_manager._connection_router.watcher_get_events_messages(
-        do_manager.email, None
-    )
+    messages = [
+        m
+        for _, m in ds_manager._connection_router.watcher_receive_events_messages(
+            do_manager.email
+        )
+    ]
     assert sorted(len(m.events) for m in messages) == [1, 2]
 
 
@@ -516,7 +525,10 @@ def test_compaction_keeps_message_size_cap(monkeypatch):
     monkeypatch.setattr(datasite_owner_syncer, "MAX_DELIVERY_MESSAGE_CONTENT", 6)
     syncer = do_manager.datasite_owner_syncer
     assert syncer.compact_outbox_if_needed(ds_manager.email, min_messages=2) == 4
-    messages = ds_manager._connection_router.watcher_get_events_messages(
-        do_manager.email, None
-    )
+    messages = [
+        m
+        for _, m in ds_manager._connection_router.watcher_receive_events_messages(
+            do_manager.email
+        )
+    ]
     assert [len(m.events) for m in messages] == [2, 2]
diff --git a/tests/unit/test_drive_connections.py b/tests/unit/test_drive_connections.py
index 1e5d5bad6d8..175b80a8ca4 100644
--- a/tests/unit/test_drive_connections.py
+++ b/tests/unit/test_drive_connections.py
@@ -24,7 +24,10 @@
 )
 from syft.sync.connections.drive.drive_connections import DriveWatcherStoreConnection
 from syft.sync.connections.drive.drive_owner_store import DriveOwnerStoreConnection
-from syft.sync.connections.drive.drive_p2p import DriveP2PConnection
+from syft.sync.connections.drive.drive_p2p import (
+    OUTBOX_LOOKBACK_SECONDS,
+    DriveP2PConnection,
+)
 from syft.sync.connections.drive.drive_peering import DrivePeeringConnection
 from syft.sync.connections.drive.gdrive_client import GDriveClient
 from syft.sync.connections.drive.mock_drive_service import (
@@ -279,7 +282,7 @@ def download_all_but_broken(self, file_id: str) -> bytes:
         return original_download(self, file_id)
 
     monkeypatch.setattr(GDriveClient, "download_file", download_all_but_broken)
-    with pytest.warns(UserWarning, match=f"Skipped inbox message {names[1]}"):
+    with pytest.warns(UserWarning, match=f"Skipped message {names[1]}"):
         received = DriveP2PConnection(do_cached_client).receive(
             DS, MessageKind.PROPOSED_CHANGES
         )
@@ -314,24 +317,121 @@ def test_file_events_sent_through_p2p_reach_the_scientist_outbox(pair):
         DS, name, b"results", MessageKind.FILE_EVENTS
     )
     assert status is SendStatus.DELIVERED
+    [message] = DriveP2PConnection(ds_cached_client).receive(
+        DO, MessageKind.FILE_EVENTS
+    )
+    assert (message.sender, message.msg_id, message.kind, message.data) == (
+        DO,
+        name,
+        MessageKind.FILE_EVENTS,
+        b"results",
+    )
+
+
+def test_outbox_messages_after_given_one_are_received_oldest_first(pair):
+    do_cached_client, ds_cached_client = pair
+    names = [_events_name(ts) for ts in (1.0, 3.0, 2.0)]
+    for name in names:
+        DriveP2PConnection(do_cached_client).send(
+            DS, name, name.encode(), MessageKind.FILE_EVENTS
+        )
     p2p = DriveP2PConnection(ds_cached_client)
-    listed = p2p.outbox_message_metadatas(DO, since_timestamp=None)
-    assert [m["file_name"] for m in listed] == [name]
-    assert p2p.fetch_message(listed[0]["file_id"]) == b"results"
-    assert p2p.outbox_messages(DO, since_timestamp=None) == [b"results"]
 
+    received = p2p.receive(
+        DO, MessageKind.FILE_EVENTS, after=names[0], already_received={names[0]}
+    )
+    assert [(m.msg_id, m.data) for m in received] == [
+        (names[2], names[2].encode()),
+        (names[1], names[1].encode()),
+    ]
 
-def test_a_p2p_copy_for_another_thread_reads_the_same_outbox(pair):
+
+def test_outbox_message_landing_after_newer_one_is_received(pair, monkeypatch):
+    do_cached_client, ds_cached_client = pair
+    do_p2p = DriveP2PConnection(do_cached_client)
+    ds_p2p = DriveP2PConnection(ds_cached_client)
+    newer, late = _events_name(2000.0), _events_name(1990.0)
+    do_p2p.send(DS, newer, b"newer", MessageKind.FILE_EVENTS)
+    ds_p2p.receive(DO, MessageKind.FILE_EVENTS)
+    do_p2p.send(DS, late, b"late", MessageKind.FILE_EVENTS)
+
+    calls = _count_drive_calls(monkeypatch)
+    received = ds_p2p.receive(
+        DO, MessageKind.FILE_EVENTS, after=newer, already_received={newer}
+    )
+
+    assert [m.msg_id for m in received] == [late]
+    assert calls["get_media"] == 1  # the newer message is not downloaded again
+
+
+def test_outbox_message_older_than_lookback_is_not_listed(pair):
+    do_cached_client, ds_cached_client = pair
+    old = _events_name(1.0)
+    after = _events_name(1.0 + OUTBOX_LOOKBACK_SECONDS + 1)
+    for name in (old, after):
+        DriveP2PConnection(do_cached_client).send(
+            DS, name, b"x", MessageKind.FILE_EVENTS
+        )
+
+    received = DriveP2PConnection(ds_cached_client).receive(
+        DO, MessageKind.FILE_EVENTS, after=after, already_received={after}
+    )
+    assert received == []
+
+
+def test_outbox_receive_stops_at_message_that_fails_to_download(pair, monkeypatch):
+    do_cached_client, ds_cached_client = pair
+    names = [_events_name(ts) for ts in (1.0, 2.0, 3.0)]
+    for name in names:
+        DriveP2PConnection(do_cached_client).send(
+            DS, name, name.encode(), MessageKind.FILE_EVENTS
+        )
+    original_download = GDriveClient.download_file
+
+    def download_all_but_second(self, file_id: str) -> bytes:
+        data = original_download(self, file_id)
+        if data == names[1].encode():
+            raise ValueError("download cut off")
+        return data
+
+    monkeypatch.setattr(GDriveClient, "download_file", download_all_but_second)
+    with pytest.warns(UserWarning, match=f"Skipped message {names[1]}"):
+        received = DriveP2PConnection(ds_cached_client).receive(
+            DO, MessageKind.FILE_EVENTS
+        )
+
+    # The third would move the reader's ``after`` past the second.
+    assert [m.msg_id for m in received] == [names[0]]
+
+
+def test_processed_outbox_message_stays_in_outbox(pair):
     do_cached_client, ds_cached_client = pair
     DriveP2PConnection(do_cached_client).send(
         DS, _events_name(1.0), b"results", MessageKind.FILE_EVENTS
     )
     p2p = DriveP2PConnection(ds_cached_client)
-    [listed] = p2p.outbox_message_metadatas(DO, since_timestamp=None)
+
+    [message] = p2p.receive(DO, MessageKind.FILE_EVENTS)
+    p2p.mark_processed(message)
+    p2p.mark_all_processed([message])
+    assert p2p.receive(DO, MessageKind.FILE_EVENTS) == [message]
+
+
+def test_receiving_from_peer_without_outbox_returns_nothing(pair):
+    do_cached_client, _ = pair
+    p2p = DriveP2PConnection(do_cached_client)
+    assert p2p.receive(DS, MessageKind.FILE_EVENTS) == []
+
+
+def test_a_p2p_copy_for_another_thread_reads_the_same_outbox(pair):
+    do_cached_client, _ = pair
+    p2p = DriveP2PConnection(do_cached_client)
+    p2p.send(DS, _events_name(1.0), b"results", MessageKind.FILE_EVENTS)
+    [listed] = p2p.own_outbox_message_metadatas(DS)
 
     copy = p2p.clone_for_thread()
-    assert copy._cached_client is not ds_cached_client
-    assert copy.fetch_message(listed["file_id"]) == b"results"
+    assert copy._cached_client is not do_cached_client
+    assert copy.fetch_message(listed["id"]) == b"results"
 
 
 def test_an_unknown_message_kind_is_refused(pair):
@@ -347,16 +447,6 @@ def test_drive_never_has_sends_waiting_for_confirmation(pair):
     assert DriveP2PConnection(do_cached_client).delivered() == []
 
 
-def test_receiving_file_events_through_p2p_is_not_built_yet(pair):
-    _, ds_cached_client = pair
-    p2p = DriveP2PConnection(ds_cached_client)
-    message = RawMessage(sender=DO, msg_id="m", kind=MessageKind.FILE_EVENTS, data=b"")
-    with pytest.raises(NotImplementedError):
-        p2p.receive(DO, MessageKind.FILE_EVENTS)
-    with pytest.raises(NotImplementedError):
-        p2p.mark_processed(message)
-
-
 # -- OwnerStoreConnection ---------------------------------------------------------
 
 
@@ -656,18 +746,64 @@ def test_the_router_marks_handled_messages_processed_through_p2p():
     message = RawMessage(
         sender=DS, msg_id="m", kind=MessageKind.PROPOSED_CHANGES, data=b""
     )
-    router.owner_mark_all_processed([message])
+    router.mark_all_processed([message])
     p2p.mark_all_processed.assert_called_once_with([message])
 
 
 def test_the_router_reads_the_outbox_through_p2p():
     router, p2p = _router_with_mock_p2p()
-    p2p.outbox_messages.return_value = []
-    p2p.outbox_message_metadatas.return_value = []
-    assert router.watcher_get_events_messages(DO, 5.0) == []
-    assert router.watcher_get_outbox_file_metadatas(DO, 5.0) == []
-    p2p.outbox_messages.assert_called_once_with(DO, 5.0)
-    p2p.outbox_message_metadatas.assert_called_once_with(DO, 5.0)
+    p2p.receive.return_value = []
+    assert (
+        router.watcher_receive_events_messages(DO, after="m", already_received={"m"})
+        == []
+    )
+    p2p.receive.assert_called_once_with(
+        DO, MessageKind.FILE_EVENTS, after="m", already_received={"m"}
+    )
+
+
+def test_router_reads_unreadable_events_message_as_none():
+    router, p2p = _router_with_mock_p2p()
+    events = FileChangeEventsMessage(events=[])
+    unreadable = RawMessage(
+        sender=DS, msg_id="bad", kind=MessageKind.FILE_EVENTS, data=b"garbage"
+    )
+    readable = RawMessage(
+        sender=DS,
+        msg_id="good",
+        kind=MessageKind.FILE_EVENTS,
+        data=events.as_compressed_data(),
+    )
+    p2p.receive.return_value = [unreadable, readable]
+
+    with pytest.warns(UserWarning, match="Skipped unreadable outbox message bad"):
+        pairs = router.watcher_receive_events_messages(DS)
+
+    assert [(raw.msg_id, message) for raw, message in pairs] == [
+        ("bad", None),
+        ("good", events),
+    ]
+
+
+def test_router_stops_at_events_message_not_readable_yet():
+    router, p2p = _router_with_mock_p2p()
+    data = FileChangeEventsMessage(events=[]).as_compressed_data()
+    # The router's store knows DS only, so a stranger is not set up yet.
+    p2p.receive.return_value = [
+        RawMessage(
+            sender=sender, msg_id=msg_id, kind=MessageKind.FILE_EVENTS, data=data
+        )
+        for sender, msg_id in (
+            (DS, "first"),
+            ("stranger@example.com", "waiting"),
+            (DS, "later"),
+        )
+    ]
+
+    with pytest.warns(UserWarning, match="Outbox messages of ds@example.com wait"):
+        pairs = router.watcher_receive_events_messages(DS)
+
+    assert [raw.msg_id for raw, _ in pairs] == ["first"]
 
 
 def test_the_router_holds_no_drive_specific_code():
diff --git a/tests/unit/test_gdrive_pagination.py b/tests/unit/test_gdrive_pagination.py
index 9869960989e..cee48169069 100644
--- a/tests/unit/test_gdrive_pagination.py
+++ b/tests/unit/test_gdrive_pagination.py
@@ -10,6 +10,7 @@
     MockDriveBackingStore,
     MockDriveService,
 )
+from syft.sync.connections.types import MessageKind
 from syft.sync.events.file_change_event import (
     FileChangeEventsMessage,
     FileChangeEventsMessageFileName,
@@ -103,9 +104,9 @@ def mock_download(file_id):
     mock_download_file.side_effect = mock_download
 
     events = [
-        FileChangeEventsMessage.from_compressed_data(data)
-        for data in DriveP2PConnection(cached_client).outbox_messages(
-            "peer@test.com", None
+        FileChangeEventsMessage.from_compressed_data(m.data)
+        for m in DriveP2PConnection(cached_client).receive(
+            "peer@test.com", MessageKind.FILE_EVENTS
         )
     ]
 
diff --git a/tests/unit/test_sync_manager.py b/tests/unit/test_sync_manager.py
index 9554379c1f5..57dfae1eb4b 100644
--- a/tests/unit/test_sync_manager.py
+++ b/tests/unit/test_sync_manager.py
@@ -5,6 +5,7 @@
 
 from syft.sync.connections.drive.drive_peering import DrivePeeringConnection
 from syft.sync.connections.connection_router import ConnectionRouter
+from syft.sync.connections.types import MessageKind
 from syft.sync.connections.drive.drive_cached_client import DriveCachedClient
 from syft.sync.encryption_policy import EncryptionPolicy
 from syft.sync.peers.peer import Peer
@@ -169,7 +170,7 @@ def test_sync_existing_datasite_state_do():
     n_files_in_cache = len(do_manager.datasite_owner_syncer.event_cache.file_connection)
     hashes_in_cache = len(do_manager.datasite_owner_syncer.event_cache.file_hashes)
 
-    n_outbox = ds_p2p.outbox_message_metadatas(do_manager.email, None)
+    n_outbox = ds_p2p.receive(do_manager.email, MessageKind.FILE_EVENTS)
     assert n_messages_in_cache >= 1  # At least 1 message with the 2 file changes
     # 2 data files + private/crypto_keys.json + private/syft_settings.json
     assert n_files_in_cache == 4
@@ -552,11 +553,12 @@ def test_job_files_sync_to_submitter_only():
     submitter_router = _peer_router(ds_manager, do_manager, submitter_email)
     do_manager.datasite_owner_syncer.sync(recipients)
 
-    messages_for_non_submitter = (
-        ds_manager._connection_router.watcher_get_events_messages(
-            do_manager.email, None
+    messages_for_non_submitter = [
+        m
+        for _, m in ds_manager._connection_router.watcher_receive_events_messages(
+            do_manager.email
         )
-    )
+    ]
 
     paths_for_non_submitter = [
         str(event.path_in_datasite)
@@ -564,9 +566,9 @@ def test_job_files_sync_to_submitter_only():
         for event in msg.events
     ]
 
-    messages_for_submitter = submitter_router.watcher_get_events_messages(
-        do_manager.email, None
-    )
+    messages_for_submitter = [
+        m for _, m in submitter_router.watcher_receive_events_messages(do_manager.email)
+    ]
     paths_for_submitter = [
         str(event.path_in_datasite)
         for msg in messages_for_submitter
@@ -634,8 +636,8 @@ def test_job_file_beside_outputs_does_not_sync_to_submitter():
 
     paths = [
         str(event.path_in_datasite)
-        for msg in ds_manager._connection_router.watcher_get_events_messages(
-            do_manager.email, None
+        for _, msg in ds_manager._connection_router.watcher_receive_events_messages(
+            do_manager.email
         )
         for event in msg.events
     ]
@@ -758,14 +760,19 @@ def test_permission_change_triggers_resend():
     recipients = [peer_a_email, peer_b_email]
     do_manager.datasite_owner_syncer.sync(recipients)
 
-    messages_for_a = ds_manager._connection_router.watcher_get_events_messages(
-        do_manager.email, None
-    )
+    messages_for_a = [
+        m
+        for _, m in ds_manager._connection_router.watcher_receive_events_messages(
+            do_manager.email
+        )
+    ]
     paths_for_a = [
         str(e.path_in_datasite) for msg in messages_for_a for e in msg.events
     ]
 
-    messages_for_b = peer_b_router.watcher_get_events_messages(do_manager.email, None)
+    messages_for_b = [
+        m for _, m in peer_b_router.watcher_receive_events_messages(do_manager.email)
+    ]
     paths_for_b = [
         str(e.path_in_datasite) for msg in messages_for_b for e in msg.events
     ]
@@ -781,9 +788,9 @@ def test_permission_change_triggers_resend():
     # Second sync: peer B should receive data.txt via resend
     do_manager.datasite_owner_syncer.sync(recipients)
 
-    messages_for_b_after = peer_b_router.watcher_get_events_messages(
-        do_manager.email, None
-    )
+    messages_for_b_after = [
+        m for _, m in peer_b_router.watcher_receive_events_messages(do_manager.email)
+    ]
     paths_for_b_after = [
         str(e.path_in_datasite) for msg in messages_for_b_after for e in msg.events
     ]
@@ -797,7 +804,7 @@ def _events_for(router: ConnectionRouter, do_manager) -> list[tuple[str, bool]]:
     """(path, is_deleted) of each event in the DO's outbox for ``router``'s peer."""
     return [
         (str(e.path_in_datasite), e.is_deleted)
-        for m in router.watcher_get_events_messages(do_manager.email, None)
+        for _, m in router.watcher_receive_events_messages(do_manager.email)
         for e in m.events
     ]
 
diff --git a/tests/unit/test_watcher_event_datasite_check.py b/tests/unit/test_watcher_event_datasite_check.py
index faa4b66389f..0bda6c93916 100644
--- a/tests/unit/test_watcher_event_datasite_check.py
+++ b/tests/unit/test_watcher_event_datasite_check.py
@@ -137,7 +137,7 @@ def test_dropped_event_is_not_restored_after_restart(tmp_path):
     restarted = DataSiteWatcherCache.from_config(config, router)
 
     assert Path(f"{VICTIM}/syft.pub.yaml") not in restarted.file_hashes
-    assert VICTIM not in restarted.last_event_timestamp_per_peer
+    assert set(restarted.last_msg_id_per_peer) == {PEER}
     assert restarted.file_hashes[Path(f"{PEER}/data/file.txt")] == legitimate.new_hash