PHPLIB-351: Cluster and DB-level change streams and startAtOperationTime #544

jmikola · 2018-06-22T21:34:58Z

https://jira.mongodb.org/browse/PHPLIB-351
https://jira.mongodb.org/browse/PHPLIB-358

jmikola · 2018-06-26T02:30:51Z

src/ChangeStream.php

+     * @param RuntimeException $exception
+     * @return boolean
+     */
+    private function isResumableError(RuntimeException $exception)


The criteria for a resumable error changed since the spec was first implemented in 1.3.0. See: Resumable Error.

jmikola · 2018-06-26T02:32:16Z

src/ChangeStream.php

+        }
+
+        if (in_array($exception->getCode(), [self::$errorCodeCappedPositionLost, self::$errorCodeCursorKilled, self::$errorCodeInterrupted])) {
+            return false;


My read of the new resumable error criteria lead me to conclude that checking the error message string for "not master" or "node is recovering" was redundant. This logic takes the into account and I've opened mongodb/specifications#365 to correct the spec.

jmikola · 2018-06-26T02:37:51Z

src/Operation/Watch.php

@@ -79,22 +92,44 @@ class Watch implements Executable
     *  * resumeAfter (document): Specifies the logical starting point for the
     *    new change stream.
     *
+     *    Using this option in conjunction with "startAtOperationTime" will
+     *    result in a server error. The options are mutually exclusive.


Per startatoperationtime in the spec, we're not supposed to raise a client-side error if both options are used.

jmikola · 2018-06-26T02:39:42Z

src/Operation/Watch.php

        }

        $this->databaseName = (string) $databaseName;
-        $this->collectionName = (string) $collectionName;
+        $this->collectionName = isset($collectionName) ? (string) $collectionName : null;


Aggregate takes care of handling the null Collection name (per PHPLIB-358).

jmikola · 2018-06-26T02:41:24Z

src/Operation/Watch.php

