perf(api): バッチ化・並列化で逐次クエリ/逐次 await を解消

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<Record<string, unknown>>;
+  scopeRow: Array<Record<string, unknown>>;
+  ghostCandidates: Array<Record<string, unknown>>;
+  onExecute: (callIndex: number) => Promise<unknown>;
+}) {
+  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<string, unknown> = {};
+    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<string, unknown> = {};
+    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<string, unknown> = {
+    select: () => makeSelectChain(),
+    update: () => makeWriteChain(),
+    insert: () => makeWriteChain(),
+    delete: () => makeWriteChain(),
+    execute: () => config.onExecute(executeCalls++),
+    transaction: (fn: (tx: unknown) => Promise<unknown>) => 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);
+  });
 });

server/api/src/services/syncAiModels.ts

@@ -189,33 +189,38 @@ async function syncOneProvider(
   const [maxRow] = await db
     .select({ maxOrder: sql<number>`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);

server/api/src/services/titleRenamePropagationService.ts

@@ -79,6 +79,21 @@ export interface PropagateTitleRenameOptions {
   invalidateDocument?: (pageId: string) => Promise<void>;
 }
 
+/**
+ * 参照元ページの書き換えを並列実行する際の最大同時実行数。各 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<void>,
+): Promise<RewriteResult> {
+  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. ベストエフォートで昇格させる。