Merge pull request #677

.travis.yml

@@ -14,7 +14,7 @@ cache:
 
 env:
   global:
-    - DRIVER_VERSION=1.6.0alpha3
+    - DRIVER_VERSION=1.6.0rc1
     - SERVER_DISTRO=ubuntu1604
     - SERVER_VERSION=4.2.0
     - DEPLOYMENT=STANDALONE

src/Client.php

@@ -170,7 +170,7 @@ class Client
             $options['typeMap'] = $this->typeMap;
         }
 
-        $server = $this->manager->selectServer(new ReadPreference(ReadPreference::RP_PRIMARY));
+        $server = select_server($this->manager, new ReadPreference(ReadPreference::RP_PRIMARY), extract_session_from_options($options));
 
         if (! isset($options['writeConcern']) && server_supports_feature($server, self::$wireVersionForWritableCommandWriteConcern)) {
             $options['writeConcern'] = $this->writeConcern;
@@ -246,7 +246,7 @@ class Client
     public function listDatabases(array $options = [])
     {
         $operation = new ListDatabases($options);
-        $server = $this->manager->selectServer(new ReadPreference(ReadPreference::RP_PRIMARY));
+        $server = select_server($this->manager, new ReadPreference(ReadPreference::RP_PRIMARY), extract_session_from_options($options));
 
         return $operation->execute($server);
     }
@@ -311,7 +311,7 @@ class Client
             $options['readPreference'] = $this->readPreference;
         }
 
-        $server = $this->manager->selectServer($options['readPreference']);
+        $server = select_server($this->manager, $options['readPreference'], extract_session_from_options($options));
 
         if (! isset($options['readConcern']) && server_supports_feature($server, self::$wireVersionForReadConcern)) {
             $options['readConcern'] = $this->readConcern;

src/Database.php

@@ -206,7 +206,7 @@ class Database
             $options['readPreference'] = new ReadPreference(ReadPreference::RP_PRIMARY);
         }
 
-        $server = $this->manager->selectServer($options['readPreference']);
+        $server = select_server($this->manager, $options['readPreference'], extract_session_from_options($options));
 
         /* MongoDB 4.2 and later supports a read concern when an $out stage is
          * being used, but earlier versions do not.
@@ -258,7 +258,7 @@ class Database
         }
 
         $operation = new DatabaseCommand($this->databaseName, $command, $options);
-        $server = $this->manager->selectServer($options['readPreference']);
+        $server = select_server($this->manager, $options['readPreference'], extract_session_from_options($options));
 
         return $operation->execute($server);
     }
@@ -280,7 +280,7 @@ class Database
             $options['typeMap'] = $this->typeMap;
         }
 
-        $server = $this->manager->selectServer(new ReadPreference(ReadPreference::RP_PRIMARY));
+        $server = select_server($this->manager, new ReadPreference(ReadPreference::RP_PRIMARY), extract_session_from_options($options));
 
         if (! isset($options['writeConcern']) && server_supports_feature($server, self::$wireVersionForWritableCommandWriteConcern)) {
             $options['writeConcern'] = $this->writeConcern;
@@ -307,7 +307,7 @@ class Database
             $options['typeMap'] = $this->typeMap;
         }
 
-        $server = $this->manager->selectServer(new ReadPreference(ReadPreference::RP_PRIMARY));
+        $server = select_server($this->manager, new ReadPreference(ReadPreference::RP_PRIMARY), extract_session_from_options($options));
 
         if (! isset($options['writeConcern']) && server_supports_feature($server, self::$wireVersionForWritableCommandWriteConcern)) {
             $options['writeConcern'] = $this->writeConcern;
@@ -335,7 +335,7 @@ class Database
             $options['typeMap'] = $this->typeMap;
         }
 
-        $server = $this->manager->selectServer(new ReadPreference(ReadPreference::RP_PRIMARY));
+        $server = select_server($this->manager, new ReadPreference(ReadPreference::RP_PRIMARY), extract_session_from_options($options));
 
         if (! isset($options['writeConcern']) && server_supports_feature($server, self::$wireVersionForWritableCommandWriteConcern)) {
             $options['writeConcern'] = $this->writeConcern;
@@ -420,7 +420,7 @@ class Database
     public function listCollections(array $options = [])
     {
         $operation = new ListCollections($this->databaseName, $options);
-        $server = $this->manager->selectServer(new ReadPreference(ReadPreference::RP_PRIMARY));
+        $server = select_server($this->manager, new ReadPreference(ReadPreference::RP_PRIMARY), extract_session_from_options($options));
 
         return $operation->execute($server);
     }
@@ -442,7 +442,7 @@ class Database
             $options['typeMap'] = $this->typeMap;
         }
 
-        $server = $this->manager->selectServer(new ReadPreference(ReadPreference::RP_PRIMARY));
+        $server = select_server($this->manager, new ReadPreference(ReadPreference::RP_PRIMARY), extract_session_from_options($options));
 
         if (! isset($options['writeConcern']) && server_supports_feature($server, self::$wireVersionForWritableCommandWriteConcern)) {
             $options['writeConcern'] = $this->writeConcern;
@@ -509,7 +509,7 @@ class Database
             $options['readPreference'] = $this->readPreference;
         }
 
-        $server = $this->manager->selectServer($options['readPreference']);
+        $server = select_server($this->manager, $options['readPreference'], extract_session_from_options($options));
 
         if (! isset($options['readConcern']) && server_supports_feature($server, self::$wireVersionForReadConcern)) {
             $options['readConcern'] = $this->readConcern;

src/Operation/Watch.php

@@ -40,6 +40,8 @@ use function is_object;
 use function is_string;
 use function MongoDB\Driver\Monitoring\addSubscriber;
 use function MongoDB\Driver\Monitoring\removeSubscriber;
+use function MongoDB\extract_session_from_options;
+use function MongoDB\select_server;
 use function MongoDB\server_supports_feature;
 
 /**
@@ -375,8 +377,11 @@ class Watch implements Executable, /* @internal */ CommandSubscriber
 
         $this->hasResumed = true;
 
