diff --git a/server/api/src/__tests__/services/titleRenamePropagationService.test.ts b/server/api/src/__tests__/services/titleRenamePropagationService.test.ts index 2ad01037..5088e30f 100644 --- a/server/api/src/__tests__/services/titleRenamePropagationService.test.ts +++ b/server/api/src/__tests__/services/titleRenamePropagationService.test.ts @@ -10,6 +10,7 @@ import * as Y from "yjs"; import { createMockDb } from "../createMockDb.js"; import { propagateTitleRename } from "../../services/titleRenamePropagationService.js"; +import { links, ghostLinks, pageContents, pages } from "../../schema/index.js"; /** * page_contents 行に入っているようなバイナリ Y.Doc を生成するヘルパー。 @@ -112,6 +113,73 @@ const OWNER_ID = "owner-user-1"; /** Default scope result: personal page owned by OWNER_ID. 個人ページ既定スコープ。 */ const PERSONAL_SCOPE_ROW = [{ noteId: null, ownerId: OWNER_ID }]; +/** + * 並列バッチ検証用の意味ベース DB モック。`createMockDb` は呼び出し順に結果を + * 返すため、複数ソースを並列処理するとクエリがインターリーブして対応が崩れる。 + * このモックは `.from(table)` で結果を決めるため順序非依存で、`execute` + * (FOR UPDATE)はソース順の呼び出し index で制御できる。 + * + * Order-independent DB mock for the parallel batch path. Results are keyed by + * the queried table (not call order), so concurrent interleaving is fine. The + * `execute` (FOR UPDATE) call is delegated to `onExecute(callIndex)` — call + * order equals source order, which lets tests fail or gate a specific source. + */ +function createParallelRenameDb(config: { + sourceIds: string[]; + makePageContentsRow: () => Array>; + scopeRow: Array>; + ghostCandidates: Array>; + onExecute: (callIndex: number) => Promise; +}) { + let executeCalls = 0; + + function resolveByTable(table: unknown): unknown { + if (table === links) return config.sourceIds.map((id) => ({ sourceId: id })); + if (table === pageContents) return config.makePageContentsRow(); + if (table === pages) return config.scopeRow; + if (table === ghostLinks) return config.ghostCandidates; + return []; + } + + function makeSelectChain() { + let table: unknown = null; + const chain: Record = {}; + for (const m of ["where", "limit", "innerJoin", "leftJoin", "orderBy", "groupBy", "offset"]) { + chain[m] = () => chain; + } + chain.from = (t: unknown) => { + table = t; + return chain; + }; + chain.then = (resolve?: (v: unknown) => unknown, reject?: (e: unknown) => unknown) => + Promise.resolve(resolveByTable(table)).then(resolve, reject); + return chain; + } + + function makeWriteChain() { + const chain: Record = {}; + for (const m of ["set", "where", "values", "onConflictDoNothing", "returning"]) { + chain[m] = () => chain; + } + chain.then = (resolve?: (v: unknown) => unknown, reject?: (e: unknown) => unknown) => + Promise.resolve([]).then(resolve, reject); + return chain; + } + + const db: Record = { + select: () => makeSelectChain(), + update: () => makeWriteChain(), + insert: () => makeWriteChain(), + delete: () => makeWriteChain(), + execute: () => config.onExecute(executeCalls++), + transaction: (fn: (tx: unknown) => Promise) => fn(db), + }; + return db; +} + +/** マクロタスク境界まで進めて保留中の microtask を全て流す。 / Flush microtasks. */ +const flushTasks = () => new Promise((resolve) => setTimeout(resolve, 0)); + describe("propagateTitleRename", () => { it("returns a zero result and skips all DB work when oldTitle or newTitle is missing", async () => { const { db, chains } = createMockDb([]); @@ -453,4 +521,84 @@ describe("propagateTitleRename", () => { expect(titles[OTHER_TARGET_ID]).toBe("Foo"); } }); + + it("rewrites multiple source pages in parallel and stays best-effort on failure", async () => { + // 3 ソースを並列処理。FOR UPDATE の 2 回目(= 2 番目のソース)だけ失敗させ、 + // 残り 2 件は成功し、集計とゴースト昇格が回帰なく動くことを検証する。 + // Three sources processed in parallel; only the 2nd FOR UPDATE (2nd source) + // fails. The other two succeed, counts aggregate, and ghost promotion runs. + const invalidate = vi.fn().mockResolvedValue(undefined); + const db = createParallelRenameDb({ + sourceIds: ["src-a", "src-b", "src-c"], + makePageContentsRow: () => [ + { pageId: "src-x", ydocState: makeYdocWithWikiLink("Foo"), version: 1 }, + ], + scopeRow: PERSONAL_SCOPE_ROW, + ghostCandidates: [], + // execute 呼び出しはソース順。index 1(2 番目)だけロック失敗させる。 + onExecute: (i) => (i === 1 ? Promise.reject(new Error("lock failed")) : Promise.resolve([])), + }); + + const result = await propagateTitleRename(db as never, PAGE_ID, "Foo", "Bar", { + invalidateDocument: invalidate, + }); + + expect(result.sourcePagesAttempted).toBe(3); + expect(result.sourcePagesSucceeded).toBe(2); + expect(result.sourcePagesFailed).toBe(1); + // 各成功ソースが 1 マークを書き換え、合算される。 / counts aggregate across sources. + expect(result.wikiLinkMarksUpdated).toBe(2); + expect(result.wikiLinkTextUpdated).toBe(2); + // 成功した 2 ソース分だけ invalidate される。 / only successful sources invalidate. + expect(invalidate).toHaveBeenCalledTimes(2); + // ゴースト昇格は失敗に関係なく実行される(候補ゼロなので 0)。 + expect(result.ghostPromotionsCount).toBe(0); + }); + + it("bounds concurrency to SOURCE_REWRITE_CONCURRENCY (4) across source pages", async () => { + // 10 ソースを投入。FOR UPDATE を手動ゲートで保留させ、最初のチャンクで + // 同時に走るのが最大 4 件であること(= 4 件だけ execute が発火)を検証する。 + // 10 sources with manually-gated FOR UPDATE: assert at most 4 run before the + // first batch resolves, proving the bound, then drain and assert all succeed. + const invalidate = vi.fn().mockResolvedValue(undefined); + const gates: Array<() => void> = []; + const db = createParallelRenameDb({ + sourceIds: Array.from({ length: 10 }, (_, i) => `src-${i}`), + makePageContentsRow: () => [ + { pageId: "src-x", ydocState: makeYdocWithWikiLink("Foo"), version: 1 }, + ], + scopeRow: PERSONAL_SCOPE_ROW, + ghostCandidates: [], + onExecute: () => + new Promise((resolve) => { + gates.push(() => resolve([])); + }), + }); + + const pending = propagateTitleRename(db as never, PAGE_ID, "Foo", "Bar", { + invalidateDocument: invalidate, + }); + + // 最初のチャンク (4 件) のみが in-flight。 / Only the first chunk of 4 has started. + await flushTasks(); + expect(gates.length).toBe(4); + + // 1 チャンク目を解放すると 2 チャンク目 (4 件) が走る。 + gates.splice(0, 4).forEach((open) => open()); + await flushTasks(); + expect(gates.length).toBe(4); + + // 2 チャンク目を解放すると 3 チャンク目 (残り 2 件) が走る。 + gates.splice(0, 4).forEach((open) => open()); + await flushTasks(); + expect(gates.length).toBe(2); + + gates.splice(0, 2).forEach((open) => open()); + const result = await pending; + + expect(result.sourcePagesAttempted).toBe(10); + expect(result.sourcePagesSucceeded).toBe(10); + expect(result.sourcePagesFailed).toBe(0); + expect(invalidate).toHaveBeenCalledTimes(10); + }); }); diff --git a/server/api/src/services/syncAiModels.ts b/server/api/src/services/syncAiModels.ts index efe7006e..24e35eeb 100644 --- a/server/api/src/services/syncAiModels.ts +++ b/server/api/src/services/syncAiModels.ts @@ -189,33 +189,38 @@ async function syncOneProvider( const [maxRow] = await db .select({ maxOrder: sql`coalesce(max(${aiModels.sortOrder}), -1)` }) .from(aiModels); - let nextSortOrder = Number(maxRow?.maxOrder ?? -1) + 1; + const nextSortOrder = Number(maxRow?.maxOrder ?? -1) + 1; + // 既存 ID を除いた新規行を 1 回のマルチバリュー INSERT でまとめて投入する。 + // sortOrder は max+1 から index で事前採番する。`existingIds` で除外済みのため + // onConflict はほぼ発生せず、仮にスキップで連番にギャップが生じても sortOrder + // は表示順(orderBy)にしか使われないため無害。 + // + // Insert all new rows (those not already present) in a single multi-value + // INSERT, pre-assigning sortOrder by index from max+1. Conflicts are + // unlikely since existing IDs are filtered out, and any gap from a skipped + // conflict is harmless because sortOrder only feeds display ordering. + const newRows = rows.filter((row) => !existingIds.has(row.id)); let upserted = 0; - for (const row of rows) { - if (existingIds.has(row.id)) continue; - - const isActive = isSonnetModel(row.provider, row.modelId) ? false : row.isActive; + if (newRows.length > 0) { + const values = newRows.map((row, idx) => ({ + id: row.id, + provider: row.provider, + modelId: row.modelId, + displayName: row.displayName, + tierRequired: row.tierRequired, + inputCostUnits: row.inputCostUnits, + outputCostUnits: row.outputCostUnits, + isActive: isSonnetModel(row.provider, row.modelId) ? false : row.isActive, + sortOrder: nextSortOrder + idx, + })); const inserted = await db .insert(aiModels) - .values({ - id: row.id, - provider: row.provider, - modelId: row.modelId, - displayName: row.displayName, - tierRequired: row.tierRequired, - inputCostUnits: row.inputCostUnits, - outputCostUnits: row.outputCostUnits, - isActive, - sortOrder: nextSortOrder, - }) + .values(values) .onConflictDoNothing({ target: aiModels.id }) .returning({ id: aiModels.id }); - if (inserted.length > 0) { - upserted += 1; - nextSortOrder += 1; - } + upserted = inserted.length; } const fetchedIds = rows.map((row) => row.id); diff --git a/server/api/src/services/titleRenamePropagationService.ts b/server/api/src/services/titleRenamePropagationService.ts index 4a8b7197..cf8c37f5 100644 --- a/server/api/src/services/titleRenamePropagationService.ts +++ b/server/api/src/services/titleRenamePropagationService.ts @@ -79,6 +79,21 @@ export interface PropagateTitleRenameOptions { invalidateDocument?: (pageId: string) => Promise; } +/** + * 参照元ページの書き換えを並列実行する際の最大同時実行数。各 source page は + * 個別の `db.transaction`(接続を 1 本消費)を張る。さらに本伝播は呼び出し元で + * fire-and-forget されるため、複数ユーザーの同時リネームで使用接続数が重畳する + * (N 件同時 → 最大 4N 接続)。DB プール上限(`db/client.ts` の `max: 20`)を + * 枯渇させて無関係な API を巻き込まないよう、保守的に 4 に固定する。 + * + * Maximum number of source-page rewrites run concurrently. Each rewrite opens + * its own transaction (one pooled connection), and propagation is fire-and- + * forget, so concurrent renames stack (N at once → up to 4N connections). Kept + * conservatively at 4 so it never exhausts the pool (`max: 20`) and starves + * unrelated requests. + */ +const SOURCE_REWRITE_CONCURRENCY = 4; + function normalizeTitle(value: string): string { return value.toLowerCase().trim(); } @@ -195,6 +210,44 @@ async function rewriteSourcePage( }); } +/** + * 1 つのソースページを書き換え、変更があれば Hocuspocus キャッシュを破棄する。 + * 並列バッチの 1 ユニットとして使う。invalidate の失敗はベストエフォートで + * warn して握りつぶし、rewrite の失敗はそのまま throw して呼び出し側の + * `Promise.allSettled` に `sourcePagesFailed` として集計させる。 + * + * Rewrite one source page and, if it changed, drop its Hocuspocus cache. + * Used as a single unit inside the bounded parallel batch. Invalidation + * failures are swallowed (best-effort warn); rewrite failures propagate so + * the caller's `Promise.allSettled` counts them as `sourcePagesFailed`. + */ +async function rewriteAndInvalidateSourcePage( + db: Database, + sourceId: string, + renamedPageId: string, + oldTitle: string, + newTitle: string, + invalidate: (pageId: string) => Promise, +): Promise { + const { changed, rewrite } = await rewriteSourcePage( + db, + sourceId, + renamedPageId, + oldTitle, + newTitle, + ); + + if (changed) { + try { + await invalidate(sourceId); + } catch (error) { + console.warn(`[RenamePropagation] Invalidation failed for source page ${sourceId}:`, error); + } + } + + return rewrite; +} + /** * 新タイトルと一致するゴーストリンクを、リネーム対象と同一スコープ内でのみ * 実体リンクへ昇格させる。スコープはリネーム対象の `pages.note_id` で決定する。 @@ -312,40 +365,47 @@ export async function propagateTitleRename( const uniqueSourceIds = Array.from(new Set(sourceRows.map((r) => r.sourceId))); - for (const sourceId of uniqueSourceIds) { - result.sourcePagesAttempted += 1; - try { - const { changed, rewrite } = await rewriteSourcePage( - db, - sourceId, - renamedPageId, - trimmedOld, - trimmedNew, - ); - result.sourcePagesSucceeded += 1; - result.wikiLinkMarksUpdated += rewrite.wikiLinkMarksUpdated; - result.wikiLinkTextUpdated += rewrite.wikiLinkTextUpdated; - result.tagMarksUpdated += rewrite.tagMarksUpdated; - result.tagTextUpdated += rewrite.tagTextUpdated; - - if (changed) { - try { - await invalidate(sourceId); - } catch (error) { - console.warn( - `[RenamePropagation] Invalidation failed for source page ${sourceId}:`, - error, - ); - } + // 各 source page は独立した `page_contents` 行をロックするため順序非依存。 + // 多数リンク時のレイテンシを抑えるため、最大 SOURCE_REWRITE_CONCURRENCY 件 + // ずつチャンクに分けて並列実行する。失敗は従来どおりベストエフォートで握り、 + // 後続の source / ghost 昇格を止めない。 + // + // Source pages are order-independent (each locks a distinct page_contents + // row), so rewrite them in bounded `Promise.allSettled` batches to keep + // latency flat as the link count grows. Per-page failures stay best-effort. + for (let i = 0; i < uniqueSourceIds.length; i += SOURCE_REWRITE_CONCURRENCY) { + const batch = uniqueSourceIds.slice(i, i + SOURCE_REWRITE_CONCURRENCY); + const settled = await Promise.allSettled( + batch.map((sourceId) => + rewriteAndInvalidateSourcePage( + db, + sourceId, + renamedPageId, + trimmedOld, + trimmedNew, + invalidate, + ), + ), + ); + + settled.forEach((outcome, idx) => { + const sourceId = batch[idx]; + result.sourcePagesAttempted += 1; + if (outcome.status === "fulfilled") { + result.sourcePagesSucceeded += 1; + result.wikiLinkMarksUpdated += outcome.value.wikiLinkMarksUpdated; + result.wikiLinkTextUpdated += outcome.value.wikiLinkTextUpdated; + result.tagMarksUpdated += outcome.value.tagMarksUpdated; + result.tagTextUpdated += outcome.value.tagTextUpdated; + } else { + result.sourcePagesFailed += 1; + console.error( + `[RenamePropagation] Failed to rewrite source page ${sourceId} ` + + `for rename ${renamedPageId} (${trimmedOld} → ${trimmedNew}):`, + outcome.reason, + ); } - } catch (error) { - result.sourcePagesFailed += 1; - console.error( - `[RenamePropagation] Failed to rewrite source page ${sourceId} ` + - `for rename ${renamedPageId} (${trimmedOld} → ${trimmedNew}):`, - error, - ); - } + }); } // 2. Promote matching ghost links. ベストエフォートで昇格させる。