minor: return an error if bulk_write is called on an unsupported server (#1117)

Author: isabelatkinson

src/operation.rs

@@ -76,6 +76,7 @@ pub(crate) use update::{Update, UpdateOrReplace};
 
 const SERVER_4_2_0_WIRE_VERSION: i32 = 8;
 const SERVER_4_4_0_WIRE_VERSION: i32 = 9;
+const SERVER_8_0_0_WIRE_VERSION: i32 = 25;
 // The maximum number of bytes that may be included in a write payload when auto-encryption is
 // enabled.
 const MAX_ENCRYPTED_WRITE_SIZE: usize = 2_097_152;

src/operation/bulk_write.rs

@@ -22,7 +22,13 @@ use crate::{
     SessionCursor,
 };
 
-use super::{ExecutionContext, Retryability, WriteResponseBody, OP_MSG_OVERHEAD_BYTES};
+use super::{
+    ExecutionContext,
+    Retryability,
+    WriteResponseBody,
+    OP_MSG_OVERHEAD_BYTES,
+    SERVER_8_0_0_WIRE_VERSION,
+};
 
 use server_responses::*;
 
@@ -174,6 +180,13 @@ where
     const NAME: &'static str = "bulkWrite";
 
     fn build(&mut self, description: &StreamDescription) -> Result<Command> {
+        if description.max_wire_version.unwrap_or(0) < SERVER_8_0_0_WIRE_VERSION {
+            return Err(ErrorKind::IncompatibleServer {
+                message: "the bulk write feature is only supported on MongoDB 8.0+".to_string(),
+            }
+            .into());
+        }
+
         let max_message_size: usize =
             Checked::new(description.max_message_size_bytes).try_into()?;
         let max_operations: usize = Checked::new(description.max_write_batch_size).try_into()?;

src/test/bulk_write.rs

@@ -564,3 +564,21 @@ async fn encryption_error() {
         Some("bulkWrite does not currently support automatic encryption".to_string())
     );
 }
+
+#[tokio::test]
+async fn unsupported_server_client_error() {
+    let client = Client::test_builder().build().await;
+
+    if client.server_version_gte(8, 0) {
+        return;
+    }
+
+    let error = client
+        .bulk_write(vec![InsertOneModel::builder()
+            .namespace(Namespace::new("db", "coll"))
+            .document(doc! { "a": "b" })
+            .build()])
+        .await
+        .unwrap_err();
+    assert!(matches!(*error.kind, ErrorKind::IncompatibleServer { .. }));
+}