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) <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 4.8
parent
4688f2f95a
commit
d8e3e9bc33
@@ -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)
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
Reference in New Issue
Block a user