-        // Select a new server using the original read preference
-        $server = $this->manager->selectServer($this->aggregateOptions['readPreference']);
+        /* Select a new server using the original read preference. While watch
+         * is not usable within transactions, we still check if there is a
+         * pinned session. This is to avoid an ambiguous error message about
+         * running a command on the wrong server. */
+        $server = select_server($this->manager, $this->aggregateOptions['readPreference'], extract_session_from_options($this->aggregateOptions));
 
         $resumeOption = isset($this->changeStreamOptions['startAfter']) && ! $hasAdvanced ? 'startAfter' : 'resumeAfter';
 

src/functions.php

@@ -19,6 +19,8 @@ namespace MongoDB;
 
 use Exception;
 use MongoDB\BSON\Serializable;
+use MongoDB\Driver\Manager;
+use MongoDB\Driver\ReadPreference;
 use MongoDB\Driver\Server;
 use MongoDB\Driver\Session;
 use MongoDB\Exception\InvalidArgumentException;
@@ -373,3 +375,32 @@ function with_transaction(Session $session, callable $callback, array $transacti
     $operation = new WithTransaction($callback, $transactionOptions);
     $operation->execute($session);
 }
+
+/**
+ * Returns the session option if it is set and valid.
+ *
+ * @internal
+ * @param array $options
+ * @return Session|null
+ */
+function extract_session_from_options(array $options)
+{
+    if (! isset($options['session']) || ! $options['session'] instanceof Session) {
+        return null;
+    }
+
+    return $options['session'];
+}
+
+/**
+ * Performs server selection, respecting the server a session may be pinned to
+ *
+ * @internal
+ * @return Server
+ */
+function select_server(Manager $manager, ReadPreference $readPreference = null, Session $session = null)
+{
+    $server = $session !== null ? $session->getServer() : null;
+
+    return $server ?: $manager->selectServer($readPreference);
+}

tests/FunctionalTestCase.php

@@ -151,7 +151,7 @@ abstract class FunctionalTestCase extends TestCase
      * @param array|stdClass $command configureFailPoint command document
      * @throws InvalidArgumentException if $command is not a configureFailPoint command
      */
