-
Type:
Bug
-
Resolution: Fixed
-
Priority:
Unknown
-
Affects Version/s: None
-
Component/s: None
-
None
-
None
-
Rust Drivers
-
Not Needed
-
None
-
None
-
None
-
None
-
None
-
None
Driver: mongo-rust-driver 3.8.2 (same code on main, driver/src/action/insert_many.rs)
Server: MongoDB 8.0
Behaviour
With w: 0, insert_many never returns. It re-sends the same batch in a loop until the future is dropped, every re-send inserts the documents again.
Suspected cause
This seems to be because the server answers an unacknowledged insert with { n: 0, ok: 1 }. insert_many advances its position by the reply's n, so the position never moves and the loop while n_attempted < ds.len() repeats the same batch.
What I expected
PyMongo advances by the number of documents encoded and sent, independent of the reply, and returns after one round of batches. insert_many should likely do the same.
Reproduction
Dependencies:
mongodb = "3.8.2"
tokio = { version = "1", features = ["full"] }
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::Duration;
use mongodb::Client;
use mongodb::bson::{Document, doc};
use mongodb::event::EventHandler;
use mongodb::event::command::CommandEvent;
use mongodb::options::{ClientOptions, DatabaseOptions, WriteConcern};
#[tokio::main]
async fn main() -> mongodb::error::Result<()> {
let uri = std::env::var("MONGODB_URI").unwrap_or("mongodb://localhost:27017".into());
let sent = Arc::new(AtomicUsize::new(0));
let counter = sent.clone();
let mut options = ClientOptions::parse(&uri).await?;
options.command_event_handler = Some(EventHandler::callback(move |event: CommandEvent| {
if let CommandEvent::Succeeded(e) = event
&& e.command_name == "insert"
&& counter.fetch_add(1, Ordering::SeqCst) == 0
{
println!("first reply: n = {:?}", e.reply.get("n"));
}
}));
let client = Client::with_options(options)?;
let db_options = DatabaseOptions::builder()
.write_concern(WriteConcern::nodes(0))
.build();
let coll = client
.database_with_options("w0_repro", db_options)
.collection::<Document>("c");
let docs = vec![doc! { "a": 1 }, doc! { "a": 2 }];
let outcome = tokio::time::timeout(Duration::from_secs(2), coll.insert_many(docs)).await;
println!("insert_many returned within 2s: {}", outcome.is_ok());
println!("insert commands sent in 2s: {}", sent.load(Ordering::SeqCst));
let stored = client
.database("w0_repro")
.collection::<Document>("c")
.count_documents(doc! {})
.await?;
println!("documents stored: {stored}");
client.database("w0_repro").drop().await?;
Ok(())
}
Output:
first reply: n = Some(Int32(0))
insert_many returned within 2s: false
insert commands sent in 2s: 5400
documents stored: 10802