PERL-792 Retryable Writes

Makefile.PL

@@ -34,6 +34,7 @@ requires 'ExtUtils::ParseXS' => 3.21;
 requires 'IO::File';
 requires 'IO::Socket' => ( $^O eq 'MSWin32' ? '1.31' : '0' );
 requires 'List::Util';
+requires 'Math::BigInt';
 requires 'MIME::Base64';
 requires 'Moo' => '2';
 requires 'Moo::Role';
@@ -68,7 +69,6 @@ test_requires 'File::Spec';
 test_requires 'File::Temp' => '0.17';
 test_requires 'FileHandle';
 test_requires 'JSON::MaybeXS' => '1.002005';
-test_requires 'Math::BigInt';
 test_requires 'Path::Tiny' => '0.054';
 test_requires 'Test::Deep' => '0.111';
 test_requires 'Test::Fatal';

devel/config/replicaset-multi-3.6.yml

@@ -0,0 +1,11 @@
+---
+type: replica
+setName: foo
+default_args: -v --noprealloc --smallfiles --bind_ip 0.0.0.0 --nssize 6 --quiet
+default_version: 3.6
+mongod:
+    - name: host1
+    - name: host2
+    - name: host3
+
+# vim: ts=4 sts=4 sw=4 et:

devel/lib/MongoDBTest/Role/Server.pm

@@ -214,6 +214,7 @@ sub _build_tempdir {
     return Path::Tiny->tempdir(
         TEMPLATE => $self->name . "-XXXXXX",
         ($ENV{DATA_DIR} ? (DIR => $ENV{DATA_DIR}) : ()),
+        ($ENV{MONGO_PRESERVE} ? (CLEANUP => 0) : ()),
     );
 }
 

devel/t-dynamic/PERL-792-replica-set-failover.t

@@ -0,0 +1,156 @@
+#
+#  Copyright 2009-2013 MongoDB, Inc.
+#
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#
+#  http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+#
+
+# Test in t-dynamic as not sure if failover should be tested on install?
+
+use strict;
+use warnings;
+use JSON::MaybeXS;
+use Path::Tiny 0.054; # basename with suffix
+use Test::More 0.88;
+use Test::Fatal;
+use boolean;
+
+use lib "t/lib";
+use lib "devel/lib";
+
+use MongoDBTest::Orchestrator; 
+
+use MongoDBTest qw/
+    build_client
+    get_test_db
+    clear_testdbs
+    get_unique_collection
+    server_version
+    server_type
+    check_min_server_version
+    get_feature_compat_version
+/;
+
+my $orc =
+MongoDBTest::Orchestrator->new(
+  config_file => "devel/config/replicaset-multi-3.6.yml" );
+$orc->start;
+
+$ENV{MONGOD} = $orc->as_uri;
+
+my @events;
+
+sub clear_events { @events = () }
+sub event_count { scalar @events }
+sub event_cb { push @events, $_[0] }
+
+my $conn = build_client(
+    retry_writes => 1,
+    heartbeat_frequency_ms => 60 * 1000,
+    # build client modifies this so we set it explicitly to the default
+    server_selection_timeout_ms => 30 * 1000,
+    server_selection_try_once => 0,
+    monitoring_callback => \&event_cb,
+);
+my $testdb         = get_test_db($conn);
+my $coll = get_unique_collection( $testdb, 'retry_failover' );
+my $server_version = server_version($conn);
+my $server_type    = server_type($conn);
+my $feat_compat_ver = get_feature_compat_version($conn);
+
+plan skip_all => "standalone servers dont support retryableWrites"
+    if $server_type eq 'Standalone';
+
+plan skip_all => "retryableWrites requires featureCompatibilityVersion 3.6 - got $feat_compat_ver"
+    if ( $feat_compat_ver < 3.6 );
+
+my $primary = $conn->_topology->current_primary;
+
+my $fail_conn = build_client( host => $primary->address );
+
+my $step_down_conn = build_client();
+
+my $ret = $coll->insert_one( { _id => 1, test => 'value' } );
+
+is $ret->inserted_id, 1, 'write succeeded';
+
+my $result = $coll->find_one( { _id => 1 } );
+
+is $result->{test}, 'value', 'Successful write';
+
+$fail_conn->send_admin_command([
+    configureFailPoint => 'onPrimaryTransactionalWrite',
+    mode => 'alwaysOn',
+]);
+
+# wrapped in eval as this will just drop connection
+eval {
+    $step_down_conn->send_admin_command([
+        replSetStepDown => 60,
+        force => true,
+    ]);
+};
+my $err = $@;
+isa_ok( $err, 'MongoDB::NetworkError', 'Step down successfully errored' );
+
+clear_events();
+
+my $post_stepdown_ret = $coll->insert_one( { _id => 2, test => 'again' } );
+
+is $post_stepdown_ret->inserted_id, 2, 'write succeeded';
+
+# All this is to make sure we dont make assumptions on position of the actual
+# event, just that the failed one comes first.
+my $first_insert_index;
+
+for my $f_idx ( 0 .. $#events - 1 ) {
+    my $event = $events[ $f_idx ];
+    if ( $event->{ commandName } eq 'insert'
+      && $event->{ type } eq 'command_started' ) {
+        my $next_event = $events[ $f_idx + 1 ];
+        is $next_event->{ commandName }, 'insert', 'found insert reply';
+        is $next_event->{ type }, 'command_failed', 'found failed reply';
+        $first_insert_index = $f_idx;
+        last;
+    }
+}
+
+ok defined( $first_insert_index ), 'found first command';
+
+my $second_insert_index;
+
+if ( $first_insert_index + 2 > $#events - 1 ) {
+  fail 'not enough events captured';
+}
+
+for my $s_idx ( $first_insert_index + 2 .. $#events - 1 ) {
+    my $event = $events[ $s_idx ];
+    if ( $event->{ commandName } eq 'insert'
+      && $event->{ type } eq 'command_started' ) {
+        my $next_event = $events[ $s_idx + 1 ];
+        is $next_event->{ commandName }, 'insert', 'found insert reply';
+        is $next_event->{ type }, 'command_succeeded', 'found success reply';
+        $second_insert_index = $s_idx;
+        last;
+    }
+}
+
+ok defined( $second_insert_index ), 'found second command';
+
+$fail_conn->send_admin_command([
+    configureFailPoint => 'onPrimaryTransactionalWrite',
+    mode => 'off',
+]);
+
+clear_testdbs;
+
+done_testing;

lib/MongoDB/BulkWrite.pm

@@ -121,6 +121,12 @@ sub _build__client {
     return $self->_database->_client;
 }
 
+has _retryable => (
+    is => 'rw',
+    isa => Bool,
+    default => 1,
+);
+
 with $_ for qw(
   MongoDB::Role::_DeprecationWarner
 );