-    protected function configureFailPoint($command)
+    public function configureFailPoint($command, Server $server = null)
     {
         if (! $this->isFailCommandSupported()) {
             $this->markTestSkipped('failCommand is only supported on mongod >= 4.0.0 and mongos >= 4.1.5.');
@@ -173,14 +173,16 @@ abstract class FunctionalTestCase extends TestCase
             throw new InvalidArgumentException('$command is not a configureFailPoint command');
         }
 
+        $failPointServer = $server ?: $this->getPrimaryServer();
+
         $operation = new DatabaseCommand('admin', $command);
-        $cursor = $operation->execute($this->getPrimaryServer());
+        $cursor = $operation->execute($failPointServer);
         $result = $cursor->toArray()[0];
 
         $this->assertCommandSucceeded($result);
 
         // Record the fail point so it can be disabled during tearDown()
-        $this->configuredFailPoints[] = $command->configureFailPoint;
+        $this->configuredFailPoints[] = [$command->configureFailPoint, $failPointServer];
     }
 
     /**
@@ -382,9 +384,16 @@ abstract class FunctionalTestCase extends TestCase
             $this->markTestSkipped('Transactions are not supported on standalone servers');
         }
 
-        // TODO: MongoDB 4.2 should support sharded clusters (see: PHPLIB-374)
         if ($this->isShardedCluster()) {
-            $this->markTestSkipped('Transactions are not supported on sharded clusters');
+            if (! $this->isShardedClusterUsingReplicasets()) {
+                $this->markTestSkipped('Transactions are not supported on sharded clusters without replica sets');
+            }
+
+            if (version_compare($this->getFeatureCompatibilityVersion(), '4.2', '<')) {
+                $this->markTestSkipped('Transactions are only supported on FCV 4.2 or higher');
+            }
+
+            return;
         }
 
         if (version_compare($this->getFeatureCompatibilityVersion(), '4.0', '<')) {
@@ -408,9 +417,7 @@ abstract class FunctionalTestCase extends TestCase
             return;
         }
 
-        $server = $this->getPrimaryServer();
-
-        foreach ($this->configuredFailPoints as $failPoint) {
+        foreach ($this->configuredFailPoints as list($failPoint, $server)) {
             $operation = new DatabaseCommand('admin', ['configureFailPoint' => $failPoint, 'mode' => 'off']);
             $operation->execute($server);
         }

tests/SpecTests/CommandExpectations.php

@@ -10,6 +10,7 @@ use MongoDB\Driver\Monitoring\CommandSubscriber;
 use MongoDB\Driver\Monitoring\CommandSucceededEvent;
 use MultipleIterator;
 use function count;
+use function in_array;
 use function key;
 use function MongoDB\Driver\Monitoring\addSubscriber;
 use function MongoDB\Driver\Monitoring\removeSubscriber;
@@ -37,6 +38,9 @@ class CommandExpectations implements CommandSubscriber
     /** @var boolean */
     private $ignoreExtraEvents = false;
 
