feat: keep the DS's watcher state on its own Drive and restore it at startup by pjwerneck · Pull Request #9613 · OpenMined/PySyft

pjwerneck · 2026-11-04T20:06:57Z

Summary

The DS saves a checkpoint and a rolling state for each DO on its own Drive, and restores them at startup. A lost local disk, such as a Colab runtime reset, no longer loses the DO files or the outbox cursor. PR 2 of "connsplit v1.3: Scientist durability", stacked on #9611.

Changes

New DriveWatcherStoreConnection (drive_watcher_store.py): /SyftBox/<me>-<version>-watcher-state, shared with nobody, with one folder per DO. Each save replaces the checkpoint or rolling-state file in place.
The DS reuses Checkpoint and RollingState from syft/sync/checkpoints/. CheckpointFile.content now holds text or bytes, and RollingState.received_msg_ids holds the ids the DS received.
Watcher state is not encrypted until key recovery exists, as the ticket says.
At startup, on-disk mode restores the files, hashes and received ids of every DO. After a pull, the DS saves the rolling state, then marks the messages processed. At 50 paths it folds into the checkpoint.
The local <syftbox>-event-messages cache is gone.
The DS drops a DO change without content or hash, and models refuse invalid base64 content. The DO compaction now raises on such a change instead of dropping the file.
Shared code: file_content.py for text-or-bytes content, Checkpoint.with_events() for the DO compaction and the DS fold, and one DEFAULT_CHECKPOINT_EVENT_THRESHOLD.
WatcherStoreConnection: the put methods take no name and return nothing, latest_checkpoint() is now get_checkpoint(), and peers_with_saved_state() is new.

Testing

New tests/unit/test_watcher_state.py: restart on an empty folder, restart with a new key, delete after a fold, refused write, binary file, email case, broken changes.
Drive store contract tests, later-version tests for Checkpoint and RollingState, and a corrupt base64 test.
The restart tests that read the local cache now restore from Drive.
Root unit, migrations, syft-rds, syft-job, syft-enclave and syft-datasets pass.

Asana task

https://app.asana.com/1/1185126988600652/project/1220031667339032/task/1219496828587155

- WatcherCheckpoint and WatcherRollingState in syft/sync/checkpoints/watcher_state.py hold the files received from one DO, the changes since then and the cursor in its outbox. They refuse a state that a later client wrote. - parse_versioned_json() checks the version of JSON text or bytes before the fields. read_versioned_json() now calls it for a file. - DriveWatcherStoreConnection keeps one folder per DO in /SyftBox/<me>-<version>-watcher-state, which is shared with nobody. A save replaces the content of the checkpoint file or the rolling-state file in place. - put_checkpoint() and put_rolling_state() take no file name, because each DO has one file of each kind.

…crypted with the DS key - watcher_save_checkpoint() and watcher_save_rolling_state() encrypt the state for the DS when encryption is required, then store it. - watcher_load_checkpoint() and watcher_load_rolling_state() return None only when nothing is saved. They raise when the state cannot be decrypted or a later client saved it. - When the DS key cannot decrypt the state, the error tells the DS to log in with the key_file of the key that saved it.

…at startup - At startup, a DS in on-disk mode restores every DO that has saved state. It writes the files of the checkpoint and the rolling state, and records their hashes and the received message ids. - After a pull that receives messages, the DS saves the rolling state of that DO, then marks the messages processed. At 50 changed paths, it folds them into the checkpoint. - The DS keeps the received message ids per DO. An event that pre_write_filter refuses is not saved, so a restore does not write it. - WatcherRollingState has no last_msg_id, because the cursor is the newest received id. - peers_with_saved_state() lists the DOs that have saved state.

… holds its state - In on-disk mode, the DS keeps its record of received messages in memory, as in-memory mode does. A restart restores only from the saved state on Drive. - The code that rebuilt the cursor and the file hashes from the local folder is gone, and so is the unused events_base_path. - The local cleanup no longer deletes a -event-messages folder. - The three tests that restarted from the local folder now check the state restored from Drive.

…checkpoint models - Watcher state is not encrypted until key recovery exists, as the ticket says. With the DS key, an enclave could not boot twice: its new key could not read the state of the last boot. - The DS saves a Checkpoint and a RollingState from syft/sync/checkpoints/, as the ticket says. watcher_state.py and parse_versioned_json are gone. - CheckpointFile.content holds str or bytes, so a binary file survives a checkpoint. RollingState.received_msg_ids holds the ids that a DS received. - The DS saves and loads the state of a DO under its email without case, so two spellings share one Drive folder. A restore writes the checkpoint files under the email as the DO's events spell it.

