diff --git a/server/indexer/git_history.py b/server/indexer/git_history.py index 33eebc9..157866e 100644 --- a/server/indexer/git_history.py +++ b/server/indexer/git_history.py @@ -17,6 +17,7 @@ ) from server.indexer.cleanup import prune_orphaned_services from server.indexer.pipeline import ProgressEvent +from server.state import get_reindex_lock from server.store.commit_store import CommitStore logger = logging.getLogger(__name__) @@ -81,116 +82,123 @@ async def index_service( if svc is None: return {"error": 1, "new": 0, "skipped": 0} - async with httpx.AsyncClient() as http_client: - commits = await list_commits( - settings.github_token, - svc.github_repo, - svc.github_ref, - root=svc.root, - max_commits=settings.git_history_max_commits, - client=http_client, - ) - - if progress_callback: - await progress_callback( - ProgressEvent( - phase="discovery", - current=len(commits), - total=len(commits), - percentage=100.0, - service=service_name, + async with get_reindex_lock(f"history:{svc.name}"): + async with httpx.AsyncClient() as http_client: + commits = await list_commits( + settings.github_token, + svc.github_repo, + svc.github_ref, + root=svc.root, + max_commits=settings.git_history_max_commits, + client=http_client, ) - ) - - existing_shas = set() if force else await self._store.get_indexed_shas(svc.name) - new_commits = [c for c in commits if c.sha not in existing_shas] - skipped = len(commits) - len(new_commits) - - shas_without_diffs = ( - set(await self._store.get_commits_without_diffs(svc.name)) - if not force - else set() - ) - commits_needing_diffs = [c for c in commits if c.sha in shas_without_diffs] - - if not new_commits and not commits_needing_diffs: - return {"new": 0, "skipped": skipped, "diff_updated": 0} - - diff_updated = 0 - - if new_commits: - logger.info( - "Fetching diffs for %d new commits in %s", - len(new_commits), - service_name, - ) - new_commits = await fetch_commits_with_diffs( - settings.github_token, svc.github_repo, new_commits - ) - texts = [_build_embedding_text(c, svc.name) for c in new_commits] - try: - vectors = await self._embedder.embed_batch(texts) - except Exception as exc: - logger.error( - "Embedding failed for %s git history: %s", service_name, exc - ) - return {"error": 1, "new": 0, "skipped": skipped, "diff_updated": 0} if progress_callback: await progress_callback( ProgressEvent( - phase="embedding", - current=len(new_commits), - total=len(new_commits), + phase="discovery", + current=len(commits), + total=len(commits), percentage=100.0, service=service_name, ) ) - payloads = [_commit_to_payload(c, svc.name) for c in new_commits] - await self._store.upsert_commits(svc.name, payloads, vectors) - logger.info("Indexed %d new commits for %s", len(new_commits), service_name) + existing_shas = ( + set() if force else await self._store.get_indexed_shas(svc.name) + ) + new_commits = [c for c in commits if c.sha not in existing_shas] + skipped = len(commits) - len(new_commits) - if progress_callback: - await progress_callback( - ProgressEvent( - phase="upserting", - current=len(new_commits), - total=len(new_commits), - percentage=100.0, - service=service_name, - ) + shas_without_diffs = ( + set(await self._store.get_commits_without_diffs(svc.name)) + if not force + else set() + ) + commits_needing_diffs = [c for c in commits if c.sha in shas_without_diffs] + + if not new_commits and not commits_needing_diffs: + return {"new": 0, "skipped": skipped, "diff_updated": 0} + + diff_updated = 0 + + if new_commits: + logger.info( + "Fetching diffs for %d new commits in %s", + len(new_commits), + service_name, + ) + new_commits = await fetch_commits_with_diffs( + settings.github_token, svc.github_repo, new_commits ) + texts = [_build_embedding_text(c, svc.name) for c in new_commits] + try: + vectors = await self._embedder.embed_batch(texts) + except Exception as exc: + logger.error( + "Embedding failed for %s git history: %s", service_name, exc + ) + return {"error": 1, "new": 0, "skipped": skipped, "diff_updated": 0} + + if progress_callback: + await progress_callback( + ProgressEvent( + phase="embedding", + current=len(new_commits), + total=len(new_commits), + percentage=100.0, + service=service_name, + ) + ) - if commits_needing_diffs: - logger.info( - "Fetching diffs for %d existing commits in %s", - len(commits_needing_diffs), - service_name, - ) - commits_needing_diffs = await fetch_commits_with_diffs( - settings.github_token, svc.github_repo, commits_needing_diffs - ) - payloads = [_commit_to_payload(c, svc.name) for c in commits_needing_diffs] - await self._store.update_commit_diffs(svc.name, payloads) - diff_updated = len(commits_needing_diffs) + payloads = [_commit_to_payload(c, svc.name) for c in new_commits] + await self._store.upsert_commits(svc.name, payloads, vectors) + logger.info( + "Indexed %d new commits for %s", len(new_commits), service_name + ) - if progress_callback: - await progress_callback( - ProgressEvent( - phase="upserting", - current=diff_updated, - total=diff_updated, - percentage=100.0, - service=service_name, + if progress_callback: + await progress_callback( + ProgressEvent( + phase="upserting", + current=len(new_commits), + total=len(new_commits), + percentage=100.0, + service=service_name, + ) ) + + if commits_needing_diffs: + logger.info( + "Fetching diffs for %d existing commits in %s", + len(commits_needing_diffs), + service_name, + ) + commits_needing_diffs = await fetch_commits_with_diffs( + settings.github_token, svc.github_repo, commits_needing_diffs ) + payloads = [ + _commit_to_payload(c, svc.name) for c in commits_needing_diffs + ] + await self._store.update_commit_diffs(svc.name, payloads) + diff_updated = len(commits_needing_diffs) + + if progress_callback: + await progress_callback( + ProgressEvent( + phase="upserting", + current=diff_updated, + total=diff_updated, + percentage=100.0, + service=service_name, + ) + ) - return { - "new": len(new_commits), - "skipped": skipped, - "diff_updated": diff_updated, - } + return { + "new": len(new_commits), + "skipped": skipped, + "diff_updated": diff_updated, + } async def index_all( self, diff --git a/server/indexer/pipeline.py b/server/indexer/pipeline.py index 9bd8aa5..39a19b8 100644 --- a/server/indexer/pipeline.py +++ b/server/indexer/pipeline.py @@ -18,6 +18,7 @@ from server.indexer.github_source import fetch_blob_content, list_github_files from server.parser.base import CodeSymbol, ParseError from server.parser.registry import parse_file +from server.state import get_reindex_lock from server.store.qdrant import QdrantStore logger = logging.getLogger(__name__) @@ -161,120 +162,131 @@ async def index_service( if svc is None: return {"error": 1, "files": 0, "chunks": 0} - async with httpx.AsyncClient() as http_client: - github_files = await list_github_files( - settings.github_token, - svc.github_repo, - svc.github_ref, - svc.name, - svc.exclude, - svc.root, - client=http_client, - ) + async with get_reindex_lock(f"code:{svc.name}"): + async with httpx.AsyncClient() as http_client: + github_files = await list_github_files( + settings.github_token, + svc.github_repo, + svc.github_ref, + svc.name, + svc.exclude, + svc.root, + client=http_client, + ) - if progress_callback: - await progress_callback( - ProgressEvent( - phase="discovery", - current=len(github_files), - total=len(github_files), - percentage=100.0, - service=service_name, + if progress_callback: + await progress_callback( + ProgressEvent( + phase="discovery", + current=len(github_files), + total=len(github_files), + percentage=100.0, + service=service_name, + ) ) - ) - existing_hashes = await self._store.get_indexed_file_hashes(svc.name) + existing_hashes = await self._store.get_indexed_file_hashes(svc.name) - indexed_files = 0 - total_chunks = 0 - skipped = 0 - total_files = len(github_files) + indexed_files = 0 + total_chunks = 0 + skipped = 0 + total_files = len(github_files) - for i, f in enumerate(github_files): - # "{service_name}/{path_in_repo}" — consistent path format across all tools - stored_path = f"{svc.name}/{f.rel_path}" + for i, f in enumerate(github_files): + # "{service_name}/{path_in_repo}" — consistent path format across all tools + stored_path = f"{svc.name}/{f.rel_path}" - # blob_sha IS the content fingerprint — no download needed to detect unchanged files - if not force and existing_hashes.get(stored_path) == f.blob_sha: - skipped += 1 - continue + # blob_sha IS the content fingerprint — no download needed to detect unchanged files + if not force and existing_hashes.get(stored_path) == f.blob_sha: + skipped += 1 + continue - try: - content = await fetch_blob_content( - settings.github_token, - svc.github_repo, - f.blob_sha, - client=http_client, - ) - except Exception as exc: - logger.error("Failed to fetch %s: %s", stored_path, exc) - continue - - try: - symbols = parse_file(content, stored_path) - except ParseError: - logger.error( - "Skipping index update for %s: parser failed, " - "existing entries preserved", - stored_path, + try: + content = await fetch_blob_content( + settings.github_token, + svc.github_repo, + f.blob_sha, + client=http_client, + ) + except Exception as exc: + logger.error("Failed to fetch %s: %s", stored_path, exc) + continue + + try: + symbols = parse_file(content, stored_path) + except ParseError: + logger.error( + "Skipping index update for %s: parser failed, " + "existing entries preserved", + stored_path, + ) + continue + + if not symbols: + # File has no indexable symbols; clean up any stale entries. + await self._store.delete_by_file(svc.name, stored_path) + continue + + texts_dense = [_build_embedding_text(s, svc.name) for s in symbols] + texts_sparse = [_build_bm25_text(s) for s in symbols] + try: + dense_vectors = await self._embedder.embed_batch(texts_dense) + sparse_vectors = await self._sparse_embedder.embed_batch( + texts_sparse + ) + except Exception as exc: + logger.error("Embedding failed for %s: %s", stored_path, exc) + continue # keep existing index entries until embedding succeeds + + payloads = [ + _symbol_to_payload(s, svc.name, f.blob_sha) for s in symbols + ] + # Upsert new/changed symbols before deleting stale ones, so there's + # never a window where the file has zero indexed symbols. + previous_ids = await self._store.get_point_ids_by_file( + svc.name, stored_path ) - continue - - if not symbols: - # File has no indexable symbols; clean up any stale entries. - await self._store.delete_by_file(svc.name, stored_path) - continue - - texts_dense = [_build_embedding_text(s, svc.name) for s in symbols] - texts_sparse = [_build_bm25_text(s) for s in symbols] - try: - dense_vectors = await self._embedder.embed_batch(texts_dense) - sparse_vectors = await self._sparse_embedder.embed_batch( - texts_sparse + new_ids = await self._store.upsert_chunks( + payloads, dense_vectors, sparse_vectors ) - except Exception as exc: - logger.error("Embedding failed for %s: %s", stored_path, exc) - continue # keep existing index entries until embedding succeeds - - payloads = [ - _symbol_to_payload(s, svc.name, f.blob_sha) for s in symbols - ] - await self._store.delete_by_file(svc.name, stored_path) - await self._store.upsert_chunks(payloads, dense_vectors, sparse_vectors) + stale_ids = previous_ids - set(new_ids) + await self._store.delete_by_ids(list(stale_ids)) + + indexed_files += 1 + total_chunks += len(symbols) + logger.info("Indexed %s: %d symbols", stored_path, len(symbols)) + + if progress_callback: + await progress_callback( + ProgressEvent( + phase="upserting", + current=i + 1, + total=total_files, + percentage=round( + (i + 1) / max(total_files, 1) * 100, 1 + ), + service=service_name, + ) + ) - indexed_files += 1 - total_chunks += len(symbols) - logger.info("Indexed %s: %d symbols", stored_path, len(symbols)) + all_stored_paths = {f"{svc.name}/{f.rel_path}" for f in github_files} + stale_paths = [p for p in existing_hashes if p not in all_stored_paths] + for stale_path in stale_paths: + await self._store.delete_by_file(svc.name, stale_path) + logger.info("Removed stale file from index: %s", stale_path) - if progress_callback: - await progress_callback( - ProgressEvent( - phase="upserting", - current=i + 1, - total=total_files, - percentage=round((i + 1) / max(total_files, 1) * 100, 1), - service=service_name, - ) + if progress_callback and stale_paths: + await progress_callback( + ProgressEvent( + phase="cleanup", + current=len(stale_paths), + total=len(stale_paths), + percentage=100.0, + service=service_name, ) - - all_stored_paths = {f"{svc.name}/{f.rel_path}" for f in github_files} - stale_paths = [p for p in existing_hashes if p not in all_stored_paths] - for stale_path in stale_paths: - await self._store.delete_by_file(svc.name, stale_path) - logger.info("Removed stale file from index: %s", stale_path) - - if progress_callback and stale_paths: - await progress_callback( - ProgressEvent( - phase="cleanup", - current=len(stale_paths), - total=len(stale_paths), - percentage=100.0, - service=service_name, ) - ) - return {"files": indexed_files, "chunks": total_chunks, "skipped": skipped} + return {"files": indexed_files, "chunks": total_chunks, "skipped": skipped} async def index_all( self, diff --git a/server/state.py b/server/state.py index a57f093..2f95af4 100644 --- a/server/state.py +++ b/server/state.py @@ -1,5 +1,8 @@ from __future__ import annotations +import asyncio +from collections import defaultdict + from server.embeddings.bm25 import BM25SparseProvider from server.store.commit_store import CommitStore from server.store.qdrant import QdrantStore @@ -7,6 +10,7 @@ _store: QdrantStore | None = None _commit_store: CommitStore | None = None _sparse_provider: BM25SparseProvider | None = None +_reindex_locks: defaultdict[str, asyncio.Lock] = defaultdict(asyncio.Lock) def get_store() -> QdrantStore: @@ -40,3 +44,10 @@ def get_sparse_provider() -> BM25SparseProvider: def set_sparse_provider(provider: BM25SparseProvider) -> None: global _sparse_provider _sparse_provider = provider + + +def get_reindex_lock(key: str) -> asyncio.Lock: + """Returns the lock guarding reindexing for `key` (e.g. "code:my-service"), + so concurrent reindex requests for the same service serialize instead of + racing each other's delete/upsert calls.""" + return _reindex_locks[key] diff --git a/server/store/qdrant.py b/server/store/qdrant.py index 1cf1e1b..7b533c9 100644 --- a/server/store/qdrant.py +++ b/server/store/qdrant.py @@ -98,8 +98,9 @@ async def upsert_chunks( chunks: list[dict[str, Any]], dense_vectors: list[list[float]], sparse_vectors: list[SparseVector], - ) -> None: + ) -> list[str]: points = [] + point_ids = [] for chunk, dense, sparse in zip(chunks, dense_vectors, sparse_vectors): point_id = _symbol_point_id( chunk["service"], @@ -107,6 +108,7 @@ async def upsert_chunks( chunk["symbol_name"], chunk["start_line"], ) + point_ids.append(point_id) points.append( PointStruct( id=point_id, @@ -116,6 +118,37 @@ async def upsert_chunks( ) if points: await self._client.upsert(collection_name=self._collection, points=points) + return point_ids + + async def get_point_ids_by_file(self, service: str, file_path: str) -> set[str]: + """Returns the ids of all points currently stored for this file.""" + ids: set[str] = set() + offset = None + scroll_filter = Filter( + must=[ + FieldCondition(key="service", match=MatchValue(value=service)), + FieldCondition(key="file_path", match=MatchValue(value=file_path)), + ] + ) + while True: + results, offset = await self._client.scroll( + collection_name=self._collection, + scroll_filter=scroll_filter, + limit=1000, + offset=offset, + with_payload=False, + with_vectors=False, + ) + ids.update(str(point.id) for point in results) + if offset is None: + break + return ids + + async def delete_by_ids(self, point_ids: list[str]) -> None: + if point_ids: + await self._client.delete( + collection_name=self._collection, points_selector=point_ids + ) async def delete_by_file(self, service: str, file_path: str) -> None: await self._client.delete( diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py index e20c6cf..f98a544 100644 --- a/tests/test_pipeline.py +++ b/tests/test_pipeline.py @@ -258,6 +258,8 @@ async def test_force_reindexes_even_when_hash_unchanged() -> None: store = AsyncMock() store.get_indexed_file_hashes.return_value = {"svc/unchanged.py": "same-sha"} + store.get_point_ids_by_file.return_value = {"old-id"} + store.upsert_chunks.return_value = ["new-id"] pipeline = _make_pipeline(store) @@ -277,8 +279,8 @@ async def test_force_reindexes_even_when_hash_unchanged() -> None: ): result = await pipeline.index_service("svc", force=True) - store.delete_by_file.assert_called_once_with("svc", "svc/unchanged.py") store.upsert_chunks.assert_called_once() + store.delete_by_ids.assert_called_once_with(["old-id"]) assert result == {"files": 1, "chunks": 1, "skipped": 0} @@ -379,7 +381,9 @@ async def test_embedding_failure_preserves_existing_index_entries() -> None: assert result["files"] == 0 -async def test_delete_by_file_called_before_upsert_for_changed_file() -> None: +async def test_upsert_called_before_deleting_stale_ids_for_changed_file() -> None: + """Regression test for GH-69: the file must never have zero indexed symbols + mid-update, so new/changed symbols are upserted before stale ones are deleted.""" svc = ServiceConfig(name="svc", github_repo="org/repo", exclude=[]) github_file = GitHubFile(rel_path="mod.py", blob_sha="new-sha") symbol = CodeSymbol( @@ -395,8 +399,14 @@ async def test_delete_by_file_called_before_upsert_for_changed_file() -> None: call_order: list[str] = [] store = AsyncMock() store.get_indexed_file_hashes.return_value = {"svc/mod.py": "old-sha"} - store.delete_by_file.side_effect = lambda *a, **k: call_order.append("delete") - store.upsert_chunks.side_effect = lambda *a, **k: call_order.append("upsert") + store.get_point_ids_by_file.return_value = {"old-id"} + + async def _upsert(*a, **k): + call_order.append("upsert") + return ["new-id"] + + store.upsert_chunks.side_effect = _upsert + store.delete_by_ids.side_effect = lambda *a, **k: call_order.append("delete") pipeline = _make_pipeline(store) @@ -416,7 +426,8 @@ async def test_delete_by_file_called_before_upsert_for_changed_file() -> None: ): await pipeline.index_service("svc", force=False) - assert call_order == ["delete", "upsert"] + assert call_order == ["upsert", "delete"] + store.delete_by_ids.assert_called_once_with(["old-id"]) async def test_unknown_service_returns_error_without_touching_github() -> None: @@ -426,3 +437,39 @@ async def test_unknown_service_returns_error_without_touching_github() -> None: result = await pipeline.index_service("missing") assert result == {"error": 1, "files": 0, "chunks": 0} + + +async def test_concurrent_index_service_calls_for_same_service_are_serialized() -> None: + """Regression test for GH-69: overlapping reindex requests for the same + service must not run concurrently and race each other's store writes.""" + import asyncio + + svc = ServiceConfig(name="concurrent-svc", github_repo="org/repo", exclude=[]) + store = AsyncMock() + store.get_indexed_file_hashes.return_value = {} + + active = 0 + max_active = 0 + + async def _list_github_files(*a, **k): + nonlocal active, max_active + active += 1 + max_active = max(max_active, active) + await asyncio.sleep(0.01) + active -= 1 + return [] + + pipeline = _make_pipeline(store) + + with ( + patch.object( + type(pipeline_module.settings), "load_services", return_value=[svc] + ), + patch.object(pipeline_module, "list_github_files", _list_github_files), + ): + await asyncio.gather( + pipeline.index_service("concurrent-svc"), + pipeline.index_service("concurrent-svc"), + ) + + assert max_active == 1