feat: isolate concurrent user syncs
Wrap SyncManager.sync_user in a per-user asyncio.Lock (raising SyncAlreadyRunning on overlap) and add sync_all_enabled() to fan out across all enabled users with per-user failure isolation via asyncio.gather(return_exceptions=True). The prior sync_user body is renamed to _sync_user_locked with no logic changes.
This commit is contained in:
@@ -17,12 +17,18 @@ from app.sync.states import SyncOutcome
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class SyncAlreadyRunning(RuntimeError):
|
||||
pass
|
||||
|
||||
|
||||
class SyncManager:
|
||||
"""Resumable single-user MyWhoosh -> Garmin sync pipeline.
|
||||
|
||||
Locking/scheduling across multiple users is layered on top of `sync_user`
|
||||
by a later task; this class only implements the state machine for one
|
||||
user's sync run.
|
||||
`_sync_user_locked` implements the state machine for one user's sync run.
|
||||
`sync_user` wraps it with a per-user `asyncio.Lock` so only one sync can
|
||||
run for a given user at a time (raising `SyncAlreadyRunning` on overlap),
|
||||
and `sync_all_enabled` fans out across all enabled users, isolating each
|
||||
user's failure to its own result rather than cancelling siblings.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
@@ -41,8 +47,32 @@ class SyncManager:
|
||||
self.mywhoosh_factory = mywhoosh_factory
|
||||
self.garmin_factory = garmin_factory
|
||||
self.fit_converter = fit_converter
|
||||
self._locks: dict[int, asyncio.Lock] = {}
|
||||
self._locks_guard = asyncio.Lock()
|
||||
|
||||
async def _lock_for(self, user_id: int) -> asyncio.Lock:
|
||||
async with self._locks_guard:
|
||||
return self._locks.setdefault(user_id, asyncio.Lock())
|
||||
|
||||
async def sync_user(self, user_id: int, mfa_code: str | None = None) -> SyncOutcome:
|
||||
lock = await self._lock_for(user_id)
|
||||
if lock.locked():
|
||||
raise SyncAlreadyRunning(f"sync already running for user {user_id}")
|
||||
async with lock:
|
||||
return await self._sync_user_locked(user_id, mfa_code)
|
||||
|
||||
def _load_enabled_user_ids(self) -> list[int]:
|
||||
with self.session_factory() as session:
|
||||
return [user.id for user in UserRepository(session).list_enabled()]
|
||||
|
||||
async def sync_all_enabled(self) -> list[SyncOutcome | Exception]:
|
||||
user_ids = self._load_enabled_user_ids()
|
||||
return await asyncio.gather(
|
||||
*(self.sync_user(user_id) for user_id in user_ids),
|
||||
return_exceptions=True,
|
||||
)
|
||||
|
||||
async def _sync_user_locked(self, user_id: int, mfa_code: str | None = None) -> SyncOutcome:
|
||||
with self.session_factory() as session:
|
||||
user = UserRepository(session).get(user_id)
|
||||
if user is None:
|
||||
|
||||
Reference in New Issue
Block a user