- Such a change blocked every later sync of the DS. A change with no hash made each fold raise, and a change with no content raised when it was applied. - The DS now drops it with a warning, as it does for an event aimed at another datasite.

…e DO and the DS - syft/sync/utils/file_content.py holds the text-or-bytes content code that ProposedFileChange, FileChangeEvent and CheckpointFile used to repeat. The JSON shape does not change. - Checkpoint.with_events() applies events to a checkpoint. The DO compaction and the DS fold both use it, and compact_incremental_checkpoints is gone. - The DO compaction now raises on a change without content or hash. Before, it dropped the file and could write "None" as a hash.

- decode_content() dropped characters outside the base64 alphabet and returned other bytes, with no error. It now raises, so the model refuses the content.

- DEFAULT_CHECKPOINT_EVENT_THRESHOLD moved to syft/sync/checkpoints/checkpoint.py. The DO syncer, the DS cache and SyftboxManager all read it. - WATCHER_CHECKPOINT_THRESHOLD and three literal defaults of 50 in SyftboxManager are gone.

- DriveWatcherStoreConnection.clone_for_thread() returned the same object, so another thread would share one Drive client. It now returns a store over a new client, as the other Drive connections do.

…ndle - WatcherStoreConnection.latest_checkpoint() is now get_checkpoint(), because each DO has one checkpoint. put_checkpoint() and put_rolling_state() return nothing, because no caller read the file id. - DriveWatcherStoreConnection names its helpers after the watcher folder and the files: _watcher_folder_id(), _save_file(), _read_file() and _find_id(). - DataSiteWatcherCache has _save_watcher_state() and _restore_watcher_state(). The DO compaction names its sorted checkpoints incrementals_in_order.

…test - test_ds_cache_handles_deletions_correctly reloaded over an empty mock Drive, so it passed even with no restore. It now reloads over the DS's own router. - The test also checks that a kept file comes back, so a restore that does nothing fails it.

koenvanderveen · 2026-11-09T13:27:01Z

+        if received and self.saves_watcher_state:
+            self._save_watcher_state(peer_email)
        self.connection_router.mark_all_processed([raw for raw, _ in received])


After a pull, sync_down() now saves the DO's rolling state, with the applied events and the received message ids, before it marks the messages processed.

Okay, are we also archiving the message when marking it processed?

No: the outbox keeps file events, and connsplit v1.4: Outbox Trim will remove them.

koenvanderveen · 2026-11-09T13:27:02Z

+    # =========================================================================
+    # WATCHER STATE (not encrypted until there is key recovery: a new key, as
+    # an enclave gets at each boot, could not read it back)
+    # =========================================================================


We should encrypt if a key is available. The user is responsible for storing the key.

Done. eb43610 encrypts the watcher state when the DS policy is REQUIRED, as for the DO's own state.

koenvanderveen · 2026-11-09T13:27:04Z

+    def _save_file(self, peer: str, name: str, data: bytes) -> None:
+        client = self._cached_client.client
+        watcher_folder_id = self._watcher_folder_id() or self._create_watcher_folder()
+        peer_folder_id = self._find_id(peer, watcher_folder_id) or client.create_folder(
+            peer, watcher_folder_id
+        )
+        payload, _ = client.create_file_payload(data)
+        file_id = self._find_id(name, peer_folder_id)


Every save looks up the watcher folder, the DO folder and the file before it writes, and none of the folder ids are cached, so one save costs about 4 Drive calls.

Why? This should always be optimized.

Done. 86700c1 caches the folder and file ids, so each later write is one Drive call.

koenvanderveen · 2026-11-09T13:27:06Z

+        )
+        self.connection_router.watcher_save_rolling_state(key, saved)
+
+    def _rolling_state(self, peer_email: str) -> RollingState:


_rolling_state() returns the rolling state held in memory for a DO, while _save_rolling_state() writes it to Drive.

Add a local postfix so we know it's the local one.

Done. ed7b0f7 renames it to _rolling_state_local().

koenvanderveen · 2026-11-09T13:27:07Z

    path: str  # Relative path in datasite (e.g., "public/syft_datasets/data.csv")
    hash: str  # Hash of the content (SHA256 hex string)
-    content: str  # File content
+    content: str | bytes  # File content: bytes for a binary file


