diff --git a/src/langbot/pkg/api/http/controller/groups/user.py b/src/langbot/pkg/api/http/controller/groups/user.py index 85704d72a..af37172db 100644 --- a/src/langbot/pkg/api/http/controller/groups/user.py +++ b/src/langbot/pkg/api/http/controller/groups/user.py @@ -453,13 +453,28 @@ class UserRouterGroup(group.RouterGroup): ) break except WorkspaceNotFoundError: - if projection_service is None or attempt == 3: + if projection_service is None: raise - elif projection_service is None or attempt == 3: + elif projection_service is None: break + if attempt == 3: + break await projection_service.sync_once() account = await self.ap.user_service.get_user_by_uuid(launch['account_uuid']) + + if access is None and projection_service is not None: + # The target event may be deeper than the bounded incremental + # page budget. One authoritative signed snapshot catches this + # process up without turning the callback into unbounded polling. + await projection_service.refresh_snapshot() + account = await self.ap.user_service.get_user_by_uuid(launch['account_uuid']) + if account is not None: + self.ap.user_service._require_active_account(account) + access = await self.ap.workspace_collaboration_service.resolve_account_workspace( + account.uuid, + launch['workspace_uuid'], + ) if account is None: raise SpaceLaunchError('Launch Account is not projected into Core') if access is None: # pragma: no cover - bounded loop resolves or raises. diff --git a/src/langbot/pkg/cloud/directory_projection.py b/src/langbot/pkg/cloud/directory_projection.py index 47fad19fd..c5fd2147d 100644 --- a/src/langbot/pkg/cloud/directory_projection.py +++ b/src/langbot/pkg/cloud/directory_projection.py @@ -125,10 +125,21 @@ class DirectoryProjectionService: # The database cursor remains the shared projection high-water mark, # while this cursor tracks what this process has actually observed. self._consumer_cursor: int | None = None + self._sync_lock = asyncio.Lock() async def initialize(self) -> None: """Block Cloud startup until one full signed snapshot is committed.""" + async with self._sync_lock: + await self._refresh_snapshot() + + async def refresh_snapshot(self) -> None: + """Refresh from one full signed snapshot within the sync single-flight.""" + + async with self._sync_lock: + await self._refresh_snapshot() + + async def _refresh_snapshot(self) -> None: last_superseded: _DirectorySnapshotSuperseded | None = None for _attempt in range(5): snapshot = await self.provider.fetch_snapshot(self.instance_uuid) @@ -159,9 +170,13 @@ class DirectoryProjectionService: delay = min(max(delay * 2, self.sync_interval_seconds), self.max_staleness_seconds / 2) async def sync_once(self) -> None: + async with self._sync_lock: + await self._sync_once() + + async def _sync_once(self) -> None: cursor = self._consumer_cursor if cursor is None: - await self.initialize() + await self._refresh_snapshot() return batch = await self.provider.fetch_events( self.instance_uuid, diff --git a/tests/integration/api/test_user_space_oauth.py b/tests/integration/api/test_user_space_oauth.py index 10a48a8c7..34c09451b 100644 --- a/tests/integration/api/test_user_space_oauth.py +++ b/tests/integration/api/test_user_space_oauth.py @@ -486,3 +486,36 @@ async def test_direct_launch_refreshes_projection_when_account_exists_before_wor assert (await response.get_json())['data']['workspace_uuid'] == WORKSPACE_UUID application.directory_projection_service.sync_once.assert_awaited_once_with() assert application.workspace_collaboration_service.resolve_account_workspace.await_count == 2 + + +@pytest.mark.asyncio +async def test_direct_launch_falls_back_to_snapshot_when_event_backlog_exceeds_page_budget(space_oauth_api): + application, client = space_oauth_api + projected_access = application.workspace_collaboration_service.resolve_account_workspace.return_value + application.workspace_collaboration_service.resolve_account_workspace = AsyncMock( + side_effect=[ + WorkspaceNotFoundError('Workspace not found'), + WorkspaceNotFoundError('Workspace not found'), + WorkspaceNotFoundError('Workspace not found'), + WorkspaceNotFoundError('Workspace not found'), + projected_access, + ] + ) + application.directory_projection_service = SimpleNamespace( + sync_once=AsyncMock(), + refresh_snapshot=AsyncMock(), + ) + + response = await client.post( + '/api/v1/user/space/callback', + json={ + 'workspace_uuid': WORKSPACE_UUID, + 'launch_assertion': 'signed-launch-token', + }, + ) + + assert response.status_code == 200 + assert (await response.get_json())['data']['workspace_uuid'] == WORKSPACE_UUID + assert application.directory_projection_service.sync_once.await_count == 3 + application.directory_projection_service.refresh_snapshot.assert_awaited_once_with() + assert application.workspace_collaboration_service.resolve_account_workspace.await_count == 5 diff --git a/tests/unit_tests/cloud/test_directory_projection.py b/tests/unit_tests/cloud/test_directory_projection.py index 69619a109..1320a9abc 100644 --- a/tests/unit_tests/cloud/test_directory_projection.py +++ b/tests/unit_tests/cloud/test_directory_projection.py @@ -1,5 +1,6 @@ from __future__ import annotations +import asyncio import datetime import logging from types import SimpleNamespace @@ -735,6 +736,72 @@ async def test_each_replica_consumes_events_with_its_own_cursor(projection_conte assert second_provider.after_cursors == [1, 2] +async def test_concurrent_sync_once_calls_are_serialized_per_service(projection_context): + application, _session_factory = projection_context + + class _ConcurrentProvider(_Provider): + def __init__(self) -> None: + super().__init__([_snapshot(1)]) + self.first_fetch_started = asyncio.Event() + self.release_first_fetch = asyncio.Event() + self.active_fetches = 0 + self.max_active_fetches = 0 + + async def fetch_events( + self, + instance_uuid: str, + after_cursor: int, + limit: int, + ) -> DirectoryEventBatch: + assert instance_uuid == INSTANCE_UUID + assert limit == 100 + self.after_cursors.append(after_cursor) + self.active_fetches += 1 + self.max_active_fetches = max(self.max_active_fetches, self.active_fetches) + try: + if len(self.after_cursors) == 1: + self.first_fetch_started.set() + await self.release_first_fetch.wait() + cursor = after_cursor + 1 + return DirectoryEventBatch( + instance_uuid=instance_uuid, + after_cursor=after_cursor, + cursor=cursor, + high_water_cursor=cursor, + events=( + DirectoryEvent( + cursor=cursor, + uuid=f'40000000-0000-4000-8000-{cursor:012d}', + aggregate_uuid=WORKSPACE_UUID, + event_type='entitlement.changed', + revision=cursor, + payload={ + 'workspace_uuid': WORKSPACE_UUID, + 'entitlement_revision': cursor, + }, + created_at=datetime.datetime(2026, 7, 24, 12, cursor, tzinfo=datetime.UTC), + ), + ), + ) + finally: + self.active_fetches -= 1 + + provider = _ConcurrentProvider() + service = DirectoryProjectionService(application, provider, INSTANCE_UUID) + await service.initialize() + + first = asyncio.create_task(service.sync_once()) + await provider.first_fetch_started.wait() + second = asyncio.create_task(service.sync_once()) + await asyncio.sleep(0) + provider.release_first_fetch.set() + await asyncio.gather(first, second) + + assert provider.max_active_fetches == 1 + assert provider.after_cursors == [1, 2] + assert service._consumer_cursor == 3 + + async def test_snapshot_coverage_allows_lagging_replica_to_replay_receipts(projection_context): application, session_factory = projection_context event_two = DirectoryEvent(