+    /** @var string[] */
+    private $ignoredCommandNames = [];
+
     private function __construct(array $events)
     {
         foreach ($events as $event) {
@@ -110,6 +114,14 @@ class CommandExpectations implements CommandSubscriber
         $o->ignoreCommandFailed = true;
         $o->ignoreCommandSucceeded = true;
 
+        /* Ignore the buildInfo and getParameter commands as they are used to
+         * check for the availability of configureFailPoint and are not expected
+         * to be called by any spec tests.
+         * configureFailPoint needs to be ignored as the targetedFailPoint
+         * operation will be caught by command monitoring and is also not
+         * present in the expected commands in spec tests. */
+        $o->ignoredCommandNames = ['buildInfo', 'getParameter', 'configureFailPoint'];
+
         return $o;
     }
 
@@ -120,7 +132,7 @@ class CommandExpectations implements CommandSubscriber
      */
     public function commandFailed(CommandFailedEvent $event)
     {
-        if ($this->ignoreCommandFailed || ($this->ignoreExtraEvents && count($this->actualEvents) === count($this->expectedEvents))) {
+        if ($this->ignoreCommandFailed || $this->isEventIgnored($event)) {
             return;
         }
 
@@ -134,7 +146,7 @@ class CommandExpectations implements CommandSubscriber
      */
     public function commandStarted(CommandStartedEvent $event)
     {
-        if ($this->ignoreCommandStarted || ($this->ignoreExtraEvents && count($this->actualEvents) === count($this->expectedEvents))) {
+        if ($this->ignoreCommandStarted || $this->isEventIgnored($event)) {
             return;
         }
 
@@ -148,7 +160,7 @@ class CommandExpectations implements CommandSubscriber
      */
     public function commandSucceeded(CommandSucceededEvent $event)
     {
-        if ($this->ignoreCommandSucceeded || ($this->ignoreExtraEvents && count($this->actualEvents) === count($this->expectedEvents))) {
+        if ($this->ignoreCommandSucceeded || $this->isEventIgnored($event)) {
             return;
         }
 
@@ -212,4 +224,10 @@ class CommandExpectations implements CommandSubscriber
             }
         }
     }
+
+    private function isEventIgnored($event)
+    {
+        return ($this->ignoreExtraEvents && count($this->actualEvents) === count($this->expectedEvents))
+            || in_array($event->getCommandName(), $this->ignoredCommandNames);
+    }
 }

tests/SpecTests/Context.php

@@ -121,7 +121,7 @@ final class Context
         return $o;
     }
 
-    public static function fromRetryableWrites(stdClass $test, $databaseName, $collectionName)
+    public static function fromRetryableWrites(stdClass $test, $databaseName, $collectionName, $useMultipleMongoses)
     {
         $o = new self($databaseName, $collectionName);
 
@@ -134,12 +134,12 @@ final class Context
             $o->outcomeCollectionName = $test->outcome->collection->name;
         }
 
-        $o->client = new Client(FunctionalTestCase::getUri(), $clientOptions);
+        $o->client = new Client(FunctionalTestCase::getUri($useMultipleMongoses), $clientOptions);
 
         return $o;
     }
 
-    public static function fromTransactions(stdClass $test, $databaseName, $collectionName)
+    public static function fromTransactions(stdClass $test, $databaseName, $collectionName, $useMultipleMongoses)
     {
         $o = new self($databaseName, $collectionName);
 
@@ -159,7 +159,7 @@ final class Context
          * re-using a previously persisted libmongoc client object. */
         $clientOptions += ['p' => mt_rand()];
 
-        $o->client = new Client(FunctionalTestCase::getUri(), $clientOptions);
+        $o->client = new Client(FunctionalTestCase::getUri($useMultipleMongoses), $clientOptions);
 
         $session0Options = isset($test->sessionOptions->session0) ? (array) $test->sessionOptions->session0 : [];
         $session1Options = isset($test->sessionOptions->session1) ? (array) $test->sessionOptions->session1 : [];

tests/SpecTests/Operation.php

@@ -9,6 +9,7 @@ use MongoDB\Database;
 use MongoDB\Driver\Cursor;
 use MongoDB\Driver\Exception\BulkWriteException;
 use MongoDB\Driver\Exception\Exception;
+use MongoDB\Driver\Server;
 use MongoDB\Driver\Session;
 use MongoDB\Driver\WriteConcern;
 use MongoDB\GridFS\Bucket;
@@ -19,6 +20,7 @@ use function array_diff_key;
 use function array_map;
 use function fclose;
 use function fopen;
+use function in_array;
 use function MongoDB\is_last_pipeline_operator_write;
 use function MongoDB\with_transaction;
 use function stream_get_contents;
@@ -37,6 +39,7 @@ final class Operation
     const OBJECT_SELECT_DATABASE = 'selectDatabase';
     const OBJECT_SESSION0 = 'session0';
     const OBJECT_SESSION1 = 'session1';
+    const OBJECT_TEST_RUNNER = 'testRunner';
 
     /** @var ErrorExpectation|null */
     public $errorExpectation;
@@ -280,6 +283,8 @@ final class Operation
                 return $this->executeForSession($context->session0, $test, $context);
             case self::OBJECT_SESSION1:
                 return $this->executeForSession($context->session1, $test, $context);
+            case self::OBJECT_TEST_RUNNER:
+                return $this->executeForTestRunner($test, $context);
             default:
                 throw new LogicException('Unsupported object: ' . $this->object);
         }
@@ -545,6 +550,42 @@ final class Operation
         }
     }
 
