From 79dcdb2662a71d9f2f463f2d86c3d7f83882a8fc Mon Sep 17 00:00:00 2001 From: Codex Luna & Lulu Date: Tue, 13 Oct 2026 18:38:16 +0900 Subject: [PATCH 1/3] feat(index): add bounded full sparse document sources MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Codex Luna doc/wiki 専用の全文 owner と独立 FTS、明示 plan/apply の有界 backfill と統計を追加する。prefix と全文の版を検証し、D1 batch/CAS と削除 trigger で更新競合・部分失敗を扱う。既定 off は旧 schema で ingest・検索・削除を維持する。 背景: #284 の保存/運用基盤を #281 の matched 検索と独立して準備する。通常の prefix corpus、graph、vector 数と応答 schema は維持する。本番の migration/apply は実行しない。 --- README.md | 3 + docs/0-requirements.ja.md | 5 + docs/0-requirements.md | 5 + docs/6-full-sparse-sources.ja.md | 21 ++ docs/installation.ja.md | 5 + docs/installation.md | 5 + migrations/0009_doc_sparse_sources.sql | 55 ++++++ src/backfill-doc-sparse.ts | 212 ++++++++++++++++++++ src/doc-sparse.test.ts | 65 +++++++ src/doc-sparse.ts | 94 +++++++++ src/doc-sparse.workers.test.ts | 258 +++++++++++++++++++++++++ src/fts.ts | 22 ++- src/index.ts | 5 + src/pipeline/embed-doc.ts | 33 +++- src/poller.ts | 6 +- src/test-env.d.ts | 1 + src/types.ts | 2 + src/webhook.ts | 10 +- vitest.workers.config.ts | 1 + wrangler.toml | 2 + 20 files changed, 791 insertions(+), 19 deletions(-) create mode 100644 docs/6-full-sparse-sources.ja.md create mode 100644 migrations/0009_doc_sparse_sources.sql create mode 100644 src/backfill-doc-sparse.ts create mode 100644 src/doc-sparse.test.ts create mode 100644 src/doc-sparse.ts create mode 100644 src/doc-sparse.workers.test.ts diff --git a/README.md b/README.md index 7fbcb5b..644c9b4 100644 --- a/README.md +++ b/README.md @@ -298,3 +298,6 @@ wrangler.toml - [Liplus-Project/github-webhook-mcp](https://github.com/Liplus-Project/github-webhook-mcp) - [Liplus-Project/liplus-language](https://github.com/Liplus-Project/liplus-language) + + +Doc/wiki full sparse source preparation is a default-off operator foundation, independent of normal search. See the [storage contract and bounded plan/apply procedure](docs/6-full-sparse-sources.ja.md). Migration 0009 precedes operator preparation and enable; default-off ingest/search remains usable without it. diff --git a/docs/0-requirements.ja.md b/docs/0-requirements.ja.md index 40a3531..36ff4c3 100644 --- a/docs/0-requirements.ja.md +++ b/docs/0-requirements.ja.md @@ -683,3 +683,8 @@ source の `docs/_Sidebar.md` は既存 Feedback Memory EN/JA ページを参照 ### Offline document-tail comparison (#279) 固定公開文書の後半発見を検討する opt-in sparse-only 実験は [Document Tail Replay](5-document-tail-replay.ja.md) の契約に従う。prefix8000 / fullbody / nonoverlap8000 の比較、入力・source hash、named target と exact literal source の別採点、静的rerank予算、節長分布を記録する。既定の検索・embedding・reranker・live索引は変更しない。 + + +### Doc/wiki 独立全文 sparse source + +#284 は既定 off の保存・operator 基盤を追加する。[全文 sparse source 契約](6-full-sparse-sources.ja.md)に従い prefix corpus、既定検索、graph、vector 数を維持する。matched 検索は #281 の別作業。 diff --git a/docs/0-requirements.md b/docs/0-requirements.md index fb505ec..18ad7fa 100644 --- a/docs/0-requirements.md +++ b/docs/0-requirements.md @@ -688,3 +688,8 @@ The source `docs/_Sidebar.md` must link the existing Feedback Memory EN/JA pages ### Offline document-tail comparison (#279) 固定公開文書の後半発見を検討する opt-in sparse-only 実験は [Document Tail Replay](5-document-tail-replay.ja.md) の契約に従う。prefix8000 / fullbody / nonoverlap8000 の比較、入力・source hash、named target と exact literal source の別採点、静的rerank予算、節長分布を記録する。既定の検索・embedding・reranker・live索引は変更しない。 + + +### Independent full sparse doc/wiki sources + +Issue #284 adds a default-off storage/operator foundation; its contract is [Full sparse sources](6-full-sparse-sources.ja.md). It preserves the prefix corpus, default retrieval, graph and vector count. Matched retrieval remains #281. diff --git a/docs/6-full-sparse-sources.ja.md b/docs/6-full-sparse-sources.ja.md new file mode 100644 index 0000000..50bfa6f --- /dev/null +++ b/docs/6-full-sparse-sources.ja.md @@ -0,0 +1,21 @@ +# Doc/wiki 全文 sparse source 基盤 + +Issue #284 の保存・運用契約。通常検索への接続は #281 の別作業。 + +`DOC_SPARSE_CONTEXT_ENABLED` は省略時 false。false では ingest は全文の生成・取得・追加 D1 呼出しを行わない。8000 UTF-16 unit の prefix、通常 nat FTS corpus/IDF、1 資料 1 vector、fetch、graph は維持する。既定 off の ingest と検索は migration 0009 の新 schema にアクセスしない。migration 0009 は operator 経路と enable より先に適用する。検索だけの旧経路は全文テーブルを読まない。 + +`search_docs` が metadata 正本。`doc_sparse_sources` は doc/wiki 専用で vector_id に一意、source_version（doc は Git blob SHA、wiki は body SHA-256）、prefix の取得時点、索引版、raw/body/source hash、UTF-8 bytes、coverage と理由を持つ。全文は path（wiki は slug）+ 2 改行 + body。同じ Worker の segmenter で作った文字列をそのまま所有し、位置表は保存しない。独立した external-content FTS は full coverage の行のみを索引する。 + +raw 256 KiB、segmented 512 KiB、owner 行の文字列 UTF-8 bytes と固定 overhead の合計 1 MiB を上限とする。超過は `out_of_range` として処理済みで prefix-only。全文/hash を取得できなかった場合は null とし、取得済みとしない。同じ版は skip、次の版または operator の `force=true` で再判定する。予算は CPU/heap の安全証明ではない。 + +有効 ingest は doc の blob hash を検証し、prefix と全文を D1 batch transaction で書いてから DO の版を進める。FTS 失敗は版を進めず、既存の ETag 保持・再試行に従う。prefix 更新は旧 owner を無効化し、削除 trigger は owner/独立 FTS を削除する。prefix 更新と backfill の競合は source_version、prefix 本文・type・repo・path・updated_at・indexed_at の compare-and-swap で上書きを防ぐ。 + +## Operator + +認証ヘッダは既存 `GITHUB_TOKEN`。`POST /admin/backfill-doc-sparse?repo=owner/repo&type=doc|wiki_doc` は plan が既定。書込みは `apply=true` のみ。`limit` は既定 5、最大 10、`max_bytes` は既定/最大 1 MiB、`cursor` は search_docs rowid。1 回の source 取得も bounded reader で制限する。doc は DO の blobSHA の Git Blob API から取得・hash 検証し、wiki は DO の extension の raw 本文を取得して bodySHA を照合する。過去 wiki を hash から取得できるとは仮定しない。store と prefix の版が異なるものや検証できないものは conflict、再取得候補として返す。embedding/Vectorize を呼ばず、unchanged/ETag を回避して準備する。 + +応答は対象 repo/type、取得時刻、各 source の版、coverage/bytes、成功・範囲外・欠損・競合・retryable error、次 cursor を持つ。retryable error または取得前に残り byte 予算が無い場合、cursor はその source を越えない。残り予算だけで取得が打ち切られた source は `byte_budget_deferred` として cursor を保持し、次の call で再試行する。call の初期予算でも取得上限で原文/hash を確認できなかった source は `source_byte_limit_unverified` conflict として進め、残りの取得を止める。`source_bytes` は取得・decode が完了した body の UTF-8 bytes、`budget_exhausted` は未取得分の上限到達も区別する。Git Blob JSON の network reader は残り decoded 予算 × 1.4 + 4096 bytes（base64 と envelope）で制限する。missing/conflict は明示報告して次へ進むため、競合解消後は cursor を戻して再実行する。plan は source を取得・検証するが保存しない。apply は同じ source の再実行で owner/FTS/vector を増やさない。 + +`POST /admin/doc-sparse-stats?repo=owner/repo&type=doc|wiki_doc` は indexed prefix population の coverage、未準備、版不一致、legacy prefix の版未検証件数、raw/segmented/行 bytes と索引版別件数を返す。これは repo tree 全体や現在の Git source 本体の件数ではない。 + +本番 apply は merge 後の別 operator 判断。対象 repo/type と plan、復旧境界を確認して migration/apply を行う。2026-10-13 の operator 確認では、全文 D1 export は FTS/virtual table 未対応で失敗しバックアップファイルは作れなかった。Time Travel info は成功し bookmark を控えたが、restore は実行していない。migration/apply 前には Time Travel info を再確認し bookmark と影響対象の metadata を控える。全 repo 一括取得は行わず cursor ごとに観察する。migration は旧 prefix/FTS/vector を変更しない。通常の fallback は enable を false に戻して全文書込みを止める。既定検索と prefix ingest は継続できる。DB 復旧は別操作であり、info で確認した Time Travel bookmark と restore の影響範囲を operator が apply 前に確認する。全 DB の restore は他 surface にも影響する別判断であり、小さい全文 backfill の通常 fallback と混同しない。空の復旧用 D1 への import/binding 切替は、完全なバックアップが得られ手順を検証した場合に限る。既存の populated D1 への export SQL import だけを復元保証にしない。この PR では DB 復元を試行・検証していない。既定 off の ingest・検索は migration 0009 が無くても使用できる。operator 経路と enable は migration 0009 が必要。報告は merge・配置・本番 plan・apply 観測を分ける。 diff --git a/docs/installation.ja.md b/docs/installation.ja.md index 0c4d790..767ebef 100644 --- a/docs/installation.ja.md +++ b/docs/installation.ja.md @@ -506,3 +506,8 @@ GitHub App の callback URL が worker 側 URL と完全一致していない。 ### Webhook verification fails `GITHUB_WEBHOOK_SECRET` が GitHub App 側設定と一致していない。 + + +## Doc/wiki 全文 sparse 準備 + +operator 経路と enable より先に migration 0009 を適用し（既定 off の ingest・検索は新 schema 不要）、準備中は `DOC_SPARSE_CONTEXT_ENABLED="false"` を保つ。認証済み `POST /admin/backfill-doc-sparse?repo=owner/repo&type=doc` は plan が既定。計画と Time Travel bookmark と復旧の影響対象を確認した後（2026-10-13 の full export は FTS/virtual table 未対応で失敗、restore 未実行）だけ `apply=true` を指定する。1 回は最大 10 資料・decoded source 1 MiB、返る `next_cursor` を `cursor` に渡し、競合・error を確認する。wiki は `type=wiki_doc` で別処理。`POST /admin/doc-sparse-stats?repo=owner/repo&type=doc` で索引済み prefix owner の coverage を観察する。[完全な契約](6-full-sparse-sources.ja.md)に予算と rollout 境界を固定している。embedding と検索応答は変わらない。 diff --git a/docs/installation.md b/docs/installation.md index 9a30d23..2bc7394 100644 --- a/docs/installation.md +++ b/docs/installation.md @@ -506,3 +506,8 @@ The GitHub App callback URL does not exactly match the worker callback URL. ### Webhook verification fails `GITHUB_WEBHOOK_SECRET` does not match the value configured in the GitHub App. + + +## Full sparse doc/wiki preparation + +Migration 0009 must precede these operator endpoints and enable. Default-off ingest and search need no new schema. Keep `DOC_SPARSE_CONTEXT_ENABLED="false"` while preparing. The authenticated `POST /admin/backfill-doc-sparse?repo=owner/repo&type=doc` endpoint defaults to plan; use `apply=true` only after reading the plan and confirming the Time Travel bookmark/recovery scope (full export failed on FTS/virtual tables in the 2026-10-13 operator check; no restore was executed). Process at most 10 sources and 1 MiB decoded source bytes per call, pass the returned `next_cursor` as `cursor`, and examine conflicts/errors before continuing. Use `type=wiki_doc` separately. `POST /admin/doc-sparse-stats?repo=owner/repo&type=doc` reports coverage of indexed prefix owners. See [the complete contract](6-full-sparse-sources.ja.md) for limits and rollout boundaries. Neither operation embeds or changes retrieval. diff --git a/migrations/0009_doc_sparse_sources.sql b/migrations/0009_doc_sparse_sources.sql new file mode 100644 index 0000000..43310a7 --- /dev/null +++ b/migrations/0009_doc_sparse_sources.sql @@ -0,0 +1,55 @@ +-- Independent doc/wiki owner. Prefix FTS tables and their corpus stay unchanged. +ALTER TABLE search_docs ADD COLUMN source_version TEXT NOT NULL DEFAULT ''; + +CREATE TABLE doc_sparse_sources ( + vector_id TEXT NOT NULL UNIQUE, + source_type TEXT NOT NULL CHECK(source_type IN ('doc', 'wiki_doc')), + source_version TEXT NOT NULL, + prefix_indexed_at TEXT NOT NULL, + index_version TEXT NOT NULL, + source_hash TEXT, + body_hash TEXT, + raw_hash TEXT, + raw TEXT, + segmented TEXT, + raw_bytes INTEGER, + segmented_bytes INTEGER, + row_bytes INTEGER NOT NULL CHECK(row_bytes BETWEEN 0 AND 1048576), + coverage TEXT NOT NULL CHECK(coverage IN ('full', 'out_of_range')), + reason TEXT NOT NULL, + observed_at TEXT NOT NULL, + CHECK(coverage <> 'full' OR (raw IS NOT NULL AND segmented IS NOT NULL + AND source_hash IS NOT NULL AND body_hash IS NOT NULL AND raw_hash IS NOT NULL + AND raw_bytes BETWEEN 0 AND 262144 AND segmented_bytes BETWEEN 0 AND 524288)) +); + +CREATE VIEW doc_sparse_content_v1 AS + SELECT rowid, segmented AS content FROM doc_sparse_sources WHERE coverage = 'full'; +CREATE VIRTUAL TABLE doc_sparse_fts_v1 USING fts5( + content, tokenize = 'porter unicode61 remove_diacritics 2', + content = 'doc_sparse_content_v1', content_rowid = 'rowid' +); +CREATE TRIGGER doc_sparse_ai AFTER INSERT ON doc_sparse_sources BEGIN + INSERT INTO doc_sparse_fts_v1(rowid, content) + SELECT new.rowid, new.segmented WHERE new.coverage = 'full'; +END; +CREATE TRIGGER doc_sparse_ad AFTER DELETE ON doc_sparse_sources BEGIN + INSERT INTO doc_sparse_fts_v1(doc_sparse_fts_v1, rowid, content) + SELECT 'delete', old.rowid, old.segmented WHERE old.coverage = 'full'; +END; +CREATE TRIGGER doc_sparse_au AFTER UPDATE ON doc_sparse_sources BEGIN + INSERT INTO doc_sparse_fts_v1(doc_sparse_fts_v1, rowid, content) + SELECT 'delete', old.rowid, old.segmented WHERE old.coverage = 'full'; + INSERT INTO doc_sparse_fts_v1(rowid, content) + SELECT new.rowid, new.segmented WHERE new.coverage = 'full'; +END; +CREATE TRIGGER doc_sparse_prefix_delete AFTER DELETE ON search_docs BEGIN + DELETE FROM doc_sparse_sources WHERE vector_id = old.vector_id; +END; +-- Even a default-off prefix update invalidates an old full snapshot. +CREATE TRIGGER doc_sparse_prefix_update AFTER UPDATE ON search_docs BEGIN + DELETE FROM doc_sparse_sources WHERE vector_id = old.vector_id + AND (old.source_version <> new.source_version OR old.content <> new.content + OR old.indexed_at <> new.indexed_at OR old.updated_at <> new.updated_at + OR old.repo <> new.repo OR old.type <> new.type OR old.doc_path <> new.doc_path); +END; diff --git a/src/backfill-doc-sparse.ts b/src/backfill-doc-sparse.ts new file mode 100644 index 0000000..3c4a7cb --- /dev/null +++ b/src/backfill-doc-sparse.ts @@ -0,0 +1,212 @@ +import type { Env, DocRecord, WikiDocRecord } from './types.js'; +import { prepareEmbeddingInput, sha256Hex } from './pipeline/hash.js'; +import { + DOC_SPARSE_INDEX_VERSION, gitBlobSha, prepareSparseSource, + sparseSourceStatement, utf8Bytes, type PrefixSnapshot, +} from './doc-sparse.js'; + +export const MAX_BACKFILL_BYTES = 1024 * 1024; +export async function handleDocSparseAdmin(request: Request, env: Env): Promise { + if (!env.GITHUB_TOKEN || request.headers.get('GITHUB_TOKEN') !== env.GITHUB_TOKEN) return new Response('Unauthorized', { status: 401 }); + const url = new URL(request.url); + const opts: SparseBackfillOptions = { + repo: url.searchParams.get('repo') ?? '', type: url.searchParams.get('type') as SparseBackfillOptions['type'], + cursor: Number(url.searchParams.get('cursor') ?? 0), limit: Number(url.searchParams.get('limit') ?? 5), + maxBytes: Number(url.searchParams.get('max_bytes') ?? MAX_BACKFILL_BYTES), + apply: url.searchParams.get('apply') === 'true', force: url.searchParams.get('force') === 'true', + }; + if (['apply', 'force'].some(key => url.searchParams.has(key) && !['true', 'false'].includes(url.searchParams.get(key)!))) { + return Response.json({ error: 'Boolean options must be true or false' }, { status: 400 }); + } + try { + return Response.json(url.pathname.endsWith('stats') ? await docSparseStats(env.DB_FTS, opts) : await backfillDocSparse(env, opts)); + } catch { + return Response.json({ error: 'Sparse operation failed; check options and migration 0009; cursor unchanged' }, { status: 400 }); + } +} + +export interface SparseBackfillOptions { + repo: string; type: 'doc' | 'wiki_doc'; limit?: number; cursor?: number; + maxBytes?: number; apply?: boolean; force?: boolean; +} +function validate(opts: SparseBackfillOptions): void { + if (!/^[\w.-]+\/[\w.-]+$/.test(opts.repo) || !['doc', 'wiki_doc'].includes(opts.type) + || !Number.isSafeInteger(opts.limit ?? 5) || (opts.limit ?? 5) < 1 || (opts.limit ?? 5) > 10 + || !Number.isSafeInteger(opts.cursor ?? 0) || (opts.cursor ?? 0) < 0 + || !Number.isSafeInteger(opts.maxBytes ?? MAX_BACKFILL_BYTES) + || (opts.maxBytes ?? MAX_BACKFILL_BYTES) < 1 || (opts.maxBytes ?? MAX_BACKFILL_BYTES) > MAX_BACKFILL_BYTES) { + throw new Error('Invalid sparse backfill options'); + } +} + +class SourceLimitError extends Error {} +class MissingSourceError extends Error {} +class SourceVersionError extends Error {} + +/** Bound network acquisition before parsing JSON or decoding the source. */ +async function readBounded(response: Response, limit: number): Promise { + if (!response.ok) { + await response.body?.cancel(); + if (response.status === 404) throw new MissingSourceError('Source missing'); + throw new Error(`Source HTTP ${response.status}`); + } + if (!response.body) throw new Error('Source has no body'); + const reader = response.body.getReader(); + const chunks: Uint8Array[] = []; let size = 0; + try { + while (true) { + const result = await reader.read(); if (result.done) break; + size += result.value.byteLength; + if (size > limit) { await reader.cancel(); throw new SourceLimitError('Source acquisition byte limit'); } + chunks.push(result.value); + } + } finally { reader.releaseLock(); } + const bytes = new Uint8Array(size); let offset = 0; + for (const chunk of chunks) { bytes.set(chunk, offset); offset += chunk.length; } + return new TextDecoder('utf-8', { fatal: true, ignoreBOM: false }).decode(bytes); +} + +export async function fetchDocBlob(repo: string, sha: string, token: string, maxBytes?: number): Promise { + if (!/^[a-f0-9]{40}$/.test(sha)) throw new SourceVersionError('Not a Git blob SHA'); + const response = await fetch(`https://api.github.com/repos/${repo}/git/blobs/${sha}`, { + headers: { Authorization: `Bearer ${token}`, Accept: 'application/vnd.github+json', + 'User-Agent': 'github-rag-mcp/0.1.0', 'X-GitHub-Api-Version': '2022-11-28' }, + }); + // Ordinary ingest retains its existing full source acquisition; only operator runs + // use the byte cap. The source owner applies its raw/segmented/row budgets after hashing. + const data = JSON.parse(await readBounded(response, maxBytes === undefined + ? Number.MAX_SAFE_INTEGER : Math.ceil(maxBytes * 1.4) + 4096)) as { + sha: string; content: string; encoding: string; + }; + if (data.sha !== sha) throw new SourceVersionError('Blob response version mismatch'); + if (data.encoding !== 'base64') throw new Error('Blob encoding mismatch'); + const binary = atob(data.content.replace(/\s/g, '')); + if (maxBytes !== undefined && binary.length > maxBytes) throw new SourceLimitError('Decoded source byte limit'); + const body = new TextDecoder('utf-8', { fatal: true, ignoreBOM: true }).decode(Uint8Array.from(binary, c => c.charCodeAt(0))); + if (await gitBlobSha(body) !== sha) throw new SourceVersionError('Blob body version mismatch'); + return body; +} + +type SnapshotRow = PrefixSnapshot & { rid: number; prepared_version: string | null; prepared_index: string | null; prepared_at: string | null }; +type StoreRecord = DocRecord | WikiDocRecord; +async function readStore(store: DurableObjectStub, prefix: PrefixSnapshot): Promise { + const path = prefix.type === 'doc' ? 'doc' : 'wiki-doc'; + const key = prefix.type === 'doc' ? 'path' : 'page'; + const response = await store.fetch(new Request(`http://store/${path}?repo=${encodeURIComponent(prefix.repo)}&${key}=${encodeURIComponent(prefix.doc_path)}`)); + if (response.status === 404) return null; + if (!response.ok) throw new Error('Source store unavailable'); + return await response.json() as StoreRecord; +} +const versionOf = (record: StoreRecord): string => 'blobSha' in record ? record.blobSha : record.contentHash; + +export async function backfillDocSparse(env: Env, opts: SparseBackfillOptions) { + validate(opts); + const limit = opts.limit ?? 5; const cursor = opts.cursor ?? 0; + const maxBytes = opts.maxBytes ?? MAX_BACKFILL_BYTES; + const store = env.ISSUE_STORE.get(env.ISSUE_STORE.idFromName('global')); + const rows = (await env.DB_FTS.prepare(`SELECT d.rowid AS rid, d.vector_id, d.repo, d.type, + d.doc_path, d.source_version, d.content, d.updated_at, d.indexed_at, + s.source_version AS prepared_version, s.index_version AS prepared_index, s.prefix_indexed_at AS prepared_at + FROM search_docs d LEFT JOIN doc_sparse_sources s ON s.vector_id=d.vector_id + WHERE d.repo=? AND d.type=? AND d.rowid>? ORDER BY d.rowid LIMIT ?`) + .bind(opts.repo, opts.type, cursor, limit).all()).results ?? []; + const results: { vector_id: string; source_version: string | null; status: string; reason?: string; + coverage?: string; raw_bytes?: number | null; segmented_bytes?: number | null; row_bytes?: number }[] = []; + let bytes = 0; let nextCursor = cursor; let stopped = false; let budgetExhausted = false; + for (const prefix of rows) { + let version: string | null = null; + try { + const record = await readStore(store, prefix); + if (!record) { + results.push({ vector_id: prefix.vector_id, source_version: null, status: 'missing', reason: 'store_missing' }); + nextCursor = prefix.rid; continue; + } + version = versionOf(record); + if (record.updatedAt !== prefix.updated_at || (prefix.source_version && prefix.source_version !== version)) { + results.push({ vector_id: prefix.vector_id, source_version: version, status: 'conflict', reason: 'store_prefix_version' }); + nextCursor = prefix.rid; continue; + } + if (!opts.force && prefix.prepared_version === version && prefix.prepared_index === DOC_SPARSE_INDEX_VERSION + && prefix.prepared_at === prefix.indexed_at) { + results.push({ vector_id: prefix.vector_id, source_version: version, status: 'unchanged' }); + nextCursor = prefix.rid; continue; + } + const remaining = maxBytes - bytes; + if (remaining < 1) { stopped = true; budgetExhausted = true; break; } + let body: string; + try { + body = prefix.type === 'doc' ? await fetchDocBlob(opts.repo, version, env.GITHUB_TOKEN, remaining) + : await readBounded(await fetch(`https://raw.githubusercontent.com/wiki/${opts.repo}/${encodeURIComponent(prefix.doc_path)}.${(record as WikiDocRecord).extension}`, { + headers: { 'User-Agent': 'github-rag-mcp/0.1.0' }, + }), remaining); + } catch (error) { + if (error instanceof SourceLimitError) { + // An unfetched body has no verified hash. Do not invent a processed snapshot. + const deferred = bytes > 0; + results.push({ vector_id: prefix.vector_id, source_version: version, + status: deferred ? 'retryable_error' : 'conflict', + reason: deferred ? 'byte_budget_deferred' : 'source_byte_limit_unverified' }); + // The failed acquisition spent the remaining envelope, even though no + // decoded body was obtained. Stop rather than spending it again per row. + budgetExhausted = true; + if (!deferred) nextCursor = prefix.rid; + stopped = true; break; + } + throw error; + } + bytes += utf8Bytes(body); + const actualVersion = prefix.type === 'doc' ? await gitBlobSha(body) : await sha256Hex(body); + if (actualVersion !== version || prepareEmbeddingInput(prefix.doc_path, body) !== prefix.content) { + results.push({ vector_id: prefix.vector_id, source_version: version, status: 'conflict', reason: 'source_prefix_version' }); + nextCursor = prefix.rid; continue; + } + const source = await prepareSparseSource(prefix.vector_id, prefix.type, prefix.doc_path, body, version, prefix.indexed_at); + // A DO version read cannot form a cross-database transaction. The D1 CAS below + // protects against concurrent prefix update/delete; this read rejects store changes too. + const current = await readStore(store, prefix); + if (!current || versionOf(current) !== version || current.updatedAt !== record.updatedAt) { + results.push({ vector_id: prefix.vector_id, source_version: version, status: 'conflict', reason: 'store_changed' }); + nextCursor = prefix.rid; continue; + } + if (opts.apply) { + const written = await sparseSourceStatement(env.DB_FTS, source, prefix).all(); + if (!written.results?.length) { + results.push({ vector_id: prefix.vector_id, source_version: version, status: 'conflict', reason: 'prefix_changed' }); + nextCursor = prefix.rid; continue; + } + } + results.push({ vector_id: prefix.vector_id, source_version: version, + status: source.coverage === 'full' ? 'success' : 'out_of_range', reason: source.reason, + coverage: source.coverage, raw_bytes: source.rawBytes, segmented_bytes: source.segmentedBytes, row_bytes: source.rowBytes }); + nextCursor = prefix.rid; + } catch (error) { + const missing = error instanceof MissingSourceError; const conflict = error instanceof SourceVersionError; + results.push({ vector_id: prefix.vector_id, source_version: version, status: missing ? 'missing' : conflict ? 'conflict' : 'retryable_error', + reason: missing ? 'source_missing' : conflict ? 'source_version_unverified' : 'source_or_write_failed' }); + if (missing || conflict) nextCursor = prefix.rid; + else { stopped = true; break; } + } + } + const counts: Record = {}; + for (const result of results) counts[result.status] = (counts[result.status] ?? 0) + 1; + return { repo: opts.repo, type: opts.type, observed_at: new Date().toISOString(), + index_version: DOC_SPARSE_INDEX_VERSION, mode: opts.apply ? 'apply' : 'plan', + source_bytes: bytes, budget_exhausted: budgetExhausted, max_bytes: maxBytes, scanned: results.length, counts, results, + next_cursor: !stopped && rows.length < limit ? null : nextCursor }; +} + +export async function docSparseStats(db: D1Database, opts: Pick) { + validate(opts); + const rows = (await db.prepare(`SELECT + CASE WHEN s.vector_id IS NULL THEN 'unprepared' + WHEN (d.source_version<>'' AND d.source_version<>s.source_version) OR d.indexed_at<>s.prefix_indexed_at THEN 'version_mismatch' + ELSE s.coverage END AS coverage, s.index_version, + COUNT(*) AS count, SUM(COALESCE(s.raw_bytes,0)) AS raw_bytes, + SUM(CASE WHEN s.vector_id IS NOT NULL AND d.source_version='' THEN 1 ELSE 0 END) AS legacy_version_unverified, + SUM(COALESCE(s.segmented_bytes,0)) AS segmented_bytes, SUM(COALESCE(s.row_bytes,0)) AS row_bytes + FROM search_docs d LEFT JOIN doc_sparse_sources s ON s.vector_id=d.vector_id + WHERE d.repo=? AND d.type=? GROUP BY coverage,s.index_version`) + .bind(opts.repo, opts.type).all()).results ?? []; + return { repo: opts.repo, type: opts.type, observed_at: new Date().toISOString(), population: 'indexed_prefix_owners', + source_count: rows.reduce((n, row) => n + Number(row.count), 0), rows }; +} diff --git a/src/doc-sparse.test.ts b/src/doc-sparse.test.ts new file mode 100644 index 0000000..a7cad57 --- /dev/null +++ b/src/doc-sparse.test.ts @@ -0,0 +1,65 @@ +import { describe, it, expect, vi, afterEach } from 'vitest'; +import { prepareSparseSource, gitBlobSha, MAX_RAW_BYTES, MAX_SEGMENTED_BYTES, MAX_ROW_BYTES, sourceRowBytes, docSparseEnabled } from './doc-sparse.js'; +import { sha256Hex } from './pipeline/hash.js'; +import { segmentForFts } from './segment.js'; +import * as segment from './segment.js'; +import type { Env } from './types.js'; +afterEach(() => vi.restoreAllMocks()); + +describe('full sparse ownership byte/version contract', () => { + it('preserves exact raw/hash/segmentation for Japanese and CRLF text', async () => { + const body = '判断構造\r\n\r\n後半の目印 😀'; + const version = await gitBlobSha(body); + const source = await prepareSparseSource('doc-id', 'doc', 'docs/a.md', body, version); + expect(source.raw).toBe(`docs/a.md\n\n${body}`); + expect(source.rawHash).toBe(await sha256Hex(source.raw!)); + expect(source.bodyHash).toBe(await sha256Hex(body)); + expect(source.sourceHash).toBe(version); + expect(source.segmented).toBe(segmentForFts(source.raw!)); + expect(source.rowBytes).toBe(sourceRowBytes(source)); + await expect(prepareSparseSource('id', 'doc', 'a', body, '0'.repeat(40))).rejects.toThrow('version'); + }); + + it('allows the exact UTF8 raw boundary and processes the next byte as out of range', async () => { + const body = 'x'.repeat(MAX_RAW_BYTES - 3); + const allowed = await prepareSparseSource('a', 'wiki_doc', 'p', body, await sha256Hex(body)); + const tooLarge = await prepareSparseSource('b', 'wiki_doc', 'p', body + 'x', await sha256Hex(body + 'x')); + expect(allowed.coverage).toBe('full'); expect(allowed.rawBytes).toBe(MAX_RAW_BYTES); + expect(tooLarge.coverage).toBe('out_of_range'); expect(tooLarge.reason).toBe('raw_bytes'); + expect(tooLarge.raw).toBeNull(); expect(tooLarge.segmented).toBeNull(); + expect(tooLarge.bodyHash).toBe(await sha256Hex(body + 'x')); + expect(tooLarge.segmentedBytes).toBeNull(); + }); + + it('counts UTF8 rather than UTF16 and enables only explicitly', async () => { + const body = '界'.repeat(Math.ceil(MAX_RAW_BYTES / 3)); + const source = await prepareSparseSource('x', 'wiki_doc', 'p', body, await sha256Hex(body)); + expect(source.coverage).toBe('out_of_range'); expect(body.length).toBeLessThan(MAX_RAW_BYTES); + expect(docSparseEnabled({} as Env)).toBe(false); + expect(docSparseEnabled({ DOC_SPARSE_CONTEXT_ENABLED: 'TRUE' } as Env)).toBe(false); + expect(docSparseEnabled({ DOC_SPARSE_CONTEXT_ENABLED: 'true' } as Env)).toBe(true); + }); + + it('enforces the segmented boundary if a segmenter emits expanded text', async () => { + const spy = vi.spyOn(segment, 'segmentForFts'); + spy.mockReturnValueOnce('s'.repeat(MAX_SEGMENTED_BYTES)); + const exact = await prepareSparseSource('x', 'wiki_doc', 'p', 'body', await sha256Hex('body')); + expect(exact.coverage).toBe('full'); expect(exact.segmentedBytes).toBe(MAX_SEGMENTED_BYTES); + spy.mockReturnValueOnce('s'.repeat(MAX_SEGMENTED_BYTES + 1)); + const over = await prepareSparseSource('x', 'wiki_doc', 'p', 'body', await sha256Hex('body')); + expect(over.coverage).toBe('out_of_range'); expect(over.reason).toBe('segmented_bytes'); + expect(over.segmented).toBeNull(); + }); + + it('checks the combined row boundary independently of raw/segmented limits', async () => { + const body = 'body'.repeat(100); + const hash = await sha256Hex(body); + const base = await prepareSparseSource('x', 'wiki_doc', 'p', body, hash); + const id = 'x'.repeat(MAX_ROW_BYTES - base.rowBytes + 1); + const exact = await prepareSparseSource(id, 'wiki_doc', 'p', body, hash); + expect(exact.rowBytes).toBe(MAX_ROW_BYTES); expect(exact.coverage).toBe('full'); + const over = await prepareSparseSource(id + 'x', 'wiki_doc', 'p', body, hash); + expect(over.coverage).toBe('out_of_range'); expect(over.reason).toBe('row_bytes'); + expect(over.rowBytes).toBeLessThan(MAX_ROW_BYTES); expect(over.raw).toBeNull(); + }); +}); diff --git a/src/doc-sparse.ts b/src/doc-sparse.ts new file mode 100644 index 0000000..f5779a2 --- /dev/null +++ b/src/doc-sparse.ts @@ -0,0 +1,94 @@ +import { segmentForFts } from './segment.js'; +import { sha256Hex } from './pipeline/hash.js'; +import type { Env } from './types.js'; + +export const DOC_SPARSE_INDEX_VERSION = 'full-v1'; +export const MAX_RAW_BYTES = 256 * 1024; +export const MAX_SEGMENTED_BYTES = 512 * 1024; +export const MAX_ROW_BYTES = 1024 * 1024; +export const utf8Bytes = (value: string): number => new TextEncoder().encode(value).byteLength; +export const docSparseEnabled = (env: Env): boolean => env.DOC_SPARSE_CONTEXT_ENABLED === 'true'; + +export interface SparseSource { + vectorId: string; type: 'doc' | 'wiki_doc'; sourceVersion: string; + prefixIndexedAt: string; indexVersion: string; + sourceHash: string | null; bodyHash: string | null; rawHash: string | null; + raw: string | null; segmented: string | null; + rawBytes: number | null; segmentedBytes: number | null; rowBytes: number; + coverage: 'full' | 'out_of_range'; reason: string; observedAt: string; +} + +/** Conservative owner budget: all TEXT bytes plus 256 bytes of scalar/header allowance. */ +export function sourceRowBytes(source: SparseSource): number { + return 256 + Object.values(source).reduce((n, v) => n + (typeof v === 'string' ? utf8Bytes(v) : 0), 0); +} + +export async function gitBlobSha(body: string): Promise { + const bytes = new TextEncoder().encode(body); + const header = new TextEncoder().encode(`blob ${bytes.length}\0`); + const all = new Uint8Array(header.length + bytes.length); + all.set(header); all.set(bytes, header.length); + return Array.from(new Uint8Array(await crypto.subtle.digest('SHA-1', all))) + .map(v => v.toString(16).padStart(2, '0')).join(''); +} + +export async function prepareSparseSource( + vectorId: string, type: SparseSource['type'], path: string, body: string, + sourceVersion: string, prefixIndexedAt = '', +): Promise { + const bodyHash = await sha256Hex(body); + const sourceHash = type === 'doc' ? await gitBlobSha(body) : bodyHash; + if (sourceHash !== sourceVersion) throw new Error('Source version mismatch'); + const raw = `${path}\n\n${body}`; + const source: SparseSource = { + vectorId, type, sourceVersion, prefixIndexedAt, indexVersion: DOC_SPARSE_INDEX_VERSION, + sourceHash, bodyHash, rawHash: await sha256Hex(raw), raw: null, segmented: null, + rawBytes: utf8Bytes(raw), segmentedBytes: null, rowBytes: 0, + coverage: 'out_of_range', reason: 'raw_bytes', observedAt: new Date().toISOString(), + }; + if (source.rawBytes! <= MAX_RAW_BYTES) { + const segmented = segmentForFts(raw); + source.segmentedBytes = utf8Bytes(segmented); + source.reason = 'segmented_bytes'; + if (source.segmentedBytes <= MAX_SEGMENTED_BYTES) { + source.raw = raw; source.segmented = segmented; source.coverage = 'full'; source.reason = ''; + source.rowBytes = sourceRowBytes(source); + if (source.rowBytes > MAX_ROW_BYTES) { + source.raw = null; source.segmented = null; source.coverage = 'out_of_range'; source.reason = 'row_bytes'; + } + } + } + source.rowBytes = sourceRowBytes(source); + return source; +} + +export interface PrefixSnapshot { + vector_id: string; repo: string; type: SparseSource['type']; doc_path: string; + source_version: string; content: string; updated_at: string; indexed_at: string; +} + +/** INSERT SELECT is the CAS: a deleted/changed canonical owner cannot accept this snapshot. */ +export function sparseSourceStatement(db: D1Database, source: SparseSource, prefix: PrefixSnapshot): D1PreparedStatement { + if (sourceRowBytes(source) > MAX_ROW_BYTES) throw new Error('Sparse owner metadata exceeds row budget'); + return db.prepare(`INSERT INTO doc_sparse_sources ( + vector_id, source_type, source_version, prefix_indexed_at, index_version, + source_hash, body_hash, raw_hash, raw, segmented, raw_bytes, segmented_bytes, + row_bytes, coverage, reason, observed_at) + SELECT ?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,? FROM search_docs + WHERE vector_id=? AND repo=? AND type=? AND doc_path=? AND source_version=? + AND content=? AND updated_at=? AND indexed_at=? + ON CONFLICT(vector_id) DO UPDATE SET + source_type=excluded.source_type, source_version=excluded.source_version, + prefix_indexed_at=excluded.prefix_indexed_at, index_version=excluded.index_version, + source_hash=excluded.source_hash, body_hash=excluded.body_hash, raw_hash=excluded.raw_hash, + raw=excluded.raw, segmented=excluded.segmented, raw_bytes=excluded.raw_bytes, + segmented_bytes=excluded.segmented_bytes, row_bytes=excluded.row_bytes, + coverage=excluded.coverage, reason=excluded.reason, observed_at=excluded.observed_at + RETURNING vector_id`).bind( + source.vectorId, source.type, source.sourceVersion, source.prefixIndexedAt, source.indexVersion, + source.sourceHash, source.bodyHash, source.rawHash, source.raw, source.segmented, + source.rawBytes, source.segmentedBytes, source.rowBytes, source.coverage, source.reason, source.observedAt, + prefix.vector_id, prefix.repo, prefix.type, prefix.doc_path, prefix.source_version, + prefix.content, prefix.updated_at, prefix.indexed_at, + ); +} diff --git a/src/doc-sparse.workers.test.ts b/src/doc-sparse.workers.test.ts new file mode 100644 index 0000000..13e1ed7 --- /dev/null +++ b/src/doc-sparse.workers.test.ts @@ -0,0 +1,258 @@ +import { describe, it, expect, beforeAll, vi, afterEach } from 'vitest'; +import { env, applyD1Migrations } from 'cloudflare:test'; +import { upsertFtsRow, prepareFtsRow, queryFts, deleteFtsRow, type FtsUpsertRow } from './fts.js'; +import { prepareSparseSource, sparseSourceStatement, gitBlobSha, MAX_RAW_BYTES, type PrefixSnapshot } from './doc-sparse.js'; +import { backfillDocSparse, docSparseStats, handleDocSparseAdmin } from './backfill-doc-sparse.js'; +import { prepareEmbeddingInput, sha256Hex } from './pipeline/hash.js'; +import { processAndUpsertDoc, processAndUpsertWikiDoc } from './pipeline/embed-doc.js'; +import type { Env, DocRecord } from './types.js'; +import { handlePushEvent } from './webhook.js'; +import { pollDocs, pollWiki } from './poller.js'; +import { docVectorId, wikiDocVectorId } from './pipeline/vector-id.js'; + +beforeAll(async () => { await applyD1Migrations(env.DB_FTS, env.TEST_MIGRATIONS); }); +afterEach(() => vi.unstubAllGlobals()); +const prefix = async (id: string) => (await env.DB_FTS.prepare('SELECT * FROM search_docs WHERE vector_id=?').bind(id).first())!; +const owner = (id: string) => env.DB_FTS.prepare('SELECT * FROM doc_sparse_sources WHERE vector_id=?').bind(id).first<{ coverage: string; source_version: string; raw: string; segmented: string; row_bytes: number }>(); +const store = () => env.ISSUE_STORE.get(env.ISSUE_STORE.idFromName('global')); +const runtime = (enabled = true) => ({ ...env, DOC_SPARSE_CONTEXT_ENABLED: enabled ? 'true' : undefined, + AI: { run: vi.fn(async () => ({ data: [[0.1, 0.2]] })) }, VECTORIZE: { upsert: vi.fn(async () => undefined) }, +}) as unknown as Env; +async function seed(repo: string, id: string, body: string, legacy = false, path = 'a.md') { + const record: DocRecord = { repo, path, blobSha: await gitBlobSha(body), updatedAt: '2026-10-13T00:00:00Z' }; + const row: FtsUpsertRow = { repo, vectorId: id, type: 'doc', state: 'active', labels: '', milestone: '', assignees: '', + updatedAt: record.updatedAt, docPath: record.path, content: prepareEmbeddingInput(record.path, body), + ...(legacy ? {} : { sourceVersion: record.blobSha }) }; + await upsertFtsRow(env.DB_FTS, row); + await store().fetch(new Request('http://store/upsert-doc', { method: 'POST', body: JSON.stringify(record) })); + return { row, record }; +} +function mockBlob(body: string, sha: string, before?: () => Promise) { + vi.stubGlobal('fetch', vi.fn(async () => { + await before?.(); + return Response.json({ sha, encoding: 'base64', content: btoa(body) }); + })); +} + +describe('independent full sparse D1 index', () => { + it('keeps default-off doc/wiki ingest and retrieval usable before migration 0009', async () => { + await applyD1Migrations(env.DB_FTS_DOC_LEGACY, env.TEST_MIGRATIONS.filter(m => !m.name.startsWith('0009'))); + const live = { ...runtime(false), DB_FTS: env.DB_FTS_DOC_LEGACY }; + expect((await processAndUpsertDoc(live, store(), 'sparse/legacy-schema', 'a.md', 'legacysentinel', 'old-standin')).embedded).toBe(true); + expect((await processAndUpsertWikiDoc(live, store(), 'sparse/legacy-schema', 'Home', 'md', 'legacywiki')).embedded).toBe(true); + expect((await queryFts(live.DB_FTS, 'legacysentinel', 10, { repo: 'sparse/legacy-schema' })).length).toBe(1); + expect((await queryFts(live.DB_FTS, 'legacywiki', 10, { repo: 'sparse/legacy-schema' })).length).toBe(1); + expect(await live.DB_FTS.prepare("SELECT name FROM sqlite_master WHERE name='doc_sparse_sources'").first()).toBeNull(); + const columns = (await live.DB_FTS.prepare('PRAGMA table_info(search_docs)').all<{ name: string }>()).results; + expect(columns.some(c => c.name === 'source_version')).toBe(false); + expect(live.VECTORIZE.upsert).toHaveBeenCalledTimes(2); + }); + + for (const route of ['push', 'doc-cron', 'wiki-cron']) it(`runs default-off ${route} on the pre-0009 schema`, async () => { + await applyD1Migrations(env.DB_FTS_DOC_LEGACY, env.TEST_MIGRATIONS.filter(m => !m.name.startsWith('0009'))); + const live = { ...runtime(false), DB_FTS: env.DB_FTS_DOC_LEGACY }; + live.VECTORIZE.deleteByIds = vi.fn(async () => ({ count: 1, mutationId: 'test' })); + const repo = `sparse/legacy-${route}`; + const body = 'legacyroute'; + vi.stubGlobal('fetch', vi.fn(async (url: string) => { + if (url.includes('/git/trees/')) return Response.json({ tree: [{ path: 'route.md', type: 'blob', sha: 'tree-standin' }], sha: 'tree', truncated: false }); + if (url.includes('/contents/')) return Response.json({ content: btoa(body), encoding: 'base64' }); + if (url.includes('.wiki.git/info/refs')) return new Response('refs'); + if (url.endsWith('/wiki/_pages')) return new Response(`Home`); + if (url.endsWith('/Home.md')) return new Response(body); + throw new Error(`Unexpected new source fetch ${url}`); + })); + if (route === 'push') { + const payload = { repository: { full_name: repo, default_branch: 'main' }, ref: 'refs/heads/main', + head_commit: { id: 'commit-standin' }, commits: [{ added: ['route.md'], modified: [], removed: [] }] }; + const response = await handlePushEvent(payload, live, store()); + expect((await response.json() as { docs: { embedded: number } }).docs.embedded).toBe(1); + expect((await handlePushEvent({ ...payload, commits: [{ added: [], modified: [], removed: ['route.md'] }] }, live, store())).status).toBe(202); + expect(await live.DB_FTS.prepare('SELECT vector_id FROM search_docs WHERE repo=?').bind(repo).first()).toBeNull(); + } else if (route === 'doc-cron') { + await pollDocs(repo, live, store()); + expect((await queryFts(live.DB_FTS, body, 10, { repo })).length).toBe(1); + } else { + expect((await pollWiki(repo, live, store())).embedded).toBe(1); + expect((await queryFts(live.DB_FTS, body, 10, { repo })).length).toBe(1); + } + expect(await live.DB_FTS.prepare("SELECT name FROM sqlite_master WHERE name='doc_sparse_sources'").first()).toBeNull(); + }); + + for (const route of ['push', 'doc-poll', 'wiki-poll']) it(`removes the full owner/index through ${route}`, async () => { + const repo = `sparse/delete-${route}`; + const live = runtime(); + live.VECTORIZE.deleteByIds = vi.fn(async () => ({ count: 1, mutationId: 'test' })); + const wiki = route === 'wiki-poll'; + const id = wiki ? await wikiDocVectorId(repo, 'Gone') : await docVectorId(repo, 'gone.md'); + if (wiki) { + await processAndUpsertWikiDoc(live, store(), repo, 'Gone', 'md', 'teardownneedle'); + await processAndUpsertWikiDoc(live, store(), repo, 'Home', 'md', 'home'); + } else { + await processAndUpsertDoc(live, store(), repo, 'gone.md', 'teardownneedle', await gitBlobSha('teardownneedle')); + } + expect(await owner(id)).not.toBeNull(); + if (route === 'push') { + const response = await handlePushEvent({ repository: { full_name: repo, default_branch: 'main' }, ref: 'refs/heads/main', + commits: [{ added: [], modified: [], removed: ['gone.md'] }] }, live, store()); + expect(response.status).toBe(202); + } else if (route === 'doc-poll') { + const keep = await seed(repo, 'sparse-kept', 'keep', false, 'keep.md'); + vi.stubGlobal('fetch', vi.fn(async () => Response.json({ sha: 'tree', tree: [{ path: 'keep.md', type: 'blob', sha: keep.record.blobSha }], truncated: false }))); + await pollDocs(repo, live, store()); + } else { + vi.stubGlobal('fetch', vi.fn(async (url: string) => { + if (url.includes('.wiki.git/info/refs')) return new Response('refs'); + if (url.endsWith('/wiki/_pages')) return new Response(`Home`); + if (url.endsWith('/Home.md')) return new Response('home'); + return new Response('missing', { status: 404 }); + })); + const summary = await pollWiki(repo, live, store()); + expect(summary.removed).toBe(1); + } + expect(await owner(id)).toBeNull(); + expect(await env.DB_FTS.prepare('SELECT vector_id FROM search_docs WHERE vector_id=?').bind(id).first()).toBeNull(); + expect((await env.DB_FTS.prepare("SELECT COUNT(*) AS n FROM doc_sparse_fts_v1 WHERE doc_sparse_fts_v1 MATCH 'teardownneedle'").first<{ n: number }>())!.n).toBe(0); + await env.DB_FTS.prepare("INSERT INTO doc_sparse_fts_v1(doc_sparse_fts_v1,rank) VALUES('integrity-check',1)").run(); + }); + + it('indexes the tail independently, remains one row, leaves prefix candidates unchanged', async () => { + const body = 'prefix '.repeat(1500) + 'tailneedle'; + const { record } = await seed('sparse/tail', 'sparse-tail', body); + const snapshot = await prefix('sparse-tail'); + const source = await prepareSparseSource('sparse-tail', 'doc', 'a.md', body, record.blobSha, snapshot.indexed_at); + await sparseSourceStatement(env.DB_FTS, source, snapshot).all(); + await sparseSourceStatement(env.DB_FTS, source, snapshot).all(); + expect((await queryFts(env.DB_FTS, 'tailneedle', 10, { repo: 'sparse/tail' })).length).toBe(0); + expect((await env.DB_FTS.prepare("SELECT COUNT(*) AS n FROM doc_sparse_fts_v1 WHERE doc_sparse_fts_v1 MATCH 'tailneedle'").first<{ n: number }>())!.n).toBe(1); + expect((await owner('sparse-tail'))?.raw).toContain('tailneedle'); + expect((await docSparseStats(env.DB_FTS, { repo: 'sparse/tail', type: 'doc' })).rows[0]).toMatchObject({ coverage: 'full', count: 1 }); + }); + + it('rejects stale snapshots after prefix update/delete and invalidates the old full row', async () => { + const { row, record } = await seed('sparse/cas', 'sparse-cas', 'oldterm'); + const old = await prefix('sparse-cas'); + const source = await prepareSparseSource(row.vectorId, 'doc', 'a.md', 'oldterm', record.blobSha, old.indexed_at); + await sparseSourceStatement(env.DB_FTS, source, old).all(); + await upsertFtsRow(env.DB_FTS, { ...row, content: 'a.md\n\nnewterm', sourceVersion: await gitBlobSha('newterm') }); + expect(await owner(row.vectorId)).toBeNull(); + expect((await sparseSourceStatement(env.DB_FTS, source, old).all()).results).toEqual([]); + await deleteFtsRow(env.DB_FTS, row.vectorId); + expect((await sparseSourceStatement(env.DB_FTS, source, old).all()).results).toEqual([]); + }); + + it('rolls prefix changes back when the same D1 batch fails', async () => { + const { row } = await seed('sparse/atomic', 'sparse-atomic', 'before'); + await expect(env.DB_FTS.batch([prepareFtsRow(env.DB_FTS, { ...row, content: 'after' }), + env.DB_FTS.prepare('INSERT INTO table_that_does_not_exist VALUES(1)')])).rejects.toThrow(); + expect((await prefix(row.vectorId)).content).toBe('a.md\n\nbefore'); + }); + + it('keeps out-of-range owners out of FTS and removes owners on repo-wide delete', async () => { + const body = 'x'.repeat(MAX_RAW_BYTES); + const { row, record } = await seed('sparse/large', 'sparse-large', body); + const snap = await prefix(row.vectorId); + await sparseSourceStatement(env.DB_FTS, await prepareSparseSource(row.vectorId, 'doc', 'a.md', body, record.blobSha, snap.indexed_at), snap).all(); + expect((await owner(row.vectorId))?.coverage).toBe('out_of_range'); + await env.DB_FTS.prepare('DELETE FROM search_docs WHERE repo=?').bind('sparse/large').run(); + expect(await owner(row.vectorId)).toBeNull(); + await env.DB_FTS.prepare("INSERT INTO doc_sparse_fts_v1(doc_sparse_fts_v1,rank) VALUES('integrity-check',1)").run(); + }); + + it('writes full doc/wiki and advances store only after D1 success; default off adds no owner', async () => { + const live = runtime(); + const docBody = 'document'; + expect((await processAndUpsertDoc(live, store(), 'sparse/ingest', 'a.md', docBody, await gitBlobSha(docBody))).embedded).toBe(true); + expect((await processAndUpsertWikiDoc(live, store(), 'sparse/ingest', 'Home', 'md', 'wiki text')).embedded).toBe(true); + const ids = (await env.DB_FTS.prepare('SELECT vector_id FROM search_docs WHERE repo=?').bind('sparse/ingest').all<{ vector_id: string }>()).results; + for (const id of ids) expect((await owner(id.vector_id))?.coverage).toBe('full'); + const bad = { ...live, DB_FTS: { prepare: env.DB_FTS.prepare.bind(env.DB_FTS), batch: async () => { throw new Error('full write failed'); } } } as unknown as Env; + const writes: Request[] = []; + const fakeStore = { fetch: async (r: Request) => { writes.push(r); return new Response('ok'); } } as unknown as DurableObjectStub; + expect((await processAndUpsertDoc(bad, fakeStore, 'sparse/failed', 'a.md', docBody, await gitBlobSha(docBody))).failed).toBe(true); + expect((await processAndUpsertWikiDoc(bad, fakeStore, 'sparse/failed', 'Home', 'md', docBody)).failed).toBe(true); + expect(writes).toHaveLength(0); + const disabled = runtime(false); + await processAndUpsertDoc(disabled, store(), 'sparse/off', 'a.md', docBody, 'standin'); + expect((await env.DB_FTS.prepare('SELECT COUNT(*) AS n FROM doc_sparse_sources s JOIN search_docs d USING(vector_id) WHERE d.repo=?').bind('sparse/off').first<{ n: number }>())!.n).toBe(0); + expect(disabled.VECTORIZE.upsert).toHaveBeenCalledTimes(1); + }); +}); + +describe('bounded operator backfill', () => { + it('resumes a source deferred by the remaining byte budget rather than skipping it', async () => { + const first = await seed('sparse/byte-resume', 'sparse-byte-first', 'abcdef', false, 'first.md'); + const second = await seed('sparse/byte-resume', 'sparse-byte-second', 'uvwxyz', false, 'second.md'); + vi.stubGlobal('fetch', vi.fn(async (url: string) => { + const one = url.endsWith(first.record.blobSha); + return Response.json({ sha: one ? first.record.blobSha : second.record.blobSha, + encoding: 'base64', content: btoa(one ? 'abcdef' : 'uvwxyz') }); + })); + const opts = { repo: first.record.repo, type: 'doc' as const, apply: true, maxBytes: 10 }; + const limited = await backfillDocSparse(runtime(false), opts); + expect(limited.results.map(r => r.status)).toEqual(['success', 'retryable_error']); + expect(limited.results[1].reason).toBe('byte_budget_deferred'); + expect(await owner(second.row.vectorId)).toBeNull(); + const resumed = await backfillDocSparse(runtime(false), { ...opts, cursor: limited.next_cursor! }); + expect(resumed.results[0].status).toBe('success'); expect(resumed.next_cursor).toBeNull(); + expect(await owner(second.row.vectorId)).not.toBeNull(); + }); + + it('requires operator authentication and rejects invalid/broad scopes', async () => { + const live = { ...runtime(), GITHUB_TOKEN: 'test-operator' } as Env; + const endpoint = 'http://test/admin/backfill-doc-sparse?repo=sparse/auth&type=doc'; + expect((await handleDocSparseAdmin(new Request(endpoint, { method: 'POST' }), live)).status).toBe(401); + for (const suffix of ['&limit=11', '&max_bytes=1048577', '&cursor=-1', '&apply=yes']) { + expect((await handleDocSparseAdmin(new Request(endpoint + suffix, { method: 'POST', headers: { GITHUB_TOKEN: 'test-operator' } }), live)).status).toBe(400); + } + const planned = await handleDocSparseAdmin(new Request(endpoint, { method: 'POST', headers: { GITHUB_TOKEN: 'test-operator' } }), live); + expect(planned.status).toBe(200); expect(await planned.json()).toMatchObject({ mode: 'plan', scanned: 0 }); + }); + + it('plans without writes, applies a legacy prefix without embedding, and skips a rerun', async () => { + const { record } = await seed('sparse/backfill', 'sparse-backfill', 'backfill body', true); + const live = runtime(false); mockBlob('backfill body', record.blobSha); + const opts = { repo: record.repo, type: 'doc' as const }; + expect((await backfillDocSparse(live, opts)).results[0].status).toBe('success'); + expect(await owner('sparse-backfill')).toBeNull(); + expect((await backfillDocSparse(live, { ...opts, apply: true })).results[0].status).toBe('success'); + expect((await backfillDocSparse(live, { ...opts, apply: true })).results[0].status).toBe('unchanged'); + expect(live.AI.run).not.toHaveBeenCalled(); expect(live.VECTORIZE.upsert).not.toHaveBeenCalled(); + }); + + it('reports a concurrent update as conflict without overwriting it', async () => { + const { row, record } = await seed('sparse/backfill-race', 'sparse-backfill-race', 'old'); + mockBlob('old', record.blobSha, async () => { + await upsertFtsRow(env.DB_FTS, { ...row, content: 'a.md\n\nnew', sourceVersion: await gitBlobSha('new') }); + }); + const result = await backfillDocSparse(runtime(false), { repo: record.repo, type: 'doc', apply: true }); + expect(result.results[0]).toMatchObject({ status: 'conflict', reason: 'prefix_changed' }); + expect(await owner(row.vectorId)).toBeNull(); + }); + + it('retains the cursor at retryable errors and resumes from the same source', async () => { + const { record } = await seed('sparse/retry', 'sparse-retry', 'retry'); + vi.stubGlobal('fetch', vi.fn(async () => { throw new Error('network'); })); + const opts = { repo: record.repo, type: 'doc' as const, apply: true, limit: 1 }; + const failed = await backfillDocSparse(runtime(false), opts); + expect(failed.next_cursor).toBe(0); expect(failed.results[0].status).toBe('retryable_error'); + mockBlob('retry', record.blobSha); + const repaired = await backfillDocSparse(runtime(false), { ...opts, cursor: failed.next_cursor! }); + expect(repaired.results[0].status).toBe('success'); + expect((await backfillDocSparse(runtime(false), { ...opts, cursor: repaired.next_cursor! })).next_cursor).toBeNull(); + }); + + it('does not invent wiki history or hashes when the bounded reader cannot acquire it', async () => { + const live = runtime(); + await processAndUpsertWikiDoc(runtime(false), store(), 'sparse/wiki-conflict', 'Home', 'md', 'past'); + vi.stubGlobal('fetch', vi.fn(async () => new Response('current'))); + const result = await backfillDocSparse(live, { repo: 'sparse/wiki-conflict', type: 'wiki_doc', apply: true }); + expect(result.results[0].status).toBe('conflict'); + vi.stubGlobal('fetch', vi.fn(async () => new Response('x'.repeat(100)))); + const large = await backfillDocSparse(live, { repo: 'sparse/wiki-conflict', type: 'wiki_doc', maxBytes: 10, apply: true }); + expect(large.results[0]).toMatchObject({ status: 'conflict', reason: 'source_byte_limit_unverified' }); + expect(large.results[0].raw_bytes).toBeUndefined(); + expect(large.source_bytes).toBe(0); expect(large.budget_exhausted).toBe(true); + expect((await docSparseStats(env.DB_FTS, { repo: 'sparse/wiki-conflict', type: 'wiki_doc' })).rows[0]).toMatchObject({ coverage: 'unprepared' }); + }); +}); diff --git a/src/fts.ts b/src/fts.ts index f5f6710..b73ae9f 100644 --- a/src/fts.ts +++ b/src/fts.ts @@ -64,6 +64,7 @@ export interface FtsUpsertRow { commitAuthor?: string; commentId?: number; reviewId?: number; + sourceVersion?: string; // doc blob SHA / wiki body SHA-256; never queried by default retrieval content: string; // tokenizable text (title+body or commit msg + path + patch) } @@ -114,16 +115,21 @@ export async function upsertFtsRow( db: D1Database, row: FtsUpsertRow, ): Promise { - const tokenizerKind: TokenizerKind = tokenizerKindForType(row.type); - const now = new Date().toISOString(); + await prepareFtsRow(db, row).run(); +} - await db +export function prepareFtsRow(db: D1Database, row: FtsUpsertRow, now = new Date().toISOString()): D1PreparedStatement { + const tokenizerKind: TokenizerKind = tokenizerKindForType(row.type); + const versionColumn = row.sourceVersion === undefined ? "" : ", source_version"; + const versionValue = row.sourceVersion === undefined ? "" : ", ?"; + const versionUpdate = row.sourceVersion === undefined ? "" : ", source_version = excluded.source_version"; + return db .prepare( `INSERT INTO search_docs ( vector_id, repo, type, state, labels, milestone, assignees, updated_at, number, tag_name, doc_path, commit_sha, file_path, file_status, - commit_date, commit_author, comment_id, review_id, tokenizer_kind, content, content_fts, indexed_at - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + commit_date, commit_author, comment_id, review_id, tokenizer_kind, content, content_fts, indexed_at${versionColumn} + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?${versionValue}) ON CONFLICT (vector_id) DO UPDATE SET repo = excluded.repo, type = excluded.type, @@ -145,7 +151,7 @@ export async function upsertFtsRow( tokenizer_kind = excluded.tokenizer_kind, content = excluded.content, content_fts = excluded.content_fts, - indexed_at = excluded.indexed_at`, + indexed_at = excluded.indexed_at${versionUpdate}`, ) .bind( row.vectorId, @@ -170,8 +176,8 @@ export async function upsertFtsRow( row.content, ftsIndexText(row.content, tokenizerKind), now, - ) - .run(); + ...(row.sourceVersion === undefined ? [] : [row.sourceVersion]), + ); } /** Outcome of one `backfillNatSegments` batch. */ diff --git a/src/index.ts b/src/index.ts index 008bf01..78eb351 100644 --- a/src/index.ts +++ b/src/index.ts @@ -31,6 +31,7 @@ */ import { backfillSourceIdentities } from "./backfill-source-identities.js"; +import { handleDocSparseAdmin } from "./backfill-doc-sparse.js"; import type { Env } from "./types.js"; import { createOAuthProvider, @@ -114,6 +115,10 @@ const innerHandler: ExportedHandler = { async fetch(request: Request, env: Env, ctx: ExecutionContext): Promise { const url = new URL(request.url); + if (request.method === "POST" && ["/admin/backfill-doc-sparse", "/admin/doc-sparse-stats"].includes(url.pathname)) { + return handleDocSparseAdmin(request, env); + } + // -- GitHub webhook receiver (IP allowlist + signature verification) -- if (request.method === "POST" && url.pathname === "/webhooks/github") { return handleWebhook(request, env); diff --git a/src/pipeline/embed-doc.ts b/src/pipeline/embed-doc.ts index 2b6bfc7..2471a8f 100644 --- a/src/pipeline/embed-doc.ts +++ b/src/pipeline/embed-doc.ts @@ -7,8 +7,9 @@ * column reuse pattern. */ +import { docSparseEnabled, prepareSparseSource, sparseSourceStatement } from "../doc-sparse.js"; import type { Env, DocRecord, WikiDocRecord } from "../types.js"; -import { upsertFtsRow } from "../fts.js"; +import { upsertFtsRow, prepareFtsRow, type FtsUpsertRow } from "../fts.js"; import { indexWikiEdges } from "../graph.js"; import { prepareEmbeddingInput, sha256Hex } from "./hash.js"; import { generateEmbedding } from "./embedding.js"; @@ -70,7 +71,7 @@ export async function processAndUpsertDoc( // Mirror into D1 FTS5 sparse index. // Not caught here: a failed FTS write must fail the whole upsert so the store // keeps the old version and the next poll retries the page (issue #282). - await upsertFtsRow(env.DB_FTS, { + await writeDocFts(env, { vectorId: dvid, repo, type: "doc", @@ -81,7 +82,8 @@ export async function processAndUpsertDoc( updatedAt: now, docPath: path, content: embeddingInput, - }); + sourceVersion: blobSha, + }, path, content); // Upsert doc record into store const record: DocRecord = { @@ -138,6 +140,7 @@ export async function processAndUpsertWikiDoc( const now = new Date().toISOString(); try { + const contentHash = await sha256Hex(content); // Generate embedding (use page name as title surrogate, content as body) const embeddingInput = prepareEmbeddingInput(pageName, content); const embedding = await generateEmbedding(env.AI, embeddingInput); @@ -169,7 +172,7 @@ export async function processAndUpsertWikiDoc( // from" field, distinguished by the row's `type='wiki_doc'`. // Not caught here: a failed FTS write must fail the whole upsert so the store // keeps the old version and the next poll retries the page (issue #282). - await upsertFtsRow(env.DB_FTS, { + await writeDocFts(env, { vectorId: wvid, repo, type: "wiki_doc", @@ -180,7 +183,8 @@ export async function processAndUpsertWikiDoc( updatedAt: now, docPath: pageName, content: embeddingInput, - }); + sourceVersion: contentHash, + }, pageName, content); // Extract + index graph "mention" edges to other wiki pages in this repo. // Additive: the default retrieval path never reads these. Failures here must @@ -194,7 +198,6 @@ export async function processAndUpsertWikiDoc( ); } - const contentHash = await sha256Hex(content); const record: WikiDocRecord = { repo, pageName, @@ -220,3 +223,21 @@ export async function processAndUpsertWikiDoc( return { embedded: false, skippedUnchanged: false, metadataUpdated: false, failed: true }; } } + +async function writeDocFts(env: Env, row: FtsUpsertRow, path: string, body: string): Promise { + if (!docSparseEnabled(env)) { + // Preserve the pre-0009 SQL shape as well as the default-off request count. + await upsertFtsRow(env.DB_FTS, { ...row, sourceVersion: undefined }); + return; + } + const indexedAt = new Date().toISOString(); + const source = await prepareSparseSource(row.vectorId, row.type as 'doc' | 'wiki_doc', path, body, row.sourceVersion!, indexedAt); + const result = await env.DB_FTS.batch([ + prepareFtsRow(env.DB_FTS, row, indexedAt), + sparseSourceStatement(env.DB_FTS, source, { + vector_id: row.vectorId, repo: row.repo, type: source.type, doc_path: path, + source_version: row.sourceVersion!, content: row.content, updated_at: row.updatedAt, indexed_at: indexedAt, + }), + ]); + if (!result[1]?.results?.length) throw new Error('Full sparse owner write did not match prefix'); +} diff --git a/src/poller.ts b/src/poller.ts index c8c3f89..064f726 100644 --- a/src/poller.ts +++ b/src/poller.ts @@ -1,3 +1,5 @@ +import { docSparseEnabled } from "./doc-sparse.js"; +import { fetchDocBlob } from "./backfill-doc-sparse.js"; /** * Cron Poller — scheduled handler for GitHub issue/PR polling and embedding pipeline. * @@ -1141,7 +1143,9 @@ export async function pollDocs( try { // Fetch file content fetchesIssued++; - const content = await fetchFileContent(repo, entry.path, env.GITHUB_TOKEN); + const content = docSparseEnabled(env) + ? await fetchDocBlob(repo, entry.sha, env.GITHUB_TOKEN) + : await fetchFileContent(repo, entry.path, env.GITHUB_TOKEN); const result = await processAndUpsertDoc(env, storeStub, repo, entry.path, content, entry.sha); diff --git a/src/test-env.d.ts b/src/test-env.d.ts index c87dfd3..28a0747 100644 --- a/src/test-env.d.ts +++ b/src/test-env.d.ts @@ -14,6 +14,7 @@ declare module "cloudflare:test" { interface ProvidedEnv { DB_FTS: D1Database; DB_FTS_MIGRATION: D1Database; + DB_FTS_DOC_LEGACY: D1Database; TEST_MIGRATIONS: TestD1Migration[]; ISSUE_STORE: DurableObjectNamespace; } diff --git a/src/types.ts b/src/types.ts index 1bb20a9..f6e07a6 100644 --- a/src/types.ts +++ b/src/types.ts @@ -226,6 +226,8 @@ export interface Env extends OAuthEnv { VECTORIZE: Vectorize; /** D1 database for full-text search (sparse side of hybrid retrieval, BM25 via FTS5) */ DB_FTS: D1Database; + /** Full doc/wiki sparse ownership; enabled only by the literal "true", default off. */ + DOC_SPARSE_CONTEXT_ENABLED?: string; AI: Ai; /** GitHub personal access token or installation token for API access (set via `wrangler secret put`) */ GITHUB_TOKEN: string; diff --git a/src/webhook.ts b/src/webhook.ts index 94234c3..988047d 100644 --- a/src/webhook.ts +++ b/src/webhook.ts @@ -1,3 +1,4 @@ +import { docSparseEnabled, gitBlobSha } from "./doc-sparse.js"; /** * GitHub webhook receiver — signature verification, IP allowlist, and event routing. * @@ -472,10 +473,11 @@ export async function handlePushEvent( try { const content = await fetchFileContent(repo, path, env.GITHUB_TOKEN); - // Use HEAD commit SHA as blob SHA stand-in. The pipeline's - // processAndUpsertDoc always embeds (caller decides freshness), - // so any unique-ish value works for the record. - const headSha = (payload.head_commit as { id: string } | null)?.id ?? "unknown"; + // Enabled ownership records the actual Git blob version of the fetched text. + // Default-off ingest keeps its existing HEAD commit stand-in. + const headSha = docSparseEnabled(env) + ? await gitBlobSha(content) + : (payload.head_commit as { id: string } | null)?.id ?? "unknown"; const result = await processAndUpsertDoc(env, storeStub, repo, path, content, headSha); if (result.embedded) embedded++; diff --git a/vitest.workers.config.ts b/vitest.workers.config.ts index 32b9045..dd7f73b 100644 --- a/vitest.workers.config.ts +++ b/vitest.workers.config.ts @@ -26,6 +26,7 @@ export default defineConfig(async () => { d1Databases: { DB_FTS: "test-fts", DB_FTS_MIGRATION: "test-fts-migration", + DB_FTS_DOC_LEGACY: "test-fts-doc-legacy", }, durableObjects: { ISSUE_STORE: { className: "IssueStore", useSQLite: true }, diff --git a/wrangler.toml b/wrangler.toml index 7ffa67f..42a4c2b 100644 --- a/wrangler.toml +++ b/wrangler.toml @@ -68,6 +68,8 @@ crons = ["0 * * * *", "15 * * * *", "30 * * * *", "45 * * * *"] # Environment variables [vars] +# Default-off ingest works before 0009. Migration and bounded preparation precede enable. +DOC_SPARSE_CONTEXT_ENABLED = "false" POLL_REPOS = "Liplus-Project/github-webhook-mcp,Liplus-Project/github-rag-mcp,Liplus-Project/liplus-desktop,Liplus-Project/liplus-language,Liplus-Project/dipper_ai,Liplus-Project/neuron-graph-rag,Liplus-Project/pullcept" [observability] From 84dd9a25001ebf12a25b8a1921116f4f37b0e5df Mon Sep 17 00:00:00 2001 From: Codex Luna & Lulu Date: Tue, 13 Oct 2026 18:39:11 +0900 Subject: [PATCH 2/3] docs(index): clarify D1 recovery boundary for sparse rollout MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Codex Luna Time Travel の全 D1 復旧と書込み静止、DO/Vectorize 非対象の境界を一次資料に結び、export 失敗と restore 未実施を維持する。#284 の operator が部分索引の fallback と全 DB recovery を混同しないため。 --- docs/6-full-sparse-sources.ja.md | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/docs/6-full-sparse-sources.ja.md b/docs/6-full-sparse-sources.ja.md index 50bfa6f..531f303 100644 --- a/docs/6-full-sparse-sources.ja.md +++ b/docs/6-full-sparse-sources.ja.md @@ -18,4 +18,6 @@ raw 256 KiB、segmented 512 KiB、owner 行の文字列 UTF-8 bytes と固定 ov `POST /admin/doc-sparse-stats?repo=owner/repo&type=doc|wiki_doc` は indexed prefix population の coverage、未準備、版不一致、legacy prefix の版未検証件数、raw/segmented/行 bytes と索引版別件数を返す。これは repo tree 全体や現在の Git source 本体の件数ではない。 -本番 apply は merge 後の別 operator 判断。対象 repo/type と plan、復旧境界を確認して migration/apply を行う。2026-10-13 の operator 確認では、全文 D1 export は FTS/virtual table 未対応で失敗しバックアップファイルは作れなかった。Time Travel info は成功し bookmark を控えたが、restore は実行していない。migration/apply 前には Time Travel info を再確認し bookmark と影響対象の metadata を控える。全 repo 一括取得は行わず cursor ごとに観察する。migration は旧 prefix/FTS/vector を変更しない。通常の fallback は enable を false に戻して全文書込みを止める。既定検索と prefix ingest は継続できる。DB 復旧は別操作であり、info で確認した Time Travel bookmark と restore の影響範囲を operator が apply 前に確認する。全 DB の restore は他 surface にも影響する別判断であり、小さい全文 backfill の通常 fallback と混同しない。空の復旧用 D1 への import/binding 切替は、完全なバックアップが得られ手順を検証した場合に限る。既存の populated D1 への export SQL import だけを復元保証にしない。この PR では DB 復元を試行・検証していない。既定 off の ingest・検索は migration 0009 が無くても使用できる。operator 経路と enable は migration 0009 が必要。報告は merge・配置・本番 plan・apply 観測を分ける。 +本番 apply は merge 後の別 operator 判断。対象 repo/type と plan、復旧境界を確認して migration/apply を行う。2026-10-13 の operator 確認では、全文 D1 export は FTS/virtual table 未対応で失敗しバックアップファイルは作れなかった。Time Travel info は成功し bookmark を控えたが、restore は実行していない。migration/apply 前には Time Travel info を再確認し bookmark と影響対象の metadata を控える。全 repo 一括取得は行わず cursor ごとに観察する。migration は旧 prefix/FTS/vector を変更しない。通常の fallback は enable を false に戻して全文書込みを止める。既定検索と prefix ingest は継続できる。DB 復旧は別操作であり、info で確認した Time Travel bookmark と restore の影響範囲を operator が apply 前に確認する。全 DB の restore は他 surface にも影響し、in-flight query/transaction を取り消す別判断であり、小さい全文 backfill の通常 fallback と混同しない。実行時は webhook/cron/admin の D1 書込みを静止させる。D1 Time Travel は DO/Vectorize を独立して復元するものではなく、復旧後の store 版・vector・prefix との整合を別途確認する。空の復旧用 D1 への import/binding 切替は、完全なバックアップが得られ手順を検証した場合に限る。既存の populated D1 への export SQL import だけを復元保証にしない。この PR では DB 復元を試行・検証していない。既定 off の ingest・検索は migration 0009 が無くても使用できる。operator 経路と enable は migration 0009 が必要。報告は merge・配置・本番 plan・apply 観測を分ける。 + +一次資料: [D1 Time Travel](https://developers.cloudflare.com/d1/reference/time-travel/)、[D1 import/export](https://developers.cloudflare.com/d1/best-practices/import-export-data/)。 From b1967d1df9efbb2c8521f1f62e13eaad7958da42 Mon Sep 17 00:00:00 2001 From: Codex Luna & Lulu Date: Tue, 13 Oct 2026 18:52:27 +0900 Subject: [PATCH 3/3] fix(index): reconcile sparse versions and isolate operator auth MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Codex Luna #284 の enable 前準備が既定 off 更新後の古い source_version で永久 conflict になる問題を修正する。store/hash/exact prefix 再検証後に source_revision CAS で版修復と全文 owner を同一 D1 batch に結び、同じ prefix/timestamp の更新や削除にも負けたら書かない。 新しい operator 2 経路は専用 DOC_SPARSE_ADMIN_TOKEN の Bearer 認証に分離し、未設定・空と GitHub credential fallback を拒否する。従来の source 取得と admin 認証は維持する。0009 前の default-off 互換、A→B の準備、競合、rollback、認証の実 D1 検証を追加した。 --- README.md | 2 + docs/0-requirements.ja.md | 2 + docs/0-requirements.md | 2 + docs/6-full-sparse-sources.ja.md | 8 +- docs/installation.ja.md | 2 + docs/installation.md | 2 + migrations/0009_doc_sparse_sources.sql | 15 ++- src/backfill-doc-sparse.ts | 44 ++++++--- src/doc-sparse.ts | 24 +++-- src/doc-sparse.workers.test.ts | 125 ++++++++++++++++++++++++- src/types.ts | 2 + wrangler.toml | 3 + 12 files changed, 206 insertions(+), 25 deletions(-) diff --git a/README.md b/README.md index 644c9b4..65e23c3 100644 --- a/README.md +++ b/README.md @@ -301,3 +301,5 @@ wrangler.toml Doc/wiki full sparse source preparation is a default-off operator foundation, independent of normal search. See the [storage contract and bounded plan/apply procedure](docs/6-full-sparse-sources.ja.md). Migration 0009 precedes operator preparation and enable; default-off ingest/search remains usable without it. + +The two new sparse operator endpoints require the dedicated `DOC_SPARSE_ADMIN_TOKEN` secret and `Authorization: Bearer `; they fail closed without it and do not accept GitHub credentials. Provision via `wrangler secret put DOC_SPARSE_ADMIN_TOKEN` after reviewed merge. diff --git a/docs/0-requirements.ja.md b/docs/0-requirements.ja.md index 36ff4c3..921986b 100644 --- a/docs/0-requirements.ja.md +++ b/docs/0-requirements.ja.md @@ -688,3 +688,5 @@ source の `docs/_Sidebar.md` は既存 Feedback Memory EN/JA ページを参照 ### Doc/wiki 独立全文 sparse source #284 は既定 off の保存・operator 基盤を追加する。[全文 sparse source 契約](6-full-sparse-sources.ja.md)に従い prefix corpus、既定検索、graph、vector 数を維持する。matched 検索は #281 の別作業。 + +新しい sparse operator の 2 経路は専用 `DOC_SPARSE_ADMIN_TOKEN` bearer secret で認証し、未設定・空は fail closed。GitHub source 取得は従来の `GITHUB_TOKEN` を使う。既定 off の更新を含め、prefix revision CAS により版修復と owner 保存を同一 D1 batch に結ぶ。 diff --git a/docs/0-requirements.md b/docs/0-requirements.md index 18ad7fa..709ddf9 100644 --- a/docs/0-requirements.md +++ b/docs/0-requirements.md @@ -693,3 +693,5 @@ The source `docs/_Sidebar.md` must link the existing Feedback Memory EN/JA pages ### Independent full sparse doc/wiki sources Issue #284 adds a default-off storage/operator foundation; its contract is [Full sparse sources](6-full-sparse-sources.ja.md). It preserves the prefix corpus, default retrieval, graph and vector count. Matched retrieval remains #281. + +The new sparse operator endpoints require the dedicated `DOC_SPARSE_ADMIN_TOKEN` bearer secret and fail closed without it; GitHub source fetching retains `GITHUB_TOKEN`. Version repair and owner writes share a D1 batch guarded by the prefix revision, including default-off updates. diff --git a/docs/6-full-sparse-sources.ja.md b/docs/6-full-sparse-sources.ja.md index 531f303..e9623f6 100644 --- a/docs/6-full-sparse-sources.ja.md +++ b/docs/6-full-sparse-sources.ja.md @@ -8,11 +8,15 @@ Issue #284 の保存・運用契約。通常検索への接続は #281 の別作 raw 256 KiB、segmented 512 KiB、owner 行の文字列 UTF-8 bytes と固定 overhead の合計 1 MiB を上限とする。超過は `out_of_range` として処理済みで prefix-only。全文/hash を取得できなかった場合は null とし、取得済みとしない。同じ版は skip、次の版または operator の `force=true` で再判定する。予算は CPU/heap の安全証明ではない。 -有効 ingest は doc の blob hash を検証し、prefix と全文を D1 batch transaction で書いてから DO の版を進める。FTS 失敗は版を進めず、既存の ETag 保持・再試行に従う。prefix 更新は旧 owner を無効化し、削除 trigger は owner/独立 FTS を削除する。prefix 更新と backfill の競合は source_version、prefix 本文・type・repo・path・updated_at・indexed_at の compare-and-swap で上書きを防ぐ。 +有効 ingest は doc の blob hash を検証し、prefix と全文を D1 batch transaction で書いてから DO の版を進める。FTS 失敗は版を進めず、既存の ETag 保持・再試行に従う。prefix 更新は旧 owner を無効化し、削除 trigger は owner/独立 FTS を削除する。prefix 更新と backfill の競合は source_version、prefix 本文・type・repo・path・updated_at・indexed_at・source_revision の compare-and-swap で上書きを防ぐ。 + +migration の source_revision は prefix の source に関わる UPDATE ごとにちょうど 1 増える。既定 off の旧 SQL 形を保ち、同じ prefix・timestamp・古い source_version が残る更新も識別する。source_revision 自体の UPDATE は generation trigger を再発火させない。owner の prefix_revision は実際の prefix generation を所有し、統計と同版 skip もこの一致を確認する。 + +既定 off で A→B が取り込まれると source_version は A のままになり得る。backfill はこれだけで永久 conflict にせず、store.updatedAt と prefix、取得した doc blob/wiki body の hash と exact prefix、store の再読をすべて確認する。plan は版修復の予定だけを返し書かない。apply は source_revision を含む snapshot CAS で canonical source_version を B へ修復し、その UPDATE の成功だけを条件に owner 保存を同一 D1 batch で行う。修復は revision を 1 増やし旧 owner を無効化してから新 owner を結ぶ。CAS が負ければ両方 0 行、書込み失敗なら版修復も rollback。legacy の空版もこの手順で結ぶ。embedding/vector は呼ばない。 ## Operator -認証ヘッダは既存 `GITHUB_TOKEN`。`POST /admin/backfill-doc-sparse?repo=owner/repo&type=doc|wiki_doc` は plan が既定。書込みは `apply=true` のみ。`limit` は既定 5、最大 10、`max_bytes` は既定/最大 1 MiB、`cursor` は search_docs rowid。1 回の source 取得も bounded reader で制限する。doc は DO の blobSHA の Git Blob API から取得・hash 検証し、wiki は DO の extension の raw 本文を取得して bodySHA を照合する。過去 wiki を hash から取得できるとは仮定しない。store と prefix の版が異なるものや検証できないものは conflict、再取得候補として返す。embedding/Vectorize を呼ばず、unchanged/ETag を回避して準備する。 +この 2 経路だけは専用 secret `DOC_SPARSE_ADMIN_TOKEN` を使い、`Authorization: Bearer ` が完全一致した時だけ認証する。未設定・空なら fail closed。`GITHUB_TOKEN` への fallback はなく、GitHub source の取得だけは従来の `GITHUB_TOKEN` を使う。既存の他 admin 経路は変更しない。reviewed merge 後、operator が `wrangler secret put DOC_SPARSE_ADMIN_TOKEN` で secret を provision する。値は [vars]・repository・chat に置かない。`POST /admin/backfill-doc-sparse?repo=owner/repo&type=doc|wiki_doc` は plan が既定。書込みは `apply=true` のみ。`limit` は既定 5、最大 10、`max_bytes` は既定/最大 1 MiB、`cursor` は search_docs rowid。1 回の source 取得も bounded reader で制限する。doc は DO の blobSHA の Git Blob API から取得・hash 検証し、wiki は DO の extension の raw 本文を取得して bodySHA を照合する。過去 wiki を hash から取得できるとは仮定しない。store と prefix の版が異なるものや検証できないものは conflict、再取得候補として返す。embedding/Vectorize を呼ばず、unchanged/ETag を回避して準備する。 応答は対象 repo/type、取得時刻、各 source の版、coverage/bytes、成功・範囲外・欠損・競合・retryable error、次 cursor を持つ。retryable error または取得前に残り byte 予算が無い場合、cursor はその source を越えない。残り予算だけで取得が打ち切られた source は `byte_budget_deferred` として cursor を保持し、次の call で再試行する。call の初期予算でも取得上限で原文/hash を確認できなかった source は `source_byte_limit_unverified` conflict として進め、残りの取得を止める。`source_bytes` は取得・decode が完了した body の UTF-8 bytes、`budget_exhausted` は未取得分の上限到達も区別する。Git Blob JSON の network reader は残り decoded 予算 × 1.4 + 4096 bytes（base64 と envelope）で制限する。missing/conflict は明示報告して次へ進むため、競合解消後は cursor を戻して再実行する。plan は source を取得・検証するが保存しない。apply は同じ source の再実行で owner/FTS/vector を増やさない。 diff --git a/docs/installation.ja.md b/docs/installation.ja.md index 767ebef..0eaf721 100644 --- a/docs/installation.ja.md +++ b/docs/installation.ja.md @@ -511,3 +511,5 @@ GitHub App の callback URL が worker 側 URL と完全一致していない。 ## Doc/wiki 全文 sparse 準備 operator 経路と enable より先に migration 0009 を適用し（既定 off の ingest・検索は新 schema 不要）、準備中は `DOC_SPARSE_CONTEXT_ENABLED="false"` を保つ。認証済み `POST /admin/backfill-doc-sparse?repo=owner/repo&type=doc` は plan が既定。計画と Time Travel bookmark と復旧の影響対象を確認した後（2026-10-13 の full export は FTS/virtual table 未対応で失敗、restore 未実行）だけ `apply=true` を指定する。1 回は最大 10 資料・decoded source 1 MiB、返る `next_cursor` を `cursor` に渡し、競合・error を確認する。wiki は `type=wiki_doc` で別処理。`POST /admin/doc-sparse-stats?repo=owner/repo&type=doc` で索引済み prefix owner の coverage を観察する。[完全な契約](6-full-sparse-sources.ja.md)に予算と rollout 境界を固定している。embedding と検索応答は変わらない。 + +reviewed merge 後、この 2 経路の呼出しより先に `wrangler secret put DOC_SPARSE_ADMIN_TOKEN` で専用 secret を provision し、`Authorization: Bearer ` を送る。未設定・空 secret、`GITHUB_TOKEN` ヘッダ、GitHub bearer credential は認証に使わない。source 取得だけは従来の `GITHUB_TOKEN` を使い、既存 admin 経路は変えない。secret の値を設定ファイルや chat に置かない。 diff --git a/docs/installation.md b/docs/installation.md index 2bc7394..a9a4ce6 100644 --- a/docs/installation.md +++ b/docs/installation.md @@ -511,3 +511,5 @@ The GitHub App callback URL does not exactly match the worker callback URL. ## Full sparse doc/wiki preparation Migration 0009 must precede these operator endpoints and enable. Default-off ingest and search need no new schema. Keep `DOC_SPARSE_CONTEXT_ENABLED="false"` while preparing. The authenticated `POST /admin/backfill-doc-sparse?repo=owner/repo&type=doc` endpoint defaults to plan; use `apply=true` only after reading the plan and confirming the Time Travel bookmark/recovery scope (full export failed on FTS/virtual tables in the 2026-10-13 operator check; no restore was executed). Process at most 10 sources and 1 MiB decoded source bytes per call, pass the returned `next_cursor` as `cursor`, and examine conflicts/errors before continuing. Use `type=wiki_doc` separately. `POST /admin/doc-sparse-stats?repo=owner/repo&type=doc` reports coverage of indexed prefix owners. See [the complete contract](6-full-sparse-sources.ja.md) for limits and rollout boundaries. Neither operation embeds or changes retrieval. + +Provision `DOC_SPARSE_ADMIN_TOKEN` with `wrangler secret put DOC_SPARSE_ADMIN_TOKEN` after reviewed merge and before these operator calls. Send `Authorization: Bearer ` to both new endpoints. They reject unset/empty secrets, a `GITHUB_TOKEN` header, and GitHub bearer credentials. GitHub source acquisition still uses `GITHUB_TOKEN`; existing older admin authentication is unchanged. Keep secret values out of configuration files and chat. diff --git a/migrations/0009_doc_sparse_sources.sql b/migrations/0009_doc_sparse_sources.sql index 43310a7..1ef3084 100644 --- a/migrations/0009_doc_sparse_sources.sql +++ b/migrations/0009_doc_sparse_sources.sql @@ -1,11 +1,13 @@ -- Independent doc/wiki owner. Prefix FTS tables and their corpus stay unchanged. ALTER TABLE search_docs ADD COLUMN source_version TEXT NOT NULL DEFAULT ''; +ALTER TABLE search_docs ADD COLUMN source_revision INTEGER NOT NULL DEFAULT 0; CREATE TABLE doc_sparse_sources ( vector_id TEXT NOT NULL UNIQUE, source_type TEXT NOT NULL CHECK(source_type IN ('doc', 'wiki_doc')), source_version TEXT NOT NULL, prefix_indexed_at TEXT NOT NULL, + prefix_revision INTEGER NOT NULL, index_version TEXT NOT NULL, source_hash TEXT, body_hash TEXT, @@ -49,7 +51,18 @@ END; -- Even a default-off prefix update invalidates an old full snapshot. CREATE TRIGGER doc_sparse_prefix_update AFTER UPDATE ON search_docs BEGIN DELETE FROM doc_sparse_sources WHERE vector_id = old.vector_id - AND (old.source_version <> new.source_version OR old.content <> new.content + AND (old.source_revision <> new.source_revision OR old.source_version <> new.source_version OR old.content <> new.content OR old.indexed_at <> new.indexed_at OR old.updated_at <> new.updated_at OR old.repo <> new.repo OR old.type <> new.type OR old.doc_path <> new.doc_path); END; + +-- The default-off application keeps its pre-0009 SQL shape. This generation +-- therefore lives in the migrated DB, not in the application upsert. UPDATE OF +-- excludes source_revision itself, so the increment does not recurse. Even a +-- same-text/same-timestamp prefix update is observable to operator CAS. +CREATE TRIGGER doc_sparse_prefix_revision +AFTER UPDATE OF repo, type, doc_path, content, updated_at, indexed_at, source_version ON search_docs +WHEN old.type IN ('doc', 'wiki_doc') OR new.type IN ('doc', 'wiki_doc') +BEGIN + UPDATE search_docs SET source_revision = old.source_revision + 1 WHERE rowid = new.rowid; +END; diff --git a/src/backfill-doc-sparse.ts b/src/backfill-doc-sparse.ts index 3c4a7cb..067a453 100644 --- a/src/backfill-doc-sparse.ts +++ b/src/backfill-doc-sparse.ts @@ -3,11 +3,15 @@ import { prepareEmbeddingInput, sha256Hex } from './pipeline/hash.js'; import { DOC_SPARSE_INDEX_VERSION, gitBlobSha, prepareSparseSource, sparseSourceStatement, utf8Bytes, type PrefixSnapshot, + prefixSnapshotGuard, } from './doc-sparse.js'; export const MAX_BACKFILL_BYTES = 1024 * 1024; export async function handleDocSparseAdmin(request: Request, env: Env): Promise { - if (!env.GITHUB_TOKEN || request.headers.get('GITHUB_TOKEN') !== env.GITHUB_TOKEN) return new Response('Unauthorized', { status: 401 }); + if (!env.DOC_SPARSE_ADMIN_TOKEN?.trim() + || request.headers.get('Authorization') !== `Bearer ${env.DOC_SPARSE_ADMIN_TOKEN}`) { + return new Response('Unauthorized', { status: 401 }); + } const url = new URL(request.url); const opts: SparseBackfillOptions = { repo: url.searchParams.get('repo') ?? '', type: url.searchParams.get('type') as SparseBackfillOptions['type'], @@ -87,7 +91,7 @@ export async function fetchDocBlob(repo: string, sha: string, token: string, max return body; } -type SnapshotRow = PrefixSnapshot & { rid: number; prepared_version: string | null; prepared_index: string | null; prepared_at: string | null }; +type SnapshotRow = PrefixSnapshot & { rid: number; source_revision: number; prepared_version: string | null; prepared_index: string | null; prepared_at: string | null; prepared_revision: number | null }; type StoreRecord = DocRecord | WikiDocRecord; async function readStore(store: DurableObjectStub, prefix: PrefixSnapshot): Promise { const path = prefix.type === 'doc' ? 'doc' : 'wiki-doc'; @@ -105,13 +109,15 @@ export async function backfillDocSparse(env: Env, opts: SparseBackfillOptions) { const maxBytes = opts.maxBytes ?? MAX_BACKFILL_BYTES; const store = env.ISSUE_STORE.get(env.ISSUE_STORE.idFromName('global')); const rows = (await env.DB_FTS.prepare(`SELECT d.rowid AS rid, d.vector_id, d.repo, d.type, - d.doc_path, d.source_version, d.content, d.updated_at, d.indexed_at, - s.source_version AS prepared_version, s.index_version AS prepared_index, s.prefix_indexed_at AS prepared_at + d.doc_path, d.source_version, d.source_revision, d.content, d.updated_at, d.indexed_at, + s.source_version AS prepared_version, s.index_version AS prepared_index, s.prefix_indexed_at AS prepared_at, + s.prefix_revision AS prepared_revision FROM search_docs d LEFT JOIN doc_sparse_sources s ON s.vector_id=d.vector_id WHERE d.repo=? AND d.type=? AND d.rowid>? ORDER BY d.rowid LIMIT ?`) .bind(opts.repo, opts.type, cursor, limit).all()).results ?? []; const results: { vector_id: string; source_version: string | null; status: string; reason?: string; - coverage?: string; raw_bytes?: number | null; segmented_bytes?: number | null; row_bytes?: number }[] = []; + coverage?: string; raw_bytes?: number | null; segmented_bytes?: number | null; row_bytes?: number; + prefix_version_repair?: boolean; prefix_source_version?: string }[] = []; let bytes = 0; let nextCursor = cursor; let stopped = false; let budgetExhausted = false; for (const prefix of rows) { let version: string | null = null; @@ -122,12 +128,12 @@ export async function backfillDocSparse(env: Env, opts: SparseBackfillOptions) { nextCursor = prefix.rid; continue; } version = versionOf(record); - if (record.updatedAt !== prefix.updated_at || (prefix.source_version && prefix.source_version !== version)) { + if (record.updatedAt !== prefix.updated_at) { results.push({ vector_id: prefix.vector_id, source_version: version, status: 'conflict', reason: 'store_prefix_version' }); nextCursor = prefix.rid; continue; } - if (!opts.force && prefix.prepared_version === version && prefix.prepared_index === DOC_SPARSE_INDEX_VERSION - && prefix.prepared_at === prefix.indexed_at) { + if (!opts.force && prefix.source_version === version && prefix.prepared_version === version && prefix.prepared_index === DOC_SPARSE_INDEX_VERSION + && prefix.prepared_at === prefix.indexed_at && prefix.prepared_revision === prefix.source_revision) { results.push({ vector_id: prefix.vector_id, source_version: version, status: 'unchanged' }); nextCursor = prefix.rid; continue; } @@ -169,13 +175,28 @@ export async function backfillDocSparse(env: Env, opts: SparseBackfillOptions) { nextCursor = prefix.rid; continue; } if (opts.apply) { - const written = await sparseSourceStatement(env.DB_FTS, source, prefix).all(); + const guard = prefixSnapshotGuard(prefix); + const reconcile = prefix.source_version !== version; + const written = reconcile + ? (await env.DB_FTS.batch([ + env.DB_FTS.prepare(`UPDATE search_docs SET source_version=? WHERE ${guard.sql} RETURNING vector_id`) + .bind(version, ...guard.params), + sparseSourceStatement(env.DB_FTS, source, { ...prefix, source_version: version, source_revision: prefix.source_revision + 1 }, true), + ]))[1] + : await sparseSourceStatement(env.DB_FTS, source, prefix).all(); if (!written.results?.length) { results.push({ vector_id: prefix.vector_id, source_version: version, status: 'conflict', reason: 'prefix_changed' }); nextCursor = prefix.rid; continue; } + } else { + const guard = prefixSnapshotGuard(prefix); + if (!await env.DB_FTS.prepare(`SELECT vector_id FROM search_docs WHERE ${guard.sql}`).bind(...guard.params).first()) { + results.push({ vector_id: prefix.vector_id, source_version: version, status: 'conflict', reason: 'prefix_changed' }); + nextCursor = prefix.rid; continue; + } } results.push({ vector_id: prefix.vector_id, source_version: version, + prefix_version_repair: prefix.source_version !== version, prefix_source_version: prefix.source_version, status: source.coverage === 'full' ? 'success' : 'out_of_range', reason: source.reason, coverage: source.coverage, raw_bytes: source.rawBytes, segmented_bytes: source.segmentedBytes, row_bytes: source.rowBytes }); nextCursor = prefix.rid; @@ -199,13 +220,14 @@ export async function docSparseStats(db: D1Database, opts: Pick'' AND d.source_version<>s.source_version) OR d.indexed_at<>s.prefix_indexed_at THEN 'version_mismatch' + WHEN (d.source_version<>'' AND d.source_version<>s.source_version) OR d.indexed_at<>s.prefix_indexed_at + OR d.source_revision<>s.prefix_revision THEN 'version_mismatch' ELSE s.coverage END AS coverage, s.index_version, COUNT(*) AS count, SUM(COALESCE(s.raw_bytes,0)) AS raw_bytes, SUM(CASE WHEN s.vector_id IS NOT NULL AND d.source_version='' THEN 1 ELSE 0 END) AS legacy_version_unverified, SUM(COALESCE(s.segmented_bytes,0)) AS segmented_bytes, SUM(COALESCE(s.row_bytes,0)) AS row_bytes FROM search_docs d LEFT JOIN doc_sparse_sources s ON s.vector_id=d.vector_id - WHERE d.repo=? AND d.type=? GROUP BY coverage,s.index_version`) + WHERE d.repo=? AND d.type=? GROUP BY 1,2`) .bind(opts.repo, opts.type).all()).results ?? []; return { repo: opts.repo, type: opts.type, observed_at: new Date().toISOString(), population: 'indexed_prefix_owners', source_count: rows.reduce((n, row) => n + Number(row.count), 0), rows }; diff --git a/src/doc-sparse.ts b/src/doc-sparse.ts index f5779a2..dd48a1a 100644 --- a/src/doc-sparse.ts +++ b/src/doc-sparse.ts @@ -65,21 +65,32 @@ export async function prepareSparseSource( export interface PrefixSnapshot { vector_id: string; repo: string; type: SparseSource['type']; doc_path: string; source_version: string; content: string; updated_at: string; indexed_at: string; + source_revision?: number; +} + +export function prefixSnapshotGuard(prefix: PrefixSnapshot): { sql: string; params: (string | number)[] } { + return { + sql: `vector_id=? AND repo=? AND type=? AND doc_path=? AND source_version=? + AND content=? AND updated_at=? AND indexed_at=?${prefix.source_revision === undefined ? '' : ' AND source_revision=?'}`, + params: [prefix.vector_id, prefix.repo, prefix.type, prefix.doc_path, prefix.source_version, + prefix.content, prefix.updated_at, prefix.indexed_at, ...(prefix.source_revision === undefined ? [] : [prefix.source_revision])], + }; } /** INSERT SELECT is the CAS: a deleted/changed canonical owner cannot accept this snapshot. */ -export function sparseSourceStatement(db: D1Database, source: SparseSource, prefix: PrefixSnapshot): D1PreparedStatement { +export function sparseSourceStatement(db: D1Database, source: SparseSource, prefix: PrefixSnapshot, requirePriorUpdate = false): D1PreparedStatement { if (sourceRowBytes(source) > MAX_ROW_BYTES) throw new Error('Sparse owner metadata exceeds row budget'); + const guard = prefixSnapshotGuard(prefix); return db.prepare(`INSERT INTO doc_sparse_sources ( - vector_id, source_type, source_version, prefix_indexed_at, index_version, + vector_id, source_type, source_version, prefix_indexed_at, index_version, prefix_revision, source_hash, body_hash, raw_hash, raw, segmented, raw_bytes, segmented_bytes, row_bytes, coverage, reason, observed_at) - SELECT ?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,? FROM search_docs - WHERE vector_id=? AND repo=? AND type=? AND doc_path=? AND source_version=? - AND content=? AND updated_at=? AND indexed_at=? + SELECT ?,?,?,?,?,source_revision,?,?,?,?,?,?,?,?,?,?,? FROM search_docs + WHERE ${guard.sql}${requirePriorUpdate ? ' AND changes()=1' : ''} ON CONFLICT(vector_id) DO UPDATE SET source_type=excluded.source_type, source_version=excluded.source_version, prefix_indexed_at=excluded.prefix_indexed_at, index_version=excluded.index_version, + prefix_revision=excluded.prefix_revision, source_hash=excluded.source_hash, body_hash=excluded.body_hash, raw_hash=excluded.raw_hash, raw=excluded.raw, segmented=excluded.segmented, raw_bytes=excluded.raw_bytes, segmented_bytes=excluded.segmented_bytes, row_bytes=excluded.row_bytes, @@ -88,7 +99,6 @@ export function sparseSourceStatement(db: D1Database, source: SparseSource, pref source.vectorId, source.type, source.sourceVersion, source.prefixIndexedAt, source.indexVersion, source.sourceHash, source.bodyHash, source.rawHash, source.raw, source.segmented, source.rawBytes, source.segmentedBytes, source.rowBytes, source.coverage, source.reason, source.observedAt, - prefix.vector_id, prefix.repo, prefix.type, prefix.doc_path, prefix.source_version, - prefix.content, prefix.updated_at, prefix.indexed_at, + ...guard.params, ); } diff --git a/src/doc-sparse.workers.test.ts b/src/doc-sparse.workers.test.ts index 13e1ed7..f9ab26c 100644 --- a/src/doc-sparse.workers.test.ts +++ b/src/doc-sparse.workers.test.ts @@ -13,7 +13,7 @@ import { docVectorId, wikiDocVectorId } from './pipeline/vector-id.js'; beforeAll(async () => { await applyD1Migrations(env.DB_FTS, env.TEST_MIGRATIONS); }); afterEach(() => vi.unstubAllGlobals()); const prefix = async (id: string) => (await env.DB_FTS.prepare('SELECT * FROM search_docs WHERE vector_id=?').bind(id).first())!; -const owner = (id: string) => env.DB_FTS.prepare('SELECT * FROM doc_sparse_sources WHERE vector_id=?').bind(id).first<{ coverage: string; source_version: string; raw: string; segmented: string; row_bytes: number }>(); +const owner = (id: string) => env.DB_FTS.prepare('SELECT * FROM doc_sparse_sources WHERE vector_id=?').bind(id).first<{ coverage: string; source_version: string; prefix_revision: number; raw: string; segmented: string; row_bytes: number }>(); const store = () => env.ISSUE_STORE.get(env.ISSUE_STORE.idFromName('global')); const runtime = (enabled = true) => ({ ...env, DOC_SPARSE_CONTEXT_ENABLED: enabled ? 'true' : undefined, AI: { run: vi.fn(async () => ({ data: [[0.1, 0.2]] })) }, VECTORIZE: { upsert: vi.fn(async () => undefined) }, @@ -180,6 +180,93 @@ describe('independent full sparse D1 index', () => { }); describe('bounded operator backfill', () => { + for (const type of ['doc', 'wiki_doc'] as const) it(`reconciles prepared A then default-off B before enable for ${type}`, async () => { + const repo = `sparse/reconcile-${type}`; + const a = 'prefix '.repeat(1400) + 'tail-a'; + const b = 'prefix '.repeat(1400) + 'tail-b'; + const live = runtime(); const disabled = { ...live, DOC_SPARSE_CONTEXT_ENABLED: undefined }; + const path = type === 'doc' ? 'a.md' : 'Home'; + const id = type === 'doc' ? await docVectorId(repo, path) : await wikiDocVectorId(repo, path); + const version = type === 'doc' ? await gitBlobSha(b) : await sha256Hex(b); + if (type === 'doc') { + await processAndUpsertDoc(live, store(), repo, path, a, await gitBlobSha(a)); + await processAndUpsertDoc(disabled, store(), repo, path, b, version); + mockBlob(b, version); + } else { + await processAndUpsertWikiDoc(live, store(), repo, path, 'md', a); + await processAndUpsertWikiDoc(disabled, store(), repo, path, 'md', b); + vi.stubGlobal('fetch', vi.fn(async () => new Response(b))); + } + expect((await prefix(id)).source_version).not.toBe(version); + expect(await owner(id)).toBeNull(); + vi.mocked(live.AI.run).mockClear(); vi.mocked(live.VECTORIZE.upsert).mockClear(); + const opts = { repo, type }; + const before = await prefix(id); + expect((await backfillDocSparse(disabled, opts)).results[0]).toMatchObject({ status: 'success', prefix_version_repair: true }); + expect((await prefix(id)).source_version).not.toBe(version); + expect((await backfillDocSparse(disabled, { ...opts, apply: true })).results[0]).toMatchObject({ status: 'success', prefix_version_repair: true }); + expect((await prefix(id)).source_version).toBe(version); + expect((await prefix(id)).source_revision).toBe(before.source_revision! + 1); + expect((await owner(id))?.source_version).toBe(version); + expect((await owner(id))?.prefix_revision).toBe(before.source_revision! + 1); + expect((await backfillDocSparse(disabled, { ...opts, apply: true })).results[0].status).toBe('unchanged'); + expect((await docSparseStats(env.DB_FTS, opts)).rows[0]).toMatchObject({ coverage: 'full', count: 1, legacy_version_unverified: 0 }); + expect(live.AI.run).not.toHaveBeenCalled(); expect(live.VECTORIZE.upsert).not.toHaveBeenCalled(); + }); + + for (const type of ['doc', 'wiki_doc'] as const) for (const race of ['same-fields-C', 'same-fields-C-plan', 'delete', 'batch-failure']) { + it(`refuses/rolls back ${race} during version repair for ${type}`, async () => { + const repo = `sparse/repair-${type}-${race}`; + const path = type === 'doc' ? 'a.md' : 'Home'; + const id = type === 'doc' ? await docVectorId(repo, path) : await wikiDocVectorId(repo, path); + const a = 'prefix '.repeat(1400) + 'tail-A'; const b = 'prefix '.repeat(1400) + 'tail-B'; + const version = type === 'doc' ? await gitBlobSha(b) : await sha256Hex(b); + const live = runtime(); const disabled = { ...live, DOC_SPARSE_CONTEXT_ENABLED: undefined }; + if (type === 'doc') { + await processAndUpsertDoc(live, store(), repo, path, a, await gitBlobSha(a)); + await processAndUpsertDoc(disabled, store(), repo, path, b, version); + } else { + await processAndUpsertWikiDoc(live, store(), repo, path, 'md', a); + await processAndUpsertWikiDoc(disabled, store(), repo, path, 'md', b); + } + const before = await prefix(id); + const duringFetch = async () => { + if (race === 'delete') await deleteFtsRow(env.DB_FTS, id); + if (race.startsWith('same-fields-C')) { + // Observe C's prefix write before its DO commit, even when its changed + // tail leaves all old text/timestamp/version fields identical. + await env.DB_FTS.prepare('UPDATE search_docs SET content=content, updated_at=updated_at, indexed_at=indexed_at WHERE vector_id=?').bind(id).run(); + } + }; + if (type === 'doc') mockBlob(b, version, duringFetch); + else vi.stubGlobal('fetch', vi.fn(async () => { await duringFetch(); return new Response(b); })); + let batchResults: D1Result[] = []; + const db = { prepare: env.DB_FTS.prepare.bind(env.DB_FTS), batch: async (statements: D1PreparedStatement[]) => { + if (race === 'batch-failure') return env.DB_FTS.batch([...statements, env.DB_FTS.prepare('INSERT INTO missing_repair_table VALUES(1)')]); + batchResults = await env.DB_FTS.batch(statements); return batchResults; + } } as unknown as D1Database; + const result = await backfillDocSparse({ ...disabled, DB_FTS: db }, { repo, type, apply: !race.endsWith('-plan') }); + expect(result.results[0].status).toBe(race === 'batch-failure' ? 'retryable_error' : 'conflict'); + expect(await owner(id)).toBeNull(); + if (race === 'delete') expect(await env.DB_FTS.prepare('SELECT vector_id FROM search_docs WHERE vector_id=?').bind(id).first()).toBeNull(); + else { + const after = await prefix(id); + expect(after.source_version).toBe(before.source_version); + expect(after.content).toBe(before.content); expect(after.indexed_at).toBe(before.indexed_at); expect(after.updated_at).toBe(before.updated_at); + expect(after.source_revision).toBe(before.source_revision! + (race.startsWith('same-fields-C') ? 1 : 0)); + } + if (race !== 'batch-failure' && !race.endsWith('-plan')) { + expect(batchResults[0].results).toEqual([]); expect(batchResults[1].results).toEqual([]); + } + await env.DB_FTS.prepare("INSERT INTO doc_sparse_fts_v1(doc_sparse_fts_v1,rank) VALUES('integrity-check',1)").run(); + if (race === 'batch-failure') { + expect(result.next_cursor).toBe(0); + expect((await backfillDocSparse(disabled, { repo, type, apply: true })).results[0].status).toBe('success'); + } + }); + } + + it('resumes a source deferred by the remaining byte budget rather than skipping it', async () => { const first = await seed('sparse/byte-resume', 'sparse-byte-first', 'abcdef', false, 'first.md'); const second = await seed('sparse/byte-resume', 'sparse-byte-second', 'uvwxyz', false, 'second.md'); @@ -199,14 +286,44 @@ describe('bounded operator backfill', () => { }); it('requires operator authentication and rejects invalid/broad scopes', async () => { - const live = { ...runtime(), GITHUB_TOKEN: 'test-operator' } as Env; + const live = { ...runtime(), GITHUB_TOKEN: 'test-github-fetch', DOC_SPARSE_ADMIN_TOKEN: 'test-sparse-operator' } as Env; const endpoint = 'http://test/admin/backfill-doc-sparse?repo=sparse/auth&type=doc'; expect((await handleDocSparseAdmin(new Request(endpoint, { method: 'POST' }), live)).status).toBe(401); + const invalidHeaders: Record[] = [ + { GITHUB_TOKEN: 'test-github-fetch' }, + { Authorization: 'Bearer test-github-fetch' }, + { Authorization: 'Bearer wrong' }, + { Authorization: 'test-sparse-operator' }, + { Authorization: 'bearer test-sparse-operator' }, + ]; + for (const headers of invalidHeaders) { + expect((await handleDocSparseAdmin(new Request(endpoint, { method: 'POST', headers }), live)).status).toBe(401); + } + for (const token of [undefined, '', ' ']) { + expect((await handleDocSparseAdmin(new Request(endpoint, { method: 'POST', headers: { Authorization: 'Bearer test-sparse-operator' } }), + { ...live, DOC_SPARSE_ADMIN_TOKEN: token })).status).toBe(401); + } for (const suffix of ['&limit=11', '&max_bytes=1048577', '&cursor=-1', '&apply=yes']) { - expect((await handleDocSparseAdmin(new Request(endpoint + suffix, { method: 'POST', headers: { GITHUB_TOKEN: 'test-operator' } }), live)).status).toBe(400); + expect((await handleDocSparseAdmin(new Request(endpoint + suffix, { method: 'POST', headers: { Authorization: 'Bearer test-sparse-operator' } }), live)).status).toBe(400); } - const planned = await handleDocSparseAdmin(new Request(endpoint, { method: 'POST', headers: { GITHUB_TOKEN: 'test-operator' } }), live); + const planned = await handleDocSparseAdmin(new Request(endpoint, { method: 'POST', headers: { Authorization: 'Bearer test-sparse-operator' } }), live); expect(planned.status).toBe(200); expect(await planned.json()).toMatchObject({ mode: 'plan', scanned: 0 }); + const stats = await handleDocSparseAdmin(new Request('http://test/admin/doc-sparse-stats?repo=sparse/auth&type=doc', { + method: 'POST', headers: { Authorization: 'Bearer test-sparse-operator' }, + }), live); + expect(stats.status).toBe(200); + expect((await handleDocSparseAdmin(new Request('http://test/admin/doc-sparse-stats?repo=sparse/auth&type=doc', { + method: 'POST', headers: { GITHUB_TOKEN: 'test-github-fetch' }, + }), live)).status).toBe(401); + const seeded = await seed('sparse/auth-fetch', 'sparse-auth-fetch', 'auth source'); + mockBlob('auth source', seeded.record.blobSha); + const applied = await handleDocSparseAdmin(new Request('http://test/admin/backfill-doc-sparse?repo=sparse/auth-fetch&type=doc&apply=true', { + method: 'POST', headers: { Authorization: 'Bearer test-sparse-operator' }, + }), live); + expect(applied.status).toBe(200); + expect(await applied.json()).toMatchObject({ counts: { success: 1 } }); + const fetchOptions = vi.mocked(fetch).mock.calls[0][1] as RequestInit; + expect((fetchOptions.headers as Record).Authorization).toBe('Bearer test-github-fetch'); }); it('plans without writes, applies a legacy prefix without embedding, and skips a rerun', async () => { diff --git a/src/types.ts b/src/types.ts index f6e07a6..9444bed 100644 --- a/src/types.ts +++ b/src/types.ts @@ -228,6 +228,8 @@ export interface Env extends OAuthEnv { DB_FTS: D1Database; /** Full doc/wiki sparse ownership; enabled only by the literal "true", default off. */ DOC_SPARSE_CONTEXT_ENABLED?: string; + /** Dedicated operator bearer secret for full sparse plan/apply/statistics only. */ + DOC_SPARSE_ADMIN_TOKEN?: string; AI: Ai; /** GitHub personal access token or installation token for API access (set via `wrangler secret put`) */ GITHUB_TOKEN: string; diff --git a/wrangler.toml b/wrangler.toml index 42a4c2b..8229c56 100644 --- a/wrangler.toml +++ b/wrangler.toml @@ -65,6 +65,9 @@ crons = ["0 * * * *", "15 * * * *", "30 * * * *", "45 * * * *"] # GITHUB_CLIENT_ID — GitHub App OAuth client ID # GITHUB_CLIENT_SECRET — GitHub App OAuth client secret # GITHUB_TOKEN — GitHub token for API polling (PAT or installation token) +# DOC_SPARSE_ADMIN_TOKEN — Dedicated bearer secret for sparse plan/apply/stats only; +# provision after reviewed merge via wrangler secret put, +# never place its value in [vars]. Unset/empty fails closed. # Environment variables [vars]