CheckpointFile.content now holds str | bytes, with a computed content_type that tells the reader to decode it from base64.

Would it make more sense to always do bytes?

Done. b45bcd3 stores checkpoint content as bytes; events keep str or bytes.

koenvanderveen · 2026-11-09T13:27:09Z

    @abstractmethod
-    def latest_checkpoint(self, peer: str) -> bytes | None:
-        """The newest snapshot of files received from ``peer``, or None."""
+    def get_checkpoint(self, peer: str) -> bytes | None:


This was latest_checkpoint().

Why is it get_checkpoint? Is it not the latest anymore?

It is still the latest, but the DS keeps only one checkpoint per DO.

koenvanderveen · 2026-11-09T13:27:11Z

+    def _save_rolling_state(self, peer_email: str) -> None:
+        key = _peer_key(peer_email)


The watcher state of a DO is saved under _peer_key(), the lowercased email, while restored checkpoint files go under the email as the DO's events spell it.

Let's make a ticket in Asana to lowercase this everywhere, and only when we first read it into an object.

koenvanderveen · 2026-11-09T13:27:13Z

                connection_router=connection_router,
                syftbox_folder=config.syftbox_folder,
                collection_specs=config.collection_specs,
+                saves_watcher_state=False,


With the in-memory cache, saves_watcher_state is False, so nothing is saved to or restored from the watcher store.

Is it also not stored in memory then? We should try to mimic the real scenario properly.

Done. eb43610 makes the in-memory cache save and restore the watcher state too.

koenvanderveen · 2026-11-09T13:27:14Z

+            cache._load_collection_hashes_from_disk()
+            cache._restore_watcher_state()


The on-disk cache no longer has the local <syftbox>-event-messages folder: it restores from the watcher state on Drive, and events_connection is in-memory in both modes.

We still have a cache though, I suppose?

Yes: the files stay on disk. Only the local copy of the received messages is gone.

koenvanderveen · 2026-11-09T13:27:16Z

