PHPLIB-441: Increment ChangeStream key consistently when resuming

Author: jmikola

src/ChangeStream.php

@@ -50,6 +50,11 @@ class ChangeStream implements Iterator
     private $resumeCallable;
     private $csIt;
     private $key = 0;
+
+    /**
+     * Whether the change stream has advanced to its first result. This is used
+     * to determine whether $key should be incremented after an iteration event.
+     */
     private $hasAdvanced = false;
 
     /**
@@ -102,7 +107,7 @@ public function next()
     {
         try {
             $this->csIt->next();
-            $this->onIteration(true);
+            $this->onIteration($this->hasAdvanced);
         } catch (RuntimeException $e) {
             if ($this->isResumableError($e)) {
                 $this->resume();
@@ -118,6 +123,9 @@ public function rewind()
     {
         try {
             $this->csIt->rewind();
+            /* Unlike next() and resume(), the decision to increment the key
+             * does not depend on whether the change stream has advanced. This
+             * ensures that multiple calls to rewind() do not alter state. */
             $this->onIteration(false);
         } catch (RuntimeException $e) {
             if ($this->isResumableError($e)) {
@@ -193,12 +201,12 @@ private function isResumableError(RuntimeException $exception)
     }
 
     /**
-     * Perform housekeeping after an iteration event (i.e. next or rewind).
+     * Perform housekeeping after an iteration event.
      *
-     * @param boolean $isNext Whether the iteration event was a call to next()
+     * @param boolean $incrementKey Increment $key if there is a current result
      * @throws ResumeTokenException
      */
-    private function onIteration($isNext)
+    private function onIteration($incrementKey)
     {
         /* If the cursorId is 0, the server has invalidated the cursor and we
          * will never perform another getMore nor need to resume since any
@@ -210,13 +218,13 @@ private function onIteration($isNext)
             $this->resumeCallable = null;
         }
 
+        /* Return early if there is not a current result. Avoid any attempt to
+         * increment the iterator's key or extract a resume token */
         if (!$this->valid()) {
             return;
         }
 
-        /* Increment the key if the iteration event was a call to next() and we
-         * have already advanced past the first result. */
-        if ($isNext && $this->hasAdvanced) {
+        if ($incrementKey) {
             $this->key++;
         }
 
@@ -234,6 +242,14 @@ private function resume()
         $newChangeStream = call_user_func($this->resumeCallable, $this->resumeToken);
         $this->csIt = $newChangeStream->csIt;
         $this->csIt->rewind();
-        $this->onIteration(false);
+        /* Note: if we are resuming after a call to ChangeStream::rewind(),
+         * $hasAdvanced will always be false. For it to be true, rewind() would
+         * need to have thrown a RuntimeException with a resumable error, which
+         * can only happen during the first call to IteratorIterator::rewind()
+         * before onIteration() has a chance to set $hasAdvanced to true.
+         * Otherwise, IteratorIterator::rewind() would either NOP (consecutive
+         * rewinds) or throw a LogicException (rewind after next), neither of
+         * which would result in a call to resume(). */
+        $this->onIteration($this->hasAdvanced);
     }
 }

tests/Operation/WatchFunctionalTest.php

@@ -379,20 +379,67 @@ public function testNoChangeAfterResumeBeforeInsert()
         $this->assertMatchesDocument($expectedResult, $changeStream->current());
     }
 
-    public function testResumeTokenIsUpdatedAfterResuming()
+    public function testResumeMultipleTimesInSuccession()
     {
-        $this->insertDocument(['_id' => 1]);
+        $operation = new CreateCollection($this->getDatabaseName(), $this->getCollectionName());
+        $operation->execute($this->getPrimaryServer());
 
         $operation = new Watch($this->manager, $this->getDatabaseName(), $this->getCollectionName(), [], $this->defaultOptions);
         $changeStream = $operation->execute($this->getPrimaryServer());
 
+        /* Killing the cursor when there are no results will test that neither
+         * the initial rewind() nor its resume attempt incremented the key. */
+        $this->killChangeStreamCursor($changeStream);
+
         $changeStream->rewind();
+        $this->assertFalse($changeStream->valid());
+        $this->assertNull($changeStream->key());
         $this->assertNull($changeStream->current());
 
+        $this->insertDocument(['_id' => 1]);
+
+        /* Killing the cursor a second time when there is a result will test
+         * that the resume attempt picks up the latest change. */
+        $this->killChangeStreamCursor($changeStream);
+
+        $changeStream->rewind();
+        $this->assertTrue($changeStream->valid());
+        $this->assertSame(0, $changeStream->key());
+
+        $expectedResult = [
+            '_id' => $changeStream->current()->_id,
+            'operationType' => 'insert',
+            'fullDocument' => ['_id' => 1],
+            'ns' => ['db' => $this->getDatabaseName(), 'coll' => $this->getCollectionName()],
+            'documentKey' => ['_id' => 1],
+        ];
+
+        $this->assertMatchesDocument($expectedResult, $changeStream->current());
+
+        /* Killing the cursor a second time will not trigger a resume until
+         * ChangeStream::next() is called. A successive call to rewind() should
+         * not change the iterator's state and preserve the current result. */
+        $this->killChangeStreamCursor($changeStream);
+
+        $changeStream->rewind();
+        $this->assertTrue($changeStream->valid());
+        $this->assertSame(0, $changeStream->key());
+
+        $expectedResult = [
+            '_id' => $changeStream->current()->_id,
+            'operationType' => 'insert',
+            'fullDocument' => ['_id' => 1],
+            'ns' => ['db' => $this->getDatabaseName(), 'coll' => $this->getCollectionName()],
+            'documentKey' => ['_id' => 1],
+        ];
+
+        $this->assertMatchesDocument($expectedResult, $changeStream->current());
+
         $this->insertDocument(['_id' => 2]);
 
         $changeStream->next();
         $this->assertTrue($changeStream->valid());
+        $this->assertSame(1, $changeStream->key());
 
         $expectedResult = [
             '_id' => $changeStream->current()->_id,
@@ -410,6 +457,7 @@ public function testResumeTokenIsUpdatedAfterResuming()
 
         $changeStream->next();
         $this->assertTrue($changeStream->valid());
+        $this->assertSame(2, $changeStream->key());
 
         $expectedResult = [
             '_id' => $changeStream->current()->_id,
@@ -431,6 +479,7 @@ public function testResumeTokenIsUpdatedAfterResuming()
 
         $changeStream->next();
         $this->assertTrue($changeStream->valid());
+        $this->assertSame(3, $changeStream->key());
 
         $expectedResult = [
             '_id' => $changeStream->current()->_id,