From ef46e86073aa8b53eb22873006d55a3b4d66cf3e Mon Sep 17 00:00:00 2001 From: Claude Lin & Lay Date: Fri, 14 Aug 2026 16:15:03 +0900 Subject: [PATCH] fix(poller): cap diff-path files per run against the subrequest budget [poller, pipeline, docs, tests] MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit diff path に file 数軸の上限が無く、file 数の多い commit 単独で Worker の 1 invocation あたり subrequest 予算を使い切っていた。#237 の merge 後も watermark が前進せず、token 軸のエラーが `Too many subrequests by single Worker invocation` に置き換わっただけだった観測（neuron-graph-rag@92eb94d、 44 file、embed batch 3 本すべて拒否）に対応する。 変更点: - `diffFileBudgetPerPhase(repoCount)` を追加。diff surface の予算取り分 (`DIFF_SUBREQUEST_BUDGET_PER_RUN` = 1000 のうち 900) を `repoCount × 2` で割り、file 数に換算する。literal を固定しないのは POLL_REPOS が通常の config commit で増えるため（#233 が 6 番目を追加）。下限 5 file を置き、 予算 0 による恒久停止ではなく回復可能な超過側に倒す。 - forward / backward 両 phase が独立の file 予算を持つ。forward が先に走る ため予算を共有すると歴史遡行が永久に進まなくなる。 - `processAndUpsertCommitDiff` に `maxFiles` / `indexedFilePaths` を追加し、 戻り値に `deferred` / `alreadyIndexed` を追加。間引きではなく分割で、 未処理 file は必ず報告される。 - 分割中の commit は新 status `partial`。watermark 不変条件の判定は `status !== "ok"` なので、最後の file が index されるまで watermark は 前進しない。#178 の不変条件は変更していない。 - 分割の再開位置は structured store の `GET /diffs` から読む。これが無いと 上限付きの run が毎 cron 同じ先頭側を再 index して収束しない。予算を超える commit だけが 1 subrequest を払う。store 側エラーは「未 index」ではなく commit の失敗として扱う。 - 要件記述と installation doc を同一 PR で更新。 webhook 経路は per-push 1 invocation で圧力が異なり、watermark を持たない ため上限を入れると file を落とすことになる。本 PR の範囲外とし、webhook が 取りこぼした commit は forward phase が redundancy として拾い、そこでは 分割が効く。 test: embed-diff の分割 / 再開 / 収束、poller の予算導出（14 repo まで 1000 subrequest 以内、下限 5 を割らない）、44 file commit の複数 run 完走、 分割中の watermark 保持、phase ごとの独立予算、resume query 失敗時の扱い。 Closes #238 --- docs/0-requirements.ja.md | 12 +- docs/0-requirements.md | 12 +- docs/installation.ja.md | 3 +- docs/installation.md | 3 +- src/pipeline/embed-diff.test.ts | 127 ++++++++++++- src/pipeline/embed-diff.ts | 57 +++++- src/poller.test.ts | 267 +++++++++++++++++++++++++- src/poller.ts | 322 ++++++++++++++++++++++++++++---- 8 files changed, 743 insertions(+), 60 deletions(-) diff --git a/docs/0-requirements.ja.md b/docs/0-requirements.ja.md index 6c67f3b..efc0176 100644 --- a/docs/0-requirements.ja.md +++ b/docs/0-requirements.ja.md @@ -188,11 +188,17 @@ commit diff poller は 2-phase 構成: - **forward phase** — `(lastPolledAt, pollStartTime]` の window を列挙し、その中の**古い側から**取り込む（webhook 取りこぼし時の redundancy）。watermark namespace は `diffs:${repo}`。 - **backward phase** — `until=oldestUnprocessedDate` で履歴を徐々に遡行する（新規 deployment や webhook 起動前の commit を backfill する経路）。watermark namespace は `diffs_backfill:${repo}`。 -1 run あたりの取り込み上限は forward / backward それぞれ 5 commits。`processAndUpsertCommitDiff` の upsert は `(repo, commit_sha, file_path)` で idempotent なので、webhook / 両 phase 間で overlap しても副作用はない。 +1 repo 1 run あたりの上限は、各 phase が 2 軸で持つ: commit 数 5 件と、file 数 `diffFileBudgetPerPhase`。`processAndUpsertCommitDiff` の upsert は `(repo, commit_sha, file_path)` で idempotent なので、webhook / 両 phase 間で overlap しても副作用はない。 -両 phase は同一の watermark 不変条件に従う: **取り込みに成功していない commit を watermark が追い越さない。** commit が「取り込み済み」とみなされるのは、detail 取得が成功し、かつ `processAndUpsertCommitDiff` の failed が 0 の場合のみ（embedding / Vectorize の失敗は throw ではなく戻り値で報告されるため、戻り値も判定に含める）。判定は dense 側のみを見る: D1 FTS mirror / store row の失敗は pipeline 側で log のみ・failed に計上しない設計で、vector は既に landing 済み・sparse index は reindex で reconcile されるため。mirror 側まで watermark の条件にすると、FTS 側の障害が diff surface 全体を止めてしまう。帰結: +commit 数は phase の消費量を縛れない。1 file の index は worst case で 3 subrequest（D1 FTS mirror 書き込み、store row、および embed batch の按分——1 batch は 2 subrequest を使い、`MAX_EMBEDDING_INPUT_CHARS` が 1 input を 8000 文字に切る一方 batch 予算は 30000 token なので、1 batch は最低 3 file を載せる）であり、1 commit は最大 300 file を運ぶため、5 commits のコストは 2 桁の幅を持つ。実測された帰結: `POLL_REPOS` の末尾 repo にある 44 file の commit で、embed batch 3 本すべてが `Too many subrequests by single Worker invocation` で拒否された。同じ loop の手前にある repo は正常に index されていた。そして後述の不変条件がその commit で watermark を止める——token 軸が起こしたのとまったく同じ、毎 cron 決定論的に再現する停止が、file 軸で起きた（issue #238）。 -- 失敗した commit と、1 run 上限で持ち越された commit は、次回 run の window に残る +そこで diff surface は invocation 予算のうち自分の取り分を明示する: `DIFF_SUBREQUEST_BUDGET_PER_RUN` = Cloudflare が許す 1000 のうち 900。diffs は専用 cron を持つので、この天井を docs / wiki / issue / release と共有はしていない。しかし `POLL_REPOS` の全 repo と両 phase では共有しており、上記の失敗を生んだのはそちらの共有である。`diffFileBudgetPerPhase` は宣言した取り分を `repoCount × 2` で割り、残りを file 数に換算する。literal を固定せず repo list から導出するのは意図的である: `POLL_REPOS` は通常の config commit で増える（issue #233 が 6 番目の repo を追加した）ため、その日の list に合わせた literal は次の追加で天井を超え、しかもその超過は原因となった変更ではなく loop が最後に到達した repo の失敗として現れる。repo list が長く予算を下回るところまで割った場合は 5 file の下限が効く——超過は次の cron が再試行するが、予算 0 は全 diff watermark を恒久的に止める。それはいま取り除こうとしている失敗そのものである。 + +file 予算は 1 commit を複数 run に分割しうる。間引きはしない: 1 run は残っている file の先頭側を index し、次の run は structured store（`GET /diffs`）から index 済み path を読み戻してそこから続ける。この読み戻しが分割を収束させる——これが無いと、上限付きの run が毎 cron 同じ先頭側を再 index し、その commit は永遠に完了しない。読み戻しを行うのは file 数が予算を超える commit だけなので、通常の commit はその判定に subrequest を払わない。またこの読み戻しが store 側エラーで失敗した場合は「まだ何も index されていない」と読むのではなく commit の失敗として扱う。前者は分割を黙って先頭からやり直させてしまう。 + +両 phase は同一の watermark 不変条件に従う: **取り込みに成功していない commit を watermark が追い越さない。** commit が「取り込み済み」とみなされるのは、detail 取得が成功し、かつ `processAndUpsertCommitDiff` の failed が 0 で、**さらに未処理の file が残っていない**場合のみ（embedding / Vectorize の失敗は throw ではなく戻り値で報告されるため、戻り値も判定に含める）。判定は dense 側のみを見る: D1 FTS mirror / store row の失敗は pipeline 側で log のみ・failed に計上しない設計で、vector は既に landing 済み・sparse index は reindex で reconcile されるため。mirror 側まで watermark の条件にすると、FTS 側の障害が diff surface 全体を止めてしまう。帰結: + +- 失敗した commit、file 予算で一部しか index されなかった commit、そして 2 つの 1 run 上限のいずれかで持ち越された commit は、いずれも次回 run の window に残る。一部だけ index された commit はエラーではなく分割の途中であり、watermark を止めることが残りの file への到達性を保つ - commit list 取得が失敗した run は watermark を動かさないので、その期間全体が再試行対象のまま残る - forward phase は window の古い端を確定してからでないと watermark を進められないため、window は 100 件/page で列挙する。page が満杯なら「列挙不能」とみなして window 終端を半分に縮める（最大 8 回）。それでも収まらない場合は watermark を保持して log に出す — silent な欠損より、観測できる stall を選ぶ。 diff --git a/docs/0-requirements.md b/docs/0-requirements.md index 9d7dcb4..fa994bc 100644 --- a/docs/0-requirements.md +++ b/docs/0-requirements.md @@ -193,11 +193,17 @@ 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}`. - **backward phase** — walks backward through history using `until=oldestUnprocessedDate`, backfilling commits that predate the webhook or a fresh deployment. Watermark namespace: `diffs_backfill:${repo}`. -Each phase ingests at most 5 commits per repo per run. Upserts through `processAndUpsertCommitDiff` are idempotent on `(repo, commit_sha, file_path)`, so overlap between webhook and either phase is safe. +Each phase is bounded on two axes per repo per run: at most 5 commits, and at most `diffFileBudgetPerPhase` files. Upserts through `processAndUpsertCommitDiff` are idempotent on `(repo, commit_sha, file_path)`, so overlap between webhook and either phase is safe. -Both phases obey one watermark invariant: **a watermark never moves past a commit that has not been successfully ingested.** A commit counts as ingested only when its detail fetch succeeded and `processAndUpsertCommitDiff` reported zero failed files — embedding and Vectorize failures are reported by return value rather than by throwing, so the return value is part of the check. The criterion is the dense side only: a D1 FTS mirror or store-row failure is logged by the pipeline and deliberately not counted, since the vector has already landed and the sparse index reconciles on reindex; gating the watermark on the mirror would let an FTS-side outage stall the whole diff surface. Consequences: +A commit count does not bound what a phase spends. Each indexed file costs 3 subrequests worst case — the D1 FTS mirror write, the store row, and the amortised share of its embed batch (a batch spends 2 and holds at least 3 files, since `MAX_EMBEDDING_INPUT_CHARS` caps an input at 8000 characters against a 30000-token batch budget) — and one commit carries up to 300 files, so 5 commits span two orders of magnitude of cost. The measured consequence: a 44-file commit in the last repo of `POLL_REPOS` had all three of its embed batches rejected with `Too many subrequests by single Worker invocation` while the repos ahead of it in the loop indexed normally, and the invariant below then held its watermark on that commit — the same deterministic, cron-after-cron stall the token axis produced, on the file axis (issue #238). -- a failed commit, and every commit the per-run cap deferred, stay inside the next run's window +The diff surface therefore declares its share of the invocation ceiling explicitly: `DIFF_SUBREQUEST_BUDGET_PER_RUN` = 900 of the 1000 Cloudflare allows. The diffs cron is the surface's own invocation, so that ceiling is not shared with docs / wiki / issues / releases — but it *is* shared across every repo in `POLL_REPOS` and both phases, which is the sharing that produced the failure above. `diffFileBudgetPerPhase` divides the declared share by `repoCount × 2` and converts the remainder to files. Deriving it from the repo list rather than fixing a literal is deliberate: `POLL_REPOS` grows by ordinary config commits (issue #233 appended the sixth repo), and a literal sized against the list of the day overruns the ceiling on the next append, surfacing as failures on whichever repo the loop reaches last rather than on the change that caused them. A floor of 5 files holds under a repo list long enough to divide the budget below it — an overrun is retried by the next cron, whereas a budget of zero stalls every diff watermark permanently, which is the failure being removed. + +The file budget can split one commit across runs. Nothing is thinned: a run indexes a prefix of the files still outstanding, and the next run reads the already-indexed paths back from the structured store (`GET /diffs`) and continues from there. That read is what makes the split converge — without it a bounded run re-indexes the same leading files every cron tick and the commit never finishes. It is spent only on a commit whose file count exceeds the budget, so an ordinary commit costs nothing to find out, and a store error on it counts the commit as failed rather than as "nothing indexed yet", which would silently restart it. + +Both phases obey one watermark invariant: **a watermark never moves past a commit that has not been successfully ingested.** A commit counts as ingested only when its detail fetch succeeded and `processAndUpsertCommitDiff` reported zero failed files *and* no files left outstanding — embedding and Vectorize failures are reported by return value rather than by throwing, so the return value is part of the check. The criterion is the dense side only: a D1 FTS mirror or store-row failure is logged by the pipeline and deliberately not counted, since the vector has already landed and the sparse index reconciles on reindex; gating the watermark on the mirror would let an FTS-side outage stall the whole diff surface. Consequences: + +- a failed commit, a commit the file budget only partly indexed, and every commit either per-run cap deferred, all stay inside the next run's window. A partly-indexed commit is not an error — it is mid-split, and holding the watermark is what keeps the rest of its files reachable - a failed commit list leaves the watermark untouched, so the whole period stays retryable - the forward phase must know the oldest end of its window before advancing, so the window is enumerated with a 100-entry page; a full page means "not enumerable" and the window end is halved (at most 8 times) until one page covers it. If even then the window overflows, the run holds the watermark and logs — a visible stall is preferred over a silent gap. diff --git a/docs/installation.ja.md b/docs/installation.ja.md index ef74626..c6335cc 100644 --- a/docs/installation.ja.md +++ b/docs/installation.ja.md @@ -212,7 +212,8 @@ POST /admin/diff-watermark?repo=owner/repo&since=2026-07-06T00:00:00Z 運用上の注意: -- 追いつき速度は 1 repo あたり 1 run 5 commits（毎時 `:30` なので約 120 commits/日）。数週間分の欠損は数日かかる +- 追いつき速度は 1 repo 1 run あたり 2 軸で決まる: commit 数 5 件と、`POLL_REPOS` の repo 数から導出される phase ごとの file 予算（6 repo なら各 18 file 前後）。毎時 `:30` なので、commit が小さい範囲では約 120 commits/日、file 数が多い commit では遅くなる——予算を超える commit は複数 run に分割され、最後の file が index されるまでその watermark は止まる。数週間分の欠損は数日かかる +- worker log の `{repo} diffs: ... [processed=.., partial=.., failed=.., deferred=..]` 行は 3 者を区別する: `partial` は分割途中の commit、`deferred` はその run が到達しなかった commit で、いずれもエラーではない - 再走査中も新規 commit は影響を受けない（webhook 経路が即時 index する） - upsert は `(repo, commit_sha, file_path)` で idempotent なので、index 済み期間を再走査しても安全 - 進捗は worker log の `{repo} diffs: forward [...]` 行、または該当期間を `type: "diff"` で検索して確認する diff --git a/docs/installation.md b/docs/installation.md index f22a513..5c01aa8 100644 --- a/docs/installation.md +++ b/docs/installation.md @@ -212,7 +212,8 @@ Authentication: Operational notes: -- catch-up rate is 5 commits per repo per cron run (~120 commits/day at the hourly `:30` trigger), so a multi-week gap takes several days to drain +- catch-up rate is bounded on two axes per repo per cron run: 5 commits, and a per-phase file budget derived from how many repos `POLL_REPOS` holds (~18 files each at six repos). At the hourly `:30` trigger that is ~120 commits/day where commits are small, and slower where they are file-heavy — a commit larger than the budget is split across runs and its watermark holds until the last of its files is indexed. A multi-week gap takes several days to drain +- the `{repo} diffs: ... [processed=.., partial=.., failed=.., deferred=..]` log line separates the three: `partial` is a commit mid-split, `deferred` is one this run never reached, and neither is an error - new commits are unaffected while a replay is in flight: the webhook path indexes them in real time - upserts are idempotent on `(repo, commit_sha, file_path)`, so re-covering an already-indexed period is safe - verify progress with the `{repo} diffs: forward [...]` line in the worker logs, or by searching `type: "diff"` for the period diff --git a/src/pipeline/embed-diff.test.ts b/src/pipeline/embed-diff.test.ts index ef9b7df..ba92b8f 100644 --- a/src/pipeline/embed-diff.test.ts +++ b/src/pipeline/embed-diff.test.ts @@ -74,7 +74,14 @@ describe("embed-diff: the batch axis is the token budget, not the file count", ( const result = await processAndUpsertCommitDiff(env, mkStore(), REPO, commit); - expect(result).toEqual({ embedded: 30, skipped: 0, failed: 0, batches: 1 }); + expect(result).toEqual({ + embedded: 30, + skipped: 0, + failed: 0, + batches: 1, + alreadyIndexed: 0, + deferred: 0, + }); expect(aiCalls).toHaveLength(1); expect(aiCalls[0]).toHaveLength(30); }); @@ -146,7 +153,123 @@ describe("embed-diff: the batch axis is the token budget, not the file count", ( const result = await processAndUpsertCommitDiff(env, mkStore(), REPO, commit); - expect(result).toEqual({ embedded: 1, skipped: 1, failed: 0, batches: 1 }); + expect(result).toEqual({ + embedded: 1, + skipped: 1, + failed: 0, + batches: 1, + alreadyIndexed: 0, + deferred: 0, + }); expect(aiCalls[0]).toHaveLength(1); }); }); + +describe("embed-diff: a file-heavy commit is split across runs, not thinned", () => { + it("indexes up to maxFiles and reports the rest as deferred", async () => { + const { env, aiCalls, upsertedIds } = mkEnv(); + const commit = mkCommit( + Array.from({ length: 44 }, (_, i) => `@@ -1 +1 @@\n+line ${i}`), + ); + + const result = await processAndUpsertCommitDiff(env, mkStore(), REPO, commit, { + maxFiles: 18, + }); + + expect(result.embedded).toBe(18); + expect(result.deferred).toBe(26); + expect(result.failed).toBe(0); + // The bound is on files indexed, not on files seen: nothing past it is touched. + expect(aiCalls.flat()).toHaveLength(18); + expect(upsertedIds.flat()).toHaveLength(18); + }); + + it("takes the leading files first so the split has a stable order", async () => { + const { env, aiCalls } = mkEnv(); + const commit = mkCommit(Array.from({ length: 5 }, (_, i) => `patch ${i}`)); + + await processAndUpsertCommitDiff(env, mkStore(), REPO, commit, { maxFiles: 2 }); + + // Input format is "{message}\n\n{path}\n\n{patch}". + const paths = aiCalls.flat().map((text) => text.split("\n\n")[1]); + expect(paths).toEqual(["src/file-0.ts", "src/file-1.ts"]); + }); + + it("resumes past the files a previous run already indexed", async () => { + const { env, aiCalls } = mkEnv(); + const commit = mkCommit(Array.from({ length: 5 }, (_, i) => `patch ${i}`)); + + const result = await processAndUpsertCommitDiff(env, mkStore(), REPO, commit, { + maxFiles: 2, + indexedFilePaths: new Set(["src/file-0.ts", "src/file-1.ts"]), + }); + + const paths = aiCalls.flat().map((text) => text.split("\n\n")[1]); + expect(paths).toEqual(["src/file-2.ts", "src/file-3.ts"]); + expect(result.alreadyIndexed).toBe(2); + expect(result.embedded).toBe(2); + expect(result.deferred).toBe(1); + }); + + it("closes the split — the last run reports nothing deferred", async () => { + // Repeated bounded calls, each fed the paths the previous ones landed, must + // reach every file exactly once. A resume set that failed to shrink the pending + // list would loop on the same prefix and never report deferred=0. + const { env } = mkEnv(); + const commit = mkCommit(Array.from({ length: 7 }, (_, i) => `patch ${i}`)); + const indexed = new Set(); + let runs = 0; + let last!: Awaited>; + + do { + if (runs++ > 10) throw new Error("split did not converge"); + last = await processAndUpsertCommitDiff(env, mkStore(), REPO, commit, { + maxFiles: 3, + indexedFilePaths: new Set(indexed), + }); + for (let i = 0; i < last.embedded; i++) { + indexed.add(`src/file-${indexed.size}.ts`); + } + } while (last.deferred > 0); + + expect(runs).toBe(3); + expect(indexed.size).toBe(7); + expect(last.deferred).toBe(0); + }); + + it("leaves the unbounded call untouched (the webhook path)", async () => { + const { env, aiCalls } = mkEnv(); + const commit = mkCommit(Array.from({ length: 12 }, (_, i) => `patch ${i}`)); + + const result = await processAndUpsertCommitDiff(env, mkStore(), REPO, commit); + + expect(result.embedded).toBe(12); + expect(result.deferred).toBe(0); + expect(result.alreadyIndexed).toBe(0); + expect(aiCalls.flat()).toHaveLength(12); + }); + + it("reports a fully-indexed commit as done rather than re-embedding it", async () => { + const { env, aiCalls } = mkEnv(); + const commit = mkCommit(Array.from({ length: 3 }, (_, i) => `patch ${i}`)); + + const result = await processAndUpsertCommitDiff(env, mkStore(), REPO, commit, { + maxFiles: 2, + indexedFilePaths: new Set([ + "src/file-0.ts", + "src/file-1.ts", + "src/file-2.ts", + ]), + }); + + expect(result).toEqual({ + embedded: 0, + skipped: 0, + failed: 0, + batches: 0, + alreadyIndexed: 3, + deferred: 0, + }); + expect(aiCalls).toHaveLength(0); + }); +}); diff --git a/src/pipeline/embed-diff.ts b/src/pipeline/embed-diff.ts index 0b3037a..1651b49 100644 --- a/src/pipeline/embed-diff.ts +++ b/src/pipeline/embed-diff.ts @@ -44,6 +44,25 @@ export interface DiffUpsertResult { failed: number; /** Number of Workers AI batch calls issued (for observability) */ batches: number; + /** Files the caller's `indexedFilePaths` set already listed, so this call left + * them alone. Zero when the caller passed no resume set. */ + alreadyIndexed: number; + /** Files with a patch, not in the resume set, that `maxFiles` kept this call + * from attempting. Non-zero means the commit is *partially* indexed and the + * caller must not treat it as done. Zero when the caller set no bound. */ + deferred: number; +} + +/** Bounds a single `processAndUpsertCommitDiff` call. Both fields absent = index + * every file with a patch, which is what the webhook path wants: it handles one + * push per invocation and has the whole subrequest budget to itself. */ +export interface DiffUpsertOptions { + /** Upper bound on files this call indexes. Files past it are reported as + * `deferred` rather than dropped — the caller resumes them on a later run. */ + maxFiles?: number; + /** File paths already on the index for this commit. Passing them is what makes + * a bounded call *resume* instead of re-indexing the same prefix forever. */ + indexedFilePaths?: ReadonlySet; } /** @@ -103,19 +122,28 @@ function normaliseFileStatus(status: string): DiffFileStatus { * * Flow: * 1. Filter `files[]` to those with a textual `patch` (binary / oversized files are skipped). - * 2. Build embedding inputs = commit message + file path + patch, truncated. - * 3. Batch-embed inputs via Workers AI (chunked by `planEmbeddingBatches`, which + * 2. Drop files the caller's `indexedFilePaths` resume set already lists, then + * take at most `options.maxFiles` of what remains. + * 3. Build embedding inputs = commit message + file path + patch, truncated. + * 4. Batch-embed inputs via Workers AI (chunked by `planEmbeddingBatches`, which * splits on an estimated token budget rather than a file count). - * 4. Upsert all vectors into Vectorize in the same chunks. - * 5. Record DiffRecord rows into the Durable Object store for each indexed file. + * 5. Upsert all vectors into Vectorize in the same chunks. + * 6. Record DiffRecord rows into the Durable Object store for each indexed file. * * Failures inside a chunk do not halt subsequent chunks — counts are tallied and * returned so the caller can log/escalate without losing partial progress. * + * `maxFiles` splits a commit across runs; it never thins one. Every file with a + * patch is either indexed by this call or counted in `deferred` for the caller to + * resume, and the two step 2 filters are what make repeated calls converge: + * without the resume set a bounded call re-indexes the same prefix every run and + * the commit never finishes (issue #238). + * * @param env - Worker env bindings (AI, VECTORIZE) * @param storeStub - Durable Object stub for IssueStore * @param repo - Repository in "owner/repo" format * @param commit - Commit detail from GitHub (from GET /repos/{repo}/commits/{sha}) + * @param options - Per-call file bound and resume set; unbounded when omitted * @returns Summary of embeddings/upserts produced */ export async function processAndUpsertCommitDiff( @@ -123,6 +151,7 @@ export async function processAndUpsertCommitDiff( storeStub: DurableObjectStub, repo: string, commit: GitHubCommitDetail, + options: DiffUpsertOptions = {}, ): Promise { const commitSha = commit.sha; const commitMessage = commit.commit.message ?? ""; @@ -137,14 +166,26 @@ export async function processAndUpsertCommitDiff( // Keep only files with a textual patch. Binary blobs, submodule changes, and // oversized diffs arrive without a patch field and cannot be embedded. - const indexable = files.filter( + const withPatch = files.filter( (f): f is typeof f & { patch: string } => typeof f.patch === "string" && f.patch.length > 0, ); - const skipped = files.length - indexable.length; + const skipped = files.length - withPatch.length; + + // Resume: an earlier run of a split commit already landed these. + const indexedPaths = options.indexedFilePaths; + const pending = indexedPaths + ? withPatch.filter((f) => !indexedPaths.has(f.filename)) + : withPatch; + const alreadyIndexed = withPatch.length - pending.length; + + // Bound what this call attempts. The remainder is reported, not dropped. + const limit = Math.max(0, options.maxFiles ?? pending.length); + const indexable = pending.slice(0, limit); + const deferred = pending.length - indexable.length; if (indexable.length === 0) { - return { embedded: 0, skipped, failed: 0, batches: 0 }; + return { embedded: 0, skipped, failed: 0, batches: 0, alreadyIndexed, deferred }; } let embedded = 0; @@ -311,5 +352,5 @@ export async function processAndUpsertCommitDiff( embedded += chunk.length; } - return { embedded, skipped, failed, batches }; + return { embedded, skipped, failed, batches, alreadyIndexed, deferred }; } diff --git a/src/poller.test.ts b/src/poller.test.ts index ca7322b..dd717c0 100644 --- a/src/poller.test.ts +++ b/src/poller.test.ts @@ -36,6 +36,8 @@ const { nextForwardDiffWatermark, nextBackfillDiffWatermark, nextIssueWatermark, + diffFileBudgetPerPhase, + parsePollRepos, MAX_EMBEDDINGS_PER_RUN, MAX_RELEASE_UPSERTS_PER_REPO_PER_RUN, MAX_DOC_FETCHES_PER_REPO_PER_RUN, @@ -68,9 +70,21 @@ function makeStore( const upserts: Array<{ number: number; bodyHash: string }> = []; /** Records written through `/upsert-release` (same retry markers, release side). */ const releaseUpserts: Array<{ tagName: string; bodyHash: string }> = []; + /** Indexed diff file paths per `${repo}@${sha}` — what `GET /diffs` reports back, + * i.e. the resume set a split commit is continued from. */ + const diffPaths = new Map>(); + /** Commit SHAs the poller asked the resume set for. */ + const diffListCalls: string[] = []; const stub = { async fetch(request: Request): Promise { const url = new URL(request.url); + if (request.method === "GET" && url.pathname === "/diffs") { + const repo = url.searchParams.get("repo") ?? ""; + const sha = url.searchParams.get("commit_sha") ?? ""; + diffListCalls.push(sha); + const paths = diffPaths.get(`${repo}@${sha}`) ?? new Set(); + return Response.json([...paths].map((filePath) => ({ filePath }))); + } if (request.method === "GET" && url.pathname === "/watermark") { const key = url.searchParams.get("repo") ?? ""; const value = watermarks.get(key); @@ -109,6 +123,8 @@ function makeStore( etags, upserts, releaseUpserts, + diffPaths, + diffListCalls, }; } @@ -301,6 +317,8 @@ beforeEach(() => { skipped: 0, failed: 0, batches: 1, + alreadyIndexed: 0, + deferred: 0, }); }); @@ -447,8 +465,8 @@ describe("poller: pollDiffs forward watermark / retry boundary", () => { processAndUpsertCommitDiffMock.mockImplementation( async (_env: unknown, _stub: unknown, _repo: string, detail: { sha: string }) => detail.sha === "c1" - ? { embedded: 1, skipped: 0, failed: 0, batches: 1 } - : { embedded: 0, skipped: 0, failed: 3, batches: 1 }, + ? { embedded: 1, skipped: 0, failed: 0, batches: 1, alreadyIndexed: 0, deferred: 0 } + : { embedded: 0, skipped: 0, failed: 3, batches: 1, alreadyIndexed: 0, deferred: 0 }, ); await pollDiffs(REPO, env, stub); @@ -515,6 +533,251 @@ describe("poller: pollDiffs forward watermark / retry boundary", () => { }); }); +describe("poller: diffFileBudgetPerPhase", () => { + /** Subrequests the surface would spend in the worst case at a given repo count, + * reproducing the arithmetic the constants document: every phase of every repo + * fills its file budget, and each file costs 3 plus the phase's fixed overhead. */ + const worstCaseSubrequests = (repoCount: number): number => + repoCount * 2 * (diffFileBudgetPerPhase(repoCount) * 3 + 20); + + it("keeps the whole invocation inside the 1000-subrequest ceiling", () => { + // The production list is 6. The ceiling holds up to 14 repos; past that the + // floor below wins on purpose and the overrun is preferred to a stall. + for (const repoCount of [1, 2, 3, 5, 6, 8, 10, 12, 14]) { + expect(worstCaseSubrequests(repoCount)).toBeLessThanOrEqual(1000); + } + }); + + it("shrinks the per-phase budget as repos are appended", () => { + // The defect this replaces: a literal cap sized against the repo list of the + // day, overrun silently by the next config commit that appends a repo. + expect(diffFileBudgetPerPhase(1)).toBeGreaterThan(diffFileBudgetPerPhase(6)); + expect(diffFileBudgetPerPhase(6)).toBeGreaterThan(diffFileBudgetPerPhase(12)); + }); + + it("clears the 44-file commit that holds the production watermark", () => { + // neuron-graph-rag@92eb94d, 44 files, at the production repo count. One run + // must not have to clear it in one pass — but it must make progress, so the + // budget has to be a usable fraction of it rather than a token few files. + expect(diffFileBudgetPerPhase(6)).toBeGreaterThanOrEqual(10); + expect(diffFileBudgetPerPhase(6)).toBeLessThan(44); + }); + + it("never returns a budget of zero, however long the repo list", () => { + // Zero would stall every diff watermark permanently — the failure shape the + // budget exists to remove. Overrunning the subrequest ceiling is recoverable + // on the next cron; a permanent stall is not. + for (const repoCount of [40, 400, 4000]) { + expect(diffFileBudgetPerPhase(repoCount)).toBeGreaterThanOrEqual(5); + } + expect(diffFileBudgetPerPhase(0)).toBeGreaterThanOrEqual(5); + }); +}); + +describe("poller: parsePollRepos", () => { + it("reads the list the cron dispatch walks", () => { + expect(parsePollRepos({ POLL_REPOS: " a/b , c/d ,, " } as unknown as Env)).toEqual([ + "a/b", + "c/d", + ]); + expect(parsePollRepos({} as unknown as Env)).toEqual([]); + }); +}); + +describe("poller: pollDiffs per-run file budget", () => { + /** Repo list long enough to drive the per-phase budget to its floor of 5, so the + * split is exercised with a commit of a size the tests can spell out. */ + const budgetEnv = { + GITHUB_TOKEN: "test-token", + POLL_REPOS: Array.from({ length: 40 }, (_, i) => `acme/r${i}`).join(","), + } as unknown as Env; + + const FILE_BUDGET = 5; + + /** + * Wire the pipeline mocks so a named commit carries `fileCount` files and the + * pipeline honours the bound and the resume set exactly as the real one does. + * + * The split arithmetic is reproduced rather than stubbed to a constant: what is + * under test is that the poller's budget, its resume query and the watermark + * invariant compose into a commit that finishes, and a stub that always claimed + * progress would pass whether or not they do. + */ + function stubSplitPipeline( + fileCounts: Record, + diffPaths: Map>, + ): void { + fetchCommitDetailMock.mockImplementation(async (_repo: string, sha: string) => ({ + sha, + commit: { message: "m" }, + files: Array.from({ length: fileCounts[sha] ?? 0 }, (_, i) => ({ + filename: `src/${sha}-f${i}.ts`, + status: "modified", + patch: "p", + })), + })); + processAndUpsertCommitDiffMock.mockImplementation( + async ( + _env: unknown, + _stub: unknown, + repo: string, + detail: { sha: string; files?: Array<{ filename: string }> }, + options: { maxFiles?: number; indexedFilePaths?: ReadonlySet } = {}, + ) => { + const all = (detail.files ?? []).map((f) => f.filename); + const pending = all.filter((p) => !options.indexedFilePaths?.has(p)); + const taken = pending.slice(0, options.maxFiles ?? pending.length); + const key = `${repo}@${detail.sha}`; + const landed = diffPaths.get(key) ?? new Set(); + for (const p of taken) landed.add(p); + diffPaths.set(key, landed); + return { + embedded: taken.length, + skipped: 0, + failed: 0, + batches: taken.length > 0 ? 1 : 0, + alreadyIndexed: all.length - pending.length, + deferred: pending.length - taken.length, + }; + }, + ); + } + + it("bounds one commit's files per run and finishes it over successive runs", async () => { + const heavy = "big"; + const { stub, watermarks, diffPaths, diffListCalls } = makeStore({ + [FORWARD_KEY]: "2026-07-01T00:00:00.000Z", + [BACKFILL_KEY]: "2026-06-01T00:00:00.000Z", + }); + stubCommitList([commit(heavy, "2026-07-01T01:00:00.000Z")]); + // 12 files against a per-phase budget of 5 — three runs to drain. + stubSplitPipeline({ [heavy]: 12 }, diffPaths); + + const indexedAfterEachRun: number[] = []; + for (let run = 0; run < 3; run++) { + await pollDiffs(REPO, budgetEnv, stub); + indexedAfterEachRun.push(diffPaths.get(`${REPO}@${heavy}`)!.size); + } + + // Monotonic progress, never more than the budget in one run. + expect(indexedAfterEachRun).toEqual([FILE_BUDGET, FILE_BUDGET * 2, 12]); + // The resume set is read for the oversized commit, which is what makes each + // run continue rather than re-index the same leading files. + expect(diffListCalls).toContain(heavy); + // The watermark stayed put while the commit was mid-split, and moves only once + // its last file is indexed. + expect(Date.parse(watermarks.get(FORWARD_KEY)!)).toBeGreaterThan( + Date.parse("2026-07-01T01:00:00.000Z"), + ); + }); + + it("holds the watermark on the run that only partly indexed the commit", async () => { + const heavy = "big"; + const start = "2026-07-01T00:00:00.000Z"; + const { stub, watermarks, diffPaths } = makeStore({ + [FORWARD_KEY]: start, + [BACKFILL_KEY]: "2026-06-01T00:00:00.000Z", + }); + stubCommitList([commit(heavy, "2026-07-01T01:00:00.000Z")]); + stubSplitPipeline({ [heavy]: 12 }, diffPaths); + + await pollDiffs(REPO, budgetEnv, stub); + + // Parked before the commit, exactly as for a failed one — a partial commit is + // not ingested, so nothing may pass it. + const wm = watermarks.get(FORWARD_KEY)!; + expect(Date.parse(wm)).toBeLessThan(Date.parse("2026-07-01T01:00:00.000Z")); + expect(Date.parse(wm)).toBeGreaterThanOrEqual(Date.parse(start)); + }); + + it("defers the commits after the one that spent the budget", async () => { + const heavy = "big"; + const { stub, diffPaths } = makeStore({ + [FORWARD_KEY]: "2026-07-01T00:00:00.000Z", + [BACKFILL_KEY]: "2026-06-01T00:00:00.000Z", + }); + stubCommitList([ + commit(heavy, "2026-07-01T01:00:00.000Z"), + commit("after1", "2026-07-01T02:00:00.000Z"), + commit("after2", "2026-07-01T03:00:00.000Z"), + ]); + stubSplitPipeline({ [heavy]: 12 }, diffPaths); + + await pollDiffs(REPO, budgetEnv, stub); + + // The oldest commit alone spent the phase's budget, so the run stops there + // instead of fanning out past the invocation's subrequest ceiling. + expect(attemptedShas()).toEqual([heavy]); + }); + + it("spends no resume query on a commit that fits the budget", async () => { + const { stub, diffPaths, diffListCalls } = makeStore({ + [FORWARD_KEY]: "2026-07-01T00:00:00.000Z", + [BACKFILL_KEY]: "2026-06-01T00:00:00.000Z", + }); + stubCommitList([commit("small", "2026-07-01T01:00:00.000Z")]); + stubSplitPipeline({ small: 3 }, diffPaths); + + await pollDiffs(REPO, budgetEnv, stub); + + expect(diffListCalls).toEqual([]); + expect(diffPaths.get(`${REPO}@small`)!.size).toBe(3); + }); + + it("gives the backfill phase its own budget rather than the forward phase's leftovers", async () => { + // A repo whose forward window is busy would otherwise starve its historical + // sweep for good, since the forward phase runs first. + const { stub, diffPaths } = makeStore({ + // Forward window is (00:45, now] — `old` sits below it, so only the backfill + // phase can reach it. Backfill walks `until=00:30`. + [FORWARD_KEY]: "2026-07-01T00:45:00.000Z", + [BACKFILL_KEY]: "2026-07-01T00:30:00.000Z", + }); + stubCommitList([ + commit("fwdheavy", "2026-07-01T01:00:00.000Z"), + commit("old", "2026-07-01T00:10:00.000Z"), + ]); + stubSplitPipeline({ fwdheavy: 12, old: 4 }, diffPaths); + + await pollDiffs(REPO, budgetEnv, stub); + + // Forward spent its whole budget on `fwdheavy` and still had nothing left, yet + // backfill indexed `old` in full — the two budgets are separate. + expect(diffPaths.get(`${REPO}@fwdheavy`)!.size).toBe(FILE_BUDGET); + expect(diffPaths.get(`${REPO}@old`)!.size).toBe(4); + }); + + it("treats a failed resume query as an unfinished commit", async () => { + const heavy = "big"; + const start = "2026-07-01T00:00:00.000Z"; + const base = makeStore({ + [FORWARD_KEY]: start, + [BACKFILL_KEY]: "2026-06-01T00:00:00.000Z", + }); + // The whole store surface, except `/diffs` is broken. An empty resume set would + // read as "nothing indexed yet" and silently restart the commit every run, so + // the query failing has to count as the commit failing. + const stub = { + async fetch(request: Request): Promise { + const url = new URL(request.url); + if (url.pathname === "/diffs") return new Response("boom", { status: 500 }); + return base.stub.fetch(request); + }, + } as unknown as DurableObjectStub; + stubCommitList([commit(heavy, "2026-07-01T01:00:00.000Z")]); + stubSplitPipeline({ [heavy]: 12 }, base.diffPaths); + + await pollDiffs(REPO, budgetEnv, stub); + + // Nothing was indexed, and the watermark is parked below the commit so the + // next run re-covers it. + expect(base.diffPaths.has(`${REPO}@${heavy}`)).toBe(false); + const wm = base.watermarks.get(FORWARD_KEY)!; + expect(Date.parse(wm)).toBeLessThan(Date.parse("2026-07-01T01:00:00.000Z")); + expect(Date.parse(wm)).toBeGreaterThanOrEqual(Date.parse(start)); + }); +}); + describe("poller: nextIssueWatermark", () => { const since = "2026-07-01T00:00:00.000Z"; const candidate = "2026-07-01T12:00:00.000Z"; diff --git a/src/poller.ts b/src/poller.ts index 7bb226b..f803e38 100644 --- a/src/poller.ts +++ b/src/poller.ts @@ -170,17 +170,86 @@ export const MAX_RELEASE_UPSERTS_PER_REPO_PER_RUN = 10; * of the diff poller per repo per run. * Forward is normally a no-op because the webhook path already indexes new * commits; this cap bounds the work when webhook delivery has stalled. - * Sized so that (forward + backward) × per-commit fan-out stays well under the - * Cloudflare Workers per-invocation subrequest limit (issue #124). */ + * + * A commit count is not what bounds the phase's subrequest spend — a commit + * carries up to 300 files and each costs several subrequests, so 5 commits span + * two orders of magnitude of cost (issue #238). `diffFileBudgetPerPhase` holds + * that axis. This cap holds the GitHub-side one: detail fetches, and how far a + * single run may walk the window (issue #124). */ const MAX_DIFF_COMMITS_FORWARD_PER_RUN = 5; /** Maximum number of commits fetched in the backward (historical backfill) phase * of the diff poller per repo per run. * Backfill walks backward through repo history one hourly run at a time; the - * cap keeps per-run API and embedding cost bounded so the total sweep spreads - * over many runs (e.g. 5 commits/run × 24 runs/day = 120 commits/day per repo). */ + * cap keeps per-run API cost bounded so the total sweep spreads over many runs + * (e.g. 5 commits/run × 24 runs/day = 120 commits/day per repo). Doubles as the + * listing's `per_page`. The embedding and upsert cost of those commits is bounded + * on the file axis by `diffFileBudgetPerPhase`, not here. */ const MAX_DIFF_COMMITS_BACKWARD_PER_RUN = 5; +/** Subrequests the commit-diff surface may spend in one `DIFFS_CRON` invocation. + * + * Cloudflare allows 1000 per invocation. Diffs get a cron of their own, so the + * surface is not sharing that ceiling with docs / wiki / issues / releases — but + * it *is* sharing it across every repo in `POLL_REPOS` and both phases, which is + * the sharing that actually bit: a 44-file commit in the last repo of the list + * had all three of its embed batches rejected with `Too many subrequests by + * single Worker invocation` while the repos ahead of it in the loop indexed + * normally (issue #238). 900 of the 1000 leaves the margin for the dispatch + * itself and for the estimates below coming in low. + * + * This constant is the diff path's declared share. `diffFileBudgetPerPhase` + * turns it into the per-phase file cap; nothing else reads it. */ +const DIFF_SUBREQUEST_BUDGET_PER_RUN = 900; + +/** Subrequests one indexed file costs, worst case. + * + * Two are fixed and per-file: the D1 FTS mirror write and the Store DO row. + * The third is the amortised batch cost — a batch spends 2 (the Workers AI call + * and its `VECTORIZE.upsert`) and holds at least 3 files, because + * `MAX_EMBEDDING_INPUT_CHARS` caps one input at 8000 characters and + * `MAX_EMBEDDING_BATCH_TOKENS` gives a batch 30000 tokens, so even CJK prose at + * ~1 token per character fits 3. That puts the true figure at 2.67 and under; 3 + * is it rounded to the safe side. */ +const DIFF_SUBREQUESTS_PER_FILE = 3; + +/** Per-phase subrequests that are not per-file: up to 5 commit detail fetches, + * up to 5 resume queries for oversized commits, and the commit listing — 1 call + * for backfill, up to 9 for a forward window that needs shrinking. */ +const DIFF_PHASE_OVERHEAD_SUBREQUESTS = 20; + +/** Floor under the per-phase file budget. + * + * A `POLL_REPOS` long enough to divide the budget below this would otherwise + * drive the cap to zero, which stalls every diff watermark permanently — the + * exact failure shape this issue exists to remove. Holding the floor overruns + * the subrequest budget instead, and an overrun is a transient failure the next + * cron retries. Choose the recoverable side. */ +const MIN_DIFF_FILES_PER_PHASE_PER_RUN = 5; + +/** + * Files one diff phase may index for one repo in one cron run. + * + * Derived rather than hard-coded because the divisor is `POLL_REPOS`, which grows + * by ordinary config commits (issue #233 added the sixth repo). A literal cap + * sized against the repo list of the day silently overruns the invocation budget + * the next time a repo is appended, and the overrun surfaces as failures on + * whichever repo the loop reaches last — not on the change that caused them. + * + * @param repoCount number of repos this invocation will walk (>=1) + */ +export function diffFileBudgetPerPhase(repoCount: number): number { + // Two phases per repo, each spending from the same invocation budget. + const phases = Math.max(1, repoCount) * 2; + const perPhase = + Math.floor(DIFF_SUBREQUEST_BUDGET_PER_RUN / phases) - + DIFF_PHASE_OVERHEAD_SUBREQUESTS; + return Math.max( + MIN_DIFF_FILES_PER_PHASE_PER_RUN, + Math.floor(perPhase / DIFF_SUBREQUESTS_PER_FILE), + ); +} + /** Page size used when the forward diff phase enumerates its commit window. * Listing is 1 subrequest regardless of page size, while the per-commit * fan-out (detail fetch + embed + Vectorize + D1 + Store DO) is what the @@ -2379,6 +2448,42 @@ async function writeWatermark( ); } +/** + * Read the file paths already indexed for one commit, from the structured store. + * + * This is the resume set for a commit the phase's file budget has to split across + * runs. Without it a bounded run re-attempts the same prefix every cron tick and + * the commit never completes, so the watermark it holds never moves — the same + * permanent stall the budget was added to prevent, reached from the other side. + * + * Throws rather than returning an empty set on a store error: an empty set reads + * as "nothing indexed yet" and would silently restart the commit from its first + * file. The caller counts the throw as a failed commit and retries next run. + */ +async function listIndexedDiffPaths( + storeStub: DurableObjectStub, + repo: string, + commitSha: string, +): Promise> { + const resp = await storeStub.fetch( + new Request( + `http://store/diffs?repo=${encodeURIComponent(repo)}` + + `&commit_sha=${encodeURIComponent(commitSha)}`, + ), + ); + if (!resp.ok) { + throw new Error( + `store /diffs returned ${resp.status} for ${repo}@${commitSha}`, + ); + } + const rows = (await resp.json()) as Array<{ filePath?: string }>; + const paths = new Set(); + for (const row of rows) { + if (row.filePath) paths.add(row.filePath); + } + return paths; +} + /** Extract the best-available ISO timestamp from a commit summary. */ function commitDateOf(summary: GitHubCommitSummary): string | undefined { return ( @@ -2394,10 +2499,17 @@ function commitDateOf(summary: GitHubCommitSummary): string | undefined { * - `ok` — the commit's every indexable file landed in Vectorize. * - `failed` — the detail fetch, an embedding batch, or a Vectorize upsert * failed for at least one file; the commit must stay retryable. + * - `partial` — every file this run attempted landed, but the phase's file budget + * left some of the commit's files unattempted. Not an error: the + * commit is mid-split and the rest lands on a later run. * - `deferred` — the commit was inside the enumerated window but was not - * attempted this run (per-run commit cap). + * attempted this run (per-run commit cap, or file budget spent). + * + * Only `ok` lets a watermark past. `partial` sits with `failed` and `deferred` on + * that axis, which is what keeps a split commit inside the next run's window until + * its last file is indexed. */ -export type DiffCommitStatus = "ok" | "failed" | "deferred"; +export type DiffCommitStatus = "ok" | "failed" | "partial" | "deferred"; /** One commit's outcome, in the order the phase walked its commits. */ export interface DiffCommitOutcome { @@ -2405,6 +2517,9 @@ export interface DiffCommitOutcome { /** Commit timestamp; absent when GitHub returned neither author nor committer date. */ date?: string; status: DiffCommitStatus; + /** Files this run attempted for the commit — what it spent from the phase's file + * budget. Absent for a commit the phase never attempted. */ + filesAttempted?: number; } /** @@ -2417,6 +2532,10 @@ export interface DiffCommitOutcome { * `since` window and was lost permanently (the backward phase only walks into * older history and never returns to that period). * + * The test is `status !== "ok"`, so a commit the file budget only partly indexed + * bounds the watermark on the same footing as a failed one. That is what lets the + * budget split a commit without the split losing files (issue #238). + * * @param since watermark the phase started from (lower bound, never regressed) * @param windowEnd upper bound of the window that was fully enumerated * @param outcomes every commit in the window, oldest-first @@ -2455,7 +2574,8 @@ export function nextForwardDiffWatermark( * Mirror image of the forward invariant: **the backfill watermark never moves * past the newest commit that has not been successfully ingested.** The phase * walks newest-first, so the watermark may only advance across the contiguous - * successful prefix; the first failure freezes it, keeping that commit inside + * successful prefix; the first commit that is not `ok` — failed, partially + * indexed, or deferred — freezes it, keeping that commit inside * the next run's `until` window. Commits already ingested after the freeze * point are re-ingested next run, which is bounded by the per-run cap and * idempotent on (repo, commit_sha, file_path). @@ -2549,32 +2669,63 @@ async function enumerateForwardWindow( * deliberately not counted, because the Vectorize upsert has already landed and * the sparse index reconciles on reindex. Gating the watermark on the sparse * mirror would let an FTS-side outage stall the whole diff surface. + * + * `fileBudget` is what the phase has left to spend. A commit carrying more files + * than that is indexed up to the bound and reported `partial`; the resume set read + * here is what makes the next run continue rather than restart it. */ async function ingestCommitDiff( repo: string, env: Env, storeStub: DurableObjectStub, summary: GitHubCommitSummary, + fileBudget: number, ): Promise { const date = commitDateOf(summary); try { const detail = await fetchCommitDetail(repo, summary.sha, env.GITHUB_TOKEN); - const result = await processAndUpsertCommitDiff(env, storeStub, repo, detail); + + // Only a commit that cannot fit the budget needs the resume set, and the test + // uses the raw file count — an upper bound on the indexable ones — so a commit + // that fits spends no subrequest finding that out. The count is a property of + // the commit, so the test answers the same on the run that finishes a split + // commit as on the run that started it. + const fileCount = detail.files?.length ?? 0; + const indexedFilePaths = + fileCount > fileBudget + ? await listIndexedDiffPaths(storeStub, repo, summary.sha) + : undefined; + + const result = await processAndUpsertCommitDiff(env, storeStub, repo, detail, { + maxFiles: fileBudget, + indexedFilePaths, + }); + // Failed files spent their embed call, so charge them to the budget too. + const filesAttempted = result.embedded + result.failed; + if (result.failed > 0) { console.error( `pollDiffs: ${repo}@${summary.sha} partially failed ` + `(embedded=${result.embedded}, failed=${result.failed}) — ` + `commit stays inside the retry window`, ); - return { sha: summary.sha, date, status: "failed" }; + return { sha: summary.sha, date, status: "failed", filesAttempted }; + } + if (result.deferred > 0) { + console.log( + `pollDiffs: ${repo}@${summary.sha} split across runs ` + + `(indexed=${result.embedded}, already=${result.alreadyIndexed}, ` + + `remaining=${result.deferred}) — watermark held until the commit completes`, + ); + return { sha: summary.sha, date, status: "partial", filesAttempted }; } - return { sha: summary.sha, date, status: "ok" }; + return { sha: summary.sha, date, status: "ok", filesAttempted }; } catch (err) { console.error( `pollDiffs: commit ${repo}@${summary.sha} failed:`, err instanceof Error ? err.message : String(err), ); - return { sha: summary.sha, date, status: "failed" }; + return { sha: summary.sha, date, status: "failed", filesAttempted: 0 }; } } @@ -2583,24 +2734,36 @@ interface DiffPhaseStats { processed: number; failed: number; deferred: number; + /** Commits indexed up to the file budget with files still outstanding. Separate + * from `failed` because nothing went wrong: the next run continues them. */ + partial: number; /** Watermark did not advance this run (list failure, or an unprocessed commit at the boundary). */ held: boolean; } +/** Tally one commit outcome into a phase's counters. */ +function tallyDiffOutcome(stats: DiffPhaseStats, status: DiffCommitStatus): void { + if (status === "ok") stats.processed++; + else if (status === "partial") stats.partial++; + else if (status === "deferred") stats.deferred++; + else stats.failed++; +} + /** * Forward phase — webhook redundancy plus gap recovery. * * Enumerates `(watermark, pollStartTime]`, processes its **oldest** - * MAX_DIFF_COMMITS_FORWARD_PER_RUN commits, and advances the watermark only - * across the contiguous successfully-ingested prefix. Anything failed or - * deferred stays inside the next run's window, so a burst larger than the - * per-run cap drains over successive runs instead of being skipped. + * MAX_DIFF_COMMITS_FORWARD_PER_RUN commits within `fileBudget` files, and advances + * the watermark only across the contiguous successfully-ingested prefix. Anything + * failed, partial or deferred stays inside the next run's window, so a burst larger + * than either cap drains over successive runs instead of being skipped. */ async function runForwardDiffPhase( repo: string, env: Env, storeStub: DurableObjectStub, pollStartTime: string, + fileBudget: number, ): Promise { const fwdKey = `diffs:${repo}`; const fwdWm = await readWatermark(storeStub, fwdKey); @@ -2610,7 +2773,13 @@ async function runForwardDiffPhase( fwdWm?.lastPolledAt ?? new Date(Date.now() - 60 * 60 * 1000).toISOString(); - const stats: DiffPhaseStats = { processed: 0, failed: 0, deferred: 0, held: false }; + const stats: DiffPhaseStats = { + processed: 0, + failed: 0, + deferred: 0, + partial: 0, + held: false, + }; if (Date.parse(since) >= Date.parse(pollStartTime)) { console.warn( @@ -2653,8 +2822,11 @@ async function runForwardDiffPhase( // over a prefix that is contiguous in commit-date order. const ordered = [...window.commits].reverse(); const outcomes: DiffCommitOutcome[] = []; + let filesLeft = fileBudget; for (const summary of ordered) { - if (outcomes.length >= MAX_DIFF_COMMITS_FORWARD_PER_RUN) { + // Two independent caps, either one deferring the rest of the window: how many + // commits this run may walk, and how many files it may index. + if (outcomes.length >= MAX_DIFF_COMMITS_FORWARD_PER_RUN || filesLeft <= 0) { outcomes.push({ sha: summary.sha, date: commitDateOf(summary), @@ -2663,10 +2835,16 @@ async function runForwardDiffPhase( stats.deferred++; continue; } - const outcome = await ingestCommitDiff(repo, env, storeStub, summary); + const outcome = await ingestCommitDiff( + repo, + env, + storeStub, + summary, + filesLeft, + ); + filesLeft -= outcome.filesAttempted ?? 0; outcomes.push(outcome); - if (outcome.status === "ok") stats.processed++; - else stats.failed++; + tallyDiffOutcome(stats, outcome.status); } const next = nextForwardDiffWatermark(since, window.windowEnd, outcomes); @@ -2688,21 +2866,32 @@ async function runForwardDiffPhase( * Backward phase — historical backfill. * * Walks `until=watermark` into older history. The watermark advances only - * across the contiguous successful prefix (newest-first), so a failed commit - * stays inside the next run's window instead of being stepped over. + * across the contiguous successful prefix (newest-first), so a failed, partial or + * deferred commit stays inside the next run's window instead of being stepped over. + * + * Gets its own `fileBudget` rather than sharing one with the forward phase, which + * runs first: a repo whose forward window is busy would otherwise spend the whole + * budget every run and the historical sweep would never advance again. */ async function runBackfillDiffPhase( repo: string, env: Env, storeStub: DurableObjectStub, pollStartTime: string, + fileBudget: number, ): Promise { const bwdKey = `diffs_backfill:${repo}`; const bwdWm = await readWatermark(storeStub, bwdKey); // First run: start walking backward from the current time. const until = bwdWm?.lastPolledAt ?? pollStartTime; - const stats: DiffPhaseStats = { processed: 0, failed: 0, deferred: 0, held: false }; + const stats: DiffPhaseStats = { + processed: 0, + failed: 0, + deferred: 0, + partial: 0, + held: false, + }; let commits: GitHubCommitSummary[]; try { @@ -2720,11 +2909,27 @@ async function runBackfillDiffPhase( } const outcomes: DiffCommitOutcome[] = []; + let filesLeft = fileBudget; for (const summary of commits) { - const outcome = await ingestCommitDiff(repo, env, storeStub, summary); + if (filesLeft <= 0) { + outcomes.push({ + sha: summary.sha, + date: commitDateOf(summary), + status: "deferred", + }); + stats.deferred++; + continue; + } + const outcome = await ingestCommitDiff( + repo, + env, + storeStub, + summary, + filesLeft, + ); + filesLeft -= outcome.filesAttempted ?? 0; outcomes.push(outcome); - if (outcome.status === "ok") stats.processed++; - else stats.failed++; + tallyDiffOutcome(stats, outcome.status); } // With 0 commits returned the repo's history is exhausted (or the token lost @@ -2769,10 +2974,17 @@ async function runBackfillDiffPhase( * deferred — all keep the commit inside the next run's window, so a transient * failure costs a retry instead of a permanent gap. * - * Each phase is capped at a small commit count (see MAX_DIFF_COMMITS_*) to - * spread cost across many cron ticks. `processAndUpsertCommitDiff` upserts - * on the (repo, commit_sha, file_path) primary key, so overlap with webhook, - * with the opposite phase, or with a retried commit is idempotent. + * Each phase is capped on two axes to spread cost across many cron ticks: a small + * commit count (see MAX_DIFF_COMMITS_*) and a file count + * (`diffFileBudgetPerPhase`). `processAndUpsertCommitDiff` upserts on the + * (repo, commit_sha, file_path) primary key, so overlap with webhook, with the + * opposite phase, or with a retried commit is idempotent. + * + * The file budget can split one commit across runs. Such a commit is `partial`, + * which the invariant treats exactly like `failed`: its watermark does not move + * until the last of its files is indexed. Nothing is thinned — a run indexes a + * prefix of what is left, reads the already-indexed paths back from the store on + * the next run, and continues from there (issue #238). * * Liveness tradeoff: a commit that fails on every attempt (e.g. a permanently * 5xx-ing detail fetch) blocks its phase's watermark. That is deliberate — a @@ -2787,17 +2999,51 @@ export async function pollDiffs( ): Promise { const pollStartTime = new Date().toISOString(); - const fwd = await runForwardDiffPhase(repo, env, storeStub, pollStartTime); - const bwd = await runBackfillDiffPhase(repo, env, storeStub, pollStartTime); + // Every repo of this invocation draws from one subrequest budget, so the share + // each phase may spend depends on how many repos the invocation walks. + const fileBudget = diffFileBudgetPerPhase(parsePollRepos(env).length); + + const fwd = await runForwardDiffPhase( + repo, + env, + storeStub, + pollStartTime, + fileBudget, + ); + const bwd = await runBackfillDiffPhase( + repo, + env, + storeStub, + pollStartTime, + fileBudget, + ); console.log( - `${repo} diffs: forward [processed=${fwd.processed}, failed=${fwd.failed}, ` + - `deferred=${fwd.deferred}, watermark=${fwd.held ? "held" : "advanced"}], ` + - `backward [processed=${bwd.processed}, failed=${bwd.failed}, ` + + `${repo} diffs (file budget ${fileBudget}/phase): ` + + `forward [processed=${fwd.processed}, partial=${fwd.partial}, ` + + `failed=${fwd.failed}, deferred=${fwd.deferred}, ` + + `watermark=${fwd.held ? "held" : "advanced"}], ` + + `backward [processed=${bwd.processed}, partial=${bwd.partial}, ` + + `failed=${bwd.failed}, deferred=${bwd.deferred}, ` + `watermark=${bwd.held ? "held" : "advanced"}]`, ); } +/** + * Repositories this invocation walks, from the `POLL_REPOS` binding. + * + * Shared by the cron dispatch and by `pollDiffs`, which needs the *count* to + * divide its subrequest budget: the two must agree, since a diff phase that sized + * its budget against a different repo list than the loop actually walks is exactly + * the overrun the budget exists to prevent. + */ +export function parsePollRepos(env: Env): string[] { + if (!env.POLL_REPOS) return []; + return env.POLL_REPOS.split(",") + .map((r) => r.trim()) + .filter((r) => r.length > 0); +} + /** Cron expression that triggers the light-surface dispatch (issues / releases / docs). */ const LIGHT_CRON = "0 * * * *"; /** Cron expression that triggers the comments-only dispatch. */ @@ -2937,11 +3183,7 @@ export async function handleScheduled( new Date(controller.scheduledTime).toISOString(), ); - const repos = env.POLL_REPOS - ? env.POLL_REPOS.split(",") - .map((r) => r.trim()) - .filter((r) => r.length > 0) - : []; + const repos = parsePollRepos(env); if (repos.length === 0) { console.warn("POLL_REPOS not configured — no repositories to poll");