From 17542df5f64502e57baf8374e53e74add1ad3498 Mon Sep 17 00:00:00 2001 From: Claude Lin & Lay Date: Mon, 3 Aug 2026 12:20:53 +0900 Subject: [PATCH] fix(poller): hold the issue watermark at the first uningested item [poller, pipeline, index, docs, tests] (#210) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit pollRepo fetches up to 200 items per run (MAX_PAGES_PER_RUN x 100) and embeds at most 50 (MAX_EMBEDDINGS_PER_RUN). The rest were stored with an empty bodyHash as a retry marker — and then the watermark advanced past them anyway, to the poll start time, or to the last fetched item's updated_at when pagination capped. The batch is sorted `updated` ascending, so both bounds sit above every deferred item: the marker was set and the marked item was never fetched again. Measured 2026-08-03, about 55% of the issue and PR history of the indexed repositories was absent from search_docs, scattered rather than in contiguous ranges because the loss follows updated_at order, not number order. Same defect the commit-diff surface had (#178 / #179), on a surface the pin was never applied to. - src/poller.ts: processIssues reports the first item it left off the retrieval surfaces (budget-deferred or embed-failed) as a retry boundary, and nextIssueWatermark pins the watermark one second before it — the same margin and never-regress shape as nextForwardDiffWatermark, since /issues documents `since` as "updated after" and would otherwise drop the boundary item. The response ETag is withheld on the same condition: storing it would make the next conditional request answer 304 and return before it saw the leftover. - src/backfill-issue-index.ts + src/index.ts: POST /admin/backfill-issue-index, a resumable per-repo sweep over the issue-number space that ingests the numbers with no search_docs issue/PR row. Numeric cursor, because a timestamp cursor over the same set reintroduces the ordering the defect exploited. dry_run measures the gap without spending the embed budget. - src/pipeline/embed-issue.ts: `force` option, set only by the backfill. Every candidate there is known to be missing a retrieval surface, and a matching hash — which an embed whose FTS5 mirror failed leaves behind — would otherwise skip it permanently. 恒久修正と backfill を同じ PR に置いたのは、watermark を直しても既存の欠落は 自力で埋まらないため。取り残された項目が再 fetch されるのは updated_at が動いた ときだけで、閉じた履歴はもう動かない。実行順序は逆にできない: backfill を先に 走らせると、その後に入る新規項目が同じ経路でまた落ちる。release surface の別種の 穴は #211 に分離してある。 Closes #210 Co-authored-by: Claude Opus 5 --- docs/0-requirements.ja.md | 6 + docs/0-requirements.md | 6 + docs/installation.ja.md | 42 +++++ docs/installation.md | 42 +++++ src/backfill-issue-index.test.ts | 311 +++++++++++++++++++++++++++++++ src/backfill-issue-index.ts | 259 +++++++++++++++++++++++++ src/index.ts | 59 ++++++ src/pipeline/embed-issue.ts | 17 +- src/poller.test.ts | 245 +++++++++++++++++++++++- src/poller.ts | 148 ++++++++++++--- 10 files changed, 1103 insertions(+), 32 deletions(-) create mode 100644 src/backfill-issue-index.test.ts create mode 100644 src/backfill-issue-index.ts diff --git a/docs/0-requirements.ja.md b/docs/0-requirements.ja.md index ef6687d..07eb68f 100644 --- a/docs/0-requirements.ja.md +++ b/docs/0-requirements.ja.md @@ -136,6 +136,10 @@ Responsibilities: 各 invocation は独立した subrequest 予算を持つ。dispatch は `controller.cron` で `handleScheduled` 内で行う。未知の cron 表現は no-op log で silent regression を防止する。 +issue / PR poller は `(lastPolledAt, now]` を `updated_at` 昇順で取得する。1 run あたり最大 `MAX_PAGES_PER_RUN` × 100 = 200 件を fetch し、そのうち embed するのは最大 `MAX_EMBEDDINGS_PER_RUN` = 50 件。この surface は commit diff と同じ watermark 不変条件に従う: **その run が retrieval surface に載せられなかった最も古い項目を watermark が追い越さない** — embedding 予算で見送った項目と、embed に失敗した項目の両方が対象。2 つの境界は別物として扱う。fetch がどこまで届いたか（poll 開始時刻、pagination が打ち切られた場合は最後に fetch した項目の `updated_at`）は上限にすぎず、取り込み境界がその下に watermark を留める。留める位置は境界の 1 秒手前で、GitHub の `since` filter が境界の項目自身を再び含むようにするため。 + +この pin が無い間、この surface は「fetch 件数 − embed 件数」の速度で取りこぼしていた。旧実装の watermark はどちらの分岐でも見送った項目より**新しい**位置に着地する — batch は昇順なので、予算はいつもその新しい端で尽きる — 一方で見送った項目に付く空の `bodyHash` は retry の印であって、以後どの `since` window もその項目を fetch しない。2026-08-03 の実測で、索引対象 repository の issue / pull request 履歴の約 55% が索引に載っていなかった。欠落が連続した番号帯ではなく散発に見えるのは、脱落が番号順ではなく `updated_at` 順に起きるため（issue #210）。**ETag も同じ条件で書き戻さない。** docs poller が tree ETag を保持するのと同じ理由で、ETag を保存すると次 run の条件付き request が 304 を返し、残りを見る前に return してしまう。 + commit diff poller は 2-phase 構成: - **forward phase** — `(lastPolledAt, pollStartTime]` の window を列挙し、その中の**古い側から**取り込む（webhook 取りこぼし時の redundancy）。watermark namespace は `diffs:${repo}`。 @@ -202,6 +206,8 @@ Responsibilities: - 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 に分割する +**索引欠落の修復.** watermark の修正は漏れを止めるだけで、既に空いた穴は埋まらない — 取り残された項目が再 fetch されるのは `updated_at` が動いたときだけで、閉じた履歴はもう動かない。`POST /admin/backfill-issue-index?repo=owner/repo`（installation guide 参照）が欠落そのものを走査する。repository の issue 番号空間は密かつ有界なので、`search_docs` に issue / PR 行が無い番号がそのまま欠落集合であり、数値 cursor が「どこまで走査したか」を厳密に表せる。同じ集合を時刻 cursor で辿ると、欠陥が突いた順序をそのまま持ち込むことになる。GitHub 側に既に無い番号（削除・transfer 済み）は 404 を返すので、retry せず計上のみ。取り込みは body-hash 判定を強制的に飛ばす: 候補はいずれも retrieval surface が欠けていると分かっている項目であり、hash が一致していると（embed 成功後に FTS5 mirror が失敗した行がこの状態になる）そのまま恒久的に skip されてしまうため。state 修復と違いこちらは embed を伴うので、1 call の予算は Workers AI の予算であり、呼び出し側が batch を跨いで sweep を進める。`dry_run=true` は予算を使わずに欠落量だけを測る。 + **取り残した 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) diff --git a/docs/0-requirements.md b/docs/0-requirements.md index c4db647..129f458 100644 --- a/docs/0-requirements.md +++ b/docs/0-requirements.md @@ -137,6 +137,10 @@ The poller runs hourly in the current deployment, split across four cron trigger Dispatch is performed inside `handleScheduled` by inspecting `controller.cron`. Unknown cron expressions fall through to a no-op log to prevent silent regressions when triggers are added later. +The issue / PR poller fetches `(lastPolledAt, now]` sorted by `updated_at` ascending, at most `MAX_PAGES_PER_RUN` × 100 = 200 items, and embeds at most `MAX_EMBEDDINGS_PER_RUN` = 50 of them. It obeys the same watermark invariant as the commit-diff surface: **the watermark never advances past the earliest item the run left off the retrieval surfaces** — deferred by the embedding budget, or failed to embed. The two bounds are separate: how far the fetch reached (the poll start time, or the last fetched item when pagination capped) is an upper bound, and the ingest boundary pins the watermark below it, one second earlier so GitHub's `since` filter still re-includes the boundary item. + +Without that pin the surface leaked at a rate of fetched-minus-embedded per run. Both of the old watermark branches landed *above* every deferred item — the batch is ascending, so the budget always runs out on its newest end — and the deferred item's empty `bodyHash` marked it for a retry that no later `since` window would ever fetch. Measured 2026-08-03, that left about 55% of the issue and pull request history of the indexed repositories absent from the index, scattered rather than in contiguous ranges because the loss follows `updated_at` order and not number order (issue #210). **The ETag is withheld on the same condition**, for the reason the docs poller withholds its tree ETag: a stored ETag makes the next run's conditional request answer 304 and return before it looks at the leftover. + The commit-diff poller runs in two phases: - **forward phase** — enumerates the window `(lastPolledAt, pollStartTime]` and ingests its **oldest** commits first, acting as redundancy when webhook delivery has stalled. Watermark namespace: `diffs:${repo}`. @@ -203,6 +207,8 @@ Responsibilities: - 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 +**Missing-entry repair.** The watermark fix stops the leak but does not fill the hole: a stranded item is only re-fetched when its `updated_at` moves, and closed history never moves again. `POST /admin/backfill-issue-index?repo=owner/repo` (see the installation guide) walks the gap directly — the repository's issue-number space is dense and bounded, so the numbers with no `search_docs` issue / PR row are exactly the missing set, and a numeric cursor states how far the sweep has reached. A timestamp cursor over the same set would reintroduce the ordering the defect exploited. Numbers GitHub no longer has (deleted or transferred) answer 404 and are counted rather than retried. The ingest is forced past the body-hash check: every candidate is known to be missing a retrieval surface, and a matching hash — which an embed whose FTS5 mirror failed leaves behind — would otherwise skip it permanently. Unlike the state repair this one embeds, so the per-call budget is a Workers AI budget and the caller drives the sweep one batch at a time; `dry_run=true` measures the gap without spending it. + **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) diff --git a/docs/installation.ja.md b/docs/installation.ja.md index f37f3a3..222ef7c 100644 --- a/docs/installation.ja.md +++ b/docs/installation.ja.md @@ -385,6 +385,48 @@ POST /admin/backfill-issue-state?repo=owner/repo - dense 側の書き込みが失敗した場合、D1 に触れる前に呼び出し全体が失敗する。中途半端な修復を残さないための設計なので、同じ `cursor` で再実行する - 確認は close 済みと分かっている項目を `state: "closed"` で検索するか、`SELECT COUNT(*) FROM search_docs WHERE repo = ? AND type IN ('issue','pull_request') AND state = 'open'` が実際の open 数と一致することを見る +## 15. 索引に一度も載らなかった issue / pull request を取り込む + +poller は 1 run で最大 200 件を fetch する一方、embed するのは最大 50 件で、旧実装は batch 全体を追い越して watermark を進めていた。embedding 予算で見送った項目には retry の印が付くが、以後どの `since` window もそれを fetch しない。結果として、索引対象 repository の issue / pull request 履歴の約 55% が一度も索引に載っていなかった（issue #210）。watermark は発生源側で修正済みだが、それは漏れを止めるだけ — 取り残された項目が再 fetch されるのは `updated_at` が動いたときだけで、閉じた履歴はもう動かない。この endpoint が欠落そのものを走査する。 + +実行は watermark 修正の deploy の**後**。順序を逆にすると、backfill 済みの索引に、同じ穴から落ちた新しい項目が混ざる。 + +Admin endpoint: + +```text +POST /admin/backfill-issue-index?repo=owner/repo +``` + +パラメータ: + +- `repo` — `owner/repo` +- `dry_run` — `true` で走査範囲の欠落量だけを測り、GitHub からは何も取得しない +- `limit` — 1 回の呼び出しで試す候補番号の数、`1..100`（既定 `25`）。dry run では無視される +- `cursor` — 再開位置の issue 番号。前回のレスポンスの `nextCursor` をそのまま渡す + +認証: + +- `GITHUB_TOKEN` ヘッダに worker secret と同じ値を送る + +レスポンス: + +```json +{ "repo": "owner/repo", "dryRun": false, "cursor": 0, "limit": 25, "maxNumber": 1690, + "scannedTo": 613, "candidates": 25, "attempted": 25, "indexed": 24, "absent": 1, + "failed": 0, "nextCursor": 613, "done": false } +``` + +運用上の注意: + +- まず `dry_run=true` で規模を測る。embedding 予算を使わず、走査した範囲（1 回あたり最大 5000 番）の `candidates` を返す +- `done` が `true` になるまで `nextCursor` を渡して繰り返し呼ぶ。既定の `limit` なら 900 件欠けている repository で 36 回 +- 上の 2 つの修復と違い、この endpoint は **embed する**。`limit` は subrequest 予算であると同時に Workers AI の予算でもあり、100 を超える指定は拒否される +- 何度実行しても安全。`search_docs` に行がある番号は fetch すらしないので、完了済みの sweep を再実行すると `candidates: 0` が返る。途中で失敗した呼び出しは同じ `cursor` から再開する +- `absent` は GitHub が 404 を返す番号の数（削除された issue、transfer で repository の外に出た番号）。これらは以後の sweep でも候補に残り続けるので、完了した repository でも `candidates` が小さな非ゼロを返すことがある +- `failed` は embed が着地しなかった候補の数。現在の呼び出し内では再試行せず、同じ範囲を次に sweep したときに拾い直す +- 取り込みは body-hash 判定を強制的に飛ばすので、vector はあるが FTS5 行が無い項目も修復される。次の poll を待つのでは代替できない理由がここ +- 確認は distinct な番号を数える: `SELECT COUNT(DISTINCT number) FROM search_docs WHERE repo = ? AND type IN ('issue','pull_request')` が実際の issue + PR 件数に近づく + ## Troubleshooting ### `GITHUB_TOKEN not configured` diff --git a/docs/installation.md b/docs/installation.md index 9d21a38..b13cf2e 100644 --- a/docs/installation.md +++ b/docs/installation.md @@ -385,6 +385,48 @@ Operational notes: - 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 +## 15. Index the issues and pull requests that never reached the index + +The poller embeds at most 50 items per run but fetches up to 200, and it used to move its watermark past the whole batch regardless. Everything the embedding budget deferred was therefore marked for a retry that no later `since` window would fetch, and roughly 55% of the issue and pull request history of the indexed repositories never reached the index (issue #210). The watermark is fixed at the source, but that only stops the leak: a stranded item is re-fetched only when its `updated_at` moves, and closed history never moves again. This endpoint walks the gap directly. + +Run it **after** deploying the watermark fix. In the other order, items indexed by the backfill are joined by new ones falling into the same hole. + +Admin endpoint: + +```text +POST /admin/backfill-issue-index?repo=owner/repo +``` + +Parameters: + +- `repo` — `owner/repo` +- `dry_run` — `true` measures the gap over the scan range and fetches nothing from GitHub +- `limit` — candidate numbers attempted per call, `1..100` (default `25`). Ignored on a dry run +- `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, "cursor": 0, "limit": 25, "maxNumber": 1690, + "scannedTo": 613, "candidates": 25, "attempted": 25, "indexed": 24, "absent": 1, + "failed": 0, "nextCursor": 613, "done": false } +``` + +Operational notes: + +- start with `dry_run=true` to size the job. It spends no embedding budget and reports `candidates` over the range it scanned (up to 5000 numbers per call) +- call it repeatedly, feeding `nextCursor` back in, until `done` is `true`. At the default `limit` a repository missing 900 items takes 36 calls +- this endpoint **embeds**, unlike the two repairs above. `limit` is a Workers AI budget as much as a subrequest budget; raising it above 100 is refused +- safe to repeat: a number that already carries a `search_docs` row is never fetched, so a re-run of a finished sweep reports `candidates: 0`. A call that fails mid-sweep is resumed from the same `cursor` +- `absent` counts numbers GitHub answers 404 for — deleted issues, and numbers whose item was transferred out. They stay candidates on every future sweep, which is why a finished repository still reports a small non-zero `candidates` +- `failed` counts candidates whose embed did not land. They are retried by the next sweep over the same range, not by the current call +- the ingest is forced past the body-hash check, so an item whose vector exists but whose FTS5 row is missing is repaired too. This is why the endpoint is not equivalent to waiting for the next poll +- verify by counting distinct numbers: `SELECT COUNT(DISTINCT number) FROM search_docs WHERE repo = ? AND type IN ('issue','pull_request')` should approach the repository's real issue + PR count + ## Troubleshooting ### `GITHUB_TOKEN not configured` diff --git a/src/backfill-issue-index.test.ts b/src/backfill-issue-index.test.ts new file mode 100644 index 0000000..8c6a3de --- /dev/null +++ b/src/backfill-issue-index.test.ts @@ -0,0 +1,311 @@ +import { describe, it, expect, vi, beforeEach, afterEach } from "vitest"; +import type { Env } from "./types.js"; + +// The embed fan-out (Workers AI + Vectorize + D1 + Store DO) is out of scope for +// the gap-walk contract under test, so the pipeline entry point is a fake and only +// the GitHub calls reach the stubbed global fetch. +const { processAndUpsertIssueMock } = vi.hoisted(() => ({ + processAndUpsertIssueMock: vi.fn(), +})); + +vi.mock("./pipeline.js", async (importOriginal) => { + const actual = await importOriginal(); + return { ...actual, processAndUpsertIssue: processAndUpsertIssueMock }; +}); + +const { + backfillIssueIndex, + fetchHighestItemNumber, + DEFAULT_INDEX_BACKFILL_LIMIT, +} = await import("./backfill-issue-index.js"); + +const REPO = "acme/widgets"; + +/** + * Stub the two GitHub surfaces the module uses: the one-entry listing that + * reports the highest number, and the per-number item fetch. + * + * `present` is the set of numbers GitHub actually has; anything else 404s, which + * is how a deleted or transferred number looks. + */ +function stubGitHub(maxNumber: number, present: Set) { + const fetched: number[] = []; + + const fetchMock = vi.fn(async (input: string | URL) => { + const url = new URL(String(input)); + const detail = url.pathname.match(/^\/repos\/.+\/issues\/(\d+)$/); + + if (detail) { + const number = Number(detail[1]); + fetched.push(number); + if (!present.has(number)) return new Response("Not Found", { status: 404 }); + return new Response( + JSON.stringify({ + number, + title: `item ${number}`, + body: "body", + state: "closed", + labels: [], + milestone: null, + assignees: [], + created_at: "2026-01-01T00:00:00Z", + updated_at: "2026-01-02T00:00:00Z", + html_url: `https://github.com/${REPO}/issues/${number}`, + }), + { status: 200 }, + ); + } + + expect(url.searchParams.get("direction")).toBe("desc"); + return new Response( + JSON.stringify(maxNumber > 0 ? [{ number: maxNumber }] : []), + { status: 200 }, + ); + }); + + vi.stubGlobal("fetch", fetchMock); + return { fetched }; +} + +/** D1 stub understanding only the indexed-number SELECT this module issues. */ +function mkDb(indexedNumbers: number[]) { + const queries: Array<{ from: number; to: number }> = []; + const db = { + prepare: (sql: string) => ({ + bind: (...args: unknown[]) => ({ + all: async () => { + if (!sql.includes("SELECT DISTINCT number")) { + throw new Error(`unexpected statement: ${sql}`); + } + const [, from, to] = args as [string, number, number]; + queries.push({ from, to }); + return { + results: indexedNumbers + .filter((n) => n > from && n <= to) + .map((n) => ({ number: n })), + }; + }, + }), + }), + } as unknown as D1Database; + return { db, queries }; +} + +function mkEnv(indexedNumbers: number[]) { + const { db, queries } = mkDb(indexedNumbers); + const env = { + GITHUB_TOKEN: "test-token", + DB_FTS: db, + ISSUE_STORE: { + idFromName: () => "id", + get: () => ({ fetch: async () => new Response("ok") }), + }, + } as unknown as Env; + return { env, queries }; +} + +/** Numbers handed to the embed pipeline. */ +const ingested = (): number[] => + processAndUpsertIssueMock.mock.calls.map((c) => Number((c[3] as { number: number }).number)); + +/** Every integer in `1..n`. */ +const range = (n: number) => Array.from({ length: n }, (_, i) => i + 1); + +beforeEach(() => { + processAndUpsertIssueMock.mockReset(); + processAndUpsertIssueMock.mockResolvedValue({ + embedded: true, + skippedUnchanged: false, + metadataUpdated: false, + failed: false, + }); +}); + +afterEach(() => { + vi.unstubAllGlobals(); +}); + +describe("backfill-issue-index: the number ceiling", () => { + it("reads the highest number from the newest-created item", async () => { + stubGitHub(1690, new Set()); + expect(await fetchHighestItemNumber(REPO, "test-token")).toBe(1690); + }); + + it("reports 0 for a repository with no issues and no pull requests", async () => { + stubGitHub(0, new Set()); + expect(await fetchHighestItemNumber(REPO, "test-token")).toBe(0); + }); + + it("surfaces a GitHub API error instead of treating it as an empty repository", async () => { + vi.stubGlobal("fetch", vi.fn(async () => new Response("nope", { status: 403 }))); + await expect(fetchHighestItemNumber(REPO, "test-token")).rejects.toThrow(/403/); + }); +}); + +describe("backfill-issue-index: gap detection", () => { + it("ingests only the numbers with no indexed row", async () => { + const { fetched } = stubGitHub(10, new Set(range(10))); + const { env } = mkEnv([1, 2, 3, 5, 8, 9, 10]); + + const summary = await backfillIssueIndex(REPO, env, {}); + + expect(summary.candidates).toBe(3); + expect(fetched).toEqual([4, 6, 7]); + expect(ingested()).toEqual([4, 6, 7]); + expect(summary.indexed).toBe(3); + expect(summary.done).toBe(true); + expect(summary.nextCursor).toBeNull(); + }); + + it("forces the ingest past the body-hash check", async () => { + stubGitHub(2, new Set([1, 2])); + const { env } = mkEnv([1]); + + await backfillIssueIndex(REPO, env, {}); + + expect(processAndUpsertIssueMock.mock.calls[0][4]).toEqual({ force: true }); + }); + + it("writes nothing when the index already covers the whole number space", async () => { + const { fetched } = stubGitHub(10, new Set(range(10))); + const { env } = mkEnv(range(10)); + + const summary = await backfillIssueIndex(REPO, env, {}); + + expect(summary.candidates).toBe(0); + expect(fetched).toEqual([]); + expect(summary.done).toBe(true); + }); + + it("counts a number GitHub does not have without calling the pipeline", async () => { + // 3 was deleted or transferred: it is missing from the index and from GitHub. + stubGitHub(4, new Set([1, 2, 4])); + const { env } = mkEnv([1, 2]); + + const summary = await backfillIssueIndex(REPO, env, {}); + + expect(summary.absent).toBe(1); + expect(summary.indexed).toBe(1); + expect(ingested()).toEqual([4]); + }); + + it("counts an embed failure separately so a later call retries it", async () => { + stubGitHub(2, new Set([1, 2])); + const { env } = mkEnv([]); + processAndUpsertIssueMock.mockImplementation( + async (_e: unknown, _s: unknown, _r: string, issue: { number: number }) => + issue.number === 1 + ? { embedded: false, skippedUnchanged: false, metadataUpdated: false, failed: true } + : { embedded: true, skippedUnchanged: false, metadataUpdated: false, failed: false }, + ); + + const summary = await backfillIssueIndex(REPO, env, {}); + + expect(summary.failed).toBe(1); + expect(summary.indexed).toBe(1); + }); + + it("surfaces a non-404 item fetch error rather than counting it as absent", async () => { + vi.stubGlobal( + "fetch", + vi.fn(async (input: string | URL) => { + const url = new URL(String(input)); + if (/\/issues\/\d+$/.test(url.pathname)) { + return new Response("rate limited", { status: 403 }); + } + return new Response(JSON.stringify([{ number: 2 }]), { status: 200 }); + }), + ); + const { env } = mkEnv([]); + + await expect(backfillIssueIndex(REPO, env, {})).rejects.toThrow(/403/); + }); +}); + +describe("backfill-issue-index: per-call budget", () => { + it("stops at the limit and hands back a cursor on the number it stopped at", async () => { + stubGitHub(10, new Set(range(10))); + const { env } = mkEnv([]); + + const summary = await backfillIssueIndex(REPO, env, { limit: 3 }); + + expect(ingested()).toEqual([1, 2, 3]); + expect(summary.nextCursor).toBe(3); + expect(summary.done).toBe(false); + }); + + it("drains the whole gap across calls when fed its own nextCursor", async () => { + const { env } = mkEnv([4, 5]); + + let cursor: number | null = 0; + let calls = 0; + while (cursor !== null) { + stubGitHub(10, new Set(range(10))); + const summary: Awaited> = + await backfillIssueIndex(REPO, env, { limit: 3, cursor }); + cursor = summary.nextCursor; + calls++; + } + + expect(ingested()).toEqual([1, 2, 3, 6, 7, 8, 9, 10]); + expect(calls).toBe(3); + }); + + it("resumes after the cursor without re-examining what came before", async () => { + const { fetched } = stubGitHub(10, new Set(range(10))); + const { env, queries } = mkEnv([]); + + await backfillIssueIndex(REPO, env, { cursor: 7 }); + + expect(fetched).toEqual([8, 9, 10]); + expect(queries[0].from).toBe(7); + }); + + it("defaults the per-call budget to DEFAULT_INDEX_BACKFILL_LIMIT", async () => { + stubGitHub(0, new Set()); + const { env } = mkEnv([]); + + const summary = await backfillIssueIndex(REPO, env, {}); + + expect(summary.limit).toBe(DEFAULT_INDEX_BACKFILL_LIMIT); + }); + + it("terminates on a cursor already past the highest number", async () => { + stubGitHub(10, new Set(range(10))); + const { env } = mkEnv([]); + + const summary = await backfillIssueIndex(REPO, env, { cursor: 99 }); + + expect(summary.candidates).toBe(0); + expect(summary.done).toBe(true); + }); + + it("chunks the indexed-set query rather than reading the whole index at once", async () => { + stubGitHub(450, new Set()); + const { env, queries } = mkEnv(range(450)); + + await backfillIssueIndex(REPO, env, {}); + + expect(queries).toEqual([ + { from: 0, to: 200 }, + { from: 200, to: 400 }, + { from: 400, to: 450 }, + ]); + }); +}); + +describe("backfill-issue-index: dry run", () => { + it("measures the gap over the whole scan range without fetching or writing", async () => { + const { fetched } = stubGitHub(10, new Set(range(10))); + const { env } = mkEnv([1, 2]); + + const summary = await backfillIssueIndex(REPO, env, { dryRun: true, limit: 3 }); + + // The fetch budget is not spent, so the measurement is not truncated by it. + expect(summary.candidates).toBe(8); + expect(summary.attempted).toBe(0); + expect(fetched).toEqual([]); + expect(ingested()).toEqual([]); + expect(summary.done).toBe(true); + }); +}); diff --git a/src/backfill-issue-index.ts b/src/backfill-issue-index.ts new file mode 100644 index 0000000..9571653 --- /dev/null +++ b/src/backfill-issue-index.ts @@ -0,0 +1,259 @@ +/** + * Re-index the issues and pull requests the poller left out of the index. + * + * Layer = L4 Operations (index repair surface) + * + * The poller advanced its watermark past every item the per-run embedding budget + * deferred, so those items were marked for retry and then never fetched again + * (issue #210). Measured 2026-08-03, that left roughly 55% of the issue / PR + * history of the indexed repositories absent from `search_docs`. The watermark + * is fixed at the source (`nextIssueWatermark` in `./poller.ts`), but the fix + * only stops the leak: an item already stranded is not re-fetched by the poller, + * because its `updated_at` no longer moves. This module walks the gap directly. + * + * Coverage is measured against `search_docs`, the sparse retrieval surface — the + * same axis the issue measured, and the one a missing row makes a search silently + * incomplete on. Walking by *issue number* rather than by timestamp is what makes + * the sweep resumable and complete: the number space is dense and bounded, so a + * numeric cursor states exactly how far the sweep has reached, while a timestamp + * cursor over the same set would reintroduce the ordering the defect exploited. + * + * Unlike `./backfill-issue-state.ts`, this repair does embed: a missing row has + * no vector to re-upsert. The per-call budget is therefore a Workers AI budget + * first, and the caller drives the sweep one batch at a time. + */ + +import type { Env } from "./types.js"; +import { processAndUpsertIssue, type GitHubIssueData } from "./pipeline.js"; + +/** Candidate numbers attempted per call, unless the caller lowers it. Each one + * costs a GitHub fetch plus (when present) an embed + Vectorize + D1 + store + * fan-out, ~5 of the invocation's 1000 subrequests. */ +export const DEFAULT_INDEX_BACKFILL_LIMIT = 25; + +/** Hard ceiling on the per-call candidate budget. 100 candidates x ~5 subrequests + * ≈ 500, half the per-invocation budget, and 100 embeds is twice the cron's own + * per-run Workers AI allowance. */ +export const MAX_INDEX_BACKFILL_LIMIT = 100; + +/** Issue numbers covered by one indexed-set query. */ +const SCAN_CHUNK = 200; + +/** Chunk queries per call. Bounds the scan on a repository whose gap sits far + * above the cursor: 25 x 200 = 5000 numbers examined before the call returns a + * cursor and lets the caller decide whether to continue. */ +const MAX_SCAN_CHUNKS = 25; + +export interface IssueIndexBackfillOptions { + /** Report the gap without fetching from GitHub or writing anything. */ + dryRun?: boolean; + /** Candidate numbers attempted (default `DEFAULT_INDEX_BACKFILL_LIMIT`). + * Ignored on a dry run, which spends no fetch budget. */ + limit?: number; + /** Issue number to resume after (exclusive). */ + cursor?: number; +} + +export interface IssueIndexBackfillSummary { + repo: string; + dryRun: boolean; + cursor: number; + limit: number; + /** Highest issue / PR number GitHub reports for the repository. */ + maxNumber: number; + /** Last number this call examined. */ + scannedTo: number; + /** Numbers in the examined range with no `search_docs` issue / PR row. */ + candidates: number; + /** Candidates this call fetched from GitHub. 0 on a dry run. */ + attempted: number; + /** Candidates now on both retrieval surfaces. */ + indexed: number; + /** Candidates GitHub does not have (deleted or transferred numbers). */ + absent: number; + /** Candidates whose embed or upsert failed; retried by a later call. */ + failed: number; + /** Pass back as `cursor` to continue; `null` once the sweep reached `maxNumber`. */ + nextCursor: number | null; + done: boolean; +} + +const GITHUB_HEADERS = (token: string): Record => ({ + Authorization: `Bearer ${token}`, + Accept: "application/vnd.github+json", + "X-GitHub-Api-Version": "2022-11-28", + "User-Agent": "github-rag-mcp/0.1.0", +}); + +/** + * Highest issue / PR number the repository has. + * + * Issues and pull requests share one ascending sequence, so the most recently + * *created* item carries the highest number — one listing entry answers it. + * Returns 0 for a repository with no issues and no pull requests. + */ +export async function fetchHighestItemNumber( + repo: string, + token: string, +): Promise { + const url = new URL(`https://api.github.com/repos/${repo}/issues`); + url.searchParams.set("state", "all"); + url.searchParams.set("sort", "created"); + url.searchParams.set("direction", "desc"); + url.searchParams.set("per_page", "1"); + + const resp = await fetch(url.toString(), { + headers: GITHUB_HEADERS(token), + 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 }>; + return items.length > 0 ? Number(items[0].number) : 0; +} + +/** Issue / PR numbers of the repository's `search_docs` rows within `(from, to]`. */ +export async function selectIndexedNumbers( + db: D1Database, + repo: string, + from: number, + to: number, +): Promise> { + const res = await db + .prepare( + `SELECT DISTINCT number + FROM search_docs + WHERE repo = ? + AND type IN ('issue', 'pull_request') + AND number > ? + AND number <= ?`, + ) + .bind(repo, from, to) + .all<{ number: number }>(); + + return new Set((res.results ?? []).map((r) => Number(r.number))); +} + +/** Fetch one issue / PR by number; `null` when GitHub does not have that number. */ +async function fetchItem( + repo: string, + token: string, + number: number, +): Promise { + const resp = await fetch( + `https://api.github.com/repos/${repo}/issues/${number}`, + { headers: GITHUB_HEADERS(token), cache: "no-store" } as RequestInit, + ); + + // A number the repository never had, or whose item was deleted. Not every + // number in the range is an issue — the sequence also skips over items moved + // out of the repository. + if (resp.status === 404 || resp.status === 410) return null; + + if (!resp.ok) { + throw new Error( + `GitHub API error ${resp.status} for ${repo}#${number}: ${await resp.text()}`, + ); + } + + return (await resp.json()) as GitHubIssueData; +} + +/** + * Index one batch of the repository's missing issue / PR numbers. + * + * Resumable and idempotent: progress is a number cursor over the ascending + * number space, an item already carrying a `search_docs` row is never fetched, + * and re-running a batch re-issues writes that are already correct. Call it + * repeatedly with the returned `nextCursor` until `done`. + * + * The ingest is forced past the body-hash check. The hash answers "did the body + * change", but every candidate here is known to be missing a retrieval surface, + * which a matching hash would otherwise skip over permanently. + */ +export async function backfillIssueIndex( + repo: string, + env: Env, + options: IssueIndexBackfillOptions = {}, +): Promise { + const dryRun = options.dryRun === true; + const limit = options.limit ?? DEFAULT_INDEX_BACKFILL_LIMIT; + const cursor = options.cursor ?? 0; + + const maxNumber = await fetchHighestItemNumber(repo, env.GITHUB_TOKEN); + + // Collect candidates chunk by chunk. A dry run spends no fetch budget, so it + // measures the whole scan range instead of stopping at `limit`. + const candidates: number[] = []; + let scannedTo = Math.min(cursor, maxNumber); + for (let chunk = 0; chunk < MAX_SCAN_CHUNKS && scannedTo < maxNumber; chunk++) { + const from = scannedTo; + const to = Math.min(from + SCAN_CHUNK, maxNumber); + const indexed = await selectIndexedNumbers(env.DB_FTS, repo, from, to); + + let stopAt: number | undefined; + for (let n = from + 1; n <= to; n++) { + if (indexed.has(n)) continue; + candidates.push(n); + if (!dryRun && candidates.length >= limit) { + // The budget is spent on this number; everything above it is unexamined. + stopAt = n; + break; + } + } + + scannedTo = stopAt ?? to; + if (stopAt !== undefined) break; + } + + let attempted = 0; + let indexed = 0; + let absent = 0; + let failed = 0; + + if (!dryRun) { + const storeId = env.ISSUE_STORE.idFromName("global"); + const storeStub = env.ISSUE_STORE.get(storeId); + + for (const number of candidates) { + attempted++; + const item = await fetchItem(repo, env.GITHUB_TOKEN, number); + if (item === null) { + absent++; + continue; + } + const result = await processAndUpsertIssue(env, storeStub, repo, item, { + force: true, + }); + if (result.embedded) indexed++; + else failed++; + } + } + + const done = scannedTo >= maxNumber; + + console.log( + `${repo} backfill-issue-index: max=${maxNumber} cursor=${cursor} ` + + `scanned_to=${scannedTo} candidates=${candidates.length} attempted=${attempted} ` + + `indexed=${indexed} absent=${absent} failed=${failed}${dryRun ? " (dry run)" : ""}`, + ); + + return { + repo, + dryRun, + cursor, + limit, + maxNumber, + scannedTo, + candidates: candidates.length, + attempted, + indexed, + absent, + failed, + nextCursor: done ? null : scannedTo, + done, + }; +} diff --git a/src/index.ts b/src/index.ts index 458f292..6f1ff0f 100644 --- a/src/index.ts +++ b/src/index.ts @@ -19,6 +19,7 @@ * 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) + * POST /admin/backfill-issue-index?repo=owner/repo[&dry_run=true][&limit=N][&cursor=N] -- Index the issue/PR numbers missing from search_docs (requires GITHUB_TOKEN header) * * Durable Objects: * RagMcpAgentV2 -- MCP server (tools: search, get_issue_context, list_recent_activity) @@ -52,6 +53,11 @@ import { MAX_ISSUE_STATE_LIMIT, backfillIssueState, } from "./backfill-issue-state.js"; +import { + DEFAULT_INDEX_BACKFILL_LIMIT, + MAX_INDEX_BACKFILL_LIMIT, + backfillIssueIndex, +} from "./backfill-issue-index.js"; // Durable Object: issue/PR state store (SQLite-backed) export { IssueStore } from "./store.js"; @@ -504,6 +510,59 @@ const innerHandler: ExportedHandler = { } } + // -- Admin: index the issue / PR numbers missing from the retrieval surfaces -- + // POST /admin/backfill-issue-index?repo=owner/repo[&dry_run=true][&limit=N][&cursor=N] + // Fills the gap the poller's watermark left behind (issue #210). Walks the repo's + // issue-number space, finds the numbers with no `search_docs` issue / PR row, and + // ingests them. Unlike `/admin/backfill-issue-state` this embeds, so the per-call + // budget is a Workers AI budget; unlike `/admin/reset-hashes` it re-embeds only the + // missing items rather than the whole repository. + // `dry_run=true` measures the gap over the scan range without spending either budget. + // 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-index") { + 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_INDEX_BACKFILL_LIMIT : Number(rawLimit); + if (!Number.isInteger(limit) || limit < 1 || limit > MAX_INDEX_BACKFILL_LIMIT) { + return new Response(`limit must be an integer in 1..${MAX_INDEX_BACKFILL_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 backfillIssueIndex(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.ts b/src/pipeline/embed-issue.ts index 23153d3..a557738 100644 --- a/src/pipeline/embed-issue.ts +++ b/src/pipeline/embed-issue.ts @@ -37,6 +37,19 @@ export interface GitHubIssueData { html_url: string; } +/** Per-call overrides for `processAndUpsertIssue`. */ +export interface ProcessIssueOptions { + /** Embed and re-upsert even when the stored hash matches. + * + * The hash check answers "did the body change", which is the wrong question + * when a retrieval surface is known to be missing the item: an embed whose + * FTS5 mirror failed leaves the hash stored and the sparse row absent, and no + * later poll reconciles it because the diff basis already matches. Only the + * index backfill sets this (issue #210); the poll and webhook paths must keep + * the hash check, or every run would re-embed the whole repository. */ + force?: boolean; +} + /** * Process and upsert a single issue/PR: check hash, embed if changed, upsert to Vectorize + Store. * @@ -44,6 +57,7 @@ export interface GitHubIssueData { * @param storeStub - Durable Object stub for IssueStore * @param repo - Repository in "owner/repo" format * @param issue - GitHub issue/PR data + * @param options - per-call overrides (see `ProcessIssueOptions`) * @returns UpsertResult indicating what happened */ export async function processAndUpsertIssue( @@ -51,6 +65,7 @@ export async function processAndUpsertIssue( storeStub: DurableObjectStub, repo: string, issue: GitHubIssueData, + options: ProcessIssueOptions = {}, ): Promise { const body = issue.body ?? ""; const title = issue.title; @@ -71,7 +86,7 @@ export async function processAndUpsertIssue( let existing: IssueRecord | null = null; if (existingResp.ok) { existing = (await existingResp.json()) as IssueRecord; - if (existing.bodyHash === bodyHash) { + if (existing.bodyHash === bodyHash && options.force !== true) { needsEmbedding = false; } } diff --git a/src/poller.test.ts b/src/poller.test.ts index b46773d..e0a3d97 100644 --- a/src/poller.test.ts +++ b/src/poller.test.ts @@ -6,9 +6,14 @@ import type { Env } from "./types.js"; // upstream of all of that, so the two pipeline entry points are replaced with // controllable fakes and only the *commit list* call reaches the stubbed global // fetch. Everything else in `./pipeline.js` stays real. -const { fetchCommitDetailMock, processAndUpsertCommitDiffMock } = vi.hoisted(() => ({ +const { + fetchCommitDetailMock, + processAndUpsertCommitDiffMock, + processAndUpsertIssueMock, +} = vi.hoisted(() => ({ fetchCommitDetailMock: vi.fn(), processAndUpsertCommitDiffMock: vi.fn(), + processAndUpsertIssueMock: vi.fn(), })); vi.mock("./pipeline.js", async (importOriginal) => { @@ -17,13 +22,16 @@ vi.mock("./pipeline.js", async (importOriginal) => { ...actual, fetchCommitDetail: fetchCommitDetailMock, processAndUpsertCommitDiff: processAndUpsertCommitDiffMock, + processAndUpsertIssue: processAndUpsertIssueMock, }; }); const { pollDiffs, + pollRepo, nextForwardDiffWatermark, nextBackfillDiffWatermark, + nextIssueWatermark, } = await import("./poller.js"); const REPO = "acme/widgets"; @@ -45,6 +53,9 @@ const commit = (sha: string, date: string): FakeCommit => ({ sha, date }); */ function makeStore(seed: Record = {}) { const watermarks = new Map(Object.entries(seed)); + const etags = new Map(); + /** Records written through `/upsert` (the empty-hash retry markers). */ + const upserts: Array<{ number: number; bodyHash: string }> = []; const stub = { async fetch(request: Request): Promise { const url = new URL(request.url); @@ -52,17 +63,30 @@ function makeStore(seed: Record = {}) { const key = url.searchParams.get("repo") ?? ""; const value = watermarks.get(key); if (!value) return new Response("not found", { status: 404 }); - return Response.json({ repo: key, lastPolledAt: value }); + const etag = etags.get(key); + return Response.json({ repo: key, lastPolledAt: value, etag }); } if (request.method === "POST" && url.pathname === "/watermark") { - const body = (await request.json()) as { repo: string; lastPolledAt: string }; + const body = (await request.json()) as { + repo: string; + lastPolledAt: string; + etag?: string; + }; watermarks.set(body.repo, body.lastPolledAt); + // The store column defaults to '' — an omitted ETag clears the stored one. + if (body.etag) etags.set(body.repo, body.etag); + else etags.delete(body.repo); + return new Response("ok"); + } + if (request.method === "POST" && url.pathname === "/upsert") { + const body = (await request.json()) as { number: number; bodyHash: string }; + upserts.push({ number: body.number, bodyHash: body.bodyHash }); return new Response("ok"); } return new Response("ok"); }, }; - return { stub: stub as unknown as DurableObjectStub, watermarks }; + return { stub: stub as unknown as DurableObjectStub, watermarks, etags, upserts }; } /** @@ -101,6 +125,75 @@ function stubCommitList(commits: FakeCommit[], opts: { fail?: boolean } = {}) { return { listQueries }; } +/** One issue as the GitHub list endpoint returns it (subset the poller reads). */ +function fakeIssue(number: number, updatedAt: string) { + return { + number, + title: `issue ${number}`, + body: "body", + state: "open" as const, + labels: [], + milestone: null, + assignees: [], + created_at: updatedAt, + updated_at: updatedAt, + html_url: `https://github.com/${REPO}/issues/${number}`, + }; +} + +/** `n` issues one minute apart, ascending — the order GitHub returns them in. */ +function issueSeries(n: number) { + return Array.from({ length: n }, (_, i) => + fakeIssue(i + 1, new Date(Date.UTC(2026, 6, 1, 0, i)).toISOString()), + ); +} + +const ISSUE_ETAG = 'W/"issues-v1"'; + +/** + * Stub the global fetch with a fake GitHub issue-list endpoint. + * + * Mirrors the contract `pollRepo` depends on: `sort=updated&direction=asc`, + * `since` filtering, `per_page` pagination, and a conditional request that + * answers 304 when the caller echoes the stored ETag back. The 304 is what makes + * an ETag stored over unfinished work fatal rather than merely wasteful. + */ +function stubIssueList(issues: ReturnType[]) { + const sinceQueries: Array = []; + + const fetchMock = vi.fn(async (input: string | URL, init?: RequestInit) => { + const url = new URL(String(input)); + const headers = (init?.headers ?? {}) as Record; + if (headers["If-None-Match"] === ISSUE_ETAG) { + return new Response(null, { status: 304 }); + } + + const since = url.searchParams.get("since") ?? undefined; + const perPage = Number(url.searchParams.get("per_page") ?? "100"); + const page = Number(url.searchParams.get("page") ?? "1"); + if (page === 1) sinceQueries.push(since); + + // GitHub documents `since` as "updated after" — model the exclusive reading, + // so a watermark pinned exactly at an item's timestamp would lose it. + const selected = issues + .filter((i) => (since ? Date.parse(i.updated_at) > Date.parse(since) : true)) + .sort((a, b) => Date.parse(a.updated_at) - Date.parse(b.updated_at)) + .slice((page - 1) * perPage, page * perPage); + + return new Response(JSON.stringify(selected), { + status: 200, + headers: { "Content-Type": "application/json", etag: ISSUE_ETAG }, + }); + }); + + vi.stubGlobal("fetch", fetchMock); + return { sinceQueries }; +} + +/** Issue numbers handed to the embed pipeline, i.e. the items a run attempted. */ +const attemptedIssues = (): number[] => + processAndUpsertIssueMock.mock.calls.map((call) => Number((call[3] as { number: number }).number)); + const env = { GITHUB_TOKEN: "test-token" } as unknown as Env; /** SHAs handed to the detail fetch, i.e. the commits a run actually attempted. */ @@ -110,6 +203,14 @@ const attemptedShas = (): string[] => beforeEach(() => { fetchCommitDetailMock.mockReset(); processAndUpsertCommitDiffMock.mockReset(); + processAndUpsertIssueMock.mockReset(); + // Default: every issue embeds cleanly. + processAndUpsertIssueMock.mockResolvedValue({ + embedded: true, + skippedUnchanged: false, + metadataUpdated: false, + failed: false, + }); // Default: every commit ingests cleanly. fetchCommitDetailMock.mockImplementation(async (_repo: string, sha: string) => ({ sha, @@ -335,6 +436,142 @@ describe("poller: pollDiffs forward watermark / retry boundary", () => { }); }); +describe("poller: nextIssueWatermark", () => { + const since = "2026-07-01T00:00:00.000Z"; + const candidate = "2026-07-01T12:00:00.000Z"; + + it("takes the fetch bound when every item was ingested", () => { + expect(nextIssueWatermark(since, candidate, undefined)).toBe(candidate); + }); + + it("pins just before the first uningested item", () => { + expect( + nextIssueWatermark(since, candidate, "2026-07-01T02:00:00.000Z"), + ).toBe("2026-07-01T01:59:59.000Z"); + }); + + it("pins on an initial sync, where there is no watermark yet", () => { + expect( + nextIssueWatermark(undefined, candidate, "2026-07-01T02:00:00.000Z"), + ).toBe("2026-07-01T01:59:59.000Z"); + }); + + it("never regresses below the watermark it started from", () => { + expect(nextIssueWatermark(since, candidate, since)).toBe(since); + }); + + it("holds the watermark when the boundary item carries no usable timestamp", () => { + expect(nextIssueWatermark(since, candidate, "not-a-date")).toBe(since); + expect(nextIssueWatermark(undefined, candidate, "not-a-date")).toBeUndefined(); + }); +}); + +describe("poller: pollRepo watermark / retry boundary", () => { + it("keeps items the embedding budget deferred inside the next run's window", async () => { + // 60 items against a per-run embedding budget of 50. + const issues = issueSeries(60); + const { stub, watermarks, upserts } = makeStore(); + stubIssueList(issues); + + await pollRepo(REPO, env, stub); + + // The budget stops at 50; the rest are marked for retry, not embedded. + expect(attemptedIssues()).toEqual(issues.slice(0, 50).map((i) => i.number)); + expect(upserts.map((u) => u.number)).toEqual( + issues.slice(50).map((i) => i.number), + ); + expect(upserts.every((u) => u.bodyHash === "")).toBe(true); + + // Watermark parked before item 51 rather than at the poll start time. + const wm = watermarks.get(REPO)!; + expect(Date.parse(wm)).toBeLessThan(Date.parse(issues[50].updated_at)); + expect(Date.parse(wm)).toBeGreaterThanOrEqual(Date.parse(issues[49].updated_at)); + + // Next run: the deferred items are back in the window. Before issue #210 the + // watermark had already moved past them and they were never fetched again. + processAndUpsertIssueMock.mockClear(); + await pollRepo(REPO, env, stub); + + expect(attemptedIssues()).toEqual(issues.slice(50).map((i) => i.number)); + }); + + it("withholds the ETag while work is left behind, so the retry is not answered 304", async () => { + const issues = issueSeries(60); + const { stub, etags } = makeStore(); + stubIssueList(issues); + + await pollRepo(REPO, env, stub); + + expect(etags.get(REPO)).toBeUndefined(); + }); + + it("advances to the poll start time and stores the ETag once everything landed", async () => { + const issues = issueSeries(3); + const { stub, watermarks, etags } = makeStore(); + stubIssueList(issues); + + await pollRepo(REPO, env, stub); + + expect(attemptedIssues()).toEqual([1, 2, 3]); + expect(Date.parse(watermarks.get(REPO)!)).toBeGreaterThan( + Date.parse(issues[2].updated_at), + ); + expect(etags.get(REPO)).toBe(ISSUE_ETAG); + }); + + it("pins the watermark before an item whose embed failed", async () => { + const issues = issueSeries(3); + const { stub, watermarks } = makeStore(); + stubIssueList(issues); + + processAndUpsertIssueMock.mockImplementation( + async ( + _env: unknown, + _stub: unknown, + _repo: string, + issue: { number: number }, + ) => + issue.number === 2 + ? { embedded: false, skippedUnchanged: false, metadataUpdated: false, failed: true } + : { embedded: true, skippedUnchanged: false, metadataUpdated: false, failed: false }, + ); + + await pollRepo(REPO, env, stub); + + const wm = watermarks.get(REPO)!; + expect(Date.parse(wm)).toBeLessThan(Date.parse(issues[1].updated_at)); + + // Item 2 is back in the next window rather than stranded behind the watermark. + processAndUpsertIssueMock.mockClear(); + processAndUpsertIssueMock.mockResolvedValue({ + embedded: true, + skippedUnchanged: false, + metadataUpdated: false, + failed: false, + }); + await pollRepo(REPO, env, stub); + + expect(attemptedIssues()).toContain(2); + }); + + it("drains a backlog larger than the budget across runs without skipping", async () => { + const issues = issueSeries(130); + const { stub } = makeStore(); + stubIssueList(issues); + + const seen = new Set(); + for (let run = 0; run < 3; run++) { + processAndUpsertIssueMock.mockClear(); + await pollRepo(REPO, env, stub); + for (const n of attemptedIssues()) seen.add(n); + } + + // Every item reached the pipeline across the three runs; none fell into the + // gap between one run's budget and the next run's `since`. + expect(seen.size).toBe(130); + }); +}); + describe("poller: pollDiffs backfill watermark", () => { it("does not step over a failed commit", async () => { const commits = [ diff --git a/src/poller.ts b/src/poller.ts index 857841e..c5f38ea 100644 --- a/src/poller.ts +++ b/src/poller.ts @@ -41,7 +41,9 @@ const PER_PAGE = 100; /** Maximum number of embeddings to generate per single cron run. * Prevents Workers AI rate-limit errors on large repos. - * Remaining issues are stored with empty bodyHash and retried next cron. */ + * Remaining issues are stored with empty bodyHash and retried next cron. + * The retry only happens because the watermark is held at the first such item + * (see `nextIssueWatermark`); the empty hash marks the item, it does not fetch it. */ const MAX_EMBEDDINGS_PER_RUN = 50; /** Maximum number of API pages to fetch per single cron run. @@ -192,6 +194,17 @@ const MAX_DIFF_FORWARD_WINDOW_SHRINKS = 8; * idempotent on (repo, commit_sha, file_path). */ const DIFF_RETRY_BOUNDARY_BACKOFF_MS = 1000; +/** Safety margin subtracted from the issue / PR poller's retry boundary, for the + * same reason as `DIFF_RETRY_BOUNDARY_BACKOFF_MS`: `/issues` documents `since` + * as "only show results that were last updated *after* the given time", so an + * item whose timestamp equals the watermark may be excluded — and items sharing + * an `updated_at` are common (one bulk label edit touches many). Without the + * margin the very item the watermark is pinned to could fall outside the next + * run's window, which is the defect being fixed, reintroduced one item wide. + * Re-fetching a few already-ingested items costs a hash comparison each and no + * embedding. */ +const ISSUE_RETRY_BOUNDARY_BACKOFF_MS = 1000; + /** Sentinel value indicating GitHub returned 304 Not Modified */ const NOT_MODIFIED = Symbol("NOT_MODIFIED"); @@ -318,23 +331,39 @@ async function fetchAllIssues( return { issues: allIssues, capped: false, notModified: false, responseEtag }; } +/** Outcome of one `processIssues` batch. */ +export interface IssueBatchStats { + processed: number; + embedded: number; + skipped: number; + failed: number; + /** `updated_at` of the first item the batch did not get onto the retrieval + * surfaces — deferred by the embedding budget, or failed to embed. Absent when + * every item landed. The caller pins the watermark before it (issue #210). */ + retryBoundary?: string; +} + /** * Process a batch of issues: compute hashes, generate embeddings for changed items, * upsert into Vectorize and IssueStore. * * Delegates per-item embedding+upsert to the shared pipeline, but manages - * batch-level concerns: embedding count cap and stats tracking. + * batch-level concerns: embedding count cap, stats tracking, and reporting the + * first item left uningested so the caller can hold the watermark before it. + * The batch arrives sorted by `updated_at` ascending, so the first such item is + * also the earliest one — pinning to it covers every later one too. */ async function processIssues( issues: GitHubIssueData[], repo: string, env: Env, storeStub: DurableObjectStub, -): Promise<{ processed: number; embedded: number; skipped: number; failed: number }> { +): Promise { let processed = 0; let embedded = 0; let skipped = 0; let failed = 0; + let retryBoundary: string | undefined; for (const issue of issues) { // Enforce per-run embedding limit to avoid Workers AI rate limits. @@ -346,8 +375,8 @@ async function processIssues( `Remaining issues will be retried next cron run.`, ); } + retryBoundary ??= issue.updated_at; // Store record with empty bodyHash to trigger retry on next poll - const body = issue.body ?? ""; const type: IssueRecord["type"] = issue.pull_request ? "pull_request" : "issue"; @@ -383,18 +412,62 @@ async function processIssues( embedded++; } else if (result.failed) { failed++; + retryBoundary ??= issue.updated_at; } processed++; } - return { processed, embedded, skipped, failed }; + return { processed, embedded, skipped, failed, retryBoundary }; +} + +/** + * Compute the next issue / PR poller watermark. + * + * Invariant, mirroring the commit-diff surface (issue #178 / #179): **the + * watermark never advances past the earliest item this run left off the + * retrieval surfaces.** Before issue #210 the watermark moved unconditionally — + * to the poll start time, or to the last fetched item's `updated_at` when + * pagination capped. Both are *newer* than every deferred item, because the + * fetch is sorted `updated` ascending and the embedding budget always runs out + * on the newest end of the batch. So an item the budget deferred fell out of + * every later `since` window: it was marked for retry (empty bodyHash) and then + * never fetched again. At 200 fetched vs 50 embedded per run, that stranded 150 + * items per run during an initial sync. + * + * @param since watermark the run started from; absent on initial sync + * @param candidate watermark the run would store if everything landed + * @param retryBoundary `updated_at` of the first uningested item, if any + * @returns the watermark to persist, or `undefined` to leave it unchanged + */ +export function nextIssueWatermark( + since: string | undefined, + candidate: string, + retryBoundary: string | undefined, +): string | undefined { + if (!retryBoundary) return candidate; + + // An item we cannot place on the timeline cannot bound the retry window; + // holding keeps the whole period retryable. + const boundaryTime = Date.parse(retryBoundary); + if (Number.isNaN(boundaryTime)) return since; + + const pinned = new Date( + boundaryTime - ISSUE_RETRY_BOUNDARY_BACKOFF_MS, + ).toISOString(); + if (since === undefined) return pinned; + + // Never regress: the boundary sits at (or before) the current watermark when + // the first item of the batch is the one that failed. + const sinceTime = Date.parse(since); + if (Number.isNaN(sinceTime)) return pinned; + return Date.parse(pinned) > sinceTime ? pinned : since; } /** * Poll a single repository for issue/PR updates. */ -async function pollRepo( +export async function pollRepo( repo: string, env: Env, storeStub: DurableObjectStub, @@ -452,40 +525,61 @@ async function pollRepo( // Process issues (embedding + store) const stats = await processIssues(issues, repo, env, storeStub); - // Watermark strategy: + // Watermark strategy, in two steps. + // + // Step 1 — how far the *fetch* reached: // - If all pages were fetched (not capped): use pollStartTime so next run // picks up anything updated during this fetch. // - If pagination was capped: use the updated_at of the last fetched issue // (sorted by updated asc) so the next cron continues from where we left off. // Using pollStartTime here would skip the remaining unfetched issues. - let nextWatermark: string; + const fetchWatermark = capped + ? issues[issues.length - 1].updated_at + : pollStartTime; if (capped) { - const lastIssue = issues[issues.length - 1]; - nextWatermark = lastIssue.updated_at; console.log( - `${repo}: pagination was capped — watermark set to last fetched issue updated_at: ${nextWatermark}`, + `${repo}: pagination was capped — fetch reached ${fetchWatermark}`, ); - } else { - nextWatermark = pollStartTime; } - // Update watermark after successful processing (with new ETag for next conditional request) - // When pagination is capped, don't store ETag — the partial fetch means the ETag - // wouldn't match the next request which starts from a different watermark position. - await storeStub.fetch( - new Request("http://store/watermark", { - method: "POST", - headers: { "Content-Type": "application/json" }, - body: JSON.stringify({ - repo, - lastPolledAt: nextWatermark, - etag: capped ? undefined : responseEtag, - }), - }), + // Step 2 — how far the *ingest* reached. The fetch bound is an upper bound + // only: the embedding budget stops well short of it, and everything past the + // stopping point must stay inside the next run's window (issue #210). + const nextWatermark = nextIssueWatermark( + since, + fetchWatermark, + stats.retryBoundary, ); + // Update watermark after successful processing (with new ETag for next conditional request). + // The ETag is withheld whenever the run is leaving work behind — pagination capped, + // or the ingest stopped short. Storing it would make the next run's conditional + // request answer 304 and return before it looked at the leftover, which is the same + // hold the docs poller applies to its tree ETag. + const holdEtag = capped || stats.retryBoundary !== undefined; + if (nextWatermark === undefined) { + console.warn( + `${repo}: watermark held — the first uningested item carries no usable updated_at`, + ); + } else { + await storeStub.fetch( + new Request("http://store/watermark", { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ + repo, + lastPolledAt: nextWatermark, + etag: holdEtag ? undefined : responseEtag, + }), + }), + ); + } + console.log( - `${repo}: ${stats.processed} processed, ${stats.embedded} embedded, ${stats.skipped} unchanged, ${stats.failed} failed`, + `${repo}: ${stats.processed} processed, ${stats.embedded} embedded, ` + + `${stats.skipped} unchanged, ${stats.failed} failed, ` + + `watermark=${nextWatermark ?? "held"}` + + `${stats.retryBoundary ? ` (pinned before ${stats.retryBoundary})` : ""}`, ); }