diff --git a/app/sync/manager.py b/app/sync/manager.py index b4e5e6c..78b37fc 100644 --- a/app/sync/manager.py +++ b/app/sync/manager.py @@ -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: diff --git a/tests/sync/test_concurrency.py b/tests/sync/test_concurrency.py new file mode 100644 index 0000000..cfcc01c --- /dev/null +++ b/tests/sync/test_concurrency.py @@ -0,0 +1,163 @@ +import asyncio + +import pytest + +from app.db.repositories import UserRepository +from app.mywhoosh.client import MyWhooshAuthError +from app.mywhoosh.models import MyWhooshActivity +from app.sync.manager import SyncAlreadyRunning, SyncManager +from tests.sync.conftest import FakeFitConverter, _create_user +from tests.sync.fakes import FakeGarminUploader + + +class BlockingMyWhooshClient: + """Fake MyWhoosh client whose list_activities() blocks on test-controlled + events, so a test can deterministically observe "sync has started but not + finished" without any production-only test hooks.""" + + def __init__(self, first_started: asyncio.Event, release: asyncio.Event) -> None: + self.first_started = first_started + self.release = release + self.list_calls = 0 + + async def list_activities(self, email: str, password: str): + self.list_calls += 1 + self.first_started.set() + await self.release.wait() + return [] + + async def download_fit(self, activity_file_id: str, email: str, password: str) -> bytes: + raise AssertionError("download_fit should not be reached in this test") + + +class ConditionalFailureMyWhooshClient: + """Fake MyWhoosh client that raises MyWhooshAuthError only for a specific + account email, letting one user's sync fail while others succeed.""" + + def __init__(self, activities, fit_bytes: bytes, failing_email: str) -> None: + self.activities = activities + self.fit_bytes = fit_bytes + self.failing_email = failing_email + self.list_calls = 0 + self.download_calls = 0 + + async def list_activities(self, email: str, password: str): + self.list_calls += 1 + if email == self.failing_email: + raise MyWhooshAuthError("simulated auth failure") + return list(self.activities) + + async def download_fit(self, activity_file_id: str, email: str, password: str) -> bytes: + self.download_calls += 1 + return self.fit_bytes + + +@pytest.fixture +def user_a(session_factory, cipher): + with session_factory() as session: + return _create_user(session, cipher) + + +@pytest.fixture +def user_b(session_factory, cipher): + with session_factory() as session: + return _create_user(session, cipher) + + +@pytest.mark.asyncio +async def test_same_user_cannot_run_twice(session_factory, cipher, settings, seeded_user) -> None: + first_started = asyncio.Event() + release_first = asyncio.Event() + mywhoosh = BlockingMyWhooshClient(first_started, release_first) + manager = SyncManager( + session_factory=session_factory, + credential_cipher=cipher, + settings=settings, + mywhoosh_factory=lambda token_store: mywhoosh, + garmin_factory=lambda email, password, tokenstore: FakeGarminUploader(), + fit_converter=FakeFitConverter(), + ) + + first = asyncio.create_task(manager.sync_user(seeded_user.id)) + await first_started.wait() + + with pytest.raises(SyncAlreadyRunning): + await manager.sync_user(seeded_user.id) + + release_first.set() + outcome = await first + + assert outcome.user_id == seeded_user.id + assert mywhoosh.list_calls == 1 + + +@pytest.mark.asyncio +async def test_different_users_can_run_concurrently(manager, user_a, user_b) -> None: + results = await asyncio.gather(manager.sync_user(user_a.id), manager.sync_user(user_b.id)) + assert {result.user_id for result in results} == {user_a.id, user_b.id} + assert all(result.status == "success" for result in results) + + +@pytest.mark.asyncio +async def test_sync_all_enabled_isolates_failures(session_factory, cipher, settings) -> None: + with session_factory() as session: + failing_user = UserRepository(session).create( + name="Failing User", + enabled=True, + mywhoosh_email_enc=cipher.encrypt("failing-mw@example.com"), + mywhoosh_password_enc=cipher.encrypt("failing-mw-pass"), + garmin_email_enc=cipher.encrypt("failing-garmin@example.com"), + garmin_password_enc=cipher.encrypt("failing-garmin-pass"), + ) + healthy_user = UserRepository(session).create( + name="Healthy User", + enabled=True, + mywhoosh_email_enc=cipher.encrypt("healthy-mw@example.com"), + mywhoosh_password_enc=cipher.encrypt("healthy-mw-pass"), + garmin_email_enc=cipher.encrypt("healthy-garmin@example.com"), + garmin_password_enc=cipher.encrypt("healthy-garmin-pass"), + ) + + remote = MyWhooshActivity( + id="mw-shared", + title="Shared Ride", + activity_file_id="file-mw-shared", + started_at=None, + ) + mywhoosh = ConditionalFailureMyWhooshClient( + activities=[remote], + fit_bytes=b"source-bytes", + failing_email="failing-mw@example.com", + ) + converter = FakeFitConverter() + garmin = FakeGarminUploader() + manager = SyncManager( + session_factory=session_factory, + credential_cipher=cipher, + settings=settings, + mywhoosh_factory=lambda token_store: mywhoosh, + garmin_factory=lambda email, password, tokenstore: garmin, + fit_converter=converter, + ) + + results = await manager.sync_all_enabled() + + assert len(results) == 2 + for result in results: + assert not isinstance(result, Exception) + + by_user = {result.user_id: result for result in results} + failing_outcome = by_user[failing_user.id] + healthy_outcome = by_user[healthy_user.id] + + # The failing user's auth error is classified by _sync_user_locked's own + # exception handling and returned as a non-success SyncOutcome rather than + # raised -- so asyncio.gather never sees an exception for this failure + # mode. It must not affect the healthy user's independent outcome. + assert failing_outcome.status != "success" + assert failing_outcome.message is not None + assert failing_outcome.discovered == 0 + + assert healthy_outcome.status == "success" + assert healthy_outcome.imported == 1 + assert healthy_outcome.failed == 0