RUST-865 Verify correct cursor behavior with SBE (#456)

Author: abr-egn

src/test/cursor.rs

@@ -1,12 +1,12 @@
 use std::time::Duration;
 
-use futures::{future::Either, StreamExt};
+use futures::{future::Either, StreamExt, TryStreamExt};
 use tokio::sync::RwLockReadGuard;
 
 use crate::{
     bson::doc,
     options::{CreateCollectionOptions, CursorType, FindOptions},
-    test::{TestClient, LOCK},
+    test::{util::EventClient, TestClient, LOCK},
     RUNTIME,
 };
 
@@ -109,3 +109,50 @@ async fn session_cursor_next() {
         );
     }
 }
+
+#[cfg_attr(feature = "tokio-runtime", tokio::test)]
+#[cfg_attr(feature = "async-std-runtime", async_std::test)]
+async fn batch_exhaustion() {
+    let _guard: RwLockReadGuard<()> = LOCK.run_concurrently().await;
+    let client = EventClient::new().await;
+
+    let coll = client
+        .create_fresh_collection(
+            "cursor_batch_exhaustion_db",
+            "cursor_batch_exhaustion_coll",
+            None,
+        )
+        .await;
+    coll.insert_many(
+        vec![
+            doc! { "foo": 1 },
+            doc! { "foo": 2 },
+            doc! { "foo": 3 },
+            doc! { "foo": 4 },
+            doc! { "foo": 5 },
+            doc! { "foo": 6 },
+        ],
+        None,
+    )
+    .await
+    .unwrap();
+
+    // Start a find where batch size will line up with limit.
+    let cursor = coll
+        .find(None, FindOptions::builder().batch_size(2).limit(4).build())
+        .await
+        .unwrap();
+    let v: Vec<_> = cursor.try_collect().await.unwrap();
+    assert_eq!(4, v.len());
+
+    // Assert that the last `getMore` response always has id 0, i.e. is exhausted.
+    let replies: Vec<_> = client
+        .get_command_events(&["getMore"])
+        .into_iter()
+        .filter_map(|e| e.as_command_succeeded().map(|e| e.reply.clone()))
+        .collect();
+    let last = replies.last().unwrap();
+    let cursor = last.get_document("cursor").unwrap();
+    let id = cursor.get_i64("id").unwrap();
+    assert_eq!(0, id);
+}

src/test/spec/command_monitoring/mod.rs

@@ -47,7 +47,7 @@ async fn run_command_monitoring_test(test_file: TestFile) {
         "A successful mixed bulk write",
         "A successful unordered bulk write with an unacknowledged write concern",
         // We can't pass this test since it relies on old OP_QUERY behavior (SPEC-1519)
-        "A successful find event with a getmore and the server kills the cursor",
+        "A successful find event with a getmore and the server kills the cursor (<= 4.4)",
     ];
 
     for test_case in test_file.tests {

src/test/spec/json/command-monitoring/legacy/find.json

@@ -413,8 +413,9 @@
       ]
     },
     {
-      "description": "A successful find event with a getmore and the server kills the cursor",
+      "description": "A successful find event with a getmore and the server kills the cursor (<= 4.4)",
       "ignore_if_server_version_less_than": "3.1",
+      "ignore_if_server_version_greater_than": "4.4",
       "ignore_if_topology_type": [
         "sharded"
       ],

src/test/spec/json/command-monitoring/legacy/find.yml

@@ -43,14 +43,13 @@ tests:
         filter: { _id: { $gt: 1 } }
         sort: { _id: 1 }
         skip: {"$numberLong": "2"}
-        modifiers:
-          $comment: "test"
-          $hint: { _id: 1 }
-          $max: { _id: 6 }
-          $maxTimeMS: 6000
-          $min: { _id: 0 }
-          $returnKey: false
-          $showDiskLoc: false
+        comment: "test"
+        hint: { _id: 1 }
+        max: { _id: 6 }
+        maxTimeMS: 6000
+        min: { _id: 0 }
+        returnKey: false
+        showRecordId: false
     expectations:
       -
         command_started_event:
@@ -195,8 +194,9 @@ tests:
               - { $numberLong: "42" }
           command_name: "killCursors"
   -
-    description: "A successful find event with a getmore and the server kills the cursor"
+    description: "A successful find event with a getmore and the server kills the cursor (<= 4.4)"
     ignore_if_server_version_less_than: "3.1"
+    ignore_if_server_version_greater_than: "4.4"
     ignore_if_topology_type:
       - "sharded"
     operation:

src/test/util/event.rs

@@ -100,14 +100,14 @@ impl CommandEvent {
         }
     }
 
-    fn as_command_started(&self) -> Option<&CommandStartedEvent> {
+    pub fn as_command_started(&self) -> Option<&CommandStartedEvent> {
         match self {
             CommandEvent::Started(e) => Some(e),
             _ => None,
         }
     }
 
-    fn as_command_succeeded(&self) -> Option<&CommandSucceededEvent> {
+    pub fn as_command_succeeded(&self) -> Option<&CommandSucceededEvent> {
         match self {
             CommandEvent::Succeeded(e) => Some(e),
             _ => None,