PHPLIB-463: Add test runner for retryable reads

tests/SpecTests/CommandExpectations.php

@@ -77,6 +77,21 @@ class CommandExpectations implements CommandSubscriber
         return $o;
     }
 
+    public static function fromRetryableReads(array $expectedEvents)
+    {
+        $o = new self($expectedEvents);
+
+        $o->ignoreCommandFailed = true;
+        $o->ignoreCommandSucceeded = true;
+
+        /* Retryable read spec tests don't include extra commands, e.g. the
+         * killCursors command issued when a change stream is garbage collected.
+         * We ignore any extra events for that reason. \*/
+        $o->ignoreExtraEvents = true;
+
+        return $o;
+    }
+
     public static function fromTransactions(array $expectedEvents)
     {
         $o = new self($expectedEvents);

tests/SpecTests/Context.php

@@ -21,6 +21,7 @@ use function mt_rand;
  */
 final class Context
 {
+    public $bucketName;
     public $client;
     public $collectionName;
     public $databaseName;
@@ -81,6 +82,19 @@ final class Context
         return $o;
     }
 
+    public static function fromRetryableReads(stdClass $test, $databaseName, $collectionName, $bucketName)
+    {
+        $o = new self($databaseName, $collectionName);
+
+        $o->bucketName = $bucketName;
+
+        $clientOptions = isset($test->clientOptions) ? (array) $test->clientOptions : [];
+
+        $o->client = new Client(FunctionalTestCase::getUri(), $clientOptions);
+
+        return $o;
+    }
+
     public static function fromRetryableWrites(stdClass $test, $databaseName, $collectionName)
     {
         $o = new self($databaseName, $collectionName);
@@ -133,6 +147,14 @@ final class Context
         return $o;
     }
 
+    /**
+     * @return Client
+     */
+    public function getClient()
+    {
+        return $this->client;
+    }
+
     public function getCollection(array $collectionOptions = [])
     {
         return $this->selectCollection(
@@ -147,6 +169,11 @@ final class Context
         return $this->selectDatabase($this->databaseName, $databaseOptions);
     }
 
+    public function getGridFSBucket(array $bucketOptions = [])
+    {
+        return $this->selectGridFSBucket($this->databaseName, $this->bucketName, $bucketOptions);
+    }
+
     /**
      * Prepare options readConcern, readPreference, and writeConcern options by
      * creating value objects.
@@ -272,6 +299,20 @@ final class Context
         );
     }
 
+    public function selectGridFSBucket($databaseName, $bucketName, array $bucketOptions = [])
+    {
+        return $this->selectDatabase($databaseName)->selectGridFSBucket($this->prepareGridFSBucketOptions($bucketOptions, $bucketName));
+    }
+
+    private function prepareGridFSBucketOptions(array $options, $bucketPrefix)
+    {
+        if ($bucketPrefix !== null) {
+            $options['bucketPrefix'] = $bucketPrefix;
+        }
+
+        return $options;
+    }
+
     private function prepareSessionOptions(array $options)
     {
         if (isset($options['defaultTransactionOptions'])) {

tests/SpecTests/ErrorExpectation.php

@@ -60,6 +60,17 @@ final class ErrorExpectation
         return $o;
     }
 
+    public static function fromRetryableReads(stdClass $operation)
+    {
+        $o = new self();
+
+        if (isset($operation->error)) {
+            $o->isExpected = $operation->error;
+        }
+
+        return $o;
+    }
+
     public static function fromRetryableWrites(stdClass $outcome)
     {
         $o = new self();

tests/SpecTests/FunctionalTestCase.php

@@ -185,31 +185,44 @@ class FunctionalTestCase extends BaseFunctionalTestCase
     {
         $context = $this->getContext();
 
-        $collection = $context->getCollection();
-        $collection->drop($context->defaultWriteOptions);
+        if ($context->bucketName !== null) {
+            $bucket = $context->getGridFSBucket($context->defaultWriteOptions);
+            $bucket->drop();
+        }
 
-        $outcomeCollection = $this->getOutcomeCollection();
+        $collection = null;
+        if ($context->collectionName !== null) {
+            $collection = $context->getCollection();
+            $collection->drop($context->defaultWriteOptions);
+        }
 
-        // Avoid redundant drop if the test and outcome collections are the same
-        if ($outcomeCollection->getNamespace() !== $collection->getNamespace()) {
-            $outcomeCollection->drop($context->defaultWriteOptions);
+        if ($context->outcomeCollectionName !== null) {
+            $outcomeCollection = $this->getOutcomeCollection();
+
+            // Avoid redundant drop if the test and outcome collections are the same
+            if ($collection === null || $outcomeCollection->getNamespace() !== $collection->getNamespace()) {
+                $outcomeCollection->drop($context->defaultWriteOptions);
+            }
         }
     }
 
     /**
      * Insert data fixtures into the test collection.
      *
-     * @param array $documents
+     * @param array       $documents
+     * @param string|null $collectionName
      */
-    protected function insertDataFixtures(array $documents)
+    protected function insertDataFixtures(array $documents, $collectionName = null)
     {
         if (empty($documents)) {
             return;
         }
 
         $context = $this->getContext();
-        $collection = $context->getCollection();
+        $collection = $collectionName ? $context->selectCollection($context->databaseName, $collectionName) : $context->getCollection();
         $collection->insertMany($documents, $context->defaultWriteOptions);
+
+        return;
     }
 
     private function getOutcomeCollection()

tests/SpecTests/ResultExpectation.php

@@ -89,6 +89,19 @@ final class ResultExpectation
         return new self($assertionType, $expectedValue);
     }
 
+    public static function fromRetryableReads(stdClass $operation, $defaultAssertionType)
+    {
+        if (property_exists($operation, 'result') && ! self::isErrorResult($operation->result)) {
+            $assertionType = $operation->result === null ? self::ASSERT_NULL : $defaultAssertionType;
+            $expectedValue = $operation->result;
+        } else {
+            $assertionType = self::ASSERT_NOTHING;
+            $expectedValue = null;
+        }
+
+        return new self($assertionType, $expectedValue);
+    }
+
     public static function fromRetryableWrites(stdClass $outcome, $defaultAssertionType)
     {
         if (property_exists($outcome, 'result')) {

tests/SpecTests/RetryableReadsSpecTest.php

+<?php
+
+namespace MongoDB\Tests\SpecTests;
+
+use stdClass;
+use function basename;
+use function file_get_contents;
+use function glob;
+use function is_object;
+use function strpos;
+
+/**
+ * Retryable reads spec tests.
+ *
+ * @see https://github.com/mongodb/specifications/tree/master/source/retryable-reads
+ */
+class RetryableReadsSpecTest extends FunctionalTestCase
+{
+    private static $skippedOperations = [
+        'listCollectionNames' => 'Not implemented',
+        'listCollectionObjects' => 'Not implemented',
+        'listDatabaseNames' => 'Not implemented',
+        'listDatabaseObjects' => 'Not implemented',
+        'listIndexNames' => 'Not implemented',
+    ];
+
+    /**
+     * 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)
+    {
+        static::assertDocumentsMatch($expected, $actual);
+    }
+
+    /**
+     * Execute an individual test case from the specification.
+     *
+     * @dataProvider provideTests
+     * @param stdClass     $test           Individual "tests[]" document
+     * @param array        $runOn          Top-level "runOn" array with server requirements
+     * @param array|object $data           Top-level "data" array to initialize collection
+     * @param string       $databaseName   Name of database under test
+     * @param string|null  $collectionName Name of collection under test
+     * @param string|null  $bucketName     Name of GridFS bucket under test
+     */
+    public function testRetryableReads(stdClass $test, array $runOn = null, $data, $databaseName, $collectionName, $bucketName)
+    {
+        if (isset($runOn)) {
+            $this->checkServerRequirements($runOn);
+        }
+
+        foreach (self::$skippedOperations as $operation => $skipReason) {
+            if (strpos($this->dataDescription(), $operation) === 0) {
+                $this->markTestSkipped($skipReason);
+            }
+        }
+
+        if (strpos($this->dataDescription(), 'changeStreams-') === 0) {
+            $this->skipIfChangeStreamIsNotSupported();
+        }
+
+        $context = Context::fromRetryableReads($test, $databaseName, $collectionName, $bucketName);
+        $this->setContext($context);
+
+        $this->dropTestAndOutcomeCollections();
+
+        if (is_object($data)) {
+            foreach ($data as $collectionName => $documents) {
+                $this->assertIsArray($documents);
+                $this->insertDataFixtures($documents, $collectionName);
+            }
+        } else {
+            $this->insertDataFixtures($data);
+        }
+
+        if (isset($test->failPoint)) {
+            $this->configureFailPoint($test->failPoint);
+        }
+
+        if (isset($test->expectations)) {
+            $commandExpectations = CommandExpectations::fromRetryableReads($test->expectations);
+            $commandExpectations->startMonitoring();
+        }
+
+        foreach ($test->operations as $operation) {
+            Operation::fromRetryableReads($operation)->assert($this, $context);
+        }
+
+        if (isset($commandExpectations)) {
+            $commandExpectations->stopMonitoring();
+            $commandExpectations->assert($this, $context);
+        }
+
+        if (isset($test->outcome->collection->data)) {
+            $this->assertOutcomeCollectionData($test->outcome->collection->data);
+        }
+    }
+
+    public function provideTests()
+    {
+        $testArgs = [];
+
+        foreach (glob(__DIR__ . '/retryable-reads/*.json') as $filename) {
+            $json = $this->decodeJson(file_get_contents($filename));
+            $group = basename($filename, '.json');
+            $runOn = isset($json->runOn) ? $json->runOn : null;
+            $data = isset($json->data) ? $json->data : [];
+            $databaseName = isset($json->database_name) ? $json->database_name : null;
+            $collectionName = isset($json->collection_name) ? $json->collection_name : null;
+            $bucketName = isset($json->bucket_name) ? $json->bucket_name : null;
+
+            foreach ($json->tests as $test) {
+                $name = $group . ': ' . $test->description;
+                $testArgs[$name] = [$test, $runOn, $data, $databaseName, $collectionName, $bucketName];
+            }
+        }
+
+        return $testArgs;
+    }
+}

tests/SpecTests/retryable-reads/aggregate-merge.json

+{
+  "runOn": [
+    {
+      "minServerVersion": "4.1.11"
+    }
+  ],
+  "database_name": "retryable-reads-tests",
+  "collection_name": "coll",
+  "data": [
+    {
+      "_id": 1,
+      "x": 11
+    },
+    {
+      "_id": 2,
+      "x": 22
+    },
+    {
+      "_id": 3,
+      "x": 33
+    }
+  ],
+  "tests": [
+    {
+      "description": "Aggregate with $merge does not retry",
+      "failPoint": {
+        "configureFailPoint": "failCommand",
+        "mode": {
+          "times": 1
+        },
+        "data": {
+          "failCommands": [
+            "aggregate"
+          ],
+          "closeConnection": true
+        }
+      },
+      "operations": [
+        {
+          "object": "collection",
+          "name": "aggregate",
+          "arguments": {
+            "pipeline": [
+              {
+                "$match": {
+                  "_id": {
+                    "$gt": 1
+                  }
+                }
+              },
+              {
+                "$sort": {
+                  "x": 1
+                }
+              },
+              {
+                "$merge": {
+                  "into": "output-collection"
+                }
+              }
+            ]
+          },
+          "error": true
+        }
+      ],
+      "expectations": [
+        {
+          "command_started_event": {
+            "command": {
+              "aggregate": "coll",
+              "pipeline": [
+                {
+                  "$match": {
+                    "_id": {
+                      "$gt": 1
+                    }
+                  }
+                },
+                {
+                  "$sort": {
+                    "x": 1
+                  }
+                },
+                {
+                  "$merge": {
+                    "into": "output-collection"
+                  }
+                }
+              ]
+            },
+            "command_name": "aggregate",
+            "database_name": "retryable-reads-tests"
+          }
+        }
+      ]
+    }
+  ]
+}

tests/SpecTests/retryable-reads/aggregate.json

+{
+  "runOn": [
+    {
+      "minServerVersion": "4.0",
+      "topology": [
+        "single",
+        "replicaset"
+      ]
+    },
+    {
+      "minServerVersion": "4.1.7",
+      "topology": [
+        "sharded"
+      ]
+    }
+  ],
+  "database_name": "retryable-reads-tests",
+  "collection_name": "coll",
+  "data": [
+    {
+      "_id": 1,
+      "x": 11
+    },
+    {
+      "_id": 2,
+      "x": 22
+    },
+    {
+      "_id": 3,
+      "x": 33
+    }
+  ],
+  "tests": [
+    {
+      "description": "Aggregate succeeds on first attempt",
+      "operations": [
+        {
+          "name": "aggregate",
+          "object": "collection",
+          "arguments": {
+            "pipeline": [
+              {
+                "$match": {
+                  "_id": {
+                    "$gt": 1
+                  }
+                }
+              },
+              {
+                "$sort": {
+                  "x": 1
+                }
+              }
+            ]
+          },
+          "result": [
+            {
+              "_id": 2,
+              "x": 22
+            },
+            {
+              "_id": 3,
+              "x": 33
+            }
+          ]
+        }
+      ],
+      "expectations": [
+        {
+          "command_started_event": {
+            "command": {
+              "aggregate": "coll",
+              "pipeline": [
+                {
+                  "$match": {
+                    "_id": {
+                      "$gt": 1
+                    }
+                  }
+                },
+                {
+                  "$sort": {
+                    "x": 1
+                  }
+                }
+              ]
+            },
+            "database_name": "retryable-reads-tests"
+          }
+        }
+      ]
+    },
+    {
+      "description": "Aggregate succeeds on second attempt",
+      "failPoint": {
+        "configureFailPoint": "failCommand",
+        "mode": {
+          "times": 1
+        },
+        "data": {
+          "failCommands": [
+            "aggregate"
+          ],
+          "closeConnection": true
+        }
+      },
+      "operations": [
+        {
+          "name": "aggregate",
+          "object": "collection",
+          "arguments": {
+            "pipeline": [
+              {
+                "$match": {
+                  "_id": {
+                    "$gt": 1
+                  }
+                }
+              },
+              {
+                "$sort": {
+                  "x": 1
+                }
+              }
+            ]
+          },
+          "result": [
+            {
+              "_id": 2,
+              "x": 22
+            },
+            {
+              "_id": 3,
+              "x": 33
+            }
+          ]
+        }
+      ],
+      "expectations": [
+        {
+          "command_started_event": {
+            "command": {
+              "aggregate": "coll",
+              "pipeline": [
+                {
+                  "$match": {
+                    "_id": {
+                      "$gt": 1
+                    }
+                  }
+                },
+                {
+                  "$sort": {
+                    "x": 1
+                  }
+                }
+              ]
+            },
+            "database_name": "retryable-reads-tests"
+          }
+        },
+        {
+          "command_started_event": {
+            "command": {
+              "aggregate": "coll",
+              "pipeline": [
+                {
+                  "$match": {
+                    "_id": {
+                      "$gt": 1
+                    }
+                  }
+                },
+                {
+                  "$sort": {
+                    "x": 1
+                  }
+                }
+              ]
+            },
+            "database_name": "retryable-reads-tests"
+          }
+        }
+      ]
+    },
+    {
+      "description": "Aggregate fails on first attempt",
+      "clientOptions": {
+        "retryReads": false
+      },
+      "failPoint": {
+        "configureFailPoint": "failCommand",
+        "mode": {
+          "times": 1
+        },
+        "data": {
+          "failCommands": [
+            "aggregate"
+          ],
+          "closeConnection": true
+        }
+      },
+      "operations": [
+        {
+          "name": "aggregate",
+          "object": "collection",
+          "arguments": {
+            "pipeline": [
+              {
+                "$match": {
+                  "_id": {
+                    "$gt": 1
+                  }
+                }
+              },
+              {
+                "$sort": {
+                  "x": 1
+                }
+              }
+            ]
+          },
+          "error": true
+        }
+      ],
+      "expectations": [
+        {
+          "command_started_event": {
+            "command": {
+              "aggregate": "coll",
+              "pipeline": [
+                {
+                  "$match": {
+                    "_id": {
+                      "$gt": 1
+                    }
+                  }
+                },
+                {
+                  "$sort": {
+                    "x": 1
+                  }
+                }
+              ]
+            },
+            "database_name": "retryable-reads-tests"
+          }
+        }
+      ]
+    },
+    {
+      "description": "Aggregate fails on second attempt",
+      "failPoint": {
+        "configureFailPoint": "failCommand",
+        "mode": {
+          "times": 2
+        },
+        "data": {
+          "failCommands": [
+            "aggregate"
+          ],
+          "closeConnection": true
+        }
+      },
+      "operations": [
+        {
+          "name": "aggregate",
+          "object": "collection",
+          "arguments": {
+            "pipeline": [
+              {
+                "$match": {
+                  "_id": {
+                    "$gt": 1
+                  }
+                }
+              },
+              {
+                "$sort": {
+                  "x": 1
+                }
+              }
+            ]
+          },
+          "error": true
+        }
+      ],
+      "expectations": [
+        {
+          "command_started_event": {
+            "command": {
+              "aggregate": "coll",
+              "pipeline": [
+                {
+                  "$match": {
+                    "_id": {
+                      "$gt": 1
+                    }
+                  }
+                },
+                {
+                  "$sort": {
+                    "x": 1
+                  }
+                }
+              ]
+            },
+            "database_name": "retryable-reads-tests"
+          }
+        },
+        {
+          "command_started_event": {
+            "command": {
+              "aggregate": "coll",
+              "pipeline": [
+                {
+                  "$match": {
+                    "_id": {
+                      "$gt": 1
+                    }
+                  }
+                },
+                {
+                  "$sort": {
+                    "x": 1
+                  }
+                }
+              ]
+            },
+            "database_name": "retryable-reads-tests"
+          }
+        }
+      ]
+    },
+    {
+      "description": "Aggregate with $out does not retry",
+      "failPoint": {
+        "configureFailPoint": "failCommand",
+        "mode": {
+          "times": 1
+        },
+        "data": {
+          "failCommands": [
+            "aggregate"
+          ],
+          "closeConnection": true
+        }
+      },
+      "operations": [
+        {
+          "name": "aggregate",
+          "object": "collection",
+          "arguments": {
+            "pipeline": [
+              {
+                "$match": {
+                  "_id": {
+                    "$gt": 1
+                  }
+                }
+              },
+              {
+                "$sort": {
+                  "x": 1
+                }
+              },
+              {
+                "$out": "output-collection"
+              }
+            ]
+          },
+          "error": true
+        }
+      ],
+      "expectations": [
+        {
+          "command_started_event": {
+            "command": {
+              "aggregate": "coll",
+              "pipeline": [
+                {
+                  "$match": {
+                    "_id": {
+                      "$gt": 1
+                    }
+                  }
+                },
+                {
+                  "$sort": {
+                    "x": 1
+                  }
+                },
+                {
+                  "$out": "output-collection"
+                }
+              ]
+            },
+            "command_name": "aggregate",
+            "database_name": "retryable-reads-tests"
+          }
+        }
+      ]
+    }
+  ]
+}

tests/SpecTests/retryable-reads/changeStreams-client.watch.json

+{
+  "runOn": [
+    {
+      "minServerVersion": "4.0",
+      "topology": [
+        "replicaset"
+      ]
+    },
+    {
+      "minServerVersion": "4.1.7",
+      "topology": [
+        "sharded"
+      ]
+    }
+  ],
+  "database_name": "retryable-reads-tests",
+  "collection_name": "coll",
+  "data": [
+    {
+      "_id": 1,
+      "x": 11
+    }
+  ],
+  "tests": [
+    {
+      "description": "client.watch succeeds on first attempt",
+      "operations": [
+        {
+          "name": "watch",
+          "object": "client"
+        }
+      ],
+      "expectations": [
+        {
+          "command_started_event": {
+            "command": {
+              "aggregate": 1,
+              "cursor": {},
+              "pipeline": [
+                {
+                  "$changeStream": {
+                    "allChangesForCluster": true
+                  }
+                }
+              ]
+            },
+            "database_name": "admin"
+          }
+        }
+      ]
+    },
+    {
+      "description": "client.watch succeeds on second attempt",
+      "failPoint": {
+        "configureFailPoint": "failCommand",
+        "mode": {
+          "times": 1
+        },
+        "data": {
+          "failCommands": [
+            "aggregate"
+          ],
+          "closeConnection": true
+        }
+      },
+      "operations": [
+        {
+          "name": "watch",
+          "object": "client"
+        }
+      ],
+      "expectations": [
+        {
+          "command_started_event": {
+            "command": {
+              "aggregate": 1,
+              "cursor": {},
+              "pipeline": [
+                {
+                  "$changeStream": {
+                    "allChangesForCluster": true
+                  }
+                }
+              ]
+            },
+            "database_name": "admin"
+          }
+        },
+        {
+          "command_started_event": {
+            "command": {
+              "aggregate": 1,
+              "cursor": {},
+              "pipeline": [
+                {
+                  "$changeStream": {
+                    "allChangesForCluster": true
+                  }
+                }
+              ]
+            },
+            "database_name": "admin"
+          }
+        }
+      ]
+    },
+    {
+      "description": "client.watch fails on first attempt",
+      "clientOptions": {
+        "retryReads": false
+      },
+      "failPoint": {
+        "configureFailPoint": "failCommand",
+        "mode": {
+          "times": 1
+        },
+        "data": {
+          "failCommands": [
+            "aggregate"
+          ],
+          "closeConnection": true
+        }
+      },
+      "operations": [
+        {
+          "name": "watch",
+          "object": "client",
+          "error": true
+        }
+      ],
+      "expectations": [
+        {
+          "command_started_event": {
+            "command": {
+              "aggregate": 1,
+              "cursor": {},
+              "pipeline": [
+                {
+                  "$changeStream": {
+                    "allChangesForCluster": true
+                  }
+                }
+              ]
+            },
+            "database_name": "admin"
+          }
+        }
+      ]
+    },
+    {
+      "description": "client.watch fails on second attempt",
+      "failPoint": {
+        "configureFailPoint": "failCommand",
+        "mode": {
+          "times": 2
+        },
+        "data": {
+          "failCommands": [
+            "aggregate"
+          ],
+          "closeConnection": true
+        }
+      },
+      "operations": [
+        {
+          "name": "watch",
+          "object": "client",
+          "error": true
+        }
+      ],
+      "expectations": [
+        {
+          "command_started_event": {
+            "command": {
+              "aggregate": 1,
+              "cursor": {},
+              "pipeline": [
+                {
+                  "$changeStream": {
+                    "allChangesForCluster": true
+                  }
+                }
+              ]
+            },
+            "database_name": "admin"
+          }
+        },
+        {
+          "command_started_event": {
+            "command": {
+              "aggregate": 1,
+              "cursor": {},
+              "pipeline": [
+                {
+                  "$changeStream": {
+                    "allChangesForCluster": true
+                  }
+                }
+              ]
+            },
+            "database_name": "admin"
+          }
+        }
+      ]
+    }
+  ]
+}

tests/SpecTests/retryable-reads/changeStreams-db.coll.watch.json

+{
+  "runOn": [
+    {
+      "minServerVersion": "4.0",
+      "topology": [
+        "replicaset"
+      ]
+    },
+    {
+      "minServerVersion": "4.1.7",
+      "topology": [
+        "sharded"
+      ]
+    }
+  ],
+  "database_name": "retryable-reads-tests",
+  "collection_name": "coll",
+  "data": [
+    {
+      "_id": 1,
+      "x": 11
+    }
+  ],
+  "tests": [
+    {
+      "description": "db.coll.watch succeeds on first attempt",
+      "operations": [
+        {
+          "name": "watch",
+          "object": "collection"
+        }
+      ],
+      "expectations": [
+        {
+          "command_started_event": {
+            "command": {
+              "aggregate": "coll",
+              "cursor": {},
+              "pipeline": [
+                {
+                  "$changeStream": {}
+                }
+              ]
+            },
+            "database_name": "retryable-reads-tests"
+          }
+        }
+      ]
+    },
+    {
+      "description": "db.coll.watch succeeds on second attempt",
+      "failPoint": {
+        "configureFailPoint": "failCommand",
+        "mode": {
+          "times": 1
+        },
+        "data": {
+          "failCommands": [
+            "aggregate"
+          ],
+          "closeConnection": true
+        }
+      },
+      "operations": [
+        {
+          "name": "watch",
+          "object": "collection"
+        }
+      ],
+      "expectations": [
+        {
+          "command_started_event": {
+            "command": {
+              "aggregate": "coll",
+              "cursor": {},
+              "pipeline": [
+                {
+                  "$changeStream": {}
+                }
+              ]
+            },
+            "database_name": "retryable-reads-tests"
+          }
+        },
+        {
+          "command_started_event": {
+            "command": {
+              "aggregate": "coll",
+              "cursor": {},
+              "pipeline": [
+                {
+                  "$changeStream": {}
+                }
+              ]
+            },
+            "database_name": "retryable-reads-tests"
+          }
+        }
+      ]
+    },
+    {
+      "description": "db.coll.watch fails on first attempt",
+      "clientOptions": {
+        "retryReads": false
+      },
+      "failPoint": {
+        "configureFailPoint": "failCommand",
+        "mode": {
+          "times": 1
+        },
+        "data": {
+          "failCommands": [
+            "aggregate"
+          ],
+          "closeConnection": true
+        }
+      },
+      "operations": [
+        {
+          "name": "watch",
+          "object": "collection",
+          "error": true
+        }
+      ],
+      "expectations": [
+        {
+          "command_started_event": {
+            "command": {
+              "aggregate": "coll",
+              "cursor": {},
+              "pipeline": [
+                {
+                  "$changeStream": {}
+                }
+              ]
+            },
+            "database_name": "retryable-reads-tests"
+          }
+        }
+      ]
+    },
+    {
+      "description": "db.coll.watch fails on second attempt",
+      "failPoint": {
+        "configureFailPoint": "failCommand",
+        "mode": {
+          "times": 2
+        },
+        "data": {
+          "failCommands": [
+            "aggregate"
+          ],
+          "closeConnection": true
+        }
+      },
+      "operations": [
+        {
+          "name": "watch",
+          "object": "collection",
+          "error": true
+        }
+      ],
+      "expectations": [
+        {
+          "command_started_event": {
+            "command": {
+              "aggregate": "coll",
+              "cursor": {},
+              "pipeline": [
+                {
+                  "$changeStream": {}
+                }
+              ]
+            },
+            "database_name": "retryable-reads-tests"
+          }
+        },
+        {
+          "command_started_event": {
+            "command": {
+              "aggregate": "coll",
+              "cursor": {},
+              "pipeline": [
+                {
+                  "$changeStream": {}
+                }
+              ]
+            },
+            "database_name": "retryable-reads-tests"
+          }
+        }
+      ]
+    }
+  ]
+}

tests/SpecTests/retryable-reads/changeStreams-db.watch.json

+{
+  "runOn": [
+    {
+      "minServerVersion": "4.0",
+      "topology": [
+        "replicaset"
+      ]
+    },
+    {
+      "minServerVersion": "4.1.7",
+      "topology": [
+        "sharded"
+      ]
+    }
+  ],
+  "database_name": "retryable-reads-tests",
+  "collection_name": "coll",
+  "data": [
+    {
+      "_id": 1,
+      "x": 11
+    }
+  ],
+  "tests": [
+    {
+      "description": "db.watch succeeds on first attempt",
+      "operations": [
+        {
+          "name": "watch",
+          "object": "database"
+        }
+      ],
+      "expectations": [
+        {
+          "command_started_event": {
+            "command": {
+              "aggregate": 1,
+              "cursor": {},
+              "pipeline": [
+                {
+                  "$changeStream": {}
+                }
+              ]
+            },
+            "database_name": "retryable-reads-tests"
+          }
+        }
+      ]
+    },
+    {
+      "description": "db.watch succeeds on second attempt",
+      "failPoint": {
+        "configureFailPoint": "failCommand",
+        "mode": {
+          "times": 1
+        },
+        "data": {
+          "failCommands": [
+            "aggregate"
+          ],
+          "closeConnection": true
+        }
+      },
+      "operations": [
+        {
+          "name": "watch",
+          "object": "database"
+        }
+      ],
+      "expectations": [
+        {
+          "command_started_event": {
+            "command": {
+              "aggregate": 1,
+              "cursor": {},
+              "pipeline": [
+                {
+                  "$changeStream": {}
+                }
+              ]
+            },
+            "database_name": "retryable-reads-tests"
+          }
+        },
+        {
+          "command_started_event": {
+            "command": {
+              "aggregate": 1,
+              "cursor": {},
+              "pipeline": [
+                {
+                  "$changeStream": {}
+                }
+              ]
+            },
+            "database_name": "retryable-reads-tests"
+          }
+        }
+      ]
+    },
+    {
+      "description": "db.watch fails on first attempt",
+      "clientOptions": {
+        "retryReads": false
+      },
+      "failPoint": {
+        "configureFailPoint": "failCommand",
+        "mode": {
+          "times": 1
+        },
+        "data": {
+          "failCommands": [
+            "aggregate"
+          ],
+          "closeConnection": true
+        }
+      },
+      "operations": [
+        {
+          "name": "watch",
+          "object": "database",
+          "error": true
+        }
+      ],
+      "expectations": [
+        {
+          "command_started_event": {
+            "command": {
+              "aggregate": 1,
+              "cursor": {},
+              "pipeline": [
+                {
+                  "$changeStream": {}
+                }
+              ]
+            },
+            "database_name": "retryable-reads-tests"
+          }
+        }
+      ]
+    },
+    {
+      "description": "db.watch fails on second attempt",
+      "failPoint": {
+        "configureFailPoint": "failCommand",
+        "mode": {
+          "times": 2
+        },
+        "data": {
+          "failCommands": [
+            "aggregate"
+          ],
+          "closeConnection": true
+        }
+      },
+      "operations": [
+        {
+          "name": "watch",
+          "object": "database",
+          "error": true
+        }
+      ],
+      "expectations": [
+        {
+          "command_started_event": {
+            "command": {
+              "aggregate": 1,
+              "cursor": {},
+              "pipeline": [
+                {
+                  "$changeStream": {}
+                }
+              ]
+            },
+            "database_name": "retryable-reads-tests"
+          }
+        },
+        {
+          "command_started_event": {
+            "command": {
+              "aggregate": 1,
+              "cursor": {},
+              "pipeline": [
+                {
+                  "$changeStream": {}
+                }
+              ]
+            },
+            "database_name": "retryable-reads-tests"
+          }
+        }
+      ]
+    }
+  ]
+}