From 560e4c18d145ea50bf264837d9b8a83cab5b7ffd Mon Sep 17 00:00:00 2001 From: Claude Lin & Lay Date: Mon, 3 Aug 2026 12:03:41 +0900 Subject: [PATCH] fix(pipeline): hold the diff baseline until the state mirror lands [pipeline, index, docs, tests] MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The metadata-only path in embed-issue advanced its own diff basis — the IssueStore record — before the Vectorize / D1 mirror writes it guards. A mirror write that failed was therefore never retried: the next poll compared GitHub against an already-updated baseline and saw nothing to do. The catch comment promised the next body change would reconcile it, which is true on the embed path and false here, because a state-only change never brings one. Rows indexed while open kept answering searches as live items after the item closed. Two changes at the source, plus a repair for the rows already stranded: - src/pipeline/embed-issue.ts: mirror writes run first, and the IssueStore upsert is skipped when either side failed, so the stale record is what makes the next delivery detect the diff again. The FTS mirror also moves out of the "vector exists" branch — nesting it there dropped the sparse state update for exactly the rows already missing a vector (#210). - src/backfill-issue-state.ts + src/index.ts: POST /admin/backfill-issue-state, a resumable per-repo repair. One paginated state=open listing supplies the truth; indexed open rows absent from it are closed on both sides. No embedding — the dense side re-upserts the existing values with only `state` replaced — so /admin/reset-hashes (full re-embed) is not needed. The backfill aborts rather than act on a truncated open listing, since absence from that listing is what marks a row closed, and it fails before touching D1 if the dense write throws, so a failed call leaves the window unrepaired rather than half-repaired. 恒久修正と backfill を同じ PR に置いたのは、順序を直しても既存の取り残しは 自力で治らないため。差分検出の基準が既に GitHub と一致していて、通常の poll では二度と到達しない。修復方向を open -> closed の一方向に限ったのは、欠陥が 生んだ方向であり、閉じた判断が生きた検討事項として再供給される害の方向でもある から。 Closes #209 Co-authored-by: Claude Opus 5 --- docs/0-requirements.ja.md | 6 +- docs/0-requirements.md | 6 +- docs/installation.ja.md | 41 ++++ docs/installation.md | 41 ++++ src/backfill-issue-state.test.ts | 279 +++++++++++++++++++++++ src/backfill-issue-state.ts | 269 ++++++++++++++++++++++ src/backfill-issue-state.workers.test.ts | 102 +++++++++ src/index.ts | 58 +++++ src/pipeline/embed-issue.test.ts | 209 +++++++++++++++++ src/pipeline/embed-issue.ts | 124 ++++++---- 10 files changed, 1089 insertions(+), 46 deletions(-) create mode 100644 src/backfill-issue-state.test.ts create mode 100644 src/backfill-issue-state.ts create mode 100644 src/backfill-issue-state.workers.test.ts create mode 100644 src/pipeline/embed-issue.test.ts 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