fix(vector-sync): sweep placeholder orphans at Pod startup (#101)
When the per-tenant nextcloud-mcp-server Pod OOMKills mid-batch, the in-memory anyio processor queue is lost but the placeholder Qdrant points (is_placeholder=true, status=pending) survive. The next Pod's scanner re-runs, sees the existing placeholders, applies the 5 × VECTOR_SYNC_SCAN_INTERVAL staleness gate (~5h with the deployed 1h scan interval), and skips them. Result: 0 documents indexed for the duration of the gate after every restart. Stamps a process-level instance_id (UUID per Pod-process) onto every placeholder write. A new sweep_orphan_placeholders helper, called once from starlette_lifespan after the Qdrant client is initialised and before the scanner / user-manager spawns, scrolls the collection and deletes any placeholder whose instance_id doesn't match the current Pod's (including placeholders with no instance_id field — back-compat for pre-fix Pod versions). The scanner's next cycle naturally re-creates fresh placeholders and queues work normally; no DocumentTask reconstruction needed. Sweep is one-shot at startup, not periodic — the existing staleness gate still covers same-Pod recovery, and the cross-Pod-restart gap was the only failure mode. Failure is non-fatal (logged via vector_sync.orphan_sweep_failed) so a transient Qdrant hiccup at boot doesn't prevent the scanner from running. Both lifespan branches (single-user BasicAuth, OAuth / multi-user BasicAuth) call the sweep via a module-local helper. A new VECTOR_SYNC_ORPHAN_SWEEP_ENABLED setting (default True) provides an escape hatch. Closes Deck #101. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Chris Coutinho
co-authored by
Claude Opus 4.7
Claude Opus 4.7
parent
32fc03cf67
commit
a5cbe91b29
@@ -129,6 +129,7 @@ from nextcloud_mcp_server.vector.oauth_sync import (
|
||||
oauth_processor_task,
|
||||
user_manager_task,
|
||||
)
|
||||
from nextcloud_mcp_server.vector.placeholder import sweep_orphan_placeholders
|
||||
from nextcloud_mcp_server.vector.processor import processor_task
|
||||
from nextcloud_mcp_server.vector.qdrant_client import get_qdrant_client
|
||||
from nextcloud_mcp_server.vector.scanner import scanner_task
|
||||
@@ -1437,6 +1438,34 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None =
|
||||
await stack.enter_async_context(_maybe_login_flow_cleanup(app))
|
||||
yield
|
||||
|
||||
async def _sweep_orphan_placeholders_if_enabled() -> None:
|
||||
"""One-shot Pod-startup sweep of cross-restart placeholder orphans.
|
||||
|
||||
See ``vector.placeholder.sweep_orphan_placeholders`` and Deck
|
||||
card #101. Both lifespan branches (single-user BasicAuth and
|
||||
OAuth / multi-user BasicAuth) call this after the qdrant
|
||||
client is initialised and before the scanner / user-manager
|
||||
tasks spawn. Failures are non-fatal — the existing staleness
|
||||
gate will eventually re-queue orphans on the slow ~5h path.
|
||||
"""
|
||||
if not settings.vector_sync_orphan_sweep_enabled:
|
||||
return
|
||||
try:
|
||||
qdrant_client = await get_qdrant_client()
|
||||
swept, kept = await sweep_orphan_placeholders(
|
||||
qdrant_client, settings.qdrant_collection
|
||||
)
|
||||
logger.info(
|
||||
"vector_sync.orphan_sweep",
|
||||
extra={
|
||||
"swept": swept,
|
||||
"kept": kept,
|
||||
"collection": settings.qdrant_collection,
|
||||
},
|
||||
)
|
||||
except Exception:
|
||||
logger.exception("vector_sync.orphan_sweep_failed")
|
||||
|
||||
@asynccontextmanager
|
||||
async def starlette_lifespan(app: Starlette):
|
||||
# Set OAuth context for OAuth login routes (ADR-004)
|
||||
@@ -1612,6 +1641,9 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None =
|
||||
f"Cannot start vector sync - Qdrant initialization failed: {e}"
|
||||
) from e
|
||||
|
||||
# Orphan-sweep before scanner starts — card #101.
|
||||
await _sweep_orphan_placeholders_if_enabled()
|
||||
|
||||
# Initialize shared state
|
||||
send_stream, receive_stream = anyio.create_memory_object_stream(
|
||||
max_buffer_size=settings.vector_sync_queue_max_size
|
||||
@@ -1774,6 +1806,11 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None =
|
||||
f"Cannot start vector sync - Qdrant initialization failed: {e}"
|
||||
) from e
|
||||
|
||||
# Orphan-sweep before scanners spawn — card #101. Runs once
|
||||
# across the shared (per-tenant) collection regardless of
|
||||
# how many per-user scanners the user-manager later starts.
|
||||
await _sweep_orphan_placeholders_if_enabled()
|
||||
|
||||
# Clean up stale app passwords at startup (BasicAuth mode only)
|
||||
if not oauth_enabled:
|
||||
try:
|
||||
|
||||
@@ -87,6 +87,12 @@ _DEFAULTS: dict[str, Any] = {
|
||||
"vector_sync_processor_workers": 3,
|
||||
"vector_sync_queue_max_size": 10000,
|
||||
"vector_sync_user_poll_interval": 60,
|
||||
# Orphan-sweep at Pod startup (card #101). When True, delete any
|
||||
# placeholders carrying a different / absent ``instance_id`` before
|
||||
# the scanner's first cycle, so a Pod restart mid-batch doesn't
|
||||
# leave work stuck behind the 5x-scan-interval staleness gate.
|
||||
# Escape hatch only — leave on by default.
|
||||
"vector_sync_orphan_sweep_enabled": True,
|
||||
# Verify-on-read concurrency cap (ADR-019)
|
||||
"verification_concurrency": 20,
|
||||
# Qdrant
|
||||
@@ -585,6 +591,7 @@ class Settings:
|
||||
vector_sync_processor_workers: int = 3
|
||||
vector_sync_queue_max_size: int = 10000
|
||||
vector_sync_user_poll_interval: int = 60 # seconds - OAuth mode user discovery
|
||||
vector_sync_orphan_sweep_enabled: bool = True # card #101
|
||||
|
||||
# Verify-on-read concurrency (ADR-019). Cap on parallel Nextcloud
|
||||
# round-trips during search-result verification fan-out. Lower this if the
|
||||
@@ -1076,6 +1083,7 @@ def get_settings() -> Settings:
|
||||
"vector_sync_processor_workers": "VECTOR_SYNC_PROCESSOR_WORKERS",
|
||||
"vector_sync_queue_max_size": "VECTOR_SYNC_QUEUE_MAX_SIZE",
|
||||
"vector_sync_user_poll_interval": "VECTOR_SYNC_USER_POLL_INTERVAL",
|
||||
"vector_sync_orphan_sweep_enabled": "VECTOR_SYNC_ORPHAN_SWEEP_ENABLED",
|
||||
# Verify-on-read (ADR-019)
|
||||
"verification_concurrency": "VERIFICATION_CONCURRENCY",
|
||||
# Qdrant settings
|
||||
|
||||
@@ -30,6 +30,13 @@ from nextcloud_mcp_server.vector.qdrant_client import get_qdrant_client
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Stamped on every placeholder this Pod-process writes. A fresh UUID per
|
||||
# process means a restarted Pod sees its predecessor's placeholders as
|
||||
# "not mine" and deletes them in ``sweep_orphan_placeholders`` at startup,
|
||||
# instead of waiting out the ``5 × VECTOR_SYNC_SCAN_INTERVAL`` staleness
|
||||
# gate (~5h with the deployed 1h scan interval). Card #101.
|
||||
_INSTANCE_ID = str(uuid.uuid4())
|
||||
|
||||
|
||||
def _generate_placeholder_id(doc_type: str, doc_id: str) -> str:
|
||||
"""Generate deterministic UUID for placeholder point.
|
||||
@@ -98,6 +105,10 @@ async def write_placeholder_point(
|
||||
"modified_at": modified_at,
|
||||
"etag": etag,
|
||||
"queued_at": int(time.time()),
|
||||
# Pod-process identity. ``sweep_orphan_placeholders`` uses
|
||||
# the (placeholder.instance_id != _INSTANCE_ID) predicate to
|
||||
# detect orphans from a crashed predecessor Pod.
|
||||
"instance_id": _INSTANCE_ID,
|
||||
}
|
||||
|
||||
# Add file_path for files
|
||||
@@ -319,3 +330,88 @@ def get_placeholder_filter() -> FieldCondition:
|
||||
key="is_placeholder",
|
||||
match=MatchValue(value=False),
|
||||
)
|
||||
|
||||
|
||||
# Batch size for the orphan-sweep scroll + delete loop. Big enough to
|
||||
# keep round-trip count down, small enough that a single delete payload
|
||||
# isn't unreasonable. Matches the order of magnitude of a per-tenant
|
||||
# placeholder count (108 in the originating incident).
|
||||
_ORPHAN_SWEEP_BATCH_SIZE = 100
|
||||
|
||||
|
||||
async def sweep_orphan_placeholders(
|
||||
qdrant_client,
|
||||
collection_name: str,
|
||||
*,
|
||||
batch_size: int = _ORPHAN_SWEEP_BATCH_SIZE,
|
||||
) -> tuple[int, int]:
|
||||
"""Delete placeholder points written by a previous Pod-process.
|
||||
|
||||
Scrolls all ``is_placeholder=true`` points in the collection,
|
||||
paginated. For each batch, partitions points by whether their
|
||||
``instance_id`` payload field matches the current Pod's
|
||||
``_INSTANCE_ID``. Orphans (different ``instance_id`` OR field
|
||||
absent — back-compat for placeholders written by pre-fix Pod
|
||||
versions) are deleted by point ID; own-Pod placeholders are
|
||||
left alone for the scanner's staleness gate to handle normally.
|
||||
|
||||
Called once at Pod startup from ``app.starlette_lifespan`` —
|
||||
NOT periodically. The own-Pod path relies on the existing
|
||||
``5 × VECTOR_SYNC_SCAN_INTERVAL`` gate; this helper only
|
||||
addresses the cross-Pod-restart gap. Card #101.
|
||||
|
||||
Args:
|
||||
qdrant_client: Async Qdrant client.
|
||||
collection_name: Target collection.
|
||||
batch_size: Scroll page size. Default 100 — small enough that
|
||||
a single delete payload is reasonable, large enough that
|
||||
round-trip count stays bounded for typical placeholder
|
||||
counts (~hundreds per tenant).
|
||||
|
||||
Returns:
|
||||
``(swept, kept)`` — number of placeholders deleted as orphans,
|
||||
and number left in place as belonging to the current Pod.
|
||||
"""
|
||||
placeholder_filter = Filter(
|
||||
must=[
|
||||
FieldCondition(key="is_placeholder", match=MatchValue(value=True)),
|
||||
]
|
||||
)
|
||||
swept = 0
|
||||
kept = 0
|
||||
offset = None
|
||||
|
||||
while True:
|
||||
points, offset = await qdrant_client.scroll(
|
||||
collection_name=collection_name,
|
||||
scroll_filter=placeholder_filter,
|
||||
limit=batch_size,
|
||||
offset=offset,
|
||||
with_payload=True,
|
||||
with_vectors=False,
|
||||
)
|
||||
if not points:
|
||||
break
|
||||
|
||||
orphan_ids = []
|
||||
for point in points:
|
||||
payload = point.payload or {}
|
||||
point_instance = payload.get("instance_id")
|
||||
if point_instance == _INSTANCE_ID:
|
||||
kept += 1
|
||||
else:
|
||||
orphan_ids.append(point.id)
|
||||
|
||||
if orphan_ids:
|
||||
await qdrant_client.delete(
|
||||
collection_name=collection_name,
|
||||
points_selector=orphan_ids,
|
||||
)
|
||||
swept += len(orphan_ids)
|
||||
|
||||
# ``offset is None`` signals the scroll cursor has been
|
||||
# exhausted — Qdrant's contract for paginated scroll.
|
||||
if offset is None:
|
||||
break
|
||||
|
||||
return swept, kept
|
||||
|
||||
@@ -0,0 +1,196 @@
|
||||
"""Unit tests for placeholder orphan-sweep (card #101).
|
||||
|
||||
When the per-tenant Pod OOMKills mid-batch the in-memory processor
|
||||
queue is lost. Placeholder points written to Qdrant survive, and the
|
||||
next Pod's scanner would skip them under the
|
||||
``5 × VECTOR_SYNC_SCAN_INTERVAL`` staleness gate (~5h with the deployed
|
||||
1h scan interval). ``sweep_orphan_placeholders`` runs once at Pod
|
||||
startup and deletes any placeholder whose ``instance_id`` payload
|
||||
doesn't match the current Pod-process, restoring throughput within
|
||||
one scan cycle.
|
||||
|
||||
The sweep contract this file pins:
|
||||
|
||||
* placeholder with ``instance_id != _INSTANCE_ID`` → deleted
|
||||
* placeholder with **absent** ``instance_id`` → deleted
|
||||
(back-compat for placeholders written by pre-fix Pod versions)
|
||||
* placeholder with ``instance_id == _INSTANCE_ID`` → kept
|
||||
* empty / no-results scroll → no-op (no spurious delete call)
|
||||
* multi-page scroll → all pages visited, deletes batched per page
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import AsyncMock
|
||||
|
||||
import pytest
|
||||
|
||||
from nextcloud_mcp_server.vector import placeholder as placeholder_module
|
||||
from nextcloud_mcp_server.vector.placeholder import sweep_orphan_placeholders
|
||||
|
||||
|
||||
def _scrolled_point(point_id, instance_id=...):
|
||||
"""Stand-in for a qdrant_client Record. Sweep reads only
|
||||
``id`` and ``payload``; passing ``instance_id=...`` (Ellipsis)
|
||||
omits the field entirely (the back-compat case)."""
|
||||
payload: dict = {"is_placeholder": True}
|
||||
if instance_id is not ...:
|
||||
payload["instance_id"] = instance_id
|
||||
return SimpleNamespace(id=point_id, payload=payload)
|
||||
|
||||
|
||||
def _make_client(scroll_pages):
|
||||
"""Build an AsyncMock qdrant client whose ``scroll`` returns the
|
||||
given ``(points, next_offset)`` pages in order, then mimics the
|
||||
real client's exhausted-cursor return of ``([], None)``."""
|
||||
client = AsyncMock()
|
||||
pages = list(scroll_pages) + [([], None)]
|
||||
client.scroll.side_effect = pages
|
||||
return client
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
async def test_sweeps_orphan_with_different_instance_id(monkeypatch):
|
||||
monkeypatch.setattr(placeholder_module, "_INSTANCE_ID", "pod-new")
|
||||
client = _make_client(
|
||||
[
|
||||
([_scrolled_point("p1", instance_id="pod-old")], None),
|
||||
]
|
||||
)
|
||||
|
||||
swept, kept = await sweep_orphan_placeholders(client, "nextcloud_content")
|
||||
|
||||
assert (swept, kept) == (1, 0)
|
||||
client.delete.assert_awaited_once()
|
||||
# Selector is the orphan's point id — own-Pod placeholders never
|
||||
# reach the delete call.
|
||||
assert client.delete.await_args.kwargs["points_selector"] == ["p1"]
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
async def test_sweeps_placeholder_with_absent_instance_id(monkeypatch):
|
||||
"""Back-compat: placeholders written by pre-fix Pod versions have
|
||||
no ``instance_id`` field. They MUST be treated as orphans so the
|
||||
first deploy of this fix doesn't leave old placeholders behind."""
|
||||
monkeypatch.setattr(placeholder_module, "_INSTANCE_ID", "pod-new")
|
||||
client = _make_client(
|
||||
[
|
||||
([_scrolled_point("legacy", instance_id=...)], None),
|
||||
]
|
||||
)
|
||||
|
||||
swept, kept = await sweep_orphan_placeholders(client, "nextcloud_content")
|
||||
|
||||
assert (swept, kept) == (1, 0)
|
||||
assert client.delete.await_args.kwargs["points_selector"] == ["legacy"]
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
async def test_keeps_own_pod_placeholders(monkeypatch):
|
||||
"""A surviving Pod's own placeholders must NOT be deleted —
|
||||
that's what the staleness gate is for, and the processor may
|
||||
still be working through them."""
|
||||
monkeypatch.setattr(placeholder_module, "_INSTANCE_ID", "pod-current")
|
||||
client = _make_client(
|
||||
[
|
||||
(
|
||||
[
|
||||
_scrolled_point("mine-1", instance_id="pod-current"),
|
||||
_scrolled_point("mine-2", instance_id="pod-current"),
|
||||
],
|
||||
None,
|
||||
),
|
||||
]
|
||||
)
|
||||
|
||||
swept, kept = await sweep_orphan_placeholders(client, "nextcloud_content")
|
||||
|
||||
assert (swept, kept) == (0, 2)
|
||||
client.delete.assert_not_awaited()
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
async def test_noop_when_no_placeholders_exist(monkeypatch):
|
||||
"""Cold-boot tenant with an empty collection — sweep does NOT
|
||||
issue a delete request, so a trivially-empty batch can't trip
|
||||
Qdrant validation on an empty selector list."""
|
||||
monkeypatch.setattr(placeholder_module, "_INSTANCE_ID", "pod-fresh")
|
||||
client = _make_client([([], None)])
|
||||
|
||||
swept, kept = await sweep_orphan_placeholders(client, "nextcloud_content")
|
||||
|
||||
assert (swept, kept) == (0, 0)
|
||||
client.delete.assert_not_awaited()
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
async def test_walks_multiple_scroll_pages(monkeypatch):
|
||||
"""Scroll cursor exhaustion is signalled by ``offset is None`` per
|
||||
Qdrant's contract. Sweep must keep paging until the cursor is
|
||||
exhausted, batching the delete per page."""
|
||||
monkeypatch.setattr(placeholder_module, "_INSTANCE_ID", "pod-current")
|
||||
client = _make_client(
|
||||
[
|
||||
(
|
||||
[
|
||||
_scrolled_point("orphan-1", instance_id="pod-prev"),
|
||||
_scrolled_point("mine", instance_id="pod-current"),
|
||||
],
|
||||
"next-cursor",
|
||||
),
|
||||
(
|
||||
[_scrolled_point("orphan-2", instance_id=...)],
|
||||
None,
|
||||
),
|
||||
]
|
||||
)
|
||||
|
||||
swept, kept = await sweep_orphan_placeholders(
|
||||
client, "nextcloud_content", batch_size=2
|
||||
)
|
||||
|
||||
assert (swept, kept) == (2, 1)
|
||||
# One delete per page (each page had at least one orphan).
|
||||
assert client.delete.await_count == 2
|
||||
delete_selectors = [
|
||||
call.kwargs["points_selector"] for call in client.delete.await_args_list
|
||||
]
|
||||
assert delete_selectors == [["orphan-1"], ["orphan-2"]]
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
async def test_write_placeholder_payload_includes_instance_id(monkeypatch):
|
||||
"""Pin the new payload contract: ``write_placeholder_point`` MUST
|
||||
stamp the current Pod's ``_INSTANCE_ID`` onto the payload it
|
||||
upserts, so the next Pod's sweep can identify these placeholders
|
||||
as belonging to a different process."""
|
||||
monkeypatch.setattr(placeholder_module, "_INSTANCE_ID", "pod-pinned")
|
||||
|
||||
fake_qdrant = AsyncMock()
|
||||
fake_settings = SimpleNamespace(get_collection_name=lambda: "nextcloud_content")
|
||||
fake_embedding = SimpleNamespace(get_dimension=lambda: 4)
|
||||
|
||||
async def fake_get_qdrant_client():
|
||||
return fake_qdrant
|
||||
|
||||
monkeypatch.setattr(placeholder_module, "get_qdrant_client", fake_get_qdrant_client)
|
||||
monkeypatch.setattr(placeholder_module, "get_settings", lambda: fake_settings)
|
||||
monkeypatch.setattr(
|
||||
placeholder_module, "get_embedding_service", lambda: fake_embedding
|
||||
)
|
||||
|
||||
await placeholder_module.write_placeholder_point(
|
||||
doc_id="d-42",
|
||||
doc_type="note",
|
||||
user_id="alice",
|
||||
modified_at=1700000000,
|
||||
etag="abc",
|
||||
)
|
||||
|
||||
fake_qdrant.upsert.assert_awaited_once()
|
||||
upserted_point = fake_qdrant.upsert.await_args.kwargs["points"][0]
|
||||
assert upserted_point.payload["instance_id"] == "pod-pinned"
|
||||
# Sanity: the other contract-pinning fields are still emitted.
|
||||
assert upserted_point.payload["is_placeholder"] is True
|
||||
assert upserted_point.payload["status"] == "pending"
|
||||
Reference in New Issue
Block a user