GithubBackup/MyGoNavi
1
0
Fork 0
mirror of https://github.com/Syngnat/GoNavi.git synced 2026-08-10 08:43:34 +08:00
Code Issues Packages Projects Releases Wiki Activity

🐛 fix(query-editor/db): 修复运行中 SQL 无法停止

Browse Source 
- 查询 ID 生成前支持终止执行，并统一刷新、翻页和 Mongo 多语句的取消代际
- 后端在连接建立和事务锁等待前登记取消，保留登记直到执行 owner 退出
- 使用 registration ID 防止旧任务清理同 ID 的新查询
- 补充前后端取消竞态与并发回归测试

Refs #754
This commit is contained in:
Syngnat
2026-08-04 11:07:08 +08:00
parent 53d0480058
commit 7ec66dc732
7 changed files with 612 additions and 106 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
frontend/src/components/QueryEditor.external-sql-save.test.tsx
View File

@@ -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 () => {

268
frontend/src/components/QueryEditor.results-and-drop.test.tsx
View File

@@ -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(<QueryEditor tab={createTab({ dbName: 'main', query: 'select 1;' })} />);
});
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(<QueryEditor tab={createTab({ dbName: 'main', query: 'select 1;' })} />);
});
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(<QueryEditor tab={createTab({ dbName: 'main', query: 'select 1 as value;' })} />);
});
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(<QueryEditor tab={createTab({ dbName: 'main', query: 'select value from items;' })} />);
});
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(<QueryEditor tab={createTab({ dbName: 'main', query })} />);
});
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 () => {

160
frontend/src/components/QueryEditor.tsx
View File

@@ -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);

29
internal/app/app.go
View File

@@ -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()

47
internal/app/methods_db.go
View File

@@ -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，导致影响行数丢失。

141
internal/app/methods_db_cancel_test.go
View File

@@ -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()

53
internal/app/methods_db_transaction.go
View File

@@ -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(