@@ -268,6 +274,7 @@ sub execute {
     my $session = $self->collection->_get_session_from_hashref( $options );
 
     my $op = MongoDB::Op::_BulkWrite->_new(
+        client                   => $self->_client,
         db_name                  => $self->_database->name,
         coll_name                => $self->collection->name,
         full_name                => $self->collection->full_name,
@@ -278,8 +285,10 @@ sub execute {
         write_concern            => $write_concern,
         session                  => $session,
         monitoring_callback      => $self->_client->monitoring_callback,
+        _retryable               => $self->_retryable,
     );
 
+    # Op::_BulkWrite internally does retryable writes
     return $self->_client->send_write_op( $op );
 }
 

lib/MongoDB/BulkWriteView.pm

@@ -126,6 +126,9 @@ sub _update {
         $doc = Tie::IxHash->new(%$doc);
     }
 
+    $self->_bulk->_retryable( 0 ) if $method eq 'update_many';
+    $self->_bulk->_retryable( 0 ) if $method eq 'delete_many';
+
     my $update = {
         q      => $self->_query,
         u      => $doc,

lib/MongoDB/Collection.pm

@@ -372,7 +372,7 @@ sub insert_one {
     MongoDB::UsageError->throw("document argument must be a reference")
       unless ref( $_[1] );
 
-    return $_[0]->client->send_write_op(
+    return $_[0]->client->send_retryable_write_op(
         MongoDB::Op::_InsertOne->_new(
             session => $_[0]->_get_session_from_hashref( $_[2] ),
             ( defined $_[2] ? (%{$_[2]}) : () ),
@@ -422,6 +422,7 @@ sub insert_many {
     MongoDB::UsageError->throw("documents argument must be an array reference")
       unless ref( $_[1] ) eq 'ARRAY';
 
+    # internally ends up performing a retryable write if possible, see OP::_BulkWrite
     my $res = $_[0]->client->send_write_op(
         MongoDB::Op::_BulkWrite->_new(
             # default
@@ -431,6 +432,8 @@ sub insert_many {
             ( defined $_[2] ? ( %{ $_[2] } ) : () ),
             # un-overridable
             queue => [ map { [ insert => $_ ] } @{ $_[1] } ],
+            # insert_many is specifically retryable (PERL-792)
+            _retryable => 1,
             %{ $_[0]->_op_args },
         )
     );
@@ -475,7 +478,7 @@ sub delete_one {
     MongoDB::UsageError->throw("filter argument must be a reference")
       unless ref( $_[1] );
 
-    return $_[0]->client->send_write_op(
+    return $_[0]->client->send_retryable_write_op(
         MongoDB::Op::_Delete->_new(
             session => $_[0]->_get_session_from_hashref( $_[2] ),
             ( defined $_[2] ? (%{$_[2]}) : () ),
@@ -557,7 +560,7 @@ sub replace_one {
     MongoDB::UsageError->throw("filter and replace arguments must be references")
       unless ref( $_[1] ) && ref( $_[2] );
 
-    return $_[0]->client->send_write_op(
+    return $_[0]->client->send_retryable_write_op(
         MongoDB::Op::_Update->_new(
             session => $_[0]->_get_session_from_hashref( $_[3] ),
             ( defined $_[3] ? (%{$_[3]}) : () ),
@@ -608,7 +611,7 @@ sub update_one {
     MongoDB::UsageError->throw("filter and update arguments must be references")
       unless ref( $_[1] ) && ref( $_[2] );
 
-    return $_[0]->client->send_write_op(
+    return $_[0]->client->send_retryable_write_op(
         MongoDB::Op::_Update->_new(
             session => $_[0]->_get_session_from_hashref( $_[3] ),
             ( defined $_[3] ? (%{$_[3]}) : () ),
@@ -940,7 +943,9 @@ sub find_one_and_delete {
         session       => $session,
     );
 
-    return $self->client->send_write_op($op);
+    return $self->write_concern->is_acknowledged
+      ? $self->client->send_retryable_write_op( $op )
+      : $self->client->send_write_op( $op );
 }
 
 =method find_one_and_replace
@@ -1647,6 +1652,7 @@ sub bulk_write {
             }
             elsif ( $method eq 'delete_many' ) {
                 $view->delete_many;
+                $bulk->_retryable( 0 );
                 next;
             }
 
@@ -1662,6 +1668,7 @@ sub bulk_write {
             }
             elsif ( $method eq 'update_many' ) {
                 $view->update_many($update_doc);
+                $bulk->_retryable( 0 );
             }
             else {
                 MongoDB::UsageError->throw("unknown bulk operation '$method'");
@@ -1733,7 +1740,9 @@ sub _find_one_and_update_or_replace {
         %{ $self->_op_args },
     );
 
-    return $self->client->send_write_op($op);
+    return $self->write_concern->is_acknowledged
+      ? $self->client->send_retryable_write_op( $op )
+      : $self->client->send_write_op( $op );
 }
 
 # Extracts a session from a provided hashref, or returns an implicit session