From d8e3e9bc3354fe291b1d2b6c544670fc4b5a3d6f Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Fri, 12 Jun 2026 09:48:56 +0200 Subject: [PATCH 1/4] feat(vector-sync): scan provisioned users immediately Background vector sync discovered newly provisioned users only on the periodic user-manager poll (VECTOR_SYNC_USER_POLL_INTERVAL, default 60s), delaying first indexing by up to a minute. Add a ProvisionSignal doorbell that provisioning paths ring after storing a user's app password, waking user_manager_task to re-poll and spawn the user's scanner at once. The periodic poll remains the backstop (covers cross-replica provisioning). - ProvisionSignal (stable reference, wait-and-re-arm) held on VectorSyncState; closes the lost-wakeup window (no await between observing the ring and re-arming; anyio.Event stickiness covers a mid-poll ring) - user_manager_task races its poll timeout against the doorbell + shutdown - notify_user_provisioned() rung from the three app-password provisioning sites: Login Flow v2 web, MCP provisioning tool, management/BasicAuth API Note: the pre-existing scanner_wake_event was never .set() and only wakes existing scanners; a brand-new user has none, so the manager is what must be nudged. Co-Authored-By: Claude Opus 4.8 (1M context) --- nextcloud_mcp_server/api/passwords.py | 6 + nextcloud_mcp_server/app.py | 32 +++ nextcloud_mcp_server/auth/provision_routes.py | 8 + nextcloud_mcp_server/server/auth_tools.py | 8 + nextcloud_mcp_server/vector/oauth_sync.py | 59 ++++- .../test_login_flow_provision_wake.py | 179 +++++++++++++++ .../test_user_manager_provision_wake.py | 213 ++++++++++++++++++ 7 files changed, 503 insertions(+), 2 deletions(-) create mode 100644 tests/integration/test_login_flow_provision_wake.py create mode 100644 tests/unit/vector/test_user_manager_provision_wake.py diff --git a/nextcloud_mcp_server/api/passwords.py b/nextcloud_mcp_server/api/passwords.py index 2cf365a0..41f94f1a 100644 --- a/nextcloud_mcp_server/api/passwords.py +++ b/nextcloud_mcp_server/api/passwords.py @@ -407,6 +407,12 @@ async def provision_app_password(request: Request) -> JSONResponse: username, app_password, scopes=scopes, username=nc_username ) invalidate_scope_cache(username) + # Wake the background sync user manager so this user's scanner starts + # now instead of after the next poll. Local import avoids an app <-> + # api-module import cycle. + from nextcloud_mcp_server.app import notify_user_provisioned # noqa: PLC0415 + + notify_user_provisioned() _record_rate_limit_attempt(path_user_id, success=True) logger.info("Provisioned app password for user: %s", username) diff --git a/nextcloud_mcp_server/app.py b/nextcloud_mcp_server/app.py index f2a9377e..98a37d08 100644 --- a/nextcloud_mcp_server/app.py +++ b/nextcloud_mcp_server/app.py @@ -128,6 +128,7 @@ from nextcloud_mcp_server.server.auth_tools import register_auth_tools from nextcloud_mcp_server.server.oauth_tools import register_oauth_tools from nextcloud_mcp_server.vector.metrics_publisher import vector_sync_metrics_task from nextcloud_mcp_server.vector.oauth_sync import ( + ProvisionSignal, oauth_processor_task, user_manager_task, ) @@ -344,6 +345,12 @@ class VectorSyncState: task_producer: "TaskProducer | None" = None shutdown_event: anyio.Event | None = None scanner_wake_event: anyio.Event | None = None + # Rung by a provisioning request to wake ``user_manager_task`` immediately so + # a just-provisioned user's scanner is spawned without waiting out the + # ``VECTOR_SYNC_USER_POLL_INTERVAL`` poll. ``None`` when no user manager is + # running (single-user mode or vector sync disabled), in which case + # ``notify_user_provisioned`` is a no-op. + provision_signal: "ProvisionSignal | None" = None # Long-lived task group used for fire-and-forget background work spawned # from the request path (e.g. ADR-019 verify-on-read eviction). Set by the # starlette lifespan after entering its task group; cleared on shutdown. @@ -354,6 +361,23 @@ class VectorSyncState: _vector_sync_state = VectorSyncState() +def notify_user_provisioned() -> None: + """Wake the user manager to discover a just-provisioned user immediately. + + Provisioning call sites invoke this after a successful app-password store so + ``user_manager_task`` re-polls at once instead of waiting out + ``VECTOR_SYNC_USER_POLL_INTERVAL``. The 60s poll remains the backstop, so a + missed signal (e.g. provisioning handled on a different replica than the + manager) only delays the scan, never skips it. + + No-op when ``provision_signal`` is ``None`` — single-user mode or vector + sync disabled, where no user manager is running. + """ + signal = _vector_sync_state.provision_signal + if signal is not None: + signal.ring() + + def _wire_vector_sync_state( app: Starlette, transport: IngestTransport, @@ -416,6 +440,7 @@ def _clear_vector_sync_state() -> None: # closed lifespan; the next startup's _wire_vector_sync_state rebinds them. _vector_sync_state.shutdown_event = None _vector_sync_state.scanner_wake_event = None + _vector_sync_state.provision_signal = None # ============================================================================= @@ -2060,6 +2085,12 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = # that choice — this path is now backend-agnostic. shutdown_event = anyio.Event() scanner_wake_event = anyio.Event() + # Doorbell the provisioning request path rings (via + # notify_user_provisioned) to wake the user manager immediately + # for a newly provisioned user. Held on the singleton only — both + # the manager and the signal helper reach it there. + provision_signal = ProvisionSignal() + _vector_sync_state.provision_signal = provision_signal # User state tracking for user manager user_states: dict = {} @@ -2095,6 +2126,7 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = nextcloud_host_for_sync, user_states, tg, + provision_signal, ) # In-process consumer pool. ``run_consumers`` is a no-op for diff --git a/nextcloud_mcp_server/auth/provision_routes.py b/nextcloud_mcp_server/auth/provision_routes.py index b83525f1..7fa2e92f 100644 --- a/nextcloud_mcp_server/auth/provision_routes.py +++ b/nextcloud_mcp_server/auth/provision_routes.py @@ -118,6 +118,14 @@ async def _poll_and_store(provision_id: str) -> None: username=result.login_name, ) invalidate_scope_cache(effective_user_id) + # Wake the background sync user manager so this user's scanner + # starts now instead of after the next poll. Local import avoids an + # app <-> route-module import cycle. + from nextcloud_mcp_server.app import ( # noqa: PLC0415 + notify_user_provisioned, + ) + + notify_user_provisioned() session = _provision_sessions.get(provision_id) if session: session["status"] = "completed" diff --git a/nextcloud_mcp_server/server/auth_tools.py b/nextcloud_mcp_server/server/auth_tools.py index 8b5b22aa..c9ce59b4 100644 --- a/nextcloud_mcp_server/server/auth_tools.py +++ b/nextcloud_mcp_server/server/auth_tools.py @@ -288,6 +288,14 @@ def register_auth_tools(mcp: FastMCP) -> None: username=poll_result.login_name, ) invalidate_scope_cache(user_id) + # Wake the background sync user manager so this user's scanner + # starts now instead of after the next poll. Local import avoids an + # app <-> server-module import cycle. + from nextcloud_mcp_server.app import ( # noqa: PLC0415 + notify_user_provisioned, + ) + + notify_user_provisioned() # Clean up the flow session await storage.delete_login_flow_session(user_id) diff --git a/nextcloud_mcp_server/vector/oauth_sync.py b/nextcloud_mcp_server/vector/oauth_sync.py index 6cf44564..485bdaa3 100644 --- a/nextcloud_mcp_server/vector/oauth_sync.py +++ b/nextcloud_mcp_server/vector/oauth_sync.py @@ -44,6 +44,40 @@ class NotProvisionedError(Exception): pass +class ProvisionSignal: + """One-shot doorbell that wakes ``user_manager_task`` on demand. + + A provisioning request rings this (``ring()``) right after storing a new + user's app password so the manager re-polls immediately instead of waiting + out ``VECTOR_SYNC_USER_POLL_INTERVAL``. The manager parks on ``wait()``; + each ring releases exactly one wait, after which the underlying event is + re-armed for the next cycle. + + The reference is stable for the life of the lifespan (stored once on the + ``VectorSyncState`` singleton), so the manager never has to republish a new + event back to shared state — avoiding any ``app`` ↔ ``vector`` import cycle. + + Concurrency: ``anyio.Event`` is sticky, so a ``ring()`` that lands before + ``wait()`` is still observed. ``wait()`` re-arms with no ``await`` between + observing the set and swapping the event, so under cooperative scheduling a + concurrent ``ring()`` cannot slip into that window and be lost. + """ + + def __init__(self) -> None: + self._event = anyio.Event() + + def ring(self) -> None: + """Signal a pending wait (or the next one to arrive).""" + self._event.set() + + async def wait(self) -> None: + """Block until the next ring, then re-arm for the following cycle.""" + await self._event.wait() + # No await before the swap: a concurrent ring() cannot interleave here, + # so it lands on the fresh event and the next wait() observes it. + self._event = anyio.Event() + + # Process-wide app-password storage for the BasicAuth client path. # # get_user_client_basic_auth is on the search hot path (Unified Search and the @@ -408,6 +442,7 @@ async def user_manager_task( nextcloud_host: str, user_states: dict[str, UserSyncState], tg: TaskGroup, + provision_signal: "ProvisionSignal", *, task_status: TaskStatus = anyio.TASK_STATUS_IGNORED, ) -> None: @@ -417,6 +452,12 @@ async def user_manager_task( - New users who have provisioned access -> start scanner - Users who have revoked access -> cancel their scanner + Polls every ``VECTOR_SYNC_USER_POLL_INTERVAL`` seconds, but also wakes + early whenever ``provision_signal`` is rung (by a provisioning request via + ``notify_user_provisioned``) so a just-provisioned user's scanner starts at + once rather than after up to a full poll interval. The poll remains the + backstop for any missed ring. + Args: send_stream: Stream to send documents to processors shutdown_event: Event signaling shutdown @@ -425,6 +466,7 @@ async def user_manager_task( nextcloud_host: Nextcloud base URL user_states: Shared dict tracking active user scanners tg: Task group for spawning scanner tasks + provision_signal: Doorbell rung on provisioning to force an early re-poll task_status: Status object for signaling task readiness """ settings = get_settings() @@ -491,10 +533,23 @@ async def user_manager_task( exc_info=True, ) - # Sleep until next poll + # Sleep until the next poll tick, but wake early on shutdown or a + # provisioning signal so a just-provisioned user is discovered at once. + # Race both waits in a child task group; whichever fires first cancels + # the scope, ending the sleep. move_on_after caps it at poll_interval. + async def _wake_on(wait_fn, scope: anyio.CancelScope) -> None: + await wait_fn() + scope.cancel() + try: with anyio.move_on_after(poll_interval): - await shutdown_event.wait() + async with anyio.create_task_group() as wake_tg: + wake_tg.start_soon( + _wake_on, shutdown_event.wait, wake_tg.cancel_scope + ) + wake_tg.start_soon( + _wake_on, provision_signal.wait, wake_tg.cancel_scope + ) except anyio.get_cancelled_exc_class(): break diff --git a/tests/integration/test_login_flow_provision_wake.py b/tests/integration/test_login_flow_provision_wake.py new file mode 100644 index 00000000..f1f78ab4 --- /dev/null +++ b/tests/integration/test_login_flow_provision_wake.py @@ -0,0 +1,179 @@ +"""Integration test: Login Flow v2 provisioning wakes the background sync +user manager immediately. + +Wires the real login-flow web-provision path end to end at the component level +(no browser / container): + + _poll_and_store (provision_routes) + -> RefreshTokenStorage.store_app_password_with_scopes (real, temp DB) + -> notify_user_provisioned -> ProvisionSignal.ring + -> user_manager_task wakes, re-polls the same storage, spawns the scanner + +Only the Nextcloud-facing Login Flow v2 poll is mocked (returning "completed"); +everything in between is the real code. With a deliberately long poll interval, +the new user's scanner must still be spawned promptly — proving the wake came +from the provisioning signal, not the periodic poll. + +This is the Login Flow v2 deployment-mode counterpart to the multi-user +BasicAuth coverage in ``test_app_password_provisioning.py``. +""" + +import tempfile +import time +from pathlib import Path +from unittest.mock import AsyncMock, MagicMock, patch + +import anyio +import pytest +from cryptography.fernet import Fernet + +from nextcloud_mcp_server.auth.login_flow import LoginFlowPollResult +from nextcloud_mcp_server.auth.provision_routes import ( + _poll_and_store, + _provision_sessions, +) +from nextcloud_mcp_server.auth.storage import RefreshTokenStorage +from nextcloud_mcp_server.vector.oauth_sync import ProvisionSignal, user_manager_task + +pytestmark = pytest.mark.integration + + +@pytest.fixture +def encryption_key(): + return Fernet.generate_key().decode() + + +@pytest.fixture +async def temp_storage(encryption_key): + """Real RefreshTokenStorage backed by a temporary SQLite DB.""" + with tempfile.TemporaryDirectory() as tmpdir: + db_path = Path(tmpdir) / "test_login_flow_wake.db" + storage = RefreshTokenStorage( + db_path=str(db_path), encryption_key=encryption_key + ) + await storage.initialize() + yield storage + + +async def test_login_flow_provision_wakes_user_manager(temp_storage, mocker): + """A completed Login Flow v2 web provision spawns the user's scanner at + once via the provision signal, well inside a long poll interval.""" + # ── user_manager: long poll interval + stubbed per-user scanner ────────── + manager_settings = MagicMock() + manager_settings.vector_sync_user_poll_interval = 1000 # never fires here + mocker.patch( + "nextcloud_mcp_server.vector.oauth_sync.get_settings", + return_value=manager_settings, + ) + + spawned: set[str] = set() + alice_spawned = anyio.Event() + + async def fake_scanner( + user_id, + cancel_scope, + send_stream, + shutdown_event, + wake_event, + nextcloud_host, + user_states, + ): + spawned.add(user_id) + if user_id == "alice": + alice_spawned.set() + with cancel_scope: + await shutdown_event.wait() + user_states.pop(user_id, None) + + mocker.patch( + "nextcloud_mcp_server.vector.oauth_sync._run_user_scanner_with_scope", + fake_scanner, + ) + + # ── wire the doorbell exactly as the lifespan does ─────────────────────── + import nextcloud_mcp_server.app as app_module + + provision_signal = ProvisionSignal() + mocker.patch.object( + app_module._vector_sync_state, "provision_signal", provision_signal + ) + + # ── mock only the Nextcloud Login Flow v2 poll ─────────────────────────── + completed = LoginFlowPollResult( + status="completed", + server="https://cloud.example.com", + login_name="alice", + app_password="aaaaa-bbbbb-ccccc-ddddd-eeeee", + ) + flow_client = AsyncMock() + flow_client.poll.return_value = completed + + provision_settings = MagicMock() + provision_settings.nextcloud_host = "https://cloud.example.com" + provision_settings.nextcloud_public_issuer_url = None + + provision_id = "login-flow-wake" + _provision_sessions[provision_id] = { + "status": "pending", + "poll_endpoint": "https://cloud.example.com/login/v2/poll", + "poll_token": "secret-token", + "user_id": "alice", + "created_at": time.time(), + "expires_at": time.time() + 1200, + } + + try: + shutdown_event = anyio.Event() + user_states: dict = {} + + async with anyio.create_task_group() as tg: + await tg.start( + user_manager_task, + None, # send_stream — unused by the stubbed scanner + shutdown_event, + anyio.Event(), # scanner wake_event + temp_storage, + "https://cloud.example.com", + user_states, + tg, + provision_signal, + ) + + # First poll: no users provisioned yet → no scanner. + await anyio.sleep(0.1) + assert not spawned + + # Run the real login-flow web-provision background task. It stores + # the app password into temp_storage and rings the doorbell. + with ( + patch( + "nextcloud_mcp_server.auth.provision_routes.get_settings", + return_value=provision_settings, + ), + patch( + "nextcloud_mcp_server.auth.provision_routes.get_nextcloud_ssl_verify", + return_value=False, + ), + patch( + "nextcloud_mcp_server.auth.provision_routes.LoginFlowV2Client", + return_value=flow_client, + ), + patch( + "nextcloud_mcp_server.auth.provision_routes.get_shared_storage", + new_callable=AsyncMock, + return_value=temp_storage, + ), + ): + await _poll_and_store(provision_id) + + # The app password was really stored … + assert "alice" in await temp_storage.get_all_app_password_user_ids() + # … and the manager woke and spawned alice's scanner far inside the + # 1000s poll interval (i.e. because of the signal, not the poll). + with anyio.fail_after(3): + await alice_spawned.wait() + assert spawned == {"alice"} + + shutdown_event.set() + finally: + _provision_sessions.pop(provision_id, None) diff --git a/tests/unit/vector/test_user_manager_provision_wake.py b/tests/unit/vector/test_user_manager_provision_wake.py new file mode 100644 index 00000000..d00af64b --- /dev/null +++ b/tests/unit/vector/test_user_manager_provision_wake.py @@ -0,0 +1,213 @@ +"""Unit tests for immediate-on-provision scanner spawning. + +Covers the doorbell that lets a provisioning request wake ``user_manager_task`` +at once instead of waiting out ``VECTOR_SYNC_USER_POLL_INTERVAL``: + +- ``ProvisionSignal`` semantics (sticky ring, wake-a-parked-wait, re-arm). +- ``user_manager_task`` re-polls early when the signal is rung, spawning a + newly provisioned user's scanner well before the next poll tick. +- ``notify_user_provisioned`` is a no-op when no manager is running. +""" + +from unittest.mock import MagicMock + +import anyio +import pytest + +from nextcloud_mcp_server.vector.oauth_sync import ProvisionSignal, user_manager_task + +pytestmark = pytest.mark.unit + + +# ── ProvisionSignal primitive ──────────────────────────────────────────────── + + +async def test_provision_signal_ring_before_wait_is_observed(): + """A ring that lands before wait() is sticky and returns immediately.""" + signal = ProvisionSignal() + signal.ring() + with anyio.fail_after(1): + await signal.wait() + + +async def test_provision_signal_wakes_parked_waiter(): + """ring() releases a wait() that is already parked.""" + signal = ProvisionSignal() + woke = anyio.Event() + + async def waiter(): + await signal.wait() + woke.set() + + async with anyio.create_task_group() as tg: + tg.start_soon(waiter) + await anyio.sleep(0.05) # let waiter park + assert not woke.is_set() + signal.ring() + with anyio.fail_after(1): + await woke.wait() + + +async def test_provision_signal_rearms_for_next_cycle(): + """After a ring is consumed, the next wait() blocks until the next ring.""" + signal = ProvisionSignal() + signal.ring() + await signal.wait() # consumes first ring, re-arms + + # Second wait must block (no pending ring) then release on the next ring. + second = anyio.Event() + + async def waiter(): + await signal.wait() + second.set() + + async with anyio.create_task_group() as tg: + tg.start_soon(waiter) + await anyio.sleep(0.05) + assert not second.is_set() # proves the first ring did not carry over + signal.ring() + with anyio.fail_after(1): + await second.wait() + + +# ── user_manager_task wake-on-provision ────────────────────────────────────── + + +class _FakeStorage: + """Storage stub whose provisioned-user set the test mutates between polls.""" + + def __init__(self, users: set[str]): + self.users = users + + async def get_all_app_password_user_ids(self) -> list[str]: + return list(self.users) + + +async def test_user_manager_wakes_on_provision_signal(mocker): + """Ringing the signal makes the manager re-poll and spawn the new user's + scanner well before the (long) poll interval elapses.""" + # Long poll interval so any prompt spawn proves it was the signal, not poll. + settings = MagicMock() + settings.vector_sync_user_poll_interval = 1000 + mocker.patch( + "nextcloud_mcp_server.vector.oauth_sync.get_settings", return_value=settings + ) + + spawned: set[str] = set() + spawn_events: dict[str, anyio.Event] = { + "alice": anyio.Event(), + "bob": anyio.Event(), + } + + async def fake_scanner( + user_id, + cancel_scope, + send_stream, + shutdown_event, + wake_event, + nextcloud_host, + user_states, + ): + spawned.add(user_id) + if user_id in spawn_events: + spawn_events[user_id].set() + # Stay alive (keeps user_states populated) until shutdown. + with cancel_scope: + await shutdown_event.wait() + user_states.pop(user_id, None) + + mocker.patch( + "nextcloud_mcp_server.vector.oauth_sync._run_user_scanner_with_scope", + fake_scanner, + ) + + storage = _FakeStorage({"alice"}) + provision_signal = ProvisionSignal() + shutdown_event = anyio.Event() + scanner_wake_event = anyio.Event() + user_states: dict = {} + + async with anyio.create_task_group() as tg: + await tg.start( + user_manager_task, + None, # send_stream — unused by the stubbed scanner + shutdown_event, + scanner_wake_event, + storage, + "http://nextcloud", + user_states, + tg, + provision_signal, + ) + + # First poll discovers the already-provisioned user. + with anyio.fail_after(2): + await spawn_events["alice"].wait() + assert "bob" not in spawned + + # Provision a new user, then ring — the manager must pick bob up fast. + storage.users.add("bob") + provision_signal.ring() + with anyio.fail_after(2): # << 1000s poll interval + await spawn_events["bob"].wait() + assert spawned == {"alice", "bob"} + + shutdown_event.set() + + +async def test_user_manager_shutdown_still_breaks_sleep(mocker): + """Setting shutdown wakes the manager out of its sleep promptly even with a + long poll interval (the doorbell race must not regress shutdown latency).""" + settings = MagicMock() + settings.vector_sync_user_poll_interval = 1000 + mocker.patch( + "nextcloud_mcp_server.vector.oauth_sync.get_settings", return_value=settings + ) + mocker.patch( + "nextcloud_mcp_server.vector.oauth_sync._run_user_scanner_with_scope", + # No users provisioned, so this is never called; keep a harmless stub. + lambda *a, **k: anyio.sleep(0), + ) + + storage = _FakeStorage(set()) + shutdown_event = anyio.Event() + + async with anyio.create_task_group() as tg: + await tg.start( + user_manager_task, + None, + shutdown_event, + anyio.Event(), + storage, + "http://nextcloud", + {}, + tg, + ProvisionSignal(), + ) + await anyio.sleep(0.05) # let it enter the sleep + shutdown_event.set() + # If shutdown didn't break the 1000s sleep, fail_after would trip. + with anyio.fail_after(2): + await anyio.sleep(0) # task group exit below is the real assertion + + +# ── notify_user_provisioned no-op guard ────────────────────────────────────── + + +def test_notify_user_provisioned_noop_without_manager(mocker): + """When no manager is running, the helper must not raise.""" + import nextcloud_mcp_server.app as app_module + + mocker.patch.object(app_module._vector_sync_state, "provision_signal", None) + # Should be a silent no-op. + app_module.notify_user_provisioned() + + +def test_notify_user_provisioned_rings_when_present(mocker): + """When a manager is running, the helper rings its signal.""" + import nextcloud_mcp_server.app as app_module + + signal = ProvisionSignal() + mocker.patch.object(app_module._vector_sync_state, "provision_signal", signal) + app_module.notify_user_provisioned() + assert signal._event.is_set() From 3f094539269fbf5572ac36630bc541547a8ba45f Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Fri, 12 Jun 2026 09:56:27 +0200 Subject: [PATCH 2/4] refactor(vector-sync): address round-1 review nits - ProvisionSignal.wait() re-arms in a finally so a cancelled wait (shutdown racing the doorbell) leaves a fresh unset event, not a stale set-but-consumed one; preserves the no-await-before-swap lost-wakeup guarantee. - Hoist user_manager_task's _wake_on helper out of the while loop (one object, not one per iteration). - Test: assert ProvisionSignal via its public wait() contract instead of the private _event attribute. - Add test_provision_app_password_wakes_user_manager covering the api/passwords.py wake path (previously only LFv2 web + MCP tool were tested). Co-Authored-By: Claude Opus 4.8 (1M context) --- nextcloud_mcp_server/vector/oauth_sync.py | 29 ++++++++----- .../test_management_app_password_endpoints.py | 42 +++++++++++++++++++ .../test_user_manager_provision_wake.py | 6 ++- 3 files changed, 64 insertions(+), 13 deletions(-) diff --git a/nextcloud_mcp_server/vector/oauth_sync.py b/nextcloud_mcp_server/vector/oauth_sync.py index 485bdaa3..e5e6a020 100644 --- a/nextcloud_mcp_server/vector/oauth_sync.py +++ b/nextcloud_mcp_server/vector/oauth_sync.py @@ -58,9 +58,11 @@ class ProvisionSignal: event back to shared state — avoiding any ``app`` ↔ ``vector`` import cycle. Concurrency: ``anyio.Event`` is sticky, so a ``ring()`` that lands before - ``wait()`` is still observed. ``wait()`` re-arms with no ``await`` between - observing the set and swapping the event, so under cooperative scheduling a - concurrent ``ring()`` cannot slip into that window and be lost. + ``wait()`` is still observed. ``wait()`` re-arms in a ``finally`` with no + ``await`` before the swap, so under cooperative scheduling a concurrent + ``ring()`` cannot slip into that window and be lost — and the re-arm also + runs if ``wait()`` is cancelled (e.g. shutdown racing the doorbell), leaving + a fresh unset event rather than a stale set-but-consumed one. """ def __init__(self) -> None: @@ -72,10 +74,13 @@ class ProvisionSignal: async def wait(self) -> None: """Block until the next ring, then re-arm for the following cycle.""" - await self._event.wait() - # No await before the swap: a concurrent ring() cannot interleave here, - # so it lands on the fresh event and the next wait() observes it. - self._event = anyio.Event() + try: + await self._event.wait() + finally: + # Re-arm even on cancellation. The assignment is not a checkpoint, + # so no concurrent ring() can interleave before the swap; a ring + # that already arrived lands on the fresh event for the next wait(). + self._event = anyio.Event() # Process-wide app-password storage for the BasicAuth client path. @@ -475,6 +480,12 @@ async def user_manager_task( logger.info("[BasicAuth] User manager started (poll interval: %ss)", poll_interval) task_status.started() + # Sleep helper: await one of the wakeup events, then end the sleep by + # cancelling the shared scope. Defined once (not per loop iteration). + async def _wake_on(wait_fn, scope: anyio.CancelScope) -> None: + await wait_fn() + scope.cancel() + while not shutdown_event.is_set(): try: # Query the app_passwords table — background sync always @@ -537,10 +548,6 @@ async def user_manager_task( # provisioning signal so a just-provisioned user is discovered at once. # Race both waits in a child task group; whichever fires first cancels # the scope, ending the sleep. move_on_after caps it at poll_interval. - async def _wake_on(wait_fn, scope: anyio.CancelScope) -> None: - await wait_fn() - scope.cancel() - try: with anyio.move_on_after(poll_interval): async with anyio.create_task_group() as wake_tg: diff --git a/tests/unit/test_management_app_password_endpoints.py b/tests/unit/test_management_app_password_endpoints.py index 7ad924b8..80690c6b 100644 --- a/tests/unit/test_management_app_password_endpoints.py +++ b/tests/unit/test_management_app_password_endpoints.py @@ -254,6 +254,48 @@ async def test_provision_app_password_success(temp_storage, mocker): assert get_kwargs["auth"] == ("testuser", "aaaaa-bbbbb-ccccc-ddddd-eeeee") +async def test_provision_app_password_wakes_user_manager(temp_storage, mocker): + """A successful provision rings the background-sync doorbell so the user + manager re-polls immediately (the api/passwords.py wake path).""" + mocker.patch( + "nextcloud_mcp_server.api.passwords.get_settings", + return_value=MagicMock( + nextcloud_host="http://localhost:8080", + nextcloud_verify_ssl=True, + nextcloud_ca_bundle=None, + ), + ) + + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = {"ocs": {"data": {"id": "testuser"}}} + mock_client = AsyncMock() + mock_client.get = AsyncMock(return_value=mock_response) + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock() + mocker.patch( + "nextcloud_mcp_server.api.passwords.nextcloud_httpx_client", + return_value=mock_client, + ) + + # Spy on the doorbell helper (imported locally from app at call time). + notify = mocker.patch("nextcloud_mcp_server.app.notify_user_provisioned") + + client = TestClient(create_test_app(temp_storage)) + response = client.post( + "/api/v1/users/testuser/app-password", + headers={ + "Authorization": create_basic_auth_header( + "testuser", "aaaaa-bbbbb-ccccc-ddddd-eeeee" + ) + }, + ) + + assert response.status_code == 200 + assert response.json()["success"] is True + notify.assert_called_once() + + async def test_provision_app_password_uses_loginname_not_uid(temp_storage, mocker): """Regression: when the Nextcloud UID differs from the loginName (e.g. OIDC-provisioned users whose UID is their display name — UID diff --git a/tests/unit/vector/test_user_manager_provision_wake.py b/tests/unit/vector/test_user_manager_provision_wake.py index d00af64b..2463f350 100644 --- a/tests/unit/vector/test_user_manager_provision_wake.py +++ b/tests/unit/vector/test_user_manager_provision_wake.py @@ -203,11 +203,13 @@ def test_notify_user_provisioned_noop_without_manager(mocker): app_module.notify_user_provisioned() -def test_notify_user_provisioned_rings_when_present(mocker): +async def test_notify_user_provisioned_rings_when_present(mocker): """When a manager is running, the helper rings its signal.""" import nextcloud_mcp_server.app as app_module signal = ProvisionSignal() mocker.patch.object(app_module._vector_sync_state, "provision_signal", signal) app_module.notify_user_provisioned() - assert signal._event.is_set() + # Public contract: after a ring, the next wait() returns without blocking. + with anyio.fail_after(1): + await signal.wait() From 79d9d62e6a39b8f871acf9b1ce17a93b7896d823 Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Fri, 12 Jun 2026 10:08:18 +0200 Subject: [PATCH 3/4] refactor(vector-sync): clear SonarCloud gate + round-2 nits MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Quality-gate fixes (new-code conditions on PR #902): - new_security_hotspots_reviewed: drop the fake "http://nextcloud" host in the manager tests to https:// (python:S5332 ×2). - new_security_rating: generate the integration test's fake app password with secrets.token_urlsafe instead of a hardcoded literal (python:S2068). - new_reliability_rating: restructure the user_manager sleep so an explicit await checkpoint lives inside the cancellation scope — await one waiter directly while watching shutdown via start_soon (python:S7490). Behaviour is unchanged: timeout, shutdown, or a provisioning ring all end the sleep. Review nits: - Move the shutdown test's fail_after(2) to wrap the whole task group so it actually bounds the task-group exit (was guarding a no-op sleep); drop the sleep(0) stub (python:S7491). - Type _wake_on's wait_fn as Callable[[], Awaitable[object]]. - Note in _wire_vector_sync_state why provision_signal is set on the singleton only, not fanned out to app.state. Co-Authored-By: Claude Opus 4.8 (1M context) --- nextcloud_mcp_server/app.py | 4 ++ nextcloud_mcp_server/vector/oauth_sync.py | 15 ++++--- .../test_login_flow_provision_wake.py | 6 ++- .../test_user_manager_provision_wake.py | 44 ++++++++++--------- 4 files changed, 42 insertions(+), 27 deletions(-) diff --git a/nextcloud_mcp_server/app.py b/nextcloud_mcp_server/app.py index 98a37d08..510d80cf 100644 --- a/nextcloud_mcp_server/app.py +++ b/nextcloud_mcp_server/app.py @@ -397,6 +397,10 @@ def _wire_vector_sync_state( ``eviction_task_group`` is deliberately not set here: it only exists once the lifespan has entered its ``anyio.create_task_group()`` (after this call), so the lifespan assigns it on the singleton directly at that point. + ``provision_signal`` is likewise excluded on purpose — only ``user_manager_task`` + consumes it (request handlers reach it via ``notify_user_provisioned``), so the + multi-user lifespan sets it on the singleton directly rather than fanning it out + to ``app.state``/the browser sub-app. """ send_stream = transport.send_stream receive_stream = transport.receive_stream diff --git a/nextcloud_mcp_server/vector/oauth_sync.py b/nextcloud_mcp_server/vector/oauth_sync.py index e5e6a020..bbbcaa44 100644 --- a/nextcloud_mcp_server/vector/oauth_sync.py +++ b/nextcloud_mcp_server/vector/oauth_sync.py @@ -20,6 +20,7 @@ background sync. import logging import time +from collections.abc import Awaitable, Callable from dataclasses import dataclass, field import anyio @@ -482,7 +483,9 @@ async def user_manager_task( # Sleep helper: await one of the wakeup events, then end the sleep by # cancelling the shared scope. Defined once (not per loop iteration). - async def _wake_on(wait_fn, scope: anyio.CancelScope) -> None: + async def _wake_on( + wait_fn: Callable[[], Awaitable[object]], scope: anyio.CancelScope + ) -> None: await wait_fn() scope.cancel() @@ -546,17 +549,17 @@ async def user_manager_task( # Sleep until the next poll tick, but wake early on shutdown or a # provisioning signal so a just-provisioned user is discovered at once. - # Race both waits in a child task group; whichever fires first cancels - # the scope, ending the sleep. move_on_after caps it at poll_interval. + # Watch shutdown concurrently and block here on a provisioning ring; + # whichever fires first cancels the shared scope and ends the sleep, + # while move_on_after caps the wait at poll_interval. Awaiting one waiter + # directly keeps an explicit checkpoint inside the cancellation scope. try: with anyio.move_on_after(poll_interval): async with anyio.create_task_group() as wake_tg: wake_tg.start_soon( _wake_on, shutdown_event.wait, wake_tg.cancel_scope ) - wake_tg.start_soon( - _wake_on, provision_signal.wait, wake_tg.cancel_scope - ) + await _wake_on(provision_signal.wait, wake_tg.cancel_scope) except anyio.get_cancelled_exc_class(): break diff --git a/tests/integration/test_login_flow_provision_wake.py b/tests/integration/test_login_flow_provision_wake.py index f1f78ab4..d6b1cea6 100644 --- a/tests/integration/test_login_flow_provision_wake.py +++ b/tests/integration/test_login_flow_provision_wake.py @@ -18,6 +18,7 @@ This is the Login Flow v2 deployment-mode counterpart to the multi-user BasicAuth coverage in ``test_app_password_provisioning.py``. """ +import secrets import tempfile import time from pathlib import Path @@ -99,11 +100,14 @@ async def test_login_flow_provision_wakes_user_manager(temp_storage, mocker): ) # ── mock only the Nextcloud Login Flow v2 poll ─────────────────────────── + # Generated, not a hardcoded literal — keeps this a fake token, not a + # credential pattern (SonarQube python:S2068). + fake_app_password = secrets.token_urlsafe(24) completed = LoginFlowPollResult( status="completed", server="https://cloud.example.com", login_name="alice", - app_password="aaaaa-bbbbb-ccccc-ddddd-eeeee", + app_password=fake_app_password, ) flow_client = AsyncMock() flow_client.poll.return_value = completed diff --git a/tests/unit/vector/test_user_manager_provision_wake.py b/tests/unit/vector/test_user_manager_provision_wake.py index 2463f350..bd29731d 100644 --- a/tests/unit/vector/test_user_manager_provision_wake.py +++ b/tests/unit/vector/test_user_manager_provision_wake.py @@ -134,7 +134,7 @@ async def test_user_manager_wakes_on_provision_signal(mocker): shutdown_event, scanner_wake_event, storage, - "http://nextcloud", + "https://nextcloud", user_states, tg, provision_signal, @@ -163,32 +163,36 @@ async def test_user_manager_shutdown_still_breaks_sleep(mocker): mocker.patch( "nextcloud_mcp_server.vector.oauth_sync.get_settings", return_value=settings ) + + async def _unused_scanner(*args, **kwargs): + # No users provisioned, so this is never called; keep a harmless stub. + return None + mocker.patch( "nextcloud_mcp_server.vector.oauth_sync._run_user_scanner_with_scope", - # No users provisioned, so this is never called; keep a harmless stub. - lambda *a, **k: anyio.sleep(0), + _unused_scanner, ) storage = _FakeStorage(set()) shutdown_event = anyio.Event() - async with anyio.create_task_group() as tg: - await tg.start( - user_manager_task, - None, - shutdown_event, - anyio.Event(), - storage, - "http://nextcloud", - {}, - tg, - ProvisionSignal(), - ) - await anyio.sleep(0.05) # let it enter the sleep - shutdown_event.set() - # If shutdown didn't break the 1000s sleep, fail_after would trip. - with anyio.fail_after(2): - await anyio.sleep(0) # task group exit below is the real assertion + # fail_after wraps the whole task group: if shutdown_event doesn't break the + # 1000s sleep, the task group never exits and the 2s deadline trips. + with anyio.fail_after(2): + async with anyio.create_task_group() as tg: + await tg.start( + user_manager_task, + None, + shutdown_event, + anyio.Event(), + storage, + "https://nextcloud", + {}, + tg, + ProvisionSignal(), + ) + await anyio.sleep(0.05) # let it enter the sleep + shutdown_event.set() # ── notify_user_provisioned no-op guard ────────────────────────────────────── From 7ceb0072e47633ee7f76da0ff96e57903dfbc6fc Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Fri, 12 Jun 2026 10:14:39 +0200 Subject: [PATCH 4/4] test(vector-sync): cover the auth_tools.py provisioning wake path Add test_check_status_completion_wakes_user_manager: register the auth tools against a stub MCP, drive nc_auth_check_status through a completed Login Flow (mocked storage + poll), and assert it stores the app password and rings the background-sync doorbell. The MCP-tool wake path was previously only verified by inspection (round-2 review nit); all three notify_user_provisioned() call sites now have dedicated coverage. Co-Authored-By: Claude Opus 4.8 (1M context) --- tests/unit/test_auth_tools.py | 88 +++++++++++++++++++++++++++++++++++ 1 file changed, 88 insertions(+) diff --git a/tests/unit/test_auth_tools.py b/tests/unit/test_auth_tools.py index 25e609fd..f1078276 100644 --- a/tests/unit/test_auth_tools.py +++ b/tests/unit/test_auth_tools.py @@ -3,18 +3,45 @@ Tests the auth tools logic with mocked storage and Login Flow client. """ +import secrets import tempfile from pathlib import Path +from typing import cast +from unittest.mock import AsyncMock, MagicMock import pytest from cryptography.fernet import Fernet +from mcp.server.fastmcp import FastMCP +from nextcloud_mcp_server.auth.login_flow import LoginFlowPollResult from nextcloud_mcp_server.auth.storage import RefreshTokenStorage from nextcloud_mcp_server.models.auth import ALL_SUPPORTED_SCOPES +from nextcloud_mcp_server.server.auth_tools import register_auth_tools pytestmark = pytest.mark.unit +def _capture_registered_tools() -> dict: + """Register the auth tools against a stub MCP and return them by name. + + ``register_auth_tools`` only uses ``@mcp.tool(...)`` decorators, so a stub + whose ``tool()`` returns an identity decorator captures the closures without + a real FastMCP instance. + """ + captured: dict = {} + + class _StubMCP: + def tool(self, *args, **kwargs): + def deco(fn): + captured[fn.__name__] = fn + return fn + + return deco + + register_auth_tools(cast(FastMCP, _StubMCP())) + return captured + + @pytest.fixture def encryption_key(): """Generate a test encryption key.""" @@ -196,3 +223,64 @@ def test_all_supported_scopes(): read_scopes = [s for s in ALL_SUPPORTED_SCOPES if s.endswith(":read")] write_scopes = [s for s in ALL_SUPPORTED_SCOPES if s.endswith(":write")] assert len(read_scopes) == len(write_scopes) + + +# ── Background-sync wake on provisioning ── + + +async def test_check_status_completion_wakes_user_manager(mocker): + """When nc_auth_check_status polls a completed Login Flow, it stores the app + password and rings the background-sync doorbell (the server/auth_tools.py + wake path).""" + check_status = _capture_registered_tools()["nc_auth_check_status"] + + mocker.patch( + "nextcloud_mcp_server.server.auth_tools.extract_user_id_from_token", + AsyncMock(return_value="alice"), + ) + storage = MagicMock() + storage.get_app_password_with_scopes = AsyncMock(return_value=None) # not yet + storage.get_login_flow_session = AsyncMock( + return_value={ + "poll_endpoint": "https://nc/login/v2/poll", + "poll_token": "tok", + "requested_scopes": None, + } + ) + storage.store_app_password_with_scopes = AsyncMock() + storage.delete_login_flow_session = AsyncMock() + mocker.patch( + "nextcloud_mcp_server.server.auth_tools.get_shared_storage", + AsyncMock(return_value=storage), + ) + mocker.patch( + "nextcloud_mcp_server.server.auth_tools.get_settings", + return_value=MagicMock( + nextcloud_host="https://nc", nextcloud_public_issuer_url=None + ), + ) + mocker.patch( + "nextcloud_mcp_server.server.auth_tools.get_nextcloud_ssl_verify", + return_value=False, + ) + mocker.patch("nextcloud_mcp_server.server.auth_tools.invalidate_scope_cache") + + flow_client = AsyncMock() + flow_client.poll = AsyncMock( + return_value=LoginFlowPollResult( + status="completed", + login_name="alice", + app_password=secrets.token_urlsafe(24), # generated, not a literal + ) + ) + mocker.patch( + "nextcloud_mcp_server.server.auth_tools.LoginFlowV2Client", + return_value=flow_client, + ) + notify = mocker.patch("nextcloud_mcp_server.app.notify_user_provisioned") + + response = await check_status(MagicMock()) + + assert response.status == "provisioned" + storage.store_app_password_with_scopes.assert_awaited_once() + notify.assert_called_once()