GODRIVER-2388 Improved Bulk Write API.

internal/driverutil/operation.go

@@ -28,4 +28,5 @@ const (
 	ListIndexesOp       = "listIndexes"       // ListIndexesOp is the name for listing indexes
 	ListDatabasesOp     = "listDatabases"     // ListDatabasesOp is the name for listing databases
 	UpdateOp            = "update"            // UpdateOp is the name for updating
+	BulkWriteOp         = "bulkWrite"         // BulkWriteOp is the name for client-level bulk write
 )

internal/integration/client_side_encryption_test.go

@@ -398,7 +398,7 @@ func TestClientSideEncryptionCustomCrypt(t *testing.T) {
 			"expected 0 calls to DecryptExplicit, got %v", cc.numDecryptExplicitCalls)
 		assert.Equal(mt, cc.numCloseCalls, 0,
 			"expected 0 calls to Close, got %v", cc.numCloseCalls)
-		assert.Equal(mt, cc.numBypassAutoEncryptionCalls, 2,
+		assert.Equal(mt, cc.numBypassAutoEncryptionCalls, 1,
 			"expected 2 calls to BypassAutoEncryption, got %v", cc.numBypassAutoEncryptionCalls)
 	})
 }

internal/integration/csot_prose_test.go

@@ -177,6 +177,60 @@ func TestCSOTProse(t *testing.T) {
 				"expected ping to fail within 150ms")
 		})
 	})
+
+	mt.RunOpts("11. multi-batch bulkWrites", mtest.NewOptions().MinServerVersion("8.0").
+		AtlasDataLake(false).Topologies(mtest.Single), func(mt *mtest.T) {
+		coll := mt.CreateCollection(mtest.Collection{DB: "db", Name: "coll"}, false)
+		err := coll.Drop(context.Background())
+		require.NoError(mt, err, "Drop error: %v", err)
+
+		mt.SetFailPoint(failpoint.FailPoint{
+			ConfigureFailPoint: "failCommand",
+			Mode: failpoint.Mode{
+				Times: 2,
+			},
+			Data: failpoint.Data{
+				FailCommands:    []string{"bulkWrite"},
+				BlockConnection: true,
+				BlockTimeMS:     1010,
+			},
+		})
+
+		var hello struct {
+			MaxBsonObjectSize   int
+			MaxMessageSizeBytes int
+		}
+		err = mt.DB.RunCommand(context.Background(), bson.D{{"hello", 1}}).Decode(&hello)
+		require.NoError(mt, err, "Hello error: %v", err)
+
+		models := &mongo.ClientWriteModels{}
+		n := hello.MaxMessageSizeBytes/hello.MaxBsonObjectSize + 1
+		for i := 0; i < n; i++ {
+			models.
+				AppendInsertOne("db", "coll", &mongo.ClientInsertOneModel{
+					Document: bson.D{{"a", strings.Repeat("b", hello.MaxBsonObjectSize-500)}},
+				})
+		}
+
+		var cnt int
+		cm := &event.CommandMonitor{
+			Started: func(_ context.Context, evt *event.CommandStartedEvent) {
+				if evt.CommandName == "bulkWrite" {
+					cnt++
+				}
+			},
+		}
+		cliOptions := options.Client().
+			SetTimeout(2 * time.Second).
+			SetMonitor(cm).
+			ApplyURI(mtest.ClusterURI())
+		integtest.AddTestServerAPIVersion(cliOptions)
+		cli, err := mongo.Connect(cliOptions)
+		require.NoError(mt, err, "Connect error: %v", err)
+		_, err = cli.BulkWrite(context.Background(), models)
+		assert.ErrorContains(mt, err, "context deadline exceeded", "expected a timeout error, got: %v", err)
+		assert.Equal(mt, 2, cnt, "expected bulkWrite calls: %d, got: %d", 2, cnt)
+	})
 }
 
 func TestCSOTProse_GridFS(t *testing.T) {