+    final public function commandSucceeded(CommandSucceededEvent $event)
+    {
+        if ($event->getCommandName() !== 'aggregate') {
+            return;


I considered throwing a LogicException here, as this class should never be registered as a CommandSubscriber except for the brief duration that we execute an aggregate command; however, I don't think that's necessary. Open to changing this if you'd prefer the extra sanity check, though.

jmikola · 2018-06-26T02:44:02Z

src/Operation/Watch.php

+
+        $reply = $event->getReply();
+
+        if (isset($reply->operationTime) && $reply->operationTime instanceof TimestampInterface) {


By this point, we've already verified that the server's wire version is >= 7. If operationTime isn't a Timestamp, something is quite wrong. I just wanted to be defensive.

Failing silently also makes sense as this resume behavior is internal behavior (akin to us creating an "implicit" session in the constructor and not bothering the user if that fails).

jmikola · 2018-06-26T02:45:56Z

tests/Operation/AggregateFunctionalTest.php

+                $operation = new Aggregate(
+                    'admin',
+                    null,
+                    [['$currentOp' => (object) []]]


I'm not specifying any options nor am I asserting the response as this is only testing that executing such a command with a null collection name results in { aggregate: 1 } in the command document.

jmikola · 2018-06-26T02:50:15Z

tests/Operation/WatchFunctionalTest.php

@@ -130,6 +131,100 @@ function(stdClass $command) use (&$commands) {
        $this->assertSame($expectedCommands, $commands);
    }

+    public function testResumeBeforeReceivingAnyResultsIncludesStartAtOperationTime()
+    {
+        $operation = new Watch($this->manager, $this->getDatabaseName(), $this->getCollectionName(), [], $this->defaultOptions);


I originally inserted a document before starting the change stream, but that lead to a strange behavior. The initial aggregate command would not find the document (of course), but it's response would include an operationTime time based on the document's insertion since it was the most recent entry in the oplog.

When we kill and resume the cursor in the subsequent test activity, that operation time would be used for startAtOperationTime in the resuming aggregate and the document would show up.

I was under the impression that using the operation time from aggregate was intended to prevent extra documents from showing up from before the launch point of the original change stream, but perhaps this is a known edge case.

Perhaps @daprahamian, @kevinAlbs, or @ShaneHarvey can chime in here.

@ShaneHarvey mentioned that he and @behackett talked about this some time ago and concluded that it was expected behavior (if no other write operation on the server occurs between the insert and aggregate nor does enough time elapse for a periodic no-op entry to be created).

They tossed around the idea of having the driver manually increment the operation time returned by aggregate, but nothing came of that. Perhaps they talked to some folks on the server team and they advised against it.

Yeah I just talked with Bernie, incrementing the operation time does not really solve the problem. We can fix the issue in SPEC-1122.

Aye, SPEC-1122 ("Update startAtOperationTime during change stream iteration" for those without access) would help provided we're not resuming because the first getMore has failed; however, I'll admit that this is a fairly contrived edge case more likely to show up in a test suite than real life.

jmikola · 2018-06-26T02:51:14Z

tests/Operation/WatchFunctionalTest.php

@@ -484,16 +579,8 @@ public function testTypeMapOption(array $typeMap, $expectedChangeDocument)

        $changeStream->next();
        $this->assertTrue($changeStream->valid());
-        $changeDocument = $changeStream->current();
-
-        // Unset the resume token and namespace, which are intentionally omitted


Thanks to assertMatchesDocument(), we can ignore extra fields in the actual document that are not present in our expected value. This allows us to do without these manual unsets.

Does that also address the extra "clusterTime" information that made tests fail?

Yes, it does.

jmikola · 2018-06-26T02:51:59Z

tests/TestCase.php

+        $extraKeys = [];
+
+        /* Avoid unsetting fields while we're iterating on the ArrayObject to
+         * work around https://bugs.php.net/bug.php?id=70246 */


I had a great time discovering this bug. 😛

For fun I would say: @sgolemon, fix yo shit ;-)

derickr

I've left some comments/questions, but LGTM.

derickr · 2018-06-26T08:29:31Z

src/Operation/Aggregate.php

@@ -289,7 +292,7 @@ public function execute(Server $server)
    private function createCommand(Server $server)
    {
        $cmd = [
-            'aggregate' => $this->collectionName,
+            'aggregate' => isset($this->collectionName) ? $this->collectionName : 1,


The use of "1" wasn't really documented: mongodb/docs#3357

derickr · 2018-06-26T08:31:51Z

src/Operation/Watch.php

+     * specifying null for both the database and collection name.
+     *
+     * @param string|null    $databaseName   Database name
+     * @param string|null    $collectionName Collection name
     * @param array          $pipeline       List of pipeline operations
     * @param array          $options        Command options
     * @param Manager        $manager        Manager instance from the driver


Why is "Manager" the last option in this list, but the first argument in the method declaration?

Good catch! Looks like an old typo: https://github.com/mongodb/mongo-php-library/blame/master/src/Operation/Watch.php#L93

I'll fix it.

derickr · 2018-06-26T08:40:41Z

tests/Operation/WatchFunctionalTest.php

@@ -484,16 +579,8 @@ public function testTypeMapOption(array $typeMap, $expectedChangeDocument)

        $changeStream->next();
        $this->assertTrue($changeStream->valid());
-        $changeDocument = $changeStream->current();
-
-        // Unset the resume token and namespace, which are intentionally omitted


Does that also address the extra "clusterTime" information that made tests fail?

derickr · 2018-06-26T08:49:18Z

tests/TestCase.php

+        $extraKeys = [];
+
+        /* Avoid unsetting fields while we're iterating on the ArrayObject to
+         * work around https://bugs.php.net/bug.php?id=70246 */


For fun I would say: @sgolemon, fix yo shit ;-)

kvwalker

LGTM other than 1 small comment!

kvwalker · 2018-06-26T14:20:51Z

src/Operation/Watch.php

        }

+        /* In the absence of an explicit session, create one to ensure that the
+         * initial aggregation and any resume attempts can use the same session
+         * ("implicit from the user's perspective per PHPLIB-342). We can ignore


I think there's a " missing after 'perspective'

MongoDB 4.0 adds a "clusterTime" field to change stream documents. Rather than add conditional logic to expect this in tests, implement a new comparison function that fields not found in our expected document.

MongoDB\Driver\Exception\RuntimeException does not conflict in this file, so we can also remove its alias.

Skip redundant error message checks, since their codes do not overlap with the three errors specifically excluded by the spec and all other server errors are resumable. Deprecate the unused class constant, which can be removed in 2.0 (PHPLIB-360).

jmikola force-pushed the phplib-351 branch 2 times, most recently from b1e8d8f to 04ddc86 Compare June 26, 2018 02:26

jmikola requested review from kvwalker and derickr June 26, 2018 02:27

jmikola changed the title ~~[WIP] PHPLIB-351: Support cluster and database-level change streams~~ PHPLIB-351: Cluster and DB-level change streams and startAtOperationTime Jun 26, 2018

jmikola force-pushed the phplib-351 branch from 04ddc86 to 48f956b Compare June 26, 2018 02:33

jmikola commented Jun 26, 2018

View reviewed changes

jmikola force-pushed the phplib-351 branch 2 times, most recently from af295aa to 20d9148 Compare June 26, 2018 02:56

derickr approved these changes Jun 26, 2018

View reviewed changes

jmikola force-pushed the phplib-351 branch from 20d9148 to 3205c3e Compare June 26, 2018 14:18

kvwalker approved these changes Jun 26, 2018

View reviewed changes

jmikola force-pushed the phplib-351 branch 2 times, most recently from 763f85d to 21e1f30 Compare June 26, 2018 18:34

jmikola added 2 commits June 26, 2018 15:49

Ignore extra fields in change stream document comparisons

930fb18

MongoDB 4.0 adds a "clusterTime" field to change stream documents. Rather than add conditional logic to expect this in tests, implement a new comparison function that fields not found in our expected document.

PHPLIB-358: Do not require collection name for Aggregate operation

2076361

jmikola force-pushed the phplib-351 branch 2 times, most recently from 97bd301 to ef31465 Compare June 26, 2018 20:05

PHPLIB-342: Explain why Watch may create its own Session object

e621613

MongoDB\Driver\Exception\RuntimeException does not conflict in this file, so we can also remove its alias.

jmikola force-pushed the phplib-351 branch from ef31465 to 8adc084 Compare June 26, 2018 20:12

jmikola added 4 commits June 26, 2018 16:14

Refactor CommandObserver to track all three APM events

0f848cb

PHPLIB-351: Cluster and DB-level change streams and startAtOperationTime

9ccfae8

PHPLIB-351: Documentation for watch() methods

eb2cd59

jmikola force-pushed the phplib-351 branch from 8adc084 to eb2cd59 Compare June 26, 2018 20:15

jmikola merged commit eb2cd59 into mongodb:master Jun 26, 2018

jmikola added a commit that referenced this pull request Jun 26, 2018

Merge pull request #544

9739904

jmikola deleted the phplib-351 branch June 26, 2018 20:26


		$reply = $event->getReply();

		if (isset($reply->operationTime) && $reply->operationTime instanceof TimestampInterface) {

PHPLIB-351: Cluster and DB-level change streams and startAtOperationTime #544

PHPLIB-351: Cluster and DB-level change streams and startAtOperationTime #544

Uh oh!

Conversation

jmikola commented Jun 22, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

jmikola Jun 26, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

derickr left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kvwalker left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

jmikola commented Jun 22, 2018 •

edited

Loading

jmikola Jun 26, 2018 •

edited

Loading