From 7ec66dc732d5dc836daa38c0c677d8b1dd0931db Mon Sep 17 00:00:00 2001 From: Syngnat Date: Tue, 4 Aug 2026 11:07:08 +0800 Subject: [PATCH] =?UTF-8?q?=F0=9F=90=9B=20fix(query-editor/db):=20?= =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E8=BF=90=E8=A1=8C=E4=B8=AD=20SQL=20=E6=97=A0?= =?UTF-8?q?=E6=B3=95=E5=81=9C=E6=AD=A2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 查询 ID 生成前支持终止执行，并统一刷新、翻页和 Mongo 多语句的取消代际 - 后端在连接建立和事务锁等待前登记取消，保留登记直到执行 owner 退出 - 使用 registration ID 防止旧任务清理同 ID 的新查询 - 补充前后端取消竞态与并发回归测试 Refs #754 --- .../QueryEditor.external-sql-save.test.tsx | 20 +- .../QueryEditor.results-and-drop.test.tsx | 268 +++++++++++++++++- frontend/src/components/QueryEditor.tsx | 160 +++++++++-- internal/app/app.go | 29 ++ internal/app/methods_db.go | 47 +-- internal/app/methods_db_cancel_test.go | 141 +++++++++ internal/app/methods_db_transaction.go | 53 ++-- 7 files changed, 612 insertions(+), 106 deletions(-) diff --git a/frontend/src/components/QueryEditor.external-sql-save.test.tsx b/frontend/src/components/QueryEditor.external-sql-save.test.tsx index 1c6c5ac6..0f6df697 100644 --- a/frontend/src/components/QueryEditor.external-sql-save.test.tsx +++ b/frontend/src/components/QueryEditor.external-sql-save.test.tsx @@ -13222,11 +13222,14 @@ WHERE GRANTEE = 'APPUSER';`; }); }); - it('shows "No running query to cancel." in English when stop is clicked before a query id exists', async () => { + it('cancels the pending run before a query id exists', async () => { storeState.languagePreference = 'en-US'; setCurrentLanguage('en-US'); - backendApp.GenerateQueryID.mockReturnValueOnce(new Promise(() => {})); + let resolveQueryId!: (queryId: string) => void; + backendApp.GenerateQueryID.mockReturnValueOnce(new Promise((resolve) => { + resolveQueryId = resolve; + })); let renderer!: ReactTestRenderer; await act(async () => { @@ -13242,8 +13245,16 @@ WHERE GRANTEE = 'APPUSER';`; await findButton(renderer, 'Stop').props.onClick(); }); - expect(messageApi.warning).toHaveBeenCalledWith('No running query to cancel.'); - expect(messageApi.warning).not.toHaveBeenCalledWith('没有正在运行的查询可取消'); + expect(messageApi.success).toHaveBeenCalledWith('Query canceled.'); + expect(messageApi.warning).not.toHaveBeenCalledWith('No running query to cancel.'); + expect(findButtons(renderer, 'Stop')).toHaveLength(0); + + await act(async () => { + resolveQueryId('query-too-late'); + await Promise.resolve(); + await Promise.resolve(); + }); + expect(backendApp.DBQueryMulti).not.toHaveBeenCalled(); }); it('shows "Query canceled." in English when stop cancels a running query', async () => { @@ -13271,6 +13282,7 @@ WHERE GRANTEE = 'APPUSER';`; expect(messageApi.success).toHaveBeenCalledWith('Query canceled.'); expect(messageApi.success).not.toHaveBeenCalledWith('查询已取消'); + expect(findButtons(renderer, 'Stop')).toHaveLength(0); }); it('shows "Failed to cancel query" in English while preserving the raw error detail', async () => { diff --git a/frontend/src/components/QueryEditor.results-and-drop.test.tsx b/frontend/src/components/QueryEditor.results-and-drop.test.tsx index b7502ae8..da3a06f8 100644 --- a/frontend/src/components/QueryEditor.results-and-drop.test.tsx +++ b/frontend/src/components/QueryEditor.results-and-drop.test.tsx @@ -2740,11 +2740,14 @@ describe('QueryEditor external SQL save', () => { }); }); - it('shows "No running query to cancel." in English when stop is clicked before a query id exists', async () => { + it('cancels the pending run before a query id exists', async () => { storeState.languagePreference = 'en-US'; setCurrentLanguage('en-US'); - backendApp.GenerateQueryID.mockReturnValueOnce(new Promise(() => {})); + let resolveQueryId!: (queryId: string) => void; + backendApp.GenerateQueryID.mockReturnValueOnce(new Promise((resolve) => { + resolveQueryId = resolve; + })); let renderer!: ReactTestRenderer; await act(async () => { @@ -2760,8 +2763,16 @@ describe('QueryEditor external SQL save', () => { await findButton(renderer, 'Stop').props.onClick(); }); - expect(messageApi.warning).toHaveBeenCalledWith('No running query to cancel.'); - expect(messageApi.warning).not.toHaveBeenCalledWith('没有正在运行的查询可取消'); + expect(messageApi.success).toHaveBeenCalledWith('Query canceled.'); + expect(messageApi.warning).not.toHaveBeenCalledWith('No running query to cancel.'); + expect(findButtons(renderer, 'Stop')).toHaveLength(0); + + await act(async () => { + resolveQueryId('query-too-late'); + await Promise.resolve(); + await Promise.resolve(); + }); + expect(backendApp.DBQueryMulti).not.toHaveBeenCalled(); }); it('shows "Query canceled." in English when stop cancels a running query', async () => { @@ -2789,6 +2800,255 @@ describe('QueryEditor external SQL save', () => { expect(messageApi.success).toHaveBeenCalledWith('Query canceled.'); expect(messageApi.success).not.toHaveBeenCalledWith('查询已取消'); + expect(findButtons(renderer, 'Stop')).toHaveLength(0); + }); + + it('keeps the newer query cancellable when the previous run finishes late', async () => { + let resolvePreviousQuery!: (value: unknown) => void; + const previousQuery = new Promise((resolve) => { + resolvePreviousQuery = resolve; + }); + const currentQuery = new Promise(() => {}); + + backendApp.GenerateQueryID + .mockResolvedValueOnce('query-previous') + .mockResolvedValueOnce('query-current'); + backendApp.DBQueryMulti + .mockReturnValueOnce(previousQuery) + .mockReturnValueOnce(currentQuery); + backendApp.CancelQuery.mockResolvedValue({ success: true }); + + let renderer!: ReactTestRenderer; + await act(async () => { + renderer = create(); + }); + + await act(async () => { + void findButton(renderer, '运行').props.onClick(); + await Promise.resolve(); + await Promise.resolve(); + }); + expect(backendApp.DBQueryMulti).toHaveBeenCalledTimes(1); + + await act(async () => { + void findButton(renderer, '运行').props.onClick(); + await Promise.resolve(); + await Promise.resolve(); + }); + expect(backendApp.DBQueryMulti).toHaveBeenCalledTimes(2); + expect(backendApp.CancelQuery).toHaveBeenCalledWith('query-previous'); + + await act(async () => { + resolvePreviousQuery({ success: false, message: 'context canceled' }); + await Promise.resolve(); + await Promise.resolve(); + }); + + backendApp.CancelQuery.mockClear(); + messageApi.warning.mockClear(); + await act(async () => { + await findButton(renderer, '停止').props.onClick(); + }); + + expect(backendApp.CancelQuery).toHaveBeenCalledWith('query-current'); + expect(messageApi.warning).not.toHaveBeenCalledWith('没有正在运行的查询可取消。'); + }); + + it('does not start a replacement run after stop cancels it while the previous query cancellation is pending', async () => { + let resolveReplacementCancel!: (value: { success: boolean }) => void; + backendApp.GenerateQueryID + .mockResolvedValueOnce('query-previous') + .mockResolvedValueOnce('query-replacement'); + backendApp.DBQueryMulti + .mockReturnValueOnce(new Promise(() => {})) + .mockResolvedValueOnce({ success: true, data: [] }); + backendApp.CancelQuery + .mockReturnValueOnce(new Promise((resolve) => { + resolveReplacementCancel = resolve; + })) + .mockResolvedValueOnce({ success: true }); + + let renderer!: ReactTestRenderer; + await act(async () => { + renderer = create(); + }); + + await act(async () => { + void findButton(renderer, '运行').props.onClick(); + await Promise.resolve(); + await Promise.resolve(); + }); + expect(backendApp.DBQueryMulti).toHaveBeenCalledTimes(1); + + await act(async () => { + void findButton(renderer, '运行').props.onClick(); + await Promise.resolve(); + await Promise.resolve(); + }); + expect(backendApp.CancelQuery).toHaveBeenCalledTimes(1); + + await act(async () => { + await findButton(renderer, '停止').props.onClick(); + }); + expect(findButtons(renderer, '停止')).toHaveLength(0); + + await act(async () => { + resolveReplacementCancel({ success: true }); + await Promise.resolve(); + await Promise.resolve(); + await Promise.resolve(); + }); + + expect(backendApp.GenerateQueryID).toHaveBeenCalledTimes(1); + expect(backendApp.DBQueryMulti).toHaveBeenCalledTimes(1); + }); + + it('cancels a pending result refresh before its query id exists', async () => { + let resolveRefreshQueryId!: (queryId: string) => void; + backendApp.GenerateQueryID + .mockResolvedValueOnce('query-initial') + .mockReturnValueOnce(new Promise((resolve) => { + resolveRefreshQueryId = resolve; + })); + backendApp.DBQueryMulti + .mockResolvedValueOnce({ + success: true, + data: [{ columns: ['value'], rows: [{ value: 1 }] }], + }) + .mockResolvedValueOnce({ + success: true, + data: [{ columns: ['value'], rows: [{ value: 2 }] }], + }); + + let renderer!: ReactTestRenderer; + await act(async () => { + renderer = create(); + }); + await act(async () => { + await findButton(renderer, '运行').props.onClick(); + await Promise.resolve(); + await Promise.resolve(); + }); + expect(dataGridState.latestProps?.onReload).toEqual(expect.any(Function)); + + await act(async () => { + void dataGridState.latestProps.onReload(); + await Promise.resolve(); + }); + await act(async () => { + await findButton(renderer, '停止').props.onClick(); + }); + + await act(async () => { + resolveRefreshQueryId('query-refresh-too-late'); + await Promise.resolve(); + await Promise.resolve(); + await Promise.resolve(); + }); + + expect(messageApi.success).toHaveBeenCalledWith('查询已中止。'); + expect(backendApp.DBQueryMulti).toHaveBeenCalledTimes(1); + }); + + it('cancels a pending page query before its query id exists', async () => { + storeState.queryOptions.maxRows = 2; + let resolvePageQueryId!: (queryId: string) => void; + backendApp.GenerateQueryID + .mockResolvedValueOnce('query-initial') + .mockReturnValueOnce(new Promise((resolve) => { + resolvePageQueryId = resolve; + })); + backendApp.DBQueryMulti + .mockResolvedValueOnce({ + success: true, + data: [{ columns: ['value'], rows: [{ value: 1 }, { value: 2 }] }], + }) + .mockResolvedValueOnce({ + success: true, + data: [{ columns: ['value'], rows: [{ value: 3 }] }], + }); + + let renderer!: ReactTestRenderer; + await act(async () => { + renderer = create(); + }); + await act(async () => { + await findButton(renderer, '运行').props.onClick(); + await Promise.resolve(); + await Promise.resolve(); + }); + expect(dataGridState.latestProps?.onPageChange).toEqual(expect.any(Function)); + + await act(async () => { + void dataGridState.latestProps.onPageChange(2, 2); + await Promise.resolve(); + }); + expect(dataGridState.latestProps?.loading).toBe(true); + await act(async () => { + await findButton(renderer, '停止').props.onClick(); + }); + expect(dataGridState.latestProps?.loading).toBe(false); + + await act(async () => { + resolvePageQueryId('query-page-too-late'); + await Promise.resolve(); + await Promise.resolve(); + await Promise.resolve(); + }); + + expect(messageApi.success).toHaveBeenCalledWith('查询已中止。'); + expect(backendApp.DBQueryMulti).toHaveBeenCalledTimes(1); + }); + + it('cancels a Mongo multi-statement run between statement query ids', async () => { + storeState.connections[0].config.type = 'mongodb'; + const query = 'db.users.find({});\ndb.logs.find({});'; + let resolveSecondQueryId!: (queryId: string) => void; + backendApp.GenerateQueryID + .mockResolvedValueOnce('query-mongo-first') + .mockReturnValueOnce(new Promise((resolve) => { + resolveSecondQueryId = resolve; + })); + backendApp.DBQueryWithCancel + .mockResolvedValueOnce({ success: true, data: [{ _id: 1 }], fields: ['_id'] }) + .mockResolvedValueOnce({ success: true, data: [{ _id: 2 }], fields: ['_id'] }); + backendApp.CancelQuery.mockResolvedValue({ success: false, message: 'query already completed' }); + + let renderer!: ReactTestRenderer; + await act(async () => { + renderer = create(); + }); + editorState.selection = { + startLineNumber: 1, + startColumn: 1, + endLineNumber: 2, + endColumn: 'db.logs.find({});'.length + 1, + positionLineNumber: 2, + positionColumn: 'db.logs.find({});'.length + 1, + }; + + await act(async () => { + void findButton(renderer, '运行').props.onClick(); + await Promise.resolve(); + await Promise.resolve(); + await Promise.resolve(); + }); + expect(backendApp.GenerateQueryID).toHaveBeenCalledTimes(2); + expect(backendApp.DBQueryWithCancel).toHaveBeenCalledTimes(1); + + await act(async () => { + await findButton(renderer, '停止').props.onClick(); + }); + await act(async () => { + resolveSecondQueryId('query-mongo-too-late'); + await Promise.resolve(); + await Promise.resolve(); + await Promise.resolve(); + }); + + expect(messageApi.success).toHaveBeenCalledWith('查询已中止。'); + expect(backendApp.CancelQuery).not.toHaveBeenCalled(); + expect(backendApp.DBQueryWithCancel).toHaveBeenCalledTimes(1); }); it('shows "Failed to cancel query" in English while preserving the raw error detail', async () => { diff --git a/frontend/src/components/QueryEditor.tsx b/frontend/src/components/QueryEditor.tsx index 65b0b84a..a594f5ee 100644 --- a/frontend/src/components/QueryEditor.tsx +++ b/frontend/src/components/QueryEditor.tsx @@ -7365,8 +7365,25 @@ const QueryEditor: React.FC<{ tab: TabData; isActive?: boolean }> = ({ tab, isAc { oceanBaseProtocol: String((config as any).oceanBaseProtocol || '') }, )).trim().toLowerCase(); + const runSeq = ++runSeqRef.current; + const isCurrentRun = () => runSeqRef.current === runSeq; + let runQueryId = ''; + setExecutionRunToken(runSeq); + setLoading(true); + try { - setLoading(true); + if (currentQueryIdRef.current) { + const previousQueryId = currentQueryIdRef.current; + try { + await CancelQuery(previousQueryId); + } catch { + // The previous query may already have completed. + } + if (!isCurrentRun()) return; + if (currentQueryIdRef.current === previousQueryId) { + clearQueryId(); + } + } // 保持与首次执行一致的后端路径，必要时复用挂起事务 let queryId: string; try { @@ -7374,6 +7391,9 @@ const QueryEditor: React.FC<{ tab: TabData; isActive?: boolean }> = ({ tab, isAc } catch { queryId = 'reload-' + Date.now(); } + if (!isCurrentRun()) return; + runQueryId = queryId; + setQueryId(queryId); const res = await executeSqlEditorMultiQuery( config, currentDb, @@ -7382,6 +7402,11 @@ const QueryEditor: React.FC<{ tab: TabData; isActive?: boolean }> = ({ tab, isAc splitSQLStatements(sql, normalizedDbType), normalizedDbType, ); + if (!isCurrentRun()) return; + if (currentQueryIdRef.current === queryId) { + clearQueryId(); + runQueryId = ''; + } if (!res?.success) { message.error(translate('query_editor.message.refresh_failed', { error: formatSqlExecutionError(res?.message || translate('common.unknown'), { translate }), @@ -7428,11 +7453,15 @@ const QueryEditor: React.FC<{ tab: TabData; isActive?: boolean }> = ({ tab, isAc : rs )); } catch (err: any) { + if (!isCurrentRun()) return; message.error(translate('query_editor.message.refresh_failed', { error: formatSqlExecutionError(err?.message || err || translate('common.unknown'), { translate }), })); } finally { - setLoading(false); + if (isCurrentRun()) setLoading(false); + if (runQueryId && currentQueryIdRef.current === runQueryId) { + clearQueryId(); + } } }; @@ -7458,7 +7487,6 @@ const QueryEditor: React.FC<{ tab: TabData; isActive?: boolean }> = ({ tab, isAc { oceanBaseProtocol: String((config as any).oceanBaseProtocol || '') }, )).toLowerCase(); const sequence = ++resultTotalCountSeqRef.current; - const requestRunSequence = runSeqRef.current; resultTotalCountRequestsRef.current[resultKey] = { sequence, queryId: '' }; setResultSets(prev => prev.map(rs => rs.key === resultKey && rs.page @@ -7468,7 +7496,6 @@ const QueryEditor: React.FC<{ tab: TabData; isActive?: boolean }> = ({ tab, isAc const countStartedAt = Date.now(); const isCurrentRequest = () => { if (resultTotalCountRequestsRef.current[resultKey]?.sequence !== sequence) return false; - if (runSeqRef.current !== requestRunSequence) return false; const currentResult = resultSetsRef.current.find((item) => item.key === resultKey); return currentResult?.page?.baseSql === target.page?.baseSql; }; @@ -7638,19 +7665,39 @@ const QueryEditor: React.FC<{ tab: TabData; isActive?: boolean }> = ({ tab, isAc sortInfo: sortInfoOverride || target.sortInfo || [], }); + const runSeq = ++runSeqRef.current; + const isCurrentRun = () => runSeqRef.current === runSeq; + let runQueryId = ''; + setExecutionRunToken(runSeq); + setLoading(true); + try { - setLoading(true); setResultSets(prev => prev.map(rs => rs.key === resultKey && rs.page ? { ...rs, page: { ...rs.page, loading: true } } : rs )); + if (currentQueryIdRef.current) { + const previousQueryId = currentQueryIdRef.current; + try { + await CancelQuery(previousQueryId); + } catch { + // The previous query may already have completed. + } + if (!isCurrentRun()) return; + if (currentQueryIdRef.current === previousQueryId) { + clearQueryId(); + } + } let queryId: string; try { queryId = await GenerateQueryID(); } catch { queryId = 'query-page-' + Date.now(); } + if (!isCurrentRun()) return; + runQueryId = queryId; + setQueryId(queryId); const res = await executeSqlEditorMultiQuery( config, currentDb, @@ -7659,6 +7706,11 @@ const QueryEditor: React.FC<{ tab: TabData; isActive?: boolean }> = ({ tab, isAc splitSQLStatements(pageSql, normalizedDbType), normalizedDbType, ); + if (!isCurrentRun()) return; + if (currentQueryIdRef.current === queryId) { + clearQueryId(); + runQueryId = ''; + } if (!res?.success) { message.error(translate('query_editor.message.page_query_failed', { error: formatSqlExecutionError(res?.message || translate('common.unknown'), { translate }), @@ -7713,16 +7765,22 @@ const QueryEditor: React.FC<{ tab: TabData; isActive?: boolean }> = ({ tab, isAc }; })); } catch (err: any) { + if (!isCurrentRun()) return; message.error(translate('query_editor.message.page_query_failed', { error: formatSqlExecutionError(err?.message || err || translate('common.unknown'), { translate }), })); } finally { - setLoading(false); - setResultSets(prev => prev.map(rs => - rs.key === resultKey && rs.page?.loading - ? { ...rs, page: { ...rs.page, loading: false } } - : rs - )); + if (isCurrentRun()) { + setLoading(false); + setResultSets(prev => prev.map(rs => + rs.key === resultKey && rs.page?.loading + ? { ...rs, page: { ...rs.page, loading: false } } + : rs + )); + } + if (runQueryId && currentQueryIdRef.current === runQueryId) { + clearQueryId(); + } } }; @@ -7988,38 +8046,46 @@ const QueryEditor: React.FC<{ tab: TabData; isActive?: boolean }> = ({ tab, isAc message.error(translate('query_editor.message.select_database_first')); return; } + + const runSeq = ++runSeqRef.current; + let runQueryId = ''; + const isCurrentRun = () => runSeqRef.current === runSeq; + setExecutionRunToken(runSeq); + setLoading(true); + setExecutionError(''); + const runStartTime = Date.now(); + await cancelResultTotalCountRequests(Object.keys(resultTotalCountRequestsRef.current)); + if (!isCurrentRun()) return; // 如果已有查询在运行，先取消它 if (currentQueryIdRef.current) { + const previousQueryID = currentQueryIdRef.current; try { - await CancelQuery(currentQueryIdRef.current); + await CancelQuery(previousQueryID); } catch (error) { // 忽略取消错误，可能查询已完成 } - // 清除旧查询ID - clearQueryId(); + if (!isCurrentRun()) return; + if (currentQueryIdRef.current === previousQueryID) { + clearQueryId(); + } } - const runSeq = ++runSeqRef.current; - setExecutionRunToken(runSeq); - setLoading(true); - setExecutionError(''); - const runStartTime = Date.now(); const conn = connections.find(c => c.id === currentConnectionId); if (!conn) { message.error(translate('query_editor.message.connection_not_found')); - if (runSeqRef.current === runSeq) setLoading(false); + if (isCurrentRun()) setLoading(false); return; } const connCaps = getDataSourceCapabilities(conn.config); if (!connCaps.supportsQueryEditor) { message.error(translate('query_editor.message.unsupported_source')); - if (runSeqRef.current === runSeq) setLoading(false); + if (isCurrentRun()) setLoading(false); return; } const restrictedStatements = findConnectionMutatingStatements(conn.config, executableSQL); if (restrictedStatements.length > 0) { message.warning(translate('query_editor.message.connection_readonly_blocked')); - if (runSeqRef.current === runSeq) setLoading(false); + if (isCurrentRun()) setLoading(false); return; } @@ -8030,8 +8096,9 @@ const QueryEditor: React.FC<{ tab: TabData; isActive?: boolean }> = ({ tab, isAc target: currentDb, translate, }); + if (!isCurrentRun()) return; if (!approved) { - if (runSeqRef.current === runSeq) setLoading(false); + setLoading(false); return; } } @@ -8116,9 +8183,16 @@ const QueryEditor: React.FC<{ tab: TabData; isActive?: boolean }> = ({ tab, isAc console.warn('GenerateQueryID failed, using local UUID fallback:', error); queryId = 'query-' + uuidv4(); } + if (!isCurrentRun()) return; + runQueryId = queryId; setQueryId(queryId); const res = await DBQueryWithCancel(buildRpcConnectionConfig(config) as any, currentDb, executedSql, queryId); + if (!isCurrentRun()) return; + if (currentQueryIdRef.current === queryId) { + clearQueryId(); + runQueryId = ''; + } const legacyResultMessages = normalizeQueryResultMessages(res?.messages); const duration = Date.now() - startTime; addSqlLog({ @@ -8313,6 +8387,7 @@ const QueryEditor: React.FC<{ tab: TabData; isActive?: boolean }> = ({ tab, isAc let exactQualifiedTable: string | undefined; for (const oracleLookupDbName of oracleLookupDbCandidates) { const oracleTables = oracleLookupDbName ? await getOracleTablesForDb(oracleLookupDbName) : []; + if (!isCurrentRun()) return; if ( isOracleBaseTableReference(statement, oracleLookupDbName, oracleTables) ) { @@ -8341,7 +8416,7 @@ const QueryEditor: React.FC<{ tab: TabData; isActive?: boolean }> = ({ tab, isAc for (let index = 0; index < sourceStatements.length; index += 1) { const statementForPlan = executedSourceStatements[index] || sourceStatements[index]; try { - statementPlans.push(await resolveQueryLocatorPlan({ + const statementPlan = await resolveQueryLocatorPlan({ statement: statementForPlan, originalStatement: sourceStatements[index], dbType: normalizedDbType, @@ -8349,8 +8424,11 @@ const QueryEditor: React.FC<{ tab: TabData; isActive?: boolean }> = ({ tab, isAc config, forceReadOnly: forceReadOnlyResult, allowOracleRowID: allowOracleRowIDByStatement[index], - })); + }); + if (!isCurrentRun()) return; + statementPlans.push(statementPlan); } catch (planError) { + if (!isCurrentRun()) return; // 行定位计划失败绝不能阻断查询执行，兜底裸计划保证结果页始终呈现。 console.warn('resolveQueryLocatorPlan failed; falling back to a bare statement plan', planError); statementPlans.push({ @@ -8384,6 +8462,8 @@ const QueryEditor: React.FC<{ tab: TabData; isActive?: boolean }> = ({ tab, isAc console.warn('GenerateQueryID failed, using local UUID fallback:', error); queryId = 'query-' + uuidv4(); } + if (!isCurrentRun()) return; + runQueryId = queryId; setQueryId(queryId); const res = useManagedTransaction @@ -8396,6 +8476,7 @@ const QueryEditor: React.FC<{ tab: TabData; isActive?: boolean }> = ({ tab, isAc executableStatements, normalizedDbType, ); + if (!isCurrentRun()) return; const duration = Date.now() - startTime; addSqlLog({ @@ -8423,7 +8504,7 @@ const QueryEditor: React.FC<{ tab: TabData; isActive?: boolean }> = ({ tab, isAc if (isCancelledError && !isTimeoutError) { clearUnpinnedResultSets(); - if (currentQueryIdRef.current) { + if (currentQueryIdRef.current === queryId) { clearQueryId(); } return; @@ -8638,6 +8719,7 @@ const QueryEditor: React.FC<{ tab: TabData; isActive?: boolean }> = ({ tab, isAc } } catch (e: any) { + if (!isCurrentRun()) return; const formattedError = formatSqlExecutionError(e?.message || e, { translate }); message.error(translate('query_editor.message.execution_failed_with_error', { error: formattedError })); addSqlLog({ @@ -8653,9 +8735,10 @@ const QueryEditor: React.FC<{ tab: TabData; isActive?: boolean }> = ({ tab, isAc setExecutionError(formattedError); clearUnpinnedResultSets(QUERY_EDITOR_SQL_LOG_TAB_KEY); } finally { - if (runSeqRef.current === runSeq) setLoading(false); - // Clear query ID after execution completes - clearQueryId(); + if (isCurrentRun()) setLoading(false); + if (runQueryId && currentQueryIdRef.current === runQueryId) { + clearQueryId(); + } } }; @@ -8664,7 +8747,22 @@ const QueryEditor: React.FC<{ tab: TabData; isActive?: boolean }> = ({ tab, isAc }; const handleCancel = async () => { + const finishCancelledRun = () => { + runSeqRef.current += 1; + setLoading(false); + setResultSets(prev => prev.map(result => + result.page?.loading + ? { ...result, page: { ...result.page, loading: false } } + : result + )); + }; + if (!currentQueryIdRef.current) { + if (loading) { + finishCancelledRun(); + message.success(translate('query_editor.message.cancel_success')); + return; + } message.warning(translate('query_editor.message.cancel_no_running')); return; } @@ -8673,9 +8771,9 @@ const QueryEditor: React.FC<{ tab: TabData; isActive?: boolean }> = ({ tab, isAc const res = await CancelQuery(queryIdToCancel); if (res.success) { message.success(translate('query_editor.message.cancel_success')); - // Clear query ID after successful cancellation if (currentQueryIdRef.current === queryIdToCancel) { - clearQueryId() + finishCancelledRun(); + clearQueryId(); } } else { message.warning(res.message); diff --git a/internal/app/app.go b/internal/app/app.go index 841a00ee..22a4120b 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -137,6 +137,7 @@ type queryContext struct { cancel context.CancelFunc started time.Time retainUntilDone bool + registrationID uint64 } type managedSQLTransaction struct { @@ -175,6 +176,7 @@ type App struct { allowApplicationQuit bool applicationQuitPromptInFlight bool queryMu sync.RWMutex + nextQueryRegistrationID uint64 dataRootApplyMu sync.Mutex configDir string secretStore secretstore.SecretStore @@ -1752,6 +1754,33 @@ func generateQueryID() string { return "query-" + uuid.New().String() } +func (a *App) registerRunningQuery(queryID string, cancel context.CancelFunc, retainUntilDone bool) func() { + a.queryMu.Lock() + if a.runningQueries == nil { + a.runningQueries = make(map[string]queryContext) + } + a.nextQueryRegistrationID++ + if a.nextQueryRegistrationID == 0 { + a.nextQueryRegistrationID++ + } + registrationID := a.nextQueryRegistrationID + a.runningQueries[queryID] = queryContext{ + cancel: cancel, + started: time.Now(), + retainUntilDone: retainUntilDone, + registrationID: registrationID, + } + a.queryMu.Unlock() + + return func() { + a.queryMu.Lock() + if current, exists := a.runningQueries[queryID]; exists && current.registrationID == registrationID { + delete(a.runningQueries, queryID) + } + a.queryMu.Unlock() + } +} + // CancelQuery cancels a running query by its ID func (a *App) CancelQuery(queryID string) connection.QueryResult { a.queryMu.Lock() diff --git a/internal/app/methods_db.go b/internal/app/methods_db.go index 2e34de59..cc7e32fa 100644 --- a/internal/app/methods_db.go +++ b/internal/app/methods_db.go @@ -1141,30 +1141,19 @@ func (a *App) dbQueryWithCancel( return connection.QueryResult{Success: false, Message: err.Error(), QueryID: queryID} } + ctx, cancel := newQueryExecutionContext(runConfig) + cleanupRunningQuery := a.registerRunningQuery(queryID, cancel, true) + defer func() { + cancel() + cleanupRunningQuery() + }() + dbInst, err := a.getDatabase(runConfig) if err != nil { logger.Error(err, "DBQuery 获取连接失败：%s", formatConnSummary(runConfig)) return connection.QueryResult{Success: false, Message: err.Error(), QueryID: queryID} } - ctx, cancel := newQueryExecutionContext(runConfig) - defer cancel() - - // Store cancel function for potential manual cancellation - a.queryMu.Lock() - a.runningQueries[queryID] = queryContext{ - cancel: cancel, - started: time.Now(), - } - a.queryMu.Unlock() - - // Ensure query is removed from tracking when done - defer func() { - a.queryMu.Lock() - delete(a.runningQueries, queryID) - a.queryMu.Unlock() - }() - isReadQuery := isReadOnlySQLQuery(runConfig.Type, query) tryQueryFirst := shouldTryQueryResultFirst(runConfig.Type, query) @@ -1333,6 +1322,13 @@ func (a *App) dbQueryMulti( return connection.QueryResult{Success: false, Message: err.Error(), QueryID: queryID} } + ctx, cancel := newQueryExecutionContext(runConfig) + cleanupRunningQuery := a.registerRunningQuery(queryID, cancel, true) + defer func() { + cancel() + cleanupRunningQuery() + }() + dbInst, err := a.getDatabase(runConfig) if err != nil { logger.Error(err, "DBQueryMulti 获取连接失败：%s", formatConnSummary(runConfig)) @@ -1345,21 +1341,6 @@ func (a *App) dbQueryMulti( } }() - ctx, cancel := newQueryExecutionContext(runConfig) - defer cancel() - - a.queryMu.Lock() - a.runningQueries[queryID] = queryContext{ - cancel: cancel, - started: time.Now(), - } - a.queryMu.Unlock() - defer func() { - a.queryMu.Lock() - delete(a.runningQueries, queryID) - a.queryMu.Unlock() - }() - // 尝试使用驱动原生多结果集支持。 // 注意：原生 conn.Query() 执行写操作（UPDATE/INSERT/DELETE）时， // sql.Rows 不暴露 RowsAffected，导致影响行数丢失。 diff --git a/internal/app/methods_db_cancel_test.go b/internal/app/methods_db_cancel_test.go index 811d2aca..2949c5e7 100644 --- a/internal/app/methods_db_cancel_test.go +++ b/internal/app/methods_db_cancel_test.go @@ -7,8 +7,35 @@ import ( "time" "GoNavi-Wails/internal/connection" + "GoNavi-Wails/internal/db" ) +type blockingConnectCancelDB struct { + db.Database + connectStarted chan struct{} + connectRelease chan struct{} + queryContextErr chan error +} + +func (f *blockingConnectCancelDB) Connect(connection.ConnectionConfig) error { + close(f.connectStarted) + <-f.connectRelease + return nil +} + +func (f *blockingConnectCancelDB) Close() error { return nil } + +func (f *blockingConnectCancelDB) Ping() error { return nil } + +func (f *blockingConnectCancelDB) QueryContext(ctx context.Context, _ string) ([]map[string]interface{}, []string, error) { + err := ctx.Err() + f.queryContextErr <- err + if err != nil { + return nil, nil, err + } + return []map[string]interface{}{{"value": 1}}, []string{"value"}, nil +} + func TestGenerateQueryID(t *testing.T) { app := NewApp() id := app.GenerateQueryID() @@ -77,6 +104,120 @@ func TestCancelQuery_ValidQuery(t *testing.T) { } } +func TestDBQueryMulti_CanBeCancelledWhileConnecting(t *testing.T) { + originalNewDatabaseFunc := newDatabaseFunc + t.Cleanup(func() { newDatabaseFunc = originalNewDatabaseFunc }) + + database := &blockingConnectCancelDB{ + connectStarted: make(chan struct{}), + connectRelease: make(chan struct{}), + queryContextErr: make(chan error, 1), + } + newDatabaseFunc = func(string) (db.Database, error) { return database, nil } + + app := NewApp() + queryID := "cancel-while-connecting" + resultCh := make(chan connection.QueryResult, 1) + go func() { + resultCh <- app.DBQueryMulti(connection.ConnectionConfig{ + Type: "mysql", + Host: "cancel-connect.test", + Port: 3306, + User: "tester", + Timeout: 5, + }, "test", "SELECT 1", queryID) + }() + + released := false + defer func() { + if !released { + close(database.connectRelease) + } + }() + select { + case <-database.connectStarted: + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for database connection attempt") + } + + firstCancel := app.CancelQuery(queryID) + secondCancel := app.CancelQuery(queryID) + close(database.connectRelease) + released = true + + var result connection.QueryResult + select { + case result = <-resultCh: + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for cancelled query to return") + } + var observedContextErr error + select { + case observedContextErr = <-database.queryContextErr: + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for query context observation") + } + + if !firstCancel.Success { + t.Errorf("first cancellation while connecting should succeed, got: %s", firstCancel.Message) + } + if !secondCancel.Success { + t.Errorf("repeated cancellation should succeed until the query owner exits, got: %s", secondCancel.Message) + } + if observedContextErr != context.Canceled { + t.Errorf("query should receive the cancellation requested during connect, got context error: %v", observedContextErr) + } + if result.Success { + t.Fatalf("query should not execute successfully after cancellation, got: %+v", result) + } + + app.queryMu.RLock() + _, stillRegistered := app.runningQueries[queryID] + app.queryMu.RUnlock() + if stillRegistered { + t.Fatal("query should be removed from runningQueries after its owner exits") + } + if thirdCancel := app.CancelQuery(queryID); thirdCancel.Success { + t.Fatal("cancellation should fail after the query owner exits") + } +} + +func TestRegisterRunningQuery_OldCleanupDoesNotDeleteReplacement(t *testing.T) { + app := NewApp() + queryID := "reused-query-id" + + firstCtx, firstCancel := context.WithCancel(context.Background()) + defer firstCancel() + cleanupFirst := app.registerRunningQuery(queryID, firstCancel, true) + + secondCtx, secondCancel := context.WithCancel(context.Background()) + defer secondCancel() + cleanupSecond := app.registerRunningQuery(queryID, secondCancel, true) + cleanupFirst() + + if result := app.CancelQuery(queryID); !result.Success { + t.Fatalf("old cleanup removed the replacement registration: %s", result.Message) + } + select { + case <-secondCtx.Done(): + case <-time.After(time.Second): + t.Fatal("replacement cancel function was not called") + } + select { + case <-firstCtx.Done(): + t.Fatal("cancelling the replacement should not cancel the old registration") + default: + } + + cleanupSecond() + app.queryMu.RLock() + _, exists := app.runningQueries[queryID] + app.queryMu.RUnlock() + if exists { + t.Fatal("replacement cleanup should remove its own registration") + } +} + func TestCleanupStaleQueries(t *testing.T) { app := NewApp() diff --git a/internal/app/methods_db_transaction.go b/internal/app/methods_db_transaction.go index b47b7f53..090fd170 100644 --- a/internal/app/methods_db_transaction.go +++ b/internal/app/methods_db_transaction.go @@ -129,27 +129,19 @@ func (a *App) DBQueryMultiTransactional(config connection.ConnectionConfig, dbNa implicitTextTransaction = true } + ctx, cancel := newQueryExecutionContext(runConfig) + cleanupRunningQuery := a.registerRunningQuery(queryID, cancel, true) + defer func() { + cancel() + cleanupRunningQuery() + }() + dbInst, err := a.getDatabase(runConfig) if err != nil { logger.Error(err, "DBQueryMultiTransactional 获取连接失败：%s", formatConnSummary(runConfig)) return connection.QueryResult{Success: false, Message: err.Error(), QueryID: queryID} } - ctx, cancel := newQueryExecutionContext(runConfig) - defer cancel() - - a.queryMu.Lock() - a.runningQueries[queryID] = queryContext{ - cancel: cancel, - started: time.Now(), - } - a.queryMu.Unlock() - defer func() { - a.queryMu.Lock() - delete(a.runningQueries, queryID) - a.queryMu.Unlock() - }() - var ( sessionExecer db.StatementExecer transactor db.TransactionExecer @@ -344,16 +336,24 @@ func (a *App) DBQueryMultiInTransaction(transactionID string, query string, quer if !ok || tx == nil || tx.execer == nil { return connection.QueryResult{Success: false, Message: a.appText("db.backend.error.transaction_not_found", nil), QueryID: queryID} } + + runConfig := tx.config + if strings.TrimSpace(runConfig.Type) == "" { + runConfig.Type = tx.dbType + } + ctx, cancel := newQueryExecutionContext(runConfig) + cleanupRunningQuery := a.registerRunningQuery(queryID, cancel, true) + defer func() { + cancel() + cleanupRunningQuery() + }() + tx.mu.Lock() defer tx.mu.Unlock() if tx.finished || tx.execer == nil { return connection.QueryResult{Success: false, Message: a.appText("db.backend.error.transaction_not_found", nil), QueryID: queryID} } - runConfig := tx.config - if strings.TrimSpace(runConfig.Type) == "" { - runConfig.Type = tx.dbType - } var queryExecutionDuration time.Duration defer func() { if !result.Success { @@ -365,21 +365,6 @@ func (a *App) DBQueryMultiInTransaction(transactionID string, query string, quer query = sanitizeSQLForPgLike(tx.dbType, query) statements := splitSQLStatementsForDialect(tx.dbType, query) - ctx, cancel := newQueryExecutionContext(runConfig) - defer cancel() - - a.queryMu.Lock() - a.runningQueries[queryID] = queryContext{ - cancel: cancel, - started: time.Now(), - } - a.queryMu.Unlock() - defer func() { - a.queryMu.Lock() - delete(a.runningQueries, queryID) - a.queryMu.Unlock() - }() - queryStartedAt := time.Now() statementAuditEvents := make([]sqlaudit.Event, 0, len(statements)) resultSets, err := executeManagedSQLTransactionStatementsWithObserver(