From ff1a0f5eda6720244dc14962b9620b61857d849f Mon Sep 17 00:00:00 2001 From: LRcoding <38545827+LwClick@users.noreply.github.com> Date: Wed, 1 Jul 2026 20:03:09 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E9=9A=94=E7=A6=BB=20MySQL=20=E5=A4=A7?= =?UTF-8?q?=E5=BA=93=E5=9C=BA=E6=99=AF=E5=89=8D=E5=8F=B0=E6=9F=A5=E8=AF=A2?= =?UTF-8?q?=E4=B8=8E=E5=90=8E=E5=8F=B0=E5=85=83=E6=95=B0=E6=8D=AE=E5=8A=A0?= =?UTF-8?q?=E8=BD=BD=20(#2354)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix: isolate data tab mysql queries from metadata loads * refactor(completion): drop unused limiter snapshot and share concurrency constant --------- Co-authored-by: zero Co-authored-by: t8y2 <1156263951@qq.com> --- .../src/components/editor/QueryEditor.vue | 6 +- .../src/composables/useDataGridActions.ts | 57 ++-- apps/desktop/src/lib/metadataTaskLimiter.ts | 70 +++++ .../connectionStore.completion.spec.ts | 47 ++++ apps/desktop/src/stores/connectionStore.ts | 243 ++++++++++-------- apps/desktop/src/stores/queryStore.ts | 2 +- .../app-tests/metadataTaskLimiter.test.ts | 74 ++++++ packages/app-tests/queryStore.test.ts | 34 +++ packages/app-tests/useDataGridActions.test.ts | 122 +++++++++ 9 files changed, 524 insertions(+), 131 deletions(-) create mode 100644 apps/desktop/src/lib/metadataTaskLimiter.ts create mode 100644 packages/app-tests/metadataTaskLimiter.test.ts create mode 100644 packages/app-tests/useDataGridActions.test.ts diff --git a/apps/desktop/src/components/editor/QueryEditor.vue b/apps/desktop/src/components/editor/QueryEditor.vue index 79bc93b3a7..6ef6e69af7 100644 --- a/apps/desktop/src/components/editor/QueryEditor.vue +++ b/apps/desktop/src/components/editor/QueryEditor.vue @@ -14,7 +14,7 @@ import { buildExecutionCandidates, hasMultipleExecutionTargets, supportsExecutio import { executableStatementRangeCacheForDoc, executableStatementRangeStartingAt as executableStatementRangeStartingAtLine, type ExecutableStatementRangeCache } from "@/lib/executableStatementRangeCache"; import { formatSqlText, type SqlFormatDialect } from "@/lib/sqlFormatter"; import { formatMongoShellText } from "@/lib/mongoFormatter"; -import { useConnectionStore } from "@/stores/connectionStore"; +import { useConnectionStore, COMPLETION_METADATA_CONCURRENCY } from "@/stores/connectionStore"; import { useSettingsStore } from "@/stores/settingsStore"; import { useTheme } from "@/composables/useTheme"; import { useToast } from "@/composables/useToast"; @@ -801,7 +801,9 @@ async function ensureForeignKeysForTables(tables: Array<{ name: string; schema?: seen.add(key); return true; }); - await Promise.all(uniqueTables.map((table) => ensureForeignKeysForTable(table))); + for (let index = 0; index < uniqueTables.length; index += COMPLETION_METADATA_CONCURRENCY) { + await Promise.all(uniqueTables.slice(index, index + COMPLETION_METADATA_CONCURRENCY).map((table) => ensureForeignKeysForTable(table))); + } } function createHoverDom(title: string, detail: string, rows: string[] = []) { diff --git a/apps/desktop/src/composables/useDataGridActions.ts b/apps/desktop/src/composables/useDataGridActions.ts index a9fcb97b95..ece2c93011 100644 --- a/apps/desktop/src/composables/useDataGridActions.ts +++ b/apps/desktop/src/composables/useDataGridActions.ts @@ -50,21 +50,35 @@ export function useDataGridActions(activeTab: ComputedRef) if (tab.mode !== "data" || !tab.connectionId || !tab.database) return; const tableMeta = tableMetaForDataTab(tab); if (!tableMeta?.tableName) return; - - console.info("[DBX][reloadData:metadata:ensure-connected:start]", { traceId: trace?.traceId, elapsed: trace?.elapsed() }); - await connectionStore.ensureConnected(tab.connectionId); - console.info("[DBX][reloadData:metadata:ensure-connected:done]", { traceId: trace?.traceId, elapsed: trace?.elapsed() }); - const config = connectionStore.getConfig(tab.connectionId); - const querySchema = metadataSchemaForConnection(config, tab.database, tableMeta.schema); - console.info("[DBX][reloadData:metadata:get-columns:start]", { traceId: trace?.traceId, elapsed: trace?.elapsed(), schema: querySchema, table: tableMeta.tableName }); - const columns = await api.getColumns(tab.connectionId, tab.database, querySchema, tableMeta.tableName); - const indexes = await api.listIndexes(tab.connectionId, tab.database, querySchema, tableMeta.tableName).catch(() => []); - console.info("[DBX][reloadData:metadata:get-columns:done]", { traceId: trace?.traceId, elapsed: trace?.elapsed(), columnCount: columns.length }); - const primaryKeys = editableRowIdentifierColumns(effectiveDatabaseTypeForConnection(config), columns, indexes, tableMeta.tableType); - queryStore.setTableMeta(tab.id, { + const target = { + tabId: tab.id, + connectionId: tab.connectionId, + database: tab.database, schema: tableMeta.schema, tableName: tableMeta.tableName, tableType: tableMeta.tableType, + }; + + console.info("[DBX][reloadData:metadata:ensure-connected:start]", { traceId: trace?.traceId, elapsed: trace?.elapsed() }); + await connectionStore.ensureConnected(target.connectionId); + console.info("[DBX][reloadData:metadata:ensure-connected:done]", { traceId: trace?.traceId, elapsed: trace?.elapsed() }); + const config = connectionStore.getConfig(target.connectionId); + const querySchema = metadataSchemaForConnection(config, target.database, target.schema); + console.info("[DBX][reloadData:metadata:get-columns:start]", { traceId: trace?.traceId, elapsed: trace?.elapsed(), schema: querySchema, table: target.tableName }); + const columns = await api.getColumns(target.connectionId, target.database, querySchema, target.tableName); + const indexes = await api.listIndexes(target.connectionId, target.database, querySchema, target.tableName).catch(() => []); + console.info("[DBX][reloadData:metadata:get-columns:done]", { traceId: trace?.traceId, elapsed: trace?.elapsed(), columnCount: columns.length }); + const current = queryStore.tabs.find((item) => item.id === target.tabId); + const currentMeta = current ? tableMetaForDataTab(current) : undefined; + if (!current || current.mode !== "data" || current.connectionId !== target.connectionId || current.database !== target.database || currentMeta?.tableName !== target.tableName || (currentMeta.schema ?? "") !== (target.schema ?? "")) { + console.info("[DBX][reloadData:metadata:stale-tab]", { traceId: trace?.traceId, elapsed: trace?.elapsed(), table: target.tableName }); + return; + } + const primaryKeys = editableRowIdentifierColumns(effectiveDatabaseTypeForConnection(config), columns, indexes, target.tableType); + queryStore.setTableMeta(target.tabId, { + schema: target.schema, + tableName: target.tableName, + tableType: target.tableType, columns, primaryKeys, }); @@ -98,16 +112,17 @@ export function useDataGridActions(activeTab: ComputedRef) queryStore.setExecuting(tab.id, true); const tableMeta = tableMetaForDataTab(tab); const metadataAgeMs = tab.tableMetaUpdatedAt ? Date.now() - tab.tableMetaUpdatedAt : Number.POSITIVE_INFINITY; - const shouldRefreshMetadata = !tableMeta?.columns.length || metadataAgeMs > DATA_TAB_METADATA_TTL_MS; + const shouldRefreshMetadata = !tab.tableMeta || !tableMeta?.columns.length || metadataAgeMs > DATA_TAB_METADATA_TTL_MS; if (shouldRefreshMetadata) { - try { - console.info("[DBX][reloadData:metadata:start]", { traceId, elapsed: elapsed(), reason: tableMeta?.columns.length ? "stale" : "missing", metadataAgeMs }); - await refreshDataTabTableMeta(tab, { traceId, elapsed }); - console.info("[DBX][reloadData:metadata:done]", { traceId, elapsed: elapsed() }); - } catch (e: any) { - console.warn("[DBX][reloadData:metadata:error]", { traceId, elapsed: elapsed(), error: e }); - toast(e?.message || String(e), 5000); - } + console.info("[DBX][reloadData:metadata:background:start]", { traceId, elapsed: elapsed(), reason: tableMeta?.columns.length ? "stale" : "missing", metadataAgeMs }); + void refreshDataTabTableMeta(tab, { traceId, elapsed }) + .then(() => { + console.info("[DBX][reloadData:metadata:background:done]", { traceId, elapsed: elapsed() }); + }) + .catch((e: any) => { + console.warn("[DBX][reloadData:metadata:background:error]", { traceId, elapsed: elapsed(), error: e }); + toast(e?.message || String(e), 5000); + }); } else { console.info("[DBX][reloadData:metadata:skip]", { traceId, elapsed: elapsed(), columnCount: tableMeta.columns.length, metadataAgeMs }); } diff --git a/apps/desktop/src/lib/metadataTaskLimiter.ts b/apps/desktop/src/lib/metadataTaskLimiter.ts new file mode 100644 index 0000000000..570fe3a704 --- /dev/null +++ b/apps/desktop/src/lib/metadataTaskLimiter.ts @@ -0,0 +1,70 @@ +export interface MetadataTaskLimiterEvent { + scope: string; + kind: string; + active: number; + queued: number; +} + +type MetadataTaskLimiterLogger = (event: MetadataTaskLimiterEvent & { event: "start" | "queued" | "done" }) => void; + +interface QueuedTask { + kind: string; + run: () => Promise; + resolve: (value: T) => void; + reject: (reason?: unknown) => void; +} + +interface ScopeState { + active: number; + queue: QueuedTask[]; +} + +export class MetadataTaskLimiter { + private readonly scopes = new Map(); + + constructor( + private readonly maxActive = 2, + private readonly logger?: MetadataTaskLimiterLogger, + ) {} + + run(scope: string, kind: string, task: () => Promise): Promise { + const normalizedScope = scope || "global"; + const state = this.stateForScope(normalizedScope); + return new Promise((resolve, reject) => { + const queued: QueuedTask = { kind, run: task, resolve, reject }; + if (state.active < this.maxActive) { + this.start(normalizedScope, state, queued); + return; + } + state.queue.push(queued as QueuedTask); + this.logger?.({ event: "queued", scope: normalizedScope, kind, active: state.active, queued: state.queue.length }); + }); + } + + private stateForScope(scope: string): ScopeState { + let state = this.scopes.get(scope); + if (!state) { + state = { active: 0, queue: [] }; + this.scopes.set(scope, state); + } + return state; + } + + private start(scope: string, state: ScopeState, task: QueuedTask) { + state.active++; + this.logger?.({ event: "start", scope, kind: task.kind, active: state.active, queued: state.queue.length }); + task + .run() + .then(task.resolve, task.reject) + .finally(() => { + state.active--; + this.logger?.({ event: "done", scope, kind: task.kind, active: state.active, queued: state.queue.length }); + const next = state.queue.shift(); + if (next) { + this.start(scope, state, next); + } else if (state.active === 0) { + this.scopes.delete(scope); + } + }); + } +} diff --git a/apps/desktop/src/stores/__tests__/connectionStore.completion.spec.ts b/apps/desktop/src/stores/__tests__/connectionStore.completion.spec.ts index b43ce25f5d..0db390cbb4 100644 --- a/apps/desktop/src/stores/__tests__/connectionStore.completion.spec.ts +++ b/apps/desktop/src/stores/__tests__/connectionStore.completion.spec.ts @@ -25,6 +25,14 @@ function postgresConnection(): ConnectionConfig { } as ConnectionConfig; } +function deferred() { + let resolve!: (value: T) => void; + const promise = new Promise((res) => { + resolve = res; + }); + return { promise, resolve }; +} + describe("connectionStore completion assistant", () => { beforeEach(() => { vi.resetModules(); @@ -112,4 +120,43 @@ describe("connectionStore completion assistant", () => { expect(dimTables).toEqual([{ name: "dim_game", schema: "dim_game_base", type: "table" }]); expect(dwsTables).toEqual([]); }); + + it("limits concurrent completion column metadata requests per connection database", async () => { + const gates = [deferred(), deferred(), deferred(), deferred()]; + let activeColumns = 0; + let maxActiveColumns = 0; + const getColumns = vi.fn((_connectionId: string, _database: string, _schema: string, table: string) => { + const index = Number(table.replace("table_", "")); + activeColumns++; + maxActiveColumns = Math.max(maxActiveColumns, activeColumns); + return gates[index].promise.finally(() => { + activeColumns--; + }); + }); + + vi.doMock("@/lib/tauriRuntime", () => ({ isTauriRuntime: () => false })); + vi.doMock("@/lib/api", () => ({ + checkConnectionHealth: vi.fn().mockResolvedValue(undefined), + completionAssistantSearch: vi.fn().mockResolvedValue({ candidates: [], incomplete: false, fallback_used: false }), + getColumns, + })); + + const { useConnectionStore } = await import("@/stores/connectionStore"); + const store = useConnectionStore(); + store.connections = [postgresConnection()]; + store.connectedIds.add("pg-1"); + + const requests = [0, 1, 2, 3].map((index) => store.listCompletionColumns("pg-1", "app", `table_${index}`, "public")); + + await vi.waitFor(() => expect(getColumns).toHaveBeenCalledTimes(2)); + expect(maxActiveColumns).toBe(2); + gates[0].resolve([{ name: "id", data_type: "integer", is_nullable: false, column_default: null, is_primary_key: true, extra: null }]); + await vi.waitFor(() => expect(getColumns).toHaveBeenCalledTimes(3)); + gates[1].resolve([{ name: "id", data_type: "integer", is_nullable: false, column_default: null, is_primary_key: true, extra: null }]); + gates[2].resolve([{ name: "id", data_type: "integer", is_nullable: false, column_default: null, is_primary_key: true, extra: null }]); + gates[3].resolve([{ name: "id", data_type: "integer", is_nullable: false, column_default: null, is_primary_key: true, extra: null }]); + + await Promise.all(requests); + expect(maxActiveColumns).toBe(2); + }); }); diff --git a/apps/desktop/src/stores/connectionStore.ts b/apps/desktop/src/stores/connectionStore.ts index a57f3981b4..96c0cbb7f1 100644 --- a/apps/desktop/src/stores/connectionStore.ts +++ b/apps/desktop/src/stores/connectionStore.ts @@ -57,6 +57,7 @@ import { completionSchemasFromTree, completionTablesFromTree } from "@/lib/compl import { kvRootNodeLabel } from "@/lib/kvRootPresentation"; import { REDIS_SCAN_PAGE_SIZE_DEFAULT } from "@/lib/redisKeyPattern"; import { appendAgentDriverUpdateHint, hasAgentDriverUpdate, type AgentDriverInstallState } from "@/lib/agentDriverInstallHint"; +import { MetadataTaskLimiter } from "@/lib/metadataTaskLimiter"; import i18n from "@/i18n"; import type { MqAdminConfig } from "@/types/mq"; @@ -68,6 +69,7 @@ const METADATA_LOAD_MIN_TIMEOUT_MS = 15_000; const METADATA_LOAD_DISABLED_QUERY_TIMEOUT_MS = 60_000; const DISCONNECT_REQUEST_TIMEOUT_MS = 5_000; const DEFAULT_KEEPALIVE_INTERVAL_SECS = 30; +export const COMPLETION_METADATA_CONCURRENCY = 2; const MONGO_LEGACY_DRIVER_PROFILE = "mongodb-legacy"; const MONGO_LEGACY_DRIVER_LABEL = "MongoDB (Legacy)"; const SUPERSEDED_CONNECTION_ATTEMPT_MESSAGE = "Connection attempt was superseded by a newer attempt"; @@ -204,6 +206,9 @@ export const useConnectionStore = defineStore("connection", () => { const completionColumnIndex = new Map(); const completionForeignKeyIndex = new Map(); const completionInFlight = new Map>(); + const completionMetadataLimiter = new MetadataTaskLimiter(COMPLETION_METADATA_CONCURRENCY, (event) => { + console.debug("[DBX][completion-metadata:limit]", event); + }); const transferSource = ref<{ connectionId: string; database: string } | null>(null); const schemaDiffSource = ref<{ connectionId: string; database: string; schema?: string } | null>(null); const dataCompareSource = ref<{ @@ -2766,10 +2771,14 @@ export const useConnectionStore = defineStore("connection", () => { for (const [oldKey] of oldest) index.delete(oldKey); } - function withCompletionInFlight(key: string, load: () => Promise): Promise { + function completionLimiterScope(connectionId: string, database = ""): string { + return `${connectionId}:${database}`; + } + + function withCompletionInFlight(key: string, load: () => Promise, limit?: { scope: string; kind: string }): Promise { const existing = completionInFlight.get(key) as Promise | undefined; if (existing) return existing; - const promise = load().finally(() => { + const promise = (limit ? completionMetadataLimiter.run(limit.scope, limit.kind, load) : load()).finally(() => { if (completionInFlight.get(key) === promise) completionInFlight.delete(key); }); completionInFlight.set(key, promise); @@ -3048,16 +3057,20 @@ export const useConnectionStore = defineStore("connection", () => { if (completionDatabasesCache.value[connectionId]) { return completionDatabasesCache.value[connectionId]; } - return withCompletionInFlight(`${connectionId}:completion-databases`, async () => { - await ensureConnected(connectionId); - const config = getConfig(connectionId); - const databases = await api.listDatabases(connectionId); - completionDatabasesCache.value[connectionId] = filterDatabaseNamesForConnection( - databases.map((database) => database.name), - config, - ); - return completionDatabasesCache.value[connectionId]; - }); + return withCompletionInFlight( + `${connectionId}:completion-databases`, + async () => { + await ensureConnected(connectionId); + const config = getConfig(connectionId); + const databases = await api.listDatabases(connectionId); + completionDatabasesCache.value[connectionId] = filterDatabaseNamesForConnection( + databases.map((database) => database.name), + config, + ); + return completionDatabasesCache.value[connectionId]; + }, + { scope: completionLimiterScope(connectionId), kind: "databases" }, + ); } async function listCompletionSchemas(connectionId: string, database: string): Promise { @@ -3142,77 +3155,81 @@ export const useConnectionStore = defineStore("connection", () => { return completionTablesCache.value[cacheKey]; } - return withCompletionInFlight(`${cacheKey}:tables`, async () => { - await ensureConnected(connectionId); + return withCompletionInFlight( + `${cacheKey}:tables`, + async () => { + await ensureConnected(connectionId); - if (isSchemaAwareDatabase(connectionId)) { - if (normalizedFilter || limit) { - let results: SqlCompletionTable[] = []; - try { - results = await listCompletionAssistantTables(connectionId, database, normalizedFilter, limit, schema); - } catch { - if (schema) { - const tables = await api.listTables(connectionId, database, schema, normalizedFilter, limit); - results = tables.map((table) => ({ - name: table.name, - schema, - type: isViewLikeTableType(table.table_type) ? ("view" as const) : ("table" as const), - })); - } else { - results = lookupLocalCompletionTables(connectionId, database, normalizedFilter, limit); - } - } - if (results.length === 0 && relaxedFilter) { - if (schema) { - try { - const tables = await api.listTables(connectionId, database, schema, relaxedFilter, expandedCompletionLimit(limit)); + if (isSchemaAwareDatabase(connectionId)) { + if (normalizedFilter || limit) { + let results: SqlCompletionTable[] = []; + try { + results = await listCompletionAssistantTables(connectionId, database, normalizedFilter, limit, schema); + } catch { + if (schema) { + const tables = await api.listTables(connectionId, database, schema, normalizedFilter, limit); results = tables.map((table) => ({ name: table.name, schema, type: isViewLikeTableType(table.table_type) ? ("view" as const) : ("table" as const), })); - } catch { - results = []; + } else { + results = lookupLocalCompletionTables(connectionId, database, normalizedFilter, limit); } - } else { - results = lookupLocalCompletionTables(connectionId, database, relaxedFilter, expandedCompletionLimit(limit)); } + if (results.length === 0 && relaxedFilter) { + if (schema) { + try { + const tables = await api.listTables(connectionId, database, schema, relaxedFilter, expandedCompletionLimit(limit)); + results = tables.map((table) => ({ + name: table.name, + schema, + type: isViewLikeTableType(table.table_type) ? ("view" as const) : ("table" as const), + })); + } catch { + results = []; + } + } else { + results = lookupLocalCompletionTables(connectionId, database, relaxedFilter, expandedCompletionLimit(limit)); + } + } + const limitedTables = limit ? dedupeCompletionTables(results).slice(0, limit) : results; + completionTablesCache.value[cacheKey] = limitedTables; + indexCompletionTables(connectionId, database, undefined, limitedTables); + evictOldestCacheEntries(completionTablesCache.value, COMPLETION_CACHE_MAX); + return completionTablesCache.value[cacheKey]; } - const limitedTables = limit ? dedupeCompletionTables(results).slice(0, limit) : results; - completionTablesCache.value[cacheKey] = limitedTables; - indexCompletionTables(connectionId, database, undefined, limitedTables); + + if (schema) { + const tables = await api.listTables(connectionId, database, schema); + completionTablesCache.value[cacheKey] = tables.map((table) => ({ + name: table.name, + schema, + type: isViewLikeTableType(table.table_type) ? ("view" as const) : ("table" as const), + })); + } else { + completionTablesCache.value[cacheKey] = lookupLocalCompletionTables(connectionId, database, normalizedFilter, limit); + } + indexCompletionTables(connectionId, database, undefined, completionTablesCache.value[cacheKey]); evictOldestCacheEntries(completionTablesCache.value, COMPLETION_CACHE_MAX); return completionTablesCache.value[cacheKey]; } - if (schema) { - const tables = await api.listTables(connectionId, database, schema); - completionTablesCache.value[cacheKey] = tables.map((table) => ({ - name: table.name, - schema, - type: isViewLikeTableType(table.table_type) ? ("view" as const) : ("table" as const), - })); - } else { - completionTablesCache.value[cacheKey] = lookupLocalCompletionTables(connectionId, database, normalizedFilter, limit); + let tables = await api.listTables(connectionId, database, database, normalizedFilter, limit); + if (tables.length === 0 && relaxedFilter) { + tables = await api.listTables(connectionId, database, database, relaxedFilter, expandedCompletionLimit(limit)); } - indexCompletionTables(connectionId, database, undefined, completionTablesCache.value[cacheKey]); + completionTablesCache.value[cacheKey] = tables.map((table) => ({ + name: table.name, + type: isViewLikeTableType(table.table_type) ? ("view" as const) : ("table" as const), + })); + completionTablesCache.value[cacheKey] = limit ? completionTablesCache.value[cacheKey].slice(0, limit) : completionTablesCache.value[cacheKey]; + indexCompletionTables(connectionId, database, schema, completionTablesCache.value[cacheKey]); evictOldestCacheEntries(completionTablesCache.value, COMPLETION_CACHE_MAX); return completionTablesCache.value[cacheKey]; - } - - let tables = await api.listTables(connectionId, database, database, normalizedFilter, limit); - if (tables.length === 0 && relaxedFilter) { - tables = await api.listTables(connectionId, database, database, relaxedFilter, expandedCompletionLimit(limit)); - } - completionTablesCache.value[cacheKey] = tables.map((table) => ({ - name: table.name, - type: isViewLikeTableType(table.table_type) ? ("view" as const) : ("table" as const), - })); - completionTablesCache.value[cacheKey] = limit ? completionTablesCache.value[cacheKey].slice(0, limit) : completionTablesCache.value[cacheKey]; - indexCompletionTables(connectionId, database, schema, completionTablesCache.value[cacheKey]); - evictOldestCacheEntries(completionTablesCache.value, COMPLETION_CACHE_MAX); - return completionTablesCache.value[cacheKey]; - }); + }, + { scope: completionLimiterScope(connectionId, database), kind: "tables" }, + ); } function relaxedCompletionTableFilter(filter: string): string | undefined { @@ -3241,13 +3258,17 @@ export const useConnectionStore = defineStore("connection", () => { const normalizedFilter = filter.trim().toLowerCase(); const cacheKey = `${connectionId}:${database}:${schema ?? ""}`; if (!completionObjectsCache.value[cacheKey]) { - await withCompletionInFlight(`${cacheKey}:objects`, async () => { - await ensureConnected(connectionId); - const objects = isSchemaAwareDatabase(connectionId) ? await listSchemaAwareCompletionObjects(connectionId, database, schema) : await api.listCompletionObjects(connectionId, database, schema || database); - completionObjectsCache.value[cacheKey] = dedupeCompletionObjects(objects.map(toSqlCompletionObject).filter((object): object is SqlCompletionObject => object != null)); - indexCompletionObjects(connectionId, database, schema, completionObjectsCache.value[cacheKey]); - evictOldestCacheEntries(completionObjectsCache.value, COMPLETION_CACHE_MAX); - }); + await withCompletionInFlight( + `${cacheKey}:objects`, + async () => { + await ensureConnected(connectionId); + const objects = isSchemaAwareDatabase(connectionId) ? await listSchemaAwareCompletionObjects(connectionId, database, schema) : await api.listCompletionObjects(connectionId, database, schema || database); + completionObjectsCache.value[cacheKey] = dedupeCompletionObjects(objects.map(toSqlCompletionObject).filter((object): object is SqlCompletionObject => object != null)); + indexCompletionObjects(connectionId, database, schema, completionObjectsCache.value[cacheKey]); + evictOldestCacheEntries(completionObjectsCache.value, COMPLETION_CACHE_MAX); + }, + { scope: completionLimiterScope(connectionId, database), kind: "objects" }, + ); } const objects = completionObjectsCache.value[cacheKey]; @@ -3257,7 +3278,7 @@ export const useConnectionStore = defineStore("connection", () => { async function listSchemaAwareCompletionObjects(connectionId: string, database: string, schema?: string): Promise { const schemas = schema ? [schema] : await listCompletionSchemas(connectionId, database); - const batchSize = 5; + const batchSize = COMPLETION_METADATA_CONCURRENCY; const results: ObjectInfo[] = []; for (let i = 0; i < schemas.length; i += batchSize) { const batch = schemas.slice(i, i + batchSize); @@ -3323,33 +3344,37 @@ export const useConnectionStore = defineStore("connection", () => { } const cacheKey = `${connectionId}:${database}:${schema || ""}:${table}`; if (!completionColumnsCache.value[cacheKey]) { - await withCompletionInFlight(`${cacheKey}:columns`, async () => { - await ensureConnected(connectionId); - try { - const assistantColumns = await listCompletionAssistantColumns(connectionId, database, table, schema); - if (assistantColumns.length > 0) { - completionColumnsCache.value[cacheKey] = assistantColumns.map((column) => ({ - name: column.name, - data_type: column.dataType ?? "", - is_nullable: column.isNullable ?? true, - column_default: null, - is_primary_key: false, - extra: null, - comment: column.comment ?? null, - numeric_precision: null, - numeric_scale: null, - character_maximum_length: null, - })); - evictOldestCacheEntries(completionColumnsCache.value, COMPLETION_CACHE_MAX); - return; + await withCompletionInFlight( + `${cacheKey}:columns`, + async () => { + await ensureConnected(connectionId); + try { + const assistantColumns = await listCompletionAssistantColumns(connectionId, database, table, schema); + if (assistantColumns.length > 0) { + completionColumnsCache.value[cacheKey] = assistantColumns.map((column) => ({ + name: column.name, + data_type: column.dataType ?? "", + is_nullable: column.isNullable ?? true, + column_default: null, + is_primary_key: false, + extra: null, + comment: column.comment ?? null, + numeric_precision: null, + numeric_scale: null, + character_maximum_length: null, + })); + evictOldestCacheEntries(completionColumnsCache.value, COMPLETION_CACHE_MAX); + return; + } + } catch { + // Fall back to the existing metadata path below. } - } catch { - // Fall back to the existing metadata path below. - } - const querySchema = metadataQuerySchema(connectionId, database, schema); - completionColumnsCache.value[cacheKey] = await api.getColumns(connectionId, database, querySchema, table); - evictOldestCacheEntries(completionColumnsCache.value, COMPLETION_CACHE_MAX); - }); + const querySchema = metadataQuerySchema(connectionId, database, schema); + completionColumnsCache.value[cacheKey] = await api.getColumns(connectionId, database, querySchema, table); + evictOldestCacheEntries(completionColumnsCache.value, COMPLETION_CACHE_MAX); + }, + { scope: completionLimiterScope(connectionId, database), kind: "columns" }, + ); } const columns = completionColumnsCache.value[cacheKey].map((column) => ({ @@ -3373,12 +3398,16 @@ export const useConnectionStore = defineStore("connection", () => { const cacheKey = `${connectionId}:${database}:${schema || ""}:${table}`; if (!completionForeignKeysCache.value[cacheKey]) { - await withCompletionInFlight(`${cacheKey}:fkeys`, async () => { - await ensureConnected(connectionId); - const querySchema = metadataQuerySchema(connectionId, database, schema); - completionForeignKeysCache.value[cacheKey] = await api.listForeignKeys(connectionId, database, querySchema, table); - evictOldestCacheEntries(completionForeignKeysCache.value, COMPLETION_CACHE_MAX); - }); + await withCompletionInFlight( + `${cacheKey}:fkeys`, + async () => { + await ensureConnected(connectionId); + const querySchema = metadataQuerySchema(connectionId, database, schema); + completionForeignKeysCache.value[cacheKey] = await api.listForeignKeys(connectionId, database, querySchema, table); + evictOldestCacheEntries(completionForeignKeysCache.value, COMPLETION_CACHE_MAX); + }, + { scope: completionLimiterScope(connectionId, database), kind: "foreignKeys" }, + ); } const foreignKeys = sqlCompletionForeignKeys(completionForeignKeysCache.value[cacheKey]); diff --git a/apps/desktop/src/stores/queryStore.ts b/apps/desktop/src/stores/queryStore.ts index c24592567b..c035cbc34f 100644 --- a/apps/desktop/src/stores/queryStore.ts +++ b/apps/desktop/src/stores/queryStore.ts @@ -1912,7 +1912,7 @@ export const useQueryStore = defineStore("query", () => { } console.info("[DBX][executeTabSql:execute-multi:start]", { traceId, elapsed: elapsed() }); - const clientSessionId = tab.mode === "query" ? tabClientSessionId(tab) : undefined; + const clientSessionId = tab.mode === "query" || tab.mode === "data" ? tabClientSessionId(tab) : undefined; const executionOptions = { ...(typeof pageLimit === "number" ? useAgentResultSession diff --git a/packages/app-tests/metadataTaskLimiter.test.ts b/packages/app-tests/metadataTaskLimiter.test.ts new file mode 100644 index 0000000000..d9223f7925 --- /dev/null +++ b/packages/app-tests/metadataTaskLimiter.test.ts @@ -0,0 +1,74 @@ +import assert from "node:assert/strict"; +import { test } from "vitest"; +import { MetadataTaskLimiter } from "../../apps/desktop/src/lib/metadataTaskLimiter.ts"; + +function deferred() { + let resolve!: (value: T) => void; + let reject!: (reason?: unknown) => void; + const promise = new Promise((res, rej) => { + resolve = res; + reject = rej; + }); + return { promise, resolve, reject }; +} + +test("metadata task limiter caps active tasks per scope and drains queued tasks", async () => { + const events: Array<{ event: string; active: number; queued: number; kind: string }> = []; + const limiter = new MetadataTaskLimiter(2, (event) => events.push(event)); + const gates = [deferred(), deferred(), deferred(), deferred()]; + let active = 0; + let maxActive = 0; + + const tasks = gates.map((gate, index) => + limiter.run("conn:db", `task-${index}`, async () => { + active++; + maxActive = Math.max(maxActive, active); + const result = await gate.promise; + active--; + return result; + }), + ); + + assert.equal(maxActive, 2); + assert.deepEqual( + events.filter((event) => event.event === "queued").map((event) => event.queued), + [1, 2], + ); + + gates[0].resolve(1); + await Promise.resolve(); + await Promise.resolve(); + assert.equal(maxActive, 2); + + gates[1].resolve(2); + gates[2].resolve(3); + gates[3].resolve(4); + + assert.deepEqual(await Promise.all(tasks), [1, 2, 3, 4]); + assert.equal(maxActive, 2); +}); + +test("metadata task limiter isolates independent scopes", async () => { + const limiter = new MetadataTaskLimiter(1); + const first = deferred(); + const second = deferred(); + let active = 0; + + const a = limiter.run("conn:a", "columns", async () => { + active++; + const result = await first.promise; + active--; + return result; + }); + const b = limiter.run("conn:b", "columns", async () => { + active++; + const result = await second.promise; + active--; + return result; + }); + + assert.equal(active, 2); + first.resolve("a"); + second.resolve("b"); + assert.deepEqual(await Promise.all([a, b]), ["a", "b"]); +}); diff --git a/packages/app-tests/queryStore.test.ts b/packages/app-tests/queryStore.test.ts index 1a356fc494..852fa4911e 100644 --- a/packages/app-tests/queryStore.test.ts +++ b/packages/app-tests/queryStore.test.ts @@ -2612,6 +2612,40 @@ test("query execution is scoped to the tab client session", async () => { } }); +test("data tab execution is scoped to the tab client session", async () => { + const restoreStorage = installMemoryStorage(); + setActivePinia(createPinia()); + const connectionStore = useConnectionStore(); + const store = useQueryStore(); + const originalFetch = globalThis.fetch; + + connectionStore.addEphemeralConnection(conn("conn-1")); + const tabId = store.createTab("conn-1", "db", "users", "data", "public"); + let executeBody: any; + + globalThis.fetch = withConnectionHealthMock(async (input, init) => { + const url = String(input); + if (url === "/api/query/execute-multi") { + executeBody = JSON.parse(String(init?.body ?? "{}")); + return new Response(JSON.stringify([{ columns: ["id"], rows: [[1]], affected_rows: 0, execution_time_ms: 1 }]), { + status: 200, + headers: { "Content-Type": "application/json" }, + }); + } + return new Response("unexpected request", { status: 500 }); + }); + + try { + await store.executeTabSql(tabId, "select * from users"); + + assert.equal(executeBody.clientSessionId, tabId); + assert.equal(executeBody.timeoutSecs, 30); + } finally { + globalThis.fetch = originalFetch; + restoreStorage(); + } +}); + test("query execution keeps automatically counting total rows in the background", async () => { const restoreStorage = installMemoryStorage(); setActivePinia(createPinia()); diff --git a/packages/app-tests/useDataGridActions.test.ts b/packages/app-tests/useDataGridActions.test.ts new file mode 100644 index 0000000000..f455631955 --- /dev/null +++ b/packages/app-tests/useDataGridActions.test.ts @@ -0,0 +1,122 @@ +import assert from "node:assert/strict"; +import { computed } from "vue"; +import { createPinia, setActivePinia } from "pinia"; +import { test, vi } from "vitest"; +import { useConnectionStore } from "../../apps/desktop/src/stores/connectionStore.ts"; +import { useQueryStore } from "../../apps/desktop/src/stores/queryStore.ts"; +import type { ColumnInfo, ConnectionConfig } from "../../apps/desktop/src/types/database.ts"; + +vi.mock("vue-i18n", async () => { + const actual = await vi.importActual("vue-i18n"); + return { + ...actual, + useI18n: () => ({ t: (key: string) => key }), + }; +}); + +vi.mock("@/composables/useToast", () => ({ + useToast: () => ({ toast: vi.fn() }), +})); + +function installMemoryStorage() { + const values = new Map(); + const original = Object.getOwnPropertyDescriptor(globalThis, "localStorage"); + Object.defineProperty(globalThis, "localStorage", { + configurable: true, + value: { + getItem: (key: string) => values.get(key) ?? null, + setItem: (key: string, value: string) => values.set(key, value), + removeItem: (key: string) => values.delete(key), + clear: () => values.clear(), + }, + }); + return () => { + if (original) Object.defineProperty(globalThis, "localStorage", original); + else Reflect.deleteProperty(globalThis, "localStorage"); + }; +} + +function conn(id: string): ConnectionConfig { + return { + id, + name: id, + db_type: "mysql", + host: "localhost", + port: 3306, + username: "root", + password: "", + }; +} + +async function waitFor(predicate: () => boolean, timeoutMs = 1000) { + const started = Date.now(); + while (!predicate()) { + if (Date.now() - started > timeoutMs) throw new Error("timed out waiting for condition"); + await new Promise((resolve) => setTimeout(resolve, 5)); + } +} + +test("data reload executes before slow metadata refresh completes", async () => { + const restoreStorage = installMemoryStorage(); + const originalFetch = globalThis.fetch; + const { useDataGridActions } = await import("../../apps/desktop/src/composables/useDataGridActions.ts"); + let executeBody: any; + let resolveColumns!: (columns: ColumnInfo[]) => void; + + globalThis.fetch = (async (input, init) => { + const url = new URL(String(input), "http://localhost"); + if (url.pathname === "/api/connection/check-health") { + return Response.json(null); + } + if (url.pathname === "/api/schema/columns") { + return Response.json(await new Promise((resolve) => (resolveColumns = resolve))); + } + if (url.pathname === "/api/schema/indexes") { + return Response.json([]); + } + if (url.pathname === "/api/query/build-table-select-sql") { + return Response.json("SELECT * FROM `users` LIMIT 50 OFFSET 0"); + } + if (url.pathname === "/api/query/execute-multi") { + executeBody = JSON.parse(String(init?.body ?? "{}")); + return Response.json([{ columns: ["id"], rows: [[1]], affected_rows: 0, execution_time_ms: 1 }]); + } + return new Response(`unexpected ${url.pathname}`, { status: 500 }); + }) as typeof fetch; + + try { + setActivePinia(createPinia()); + const connectionStore = useConnectionStore(); + const queryStore = useQueryStore(); + connectionStore.addEphemeralConnection(conn("mysql-1")); + const tabId = queryStore.createTab("mysql-1", "app", "users", "data"); + queryStore.setTableMeta(tabId, { tableName: "users", tableType: "TABLE", columns: [], primaryKeys: [] }); + const tab = queryStore.tabs.find((item) => item.id === tabId); + assert.ok(tab); + + const actions = useDataGridActions(computed(() => tab)); + const reload = actions.onReloadData(undefined, undefined, undefined, undefined, 50, 0); + + await waitFor(() => !!executeBody); + assert.equal(executeBody.clientSessionId, tabId); + assert.deepEqual(tab.result?.rows, [[1]]); + + await waitFor(() => typeof resolveColumns === "function"); + resolveColumns([ + { + name: "id", + data_type: "int", + is_nullable: false, + column_default: null, + is_primary_key: true, + extra: null, + }, + ]); + await reload; + await waitFor(() => tab.tableMeta?.columns.length === 1); + assert.equal(tab.tableMeta?.primaryKeys[0], "id"); + } finally { + globalThis.fetch = originalFetch; + restoreStorage(); + } +});