fix(mcp): refresh desktop connection list after MCP mutations

This commit is contained in:
ezreal
2026-09-29 01:18:36 +08:00
committed by GitHub
parent 914a97326c
commit 0be8a38984
8 changed files with 223 additions and 10 deletions
@@ -150,6 +150,7 @@ function createStore() {
moveConnectionToGroup: vi.fn(),
createConnectionGroup: vi.fn(() => "group-new"),
refreshAllTree: vi.fn().mockResolvedValue(undefined),
reloadFromDisk: vi.fn().mockResolvedValue(undefined),
removeConnections: vi.fn().mockResolvedValue(undefined),
disconnect: vi.fn().mockResolvedValue(undefined),
});
@@ -196,6 +197,31 @@ describe("AppSidebar connection multi-select moves", () => {
}
});
it("waits for persisted connections before refreshing the tree", async () => {
let finishReload!: () => void;
mocks.store.reloadFromDisk.mockReturnValue(
new Promise<void>((resolve) => {
finishReload = resolve;
}),
);
const host = await mountSidebar();
click(host.querySelector('[data-tooltip="contextMenu.refreshChildren"] button'));
expect(mocks.store.reloadFromDisk).toHaveBeenCalledOnce();
expect(mocks.store.refreshAllTree).not.toHaveBeenCalled();
finishReload();
await nextTick();
expect(mocks.store.refreshAllTree).toHaveBeenCalledOnce();
});
it("reports a failed connection reload without refreshing stale nodes", async () => {
mocks.store.reloadFromDisk.mockRejectedValue(new Error("storage unavailable"));
const host = await mountSidebar();
click(host.querySelector('[data-tooltip="contextMenu.refreshChildren"] button'));
await nextTick();
expect(mocks.store.refreshAllTree).not.toHaveBeenCalled();
expect(mocks.toast).toHaveBeenCalledOnce();
});
it("releases an existing-group batch including filtered connections before the next batch", async () => {
setSelection(["conn-visible", "conn-hidden"]);
const host = await mountSidebar();
@@ -87,6 +87,7 @@ const moveGroupItems = computed(() => [
async function refreshTree() {
try {
await connectionStore.reloadFromDisk();
await connectionStore.refreshAllTree();
} catch (e: any) {
toast(t("connection.connectFailed", { message: translateBackendError(t, e) }), 5000);
@@ -2,6 +2,7 @@ import { beforeEach, describe, expect, it, vi } from "vitest";
const listeners = new Map<string, (event: { payload: unknown }) => void>();
const unlisten = vi.fn();
const reloadFromDisk = vi.fn();
let listenerRegistrationBarrier: Promise<void> | null = null;
const listen = vi.fn(async (event: string, handler: (event: { payload: unknown }) => void) => {
await listenerRegistrationBarrier;
@@ -14,7 +15,7 @@ vi.mock("@tauri-apps/api/event", () => ({
}));
vi.mock("@/stores/connectionStore", () => ({
useConnectionStore: () => ({ connections: [], initFromDisk: vi.fn(), getConfig: vi.fn(), ensureConnected: vi.fn() }),
useConnectionStore: () => ({ connections: [], initFromDisk: vi.fn(), reloadFromDisk, getConfig: vi.fn(), ensureConnected: vi.fn() }),
}));
vi.mock("@/stores/queryStore", () => ({
@@ -28,6 +29,7 @@ describe("useTauriEvents", () => {
listeners.clear();
listen.mockClear();
unlisten.mockClear();
reloadFromDisk.mockClear();
listenerRegistrationBarrier = null;
});
@@ -60,6 +62,8 @@ describe("useTauriEvents", () => {
expect(setupResolved).toBe(true);
expect(listeners.has("dbx-open-connection-links")).toBe(true);
await listeners.get("mcp-reload-connections")!({ payload: undefined });
expect(reloadFromDisk).toHaveBeenCalledOnce();
});
it("routes the native macOS close-tab menu event to the active surface", async () => {
@@ -102,7 +102,7 @@ export function useTauriEvents(deps: {
track(
listen("mcp-reload-connections", async () => {
try {
await connectionStore.initFromDisk();
await connectionStore.reloadFromDisk();
} catch (e) {
console.error("[DBX] mcp-reload-connections error:", e);
}
@@ -117,6 +117,49 @@ describe("connectionStore one_time runtime cleanup", () => {
expect(store.connections).toEqual([]);
});
it("reloads externally added, copied and deleted connections while preserving existing tree state", async () => {
installApiMocks();
const { loadConnections } = await import("@/lib/backend/api");
const { useConnectionStore } = await import("@/stores/connectionStore");
const store = useConnectionStore();
const kept = previewConnection({ id: "kept", one_time: false });
const temporary = previewConnection({ id: "temporary" });
store.connections = [kept, previewConnection({ id: "removed", one_time: false }), temporary];
store.connectedIds.add("kept");
store.treeNodes = [{ id: "kept", label: kept.name, type: "connection", connectionId: "kept", isExpanded: true, children: [{ id: "kept:app", label: "app", type: "database", connectionId: "kept", database: "app", isExpanded: true }] }];
vi.mocked(loadConnections).mockResolvedValue([kept, previewConnection({ id: "added", one_time: false }), previewConnection({ id: "copied", one_time: false })]);
await store.reloadFromDisk();
expect(store.connections.map((connection) => connection.id)).toEqual(["kept", "added", "copied", "temporary"]);
expect(store.treeNodes.map((node) => node.id)).toEqual(["kept", "added", "copied", "temporary"]);
expect(store.treeNodes[0]).toMatchObject({ isExpanded: true, children: expect.arrayContaining([expect.objectContaining({ id: "kept:app", isExpanded: true })]) });
expect(store.connectedIds.has("kept")).toBe(true);
});
it("reads a fresh snapshot when an external change arrives during an in-flight reload", async () => {
installApiMocks();
const { loadConnections } = await import("@/lib/backend/api");
let finishFirstRead!: (connections: ConnectionConfig[]) => void;
vi.mocked(loadConnections)
.mockReturnValueOnce(
new Promise((resolve) => {
finishFirstRead = resolve;
}),
)
.mockResolvedValueOnce([previewConnection({ id: "new", one_time: false })]);
const { useConnectionStore } = await import("@/stores/connectionStore");
const store = useConnectionStore();
const initialLoad = store.initFromDisk();
await vi.waitFor(() => expect(loadConnections).toHaveBeenCalledOnce());
const externalReload = store.reloadFromDisk();
finishFirstRead([]);
await Promise.all([initialLoad, externalReload]);
expect(loadConnections).toHaveBeenCalledTimes(2);
expect(store.getConfig("new")).toBeDefined();
});
it("removeConnection leaves saved connections to the save_connections sync", async () => {
installApiMocks();
const { useConnectionStore } = await import("@/stores/connectionStore");
@@ -9995,6 +9995,13 @@ export const useConnectionStore = defineStore("connection", () => {
updateLayoutAndRebuild(reconciledLayout);
}
async function reloadFromDisk() {
// An external change may arrive after an in-flight reload read its snapshot.
// Wait for it, then read again instead of joining that stale snapshot.
await initFromDiskPromise;
await initFromDisk();
}
async function initFromDisk() {
// Connection normalization and timeout migration depend on persisted global
// settings. Startup helpers may initialize connections before App.initApp().
@@ -10067,6 +10074,7 @@ export const useConnectionStore = defineStore("connection", () => {
replacePinnedTreeNode,
removeTreeNode,
refreshAllTree,
reloadFromDisk,
collapseAllTreeNodes,
refreshSidebarObjectPagination,
refreshTreeNode,
+9 -8
View File
@@ -430,7 +430,7 @@ mod tests {
options.null_literal = dbx_formats::csv_export::DEFAULT_CSV_NULL_LITERAL.to_string();
let mut pages = vec![result_with_a_null_and_an_empty_string()].into_iter();
let mut output = Vec::new();
write_table_csv_pages(&mut output, DatabaseType::Postgres, &options, None, |_, _| {
write_table_csv_pages(&mut output, DatabaseType::Postgres, None, &options, None, |_, _| {
std::future::ready(Ok(pages.next().expect("single page")))
})
.await
@@ -441,7 +441,7 @@ mod tests {
options.null_literal = String::new();
let mut pages = vec![result_with_a_null_and_an_empty_string()].into_iter();
let mut output = Vec::new();
write_table_csv_pages(&mut output, DatabaseType::Postgres, &options, None, |_, _| {
write_table_csv_pages(&mut output, DatabaseType::Postgres, None, &options, None, |_, _| {
std::future::ready(Ok(pages.next().expect("single page")))
})
.await
@@ -479,12 +479,13 @@ mod tests {
options.null_literal = dbx_formats::csv_export::DEFAULT_CSV_NULL_LITERAL.to_string();
let mut exported_bytes = "\u{FEFF}".as_bytes().to_vec();
let exported = write_table_csv_pages(&mut exported_bytes, DatabaseType::Sqlite, &options, None, |sql, _| {
let sqlite = sqlite.clone();
async move { crate::db::sqlite::execute_query(&sqlite, &sql).await.map_err(|error| error.to_string()) }
})
.await
.unwrap();
let exported =
write_table_csv_pages(&mut exported_bytes, DatabaseType::Sqlite, None, &options, None, |sql, _| {
let sqlite = sqlite.clone();
async move { crate::db::sqlite::execute_query(&sqlite, &sql).await.map_err(|error| error.to_string()) }
})
.await
.unwrap();
assert_eq!(exported, 2);
// 导出的 NULL 与空串必须已经可区分:空串 `""`、NULL 裸写 `\N`
assert_eq!(
+130
View File
@@ -682,6 +682,16 @@ fn plugin_lacks_mcp_surface(err: &str) -> bool {
const PLUGIN_TOOLS_PROBE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(8);
impl LocalBackend {
async fn notify_connections_changed(&self) {
// Persistence has already succeeded. A closed or unresponsive desktop
// must not turn the mutation into an error (or block headless MCP).
let _ = tokio::time::timeout(
std::time::Duration::from_secs(2),
self.bridge_request("/reload-connections", json!({})),
)
.await;
}
fn spawn_connection_lifecycle_watcher(
&self,
lifecycle: ConnectionLifecycleSnapshot,
@@ -1242,6 +1252,7 @@ impl DbxBackend for LocalBackend {
async fn add_connection_for_mcp(&self, config: ConnectionConfig) -> Result<ConnectionConfig, String> {
let config = self.state.storage.add_connection_for_mcp(config).await?;
self.state.configs.write().await.insert(config.id.clone(), config.clone());
self.notify_connections_changed().await;
Ok(config)
}
@@ -1253,6 +1264,7 @@ impl DbxBackend for LocalBackend {
) -> Result<ConnectionConfig, String> {
let config = self.state.storage.duplicate_connection_for_mcp(source_id, copy_id, copy_name).await?;
self.state.configs.write().await.insert(config.id.clone(), config.clone());
self.notify_connections_changed().await;
Ok(config)
}
@@ -1262,6 +1274,7 @@ impl DbxBackend for LocalBackend {
self.state.configs.write().await.remove(connection_id);
self.transaction_owners.invalidate_connection(connection_id);
self.state.remove_connection_pools_detached(connection_id).await;
self.notify_connections_changed().await;
}
Ok(removed)
}
@@ -2809,6 +2822,123 @@ mod tests {
time::Duration,
};
async fn connection_mutation_backend(data_dir: &Path) -> LocalBackend {
let storage = Storage::open(&data_dir.join("dbx.db"))
.await
.unwrap()
.with_secret_key_policy(dbx_core::persistence::secret_codec::SecretKeyPolicy::ManagedDataDir);
storage.save_mcp_global_policy(&McpGlobalPolicy { read_only: false, ..Default::default() }).await.unwrap();
LocalBackend::from_app_state(Arc::new(AppState::new(storage)), data_dir.to_path_buf())
}
fn mutation_test_connection() -> ConnectionConfig {
new_connection_config(
"source".into(),
"Source".into(),
DatabaseType::Postgres,
"127.0.0.1".into(),
5432,
"postgres".into(),
"test-password".into(),
None,
false,
None,
)
.unwrap()
}
#[tokio::test]
async fn local_connection_mutations_notify_desktop_only_after_success() {
let data_dir = tempfile::tempdir().unwrap();
let backend = connection_mutation_backend(data_dir.path()).await;
let (sent, mut received) = tokio::sync::mpsc::unbounded_channel();
let storage = backend.state.storage.clone();
let app = axum::Router::new().route(
"/reload-connections",
axum::routing::post(move || {
let sent = sent.clone();
let storage = storage.clone();
async move {
// The desktop must observe the committed state, not the old list.
let ids: Vec<_> = storage.load_connections().await.unwrap().into_iter().map(|c| c.id).collect();
sent.send(ids).unwrap();
axum::http::StatusCode::OK
}
}),
);
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
tokio::fs::write(data_dir.path().join("mcp-bridge-port"), listener.local_addr().unwrap().port().to_string())
.await
.unwrap();
let server = tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
backend.add_connection_for_mcp(mutation_test_connection()).await.unwrap();
assert_eq!(received.try_recv().unwrap(), vec!["source"]);
backend.duplicate_connection_for_mcp("source", "copy", "Copy").await.unwrap();
let mut ids = received.try_recv().unwrap();
ids.sort();
assert_eq!(ids, vec!["copy", "source"]);
assert!(backend.remove_connection_for_mcp("copy").await.unwrap());
assert_eq!(received.try_recv().unwrap(), vec!["source"]);
assert!(!backend.remove_connection_for_mcp("missing").await.unwrap());
assert!(backend.duplicate_connection_for_mcp("missing", "other", "Other").await.is_err());
backend
.state
.storage
.save_mcp_global_policy(&McpGlobalPolicy { read_only: true, ..Default::default() })
.await
.unwrap();
assert!(backend.add_connection_for_mcp(mutation_test_connection()).await.is_err());
assert!(backend.remove_connection_for_mcp("source").await.is_err());
assert!(received.try_recv().is_err());
server.abort();
}
#[tokio::test]
async fn local_connection_mutations_succeed_without_desktop() {
let data_dir = tempfile::tempdir().unwrap();
let backend = connection_mutation_backend(data_dir.path()).await;
backend.add_connection_for_mcp(mutation_test_connection()).await.unwrap();
backend.duplicate_connection_for_mcp("source", "copy", "Copy").await.unwrap();
assert!(backend.remove_connection_for_mcp("source").await.unwrap());
let connections = backend.load_connections().await.unwrap();
assert_eq!(connections.len(), 1);
assert_eq!(connections[0].id, "copy");
assert_eq!(connections[0].password, "test-password");
}
#[tokio::test]
async fn local_connection_mutations_succeed_when_desktop_fails_or_stalls() {
for stall in [false, true] {
let data_dir = tempfile::tempdir().unwrap();
let backend = connection_mutation_backend(data_dir.path()).await;
let app = axum::Router::new().route(
"/reload-connections",
axum::routing::post(move || async move {
if stall {
std::future::pending::<()>().await;
}
axum::http::StatusCode::INTERNAL_SERVER_ERROR
}),
);
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
tokio::fs::write(
data_dir.path().join("mcp-bridge-port"),
listener.local_addr().unwrap().port().to_string(),
)
.await
.unwrap();
let server = tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
tokio::time::timeout(Duration::from_secs(5), backend.add_connection_for_mcp(mutation_test_connection()))
.await
.expect("desktop notification must be bounded")
.unwrap();
assert_eq!(backend.load_connections().await.unwrap().len(), 1);
server.abort();
}
}
fn policy_state(configured: bool, read_only: bool) -> McpGlobalPolicyState {
McpGlobalPolicyState {
configured,