CDRIVER-5869 fix bulk write crash with verbose results (#1843)

* do not use `bson_iter_t` to store `_id` location

The `bson_iter_t` stores a pointer to the referred data. The payload buffer `mongoc_bulkwrite_t::ops` may be reallocated and invalidate existing iterators. Instead, store integral offsets into the payload.

* fix comment describing `ops_byte_len` and `ops_doc_len`

---------

Co-authored-by: Ezra Chung <[email protected]>

Author: kevinAlbs

src/libmongoc/src/mongoc/mongoc-bulkwrite.c

@@ -139,7 +139,12 @@ mongoc_bulkwriteopts_destroy (mongoc_bulkwriteopts_t *self)
 typedef enum { MODEL_OP_INSERT, MODEL_OP_UPDATE, MODEL_OP_DELETE } model_op_t;
 typedef struct {
    model_op_t op;
-   bson_iter_t id_iter;
+   // `id_loc` locates the "_id" field of an insert document.
+   struct {
+      size_t op_start;    // Offset in `mongoc_bulkwrite_t::ops` to the BSON for the insert op: { "document": ... }
+      size_t op_len;      // Length of insert op.
+      uint32_t id_offset; // Offset in the insert op to the "_id" field.
+   } id_loc;
    char *ns;
 } modeldata_t;
 
@@ -277,19 +282,14 @@ mongoc_bulkwrite_append_insertone (mongoc_bulkwrite_t *self,
       persisted_id_offset += existing_id_offset;
    }
 
-   BSON_ASSERT (_mongoc_buffer_append (&self->ops, bson_get_data (&op), op.len));
-
-   // Store an iterator to the document's `_id` in the persisted payload:
-   bson_iter_t persisted_id_iter;
-   {
-      BSON_ASSERT (mcommon_in_range_size_t_unsigned (op.len));
-      size_t start = self->ops.len - (size_t) op.len;
-      BSON_ASSERT (bson_iter_init_from_data_at_offset (
-         &persisted_id_iter, self->ops.data + start, (size_t) op.len, persisted_id_offset, strlen ("_id")));
-   }
+   size_t op_start = self->ops.len; // Save location of `op` to retrieve `_id` later.
+   BSON_ASSERT (mcommon_in_range_size_t_unsigned (op.len));
+   BSON_ASSERT (_mongoc_buffer_append (&self->ops, bson_get_data (&op), (size_t) op.len));
 
    self->n_ops++;
-   modeldata_t md = {.op = MODEL_OP_INSERT, .id_iter = persisted_id_iter, .ns = bson_strdup (ns)};
+   modeldata_t md = {.op = MODEL_OP_INSERT,
+                     .id_loc = {.op_start = op_start, .op_len = (size_t) op.len, .id_offset = persisted_id_offset},
+                     .ns = bson_strdup (ns)};
    _mongoc_array_append_val (&self->arrayof_modeldata, md);
    bson_destroy (&op);
    return true;
@@ -1340,7 +1340,8 @@ static bool
 _bulkwritereturn_apply_result (mongoc_bulkwritereturn_t *self,
                                const bson_t *result,
                                size_t ops_doc_offset,
-                               const mongoc_array_t *arrayof_modeldata)
+                               const mongoc_array_t *arrayof_modeldata,
+                               const mongoc_buffer_t *ops)
 {
    BSON_ASSERT_PARAM (self);
    BSON_ASSERT_PARAM (result);
@@ -1458,7 +1459,10 @@ _bulkwritereturn_apply_result (mongoc_bulkwritereturn_t *self,
          break;
       }
       case MODEL_OP_INSERT: {
-         _bulkwriteresult_set_insertresult (self->res, &md->id_iter, models_idx);
+         bson_iter_t id_iter;
+         BSON_ASSERT (bson_iter_init_from_data_at_offset (
+            &id_iter, ops->data + md->id_loc.op_start, md->id_loc.op_len, md->id_loc.id_offset, strlen ("_id")));
+         _bulkwriteresult_set_insertresult (self->res, &id_iter, models_idx);
          break;
       }
       default:
@@ -1705,9 +1709,9 @@ mongoc_bulkwrite_execute (mongoc_bulkwrite_t *self, const mongoc_bulkwriteopts_t
       bool batch_ok = false;
       bson_t cmd_reply = BSON_INITIALIZER;
       mongoc_cursor_t *reply_cursor = NULL;
-      // `ops_byte_len` is the number of documents from `ops` to send in this batch.
+      // `ops_byte_len` is the number of bytes from `ops` to send in this batch.
       size_t ops_byte_len = 0;
-      // `ops_doc_len` is the number of bytes from `ops` to send in this batch.
+      // `ops_doc_len` is the number of documents from `ops` to send in this batch.
       size_t ops_doc_len = 0;
 
       if (ops_byte_offset == self->ops.len) {
@@ -1903,7 +1907,8 @@ mongoc_bulkwrite_execute (mongoc_bulkwrite_t *self, const mongoc_bulkwriteopts_t
                // Iterate over cursor results.
                const bson_t *result;
                while (mongoc_cursor_next (reply_cursor, &result)) {
-                  if (!_bulkwritereturn_apply_result (&ret, result, ops_doc_offset, &self->arrayof_modeldata)) {
+                  if (!_bulkwritereturn_apply_result (
+                         &ret, result, ops_doc_offset, &self->arrayof_modeldata, &self->ops)) {
                      goto batch_fail;
                   }
                }

src/libmongoc/tests/test-mongoc-bulkwrite.c

@@ -629,6 +629,59 @@ test_bulkwrite_execute_requires_client (void *ctx)
    mongoc_client_destroy (client);
 }
 
+// `test_bulkwrite_two_large_inserts` is a regression test for CDRIVER-5869.
+static void
+test_bulkwrite_two_large_inserts (void *unused)
+{
+   BSON_UNUSED (unused);
+
+   bson_error_t error;
+   mongoc_client_t *client = test_framework_new_default_client ();
+
+   // Drop prior collection:
+   {
+      mongoc_collection_t *coll = mongoc_client_get_collection (client, "db", "coll");
+      mongoc_collection_drop (coll, NULL);
+      mongoc_collection_destroy (coll);
+   }
+
+   // Allocate a large string:
+   size_t large_len = 2095652;
+   char *large_string = bson_malloc (large_len + 1);
+   memset (large_string, 'a', large_len);
+   large_string[large_len] = '\0';
+   ASSERT (mcommon_in_range_unsigned (int, large_len));
+
+   // Create two large documents:
+   bson_t *docs[2];
+   docs[0] = BCON_NEW ("_id", "over_2mib_1");
+   bson_append_utf8 (docs[0], "unencrypted", -1, large_string, (int) large_len);
+   docs[1] = BCON_NEW ("_id", "over_2mib_2");
+   bson_append_utf8 (docs[1], "unencrypted", -1, large_string, (int) large_len);
+
+   mongoc_bulkwriteopts_t *bw_opts = mongoc_bulkwriteopts_new ();
+   mongoc_bulkwriteopts_set_verboseresults (bw_opts, true);
+
+   mongoc_bulkwrite_t *bw = mongoc_client_bulkwrite_new (client);
+   ASSERT_OR_PRINT (mongoc_bulkwrite_append_insertone (bw, "db.coll", docs[0], NULL, &error), error);
+   ASSERT_OR_PRINT (mongoc_bulkwrite_append_insertone (bw, "db.coll", docs[1], NULL, &error), error);
+
+   mongoc_bulkwritereturn_t bwr = mongoc_bulkwrite_execute (bw, bw_opts);
+   ASSERT_NO_BULKWRITEEXCEPTION (bwr);
+   ASSERT (bwr.res);
+   const bson_t *insertresults = mongoc_bulkwriteresult_insertresults (bwr.res);
+   ASSERT_MATCH (insertresults,
+                 BSON_STR ({"0" : {"insertedId" : "over_2mib_1"}}, {"1" : {"insertedId" : "over_2mib_2"}}));
+   bson_destroy (docs[0]);
+   bson_destroy (docs[1]);
+   mongoc_bulkwrite_destroy (bw);
+   mongoc_bulkwriteresult_destroy (bwr.res);
+   mongoc_bulkwriteexception_destroy (bwr.exc);
+   mongoc_bulkwriteopts_destroy (bw_opts);
+   mongoc_client_destroy (client);
+   bson_free (large_string);
+}
+
 void
 test_bulkwrite_install (TestSuite *suite)
 {
@@ -722,4 +775,12 @@ test_bulkwrite_install (TestSuite *suite)
                       NULL /* ctx */,
                       test_framework_skip_if_max_wire_version_less_than_25 // require server 8.0
    );
+
+   TestSuite_AddFull (suite,
+                      "/bulkwrite/two_large_inserts",
+                      test_bulkwrite_two_large_inserts,
+                      NULL /* dtor */,
+                      NULL /* ctx */,
+                      test_framework_skip_if_max_wire_version_less_than_25 // require server 8.0
+   );
 }