diff --git a/docs/0-requirements.ja.md b/docs/0-requirements.ja.md index 239375d..ef6687d 100644 --- a/docs/0-requirements.ja.md +++ b/docs/0-requirements.ja.md @@ -196,10 +196,14 @@ Responsibilities: - vector と metadata を Vectorize に upsert する - 同じ content を D1 FTS5 の `search_docs` table にも upsert する（sparse 側同期、tokenizer_kind は type に応じて `nat` / `code` を自動選択） - embedding 失敗時も次回 retry できる状態を保つ -- D1 FTS5 upsert 失敗は Vectorize upsert を無効化しない（次回 reindex で reconcile） +- embed 経路では D1 FTS5 upsert 失敗は Vectorize upsert を無効化しない。保存した bodyHash が次回の試行を駆動し、次の reindex で sparse 側が reconcile される +- metadata のみの経路（body は変わらず state / labels / milestone / assignees が変わった場合）では、mirror 書き込みの失敗を best-effort 扱いに**しない**。差分検出の基準を進めずに保持し、次の poll / webhook 配信で再試行させる。基準は IssueStore の record そのものなので、失敗した mirror を追い越して基準を進めると取り残しが恒久化する — state だけの変更は、embed 経路が待っている body 変更を二度と連れてこない（issue #209） +- この経路の dense / sparse mirror は互いに独立して書く。vector が欠けている行（issue #210）でも sparse 側の state は更新される - commit diff は 1 commit 分の file リストを batch embed（Workers AI の `text: string[]` 対応を利用）し、1 回の Vectorize upsert で N vector を書き込む - batch size は `MAX_EMBEDDING_BATCH_SIZE`（既定 20）で上限。これを超える commit は複数 batch call に分割する +**取り残した state の修復.** 上の順序欠陥が残した行は、生きている項目として検索に出続ける。通常の poll では到達できない — 差分検出の基準がすでに GitHub と一致しているため。`POST /admin/backfill-issue-state?repo=owner/repo`（installation guide 参照）が repository 単位でこれを揃える。ページングした `state=open` 一覧を正とし、そこに無い索引済み `open` 行を dense / sparse 両側で `closed` にする。再 embed は伴わない（dense 側は既存の値をそのまま再 upsert し、`state` だけ差し替える）。修復は一方向（`open` → `closed`）で、欠陥が生んだ方向に一致する。open 一覧が打ち切られる場合は何もせず中断する — 「一覧に無いこと」が close の根拠だから。 + ### 5. Vector Store (Dense) Vectorize は hybrid retrieval の dense 側を担う。次の metadata を伴う semantic embedding を保持する。 diff --git a/docs/0-requirements.md b/docs/0-requirements.md index b47d2d7..c4db647 100644 --- a/docs/0-requirements.md +++ b/docs/0-requirements.md @@ -197,10 +197,14 @@ Responsibilities: - upsert vectors with metadata into Vectorize - mirror the same content into the D1 FTS5 `search_docs` table for the sparse (BM25) side, choosing tokenizer_kind `nat` or `code` by surface type - keep retryable failures detectable on the next run -- D1 FTS5 upsert failures do not invalidate a successful Vectorize upsert; the next reindex reconciles the sparse side +- D1 FTS5 upsert failures do not invalidate a successful Vectorize upsert on the embed path; the stored bodyHash drives the next attempt and the next reindex reconciles the sparse side +- on the metadata-only path (state / labels / milestone / assignees changed, body did not) a failed mirror write is **not** best-effort: the diff baseline is held so the next poll or webhook delivery retries. The baseline is the IssueStore record itself, so advancing it past a failed mirror makes the miss permanent — a state-only change never brings the body change the embed path waits for (issue #209) +- the dense and sparse mirrors on that path are written independently: a row with no vector (issue #210) still gets its sparse state updated - for commit diffs: batch-embed a commit's file list in a single Workers AI call (`text: string[]`) and upsert the resulting N vectors in one `VECTORIZE.upsert` call - batch size is capped by `MAX_EMBEDDING_BATCH_SIZE` (default 20); commits exceeding it are split across multiple batch calls +**Stale-state repair.** Rows left behind by the ordering defect above keep answering searches as live items, and no ordinary poll reaches them: their diff baseline already matches GitHub. `POST /admin/backfill-issue-state?repo=owner/repo` (see the installation guide) reconciles them per repository — one paginated `state=open` listing supplies the truth, indexed `open` rows absent from it are set to `closed` on both sides, and nothing is re-embedded (the dense side re-upserts the existing values with only `state` replaced). The repair is one-way (`open` → `closed`), which is the direction the defect produced; it aborts rather than act on a truncated open listing, since absence from that listing is what marks a row closed. + ### 5. Vector Store (Dense) Vectorize is the dense side of hybrid retrieval. It stores semantic embeddings and metadata for: diff --git a/docs/installation.ja.md b/docs/installation.ja.md index 9664147..f37f3a3 100644 --- a/docs/installation.ja.md +++ b/docs/installation.ja.md @@ -344,6 +344,47 @@ POST /admin/purge-legacy-vectors?repo=owner/repo - 残差は増えない — 旧形式が書き込まれた期間は移行時点で閉じている。したがってこの endpoint は repository ごとに一回性で、定期実行するものではない - 確認は二重索引されていたファイルを検索し、移行前のコピー（古い内容・dense のみ）が出なくなることを見る +## 14. close 済みなのに `open` のまま残っている索引行を揃える + +issue / PR が open の状態で索引された行は `state: "open"` を持つ。その後の state 変更を索引へ反映するのは metadata のみを更新する経路だが、この経路は自分が守っている mirror 書き込みより**先に**差分検出の基準を進めていた。そのため mirror が失敗しても二度と再試行されず、行は「まだ生きている項目」として検索に出続けた（issue #209）。順序は発生源側で修正済み。この endpoint は旧順序が残した行を修復する。 + +再 embed は伴わない。実際の state は repo ごとに 1 回の `state=open` 一覧から取り、sparse 側は `UPDATE`、dense 側は既存の vector 値をそのまま再 upsert して metadata の `state` だけ差し替える。`/admin/reset-hashes` はこの用途には使えない — repository 全体の再 embed を起こす。 + +Admin endpoint: + +```text +POST /admin/backfill-issue-state?repo=owner/repo +``` + +パラメータ: + +- `repo` — `owner/repo` +- `dry_run` — `true` で件数だけ返し、D1 にも Vectorize にも書かない +- `limit` — 1 回の呼び出しで見る行数、`1..1000`（既定 `200`） +- `cursor` — 再開位置の issue 番号。前回のレスポンスの `nextCursor` をそのまま渡す + +認証: + +- `GITHUB_TOKEN` ヘッダに worker secret と同じ値を送る + +レスポンス: + +```json +{ "repo": "owner/repo", "dryRun": false, "openOnGitHub": 33, "cursor": 0, "limit": 200, + "scanned": 165, "stale": 132, "ftsUpdated": 132, "vectorsUpdated": 130, + "vectorsMissing": 2, "nextCursor": null, "done": true } +``` + +運用上の注意: + +- `done` が `true` になるまで `nextCursor` を渡して繰り返し呼ぶ +- 何度実行しても安全。GitHub 側でまだ open な行は書き込みなしでスキップされるので、完了済みの修復を再実行すると `stale: 0` が返る +- 方向は一方向（`open` → `closed`）。欠陥が生んだ方向であり、検索を害する方向（閉じた判断が生きた検討事項として再供給される）でもある。走査対象が索引全体でなく open 集合の大きさに比例する点も、この方向に限る理由 +- open 一覧が 50 ページを超える場合、何も閉じずにエラーで中断する。「一覧に無いこと」が close の根拠なので、部分的な一覧を使ってはならない +- `vectorsMissing` は対応する vector が無い stale 行の数。これは索引欠落側（issue #210）の面で本 endpoint の範囲外。sparse 側は存在するので、そちらは修復する +- dense 側の書き込みが失敗した場合、D1 に触れる前に呼び出し全体が失敗する。中途半端な修復を残さないための設計なので、同じ `cursor` で再実行する +- 確認は close 済みと分かっている項目を `state: "closed"` で検索するか、`SELECT COUNT(*) FROM search_docs WHERE repo = ? AND type IN ('issue','pull_request') AND state = 'open'` が実際の open 数と一致することを見る + ## Troubleshooting ### `GITHUB_TOKEN not configured` diff --git a/docs/installation.md b/docs/installation.md index f8e9f04..9d21a38 100644 --- a/docs/installation.md +++ b/docs/installation.md @@ -344,6 +344,47 @@ Operational notes: - the residue is closed, not growing — the legacy scheme stopped writing at the migration — so this endpoint is a one-off per repository, not a recurring job - verify by searching for a file that was double-indexed: the pre-migration copy (old content, dense-only) should stop appearing +## 14. Close indexed rows that stayed `open` after the item was closed + +A row indexed while its issue or PR was open carries `state: "open"`. The state change is mirrored onto the index by the metadata-only path, and that path used to advance its own diff baseline before the mirror writes it guards — so a mirror write that failed was never retried, and the row kept answering searches as a live item (issue #209). The ordering is fixed at the source; this endpoint repairs the rows the old ordering left behind. + +Nothing is re-embedded: the true state comes from one `state=open` listing per repo, the sparse side is an `UPDATE`, and the dense side re-upserts the existing vector values with only `state` replaced. `/admin/reset-hashes` is the wrong tool here — it triggers a full re-embedding of the repository. + +Admin endpoint: + +```text +POST /admin/backfill-issue-state?repo=owner/repo +``` + +Parameters: + +- `repo` — `owner/repo` +- `dry_run` — `true` reports the counts and writes to neither D1 nor Vectorize +- `limit` — rows examined per call, `1..1000` (default `200`) +- `cursor` — issue number to resume after; pass back the `nextCursor` of the previous response + +Authentication: + +- send the same `GITHUB_TOKEN` value in the `GITHUB_TOKEN` header + +Response: + +```json +{ "repo": "owner/repo", "dryRun": false, "openOnGitHub": 33, "cursor": 0, "limit": 200, + "scanned": 165, "stale": 132, "ftsUpdated": 132, "vectorsUpdated": 130, + "vectorsMissing": 2, "nextCursor": null, "done": true } +``` + +Operational notes: + +- call it repeatedly, feeding `nextCursor` back in, until `done` is `true` +- safe to repeat: a row GitHub still lists as open is skipped without a write, so a re-run of a finished repair reports `stale: 0` +- the direction is one-way (`open` → `closed`). That is the direction the defect produced and the one that harms retrieval — a closed decision resurfacing as a live one. It also keeps the scan proportional to the open set rather than to the whole index +- the run aborts with an error rather than closing anything if the open-item listing would exceed 50 pages. Absence from that listing is what marks a row closed, so a partial listing must never be used +- `vectorsMissing` counts stale rows with no vector to refresh. Those are the missing-index-entry surface (issue #210) and are out of scope here; their sparse half still exists and is still repaired +- if the dense write fails, the whole call fails before D1 is touched, leaving the window unrepaired rather than half-repaired. Retry with the same `cursor` +- verify with a `state: "closed"` search for an item you know was closed, or by counting `search_docs` rows: `SELECT COUNT(*) FROM search_docs WHERE repo = ? AND type IN ('issue','pull_request') AND state = 'open'` should match the repository's real open count + ## Troubleshooting ### `GITHUB_TOKEN not configured` diff --git a/src/backfill-issue-state.test.ts b/src/backfill-issue-state.test.ts new file mode 100644 index 0000000..270f6d8 --- /dev/null +++ b/src/backfill-issue-state.test.ts @@ -0,0 +1,279 @@ +import { describe, it, expect, vi, afterEach } from "vitest"; +import type { Env } from "./types.js"; +import { + backfillIssueState, + fetchOpenItemNumbers, + DEFAULT_ISSUE_STATE_LIMIT, +} from "./backfill-issue-state.js"; + +const REPO = "acme/widgets"; + +/** Stub the GitHub `state=open` listing, paginating at 100 like the real API. */ +function stubOpenListing(openNumbers: number[]) { + const pages: Array> = []; + for (let i = 0; i < openNumbers.length; i += 100) { + pages.push(openNumbers.slice(i, i + 100).map((number) => ({ number }))); + } + // A full last page must be followed by an empty one, or the walk cannot stop. + if (pages.length === 0 || pages[pages.length - 1].length === 100) pages.push([]); + + const fetchMock = vi.fn(async (input: string | URL) => { + const url = new URL(String(input)); + expect(url.pathname).toBe(`/repos/${REPO}/issues`); + expect(url.searchParams.get("state")).toBe("open"); + const page = Number(url.searchParams.get("page")); + return new Response(JSON.stringify(pages[page - 1] ?? []), { status: 200 }); + }); + vi.stubGlobal("fetch", fetchMock); + return fetchMock; +} + +/** Minimal D1 stub over an in-memory row list; understands only the two + * statements this module issues (the ordered open-row SELECT and the state + * UPDATE by vector_id). */ +function mkDb(rows: Array<{ vector_id: string; number: number; state: string }>) { + const updated: string[] = []; + const batches: number[] = []; + + const db = { + prepare: (sql: string) => ({ + bind: (...args: unknown[]) => ({ + all: async () => { + const [, cursor, limit] = args as [string, number, number]; + const results = rows + .filter((r) => r.state === "open" && r.number > cursor) + .sort((a, b) => a.number - b.number) + .slice(0, limit) + .map((r) => ({ vector_id: r.vector_id, number: r.number })); + return { results }; + }, + // Returned by db.batch below, not executed directly. + __apply: () => { + if (!sql.includes("UPDATE search_docs SET state = 'closed'")) { + throw new Error(`unexpected statement: ${sql}`); + } + const [vectorId] = args as [string]; + const row = rows.find((r) => r.vector_id === vectorId); + if (row) row.state = "closed"; + updated.push(vectorId); + }, + }), + }), + batch: async (stmts: Array<{ __apply: () => void }>) => { + batches.push(stmts.length); + for (const s of stmts) s.__apply(); + return []; + }, + } as unknown as D1Database; + + return { db, rows, updated, batches }; +} + +function mkEnv( + rows: Array<{ vector_id: string; number: number; state: string }>, + opts: { missingVectors?: string[]; vectorizeThrows?: boolean } = {}, +) { + const store = mkDb(rows); + const upserted: Array<{ id: string; metadata: Record }> = []; + const getBatchSizes: number[] = []; + + const env = { + GITHUB_TOKEN: "test-token", + DB_FTS: store.db, + VECTORIZE: { + getByIds: vi.fn(async (ids: string[]) => { + getBatchSizes.push(ids.length); + if (opts.vectorizeThrows) throw new Error("vectorize down"); + return ids + .filter((id) => !(opts.missingVectors ?? []).includes(id)) + .map((id) => ({ id, values: [0.1, 0.2], metadata: { repo: REPO, state: "open" } })); + }), + upsert: vi.fn(async (vectors: Array<{ id: string; metadata: Record }>) => { + upserted.push(...vectors); + }), + }, + } as unknown as Env; + + return { env, ...store, upserted, getBatchSizes }; +} + +/** `n` indexed rows numbered 1..n, all stored as open. */ +function openRows(n: number, offset = 0) { + return Array.from({ length: n }, (_, i) => ({ + vector_id: `i:${i + 1 + offset}`, + number: i + 1 + offset, + state: "open", + })); +} + +afterEach(() => { + vi.unstubAllGlobals(); +}); + +describe("backfill-issue-state: the open set", () => { + it("paginates until a short page and returns every open number", async () => { + const numbers = Array.from({ length: 150 }, (_, i) => i + 1); + const fetchMock = stubOpenListing(numbers); + + const open = await fetchOpenItemNumbers(REPO, "test-token"); + + expect(open.size).toBe(150); + expect(fetchMock).toHaveBeenCalledTimes(2); + }); + + it("refuses to answer from a truncated listing", async () => { + // Every page full: the walk can never conclude, and guessing would close + // items that are actually open. + vi.stubGlobal( + "fetch", + vi.fn(async () => + new Response( + JSON.stringify(Array.from({ length: 100 }, (_, i) => ({ number: i + 1 }))), + { status: 200 }, + ), + ), + ); + + await expect(fetchOpenItemNumbers(REPO, "test-token")).rejects.toThrow(/partial open set/); + }); + + it("surfaces a GitHub API error instead of treating it as an empty open set", async () => { + vi.stubGlobal("fetch", vi.fn(async () => new Response("nope", { status: 403 }))); + + await expect(fetchOpenItemNumbers(REPO, "test-token")).rejects.toThrow(/403/); + }); +}); + +describe("backfill-issue-state: stale detection", () => { + it("closes only the rows GitHub no longer lists as open", async () => { + stubOpenListing([2, 4]); + const { env, rows, updated, upserted } = mkEnv(openRows(5)); + + const summary = await backfillIssueState(REPO, env, {}); + + expect(summary.openOnGitHub).toBe(2); + expect(summary.scanned).toBe(5); + expect(summary.stale).toBe(3); + expect(updated).toEqual(["i:1", "i:3", "i:5"]); + expect(rows.filter((r) => r.state === "open").map((r) => r.number)).toEqual([2, 4]); + expect(upserted.map((v) => v.id)).toEqual(["i:1", "i:3", "i:5"]); + expect(upserted.every((v) => v.metadata.state === "closed")).toBe(true); + // Only `state` moves; the rest of the metadata is carried through as-is. + expect(upserted[0].metadata.repo).toBe(REPO); + }); + + it("writes nothing when every indexed open row is still open", async () => { + stubOpenListing([1, 2, 3]); + const { env, updated, upserted } = mkEnv(openRows(3)); + + const summary = await backfillIssueState(REPO, env, {}); + + expect(summary.stale).toBe(0); + expect(updated).toEqual([]); + expect(upserted).toEqual([]); + expect(summary.done).toBe(true); + }); + + it("is idempotent — a second pass finds nothing left to do", async () => { + stubOpenListing([2]); + const { env, updated } = mkEnv(openRows(3)); + + await backfillIssueState(REPO, env, {}); + stubOpenListing([2]); + const second = await backfillIssueState(REPO, env, {}); + + expect(second.stale).toBe(0); + expect(updated).toEqual(["i:1", "i:3"]); + }); +}); + +describe("backfill-issue-state: dry run", () => { + it("reports the stale count without writing to either side", async () => { + stubOpenListing([2]); + const { env, rows, updated, upserted } = mkEnv(openRows(4)); + + const summary = await backfillIssueState(REPO, env, { dryRun: true }); + + expect(summary.dryRun).toBe(true); + expect(summary.stale).toBe(3); + expect(summary.ftsUpdated).toBe(0); + expect(summary.vectorsUpdated).toBe(0); + expect(updated).toEqual([]); + expect(upserted).toEqual([]); + expect(rows.every((r) => r.state === "open")).toBe(true); + }); +}); + +describe("backfill-issue-state: per-call budget", () => { + it("caps the scan and hands back a resumable cursor", async () => { + stubOpenListing([]); + const { env } = mkEnv(openRows(7)); + + const summary = await backfillIssueState(REPO, env, { limit: 3 }); + + expect(summary.scanned).toBe(3); + expect(summary.nextCursor).toBe(3); + expect(summary.done).toBe(false); + }); + + it("drains the rest across calls when fed its own nextCursor", async () => { + const { env, updated } = mkEnv(openRows(7)); + + let cursor: number | null = 0; + let calls = 0; + while (cursor !== null) { + stubOpenListing([]); + const summary: Awaited> = + await backfillIssueState(REPO, env, { limit: 3, cursor }); + cursor = summary.nextCursor; + calls++; + } + + expect(calls).toBe(3); + expect(updated).toEqual(["i:1", "i:2", "i:3", "i:4", "i:5", "i:6", "i:7"]); + }); + + it("defaults the per-call budget to DEFAULT_ISSUE_STATE_LIMIT", async () => { + stubOpenListing([]); + const { env } = mkEnv(openRows(1)); + + const summary = await backfillIssueState(REPO, env, {}); + + expect(summary.limit).toBe(DEFAULT_ISSUE_STATE_LIMIT); + }); + + it("batches the Vectorize reads rather than one call per row", async () => { + stubOpenListing([]); + const { env, getBatchSizes } = mkEnv(openRows(120)); + + await backfillIssueState(REPO, env, { limit: 200 }); + + expect(getBatchSizes).toEqual([50, 50, 20]); + }); +}); + +describe("backfill-issue-state: partial index states", () => { + it("counts a missing vector without letting it block the sparse repair", async () => { + // A row with no vector is the issue #210 surface; its sparse half still exists + // and is still wrong, so it gets fixed. + stubOpenListing([]); + const { env, updated, upserted } = mkEnv(openRows(3), { missingVectors: ["i:2"] }); + + const summary = await backfillIssueState(REPO, env, {}); + + expect(summary.vectorsMissing).toBe(1); + expect(summary.vectorsUpdated).toBe(2); + expect(upserted.map((v) => v.id)).toEqual(["i:1", "i:3"]); + expect(updated).toEqual(["i:1", "i:2", "i:3"]); + }); + + it("leaves the sparse rows open when the dense write fails, so the retry re-covers them", async () => { + stubOpenListing([]); + const { env, rows, updated } = mkEnv(openRows(3), { vectorizeThrows: true }); + + await expect(backfillIssueState(REPO, env, {})).rejects.toThrow(/vectorize down/); + + expect(updated).toEqual([]); + expect(rows.every((r) => r.state === "open")).toBe(true); + }); +}); diff --git a/src/backfill-issue-state.ts b/src/backfill-issue-state.ts new file mode 100644 index 0000000..d94847e --- /dev/null +++ b/src/backfill-issue-state.ts @@ -0,0 +1,269 @@ +/** + * Reconcile the indexed `state` of issue / PR rows that were indexed while open + * and never followed the item to `closed`. + * + * Layer = L4 Operations (index repair surface) + * + * The metadata-only path in `./pipeline/embed-issue.ts` is what carries a state + * change onto the retrieval surfaces. It used to advance its own diff baseline + * (the IssueStore record) before the mirror writes it guards, so a mirror write + * that failed was never retried: the next poll compared GitHub against an + * already-updated baseline and saw nothing to do. A body change would have + * reconciled it, but a state-only change never brings one, so the rows stayed + * `open` permanently (issue #209). The ordering is fixed at the source; this + * module repairs the rows the old ordering left behind. + * + * No embedding is involved, so the repair is free of Workers AI cost: + * - the true state comes from one paginated `state=open` listing per repo; + * - the sparse side is a plain `UPDATE search_docs SET state = 'closed'`; + * - the dense side re-upserts the existing vector values fetched via + * `getByIds`, with only the `state` field of the metadata replaced. + * + * Direction is deliberately one-way (`open` -> `closed`). That is the direction + * the defect produced and the direction that harms retrieval: a closed decision + * resurfacing as a live one. Rows whose stored state is already `closed` are not + * examined, which also keeps the scan proportional to the (small) open set + * rather than to the whole index. + * + * Rows with no vector in Vectorize are counted and left to issue #210 (missing + * index entries); their sparse state is still repaired, since that half exists. + */ + +import type { Env } from "./types.js"; + +/** Rows examined per call, unless the caller lowers it. */ +export const DEFAULT_ISSUE_STATE_LIMIT = 200; + +/** Hard ceiling on the per-call row budget a caller may request. */ +export const MAX_ISSUE_STATE_LIMIT = 1000; + +/** Vector IDs per `getByIds` / `upsert` call. Well inside the documented 1000-vector + * batch cap, and small enough that one batch's payload (values + metadata) stays + * modest for a 1024-dimension index. */ +const VECTOR_BATCH_SIZE = 50; + +/** Items per page of the GitHub open-item listing. */ +const OPEN_LIST_PER_PAGE = 100; + +/** Page cap on the open-item listing. The listing MUST be complete: a truncated + * open set would report live items as absent and close them. At 100 per page this + * allows 5000 open items, and the walk throws rather than guess past it. */ +const MAX_OPEN_LIST_PAGES = 50; + +export interface IssueStateBackfillOptions { + /** Report what would change without writing to D1 or Vectorize. */ + dryRun?: boolean; + /** Rows examined in this call (default `DEFAULT_ISSUE_STATE_LIMIT`). */ + limit?: number; + /** Issue number to resume after (exclusive). */ + cursor?: number; +} + +export interface IssueStateBackfillSummary { + repo: string; + dryRun: boolean; + /** Issues + PRs GitHub currently reports as open. */ + openOnGitHub: number; + cursor: number; + limit: number; + /** `state = 'open'` issue / PR rows examined in this call. */ + scanned: number; + /** Examined rows GitHub no longer lists as open. */ + stale: number; + /** Stale rows whose `search_docs.state` was set to `closed`. 0 on a dry run. */ + ftsUpdated: number; + /** Stale rows whose Vectorize metadata was re-upserted. 0 on a dry run. */ + vectorsUpdated: number; + /** Stale rows with no vector to refresh (issue #210 surface). */ + vectorsMissing: number; + /** Pass back as `cursor` to continue; `null` once the scan is exhausted. */ + nextCursor: number | null; + done: boolean; +} + +/** One `state = 'open'` issue / PR row of the sparse index. */ +export interface OpenRow { + vectorId: string; + number: number; +} + +/** + * One ordered page of the repo's `state = 'open'` issue / PR rows, resuming after + * issue number `cursor`. Ordering by number is what makes the cursor resumable. + */ +export async function selectIndexedOpenRows( + db: D1Database, + repo: string, + cursor: number, + limit: number, +): Promise { + const res = await db + .prepare( + `SELECT vector_id, number + FROM search_docs + WHERE repo = ? + AND type IN ('issue', 'pull_request') + AND state = 'open' + AND number > ? + ORDER BY number + LIMIT ?`, + ) + .bind(repo, cursor, limit) + .all<{ vector_id: string; number: number }>(); + + return (res.results ?? []).map((r) => ({ + vectorId: String(r.vector_id ?? ""), + number: Number(r.number ?? 0), + })); +} + +/** + * Flip the given rows to `closed`. + * + * Batched statements rather than one bulk UPDATE: the `search_docs` UPDATE trigger + * re-syncs the FTS5 index per row, and a single statement spanning the whole window + * would fire it for every row at once. `content` / `content_fts` are untouched, so + * the trigger's delete-then-reinsert replays the exact text the index holds. + */ +export async function markRowsClosed( + db: D1Database, + vectorIds: string[], +): Promise { + if (vectorIds.length === 0) return; + const update = db.prepare( + `UPDATE search_docs SET state = 'closed' WHERE vector_id = ?`, + ); + await db.batch(vectorIds.map((id) => update.bind(id))); +} + +/** + * Every issue / PR number GitHub currently reports as open for a repo. + * + * `/issues` returns pull requests as well, so one listing covers both indexed + * types. Throws when the walk would exceed `MAX_OPEN_LIST_PAGES`: a partial open + * set is worse than no answer, because absence from it is what marks a row closed. + */ +export async function fetchOpenItemNumbers( + repo: string, + token: string, +): Promise> { + const open = new Set(); + + for (let page = 1; page <= MAX_OPEN_LIST_PAGES; page++) { + const url = new URL(`https://api.github.com/repos/${repo}/issues`); + url.searchParams.set("state", "open"); + url.searchParams.set("per_page", String(OPEN_LIST_PER_PAGE)); + url.searchParams.set("page", String(page)); + + const resp = await fetch(url.toString(), { + headers: { + Authorization: `Bearer ${token}`, + Accept: "application/vnd.github+json", + "X-GitHub-Api-Version": "2022-11-28", + "User-Agent": "github-rag-mcp/0.1.0", + }, + cache: "no-store", + } as RequestInit); + + if (!resp.ok) { + throw new Error(`GitHub API error ${resp.status}: ${await resp.text()}`); + } + + const items = (await resp.json()) as Array<{ number: number }>; + for (const item of items) open.add(item.number); + + if (items.length < OPEN_LIST_PER_PAGE) return open; + } + + throw new Error( + `open-item listing for ${repo} exceeded ${MAX_OPEN_LIST_PAGES} pages — ` + + `refusing to close rows against a partial open set`, + ); +} + +/** + * Repair one batch of stale `open` rows for a repo. + * + * Resumable and idempotent: progress is an issue-number cursor over the ordered + * `state = 'open'` rows, a row that is genuinely open is skipped without a write, + * and re-running a batch re-issues writes that are already correct. Vectorize is + * updated before D1 on purpose — when the dense write throws, the sparse rows are + * still `open`, so the same window is re-covered on the next call instead of + * leaving the two sides split. + */ +export async function backfillIssueState( + repo: string, + env: Env, + options: IssueStateBackfillOptions = {}, +): Promise { + const dryRun = options.dryRun === true; + const limit = options.limit ?? DEFAULT_ISSUE_STATE_LIMIT; + const cursor = options.cursor ?? 0; + + const openNumbers = await fetchOpenItemNumbers(repo, env.GITHUB_TOKEN); + + const rows = await selectIndexedOpenRows(env.DB_FTS, repo, cursor, limit); + const stale = rows.filter((r) => !openNumbers.has(r.number)); + + let vectorsUpdated = 0; + let vectorsMissing = 0; + let ftsUpdated = 0; + + if (!dryRun && stale.length > 0) { + // Dense side: keep the existing values, replace only `state` in the metadata. + for (let i = 0; i < stale.length; i += VECTOR_BATCH_SIZE) { + const batch = stale.slice(i, i + VECTOR_BATCH_SIZE); + const found = await env.VECTORIZE.getByIds(batch.map((r) => r.vectorId)); + const byId = new Map(found.map((v) => [v.id, v])); + + const refreshed: VectorizeVector[] = []; + for (const row of batch) { + const vector = byId.get(row.vectorId); + if (!vector || !vector.values) { + vectorsMissing++; + continue; + } + refreshed.push({ + id: row.vectorId, + values: vector.values as number[], + metadata: { ...(vector.metadata ?? {}), state: "closed" }, + }); + } + + if (refreshed.length > 0) { + await env.VECTORIZE.upsert(refreshed); + vectorsUpdated += refreshed.length; + } + } + + // Sparse side. + await markRowsClosed(env.DB_FTS, stale.map((r) => r.vectorId)); + ftsUpdated = stale.length; + } + + // A short page means the scan reached the end of the open rows for this repo. + const done = rows.length < limit; + const nextCursor = done ? null : rows[rows.length - 1].number; + + console.log( + `${repo} backfill-issue-state: open_on_github=${openNumbers.size} cursor=${cursor} ` + + `scanned=${rows.length} stale=${stale.length} fts_updated=${ftsUpdated} ` + + `vectors_updated=${vectorsUpdated} vectors_missing=${vectorsMissing}` + + `${dryRun ? " (dry run)" : ""}`, + ); + + return { + repo, + dryRun, + openOnGitHub: openNumbers.size, + cursor, + limit, + scanned: rows.length, + stale: stale.length, + ftsUpdated, + vectorsUpdated, + vectorsMissing, + nextCursor, + done, + }; +} diff --git a/src/backfill-issue-state.workers.test.ts b/src/backfill-issue-state.workers.test.ts new file mode 100644 index 0000000..fa17e33 --- /dev/null +++ b/src/backfill-issue-state.workers.test.ts @@ -0,0 +1,102 @@ +import { describe, it, expect, beforeAll } from "vitest"; +import { env, applyD1Migrations } from "cloudflare:test"; +import { upsertFtsRow, queryFts, type FtsUpsertRow } from "./fts.js"; +import { selectIndexedOpenRows, markRowsClosed } from "./backfill-issue-state.js"; + +// Shared local D1, no per-test isolation — see the note in fts.workers.test.ts. +// Every vector_id and repo here is unique to this file. +beforeAll(async () => { + await applyD1Migrations(env.DB_FTS, env.TEST_MIGRATIONS); +}); + +function mkRow( + overrides: Partial & + Pick, +): FtsUpsertRow { + return { + state: "open", + labels: "", + milestone: "", + assignees: "", + updatedAt: "2026-01-01T00:00:00Z", + ...overrides, + }; +} + +async function readState(vectorId: string): Promise { + const row = await env.DB_FTS + .prepare(`SELECT state FROM search_docs WHERE vector_id = ?`) + .bind(vectorId) + .first<{ state: string }>(); + return String(row?.state ?? ""); +} + +describe("backfill-issue-state D1: candidate selection", () => { + it("returns open issue / PR rows in number order, and nothing else", async () => { + const repo = "t/state-select"; + await upsertFtsRow(env.DB_FTS, mkRow({ vectorId: "i:sel-2", type: "issue", repo, number: 2, content: "second" })); + await upsertFtsRow(env.DB_FTS, mkRow({ vectorId: "i:sel-1", type: "issue", repo, number: 1, content: "first" })); + await upsertFtsRow(env.DB_FTS, mkRow({ vectorId: "p:sel-3", type: "pull_request", repo, number: 3, content: "third" })); + // Excluded: already closed, a different surface type, and another repo. + await upsertFtsRow(env.DB_FTS, mkRow({ vectorId: "i:sel-4", type: "issue", repo, number: 4, state: "closed", content: "fourth" })); + await upsertFtsRow(env.DB_FTS, mkRow({ vectorId: "r:sel-5", type: "release", repo, number: 5, content: "fifth" })); + await upsertFtsRow(env.DB_FTS, mkRow({ vectorId: "i:sel-6", type: "issue", repo: "t/other-repo", number: 6, content: "sixth" })); + + const rows = await selectIndexedOpenRows(env.DB_FTS, repo, 0, 50); + + expect(rows.map((r) => r.number)).toEqual([1, 2, 3]); + expect(rows.map((r) => r.vectorId)).toEqual(["i:sel-1", "i:sel-2", "p:sel-3"]); + }); + + it("resumes after the cursor and honours the row budget", async () => { + const repo = "t/state-cursor"; + for (const n of [1, 2, 3, 4]) { + await upsertFtsRow(env.DB_FTS, mkRow({ vectorId: `i:cur-${n}`, type: "issue", repo, number: n, content: `row ${n}` })); + } + + const first = await selectIndexedOpenRows(env.DB_FTS, repo, 0, 2); + const second = await selectIndexedOpenRows(env.DB_FTS, repo, first[first.length - 1].number, 2); + + expect(first.map((r) => r.number)).toEqual([1, 2]); + expect(second.map((r) => r.number)).toEqual([3, 4]); + }); +}); + +describe("backfill-issue-state D1: closing rows", () => { + it("flips the stored state and drops the row out of the candidate set", async () => { + const repo = "t/state-close"; + await upsertFtsRow(env.DB_FTS, mkRow({ vectorId: "i:close-1", type: "issue", repo, number: 1, content: "stale open row" })); + await upsertFtsRow(env.DB_FTS, mkRow({ vectorId: "i:close-2", type: "issue", repo, number: 2, content: "genuinely open row" })); + + await markRowsClosed(env.DB_FTS, ["i:close-1"]); + + expect(await readState("i:close-1")).toBe("closed"); + expect(await readState("i:close-2")).toBe("open"); + expect((await selectIndexedOpenRows(env.DB_FTS, repo, 0, 50)).map((r) => r.number)).toEqual([2]); + }); + + it("keeps the FTS5 index intact and moves the row across the state filter", async () => { + // The UPDATE trigger deletes and reinserts the row in the FTS5 index. This is the + // path that used to throw (migration 0005 / issue #175), and a corrupt vtab would + // surface here as a failing or empty bm25 query. + const repo = "t/state-filter"; + await upsertFtsRow( + env.DB_FTS, + mkRow({ vectorId: "i:filter-1", type: "issue", repo, number: 1, content: "kubernetes scheduler regression" }), + ); + + expect((await queryFts(env.DB_FTS, "scheduler", 10, { repo, state: "open" })).map((h) => h.vectorId)) + .toEqual(["i:filter-1"]); + + await markRowsClosed(env.DB_FTS, ["i:filter-1"]); + + expect(await queryFts(env.DB_FTS, "scheduler", 10, { repo, state: "open" })).toEqual([]); + const closed = await queryFts(env.DB_FTS, "scheduler", 10, { repo, state: "closed" }); + expect(closed.map((h) => h.vectorId)).toEqual(["i:filter-1"]); + expect(closed[0].content).toBe("kubernetes scheduler regression"); + }); + + it("is a no-op on an empty id list", async () => { + await expect(markRowsClosed(env.DB_FTS, [])).resolves.toBeUndefined(); + }); +}); diff --git a/src/index.ts b/src/index.ts index 619cc29..458f292 100644 --- a/src/index.ts +++ b/src/index.ts @@ -18,6 +18,7 @@ * POST /admin/backfill-fts-segments[?repo=owner/repo][&cursor=N][&limit=N] -- Re-segment one batch of natural-language FTS rows (requires GITHUB_TOKEN header) * POST /admin/backfill-wiki?repo=owner/repo[&limit=N][&cursor=SLUG] -- Walk one batch of a repo's wiki without waiting for the :45 cron (requires GITHUB_TOKEN header) * POST /admin/purge-legacy-vectors?repo=owner/repo[&dry_run=true][&surface=doc][&limit=N][&cursor=N] -- Delete pre-migration doc vectors the reap cannot name (requires GITHUB_TOKEN header) + * POST /admin/backfill-issue-state?repo=owner/repo[&dry_run=true][&limit=N][&cursor=N] -- Close indexed issue/PR rows GitHub no longer lists as open (requires GITHUB_TOKEN header) * * Durable Objects: * RagMcpAgentV2 -- MCP server (tools: search, get_issue_context, list_recent_activity) @@ -46,6 +47,11 @@ import { MAX_PURGE_LIMIT, purgeLegacyDocVectors, } from "./purge-legacy.js"; +import { + DEFAULT_ISSUE_STATE_LIMIT, + MAX_ISSUE_STATE_LIMIT, + backfillIssueState, +} from "./backfill-issue-state.js"; // Durable Object: issue/PR state store (SQLite-backed) export { IssueStore } from "./store.js"; @@ -446,6 +452,58 @@ const innerHandler: ExportedHandler = { } } + // -- Admin: close indexed issue / PR rows GitHub no longer lists as open -- + // POST /admin/backfill-issue-state?repo=owner/repo[&dry_run=true][&limit=N][&cursor=N] + // Repairs rows indexed while the item was open, whose state mirror never landed + // (issue #209). No embedding: the state comes from one `state=open` listing, the + // sparse side is an UPDATE, and the dense side re-upserts the existing values with + // only `state` replaced. `/admin/reset-hashes` cannot be used for this — it would + // trigger a full re-embedding of the repo. + // Call repeatedly, passing the returned `nextCursor` back, until `done` is true. + // Requires GITHUB_TOKEN header for authentication. + if (request.method === "POST" && url.pathname === "/admin/backfill-issue-state") { + const authHeader = request.headers.get("GITHUB_TOKEN"); + if (!authHeader || authHeader !== env.GITHUB_TOKEN) { + return new Response("Unauthorized", { status: 401 }); + } + + const repo = url.searchParams.get("repo"); + if (!repo) { + return new Response("missing repo query parameter", { status: 400 }); + } + + const dryRun = url.searchParams.get("dry_run") === "true"; + + const rawLimit = url.searchParams.get("limit"); + const limit = rawLimit === null ? DEFAULT_ISSUE_STATE_LIMIT : Number(rawLimit); + if (!Number.isInteger(limit) || limit < 1 || limit > MAX_ISSUE_STATE_LIMIT) { + return new Response(`limit must be an integer in 1..${MAX_ISSUE_STATE_LIMIT}`, { + status: 400, + }); + } + + const rawCursor = url.searchParams.get("cursor"); + const cursor = rawCursor === null ? 0 : Number(rawCursor); + if (!Number.isInteger(cursor) || cursor < 0) { + return new Response("cursor must be a non-negative integer", { status: 400 }); + } + + try { + const summary = await backfillIssueState(repo, env, { dryRun, limit, cursor }); + return new Response(JSON.stringify(summary), { + status: 200, + headers: { "Content-Type": "application/json" }, + }); + } catch (err) { + return new Response( + JSON.stringify({ + error: err instanceof Error ? err.message : String(err), + }), + { status: 500, headers: { "Content-Type": "application/json" } }, + ); + } + } + // -- MCP endpoint (OAuth-protected, ctx.props set by OAuthProvider) -- if (url.pathname.startsWith("/mcp")) { const props = readGitHubProps(ctx); diff --git a/src/pipeline/embed-issue.test.ts b/src/pipeline/embed-issue.test.ts new file mode 100644 index 0000000..0a1436e --- /dev/null +++ b/src/pipeline/embed-issue.test.ts @@ -0,0 +1,209 @@ +import { describe, it, expect, vi } from "vitest"; +import type { Env, IssueRecord } from "../types.js"; +import { processAndUpsertIssue, type GitHubIssueData } from "./embed-issue.js"; +import { computeBodyHash } from "./hash.js"; +import { vectorId } from "./vector-id.js"; + +const REPO = "acme/widgets"; + +function mkIssue(overrides: Partial = {}): GitHubIssueData { + return { + number: 42, + title: "a title", + body: "a body", + state: "open", + labels: [{ name: "bug" }], + milestone: null, + assignees: [], + created_at: "2026-01-01T00:00:00Z", + updated_at: "2026-01-02T00:00:00Z", + html_url: `https://github.com/${REPO}/issues/42`, + ...overrides, + }; +} + +/** + * IssueStore stub holding one record. `upserts` records every write so a test can + * assert whether the diff baseline advanced — the whole point of issue #209. + */ +function mkStore(existing: IssueRecord | null) { + const upserts: IssueRecord[] = []; + const stub = { + fetch: vi.fn(async (req: Request) => { + const url = new URL(req.url); + if (url.pathname === "/issue") { + if (!existing) return new Response("not found", { status: 404 }); + return new Response(JSON.stringify(existing), { status: 200 }); + } + if (url.pathname === "/upsert") { + upserts.push((await req.json()) as IssueRecord); + return new Response("{}", { status: 200 }); + } + throw new Error(`unexpected store path: ${url.pathname}`); + }), + } as unknown as DurableObjectStub; + return { stub, upserts }; +} + +interface EnvStubOptions { + /** Vector returned by getByIds; null models a row with no vector (issue #210). */ + vector?: { values: number[]; metadata?: Record } | null; + vectorizeThrows?: boolean; + ftsThrows?: boolean; +} + +function mkEnv(options: EnvStubOptions = {}) { + const vector = options.vector === undefined ? { values: [0.1, 0.2] } : options.vector; + const upserted: Array<{ id: string; metadata: Record }> = []; + const ftsWrites: unknown[][] = []; + + const env = { + GITHUB_TOKEN: "t", + VECTORIZE: { + getByIds: vi.fn(async (ids: string[]) => { + if (options.vectorizeThrows) throw new Error("vectorize down"); + return vector ? [{ id: ids[0], ...vector }] : []; + }), + upsert: vi.fn(async (vectors: Array<{ id: string; metadata: Record }>) => { + upserted.push(...vectors); + }), + }, + DB_FTS: { + prepare: () => ({ + bind: (...args: unknown[]) => ({ + run: async () => { + if (options.ftsThrows) throw new Error("d1 down"); + ftsWrites.push(args); + return {}; + }, + }), + }), + }, + } as unknown as Env; + + return { env, upserted, ftsWrites }; +} + +/** Stored record whose bodyHash matches `issue`, so the hash-skip path is taken. */ +async function mkExisting( + issue: GitHubIssueData, + overrides: Partial = {}, +): Promise { + return { + repo: REPO, + number: issue.number, + type: "issue", + state: issue.state, + title: issue.title, + labels: issue.labels.map((l) => l.name), + milestone: issue.milestone?.title ?? "", + assignees: issue.assignees.map((a) => a.login), + bodyHash: await computeBodyHash(issue.title, issue.body ?? ""), + createdAt: issue.created_at, + updatedAt: issue.updated_at, + ...overrides, + }; +} + +describe("embed-issue: metadata-only path mirrors the state change", () => { + it("writes both the dense and the sparse side, then advances the baseline", async () => { + const issue = mkIssue({ state: "closed" }); + const { stub, upserts } = mkStore(await mkExisting(issue, { state: "open" })); + const { env, upserted, ftsWrites } = mkEnv(); + + const result = await processAndUpsertIssue(env, stub, REPO, issue); + + expect(result).toEqual({ + embedded: false, + skippedUnchanged: false, + metadataUpdated: true, + failed: false, + }); + expect(upserted).toHaveLength(1); + expect(upserted[0].metadata.state).toBe("closed"); + expect(upserted[0].id).toBe(await vectorId(REPO, issue.number)); + expect(ftsWrites).toHaveLength(1); + expect(ftsWrites[0]).toContain("closed"); + expect(upserts).toHaveLength(1); + expect(upserts[0].state).toBe("closed"); + }); + + it("skips both mirrors and still advances the baseline when nothing changed", async () => { + const issue = mkIssue(); + const { stub, upserts } = mkStore(await mkExisting(issue)); + const { env, upserted, ftsWrites } = mkEnv(); + + const result = await processAndUpsertIssue(env, stub, REPO, issue); + + expect(result.skippedUnchanged).toBe(true); + expect(result.metadataUpdated).toBe(false); + expect(upserted).toEqual([]); + expect(ftsWrites).toEqual([]); + expect(upserts).toHaveLength(1); + }); +}); + +describe("embed-issue: a failed mirror must stay retryable (issue #209)", () => { + // The baseline `metadataChanged` is measured against is the IssueStore record. + // Advancing it past a failed mirror write made the miss permanent: a state-only + // change never brings the body change the old code was waiting for. + it("holds the IssueStore baseline when the sparse write fails", async () => { + const issue = mkIssue({ state: "closed" }); + const { stub, upserts } = mkStore(await mkExisting(issue, { state: "open" })); + const { env } = mkEnv({ ftsThrows: true }); + + const result = await processAndUpsertIssue(env, stub, REPO, issue); + + expect(result.failed).toBe(true); + expect(result.metadataUpdated).toBe(false); + expect(upserts).toEqual([]); + }); + + it("holds the IssueStore baseline when the dense write fails", async () => { + const issue = mkIssue({ state: "closed" }); + const { stub, upserts } = mkStore(await mkExisting(issue, { state: "open" })); + const { env } = mkEnv({ vectorizeThrows: true }); + + const result = await processAndUpsertIssue(env, stub, REPO, issue); + + expect(result.failed).toBe(true); + expect(upserts).toEqual([]); + }); + + it("retries on the next delivery because the baseline is still stale", async () => { + const issue = mkIssue({ state: "closed" }); + const existing = await mkExisting(issue, { state: "open" }); + const failing = mkStore(existing); + await processAndUpsertIssue(mkEnv({ ftsThrows: true }).env, failing.stub, REPO, issue); + expect(failing.upserts).toEqual([]); + + // Same stale record, mirror healthy this time. + const recovered = mkStore(existing); + const { env, upserted, ftsWrites } = mkEnv(); + const result = await processAndUpsertIssue(env, recovered.stub, REPO, issue); + + expect(result.metadataUpdated).toBe(true); + expect(upserted).toHaveLength(1); + expect(ftsWrites).toHaveLength(1); + expect(recovered.upserts[0].state).toBe("closed"); + }); +}); + +describe("embed-issue: a missing vector must not swallow the sparse update", () => { + // The sparse mirror used to be nested inside the "vector exists" branch, so rows + // missing a vector (issue #210) lost their state update on both sides at once. + it("still writes the sparse state and advances the baseline", async () => { + const issue = mkIssue({ state: "closed" }); + const { stub, upserts } = mkStore(await mkExisting(issue, { state: "open" })); + const { env, upserted, ftsWrites } = mkEnv({ vector: null }); + + const result = await processAndUpsertIssue(env, stub, REPO, issue); + + expect(upserted).toEqual([]); + expect(ftsWrites).toHaveLength(1); + expect(ftsWrites[0]).toContain("closed"); + expect(result.failed).toBe(false); + expect(upserts).toHaveLength(1); + expect(upserts[0].state).toBe("closed"); + }); +}); diff --git a/src/pipeline/embed-issue.ts b/src/pipeline/embed-issue.ts index 570da42..23153d3 100644 --- a/src/pipeline/embed-issue.ts +++ b/src/pipeline/embed-issue.ts @@ -3,8 +3,16 @@ * * Owns the GitHub issue / PR data shape (`GitHubIssueData`) and the * `processAndUpsertIssue` flow: hash-based change detection, metadata-only - * Vectorize refresh when only labels / state changed, and full embed + upsert - * when the body changed. FTS5 mirror writes are best-effort. + * Vectorize + D1 refresh when only labels / state changed, and full embed + + * upsert when the body changed. + * + * Mirror-failure handling differs per path, because what makes a retry happen + * differs. On the embed path a failed FTS5 write is best-effort: the stored + * bodyHash is what drives the next attempt, and the next body change reconciles + * the sparse side. On the metadata-only path there is no next body change to + * wait for, so a failed mirror write holds the IssueStore baseline instead — + * that record is the diff basis, and advancing it would make the miss permanent + * (issue #209). */ import type { Env, IssueRecord } from "../types.js"; @@ -69,7 +77,8 @@ export async function processAndUpsertIssue( } if (!needsEmbedding) { - // Hash matched — skip embedding but update IssueStore (metadata may have changed) + // Hash matched — skip embedding, but state / labels / milestone / assignees may + // still have changed and the retrieval surfaces have to follow. const labelNames = issue.labels.map((l) => l.name); const assigneeLogins = issue.assignees.map((a) => a.login); const milestoneTitle = issue.milestone?.title ?? ""; @@ -88,16 +97,8 @@ export async function processAndUpsertIssue( updatedAt: issue.updated_at, }; - await storeStub.fetch( - new Request("http://store/upsert", { - method: "POST", - headers: { "Content-Type": "application/json" }, - body: JSON.stringify(record), - }), - ); - - // Check if metadata changed — if so, update Vectorize metadata too - // (Vectorize state/labels/assignees must stay in sync with GitHub) + // Check if metadata changed — if so, the dense and sparse sides need updating + // (Vectorize / search_docs state, labels, assignees must stay in sync with GitHub). const sortedLabels = [...labelNames].sort(); const metadataChanged = existing !== null && ( existing.state !== issue.state || @@ -107,12 +108,21 @@ export async function processAndUpsertIssue( [...existing.assignees].sort().join(",") !== [...assigneeLogins].sort().join(",") ); + // The IssueStore record is the *baseline* `metadataChanged` is measured against, so + // it must not advance until the mirrors it guards have landed. Advancing it first + // made a failed mirror write permanent: the next poll compares GitHub against an + // already-updated baseline, sees no diff, and never retries. A body change would + // reconcile it, but a state-only change never brings one (issue #209). + let mirrorFailed = false; + if (metadataChanged) { + const vid = await vectorId(repo, issue.number); + + // Dense side. A row with no vector is the #210 (missing index entry) surface, + // not a mirror failure — nothing here can rebuild it without embedding, so it + // must not hold the baseline hostage. try { - // Retrieve existing vector values to re-upsert with updated metadata - const vid = await vectorId(repo, issue.number); const vectors = await env.VECTORIZE.getByIds([vid]); - if (vectors.length > 0 && vectors[0].values) { const metadata: Record = { repo, @@ -138,43 +148,69 @@ export async function processAndUpsertIssue( metadata, }, ]); - - // Mirror the metadata change onto D1 FTS5 so sparse retrieval stays filterable. - // Content stays the same (no body change), but labels/state/milestone etc. - // on the sparse side must match the dense side for pre-filter consistency. - try { - await upsertFtsRow(env.DB_FTS, { - vectorId: vid, - repo, - type, - state: issue.state, - labels: sortedLabels.join(","), - milestone: milestoneTitle, - assignees: assigneeLogins.join(","), - updatedAt: issue.updated_at, - number: issue.number, - content: prepareEmbeddingInput(title, issue.body), - }); - } catch (ftsErr) { - console.error( - `Failed to update FTS5 metadata for ${repo}#${issue.number}:`, - ftsErr instanceof Error ? ftsErr.message : String(ftsErr), - ); - // Non-fatal: sparse side will catch up on next body change. - } - - return { embedded: false, skippedUnchanged: false, metadataUpdated: true, failed: false }; } } catch (err) { console.error( `Failed to update Vectorize metadata for ${repo}#${issue.number}:`, err instanceof Error ? err.message : String(err), ); - // IssueStore was already updated — Vectorize metadata will catch up on next body change + mirrorFailed = true; } + + // Sparse side. Deliberately outside the dense branch above: the two stores fail + // independently, and nesting this call inside "the vector exists" skipped the + // sparse state update for exactly the rows that were already missing a vector. + // Content stays the same (no body change); only the filterable columns move. + try { + await upsertFtsRow(env.DB_FTS, { + vectorId: vid, + repo, + type, + state: issue.state, + labels: sortedLabels.join(","), + milestone: milestoneTitle, + assignees: assigneeLogins.join(","), + updatedAt: issue.updated_at, + number: issue.number, + content: prepareEmbeddingInput(title, issue.body), + }); + } catch (ftsErr) { + console.error( + `Failed to update FTS5 metadata for ${repo}#${issue.number}:`, + ftsErr instanceof Error ? ftsErr.message : String(ftsErr), + ); + mirrorFailed = true; + } + } + + if (mirrorFailed) { + // Baseline held on purpose: the stale IssueStore record is what makes the next + // poll / webhook delivery for this item detect the diff again and retry. + console.warn( + `Holding IssueStore baseline for ${repo}#${issue.number} so the metadata mirror is retried`, + ); + return { + embedded: false, + skippedUnchanged: false, + metadataUpdated: false, + failed: true, + }; } - return { embedded: false, skippedUnchanged: true, metadataUpdated: false, failed: false }; + await storeStub.fetch( + new Request("http://store/upsert", { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify(record), + }), + ); + + return { + embedded: false, + skippedUnchanged: !metadataChanged, + metadataUpdated: metadataChanged, + failed: false, + }; } // Content changed — generate embedding