fix(schema-diff): stop comparison and release pools on disconnect

Closes #11264

apps/desktop/src/composables/__tests__/useSchemaDiffSession.spec.ts

@@ -22,6 +22,42 @@ vi.mock("@/lib/schema/schemaDiffMetadataLoad", () => ({ loadSchemaDetails: vi.fn
 
 const { startSchemaDiffSession } = await import("../useSchemaDiffSession.ts");
 
+test("disconnecting either side stops the compare after its issued table-list requests settle", async () => {
+  const { cancelSchemaDiffTasksForConnection } = await import("@/lib/schema/schemaDiffCancellation");
+  const { loadSchemaDetails } = await import("@/lib/schema/schemaDiffMetadataLoad");
+  vi.clearAllMocks();
+  const finish: Array<(tables: never[]) => void> = [];
+  const session = startSchemaDiffSession(
+    {
+      sourceConnectionId: "cancel-source",
+      sourceDatabase: "app",
+      sourceSchema: "",
+      targetConnectionId: "cancel-target",
+      targetDatabase: "app",
+      targetSchema: "",
+      sourceDbType: "mysql",
+      targetDbType: "mysql",
+      options: {},
+      ignoreComments: false,
+      label: "cancel compare",
+    },
+    { tableListLoader: { load: vi.fn(() => new Promise<never[]>((resolve) => finish.push(resolve))) } },
+  );
+  const cancellation = cancelSchemaDiffTasksForConnection("cancel-target", new Error("connection disconnected"));
+  assert.ok(cancellation);
+  finish[0]([]);
+  await new Promise((resolve) => setTimeout(resolve, 0));
+  assert.equal(session.status, "running");
+  finish[1]([]);
+  await cancellation;
+  assert.equal(session.status, "failed");
+  assert.equal(session.error, "connection disconnected");
+  assert.equal(vi.mocked(loadSchemaDetails).mock.calls.length, 0);
+  assert.equal(apiMock.prepareSchemaDiff.mock.calls.length, 0);
+  assert.equal(trackerMock.updateCompareTask.mock.calls.at(-1)?.[1].status, "Error");
+  vi.clearAllMocks();
+});
+
 test("applies every exclude rule before loading source and target details", async () => {
   const { loadSchemaDetails } = await import("@/lib/schema/schemaDiffMetadataLoad");
   vi.mocked(loadSchemaDetails).mockClear();

apps/desktop/src/composables/useSchemaDiffSession.ts

@@ -7,6 +7,7 @@ import { filterSchemaDiffTables } from "@/lib/schema/schemaDiffTableFilter";
 import { compileSchemaDiffTableFilter } from "@/lib/schema/schemaDiffTableFilter";
 import { filterSchemaDiffFunctions } from "@/lib/schema/schemaDiffRoutine";
 import { loadSchemaDetails } from "@/lib/schema/schemaDiffMetadataLoad";
+import { registerSchemaDiffTask } from "@/lib/schema/schemaDiffCancellation";
 import type { SchemaDiffTableIdentity, SchemaDiffTableListLoader } from "@/lib/schema/schemaDiffTableList";
 import { getSchemaDiffNextProgressStep, isSchemaDiffPostgresLike, shouldLoadSchemaDiffExtraObjectPhase, shouldLoadSchemaDiffRoutines, type SchemaDiffProgressPhase } from "@/lib/schema/schemaDiffProgress";
 import { schemaDiffRoutineObjectTypesIntersection } from "@/lib/database/databaseObjectCapabilities";
@@ -112,7 +113,15 @@ function publishProgress(session: SchemaDiffSession, progress: SchemaDiffSession
   });
 }
 
-async function runSchemaDiffSession(session: SchemaDiffSession, dependencies: SchemaDiffSessionDependencies): Promise<void> {
+async function settleRequests<T>(requests: Promise<T>[]): Promise<T[]> {
+  const results = await Promise.allSettled(requests);
+  return results.map((result) => {
+    if (result.status === "rejected") throw result.reason;
+    return result.value;
+  });
+}
+
+async function runSchemaDiffSession(session: SchemaDiffSession, dependencies: SchemaDiffSessionDependencies, signal: AbortSignal): Promise<void> {
   const input = session.config;
   const tracker = useExportTracker();
   const options = normalizeSchemaDiffCompareOptions(input.options, input.targetDbType);
@@ -143,7 +152,8 @@ async function runSchemaDiffSession(session: SchemaDiffSession, dependencies: Sc
         database: input.targetDatabase,
         schema: input.targetSchema,
       };
-      const [sourceTableList, targetTableList] = await Promise.all([dependencies.tableListLoader.load(sourceTableIdentity, { refresh: true }), dependencies.tableListLoader.load(targetTableIdentity, { refresh: true })]);
+      const [sourceTableList, targetTableList] = await settleRequests([dependencies.tableListLoader.load(sourceTableIdentity, { refresh: true }), dependencies.tableListLoader.load(targetTableIdentity, { refresh: true })]);
+      signal.throwIfAborted();
       ({ sourceTables, targetTables } = filterSchemaDiffTables(sourceTableList, targetTableList, tableFilter, sessionOptions, sessionOptions.selectedTables));
 
       publishProgress(session, { phase: "loading-source-details", current: 0, total: sourceTables.length });
@@ -155,11 +165,13 @@ async function runSchemaDiffSession(session: SchemaDiffSession, dependencies: Sc
           schema: input.sourceSchema,
           dbType: input.sourceDbType,
           options: sessionOptions,
+          signal,
           onProgress: (progress) => publishProgress(session, { phase: "loading-source-details", ...progress }),
         },
         api,
       );
 
+      signal.throwIfAborted();
       publishProgress(session, { phase: "loading-target-details", current: 0, total: targetTables.length });
       targetDetails = await loadSchemaDetails(
         targetTables,
@@ -169,12 +181,14 @@ async function runSchemaDiffSession(session: SchemaDiffSession, dependencies: Sc
           schema: input.targetSchema,
           dbType: input.targetDbType,
           options: sessionOptions,
+          signal,
           onProgress: (progress) => publishProgress(session, { phase: "loading-target-details", ...progress }),
         },
         api,
       );
     }
 
+    signal.throwIfAborted();
     const promises: Promise<unknown>[] = [];
     if (loadRoutines) {
       promises.push(api.listFunctions(input.sourceConnectionId, input.sourceDatabase, input.sourceSchema));
@@ -194,7 +208,8 @@ async function runSchemaDiffSession(session: SchemaDiffSession, dependencies: Sc
     }
 
     if (hasExtraObjectPhase) publishProgress(session, { phase: "loading-extra-objects" });
-    const extraObjects = await Promise.all(promises);
+    const extraObjects = await settleRequests(promises);
+    signal.throwIfAborted();
     let index = 0;
     const listedSourceFunctions = loadRoutines ? filterFunctionsByRoutineKinds(extraObjects[index++] as FunctionInfo[], routineKinds) : [];
     const listedTargetFunctions = loadRoutines ? filterFunctionsByRoutineKinds(extraObjects[index++] as FunctionInfo[], routineKinds) : [];
@@ -251,6 +266,7 @@ async function runSchemaDiffSession(session: SchemaDiffSession, dependencies: Sc
         })) || [],
     });
 