+def _is_complete(event: FileChangeEvent) -> bool:
+    """Whether ``event`` is a delete, or has the content and the hash of its file."""
+    return event.is_deleted or (
+        event.content is not None and event.new_hash is not None
+    )
+
+
+def _accepted_events(


_is_complete() counts an event as complete when it is a delete or has both content and a hash, and _accepted_events() drops the incomplete ones with a warning.

Isn't a change without content a file delete? I also think _accepted_events is a weird name: it's actually a filter, not something that was computed before.

Done. A delete sets is_deleted; c79c51b moves the check into FileChangeEvent and renames the filter back to _events_within_peer_datasite().

koenvanderveen · 2026-11-09T13:27:18Z

+            if self.saves_watcher_state:
+                self._rolling_state(peer_email).add_event(event)
+
+    def _write_denied(self, event: FileChangeEvent) -> bool:


apply_event_message() now asks _write_denied(), which calls the cache's pre_write_filter, before it writes an event or adds it to the rolling state.

Did this already exist as a callback? Why does it need to be a callback?

No, it doesn't need to be a callback. Changed to a config option.

koenvanderveen · 2026-11-09T13:27:19Z

+            elif event.content is None or event.new_hash is None:
+                raise ValueError(
+                    f"The change of {path} in {self.email} has no content or no hash."


Checkpoint.with_events() raises on a change without content or hash, so DatasiteOwnerSyncer.compact_checkpoints() now fails on such a change where it used to leave the file out.

Let's double check if these are produced in the code (maybe delete?). I think we want to think about error handling here, because now it means that a wrong event is an error we cannot recover from?

Done. No code makes them; since c79c51b the DS skips such a message with a warning.

koenvanderveen · 2026-11-09T13:27:21Z


+    # --- State saved in the DS's storage ---
+
+    def _restore_watcher_state(self) -> None:


_restore_watcher_state() loads each DO's checkpoint and rolling state and applies both inline.

Make two helpers here for readability.

Done. a0e19ed adds _restore_checkpoint() and _restore_rolling_state().

… helpers - _restore_watcher_state() calls _restore_checkpoint() and then _restore_rolling_state() for each DO with saved state. The checkpoint goes first, because the rolling state holds the changes after it.

…al() - _rolling_state() is now _rolling_state_local(), so its name says that it returns the copy in memory. _save_rolling_state() writes that copy to the DS's storage.

- DriveWatcherStoreConnection keeps the ids of the watcher folder, of each DO folder and of each file once it finds or creates them. After the first save, a save is one Drive call and a load is one download. Before, each save looked up three ids first. - reset(peer) drops the ids of that DO, reset_caches() drops all of them, and clone_for_thread() copies them.

…the DS - FileChangeEvent refuses a change that is not a delete and has no content or no hash. No code builds one, and a message that holds one cannot be read: the DS skips it with a warning, as any unreadable outbox message. - The DS no longer checks each event for content. _accepted_events() is _events_within_peer_datasite() again. - Checkpoint.with_events() keeps its check as a guard of that rule.

- CheckpointFile.content is always bytes, as base64 in JSON. A checkpoint built from text encodes it as UTF-8, so the hashes do not change. - The content_type field and its two hooks are gone from CheckpointFile. Events and proposals keep str or bytes, because peers exchange them.

- The router encrypts the watcher state with the DS key when the DS requires encryption, as it does the DO's own state. A key that cannot decrypt it raises an error that says to log in with the key_file of the key that saved it. - persist_watcher_state, next to persist_owner_state, turns the watcher state on or off, in both cache modes. An enclave gets a new key at each boot, so for_enclave() sets both from the same setting, which is off by default. - The in-memory cache now saves and restores the watcher state too, so test clients run it as a real client does.

- peers_with_saved_state() no longer fills the folder-id cache from its listing. That listing skipped the check that a DO has exactly one folder, so two folders made at the same time were read as one without an error. - The test of the id cache counts Drive calls with _count_drive_calls(), which sees every list request.

…_content - file_content.content_bytes() encodes text as UTF-8. The checkpoint and the send queue use it, instead of a copy each. - The send queue reads ContentType from file_content, instead of its own copy.

- With freeze_private_datasets on, the watcher does not change or delete a file that it already has in an immutable collection. - The enclave flow turns the flag on, so syft_enclaves.immutability is gone.

pjwerneck added 12 commits November 3, 2026 17:39

fix: refuse file content that is not valid base64

ff407d1

- decode_content() dropped characters outside the base64 alphabet and returned other bytes, with no error. It now raises, so the model refuses the content.

fix: give the Drive watcher store its own Drive client in another thread

2e13bb9

- DriveWatcherStoreConnection.clone_for_thread() returned the same object, so another thread would share one Drive client. It now returns a store over a new client, as the other Drive connections do.

github-actions Bot added feature pkg:syft labels Nov 4, 2026

koenvanderveen reviewed Nov 9, 2026

View reviewed changes

pjwerneck added 3 commits November 10, 2026 00:08

Merge branch 'pjwerneck/ds-spool' into pjwerneck/ds-watcher-storage

928ddec

refactor: restore the checkpoint and the rolling state of a DO in two…

a0e19ed

… helpers - _restore_watcher_state() calls _restore_checkpoint() and then _restore_rolling_state() for each DO with saved state. The checkpoint goes first, because the rolling state holds the changes after it.

refactor: name the in-memory rolling state of a DO _rolling_state_loc…

ed7b0f7

…al() - _rolling_state() is now _rolling_state_local(), so its name says that it returns the copy in memory. _save_rolling_state() writes that copy to the DS's storage.

pjwerneck added 7 commits November 10, 2026 19:40

refactor: share the text-to-bytes helper and ContentType through file…

dde6d12

…_content - file_content.content_bytes() encodes text as UTF-8. The checkpoint and the send queue use it, instead of a copy each. - The send queue reads ContentType from file_content, instead of its own copy.

Merge branch 'pjwerneck/ds-spool' into pjwerneck/ds-watcher-storage

35c100f

github-actions Bot added the pkg:syft-enclave label Nov 11, 2026

refactor: replace the pre_write_filter callback with a config flag

0cc48d2

- With freeze_private_datasets on, the watcher does not change or delete a file that it already has in an immutable collection. - The enclave flow turns the flag on, so syft_enclaves.immutability is gone.

		def _save_rolling_state(self, peer_email: str) -> None:
		key = _peer_key(peer_email)

		cache._load_collection_hashes_from_disk()
		cache._restore_watcher_state()


		# --- State saved in the DS's storage ---

		def _restore_watcher_state(self) -> None:

Conversation

pjwerneck commented Nov 4, 2026

Summary

Changes

Testing

Asana task

Uh oh!

Choose a reason for hiding this comment

Uh oh!

pjwerneck Nov 12, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

This branch has not been deployed

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

pjwerneck Nov 12, 2026 •

edited

Loading