feat(ChangeStream): update default startAtOperationTime

Updates the setting of a default startAtOperationTime to be spec-compliant. Also re-syncs change-stream tests. Fixes NODE-1520
mongodb · Jun 20, 2018 · 50a9f65 · 50a9f65
1 parent 770db7f
commit 50a9f65
Show file tree

Hide file tree

Showing 4 changed files with 36 additions and 114 deletions.
diff --git a/lib/change_stream.js b/lib/change_stream.js
@@ -281,25 +281,36 @@ var createChangeStreamCursor = function(self) {
   return changeStreamCursor;
 };
 
+function getResumeToken(self) {
+  return self.resumeToken || self.options.resumeAfter;
+}
+
+function getStartAtOperationTime(self) {
+  const isMaster = self.serverConfig.lastIsMaster() || {};
+  return (
+    isMaster.maxWireVersion && isMaster.maxWireVersion >= 7 && self.options.startAtOperationTime
+  );
+}
+
 var buildChangeStreamAggregationCommand = function(self) {
   const serverConfig = self.serverConfig;
   const namespace = self.namespace;
   const pipeline = self.pipeline;
-  const resumeToken = self.resumeToken;
   const options = self.options;
   const cursorNamespace = self.cursorNamespace;
 
-  const isMaster = serverConfig.lastIsMaster() || {};
-
   var changeStreamStageOptions = {
     fullDocument: options.fullDocument || 'default'
   };
 
-  if (resumeToken || options.resumeAfter) {
-    changeStreamStageOptions.resumeAfter = resumeToken || options.resumeAfter;
-  } else if (isMaster.maxWireVersion && isMaster.maxWireVersion >= 7) {
-    const ts = options.startAtOperationTime || self.operationTime;
-    changeStreamStageOptions.startAtOperationTime = ts;
+  const resumeToken = getResumeToken(self);
+  const startAtOperationTime = getStartAtOperationTime(self);
+  if (resumeToken) {
+    changeStreamStageOptions.resumeAfter = resumeToken;
+  }
+
+  if (startAtOperationTime) {
+    changeStreamStageOptions.startAtOperationTime = startAtOperationTime;
   }
 
   // Map cursor options
@@ -338,6 +349,12 @@ var processNewChange = function(self, err, change, callback) {
     // Handle resumable MongoNetworkErrors
     if (isResumableError(err) && !self.attemptingResume) {
       self.attemptingResume = true;
+
+      if (!(getResumeToken(self) || getStartAtOperationTime(self))) {
+        const startAtOperationTime = self.cursor.cursorState.operationTime;
+        self.options = Object.assign({ startAtOperationTime }, self.options);
+      }
+
       if (callback) {
         return self.cursor.close(function(closeErr) {
           if (closeErr) {

diff --git a/test/functional/change_stream_tests.js b/test/functional/change_stream_tests.js
@@ -1588,6 +1588,10 @@ describe('Change Streams', function() {
           firstBatch: [],
           id: new Long('9064341847921713401'),
           ns: 'test.test'
+        },
+        operationTime: OPERATION_TIME,
+        $clusterTime: {
+          clusterTime: OPERATION_TIME
         }
       };
 
@@ -1652,6 +1656,11 @@ describe('Change Streams', function() {
                 .to.have.nested.property('pipeline[0].$changeStream.startAtOperationTime')
                 .that.deep.equals(OPERATION_TIME);
               expect(doc).to.not.have.nested.property('pipeline[0].$changeStream.resumeAfter');
+            } else {
+              expect(doc).to.not.have.nested.property(
+                'pipeline[0].$changeStream.startAtOperationTime'
+              );
+              expect(doc).to.not.have.nested.property('pipeline[0].$changeStream.resumeAfter');
             }
             return request.reply(AGGREGATE_RESPONSE);
           } else if (doc.getMore) {

diff --git a/test/functional/spec/change-stream/change-streams.json b/test/functional/spec/change-stream/change-streams.json
@@ -246,56 +246,6 @@
         ]
       }
     },
-    {
-      "description": "A fresh ChangeStream against a server >=4.0 will always include startAtOperationTime in the $changeStream stage.",
-      "minServerVersion": "3.8.0",
-      "target": "collection",
-      "topology": [
-        "replicaset"
-      ],
-      "changeStreamPipeline": [],
-      "changeStreamOptions": {},
-      "operations": [
-        {
-          "database": "change-stream-tests",
-          "collection": "test",
-          "name": "insertOne",
-          "arguments": {
-            "document": {
-              "x": 1
-            }
-          }
-        }
-      ],
-      "expectations": [
-        {
-          "command_started_event": {
-            "command": {
-              "aggregate": "test",
-              "cursor": {},
-              "pipeline": [
-                {
-                  "$changeStream": {
-                    "fullDocument": "default",
-                    "startAtOperationTime": {
-                      "$timestamp": {
-                        "i": 42,
-                        "t": 42
-                      }
-                    }
-                  }
-                }
-              ]
-            },
-            "command_name": "aggregate",
-            "database_name": "change-stream-tests"
-          }
-        }
-      ],
-      "result": {
-        "success": []
-      }
-    },
     {
       "description": "Executing a watch helper on a Database results in notifications for changes to all collections in the specified database.",
       "minServerVersion": "3.8.0",
@@ -348,13 +298,7 @@
               "pipeline": [
                 {
                   "$changeStream": {
-                    "fullDocument": "default",
-                    "startAtOperationTime": {
-                      "$timestamp": {
-                        "i": 42,
-                        "t": 42
-                      }
-                    }
+                    "fullDocument": "default"
                   }
                 }
               ]
@@ -446,13 +390,7 @@
                 {
                   "$changeStream": {
                     "fullDocument": "default",
-                    "allChangesForCluster": true,
-                    "startAtOperationTime": {
-                      "$timestamp": {
-                        "i": 42,
-                        "t": 42
-                      }
-                    }
+                    "allChangesForCluster": true
                   }
                 }
               ]

diff --git a/test/functional/spec/change-stream/change-streams.yml b/test/functional/spec/change-stream/change-streams.yml
@@ -166,40 +166,6 @@ tests:
           fullDocument:
             z:
               $numberInt: "3"
-  -
-    description:  A fresh ChangeStream against a server >=4.0 will always include startAtOperationTime in the $changeStream stage.
-    minServerVersion: "3.8.0"
-    target: collection
-    topology:
-      - replicaset
-    changeStreamPipeline: []
-    changeStreamOptions: {}
-    operations:
-      -
-        database: *database_name
-        collection: *collection_name
-        name: insertOne
-        arguments:
-          document:
-            x: 1
-    expectations:
-      - 
-        command_started_event:
-          command:
-            aggregate: *collection_name
-            cursor: {}
-            pipeline:
-              - 
-                $changeStream:
-                  fullDocument: default
-                  startAtOperationTime:
-                    $timestamp:
-                      i: 42
-                      t: 42
-          command_name: aggregate
-          database_name: *database_name
-    result:
-      success: []
   -
     description: Executing a watch helper on a Database results in notifications for changes to all collections in the specified database.
     minServerVersion: "3.8.0"
@@ -241,10 +207,6 @@ tests:
               - 
                 $changeStream:
                   fullDocument: default
-                  startAtOperationTime:
-                    $timestamp:
-                      i: 42
-                      t: 42
           command_name: aggregate
           database_name: *database_name
     result:
@@ -307,10 +269,6 @@ tests:
                 $changeStream:
                   fullDocument: default
                   allChangesForCluster: true
-                  startAtOperationTime:
-                    $timestamp:
-                      i: 42
-                      t: 42
           command_name: aggregate
           database_name: admin
     result: