Skip to content

refactor: share one connection router across a manager's parts #9597

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
koenvanderveen merged 1 commit into from
Oct 29, 2026
+160 −113
Merged

refactor: share one connection router across a manager's parts #9597

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 7 additions & 2 deletions syft/sync/connections/connection_router.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,13 +61,18 @@ def transport(self) -> Transport:
return self._transport

@classmethod
def from_configs(cls, email: str, connection_configs: List[ConnectionConfig]):
def from_configs(
cls, connection_configs: List[ConnectionConfig], peer_store: PeerStore
) -> "ConnectionRouter":
"""The one router of a manager. ``SyftboxManager.from_config`` builds
it and hands it to every part, so they share one connection, one
transport and one peer store."""
return cls(
connections=[
SyftboxPlatformConnection.from_config(config)
for config in connection_configs
],
peer_store=PeerStore(email=email),
peer_store=peer_store,
)

def add_connection(self, connection: SyftboxPlatformConnection):
Expand Down
101 changes: 36 additions & 65 deletions syft/sync/syftbox_manager.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@

from syft.sync.callback_mixin import BaseModelCallbackMixin
from syft.sync.connections.base_connection import (
ConnectionConfig,
SyftboxPlatformConnection,
)
from syft.sync.connections.connection_router import ConnectionRouter
Expand Down Expand Up @@ -54,6 +55,7 @@
random_email,
random_syftbox_folder_for_testing,
)
from syft.sync.peers.peer_store import PeerStore
from syft.sync.version.peer_manager import (
PeerManager,
PeerManagerConfig,
Expand Down Expand Up @@ -122,6 +124,10 @@ class SyftboxManagerConfig(BaseModel):
# log, rolling state, checkpoints. Off for datasites that can never
# restore (see syft-enclave's EnclaveSettings).
persist_owner_state: bool = True
# The backends to connect to. Only the manager reads them: it builds one
# connection router from them and shares it with every part. Empty for
# tests that add a mock connection afterwards (see _add_connection).
connection_configs: List[ConnectionConfig] = []

datasite_owner_syncer_config: DatasiteOwnerSyncerConfig
peer_manager_config: PeerManagerConfig
Expand Down Expand Up @@ -163,7 +169,6 @@ def for_colab(
syftbox_folder=syftbox_folder,
collections_folder=collections_folder,
collection_specs=collection_specs,
connection_configs=connection_configs,
cache_config=DataSiteOwnerEventCacheConfig(
email=email,
use_in_memory_cache=use_in_memory_cache,
Expand All @@ -175,19 +180,16 @@ def for_colab(
datasite_watcher_syncer_config = DatasiteWatcherSyncerConfig(
syftbox_folder=syftbox_folder,
email=email,
connection_configs=connection_configs,
datasite_watcher_cache_config=DataSiteWatcherCacheConfig(
email=email,
use_in_memory_cache=use_in_memory_cache,
syftbox_folder=syftbox_folder,
collection_specs=collection_specs,
connection_configs=connection_configs,
),
)
peer_manager_config = PeerManagerConfig(
syftbox_folder=syftbox_folder,
email=email,
connection_configs=connection_configs,
has_do_role=has_do_role,
has_ds_role=has_ds_role,
encryption_policy=EncryptionPolicy.from_bool(encryption),
Expand All @@ -197,9 +199,9 @@ def for_colab(
return cls(
email=email,
syftbox_folder=syftbox_folder,
connection_configs=connection_configs,
has_ds_role=has_ds_role,
has_do_role=has_do_role,
connection_configs=connection_configs,
use_in_memory_cache=False,
persist_owner_state=persist_owner_state,
datasite_owner_syncer_config=datasite_owner_syncer_config,
Expand Down Expand Up @@ -246,32 +248,27 @@ def for_jupyter(
syftbox_folder=syftbox_folder,
collections_folder=collections_folder,
collection_specs=collection_specs,
connection_configs=connection_configs,
cache_config=DataSiteOwnerEventCacheConfig(
email=email,
use_in_memory_cache=False,
syftbox_folder=syftbox_folder,
collections_folder=collections_folder,
collection_variants=collection_variants,
connection_configs=connection_configs,
),
)
datasite_watcher_syncer_config = DatasiteWatcherSyncerConfig(
syftbox_folder=syftbox_folder,
email=email,
connection_configs=connection_configs,
datasite_watcher_cache_config=DataSiteWatcherCacheConfig(
email=email,
use_in_memory_cache=False,
syftbox_folder=syftbox_folder,
collection_specs=collection_specs,
connection_configs=connection_configs,
),
)
peer_manager_config = PeerManagerConfig(
syftbox_folder=syftbox_folder,
email=email,
connection_configs=connection_configs,
has_do_role=has_do_role,
has_ds_role=has_ds_role,
encryption_policy=EncryptionPolicy.from_bool(encryption),
Expand All @@ -281,6 +278,7 @@ def for_jupyter(
return cls(
email=email,
syftbox_folder=syftbox_folder,
connection_configs=connection_configs,
has_ds_role=has_ds_role,
has_do_role=has_do_role,
use_in_memory_cache=False,
Expand Down Expand Up @@ -346,7 +344,6 @@ def _base_config_for_testing(
peer_manager_config = PeerManagerConfig(
syftbox_folder=syftbox_folder,
email=email,
connection_configs=[], # Empty for in-memory, connections added later
n_threads=2, # Use fewer threads for testing
force_ignore_peer_version=not check_versions,
has_do_role=has_do_role,
Expand Down Expand Up @@ -402,7 +399,6 @@ def for_google_drive_testing_connection(
syftbox_folder=syftbox_folder,
collections_folder=collections_folder,
collection_specs=collection_specs,
connection_configs=connection_configs,
cache_config=DataSiteOwnerEventCacheConfig(
email=email,
use_in_memory_cache=use_in_memory_cache,
Expand All @@ -414,27 +410,25 @@ def for_google_drive_testing_connection(
datasite_watcher_syncer_config = DatasiteWatcherSyncerConfig(
syftbox_folder=syftbox_folder,
email=email,
connection_configs=connection_configs,
datasite_watcher_cache_config=DataSiteWatcherCacheConfig(
email=email,
use_in_memory_cache=use_in_memory_cache,
syftbox_folder=syftbox_folder,
collection_specs=collection_specs,
connection_configs=connection_configs,
),
)

peer_manager_config = PeerManagerConfig(
syftbox_folder=syftbox_folder,
email=email,
connection_configs=connection_configs,
force_ignore_peer_version=not check_versions,
has_do_role=has_do_role,
has_ds_role=has_ds_role,
)
return cls(
email=email,
syftbox_folder=syftbox_folder,
connection_configs=connection_configs,
write_files=write_files,
persist_owner_state=persist_owner_state,
datasite_owner_syncer_config=datasite_owner_syncer_config,
Expand Down Expand Up @@ -466,7 +460,6 @@ class SyftboxManager(BaseModelCallbackMixin):
_executor: ThreadPoolExecutor = PrivateAttr(
default_factory=lambda: ThreadPoolExecutor(max_workers=10)
)
_peer_store: object = PrivateAttr(default=None)

_PUBLIC_API = (
"email",
Expand Down Expand Up @@ -524,6 +517,20 @@ def peers(self) -> PeerList:

@classmethod
def from_config(cls, config: SyftboxManagerConfig):
"""Build a manager and its parts from ``config``.

Configs hold plain values only (strings, paths, flags), never live
objects or state. A live object that several parts share is built here,
once, and passed to each part's ``from_config`` next to its config:
the peer store and the connection router that holds it, and through
the router one connection and one transport for the whole manager.
"""
peer_store = PeerManager.peer_store_from_config(
config.peer_manager_config, email=config.email
)
connection_router = ConnectionRouter.from_configs(
config.connection_configs, peer_store
)
file_writer = FileWriter(
base_path=config.syftbox_folder, write_files=config.write_files
)
Expand All @@ -534,18 +541,18 @@ def from_config(cls, config: SyftboxManagerConfig):

if config.has_do_role:
datasite_owner_syncer = DatasiteOwnerSyncer.from_config(
config.datasite_owner_syncer_config
config.datasite_owner_syncer_config, connection_router
)

job_file_change_handler = JobFileChangeHandler()

if config.has_ds_role:
datasite_watcher_syncer = DatasiteWatcherSyncer.from_config(
config.datasite_watcher_syncer_config
config.datasite_watcher_syncer_config, connection_router
)

peer_manager = PeerManager.from_config(
config.peer_manager_config, email=config.email
config.peer_manager_config, connection_router
)

manager_res = cls(
Expand All @@ -560,30 +567,8 @@ def from_config(cls, config: SyftboxManagerConfig):
has_ds_role=config.has_ds_role,
config=config,
)

# PeerManager.from_config built the peer store from peer_manager_config;
# share that single store across every syncer's router so all sync paths
# see the same peers, peer policies and keys.
manager_res._set_peer_store(peer_manager.peer_store)

return manager_res

def _set_peer_store(self, peer_store) -> None:
"""Wire shared peer_store into all connection routers."""
from syft.sync.peers.peer_store import PeerStore

if not isinstance(peer_store, PeerStore):
return
self._peer_store = peer_store
if self.peer_manager:
self.peer_manager.peer_store = peer_store
self.peer_manager.connection_router.peer_store = peer_store
if self.datasite_owner_syncer:
self.datasite_owner_syncer.connection_router.peer_store = peer_store
if self.datasite_watcher_syncer:
self.datasite_watcher_syncer.connection_router.peer_store = peer_store
self.datasite_watcher_syncer.datasite_watcher_cache.connection_router.peer_store = peer_store

@classmethod
def for_colab(
cls,
Expand Down Expand Up @@ -1088,16 +1073,7 @@ def _add_connection(self, connection: SyftboxPlatformConnection):
"Only MockDriveService connections can be added to the manager"
)

if self.datasite_owner_syncer is not None:
self.datasite_owner_syncer.connection_router.add_connection(connection)
if self.datasite_watcher_syncer is not None:
self.datasite_watcher_syncer.connection_router.add_connection(connection)
self.datasite_watcher_syncer.datasite_watcher_cache.connection_router.add_connection(
connection
)

# Add connection to version manager's router
self.peer_manager.connection_router.add_connection(connection)
self._connection_router.add_connection(connection)

def _send_file_change(self, path: str | Path, content: str):
self.file_writer.write_file(path, content)
Expand All @@ -1109,22 +1085,17 @@ def _get_all_accepted_events_do(self) -> List[FileChangeEventsMessage]:

@property
def _connection_router(self) -> ConnectionRouter:
# for DOs we have a syncer, for DSs we have a watcher syncer
if self.datasite_owner_syncer is not None:
return self.datasite_owner_syncer.connection_router
else:
return self.datasite_watcher_syncer.connection_router
# Every part holds the same router (see from_config); the peer manager
# is the one part every manager has.
return self.peer_manager.connection_router

@property
def _peer_store(self) -> PeerStore:
return self._connection_router.peer_store

def reset_all_connection_caches(self):
"""Reset GDrive caches on all connection router instances."""
if self.peer_manager:
self.peer_manager.connection_router.reset_caches()
if self.datasite_owner_syncer:
self.datasite_owner_syncer.connection_router.reset_caches()
if self.datasite_watcher_syncer:
self.datasite_watcher_syncer.connection_router.reset_caches()
if hasattr(self.datasite_watcher_syncer, "datasite_watcher_cache"):
self.datasite_watcher_syncer.datasite_watcher_cache.connection_router.reset_caches()
"""Reset the Drive caches of the manager's one connection router."""
self._connection_router.reset_caches()

def _clear_caches(self):
if self.datasite_owner_syncer is not None:
Expand Down
20 changes: 9 additions & 11 deletions syft/sync/sync/caches/datasite_watcher_cache.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,6 @@
FileChangeEventsMessage,
)
from syft.sync.connections.connection_router import ConnectionRouter
from syft.sync.connections.base_connection import ConnectionConfig
from syft.sync.sync.caches.cache_file_writer_connection import (
CacheFileConnection,
InMemoryCacheFileConnection,
Expand Down Expand Up @@ -58,7 +57,6 @@ class DataSiteWatcherCacheConfig(BaseModel):
use_in_memory_cache: bool = True
syftbox_folder: Path | None = None
events_base_path: Path | None = None
connection_configs: List[ConnectionConfig] = []
# Collections to sync down (prefix + local subpath per collection type)
collection_specs: list[CollectionSyncSpec] = []

Expand Down Expand Up @@ -90,15 +88,18 @@ class DataSiteWatcherCache(BaseModel):
pre_write_filter: Callable[[str, bool], bool] | None = None

@classmethod
def from_config(cls, config: DataSiteWatcherCacheConfig):
def from_config(
cls, config: DataSiteWatcherCacheConfig, connection_router: ConnectionRouter
):
"""``config`` holds plain values only. ``connection_router`` is live and
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``)."""
if config.use_in_memory_cache:
res = cls(
events_connection=InMemoryCacheFileConnection[FileChangeEvent](),
file_connection=InMemoryCacheFileConnection[str](),
connection_router=ConnectionRouter.from_configs(
email=config.email,
connection_configs=config.connection_configs,
),
connection_router=connection_router,
syftbox_folder=config.syftbox_folder,
collection_specs=config.collection_specs,
)
Expand All @@ -116,10 +117,7 @@ def from_config(cls, config: DataSiteWatcherCacheConfig):
base_dir=events_folder, dtype=FileChangeEventsMessage
),
file_connection=FSFileConnection(base_dir=config.syftbox_folder),
connection_router=ConnectionRouter.from_configs(
email=config.email,
connection_configs=config.connection_configs,
),
connection_router=connection_router,
syftbox_folder=config.syftbox_folder,
collection_specs=config.collection_specs,
)
Expand Down
Loading
Loading