+    private function executeForTestRunner(FunctionalTestCase $test, Context $context)
+    {
+        $args = $context->prepareOptions($this->arguments);
+        $context->replaceArgumentSessionPlaceholder($args);
+
+        switch ($this->name) {
+            case 'assertSessionPinned':
+                $test->assertInstanceOf(Session::class, $args['session']);
+                $test->assertInstanceOf(Server::class, $args['session']->getServer());
+
+                return null;
+            case 'assertSessionTransactionState':
+                $test->assertInstanceOf(Session::class, $args['session']);
+                /* PHPC currently does not expose the exact session state, but
+                 * instead exposes a bool to let us know whether a transaction
+                 * is currently in progress. This code may fail down the line
+                 * and should be adjusted once PHPC-1438 is implemented. */
+                $expected = in_array($this->arguments['state'], ['in_progress', 'starting']);
+                $test->assertSame($expected, $args['session']->isInTransaction());
+
+                return null;
+            case 'assertSessionUnpinned':
+                $test->assertInstanceOf(Session::class, $args['session']);
+                $test->assertNull($args['session']->getServer());
+
+                return null;
+            case 'targetedFailPoint':
+                $test->assertInstanceOf(Session::class, $args['session']);
+                $test->configureFailPoint($this->arguments['failPoint'], $args['session']->getServer());
+
+                return null;
+            default:
+                throw new LogicException('Unsupported test runner operation: ' . $this->name);
+        }
+    }
+
     /**
      * @throws LogicException if the operation object is unsupported
      */
@@ -561,6 +602,7 @@ final class Operation
                 return ResultExpectation::ASSERT_SAME;
             case self::OBJECT_SESSION0:
             case self::OBJECT_SESSION1:
+            case self::OBJECT_TEST_RUNNER:
                 return ResultExpectation::ASSERT_NOTHING;
             default:
                 throw new LogicException('Unsupported object: ' . $this->object);

tests/SpecTests/RetryableWritesSpecTest.php

@@ -2,7 +2,6 @@
 
 namespace MongoDB\Tests\SpecTests;
 
-use MongoDB\Driver\Manager;
 use stdClass;
 use function basename;
 use function file_get_contents;
@@ -29,15 +28,13 @@ class RetryableWritesSpecTest extends FunctionalTestCase
             $this->markTestSkipped('Transaction numbers are only allowed on a replica set member or mongos (PHPC-1415)');
         }
 
-        if (isset($test->useMultipleMongoses) && $test->useMultipleMongoses && $this->isShardedCluster()) {
-            $this->manager = new Manager(static::getUri(true));
-        }
+        $useMultipleMongoses = isset($test->useMultipleMongoses) && $test->useMultipleMongoses && $this->isShardedCluster();
 
         if (isset($runOn)) {
             $this->checkServerRequirements($runOn);
         }
 
-        $context = Context::fromRetryableWrites($test, $this->getDatabaseName(), $this->getCollectionName());
+        $context = Context::fromRetryableWrites($test, $this->getDatabaseName(), $this->getCollectionName(), $useMultipleMongoses);
         $this->setContext($context);
 
         $this->dropTestAndOutcomeCollections();

tests/SpecTests/TransactionsSpecTest.php

@@ -4,6 +4,7 @@ namespace MongoDB\Tests\SpecTests;
 
 use MongoDB\BSON\Int64;
 use MongoDB\BSON\Timestamp;
+use MongoDB\Client;
 use MongoDB\Driver\Command;
 use MongoDB\Driver\Exception\ServerException;
 use MongoDB\Driver\Manager;
@@ -11,7 +12,9 @@ use MongoDB\Driver\ReadPreference;
 use MongoDB\Driver\Server;
 use stdClass;
 use Symfony\Bridge\PhpUnit\SetUpTearDownTrait;
+use function array_unique;
 use function basename;
+use function count;
 use function dirname;
 use function file_get_contents;
 use function get_object_vars;
