Merge pull request #848 from cbcoutinho/feat/vector-sync-cross-user-dedup

feat: dedup shared-file parsing/embedding across users in vector sync
This commit is contained in:
Chris Coutinho
2026-06-04 17:59:31 +02:00
committed by GitHub
9 changed files with 690 additions and 69 deletions
+15 -4
View File
@@ -162,10 +162,14 @@ def build_ownership_filter(
"""Build the Qdrant ``Filter`` constraining a search to readable points.
Matches points whose ``owner_id`` is in ``accessible_owners`` (excluding
self) OR whose ``user_id`` equals ``user_id``. The ``user_id`` branch covers
*all* of the caller's own content — both new points (where
``owner_id == user_id``) and legacy points indexed before ``owner_id``
existed — so self is intentionally NOT repeated in the ``owner_id`` branch.
self) OR whose ``user_id`` equals ``user_id`` OR whose ``acl_principals``
contains ``user:<user_id>``. The ``user_id`` branch covers *all* of the
caller's own content — both new points (where ``owner_id == user_id``) and
legacy points indexed before ``owner_id`` existed — so self is intentionally
NOT repeated in the ``owner_id`` branch. The ``acl_principals`` branch covers
files that were indexed once and deduplicated across users (user-agnostic
point IDs): such a point's ``user_id``/``owner_id`` are the first indexer's,
so only the observed-access principal set surfaces it to other readers.
Args:
user_id: Querying user (matched by the ``user_id`` branch, which is the
@@ -188,6 +192,13 @@ def build_ownership_filter(
other_owners = [owner for owner in owners if owner != user_id]
conditions: list[Condition] = [
FieldCondition(key="user_id", match=MatchValue(value=user_id)),
# Observed-access branch: a file deduplicated across users carries one
# user-agnostic point set whose ``user_id``/``owner_id`` are the first
# indexer's. ``acl_principals`` lists every user whose scanner has seen
# (hence can read) the file, so this branch surfaces a shared/group-folder
# file to every reader even when they were not the indexer. Verify-on-read
# (_verify_files) is the precise ACL gate on the returned candidates.
FieldCondition(key="acl_principals", match=MatchAny(any=[f"user:{user_id}"])),
]
if other_owners:
conditions.insert(
+10 -23
View File
@@ -9,10 +9,7 @@ simply re-verify and re-attempt.
import logging
from qdrant_client.models import FieldCondition, Filter, MatchValue
from nextcloud_mcp_server.config import get_settings
from nextcloud_mcp_server.vector.qdrant_client import get_qdrant_client
from nextcloud_mcp_server.vector.sharing_state import release_document_for_user
logger = logging.getLogger(__name__)
@@ -22,39 +19,29 @@ async def delete_document_points(
doc_type: str,
user_id: str,
) -> None:
"""Remove all Qdrant points for a single document.
"""Revoke one user's access to a document's points (verify-on-read eviction).
Deletes both real chunk points and any leftover placeholder points for the
given (user_id, doc_id, doc_type) tuple. Safe to call when the document is
A document can be indexed once and shared across users (user-agnostic point
IDs), so eviction must *release* this user rather than blindly delete: it
drops ``user:<user_id>`` from the point's ``acl_principals`` and removes the
points only when no reader remains. Legacy points without a principal set
fall back to the original per-user delete. Safe to call when the document is
not present — Qdrant returns successfully with zero points affected.
Args:
doc_id: Document ID (str — keyword-indexed in Qdrant payload)
doc_type: Document type (note, file, deck_card, news_item)
user_id: Owner of the points being evicted
user_id: User whose access is being revoked
Raises:
Exception: If the underlying Qdrant client raises. Callers in the
search hot path should catch and log; eviction failures must not
block search responses.
"""
qdrant_client = await get_qdrant_client()
settings = get_settings()
await qdrant_client.delete(
collection_name=settings.get_collection_name(),
points_selector=Filter(
must=[
FieldCondition(key="user_id", match=MatchValue(value=user_id)),
FieldCondition(key="doc_id", match=MatchValue(value=doc_id)),
FieldCondition(key="doc_type", match=MatchValue(value=doc_type)),
]
),
)
await release_document_for_user(doc_id, doc_type, user_id)
logger.info(
"Evicted Qdrant points for %s_%s (user=%s); "
"document was inaccessible at verification time",
"Released %s_%s for user=%s; document was inaccessible at verification time",
doc_type,
doc_id,
user_id,
+67 -21
View File
@@ -11,7 +11,7 @@ from typing import Any, cast
import anyio
from anyio.abc import TaskStatus
from anyio.streams.memory import MemoryObjectReceiveStream
from qdrant_client.models import FieldCondition, Filter, MatchValue, PointStruct
from qdrant_client.models import PointStruct
from nextcloud_mcp_server.acl_hash import compute_acl_hash
from nextcloud_mcp_server.client import NextcloudClient
@@ -33,6 +33,11 @@ from nextcloud_mcp_server.vector.html_processor import html_to_markdown
from nextcloud_mcp_server.vector.placeholder import delete_placeholder_point
from nextcloud_mcp_server.vector.qdrant_client import get_qdrant_client
from nextcloud_mcp_server.vector.scanner import DocumentTask
from nextcloud_mcp_server.vector.sharing_state import (
claim_existing_index,
existing_principals,
release_document_for_user,
)
logger = logging.getLogger(__name__)
@@ -192,28 +197,15 @@ async def process_document(
):
try:
qdrant_client = await get_qdrant_client()
settings = get_settings()
# Handle deletion
if doc_task.operation == "delete":
await qdrant_client.delete(
collection_name=settings.get_collection_name(),
points_selector=Filter(
must=[
FieldCondition(
key="user_id",
match=MatchValue(value=doc_task.user_id),
),
FieldCondition(
key="doc_id",
match=MatchValue(value=doc_task.doc_id),
),
FieldCondition(
key="doc_type",
match=MatchValue(value=doc_task.doc_type),
),
]
),
# Release this user rather than blind-delete: a file shared across
# users has one user-agnostic point set referenced by multiple
# principals, so the points are removed only once the last reader
# is gone (see vector/sharing_state.release_document_for_user).
await release_document_for_user(
doc_task.doc_id, doc_task.doc_type, doc_task.user_id
)
logger.info(
"Deleted %s_%s for %s",
@@ -479,6 +471,28 @@ async def _index_document(
)
file_path = doc_task.file_path
# Cross-worker dedup race-guard: two users' tasks for the same shared
# file can be enqueued before either finishes. If another worker has
# already indexed this exact content (fileid + etag + embedding model)
# in the tenant, claim it for this user (observed-access ACL) and skip
# the expensive fetch/parse/embed entirely.
if doc_task.etag and await claim_existing_index(
doc_task.doc_id, "file", doc_task.etag, doc_task.user_id
):
await delete_placeholder_point(
doc_id=doc_task.doc_id,
doc_type="file",
user_id=doc_task.user_id,
)
logger.info(
"Dedup hit for file %s (etag=%s); claimed for user %s "
"without reprocessing",
doc_task.doc_id,
doc_task.etag,
doc_task.user_id,
)
return
# Read file content via WebDAV
content_bytes, content_type = await nc_client.webdav.read_file(file_path)
else:
@@ -510,7 +524,10 @@ async def _index_document(
content = result.text
file_metadata = result.metadata
title = file_metadata.get("title") or file_path.split("/")[-1]
etag = "" # WebDAV read_file doesn't return etag
# etag comes from the scanner's tag REPORT (threaded via the
# DocumentTask); read_file itself returns no etag. It is the
# tenant-wide content-dedup key, so it must be persisted.
etag = doc_task.etag or ""
# Diagnostic: Log page boundary information if available
if "page_boundaries" in file_metadata:
@@ -763,6 +780,28 @@ async def _index_document(
_embedding_identity = settings.get_embedding_model_name()
_acl_hash = compute_acl_hash([("user", doc_task.user_id)])
# Observed-access ACL principals (computed once per document, not per chunk).
# Seed with the indexer (and owner, if distinct). For files — the only type
# with cross-user dedup and globally-unique IDs (Nextcloud fileid) — union in
# any principals already recorded so re-indexing after a content change
# preserves visibility for readers who had previously claimed the file. For
# note/news_item/deck_card, IDs are per-user (not globally unique) and point
# IDs are user-agnostic, so merging another user's principals on an ID
# collision would wrongly cross-surface their content; those types are
# seeded with the indexer only.
_prior_principals = (
await existing_principals(doc_task.doc_id, doc_task.doc_type)
if doc_task.doc_type == "file"
else []
)
_acl_principals = sorted(
set(_prior_principals)
| {
f"user:{doc_task.user_id}",
f"user:{doc_task.owner_id or doc_task.user_id}",
}
)
# Surface deck card data quality issues at indexing time rather than
# only at verification time (where _verify_deck_cards falls through to
# legacy-data pass-through when board_id/stack_id are missing). This is
@@ -807,6 +846,13 @@ async def _index_document(
# crawl shared-with-them content can set owner_id to the
# true owner without losing the "who indexed this" trail.
"owner_id": doc_task.owner_id or doc_task.user_id,
# Observed-access ACL set: every user whose scanner has seen
# (hence can read) this document. Seeded with the indexer (and
# owner, if distinct); grown lazily as other readers' scanners
# hit the tenant-wide dedup path. Search ORs a
# MatchAny(acl_principals, ["user:<me>"]) branch so a
# deduplicated shared file stays findable by every reader.
"acl_principals": _acl_principals,
"doc_id": doc_task.doc_id,
"doc_type": doc_task.doc_type,
"is_placeholder": False, # Real indexed document (not placeholder)
@@ -70,6 +70,19 @@ _PAYLOAD_INDEX_FIELDS: dict[str, PayloadSchemaType] = {
# migration like modified_at. Local/embedded qdrant-client matches by
# substring without an index, so dev stacks work without it too.
"file_path": PayloadSchemaType.TEXT,
# etag is the tenant-wide content-dedup key: the scanner/processor scroll for
# a non-placeholder point matching (doc_id, doc_type, etag) to decide whether
# a file's content is already indexed and reprocessing can be skipped (see
# vector/sharing_state.find_indexed_content). KEYWORD for exact match; the
# value is the Nextcloud etag already written to every point. Idempotent
# startup migration like the fields above.
"etag": PayloadSchemaType.KEYWORD,
# acl_principals is the observed-access ACL set ("user:<uid>" entries). Search
# ORs MatchAny(key="acl_principals", any=["user:<me>"]) so a deduplicated
# shared point reaches every reader's candidate set (see
# search/access_filter.build_ownership_filter); KEYWORD indexes match array
# membership element-wise. Idempotent startup migration.
"acl_principals": PayloadSchemaType.KEYWORD,
}
# Sentinel point that records "this collection has been backfilled to str
+33 -1
View File
@@ -31,6 +31,7 @@ from nextcloud_mcp_server.vector.placeholder import (
)
from nextcloud_mcp_server.vector.qdrant_client import get_qdrant_client
from nextcloud_mcp_server.vector.queue.ports import TaskProducer
from nextcloud_mcp_server.vector.sharing_state import claim_existing_index
logger = logging.getLogger(__name__)
@@ -360,7 +361,16 @@ async def scan_user_documents(
return
# Scan tagged PDF files (after notes)
# Get indexed file IDs from Qdrant (for deletion tracking)
# Get indexed file IDs from Qdrant (for deletion tracking).
# NOTE: this is filtered by user_id, so a "pure claimer" — a user who
# gained access to a shared file via the tenant-wide dedup path
# (claim_existing_index) without ever indexing it themselves — is NOT in
# this set (the points carry the first indexer's user_id, only the
# claimer's user:<uid> in acl_principals). Such a user is therefore never
# enqueued for deletion by the grace-period sweep below; their stale
# acl_principals entry is reclaimed lazily by verify-on-read eviction
# (release_document_for_user) when a search surfaces a now-inaccessible
# result. A future scanner-side cleanup could scroll acl_principals too.
indexed_file_ids = set()
if not initial_sync:
assert qdrant_client is not None # narrow for the type checker
@@ -448,6 +458,24 @@ async def scan_user_documents(
except (ValueError, KeyError):
pass
# Tenant-wide content dedup (Layer 1 / observed-access ACL): if
# this exact file content (fileid + etag) is already indexed under
# the current embedding model by ANY user in the tenant, skip
# re-parsing/re-embedding and just record that this user can read
# it. Eliminates the per-user reprocessing ping-pong that arises
# because chunk point IDs are user-agnostic (note 386945 #5).
etag = str(file_info.get("etag") or "")
if etag and await claim_existing_index(file_id, "file", etag, user_id):
_potentially_deleted.pop((user_id, file_id), None)
logger.debug(
"Dedup: file %s (ID: %s) already indexed in tenant; "
"granted access to %s without reprocessing",
file_path,
file_id,
user_id,
)
continue
if initial_sync:
# Send everything on first sync - write placeholder first
await write_placeholder_point(
@@ -455,6 +483,7 @@ async def scan_user_documents(
doc_type="file",
user_id=user_id,
modified_at=modified_at,
etag=etag,
file_path=file_path,
)
await send_stream.send(
@@ -465,6 +494,7 @@ async def scan_user_documents(
operation="index",
modified_at=modified_at,
file_path=file_path,
etag=etag,
)
)
file_queued += 1
@@ -525,6 +555,7 @@ async def scan_user_documents(
doc_type="file",
user_id=user_id,
modified_at=modified_at,
etag=etag,
file_path=file_path,
)
await send_stream.send(
@@ -535,6 +566,7 @@ async def scan_user_documents(
operation="index",
modified_at=modified_at,
file_path=file_path,
etag=etag,
)
)
file_queued += 1
@@ -0,0 +1,286 @@
"""Tenant-wide content dedup + observed-access ACL state for the vector index.
A file shared across users (directly, or via a group folder shared to a group)
has one Nextcloud ``fileid`` and one ``etag`` for everyone, and chunk point IDs
are user-agnostic (``uuid5(tenant_id, doc_id, chunk_index)`` — see
``vector/payload_keys.py``). So two users indexing the same file produce the
*same* points. The per-user freshness gate (filtered by ``user_id``) nonetheless
made them re-parse + re-embed the identical content on every scan (note 386945,
finding #5). This module lets the pipeline detect "already indexed by someone in
this tenant" and skip the expensive work.
Visibility is handled by an *observed-access* model rather than push-enumeration
of share/group-folder grants (which the server cannot read without admin creds —
group membership and the GroupFolders API are admin-only, and WebDAV PROPFIND
carries no ACL). The per-user scanner crawl is itself the access oracle: a tagged
file appears in a user's ``find_files_by_tag`` REPORT **iff** that user can read
it. So each point carries ``acl_principals`` — the set of ``user:<uid>`` whose
scanner has observed (hence can access) the file. The search filter ORs a
``MatchAny(acl_principals, ["user:<me>"])`` branch, and ``_verify_files`` (the
verify-on-read gate) re-checks each result against the user's tagged REPORT, so
an over-broad principal match can never leak content.
All point IDs are user-agnostic, so deletion must *release one user* (drop their
principal) and only remove the points when the principal set empties — otherwise
one user untagging a shared file would evict it for everyone still reading it.
"""
from __future__ import annotations
import logging
from qdrant_client.models import FieldCondition, Filter, MatchValue
from nextcloud_mcp_server.config import get_settings
from nextcloud_mcp_server.vector import payload_keys
from nextcloud_mcp_server.vector.placeholder import get_placeholder_filter
from nextcloud_mcp_server.vector.qdrant_client import get_qdrant_client
logger = logging.getLogger(__name__)
ACL_PRINCIPALS_KEY = "acl_principals"
def user_principal(user_id: str) -> str:
"""The ``acl_principals`` entry representing a single user's read access."""
return f"user:{user_id}"
def _document_filter(doc_id: str, doc_type: str, *, real_only: bool) -> Filter:
"""Match every chunk of one document; optionally exclude placeholder points."""
must: list = [
FieldCondition(key="doc_id", match=MatchValue(value=doc_id)),
FieldCondition(key="doc_type", match=MatchValue(value=doc_type)),
]
if real_only:
must.append(get_placeholder_filter())
return Filter(must=must)
async def find_indexed_content(
doc_id: str,
doc_type: str,
etag: str,
embedding_identity: str,
) -> dict | None:
"""Return a real point's payload if this exact content is already indexed.
Looks tenant-wide (no ``user_id`` filter) for a non-placeholder point with
the given ``doc_id``/``doc_type``/``etag``. The match is gated on
``embedding_identity`` in Python (not the Qdrant filter, to avoid requiring an
index on that field): since point IDs are model-agnostic, a model switch
overwrites the same points, so all live points for a doc share one identity —
a mismatch means the existing vectors were produced by a different model and
must be re-embedded, so we report "not indexed".
Returns the payload dict (including ``acl_principals``) on a hit, else None.
"""
if not etag:
return None
qdrant_client = await get_qdrant_client()
settings = get_settings()
points, _ = await qdrant_client.scroll(
collection_name=settings.get_collection_name(),
scroll_filter=Filter(
must=[
FieldCondition(key="doc_id", match=MatchValue(value=doc_id)),
FieldCondition(key="doc_type", match=MatchValue(value=doc_type)),
FieldCondition(key="etag", match=MatchValue(value=etag)),
get_placeholder_filter(),
]
),
limit=1,
with_payload=True,
with_vectors=False,
)
if not points:
return None
payload = dict(points[0].payload or {})
if payload.get(payload_keys.EMBEDDING_IDENTITY) != embedding_identity:
# Existing vectors were produced by a different embedding model — a
# re-embed is required, so this content is not reusable as-is.
return None
return payload
async def existing_principals(doc_id: str, doc_type: str) -> list[str]:
"""Return the ``acl_principals`` already recorded for a document (or []).
Used when re-indexing after a content change (etag differs, so the dedup
race-guard misses and the points are overwritten): seeding the new points
with the prior principal set preserves visibility for readers who had
already claimed the file, instead of resetting it to just the indexer.
"""
qdrant_client = await get_qdrant_client()
settings = get_settings()
points, _ = await qdrant_client.scroll(
collection_name=settings.get_collection_name(),
scroll_filter=_document_filter(doc_id, doc_type, real_only=True),
limit=1,
with_payload=True,
with_vectors=False,
)
if not points:
return []
return list(dict(points[0].payload or {}).get(ACL_PRINCIPALS_KEY) or [])
async def add_principal(
doc_id: str,
doc_type: str,
user_id: str,
current_principals: list[str] | None,
) -> bool:
"""Record that ``user_id`` can read this document (observed-access ACL).
No-op (returns False) when the user's principal is already present — so the
steady state of a repeat scan writes nothing. Otherwise unions the principal
onto every real chunk of the document via a single ``set_payload`` and
returns True. Concurrent adds race to a last-writer-wins union; a dropped
add is re-applied on the losing user's next scan, and verify-on-read gates
correctness in the meantime.
"""
principal = user_principal(user_id)
existing = current_principals or []
if principal in existing:
return False
new_principals = sorted(set(existing) | {principal})
qdrant_client = await get_qdrant_client()
settings = get_settings()
await qdrant_client.set_payload(
collection_name=settings.get_collection_name(),
payload={ACL_PRINCIPALS_KEY: new_principals},
points=_document_filter(doc_id, doc_type, real_only=True),
wait=True,
)
logger.debug(
"Granted read principal %s on %s_%s (now %d principal(s))",
principal,
doc_type,
doc_id,
len(new_principals),
)
return True
async def claim_existing_index(
doc_id: str,
doc_type: str,
etag: str,
user_id: str,
) -> bool:
"""Tenant-wide dedup claim: skip reprocessing if content is already indexed.
Returns True when a non-placeholder point for this exact content (fileid +
etag + current embedding model) already exists for some user in the tenant —
in which case ``user_id`` is added to ``acl_principals`` (so the file remains
searchable for them) and the caller should skip fetch/parse/embed. Returns
False when nothing reusable exists and the document must be processed.
Fail-safe: a Qdrant error during the lookup degrades to False (process the
document normally) rather than aborting the scan — the dedup is an
optimisation, never a correctness gate. A failure to record the principal
after a confirmed hit is non-fatal (logged, not raised): verify-on-read still
gates access and the user's next scan re-claims it.
"""
embedding_identity = get_settings().get_embedding_model_name()
try:
existing = await find_indexed_content(
doc_id, doc_type, etag, embedding_identity
)
except Exception as exc: # noqa: BLE001 — degrade to "process normally"
logger.warning(
"Dedup lookup failed for %s_%s (%s); processing without dedup",
doc_type,
doc_id,
exc,
)
return False
if existing is None:
return False
try:
await add_principal(doc_id, doc_type, user_id, existing.get(ACL_PRINCIPALS_KEY))
except Exception as exc: # noqa: BLE001 — non-fatal; recovered on next scan
logger.warning(
"Failed to grant read principal user:%s on %s_%s (%s); "
"verify-on-read and the next scan will reconcile",
user_id,
doc_type,
doc_id,
exc,
)
return True
async def release_document_for_user(
doc_id: str,
doc_type: str,
user_id: str,
) -> None:
"""Drop ``user_id``'s access to a document; delete points only when orphaned.
Replaces a blind per-document delete. Because point IDs are user-agnostic, a
shared document has one point set referenced by multiple principals; removing
one user must not evict it for the others. Removes the user's principal and
deletes the points only once no principal remains.
Legacy points written before ``acl_principals`` existed have no principal
set; for those we preserve the original behaviour (delete by
``user_id``/``doc_id``/``doc_type``) so a single-owner delete still works.
"""
qdrant_client = await get_qdrant_client()
settings = get_settings()
collection = settings.get_collection_name()
points, _ = await qdrant_client.scroll(
collection_name=collection,
scroll_filter=_document_filter(doc_id, doc_type, real_only=True),
limit=1,
with_payload=True,
with_vectors=False,
)
principals = (
(dict(points[0].payload or {}).get(ACL_PRINCIPALS_KEY)) if points else None
)
if not principals:
# No real points, or legacy points without a principal set: fall back to
# the original per-user delete (also clears this user's placeholder).
await qdrant_client.delete(
collection_name=collection,
points_selector=Filter(
must=[
FieldCondition(key="user_id", match=MatchValue(value=user_id)),
FieldCondition(key="doc_id", match=MatchValue(value=doc_id)),
FieldCondition(key="doc_type", match=MatchValue(value=doc_type)),
]
),
)
return
remaining = sorted(p for p in principals if p != user_principal(user_id))
if not remaining:
# Last reader released — remove every point (real + placeholder).
await qdrant_client.delete(
collection_name=collection,
points_selector=_document_filter(doc_id, doc_type, real_only=False),
)
logger.info(
"Released last principal for %s_%s — document removed from index",
doc_type,
doc_id,
)
else:
await qdrant_client.set_payload(
collection_name=collection,
payload={ACL_PRINCIPALS_KEY: remaining},
points=_document_filter(doc_id, doc_type, real_only=True),
wait=True,
)
logger.info(
"Released principal %s for %s_%s%d reader(s) remain",
user_principal(user_id),
doc_type,
doc_id,
len(remaining),
)