Skip to content

Commit aba7632

Browse files
GoodbyePlanetGoodbyePlanet
authored
Merge pull request #91 from GoodbyePlanet/fix/reindex-atomic-upsert-and-lock
fix: upsert-before-delete-stale and per-service reindex lock
2 parents f80c61e + 2cdf49c commit aba7632

5 files changed

Lines changed: 310 additions & 199 deletions

File tree

‎server/indexer/git_history.py‎

Lines changed: 101 additions & 93 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@
1717
)
1818
from server.indexer.cleanup import prune_orphaned_services
1919
from server.indexer.pipeline import ProgressEvent
20+
from server.state import get_reindex_lock
2021
from server.store.commit_store import CommitStore
2122

2223
logger = logging.getLogger(__name__)
@@ -81,116 +82,123 @@ async def index_service(
8182
if svc is None:
8283
return {"error": 1, "new": 0, "skipped": 0}
8384

84-
async with httpx.AsyncClient() as http_client:
85-
commits = await list_commits(
86-
settings.github_token,
87-
svc.github_repo,
88-
svc.github_ref,
89-
root=svc.root,
90-
max_commits=settings.git_history_max_commits,
91-
client=http_client,
92-
)
93-
94-
if progress_callback:
95-
await progress_callback(
96-
ProgressEvent(
97-
phase="discovery",
98-
current=len(commits),
99-
total=len(commits),
100-
percentage=100.0,
101-
service=service_name,
85+
async with get_reindex_lock(f"history:{svc.name}"):
86+
async with httpx.AsyncClient() as http_client:
87+
commits = await list_commits(
88+
settings.github_token,
89+
svc.github_repo,
90+
svc.github_ref,
91+
root=svc.root,
92+
max_commits=settings.git_history_max_commits,
93+
client=http_client,
10294
)
103-
)
104-
105-
existing_shas = set() if force else await self._store.get_indexed_shas(svc.name)
106-
new_commits = [c for c in commits if c.sha not in existing_shas]
107-
skipped = len(commits) - len(new_commits)
108-
109-
shas_without_diffs = (
110-
set(await self._store.get_commits_without_diffs(svc.name))
111-
if not force
112-
else set()
113-
)
114-
commits_needing_diffs = [c for c in commits if c.sha in shas_without_diffs]
115-
116-
if not new_commits and not commits_needing_diffs:
117-
return {"new": 0, "skipped": skipped, "diff_updated": 0}
118-
119-
diff_updated = 0
120-
121-
if new_commits:
122-
logger.info(
123-
"Fetching diffs for %d new commits in %s",
124-
len(new_commits),
125-
service_name,
126-
)
127-
new_commits = await fetch_commits_with_diffs(
128-
settings.github_token, svc.github_repo, new_commits
129-
)
130-
texts = [_build_embedding_text(c, svc.name) for c in new_commits]
131-
try:
132-
vectors = await self._embedder.embed_batch(texts)
133-
except Exception as exc:
134-
logger.error(
135-
"Embedding failed for %s git history: %s", service_name, exc
136-
)
137-
return {"error": 1, "new": 0, "skipped": skipped, "diff_updated": 0}
13895

13996
if progress_callback:
14097
await progress_callback(
14198
ProgressEvent(
142-
phase="embedding",
143-
current=len(new_commits),
144-
total=len(new_commits),
99+
phase="discovery",
100+
current=len(commits),
101+
total=len(commits),
145102
percentage=100.0,
146103
service=service_name,
147104
)
148105
)
149106

150-
payloads = [_commit_to_payload(c, svc.name) for c in new_commits]
151-
await self._store.upsert_commits(svc.name, payloads, vectors)
152-
logger.info("Indexed %d new commits for %s", len(new_commits), service_name)
107+
existing_shas = (
108+
set() if force else await self._store.get_indexed_shas(svc.name)
109+
)
110+
new_commits = [c for c in commits if c.sha not in existing_shas]
111+
skipped = len(commits) - len(new_commits)
153112

154-
if progress_callback:
155-
await progress_callback(
156-
ProgressEvent(
157-
phase="upserting",
158-
current=len(new_commits),
159-
total=len(new_commits),
160-
percentage=100.0,
161-
service=service_name,
162-
)
113+
shas_without_diffs = (
114+
set(await self._store.get_commits_without_diffs(svc.name))
115+
if not force
116+
else set()
117+
)
118+
commits_needing_diffs = [c for c in commits if c.sha in shas_without_diffs]
119+
120+
if not new_commits and not commits_needing_diffs:
121+
return {"new": 0, "skipped": skipped, "diff_updated": 0}
122+
123+
diff_updated = 0
124+
125+
if new_commits:
126+
logger.info(
127+
"Fetching diffs for %d new commits in %s",
128+
len(new_commits),
129+
service_name,
130+
)
131+
new_commits = await fetch_commits_with_diffs(
132+
settings.github_token, svc.github_repo, new_commits
163133
)
134+
texts = [_build_embedding_text(c, svc.name) for c in new_commits]
135+
try:
136+
vectors = await self._embedder.embed_batch(texts)
137+
except Exception as exc:
138+
logger.error(
139+
"Embedding failed for %s git history: %s", service_name, exc
140+
)
141+
return {"error": 1, "new": 0, "skipped": skipped, "diff_updated": 0}
142+
143+
if progress_callback:
144+
await progress_callback(
145+
ProgressEvent(
146+
phase="embedding",
147+
current=len(new_commits),
148+
total=len(new_commits),
149+
percentage=100.0,
150+
service=service_name,
151+
)
152+
)
164153

165-
if commits_needing_diffs:
166-
logger.info(
167-
"Fetching diffs for %d existing commits in %s",
168-
len(commits_needing_diffs),
169-
service_name,
170-
)
171-
commits_needing_diffs = await fetch_commits_with_diffs(
172-
settings.github_token, svc.github_repo, commits_needing_diffs
173-
)
174-
payloads = [_commit_to_payload(c, svc.name) for c in commits_needing_diffs]
175-
await self._store.update_commit_diffs(svc.name, payloads)
176-
diff_updated = len(commits_needing_diffs)
154+
payloads = [_commit_to_payload(c, svc.name) for c in new_commits]
155+
await self._store.upsert_commits(svc.name, payloads, vectors)
156+
logger.info(
157+
"Indexed %d new commits for %s", len(new_commits), service_name
158+
)
177159

178-
if progress_callback:
179-
await progress_callback(
180-
ProgressEvent(
181-
phase="upserting",
182-
current=diff_updated,
183-
total=diff_updated,
184-
percentage=100.0,
185-
service=service_name,
160+
if progress_callback:
161+
await progress_callback(
162+
ProgressEvent(
163+
phase="upserting",
164+
current=len(new_commits),
165+
total=len(new_commits),
166+
percentage=100.0,
167+
service=service_name,
168+
)
186169
)
170+
171+
if commits_needing_diffs:
172+
logger.info(
173+
"Fetching diffs for %d existing commits in %s",
174+
len(commits_needing_diffs),
175+
service_name,
176+
)
177+
commits_needing_diffs = await fetch_commits_with_diffs(
178+
settings.github_token, svc.github_repo, commits_needing_diffs
187179
)
180+
payloads = [
181+
_commit_to_payload(c, svc.name) for c in commits_needing_diffs
182+
]
183+
await self._store.update_commit_diffs(svc.name, payloads)
184+
diff_updated = len(commits_needing_diffs)
185+
186+
if progress_callback:
187+
await progress_callback(
188+
ProgressEvent(
189+
phase="upserting",
190+
current=diff_updated,
191+
total=diff_updated,
192+
percentage=100.0,
193+
service=service_name,
194+
)
195+
)
188196

189-
return {
190-
"new": len(new_commits),
191-
"skipped": skipped,
192-
"diff_updated": diff_updated,
193-
}
197+
return {
198+
"new": len(new_commits),
199+
"skipped": skipped,
200+
"diff_updated": diff_updated,
201+
}
194202

195203
async def index_all(
196204
self,

0 commit comments

Comments
 (0)