Skip to content

fix: upsert-before-delete-stale and per-service reindex lock #91

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
GoodbyePlanet merged 1 commit into from
Jul 14, 2026
Merged

fix: upsert-before-delete-stale and per-service reindex lock #91

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
194 changes: 101 additions & 93 deletions server/indexer/git_history.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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__)
Expand Down Expand Up @@ -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,
Expand Down
Loading
Loading