✨ feat(connection): 支持自定义 SQL 探活

- 新增连接级自定义探活 SQL 配置与多语言表单校验
- 限制单条 SELECT/WITH 并使用可取消的超时查询
- 隔离陈旧探活策略，避免配置错误驱逐健康连接

Fixes #611

internal/app/app.go

@@ -50,12 +50,17 @@ var (
 )
 
 type cachedDatabase struct {
-	inst              db.Database
-	lastPing          time.Time
-	config            connection.ConnectionConfig
-	keepAliveEnabled  bool
-	keepAliveInterval time.Duration
-	keepAliveInFlight bool
+	inst                      db.Database
+	lastPing                  time.Time
+	lastKeepAliveAt           time.Time
+	config                    connection.ConnectionConfig
+	keepAliveEnabled          bool
+	keepAliveInterval         time.Duration
+	keepAliveSQL              string
+	keepAliveDBType           string
+	keepAliveRevision         uint64
+	keepAliveInFlight         bool
+	keepAliveInFlightRevision uint64
 }
 
 type cachedConnectFailure struct {
@@ -374,6 +379,7 @@ func normalizeCacheKeyConfig(config connection.ConnectionConfig) connection.Conn
 	// keepalive 仅影响后台保活策略，不应参与物理连接复用键。
 	normalized.KeepAliveEnabled = false
 	normalized.KeepAliveIntervalMinutes = 0
+	normalized.KeepAliveSQL = ""
 	normalized.SavePassword = false
 
 	if !normalized.UseSSH {
@@ -918,13 +924,31 @@ func (a *App) getDatabaseWithPing(config connection.ConnectionConfig, forcePing
 	a.mu.RUnlock()
 	if ok {
 		keepAliveEnabled, keepAliveInterval := resolveConnectionKeepAliveSettings(effectiveConfig)
-		if entry.keepAliveEnabled != keepAliveEnabled || entry.keepAliveInterval != keepAliveInterval {
+		keepAliveSQL, keepAliveDBType := resolveConnectionKeepAliveSQL(effectiveConfig)
+		if entry.keepAliveEnabled != keepAliveEnabled ||
+			entry.keepAliveInterval != keepAliveInterval ||
+			entry.keepAliveSQL != keepAliveSQL ||
+			entry.keepAliveDBType != keepAliveDBType {
 			a.mu.Lock()
 			if cur, exists := a.dbCache[key]; exists && cur.inst == entry.inst {
-				cur.keepAliveEnabled = keepAliveEnabled
-				cur.keepAliveInterval = keepAliveInterval
-				if !keepAliveEnabled {
-					cur.keepAliveInFlight = false
+				policyChanged := cur.keepAliveEnabled != keepAliveEnabled ||
+					cur.keepAliveInterval != keepAliveInterval ||
+					cur.keepAliveSQL != keepAliveSQL ||
+					cur.keepAliveDBType != keepAliveDBType
+				if policyChanged {
+					wasKeepAliveEnabled := cur.keepAliveEnabled
+					cur.keepAliveRevision = nextConnectionKeepAliveRevision(cur.keepAliveRevision)
+					cur.keepAliveEnabled = keepAliveEnabled
+					cur.keepAliveInterval = keepAliveInterval
+					cur.keepAliveSQL = keepAliveSQL
+					cur.keepAliveDBType = keepAliveDBType
+					if !keepAliveEnabled {
+						cur.keepAliveInFlight = false
+						cur.keepAliveInFlightRevision = 0
+						cur.lastKeepAliveAt = time.Time{}
+					} else if !wasKeepAliveEnabled || cur.lastKeepAliveAt.IsZero() {
+						cur.lastKeepAliveAt = time.Now()
+					}
 				}
 				a.dbCache[key] = cur
 				entry = cur
@@ -1014,9 +1038,30 @@ func (a *App) getDatabaseWithPing(config connection.ConnectionConfig, forcePing
 
 	now := time.Now()
 	keepAliveEnabled, keepAliveInterval := resolveConnectionKeepAliveSettings(effectiveConfig)
+	keepAliveSQL, keepAliveDBType := resolveConnectionKeepAliveSQL(effectiveConfig)
 
 	a.mu.Lock()
 	if existing, exists := a.dbCache[key]; exists && existing.inst != nil {
+		policyChanged := existing.keepAliveEnabled != keepAliveEnabled ||
+			existing.keepAliveInterval != keepAliveInterval ||
+			existing.keepAliveSQL != keepAliveSQL ||
+			existing.keepAliveDBType != keepAliveDBType
+		if policyChanged {
+			wasKeepAliveEnabled := existing.keepAliveEnabled
+			existing.keepAliveRevision = nextConnectionKeepAliveRevision(existing.keepAliveRevision)
+			existing.keepAliveEnabled = keepAliveEnabled
+			existing.keepAliveInterval = keepAliveInterval
+			existing.keepAliveSQL = keepAliveSQL
+			existing.keepAliveDBType = keepAliveDBType
+			if !keepAliveEnabled {
+				existing.keepAliveInFlight = false
+				existing.keepAliveInFlightRevision = 0
+				existing.lastKeepAliveAt = time.Time{}
+			} else if !wasKeepAliveEnabled || existing.lastKeepAliveAt.IsZero() {
+				existing.lastKeepAliveAt = now
+			}
+		}
+		a.dbCache[key] = existing
 		a.mu.Unlock()
 		// Prefer existing cached connection to avoid cache racing duplicates.
 		_ = dbInst.Close()
@@ -1028,9 +1073,13 @@ func (a *App) getDatabaseWithPing(config connection.ConnectionConfig, forcePing
 	a.dbCache[key] = cachedDatabase{
 		inst:              dbInst,
 		lastPing:          now,
+		lastKeepAliveAt:   now,
 		config:            normalizeCacheKeyConfig(effectiveConfig),
 		keepAliveEnabled:  keepAliveEnabled,
 		keepAliveInterval: keepAliveInterval,
+		keepAliveSQL:      keepAliveSQL,
+		keepAliveDBType:   keepAliveDBType,
+		keepAliveRevision: 1,
 	}
 	a.mu.Unlock()
 

internal/app/app_cache_key_test.go

@@ -53,10 +53,12 @@ func TestGetCacheKey_IgnoreKeepAliveSettings(t *testing.T) {
 		Database:                 "app",
 		KeepAliveEnabled:         false,
 		KeepAliveIntervalMinutes: 240,
+		KeepAliveSQL:             "SELECT 1",
 	}
 	modified := base
 	modified.KeepAliveEnabled = true
 	modified.KeepAliveIntervalMinutes = 15
+	modified.KeepAliveSQL = "SELECT current_timestamp"
 
 	left := getCacheKey(base)
 	right := getCacheKey(modified)

internal/app/app_keepalive.go

@@ -2,11 +2,15 @@ package app
 
 import (
 	"context"
+	"errors"
+	"strings"
 	"time"
+	"unicode/utf8"
 
 	"GoNavi-Wails/internal/connection"
 	"GoNavi-Wails/internal/db"
 	"GoNavi-Wails/internal/logger"
+	"GoNavi-Wails/internal/sqlaudit"
 )
 
 const (
@@ -14,12 +18,75 @@ const (
 	minConnectionKeepAliveIntervalMinutes     = 1
 	maxConnectionKeepAliveIntervalMinutes     = 1440
 	connectionKeepAliveScanInterval           = 30 * time.Second
+	connectionKeepAliveQueryTimeout           = 30 * time.Second
+	maxConnectionKeepAliveSQLLength           = 4096
+)
+
+var (
+	errInvalidConnectionKeepAliveSQL              = errors.New("custom keep-alive SQL must be one SELECT or WITH statement without write operations")
+	errConnectionKeepAliveQueryContextUnsupported = errors.New("database driver does not support cancellable custom keep-alive SQL")
 )
 
 type cachedDatabaseKeepAliveTarget struct {
-	key    string
-	inst   db.Database
-	config connection.ConnectionConfig
+	key      string
+	inst     db.Database
+	config   connection.ConnectionConfig
+	sql      string
+	dbType   string
+	revision uint64
+}
+
+func nextConnectionKeepAliveRevision(current uint64) uint64 {
+	next := current + 1
+	if next == 0 {
+		return 1
+	}
+	return next
+}
+
+func supportsConnectionKeepAliveSQL(config connection.ConnectionConfig) bool {
+	dbType := resolveDDLDBType(config)
+	if dbType == "mongodb" || isFileDatabaseType(dbType) {
+		return false
+	}
+	_, supported := connectionReadOnlySupportedTypes[dbType]
+	return supported
+}
+
+func resolveConnectionKeepAliveSQL(config connection.ConnectionConfig) (string, string) {
+	if !config.KeepAliveEnabled || !supportsConnectionKeepAliveSQL(config) {
+		return "", ""
+	}
+	return strings.TrimSpace(config.KeepAliveSQL), resolveDDLDBType(config)
+}
+
+func executeConnectionKeepAlive(ctx context.Context, target cachedDatabaseKeepAliveTarget) error {
+	if strings.TrimSpace(target.sql) == "" {
+		return target.inst.Ping()
+	}
+	if utf8.RuneCountInString(target.sql) > maxConnectionKeepAliveSQLLength ||
+		!isSafeExplainQuery(target.dbType, target.sql) {
+		return errInvalidConnectionKeepAliveSQL
+	}
+
+	if ctx == nil {
+		ctx = context.Background()
+	}
+	queryCtx, cancel := context.WithTimeout(ctx, connectionKeepAliveQueryTimeout)
+	defer cancel()
+	queryer, ok := target.inst.(interface {
+		QueryContext(context.Context, string) ([]map[string]interface{}, []string, error)
+	})
+	if !ok {
+		return errConnectionKeepAliveQueryContextUnsupported
+	}
+	_, _, err := queryer.QueryContext(queryCtx, target.sql)
+	return err
+}
+
+func isConnectionKeepAlivePolicyError(err error) bool {
+	return errors.Is(err, errInvalidConnectionKeepAliveSQL) ||
+		errors.Is(err, errConnectionKeepAliveQueryContextUnsupported)
 }
 
 func resolveConnectionKeepAliveSettings(config connection.ConnectionConfig) (bool, time.Duration) {
@@ -61,7 +128,7 @@ func (a *App) startConnectionKeepAliveLoop() {
 			case <-ctx.Done():
 				return
 			case now := <-ticker.C:
-				a.runConnectionKeepAliveTick(now)
+				a.runConnectionKeepAliveTickContext(ctx, now)
 			}
 		}
 	}()
@@ -86,22 +153,103 @@ func (a *App) stopConnectionKeepAliveLoop() {
 }
 
 func (a *App) runConnectionKeepAliveTick(now time.Time) {
-	for _, target := range a.collectDueConnectionKeepAliveTargets(now) {
-		if target.inst == nil {
+	a.runConnectionKeepAliveTickContext(context.Background(), now)
+}
+
+func (a *App) runConnectionKeepAliveTickContext(ctx context.Context, now time.Time) {
+	if ctx == nil {
+		ctx = context.Background()
+	}
+	targets := a.collectDueConnectionKeepAliveTargets(now)
+	for index, target := range targets {
+		if ctx.Err() != nil {
+			a.releaseCachedDatabaseKeepAliveTargets(targets[index:])
+			return
+		}
+		if target.inst == nil || !a.isCachedDatabaseKeepAliveTargetCurrent(target) {
 			continue
 		}
-		if err := target.inst.Ping(); err != nil {
+		if err := executeConnectionKeepAlive(ctx, target); err != nil {
+			if ctx.Err() != nil {
+				a.releaseCachedDatabaseKeepAliveTargets(targets[index:])
+				return
+			}
+			if isConnectionKeepAlivePolicyError(err) {
+				if a.markCachedDatabaseKeepAliveSkipped(target, time.Now()) {
+					logger.Warnf(
+						"连接自定义保活配置无效，已跳过本次探活：%s 缓存Key=%s 原因=%s",
+						formatConnSummary(target.config),
+						shortCacheKey(target.key),
+						sqlaudit.RedactError(normalizeErrorMessage(err)),
+					)
+				}
+				continue
+			}
 			if closed, summary := a.evictCachedDatabaseAfterKeepAliveFailure(target); closed {
 				logger.Warnf(
 					"连接保活失败，已清理缓存连接：%s 缓存Key=%s 原因=%s",
 					summary,
 					shortCacheKey(target.key),
-					normalizeErrorMessage(err),
+					sqlaudit.RedactError(normalizeErrorMessage(err)),
 				)
 			}
 			continue
 		}
-		a.markCachedDatabaseKeepAliveSuccess(target.key, target.inst, time.Now())
+		a.markCachedDatabaseKeepAliveSuccess(target, time.Now())
+	}
+}
+
+func cachedDatabaseKeepAliveTargetOwnsInFlight(entry cachedDatabase, target cachedDatabaseKeepAliveTarget) bool {
+	return entry.inst == target.inst &&
+		entry.keepAliveInFlight &&
+		entry.keepAliveInFlightRevision == target.revision
+}
+
+func cachedDatabaseKeepAliveTargetMatches(entry cachedDatabase, target cachedDatabaseKeepAliveTarget) bool {
+	return cachedDatabaseKeepAliveTargetOwnsInFlight(entry, target) &&
+		entry.keepAliveEnabled &&
+		entry.keepAliveRevision == target.revision &&
+		entry.keepAliveSQL == target.sql &&
+		entry.keepAliveDBType == target.dbType
+}
+
+func (a *App) isCachedDatabaseKeepAliveTargetCurrent(target cachedDatabaseKeepAliveTarget) bool {
+	if a == nil {
+		return false
+	}
+
+	a.mu.Lock()
+	defer a.mu.Unlock()
+	entry, exists := a.dbCache[target.key]
+	if !exists {
+		return false
+	}
+	if cachedDatabaseKeepAliveTargetMatches(entry, target) {
+		return true
+	}
+	if cachedDatabaseKeepAliveTargetOwnsInFlight(entry, target) {
+		entry.keepAliveInFlight = false
+		entry.keepAliveInFlightRevision = 0
+		a.dbCache[target.key] = entry
+	}
+	return false
+}
+
+func (a *App) releaseCachedDatabaseKeepAliveTargets(targets []cachedDatabaseKeepAliveTarget) {
+	if a == nil || len(targets) == 0 {
+		return
+	}
+
+	a.mu.Lock()
+	defer a.mu.Unlock()
+	for _, target := range targets {
+		entry, exists := a.dbCache[target.key]
+		if !exists || !cachedDatabaseKeepAliveTargetOwnsInFlight(entry, target) {
+			continue
+		}
+		entry.keepAliveInFlight = false
+		entry.keepAliveInFlightRevision = 0
+		a.dbCache[target.key] = entry
 	}
 }
 
@@ -118,23 +266,27 @@ func (a *App) collectDueConnectionKeepAliveTargets(now time.Time) []cachedDataba
 		if entry.inst == nil || !entry.keepAliveEnabled || entry.keepAliveInterval <= 0 || entry.keepAliveInFlight {
 			continue
 		}
-		if !entry.lastPing.IsZero() && now.Sub(entry.lastPing) < entry.keepAliveInterval {
+		if !entry.lastKeepAliveAt.IsZero() && now.Sub(entry.lastKeepAliveAt) < entry.keepAliveInterval {
 			continue
 		}
 
 		entry.keepAliveInFlight = true
+		entry.keepAliveInFlightRevision = entry.keepAliveRevision
 		a.dbCache[key] = entry
 		targets = append(targets, cachedDatabaseKeepAliveTarget{
-			key:    key,
-			inst:   entry.inst,
-			config: entry.config,
+			key:      key,
+			inst:     entry.inst,
+			config:   entry.config,
+			sql:      entry.keepAliveSQL,
+			dbType:   entry.keepAliveDBType,
+			revision: entry.keepAliveRevision,
 		})
 	}
 
 	return targets
 }
 
-func (a *App) markCachedDatabaseKeepAliveSuccess(key string, inst db.Database, pingedAt time.Time) {
+func (a *App) markCachedDatabaseKeepAliveSuccess(target cachedDatabaseKeepAliveTarget, pingedAt time.Time) {
 	if a == nil {
 		return
 	}
@@ -142,14 +294,50 @@ func (a *App) markCachedDatabaseKeepAliveSuccess(key string, inst db.Database, p
 	a.mu.Lock()
 	defer a.mu.Unlock()
 
-	entry, exists := a.dbCache[key]
-	if !exists || entry.inst != inst {
+	entry, exists := a.dbCache[target.key]
+	if !exists {
 		return
 	}
+	if cachedDatabaseKeepAliveTargetMatches(entry, target) {
+		entry.keepAliveInFlight = false
+		entry.keepAliveInFlightRevision = 0
+		entry.lastPing = pingedAt
+		entry.lastKeepAliveAt = pingedAt
+		a.dbCache[target.key] = entry
+		return
+	}
+	if cachedDatabaseKeepAliveTargetOwnsInFlight(entry, target) {
+		entry.keepAliveInFlight = false
+		entry.keepAliveInFlightRevision = 0
+		a.dbCache[target.key] = entry
+	}
+}
 
-	entry.keepAliveInFlight = false
-	entry.lastPing = pingedAt
-	a.dbCache[key] = entry
+func (a *App) markCachedDatabaseKeepAliveSkipped(target cachedDatabaseKeepAliveTarget, attemptedAt time.Time) bool {
+	if a == nil {
+		return false
+	}
+
+	a.mu.Lock()
+	defer a.mu.Unlock()
+
+	entry, exists := a.dbCache[target.key]
+	if !exists {
+		return false
+	}
+	if cachedDatabaseKeepAliveTargetMatches(entry, target) {
+		entry.keepAliveInFlight = false
+		entry.keepAliveInFlightRevision = 0
+		entry.lastKeepAliveAt = attemptedAt
+		a.dbCache[target.key] = entry
+		return true
+	}
+	if cachedDatabaseKeepAliveTargetOwnsInFlight(entry, target) {
+		entry.keepAliveInFlight = false
+		entry.keepAliveInFlightRevision = 0
+		a.dbCache[target.key] = entry
+	}
+	return false
 }
 
 func (a *App) evictCachedDatabaseAfterKeepAliveFailure(target cachedDatabaseKeepAliveTarget) (bool, string) {
@@ -164,10 +352,14 @@ func (a *App) evictCachedDatabaseAfterKeepAliveFailure(target cachedDatabaseKeep
 
 	a.mu.Lock()
 	entry, exists := a.dbCache[target.key]
-	if exists && entry.inst == target.inst {
+	if exists && cachedDatabaseKeepAliveTargetMatches(entry, target) {
 		inst = entry.inst
 		summary = formatConnSummary(entry.config)
 		delete(a.dbCache, target.key)
+	} else if exists && cachedDatabaseKeepAliveTargetOwnsInFlight(entry, target) {
+		entry.keepAliveInFlight = false
+		entry.keepAliveInFlightRevision = 0
+		a.dbCache[target.key] = entry
 	}
 	a.mu.Unlock()
 

internal/app/app_keepalive_test.go

@@ -1,20 +1,33 @@
 package app
 
 import (
+	"context"
 	"errors"
 	"testing"
 	"time"
 
 	"GoNavi-Wails/internal/connection"
+	"GoNavi-Wails/internal/db"
 )
 
 type keepAliveRecordingDB struct {
-	closed  int
-	pings   int
-	pingErr error
+	closed               int
+	pings                int
+	pingErr              error
+	queries              []string
+	queryErr             error
+	queryContextCalls    int
+	queryContextDeadline bool
+	queryContextHook     func(context.Context, string) error
+	connectHook          func()
 }
 
-func (f *keepAliveRecordingDB) Connect(config connection.ConnectionConfig) error { return nil }
+func (f *keepAliveRecordingDB) Connect(config connection.ConnectionConfig) error {
+	if f.connectHook != nil {
+		f.connectHook()
+	}
+	return nil
+}
 func (f *keepAliveRecordingDB) Close() error {
 	f.closed++
 	return nil
@@ -24,7 +37,17 @@ func (f *keepAliveRecordingDB) Ping() error {
 	return f.pingErr
 }
 func (f *keepAliveRecordingDB) Query(query string) ([]map[string]interface{}, []string, error) {
-	return nil, nil, nil
+	f.queries = append(f.queries, query)
+	return nil, nil, f.queryErr
+}
+func (f *keepAliveRecordingDB) QueryContext(ctx context.Context, query string) ([]map[string]interface{}, []string, error) {
+	f.queryContextCalls++
+	_, f.queryContextDeadline = ctx.Deadline()
+	f.queries = append(f.queries, query)
+	if f.queryContextHook != nil {
+		return nil, nil, f.queryContextHook(ctx, query)
+	}
+	return nil, nil, f.queryErr
 }
 func (f *keepAliveRecordingDB) Exec(query string) (int64, error)          { return 0, nil }
 func (f *keepAliveRecordingDB) GetDatabases() ([]string, error)           { return nil, nil }
@@ -75,6 +98,305 @@ func TestRunConnectionKeepAliveTick_PingsDueCachedConnection(t *testing.T) {
 	if entry.lastPing.IsZero() {
 		t.Fatal("expected keepalive success to update lastPing")
 	}
+	if entry.lastKeepAliveAt.IsZero() {
+		t.Fatal("expected keepalive success to update lastKeepAliveAt")
+	}
+}
+
+func TestRunConnectionKeepAliveTick_ExecutesCustomReadOnlySQL(t *testing.T) {
+	app := NewApp()
+	config := connection.ConnectionConfig{
+		Type:             "mysql",
+		Host:             "db.local",
+		Port:             3306,
+		User:             "readonly",
+		KeepAliveEnabled: true,
+		KeepAliveSQL:     "  SELECT 1  ",
+	}
+	key := getCacheKey(config)
+	dbInst := &keepAliveRecordingDB{}
+
+	app.dbCache[key] = cachedDatabase{
+		inst:              dbInst,
+		lastPing:          time.Now().Add(-5 * time.Hour),
+		config:            normalizeCacheKeyConfig(config),
+		keepAliveEnabled:  true,
+		keepAliveInterval: 4 * time.Hour,
+		keepAliveSQL:      "SELECT 1",
+		keepAliveDBType:   "mysql",
+	}
+
+	app.runConnectionKeepAliveTick(time.Now())
+
+	if dbInst.pings != 0 {
+		t.Fatalf("expected custom SQL to replace Ping, got %d pings", dbInst.pings)
+	}
+	if len(dbInst.queries) != 1 || dbInst.queries[0] != "SELECT 1" {
+		t.Fatalf("expected trimmed custom SQL once, got %#v", dbInst.queries)
+	}
+	if dbInst.queryContextCalls != 1 || !dbInst.queryContextDeadline {
+		t.Fatalf("expected cancellable QueryContext with deadline, calls=%d deadline=%t", dbInst.queryContextCalls, dbInst.queryContextDeadline)
+	}
+}
+
+func TestResolveConnectionKeepAliveSQL_NormalizesSupportedSQLConnections(t *testing.T) {
+	sql, dbType := resolveConnectionKeepAliveSQL(connection.ConnectionConfig{
+		Type:             "mysql",
+		KeepAliveEnabled: true,
+		KeepAliveSQL:     "  SELECT 1  ",
+	})
+	if sql != "SELECT 1" || dbType != "mysql" {
+		t.Fatalf("expected normalized MySQL keepalive SQL, sql=%q dbType=%q", sql, dbType)
+	}
+
+	sql, dbType = resolveConnectionKeepAliveSQL(connection.ConnectionConfig{
+		Type:             "redis",
+		KeepAliveEnabled: true,
+		KeepAliveSQL:     "SELECT 1",
+	})
+	if sql != "" || dbType != "" {
+		t.Fatalf("expected non-SQL datasource to ignore custom SQL, sql=%q dbType=%q", sql, dbType)
+	}
+}
+
+func TestExecuteConnectionKeepAlive_RejectsUnsafeSQL(t *testing.T) {
+	tests := []string{
+		"DELETE FROM accounts",
+		"SELECT 1; SELECT 2",
+		"SELECT 1; DELETE FROM accounts",
+		"/*!50000 DELETE FROM accounts */ SELECT 1",
+		"SELECT /*!50000 SQL_NO_CACHE */ 1",
+		"SELECT ';' AS probe",
+		"SELECT 1 /* ; */",
+	}
+
+	for _, query := range tests {
+		t.Run(query, func(t *testing.T) {
+			dbInst := &keepAliveRecordingDB{}
+			err := executeConnectionKeepAlive(context.Background(), cachedDatabaseKeepAliveTarget{
+				inst:   dbInst,
+				sql:    query,
+				dbType: "mysql",
+			})
+			if !errors.Is(err, errInvalidConnectionKeepAliveSQL) {
+				t.Fatalf("expected unsafe SQL rejection, got %v", err)
+			}
+			if len(dbInst.queries) != 0 || dbInst.pings != 0 {
+				t.Fatalf("expected unsafe SQL not to reach database, queries=%#v pings=%d", dbInst.queries, dbInst.pings)
+			}
+		})
+	}
+}
+
+func TestExecuteConnectionKeepAlive_RequiresCancellableQuery(t *testing.T) {
+	dbInst := &keepAliveRecordingDB{}
+	dbWithoutQueryContext := struct{ db.Database }{Database: dbInst}
+
+	err := executeConnectionKeepAlive(context.Background(), cachedDatabaseKeepAliveTarget{
+		inst:   dbWithoutQueryContext,
+		sql:    "SELECT 1",
+		dbType: "mysql",
+	})
+	if !errors.Is(err, errConnectionKeepAliveQueryContextUnsupported) {
+		t.Fatalf("expected drivers without QueryContext to be rejected, got %v", err)
+	}
+	if len(dbInst.queries) != 0 || dbInst.pings != 0 {
+		t.Fatalf("expected no unbounded query fallback, queries=%#v pings=%d", dbInst.queries, dbInst.pings)
+	}
+}
+
+func TestRunConnectionKeepAliveTick_KeepsHealthyConnectionOnPolicyError(t *testing.T) {
+	app := NewApp()
+	config := connection.ConnectionConfig{Type: "mysql", Host: "db.local", Port: 3306, User: "readonly"}
+	key := getCacheKey(config)
+	dbInst := &keepAliveRecordingDB{}
+
+	app.dbCache[key] = cachedDatabase{
+		inst:              dbInst,
+		config:            normalizeCacheKeyConfig(config),
+		keepAliveEnabled:  true,
+		keepAliveInterval: 4 * time.Hour,
+		keepAliveSQL:      "DELETE FROM accounts",
+		keepAliveDBType:   "mysql",
+	}
+
+	app.runConnectionKeepAliveTick(time.Now())
+
+	entry, exists := app.dbCache[key]
+	if !exists || entry.inst != dbInst {
+		t.Fatal("expected policy error not to evict the healthy cached connection")
+	}
+	if entry.keepAliveInFlight || entry.lastKeepAliveAt.IsZero() {
+		t.Fatalf("expected skipped policy to finish and advance its schedule, inFlight=%t lastKeepAliveAt=%s", entry.keepAliveInFlight, entry.lastKeepAliveAt)
+	}
+	if dbInst.closed != 0 || len(dbInst.queries) != 0 || dbInst.pings != 0 {
+		t.Fatalf("expected invalid policy not to touch the database, closed=%d queries=%#v pings=%d", dbInst.closed, dbInst.queries, dbInst.pings)
+	}
+}
+
+func TestRunConnectionKeepAliveTickContext_CancelsCustomQuery(t *testing.T) {
+	app := NewApp()
+	config := connection.ConnectionConfig{Type: "postgres", Host: "db.local", Port: 5432, User: "readonly"}
+	key := getCacheKey(config)
+	queryStarted := make(chan struct{})
+	dbInst := &keepAliveRecordingDB{
+		queryContextHook: func(ctx context.Context, _ string) error {
+			close(queryStarted)
+			<-ctx.Done()
+			return ctx.Err()
+		},
+	}
+	app.dbCache[key] = cachedDatabase{
+		inst:              dbInst,
+		config:            normalizeCacheKeyConfig(config),
+		keepAliveEnabled:  true,
+		keepAliveInterval: 4 * time.Hour,
+		keepAliveSQL:      "SELECT 1",
+		keepAliveDBType:   "postgres",
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		app.runConnectionKeepAliveTickContext(ctx, time.Now())
+	}()
+	<-queryStarted
+	cancel()
+
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		t.Fatal("expected custom keepalive query to stop after loop cancellation")
+	}
+	entry, exists := app.dbCache[key]
+	if !exists || entry.inst != dbInst || entry.keepAliveInFlight {
+		t.Fatalf("expected cancelled keepalive to preserve and release cached connection, exists=%t inFlight=%t", exists, entry.keepAliveInFlight)
+	}
+	if dbInst.closed != 0 {
+		t.Fatalf("expected shutdown cancellation not to evict the connection, closed=%d", dbInst.closed)
+	}
+}
+
+func TestCachedDatabaseKeepAliveTarget_SkipsQueuedStalePolicy(t *testing.T) {
+	app := NewApp()
+	config := connection.ConnectionConfig{Type: "mysql", Host: "db.local", Port: 3306, User: "readonly"}
+	key := getCacheKey(config)
+	dbInst := &keepAliveRecordingDB{}
+	app.dbCache[key] = cachedDatabase{
+		inst:              dbInst,
+		config:            normalizeCacheKeyConfig(config),
+		keepAliveEnabled:  true,
+		keepAliveInterval: 4 * time.Hour,
+		keepAliveSQL:      "SELECT 1",
+		keepAliveDBType:   "mysql",
+		keepAliveRevision: 1,
+	}
+
+	targets := app.collectDueConnectionKeepAliveTargets(time.Now())
+	if len(targets) != 1 {
+		t.Fatalf("expected one due target, got %d", len(targets))
+	}
+	entry := app.dbCache[key]
+	entry.keepAliveRevision = nextConnectionKeepAliveRevision(entry.keepAliveRevision)
+	entry.keepAliveSQL = "SELECT 2"
+	app.dbCache[key] = entry
+
+	if app.isCachedDatabaseKeepAliveTargetCurrent(targets[0]) {
+		t.Fatal("expected queued target to become stale after policy update")
+	}
+	entry = app.dbCache[key]
+	if entry.keepAliveInFlight || entry.keepAliveSQL != "SELECT 2" || entry.keepAliveRevision != 2 {
+		t.Fatalf("expected stale target release without changing new policy, inFlight=%t sql=%q revision=%d", entry.keepAliveInFlight, entry.keepAliveSQL, entry.keepAliveRevision)
+	}
+	if len(dbInst.queries) != 0 {
+		t.Fatalf("expected stale queued SQL not to execute, queries=%#v", dbInst.queries)
+	}
+}
+
+func TestRunConnectionKeepAliveTick_DoesNotApplyStalePolicyResult(t *testing.T) {
+	app := NewApp()
+	config := connection.ConnectionConfig{Type: "postgres", Host: "db.local", Port: 5432, User: "readonly"}
+	key := getCacheKey(config)
+	queryStarted := make(chan struct{})
+	finishQuery := make(chan struct{})
+	dbInst := &keepAliveRecordingDB{
+		queryContextHook: func(_ context.Context, _ string) error {
+			close(queryStarted)
+			<-finishQuery
+			return errors.New("old probe failed")
+		},
+	}
+	lastKeepAliveAt := time.Now().Add(-5 * time.Hour)
+	app.dbCache[key] = cachedDatabase{
+		inst:              dbInst,
+		lastKeepAliveAt:   lastKeepAliveAt,
+		config:            normalizeCacheKeyConfig(config),
+		keepAliveEnabled:  true,
+		keepAliveInterval: 4 * time.Hour,
+		keepAliveSQL:      "SELECT 1",
+		keepAliveDBType:   "postgres",
+		keepAliveRevision: 1,
+	}
+
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		app.runConnectionKeepAliveTick(time.Now())
+	}()
+	<-queryStarted
+	app.mu.Lock()
+	entry := app.dbCache[key]
+	entry.keepAliveRevision = nextConnectionKeepAliveRevision(entry.keepAliveRevision)
+	entry.keepAliveSQL = "SELECT 2"
+	app.dbCache[key] = entry
+	app.mu.Unlock()
+	close(finishQuery)
+
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		t.Fatal("expected stale custom keepalive query to finish")
+	}
+	entry, exists := app.dbCache[key]
+	if !exists || entry.inst != dbInst {
+		t.Fatal("expected stale query failure not to evict the current cached connection")
+	}
+	if entry.keepAliveInFlight || entry.keepAliveSQL != "SELECT 2" || entry.keepAliveRevision != 2 {
+		t.Fatalf("expected current policy to remain unchanged, inFlight=%t sql=%q revision=%d", entry.keepAliveInFlight, entry.keepAliveSQL, entry.keepAliveRevision)
+	}
+	if !entry.lastKeepAliveAt.Equal(lastKeepAliveAt) {
+		t.Fatalf("expected stale result not to advance current policy schedule, before=%s after=%s", lastKeepAliveAt, entry.lastKeepAliveAt)
+	}
+	if dbInst.closed != 0 {
+		t.Fatalf("expected stale query failure not to close the current connection, closed=%d", dbInst.closed)
+	}
+}
+
+func TestRunConnectionKeepAliveTick_RemovesFailedCustomSQLConnection(t *testing.T) {
+	app := NewApp()
+	config := connection.ConnectionConfig{Type: "mysql", Host: "db.local", Port: 3306, User: "readonly"}
+	key := getCacheKey(config)
+	dbInst := &keepAliveRecordingDB{queryErr: errors.New("token expired")}
+
+	app.dbCache[key] = cachedDatabase{
+		inst:              dbInst,
+		lastPing:          time.Now().Add(-5 * time.Hour),
+		config:            normalizeCacheKeyConfig(config),
+		keepAliveEnabled:  true,
+		keepAliveInterval: 4 * time.Hour,
+		keepAliveSQL:      "SELECT 1",
+		keepAliveDBType:   "mysql",
+	}
+
+	app.runConnectionKeepAliveTick(time.Now())
+
+	if len(dbInst.queries) != 1 || dbInst.pings != 0 {
+		t.Fatalf("expected one custom query and no Ping, queries=%#v pings=%d", dbInst.queries, dbInst.pings)
+	}
+	if dbInst.closed != 1 || len(app.dbCache) != 0 {
+		t.Fatalf("expected failed custom keepalive to evict and close connection, closed=%d cache=%d", dbInst.closed, len(app.dbCache))
+	}
 }
 
 func TestRunConnectionKeepAliveTick_RemovesFailedCachedConnection(t *testing.T) {
@@ -121,6 +443,7 @@ func TestGetDatabaseWithPing_UpdatesCachedKeepAliveSettings(t *testing.T) {
 		User:                     "postgres",
 		KeepAliveEnabled:         true,
 		KeepAliveIntervalMinutes: 15,
+		KeepAliveSQL:             " SELECT current_timestamp ",
 	}
 	key := getCacheKey(config)
 	dbInst := &keepAliveRecordingDB{}
@@ -146,4 +469,113 @@ func TestGetDatabaseWithPing_UpdatesCachedKeepAliveSettings(t *testing.T) {
 	if entry.keepAliveInterval != 15*time.Minute {
 		t.Fatalf("expected keepalive interval 15m, got %s", entry.keepAliveInterval)
 	}
+	if entry.keepAliveSQL != "SELECT current_timestamp" {
+		t.Fatalf("expected cached custom SQL to be updated, got %q", entry.keepAliveSQL)
+	}
+	if entry.keepAliveDBType != "postgres" {
+		t.Fatalf("expected cached custom SQL dialect postgres, got %q", entry.keepAliveDBType)
+	}
+}
+
+func TestGetDatabaseWithPing_ConcurrentCacheWinnerReceivesKeepAliveSettings(t *testing.T) {
+	originalNewDatabaseFunc := newDatabaseFunc
+	originalDriverRuntimeSupportStatusFunc := driverRuntimeSupportStatusFunc
+	originalVerifyDriverAgentRevisionFunc := verifyDriverAgentRevisionFunc
+	defer func() {
+		newDatabaseFunc = originalNewDatabaseFunc
+		driverRuntimeSupportStatusFunc = originalDriverRuntimeSupportStatusFunc
+		verifyDriverAgentRevisionFunc = originalVerifyDriverAgentRevisionFunc
+	}()
+	driverRuntimeSupportStatusFunc = func(string) (bool, string) { return true, "" }
+	verifyDriverAgentRevisionFunc = func(connection.ConnectionConfig) error { return nil }
+
+	app := NewApp()
+	config := connection.ConnectionConfig{
+		Type:                     "postgres",
+		Host:                     "db.local",
+		Port:                     5432,
+		User:                     "postgres",
+		KeepAliveEnabled:         true,
+		KeepAliveIntervalMinutes: 15,
+		KeepAliveSQL:             " SELECT current_timestamp ",
+	}
+	key := getCacheKey(config)
+	winner := &keepAliveRecordingDB{}
+	created := &keepAliveRecordingDB{}
+	created.connectHook = func() {
+		app.mu.Lock()
+		app.dbCache[key] = cachedDatabase{
+			inst:     winner,
+			lastPing: time.Now(),
+			config:   normalizeCacheKeyConfig(config),
+		}
+		app.mu.Unlock()
+	}
+	newDatabaseFunc = func(string) (db.Database, error) { return created, nil }
+
+	inst, err := app.getDatabaseWithPing(config, false)
+	if err != nil {
+		t.Fatalf("expected concurrent cache winner lookup to succeed, got %v", err)
+	}
+	if inst != winner {
+		t.Fatal("expected the concurrent cache winner to be reused")
+	}
+	if created.closed != 1 {
+		t.Fatalf("expected duplicate created connection to be closed once, got %d", created.closed)
+	}
+	entry := app.dbCache[key]
+	if !entry.keepAliveEnabled || entry.keepAliveInterval != 15*time.Minute {
+		t.Fatalf("expected winner keepalive policy to update, enabled=%t interval=%s", entry.keepAliveEnabled, entry.keepAliveInterval)
+	}
+	if entry.keepAliveSQL != "SELECT current_timestamp" || entry.keepAliveDBType != "postgres" {
+		t.Fatalf("expected winner custom SQL policy to update, sql=%q dbType=%q", entry.keepAliveSQL, entry.keepAliveDBType)
+	}
+	if entry.lastKeepAliveAt.IsZero() {
+		t.Fatal("expected winner keepalive schedule to start from the policy update")
+	}
+}
+
+func TestGetDatabaseWithPing_ForegroundPingDoesNotDelayCustomKeepAliveSQL(t *testing.T) {
+	originalDriverRuntimeSupportStatusFunc := driverRuntimeSupportStatusFunc
+	defer func() {
+		driverRuntimeSupportStatusFunc = originalDriverRuntimeSupportStatusFunc
+	}()
+	driverRuntimeSupportStatusFunc = func(string) (bool, string) { return true, "" }
+
+	app := NewApp()
+	config := connection.ConnectionConfig{
+		Type:                     "postgres",
+		Host:                     "db.local",
+		Port:                     5432,
+		User:                     "postgres",
+		KeepAliveEnabled:         true,
+		KeepAliveIntervalMinutes: 15,
+		KeepAliveSQL:             "SELECT 1",
+	}
+	key := getCacheKey(config)
+	dbInst := &keepAliveRecordingDB{}
+	lastKeepAliveAt := time.Now().Add(-20 * time.Minute)
+	app.dbCache[key] = cachedDatabase{
+		inst:              dbInst,
+		lastPing:          time.Now().Add(-5 * time.Minute),
+		lastKeepAliveAt:   lastKeepAliveAt,
+		config:            normalizeCacheKeyConfig(config),
+		keepAliveEnabled:  true,
+		keepAliveInterval: 15 * time.Minute,
+		keepAliveSQL:      "SELECT 1",
+		keepAliveDBType:   "postgres",
+	}
+
+	if _, err := app.getDatabaseWithPing(config, false); err != nil {
+		t.Fatalf("expected foreground cache health Ping to succeed, got %v", err)
+	}
+	entry := app.dbCache[key]
+	if !entry.lastKeepAliveAt.Equal(lastKeepAliveAt) {
+		t.Fatalf("expected foreground Ping not to move keepalive schedule, before=%s after=%s", lastKeepAliveAt, entry.lastKeepAliveAt)
+	}
+
+	app.runConnectionKeepAliveTick(time.Now())
+	if dbInst.pings != 1 || len(dbInst.queries) != 1 {
+		t.Fatalf("expected foreground Ping followed by due custom SQL, pings=%d queries=%#v", dbInst.pings, dbInst.queries)
+	}
 }