PHPLIB-537: Use whitelist to check if a change stream is resumable

src/ChangeStream.php

@@ -42,13 +42,31 @@ class ChangeStream implements Iterator
      */
     const CURSOR_NOT_FOUND = 43;
 
-    /** @var array */
-    private static $nonResumableErrorCodes = [
-        136, // CappedPositionLost
-        237, // CursorKilled
-        11601, // Interrupted
+    /** @var int[] */
+    private static $resumableErrorCodes = [
+        6, // HostUnreachable
+        7, // HostNotFound
+        89, // NetworkTimeout
+        91, // ShutdownInProgress
+        189, // PrimarySteppedDown
+        262, // ExceededTimeLimit
+        9001, // SocketException
+        10107, // NotMaster
+        11600, // InterruptedAtShutdown
+        11602, // InterruptedDueToReplStateChange
+        13435, // NotMasterNoSlaveOk
+        13436, // NotMasterOrSecondary
+        63, // StaleShardVersion
+        150, // StaleEpoch
+        13388, // StaleConfig
+        234, // RetryChangeStream
+        133, // FailedToSatisfyReadPreference
+        216, // ElectionInProgress
     ];
 
+    /** @var int */
+    private static $wireVersionForResumableChangeStreamError = 9;
+
     /** @var callable */
     private $resumeCallable;
 
@@ -180,15 +198,11 @@ class ChangeStream implements Iterator
             return false;
         }
 
-        if ($exception->hasErrorLabel('NonResumableChangeStreamError')) {
-            return false;
-        }
-
-        if (in_array($exception->getCode(), self::$nonResumableErrorCodes)) {
-            return false;
+        if (server_supports_feature($this->iterator->getServer(), self::$wireVersionForResumableChangeStreamError)) {
+            return $exception->hasErrorLabel('ResumableChangeStreamError');
         }
 
-        return true;
+        return in_array($exception->getCode(), self::$resumableErrorCodes);
     }
 
     /**

src/Model/ChangeStreamIterator.php

@@ -24,6 +24,7 @@ use MongoDB\Driver\Monitoring\CommandFailedEvent;
 use MongoDB\Driver\Monitoring\CommandStartedEvent;
 use MongoDB\Driver\Monitoring\CommandSubscriber;
 use MongoDB\Driver\Monitoring\CommandSucceededEvent;
+use MongoDB\Driver\Server;
 use MongoDB\Exception\InvalidArgumentException;
 use MongoDB\Exception\ResumeTokenException;
 use MongoDB\Exception\UnexpectedValueException;
@@ -63,6 +64,9 @@ class ChangeStreamIterator extends IteratorIterator implements CommandSubscriber
     /** @var array|object|null */
     private $resumeToken;
 
+    /** @var Server */
+    private $server;
+
     /**
      * @internal
      * @param Cursor            $cursor
@@ -90,6 +94,7 @@ class ChangeStreamIterator extends IteratorIterator implements CommandSubscriber
         $this->isRewindNop = ($firstBatchSize === 0);
         $this->postBatchResumeToken = $postBatchResumeToken;
         $this->resumeToken = $initialResumeToken;
+        $this->server = $cursor->getServer();
     }
 
     /** @internal */
@@ -152,6 +157,14 @@ class ChangeStreamIterator extends IteratorIterator implements CommandSubscriber
         return $this->resumeToken;
     }
 