+    signal.throwIfAborted();
     publishProgress(session, { phase: "generating" });
     session.result = result;
     session.progress = { phase: "complete" };
@@ -265,7 +281,7 @@ async function runSchemaDiffSession(session: SchemaDiffSession, dependencies: Sc
       compareResultCount: resultObjectCount(result),
     });
   } catch (error: unknown) {
-    const message = sessionError(error);
+    const message = sessionError(signal.aborted ? signal.reason : error);
     session.error = message;
     session.progress = null;
     session.version += 1;
@@ -302,7 +318,9 @@ export function startSchemaDiffSession(input: Omit<SchemaDiffSessionConfig, "opt
     () => openSchemaDiffSession(id),
     () => removeSchemaDiffSession(id),
   );
-  void runSchemaDiffSession(session, dependencies);
+  const controller = new AbortController();
+  const settled = runSchemaDiffSession(session, dependencies, controller.signal);
+  registerSchemaDiffTask([input.sourceConnectionId, input.targetConnectionId], controller, settled);
   return session;
 }
 

apps/desktop/src/i18n/locales/az.ts

@@ -6822,6 +6822,7 @@ export default withEnglishFallback({
     rollbackSql: "Geri qaytarma SQL-i",
   },
   schemaDiff: {
+    connectionDisconnected: "Verilənlər bazası bağlantısı kəsildiyi üçün sxem müqayisəsi dayandırıldı.",
     optionsTitle: "Müqayisə seçimləri",
     selectConfig: "Konfiqurasiya seç",
     newConfigName: "Yeni konfiqurasiya",

apps/desktop/src/i18n/locales/en.ts

@@ -7780,6 +7780,7 @@ export default {
     rollbackSql: "Rollback SQL",
   },
   schemaDiff: {
+    connectionDisconnected: "Schema comparison stopped because a database connection was disconnected.",
     optionsTitle: "Compare Options",
     selectConfig: "Select config",
     newConfigName: "New Config",

apps/desktop/src/i18n/locales/es.ts

@@ -7068,6 +7068,7 @@ export default withEnglishFallback({
     rollbackSql: "SQL de Reversión",
   },
   schemaDiff: {
+    connectionDisconnected: "La comparación de esquemas se detuvo porque se desconectó una conexión de base de datos.",
     optionsTitle: "Opciones de comparación",
     selectConfig: "Select config",
     newConfigName: "New Config",

apps/desktop/src/i18n/locales/id.ts

@@ -7537,6 +7537,7 @@ export default withEnglishFallback({
     rollbackSql: "SQL Rollback",
   },
   schemaDiff: {
+    connectionDisconnected: "Perbandingan skema dihentikan karena koneksi database terputus.",
     optionsTitle: "Opsi Perbandingan",
     selectConfig: "Pilih konfigurasi",
     newConfigName: "Konfigurasi Baru",

apps/desktop/src/i18n/locales/it.ts

@@ -6850,6 +6850,7 @@ export default withEnglishFallback({
     rollbackSql: "SQL di Rollback",
   },
   schemaDiff: {
+    connectionDisconnected: "Il confronto degli schemi è stato interrotto perché una connessione al database è stata disconnessa.",
     optionsTitle: "Opzioni di Confronto",
     selectConfig: "Seleziona configurazione",
     newConfigName: "Nuova Configurazione",

apps/desktop/src/i18n/locales/ja.ts

@@ -6880,6 +6880,7 @@ export default withEnglishFallback({
     rollbackSql: "ロールバックSQL",
   },
   schemaDiff: {
+    connectionDisconnected: "データベース接続が切断されたため、スキーマ比較を停止しました。",
     optionsTitle: "比較オプション",
     selectConfig: "設定を選択",
     newConfigName: "新しい設定",

apps/desktop/src/i18n/locales/ko.ts

@@ -6691,6 +6691,7 @@ export default withEnglishFallback({
     rollbackSql: "롤백 SQL",
   },
   schemaDiff: {
+    connectionDisconnected: "데이터베이스 연결이 끊어져 스키마 비교가 중지되었습니다.",
     optionsTitle: "비교 옵션",
     selectConfig: "설정 선택",
     newConfigName: "새 설정",

apps/desktop/src/i18n/locales/pt-BR.ts

@@ -6851,6 +6851,7 @@ export default withEnglishFallback({
     rollbackSql: "SQL de Reversão",
   },
   schemaDiff: {
+    connectionDisconnected: "A comparação de esquemas foi interrompida porque uma conexão com o banco de dados foi desconectada.",
     optionsTitle: "Opções de Comparação",
     selectConfig: "Selecionar configuração",
     newConfigName: "Nova Configuração",

apps/desktop/src/i18n/locales/ru.ts

@@ -8093,6 +8093,7 @@ export default withEnglishFallback({
     rollbackSql: "SQL отката",
   },
   schemaDiff: {
+    connectionDisconnected: "Сравнение схем остановлено, поскольку соединение с базой данных было отключено.",
     optionsTitle: "Параметры сравнения",
     selectConfig: "Выбрать конфигурацию",
     newConfigName: "Новая конфигурация",

apps/desktop/src/i18n/locales/tr.ts

@@ -6703,6 +6703,7 @@ export default withEnglishFallback({
     rollbackSql: "Geri Alma SQL'i",
   },
   schemaDiff: {
+    connectionDisconnected: "Veritabanı bağlantısı kesildiği için şema karşılaştırması durduruldu.",
     optionsTitle: "Karşılaştırma Seçenekleri",
     selectConfig: "Yapılandırma seçin",
     newConfigName: "Yeni Yapılandırma",

apps/desktop/src/i18n/locales/zh-CN.ts

@@ -7766,6 +7766,7 @@ export default withEnglishFallback({
     rollbackSql: "回滚 SQL",
   },
   schemaDiff: {
+    connectionDisconnected: "数据库连接已断开，结构比对已停止。",
     optionsTitle: "比较选项",
     selectConfig: "选择配置",
     newConfigName: "新配置",

apps/desktop/src/i18n/locales/zh-TW.ts

@@ -6205,6 +6205,7 @@ export default withEnglishFallback({
     rollbackSql: "回溯 SQL",
   },
   schemaDiff: {
+    connectionDisconnected: "資料庫連線已中斷，結構比對已停止。",
     optionsTitle: "比較選項",
     selectConfig: "選擇配置",
     newConfigName: "新配置",

apps/desktop/src/lib/__tests__/schema/schemaDiffMetadataLoad.spec.ts

@@ -1,4 +1,4 @@
-import { describe, expect, it } from "vitest";
+import { describe, expect, it, vi } from "vitest";
 import {
   createConcurrencyLimiter,
   loadSchemaDetails,
@@ -18,6 +18,46 @@ function wait(ms: number) {
   return new Promise<void>((resolve) => setTimeout(resolve, ms));
 }
 
+it("stops queued metadata after cancellation and waits for every issued request", async () => {
+  const controller = new AbortController();
+  const finish: Array<() => void> = [];
+  const request = vi.fn(() => new Promise<never[]>((resolve) => finish.push(() => resolve([]))));
+  const api: SchemaDiffMetadataApi = {
+    getTableDdl: async () => {
+      await request();
+      return "";
+    },
+    getColumns: request,
+    listIndexes: request,
+    listForeignKeys: request,
+    listTriggers: request,
+  };
+  const loading = loadSchemaDetails(
+    Array.from({ length: 40 }, (_, i) => ({ name: `t${i}`, table_type: "BASE TABLE" }) as TableInfo),
+    { connectionId: "cancelled", database: "app", schema: "", dbType: "mysql", options: DEFAULT_MYSQL_OPTIONS, signal: controller.signal },
+    api,
+  );
+  const rejection = expect(loading).rejects.toThrow("disconnected");
+  await vi.waitFor(() => expect(request).toHaveBeenCalledTimes(2));
+  let settled = false;
+  void loading.then(
+    () => {
+      settled = true;
+    },
+    () => {
+      settled = true;
+    },
+  );
+  controller.abort(new Error("disconnected"));
+  finish[0]();
+  await new Promise((resolve) => setTimeout(resolve, 0));
+  expect(request).toHaveBeenCalledTimes(2);
+  expect(settled).toBe(false);
+  finish[1]();
+  await rejection;
+  expect(schemaDiffMetadataLaneCountForTests()).toBe(0);
+});
+
 const METADATA_POOL_BUSY = "DBX metadata pool is busy; please retry";
 
 /**

apps/desktop/src/lib/schema/schemaDiffCancellation.ts

@@ -0,0 +1,24 @@
+interface SchemaDiffTask {
+  connectionIds: readonly string[];
+  controller: AbortController;
+  settled: Promise<void>;
+}
+
+const tasks = new Set<SchemaDiffTask>();
+
+export function registerSchemaDiffTask(connectionIds: readonly string[], controller: AbortController, settled: Promise<void>): void {
+  const task = { connectionIds, controller, settled };
+  tasks.add(task);
+  void settled.then(
+    () => tasks.delete(task),
+    () => tasks.delete(task),
+  );
+}
+
+/** Stop new metadata requests and drain issued requests before closing their pools. */
+export function cancelSchemaDiffTasksForConnection(connectionId: string, reason: Error): Promise<void> | undefined {
+  const matching = [...tasks].filter((task) => task.connectionIds.includes(connectionId));
+  if (matching.length === 0) return undefined;
+  for (const task of matching) task.controller.abort(reason);
+  return Promise.allSettled(matching.map((task) => task.settled)).then(() => undefined);
+}

apps/desktop/src/lib/schema/schemaDiffMetadataLoad.ts

@@ -229,6 +229,7 @@ export interface SchemaDetailLoadContext {
   schema: string;
   dbType: string;
   options: SchemaDiffCompareOptions;
+  signal?: AbortSignal;
   onProgress?: (progress: SchemaDiffMetadataProgress) => void;
 }
 
@@ -244,21 +245,33 @@ function isViewOrMaterializedView(tableType: string): ObjectSourceKind | undefin
 }
 
 export async function loadSchemaDetails(tables: TableInfo[], context: SchemaDetailLoadContext, api: SchemaDiffMetadataApi): Promise<TableSchemaDetail[]> {
+  context.signal?.throwIfAborted();
   const concurrency = schemaDiffMetadataConcurrency(context.dbType, tables.length);
   // Hold the lane from before the first await so a compare starting while this
   // one is between tables shares this lane instead of building a second one.
   const lane = acquireSchemaDiffMetadataLane(context.connectionId, context.database, concurrency);
+  const pending = new Set<Promise<unknown>>();
   const runMetadataQuery = <T>(task: () => Promise<T>): Promise<T> => {
     lane.requests += 1;
-    return lane.run(task).finally(() => {
-      lane.requests -= 1;
-      evictSchemaDiffMetadataLaneIfUnused(lane);
-    });
+    const request = lane
+      .run(() => {
+        // A queued request must recheck after acquiring the shared lane.
+        context.signal?.throwIfAborted();
+        return task();
+      })
+      .finally(() => {
+        lane.requests -= 1;
+        pending.delete(request);
+        evictSchemaDiffMetadataLaneIfUnused(lane);
+      });
+    pending.add(request);
+    return request;
   };
   let completed = 0;
 
   try {
     return await mapWithConcurrency(tables, concurrency, async (table) => {
+      context.signal?.throwIfAborted();
       const objectType = isViewOrMaterializedView(table.table_type);
       const loadPlan = schemaDiffMetadataLoadPlan(isSchemaDiffView(table), context.options);
       const ddlPromise = loadPlan.ddl ? runMetadataQuery(() => api.getTableDdl(context.connectionId, context.database, context.schema, table.name, objectType)) : Promise.resolve("");
@@ -271,10 +284,13 @@ export async function loadSchemaDetails(tables: TableInfo[], context: SchemaDeta
       ]);
 
       const detail = { name: table.name, columns, indexes, foreignKeys, triggers, ddl };
+      context.signal?.throwIfAborted();
       context.onProgress?.({ current: ++completed, total: tables.length, objectName: table.name });
       return detail;
     });
   } finally {
+    // Cancellable compares must drain siblings before disconnect can close pools.
+    if (context.signal) await Promise.allSettled(pending);
     releaseSchemaDiffMetadataLane(lane);
   }
 }

apps/desktop/src/stores/__tests__/connectionStore.oneTimeCleanup.spec.ts

@@ -72,6 +72,35 @@ describe("connectionStore one_time runtime cleanup", () => {
     vi.restoreAllMocks();
   });
 
+  it("cancels source and target schema compares and drains metadata before disconnecting pools", async () => {
+    installApiMocks();
+    const { registerSchemaDiffTask } = await import("@/lib/schema/schemaDiffCancellation");
+    const { useConnectionStore } = await import("@/stores/connectionStore");
+    const { disconnectDb } = await import("@/lib/backend/api");
+    const store = useConnectionStore();
+    store.connections = [previewConnection({ id: "saved-1", one_time: false })];
+    const controllers = [new AbortController(), new AbortController(), new AbortController()];
+    const finish: Array<() => void> = [];
+    for (const [i, ids] of [
+      ["saved-1", "target"],
+      ["source", "saved-1"],
+      ["other", "target"],
+    ].entries()) {
+      registerSchemaDiffTask(ids, controllers[i], new Promise<void>((resolve) => finish.push(resolve)));
+    }
+    const disconnecting = store.disconnect("saved-1");
+    expect(controllers.map((controller) => controller.signal.aborted)).toEqual([true, true, false]);
+    expect(disconnectDb).not.toHaveBeenCalled();
+    finish[0]();
+    await new Promise((resolve) => setTimeout(resolve, 0));
+    expect(disconnectDb).not.toHaveBeenCalled();
+    finish[1]();
+    await disconnecting;
+    expect(disconnectDb).toHaveBeenCalledOnce();
+    expect(disconnectDb).toHaveBeenCalledWith("saved-1", undefined);
+    finish[2]();
+  });
+
   // One-time connections are never persisted, so the backend's save_connections
   // sync never reclaims them; disconnect_db is the only reclaim point, which makes
   // an explicit disconnect on removal mandatory.

apps/desktop/src/stores/connectionStore.ts

@@ -1,4 +1,5 @@
 import { defineStore } from "pinia";
+import { cancelSchemaDiffTasksForConnection } from "@/lib/schema/schemaDiffCancellation";
 import type { SqlFilePreview } from "@/lib/backend/api";
 import { uuid } from "@/lib/common/utils";
 import { containsHan, orderedSubsequenceSpan, pinyinFirstLetters } from "@/lib/common/pinyin";
@@ -1120,11 +1121,11 @@ export const useConnectionStore = defineStore("connection", () => {
     return bounded;
   }
 
-  function startDisconnectRequest(connectionId: string): Promise<void> {
+  function startDisconnectRequest(connectionId: string, metadataDrain?: Promise<void>): Promise<void> {
     const clientAttempt = activeLocalConnectionAttempts.get(connectionId) ?? successfulLocalConnectionAttempts.get(connectionId);
     let request: Promise<void>;
     try {
-      request = api.disconnectDb(connectionId, clientAttempt);
+      request = metadataDrain ? metadataDrain.then(() => api.disconnectDb(connectionId, clientAttempt)) : api.disconnectDb(connectionId, clientAttempt);
     } catch (error) {
       request = Promise.reject(error);
     }
@@ -4772,10 +4773,11 @@ export const useConnectionStore = defineStore("connection", () => {
    * `disconnectTabHandlingMode`，否则会把刚保留下来的 SQL 页签又关掉。
    */
   async function disconnect(connectionId: string, options: { skipTabHandling?: boolean } = {}) {
+    const metadataDrain = cancelSchemaDiffTasksForConnection(connectionId, new Error(i18n.global.t("schemaDiff.connectionDisconnected")));
     const stateRevision = bumpConnectionStateRevision(connectionId);
     const shouldRemoveOneTimeConnection = getConfig(connectionId)?.one_time === true;
     if (hasSqlServerActivityTraceForConnection(connectionId)) await disposeSqlServerActivityTracesForConnection(connectionId);
-    const disconnectRequest = startDisconnectRequest(connectionId);
+    const disconnectRequest = startDisconnectRequest(connectionId, metadataDrain);
     cancelLocalConnectionAttempt(connectionId);
 
     connectedIds.value.delete(connectionId);