diff --git a/crates/dbx-core/src/data/mongodb_dump.rs b/crates/dbx-core/src/data/mongodb_dump.rs index 8178e23236..beba331d87 100644 --- a/crates/dbx-core/src/data/mongodb_dump.rs +++ b/crates/dbx-core/src/data/mongodb_dump.rs @@ -21,10 +21,15 @@ use std::sync::{ use std::time::{Duration, Instant}; use futures::TryStreamExt; -use mongodb::bson::{doc, Document}; +use mongodb::bson::{doc, Document, RawDocumentBuf}; use serde::{Deserialize, Serialize}; use crate::connection::{AppState, PoolKind}; +use crate::db::agent_driver::{AgentCapability, PooledAgentClient}; +use crate::db::mongo_driver::{ + agent_insert_outcome, document_to_canonical_extended_json, insert_bson_documents, + json_object_to_document_extended_json, MongoCollectionKind, MongoDocumentResult, MongoInsertOutcome, +}; use crate::mongodb_import_export::{export_mongodb_query_core, MongoExportFormat, MongoExportRequest}; use metadata::CollectionMetadata; @@ -206,33 +211,205 @@ fn catalog(entries: &[PreparedCollection], exact: bool) -> MongoDumpCatalog { } } -async fn native_client(state: &AppState, id: &str, database: &str) -> Result { - metadata::validate_database(database)?; - state.get_or_create_pool(id, Some(database)).await?; - match state.pool_handle(id).await { - Some(PoolKind::MongoDb(client)) => Ok(client.clone()), - _ => Err("Database dump/restore requires the native MongoDB driver".into()), +/// A `listCollections` entry with what dump/restore needs from it. +pub(crate) struct DumpCollectionSpec { + pub name: String, + pub kind: MongoCollectionKind, + pub options: Document, +} + +/// The connection a dump or restore runs over. Both drivers expose the same handful of +/// commands; the legacy agent runs them through its generic `runCommand` so nothing has to +/// be added to the agent itself. +pub(crate) enum DumpClient { + Native(mongodb::Client), + Agent(Arc), +} + +impl DumpClient { + async fn run_command(&self, database: &str, command: Document) -> Result { + match self { + Self::Native(client) => client.database(database).run_command(command).await.map_err(|e| e.to_string()), + Self::Agent(client) => { + let mut client = client.lock().await; + let result: MongoDocumentResult = client + .mongo_run_command(serde_json::json!({ + "database": database, + "command_json": document_to_canonical_extended_json(&command).to_string(), + })) + .await?; + let response = result + .extended_documents + .as_ref() + .and_then(|documents| documents.first()) + .or_else(|| result.documents.first()) + .ok_or("MongoDB Legacy Agent returned an empty runCommand response")?; + json_object_to_document_extended_json(response) + } + } + } + + /// Drains a command cursor (`listCollections`, `listIndexes`) through `getMore` so a database + /// with more entries than one batch is still read completely on the agent path. + async fn command_cursor(&self, database: &str, command: Document) -> Result, String> { + let mut response = self.run_command(database, command).await?; + let mut documents = Vec::new(); + loop { + let cursor = response.get_document_mut("cursor").map_err(|e| e.to_string())?; + let batch = if cursor.contains_key("firstBatch") { "firstBatch" } else { "nextBatch" }; + if let Ok(entries) = cursor.get_array_mut(batch) { + documents.extend(std::mem::take(entries).into_iter().filter_map(|entry| match entry { + mongodb::bson::Bson::Document(document) => Some(document), + _ => None, + })); + } + let id = cursor.get_i64("id").unwrap_or(0); + if id == 0 { + return Ok(documents); + } + let namespace = cursor.get_str("ns").map_err(|e| e.to_string())?; + let collection = + namespace.strip_prefix(database).and_then(|rest| rest.strip_prefix('.')).unwrap_or(namespace); + response = self.run_command(database, doc! { "getMore": id, "collection": collection }).await?; + } + } + + pub(crate) async fn list_collections(&self, database: &str) -> Result, String> { + match self { + Self::Native(client) => { + let mut cursor = client.database(database).list_collections().await.map_err(|e| e.to_string())?; + let mut specs = Vec::new(); + while let Some(spec) = cursor.try_next().await.map_err(|e| e.to_string())? { + specs.push(DumpCollectionSpec { + name: spec.name, + kind: MongoCollectionKind::from_driver_type(&spec.collection_type), + options: mongodb::bson::to_document(&spec.options).map_err(|e| e.to_string())?, + }); + } + Ok(specs) + } + Self::Agent(_) => self + .command_cursor(database, doc! { "listCollections": 1 }) + .await? + .into_iter() + .map(|mut spec| { + let name = spec.get_str("name").map_err(|e| format!("listCollections entry: {e}"))?.to_string(); + let kind = MongoCollectionKind::from_metadata_kind(spec.get_str("type").ok()); + let options = spec.remove("options").and_then(|o| o.as_document().cloned()).unwrap_or_default(); + Ok(DumpCollectionSpec { name, kind, options }) + }) + .collect(), + } + } + + async fn list_indexes(&self, database: &str, collection: &str) -> Result, String> { + match self { + Self::Native(client) => { + let mut indexes = client + .database(database) + .collection::(collection) + .list_indexes() + .await + .map_err(|e| e.to_string())?; + let mut documents = Vec::new(); + while let Some(index) = indexes.try_next().await.map_err(|e| e.to_string())? { + documents.push(mongodb::bson::to_document(&index).map_err(|e| e.to_string())?); + } + Ok(documents) + } + Self::Agent(_) => self.command_cursor(database, doc! { "listIndexes": collection }).await, + } + } + + pub(crate) async fn drop_collection(&self, database: &str, collection: &str) -> Result<(), String> { + match self { + Self::Native(client) => { + client.database(database).collection::(collection).drop().await.map_err(|e| e.to_string()) + } + Self::Agent(_) => self.run_command(database, doc! { "drop": collection }).await.map(|_| ()), + } + } + + pub(crate) async fn create_collection(&self, database: &str, command: Document) -> Result<(), String> { + self.run_command(database, command).await.map(|_| ()) + } + + /// Inserts one restore batch. The agent takes documents as canonical Extended JSON, the same + /// lossless form collection-level BSON import already uses on that path. + pub(crate) async fn insert_documents( + &self, + database: &str, + collection: &str, + documents: Vec, + ) -> Result { + match self { + Self::Native(client) => { + insert_bson_documents(client, database, collection, documents).await.map_err(|e| e.message) + } + Self::Agent(client) => { + let documents = documents + .iter() + .map(|raw| { + Document::try_from(raw.as_ref()).map(|document| document_to_canonical_extended_json(&document)) + }) + .collect::, _>>() + .map_err(|e| e.to_string())?; + let docs_json = serde_json::to_string(&documents).map_err(|e| e.to_string())?; + let mut client = client.lock().await; + let result: serde_json::Value = client + .mongo_insert_documents(serde_json::json!({ + "database": database, + "collection": collection, + "docs_json": docs_json, + "ordered": false, + })) + .await?; + agent_insert_outcome(&result) + } + } + } + + async fn create_index(&self, database: &str, collection: &str, index: Document) -> Result<(), String> { + self.run_command(database, doc! { "createIndexes": collection, "indexes": [index] }).await.map(|_| ()) + } + + async fn server_version(&self, database: &str) -> Result { + Ok(self.run_command(database, doc! { "buildInfo": 1 }).await?.get_str("version").unwrap_or("").to_string()) } } -async fn database_entries(client: &mongodb::Client, database: &str) -> Result, String> { - let db = client.database(database); - let mut cursor = db.list_collections().await.map_err(|e| e.to_string())?; +async fn dump_client(state: &AppState, id: &str, database: &str) -> Result { + metadata::validate_database(database)?; + state.get_or_create_pool(id, Some(database)).await?; + match state.pool_handle(id).await { + Some(PoolKind::MongoDb(client)) => Ok(DumpClient::Native(client.clone())), + Some(PoolKind::Agent(client)) => { + let supported = { + let client = client.lock().await; + client.supports_capability(AgentCapability::MongoRunCommand) + && client.supports_capability(AgentCapability::MongoInsertDocuments) + }; + if !supported { + return Err("MongoDB Legacy Agent does not support database dump/restore; upgrade or reinstall the MongoDB Legacy driver".into()); + } + Ok(DumpClient::Agent(client.clone())) + } + _ => Err("Database dump/restore requires a MongoDB connection".into()), + } +} + +async fn database_entries(client: &DumpClient, database: &str) -> Result, String> { let mut entries = Vec::new(); - while let Some(spec) = cursor.try_next().await.map_err(|e| e.to_string())? { + for spec in client.list_collections(database).await? { if spec.name.starts_with("system.") && spec.name != "system.js" { continue; } let mut metadata = CollectionMetadata::empty(&spec.name); - metadata.kind = - crate::db::mongo_driver::MongoCollectionKind::from_driver_type(&spec.collection_type).as_str().into(); - metadata.options = mongodb::bson::to_document(&spec.options).map_err(|e| e.to_string())?; + metadata.kind = spec.kind.as_str().into(); + metadata.options = spec.options; metadata.validate()?; if !metadata.is_view() { - let mut indexes = db.collection::(&spec.name).list_indexes().await.map_err(|e| e.to_string())?; - while let Some(index) = indexes.try_next().await.map_err(|e| e.to_string())? { - metadata.indexes.push(mongodb::bson::to_document(&index).map_err(|e| e.to_string())?); - } + metadata.indexes = client.list_indexes(database, &spec.name).await?; } metadata.validate()?; entries.push(PreparedCollection { @@ -253,7 +430,7 @@ pub async fn inspect_mongodb_database_dump( id: &str, database: &str, ) -> Result { - let client = native_client(state, id, database).await?; + let client = dump_client(state, id, database).await?; let entries = database_entries(&client, database).await?; let mut catalog = catalog(&entries, false); catalog.databases = vec![database.into()]; @@ -370,7 +547,7 @@ where if is_cancelled(&request.task_id).await { return Err("MongoDB dump/restore cancelled".into()); } - let client = native_client(state, &request.connection_id, &request.database).await?; + let client = dump_client(state, &request.connection_id, &request.database).await?; let mut entries = select_entries(&database_entries(&client, &request.database).await?, &request.collections)?; progress.collections_total = entries.len(); let target = PathBuf::from(&request.file_path); @@ -428,14 +605,7 @@ where } progress.phase = "archive".into(); progress.emit(started, &mut on_progress); - let server_version = client - .database(&request.database) - .run_command(doc! { "buildInfo": 1 }) - .await - .map_err(|e| e.to_string())? - .get_str("version") - .unwrap_or("") - .to_string(); + let server_version = client.server_version(&request.database).await?; let database = request.database.clone(); let gzip = request.gzip; let format = request.format; @@ -507,21 +677,13 @@ where let entries = ordered_entries(&select_entries(&source_entries, &request.collections)?)?; progress.collections_total = entries.len(); source.check_ready(&entries)?; - let client = native_client(state, &request.connection_id, &request.database).await?; - let db = client.database(&request.database); - let existing = db - .list_collections() - .await - .map_err(|e| e.to_string())? - .try_collect::>() - .await - .map_err(|e| e.to_string())?; + let client = dump_client(state, &request.connection_id, &request.database).await?; + let existing = client.list_collections(&request.database).await?; for entry in &entries { entry.metadata.validate()?; if let Some(target) = existing.iter().find(|target| target.name == entry.metadata.collection_name) { if !request.drop_existing - && (entry.metadata.is_view() - || !matches!(target.collection_type, mongodb::results::CollectionType::Collection)) + && (entry.metadata.is_view() || target.kind != MongoCollectionKind::Collection) { return Err(format!( "{}: replacing an existing view or non-regular collection requires dropExisting", @@ -571,9 +733,7 @@ where if is_cancelled(&request.task_id).await { return Err("MongoDB dump/restore cancelled".into()); } - db.run_command(doc! { "createIndexes": name, "indexes": [index] }) - .await - .map_err(|e| e.to_string())?; + client.create_index(&request.database, name, index).await?; progress.indexes_created += 1; } } diff --git a/crates/dbx-core/src/data/mongodb_dump/stream.rs b/crates/dbx-core/src/data/mongodb_dump/stream.rs index 3803fa351d..8e3f8ce956 100644 --- a/crates/dbx-core/src/data/mongodb_dump/stream.rs +++ b/crates/dbx-core/src/data/mongodb_dump/stream.rs @@ -9,10 +9,10 @@ use mongodb::bson::{Document, RawDocumentBuf}; use tokio::sync::mpsc; use super::{ - archive, cancelled, source::CatalogSource, CancelFuture, MongoDatabaseDumpProgress, MongoDatabaseRestoreRequest, - MongoDumpFormat, PreparedCollection, + archive, cancelled, source::CatalogSource, CancelFuture, DumpClient, MongoDatabaseDumpProgress, + MongoDatabaseRestoreRequest, MongoDumpFormat, PreparedCollection, }; -use crate::{connection::AppState, db::mongo_driver::insert_bson_documents}; +use crate::connection::AppState; const BATCH_BYTES: usize = 8 * 1024 * 1024; @@ -142,17 +142,16 @@ fn produce( } pub(super) async fn initialize_collection( - client: &mongodb::Client, + client: &DumpClient, request: &MongoDatabaseRestoreRequest, entry: &PreparedCollection, exists: bool, ) -> Result<(), String> { - let db = client.database(&request.database); if request.drop_existing && exists { - db.collection::(&entry.metadata.collection_name).drop().await.map_err(|e| e.to_string())?; + client.drop_collection(&request.database, &entry.metadata.collection_name).await?; } if !exists || request.drop_existing { - db.run_command(entry.metadata.create_command(request.restore_options)).await.map_err(|e| e.to_string())?; + client.create_collection(&request.database, entry.metadata.create_command(request.restore_options)).await?; } Ok(()) } @@ -167,7 +166,7 @@ pub(super) async fn writable(state: &AppState, id: &str) -> Result<(), String> { #[allow(clippy::too_many_arguments)] pub(super) async fn restore_data( state: &AppState, - client: &mongodb::Client, + client: &DumpClient, request: &MongoDatabaseRestoreRequest, source: Arc, entries: &[PreparedCollection], @@ -238,9 +237,7 @@ pub(super) async fn restore_data( progress.documents_validated += documents.len() as u64; } let outcome = - insert_bson_documents(client, &request.database, &entry.metadata.collection_name, documents) - .await - .map_err(|e| e.message)?; + client.insert_documents(&request.database, &entry.metadata.collection_name, documents).await?; progress.documents_written += outcome.inserted; progress.documents_failed += outcome.errors.len() as u64; progress.emit(started, on_progress); diff --git a/crates/dbx-core/src/data/mongodb_import_export.rs b/crates/dbx-core/src/data/mongodb_import_export.rs index 8d913fcd68..68a266e84a 100644 --- a/crates/dbx-core/src/data/mongodb_import_export.rs +++ b/crates/dbx-core/src/data/mongodb_import_export.rs @@ -2849,12 +2849,7 @@ where if is_cancelled(&request.export_id).await { return Err("Export cancelled".to_string()); } - // BSON dumps must carry native BSON types, so they are written straight from the native - // driver's documents; the legacy agent's canonical Extended JSON round trip does not apply. - let client = match state.pool_handle(&request.connection_id).await { - Some(PoolKind::MongoDb(client)) => client, - _ => return Err("BSON dump export requires the native MongoDB driver".into()), - }; + let pool = state.pool_handle(&request.connection_id).await.ok_or_else(|| "Not found".to_string())?; let file = File::create(temp).map_err(|error| error.to_string())?; let buffered = BufWriter::new(file); let mut writer = if request.gzip { @@ -2864,48 +2859,64 @@ where }; let mut documents_read = 0u64; let mut bytes_written = 0u64; + let mut write_document = |document: Document| -> Result<(), String> { + let bytes = mongodb::bson::to_vec(&document).map_err(|error| error.to_string())?; + writer.write_all(&bytes).map_err(|error| error.to_string())?; + bytes_written += bytes.len() as u64; + 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(()) + }; - let export = 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 bytes = mongodb::bson::to_vec(&document).map_err(|error| error.to_string())?; - writer.write_all(&bytes).map_err(|error| error.to_string())?; - bytes_written += bytes.len() as u64; - 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(()) - }, - ); - { - tokio::pin!(export); - let mut poll_cancel = tokio::time::interval(std::time::Duration::from_millis(100)); - loop { - tokio::select! { - result = &mut export => { result?; break; } - _ = poll_cancel.tick() => { - if is_cancelled(&request.export_id).await { - return Err("Export cancelled".to_string()); + match &pool { + PoolKind::MongoDb(client) => { + let export = 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, + &mut write_document, + ); + tokio::pin!(export); + let mut poll_cancel = tokio::time::interval(std::time::Duration::from_millis(100)); + loop { + tokio::select! { + result = &mut export => { result?; break; } + _ = poll_cancel.tick() => { + if is_cancelled(&request.export_id).await { + return Err("Export cancelled".to_string()); + } } } } } + PoolKind::Agent(client) => { + // A BSON dump must keep every BSON type. The agent's find cursor streams canonical + // Extended JSON, which round-trips losslessly; the older page-based fallback is + // relaxed JSON (Int32/Int64 collapse), so it is not accepted here. + if !client.lock().await.supports_capability(AgentCapability::MongoFindCursor) { + return Err("BSON export over the MongoDB Legacy Agent needs find-cursor support; upgrade or reinstall the MongoDB Legacy driver".into()); + } + for_each_agent_export_document(client, request, is_cancelled, |value| { + write_document(mongo_driver::json_object_to_document_extended_json(&value)?) + }) + .await?; + } + _ => return Err("Not a MongoDB connection".to_string()), } writer.finish()?; if is_cancelled(&request.export_id).await { diff --git a/crates/dbx-core/tests/mongodb_dump_restore.rs b/crates/dbx-core/tests/mongodb_dump_restore.rs index c4a81a5bef..5ebc3d0cad 100644 --- a/crates/dbx-core/tests/mongodb_dump_restore.rs +++ b/crates/dbx-core/tests/mongodb_dump_restore.rs @@ -721,6 +721,193 @@ async fn official_database_tools_round_trip_through_dbx() { } } +#[tokio::test] +#[ignore = "opt-in: DBX_MONGO_LEGACY_DUMP_TEST_HOST (host:port, MongoDB 3.6+ without auth) and an installed MongoDB Legacy Agent; creates temporary databases"] +async fn legacy_agent_database_round_trip_through_dbx() { + use dbx_core::{ + connection::{AppState, PoolKind}, + models::connection::ConnectionConfig, + mongo_ops::mongo_run_command_core, + mongodb_dump::*, + storage::Storage, + }; + // Everything, including seeding and verification, goes through the legacy agent: the native + // Rust driver refuses servers older than 4.2, which is the very case the agent exists for. + async fn command(state: &AppState, id: &str, database: &str, command: Document) -> Document { + let json = mongodb::bson::Bson::Document(command).into_canonical_extjson().to_string(); + let result = mongo_run_command_core(state, id, database, &json).await.unwrap(); + let response = result.extended_documents.unwrap().remove(0); + let response = dbx_core::db::mongo_driver::json_object_to_document_extended_json(&response).unwrap(); + assert_eq!( + response.get_f64("ok").ok().or_else(|| response.get_i32("ok").ok().map(f64::from)), + Some(1.0), + "{response}" + ); + response + } + fn batch(response: &Document) -> Vec { + let cursor = response.get_document("cursor").unwrap(); + cursor.get_array("firstBatch").unwrap().iter().map(|b| b.as_document().unwrap().clone()).collect() + } + async fn documents(state: &AppState, id: &str, database: &str, collection: &str) -> Vec> { + batch(&command(state, id, database, doc! { "find": collection, "sort": { "_id": 1 } }).await) + .iter() + .map(|document| mongodb::bson::to_vec(document).unwrap()) + .collect() + } + async fn specs(state: &AppState, id: &str, database: &str, collection: &str) -> (Document, Vec) { + let listed = + batch(&command(state, id, database, doc! { "listCollections": 1, "filter": { "name": collection } }).await); + let spec = listed.first().unwrap_or_else(|| panic!("{database}.{collection} is missing")); + let options = spec.get_document("options").cloned().unwrap_or_default(); + let mut indexes = Vec::new(); + if spec.get_str("type").unwrap_or("collection") != "view" { + for mut index in batch(&command(state, id, database, doc! { "listIndexes": collection }).await) { + index.remove("ns"); + index.remove("v"); + indexes.push(index); + } + indexes.sort_by_key(|index| index.get_str("name").unwrap().to_string()); + } + (options, indexes) + } + + let endpoint = std::env::var("DBX_MONGO_LEGACY_DUMP_TEST_HOST").expect("DBX_MONGO_LEGACY_DUMP_TEST_HOST"); + let (host, port) = endpoint.split_once(':').expect("host:port"); + let files = tempfile::tempdir().unwrap(); + let database = format!("dbx_legacy_{}", uuid::Uuid::new_v4().simple()); + let state = AppState::new(Storage::open(&files.path().join("storage.db")).await.unwrap()); + let id = "legacy-dump-test"; + let config: ConnectionConfig = serde_json::from_value(serde_json::json!({ "id": id, "name": "Legacy dump test", "db_type": "mongodb", "host": host, "port": port.parse::().unwrap(), "username": "", "password": "", "database": database, "driver_profile": "mongodb-legacy" })).unwrap(); + state.configs.write().await.insert(id.into(), config); + let key = state.get_or_create_pool(id, Some(&database)).await.unwrap(); + let kind = match state.pool_handle(&key).await { + Some(PoolKind::Agent(_)) => "agent", + Some(PoolKind::MongoDb(_)) => "native", + Some(_) => "other", + None => "none", + }; + assert_eq!(kind, "agent", "test must run over the legacy agent (pool key {key})"); + + command(&state, id, &database, doc! { "create": "records", "validator": { "score": { "$gte": 0 } }, "validationLevel": "strict", "collation": { "locale": "en", "strength": 2 } }).await; + command(&state, id, &database, doc! { "insert": "records", "documents": [ + { "_id": 1, "score": 1i64, "name": "first", "when": mongodb::bson::DateTime::from_millis(1_700_000_000_000) }, + { "_id": 2, "score": 2i64, "name": "second", "ratio": 0.5f64, "tags": ["a", "b"] }, + ] }).await; + command(&state, id, &database, doc! { "createIndexes": "records", "indexes": [ { "key": { "name": 1, "score": -1 }, "name": "compound_unique", "unique": true }, { "key": { "expires": 1 }, "name": "ttl", "expireAfterSeconds": 3600 }] }).await; + command(&state, id, &database, doc! { "create": "empty" }).await; + command(&state, id, &database, doc! { "create": "capped", "capped": true, "size": 65536i64, "max": 100i64 }).await; + command(&state, id, &database, doc! { "insert": "capped", "documents": [ { "_id": 1, "n": 1 } ] }).await; + command(&state, id, &database, doc! { "insert": "odd/name", "documents": [ { "_id": mongodb::bson::oid::ObjectId::new(), "binary": mongodb::bson::Binary { subtype: mongodb::bson::spec::BinarySubtype::Generic, bytes: vec![1, 2, 3] }, "big": i64::MAX } ] }).await; + command(&state, id, &database, doc! { "create": "z_view", "viewOn": "records", "pipeline": [ { "$match": { "score": { "$gte": 1 } } } ], "collation": { "locale": "en", "strength": 2 } }).await; + command(&state, id, &database, doc! { "create": "a_nested", "viewOn": "z_view", "pipeline": [], "collation": { "locale": "en", "strength": 2 } }).await; + let collections = ["records", "empty", "capped", "odd/name", "z_view", "a_nested"]; + + let catalog = inspect_mongodb_database_dump(&state, id, &database).await.unwrap(); + assert_eq!(catalog.collections.len(), 6); + + let mut databases = vec![database.clone()]; + for format in [MongoDumpFormat::Directory, MongoDumpFormat::Archive] { + for gzip in [false, true] { + let tag = format!("{format:?}_{gzip}"); + let output = files.path().join(format!("legacy-{tag}")); + let dump = MongoDatabaseDumpRequest { + task_id: uuid::Uuid::new_v4().to_string(), + connection_id: id.into(), + database: database.clone(), + file_path: output.to_str().unwrap().into(), + format, + gzip, + collections: None, + }; + let dumped = dump_mongodb_database(&state, &dump, |_| Box::pin(async { false }), |_| {}).await.unwrap(); + assert_eq!(dumped.collections_done, 6); + assert_eq!(dumped.documents_read, 4); + + let preview = prepare_mongodb_restore_source(MongoRestoreSourceRequest { + path: output.to_str().unwrap().into(), + format, + gzip, + }) + .await + .unwrap(); + let restored = format!("{database}_{tag}"); + databases.push(restored.clone()); + let request = MongoDatabaseRestoreRequest { + task_id: uuid::Uuid::new_v4().to_string(), + connection_id: id.into(), + database: restored.clone(), + source_database: database.clone(), + source_ref: preview.source_ref.clone(), + collections: None, + drop_existing: false, + restore_options: true, + restore_indexes: true, + stop_on_error: true, + objcheck: gzip, + batch_size: 500, + execution_id: None, + }; + let result = + restore_mongodb_database(&state, &request, |_| Box::pin(async { false }), |_| {}).await.unwrap(); + assert_eq!(result.collections_done, 6); + assert_eq!(result.documents_written, 4); + assert_eq!(result.documents_failed, 0); + assert_eq!(result.indexes_created, 2, "records: compound_unique + ttl; _id_ indexes are never recreated"); + for collection in collections { + assert_eq!( + documents(&state, id, &database, collection).await, + documents(&state, id, &restored, collection).await, + "data: {tag} {collection}" + ); + assert_eq!( + specs(&state, id, &database, collection).await, + specs(&state, id, &restored, collection).await, + "metadata: {tag} {collection}" + ); + } + + // Selective restore with dropExisting replaces only the chosen collection. + command(&state, id, &restored, doc! { "insert": "unselected", "documents": [ { "_id": "keep" } ] }).await; + command( + &state, + id, + &restored, + doc! { "insert": "records", "documents": [ { "_id": 999, "score": 999, "name": "remove" } ] }, + ) + .await; + let mut selected = request.clone(); + selected.drop_existing = true; + selected.collections = Some(vec!["records".into()]); + restore_mongodb_database(&state, &selected, |_| Box::pin(async { false }), |_| {}).await.unwrap(); + assert_eq!( + documents(&state, id, &database, "records").await, + documents(&state, id, &restored, "records").await + ); + assert_eq!(documents(&state, id, &restored, "unselected").await.len(), 1); + + // Appending onto existing _ids reports each duplicate instead of silently succeeding. + let mut duplicate = request.clone(); + duplicate.collections = Some(vec!["records".into()]); + duplicate.stop_on_error = false; + let appended = + restore_mongodb_database(&state, &duplicate, |_| Box::pin(async { false }), |_| {}).await.unwrap(); + assert_eq!(appended.documents_written, 0); + assert_eq!(appended.documents_failed, 2); + assert_eq!( + documents(&state, id, &database, "records").await, + documents(&state, id, &restored, "records").await + ); + + assert!(release_mongodb_restore_source(&preview.source_ref)); + println!("legacy agent {tag}: six collections/views, BSON, options and indexes match"); + } + } + for database in databases { + command(&state, id, &database, doc! { "dropDatabase": 1 }).await; + } +} + #[tokio::test] async fn database_source_rejects_invalid_metadata_and_view_cycles() { use dbx_core::mongodb_dump::*; diff --git a/docs/content/docs/mongodb.cn.mdx b/docs/content/docs/mongodb.cn.mdx index bd7010857a..5e9b29c33d 100644 --- a/docs/content/docs/mongodb.cn.mdx +++ b/docs/content/docs/mongodb.cn.mdx @@ -124,7 +124,7 @@ CSV 类型保真：`Double` 导出再导入会变为 `Decimal128`（十进制文 ### 数据库备份与恢复 -使用 MongoDB 原生驱动时，在数据库节点右键选择「备份数据库」或「恢复数据库」。数据库备份包含选中集合、集合选项、索引和视图。 +在数据库节点右键选择「备份数据库」或「恢复数据库」。数据库备份包含选中集合、集合选项、索引和视图。原生驱动和 `mongodb-legacy` Agent 均支持；Agent 路径需要较新的 Agent 版本（支持 find 游标），以保证 BSON 类型在传输中不丢失。 - 桌面端支持官方目录布局与 MongoDB archive 格式，均可使用 gzip。Web 端导出 archive 文件，恢复时支持 archive 或上传目录。 - 目录备份包含 `<数据库>/<集合>.bson` 和 `.metadata.json`，或对应的 `.gz` 文件。MongoDB archive 是独立的流式格式，不是 ZIP 或 tar。 diff --git a/docs/content/docs/mongodb.mdx b/docs/content/docs/mongodb.mdx index 308b920f10..df4fe0b5c0 100644 --- a/docs/content/docs/mongodb.mdx +++ b/docs/content/docs/mongodb.mdx @@ -124,7 +124,7 @@ Collection-import limits: append only (no truncate or upsert), CSV export discov ### Database backups -With the native MongoDB driver, right-click a database and choose **Back up database** or **Restore database**. Database backups include selected collections, collection options, indexes and views. +Right-click a database and choose **Back up database** or **Restore database**. Database backups include selected collections, collection options, indexes and views. Both the native driver and the `mongodb-legacy` Agent are supported; the Agent path needs a current Agent build (find-cursor support) so BSON types survive the transfer. - Desktop supports the official dump directory layout and MongoDB archive format, optionally gzip-compressed. Web exports archive files and restores either archives or uploaded directories. - A dump directory contains `/.bson` and `.metadata.json`, or their `.gz` variants. MongoDB archive is a separate stream format, not ZIP or tar.