diff --git a/frontend/src/components/DataViewer.primary-key.test.tsx b/frontend/src/components/DataViewer.primary-key.test.tsx index d285b4d8..babdc7bb 100644 --- a/frontend/src/components/DataViewer.primary-key.test.tsx +++ b/frontend/src/components/DataViewer.primary-key.test.tsx @@ -624,7 +624,9 @@ describe('DataViewer safe editing locator', () => { }); await flushPromises(); - expect(backendApp.DBQuery.mock.calls.some((call: any[]) => /count\s*\(/i.test(String(call[2] || '')))).toBe(true); + const countCalls = backendApp.DBQuery.mock.calls.filter((call: any[]) => /count\s*\(/i.test(String(call[2] || ''))); + expect(countCalls.length).toBeGreaterThan(0); + expect(countCalls.every((call: any[]) => Number(call[0]?.queryTimeout) > 0)).toBe(true); expect(dataGridState.latestProps?.pagination).toMatchObject({ total: 500, totalKnown: true, diff --git a/frontend/src/components/DataViewer.tsx b/frontend/src/components/DataViewer.tsx index 5e39eda2..9e5cc211 100644 --- a/frontend/src/components/DataViewer.tsx +++ b/frontend/src/components/DataViewer.tsx @@ -543,7 +543,7 @@ const DataViewer: React.FC<{ tab: TabData; isActive?: boolean }> = React.memo(({ const countSeq = ++manualCountSeqRef.current; const countStart = Date.now(); setPagination(prev => ({ ...prev, totalCountLoading: true, totalCountCancelled: false })); - const countConfig = buildRpcConnectionConfig(config, { timeout: 120 }); + const countConfig = buildRpcConnectionConfig(config, { timeout: 120, queryTimeout: 120 }); try { const resCount = await DBQuery(countConfig as any, dbName, countSql); @@ -781,7 +781,7 @@ const DataViewer: React.FC<{ tab: TabData; isActive?: boolean }> = React.memo(({ latestCountKeyRef.current = countKey; const countStart = Date.now(); - const countConfig = buildRpcConnectionConfig(config, { timeout: 120 }); + const countConfig = buildRpcConnectionConfig(config, { timeout: 120, queryTimeout: 120 }); try { const resCount = await DBQuery(countConfig as any, dbName, countSql); addSqlLog({ @@ -1093,7 +1093,7 @@ const DataViewer: React.FC<{ tab: TabData; isActive?: boolean }> = React.memo(({ const countStart = Date.now(); // Large-table COUNT(*) can be slow and may delay later operations in some runtimes. // DuckDB large-file scenarios disable background COUNT because it can slow pagination significantly. - const countConfig = buildRpcConnectionConfig(config, { timeout: 5 }); + const countConfig = buildRpcConnectionConfig(config, { timeout: 5, queryTimeout: 5 }); setPagination(prev => ({ ...prev, totalCountLoading: true, totalCountCancelled: false })); DBQuery(countConfig, dbName, countSql) @@ -1160,7 +1160,7 @@ const DataViewer: React.FC<{ tab: TabData; isActive?: boolean }> = React.memo(({ const { schemaName, pureTableName } = resolveDuckDBSchemaAndTable(dbName, tableName); const escapedSchema = escapeSQLLiteral(schemaName); const escapedTable = escapeSQLLiteral(pureTableName); - const approxConfig = buildRpcConnectionConfig(config, { timeout: 3 }); + const approxConfig = buildRpcConnectionConfig(config, { timeout: 3, queryTimeout: 3 }); const approxSqlCandidates = [ `SELECT estimated_size AS approx_total FROM duckdb_tables() WHERE schema_name='${escapedSchema}' AND table_name='${escapedTable}' LIMIT 1`, `SELECT estimated_size AS approx_total FROM duckdb_tables() WHERE table_name='${escapedTable}' ORDER BY CASE WHEN schema_name='${escapedSchema}' THEN 0 ELSE 1 END LIMIT 1`, @@ -1201,7 +1201,7 @@ const DataViewer: React.FC<{ tab: TabData; isActive?: boolean }> = React.memo(({ if (approximateCountStrategy === 'oracle-num-rows' && oracleApproxKeyRef.current !== countKey) { oracleApproxKeyRef.current = countKey; const approxSeq = ++oracleApproxSeqRef.current; - const approxConfig = buildRpcConnectionConfig(config, { timeout: 3 }); + const approxConfig = buildRpcConnectionConfig(config, { timeout: 3, queryTimeout: 3 }); const approxSql = buildOracleApproximateTotalSql({ dbName, tableName }); DBQuery(approxConfig as any, dbName, approxSql) diff --git a/frontend/src/components/QueryEditor.tsx b/frontend/src/components/QueryEditor.tsx index 651b3ea0..57b81343 100644 --- a/frontend/src/components/QueryEditor.tsx +++ b/frontend/src/components/QueryEditor.tsx @@ -199,6 +199,7 @@ import { resolveOracleLikeLookupSchemaCandidates, resolveQueryEditorFormatterLanguage, resolveQueryEditorCompletionFilterText, + resolveQueryEditorConnectionTimeout, resolveQueryEditorMonacoLanguage, resolveQueryEditorHoverTarget, resolveQueryEditorNavigationDecorations, @@ -7571,7 +7572,7 @@ const QueryEditor: React.FC<{ tab: TabData; isActive?: boolean }> = ({ tab, isAc database: conn.config.database || '', useSSH: conn.config.useSSH || false, ssh: conn.config.ssh || { host: '', port: 22, user: '', password: '', keyPath: '' }, - timeout: Math.max(Number(conn.config.timeout) || 30, 120), + timeout: resolveQueryEditorConnectionTimeout(conn.config), }; const normalizedDbType = String(resolveSqlDialect( String(config.type || 'mysql'), @@ -7962,7 +7963,7 @@ const QueryEditor: React.FC<{ tab: TabData; isActive?: boolean }> = ({ tab, isAc database: conn.config.database || '', useSSH: conn.config.useSSH || false, ssh: conn.config.ssh || { host: '', port: 22, user: '', password: '', keyPath: '' }, - timeout: Math.max(Number(conn.config.timeout) || 30, 120), + timeout: resolveQueryEditorConnectionTimeout(conn.config), }) as any; const runSeq = ++runSeqRef.current; @@ -8202,7 +8203,7 @@ const QueryEditor: React.FC<{ tab: TabData; isActive?: boolean }> = ({ tab, isAc database: conn.config.database || "", useSSH: conn.config.useSSH || false, ssh: conn.config.ssh || { host: "", port: 22, user: "", password: "", keyPath: "" }, - timeout: Math.max(Number(conn.config.timeout) || 30, 120), + timeout: resolveQueryEditorConnectionTimeout(conn.config), }; try { diff --git a/frontend/src/components/queryEditor/QueryEditorHelpers.test.ts b/frontend/src/components/queryEditor/QueryEditorHelpers.test.ts index f1017bc7..fd9a7af0 100644 --- a/frontend/src/components/queryEditor/QueryEditorHelpers.test.ts +++ b/frontend/src/components/queryEditor/QueryEditorHelpers.test.ts @@ -14,6 +14,7 @@ import { materializeBoundedQueryEditorCompletionBatches, rankQueryEditorCompletionCandidate, resolveQueryEditorCompletionFilterText, + resolveQueryEditorConnectionTimeout, resolveOracleLikeDefaultSchemaName, resolveOracleLikeExecutionSchemaName, resolveOracleLikeLookupSchemaCandidates, @@ -24,6 +25,29 @@ import { splitCompletionSchemaAndTable, } from './QueryEditorHelpers'; +describe('QueryEditor connection timeout', () => { + it('keeps the configured MySQL timeout instead of forcing a 120 second minimum', () => { + expect(resolveQueryEditorConnectionTimeout({ type: 'mysql' })).toBe(30); + expect(resolveQueryEditorConnectionTimeout({ type: 'mysql', timeout: 45 })).toBe(45); + expect(resolveQueryEditorConnectionTimeout({ type: 'mysql', timeout: 300 })).toBe(300); + }); + + it.each([ + [{ type: 'goldendb', timeout: 45 }, 45], + [{ type: 'custom', driver: 'gdb', timeout: 60 }, 60], + ])('keeps the configured connection timeout for compatible config %#', (config, expected) => { + expect(resolveQueryEditorConnectionTimeout(config)).toBe(expected); + }); + + it.each([ + [{ type: 'postgres', timeout: 30 }, 30], + [{ type: 'oracle', timeout: 45 }, 45], + [{ type: 'elasticsearch', timeout: 60 }, 60], + ])('keeps the configured connection timeout for every data source %#', (config, expected) => { + expect(resolveQueryEditorConnectionTimeout(config)).toBe(expected); + }); +}); + describe('QueryEditor result merge identity', () => { it('keeps zero-based Elasticsearch request indexes distinct', () => { const first = buildQueryEditorResultSetMergeKey({ diff --git a/frontend/src/components/queryEditor/QueryEditorHelpers.ts b/frontend/src/components/queryEditor/QueryEditorHelpers.ts index 8ad43ee7..7fad7e3b 100644 --- a/frontend/src/components/queryEditor/QueryEditorHelpers.ts +++ b/frontend/src/components/queryEditor/QueryEditorHelpers.ts @@ -861,6 +861,11 @@ export const normalizeMetadataDialect = (conn: any): string => { return String(dialect || '').toLowerCase(); }; +export const resolveQueryEditorConnectionTimeout = (config: Record): number => { + const rawTimeout = Number(config?.timeout); + return Number.isFinite(rawTimeout) && rawTimeout > 0 ? rawTimeout : 30; +}; + export type QueryEditorMonacoLanguage = 'sql' | 'mysql' | 'elasticsearch-console'; export const resolveQueryEditorMonacoLanguage = (conn: any): QueryEditorMonacoLanguage => { diff --git a/frontend/src/types.ts b/frontend/src/types.ts index 752cbc4a..9187e153 100644 --- a/frontend/src/types.ts +++ b/frontend/src/types.ts @@ -306,6 +306,7 @@ export interface ConnectionConfig { dsn?: string; connectionParams?: string; timeout?: number; + queryTimeout?: number; // transient per-request override; not a saved connection setting keepAliveEnabled?: boolean; keepAliveIntervalMinutes?: number; keepAliveSQL?: string; diff --git a/frontend/src/utils/connectionRpcConfig.test.ts b/frontend/src/utils/connectionRpcConfig.test.ts index 85fd6795..5c81eca0 100644 --- a/frontend/src/utils/connectionRpcConfig.test.ts +++ b/frontend/src/utils/connectionRpcConfig.test.ts @@ -39,6 +39,7 @@ describe('buildRpcConnectionConfig', () => { expect(result.ssh?.port).toBe(2222); expect(result.proxy?.port).toBe(8080); expect(result.timeout).toBe(120); + expect(result.queryTimeout).toBeUndefined(); expect(result.redisDB).toBe(6); expect(result.database).toBe('app'); expect(result.keepAliveEnabled).toBe(true); @@ -46,6 +47,22 @@ describe('buildRpcConnectionConfig', () => { expect(result.keepAliveSQL).toBe('SELECT 1'); }); + it('preserves a per-request query timeout override separately from connection timeout', () => { + const result = buildRpcConnectionConfig({ + type: 'mysql', + host: 'db.local', + port: 3306, + user: 'root', + timeout: 30, + } as any, { + timeout: 5, + queryTimeout: 5, + }); + + expect(result.timeout).toBe(5); + expect(result.queryTimeout).toBe(5); + }); + it('preserves ClickHouse protocol override for RPC calls', () => { const result = buildRpcConnectionConfig({ id: 'conn-clickhouse', diff --git a/frontend/src/utils/connectionRpcConfig.ts b/frontend/src/utils/connectionRpcConfig.ts index af1c5011..e1111960 100644 --- a/frontend/src/utils/connectionRpcConfig.ts +++ b/frontend/src/utils/connectionRpcConfig.ts @@ -123,6 +123,7 @@ export function buildRpcConnectionConfig( const baseId = toStringValue(config.id).trim() || toStringValue(overrides.id).trim() || undefined; const timeout = toOptionalInteger(rpcMerged.timeout, toOptionalInteger(config.timeout)); + const queryTimeout = toOptionalInteger(rpcMerged.queryTimeout, toOptionalInteger(config.queryTimeout)); const redisDB = toOptionalInteger(rpcMerged.redisDB, toOptionalInteger(config.redisDB)); const protection = resolveConnectionProtectionConfig({ type: toStringValue(rpcMerged.type), @@ -149,6 +150,7 @@ export function buildRpcConnectionConfig( useHttpTunnel: rpcMerged.useHttpTunnel === true, httpTunnel: normalizeHttpTunnelConfig(rpcMerged.httpTunnel), timeout, + queryTimeout, redisDB, }) as RpcConnectionConfig; diff --git a/frontend/wailsjs/go/models.ts b/frontend/wailsjs/go/models.ts index 74ebfa5f..14a9db1f 100755 --- a/frontend/wailsjs/go/models.ts +++ b/frontend/wailsjs/go/models.ts @@ -1884,6 +1884,7 @@ export namespace connection { dsn?: string; connectionParams?: string; timeout?: number; + queryTimeout?: number; keepAliveEnabled?: boolean; keepAliveIntervalMinutes?: number; keepAliveSQL?: string; @@ -1938,6 +1939,7 @@ export namespace connection { this.dsn = source["dsn"]; this.connectionParams = source["connectionParams"]; this.timeout = source["timeout"]; + this.queryTimeout = source["queryTimeout"]; this.keepAliveEnabled = source["keepAliveEnabled"]; this.keepAliveIntervalMinutes = source["keepAliveIntervalMinutes"]; this.keepAliveSQL = source["keepAliveSQL"]; diff --git a/internal/app/app.go b/internal/app/app.go index 22a4120b..f1a699ab 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -518,8 +518,9 @@ func normalizeCacheKeyConfig(config connection.ConnectionConfig) connection.Conn normalized.ConnectionParams = normalizeOceanBaseConnectionParamsForCacheWithProtocol(normalized.ConnectionParams, protocol) normalized.OceanBaseProtocol = "" } - // timeout 仅用于 Query/Ping 控制，不应作为物理连接复用键的一部分。 + // Connection/query timeouts affect operations, not physical connection identity. normalized.Timeout = 0 + normalized.QueryTimeout = 0 // keepalive 仅影响后台保活策略，不应参与物理连接复用键。 normalized.KeepAliveEnabled = false normalized.KeepAliveIntervalMinutes = 0 diff --git a/internal/app/app_cache_key_test.go b/internal/app/app_cache_key_test.go index f70c9065..89c5ccfa 100644 --- a/internal/app/app_cache_key_test.go +++ b/internal/app/app_cache_key_test.go @@ -24,6 +24,24 @@ func TestGetCacheKey_IgnoreTimeout(t *testing.T) { } } +func TestGetCacheKey_IgnoreQueryTimeout(t *testing.T) { + base := connection.ConnectionConfig{ + Type: "mysql", + Host: "127.0.0.1", + Port: 3306, + User: "root", + QueryTimeout: 0, + } + modified := base + modified.QueryTimeout = 600 + + left := getCacheKey(base) + right := getCacheKey(modified) + if left != right { + t.Fatalf("expected same cache key when only query timeout differs, got %s vs %s", left, right) + } +} + func TestGetCacheKey_IgnoreConnectionID(t *testing.T) { base := connection.ConnectionConfig{ ID: "conn-1", diff --git a/internal/app/methods_db.go b/internal/app/methods_db.go index bab56cab..7bb28268 100644 --- a/internal/app/methods_db.go +++ b/internal/app/methods_db.go @@ -29,14 +29,13 @@ func normalizeTestConnectionConfig(config connection.ConnectionConfig) connectio } func newQueryExecutionContext(config connection.ConnectionConfig) (context.Context, context.CancelFunc) { - if strings.EqualFold(strings.TrimSpace(config.Type), "duckdb") { - return context.WithCancel(context.Background()) + if config.QueryTimeout > 0 { + return utils.ContextWithTimeout(time.Duration(config.QueryTimeout) * time.Second) } - timeoutSeconds := config.Timeout - if timeoutSeconds <= 0 { - timeoutSeconds = 30 - } - return utils.ContextWithTimeout(time.Duration(timeoutSeconds) * time.Second) + + // Connection timeout is only for establishing the connection. Do not reuse it + // as a query deadline; long-running queries remain cancellable via CancelQuery. + return context.WithCancel(context.Background()) } func validateTestConnectionInput(config connection.ConnectionConfig) error { diff --git a/internal/app/methods_db_cancel_test.go b/internal/app/methods_db_cancel_test.go index 2949c5e7..dcfc2956 100644 --- a/internal/app/methods_db_cancel_test.go +++ b/internal/app/methods_db_cancel_test.go @@ -2,6 +2,7 @@ package app import ( "context" + "errors" "strings" "testing" "time" @@ -292,13 +293,17 @@ func TestDBQueryWithCancel_QueryIDPropagation(t *testing.T) { } } -func TestNewQueryExecutionContext_UsesTimeoutForNetworkDatabases(t *testing.T) { - ctx, cancel := newQueryExecutionContext(connection.ConnectionConfig{Type: "mysql", Timeout: 7}) +func TestNewQueryExecutionContext_UsesExplicitQueryTimeout(t *testing.T) { + ctx, cancel := newQueryExecutionContext(connection.ConnectionConfig{ + Type: "mysql", + Timeout: 1, + QueryTimeout: 7, + }) defer cancel() deadline, ok := ctx.Deadline() if !ok { - t.Fatal("expected network database query context to carry a deadline") + t.Fatal("expected explicit query timeout to carry a deadline") } remaining := time.Until(deadline) if remaining <= 0 || remaining > 8*time.Second { @@ -306,6 +311,36 @@ func TestNewQueryExecutionContext_UsesTimeoutForNetworkDatabases(t *testing.T) { } } +func TestNewQueryExecutionContext_AllDataSourcesDoNotApplyConnectTimeout(t *testing.T) { + tests := []struct { + name string + config connection.ConnectionConfig + }{ + {name: "mysql", config: connection.ConnectionConfig{Type: "mysql", Timeout: 7}}, + {name: "goldendb", config: connection.ConnectionConfig{Type: "goldendb", Timeout: 7}}, + {name: "custom gdb", config: connection.ConnectionConfig{Type: "custom", Driver: "gdb", Timeout: 7}}, + {name: "postgres", config: connection.ConnectionConfig{Type: "postgres", Timeout: 7}}, + {name: "oracle", config: connection.ConnectionConfig{Type: "oracle", Timeout: 7}}, + {name: "sqlserver", config: connection.ConnectionConfig{Type: "sqlserver", Timeout: 7}}, + {name: "elasticsearch", config: connection.ConnectionConfig{Type: "elasticsearch", Timeout: 7}}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctx, cancel := newQueryExecutionContext(tt.config) + + if _, ok := ctx.Deadline(); ok { + cancel() + t.Fatal("expected query context to avoid inheriting the connection-timeout deadline") + } + cancel() + if !errors.Is(ctx.Err(), context.Canceled) { + t.Fatalf("expected manual cancellation to remain effective, got %v", ctx.Err()) + } + }) + } +} + func TestNewQueryExecutionContext_DoesNotApplyConnectTimeoutToDuckDBQueries(t *testing.T) { ctx, cancel := newQueryExecutionContext(connection.ConnectionConfig{Type: "duckdb", Timeout: 1}) defer cancel() diff --git a/internal/app/methods_db_multi_test.go b/internal/app/methods_db_multi_test.go index ac8361d6..80c6e127 100644 --- a/internal/app/methods_db_multi_test.go +++ b/internal/app/methods_db_multi_test.go @@ -988,6 +988,41 @@ func TestDBQueryWithCancel_DuckDBQueriesDoNotInheritConnectTimeout(t *testing.T) } } +func TestDBQueryMulti_MySQLQueriesDoNotInheritConnectTimeout(t *testing.T) { + originalNewDatabaseFunc := newDatabaseFunc + t.Cleanup(func() { + newDatabaseFunc = originalNewDatabaseFunc + }) + + query := "SELECT 1" + fakeDB := &fakeBatchWriteDB{ + queryMap: map[string][]map[string]interface{}{ + query: {{"value": 1}}, + }, + fieldMap: map[string][]string{ + query: {"value"}, + }, + queryErr: map[string]error{}, + } + newDatabaseFunc = func(dbType string) (db.Database, error) { + return fakeDB, nil + } + + app := NewAppWithSecretStore(secretstore.NewUnavailableStore("test")) + config := connection.ConnectionConfig{Type: "mysql", Host: "127.0.0.1", Port: 3306, Timeout: 1} + + result := app.DBQueryMulti(config, "testdb", query, "mysql-no-connect-deadline-test") + if !result.Success { + t.Fatalf("expected MySQL DBQueryMulti success, got failure: %s", result.Message) + } + if fakeDB.lastCtx == nil { + t.Fatal("expected MySQL query path to receive a context") + } + if _, ok := fakeDB.lastCtx.Deadline(); ok { + t.Fatal("expected MySQL query context to avoid connection-timeout deadline") + } +} + func TestDBQueryMultiPreservesPerStatementResultsForMultipleWriteStatements(t *testing.T) { originalNewDatabaseFunc := newDatabaseFunc t.Cleanup(func() { diff --git a/internal/connection/types.go b/internal/connection/types.go index c7512a05..6acc01ee 100644 --- a/internal/connection/types.go +++ b/internal/connection/types.go @@ -112,6 +112,7 @@ type ConnectionConfig struct { DSN string `json:"dsn,omitempty"` // For custom connection ConnectionParams string `json:"connectionParams,omitempty"` // Extra URI query parameters for built-in drivers Timeout int `json:"timeout,omitempty"` // Connection timeout in seconds (default: 30) + QueryTimeout int `json:"queryTimeout,omitempty"` // Per-request query timeout in seconds; 0 disables the automatic query deadline KeepAliveEnabled bool `json:"keepAliveEnabled,omitempty"` // Enable background keep-alive ping for long-lived cached connections KeepAliveIntervalMinutes int `json:"keepAliveIntervalMinutes,omitempty"` // Keep-alive ping interval in minutes (default: 240) KeepAliveSQL string `json:"keepAliveSQL,omitempty"` // Optional single SELECT/WITH probe used instead of the driver ping diff --git a/internal/db/chroma_impl.go b/internal/db/chroma_impl.go index 262ebed2..c40d8475 100644 --- a/internal/db/chroma_impl.go +++ b/internal/db/chroma_impl.go @@ -467,16 +467,20 @@ func chromaAuthHeaders(config connection.ConnectionConfig) map[string]string { func buildChromaHTTPClient(config connection.ConnectionConfig) *http.Client { transport := http.DefaultTransport.(*http.Transport).Clone() + dialTimeout := getConnectTimeout(config) + transport.DialContext = (&net.Dialer{Timeout: dialTimeout, KeepAlive: 30 * time.Second}).DialContext if tlsConfig, err := resolveGenericTLSConfig(config); err == nil && tlsConfig != nil { transport.TLSClientConfig = tlsConfig } if config.UseProxy { proxyCfg := config.Proxy transport.DialContext = func(ctx context.Context, network, addr string) (net.Conn, error) { - return proxytunnel.DialContext(ctx, proxyCfg, network, addr) + dialCtx, cancel := context.WithTimeout(ctx, dialTimeout) + defer cancel() + return proxytunnel.DialContext(dialCtx, proxyCfg, network, addr) } } - return &http.Client{Transport: transport, Timeout: getConnectTimeout(config)} + return &http.Client{Transport: transport} } func (c *ChromaDB) detectVersion(ctx context.Context) error { diff --git a/internal/db/clickhouse_impl.go b/internal/db/clickhouse_impl.go index 841b3491..d0ee5651 100644 --- a/internal/db/clickhouse_impl.go +++ b/internal/db/clickhouse_impl.go @@ -32,8 +32,10 @@ const ( defaultClickHousePort = 9000 defaultClickHouseUser = "default" defaultClickHouseDatabase = "default" - minClickHouseReadTimeout = 5 * time.Minute - clickHouseHTTPPortHint = "8123/8125/8132/8443" + // clickhouse-go replaces a zero ReadTimeout with five minutes. Max duration + // keeps context cancellation as the only practical automatic query deadline. + clickHouseNoAutomaticReadTimeout = time.Duration(1<<63 - 1) + clickHouseHTTPPortHint = "8123/8125/8132/8443" clickHouseProtocolAuto = "auto" clickHouseProtocolHTTP = "http" @@ -179,10 +181,6 @@ func (c *ClickHouseDB) buildClickHouseOptions(config connection.ConnectionConfig func (c *ClickHouseDB) buildClickHouseOptionsWithHTTPCompatibility(config connection.ConnectionConfig, stripHTTPClientProtocolVersion bool) (*clickhouse.Options, error) { connectTimeout := getConnectTimeout(config) - readTimeout := connectTimeout - if readTimeout < minClickHouseReadTimeout { - readTimeout = minClickHouseReadTimeout - } protocol := detectClickHouseProtocol(config) opts := &clickhouse.Options{ Protocol: protocol, @@ -195,7 +193,7 @@ func (c *ClickHouseDB) buildClickHouseOptionsWithHTTPCompatibility(config connec Password: config.Password, }, DialTimeout: connectTimeout, - ReadTimeout: readTimeout, + ReadTimeout: clickHouseNoAutomaticReadTimeout, } tlsConfig, err := resolveGenericTLSConfig(config) if err != nil { diff --git a/internal/db/dsn_test.go b/internal/db/dsn_test.go index 5adf2cee..564e9875 100644 --- a/internal/db/dsn_test.go +++ b/internal/db/dsn_test.go @@ -642,7 +642,7 @@ func TestClickHouseOptions_UsesStructuredTimeoutAndAuth(t *testing.T) { if opts.DialTimeout != 15*time.Second { t.Fatalf("dial timeout 不符合预期：%s", opts.DialTimeout) } - if opts.ReadTimeout != minClickHouseReadTimeout { + if opts.ReadTimeout != clickHouseNoAutomaticReadTimeout { t.Fatalf("read timeout 不符合预期：%s", opts.ReadTimeout) } if _, ok := opts.Settings["write_timeout"]; ok { @@ -687,7 +687,7 @@ func TestClickHouseOptions_MergesConnectionParamsIntoOptionsAndSettings(t *testi } } -func TestClickHouseOptions_ReadTimeoutUsesLargerConfiguredTimeout(t *testing.T) { +func TestClickHouseOptions_ConnectionTimeoutDoesNotBecomeReadTimeout(t *testing.T) { c := &ClickHouseDB{} cfg := normalizeClickHouseConfig(connection.ConnectionConfig{ Type: "clickhouse", @@ -709,7 +709,7 @@ func TestClickHouseOptions_ReadTimeoutUsesLargerConfiguredTimeout(t *testing.T) if opts.DialTimeout != 900*time.Second { t.Fatalf("dial timeout 不符合预期：%s", opts.DialTimeout) } - if opts.ReadTimeout != 900*time.Second { + if opts.ReadTimeout != clickHouseNoAutomaticReadTimeout { t.Fatalf("read timeout 不符合预期：%s", opts.ReadTimeout) } } diff --git a/internal/db/elasticsearch_helpers.go b/internal/db/elasticsearch_helpers.go index 827fa40f..6fcd253d 100644 --- a/internal/db/elasticsearch_helpers.go +++ b/internal/db/elasticsearch_helpers.go @@ -678,26 +678,23 @@ func buildESClientConfig(config connection.ConnectionConfig) elasticsearch.Confi } } - // 代理支持 - if config.UseProxy { - transport, ok := cfg.Transport.(*http.Transport) - if !ok { - transport = http.DefaultTransport.(*http.Transport).Clone() - } - proxyCfg := config.Proxy - transport.DialContext = func(ctx context.Context, network, addr string) (net.Conn, error) { - return proxytunnel.DialContext(ctx, proxyCfg, network, addr) - } - cfg.Transport = transport - } - - // 超时设置 + // Keep the connection dial bounded, but let the request context own query + // cancellation. A connection timeout must not become a response deadline. timeout := getConnectTimeout(config) if cfg.Transport == nil { cfg.Transport = http.DefaultTransport.(*http.Transport).Clone() } if transport, ok := cfg.Transport.(*http.Transport); ok { - transport.ResponseHeaderTimeout = timeout + transport.DialContext = (&net.Dialer{Timeout: timeout, KeepAlive: 30 * time.Second}).DialContext + if config.UseProxy { + proxyCfg := config.Proxy + transport.DialContext = func(ctx context.Context, network, addr string) (net.Conn, error) { + dialCtx, cancel := context.WithTimeout(ctx, timeout) + defer cancel() + return proxytunnel.DialContext(dialCtx, proxyCfg, network, addr) + } + } + transport.ResponseHeaderTimeout = 0 } // 包装 transport：注入 X-Elastic-Product 头以兼容 ES 6.x / 7.x 早期版本。 diff --git a/internal/db/elasticsearch_impl_test.go b/internal/db/elasticsearch_impl_test.go index 01d6415f..c7d6fb62 100644 --- a/internal/db/elasticsearch_impl_test.go +++ b/internal/db/elasticsearch_impl_test.go @@ -59,6 +59,29 @@ func newTestESDB(t *testing.T, serverURL, defaultIndex string) *ElasticsearchDB } } +func TestBuildESClientConfigSeparatesConnectionAndRequestTimeout(t *testing.T) { + config := buildESClientConfig(connection.ConnectionConfig{ + Type: "elasticsearch", + Host: "127.0.0.1", + Port: defaultEsPort, + Timeout: 1, + }) + wrapped, ok := config.Transport.(*esProductCheckBypassTransport) + if !ok { + t.Fatalf("expected product-check transport wrapper, got %T", config.Transport) + } + transport, ok := wrapped.inner.(*http.Transport) + if !ok { + t.Fatalf("expected HTTP transport, got %T", wrapped.inner) + } + if transport.ResponseHeaderTimeout != 0 { + t.Fatalf("connection timeout leaked into Elasticsearch response timeout: %s", transport.ResponseHeaderTimeout) + } + if transport.DialContext == nil { + t.Fatal("expected bounded connection dial") + } +} + // buildMockESMappingResponse 构造模拟的 mapping 响应 JSON。 func buildMockESMappingResponse(indexName string, fields map[string]string) map[string]interface{} { properties := make(map[string]interface{}) diff --git a/internal/db/iotdb_impl.go b/internal/db/iotdb_impl.go index 64dbf9f5..9fca71b1 100644 --- a/internal/db/iotdb_impl.go +++ b/internal/db/iotdb_impl.go @@ -201,8 +201,11 @@ func (i *IoTDBDB) QueryContext(ctx context.Context, query string) ([]map[string] if text == "" { return nil, nil, fmt.Errorf("查询语句不能为空") } - timeoutMs := int64(i.effectiveTimeout().Milliseconds()) - ds, err := i.session.Query(ctx, text, &timeoutMs) + var timeoutMs *int64 + if remaining := timeoutMsFromContext(ctx); remaining > 0 { + timeoutMs = &remaining + } + ds, err := i.session.Query(ctx, text, timeoutMs) if err != nil { return nil, nil, err } diff --git a/internal/db/iotdb_impl_test.go b/internal/db/iotdb_impl_test.go index d648f010..7fb82bd1 100644 --- a/internal/db/iotdb_impl_test.go +++ b/internal/db/iotdb_impl_test.go @@ -9,6 +9,7 @@ import ( "strconv" "strings" "testing" + "time" "GoNavi-Wails/internal/connection" @@ -16,13 +17,19 @@ import ( ) type fakeIoTDBSession struct { - queryResults map[string][]map[string]interface{} - execs []string + queryResults map[string][]map[string]interface{} + execs []string + queryTimeouts []int64 } func (f *fakeIoTDBSession) Close() error { return nil } -func (f *fakeIoTDBSession) Query(_ context.Context, sql string, _ *int64) (iotdbDataSet, error) { +func (f *fakeIoTDBSession) Query(_ context.Context, sql string, timeoutMs *int64) (iotdbDataSet, error) { + timeout := int64(-1) + if timeoutMs != nil { + timeout = *timeoutMs + } + f.queryTimeouts = append(f.queryTimeouts, timeout) rows := f.queryResults[sql] return &fakeIoTDBDataSet{rows: rows, columns: fakeIoTDBColumns(rows)}, nil } @@ -207,6 +214,32 @@ func TestNormalizeIoTDBValueConvertsBinaryText(t *testing.T) { } } +func TestIoTDBQueryContextOnlySendsExplicitDeadlineToServer(t *testing.T) { + session := &fakeIoTDBSession{queryResults: map[string][]map[string]interface{}{ + "SELECT * FROM root.sg.d1": {}, + }} + client := &IoTDBDB{session: session, pingTimeout: time.Second} + + if _, _, err := client.QueryContext(context.Background(), "SELECT * FROM root.sg.d1"); err != nil { + t.Fatalf("QueryContext without deadline: %v", err) + } + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + if _, _, err := client.QueryContext(ctx, "SELECT * FROM root.sg.d1"); err != nil { + t.Fatalf("QueryContext with deadline: %v", err) + } + + if len(session.queryTimeouts) != 2 { + t.Fatalf("query timeout calls = %v", session.queryTimeouts) + } + if session.queryTimeouts[0] != -1 { + t.Fatalf("connection timeout leaked into IoTDB query timeout: %dms", session.queryTimeouts[0]) + } + if session.queryTimeouts[1] <= 0 || session.queryTimeouts[1] > 2000 { + t.Fatalf("explicit context deadline was not propagated: %dms", session.queryTimeouts[1]) + } +} + func TestIoTDBLiveSmoke(t *testing.T) { addr := strings.TrimSpace(os.Getenv("GONAVI_IOTDB_TEST_ADDR")) if addr == "" { diff --git a/internal/db/kafka_impl.go b/internal/db/kafka_impl.go index e8ab3321..bc0a2174 100644 --- a/internal/db/kafka_impl.go +++ b/internal/db/kafka_impl.go @@ -705,8 +705,10 @@ func newKafkaGoRuntime(config connection.ConnectionConfig) (kafkaRuntime, error) SASL: mechanism, } client := &kafka.Client{ - Addr: kafka.TCP(brokers...), - Timeout: timeout, + Addr: kafka.TCP(brokers...), + // Client.Timeout is a per-request deadline. Keep it unset so the + // caller's context (including an explicit queryTimeout) owns it. + Timeout: 0, Transport: transport, } return &kafkaGoRuntime{ @@ -1060,12 +1062,12 @@ type kafkaParsedSQL struct { } var ( - kafkaSQLFromRE = regexp.MustCompile(`(?i)\bFROM\s+(?:"([^"]+)"|` + "`" + `([^` + "`" + `]+)` + "`" + `|([a-zA-Z0-9_.\-]+))`) - kafkaSQLLimitRE = regexp.MustCompile(`(?i)\bLIMIT\s+(\d+)`) - kafkaSQLOffsetRE = regexp.MustCompile(`(?i)\bOFFSET\s+(\d+)`) - kafkaShowTopicsRE = regexp.MustCompile(`(?i)^\s*SHOW\s+TOPICS(?:\s+LIMIT\s+(\d+))?\s*$`) - kafkaDescribeTopicRE = regexp.MustCompile(`(?i)^\s*(?:SHOW|DESCRIBE)\s+TOPIC\s+(?:"([^"]+)"|` + "`" + `([^` + "`" + `]+)` + "`" + `|([a-zA-Z0-9_.\-]+))\s*$`) - kafkaConsumeTopicRE = regexp.MustCompile(`(?i)^\s*CONSUME(?:\s+GROUP\s+(?:"([^"]+)"|` + "`" + `([^` + "`" + `]+)` + "`" + `|([a-zA-Z0-9_.\-]+)))?\s+FROM\s+(?:"([^"]+)"|` + "`" + `([^` + "`" + `]+)` + "`" + `|([a-zA-Z0-9_.\-]+))`) + kafkaSQLFromRE = regexp.MustCompile(`(?i)\bFROM\s+(?:"([^"]+)"|` + "`" + `([^` + "`" + `]+)` + "`" + `|([a-zA-Z0-9_.\-]+))`) + kafkaSQLLimitRE = regexp.MustCompile(`(?i)\bLIMIT\s+(\d+)`) + kafkaSQLOffsetRE = regexp.MustCompile(`(?i)\bOFFSET\s+(\d+)`) + kafkaShowTopicsRE = regexp.MustCompile(`(?i)^\s*SHOW\s+TOPICS(?:\s+LIMIT\s+(\d+))?\s*$`) + kafkaDescribeTopicRE = regexp.MustCompile(`(?i)^\s*(?:SHOW|DESCRIBE)\s+TOPIC\s+(?:"([^"]+)"|` + "`" + `([^` + "`" + `]+)` + "`" + `|([a-zA-Z0-9_.\-]+))\s*$`) + kafkaConsumeTopicRE = regexp.MustCompile(`(?i)^\s*CONSUME(?:\s+GROUP\s+(?:"([^"]+)"|` + "`" + `([^` + "`" + `]+)` + "`" + `|([a-zA-Z0-9_.\-]+)))?\s+FROM\s+(?:"([^"]+)"|` + "`" + `([^` + "`" + `]+)` + "`" + `|([a-zA-Z0-9_.\-]+))`) ) func parseKafkaSQL(sqlText string, defaultLatest bool) (kafkaParsedSQL, bool) { diff --git a/internal/db/kafka_impl_test.go b/internal/db/kafka_impl_test.go index f5c1e534..aa61349c 100644 --- a/internal/db/kafka_impl_test.go +++ b/internal/db/kafka_impl_test.go @@ -5,6 +5,7 @@ import ( "reflect" "strings" "testing" + "time" "GoNavi-Wails/internal/connection" @@ -12,13 +13,37 @@ import ( ) type fakeKafkaRuntime struct { - listTopicsResult []kafkaTopicInfo - describeResult kafkaTopicDescription - fetchResult []kafkaMessageRecord - publishAffected int64 - lastDescribeTopic string - lastFetchRequest kafkaFetchRequest - lastPublishCommand kafkaPublishCommand + listTopicsResult []kafkaTopicInfo + describeResult kafkaTopicDescription + fetchResult []kafkaMessageRecord + publishAffected int64 + lastDescribeTopic string + lastFetchRequest kafkaFetchRequest + lastPublishCommand kafkaPublishCommand +} + +func TestKafkaRuntimeDoesNotDeriveRequestTimeoutFromConnectionTimeout(t *testing.T) { + runtime, err := newKafkaGoRuntime(connection.ConnectionConfig{ + Type: "kafka", + Host: "127.0.0.1", + Port: 9092, + Timeout: 1, + }) + if err != nil { + t.Fatalf("newKafkaGoRuntime: %v", err) + } + defer runtime.Close() + + concrete, ok := runtime.(*kafkaGoRuntime) + if !ok { + t.Fatalf("runtime type = %T", runtime) + } + if concrete.client.Timeout != 0 { + t.Fatalf("connection timeout leaked into Kafka request timeout: %s", concrete.client.Timeout) + } + if concrete.dialer.Timeout != time.Second { + t.Fatalf("Kafka dial timeout = %s, want 1s", concrete.dialer.Timeout) + } } type kafkaOffsetSeekerRecorder struct { diff --git a/internal/db/milvus_impl.go b/internal/db/milvus_impl.go index 263614ca..72b38260 100644 --- a/internal/db/milvus_impl.go +++ b/internal/db/milvus_impl.go @@ -515,16 +515,20 @@ func milvusAuthHeaders(config connection.ConnectionConfig) map[string]string { func buildMilvusHTTPClient(config connection.ConnectionConfig) *http.Client { transport := http.DefaultTransport.(*http.Transport).Clone() + dialTimeout := getConnectTimeout(config) + transport.DialContext = (&net.Dialer{Timeout: dialTimeout, KeepAlive: 30 * time.Second}).DialContext if tlsConfig, err := resolveGenericTLSConfig(config); err == nil && tlsConfig != nil { transport.TLSClientConfig = tlsConfig } if config.UseProxy { proxyConfig := config.Proxy transport.DialContext = func(ctx context.Context, network, address string) (net.Conn, error) { - return proxytunnel.DialContext(ctx, proxyConfig, network, address) + dialCtx, cancel := context.WithTimeout(ctx, dialTimeout) + defer cancel() + return proxytunnel.DialContext(dialCtx, proxyConfig, network, address) } } - return &http.Client{Transport: transport, Timeout: getConnectTimeout(config)} + return &http.Client{Transport: transport} } func (m *MilvusDB) doJSON(ctx context.Context, method, path string, body interface{}, out interface{}) error { diff --git a/internal/db/qdrant_impl.go b/internal/db/qdrant_impl.go index 394cb88c..29d8a59f 100644 --- a/internal/db/qdrant_impl.go +++ b/internal/db/qdrant_impl.go @@ -461,16 +461,20 @@ func qdrantAuthHeaders(config connection.ConnectionConfig) map[string]string { func buildQdrantHTTPClient(config connection.ConnectionConfig) *http.Client { transport := http.DefaultTransport.(*http.Transport).Clone() + dialTimeout := getConnectTimeout(config) + transport.DialContext = (&net.Dialer{Timeout: dialTimeout, KeepAlive: 30 * time.Second}).DialContext if tlsConfig, err := resolveGenericTLSConfig(config); err == nil && tlsConfig != nil { transport.TLSClientConfig = tlsConfig } if config.UseProxy { proxyCfg := config.Proxy transport.DialContext = func(ctx context.Context, network, addr string) (net.Conn, error) { - return proxytunnel.DialContext(ctx, proxyCfg, network, addr) + dialCtx, cancel := context.WithTimeout(ctx, dialTimeout) + defer cancel() + return proxytunnel.DialContext(dialCtx, proxyCfg, network, addr) } } - return &http.Client{Transport: transport, Timeout: getConnectTimeout(config)} + return &http.Client{Transport: transport} } func (q *QdrantDB) doJSON(ctx context.Context, method, path string, body interface{}, out interface{}) error { diff --git a/internal/db/rabbitmq_impl.go b/internal/db/rabbitmq_impl.go index 4a1c77eb..a00780fc 100644 --- a/internal/db/rabbitmq_impl.go +++ b/internal/db/rabbitmq_impl.go @@ -613,16 +613,20 @@ func buildRabbitMQBaseURL(config connection.ConnectionConfig) string { func buildRabbitMQHTTPClient(config connection.ConnectionConfig) *http.Client { transport := http.DefaultTransport.(*http.Transport).Clone() + dialTimeout := getConnectTimeout(config) + transport.DialContext = (&net.Dialer{Timeout: dialTimeout, KeepAlive: 30 * time.Second}).DialContext if tlsConfig, err := resolveGenericTLSConfig(config); err == nil && tlsConfig != nil { transport.TLSClientConfig = tlsConfig } if config.UseProxy { proxyCfg := config.Proxy transport.DialContext = func(ctx context.Context, network, addr string) (net.Conn, error) { - return proxytunnel.DialContext(ctx, proxyCfg, network, addr) + dialCtx, cancel := context.WithTimeout(ctx, dialTimeout) + defer cancel() + return proxytunnel.DialContext(dialCtx, proxyCfg, network, addr) } } - return &http.Client{Transport: transport, Timeout: getConnectTimeout(config)} + return &http.Client{Transport: transport} } func rabbitmqAuthHeaders(config connection.ConnectionConfig) map[string]string { diff --git a/internal/db/timeout_policy_test.go b/internal/db/timeout_policy_test.go new file mode 100644 index 00000000..9be91609 --- /dev/null +++ b/internal/db/timeout_policy_test.go @@ -0,0 +1,34 @@ +package db + +import ( + "net/http" + "testing" + + "GoNavi-Wails/internal/connection" +) + +func TestHTTPDataSourceClientsDoNotUseConnectTimeoutAsRequestTimeout(t *testing.T) { + config := connection.ConnectionConfig{Timeout: 1} + tests := []struct { + name string + build func(connection.ConnectionConfig) *http.Client + }{ + {name: "chroma", build: buildChromaHTTPClient}, + {name: "qdrant", build: buildQdrantHTTPClient}, + {name: "milvus", build: buildMilvusHTTPClient}, + {name: "rabbitmq", build: buildRabbitMQHTTPClient}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + client := tt.build(config) + if client.Timeout != 0 { + t.Fatalf("connection timeout leaked into HTTP request timeout: %s", client.Timeout) + } + transport, ok := client.Transport.(*http.Transport) + if !ok || transport.DialContext == nil { + t.Fatal("expected HTTP transport to retain a bounded connection dial") + } + }) + } +} diff --git a/internal/db/trino_impl.go b/internal/db/trino_impl.go index 4743e633..6764876a 100644 --- a/internal/db/trino_impl.go +++ b/internal/db/trino_impl.go @@ -261,6 +261,9 @@ func buildTrinoDSN(config connection.ConnectionConfig, customClientName string) } params := connectionParamsFromText(config.ConnectionParams) + if params == nil { + params = url.Values{} + } catalog, schema := resolveTrinoNamespace(config.Database, "") if catalog != "" { params.Set("catalog", catalog) @@ -274,9 +277,8 @@ func buildTrinoDSN(config connection.ConnectionConfig, customClientName string) if strings.TrimSpace(params.Get("explicitPrepare")) == "" { params.Set("explicitPrepare", "false") } - if strings.TrimSpace(params.Get("query_timeout")) == "" { - params.Set("query_timeout", fmt.Sprintf("%ds", getConnectTimeoutSeconds(config))) - } + // Do not derive Trino's server-side query_timeout from the connection + // timeout. The request context is the sole automatic query deadline. if strings.TrimSpace(customClientName) != "" { params.Set("custom_client", strings.TrimSpace(customClientName)) } diff --git a/internal/db/trino_impl_test.go b/internal/db/trino_impl_test.go index 95891cf0..a6406ef3 100644 --- a/internal/db/trino_impl_test.go +++ b/internal/db/trino_impl_test.go @@ -6,8 +6,11 @@ import ( "database/sql" "database/sql/driver" "errors" + "net/url" "sync" "testing" + + "GoNavi-Wails/internal/connection" ) var ( @@ -63,3 +66,24 @@ func TestTrinoCloseCleansStateWhenDatabaseCloseFails(t *testing.T) { t.Fatalf("Close() namespace = %q, want empty", trino.namespace) } } + +func TestBuildTrinoDSNDoesNotDeriveQueryTimeoutFromConnectionTimeout(t *testing.T) { + dsn, err := buildTrinoDSN(connection.ConnectionConfig{ + Type: "trino", + Host: "127.0.0.1", + Port: 8080, + User: "alice", + Database: "hive.analytics", + Timeout: 1, + }, "") + if err != nil { + t.Fatalf("buildTrinoDSN: %v", err) + } + parsed, err := url.Parse(dsn) + if err != nil { + t.Fatalf("parse Trino DSN: %v", err) + } + if got := parsed.Query().Get("query_timeout"); got != "" { + t.Fatalf("connection timeout leaked into Trino query_timeout=%q", got) + } +}