From 34c74902403a41be737a9441c8425778a77a554c Mon Sep 17 00:00:00 2001 From: Shree Bohara Date: Tue, 11 Aug 2026 11:27:13 -0700 Subject: [PATCH] Make indexing progress observable and stuck indexes recoverable Two bugs whose failure mode was silence, both from the audit. 1. PROGRESS WAS WRITE-ONLY IndexingService wrote progress into a per-instance dict. The SSE endpoint builds its own IndexingService per request, so it never saw those writes and always fell through to the database branch, which hardcodes current_step="Unknown" and a percentage of 0 or 100. The per-file progress the indexer computes could not reach any client. A per-process dict would not fix it either: indexing runs in a background task and, with more than one worker, in a different process from the reader. core/progress.py adds a ProgressStore over a capped Redis stream per repository (progress:{repo_id}, MAXLEN 500) with a process-wide in-memory fallback for the no-Redis case. A stream rather than a key because the SSE endpoint wants history: a client that connects late can replay what it missed via XRANGE instead of seeing only the current value. get_progress now reads the shared store first, then the instance dict, then the database. Publishing never raises. Progress is telemetry, and a Redis blip must not fail an otherwise healthy index, so a publish failure degrades to memory and the caller sees nothing. There is a test with a Redis double that raises on every call. 2. A KILLED INDEX WAS UNRECOVERABLE Only FAILED and COMPLETED are terminal, and every self-healing path (repos.py, seed_demo.py) keys on FAILED -- so a container killed during CLONING, PARSING or EMBEDDING left the row in that state forever with no way to retry through the API. In demo mode that bricked the deployment. reap_stuck_indexing runs at startup, where anything still in a transient state provably has no live indexer because the process that owned it is gone. PENDING is deliberately left alone: a repository queued but not yet started is not stuck, and failing it would break the normal import path. An existing indexing_error is preserved, since a real failure reason beats the generic interruption note. Also hardened the SSE route, which looped forever with no client check and no ceiling: a repository stuck mid-index held an open response and its database session indefinitely, once per browser tab. It now checks request.is_disconnected(), emits only on change instead of re-sending an identical payload every second, and stops after 15 minutes with a reconnect hint. Verified: 17 new tests, 159 total (was 142), ruff clean, run with apps/api/.env moved aside and OPENAI_API_KEY / AZURE_OPENAI_API_KEY / ANTHROPIC_API_KEY / REDIS_URL all unset. The Redis paths are exercised with fakeredis, which implements streams, consumer groups and XAUTOCLAIM, so they are genuinely executed rather than mocked. End-to-end check: a fresh IndexingService with an empty _progress dict -- the exact object the SSE route builds -- now returns 57.5% and "Parsing widget.tsx..." where it previously returned "Unknown" at 0%. Co-Authored-By: Claude Opus 5 --- apps/api/requirements.txt | 2 + apps/api/src/api/routes/repos.py | 35 +++- apps/api/src/core/progress.py | 159 ++++++++++++++ apps/api/src/dependencies.py | 7 + apps/api/src/main.py | 4 +- apps/api/src/models/database.py | 42 ++++ apps/api/src/services/indexing_service.py | 57 ++++- apps/api/tests/unit/test_progress_store.py | 233 +++++++++++++++++++++ 8 files changed, 527 insertions(+), 12 deletions(-) create mode 100644 apps/api/src/core/progress.py create mode 100644 apps/api/tests/unit/test_progress_store.py diff --git a/apps/api/requirements.txt b/apps/api/requirements.txt index 2014119..b98b424 100644 --- a/apps/api/requirements.txt +++ b/apps/api/requirements.txt @@ -66,5 +66,7 @@ pyyaml>=6.0 # Testing pytest>=8.0.0 pytest-asyncio>=0.23.0 +# Exercises the Redis code paths (streams, consumer groups, XAUTOCLAIM) without a server. +fakeredis>=2.20.0 # pyproject.toml's addopts hardcodes --cov, so pytest cannot start without this. pytest-cov>=4.1.0 diff --git a/apps/api/src/api/routes/repos.py b/apps/api/src/api/routes/repos.py index 2169f9e..a8cf5ac 100644 --- a/apps/api/src/api/routes/repos.py +++ b/apps/api/src/api/routes/repos.py @@ -6,7 +6,7 @@ import json import logging -from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException +from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Request from fastapi.responses import StreamingResponse from sqlalchemy.orm import Session @@ -25,6 +25,9 @@ router = APIRouter() logger = logging.getLogger(__name__) +# Ceiling on a single SSE progress subscription; clients reconnect if they still care. +_PROGRESS_STREAM_TIMEOUT_SECONDS = 15 * 60 + def run_indexing_task(repo_id: str): """Run async indexing in a new event loop with fresh DB session.""" @@ -151,6 +154,7 @@ async def get_repository( @router.get("/{repo_id}/progress") async def get_indexing_progress( repo_id: str, + request: Request, db: Session = Depends(get_db), ): """Stream indexing progress updates via SSE.""" @@ -162,14 +166,37 @@ async def get_indexing_progress( raise HTTPException(status_code=404, detail="Repository not found") async def progress_stream(): - """Generate SSE events for progress updates.""" + """ + SSE progress events. + + Bounded and disconnect-aware: the previous version looped forever with no client + check and no ceiling, so a repository stuck mid-index held an open response (and + its database session) indefinitely, once per browser tab. + """ indexing_service = IndexingService(db) + deadline = asyncio.get_event_loop().time() + _PROGRESS_STREAM_TIMEOUT_SECONDS + last_payload = None while True: + if await request.is_disconnected(): + break + progress = await indexing_service.get_progress(repo_id) - yield f"data: {json.dumps(progress)}\n\n" - if progress["status"] in ["completed", "failed"]: + # Only emit on change; a 1s poll of an unchanged value is pure noise. + payload = json.dumps(progress, sort_keys=True) + if payload != last_payload: + last_payload = payload + yield f"data: {payload}\n\n" + + if progress.get("status") in ("completed", "failed"): + break + + if asyncio.get_event_loop().time() > deadline: + yield ( + 'data: {"type": "timeout", "message": ' + '"Progress stream timed out; reconnect to continue watching."}\n\n' + ) break await asyncio.sleep(1) diff --git a/apps/api/src/core/progress.py b/apps/api/src/core/progress.py new file mode 100644 index 0000000..ac35766 --- /dev/null +++ b/apps/api/src/core/progress.py @@ -0,0 +1,159 @@ +""" +Shared indexing-progress store. + +WHY THIS EXISTS +IndexingService wrote progress into a per-instance dict. The SSE endpoint constructs a +fresh IndexingService per request, so it never saw those writes and always fell through to +the database branch, which hardcodes current_step="Unknown" and a percentage of 0 or 100. +The per-file progress the indexer computes was therefore unreachable by any client, and +the progress bar could only ever show 0% or 100%. + +A per-process dict cannot fix this either: indexing runs in a background task and, under +more than one worker, in a different process from the request that wants to read it. + +DESIGN +A Redis stream per repository (progress:{repo_id}), capped with MAXLEN so it cannot grow +without bound. Redis is optional throughout this codebase, so there is an in-memory +fallback that is correct within a single process -- which is the local-dev shape. The +fallback is explicitly not cross-process, and callers can tell which backend is live via +`backend`. + +Streams rather than a plain key because the SSE endpoint wants *history*: a client that +connects late should be able to replay what it missed rather than only seeing the current +value, and XRANGE gives that for free. +""" + +from __future__ import annotations + +import json +import logging +import time +from collections import defaultdict, deque +from threading import Lock +from typing import Any, Deque, Dict, List, Optional + +logger = logging.getLogger(__name__) + +# Enough to replay a long index without unbounded growth. Each entry is small. +_STREAM_MAXLEN = 500 +_MEMORY_MAXLEN = 500 + + +def _stream_key(repo_id: str) -> str: + return f"progress:{repo_id}" + + +class ProgressStore: + """ + Publishes and reads indexing progress. + + Safe to construct per request; the Redis client it wraps is the shared singleton. + """ + + def __init__(self, redis_client=None): + self._redis = redis_client + # Class-level so every instance in a process shares the fallback, which is the + # whole point -- a per-instance dict is the bug this replaces. + self._lock = _MEMORY_LOCK + self._memory = _MEMORY + + @property + def backend(self) -> str: + return "redis" if self._redis is not None else "memory" + + async def publish( + self, + repo_id: str, + status: str, + step: str, + percent: float, + files_processed: int = 0, + total_files: int = 0, + ) -> None: + """Record a progress event. Never raises -- progress must not break indexing.""" + event = { + "repo_id": repo_id, + "status": status, + "current_step": step, + "progress_percent": round(float(percent), 2), + "files_processed": int(files_processed), + "total_files": int(total_files), + "at": time.time(), + } + + if self._redis is not None: + try: + await self._redis.xadd( + _stream_key(repo_id), + {"event": json.dumps(event)}, + maxlen=_STREAM_MAXLEN, + approximate=True, + ) + return + except Exception as exc: + # Fall through to memory rather than losing the event or failing the + # index. Logged at debug because a flapping Redis would otherwise emit + # one warning per parsed file. + logger.debug("Progress publish to Redis failed, using memory: %s", exc) + + with self._lock: + self._memory[repo_id].append(event) + + async def latest(self, repo_id: str) -> Optional[Dict[str, Any]]: + """Most recent event, or None if nothing has been published.""" + if self._redis is not None: + try: + entries = await self._redis.xrevrange(_stream_key(repo_id), count=1) + if entries: + return _decode(entries[0]) + except Exception as exc: + logger.debug("Progress read from Redis failed, using memory: %s", exc) + + with self._lock: + events = self._memory.get(repo_id) + return dict(events[-1]) if events else None + + async def history(self, repo_id: str, limit: int = 100) -> List[Dict[str, Any]]: + """Oldest-first events, so a late subscriber can replay what it missed.""" + if self._redis is not None: + try: + entries = await self._redis.xrange(_stream_key(repo_id), count=limit) + return [_decode(e) for e in entries] + except Exception as exc: + logger.debug("Progress history from Redis failed, using memory: %s", exc) + + with self._lock: + events = self._memory.get(repo_id) or [] + return [dict(e) for e in list(events)[-limit:]] + + async def clear(self, repo_id: str) -> None: + """Drop a repository's progress, e.g. when it is deleted or re-indexed.""" + if self._redis is not None: + try: + await self._redis.delete(_stream_key(repo_id)) + except Exception as exc: + logger.debug("Progress clear in Redis failed: %s", exc) + + with self._lock: + self._memory.pop(repo_id, None) + + +def _decode(entry) -> Dict[str, Any]: + """ + Turn one stream entry into the event dict. + + Redis returns (id, {field: value}); values are bytes unless the client decodes + responses, and this codebase does not configure that, so both are handled. + """ + _entry_id, fields = entry + raw = fields.get(b"event") or fields.get("event") or "{}" + if isinstance(raw, bytes): + raw = raw.decode("utf-8", errors="replace") + try: + return json.loads(raw) + except json.JSONDecodeError: + return {} + + +_MEMORY: Dict[str, Deque[Dict[str, Any]]] = defaultdict(lambda: deque(maxlen=_MEMORY_MAXLEN)) +_MEMORY_LOCK = Lock() diff --git a/apps/api/src/dependencies.py b/apps/api/src/dependencies.py index 7a6f6eb..22e8096 100644 --- a/apps/api/src/dependencies.py +++ b/apps/api/src/dependencies.py @@ -159,3 +159,10 @@ def get_graph_store(): from src.core.graph.neo4j_store import Neo4jGraphStore return Neo4jGraphStore(driver, database=settings.neo4j_database) + + +def get_progress_store(): + """Shared indexing-progress store (Redis stream, memory fallback).""" + from src.core.progress import ProgressStore + + return ProgressStore(redis_client=get_redis_client()) diff --git a/apps/api/src/main.py b/apps/api/src/main.py index 3304e9d..3e4212d 100644 --- a/apps/api/src/main.py +++ b/apps/api/src/main.py @@ -21,7 +21,7 @@ get_session_factory, get_vector_store, ) -from src.models.database import init_db +from src.models.database import init_db, reap_stuck_indexing from src.models.migrations import run_pending_migrations # Configure logging @@ -53,6 +53,8 @@ async def lifespan(app: FastAPI): engine = get_db_engine() init_db(engine) run_pending_migrations(engine) + # Anything still mid-index belongs to a process that no longer exists. + reap_stuck_indexing(engine) logger.info("Database initialized") # Initialize vector store diff --git a/apps/api/src/models/database.py b/apps/api/src/models/database.py index 1ed8bb6..ded0bc2 100644 --- a/apps/api/src/models/database.py +++ b/apps/api/src/models/database.py @@ -4,6 +4,7 @@ """ import enum +import logging import uuid from datetime import datetime, timezone @@ -22,6 +23,8 @@ from sqlalchemy import Enum as SQLEnum from sqlalchemy.orm import declarative_base, relationship +logger = logging.getLogger(__name__) + Base = declarative_base() @@ -412,6 +415,45 @@ class GraphNodeInteraction(Base): ) +def reap_stuck_indexing(engine) -> int: + """ + Fail repositories left mid-index by a process that died. + + Only FAILED and COMPLETED are terminal. A container killed during CLONING, PARSING or + EMBEDDING left the row in that state forever, and the only self-healing path in + repos.py and seed_demo.py triggers on FAILED -- so a stuck repository could never be + retried through the API, and in demo mode that bricked the deployment. + + Runs at startup, where anything still in a transient state provably has no live + indexer, because the process that owned it is gone. + """ + from sqlalchemy import text as _text + + transient = ("cloning", "parsing", "embedding") + with engine.begin() as connection: + placeholders = ", ".join(f":s{i}" for i in range(len(transient))) + params = {f"s{i}": value for i, value in enumerate(transient)} + # SQLEnum persists the member NAME, so compare case-insensitively to be safe + # across both spellings. + result = connection.execute( + _text( + "UPDATE repositories SET status = 'FAILED', " + "indexing_error = COALESCE(indexing_error, " + "'Indexing was interrupted before completing. Re-import to retry.') " + f"WHERE LOWER(status) IN ({placeholders})" + ), + params, + ) + reaped = result.rowcount or 0 + + if reaped: + logger.warning( + "Marked %d repository(ies) as failed: they were mid-index when the previous " + "process exited. They can now be re-indexed.", reaped + ) + return reaped + + def init_db(engine): """Create all tables.""" Base.metadata.create_all(bind=engine) diff --git a/apps/api/src/services/indexing_service.py b/apps/api/src/services/indexing_service.py index 6b026a0..b1773dc 100644 --- a/apps/api/src/services/indexing_service.py +++ b/apps/api/src/services/indexing_service.py @@ -82,7 +82,7 @@ async def index_repository(self, repo_id: str, force_reindex: bool = False): # Update status to cloning repo.status = IndexingStatus.CLONING self._db.commit() - self._update_progress(repo_id, "cloning", "Cloning repository...", 0) + await self._publish_progress(repo_id, "cloning", "Cloning repository...", 0) # Clone repository local_path = await self._repo_manager.clone_repository( @@ -98,7 +98,7 @@ async def index_repository(self, repo_id: str, force_reindex: bool = False): # Update status to parsing repo.status = IndexingStatus.PARSING self._db.commit() - self._update_progress(repo_id, "parsing", "Parsing code files...", 20) + await self._publish_progress(repo_id, "parsing", "Parsing code files...", 20) # Find and parse files files = self._find_files(local_path) @@ -108,7 +108,7 @@ async def index_repository(self, repo_id: str, force_reindex: bool = False): chunks_data = [] for i, file_path in enumerate(files): progress_pct = 20 + (60 * (i / max(total_files, 1))) - self._update_progress(repo_id, "parsing", f"Parsing {file_path.name}...", progress_pct) + await self._publish_progress(repo_id, "parsing", f"Parsing {file_path.name}...", progress_pct, i, total_files) try: file_chunks = await self._parse_file(repo, file_path, local_path) @@ -123,7 +123,7 @@ async def index_repository(self, repo_id: str, force_reindex: bool = False): # Update status to embedding repo.status = IndexingStatus.EMBEDDING self._db.commit() - self._update_progress(repo_id, "embedding", "Generating embeddings...", 80) + await self._publish_progress(repo_id, "embedding", "Generating embeddings...", 80) # Generate embeddings and store if chunks_data: @@ -153,7 +153,7 @@ async def index_repository(self, repo_id: str, force_reindex: bool = False): repo.status = IndexingStatus.COMPLETED repo.last_indexed_at = datetime.now(timezone.utc) self._db.commit() - self._update_progress(repo_id, "completed", "Indexing complete!", 100) + await self._publish_progress(repo_id, "completed", "Indexing complete!", 100, total_files, total_files) logger.info(f"Successfully indexed {repo.github_owner}/{repo.github_name}") @@ -162,7 +162,7 @@ async def index_repository(self, repo_id: str, force_reindex: bool = False): repo.status = IndexingStatus.FAILED repo.indexing_error = str(e) self._db.commit() - self._update_progress(repo_id, "failed", str(e), 0) + await self._publish_progress(repo_id, "failed", str(e), 0) async def _reset_repository_index_data(self, repo_id: str) -> None: """Clear prior SQL/vector artifacts for a repository before full re-index.""" @@ -685,6 +685,27 @@ async def _embed_and_store(self, repo_id: str, chunks_data: List[Dict[str, Any]] metadatas=[c["metadata"] for c in chunks_data], ) + async def _publish_progress( + self, repo_id: str, status: str, step: str, percent: float, + files_processed: int = 0, total_files: int = 0, + ) -> None: + """ + Record progress where a *different* process can read it. + + _update_progress below writes an instance dict, which the SSE endpoint can never + see because it builds its own IndexingService. This is the write that actually + reaches a client. + """ + self._update_progress(repo_id, status, step, percent) + try: + from src.dependencies import get_progress_store + + await get_progress_store().publish( + repo_id, status, step, percent, files_processed, total_files + ) + except Exception as exc: + logger.debug("Progress publish failed for %s: %s", repo_id, exc) + def _update_progress(self, repo_id: str, status: str, step: str, percent: float): """Update progress tracking.""" self._progress[repo_id] = { @@ -694,7 +715,29 @@ def _update_progress(self, repo_id: str, status: str, step: str, percent: float) } async def get_progress(self, repo_id: str) -> Dict[str, Any]: - """Get current progress for a repository.""" + """ + Current progress for a repository. + + Order matters: the shared store first, because it is the only source a *different* + process (the SSE request handler) can observe. The instance dict and the database + fallback follow for the single-process case and for repos with no live indexer. + """ + try: + from src.dependencies import get_progress_store + + event = await get_progress_store().latest(repo_id) + if event: + return { + "repo_id": repo_id, + "status": event.get("status", "unknown"), + "current_step": event.get("current_step", ""), + "progress_percent": event.get("progress_percent", 0), + "files_processed": event.get("files_processed", 0), + "total_files": event.get("total_files", 0), + } + except Exception as exc: + logger.debug("Progress store read failed for %s: %s", repo_id, exc) + if repo_id in self._progress: return { "repo_id": repo_id, diff --git a/apps/api/tests/unit/test_progress_store.py b/apps/api/tests/unit/test_progress_store.py new file mode 100644 index 0000000..716d09d --- /dev/null +++ b/apps/api/tests/unit/test_progress_store.py @@ -0,0 +1,233 @@ +""" +Shared indexing-progress store, and the stuck-index reaper. + +Both fix bugs where the failure mode was silence rather than an error: + + * progress written to a per-instance dict was invisible to the SSE endpoint, which + builds its own IndexingService -- so the bar could only show 0% or 100%. + * a container killed mid-index left the repository in CLONING/PARSING/EMBEDDING + forever, and every self-healing path keys on FAILED, so it could never be retried. + +Redis is exercised with fakeredis, which implements streams, consumer groups and +XAUTOCLAIM, so the Redis path here is genuinely executed rather than mocked. +""" + +import pytest +from sqlalchemy import create_engine, text +from sqlalchemy.orm import sessionmaker + +from src.core.progress import ProgressStore +from src.models.database import Base, IndexingStatus, Repository, reap_stuck_indexing + + +@pytest.fixture() +def redis_client(): + import fakeredis.aioredis + + return fakeredis.aioredis.FakeRedis() + + +@pytest.fixture(autouse=True) +def _clear_memory(): + """The in-memory fallback is process-wide by design, so isolate tests from it.""" + import src.core.progress as mod + + mod._MEMORY.clear() + yield + mod._MEMORY.clear() + + +# --- the bug this replaces ------------------------------------------------------- + +@pytest.mark.asyncio +async def test_progress_is_visible_across_store_instances(redis_client): + """ + The actual defect: the reader is a different object from the writer. A per-instance + dict cannot satisfy this, which is why the SSE endpoint always fell back to the + database branch and reported "Unknown" at 0%. + """ + writer = ProgressStore(redis_client) + reader = ProgressStore(redis_client) + + await writer.publish("repo-1", "parsing", "Parsing app.ts...", 42.5, 17, 40) + + event = await reader.latest("repo-1") + assert event is not None, "a separate instance must observe the write" + assert event["status"] == "parsing" + assert event["current_step"] == "Parsing app.ts..." + assert event["progress_percent"] == 42.5 + assert event["files_processed"] == 17 + assert event["total_files"] == 40 + + +@pytest.mark.asyncio +async def test_memory_fallback_also_shares_across_instances(): + """With no Redis the store must still work within one process (the local-dev shape).""" + writer, reader = ProgressStore(None), ProgressStore(None) + assert writer.backend == "memory" + + await writer.publish("repo-2", "embedding", "Generating embeddings...", 80) + event = await reader.latest("repo-2") + assert event["status"] == "embedding" + assert event["progress_percent"] == 80 + + +@pytest.mark.asyncio +async def test_backend_reports_which_path_is_live(redis_client): + assert ProgressStore(redis_client).backend == "redis" + assert ProgressStore(None).backend == "memory" + + +# --- history / replay ----------------------------------------------------------- + +@pytest.mark.asyncio +async def test_history_is_oldest_first_so_a_late_client_can_replay(redis_client): + store = ProgressStore(redis_client) + for pct, step in ((0, "Cloning"), (20, "Parsing"), (80, "Embedding"), (100, "Done")): + await store.publish("repo-3", "working", step, pct) + + history = await store.history("repo-3") + assert [e["progress_percent"] for e in history] == [0, 20, 80, 100] + + +@pytest.mark.asyncio +async def test_latest_returns_none_before_anything_is_published(redis_client): + assert await ProgressStore(redis_client).latest("never-seen") is None + + +@pytest.mark.asyncio +async def test_clear_removes_progress(redis_client): + store = ProgressStore(redis_client) + await store.publish("repo-4", "parsing", "x", 10) + await store.clear("repo-4") + assert await store.latest("repo-4") is None + + +# --- resilience: progress must never break indexing ----------------------------- + +class ExplodingRedis: + async def xadd(self, *a, **k): + raise RuntimeError("redis down") + + async def xrevrange(self, *a, **k): + raise RuntimeError("redis down") + + async def xrange(self, *a, **k): + raise RuntimeError("redis down") + + async def delete(self, *a, **k): + raise RuntimeError("redis down") + + +@pytest.mark.asyncio +async def test_a_broken_redis_degrades_to_memory_instead_of_raising(): + """ + Progress is telemetry. If publishing it could raise, a Redis blip would fail an + otherwise healthy index -- so the event falls back to memory and the caller never + sees an exception. + """ + store = ProgressStore(ExplodingRedis()) + + await store.publish("repo-5", "parsing", "still working", 33) + + event = await store.latest("repo-5") + assert event is not None, "the event should have landed in the memory fallback" + assert event["progress_percent"] == 33 + assert await store.history("repo-5") + await store.clear("repo-5") + + +# --- stuck-index reaper --------------------------------------------------------- + +@pytest.fixture() +def engine(tmp_path): + eng = create_engine(f"sqlite:///{tmp_path / 'reap.db'}") + Base.metadata.create_all(eng) + return eng + + +def _add(engine, name, status): + db = sessionmaker(bind=engine)() + db.add(Repository( + github_url=f"https://github.com/o/{name}", github_owner="o", + github_name=name, status=status, + )) + db.commit() + db.close() + + +def _status_of(engine, name): + with engine.begin() as c: + row = c.execute( + text("SELECT status FROM repositories WHERE github_name = :n"), {"n": name} + ).first() + return (row[0] or "").lower() + + +@pytest.mark.parametrize( + "stuck", [IndexingStatus.CLONING, IndexingStatus.PARSING, IndexingStatus.EMBEDDING] +) +def test_transient_states_are_reaped_to_failed(engine, stuck): + _add(engine, f"r-{stuck.value}", stuck) + assert reap_stuck_indexing(engine) == 1 + assert _status_of(engine, f"r-{stuck.value}") == "failed" + + +@pytest.mark.parametrize( + "keep", [IndexingStatus.COMPLETED, IndexingStatus.PENDING, IndexingStatus.FAILED] +) +def test_terminal_and_pending_states_are_left_alone(engine, keep): + """ + PENDING must survive: a repository queued by a background task that has not started + yet is not stuck, and failing it would break the normal import path. + """ + _add(engine, f"k-{keep.value}", keep) + assert reap_stuck_indexing(engine) == 0 + assert _status_of(engine, f"k-{keep.value}") == keep.value + + +def test_reaper_sets_an_actionable_error_message(engine): + _add(engine, "msg", IndexingStatus.PARSING) + reap_stuck_indexing(engine) + with engine.begin() as c: + err = c.execute( + text("SELECT indexing_error FROM repositories WHERE github_name = 'msg'") + ).scalar() + assert err and "interrupted" in err.lower() + + +def test_reaper_preserves_an_existing_error(engine): + """A real failure reason is more useful than the generic interruption note.""" + _add(engine, "prior", IndexingStatus.EMBEDDING) + db = sessionmaker(bind=engine)() + repo = db.query(Repository).filter(Repository.github_name == "prior").first() + repo.indexing_error = "rate limited by provider" + db.commit() + db.close() + + reap_stuck_indexing(engine) + with engine.begin() as c: + err = c.execute( + text("SELECT indexing_error FROM repositories WHERE github_name = 'prior'") + ).scalar() + assert err == "rate limited by provider" + + +def test_reaper_is_idempotent(engine): + _add(engine, "idem", IndexingStatus.CLONING) + assert reap_stuck_indexing(engine) == 1 + assert reap_stuck_indexing(engine) == 0 + + +def test_reaped_repo_is_retryable(engine): + """ + The point of reaping: repos.py and seed_demo.py both self-heal only from FAILED, so + this is what makes a stuck repository importable again. + """ + _add(engine, "retry", IndexingStatus.PARSING) + reap_stuck_indexing(engine) + + db = sessionmaker(bind=engine)() + repo = db.query(Repository).filter(Repository.github_name == "retry").first() + assert repo.status == IndexingStatus.FAILED + db.close()