Retain bulk execution replies on errors

Expose an optional owned BSON output before propagating the driver error, including through collection sessions.

AI disclosure: This work was prepared with assistance from OpenAI Codex.

Author: quinnj

docs/src/api.md

@@ -66,6 +66,8 @@ Mongoc.find_one_and_delete
 Mongoc.find_one_and_replace
 Mongoc.find_one_and_update
 Mongoc.replace_one
+Mongoc.insert_many
+Mongoc.execute!
 ```
 
 ## Aggregation

src/collection.jl

@@ -149,22 +149,40 @@ function BulkOperationResult(reply::BSON, server_id::UInt32)
     BulkOperationResult(reply, server_id, Vector{Union{Nothing, BSONObjectId}}())
 end
 
-function execute!(bulk_operation::BulkOperation) :: BulkOperationResult
+"""
+    execute!(bulk_operation::BulkOperation; reply=nothing) :: BulkOperationResult
+
+Execute a bulk operation once and release its native handle, including on failure.
+Errors throw `BSONError` even when some writes succeeded.
+
+Pass `reply=Ref{BSON}()` to retain the driver's reply on success or failure. The
+reference is assigned after the driver returns, before checking for an error.
+It remains unchanged if execution is rejected before calling the driver. A reply
+may be empty on failure; it does not by itself indicate success. On success,
+`reply[]` is the same document as the returned result's `reply` field. The reply
+remains usable after the bulk operation's native handle is released.
+"""
+function execute!(bulk_operation::BulkOperation;
+                  reply::Union{Nothing, Ref{BSON}}=nothing) :: BulkOperationResult
     if bulk_operation.executed
         error("Bulk operation was already executed.")
     end
 
     try
-        reply = BSON()
+        reply_bson = BSON()
         err_ref = Ref{BSONError}()
 
-        bulk_operation_result = mongoc_bulk_operation_execute(bulk_operation.handle,
-            reply.handle, err_ref)
+        bulk_operation_result = GC.@preserve bulk_operation reply_bson begin
+            mongoc_bulk_operation_execute(bulk_operation.handle, reply_bson.handle, err_ref)
+        end
+        if reply !== nothing
+            reply[] = reply_bson
+        end
 
         if bulk_operation_result == 0
             throw(err_ref[])
         end
-        return BulkOperationResult(reply, bulk_operation_result)
+        return BulkOperationResult(reply_bson, bulk_operation_result)
     finally
         destroy!(bulk_operation)
     end
@@ -222,8 +240,23 @@ function bulk_update_one!(
 end
 
 
+"""
+    insert_many(collection, documents::Vector{BSON};
+                bulk_options=nothing, insert_options=nothing, reply=nothing)
+
+Insert documents into a `Collection` or `CollectionSession`, returning a
+`BulkOperationResult`. With `bulk_options=BSON("ordered" => false)`, the server
+continues after individual write errors, but this function still throws
+`BSONError` if any write fails.
+
+Pass `reply=Ref{BSON}()` to retain the bulk reply when execution throws. The reply
+contains counts and `writeErrors` when available; error indices are zero-based
+positions in `documents`. Setup errors before bulk execution leave the reference
+unchanged. See [`execute!`](@ref) for the reply's lifetime and success behavior.
+"""
 function insert_many(collection::Collection, documents::Vector{BSON};
-        bulk_options::Union{Nothing, BSON}=nothing, insert_options::Union{Nothing, BSON}=nothing)
+        bulk_options::Union{Nothing, BSON}=nothing, insert_options::Union{Nothing, BSON}=nothing,
+        reply::Union{Nothing, Ref{BSON}}=nothing)
 
     inserted_oids = Vector{Union{Nothing, BSONObjectId}}()
 
@@ -233,7 +266,7 @@ function insert_many(collection::Collection, documents::Vector{BSON};
         bulk_insert!(bulk_operation, doc, options=insert_options)
         push!(inserted_oids, inserted_oid)
     end
-    result = execute!(bulk_operation)
+    result = execute!(bulk_operation; reply=reply)
     append!(result.inserted_oids, inserted_oids)
     return result
 end

src/session.jl

@@ -82,10 +82,11 @@ end
 
 function insert_many(collection::CollectionSession, documents::Vector{BSON};
                      bulk_options::Union{Nothing, BSON}=nothing,
-                     insert_options::Union{Nothing, BSON}=nothing)
+                     insert_options::Union{Nothing, BSON}=nothing,
+                     reply::Union{Nothing, Ref{BSON}}=nothing)
     bulk_options_with_session = _join(bulk_options, get_session(collection))
     insert_many(collection.collection, documents,
-                bulk_options=bulk_options_with_session, insert_options=insert_options)
+                bulk_options=bulk_options_with_session, insert_options=insert_options, reply=reply)
 end
 
 function find(collection::CollectionSession, bson_filter::BSON=BSON();

test/mongodb_tests.jl

@@ -252,6 +252,99 @@ const DB_NAME = "mongoc"
             Mongoc.drop(collection)
         end
 
+        @testset "Bulk replies on success and failure" begin
+            for ordered in (true, false), direct in (true, false)
+                collection = client[DB_NAME]["bulk_reply_$(ordered)_$(direct)"]
+                bulk = nothing
+                reply = Ref{Mongoc.BSON}()
+                try
+                    Mongoc.insert_one(collection, Mongoc.BSON("_id" => 1, "original" => true))
+                    documents = [Mongoc.BSON("_id" => i) for i in 1:3]
+                    err = try
+                        if direct
+                            bulk = Mongoc.BulkOperation(collection; options=Mongoc.BSON("ordered" => ordered))
+                            for document in documents
+                                Mongoc.bulk_insert!(bulk, document)
+                            end
+                            Mongoc.execute!(bulk; reply=reply)
+                        else
+                            Mongoc.insert_many(collection, documents;
+                                bulk_options=Mongoc.BSON("ordered" => ordered), reply=reply)
+                        end
+                        nothing
+                    catch caught
+                        caught
+                    end
+                    @test err isa Mongoc.BSONError
+                    @test err.code == 11000
+                    @test isassigned(reply)
+                    GC.gc()
+                    @test reply[]["nInserted"] == (ordered ? 0 : 2)
+                    errors = reply[]["writeErrors"]
+                    @test length(errors) == 1
+                    @test errors[1]["index"] == 0
+                    @test errors[1]["code"] == 11000
+                    @test Mongoc.count_documents(collection) == (ordered ? 1 : 3)
+                    @test Mongoc.find_one(collection, Mongoc.BSON("_id" => 1))["original"] === true
+                    if direct
+                        @test bulk.executed
+                        @test bulk.handle == C_NULL
+                        retained_reply = reply[]
+                        @test_throws ErrorException Mongoc.execute!(bulk; reply=reply)
+                        @test reply[] === retained_reply
+                    end
+                    @test_throws Mongoc.BSONError Mongoc.insert_many(collection, [Mongoc.BSON("_id" => 1)])
+                finally
+                    bulk === nothing || Mongoc.destroy!(bulk)
+                    Mongoc.drop(collection)
+                    Mongoc.destroy!(collection)
+                end
+            end
+
+            collection = client[DB_NAME]["bulk_reply_success"]
+            try
+                previous = Mongoc.BSON("original" => true)
+                reply = Ref(previous)
+                result = Mongoc.insert_many(collection, [Mongoc.BSON("_id" => 1)]; reply=reply)
+                @test reply[] === result.reply
+                @test reply[] !== previous
+                @test previous["original"] === true
+                @test reply[]["nInserted"] == 1
+                @test isempty(reply[]["writeErrors"])
+                bulk = Mongoc.BulkOperation(collection)
+                empty_reply = Ref{Mongoc.BSON}()
+                @test_throws Mongoc.BSONError Mongoc.execute!(bulk; reply=empty_reply)
+                @test isassigned(empty_reply)
+                @test isempty(empty_reply[])
+                @test bulk.handle == C_NULL
+            finally
+                Mongoc.drop(collection)
+                Mongoc.destroy!(collection)
+            end
+
+            options = Mongoc.SessionOptions()
+            session = Mongoc.Session(client; options=options)
+            collection = session[DB_NAME]["bulk_reply_session"]
+            try
+                reply = Ref{Mongoc.BSON}()
+                result = Mongoc.insert_many(collection, [Mongoc.BSON("_id" => 1)]; reply=reply)
+                @test reply[] === result.reply
+                @test reply[]["nInserted"] == 1
+                @test_throws Mongoc.BSONError Mongoc.insert_many(collection,
+                    [Mongoc.BSON("_id" => 1), Mongoc.BSON("_id" => 2)];
+                    bulk_options=Mongoc.BSON("ordered" => false), reply=reply)
+                @test reply[]["nInserted"] == 1
+                @test reply[]["writeErrors"][1]["index"] == 0
+                @test Mongoc.count_documents(collection) == 2
+            finally
+                Mongoc.drop(collection)
+                Mongoc.destroy!(collection.collection)
+                Mongoc.destroy!(collection.database_session.database)
+                Mongoc.destroy!(session)
+                Mongoc.destroy!(options)
+            end
+        end
+
         @testset "bulk_replace_one" begin
             collection = client[DB_NAME]["replace_many"]
             push!(collection, Mongoc.BSON("""{ "x": 1, "y": 100 }"""))