From 7ab781ce191107cbefe2ebe4a40a317df23e7378 Mon Sep 17 00:00:00 2001 From: onenewcode Date: Wed, 16 Sep 2026 13:20:08 +0800 Subject: [PATCH] feat(mongodb): stream legacy agent exports via find cursors --- .../java/com/dbx/agent/AgentProtocol.java | 14 +- .../src/main/resources/agent-protocol-v1.json | 4 +- .../src/main/resources/agent-protocol-v2.json | 4 +- .../com/dbx/agent/mongodb/MongoAgent.java | 200 +++- .../com/dbx/agent/mongodb/MongoAgentTest.java | 446 +++++++ .../components/document/MongoImportDialog.vue | 3 +- .../src/i18n/__tests__/backendErrors.spec.ts | 42 + apps/desktop/src/i18n/backend-errors.ts | 72 +- apps/desktop/src/i18n/locales/az.ts | 8 + apps/desktop/src/i18n/locales/en.ts | 8 + apps/desktop/src/i18n/locales/es.ts | 8 + apps/desktop/src/i18n/locales/it.ts | 8 + apps/desktop/src/i18n/locales/ja.ts | 8 + apps/desktop/src/i18n/locales/ko.ts | 8 + apps/desktop/src/i18n/locales/pt-BR.ts | 8 + apps/desktop/src/i18n/locales/tr.ts | 8 + apps/desktop/src/i18n/locales/zh-CN.ts | 8 + apps/desktop/src/i18n/locales/zh-TW.ts | 8 + apps/desktop/src/lib/backend/errorUtils.ts | 8 +- crates/dbx-core/assets/agent-protocol-v1.json | 4 +- crates/dbx-core/assets/agent-protocol-v2.json | 4 +- crates/dbx-core/src/db/agent_driver.rs | 55 +- crates/dbx-core/src/db/mongo_driver.rs | 35 + crates/dbx-core/src/mongo_ops.rs | 65 +- crates/dbx-core/src/mongodb_import_export.rs | 1062 +++++++++++++++-- crates/dbx-core/src/transfer.rs | 4 +- 26 files changed, 1942 insertions(+), 160 deletions(-) diff --git a/agents/common/src/main/java/com/dbx/agent/AgentProtocol.java b/agents/common/src/main/java/com/dbx/agent/AgentProtocol.java index d3fceaf6da..474fee0935 100644 --- a/agents/common/src/main/java/com/dbx/agent/AgentProtocol.java +++ b/agents/common/src/main/java/com/dbx/agent/AgentProtocol.java @@ -68,6 +68,9 @@ public final class AgentProtocol { public static final String MONGO_METHOD_DROP_DATABASE = "drop_database"; public static final String MONGO_METHOD_INSERT_DOCUMENT = "insert_document"; public static final String MONGO_METHOD_INSERT_DOCUMENTS = "insert_documents"; + public static final String MONGO_METHOD_START_FIND_CURSOR = "start_find_cursor"; + public static final String MONGO_METHOD_FETCH_FIND_CURSOR = "fetch_find_cursor"; + public static final String MONGO_METHOD_CLOSE_FIND_CURSOR = "close_find_cursor"; public static final String MONGO_METHOD_UPDATE_DOCUMENT = "update_document"; public static final String MONGO_METHOD_UPDATE_DOCUMENTS = "update_documents"; public static final String MONGO_METHOD_REPLACE_DOCUMENT = "replace_document"; @@ -129,6 +132,7 @@ public final class AgentProtocol { public static final String CAPABILITY_MONGO_RUN_COMMAND = "mongo_run_command"; public static final String CAPABILITY_MONGO_INSERT_DOCUMENTS = "mongo_insert_documents"; public static final String CAPABILITY_MONGO_REPLACE_DOCUMENT = "mongo_replace_document"; + public static final String CAPABILITY_MONGO_FIND_CURSOR = "mongo_find_cursor"; public static final String CAPABILITY_MULTI_SESSION = "multi_session"; public static final String CAPABILITY_STRUCTURED_ERROR_V1 = "structured_error_v1"; @@ -165,7 +169,8 @@ public final class AgentProtocol { CAPABILITY_MONGO_CLONE_COLLECTION, CAPABILITY_MONGO_RUN_COMMAND, CAPABILITY_MONGO_INSERT_DOCUMENTS, - CAPABILITY_MONGO_REPLACE_DOCUMENT + CAPABILITY_MONGO_REPLACE_DOCUMENT, + CAPABILITY_MONGO_FIND_CURSOR )); public static final List MULTI_SESSION_CAPABILITIES; @@ -230,6 +235,7 @@ public final class AgentProtocol { mongoCapabilities.add(CAPABILITY_MONGO_RUN_COMMAND); mongoCapabilities.add(CAPABILITY_MONGO_INSERT_DOCUMENTS); mongoCapabilities.add(CAPABILITY_MONGO_REPLACE_DOCUMENT); + mongoCapabilities.add(CAPABILITY_MONGO_FIND_CURSOR); MONGO_LEGACY_CAPABILITIES = Collections.unmodifiableList(mongoCapabilities); List mongoMultiSessionCapabilities = new java.util.ArrayList<>(MULTI_SESSION_CAPABILITIES); @@ -238,6 +244,7 @@ public final class AgentProtocol { mongoMultiSessionCapabilities.add(CAPABILITY_MONGO_RUN_COMMAND); mongoMultiSessionCapabilities.add(CAPABILITY_MONGO_INSERT_DOCUMENTS); mongoMultiSessionCapabilities.add(CAPABILITY_MONGO_REPLACE_DOCUMENT); + mongoMultiSessionCapabilities.add(CAPABILITY_MONGO_FIND_CURSOR); MONGO_LEGACY_MULTI_SESSION_CAPABILITIES = Collections.unmodifiableList(mongoMultiSessionCapabilities); List jdbcCapabilities = new java.util.ArrayList<>(MULTI_SESSION_CAPABILITIES); @@ -282,7 +289,10 @@ public final class AgentProtocol { MONGO_METHOD_REPLACE_DOCUMENT, MONGO_METHOD_DELETE_DOCUMENT, MONGO_METHOD_DELETE_DOCUMENTS, - MONGO_METHOD_RUN_COMMAND + MONGO_METHOD_RUN_COMMAND, + MONGO_METHOD_START_FIND_CURSOR, + MONGO_METHOD_FETCH_FIND_CURSOR, + MONGO_METHOD_CLOSE_FIND_CURSOR )); public static final List KV_METHODS = Collections.unmodifiableList(Arrays.asList( diff --git a/agents/common/src/main/resources/agent-protocol-v1.json b/agents/common/src/main/resources/agent-protocol-v1.json index 716e9eb035..abb7ebbc41 100644 --- a/agents/common/src/main/resources/agent-protocol-v1.json +++ b/agents/common/src/main/resources/agent-protocol-v1.json @@ -2,7 +2,7 @@ "protocolVersion": 1, "handshakeMethod": "handshake", "handshakeResponseFields": ["protocolVersion", "agentProtocolVersion", "capabilities"], - "allCapabilities": ["connect", "test_connection", "metadata", "query", "paged_query", "transaction", "ddl", "kv", "kv_ttl", "kv_cas", "kv_list_values", "kv_status", "kv_history", "etcd_compaction", "etcd_defrag", "etcd_watch", "etcd_lease", "etcd_auth", "mongo_drop_database", "mongo_clone_collection", "mongo_run_command", "mongo_insert_documents", "mongo_replace_document"], + "allCapabilities": ["connect", "test_connection", "metadata", "query", "paged_query", "transaction", "ddl", "kv", "kv_ttl", "kv_cas", "kv_list_values", "kv_status", "kv_history", "etcd_compaction", "etcd_defrag", "etcd_watch", "etcd_lease", "etcd_auth", "mongo_drop_database", "mongo_clone_collection", "mongo_run_command", "mongo_insert_documents", "mongo_replace_document", "mongo_find_cursor"], "capabilities": ["connect", "test_connection", "metadata", "query", "paged_query", "transaction", "ddl"], "defaultSqlCapabilities": ["connect", "test_connection", "metadata", "query", "paged_query", "transaction", "ddl"], "commonMethods": [ @@ -42,6 +42,6 @@ "disconnect", "shutdown" ], - "mongoLegacyMethods": ["list_databases", "list_collections", "find_documents", "find_one", "explain_find", "aggregate_documents", "find_documents_extended_json", "count_documents", "server_version", "create_index", "create_user", "drop_indexes", "drop_collection", "clone_collection", "drop_database", "insert_document", "insert_documents", "update_document", "update_documents", "replace_document", "delete_document", "delete_documents", "run_command"], + "mongoLegacyMethods": ["list_databases", "list_collections", "find_documents", "find_one", "explain_find", "aggregate_documents", "find_documents_extended_json", "count_documents", "server_version", "create_index", "create_user", "drop_indexes", "drop_collection", "clone_collection", "drop_database", "insert_document", "insert_documents", "update_document", "update_documents", "replace_document", "delete_document", "delete_documents", "run_command", "start_find_cursor", "fetch_find_cursor", "close_find_cursor"], "kvMethods": ["kv_list_prefix", "kv_get", "kv_put", "kv_delete", "kv_rename", "kv_history", "kv_status", "etcd_compact", "etcd_defrag", "etcd_watch_start", "etcd_watch_poll", "etcd_watch_stop", "etcd_lease_list", "etcd_lease_get", "etcd_lease_grant", "etcd_lease_keepalive_once", "etcd_lease_revoke", "etcd_auth_user_list", "etcd_auth_user_get", "etcd_auth_user_add", "etcd_auth_user_delete", "etcd_auth_user_change_password", "etcd_auth_user_grant_role", "etcd_auth_user_revoke_role", "etcd_auth_role_list", "etcd_auth_role_get", "etcd_auth_role_add", "etcd_auth_role_delete", "etcd_auth_role_grant_permission", "etcd_auth_role_revoke_permission"] } diff --git a/agents/common/src/main/resources/agent-protocol-v2.json b/agents/common/src/main/resources/agent-protocol-v2.json index 1b1a2017a9..c1ef26af18 100644 --- a/agents/common/src/main/resources/agent-protocol-v2.json +++ b/agents/common/src/main/resources/agent-protocol-v2.json @@ -2,7 +2,7 @@ "protocolVersion": 2, "handshakeMethod": "handshake", "handshakeResponseFields": ["protocolVersion", "agentProtocolVersion", "capabilities"], - "allCapabilities": ["connect", "test_connection", "metadata", "query", "paged_query", "transaction", "ddl", "kv", "kv_ttl", "kv_cas", "kv_list_values", "kv_status", "kv_history", "etcd_compaction", "etcd_defrag", "etcd_watch", "etcd_lease", "etcd_auth", "mongo_drop_database", "mongo_clone_collection", "mongo_run_command", "mongo_insert_documents", "mongo_replace_document", "multi_session", "structured_error_v1"], + "allCapabilities": ["connect", "test_connection", "metadata", "query", "paged_query", "transaction", "ddl", "kv", "kv_ttl", "kv_cas", "kv_list_values", "kv_status", "kv_history", "etcd_compaction", "etcd_defrag", "etcd_watch", "etcd_lease", "etcd_auth", "mongo_drop_database", "mongo_clone_collection", "mongo_run_command", "mongo_insert_documents", "mongo_replace_document", "mongo_find_cursor", "multi_session", "structured_error_v1"], "capabilities": ["connect", "test_connection", "metadata", "query", "paged_query", "transaction", "ddl", "multi_session"], "defaultSqlCapabilities": ["connect", "test_connection", "metadata", "query", "paged_query", "transaction", "ddl", "multi_session", "structured_error_v1"], "commonMethods": [ @@ -46,7 +46,7 @@ "disconnect", "shutdown" ], - "mongoLegacyMethods": ["list_databases", "list_collections", "find_documents", "find_one", "explain_find", "aggregate_documents", "find_documents_extended_json", "count_documents", "server_version", "create_index", "create_user", "drop_indexes", "drop_collection", "clone_collection", "drop_database", "insert_document", "insert_documents", "update_document", "update_documents", "replace_document", "delete_document", "delete_documents", "run_command"], + "mongoLegacyMethods": ["list_databases", "list_collections", "find_documents", "find_one", "explain_find", "aggregate_documents", "find_documents_extended_json", "count_documents", "server_version", "create_index", "create_user", "drop_indexes", "drop_collection", "clone_collection", "drop_database", "insert_document", "insert_documents", "update_document", "update_documents", "replace_document", "delete_document", "delete_documents", "run_command", "start_find_cursor", "fetch_find_cursor", "close_find_cursor"], "kvMethods": ["kv_list_prefix", "kv_get", "kv_put", "kv_delete", "kv_rename", "kv_history", "kv_status", "etcd_compact", "etcd_defrag", "etcd_watch_start", "etcd_watch_poll", "etcd_watch_stop", "etcd_lease_list", "etcd_lease_get", "etcd_lease_grant", "etcd_lease_keepalive_once", "etcd_lease_revoke", "etcd_auth_user_list", "etcd_auth_user_get", "etcd_auth_user_add", "etcd_auth_user_delete", "etcd_auth_user_change_password", "etcd_auth_user_grant_role", "etcd_auth_user_revoke_role", "etcd_auth_role_list", "etcd_auth_role_get", "etcd_auth_role_add", "etcd_auth_role_delete", "etcd_auth_role_grant_permission", "etcd_auth_role_revoke_permission"], "sessionField": "agentSessionId", "cursorSessionField": "sessionId" diff --git a/agents/drivers/mongodb/src/main/java/com/dbx/agent/mongodb/MongoAgent.java b/agents/drivers/mongodb/src/main/java/com/dbx/agent/mongodb/MongoAgent.java index 4f380c33ad..3f6f1b94bc 100644 --- a/agents/drivers/mongodb/src/main/java/com/dbx/agent/mongodb/MongoAgent.java +++ b/agents/drivers/mongodb/src/main/java/com/dbx/agent/mongodb/MongoAgent.java @@ -10,10 +10,14 @@ import com.google.gson.JsonNull; import com.google.gson.JsonObject; import com.google.gson.JsonParser; import com.mongodb.ConnectionString; +import com.mongodb.MongoBulkWriteException; import com.mongodb.MongoCredential; import com.mongodb.MongoClientSettings; import com.mongodb.ServerAddress; +import com.mongodb.bulk.BulkWriteError; +import com.mongodb.bulk.WriteConcernError; import com.mongodb.client.AggregateIterable; +import com.mongodb.client.FindIterable; import com.mongodb.client.MongoClient; import com.mongodb.client.MongoClients; import com.mongodb.client.MongoCollection; @@ -58,6 +62,7 @@ import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.Set; +import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; @@ -81,11 +86,22 @@ public final class MongoAgent { private static final JsonWriterSettings EXTENDED_JSON_SETTINGS = JsonWriterSettings.builder() .outputMode(JsonMode.RELAXED) .build(); + // Driver 3.x names the canonical Extended JSON v2 output "EXTENDED"; RELAXED is its lossy + // sibling (plain numbers, ISO dates), so exports must use this one to keep BSON types. + private static final JsonWriterSettings CANONICAL_JSON_SETTINGS = JsonWriterSettings.builder() + .outputMode(JsonMode.EXTENDED) + .build(); private static final String LEGACY_SESSION_ID = "__legacy__"; private static final String DEFAULT_ID_INDEX_NAME = "_id_"; private static final int CLONE_INSERT_BATCH_SIZE = 1_000; + private static final int DEFAULT_FIND_CURSOR_BATCH_SIZE = 1_000; + private static final int MAX_FIND_CURSORS_PER_OWNER = 16; + private static final int MAX_FIND_CURSORS_TOTAL = 64; private static final int MAX_SESSIONS = 256; private static final ThreadLocal CURRENT_CLIENT = new ThreadLocal<>(); + private static final ThreadLocal> CURRENT_FIND_CURSOR_OWNER = new ThreadLocal<>(); + private static final ConcurrentHashMap> FIND_CURSORS = new ConcurrentHashMap<>(); + private static final Set LEGACY_FIND_CURSOR_IDS = ConcurrentHashMap.newKeySet(); private static MongoClient legacyClient; private MongoAgent() { @@ -1489,8 +1505,174 @@ public final class MongoAgent { MongoClient client = requireClient(); String database = params.get("database").getAsString(); String collection = params.get("collection").getAsString(); - client.getDatabase(database).getCollection(collection).insertMany(documents); - return Collections.singletonMap("affected_rows", documents.size()); + InsertManyOptions options = new InsertManyOptions(); + if (params.has("ordered") && !params.get("ordered").isJsonNull()) { + JsonElement ordered = params.get("ordered"); + if (!ordered.isJsonPrimitive() || !ordered.getAsJsonPrimitive().isBoolean()) { + throw new IllegalArgumentException("ordered must be a boolean"); + } + options.ordered(ordered.getAsBoolean()); + } + try { + client.getDatabase(database).getCollection(collection).insertMany(documents, options); + return Collections.singletonMap("affected_rows", documents.size()); + } catch (MongoBulkWriteException error) { + return bulkInsertResult(error); + } + } + + static Map bulkInsertResult(MongoBulkWriteException error) { + Map result = new LinkedHashMap<>(); + result.put("affected_rows", error.getWriteResult().getInsertedCount()); + List> errors = new ArrayList<>(); + for (BulkWriteError writeError : error.getWriteErrors()) { + Map item = new LinkedHashMap<>(); + item.put("index", writeError.getIndex()); + item.put("code", writeError.getCode()); + item.put("message", writeError.getMessage()); + errors.add(item); + } + // The driver raises the same exception for write concern failures only (for example a + // wtimeout): there are no per-document errors then, but the batch must not look + // successful, so report the write concern error as a batch-wide rejection without index. + WriteConcernError writeConcernError = error.getWriteConcernError(); + if (errors.isEmpty() && writeConcernError != null) { + Map item = new LinkedHashMap<>(); + item.put("code", writeConcernError.getCode()); + item.put("message", writeConcernError.getMessage()); + errors.add(item); + } + result.put("errors", errors); + return result; + } + + private static Object startFindCursor(JsonObject params) { + Set owner = CURRENT_FIND_CURSOR_OWNER.get(); + if (owner != null && owner.size() >= MAX_FIND_CURSORS_PER_OWNER) { + throw new IllegalStateException("MongoDB find cursor limit reached: " + MAX_FIND_CURSORS_PER_OWNER); + } + // The per-owner budget keeps one session from starving the others; the process-wide cap + // only bounds how many server cursors this agent can hold in total. + if (FIND_CURSORS.size() >= MAX_FIND_CURSORS_TOTAL) { + throw new IllegalStateException("MongoDB find cursor limit reached: " + MAX_FIND_CURSORS_TOTAL); + } + MongoClient client = requireClient(); + String database = params.get("database").getAsString(); + String collection = params.get("collection").getAsString(); + Document filterDoc = documentOrNull(params, "filter"); + Document projectionDoc = documentOrNull(params, "projection"); + Document sortDoc = documentOrNull(params, "sort"); + Collation collation = collationOrNull(documentOrNull(params, "collation")); + int batchSize = findCursorBatchSize(params); + + FindIterable iterable = client.getDatabase(database) + .getCollection(collection) + .find(filterDoc == null ? new Document() : filterDoc) + .batchSize(batchSize); + if (projectionDoc != null) { + iterable = iterable.projection(projectionDoc); + } + if (sortDoc != null) { + iterable = iterable.sort(sortDoc); + } + if (collation != null) { + iterable = iterable.collation(collation); + } + + String cursorId = UUID.randomUUID().toString(); + FIND_CURSORS.put(cursorId, iterable.iterator()); + if (owner != null) { + owner.add(cursorId); + } + Map result = new LinkedHashMap<>(); + result.put("cursor_id", cursorId); + result.put("batch_size", batchSize); + return result; + } + + private static Object fetchFindCursor(JsonObject params) { + String cursorId = requiredCursorId(params); + MongoCursor cursor = FIND_CURSORS.get(cursorId); + if (cursor == null) { + throw new IllegalStateException("Find cursor not found"); + } + int limit = params.has("limit") ? params.get("limit").getAsInt() : DEFAULT_FIND_CURSOR_BATCH_SIZE; + if (limit <= 0) { + throw new IllegalArgumentException("Find cursor limit must be a positive integer"); + } + List documents = new ArrayList<>(); + try { + while (documents.size() < limit && cursor.hasNext()) { + documents.add(bsonToCanonicalExtendedJson(cursor.next())); + } + } catch (RuntimeException error) { + // A cursor that fails mid-iteration (server-side idle timeout, lost connection) can + // never be resumed, so drop it instead of leaking a slot in the cursor budgets. + closeFindCursorById(cursorId); + throw error; + } + boolean exhausted = !cursor.hasNext(); + if (exhausted) { + closeFindCursorById(cursorId); + } + Map result = new LinkedHashMap<>(); + result.put("documents", documents); + result.put("exhausted", exhausted); + return result; + } + + private static Object closeFindCursor(JsonObject params) { + closeFindCursorById(requiredCursorId(params)); + return Collections.singletonMap("ok", true); + } + + private static String requiredCursorId(JsonObject params) { + String cursorId = stringOrNull(params, "cursor_id"); + if (cursorId == null || cursorId.isBlank()) { + throw new IllegalArgumentException("cursor_id is required"); + } + return cursorId; + } + + private static int findCursorBatchSize(JsonObject params) { + if (!params.has("batch_size") || params.get("batch_size").isJsonNull()) { + return DEFAULT_FIND_CURSOR_BATCH_SIZE; + } + int batchSize = params.get("batch_size").getAsInt(); + if (batchSize <= 0) { + throw new IllegalArgumentException("batch_size must be a positive integer"); + } + return batchSize; + } + + private static void closeFindCursorById(String cursorId) { + MongoCursor cursor = FIND_CURSORS.remove(cursorId); + if (cursor != null) { + try { + cursor.close(); + } catch (RuntimeException ignored) { + // Closing a server cursor is best-effort; the client may already be gone. + } + } + Set owner = CURRENT_FIND_CURSOR_OWNER.get(); + if (owner != null) { + owner.remove(cursorId); + } + LEGACY_FIND_CURSOR_IDS.remove(cursorId); + } + + private static void closeFindCursors(Set cursorIds) { + for (String cursorId : new ArrayList<>(cursorIds)) { + closeFindCursorById(cursorId); + } + cursorIds.clear(); + } + + static void resetFindCursorsForTests() { + for (String cursorId : new ArrayList<>(FIND_CURSORS.keySet())) { + closeFindCursorById(cursorId); + } + LEGACY_FIND_CURSOR_IDS.clear(); } static Object parseId(String id) { @@ -1787,6 +1969,10 @@ public final class MongoAgent { return preserveUnsafeLongsForJsonClients(doc, relaxed).getAsJsonObject(); } + static JsonObject bsonToCanonicalExtendedJson(Document doc) { + return JsonParser.parseString(doc.toJson(CANONICAL_JSON_SETTINGS)).getAsJsonObject(); + } + private static JsonElement preserveUnsafeLongsForJsonClients(Object bsonValue, JsonElement relaxedValue) { if ( bsonValue instanceof Long longValue && @@ -1914,6 +2100,9 @@ public final class MongoAgent { case AgentProtocol.MONGO_METHOD_DROP_DATABASE -> dropDatabase(params); case AgentProtocol.MONGO_METHOD_INSERT_DOCUMENT -> insertDocument(params); case AgentProtocol.MONGO_METHOD_INSERT_DOCUMENTS -> insertDocuments(params); + case AgentProtocol.MONGO_METHOD_START_FIND_CURSOR -> startFindCursor(params); + case AgentProtocol.MONGO_METHOD_FETCH_FIND_CURSOR -> fetchFindCursor(params); + case AgentProtocol.MONGO_METHOD_CLOSE_FIND_CURSOR -> closeFindCursor(params); case AgentProtocol.MONGO_METHOD_UPDATE_DOCUMENT -> updateDocument(params); case AgentProtocol.MONGO_METHOD_UPDATE_DOCUMENTS -> updateDocuments(params); case AgentProtocol.MONGO_METHOD_REPLACE_DOCUMENT -> replaceDocument(params); @@ -1921,6 +2110,7 @@ public final class MongoAgent { case AgentProtocol.MONGO_METHOD_DELETE_DOCUMENTS -> deleteDocuments(params); case AgentProtocol.MONGO_METHOD_RUN_COMMAND -> runCommand(params); case AgentProtocol.METHOD_DISCONNECT, AgentProtocol.METHOD_SHUTDOWN -> { + closeFindCursors(LEGACY_FIND_CURSOR_IDS); closeLegacyClient(); if (AgentProtocol.METHOD_SHUTDOWN.equals(method)) { System.exit(0); @@ -1968,6 +2158,7 @@ public final class MongoAgent { static String handleRequest(String line, MongoClient client) { if (client != null) { CURRENT_CLIENT.set(client); + CURRENT_FIND_CURSOR_OWNER.set(LEGACY_FIND_CURSOR_IDS); } try { JsonObject req = JsonParser.parseString(line).getAsJsonObject(); @@ -1995,6 +2186,7 @@ public final class MongoAgent { } finally { if (client != null) { CURRENT_CLIENT.remove(); + CURRENT_FIND_CURSOR_OWNER.remove(); } } } @@ -2120,6 +2312,7 @@ public final class MongoAgent { private static final class Session { private final MongoClient client; + private final Set findCursorIds = ConcurrentHashMap.newKeySet(); private Session(MongoClient client) { this.client = client; @@ -2127,10 +2320,12 @@ public final class MongoAgent { private synchronized Object handle(String method, JsonObject params) { CURRENT_CLIENT.set(client); + CURRENT_FIND_CURSOR_OWNER.set(findCursorIds); try { return dispatch(method, params); } finally { CURRENT_CLIENT.remove(); + CURRENT_FIND_CURSOR_OWNER.remove(); } } @@ -2144,6 +2339,7 @@ public final class MongoAgent { } private synchronized void close() { + closeFindCursors(findCursorIds); client.close(); } } diff --git a/agents/drivers/mongodb/src/test/java/com/dbx/agent/mongodb/MongoAgentTest.java b/agents/drivers/mongodb/src/test/java/com/dbx/agent/mongodb/MongoAgentTest.java index 8b287bd2e5..82c0fead96 100644 --- a/agents/drivers/mongodb/src/test/java/com/dbx/agent/mongodb/MongoAgentTest.java +++ b/agents/drivers/mongodb/src/test/java/com/dbx/agent/mongodb/MongoAgentTest.java @@ -1,5 +1,6 @@ package com.dbx.agent.mongodb; +import static org.junit.jupiter.api.Assertions.assertArrayEquals; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNotNull; @@ -12,7 +13,12 @@ import com.google.gson.Gson; import com.google.gson.JsonArray; import com.google.gson.JsonObject; import com.google.gson.JsonParser; +import com.mongodb.MongoBulkWriteException; import com.mongodb.MongoClientSettings; +import com.mongodb.ServerAddress; +import com.mongodb.bulk.BulkWriteError; +import com.mongodb.bulk.BulkWriteResult; +import com.mongodb.bulk.WriteConcernError; import com.mongodb.client.AggregateIterable; import com.mongodb.client.FindIterable; import com.mongodb.client.ListCollectionsIterable; @@ -40,8 +46,12 @@ import java.util.Date; import java.util.List; import java.util.Map; import java.util.concurrent.TimeUnit; +import org.bson.BsonDocument; +import org.bson.BsonRegularExpression; import org.bson.Document; +import org.bson.types.Binary; import org.bson.types.ObjectId; +import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; @@ -99,6 +109,11 @@ class MongoAgentTest { Files.writeString(clientKeyPath, pkcs8Pem); } + @AfterEach + void resetFindCursors() { + MongoAgent.resetFindCursorsForTests(); + } + // ─── existing tests ─── @Test @@ -128,6 +143,7 @@ class MongoAgentTest { assertTrue(containsCapability(result.getAsJsonArray("capabilities"), AgentProtocol.CAPABILITY_MONGO_CLONE_COLLECTION)); assertTrue(containsCapability(result.getAsJsonArray("capabilities"), AgentProtocol.CAPABILITY_MONGO_RUN_COMMAND)); assertTrue(containsCapability(result.getAsJsonArray("capabilities"), AgentProtocol.CAPABILITY_MONGO_INSERT_DOCUMENTS)); + assertTrue(containsCapability(result.getAsJsonArray("capabilities"), AgentProtocol.CAPABILITY_MONGO_FIND_CURSOR)); } @Test @@ -150,6 +166,7 @@ class MongoAgentTest { assertTrue(containsCapability(result.getAsJsonArray("capabilities"), AgentProtocol.CAPABILITY_MONGO_CLONE_COLLECTION)); assertTrue(containsCapability(result.getAsJsonArray("capabilities"), AgentProtocol.CAPABILITY_MONGO_RUN_COMMAND)); assertTrue(containsCapability(result.getAsJsonArray("capabilities"), AgentProtocol.CAPABILITY_MONGO_INSERT_DOCUMENTS)); + assertTrue(containsCapability(result.getAsJsonArray("capabilities"), AgentProtocol.CAPABILITY_MONGO_FIND_CURSOR)); } @Test @@ -894,6 +911,10 @@ class MongoAgentTest { assertEquals("12", batches.get(0).get(1).getString("refid")); assertTrue(AgentProtocol.MONGO_LEGACY_METHODS.contains(AgentProtocol.MONGO_METHOD_INSERT_DOCUMENTS)); assertTrue(AgentProtocol.MONGO_LEGACY_CAPABILITIES.contains(AgentProtocol.CAPABILITY_MONGO_INSERT_DOCUMENTS)); + assertTrue(AgentProtocol.MONGO_LEGACY_CAPABILITIES.contains(AgentProtocol.CAPABILITY_MONGO_FIND_CURSOR)); + assertTrue(AgentProtocol.MONGO_LEGACY_METHODS.contains(AgentProtocol.MONGO_METHOD_START_FIND_CURSOR)); + assertTrue(AgentProtocol.MONGO_LEGACY_METHODS.contains(AgentProtocol.MONGO_METHOD_FETCH_FIND_CURSOR)); + assertTrue(AgentProtocol.MONGO_LEGACY_METHODS.contains(AgentProtocol.MONGO_METHOD_CLOSE_FIND_CURSOR)); } @Test @@ -939,6 +960,293 @@ class MongoAgentTest { } } + @Test + void documentForWritePreservesExtendedJsonBsonTypes() { + Document document = MongoAgent.documentForWrite( + "{\"_id\":{\"$oid\":\"507f1f77bcf86cd799439011\"}," + + "\"count\":{\"$numberLong\":\"9007199254740993\"}," + + "\"when\":{\"$date\":\"2020-01-02T03:04:05.000Z\"}}" + ); + + assertEquals(new ObjectId("507f1f77bcf86cd799439011"), document.getObjectId("_id")); + assertEquals(9007199254740993L, document.get("count")); + assertTrue(document.get("when") instanceof Date); + assertEquals(Date.from(java.time.Instant.parse("2020-01-02T03:04:05.000Z")), document.getDate("when")); + } + + @Test + void insertDocumentsRpcHonoursExplicitUnorderedInserts() { + List> batches = new ArrayList<>(); + List calls = new ArrayList<>(); + List ordered = new ArrayList<>(); + MongoClient client = recordingInsertMongoClient(calls, batches, ordered); + + JsonObject params = new JsonObject(); + params.addProperty("database", "app"); + params.addProperty("collection", "users"); + params.addProperty("docs_json", "[{\"name\":\"Ada\"}]"); + params.addProperty("ordered", false); + JsonObject request = new JsonObject(); + request.addProperty("jsonrpc", "2.0"); + request.addProperty("id", 44); + request.addProperty("method", "insert_documents"); + request.add("params", params); + + JsonObject response = JsonParser.parseString(MongoAgent.handleRequest(request.toString(), client)).getAsJsonObject(); + + assertFalse(response.has("error"), response.toString()); + assertEquals(List.of(false), ordered); + assertEquals(1, response.getAsJsonObject("result").get("affected_rows").getAsInt()); + } + + @Test + void insertDocumentsRpcPreservesExtendedJsonTypesInTheInsertManyBatch() { + List> batches = new ArrayList<>(); + List calls = new ArrayList<>(); + MongoClient client = recordingInsertMongoClient(calls, batches); + String documents = "[{\"_id\":{\"$oid\":\"507f1f77bcf86cd799439011\"}," + + "\"count\":{\"$numberLong\":\"9007199254740993\"}}]"; + + JsonObject response = insertDocumentsRpc(client, 45, "users", documents); + + assertFalse(response.has("error"), response.toString()); + assertEquals(1, batches.size()); + Document written = batches.get(0).get(0); + assertEquals(new ObjectId("507f1f77bcf86cd799439011"), written.getObjectId("_id")); + assertEquals(9007199254740993L, written.get("count")); + assertTrue(written.get("count") instanceof Long); + } + + @Test + void insertDocumentsRpcParsesCanonicalBinaryAndRegularExpression() { + List> batches = new ArrayList<>(); + MongoClient client = recordingInsertMongoClient(new ArrayList<>(), batches); + String documents = "[{\"bin\":{\"$binary\":{\"base64\":\"AQID\",\"subType\":\"00\"}}," + + "\"re\":{\"$regularExpression\":{\"pattern\":\"^test$\",\"options\":\"i\"}}}]"; + + JsonObject response = insertDocumentsRpc(client, 46, "users", documents); + + assertFalse(response.has("error"), response.toString()); + Document written = batches.get(0).get(0); + assertTrue(written.get("bin") instanceof Binary, String.valueOf(written.get("bin"))); + Binary binary = (Binary) written.get("bin"); + assertEquals(0, binary.getType()); + assertArrayEquals(new byte[] {1, 2, 3}, binary.getData()); + assertTrue(written.get("re") instanceof BsonRegularExpression, String.valueOf(written.get("re"))); + BsonRegularExpression regex = (BsonRegularExpression) written.get("re"); + assertEquals("^test$", regex.getPattern()); + assertEquals("i", regex.getOptions()); + } + + @Test + void bulkInsertResultKeepsInsertedCountAndPerDocumentErrors() { + BulkWriteResult writeResult = BulkWriteResult.acknowledged(1, 0, 0, 0, List.of()); + MongoBulkWriteException error = new MongoBulkWriteException( + writeResult, + List.of(new BulkWriteError(11000, "E11000 duplicate key", new BsonDocument(), 1)), + null, + new ServerAddress() + ); + + Map result = MongoAgent.bulkInsertResult(error); + + assertEquals(1, result.get("affected_rows")); + @SuppressWarnings("unchecked") + List> errors = (List>) result.get("errors"); + assertEquals(1, errors.size()); + assertEquals(1, errors.get(0).get("index")); + assertEquals(11000, errors.get(0).get("code")); + assertEquals("E11000 duplicate key", errors.get(0).get("message")); + } + + @Test + void bulkInsertResultReportsWriteConcernFailures() { + // The driver raises MongoBulkWriteException with no write errors when only the write + // concern failed, so the batch must not come back looking successful. + MongoBulkWriteException error = new MongoBulkWriteException( + BulkWriteResult.acknowledged(0, 0, 0, 0, List.of()), + List.of(), + new WriteConcernError(64, "WriteConcernFailed", "waiting for replication timed out", new BsonDocument()), + new ServerAddress() + ); + + Map result = MongoAgent.bulkInsertResult(error); + + assertEquals(0, result.get("affected_rows")); + @SuppressWarnings("unchecked") + List> errors = (List>) result.get("errors"); + assertEquals(1, errors.size()); + assertEquals(64, errors.get(0).get("code")); + assertEquals("waiting for replication timed out", errors.get(0).get("message")); + assertFalse(errors.get(0).containsKey("index")); + } + + @Test + void findCursorRpcPagesExtendedJsonDocumentsThenExhausts() { + List calls = new ArrayList<>(); + List documents = List.of( + new Document("_id", new ObjectId("507f1f77bcf86cd799439011")) + .append("n", 1L) + .append("bin", new Binary((byte) 0, new byte[] {1, 2, 3})) + .append("re", new BsonRegularExpression("^test$", "i")), + new Document("_id", new ObjectId("507f1f77bcf86cd799439012")).append("n", 2L) + ); + MongoClient client = recordingFindMongoClient(calls, documents); + + JsonObject start = findCursorRpc(client, 50, "start_find_cursor", cursorParams("users", 1, null)); + assertFalse(start.has("error"), start.toString()); + String cursorId = start.getAsJsonObject("result").get("cursor_id").getAsString(); + assertEquals(1, start.getAsJsonObject("result").get("batch_size").getAsInt()); + + JsonObject first = findCursorRpc(client, 51, "fetch_find_cursor", cursorParams(null, 0, cursorId)); + assertFalse(first.has("error"), first.toString()); + assertEquals(1, first.getAsJsonObject("result").getAsJsonArray("documents").size()); + assertFalse(first.getAsJsonObject("result").get("exhausted").getAsBoolean()); + JsonObject firstDocument = first.getAsJsonObject("result").getAsJsonArray("documents").get(0).getAsJsonObject(); + assertEquals( + "507f1f77bcf86cd799439011", + firstDocument.getAsJsonObject("_id").get("$oid").getAsString() + ); + assertEquals("1", firstDocument.getAsJsonObject("n").get("$numberLong").getAsString()); + JsonObject binary = firstDocument.getAsJsonObject("bin").getAsJsonObject("$binary"); + assertEquals("AQID", binary.get("base64").getAsString(), firstDocument.toString()); + assertEquals("00", binary.get("subType").getAsString(), firstDocument.toString()); + JsonObject regex = firstDocument.getAsJsonObject("re").getAsJsonObject("$regularExpression"); + assertEquals("^test$", regex.get("pattern").getAsString(), firstDocument.toString()); + assertEquals("i", regex.get("options").getAsString(), firstDocument.toString()); + + JsonObject second = findCursorRpc(client, 52, "fetch_find_cursor", cursorParams(null, 0, cursorId)); + assertFalse(second.has("error"), second.toString()); + assertEquals(1, second.getAsJsonObject("result").getAsJsonArray("documents").size()); + assertTrue(second.getAsJsonObject("result").get("exhausted").getAsBoolean()); + + JsonObject missing = findCursorRpc(client, 53, "fetch_find_cursor", cursorParams(null, 0, cursorId)); + assertTrue(missing.has("error"), missing.toString()); + assertEquals("Find cursor not found", missing.getAsJsonObject("error").get("message").getAsString()); + assertTrue(calls.contains("find")); + assertTrue(calls.contains("batchSize:1")); + assertTrue(calls.contains("iterator")); + assertTrue(calls.contains("close")); + } + + @Test + void closeFindCursorRpcIsIdempotent() { + List calls = new ArrayList<>(); + MongoClient client = recordingFindMongoClient(calls, List.of(new Document("n", 1))); + JsonObject start = findCursorRpc(client, 54, "start_find_cursor", cursorParams("users", 10, null)); + String cursorId = start.getAsJsonObject("result").get("cursor_id").getAsString(); + + JsonObject closed = findCursorRpc(client, 55, "close_find_cursor", cursorParams(null, 0, cursorId)); + JsonObject again = findCursorRpc(client, 56, "close_find_cursor", cursorParams(null, 0, cursorId)); + + assertFalse(closed.has("error"), closed.toString()); + assertFalse(again.has("error"), again.toString()); + assertEquals(1, calls.stream().filter("close"::equals).count()); + } + + @Test + void bsonToCanonicalExtendedJsonKeepsInt64DistinctFromJsonNumber() { + JsonObject json = MongoAgent.bsonToCanonicalExtendedJson( + new Document("n", 5L) + .append("bin", new Binary((byte) 0, new byte[] {1, 2, 3})) + .append("re", new BsonRegularExpression("^test$", "i")) + ); + + assertEquals("5", json.getAsJsonObject("n").get("$numberLong").getAsString()); + assertFalse(json.get("n").isJsonPrimitive(), json.toString()); + JsonObject binary = json.getAsJsonObject("bin").getAsJsonObject("$binary"); + assertEquals("AQID", binary.get("base64").getAsString(), json.toString()); + assertEquals("00", binary.get("subType").getAsString(), json.toString()); + assertFalse(json.getAsJsonObject("bin").has("$type"), json.toString()); + JsonObject regex = json.getAsJsonObject("re").getAsJsonObject("$regularExpression"); + assertEquals("^test$", regex.get("pattern").getAsString(), json.toString()); + assertEquals("i", regex.get("options").getAsString(), json.toString()); + assertFalse(json.getAsJsonObject("re").has("$regex"), json.toString()); + } + + @Test + void disconnectClosesOpenFindCursors() { + List calls = new ArrayList<>(); + MongoClient client = recordingFindMongoClient(calls, List.of(new Document("n", 1L))); + JsonObject start = findCursorRpc(client, 57, "start_find_cursor", cursorParams("users", 10, null)); + String cursorId = start.getAsJsonObject("result").get("cursor_id").getAsString(); + + JsonObject disconnect = JsonParser.parseString(MongoAgent.handleRequest( + "{\"jsonrpc\":\"2.0\",\"id\":58,\"method\":\"disconnect\",\"params\":{}}" + )).getAsJsonObject(); + JsonObject fetch = findCursorRpc(client, 59, "fetch_find_cursor", cursorParams(null, 0, cursorId)); + + assertFalse(disconnect.has("error"), disconnect.toString()); + assertTrue(fetch.has("error"), fetch.toString()); + assertEquals("Find cursor not found", fetch.getAsJsonObject("error").get("message").getAsString()); + assertTrue(calls.contains("close")); + } + + @Test + void fetchFailureReleasesTheCursorSlot() { + List calls = new ArrayList<>(); + MongoCursor failing = new MongoCursor<>() { + @Override + public void close() { + calls.add("close"); + } + + @Override + public boolean hasNext() { + throw new IllegalStateException("cursor id 42 not found"); + } + + @Override + public Document next() { + throw new IllegalStateException("cursor id 42 not found"); + } + + @Override + public Document tryNext() { + throw new IllegalStateException("cursor id 42 not found"); + } + + @Override + public ServerAddress getServerAddress() { + return new ServerAddress(); + } + + @Override + public com.mongodb.ServerCursor getServerCursor() { + return null; + } + }; + MongoClient client = findMongoClientWithCursor(calls, failing); + + JsonObject start = findCursorRpc(client, 60, "start_find_cursor", cursorParams("users", 10, null)); + String cursorId = start.getAsJsonObject("result").get("cursor_id").getAsString(); + JsonObject fetch = findCursorRpc(client, 61, "fetch_find_cursor", cursorParams(null, 0, cursorId)); + JsonObject again = findCursorRpc(client, 62, "fetch_find_cursor", cursorParams(null, 0, cursorId)); + + assertTrue(fetch.has("error"), fetch.toString()); + assertTrue(calls.contains("close")); + // A dead cursor is dropped, so the slot is free again instead of being held until disconnect. + assertEquals("Find cursor not found", again.getAsJsonObject("error").get("message").getAsString()); + } + + @Test + void startFindCursorEnforcesPerOwnerBudget() { + List calls = new ArrayList<>(); + MongoClient client = recordingFindMongoClient(calls, List.of(new Document("n", 1))); + + for (int index = 0; index < 16; index++) { + JsonObject started = findCursorRpc(client, 100 + index, "start_find_cursor", cursorParams("users", 100, null)); + assertFalse(started.has("error"), started.toString()); + } + JsonObject rejected = findCursorRpc(client, 200, "start_find_cursor", cursorParams("users", 100, null)); + + assertTrue(rejected.has("error"), rejected.toString()); + assertEquals( + "MongoDB find cursor limit reached: 16", + rejected.getAsJsonObject("error").get("message").getAsString() + ); + } + @Test void updateDocumentsRpcUsesDocumentOverloads() { List calls = new ArrayList<>(); @@ -1458,9 +1766,140 @@ class MongoAgentTest { return JsonParser.parseString(MongoAgent.handleRequest(request.toString(), client)).getAsJsonObject(); } + private static JsonObject cursorParams(String collection, int batchSize, String cursorId) { + JsonObject params = new JsonObject(); + params.addProperty("database", "app"); + if (collection != null) { + params.addProperty("collection", collection); + } + if (batchSize > 0) { + params.addProperty("batch_size", batchSize); + } + if (cursorId != null) { + params.addProperty("cursor_id", cursorId); + params.addProperty("limit", 1); + } + return params; + } + + private static JsonObject findCursorRpc(MongoClient client, int id, String method, JsonObject params) { + JsonObject request = new JsonObject(); + request.addProperty("jsonrpc", "2.0"); + request.addProperty("id", id); + request.addProperty("method", method); + request.add("params", params); + return JsonParser.parseString(MongoAgent.handleRequest(request.toString(), client)).getAsJsonObject(); + } + + @SuppressWarnings("unchecked") + private static MongoClient recordingFindMongoClient(List calls, List documents) { + MongoCursor cursor = new MongoCursor<>() { + private int index; + private boolean closed; + + @Override + public void close() { + closed = true; + calls.add("close"); + } + + @Override + public boolean hasNext() { + return !closed && index < documents.size(); + } + + @Override + public Document next() { + return documents.get(index++); + } + + @Override + public Document tryNext() { + return hasNext() ? next() : null; + } + + @Override + public com.mongodb.ServerAddress getServerAddress() { + return new com.mongodb.ServerAddress(); + } + + @Override + public com.mongodb.ServerCursor getServerCursor() { + return null; + } + }; + return findMongoClientWithCursor(calls, cursor); + } + + /** Same proxy chain as [recordingFindMongoClient], for cursors that fail mid-iteration. */ + @SuppressWarnings("unchecked") + private static MongoClient findMongoClientWithCursor(List calls, MongoCursor cursor) { + FindIterable iterable = (FindIterable) Proxy.newProxyInstance( + FindIterable.class.getClassLoader(), + new Class[] {FindIterable.class}, + (proxy, method, args) -> { + switch (method.getName()) { + case "batchSize" -> { + calls.add("batchSize:" + args[0]); + return proxy; + } + case "projection", "sort", "collation", "filter", "skip", "limit" -> { + calls.add(method.getName()); + return proxy; + } + case "iterator" -> { + calls.add("iterator"); + return cursor; + } + default -> throw new UnsupportedOperationException(method.getName()); + } + } + ); + MongoCollection collection = (MongoCollection) Proxy.newProxyInstance( + MongoCollection.class.getClassLoader(), + new Class[] {MongoCollection.class}, + (proxy, method, args) -> { + if ("find".equals(method.getName())) { + calls.add("find"); + return iterable; + } + throw new UnsupportedOperationException(method.getName()); + } + ); + MongoDatabase database = (MongoDatabase) Proxy.newProxyInstance( + MongoDatabase.class.getClassLoader(), + new Class[] {MongoDatabase.class}, + (proxy, method, args) -> { + if ("getCollection".equals(method.getName())) { + calls.add("getCollection:" + args[0]); + return collection; + } + throw new UnsupportedOperationException(method.getName()); + } + ); + return (MongoClient) Proxy.newProxyInstance( + MongoClient.class.getClassLoader(), + new Class[] {MongoClient.class}, + (proxy, method, args) -> { + if ("getDatabase".equals(method.getName())) { + calls.add("getDatabase:" + args[0]); + return database; + } + throw new UnsupportedOperationException(method.getName()); + } + ); + } + @SuppressWarnings("unchecked") private static MongoClient recordingInsertMongoClient( List calls, List> batches + ) { + return recordingInsertMongoClient(calls, batches, null); + } + + @SuppressWarnings("unchecked") + private static MongoClient recordingInsertMongoClient( + List calls, List> batches, List ordered ) { MongoCollection collection = (MongoCollection) Proxy.newProxyInstance( MongoCollection.class.getClassLoader(), @@ -1470,6 +1909,13 @@ class MongoAgentTest { List batch = ((List) args[0]).stream().map(Document::new).toList(); batches.add(batch); calls.add("insertMany:" + batch.size()); + if (ordered != null) { + boolean isOrdered = true; + if (args.length > 1 && args[1] instanceof InsertManyOptions options) { + isOrdered = options.isOrdered(); + } + ordered.add(isOrdered); + } return null; } throw new UnsupportedOperationException(method.getName()); diff --git a/apps/desktop/src/components/document/MongoImportDialog.vue b/apps/desktop/src/components/document/MongoImportDialog.vue index 31ada2b70a..43e63ecb67 100644 --- a/apps/desktop/src/components/document/MongoImportDialog.vue +++ b/apps/desktop/src/components/document/MongoImportDialog.vue @@ -16,6 +16,7 @@ import { executeWithProductionContextGuard } from "@/lib/database/productionExec import { TABLE_IMPORT_ENCODING_OPTIONS } from "@/lib/table/tableImport"; import { importPreviewInput, importSourceDisplayName, importTextDelimiterForName, uploadedImportSourceFromPreview, type UploadedImportSource } from "@/lib/import/importSource"; import { useToast } from "@/composables/useToast"; +import { translateBackendError } from "@/i18n/backend-errors"; import * as api from "@/lib/backend/api"; const { t } = useI18n(); @@ -267,7 +268,7 @@ async function startImport() { running.value = false; open.value = false; } catch (error) { - errorMessage.value = error instanceof Error ? error.message : String(error); + errorMessage.value = translateBackendError(t, error); running.value = false; } finally { cancelling.value = false; diff --git a/apps/desktop/src/i18n/__tests__/backendErrors.spec.ts b/apps/desktop/src/i18n/__tests__/backendErrors.spec.ts index 66a87a8fad..745b18becd 100644 --- a/apps/desktop/src/i18n/__tests__/backendErrors.spec.ts +++ b/apps/desktop/src/i18n/__tests__/backendErrors.spec.ts @@ -101,6 +101,34 @@ const CASES: { name: string; message: string; key: string; params?: Record): " envelope. + name: "MongoDB Legacy export cursor expired", + message: "Agent RPC error (-1): Find cursor not found", + key: "mongo.import.legacyCursorInvalid", + }, { name: "DuckDB draining", message: "The previous DuckDB query is still stopping. Please try again shortly.", @@ -401,6 +429,17 @@ describe("backend error translation", () => { expect(translateBackendError(t, error)).toBe(`${t("backendErrors.legacy")}\n\nlegacy backend failure`); }); + // Tauri wraps every backend rejection in a legacy envelope, so a message the catalog knows must + // still be phrased by the catalog rather than surfaced as raw English under a generic summary. + test("phrases a known message that arrived inside a legacy envelope", () => { + const t = translatorFor("zh-CN"); + const wrap = (message: string) => new BackendErrorException(message); + + expect(translateBackendError(t, wrap("row 1: MongoDB Legacy Agent does not support insertMany; upgrade or reinstall the MongoDB Legacy driver"))).toBe(t("mongo.import.legacyInsertUnsupported")); + expect(translateBackendError(t, wrap("Agent RPC error (-1): Find cursor not found"))).toBe(t("mongo.import.legacyCursorInvalid")); + expect(translateBackendError(t, wrap("MongoDB Legacy Agent rejected 2 of 500 documents: E11000 duplicate key"))).toBe(t("mongo.insert.partialFailure", { failed: "2", total: "500", message: "E11000 duplicate key" })); + }); + test("keeps an explicit original detail when a structured legacy error omits detail", () => { const t = translatorFor("zh-CN"); const error = { @@ -529,6 +568,9 @@ describe("backend error wording is pinned to the Rust sources", () => { test.each([ ["crates/dbx-core/src/query_result_export.rs", "Streaming export is unsupported for this query. Simplify it or use a supported driver."], ["crates/dbx-core/src/query_result_export.rs", "Streaming export needs a result-set session, but this driver returned no session_id."], + ["crates/dbx-core/src/mongodb_import_export.rs", "MongoDB Legacy Agent does not support insertMany; upgrade or reinstall the MongoDB Legacy driver"], + ["crates/dbx-core/src/mongodb_import_export.rs", "MongoDB Legacy Agent returned an invalid find cursor"], + ["crates/dbx-core/src/mongo_ops.rs", "MongoDB Legacy Agent rejected "], ["crates/dbx-core/src/agent_service.rs", "Failed to remove the old JRE directory: "], ["crates/dbx-core/src/agent_service.rs", "is in use by drivers: "], ["crates/dbx-core/src/agent_service.rs", "agent-registry.json not found in the ZIP; not a valid offline driver package."], diff --git a/apps/desktop/src/i18n/backend-errors.ts b/apps/desktop/src/i18n/backend-errors.ts index 71a5681508..1c045d3fac 100644 --- a/apps/desktop/src/i18n/backend-errors.ts +++ b/apps/desktop/src/i18n/backend-errors.ts @@ -1,4 +1,4 @@ -import { GENERIC_TRANSPORT_FAILURE_MESSAGE, normalizeBackendError, sanitizeBackendErrorMessage, type BackendError } from "@/lib/backend/errorUtils"; +import { GENERIC_TRANSPORT_FAILURE_MESSAGE, LEGACY_BACKEND_ERROR_CODE, normalizeBackendError, sanitizeBackendErrorMessage, type BackendError } from "@/lib/backend/errorUtils"; import { PHOENIX_DRIVER_NOT_INSTALLED_ERROR, PHOENIX_JDBC_PLUGIN_NOT_INSTALLED_ERROR } from "@/lib/database/phoenixConnection"; /** @@ -127,6 +127,17 @@ const patterns: [RegExp, string][] = [ // Query result export limits (crates/dbx-core/src/query_result_export.rs) [/^Streaming export is unsupported for this query\. Simplify it or use a supported driver\.$/, "exportProgress.streamingUnsupported"], + // MongoDB Legacy agent failures: messages raised by the agent arrive wrapped in "Agent RPC + // error (): ", and the import path prefixes the row a batch-level failure belongs to. + [/^(?:Agent RPC error \(-?\d+\): )?MongoDB Legacy Agent does not support insertMany; upgrade or reinstall the MongoDB Legacy driver$/, "mongo.import.legacyInsertUnsupported"], + [/^(?:row \d+: )?MongoDB Legacy Agent does not support insertMany; upgrade or reinstall the MongoDB Legacy driver$/, "mongo.import.legacyInsertUnsupported"], + [/^MongoDB Legacy Agent rejected (\d+) of (\d+) documents: ([\s\S]+)$/, "mongo.insert.partialFailure"], + [/^(?:Agent RPC error \(-?\d+\): )?MongoDB Legacy Agent returned an invalid find cursor$/, "mongo.import.legacyCursorInvalid"], + [/^(?:Agent RPC error \(-?\d+\): )?MongoDB Legacy Agent returned an invalid find cursor page$/, "mongo.import.legacyCursorInvalid"], + [/^(?:Agent RPC error \(-?\d+\): )?Find cursor not found$/, "mongo.import.legacyCursorInvalid"], +[/^(?:Agent RPC error \(-?\d+\): )?MongoDB Legacy Agent does not support type-preserving export; upgrade or reinstall the MongoDB Legacy driver$/, "mongo.import.legacyExportUnsupported"], +[/^(?:Agent RPC error \(-?\d+\): )?MongoDB Legacy Agent returned the same export page twice; upgrade or reinstall the MongoDB Legacy driver$/, "mongo.import.legacyExportStalled"], +[/^(?:Agent RPC error \(-?\d+\): )?MongoDB Legacy Agent returned a document without _id$/, "mongo.import.legacyExportMissingId"], [/^Streaming export needs a result-set session, but this driver returned no session_id\.$/, "exportProgress.agentSessionMissing"], // Legacy bundled DuckDB error kept for compatibility with older backends. @@ -157,6 +168,7 @@ const paramNames: Record = { "connection.jreNotInstalled": "jre", "ai.configNameExists": "name", "nacos.nacosManagedNamespaceAccessDenied": "detail", + "mongo.insert.partialFailure": ["failed", "total", "message"], "settings.tunnelsHttpTestSuccess": "code", "settings.tunnelsProxyTimedOut": "duration", "settings.tunnelsProxyConnectFailed": "error", @@ -188,27 +200,11 @@ function usableFallbackDetail(value: unknown): string | undefined { return detail; } -function translateStructuredBackendError(t: BackendErrorTranslate, error: BackendError, fallbackDetail?: unknown): string { - const translated = t(error.messageKey, error.messageParams); - const summary = translated !== error.messageKey ? translated : t("backendErrors.unknown"); - let detail = error.detail ? sanitizeBackendErrorMessage(error.detail).trim() : usableFallbackDetail(fallbackDetail); - // Callers may pass an already-composited "summary\n\noriginal" cell as the - // fallback (query history re-translation of a failed result grid); strip the - // repeated summary so the message stays singular. - if (detail && detail.startsWith(`${summary}\n\n`)) { - detail = detail.slice(summary.length + 2).trim() || undefined; - } - const rawAdapterCode = error.diagnostics?.adapterCode; - const adapterCode = typeof rawAdapterCode === "string" && /^[A-Za-z][A-Za-z0-9_.-]{0,63}$/.test(rawAdapterCode) ? rawAdapterCode : undefined; - const diagnosticDetail = detail && adapterCode ? `[${adapterCode}] ${detail}` : (detail ?? adapterCode); - return diagnosticDetail && diagnosticDetail !== summary ? `${summary}\n\n${diagnosticDetail}` : summary; -} - -export function translateBackendError(t: BackendErrorTranslate, error: unknown, fallbackDetail?: unknown): string { - const structured = normalizeBackendError(error); - if (structured) return translateStructuredBackendError(t, structured, fallbackDetail); - - const message = backendErrorMessage(error); +/** + * Resolves a raw backend message through the known-message catalog, or `null` when the message is + * not one the UI knows how to phrase. + */ +function translateKnownMessage(t: BackendErrorTranslate, message: string): string | null { const exactKey = exactMessageKeys[message]; if (exactKey) return t(exactKey); @@ -238,5 +234,35 @@ export function translateBackendError(t: BackendErrorTranslate, error: unknown, return t(key); } } - return message; + return null; +} + +function translateStructuredBackendError(t: BackendErrorTranslate, error: BackendError, fallbackDetail?: unknown): string { + const translated = t(error.messageKey, error.messageParams); + const summary = translated !== error.messageKey ? translated : t("backendErrors.unknown"); + let detail = error.detail ? sanitizeBackendErrorMessage(error.detail).trim() : usableFallbackDetail(fallbackDetail); + // Callers may pass an already-composited "summary\n\noriginal" cell as the + // fallback (query history re-translation of a failed result grid); strip the + // repeated summary so the message stays singular. + if (detail && detail.startsWith(`${summary}\n\n`)) { + detail = detail.slice(summary.length + 2).trim() || undefined; + } + // An unclassified message still goes through the same catalog the raw transports use, so a + // message DBX knows reads the same whether it arrived wrapped or bare. + if (error.code === LEGACY_BACKEND_ERROR_CODE && detail) { + const known = translateKnownMessage(t, detail); + if (known) return known; + } + const rawAdapterCode = error.diagnostics?.adapterCode; + const adapterCode = typeof rawAdapterCode === "string" && /^[A-Za-z][A-Za-z0-9_.-]{0,63}$/.test(rawAdapterCode) ? rawAdapterCode : undefined; + const diagnosticDetail = detail && adapterCode ? `[${adapterCode}] ${detail}` : (detail ?? adapterCode); + return diagnosticDetail && diagnosticDetail !== summary ? `${summary}\n\n${diagnosticDetail}` : summary; +} + +export function translateBackendError(t: BackendErrorTranslate, error: unknown, fallbackDetail?: unknown): string { + const structured = normalizeBackendError(error); + if (structured) return translateStructuredBackendError(t, structured, fallbackDetail); + + const message = backendErrorMessage(error); + return translateKnownMessage(t, message) ?? message; } diff --git a/apps/desktop/src/i18n/locales/az.ts b/apps/desktop/src/i18n/locales/az.ts index 66cf864833..cf53640831 100644 --- a/apps/desktop/src/i18n/locales/az.ts +++ b/apps/desktop/src/i18n/locales/az.ts @@ -5406,6 +5406,11 @@ export default withEnglishFallback({ start: "İdxal et", success: "{count} sənəd idxal olundu", stillRunning: "İdxal hələ işləyir. Yeni paketləri dayandırmaq üçün Ləğv et düyməsini istifadə edin.", + legacyInsertUnsupported: "Bu MongoDB Legacy sürücüsü sənədləri toplu daxil edə bilmir. MongoDB Legacy sürücüsünü yeniləyin və ya yenidən quraşdırın.", + legacyCursorInvalid: "MongoDB Legacy ixrac kursoru artıq əlçatan deyil. İxracı yenidən cəhd edin.", + legacyExportUnsupported: "Bu MongoDB Legacy sürücüsü ixrac zamanı BSON tiplərini qorumağı dəstəkləmir. MongoDB Legacy sürücüsünü yeniləyin və ya yenidən quraşdırın.", + legacyExportStalled: "MongoDB Legacy sürücüsü eyni ixrac səhifəsini iki dəfə qaytardı. Sürücünü yeniləyin və ya yenidən quraşdırıb ixracı yenidən cəhd edin.", + legacyExportMissingId: "MongoDB Legacy sürücüsü _id olmayan sənəd qaytardı; ixrac təhlükəsiz şəkildə səhifələnə bilmir.", rowsRead: "Oxunan sətirlər", rowsInserted: "Əlavə olunan sətirlər", rowsFailed: "Uğursuz sətirlər", @@ -5417,6 +5422,9 @@ export default withEnglishFallback({ done: "Bitdi", }, }, + insert: { + partialFailure: "MongoDB Legacy drayveri {total} sənədin {failed} hissəsini rədd etdi: {message}", + }, }, meilisearch: { ...meilisearchManagementAz, diff --git a/apps/desktop/src/i18n/locales/en.ts b/apps/desktop/src/i18n/locales/en.ts index 9a7679b502..44ab6375f4 100644 --- a/apps/desktop/src/i18n/locales/en.ts +++ b/apps/desktop/src/i18n/locales/en.ts @@ -5558,6 +5558,11 @@ export default { start: "Import", success: "Imported {count} documents", stillRunning: "Import is still running. Use Cancel to stop new batches.", + legacyInsertUnsupported: "This MongoDB Legacy driver cannot insert documents in bulk. Upgrade or reinstall the MongoDB Legacy driver.", + legacyCursorInvalid: "The MongoDB Legacy export cursor is no longer available. Retry the export.", + legacyExportUnsupported: "This MongoDB Legacy driver cannot preserve BSON types during export. Upgrade or reinstall the MongoDB Legacy driver.", + legacyExportStalled: "The MongoDB Legacy driver returned the same export page twice. Upgrade or reinstall the MongoDB Legacy driver and retry the export.", + legacyExportMissingId: "The MongoDB Legacy driver returned a document without _id, so the export cannot page safely.", rowsRead: "Rows read", rowsInserted: "Rows inserted", rowsFailed: "Rows failed", @@ -5569,6 +5574,9 @@ export default { done: "Done", }, }, + insert: { + partialFailure: "The MongoDB Legacy driver rejected {failed} of {total} documents: {message}", + }, }, meilisearch: { ...meilisearchManagementEn, diff --git a/apps/desktop/src/i18n/locales/es.ts b/apps/desktop/src/i18n/locales/es.ts index 31e1c7c7ce..101889ffa7 100644 --- a/apps/desktop/src/i18n/locales/es.ts +++ b/apps/desktop/src/i18n/locales/es.ts @@ -5260,6 +5260,11 @@ export default withEnglishFallback({ start: "Importar", success: "Se importaron {count} documentos", stillRunning: "La importación sigue en curso. Use Cancelar para detener nuevos lotes.", + legacyInsertUnsupported: "Este controlador MongoDB Legacy no puede insertar documentos en lote. Actualice o reinstale el controlador MongoDB Legacy.", + legacyCursorInvalid: "El cursor de exportación de MongoDB Legacy ya no está disponible. Vuelva a intentar la exportación.", + legacyExportUnsupported: "Este controlador MongoDB Legacy no admite la preservación de tipos BSON al exportar. Actualice o reinstale el controlador MongoDB Legacy.", + legacyExportStalled: "El controlador MongoDB Legacy devolvió dos veces la misma página de exportación. Actualice o reinstale el controlador y vuelva a intentar la exportación.", + legacyExportMissingId: "El controlador MongoDB Legacy devolvió un documento sin _id, por lo que la exportación no puede paginar de forma segura.", rowsRead: "Filas leídas", rowsInserted: "Filas insertadas", rowsFailed: "Filas fallidas", @@ -5271,6 +5276,9 @@ export default withEnglishFallback({ done: "Hecho", }, }, + insert: { + partialFailure: "El controlador MongoDB Legacy rechazó {failed} de {total} documentos: {message}", + }, }, meilisearch: { ...meilisearchManagementEs, diff --git a/apps/desktop/src/i18n/locales/it.ts b/apps/desktop/src/i18n/locales/it.ts index 8eab3a392d..e0fc35e270 100644 --- a/apps/desktop/src/i18n/locales/it.ts +++ b/apps/desktop/src/i18n/locales/it.ts @@ -5258,6 +5258,11 @@ export default withEnglishFallback({ start: "Importa", success: "Importati {count} documenti", stillRunning: "L'importazione è ancora in esecuzione. Usa Annulla per fermare i nuovi batch.", + legacyInsertUnsupported: "Questo driver MongoDB Legacy non può inserire documenti in blocco. Aggiorna o reinstalla il driver MongoDB Legacy.", + legacyCursorInvalid: "Il cursore di esportazione MongoDB Legacy non è più disponibile. Riprova l'esportazione.", + legacyExportUnsupported: "Questo driver MongoDB Legacy non supporta la conservazione dei tipi BSON durante l'esportazione. Aggiorna o reinstalla il driver MongoDB Legacy.", + legacyExportStalled: "Il driver MongoDB Legacy ha restituito due volte la stessa pagina di esportazione. Aggiorna o reinstalla il driver e riprova l'esportazione.", + legacyExportMissingId: "Il driver MongoDB Legacy ha restituito un documento senza _id, quindi l'esportazione non può impaginare in modo sicuro.", rowsRead: "Righe lette", rowsInserted: "Righe inserite", rowsFailed: "Righe non riuscite", @@ -5269,6 +5274,9 @@ export default withEnglishFallback({ done: "Completato", }, }, + insert: { + partialFailure: "Il driver MongoDB Legacy ha rifiutato {failed} di {total} documenti: {message}", + }, }, meilisearch: { ...meilisearchManagementIt, diff --git a/apps/desktop/src/i18n/locales/ja.ts b/apps/desktop/src/i18n/locales/ja.ts index fc9daefad5..45e03987c7 100644 --- a/apps/desktop/src/i18n/locales/ja.ts +++ b/apps/desktop/src/i18n/locales/ja.ts @@ -5287,6 +5287,11 @@ export default withEnglishFallback({ start: "インポート", success: "{count} 件のドキュメントをインポートしました", stillRunning: "インポートは実行中です。以降のバッチを止めるにはキャンセルしてください。", + legacyInsertUnsupported: "この MongoDB Legacy ドライバーは一括挿入に対応していません。ドライバーをアップグレードまたは再インストールしてください。", + legacyCursorInvalid: "MongoDB Legacy のエクスポートカーソルは利用できなくなりました。エクスポートを再試行してください。", + legacyExportUnsupported: "この MongoDB Legacy ドライバーはエクスポート時の BSON 型保持に対応していません。ドライバーをアップグレードまたは再インストールしてください。", + legacyExportStalled: "MongoDB Legacy ドライバーが同じエクスポートページを 2 回返しました。ドライバーをアップグレードまたは再インストールしてから、エクスポートを再試行してください。", + legacyExportMissingId: "MongoDB Legacy ドライバーが _id のないドキュメントを返したため、エクスポートを安全にページングできません。", rowsRead: "読み取り行数", rowsInserted: "挿入行数", rowsFailed: "失敗行数", @@ -5298,6 +5303,9 @@ export default withEnglishFallback({ done: "完了", }, }, + insert: { + partialFailure: "MongoDB Legacy ドライバーは {total} 件中 {failed} 件のドキュメントを拒否しました: {message}", + }, }, meilisearch: { ...meilisearchManagementJa, diff --git a/apps/desktop/src/i18n/locales/ko.ts b/apps/desktop/src/i18n/locales/ko.ts index 748479f9ee..675a671d34 100644 --- a/apps/desktop/src/i18n/locales/ko.ts +++ b/apps/desktop/src/i18n/locales/ko.ts @@ -4895,6 +4895,11 @@ export default withEnglishFallback({ start: "가져오기", success: "{count}개 문서를 가져왔습니다", stillRunning: "가져오기가 계속 실행 중입니다. 이후 배치를 멈추려면 취소를 사용하세요.", + legacyInsertUnsupported: "이 MongoDB Legacy 드라이버는 문서 일괄 삽입을 지원하지 않습니다. MongoDB Legacy 드라이버를 업그레이드하거나 다시 설치하세요.", + legacyCursorInvalid: "MongoDB Legacy 내보내기 커서를 더 이상 사용할 수 없습니다. 내보내기를 다시 시도하세요.", + legacyExportUnsupported: "이 MongoDB Legacy 드라이버는 내보내기 시 BSON 타입 유지를 지원하지 않습니다. MongoDB Legacy 드라이버를 업그레이드하거나 다시 설치하세요.", + legacyExportStalled: "MongoDB Legacy 드라이버가 동일한 내보내기 페이지를 두 번 반환했습니다. 드라이버를 업그레이드하거나 다시 설치한 후 다시 시도하세요.", + legacyExportMissingId: "MongoDB Legacy 드라이버가 _id 없는 문서를 반환해 안전하게 페이지 매김할 수 없습니다.", rowsRead: "읽은 행", rowsInserted: "삽입된 행", rowsFailed: "실패한 행", @@ -4906,6 +4911,9 @@ export default withEnglishFallback({ done: "완료", }, }, + insert: { + partialFailure: "MongoDB Legacy 드라이버가 {total}개 문서 중 {failed}개를 거부했습니다: {message}", + }, }, meilisearch: { ...meilisearchManagementKo, diff --git a/apps/desktop/src/i18n/locales/pt-BR.ts b/apps/desktop/src/i18n/locales/pt-BR.ts index 61e52b5d7d..9ac79fde97 100644 --- a/apps/desktop/src/i18n/locales/pt-BR.ts +++ b/apps/desktop/src/i18n/locales/pt-BR.ts @@ -5260,6 +5260,11 @@ export default withEnglishFallback({ start: "Importar", success: "{count} documentos importados", stillRunning: "A importação ainda está em execução. Use Cancelar para interromper novos lotes.", + legacyInsertUnsupported: "Este driver MongoDB Legacy não pode inserir documentos em lote. Atualize ou reinstale o driver MongoDB Legacy.", + legacyCursorInvalid: "O cursor de exportação do MongoDB Legacy não está mais disponível. Tente exportar novamente.", + legacyExportUnsupported: "Este driver MongoDB Legacy não é compatível com a preservação de tipos BSON na exportação. Atualize ou reinstale o driver MongoDB Legacy.", + legacyExportStalled: "O driver MongoDB Legacy retornou a mesma página de exportação duas vezes. Atualize ou reinstale o driver e tente exportar novamente.", + legacyExportMissingId: "O driver MongoDB Legacy retornou um documento sem _id, então a exportação não pode paginar com segurança.", rowsRead: "Linhas lidas", rowsInserted: "Linhas inseridas", rowsFailed: "Linhas com falha", @@ -5271,6 +5276,9 @@ export default withEnglishFallback({ done: "Concluído", }, }, + insert: { + partialFailure: "O driver MongoDB Legacy rejeitou {failed} de {total} documentos: {message}", + }, }, meilisearch: { ...meilisearchManagementPtBR, diff --git a/apps/desktop/src/i18n/locales/tr.ts b/apps/desktop/src/i18n/locales/tr.ts index 316cd3ef8c..0574c0fbcb 100644 --- a/apps/desktop/src/i18n/locales/tr.ts +++ b/apps/desktop/src/i18n/locales/tr.ts @@ -5300,6 +5300,11 @@ export default withEnglishFallback({ start: "İçe aktar", success: "{count} belge içe aktarıldı", stillRunning: "İçe aktarma hâlâ çalışıyor. Yeni toplu işleri durdurmak için İptal’i kullanın.", + legacyInsertUnsupported: "Bu MongoDB Legacy sürücüsü belgeleri toplu ekleyemez. MongoDB Legacy sürücüsünü yükseltin veya yeniden yükleyin.", + legacyCursorInvalid: "MongoDB Legacy dışa aktarma imleci artık kullanılamıyor. Dışa aktarmayı yeniden deneyin.", + legacyExportUnsupported: "Bu MongoDB Legacy sürücüsü dışa aktarma sırasında BSON türlerini korumayı desteklemiyor. MongoDB Legacy sürücüsünü yükseltin veya yeniden yükleyin.", + legacyExportStalled: "MongoDB Legacy sürücüsü aynı dışa aktarma sayfasını iki kez döndürdü. Sürücüyü yükseltin veya yeniden yükleyin ve dışa aktarmayı yeniden deneyin.", + legacyExportMissingId: "MongoDB Legacy sürücüsü _id içermeyen bir belge döndürdü; dışa aktarma güvenli şekilde sayfalanamıyor.", rowsRead: "Okunan satırlar", rowsInserted: "Eklenen satırlar", rowsFailed: "Başarısız satırlar", @@ -5311,6 +5316,9 @@ export default withEnglishFallback({ done: "Bitti", }, }, + insert: { + partialFailure: "MongoDB Legacy sürücüsü {total} belgenin {failed} tanesini reddetti: {message}", + }, }, meilisearch: { ...meilisearchManagementTr, diff --git a/apps/desktop/src/i18n/locales/zh-CN.ts b/apps/desktop/src/i18n/locales/zh-CN.ts index e2c4691b03..d31ba9be9f 100644 --- a/apps/desktop/src/i18n/locales/zh-CN.ts +++ b/apps/desktop/src/i18n/locales/zh-CN.ts @@ -5533,6 +5533,11 @@ export default withEnglishFallback({ start: "开始导入", success: "已导入 {count} 个文档", stillRunning: "导入仍在运行。如需停止后续批次，请点取消。", + legacyInsertUnsupported: "当前 MongoDB Legacy 驱动不支持批量插入文档。请升级或重新安装 MongoDB Legacy 驱动。", + legacyCursorInvalid: "MongoDB Legacy 导出游标已失效，请重试导出。", + legacyExportUnsupported: "当前 MongoDB Legacy 驱动不支持导出时保留 BSON 类型。请升级或重新安装 MongoDB Legacy 驱动。", + legacyExportStalled: "MongoDB Legacy 驱动返回了重复的导出分页。请升级或重新安装 MongoDB Legacy 驱动后重试导出。", + legacyExportMissingId: "MongoDB Legacy 驱动返回了缺少 _id 的文档，导出无法安全分页。", rowsRead: "已读取", rowsInserted: "已插入", rowsFailed: "失败", @@ -5544,6 +5549,9 @@ export default withEnglishFallback({ done: "完成", }, }, + insert: { + partialFailure: "MongoDB Legacy 驱动拒绝了 {total} 个文档中的 {failed} 个：{message}", + }, }, meilisearch: { ...meilisearchManagementZhCN, diff --git a/apps/desktop/src/i18n/locales/zh-TW.ts b/apps/desktop/src/i18n/locales/zh-TW.ts index 1ef4721286..a37c6828d2 100644 --- a/apps/desktop/src/i18n/locales/zh-TW.ts +++ b/apps/desktop/src/i18n/locales/zh-TW.ts @@ -4575,6 +4575,11 @@ export default withEnglishFallback({ start: "開始匯入", success: "已匯入 {count} 個文件", stillRunning: "匯入仍在執行。如需停止後續批次，請點取消。", + legacyInsertUnsupported: "目前的 MongoDB Legacy 驅動程式不支援批次插入文件。請升級或重新安裝 MongoDB Legacy 驅動程式。", + legacyCursorInvalid: "MongoDB Legacy 匯出游標已失效，請重試匯出。", + legacyExportUnsupported: "目前的 MongoDB Legacy 驅動程式不支援匯出時保留 BSON 類型。請升級或重新安裝 MongoDB Legacy 驅動程式。", + legacyExportStalled: "MongoDB Legacy 驅動程式重複回傳了同一個匯出分頁。請升級或重新安裝 MongoDB Legacy 驅動程式後重試匯出。", + legacyExportMissingId: "MongoDB Legacy 驅動程式回傳了缺少 _id 的文件，匯出無法安全分頁。", rowsRead: "已讀取", rowsInserted: "已插入", rowsFailed: "失敗", @@ -4591,6 +4596,9 @@ export default withEnglishFallback({ nativeJsonHint: "BSON 值請使用 MongoDB Extended JSON，例如 $oid、$date、$numberLong。", invalidNativeJson: "原生 JSON 格式無效：{error}", nativeDocumentRequired: "原生 JSON 的最上層必須是文件物件。", + insert: { + partialFailure: "MongoDB Legacy 驅動程式拒絕了 {total} 個文件中的 {failed} 個：{message}", + }, }, meilisearch: { ...meilisearchManagementZhTW, diff --git a/apps/desktop/src/lib/backend/errorUtils.ts b/apps/desktop/src/lib/backend/errorUtils.ts index 5de31d1ee7..657940dc8b 100644 --- a/apps/desktop/src/lib/backend/errorUtils.ts +++ b/apps/desktop/src/lib/backend/errorUtils.ts @@ -182,6 +182,12 @@ function normalizeBackendErrorAtDepth(error: unknown, seen: WeakSet, dep export const GENERIC_TRANSPORT_FAILURE_MESSAGE = "Backend request failed"; +/** + * Marks a backend message the UI could not classify, so it is carried as raw text inside a + * structured envelope. Callers may still match that text against the known-message catalog. + */ +export const LEGACY_BACKEND_ERROR_CODE = "DBX-LEGACY-0001"; + export class BackendErrorException extends Error { readonly backendError: BackendError; @@ -193,7 +199,7 @@ export class BackendErrorException extends Error { this.name = "BackendErrorException"; this.backendError = backendError ?? { version: 1, - code: "DBX-LEGACY-0001", + code: LEGACY_BACKEND_ERROR_CODE, messageKey: "backendErrors.legacy", messageParams: {}, source: "legacyBackend", diff --git a/crates/dbx-core/assets/agent-protocol-v1.json b/crates/dbx-core/assets/agent-protocol-v1.json index 716e9eb035..abb7ebbc41 100644 --- a/crates/dbx-core/assets/agent-protocol-v1.json +++ b/crates/dbx-core/assets/agent-protocol-v1.json @@ -2,7 +2,7 @@ "protocolVersion": 1, "handshakeMethod": "handshake", "handshakeResponseFields": ["protocolVersion", "agentProtocolVersion", "capabilities"], - "allCapabilities": ["connect", "test_connection", "metadata", "query", "paged_query", "transaction", "ddl", "kv", "kv_ttl", "kv_cas", "kv_list_values", "kv_status", "kv_history", "etcd_compaction", "etcd_defrag", "etcd_watch", "etcd_lease", "etcd_auth", "mongo_drop_database", "mongo_clone_collection", "mongo_run_command", "mongo_insert_documents", "mongo_replace_document"], + "allCapabilities": ["connect", "test_connection", "metadata", "query", "paged_query", "transaction", "ddl", "kv", "kv_ttl", "kv_cas", "kv_list_values", "kv_status", "kv_history", "etcd_compaction", "etcd_defrag", "etcd_watch", "etcd_lease", "etcd_auth", "mongo_drop_database", "mongo_clone_collection", "mongo_run_command", "mongo_insert_documents", "mongo_replace_document", "mongo_find_cursor"], "capabilities": ["connect", "test_connection", "metadata", "query", "paged_query", "transaction", "ddl"], "defaultSqlCapabilities": ["connect", "test_connection", "metadata", "query", "paged_query", "transaction", "ddl"], "commonMethods": [ @@ -42,6 +42,6 @@ "disconnect", "shutdown" ], - "mongoLegacyMethods": ["list_databases", "list_collections", "find_documents", "find_one", "explain_find", "aggregate_documents", "find_documents_extended_json", "count_documents", "server_version", "create_index", "create_user", "drop_indexes", "drop_collection", "clone_collection", "drop_database", "insert_document", "insert_documents", "update_document", "update_documents", "replace_document", "delete_document", "delete_documents", "run_command"], + "mongoLegacyMethods": ["list_databases", "list_collections", "find_documents", "find_one", "explain_find", "aggregate_documents", "find_documents_extended_json", "count_documents", "server_version", "create_index", "create_user", "drop_indexes", "drop_collection", "clone_collection", "drop_database", "insert_document", "insert_documents", "update_document", "update_documents", "replace_document", "delete_document", "delete_documents", "run_command", "start_find_cursor", "fetch_find_cursor", "close_find_cursor"], "kvMethods": ["kv_list_prefix", "kv_get", "kv_put", "kv_delete", "kv_rename", "kv_history", "kv_status", "etcd_compact", "etcd_defrag", "etcd_watch_start", "etcd_watch_poll", "etcd_watch_stop", "etcd_lease_list", "etcd_lease_get", "etcd_lease_grant", "etcd_lease_keepalive_once", "etcd_lease_revoke", "etcd_auth_user_list", "etcd_auth_user_get", "etcd_auth_user_add", "etcd_auth_user_delete", "etcd_auth_user_change_password", "etcd_auth_user_grant_role", "etcd_auth_user_revoke_role", "etcd_auth_role_list", "etcd_auth_role_get", "etcd_auth_role_add", "etcd_auth_role_delete", "etcd_auth_role_grant_permission", "etcd_auth_role_revoke_permission"] } diff --git a/crates/dbx-core/assets/agent-protocol-v2.json b/crates/dbx-core/assets/agent-protocol-v2.json index 1b1a2017a9..c1ef26af18 100644 --- a/crates/dbx-core/assets/agent-protocol-v2.json +++ b/crates/dbx-core/assets/agent-protocol-v2.json @@ -2,7 +2,7 @@ "protocolVersion": 2, "handshakeMethod": "handshake", "handshakeResponseFields": ["protocolVersion", "agentProtocolVersion", "capabilities"], - "allCapabilities": ["connect", "test_connection", "metadata", "query", "paged_query", "transaction", "ddl", "kv", "kv_ttl", "kv_cas", "kv_list_values", "kv_status", "kv_history", "etcd_compaction", "etcd_defrag", "etcd_watch", "etcd_lease", "etcd_auth", "mongo_drop_database", "mongo_clone_collection", "mongo_run_command", "mongo_insert_documents", "mongo_replace_document", "multi_session", "structured_error_v1"], + "allCapabilities": ["connect", "test_connection", "metadata", "query", "paged_query", "transaction", "ddl", "kv", "kv_ttl", "kv_cas", "kv_list_values", "kv_status", "kv_history", "etcd_compaction", "etcd_defrag", "etcd_watch", "etcd_lease", "etcd_auth", "mongo_drop_database", "mongo_clone_collection", "mongo_run_command", "mongo_insert_documents", "mongo_replace_document", "mongo_find_cursor", "multi_session", "structured_error_v1"], "capabilities": ["connect", "test_connection", "metadata", "query", "paged_query", "transaction", "ddl", "multi_session"], "defaultSqlCapabilities": ["connect", "test_connection", "metadata", "query", "paged_query", "transaction", "ddl", "multi_session", "structured_error_v1"], "commonMethods": [ @@ -46,7 +46,7 @@ "disconnect", "shutdown" ], - "mongoLegacyMethods": ["list_databases", "list_collections", "find_documents", "find_one", "explain_find", "aggregate_documents", "find_documents_extended_json", "count_documents", "server_version", "create_index", "create_user", "drop_indexes", "drop_collection", "clone_collection", "drop_database", "insert_document", "insert_documents", "update_document", "update_documents", "replace_document", "delete_document", "delete_documents", "run_command"], + "mongoLegacyMethods": ["list_databases", "list_collections", "find_documents", "find_one", "explain_find", "aggregate_documents", "find_documents_extended_json", "count_documents", "server_version", "create_index", "create_user", "drop_indexes", "drop_collection", "clone_collection", "drop_database", "insert_document", "insert_documents", "update_document", "update_documents", "replace_document", "delete_document", "delete_documents", "run_command", "start_find_cursor", "fetch_find_cursor", "close_find_cursor"], "kvMethods": ["kv_list_prefix", "kv_get", "kv_put", "kv_delete", "kv_rename", "kv_history", "kv_status", "etcd_compact", "etcd_defrag", "etcd_watch_start", "etcd_watch_poll", "etcd_watch_stop", "etcd_lease_list", "etcd_lease_get", "etcd_lease_grant", "etcd_lease_keepalive_once", "etcd_lease_revoke", "etcd_auth_user_list", "etcd_auth_user_get", "etcd_auth_user_add", "etcd_auth_user_delete", "etcd_auth_user_change_password", "etcd_auth_user_grant_role", "etcd_auth_user_revoke_role", "etcd_auth_role_list", "etcd_auth_role_get", "etcd_auth_role_add", "etcd_auth_role_delete", "etcd_auth_role_grant_permission", "etcd_auth_role_revoke_permission"], "sessionField": "agentSessionId", "cursorSessionField": "sessionId" diff --git a/crates/dbx-core/src/db/agent_driver.rs b/crates/dbx-core/src/db/agent_driver.rs index 2b6309fbf1..bd4247d231 100644 --- a/crates/dbx-core/src/db/agent_driver.rs +++ b/crates/dbx-core/src/db/agent_driver.rs @@ -1042,6 +1042,7 @@ pub enum AgentCapability { MongoRunCommand, MongoInsertDocuments, MongoReplaceDocument, + MongoFindCursor, MultiSession, StructuredErrorV1, } @@ -1124,7 +1125,7 @@ fn parse_agent_rpc_error_header(header: &str) -> (Option, String) { } impl AgentCapability { - pub const ALL: [Self; 25] = [ + pub const ALL: [Self; 26] = [ Self::Connect, Self::TestConnection, Self::Metadata, @@ -1148,6 +1149,7 @@ impl AgentCapability { Self::MongoRunCommand, Self::MongoInsertDocuments, Self::MongoReplaceDocument, + Self::MongoFindCursor, Self::MultiSession, Self::StructuredErrorV1, ]; @@ -1177,6 +1179,7 @@ impl AgentCapability { Self::MongoRunCommand => "mongo_run_command", Self::MongoInsertDocuments => "mongo_insert_documents", Self::MongoReplaceDocument => "mongo_replace_document", + Self::MongoFindCursor => "mongo_find_cursor", Self::MultiSession => "multi_session", Self::StructuredErrorV1 => "structured_error_v1", } @@ -1383,10 +1386,13 @@ pub enum MongoAgentMethod { DeleteDocument, DeleteDocuments, RunCommand, + StartFindCursor, + FetchFindCursor, + CloseFindCursor, } impl MongoAgentMethod { - pub const ALL: [Self; 23] = [ + pub const ALL: [Self; 26] = [ Self::ListDatabases, Self::ListCollections, Self::FindDocuments, @@ -1410,6 +1416,9 @@ impl MongoAgentMethod { Self::DeleteDocument, Self::DeleteDocuments, Self::RunCommand, + Self::StartFindCursor, + Self::FetchFindCursor, + Self::CloseFindCursor, ]; pub fn as_str(self) -> &'static str { @@ -1437,6 +1446,9 @@ impl MongoAgentMethod { Self::DeleteDocument => "delete_document", Self::DeleteDocuments => "delete_documents", Self::RunCommand => "run_command", + Self::StartFindCursor => "start_find_cursor", + Self::FetchFindCursor => "fetch_find_cursor", + Self::CloseFindCursor => "close_find_cursor", } } } @@ -2795,6 +2807,27 @@ impl AgentDriverClient { self.call_mongo_method(MongoAgentMethod::InsertDocuments, params).await } + pub async fn mongo_start_find_cursor( + &mut self, + params: Value, + ) -> Result { + self.call_mongo_method(MongoAgentMethod::StartFindCursor, params).await + } + + pub async fn mongo_fetch_find_cursor( + &mut self, + params: Value, + ) -> Result { + self.call_mongo_method(MongoAgentMethod::FetchFindCursor, params).await + } + + pub async fn mongo_close_find_cursor( + &mut self, + params: Value, + ) -> Result { + self.call_mongo_method(MongoAgentMethod::CloseFindCursor, params).await + } + pub async fn mongo_update_document( &mut self, params: Value, @@ -2972,6 +3005,7 @@ pub fn agent_supports_capability(handshake: Option<&AgentHandshake>, capability: | AgentCapability::MongoRunCommand | AgentCapability::MongoInsertDocuments | AgentCapability::MongoReplaceDocument + | AgentCapability::MongoFindCursor ) { return handshake.map(|value| value.supports(capability)).unwrap_or(false); } @@ -5006,9 +5040,10 @@ for line in sys.stdin: assert_eq!(AgentCapability::MongoRunCommand.as_str(), "mongo_run_command"); assert_eq!(AgentCapability::MongoInsertDocuments.as_str(), "mongo_insert_documents"); assert_eq!(AgentCapability::MongoReplaceDocument.as_str(), "mongo_replace_document"); + assert_eq!(AgentCapability::MongoFindCursor.as_str(), "mongo_find_cursor"); assert_eq!(AgentCapability::MultiSession.as_str(), "multi_session"); assert_eq!(AgentCapability::StructuredErrorV1.as_str(), "structured_error_v1"); - assert_eq!(AgentCapability::ALL.len(), 25); + assert_eq!(AgentCapability::ALL.len(), 26); } #[test] @@ -5073,6 +5108,9 @@ for line in sys.stdin: assert_eq!(MongoAgentMethod::ReplaceDocument.as_str(), "replace_document"); assert_eq!(MongoAgentMethod::DeleteDocument.as_str(), "delete_document"); assert_eq!(MongoAgentMethod::DeleteDocuments.as_str(), "delete_documents"); + assert_eq!(MongoAgentMethod::StartFindCursor.as_str(), "start_find_cursor"); + assert_eq!(MongoAgentMethod::FetchFindCursor.as_str(), "fetch_find_cursor"); + assert_eq!(MongoAgentMethod::CloseFindCursor.as_str(), "close_find_cursor"); } #[test] @@ -5138,6 +5176,9 @@ for line in sys.stdin: let _mongo_clone_collection = AgentDriverClient::mongo_clone_collection::; let _mongo_drop_database = AgentDriverClient::mongo_drop_database::; let _mongo_insert_document = AgentDriverClient::mongo_insert_document::; + let _mongo_start_find_cursor = AgentDriverClient::mongo_start_find_cursor::; + let _mongo_fetch_find_cursor = AgentDriverClient::mongo_fetch_find_cursor::; + let _mongo_close_find_cursor = AgentDriverClient::mongo_close_find_cursor::; let _mongo_update_document = AgentDriverClient::mongo_update_document::; let _mongo_update_documents = AgentDriverClient::mongo_update_documents::; let _mongo_delete_document = AgentDriverClient::mongo_delete_document::; @@ -5389,6 +5430,8 @@ for line in sys.stdin: assert!(!agent_supports_capability(Some(&handshake), AgentCapability::MongoRunCommand)); assert!(!agent_supports_capability(None, AgentCapability::MongoInsertDocuments)); assert!(!agent_supports_capability(Some(&handshake), AgentCapability::MongoInsertDocuments)); + assert!(!agent_supports_capability(None, AgentCapability::MongoFindCursor)); + assert!(!agent_supports_capability(Some(&handshake), AgentCapability::MongoFindCursor)); let mongo_handshake = AgentHandshake { capabilities: vec![AgentCapability::MongoDropDatabase.as_str().to_string()], ..handshake }; @@ -5414,6 +5457,12 @@ for line in sys.stdin: Some(&mongo_insert_documents_handshake), AgentCapability::MongoInsertDocuments )); + + let mongo_find_cursor_handshake = AgentHandshake { + capabilities: vec![AgentCapability::MongoFindCursor.as_str().to_string()], + ..mongo_insert_documents_handshake + }; + assert!(agent_supports_capability(Some(&mongo_find_cursor_handshake), AgentCapability::MongoFindCursor)); } #[test] diff --git a/crates/dbx-core/src/db/mongo_driver.rs b/crates/dbx-core/src/db/mongo_driver.rs index bac813b5e1..e9a10ec807 100644 --- a/crates/dbx-core/src/db/mongo_driver.rs +++ b/crates/dbx-core/src/db/mongo_driver.rs @@ -2649,6 +2649,41 @@ fn is_retryable_mongo_write_code(code: i32) -> bool { !matches!(code, 11000 | 11001 | 12582) } +/// Reads the `insert_documents` result of a MongoDB legacy agent. The agent reports a partly +/// applied batch as a success carrying one entry per rejected document, so callers must inspect +/// [`MongoInsertOutcome::errors`] instead of trusting a bare affected-row count. +pub fn agent_insert_outcome(result: &serde_json::Value) -> Result { + let inserted = result + .get("affected_rows") + .and_then(serde_json::Value::as_u64) + .ok_or_else(|| "MongoDB Legacy Agent returned an invalid insertMany result".to_string())?; + let errors = result + .get("errors") + .and_then(serde_json::Value::as_array) + .map(|errors| errors.iter().map(agent_insert_error).collect()) + .unwrap_or_default(); + Ok(MongoInsertOutcome { inserted, errors }) +} + +/// One rejected document, or a batch-wide rejection the agent could not attribute to a document +/// (a write concern failure) — those carry no index and keep [`MongoBulkWriteError::index`] `None`. +fn agent_insert_error(value: &serde_json::Value) -> MongoBulkWriteError { + // A rejection is never dropped for lack of a message: dropping it would make a document the + // server refused look like a successful insert. + let message = value + .get("message") + .and_then(serde_json::Value::as_str) + .unwrap_or("MongoDB Legacy Agent rejected a document without a message") + .to_string(); + let code = value.get("code").and_then(serde_json::Value::as_i64).map(|code| code as i32); + MongoBulkWriteError { + message, + index: value.get("index").and_then(serde_json::Value::as_u64).map(|index| index as usize), + code, + retryable: code.is_some_and(is_retryable_mongo_write_code), + } +} + #[allow(clippy::too_many_arguments)] pub async fn for_each_find_document( client: &Client, diff --git a/crates/dbx-core/src/mongo_ops.rs b/crates/dbx-core/src/mongo_ops.rs index 0b9c34adc4..473c6529f1 100644 --- a/crates/dbx-core/src/mongo_ops.rs +++ b/crates/dbx-core/src/mongo_ops.rs @@ -439,11 +439,18 @@ pub async fn mongo_find_documents_extended_json_core( } } -fn is_unknown_agent_method_error(error: &str, method: &str) -> bool { +pub(crate) fn is_unknown_agent_method_error(error: &str, method: &str) -> bool { let lower = error.to_ascii_lowercase(); lower.contains(method) && (lower.contains("unknown method") || lower.contains("method not found")) } +/// True only when a legacy agent refused a bulk write because it has no `insertMany` at all, which +/// is the one failure a document-by-document retry can still get through. It must stay precise: +/// every other legacy-agent failure, a rejected document included, has to surface as-is. +pub(crate) fn is_legacy_agent_insert_many_unsupported(error: &str) -> bool { + error.contains("does not support insertMany") || error.contains("does not support bulk insertMany") +} + pub async fn mongo_aggregate_documents_core( state: &AppState, connection_id: &str, @@ -730,10 +737,18 @@ pub async fn mongo_insert_documents_core( "docs_json": docs_json, })) .await?; - result - .get("affected_rows") - .and_then(serde_json::Value::as_u64) - .ok_or_else(|| "MongoDB Legacy Agent returned an invalid insertMany result".to_string()) + let outcome = mongo_driver::agent_insert_outcome(&result)?; + // The agent reports a partly applied batch as a success, so a rejected document has to + // fail the call here exactly like the native driver's insert_many does. + if let Some(first) = outcome.errors.first() { + return Err(format!( + "MongoDB Legacy Agent rejected {} of {} documents: {}", + outcome.errors.len(), + documents.len(), + first.message + )); + } + Ok(outcome.inserted) } _ => Err("Not a MongoDB connection".to_string()), } @@ -1758,6 +1773,46 @@ for line in sys.stdin: assert_eq!(affected, 2); } + #[cfg(unix)] + #[tokio::test] + async fn mongo_insert_many_fails_when_the_legacy_agent_rejects_a_document() { + let documents = r#"[{"_id":1},{"_id":1}]"#; + let (state, _directory) = legacy_mongo_state_with_capabilities( + "insert_documents", + serde_json::json!({ + "database": "app", + "collection": "users", + "docs_json": documents, + }), + serde_json::json!({ + "affected_rows": 1, + "errors": [{ "index": 1, "code": 11000, "message": "E11000 duplicate key" }], + }), + &["mongo_insert_documents"], + ) + .await; + + let error = mongo_insert_documents_core(&state, "legacy", "app", "users", documents).await.unwrap_err(); + + // The agent reports a partly applied batch as a success, so a rejected document has to + // fail the call here rather than being rounded up to "inserted". + assert_eq!(error, "MongoDB Legacy Agent rejected 1 of 2 documents: E11000 duplicate key"); + } + + #[test] + fn legacy_insert_many_heuristic_covers_only_the_missing_capability() { + assert!(is_legacy_agent_insert_many_unsupported( + "MongoDB Legacy Agent does not support insertMany; upgrade or reinstall the MongoDB Legacy driver" + )); + assert!(is_legacy_agent_insert_many_unsupported( + "MongoDB legacy agent does not support bulk insertMany/insertOne writes" + )); + // A rejected document must not be retried one by one: the rest of the batch is written. + assert!(!is_legacy_agent_insert_many_unsupported( + "MongoDB Legacy Agent rejected 1 of 2 documents: E11000 duplicate key" + )); + } + #[cfg(unix)] #[tokio::test] async fn mongo_insert_many_requires_an_explicit_legacy_agent_capability() { diff --git a/crates/dbx-core/src/mongodb_import_export.rs b/crates/dbx-core/src/mongodb_import_export.rs index 9e1d31b038..57fbbb204d 100644 --- a/crates/dbx-core/src/mongodb_import_export.rs +++ b/crates/dbx-core/src/mongodb_import_export.rs @@ -12,10 +12,10 @@ use serde::{Deserialize, Serialize}; use crate::connection::{task_client_session_id, AppState, PoolKind}; use crate::csv_export::{push_csv_field, CsvQuoteMode}; -use crate::db::agent_driver::AgentCapability; +use crate::db::agent_driver::{AgentCapability, PooledAgentClient}; use crate::db::mongo_driver::{ self, document_to_canonical_extended_json, for_each_find_document, insert_bson_documents, - json_object_to_document_extended_json, MongoBulkWriteError, MongoInsertOutcome, + json_object_to_document_extended_json, MongoBulkWriteError, MongoDocumentResult, MongoInsertOutcome, }; use crate::table_import::{open_transcoded_text_file, TableImportTextEncoding}; @@ -1569,7 +1569,6 @@ async fn insert_documents_batch( database: &str, collection: &str, documents: Vec, - require_bson_types: bool, ) -> Result { let pool = state.pool_handle(connection_id).await.ok_or_else(|| MongoImportIssue::new("CONNECTION", "Not found"))?; @@ -1578,12 +1577,6 @@ async fn insert_documents_batch( insert_bson_documents(client, database, collection, documents).await.map_err(bulk_write_issue) } PoolKind::Agent(client) => { - if require_bson_types { - return Err(MongoImportIssue::new( - "LEGACY_AGENT", - "MongoDB Legacy Agent cannot preserve BSON types during file import; use the native MongoDB driver", - )); - } let mut client = client.lock().await; if !client.supports_capability(AgentCapability::MongoInsertDocuments) { return Err(MongoImportIssue::new( @@ -1599,18 +1592,20 @@ async fn insert_documents_batch( "database": database, "collection": collection, "docs_json": docs_json, + "ordered": false, })) .await .map_err(|error| MongoImportIssue::new("PERMISSION", error).retryable())?; - let inserted = result.get("affected_rows").and_then(serde_json::Value::as_u64).ok_or_else(|| { - MongoImportIssue::new("CONNECTION", "MongoDB Legacy Agent returned an invalid insertMany result") - })?; - Ok(MongoInsertOutcome { inserted, errors: Vec::new() }) + agent_insert_outcome(&result) } _ => Err(MongoImportIssue::new("CONNECTION", "Not a MongoDB connection")), } } +fn agent_insert_outcome(result: &serde_json::Value) -> Result { + mongo_driver::agent_insert_outcome(result).map_err(|error| MongoImportIssue::new("CONNECTION", error)) +} + /// Rewrites a write issue's batch-relative index into the source file row it came from, so the /// user can find the offending record. fn located_in_batch(mut issue: MongoImportIssue, rows: &[u64], batch: u64) -> MongoImportIssue { @@ -1644,7 +1639,6 @@ where { let started_at = Instant::now(); let batch_size = clamp_batch_size(if request.batch_size == 0 { DEFAULT_BATCH_SIZE } else { request.batch_size })?; - let require_bson_types = !matches!(request.parse_options.type_mode(), MongoImportTypeMode::String); on_progress(progress( &request.import_id, MongoImportPhase::Preparing, @@ -1788,7 +1782,6 @@ where &request.database, &request.collection, documents, - require_bson_types, ) .await { @@ -2236,28 +2229,12 @@ where state.get_or_create_pool(&request.connection_id, Some(&request.database)).await?; let pool = state.pool_handle(&request.connection_id).await.ok_or_else(|| "Not found".to_string())?; - let client = match &pool { - PoolKind::MongoDb(client) => client.clone(), - PoolKind::Agent(_) => { - return Err( - "MongoDB Legacy Agent does not support cursor export of the full query; use the native MongoDB driver" - .to_string(), - ); - } + match &pool { + PoolKind::MongoDb(_) | PoolKind::Agent(_) => {} _ => return Err("Not a MongoDB connection".to_string()), - }; + } - let total_documents = if request - .filter - .as_deref() - .is_none_or(|filter| filter.trim().is_empty() || filter.trim() == "{}") - { - mongo_driver::count_documents(&client, &request.database, &request.collection, request.filter.as_deref(), false) - .await - .ok() - } else { - None - }; + let total_documents = count_export_documents(state, request).await; let target = PathBuf::from(&request.file_path); if let Some(parent) = target.parent() { @@ -2268,11 +2245,10 @@ where let result = match request.format { MongoExportFormat::Ndjson => { - export_ndjson(&client, request, &temp, total_documents, started_at, &mut is_cancelled, &mut on_progress) - .await + export_ndjson(state, request, &temp, total_documents, started_at, &mut is_cancelled, &mut on_progress).await } MongoExportFormat::Csv => { - export_csv(&client, request, &temp, total_documents, started_at, &mut is_cancelled, &mut on_progress).await + export_csv(state, request, &temp, total_documents, started_at, &mut is_cancelled, &mut on_progress).await } }; @@ -2325,8 +2301,349 @@ where } } +async fn count_export_documents(state: &AppState, request: &MongoExportRequest) -> Option { + if request.filter.as_deref().is_some_and(|filter| !filter.trim().is_empty() && filter.trim() != "{}") { + return None; + } + let pool = state.pool_handle(&request.connection_id).await?; + match pool { + PoolKind::MongoDb(client) => mongo_driver::count_documents( + &client, + &request.database, + &request.collection, + request.filter.as_deref(), + false, + ) + .await + .ok(), + PoolKind::Agent(client) => { + let mut client = client.lock().await; + client + .mongo_count_documents::(serde_json::json!({ + "database": request.database, + "collection": request.collection, + "filter": request.filter, + "accurate": false, + })) + .await + .ok() + } + _ => None, + } +} + +async fn for_each_export_json_document( + state: &AppState, + request: &MongoExportRequest, + is_cancelled: &mut C, + mut on_document: F, +) -> Result<(), String> +where + C: FnMut(&str) -> std::pin::Pin + Send>>, + F: FnMut(serde_json::Value) -> Result<(), String>, +{ + let pool = state.pool_handle(&request.connection_id).await.ok_or_else(|| "Not found".to_string())?; + match pool { + PoolKind::MongoDb(client) => { + for_each_find_document( + &client, + &request.database, + &request.collection, + request.filter.as_deref(), + request.projection.as_deref(), + request.sort.as_deref(), + request.collation.as_deref(), + DEFAULT_EXPORT_BATCH_SIZE, + |document| on_document(document_to_canonical_extended_json(&document)), + ) + .await + } + PoolKind::Agent(client) => for_each_agent_export_document(&client, request, is_cancelled, on_document).await, + _ => Err("Not a MongoDB connection".to_string()), + } +} + +async fn for_each_agent_export_document( + client: &PooledAgentClient, + request: &MongoExportRequest, + is_cancelled: &mut C, + mut on_document: F, +) -> Result<(), String> +where + C: FnMut(&str) -> std::pin::Pin + Send>>, + F: FnMut(serde_json::Value) -> Result<(), String>, +{ + // The lock is taken per RPC, not for the whole export: a long export must not block every + // other operation that shares this pooled agent connection. + let supports_cursor = client.lock().await.supports_capability(AgentCapability::MongoFindCursor); + if supports_cursor { + export_agent_find_cursor(client, request, is_cancelled, &mut on_document).await + } else { + export_agent_find_pages(client, request, is_cancelled, &mut on_document).await + } +} + +/// The caller's filter as a JSON object when the export can page by `_id`, or `None` when it has +/// to keep offset paging: an explicit order, a collation, or a projection that drops `_id` all +/// make the `_id` keyset unusable. +fn agent_keyset_base_filter(request: &MongoExportRequest) -> Option> { + if request.sort.is_some() || request.collation.is_some() || !projection_keeps_id(request.projection.as_deref()) { + return None; + } + match request.filter.as_deref() { + None => Some(serde_json::Map::new()), + Some(filter) => { + let trimmed = filter.trim(); + if trimmed.is_empty() || trimmed == "{}" { + return Some(serde_json::Map::new()); + } + match serde_json::from_str::(trimmed) { + Ok(serde_json::Value::Object(object)) => Some(object), + _ => None, + } + } + } +} + +/// A keyset export has to read each page's last `_id`, so an unparseable projection counts as +/// unsafe and keeps the export on offset paging. +fn projection_keeps_id(projection: Option<&str>) -> bool { + let Some(projection) = projection else { + return true; + }; + let Ok(serde_json::Value::Object(object)) = serde_json::from_str::(projection) else { + return false; + }; + match object.get("_id") { + Some(serde_json::Value::Bool(false)) => false, + Some(serde_json::Value::Number(excluded)) => excluded.as_i64() != Some(0), + _ => true, + } +} + +/// Builds one `_id`-keyset page request. The keyset is ANDed with the caller's filter rather than +/// merged into it, so a filter that constrains `_id` itself keeps its own bounds. +fn agent_keyset_find_params( + request: &MongoExportRequest, + base: &serde_json::Map, + last_id: Option<&serde_json::Value>, +) -> serde_json::Value { + let filter = match last_id { + None => serde_json::Value::Object(base.clone()), + Some(last_id) if base.is_empty() => serde_json::json!({ "_id": { "$gt": last_id } }), + Some(last_id) => { + serde_json::json!({ "$and": [serde_json::Value::Object(base.clone()), { "_id": { "$gt": last_id } }] }) + } + }; + let mut params = serde_json::json!({ + "database": request.database, + "collection": request.collection, + "skip": 0, + "limit": DEFAULT_EXPORT_BATCH_SIZE, + "filter": filter.to_string(), + "sort": "{\"_id\":1}", + "batch_size": DEFAULT_EXPORT_BATCH_SIZE, + }); + if let Some(projection) = &request.projection { + params["projection"] = serde_json::json!(projection); + } + params +} + +fn agent_find_params(request: &MongoExportRequest, skip: u64, limit: u32) -> serde_json::Value { + let mut params = serde_json::json!({ + "database": request.database, + "collection": request.collection, + "skip": skip, + "limit": limit, + "filter": request.filter, + "sort": request.sort, + "batch_size": limit, + }); + if let Some(projection) = &request.projection { + params["projection"] = serde_json::json!(projection); + } + if let Some(collation) = &request.collation { + params["collation"] = serde_json::json!(collation); + } + params +} + +/// Cursor start request. `start_find_cursor` pages server-side, so it neither needs nor honors +/// the offset paging's `skip`/`limit`; sending them would invite a silent split brain the day a +/// caller-facing limit exists. +fn agent_cursor_find_params(request: &MongoExportRequest) -> serde_json::Value { + let mut params = serde_json::json!({ + "database": request.database, + "collection": request.collection, + "filter": request.filter, + "sort": request.sort, + "batch_size": DEFAULT_EXPORT_BATCH_SIZE, + }); + if let Some(projection) = &request.projection { + params["projection"] = serde_json::json!(projection); + } + if let Some(collation) = &request.collation { + params["collation"] = serde_json::json!(collation); + } + params +} + +async fn export_agent_find_cursor( + client: &PooledAgentClient, + request: &MongoExportRequest, + is_cancelled: &mut C, + on_document: &mut F, +) -> Result<(), String> +where + C: FnMut(&str) -> std::pin::Pin + Send>>, + F: FnMut(serde_json::Value) -> Result<(), String>, +{ + if is_cancelled(&request.export_id).await { + return Err("Export cancelled".to_string()); + } + let started: serde_json::Value = { + let mut guard = client.lock().await; + match guard.mongo_start_find_cursor(agent_cursor_find_params(request)).await { + Ok(started) => started, + Err(error) if crate::mongo_ops::is_unknown_agent_method_error(&error, "start_find_cursor") => { + return export_agent_find_pages(client, request, is_cancelled, on_document).await; + } + Err(error) => return Err(error), + } + }; + let cursor_id = started + .get("cursor_id") + .and_then(serde_json::Value::as_str) + .ok_or_else(|| "MongoDB Legacy Agent returned an invalid find cursor".to_string())? + .to_string(); + let result = fetch_agent_find_cursor(client, request, &cursor_id, is_cancelled, on_document).await; + let _: Result = { + let mut guard = client.lock().await; + guard.mongo_close_find_cursor(serde_json::json!({ "cursor_id": cursor_id })).await + }; + result +} + +async fn fetch_agent_find_cursor( + client: &PooledAgentClient, + request: &MongoExportRequest, + cursor_id: &str, + is_cancelled: &mut C, + on_document: &mut F, +) -> Result<(), String> +where + C: FnMut(&str) -> std::pin::Pin + Send>>, + F: FnMut(serde_json::Value) -> Result<(), String>, +{ + loop { + if is_cancelled(&request.export_id).await { + return Err("Export cancelled".to_string()); + } + let mut page: serde_json::Value = { + let mut guard = client.lock().await; + guard + .mongo_fetch_find_cursor(serde_json::json!({ + "cursor_id": cursor_id, + "limit": DEFAULT_EXPORT_BATCH_SIZE, + })) + .await? + }; + let documents = page + .get_mut("documents") + .and_then(serde_json::Value::as_array_mut) + .map(std::mem::take) + .ok_or_else(|| "MongoDB Legacy Agent returned an invalid find cursor page".to_string())?; + let exhausted = page.get("exhausted").and_then(serde_json::Value::as_bool).unwrap_or(documents.is_empty()); + let empty = documents.is_empty(); + for document in documents { + on_document(document)?; + } + if exhausted || empty { + return Ok(()); + } + } +} + +async fn export_agent_find_pages( + client: &PooledAgentClient, + request: &MongoExportRequest, + is_cancelled: &mut C, + on_document: &mut F, +) -> Result<(), String> +where + C: FnMut(&str) -> std::pin::Pin + Send>>, + F: FnMut(serde_json::Value) -> Result<(), String>, +{ + let keyset_base = agent_keyset_base_filter(request); + // Offset paging costs the server one scan of every skipped document per page, so it is only + // used when the export needs the caller's own ordering; `_id` keyset paging is linear and + // cannot duplicate or drop documents when the collection changes mid-export. Neither order is + // the natural order an agent with cursors returns, but both are "unspecified" to the caller. + let mut skip = 0u64; + let mut last_id: Option = None; + loop { + if is_cancelled(&request.export_id).await { + return Err("Export cancelled".to_string()); + } + let params = match &keyset_base { + Some(base) => agent_keyset_find_params(request, base, last_id.as_ref()), + None => agent_find_params(request, skip, DEFAULT_EXPORT_BATCH_SIZE), + }; + let mut page: MongoDocumentResult = { + let mut guard = client.lock().await; + match guard.mongo_find_documents_extended_json(params).await { + Ok(page) => page, + Err(error) + if crate::mongo_ops::is_unknown_agent_method_error(&error, "find_documents_extended_json") => + { + return Err( + "MongoDB Legacy Agent does not support type-preserving export; upgrade or reinstall the MongoDB Legacy driver" + .to_string(), + ); + } + Err(error) => return Err(error), + } + }; + // The shipped legacy agent answers this method with `documentQueryResult`, so + // `extended_documents` is normally absent and `documents` (relaxed Extended JSON, which + // keeps ObjectId/date/binary but not integer widths) is what gets exported. + let documents = match page.extended_documents.take() { + Some(extended) if extended.len() == page.documents.len() => extended, + _ => std::mem::take(&mut page.documents), + }; + if documents.is_empty() { + return Ok(()); + } + let count = documents.len() as u64; + if keyset_base.is_some() { + let Some(next_id) = documents.last().and_then(|document| document.get("_id")).cloned() else { + return Err("MongoDB Legacy Agent returned a document without _id".to_string()); + }; + // A page that does not move the keyset forward would be re-requested forever, which + // offset paging cannot hit because it always advances `skip`. + if last_id.as_ref() == Some(&next_id) { + return Err( + "MongoDB Legacy Agent returned the same export page twice; upgrade or reinstall the MongoDB Legacy driver" + .to_string(), + ); + } + last_id = Some(next_id); + } + for document in documents { + on_document(document)?; + } + if count < u64::from(DEFAULT_EXPORT_BATCH_SIZE) { + return Ok(()); + } + if keyset_base.is_none() { + skip = skip.saturating_add(count); + } + } +} + async fn export_ndjson( - client: &mongodb::Client, + state: &AppState, request: &MongoExportRequest, temp: &Path, total_documents: Option, @@ -2342,35 +2659,24 @@ where let mut writer = BufWriter::new(file); let mut documents_read = 0u64; let mut bytes_written = 0u64; - for_each_find_document( - client, - &request.database, - &request.collection, - request.filter.as_deref(), - request.projection.as_deref(), - request.sort.as_deref(), - request.collation.as_deref(), - DEFAULT_EXPORT_BATCH_SIZE, - |document| { - let json = document_to_canonical_extended_json(&document); - let mut line = json.to_string(); - line.push('\n'); - bytes_written += write_export_line(&mut writer, &line)?; - documents_read += 1; - if documents_read == 1 || documents_read.is_multiple_of(500) { - on_progress(export_progress( - &request.export_id, - MongoExportStatus::Running, - documents_read, - bytes_written, - total_documents, - None, - started_at, - )); - } - Ok(()) - }, - ) + for_each_export_json_document(state, request, is_cancelled, |json| { + let mut line = json.to_string(); + line.push('\n'); + bytes_written += write_export_line(&mut writer, &line)?; + documents_read += 1; + if documents_read == 1 || documents_read.is_multiple_of(500) { + on_progress(export_progress( + &request.export_id, + MongoExportStatus::Running, + documents_read, + bytes_written, + total_documents, + None, + started_at, + )); + } + Ok(()) + }) .await?; writer.flush().map_err(|error| error.to_string())?; if is_cancelled(&request.export_id).await { @@ -2382,7 +2688,7 @@ where const CSV_FIELD_DISCOVERY_DOCS: usize = 10_000; async fn export_csv( - client: &mongodb::Client, + state: &AppState, request: &MongoExportRequest, temp: &Path, total_documents: Option, @@ -2406,47 +2712,20 @@ where let mut documents_read = 0u64; let mut bytes_written = 0u64; - for_each_find_document( - client, - &request.database, - &request.collection, - request.filter.as_deref(), - request.projection.as_deref(), - request.sort.as_deref(), - request.collation.as_deref(), - DEFAULT_EXPORT_BATCH_SIZE, - |document| { - if !header_ready { - let json = document_to_canonical_extended_json(&document); - collect_csv_fields(&json, &mut fields, &mut seen)?; - buffered.push(json); - if buffered.len() >= CSV_FIELD_DISCOVERY_DOCS { - write_csv_header_and_buffer( - request.include_header, - &mut fields, - &mut buffered, - &mut writer, - &mut bytes_written, - &mut documents_read, - )?; - header_ready = true; - on_progress(export_progress( - &request.export_id, - MongoExportStatus::Running, - documents_read, - bytes_written, - total_documents, - None, - started_at, - )); - } - return Ok(()); - } - let json = document_to_canonical_extended_json(&document); - let line = format_csv_document_line(&fields, &json); - bytes_written += write_export_line(&mut writer, &line)?; - documents_read += 1; - if documents_read.is_multiple_of(500) { + for_each_export_json_document(state, request, is_cancelled, |json| { + if !header_ready { + collect_csv_fields(&json, &mut fields, &mut seen)?; + buffered.push(json); + if buffered.len() >= CSV_FIELD_DISCOVERY_DOCS { + write_csv_header_and_buffer( + request.include_header, + &mut fields, + &mut buffered, + &mut writer, + &mut bytes_written, + &mut documents_read, + )?; + header_ready = true; on_progress(export_progress( &request.export_id, MongoExportStatus::Running, @@ -2457,9 +2736,24 @@ where started_at, )); } - Ok(()) - }, - ) + return Ok(()); + } + let line = format_csv_document_line(&fields, &json); + bytes_written += write_export_line(&mut writer, &line)?; + documents_read += 1; + if documents_read.is_multiple_of(500) { + on_progress(export_progress( + &request.export_id, + MongoExportStatus::Running, + documents_read, + bytes_written, + total_documents, + None, + started_at, + )); + } + Ok(()) + }) .await?; if !header_ready { write_csv_header_and_buffer( @@ -3317,4 +3611,546 @@ mod tests { let error = csv_fields_from_extended_documents(&[serde_json::Value::Object(object)]).unwrap_err(); assert!(error.contains("NDJSON")); } + + #[test] + fn agent_insert_outcome_maps_partial_failures() { + let outcome = agent_insert_outcome(&serde_json::json!({ + "affected_rows": 1, + "errors": [{ "index": 1, "code": 11000, "message": "E11000 duplicate key" }] + })) + .unwrap(); + assert_eq!(outcome.inserted, 1); + assert_eq!(outcome.errors.len(), 1); + assert_eq!(outcome.errors[0].index, Some(1)); + assert_eq!(outcome.errors[0].code, Some(11000)); + assert!(!outcome.errors[0].retryable); + } + + #[test] + fn agent_insert_outcome_keeps_batch_wide_and_message_less_rejections() { + // A write concern failure is reported without a document index, and a rejection without a + // message must still count as a failure rather than disappear. + let outcome = agent_insert_outcome(&serde_json::json!({ + "affected_rows": 0, + "errors": [{ "code": 64, "message": "waiting for replication timed out" }, {}] + })) + .unwrap(); + assert_eq!(outcome.inserted, 0); + assert_eq!(outcome.errors.len(), 2); + assert_eq!(outcome.errors[0].index, None); + assert_eq!(outcome.errors[0].code, Some(64)); + assert!(outcome.errors[0].retryable); + assert!(outcome.errors[1].message.contains("without a message"), "{:?}", outcome.errors[1]); + } + + #[cfg(unix)] + async fn legacy_mongo_import_export_state( + script: &str, + capabilities: &[&str], + ) -> (crate::connection::AppState, tempfile::TempDir) { + use std::io::Write; + + use crate::db::agent_driver::{AgentDriverClient, AgentLaunchSpec}; + use crate::models::connection::ConnectionConfig; + use crate::storage::Storage; + + let directory = tempfile::tempdir().unwrap(); + let mut file = tempfile::NamedTempFile::new_in(directory.path()).unwrap(); + let capabilities = serde_json::to_string(capabilities).unwrap(); + write!( + file, + r#"import json +import sys + +CAPABILITIES = {capabilities} +{script} +print(json.dumps({{"ready": True}}), flush=True) +for line in sys.stdin: + request = json.loads(line) + result = handle(request) + if isinstance(result, dict) and "__rpc_error" in result: + print(json.dumps({{"jsonrpc": "2.0", "id": request["id"], "error": {{"code": -1, "message": result["__rpc_error"]}}}}), flush=True) + elif result is None: + print(json.dumps({{"jsonrpc": "2.0", "id": request["id"], "error": {{"code": -1, "message": "unexpected MongoDB RPC"}}}}), flush=True) + else: + print(json.dumps({{"jsonrpc": "2.0", "id": request["id"], "result": result}}), flush=True) +"# + ) + .unwrap(); + file.flush().unwrap(); + let (_, script_path) = file.keep().unwrap(); + let mut client = AgentDriverClient::spawn( + AgentLaunchSpec::new("python3").with_args([script_path.to_string_lossy().to_string()]), + ) + .await + .unwrap(); + client.try_optional_handshake("test").await.unwrap(); + let storage = Storage::open(&directory.path().join("storage.db")).await.unwrap(); + let state = crate::connection::AppState::new(storage); + let config: ConnectionConfig = serde_json::from_value(serde_json::json!({ + "id": "legacy", + "name": "Legacy MongoDB", + "db_type": "mongodb", + "driver_profile": "mongodb-legacy", + "host": "localhost", + "port": 27017, + "username": "", + "password": "", + "database": null, + })) + .unwrap(); + state.configs.write().await.insert("legacy".to_string(), config); + let pool = PoolKind::agent(client); + state + .update_connection_pools(|connections| { + connections.insert("legacy".to_string(), pool.clone()); + connections.insert("legacy:app".to_string(), pool); + }) + .await; + (state, directory) + } + + #[cfg(unix)] + #[tokio::test] + async fn legacy_agent_imports_extended_json_with_unordered_insert_many() { + let source_dir = std::env::temp_dir().join(format!("dbx-mongo-legacy-import-{}", uuid::Uuid::new_v4())); + std::fs::create_dir_all(&source_dir).unwrap(); + let path = source_dir.join("data.ndjson"); + std::fs::write(&path, "{\"_id\":{\"$oid\":\"507f1f77bcf86cd799439011\"},\"name\":\"Ada\"}\n").unwrap(); + + let (state, _directory) = legacy_mongo_import_export_state( + r#" +def handle(request): + method = request.get("method") + if method == "handshake": + return {"protocolVersion": 1, "agentProtocolVersion": 1, "capabilities": CAPABILITIES} + if method in ("validate_connection", "connect"): + return {"ok": True} + if method == "insert_documents": + params = request.get("params") or {} + if params.get("ordered") is not False: + return None + docs = json.loads(params.get("docs_json") or "[]") + if not docs or docs[0].get("_id", {}).get("$oid") != "507f1f77bcf86cd799439011": + return None + return {"affected_rows": 1, "errors": []} + return None +"#, + &["mongo_insert_documents"], + ) + .await; + + let summary = import_mongodb_file_core( + &state, + &MongoImportRequest { + import_id: "import-1".to_string(), + connection_id: "legacy".to_string(), + database: "app".to_string(), + collection: "users".to_string(), + file_path: path.to_string_lossy().to_string(), + source_ref: None, + format: MongoImportFormat::Ndjson, + parse_options: MongoImportParseOptions { + type_mode: Some(MongoImportTypeMode::ExtendedJson), + ..MongoImportParseOptions::default() + }, + batch_size: 500, + execution_id: None, + }, + |_| Box::pin(async { false }), + |_| {}, + ) + .await + .unwrap(); + assert_eq!(summary.rows_inserted, 1); + assert_eq!(summary.rows_failed, 0); + let _ = std::fs::remove_dir_all(source_dir); + } + + #[cfg(unix)] + #[tokio::test] + async fn legacy_agent_exports_ndjson_through_find_cursor() { + let target = std::env::temp_dir().join(format!("dbx-mongo-legacy-export-{}.ndjson", uuid::Uuid::new_v4())); + let (state, _directory) = legacy_mongo_import_export_state( + r#" +def handle(request): + method = request.get("method") + if method == "handshake": + return {"protocolVersion": 1, "agentProtocolVersion": 1, "capabilities": CAPABILITIES} + if method in ("validate_connection", "connect"): + return {"ok": True} + if method == "count_documents": + return 1 + if method == "start_find_cursor": + return {"cursor_id": "c1", "batch_size": 1000} + if method == "fetch_find_cursor": + return {"documents": [{"_id": {"$oid": "507f1f77bcf86cd799439011"}, "name": "Ada"}], "exhausted": True} + if method == "close_find_cursor": + return {"ok": True} + return None +"#, + &["mongo_find_cursor"], + ) + .await; + + let summary = export_mongodb_query_core( + &state, + &MongoExportRequest { + export_id: "export-1".to_string(), + connection_id: "legacy".to_string(), + database: "app".to_string(), + collection: "users".to_string(), + filter: None, + sort: None, + projection: None, + collation: None, + format: MongoExportFormat::Ndjson, + include_header: true, + file_path: target.to_string_lossy().to_string(), + execution_id: None, + }, + |_| Box::pin(async { false }), + |_| {}, + ) + .await + .unwrap(); + let body = std::fs::read_to_string(&target).unwrap(); + let _ = std::fs::remove_file(&target); + assert_eq!(summary.documents_exported, 1); + assert!(body.contains("507f1f77bcf86cd799439011")); + assert!(body.contains("Ada")); + } + + #[cfg(unix)] + #[tokio::test] + async fn legacy_agent_export_falls_back_to_paged_find_without_cursor_capability() { + let target = std::env::temp_dir().join(format!("dbx-mongo-legacy-export-page-{}.ndjson", uuid::Uuid::new_v4())); + let (state, _directory) = legacy_mongo_import_export_state( + r#" +def handle(request): + method = request.get("method") + if method == "handshake": + return {"protocolVersion": 1, "agentProtocolVersion": 1, "capabilities": CAPABILITIES} + if method in ("validate_connection", "connect"): + return {"ok": True} + if method == "count_documents": + return 1 + if method == "find_documents_extended_json": + # The shipped legacy agent answers with documentQueryResult: relaxed documents only. + return {"documents": [{"_id": {"$oid": "507f1f77bcf86cd799439011"}, "name": "Ada"}], "total": 1} + return None +"#, + &["mongo_insert_documents"], + ) + .await; + + let summary = export_mongodb_query_core( + &state, + &MongoExportRequest { + export_id: "export-2".to_string(), + connection_id: "legacy".to_string(), + database: "app".to_string(), + collection: "users".to_string(), + filter: None, + sort: None, + projection: None, + collation: None, + format: MongoExportFormat::Ndjson, + include_header: true, + file_path: target.to_string_lossy().to_string(), + execution_id: None, + }, + |_| Box::pin(async { false }), + |_| {}, + ) + .await + .unwrap(); + let body = std::fs::read_to_string(&target).unwrap(); + let _ = std::fs::remove_file(&target); + assert_eq!(summary.documents_exported, 1); + assert!(body.contains("Ada")); + } + + #[cfg(unix)] + #[tokio::test] + async fn legacy_agent_export_pages_by_id_without_skipping() { + let target = + std::env::temp_dir().join(format!("dbx-mongo-legacy-export-keyset-{}.ndjson", uuid::Uuid::new_v4())); + let (state, _directory) = legacy_mongo_import_export_state( + r#" +PAGE = {"index": 0} + +def page_documents(request): + params = request.get("params") or {} + if params.get("skip") not in (0, None): + return {"__rpc_error": "keyset paging must not use skip"} + if params.get("sort") != '{"_id":1}': + return {"__rpc_error": "keyset paging must sort by _id, got " + str(params.get("sort"))} + index = PAGE["index"] + PAGE["index"] = index + 1 + if index == 0: + return {"documents": [{"_id": value} for value in range(1000)], "total": 2001} + if index == 1: + keyset = params.get("filter") or "" + if "$gt" not in keyset or "999" not in keyset: + return {"__rpc_error": "second page must continue after _id 999, got " + keyset} + return {"documents": [{"_id": value} for value in range(1000, 2000)], "total": 2001} + if index == 2: + return {"documents": [{"_id": 2000}], "total": 2001} + return {"__rpc_error": "unexpected extra page"} + +def handle(request): + method = request.get("method") + if method == "handshake": + return {"protocolVersion": 1, "agentProtocolVersion": 1, "capabilities": CAPABILITIES} + if method in ("validate_connection", "connect"): + return {"ok": True} + if method == "count_documents": + return 2001 + if method == "find_documents_extended_json": + return page_documents(request) + return None +"#, + &["mongo_insert_documents"], + ) + .await; + + let summary = export_mongodb_query_core( + &state, + &MongoExportRequest { + export_id: "export-keyset".to_string(), + connection_id: "legacy".to_string(), + database: "app".to_string(), + collection: "users".to_string(), + filter: None, + sort: None, + projection: None, + collation: None, + format: MongoExportFormat::Ndjson, + include_header: true, + file_path: target.to_string_lossy().to_string(), + execution_id: None, + }, + |_| Box::pin(async { false }), + |_| {}, + ) + .await + .unwrap(); + let body = std::fs::read_to_string(&target).unwrap(); + let _ = std::fs::remove_file(&target); + assert_eq!(summary.documents_exported, 2001); + assert_eq!(body.lines().count(), 2001); + } + + #[cfg(unix)] + #[tokio::test] + async fn legacy_agent_export_stops_when_a_keyset_page_repeats() { + let target = + std::env::temp_dir().join(format!("dbx-mongo-legacy-export-stall-{}.ndjson", uuid::Uuid::new_v4())); + let (state, _directory) = legacy_mongo_import_export_state( + r#" +def handle(request): + method = request.get("method") + if method == "handshake": + return {"protocolVersion": 1, "agentProtocolVersion": 1, "capabilities": CAPABILITIES} + if method in ("validate_connection", "connect"): + return {"ok": True} + if method == "count_documents": + return 2000 + if method == "find_documents_extended_json": + # An agent that ignores the keyset filter would otherwise be asked for this page forever. + return {"documents": [{"_id": value} for value in range(1000)], "total": 2000} + return None +"#, + &["mongo_insert_documents"], + ) + .await; + + let error = export_mongodb_query_core( + &state, + &MongoExportRequest { + export_id: "export-stall".to_string(), + connection_id: "legacy".to_string(), + database: "app".to_string(), + collection: "users".to_string(), + filter: None, + sort: None, + projection: None, + collation: None, + format: MongoExportFormat::Ndjson, + include_header: true, + file_path: target.to_string_lossy().to_string(), + execution_id: None, + }, + |_| Box::pin(async { false }), + |_| {}, + ) + .await + .unwrap_err(); + let _ = std::fs::remove_file(&target); + + assert!(error.contains("same export page twice"), "{error}"); + } + + #[cfg(unix)] + #[tokio::test] + async fn legacy_agent_export_keeps_offset_paging_for_an_explicit_sort() { + let target = + std::env::temp_dir().join(format!("dbx-mongo-legacy-export-offset-{}.ndjson", uuid::Uuid::new_v4())); + let (state, _directory) = legacy_mongo_import_export_state( + r#" +def handle(request): + method = request.get("method") + if method == "handshake": + return {"protocolVersion": 1, "agentProtocolVersion": 1, "capabilities": CAPABILITIES} + if method in ("validate_connection", "connect"): + return {"ok": True} + if method == "count_documents": + return 1001 + if method == "find_documents_extended_json": + params = request.get("params") or {} + if params.get("sort") != '{"name":1}': + return {"__rpc_error": "an explicit sort must be preserved, got " + str(params.get("sort"))} + if params.get("skip") == 0: + return {"documents": [{"_id": value, "name": "Ada"} for value in range(1000)], "total": 1001} + if params.get("skip") == 1000: + return {"documents": [{"_id": 1000, "name": "Ada"}], "total": 1001} + return {"__rpc_error": "unexpected skip " + str(params.get("skip"))} + return None +"#, + &["mongo_insert_documents"], + ) + .await; + + let summary = export_mongodb_query_core( + &state, + &MongoExportRequest { + export_id: "export-offset".to_string(), + connection_id: "legacy".to_string(), + database: "app".to_string(), + collection: "users".to_string(), + filter: None, + sort: Some("{\"name\":1}".to_string()), + projection: None, + collation: None, + format: MongoExportFormat::Ndjson, + include_header: true, + file_path: target.to_string_lossy().to_string(), + execution_id: None, + }, + |_| Box::pin(async { false }), + |_| {}, + ) + .await + .unwrap(); + let _ = std::fs::remove_file(&target); + assert_eq!(summary.documents_exported, 1001); + } + + #[cfg(unix)] + #[tokio::test] + async fn legacy_agent_export_does_not_use_display_find_documents() { + let target = + std::env::temp_dir().join(format!("dbx-mongo-legacy-export-display-{}.ndjson", uuid::Uuid::new_v4())); + let (state, _directory) = legacy_mongo_import_export_state( + r#" +def handle(request): + method = request.get("method") + if method == "handshake": + return {"protocolVersion": 1, "agentProtocolVersion": 1, "capabilities": CAPABILITIES} + if method in ("validate_connection", "connect"): + return {"ok": True} + if method == "count_documents": + return 1 + if method == "find_documents_extended_json": + return {"__rpc_error": "Unknown method: find_documents_extended_json"} + if method == "find_documents": + return {"documents": [{"_id": "not-an-objectid"}], "total": 1} + return None +"#, + &["mongo_insert_documents"], + ) + .await; + + let error = export_mongodb_query_core( + &state, + &MongoExportRequest { + export_id: "export-3".to_string(), + connection_id: "legacy".to_string(), + database: "app".to_string(), + collection: "users".to_string(), + filter: None, + sort: None, + projection: None, + collation: None, + format: MongoExportFormat::Ndjson, + include_header: true, + file_path: target.to_string_lossy().to_string(), + execution_id: None, + }, + |_| Box::pin(async { false }), + |_| {}, + ) + .await + .unwrap_err(); + let _ = std::fs::remove_file(&target); + assert!(error.contains("type-preserving export"), "{error}"); + assert!(!error.contains("ISODate"), "{error}"); + } + + #[cfg(unix)] + #[tokio::test] + async fn legacy_agent_import_records_partial_insert_errors_when_skipping_rows() { + let source_dir = std::env::temp_dir().join(format!("dbx-mongo-legacy-import-errors-{}", uuid::Uuid::new_v4())); + std::fs::create_dir_all(&source_dir).unwrap(); + let path = source_dir.join("data.ndjson"); + std::fs::write( + &path, + "{\"_id\":{\"$oid\":\"507f1f77bcf86cd799439011\"},\"name\":\"Ada\"}\n{\"_id\":{\"$oid\":\"507f1f77bcf86cd799439012\"},\"name\":\"Grace\"}\n", + ) + .unwrap(); + + let (state, _directory) = legacy_mongo_import_export_state( + r#" +def handle(request): + method = request.get("method") + if method == "handshake": + return {"protocolVersion": 1, "agentProtocolVersion": 1, "capabilities": CAPABILITIES} + if method in ("validate_connection", "connect"): + return {"ok": True} + if method == "insert_documents": + return {"affected_rows": 1, "errors": [{"index": 1, "code": 11000, "message": "E11000 duplicate key"}]} + return None +"#, + &["mongo_insert_documents"], + ) + .await; + + let summary = import_mongodb_file_core( + &state, + &MongoImportRequest { + import_id: "import-2".to_string(), + connection_id: "legacy".to_string(), + database: "app".to_string(), + collection: "users".to_string(), + file_path: path.to_string_lossy().to_string(), + source_ref: None, + format: MongoImportFormat::Ndjson, + parse_options: MongoImportParseOptions { + type_mode: Some(MongoImportTypeMode::ExtendedJson), + skip_error_rows: Some(true), + ..MongoImportParseOptions::default() + }, + batch_size: 500, + execution_id: None, + }, + |_| Box::pin(async { false }), + |_| {}, + ) + .await + .unwrap(); + let _ = std::fs::remove_dir_all(source_dir); + assert_eq!(summary.rows_inserted, 1); + assert_eq!(summary.rows_failed, 1); + } } diff --git a/crates/dbx-core/src/transfer.rs b/crates/dbx-core/src/transfer.rs index 413043bb6c..c2eaf9a44f 100644 --- a/crates/dbx-core/src/transfer.rs +++ b/crates/dbx-core/src/transfer.rs @@ -5421,7 +5421,7 @@ async fn insert_mongo_documents_for_transfer( let docs_json = serde_json::to_string(documents).map_err(|e| format!("Failed to encode MongoDB documents: {e}"))?; match crate::mongo_ops::mongo_insert_documents_core(state, connection_id, database, collection, &docs_json).await { Ok(count) => Ok(count), - Err(error) if error.to_ascii_lowercase().contains("legacy agent") => { + Err(error) if crate::mongo_ops::is_legacy_agent_insert_many_unsupported(&error) => { let mut inserted = 0; for document in documents { let doc_json = @@ -5457,7 +5457,7 @@ async fn insert_mongo_documents_extended_json_for_transfer( .await { Ok(count) => Ok(count), - Err(error) if error.to_ascii_lowercase().contains("legacy agent") => { + Err(error) if crate::mongo_ops::is_legacy_agent_insert_many_unsupported(&error) => { insert_mongo_documents_for_transfer(state, connection_id, database, collection, documents).await } Err(error) => Err(error),