+    /**
+     * Returns the server the cursor is running on.
+     */
+    public function getServer() : Server
+    {
+        return $this->server;
+    }
+
     /**
      * @see https://php.net/iteratoriterator.key
      * @return mixed

tests/Operation/WatchFunctionalTest.php

@@ -155,49 +155,6 @@ class WatchFunctionalTest extends FunctionalTestCase
         $this->assertSameDocument($postBatchResumeToken, $changeStream->getResumeToken());
     }
 
-    /**
-     * Prose test 10: "ChangeStream will resume after a killCursors command is
-     * issued for its child cursor."
-     */
-    public function testNextResumesAfterCursorNotFound()
-    {
-        $operation = new Watch($this->manager, $this->getDatabaseName(), $this->getCollectionName(), [], $this->defaultOptions);
-        $changeStream = $operation->execute($this->getPrimaryServer());
-
-        $changeStream->rewind();
-        $this->assertFalse($changeStream->valid());
-
-        $this->insertDocument(['_id' => 1, 'x' => 'foo']);
-
-        $this->advanceCursorUntilValid($changeStream);
-
-        $expectedResult = [
-            '_id' => $changeStream->current()->_id,
-            'operationType' => 'insert',
-            'fullDocument' => ['_id' => 1, 'x' => 'foo'],
-            'ns' => ['db' => $this->getDatabaseName(), 'coll' => $this->getCollectionName()],
-            'documentKey' => ['_id' => 1],
-        ];
-
-        $this->assertMatchesDocument($expectedResult, $changeStream->current());
-
-        $this->killChangeStreamCursor($changeStream);
-
-        $this->insertDocument(['_id' => 2, 'x' => 'bar']);
-
-        $this->advanceCursorUntilValid($changeStream);
-
-        $expectedResult = [
-            '_id' => $changeStream->current()->_id,
-            'operationType' => 'insert',
-            'fullDocument' => ['_id' => 2, 'x' => 'bar'],
-            'ns' => ['db' => $this->getDatabaseName(), 'coll' => $this->getCollectionName()],
-            'documentKey' => ['_id' => 2],
-        ];
-
-        $this->assertMatchesDocument($expectedResult, $changeStream->current());
-    }
-
     public function testNextResumesAfterConnectionException()
     {
         /* In order to trigger a dropped connection, we'll use a new client with
@@ -731,41 +688,6 @@ class WatchFunctionalTest extends FunctionalTestCase
         $this->assertFalse($cursor->isDead());
     }
 
-    /**
-     * Prose test 5: "ChangeStream will not attempt to resume after encountering
-     * error code 11601 (Interrupted), 136 (CappedPositionLost), or 237
-     * (CursorKilled) while executing a getMore command."
-     *
-     * @dataProvider provideNonResumableErrorCodes
-     */
-    public function testNonResumableErrorCodes($errorCode)
-    {
-        $this->configureFailPoint([
-            'configureFailPoint' => 'failCommand',
-            'mode' => ['times' => 1],
-            'data' => ['failCommands' => ['getMore'], 'errorCode' => $errorCode],
-        ]);
-
-        $this->insertDocument(['x' => 1]);
-
-        $operation = new Watch($this->manager, $this->getDatabaseName(), $this->getCollectionName(), []);
-        $changeStream = $operation->execute($this->getPrimaryServer());
-        $changeStream->rewind();
-
-        $this->expectException(ServerException::class);
-        $this->expectExceptionCode($errorCode);
-        $changeStream->next();
-    }
-
-    public function provideNonResumableErrorCodes()
-    {
-        return [
-            'CappedPositionLost' => [136],
-            'CursorKilled' => [237],
-            'Interrupted' => [11601],
-        ];
-    }
-
     /**
      * Prose test 2: "ChangeStream will throw an exception if the server
      * response is missing the resume token (if wire version is < 8, this is a

tests/SpecTests/ChangeStreamsSpecTest.php

@@ -4,6 +4,7 @@ namespace MongoDB\Tests\SpecTests;
 
 use ArrayIterator;
 use LogicException;
+use MongoDB\BSON\Int64;
 use MongoDB\ChangeStream;
 use MongoDB\Driver\Exception\Exception;
 use MongoDB\Model\BSONDocument;
@@ -22,16 +23,27 @@ use function glob;
 class ChangeStreamsSpecTest extends FunctionalTestCase
 {
     /** @var array */
-    private static $incompleteTests = ['change-streams-errors: Change Stream should error when _id is projected out' => 'PHPC-1419'];
+    private static $incompleteTests = [];
 
     /**
      * Assert that the expected and actual command documents match.
      *
+     * Note: this method may modify the $expected object.
+     *
      * @param stdClass $expected Expected command document
      * @param stdClass $actual   Actual command document
      */
     public static function assertCommandMatches(stdClass $expected, stdClass $actual)
     {
+        if (isset($expected->getMore) && $expected->getMore === 42) {
+            static::assertObjectHasAttribute('getMore', $actual);
+            static::assertThat($actual->getMore, static::logicalOr(
+                static::isInstanceOf(Int64::class),
+                static::isType('integer')
+            ));
+            unset($expected->getMore);
+        }
+
         static::assertDocumentsMatch($expected, $actual);
     }
 
@@ -76,7 +88,7 @@ class ChangeStreamsSpecTest extends FunctionalTestCase
 
         $this->checkServerRequirements($this->createRunOn($test));
 
-        if (! isset($databaseName, $collectionName, $database2Name, $collection2Name)) {
+        if (! isset($databaseName, $collectionName)) {
             $this->fail('Required database and collection names are unset');
         }
 
@@ -84,7 +96,10 @@ class ChangeStreamsSpecTest extends FunctionalTestCase
         $this->setContext($context);
 
         $this->dropDatabasesAndCreateCollection($databaseName, $collectionName);
-        $this->dropDatabasesAndCreateCollection($database2Name, $collection2Name);
+
+        if (isset($database2Name, $collection2Name)) {
+            $this->dropDatabasesAndCreateCollection($database2Name, $collection2Name);
+        }
 
         if (isset($test->failPoint)) {
             $this->configureFailPoint($test->failPoint);

tests/SpecTests/change-streams/change-streams-errors.json

@@ -54,9 +54,7 @@
               "cursor": {},
               "pipeline": [
                 {
-                  "$changeStream": {
-                    "fullDocument": "default"
-                  }
+                  "$changeStream": {}
                 },
                 {
                   "$unsupported": "foo"
@@ -104,10 +102,54 @@
       ],
       "result": {
         "error": {
-          "code": 280,
-          "errorLabels": [
-            "NonResumableChangeStreamError"
-          ]
+          "code": 280
+        }
+      }
+    },
+    {
+      "description": "change stream errors on MaxTimeMSExpired",
+      "minServerVersion": "4.2",
+      "failPoint": {
+        "configureFailPoint": "failCommand",
+        "mode": {
+          "times": 1
+        },
+        "data": {
+          "failCommands": [
+            "getMore"
+          ],
+          "errorCode": 50,
+          "closeConnection": false
+        }
+      },
+      "target": "collection",
+      "topology": [
+        "replicaset",
+        "sharded"
+      ],
+      "changeStreamPipeline": [
+        {
+          "$project": {
+            "_id": 0
+          }
+        }
+      ],
+      "changeStreamOptions": {},
+      "operations": [
+        {
+          "database": "change-stream-tests",
+          "collection": "test",
+          "name": "insertOne",
+          "arguments": {
+            "document": {
+              "z": 3
+            }
+          }
+        }
+      ],
+      "result": {
+        "error": {
+          "code": 50
         }
       }
     }

tests/SpecTests/change-streams/change-streams-errors.yml

-collection_name: &collection_name "test"
-database_name: &database_name "change-stream-tests"
-collection2_name: &collection2_name "test2"
-database2_name: &database2_name "change-stream-tests-2"
-tests:
-  -
-    description: The watch helper must not throw a custom exception when executed against a single server topology, but instead depend on a server error
-    minServerVersion: "3.6.0"
-    target: collection
-    topology:
-      - single
-    changeStreamPipeline: []
-    changeStreamOptions: {}
-    operations: []
-    expectations: []
-    result:
-      error:
-        code: 40573
-  -
-    description: Change Stream should error when an invalid aggregation stage is passed in
-    minServerVersion: "3.6.0"
-    target: collection
-    topology:
-      - replicaset
-    changeStreamPipeline:
-      -
-        $unsupported: foo
-    changeStreamOptions: {}
-    operations:
-      -
-        database: *database_name
-        collection: *collection_name
-        name: insertOne
-        arguments:
-          document:
-            z: 3
-    expectations:
-      -
-        command_started_event:
-          command:
-            aggregate: *collection_name
-            cursor: {}
-            pipeline:
-              - 
-                $changeStream:
-                  fullDocument: default
-              -
-                $unsupported: foo
-          command_name: aggregate
-          database_name: *database_name
-    result:
-      error:
-        code: 40324
-  -
-    description: Change Stream should error when _id is projected out
-    minServerVersion: "4.1.11"
-    target: collection
-    topology:
-      - replicaset
-      - sharded
-    changeStreamPipeline:
-      -
-        $project: { _id: 0 }
-    changeStreamOptions: {}
-    operations:
-      -
-        database: *database_name
-        collection: *collection_name
-        name: insertOne
-        arguments:
-          document:
-            z: 3
-    result:
-      error:
-        code: 280
-        errorLabels: [ "NonResumableChangeStreamError" ]

tests/SpecTests/change-streams/change-streams.json

@@ -33,9 +33,7 @@
               "cursor": {},
               "pipeline": [
                 {
-                  "$changeStream": {
-                    "fullDocument": "default"
-                  }
+                  "$changeStream": {}
                 }
               ]
             },
@@ -153,9 +151,7 @@
               "cursor": {},
               "pipeline": [
                 {
-                  "$changeStream": {
-                    "fullDocument": "default"
-                  }
+                  "$changeStream": {}
                 }
               ]
             },
@@ -226,9 +222,7 @@
               "cursor": {},
               "pipeline": [
                 {
-                  "$changeStream": {
-                    "fullDocument": "default"
-                  }
+                  "$changeStream": {}
                 },
                 {
                   "$match": {
@@ -312,9 +306,7 @@
               "cursor": {},
               "pipeline": [
                 {
-                  "$changeStream": {
-                    "fullDocument": "default"
-                  }
+                  "$changeStream": {}
                 }
               ]
             },
@@ -404,7 +396,6 @@
               "pipeline": [
                 {
                   "$changeStream": {
-                    "fullDocument": "default",
                     "allChangesForCluster": true
                   }
                 }
@@ -523,9 +514,7 @@
               "cursor": {},
               "pipeline": [
                 {
-                  "$changeStream": {
-                    "fullDocument": "default"
-                  }
+                  "$changeStream": {}
                 }
               ]
             },
@@ -611,9 +600,7 @@
               "cursor": {},
               "pipeline": [
                 {
-                  "$changeStream": {
-                    "fullDocument": "default"
-                  }
+                  "$changeStream": {}
                 }
               ]
             },
@@ -665,9 +652,7 @@
               "cursor": {},
               "pipeline": [
                 {
-                  "$changeStream": {
-                    "fullDocument": "default"
-                  }
+                  "$changeStream": {}
                 }
               ]
             },
@@ -756,9 +741,7 @@
               },
               "pipeline": [
                 {
-                  "$changeStream": {
-                    "fullDocument": "default"
-                  }
+                  "$changeStream": {}
                 }
               ]
             },