@@ -35,6 +38,9 @@ class TransactionsSpecTest extends FunctionalTestCase
      * @var array
      */
     private static $incompleteTests = [
+        'transactions/mongos-recovery-token: commitTransaction retry fails on new mongos' => 'isMaster failpoints cannot be disabled',
+        'transactions/pin-mongos: remain pinned after non-transient error on commit' => 'Blocked on SPEC-1320',
+        'transactions/pin-mongos: unpin after transient error within a transaction and commit' => 'isMaster failpoints cannot be disabled',
         'transactions/read-pref: default readPreference' => 'PHPLIB does not properly inherit readPreference for transactions (PHPLIB-473)',
         'transactions/read-pref: primary readPreference' => 'PHPLIB does not properly inherit readPreference for transactions (PHPLIB-473)',
         'transactions/run-command: run command with secondary read preference in client option and primary read preference in transaction options' => 'PHPLIB does not properly inherit readPreference for transactions (PHPLIB-473)',
@@ -48,6 +54,8 @@ class TransactionsSpecTest extends FunctionalTestCase
         parent::setUp();
 
         static::killAllSessions();
+
+        $this->skipIfTransactionsAreNotSupported();
     }
 
     private function doTearDown()
@@ -128,13 +136,11 @@ class TransactionsSpecTest extends FunctionalTestCase
             $this->markTestIncomplete(self::$incompleteTests[$this->dataDescription()]);
         }
 
