feat(connection): notify sidebar when keepalive loses connection

This commit is contained in:
Abeautifulsnow
2026-09-30 19:49:52 +08:00
committed by GitHub
parent 3ffff92c8f
commit dc503877f9
13 changed files with 864 additions and 27 deletions
+39
View File
@@ -371,6 +371,9 @@ let updateCheckTimer: ReturnType<typeof setInterval> | undefined;
const needsAuth = ref(!isDesktop && (startupProps.startupAuthentication?.required ?? true));
const authenticated = ref(isDesktop || (startupProps.startupAuthentication?.authenticated ?? false));
const setupRequired = ref(!isDesktop && (startupProps.startupAuthentication?.setup_required ?? false));
// Mirrors the template gate above the app shell. The backend liveness stream is registered
// against it so the web runtime only opens an authenticated subscription.
const appReady = computed(() => !setupRequired.value && (!needsAuth.value || authenticated.value));
const showConnectionDialog = ref(false);
const connectionDialogPrefill = ref<ConnectionDeepLinkDraft | null>(null);
@@ -4078,6 +4081,40 @@ function runUpdateNotificationChecks() {
void componentUpdates.refresh();
}
// Backend-confirmed connection liveness losses (#4339). Registered through the forwarded
// `api` layer rather than the Tauri-only `listen` helpers so both runtimes subscribe, and
// keyed off `appReady` so the web runtime never opens the SSE stream before it is
// authenticated (the whole /api surface sits behind the auth middleware).
let connectionLivenessUnlisten: (() => void) | null = null;
let connectionLivenessSubscribing = false;
async function syncConnectionLivenessSubscription(active: boolean): Promise<void> {
if (!active) {
connectionLivenessUnlisten?.();
connectionLivenessUnlisten = null;
return;
}
if (connectionLivenessUnlisten || connectionLivenessSubscribing) return;
connectionLivenessSubscribing = true;
try {
const unlisten = await api.subscribeConnectionLiveness((message) => {
void connectionStore.handleConnectionLivenessMessage(message);
});
if (!appReady.value) {
// Auth flipped off while the subscription was being established.
unlisten();
return;
}
connectionLivenessUnlisten = unlisten;
} catch (error) {
console.error("[DBX] subscribeConnectionLiveness error:", error);
} finally {
connectionLivenessSubscribing = false;
}
}
watch(appReady, (ready) => void syncConnectionLivenessSubscription(ready), { immediate: true });
onMounted(async () => {
clearStartupPreloadRetry();
markStartupPhase("app-mounted");
@@ -4182,6 +4219,8 @@ onMounted(async () => {
onUnmounted(() => {
disposeUpdater();
updatePreparation?.dispose();
connectionLivenessUnlisten?.();
connectionLivenessUnlisten = null;
detachedEventUnlisteners.forEach((unlisten) => unlisten());
detachedEventUnlisteners = [];
cleanupTauriListeners();
+2
View File
@@ -75,6 +75,8 @@ export const sessionCredentialStatus = forward("sessionCredentialStatus");
export const forgetSessionCredential = forward("forgetSessionCredential");
export const replaceNacosSessionCredential = forward("replaceNacosSessionCredential");
export const checkConnectionHealth = forward("checkConnectionHealth");
export const connectionIsOpen = forward("connectionIsOpen");
export const subscribeConnectionLiveness = forward("subscribeConnectionLiveness");
export const prewarmConnection = forward("prewarmConnection");
export const connectionIdentifierQuote = forward("connectionIdentifierQuote");
export const closeDatabaseConnection = forward("closeDatabaseConnection");
+26
View File
@@ -249,6 +249,7 @@ import type { DataCompareFromTablesOptions, DataCompareFromTablesPreparation, Da
import { apiUrl, apiWebSocketUrl } from "@/lib/common/webPath";
import type {
ActivePluginSession,
ConnectionLivenessMessage,
PluginBinaryEvent,
PluginConnectionActionResult,
PluginEvent,
@@ -556,6 +557,17 @@ export async function checkConnectionHealth(connectionId: string): Promise<void>
return post("/api/connection/check-health", { connectionId });
}
/**
* Read-only counterpart of `checkConnectionHealth`: reports whether the connection still has a
* pool, without probing or mutating anything (#4339).
*
* Liveness events must be confirmed through this, never through `checkConnectionHealth`: the
* latter removes unhealthy pools and is the path `ensureConnected` uses to trigger a reconnect.
*/
export async function connectionIsOpen(connectionId: string): Promise<boolean> {
return post("/api/connection/is-open", { connectionId });
}
export async function prewarmConnection(connectionId: string, database?: string, catalog?: string, clientSessionId?: string): Promise<void> {
return post("/api/connection/prewarm", { connectionId, database, catalog, clientSessionId });
}
@@ -785,6 +797,20 @@ export async function subscribePluginEvents(onEvent: (event: PluginEvent) => voi
return () => source.close();
}
/**
* Subscribe to backend connection-liveness messages (#4339).
*
* The SSE payload is one `ConnectionLivenessMessage`: the plugin stream's `{ kind }` wrapper
* multiplexes two streams, whereas here `kind` is the message's own discriminator.
*/
export async function subscribeConnectionLiveness(onEvent: (event: ConnectionLivenessMessage) => void): Promise<() => void> {
const source = new EventSource(apiUrl("/api/connection/liveness-events"));
source.onmessage = (message) => {
onEvent(JSON.parse(message.data) as ConnectionLivenessMessage);
};
return () => source.close();
}
async function pluginAssetPayload(path: string): Promise<PluginUiAssetPayload> {
const response = await fetch(apiUrl(path));
if (!response.ok) throw new Error(await response.text());
+22
View File
@@ -122,6 +122,7 @@ import type { QueryEditability } from "@/lib/sql/sqlAnalysis";
import { isTerminalTransferProgress } from "@/lib/backend/transferProgress";
import type {
ActivePluginSession,
ConnectionLivenessMessage,
PluginBinaryEvent,
PluginConnectionActionResult,
PluginEvent,
@@ -1563,6 +1564,17 @@ export async function checkConnectionHealth(connectionId: string): Promise<void>
return invokeBackend("check_connection_health", { connectionId });
}
/**
* Read-only counterpart of `checkConnectionHealth`: reports whether the connection still has a
* pool, without probing or mutating anything (#4339).
*
* Liveness events must be confirmed through this, never through `checkConnectionHealth`: the
* latter removes unhealthy pools and is the path `ensureConnected` uses to trigger a reconnect.
*/
export async function connectionIsOpen(connectionId: string): Promise<boolean> {
return invokeBackend("connection_is_open", { connectionId });
}
export async function prewarmConnection(connectionId: string, database?: string, catalog?: string, clientSessionId?: string): Promise<void> {
return invokeBackend("prewarm_connection", { connectionId, database, catalog, clientSessionId });
}
@@ -2853,6 +2865,16 @@ export async function subscribePluginEvents(onEvent: (event: PluginEvent) => voi
};
}
/**
* Subscribe to backend connection-liveness messages (#4339).
*
* The listener receives the message unwrapped; the web transport's SSE handler parses the same
* shape, so the store can stay transport-agnostic.
*/
export async function subscribeConnectionLiveness(onEvent: (event: ConnectionLivenessMessage) => void): Promise<UnlistenFn> {
return listen<ConnectionLivenessMessage>("dbx-connection-liveness", (event) => onEvent(event.payload));
}
export async function listJdbcDrivers(): Promise<JdbcDriverInfo[]> {
return invoke("list_jdbc_drivers");
}
@@ -0,0 +1,163 @@
import { createPinia, setActivePinia } from "pinia";
import { beforeEach, describe, expect, it, vi } from "vitest";
import i18n from "@/i18n";
const connectionIsOpen = vi.hoisted(() => vi.fn());
function installLocalStorage() {
const data = new Map<string, string>();
vi.stubGlobal("localStorage", {
getItem: vi.fn((key: string) => data.get(key) ?? null),
setItem: vi.fn((key: string, value: string) => data.set(key, value)),
removeItem: vi.fn((key: string) => data.delete(key)),
});
}
/**
* Backend connection-liveness messages (#4339).
*
* The backend publishes `lost` only once a connection has no pools left, but the message is
* asynchronous: it can be delivered after the user already reconnected. The store therefore
* confirms with the read-only `connectionIsOpen` before it touches the sidebar, and must never
* confirm with `checkConnectionHealth` (which removes pools and reconnects).
*
* `resync` is sent when the transport dropped messages; those are unrecoverable, so the store
* re-checks everything it still shows as connected.
*/
describe("connectionStore keepalive liveness", () => {
beforeEach(() => {
vi.resetModules();
vi.unstubAllGlobals();
installLocalStorage();
setActivePinia(createPinia());
i18n.global.locale.value = "en";
connectionIsOpen.mockReset();
vi.doMock("@/lib/backend/tauriRuntime", () => ({ isTauriRuntime: () => false }));
vi.doMock("@/lib/backend/api", () => ({
connectionIsOpen,
disconnectDb: vi.fn().mockResolvedValue(undefined),
checkConnectionHealth: vi.fn().mockResolvedValue(undefined),
listInstalledAgents: vi.fn().mockResolvedValue([]),
loadSchemaCache: vi.fn().mockResolvedValue(null),
saveConnections: vi.fn().mockResolvedValue(undefined),
saveSchemaCache: vi.fn().mockResolvedValue(undefined),
saveSidebarLayout: vi.fn().mockResolvedValue(undefined),
}));
});
async function loadStore() {
const { useConnectionStore } = await import("@/stores/connectionStore");
return useConnectionStore();
}
it("flips the sidebar offline only once the read-only confirm reports no pool", async () => {
connectionIsOpen.mockResolvedValue(false);
const store = await loadStore();
store.connectedIds.add("c1");
await store.handleConnectionLivenessMessage({ kind: "lost", connectionId: "c1", failureKind: "probe_failed" });
expect(connectionIsOpen).toHaveBeenCalledWith("c1");
expect(store.connectedIds.has("c1")).toBe(false);
});
it("keeps the connection when the confirm reports it open again", async () => {
// A reconnect that finished before the message was delivered: the pool is back, so a late
// message must not grey out a connection the user is actively using.
connectionIsOpen.mockResolvedValue(true);
const store = await loadStore();
store.connectedIds.add("c1");
await store.handleConnectionLivenessMessage({ kind: "lost", connectionId: "c1", failureKind: "timed_out" });
expect(store.connectedIds.has("c1")).toBe(true);
});
it("ignores messages for connections this frontend does not consider connected", async () => {
const store = await loadStore();
await store.handleConnectionLivenessMessage({
kind: "lost",
connectionId: "never-connected",
failureKind: "probe_failed",
});
expect(connectionIsOpen).not.toHaveBeenCalled();
expect(store.connectedIds.size).toBe(0);
});
it("leaves the sidebar untouched when the confirm itself fails", async () => {
// A failed confirm says nothing about the connection, so the status must not be guessed.
connectionIsOpen.mockRejectedValue(new Error("ipc unavailable"));
const store = await loadStore();
store.connectedIds.add("c1");
await store.handleConnectionLivenessMessage({ kind: "lost", connectionId: "c1", failureKind: "probe_failed" });
expect(store.connectedIds.has("c1")).toBe(true);
});
it("re-checks every connected connection on a resync instead of keeping a dropped loss", async () => {
// The transport dropped messages, so the ids whose loss was dropped are unknown: the store
// has to re-check the whole connected set, or one sidebar stays green indefinitely.
const store = await loadStore();
store.connectedIds.add("dead");
store.connectedIds.add("alive");
connectionIsOpen.mockImplementation((connectionId: string) => Promise.resolve(connectionId === "alive"));
await store.handleConnectionLivenessMessage({ kind: "resync" });
expect(connectionIsOpen).toHaveBeenCalledWith("dead");
expect(connectionIsOpen).toHaveBeenCalledWith("alive");
expect(store.connectedIds.has("dead")).toBe(false);
expect(store.connectedIds.has("alive")).toBe(true);
});
it("does not query the backend on a resync when nothing is connected", async () => {
const store = await loadStore();
await store.handleConnectionLivenessMessage({ kind: "resync" });
expect(connectionIsOpen).not.toHaveBeenCalled();
});
it("drops a stale confirm that lands after a newer connection attempt", async () => {
// The revision guard: a `lost` message captured before a newer connect/disconnect attempt
// must not grey out the connection that attempt produced. Only `disconnect` (and the
// connect path it mirrors) advances the generation, so the test drives it for real.
let resolveConfirm: (open: boolean) => void = () => {};
connectionIsOpen.mockImplementation(
() =>
new Promise<boolean>((resolve) => {
resolveConfirm = resolve;
}),
);
vi.doMock("@/stores/settingsStore", () => ({
useSettingsStore: () => ({ editorSettings: { disconnectTabHandlingMode: "keep-tabs" } }),
}));
vi.doMock("@/stores/queryStore", () => ({
useQueryStore: () => ({
closeConnectionTabs: vi.fn(),
releaseConnectionTabs: vi.fn(),
flushPendingPersist: vi.fn().mockResolvedValue(undefined),
}),
}));
const store = await loadStore();
store.connectedIds.add("c1");
const pending = store.handleConnectionLivenessMessage({
kind: "lost",
connectionId: "c1",
failureKind: "probe_failed",
});
// The user drops and re-establishes the connection while the probe is still in flight.
await store.disconnect("c1");
store.connectedIds.add("c1");
resolveConfirm(false);
await pending;
expect(store.connectedIds.has("c1")).toBe(true);
});
});
@@ -12,6 +12,7 @@ import type {
CompletionAssistantObjectKind,
CompletionAssistantRequest,
ConnectionConfig,
ConnectionLivenessMessage,
DatabaseType,
DatabaseConnectionInfo,
DatabaseStorageInfo,
@@ -4676,6 +4677,60 @@ export const useConnectionStore = defineStore("connection", () => {
clearConnectionPrewarmState(connectionId);
}
/**
* Re-check one connection against the backend and grey it out when it has no pool left.
*
* The confirmation is the read-only `api.connectionIsOpen`, never `checkConnectionHealth`:
* the latter removes unhealthy pools and is the path `ensureConnected` uses to reconnect, so
* confirming with it would turn a background probe into a reconnect and an error banner.
*
* A failed confirm says nothing about the connection, so it leaves the sidebar untouched.
*/
async function confirmConnectionLiveness(connectionId: string): Promise<void> {
if (!connectionId || !connectedIds.value.has(connectionId)) return;
const stateRevision = connectionStateRevision(connectionId);
let open: boolean;
try {
open = await api.connectionIsOpen(connectionId);
} catch (error) {
console.warn("[DBX] connection liveness confirm failed:", error);
return;
}
if (open) return;
// A late confirm must not undo an explicit disconnect or a newer reconnect.
if (!isCurrentConnectionStateRevision(connectionId, stateRevision)) return;
if (!connectedIds.value.has(connectionId)) return;
markConnectionOffline(connectionId);
}
/**
* Apply a message from the backend connection-liveness channel (#4339).
*
* `lost` is published only once the connection has no pools left, but it is asynchronous, so
* it is confirmed before the sidebar changes: a message generated before a reconnect but
* delivered after it confirms as "open" and is dropped.
*
* `resync` is sent when the transport skipped messages. Those are gone for good, so every
* connection this frontend still shows as connected is re-checked. Without it a dropped loss
* would leave its sidebar green until the user happened to touch that connection — the very
* bug this channel exists to fix.
*/
async function handleConnectionLivenessMessage(message: ConnectionLivenessMessage): Promise<void> {
if (!message) return;
if (message.kind === "resync") {
// Sequential on purpose: resync only happens after a transport lag, and one probe in
// flight at a time keeps a large connected set from stampeding the backend.
// Snapshot the ids: the confirm below removes entries from this set as it flips them.
for (const connectionId of Array.from(connectedIds.value)) {
await confirmConnectionLiveness(connectionId);
}
return;
}
// Defensive against a newer backend sending a kind this build does not know.
if (message.kind !== "lost") return;
await confirmConnectionLiveness(message.connectionId);
}
/**
* 断开连接。
*
@@ -10181,6 +10236,7 @@ export const useConnectionStore = defineStore("connection", () => {
disconnect,
hasDisconnectInFlight,
markConnectionOffline,
handleConnectionLivenessMessage,
metadataGenerationFor,
disconnectAndForgetConnectionPassword,
hasSessionCredential,
+20
View File
@@ -190,6 +190,26 @@ export interface ConnectionTestResult {
databaseInfo?: DatabaseConnectionInfo;
}
/**
* Why the backend declared a connection's pools dead while the app was idle (#4339).
* A stable enum by design: raw driver/network error text must not reach a background
* UI notification.
*/
export type ConnectionLivenessFailureKind = "probe_failed" | "timed_out";
/**
* A message on the backend connection-liveness channel (#4339).
*
* Mirrors the Rust `ConnectionLivenessMessage`, and both transports deliver this exact shape:
* the desktop shell forwards it unwrapped and the web SSE stream sends the object itself, so
* there is no envelope beyond the message's own `kind` discriminator.
*
* `resync` means the transport skipped messages, so the frontend must re-check every
* connection it still shows as connected — without it, a dropped `lost` would leave a
* sidebar green indefinitely.
*/
export type ConnectionLivenessMessage = { kind: "lost"; connectionId: string; failureKind: ConnectionLivenessFailureKind } | { kind: "resync" };
export type TransportLayerConfig = ({ type: "ssh" } & SshTunnelConfig) | ({ type: "proxy" } & ProxyTunnelConfig) | ({ type: "http_tunnel" } & HttpTunnelConfig);
/**
+402 -23
View File
@@ -33,7 +33,8 @@ use crate::db::proxy_tunnel::ProxyTunnelManager;
use crate::db::ssh_tunnel::TunnelManager;
use crate::models::connection::{
database_info_from_protocol_value, parse_jdbc_host_port, parse_mongo_first_host, rewrite_jdbc_url_host,
ConnectionConfig, ConnectionTestResult, DatabaseConnectionInfo, DatabaseType, TransportLayerConfig,
ConnectionConfig, ConnectionLivenessFailureKind, ConnectionLivenessMessage, ConnectionTestResult,
DatabaseConnectionInfo, DatabaseType, TransportLayerConfig,
};
use crate::mongo_oidc::MongoOidcBrowserOpener;
use crate::nacos::config::{NACOS_CONSOLE_SESSION_PASSWORD, NACOS_PRIMARY_SESSION_PASSWORD};
@@ -428,6 +429,10 @@ pub struct AppState {
/// In-memory, never-persisted 1/5 minute write overrides for read-only connections.
pub write_unlock_windows: crate::write_unlock::WriteUnlockWindows,
metadata_gates: Arc<Mutex<HashMap<String, Arc<Semaphore>>>>,
/// Backend-confirmed liveness losses plus resync requests (#4339). The keepalive task
/// publishes here once a connection has no pools left; each shell relays it to its own
/// frontend, so DBX core never needs a UI handle of its own.
connection_liveness: tokio::sync::broadcast::Sender<ConnectionLivenessMessage>,
mongo_oidc_browser_opener: std::sync::RwLock<Option<MongoOidcBrowserOpener>>,
salesforce_browser_opener: std::sync::RwLock<Option<SfBrowserOpener>>,
#[cfg(feature = "mq-admin")]
@@ -1429,6 +1434,15 @@ impl AppState {
self.connections.read().await.keys().any(|key| pool_key_belongs_to_connection(key, connection_id))
}
/// Subscribe to backend-confirmed connection liveness messages (#4339).
///
/// Each shell relays these to its own frontend. A fresh receiver only sees messages
/// published after it subscribed, so shells subscribe once at startup and keep the
/// receiver alive rather than re-subscribing per request.
pub fn subscribe_connection_liveness(&self) -> tokio::sync::broadcast::Receiver<ConnectionLivenessMessage> {
self.connection_liveness.subscribe()
}
/// Find an already-registered metadata/workload pool for a metadata read.
/// Unlike `get_or_create_metadata_pool_for_session`, this is a pure lookup:
/// it never validates credentials, starts an agent, or opens a transport.
@@ -1623,6 +1637,10 @@ impl AppState {
session_credentials: SessionCredentialStore::new(),
write_unlock_windows: crate::write_unlock::WriteUnlockWindows::default(),
metadata_gates: Arc::new(Mutex::new(HashMap::new())),
// Bounded so a shell that stops draining cannot grow memory without bound. A
// lagged receiver re-syncs from the read-only status check on its next use, so
// dropping events here is safe.
connection_liveness: tokio::sync::broadcast::Sender::new(64),
mongo_oidc_browser_opener: std::sync::RwLock::new(None),
salesforce_browser_opener: std::sync::RwLock::new(None),
#[cfg(feature = "mq-admin")]
@@ -2167,6 +2185,10 @@ impl AppState {
let routing = self.pool_routing_control();
let connections = self.connections.clone();
let running_queries = self.running_queries.clone();
let liveness = self.connection_liveness.clone();
// Liveness is keyed by connection, not by pool: pool keys are per session/role/
// database derived, so only the connection id identifies "this connection is gone".
let connection_id = config.id.clone();
// MQ pool markers are empty; close_pool_kind is a no-op, so keepalive must
// drop the registry adapter or reconnect would reuse a dead agent.
#[cfg(feature = "mq-admin")]
@@ -2209,22 +2231,25 @@ impl AppState {
}
#[cfg(feature = "mq-admin")]
let drop_mq = matches!(target, KeepaliveTarget::MessageQueue(_));
if !detach_keepalive_target_if_current(
// Only a pool we actually detached may report liveness loss; a
// probe failure on an already-replaced pool must not.
let detached = detach_and_report_keepalive_loss(
&routing,
&connections,
&liveness,
&key,
target,
&connection_id,
ConnectionLivenessFailureKind::ProbeFailed,
replace_runtime,
)
.await
{
log::debug!("Skipping stale keepalive result for replaced pool '{key}'");
} else {
#[cfg(feature = "mq-admin")]
if drop_mq {
mq_registry.drop_connection(&mq_connection_id).await;
}
.await;
#[cfg(feature = "mq-admin")]
if detached && drop_mq {
mq_registry.drop_connection(&mq_connection_id).await;
}
#[cfg(not(feature = "mq-admin"))]
let _ = detached;
break;
}
Err(_) => {
@@ -2241,14 +2266,26 @@ impl AppState {
}
#[cfg(feature = "mq-admin")]
let drop_mq = matches!(target, KeepaliveTarget::MessageQueue(_));
if !detach_keepalive_target_if_current(&routing, &connections, &key, target, false).await {
log::debug!("Skipping stale keepalive timeout for replaced pool '{key}'");
} else {
#[cfg(feature = "mq-admin")]
if drop_mq {
mq_registry.drop_connection(&mq_connection_id).await;
}
// The pool is torn down either way, so a timed-out probe is reported
// too: staying silent would keep the sidebar claiming the connection
// is usable.
let detached = detach_and_report_keepalive_loss(
&routing,
&connections,
&liveness,
&key,
target,
&connection_id,
ConnectionLivenessFailureKind::TimedOut,
false,
)
.await;
#[cfg(feature = "mq-admin")]
if detached && drop_mq {
mq_registry.drop_connection(&mq_connection_id).await;
}
#[cfg(not(feature = "mq-admin"))]
let _ = detached;
break;
}
}
@@ -5965,6 +6002,50 @@ fn keepalive_failure_proves_pool_dead(error: &str) -> bool {
!crate::query::is_pool_saturation_error(error)
}
/// Tells the shells that `connection_id` has no pools left, so the frontend can stop
/// claiming the connection is connected (#4339).
///
/// Pool keys are per session/role/database derived, so one dying derived pool says nothing
/// about the connection root: only the last pool to go makes the connection really unusable.
/// Gating on "no pools left" is also what collapses N per-pool keepalive failures into one
/// logical disconnect. Two pools of the same connection can both finish detaching before
/// either evaluates this, publishing twice; the frontend path is idempotent.
async fn report_connection_liveness_lost(
connections: &Arc<RwLock<ConnectionPoolRegistry>>,
liveness: &tokio::sync::broadcast::Sender<ConnectionLivenessMessage>,
connection_id: &str,
failure_kind: ConnectionLivenessFailureKind,
) {
if connections.read().await.keys().any(|key| pool_key_belongs_to_connection(key, connection_id)) {
return;
}
// A missing subscriber (headless shell) is not an error worth reporting.
let _ = liveness.send(ConnectionLivenessMessage::Lost { connection_id: connection_id.to_string(), failure_kind });
}
/// Detaches a probe's pool when it is still the current one and reports the loss if the
/// connection has nothing left (#4339).
///
/// Returns whether this probe's pool was still current and was detached. A stale probe (its
/// pool was already replaced) reports nothing: its result says nothing about the connection.
async fn detach_and_report_keepalive_loss(
routing: &PoolRoutingControl,
connections: &Arc<RwLock<ConnectionPoolRegistry>>,
liveness: &tokio::sync::broadcast::Sender<ConnectionLivenessMessage>,
pool_key: &str,
target: &KeepaliveTarget,
connection_id: &str,
failure_kind: ConnectionLivenessFailureKind,
replace_agent_runtime: bool,
) -> bool {
if !detach_keepalive_target_if_current(routing, connections, pool_key, target, replace_agent_runtime).await {
log::debug!("Skipping stale keepalive {failure_kind:?} result for replaced pool '{pool_key}'");
return false;
}
report_connection_liveness_lost(connections, liveness, connection_id, failure_kind).await;
true
}
impl KeepaliveTarget {
fn matches_pool(&self, pool: &PoolKind) -> bool {
match (self, pool) {
@@ -6945,11 +7026,11 @@ mod tests {
kafka_single_loopback_bootstrap_endpoint, keepalive_failure_proves_pool_dead, metadata_connection_config,
metadata_pool_database, mysql_metadata_fallback_url, mysql_pool_setup_queries, oceanbase_mysql_setup_queries,
prestosql_jdbc_config_for_endpoint, redacted_connection_url_for_endpoint, redis_sentinel_transport_id,
redis_sentinel_transport_prefix, sqlserver_legacy_agent_config, sqlserver_legacy_driver_error,
sqlserver_uses_legacy_driver, task_client_session_id, transport_layers_through_last_ssh,
upsert_connection_url_param, uses_bare_mysql_pool, uses_tcp_probe, validate_connection_url_params,
validate_h2_database_path, AppState, MysqlMode, PoolKind, TxnConnection, GAUSSDB_M_JDBC_DRIVER_CLASS,
GAUSSDB_M_JDBC_DRIVER_PROFILE, PRESTOSQL_JDBC_DRIVER_CLASS,
redis_sentinel_transport_prefix, report_connection_liveness_lost, sqlserver_legacy_agent_config,
sqlserver_legacy_driver_error, sqlserver_uses_legacy_driver, task_client_session_id,
transport_layers_through_last_ssh, upsert_connection_url_param, uses_bare_mysql_pool, uses_tcp_probe,
validate_connection_url_params, validate_h2_database_path, AppState, MysqlMode, PoolKind, TxnConnection,
GAUSSDB_M_JDBC_DRIVER_CLASS, GAUSSDB_M_JDBC_DRIVER_PROFILE, PRESTOSQL_JDBC_DRIVER_CLASS,
};
use crate::agent_connection::{
agent_connect_params, mongo_legacy_error_with_auth_hint, mongo_uses_legacy_driver,
@@ -6960,7 +7041,8 @@ mod tests {
use crate::db;
use crate::models::connection::{
default_connect_timeout_secs, default_redis_key_separator, AttachedDatabaseConfig, ConnectionConfig,
DatabaseType, HttpTunnelConfig, ProxyTunnelConfig, ProxyType, SshTunnelConfig, TransportLayerConfig,
ConnectionLivenessFailureKind, ConnectionLivenessMessage, DatabaseType, HttpTunnelConfig, ProxyTunnelConfig,
ProxyType, SshTunnelConfig, TransportLayerConfig,
};
use crate::query;
use crate::schema;
@@ -9432,6 +9514,303 @@ sleep 30
let _ = std::fs::remove_dir_all(dir);
}
/// Minimal `ConnectionConfig` for driving `start_keepalive_task` directly.
///
/// The keepalive path only reads `id`, the effective timeouts and
/// `keepalive_interval_secs`; the rest is filler so the struct can be built without a
/// live database. `connect_timeout_secs` is 1 so a failing probe resolves fast.
fn keepalive_test_config(id: &str, interval_secs: u64) -> ConnectionConfig {
ConnectionConfig {
docs_notes_path: None,
id: id.to_string(),
name: id.to_string(),
note: String::new(),
db_type: DatabaseType::Oracle,
driver_profile: None,
driver_label: None,
url_params: None,
agent_java_options: Vec::new(),
host: "127.0.0.1".to_string(),
port: 1521,
username: "dbx".to_string(),
password: "dbx".to_string(),
database: None,
default_schema: None,
visible_databases: None,
visible_database_patterns: None,
visible_schemas: None,
attached_databases: Vec::new(),
init_script: None,
color: None,
transport_layers: Vec::new(),
connect_timeout_secs: 1,
query_timeout_secs: 30,
idle_timeout_secs: 60,
keepalive_interval_secs: interval_secs,
ssl: false,
ca_cert_path: String::new(),
client_cert_path: String::new(),
client_key_path: String::new(),
sysdba: false,
oracle_connection_type: None,
connection_string: None,
redis_connection_mode: None,
redis_sentinel_master: String::new(),
redis_sentinel_nodes: String::new(),
redis_sentinel_username: String::new(),
redis_sentinel_password: String::new(),
redis_sentinel_tls: false,
redis_cluster_nodes: String::new(),
redis_key_separator: crate::models::connection::default_redis_key_separator(),
redis_scan_page_size: None,
redis_database_aliases: Default::default(),
redis_key_templates: Vec::new(),
redis_key_grouping: None,
etcd_endpoints: String::new(),
gbase_server: String::new(),
informix_server: String::new(),
external_config: None,
plugin_id: None,
plugin_connection_provider: None,
plugin_connection_type: None,
connection_secrets: Default::default(),
jdbc_driver_class: None,
jdbc_driver_paths: Vec::new(),
one_time: false,
save_password: true,
read_only: false,
is_production: false,
production_databases: vec![],
show_system_schemas: false,
sidebar_auto_load_all_tables: false,
database_info: None,
}
}
/// Fake agent whose `validate_connection` reports a real failure.
///
/// It must NOT answer with an unsupported-method error: `ping_keepalive_target` treats that
/// as a healthy connection (`is_agent_validate_connection_unsupported`), which would make
/// this fixture silently green.
const KEEPALIVE_FAILING_AGENT: &str = r#"import json, sys
print(json.dumps({'ready': True}), flush=True)
for line in sys.stdin:
req = json.loads(line)
if req['method'] == 'handshake':
result = {'protocolVersion': 2, 'agentProtocolVersion': 2, 'capabilities': ['multi_session']}
print(json.dumps({'jsonrpc': '2.0', 'id': req['id'], 'result': result}), flush=True)
elif 'validate_connection' in req['method']:
print(json.dumps({'jsonrpc': '2.0', 'id': req['id'], 'error': {'code': -32000, 'message': 'connection refused by keepalive fixture'}}), flush=True)
else:
print(json.dumps({'jsonrpc': '2.0', 'id': req['id'], 'result': {}}), flush=True)
"#;
async fn spawn_keepalive_agent(
dir: &std::path::Path,
name: &str,
) -> std::sync::Arc<crate::db::agent_driver::PooledAgentClient> {
let script_path = dir.join(name);
std::fs::write(&script_path, KEEPALIVE_FAILING_AGENT).unwrap();
let python = if cfg!(windows) { "python" } else { "python3" };
let runtime = crate::db::agent_driver::AgentRuntimeClient::spawn(
crate::db::agent_driver::AgentLaunchSpec::new(python)
.with_args([script_path.to_string_lossy().to_string()]),
"test",
)
.await
.unwrap();
runtime.increment_session_count();
std::sync::Arc::new(crate::db::agent_driver::PooledAgentClient::new(
crate::db::agent_driver::AgentDriverClient::shared_session(runtime, format!("{name}-session")),
))
}
/// A failing probe on the connection's current pool must reach the UI (#4339). This drives
/// the real keepalive task rather than the publish helper, so it pins the wiring between
/// the probe's failure branch and the report.
#[tokio::test]
async fn keepalive_probe_failure_reports_a_lost_connection() {
let (state, dir) = test_app_state().await;
let config = keepalive_test_config("conn", 1);
state.configs.write().await.insert(config.id.clone(), config.clone());
let client = spawn_keepalive_agent(&dir, "failing-keepalive-agent.py").await;
state.connections.write().await.insert("conn".to_string(), PoolKind::Agent(std::sync::Arc::clone(&client)));
let pool = PoolKind::Agent(client);
let mut events = state.subscribe_connection_liveness();
state.start_keepalive_task(
"conn",
&pool,
&config,
#[cfg(feature = "mq-admin")]
None,
);
let message = tokio::time::timeout(Duration::from_secs(10), events.recv())
.await
.expect("a failing probe must report a liveness loss")
.expect("the channel must stay open");
assert_eq!(
message,
ConnectionLivenessMessage::Lost {
connection_id: "conn".to_string(),
failure_kind: ConnectionLivenessFailureKind::ProbeFailed,
}
);
state.stop_keepalive_task("conn").await;
let _ = std::fs::remove_dir_all(dir);
}
/// A pool that is busy with a running query is not evidence of a dead connection: the probe
/// is skipped entirely, so nothing may be reported (#4339).
#[tokio::test]
async fn keepalive_busy_pool_reports_nothing() {
let (state, dir) = test_app_state().await;
let config = keepalive_test_config("conn", 1);
state.configs.write().await.insert(config.id.clone(), config.clone());
let client = spawn_keepalive_agent(&dir, "busy-keepalive-agent.py").await;
state.connections.write().await.insert("conn".to_string(), PoolKind::Agent(std::sync::Arc::clone(&client)));
let pool = PoolKind::Agent(client);
// `is_pool_active` is true whenever a registered execution owns this pool key.
let registered = state.running_queries.register("keepalive-busy-exec".to_string());
state.running_queries.set_pool_key("keepalive-busy-exec", "conn");
assert!(state.running_queries.is_pool_active("conn"));
let mut events = state.subscribe_connection_liveness();
state.start_keepalive_task(
"conn",
&pool,
&config,
#[cfg(feature = "mq-admin")]
None,
);
// Long enough for more than one interval to elapse.
let outcome = tokio::time::timeout(Duration::from_millis(1500), events.recv()).await;
assert!(outcome.is_err(), "a busy pool must not report a liveness loss: {outcome:?}");
state.stop_keepalive_task("conn").await;
drop(registered);
let _ = std::fs::remove_dir_all(dir);
}
/// A probe whose pool was already replaced reports nothing: its result says nothing about
/// the connection that is live now (#4339).
#[tokio::test]
async fn keepalive_stale_target_reports_nothing() {
let (state, dir) = test_app_state().await;
let config = keepalive_test_config("conn", 1);
state.configs.write().await.insert(config.id.clone(), config.clone());
// The registry holds `current`, but the keepalive task is started for a different client,
// so the detach cannot match the live pool and must stay silent.
let current = spawn_keepalive_agent(&dir, "stale-current-agent.py").await;
let stale = spawn_keepalive_agent(&dir, "stale-replaced-agent.py").await;
state.connections.write().await.insert("conn".to_string(), PoolKind::Agent(current));
let mut events = state.subscribe_connection_liveness();
state.start_keepalive_task(
"conn",
&PoolKind::Agent(stale),
&config,
#[cfg(feature = "mq-admin")]
None,
);
let outcome = tokio::time::timeout(Duration::from_millis(1500), events.recv()).await;
assert!(outcome.is_err(), "a stale probe must not report a liveness loss: {outcome:?}");
state.stop_keepalive_task("conn").await;
let _ = std::fs::remove_dir_all(dir);
}
/// Stopping the keepalive task (explicit disconnect / reconnect) must not report a loss for
/// a pool that was never detached (#4339).
#[tokio::test]
async fn keepalive_stop_reports_nothing() {
let (state, dir) = test_app_state().await;
let config = keepalive_test_config("conn", 1);
state.configs.write().await.insert(config.id.clone(), config.clone());
let client = spawn_keepalive_agent(&dir, "stopped-keepalive-agent.py").await;
state.connections.write().await.insert("conn".to_string(), PoolKind::Agent(std::sync::Arc::clone(&client)));
let pool = PoolKind::Agent(client);
let mut events = state.subscribe_connection_liveness();
state.start_keepalive_task(
"conn",
&pool,
&config,
#[cfg(feature = "mq-admin")]
None,
);
state.stop_keepalive_task("conn").await;
let outcome = tokio::time::timeout(Duration::from_millis(1500), events.recv()).await;
assert!(outcome.is_err(), "a stopped keepalive must not report a liveness loss: {outcome:?}");
let _ = std::fs::remove_dir_all(dir);
}
/// The sidebar must not grey out a connection because one derived pool died: pool keys
/// are per session/role/database derived, so only the last pool going makes the
/// connection really unusable. This gate is also what collapses N per-pool keepalive
/// failures into a single logical disconnect (#4339).
#[tokio::test]
async fn liveness_is_published_only_after_the_connections_last_pool_is_gone() {
let (state, dir) = test_app_state().await;
let pool = crate::db::sqlite::connect_path(":memory:").await.unwrap();
let mut events = state.subscribe_connection_liveness();
state
.update_connection_pools(|connections| {
connections.insert("conn".to_string(), PoolKind::Sqlite(pool.clone()));
connections.insert("conn:analytics:session:tab-1".to_string(), PoolKind::Sqlite(pool.clone()));
// A sibling id sharing the prefix is a different connection and must not
// keep "conn" alive.
connections.insert("conn-2".to_string(), PoolKind::Sqlite(pool));
})
.await;
// A derived pool dying while another pool of the same connection survives says
// nothing about the connection root.
state.update_connection_pools(|connections| connections.remove("conn:analytics:session:tab-1")).await;
report_connection_liveness_lost(
&state.connections,
&state.connection_liveness,
"conn",
ConnectionLivenessFailureKind::ProbeFailed,
)
.await;
assert!(
matches!(events.try_recv(), Err(tokio::sync::broadcast::error::TryRecvError::Empty)),
"a surviving pool must suppress the liveness message"
);
// The last pool going is a real disconnect, and only then is it reported.
state.update_connection_pools(|connections| connections.remove("conn")).await;
report_connection_liveness_lost(
&state.connections,
&state.connection_liveness,
"conn",
ConnectionLivenessFailureKind::TimedOut,
)
.await;
let message = events.try_recv().expect("the last pool going must report a liveness loss");
assert_eq!(
message,
ConnectionLivenessMessage::Lost {
connection_id: "conn".to_string(),
failure_kind: ConnectionLivenessFailureKind::TimedOut,
}
);
assert!(
matches!(events.try_recv(), Err(tokio::sync::broadcast::error::TryRecvError::Empty)),
"one outage publishes exactly one message"
);
let _ = std::fs::remove_dir_all(dir);
}
#[tokio::test]
async fn connection_slot_recovery_waits_for_agent_metadata_and_confirms_close() {
let (state, dir) = test_app_state().await;
+29
View File
@@ -12,6 +12,35 @@ pub enum IdentifierCase {
Mixed,
}
/// Why the backend declared a connection's pools dead while nobody was using it.
///
/// A stable enum on purpose: the frontend only needs to tell the two cases apart, and raw
/// driver/network error text must not leak into a background UI notification (#4339).
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum ConnectionLivenessFailureKind {
/// The keepalive probe ran and reported a dead connection.
ProbeFailed,
/// The keepalive probe exceeded its budget.
TimedOut,
}
/// A message on the connection-liveness channel (#4339).
///
/// One shape for both shells and both transports, discriminated by `kind`. The `Resync`
/// variant exists because a broadcast subscriber can fall behind: without a way to say
/// "you skipped messages", a dropped `Lost` would leave a sidebar claiming a dead
/// connection is connected until the user happened to act on it.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(tag = "kind", rename_all = "snake_case", rename_all_fields = "camelCase")]
pub enum ConnectionLivenessMessage {
/// The connection has no pools left, so the sidebar must stop showing it as connected.
Lost { connection_id: String, failure_kind: ConnectionLivenessFailureKind },
/// The transport skipped messages and must re-check every connection it currently
/// shows as connected.
Resync,
}
#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub struct DatabaseConnectionInfo {
+2
View File
@@ -531,6 +531,8 @@ async fn serve() {
.route("/connection/final-proxy-port", post(routes::connection::connection_final_proxy_port))
.route("/connection/disconnect", post(routes::connection::disconnect_db))
.route("/connection/check-health", post(routes::connection::check_connection_health))
.route("/connection/is-open", post(routes::connection::connection_is_open))
.route("/connection/liveness-events", get(routes::connection::connection_liveness_events))
.route("/connection/prewarm", post(routes::connection::prewarm_connection))
.route("/connection/session-credential-status", post(routes::connection::session_credential_status))
.route("/connection/forget-session-credential", post(routes::connection::forget_session_credential))
+60 -1
View File
@@ -1,13 +1,18 @@
use std::collections::HashSet;
use std::convert::Infallible;
use std::sync::Arc;
use async_stream::stream;
use axum::extract::{Query, State};
use axum::http::HeaderMap;
use axum::response::sse::{Event, KeepAlive, Sse};
use axum::Json;
use dbx_core::connection::{
connection_configs_pool_equivalent, connection_configs_session_credentials_compatible, AppState, PoolKind,
};
use dbx_core::models::connection::{ConnectionConfig, ConnectionTestResult, DatabaseConnectionInfo, DatabaseType};
use dbx_core::models::connection::{
ConnectionConfig, ConnectionLivenessMessage, ConnectionTestResult, DatabaseConnectionInfo, DatabaseType,
};
use dbx_core::nacos::config::{
take_transient_passwords, NACOS_CONSOLE_SESSION_PASSWORD, NACOS_PRIMARY_SESSION_PASSWORD,
};
@@ -625,6 +630,60 @@ pub async fn check_connection_health(
Ok(Json(()))
}
/// Read-only counterpart of `check_connection_health`: reports whether the connection still
/// has a pool, without probing, mutating, or triggering a reconnect (#4339).
pub async fn connection_is_open(
State(state): State<Arc<WebState>>,
Json(body): Json<DisconnectRequest>,
) -> Result<Json<bool>, AppError> {
Ok(Json(state.app.is_connection_open(&body.connection_id).await))
}
/// Serialise one liveness message for the SSE stream, or `None` when serialisation fails.
fn liveness_sse_event(message: &ConnectionLivenessMessage) -> Option<Event> {
match serde_json::to_string(message) {
Ok(payload) => Some(Event::default().data(payload)),
// The payload is a tagged enum of plain fields, so this is unreachable in practice.
Err(error) => {
log::warn!("Connection liveness message could not be serialised: {error}");
None
}
}
}
/// Stream backend-confirmed liveness messages to the browser (#4339).
///
/// The payload is the message itself, byte-identical to what the desktop shell emits as
/// `dbx-connection-liveness`, so the frontend parses one shape on both transports.
pub async fn connection_liveness_events(
State(state): State<Arc<WebState>>,
) -> Sse<impl futures::Stream<Item = Result<Event, Infallible>>> {
let mut events = state.app.subscribe_connection_liveness();
let stream = stream! {
loop {
match events.recv().await {
Ok(message) => {
if let Some(event) = liveness_sse_event(&message) {
yield Ok(event);
}
}
Err(tokio::sync::broadcast::error::RecvError::Lagged(skipped)) => {
// The skipped messages are gone for good, so ask the client to re-check
// every connection it still shows as connected. Dropping the transition
// silently would leave a sidebar green indefinitely — exactly the state
// this stream exists to prevent.
log::warn!("Web connection liveness stream skipped {skipped} messages; requesting a resync");
if let Some(event) = liveness_sse_event(&ConnectionLivenessMessage::Resync) {
yield Ok(event);
}
}
Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
}
}
};
Sse::new(stream).keep_alive(KeepAlive::default())
}
pub async fn prewarm_connection(
State(state): State<Arc<WebState>>,
Json(body): Json<PrewarmConnectionRequest>,
+41 -3
View File
@@ -1,6 +1,6 @@
use std::collections::HashSet;
use std::sync::Arc;
use tauri::State;
use tauri::{Emitter, State};
pub use dbx_core::agent_connection::{
agent_connect_params, mongo_legacy_error_with_auth_hint, mongo_uses_legacy_driver, oracle_alternate_connect_config,
@@ -16,8 +16,8 @@ use dbx_core::database_capabilities;
use dbx_core::db;
use dbx_core::db::agent_driver::{AgentDriverClient, AgentMethod};
use dbx_core::models::connection::{
database_info_from_protocol_value, rewrite_jdbc_url_host, ConnectionConfig, ConnectionTestResult,
DatabaseConnectionInfo, DatabaseType,
database_info_from_protocol_value, rewrite_jdbc_url_host, ConnectionConfig, ConnectionLivenessMessage,
ConnectionTestResult, DatabaseConnectionInfo, DatabaseType,
};
pub use dbx_core::path_utils::expand_tilde;
use dbx_core::runtime_config::{release_runtime_config_on_disconnect, should_retain_runtime_config};
@@ -2306,6 +2306,44 @@ pub async fn check_connection_health(state: State<'_, Arc<AppState>>, connection
state.check_connection_health(&connection_id).await
}
/// Read-only counterpart of `check_connection_health`: reports whether the connection still
/// has a pool, without probing, mutating, or triggering a reconnect.
///
/// The frontend uses it to confirm a keepalive liveness event before greying the sidebar
/// (#4339). `check_connection_health` must never be used for that confirmation: it removes
/// unhealthy pools and is the path `ensureConnected` uses to reconnect.
#[tauri::command]
pub async fn connection_is_open(state: State<'_, Arc<AppState>>, connection_id: String) -> Result<bool, String> {
Ok(state.is_connection_open(&connection_id).await)
}
/// Relay backend-confirmed liveness losses to the frontend (#4339).
///
/// Mirrors `install_plugin_event_bridge`: the core publishes to a broadcast channel and each
/// shell owns its transport, so core never needs a UI handle.
pub fn install_connection_liveness_bridge(app: &tauri::AppHandle, state: Arc<AppState>) {
let app_handle = app.clone();
let mut events = state.subscribe_connection_liveness();
tauri::async_runtime::spawn(async move {
loop {
match events.recv().await {
Ok(message) => {
let _ = app_handle.emit("dbx-connection-liveness", message);
}
Err(tokio::sync::broadcast::error::RecvError::Lagged(skipped)) => {
// The skipped messages are gone for good, so ask the frontend to re-check
// every connection it still shows as connected. Dropping the transition
// silently would leave a sidebar green indefinitely — exactly the state
// this bridge exists to prevent.
log::warn!("Desktop connection liveness bridge skipped {skipped} messages; requesting a resync");
let _ = app_handle.emit("dbx-connection-liveness", ConnectionLivenessMessage::Resync);
}
Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
}
}
});
}
/// Warm the driver and connection pool for a connection a tab is opening, so the
/// first Run does not pay pool creation, tunnel setup, or external-driver (JDBC
/// agent) startup while the user waits. Never removes an existing pool.
+2
View File
@@ -1699,6 +1699,7 @@ pub fn run() {
let state = Arc::new(state);
app.manage(state.clone());
commands::plugins::install_plugin_event_bridge(app.handle(), state.clone());
commands::connection::install_connection_liveness_bridge(app.handle(), state.clone());
let backups = tauri::async_runtime::block_on(async {
background_backup::BackgroundBackup::new(state.clone(), data_dir.clone())
});
@@ -1968,6 +1969,7 @@ pub fn run() {
commands::connection::clear_all_session_credentials,
commands::connection::refresh_connections,
commands::connection::check_connection_health,
commands::connection::connection_is_open,
commands::connection::prewarm_connection,
commands::connection::connection_identifier_quote,
commands::connection::connection_database_info,