xiaoxue_admin
/
dbx
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix(connection): recover connection state after VPN failures (#641)

This commit is contained in:
不吃西葫芦 2026-06-24 01:44:46 +08:00 committed by GitHub
parent 3ebfae05a2
commit 9a0a32e04c
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
8 changed files with 469 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
apps/desktop/src/components/sidebar/TreeItem.vue
View File

@ -425,7 +425,10 @@ function isTooltipDisabled(): boolean {
async function toggle() {
const node = props.node;
if (node.isLoading) return;
if (node.isLoading) {
if (node.type !== "connection") return;
node.isLoading = false;
}
emit("search-toggle", node);
const wasExpanded = !!node.isExpanded;

1
apps/desktop/src/lib/sidebarSearchTree.ts
View File

@ -51,6 +51,7 @@ function filterSidebarTreeWithMatcher(nodes: TreeNode[], matchLabel: SidebarLabe
node: {
...node,
children,
isLoading: node.type === "connection" ? false : node.isLoading,
isExpanded: children.length > 0 && !collapsedIds.has(node.id),
},
score: selfMatch?.score ?? 0,

98
apps/desktop/src/stores/connectionStore.ts
View File

@ -56,6 +56,9 @@ import { completionSchemasFromTree, completionTablesFromTree } from "@/lib/compl
const PINNED_TREE_NODES_STORAGE_KEY = "dbx-pinned-tree-nodes";
const ACTIVE_CONNECTION_STORAGE_KEY = "dbx-active-connection";
const CONNECTION_HEALTH_CHECK_TTL_MS = 2000;
const METADATA_LOAD_MIN_TIMEOUT_MS = 15_000;
const METADATA_LOAD_DISABLED_QUERY_TIMEOUT_MS = 60_000;
const DISCONNECT_REQUEST_TIMEOUT_MS = 5_000;
const MONGO_LEGACY_DRIVER_PROFILE = "mongodb-legacy";
const MONGO_LEGACY_DRIVER_LABEL = "MongoDB (Legacy)";
function sidebarObjectGroupPageSize(): number {
@ -273,6 +276,57 @@ export const useConnectionStore = defineStore("connection", () => {
return typeof checkedAt === "number" && Date.now() - checkedAt < CONNECTION_HEALTH_CHECK_TTL_MS;
}
function clearConnectionNodeLoading(connectionId: string) {
const node = findNode(treeNodes.value, connectionId);
if (node) node.isLoading = false;
}
function metadataLoadTimeoutMs(config?: ConnectionConfig): number {
const queryTimeoutSecs = Number(config?.query_timeout_secs);
if (queryTimeoutSecs === 0) return METADATA_LOAD_DISABLED_QUERY_TIMEOUT_MS;
const boundedTimeoutSecs = Number.isFinite(queryTimeoutSecs) && queryTimeoutSecs > 0 ? queryTimeoutSecs + 5 : 35;
return Math.max(METADATA_LOAD_MIN_TIMEOUT_MS, boundedTimeoutSecs * 1000);
}
async function withMetadataLoadTimeout<T>(connectionId: string, promise: Promise<T>, label: string): Promise<T> {
const timeoutMs = metadataLoadTimeoutMs(getConfig(connectionId));
let timer: ReturnType<typeof setTimeout> | undefined;
try {
return await Promise.race([
promise,
new Promise<never>((_, reject) => {
timer = setTimeout(() => {
reject(new Error(`Connection timed out while loading ${label} after ${Math.ceil(timeoutMs / 1000)}s. Please check the network or VPN and try again.`));
}, timeoutMs);
}),
]);
} finally {
if (timer) clearTimeout(timer);
}
}
async function withDisconnectRequestTimeout(connectionId: string, promise: Promise<void>): Promise<void> {
let timedOut = false;
let timer: ReturnType<typeof setTimeout> | undefined;
void promise.catch((error) => {
if (timedOut) console.warn("[DBX][connection:disconnect-late-error]", { connectionId, error });
});
try {
await Promise.race([
promise,
new Promise<void>((resolve) => {
timer = setTimeout(() => {
timedOut = true;
console.warn("[DBX][connection:disconnect-timeout]", { connectionId, timeoutMs: DISCONNECT_REQUEST_TIMEOUT_MS });
resolve();
}, DISCONNECT_REQUEST_TIMEOUT_MS);
}),
]);
} finally {
if (timer) clearTimeout(timer);
}
}
function recordConnectionError(connectionId: string, error: unknown): string {
const message = connectionErrorMessage(error);
setConnectionError(connectionId, message);
@ -281,6 +335,7 @@ export const useConnectionStore = defineStore("connection", () => {
function markConnectionLost(connectionId: string, error: unknown) {
connectedIds.value.delete(connectionId);
clearConnectionNodeLoading(connectionId);
clearConnectionHealthCheck(connectionId);
if (activeConnectionId.value === connectionId) activeConnectionId.value = null;
recordConnectionError(connectionId, error);
@ -305,12 +360,22 @@ export const useConnectionStore = defineStore("connection", () => {
const timeoutMessage = connectionAttemptTimeoutMessage(timeoutMs);
let timedOut = false;
let timer: ReturnType<typeof setTimeout> | undefined;
void promise.catch((error) => {
if (!timedOut) return;
const current = connectionErrors.value[config.id];
if (current !== timeoutMessage) return;
setConnectionError(config.id, connectionAttemptOriginalErrorMessage(timeoutMessage, connectionErrorMessage(error)));
});
void promise.then(
(connectionId) => {
if (!timedOut) return;
const cleanupConnectionId = typeof connectionId === "string" && connectionId ? connectionId : config.id;
if (connectedIds.value.has(cleanupConnectionId)) return;
void api.disconnectDb(cleanupConnectionId).catch((error) => {
console.warn("[DBX][connection:timeout-cleanup-failed]", { connectionId: cleanupConnectionId, error });
});
},
(error) => {
if (!timedOut) return;
const current = connectionErrors.value[config.id];
if (current !== timeoutMessage) return;
setConnectionError(config.id, connectionAttemptOriginalErrorMessage(timeoutMessage, connectionErrorMessage(error)));
},
);
try {
return await Promise.race([
promise,
@ -928,7 +993,7 @@ export const useConnectionStore = defineStore("connection", () => {
async function disconnect(connectionId: string) {
const shouldRemoveOneTimeConnection = getConfig(connectionId)?.one_time === true;
await api.disconnectDb(connectionId);
await withDisconnectRequestTimeout(connectionId, api.disconnectDb(connectionId));
clearConnectionError(connectionId);
const { useQueryStore } = await import("@/stores/queryStore");
const queryStore = useQueryStore();
@ -946,6 +1011,7 @@ export const useConnectionStore = defineStore("connection", () => {
clearConnectionHealthCheck(connectionId);
const node = findNode(treeNodes.value, connectionId);
if (node) {
node.isLoading = false;
node.isExpanded = false;
node.children = [];
}
@ -1043,7 +1109,7 @@ export const useConnectionStore = defineStore("connection", () => {
return;
}
}
const [databases, schemas] = await Promise.all([api.listDatabases(connectionId), api.listSchemas(connectionId, "main")]);
const [databases, schemas] = await Promise.all([withMetadataLoadTimeout(connectionId, api.listDatabases(connectionId), "databases"), withMetadataLoadTimeout(connectionId, api.listSchemas(connectionId, "main"), "schemas")]);
const children = withSavedSqlRoot(connectionId, buildDuckDbConnectionTreeNodes(connectionId, databases, schemas), node);
setChildren(node, children);
await savePersistedTreeChildren(cacheKey, children);
@ -1057,7 +1123,7 @@ export const useConnectionStore = defineStore("connection", () => {
return;
}
}
const schemas = await api.listSchemas(connectionId, effectiveDb);
const schemas = await withMetadataLoadTimeout(connectionId, api.listSchemas(connectionId, effectiveDb), "schemas");
const visibleSchemas = filterDatabaseNamesForConnection(schemas, config);
const schemaNodes: TreeNode[] = sortSidebarNames(visibleSchemas).map((s) => ({
id: `${connectionId}:${s}:${s}`,
@ -1080,7 +1146,7 @@ export const useConnectionStore = defineStore("connection", () => {
return;
}
}
const databases = await api.listDatabases(connectionId);
const databases = await withMetadataLoadTimeout(connectionId, api.listDatabases(connectionId), "databases");
const visibleNames = filterDatabaseNamesForConnection(
databases.map((database) => database.name),
config,
@ -1092,7 +1158,7 @@ export const useConnectionStore = defineStore("connection", () => {
includeDefaultWhenEmpty: usesTreeSchemaMode(effectiveDbType) || shouldIncludeDefaultDatabaseNode(config, visibleDatabases),
});
if (config?.db_type === "sqlserver") {
const linkedServers = await api.listSqlServerLinkedServers(connectionId).catch(() => []);
const linkedServers = await withMetadataLoadTimeout(connectionId, api.listSqlServerLinkedServers(connectionId), "linked servers").catch(() => []);
const linkedDatabase = sqlServerLinkedRuntimeDatabase(config);
databaseNodes.push({
...sqlServerLinkedRootNode(connectionId, linkedDatabase),
@ -1130,7 +1196,7 @@ export const useConnectionStore = defineStore("connection", () => {
node.isLoading = true;
try {
await ensureConnected(connectionId);
const dbs = await api.redisListDatabases(connectionId);
const dbs = await withMetadataLoadTimeout(connectionId, api.redisListDatabases(connectionId), "Redis databases");
const config = getConfig(connectionId);
const visibleNames = filterVisibleDatabaseNames(
dbs.map((db) => String(db.db)),
@ -1209,7 +1275,7 @@ export const useConnectionStore = defineStore("connection", () => {
await ensureConnected(connectionId);
if (useCachedChildren(node, options)) return;
const tenants = await api.mqListTenants(connectionId);
const tenants = await withMetadataLoadTimeout(connectionId, api.mqListTenants(connectionId), "message queue tenants");
const tenantNames = sortSidebarNames(tenants.map((tenant) => tenant.name).filter((name) => !!name.trim()));
setChildren(
node,
@ -1307,7 +1373,7 @@ export const useConnectionStore = defineStore("connection", () => {
node.isLoading = true;
try {
await ensureConnected(connectionId);
const dbs = await api.mongoListDatabases(connectionId);
const dbs = await withMetadataLoadTimeout(connectionId, api.mongoListDatabases(connectionId), "MongoDB databases");
const config = getConfig(connectionId);
const visibleDbs = filterDatabaseNamesForConnection(dbs, config);
setChildren(
@ -1342,7 +1408,7 @@ export const useConnectionStore = defineStore("connection", () => {
node.isLoading = true;
try {
await ensureConnected(connectionId);
const indices = await api.elasticsearchListIndices(connectionId);
const indices = await withMetadataLoadTimeout(connectionId, api.elasticsearchListIndices(connectionId), "Elasticsearch indices");
setChildren(
node,
withSavedSqlRoot(
@ -1374,7 +1440,7 @@ export const useConnectionStore = defineStore("connection", () => {
node.isLoading = true;
try {
await ensureConnected(connectionId);
const collections = await api.vectorListCollections(connectionId);
const collections = await withMetadataLoadTimeout(connectionId, api.vectorListCollections(connectionId), "vector collections");
setChildren(
node,
withSavedSqlRoot(

110
crates/dbx-core/src/connection.rs
View File

@ -126,6 +126,7 @@ pub struct AppState {
pub connections: Arc<RwLock<HashMap<String, PoolKind>>>,
keepalive_tasks: Arc<RwLock<HashMap<String, JoinHandle<()>>>>,
pool_activity: Arc<RwLock<HashMap<String, PoolActivity>>>,
connection_attempts: RwLock<HashMap<String, u64>>,
pub configs: RwLock<HashMap<String, ConnectionConfig>>,
pub running_queries: RunningQueries,
pub tunnels: TunnelManager,
@ -351,6 +352,7 @@ impl AppState {
connections: Arc::new(RwLock::new(HashMap::new())),
keepalive_tasks: Arc::new(RwLock::new(HashMap::new())),
pool_activity: Arc::new(RwLock::new(HashMap::new())),
connection_attempts: RwLock::new(HashMap::new()),
configs: RwLock::new(HashMap::new()),
running_queries: RunningQueries::default(),
tunnels: TunnelManager::new(),
@ -423,6 +425,48 @@ impl AppState {
}
}
pub async fn begin_connection_attempt(&self, connection_id: &str) -> u64 {
let mut attempts = self.connection_attempts.write().await;
let next = attempts.get(connection_id).copied().unwrap_or(0).wrapping_add(1);
attempts.insert(connection_id.to_string(), next);
next
}
pub async fn supersede_connection_attempt(&self, connection_id: &str) {
self.begin_connection_attempt(connection_id).await;
}
async fn connection_attempt_is_current(&self, connection_id: &str, attempt: u64) -> bool {
self.connection_attempts.read().await.get(connection_id).copied() == Some(attempt)
}
async fn ensure_current_connection_attempt(&self, connection_id: &str, attempt: Option<u64>) -> Result<(), String> {
let Some(attempt) = attempt else {
return Ok(());
};
if self.connection_attempt_is_current(connection_id, attempt).await {
Ok(())
} else {
Err("Connection attempt was superseded by a newer attempt".to_string())
}
}
pub async fn insert_connection_pool_for_attempt(
&self,
connection_id: &str,
attempt: u64,
pool_key: String,
pool: PoolKind,
config: &ConnectionConfig,
) -> Result<(), String> {
if let Err(err) = self.ensure_current_connection_attempt(connection_id, Some(attempt)).await {
close_pool_kind(pool).await;
return Err(err);
}
self.insert_connection_pool(pool_key, pool, config).await;
Ok(())
}
async fn start_keepalive_task(&self, pool_key: &str, pool: &PoolKind, config: &ConnectionConfig) {
let interval_secs = config.keepalive_interval_secs;
let idle_timeout_secs = config.idle_timeout_secs;
@ -537,11 +581,30 @@ impl AppState {
self.get_or_create_pool_for_session(connection_id, database, None).await
}
pub async fn get_or_create_pool_for_connection_attempt(
&self,
connection_id: &str,
database: Option<&str>,
attempt: u64,
) -> Result<String, String> {
self.get_or_create_pool_for_session_inner(connection_id, database, None, Some(attempt)).await
}
pub async fn get_or_create_pool_for_session(
&self,
connection_id: &str,
database: Option<&str>,
client_session_id: Option<&str>,
) -> Result<String, String> {
self.get_or_create_pool_for_session_inner(connection_id, database, client_session_id, None).await
}
async fn get_or_create_pool_for_session_inner(
&self,
connection_id: &str,
database: Option<&str>,
client_session_id: Option<&str>,
connection_attempt: Option<u64>,
) -> Result<String, String> {
let db_type = {
let configs = self.configs.read().await;
@ -693,6 +756,7 @@ impl AppState {
if self.connections.read().await.contains_key(&pool_key) {
return Ok(pool_key);
}
self.ensure_current_connection_attempt(connection_id, connection_attempt).await?;
self.insert_connection_pool(pool_key.clone(), PoolKind::MongoDb(client), &db_config)
.await;
return Ok(pool_key);
@ -924,6 +988,10 @@ impl AppState {
}
};
if let Err(err) = self.ensure_current_connection_attempt(connection_id, connection_attempt).await {
close_pool_kind(pool).await;
return Err(err);
}
self.insert_connection_pool(pool_key.clone(), pool, &db_config).await;
Ok(pool_key)
}
@ -2679,6 +2747,48 @@ mod tests {
let _ = std::fs::remove_dir_all(dir);
}
#[tokio::test]
async fn stale_connection_attempt_cannot_replace_newer_pool() {
let (state, dir) = test_app_state().await;
let mut config = mysql_config(None);
config.name = "SQLite".to_string();
config.db_type = DatabaseType::Sqlite;
config.host = dir.join("current.db").to_string_lossy().to_string();
let old_attempt = state.begin_connection_attempt("conn").await;
let new_attempt = state.begin_connection_attempt("conn").await;
let current_pool =
db::sqlite::connect_path_create_if_missing(&dir.join("current.db").to_string_lossy()).await.unwrap();
let stale_pool =
db::sqlite::connect_path_create_if_missing(&dir.join("stale.db").to_string_lossy()).await.unwrap();
state
.insert_connection_pool_for_attempt(
"conn",
new_attempt,
"conn".to_string(),
PoolKind::Sqlite(current_pool),
&config,
)
.await
.unwrap();
let result = state
.insert_connection_pool_for_attempt(
"conn",
old_attempt,
"conn".to_string(),
PoolKind::Sqlite(stale_pool),
&config,
)
.await;
assert!(result.is_err());
let conns = state.connections.read().await;
assert!(matches!(conns.get("conn"), Some(PoolKind::Sqlite(_))));
assert_eq!(conns.len(), 1);
let _ = std::fs::remove_dir_all(dir);
}
#[tokio::test]
async fn jdbc_plugin_env_uses_managed_jre_when_installed() {
let dir = std::env::temp_dir().join(format!("dbx-core-jdbc-managed-jre-{}", uuid::Uuid::new_v4()));

10
crates/dbx-web/src/routes/connection.rs
View File

@ -67,12 +67,13 @@ pub async fn connect_db(
let config = body.config;
let app = &state.app;
let connection_id = config.id.clone();
let attempt = app.begin_connection_attempt(&connection_id).await;
app.remove_connection_pools(&connection_id).await;
app.remove_connection_pools_detached(&connection_id).await;
app.reset_connection_transport_for_config(&connection_id, &config).await;
app.configs.write().await.insert(connection_id.clone(), config.clone());
app.get_or_create_pool(&connection_id, None).await.map_err(AppError)?;
app.get_or_create_pool_for_connection_attempt(&connection_id, None, attempt).await.map_err(AppError)?;
Ok(Json(connection_id))
}
@ -101,7 +102,8 @@ pub async fn disconnect_db(
) -> Result<Json<()>, AppError> {
let app = &state.app;
app.remove_connection_pools(&body.connection_id).await;
app.supersede_connection_attempt(&body.connection_id).await;
app.remove_connection_pools_detached(&body.connection_id).await;
app.nacos_registry.drop_connection(&body.connection_id).await;
#[cfg(feature = "mq-admin")]
app.mq_registry.drop_connection(&body.connection_id).await;
@ -225,7 +227,7 @@ async fn drop_mq_adapters_for_connection_ids(_state: &WebState, _connection_ids:
async fn remove_connection_pools_for_connection_ids(state: &WebState, connection_ids: &[String]) {
for connection_id in connection_ids {
state.app.remove_connection_pools(connection_id).await;
state.app.remove_connection_pools_detached(connection_id).await;
}
}

225
packages/app-tests/connectionStoreErrorState.test.ts
View File

@ -84,6 +84,50 @@ test("failed disconnect keeps the existing connection error", async () => {
}
});
test("hanging disconnect request still clears local connection state", async () => {
vi.useFakeTimers();
const restoreStorage = installMemoryStorage();
const originalFetch = globalThis.fetch;
globalThis.fetch = (async (input) => {
if (String(input) === "/api/connection/disconnect") {
return new Promise<Response>(() => {});
}
return new Response("unexpected request", { status: 500 });
}) as typeof fetch;
try {
setActivePinia(createPinia());
const store = useConnectionStore();
store.addEphemeralConnection(conn("conn-1"));
store.activeConnectionId = "conn-1";
store.recordConnectionError("conn-1", new Error("metadata failed"));
store.treeNodes.push({
id: "conn-1",
label: "conn-1",
type: "connection",
connectionId: "conn-1",
isLoading: true,
isExpanded: true,
children: [{ id: "conn-1:db", label: "db", type: "database", connectionId: "conn-1", database: "db" }],
});
const disconnectPromise = store.disconnect("conn-1");
await vi.advanceTimersByTimeAsync(5000);
await disconnectPromise;
assert.equal(store.connectionErrors["conn-1"], undefined);
assert.equal(store.connectedIds.has("conn-1"), false);
assert.equal(store.activeConnectionId, null);
assert.equal(store.treeNodes[0].isLoading, false);
assert.equal(store.treeNodes[0].isExpanded, false);
assert.deepEqual(store.treeNodes[0].children, []);
} finally {
vi.useRealTimers();
globalThis.fetch = originalFetch;
restoreStorage();
}
});
test("query errors mentioning connection do not mark the connection disconnected", async () => {
const restoreStorage = installMemoryStorage();
try {
@ -142,11 +186,20 @@ test("explicit lost-connection marker clears state without relying on error text
const store = useConnectionStore();
store.addEphemeralConnection(conn("conn-1"));
store.activeConnectionId = "conn-1";
store.treeNodes.push({
id: "conn-1",
label: "conn-1",
type: "connection",
connectionId: "conn-1",
isLoading: true,
children: [],
});
store.markConnectionLost("conn-1", new Error("连接可能已断开，请刷新数据重试"));
assert.equal(store.connectedIds.has("conn-1"), false);
assert.equal(store.activeConnectionId, null);
assert.equal(store.treeNodes[0].isLoading, false);
assert.equal(store.connectionErrors["conn-1"], "连接可能已断开，请刷新数据重试");
await new Promise((resolve) => setTimeout(resolve, 0));
} finally {
@ -189,3 +242,175 @@ test("late original connect errors replace the generic timeout detail", async ()
restoreStorage();
}
});
test("late original connect success is disconnected after the UI timeout", async () => {
vi.useFakeTimers();
const restoreStorage = installMemoryStorage();
const originalFetch = globalThis.fetch;
const disconnected: string[] = [];
globalThis.fetch = (async (input, init) => {
if (String(input) === "/api/connection/connect") {
return new Promise<Response>((resolve) => {
setTimeout(() => resolve(new Response(JSON.stringify("conn-1"), { status: 200, headers: { "Content-Type": "application/json" } })), 3500);
});
}
if (String(input) === "/api/connection/disconnect") {
const body = JSON.parse(String(init?.body ?? "{}")) as { connectionId?: string };
if (body.connectionId) disconnected.push(body.connectionId);
return new Response("null", { status: 200, headers: { "Content-Type": "application/json" } });
}
return new Response("unexpected request", { status: 500 });
}) as typeof fetch;
try {
setActivePinia(createPinia());
const store = useConnectionStore();
const config = { ...conn("conn-1"), connect_timeout_secs: 1 };
const connectPromise = store.connect(config);
const timeoutRejection = assert.rejects(() => connectPromise, /Connection attempt timed out after 3s/);
await vi.advanceTimersByTimeAsync(3000);
await timeoutRejection;
assert.equal(store.connectedIds.has("conn-1"), false);
await vi.advanceTimersByTimeAsync(500);
await Promise.resolve();
assert.deepEqual(disconnected, ["conn-1"]);
assert.equal(store.connectedIds.has("conn-1"), false);
} finally {
vi.useRealTimers();
globalThis.fetch = originalFetch;
restoreStorage();
}
});
test("late original connect success does not disconnect a newer connected state", async () => {
vi.useFakeTimers();
const restoreStorage = installMemoryStorage();
const originalFetch = globalThis.fetch;
const disconnected: string[] = [];
globalThis.fetch = (async (input, init) => {
if (String(input) === "/api/connection/connect") {
return new Promise<Response>((resolve) => {
setTimeout(() => resolve(new Response(JSON.stringify("conn-1"), { status: 200, headers: { "Content-Type": "application/json" } })), 3500);
});
}
if (String(input) === "/api/connection/disconnect") {
const body = JSON.parse(String(init?.body ?? "{}")) as { connectionId?: string };
if (body.connectionId) disconnected.push(body.connectionId);
return new Response("null", { status: 200, headers: { "Content-Type": "application/json" } });
}
return new Response("unexpected request", { status: 500 });
}) as typeof fetch;
try {
setActivePinia(createPinia());
const store = useConnectionStore();
const config = { ...conn("conn-1"), connect_timeout_secs: 1 };
const connectPromise = store.connect(config);
const timeoutRejection = assert.rejects(() => connectPromise, /Connection attempt timed out after 3s/);
await vi.advanceTimersByTimeAsync(3000);
await timeoutRejection;
store.connectedIds.add("conn-1");
await vi.advanceTimersByTimeAsync(500);
await Promise.resolve();
assert.deepEqual(disconnected, []);
assert.equal(store.connectedIds.has("conn-1"), true);
} finally {
vi.useRealTimers();
globalThis.fetch = originalFetch;
restoreStorage();
}
});
test("hanging database metadata load times out and clears loading state", async () => {
vi.useFakeTimers();
const restoreStorage = installMemoryStorage();
const originalFetch = globalThis.fetch;
globalThis.fetch = (async (input) => {
if (String(input).startsWith("/api/schema/databases?")) {
return new Promise<Response>(() => {});
}
return new Response("unexpected request", { status: 500 });
}) as typeof fetch;
try {
setActivePinia(createPinia());
const store = useConnectionStore();
store.addEphemeralConnection(conn("conn-1"));
store.activeConnectionId = "conn-1";
store.treeNodes.push({
id: "conn-1",
label: "conn-1",
type: "connection",
connectionId: "conn-1",
children: [],
});
const loadPromise = store.loadDatabases("conn-1");
const timeoutRejection = assert.rejects(() => loadPromise, /Connection timed out while loading databases after 35s/);
await vi.advanceTimersByTimeAsync(35000);
await timeoutRejection;
assert.equal(store.treeNodes[0].isLoading, false);
assert.equal(store.connectedIds.has("conn-1"), false);
assert.equal(store.activeConnectionId, null);
assert.match(store.connectionErrors["conn-1"], /Connection timed out while loading databases/);
} finally {
vi.useRealTimers();
globalThis.fetch = originalFetch;
restoreStorage();
}
});
test.each([
["redis", "loadRedisDatabases", "/api/redis/list-databases", "Redis databases"],
["mq", "loadMqTenants", "/api/mq/tenants/list", "message queue tenants"],
["mongodb", "loadMongoDatabases", "/api/mongo/list-databases", "MongoDB databases"],
["elasticsearch", "loadElasticsearchIndices", "/api/mongo/list-collections", "Elasticsearch indices"],
["qdrant", "loadVectorCollections", "/api/mongo/list-collections", "vector collections"],
] as const)("hanging %s root metadata load times out and clears loading state", async (dbType, loader, endpoint, label) => {
vi.useFakeTimers();
const restoreStorage = installMemoryStorage();
const originalFetch = globalThis.fetch;
globalThis.fetch = (async (input) => {
if (String(input) === endpoint) {
return new Promise<Response>(() => {});
}
return new Response("unexpected request", { status: 500 });
}) as typeof fetch;
try {
setActivePinia(createPinia());
const store = useConnectionStore();
store.addEphemeralConnection({ ...conn("conn-1"), db_type: dbType });
store.activeConnectionId = "conn-1";
store.treeNodes.push({
id: "conn-1",
label: "conn-1",
type: "connection",
connectionId: "conn-1",
children: [],
});
const loadPromise = store[loader]("conn-1");
const timeoutRejection = assert.rejects(() => loadPromise, new RegExp(`Connection timed out while loading ${label} after 35s`));
await vi.advanceTimersByTimeAsync(35000);
await timeoutRejection;
assert.equal(store.treeNodes[0].isLoading, false);
assert.equal(store.connectedIds.has("conn-1"), false);
assert.equal(store.activeConnectionId, null);
assert.match(store.connectionErrors["conn-1"], new RegExp(`Connection timed out while loading ${label}`));
} finally {
vi.useRealTimers();
globalThis.fetch = originalFetch;
restoreStorage();
}
});

26
packages/app-tests/sidebarSearchTree.test.ts
View File

@ -145,3 +145,29 @@ test("connection search results stay visible before connecting", () => {
["conn:1"],
);
});
test("connection search copies do not keep stale loading state", () => {
const nodes: TreeNode[] = [
{
id: "conn:1",
label: "Orders local",
type: "connection",
connectionId: "conn:1",
isLoading: true,
children: [
{
id: "conn:1:db",
label: "orders",
type: "database",
connectionId: "conn:1",
database: "orders",
},
],
},
];
const filtered = filterSidebarTree(nodes, "orders", new Set());
assert.equal(filtered[0]?.type, "connection");
assert.equal(filtered[0]?.isLoading, false);
});

20
src-tauri/src/commands/connection.rs
View File

@ -512,7 +512,7 @@ async fn drop_mq_adapters_for_connection_ids(_state: &AppState, _connection_ids:
async fn remove_connection_pools_for_connection_ids(state: &AppState, connection_ids: &[String]) {
for connection_id in connection_ids {
state.remove_connection_pools(connection_id).await;
state.remove_connection_pools_detached(connection_id).await;
}
}
@ -832,10 +832,11 @@ pub async fn connect_db(state: State<'_, Arc<AppState>>, config: ConnectionConfi
let config = config.canonicalized();
let id = config.id.clone();
let db_config = metadata_connection_config(&config);
let attempt = state.begin_connection_attempt(&id).await;
let mut connected_config = config.clone();
let mut connected_db_config = db_config.clone();
state.remove_connection_pools(&id).await;
state.remove_connection_pools_detached(&id).await;
state.reset_connection_transport_for_config(&id, &db_config).await;
let (host, port) = state.connection_host_port(&id, &db_config).await?;
@ -922,8 +923,16 @@ pub async fn connect_db(state: State<'_, Arc<AppState>>, config: ConnectionConfi
.await
{
Ok(()) => {
state
.insert_connection_pool_for_attempt(
&id,
attempt,
id.clone(),
PoolKind::MongoDb(client),
&db_config,
)
.await?;
state.configs.write().await.insert(id.clone(), config);
state.insert_connection_pool(id.clone(), PoolKind::MongoDb(client), &db_config).await;
return Ok(id);
}
Err(e) => e,
@ -1085,7 +1094,7 @@ pub async fn connect_db(state: State<'_, Arc<AppState>>, config: ConnectionConfi
db_type => return Err(format!("Unsupported database type: {db_type:?}")),
};
state.insert_connection_pool(id.clone(), pool, &connected_db_config).await;
state.insert_connection_pool_for_attempt(&id, attempt, id.clone(), pool, &connected_db_config).await?;
state.configs.write().await.insert(id.clone(), connected_config);
Ok(id)
@ -1111,7 +1120,8 @@ pub async fn connection_final_proxy_port(
#[tauri::command]
pub async fn disconnect_db(state: State<'_, Arc<AppState>>, connection_id: String) -> Result<(), String> {
state.remove_connection_pools(&connection_id).await;
state.supersede_connection_attempt(&connection_id).await;
state.remove_connection_pools_detached(&connection_id).await;
drop_nacos_adapters_for_connection_ids(state.inner(), std::slice::from_ref(&connection_id)).await;
drop_mq_adapters_for_connection_ids(state.inner(), std::slice::from_ref(&connection_id)).await;
state.reset_connection_transport(&connection_id).await;