-        if ($this->isShardedCluster()) {
-            $this->markTestSkipped('PHP MongoDB driver 1.6.0alpha2 does not support running multi-document transactions on sharded clusters');
+        if (isset($test->skipReason)) {
+            $this->markTestSkipped($test->skipReason);
         }
 
-        if (isset($test->useMultipleMongoses) && $test->useMultipleMongoses && $this->isShardedCluster()) {
-            $this->manager = new Manager(static::getUri(true));
-        }
+        $useMultipleMongoses = isset($test->useMultipleMongoses) && $test->useMultipleMongoses && $this->isShardedCluster();
 
         if (isset($runOn)) {
             $this->checkServerRequirements($runOn);
@@ -147,7 +153,7 @@ class TransactionsSpecTest extends FunctionalTestCase
         $databaseName = isset($databaseName) ? $databaseName : $this->getDatabaseName();
         $collectionName = isset($collectionName) ? $collectionName : $this->getCollectionName();
 
-        $context = Context::fromTransactions($test, $databaseName, $collectionName);
+        $context = Context::fromTransactions($test, $databaseName, $collectionName, $useMultipleMongoses);
         $this->setContext($context);
 
         $this->dropTestAndOutcomeCollections();
@@ -202,6 +208,80 @@ class TransactionsSpecTest extends FunctionalTestCase
         return $testArgs;
     }
 
+    /**
+     * Prose test 1: Test that starting a new transaction on a pinned
+     * ClientSession unpins the session and normal server selection is performed
+     * for the next operation.
+     */
+    public function testStartingNewTransactionOnPinnedSessionUnpinsSession()
+    {
+        if (! $this->isShardedClusterUsingReplicasets()) {
+            $this->markTestSkipped('Mongos pinning tests can only run on sharded clusters using replica sets');
+        }
+
+        $client = new Client($this->getUri(true));
+
+        $session = $client->startSession();
+        $collection = $client->selectCollection($this->getDatabaseName(), $this->getCollectionName());
+
+        // Create collection before transaction
+        $collection->insertOne([]);
+
+        $session->startTransaction([]);
+        $collection->insertOne([], ['session' => $session]);
+        $session->commitTransaction();
+
+        $servers = [];
+        for ($i = 0; $i < 50; $i++) {
+            $session->startTransaction([]);
+            $cursor = $collection->find([], ['session' => $session]);
+            $servers[] = $cursor->getServer()->getHost() . ':' . $cursor->getServer()->getPort();
+            $this->assertInstanceOf(Server::class, $session->getServer());
+            $session->commitTransaction();
+        }
+
+        $servers = array_unique($servers);
+        $this->assertGreaterThan(1, count($servers));
+
+        $session->endSession();
+    }
+
+    /**
+     * Prose test 2: Test non-transaction operations using a pinned
+     * ClientSession unpins the session and normal server selection is
+     * performed.
+     */
+    public function testRunningNonTransactionOperationOnPinnedSessionUnpinsSession()
+    {
+        if (! $this->isShardedClusterUsingReplicasets()) {
+            $this->markTestSkipped('Mongos pinning tests can only run on sharded clusters using replica sets');
+        }
+
+        $client = new Client($this->getUri(true));
+
+        $session = $client->startSession();
+        $collection = $client->selectCollection($this->getDatabaseName(), $this->getCollectionName());
+
+        // Create collection before transaction
+        $collection->insertOne([]);
+
+        $session->startTransaction([]);
+        $collection->insertOne([], ['session' => $session]);
+        $session->commitTransaction();
+
+        $servers = [];
+        for ($i = 0; $i < 50; $i++) {
+            $cursor = $collection->find([], ['session' => $session]);
+            $servers[] = $cursor->getServer()->getHost() . ':' . $cursor->getServer()->getPort();
+            $this->assertNull($session->getServer());
+        }
+
+        $servers = array_unique($servers);
+        $this->assertGreaterThan(1, count($servers));
+
+        $session->endSession();
+    }
+
     /**
      * Create the collection, since it cannot be created within a transaction.
      */

tests/SpecTests/transactions/errors-client.json

+{
+  "runOn": [
+    {
+      "minServerVersion": "4.0",
+      "topology": [
+        "replicaset"
+      ]
+    },
+    {
+      "minServerVersion": "4.1.8",
+      "topology": [
+        "sharded"
+      ]
+    }
+  ],
+  "database_name": "transaction-tests",
+  "collection_name": "test",
+  "data": [],
+  "tests": [
+    {
+      "description": "Client side error in command starting transaction",
+      "operations": [
+        {
+          "name": "startTransaction",
+          "object": "session0"
+        },
+        {
+          "name": "insertOne",
+          "object": "collection",
+          "arguments": {
+            "session": "session0",
+            "document": {
+              "_id": {
+                ".": "."
+              }
+            }
+          },
+          "error": true
+        },
+        {
+          "name": "assertSessionTransactionState",
+          "object": "testRunner",
+          "arguments": {
+            "session": "session0",
+            "state": "starting"
+          }
+        }
+      ]
+    },
+    {
+      "description": "Client side error when transaction is in progress",
+      "operations": [
+        {
+          "name": "startTransaction",
+          "object": "session0"
+        },
+        {
+          "name": "insertOne",
+          "object": "collection",
+          "arguments": {
+            "session": "session0",
+            "document": {
+              "_id": 4
+            }
+          },
+          "result": {
+            "insertedId": 4
+          }
+        },
+        {
+          "name": "insertOne",
+          "object": "collection",
+          "arguments": {
+            "session": "session0",
+            "document": {
+              "_id": {
+                ".": "."
+              }
+            }
+          },
+          "error": true
+        },
+        {
+          "name": "assertSessionTransactionState",
+          "object": "testRunner",
+          "arguments": {
+            "session": "session0",
+            "state": "in_progress"
+          }
+        }
+      ]
+    }
+  ]
+}

tests/SpecTests/transactions/retryable-abort.json

@@ -705,7 +705,7 @@
       }
     },
     {
-      "description": "abortTransaction succeeds after InterruptedDueToStepDown",
+      "description": "abortTransaction succeeds after InterruptedDueToReplStateChange",
       "failPoint": {
         "configureFailPoint": "failCommand",
         "mode": {
@@ -1627,7 +1627,7 @@
       }
     },
     {
-      "description": "abortTransaction succeeds after WriteConcernError InterruptedDueToStepDown",
+      "description": "abortTransaction succeeds after WriteConcernError InterruptedDueToReplStateChange",
       "failPoint": {
         "configureFailPoint": "failCommand",
         "mode": {

tests/SpecTests/transactions/retryable-commit.json

@@ -942,7 +942,7 @@
       }
     },
     {
-      "description": "commitTransaction succeeds after InterruptedDueToStepDown",
+      "description": "commitTransaction succeeds after InterruptedDueToReplStateChange",
       "failPoint": {
         "configureFailPoint": "failCommand",
         "mode": {
@@ -1925,7 +1925,7 @@
       }
     },
     {
-      "description": "commitTransaction succeeds after WriteConcernError InterruptedDueToStepDown",
+      "description": "commitTransaction succeeds after WriteConcernError InterruptedDueToReplStateChange",
       "failPoint": {
         "configureFailPoint": "failCommand",
         "mode": {