PERL-875 - Transaction Support

Makefile.PL

@@ -43,7 +43,7 @@ my %WriteMakefileArgs = (
     "Moo" => 2,
     "Moo::Role" => 0,
     "Net::DNS" => 0,
-    "Safe::Isa" => 0,
+    "Safe::Isa" => '1.000007',
     "Scalar::Util" => 0,
     "Socket" => 0,
     "Sub::Quote" => 0,

devel/config/replicaset-multi-4.0-w_arbiter.yml

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

devel/config/replicaset-multi-4.0.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: 4.0
+mongod:
+    - name: host1
+    - name: host2
+    - name: host3
+
+# vim: ts=4 sts=4 sw=4 et:

lib/MongoDB/ClientSession.pm

@@ -24,15 +24,19 @@ our $VERSION = 'v1.999.1';
 use MongoDB::Error;
 
 use Moo;
+use MongoDB::_Constants;
 use MongoDB::_Types qw(
     Document
     BSONTimestamp
+    TransactionState
+    Boolish
 );
 use Types::Standard qw(
     Maybe
     HashRef
     InstanceOf
 );
+use MongoDB::_TransactionOptions;
 use namespace::clean -except => 'meta';
 
 =attr client
@@ -66,13 +70,17 @@ has cluster_time => (
 
 Options provided for this particular session. Available options include:
 
-=for :list 
+=for :list
 * C<causalConsistency> - If true, will enable causalConsistency for
   this session. For more information, see L<MongoDB documentation on Causal
   Consistency|https://docs.mongodb.com/manual/core/read-isolation-consistency-recency/#causal-consistency>.
   Note that causalConsistency does not apply for unacknowledged writes.
   Defaults to true.
-
+* C<defaultTransactionOptions> - Options to use by default for transactions
+  created with this session. If when creating a transaction, none or only some of
+  the transaction options are defined, these options will be used as a fallback.
+  Defaults to inheriting from the parent client. See L</start_transaction> for
+  available options.
 
 =cut
 
@@ -83,10 +91,16 @@ has options => (
     # Shallow copy to prevent action at a distance.
     # Upgrade to use Storable::dclone if a more complex option is required
     coerce => sub {
-      $_[0] = {
-        causalConsistency => 1,
-        %{ $_[0] }
-      };
+        # Will cause the isa requirement to fire
+        return unless defined( $_[0] ) && ref( $_[0] ) eq 'HASH';
+        $_[0] = {
+            causalConsistency => defined $_[0]->{causalConsistency}
+                ? $_[0]->{causalConsistency}
+                : 1,
+            defaultTransactionOptions => {
+                %{ $_[0]->{defaultTransactionOptions} || {} }
+            },
+        };
     },
 );
 
@@ -98,6 +112,37 @@ has _server_session => (
     clearer => '__clear_server_session',
 );
 
+has _current_transaction_options => (
+    is => 'rwp',
+    isa => InstanceOf[ 'MongoDB::_TransactionOptions' ],
+    handles  => {
+        _get_transaction_write_concern   => 'write_concern',
+        _get_transaction_read_concern    => 'read_concern',
+        _get_transaction_read_preference => 'read_preference',
+    },
+);
+
+has _transaction_state => (
+    is => 'rwp',
+    isa => TransactionState,
+    default => 'none',
+);
+
+# Flag used to say we are still in a transaction
+has _active_transaction => (
+    is => 'rwp',
+    isa => Boolish,
+    default => 0,
+);
+
+# Flag used to say whether any operations have been performed on the
+# transaction
+has _has_transaction_operations => (
+    is => 'rwp',
+    isa => Boolish,
+    default => 0,
+);
+
 =attr operation_time
 
 The last operation time. This is updated when an operation is performed during
@@ -215,6 +260,196 @@ sub advance_operation_time {
     return;
 }
 
+# Returns 1 if the session is in one of the specified transaction states.
+# Returns a false value if not in any of the states defined as an argument.
+sub _in_transaction_state {
+    my ( $self, @states ) = @_;
+    return 1 if scalar ( grep { $_ eq $self->_transaction_state } @states );
+    return;
+}
+
+=method start_transaction
+
+Start a transaction in this session. Takes a hashref of options which can contain the following options:
+
+=for :list
+* C<readConcern> - The read concern to use for the first command in this
+  transaction. If not defined here or in the C<defaultTransactionOptions> in
+  L</options>, will inherit from the parent client.
+* C<writeConcern> - The write concern to use for committing or aborting this
+  transaction. As per C<readConcern>, if not defined here then the value defined
+  in C<defaultTransactionOptions> will be used, or the parent client if not
+  defined.
+* C<readPreference> - The read preference to use for all read operations in
+  this transaction. If not defined, then will inherit from
+  C<defaultTransactionOptions> or from the parent client. This value will
+  override all other read preferences set in any subsequent commands inside this
+  transaction.
+
+=cut
+
+sub start_transaction {
+    my ( $self, $opts ) = @_;
+
+    MongoDB::UsageError->throw("Transaction already in progress")
+        if $self->_in_transaction_state( TXN_STARTING, TXN_IN_PROGRESS );
+
+    MongoDB::ConfigurationError->throw("Transactions are unsupported on this deployment")
+        unless $self->client->_topology->_supports_transactions;
+
+    $opts ||= {};
+    my $trans_opts = MongoDB::_TransactionOptions->new(
+        client => $self->client,
+        options => $opts,
+        default_options => $self->options->{defaultTransactionOptions},
+    );
+
+    $self->_set__current_transaction_options( $trans_opts );
+
+    $self->_set__transaction_state( TXN_STARTING );
+
+    $self->_increment_transaction_id;
+
+    $self->_set__active_transaction( 1 );
+    $self->_set__has_transaction_operations( 0 );
+
+    return;
+}
+
+sub _increment_transaction_id {
+    my $self = shift;
+    return if $self->_active_transaction;
+
+    $self->_server_session->transaction_id->binc();
+}
+
+=method commit_transaction
+
+Commit the current transaction. This will use the writeConcern set on this transaction.
+
+=cut
+
+sub commit_transaction {
+    my $self = shift;
+
+    MongoDB::UsageError->throw("No transaction started")
+        if $self->_in_transaction_state( TXN_NONE );
+
+    # Error message tweaked to use our function names
+    MongoDB::UsageError->throw("Cannot call commit_transaction after calling abort_transaction")
+        if $self->_in_transaction_state( TXN_ABORTED );
+
+    # Commit can be called multiple times - even if the transaction completes
+    # correctly. Setting this here makes sure we dont increment transaction id
+    # until after another command has been called using this session
+    $self->_set__active_transaction( 1 );
+
+    eval {
+        $self->_send_end_transaction_command( TXN_COMMITTED, [ commitTransaction => 1 ] );
+    };
+    if ( my $err = $@ ) {
+        # catch and re-throw after retryable errors
+        # TODO maybe need better checking logic that theres actually an error code in output?
+        my $err_code_name;
+        my $err_code;
+        if ( $err->can('result') ) {
+            if ( $err->result->can('output') ) {
+                $err_code_name = $err->result->output->{codeName};
+                $err_code = $err->result->output->{code};
+                $err_code_name ||= $err->result->output->{writeConcernError}
+                    ? $err->result->output->{writeConcernError}->{codeName}
+                    : ''; # Empty string just in case
+                $err_code ||= $err->result->output->{writeConcernError}
+                    ? $err->result->output->{writeConcernError}->{code}
+                    : 0; # just in case
+            }
+        }
+        # If its a write concern error, retrying a commit would still error
+        unless (
+            ( defined( $err_code_name ) && grep { $_ eq $err_code_name } qw/
+                CannotSatisfyWriteConcern
+                UnsatisfiableWriteConcern
+                UnknownReplWriteConcern
+                NoSuchTransaction
+            / )
+            # Spec tests include code numbers only with no codeName
+            || ( defined ( $err_code ) && grep { $_ == $err_code }
+                100, # UnsatisfiableWriteConcern/CannotSatisfyWriteConcern
+                79,  # UnknownReplWriteConcern
+                251, # NoSuchTransaction
+            )
+        ) {
+            push @{ $err->error_labels }, 'UnknownTransactionCommitResult';
+        }
+        die $err;
+    }
+
+    return;
+}
+
+=method abort_transaction
+
+Abort the current transaction. This will use the writeConcern set on this transaction.
+
+=cut
+
+sub abort_transaction {
+    my $self = shift;
+
+    MongoDB::UsageError->throw("No transaction started")
+        if $self->_in_transaction_state( TXN_NONE );
+
+    # Error message tweaked to use our function names
+    MongoDB::UsageError->throw("Cannot call abort_transaction after calling commit_transaction")
+        if $self->_in_transaction_state( TXN_COMMITTED );
+
+    # Error message tweaked to use our function names
+    MongoDB::UsageError->throw("Cannot call abort_transaction twice")
+        if $self->_in_transaction_state( TXN_ABORTED );
+
+    eval {
+        $self->_send_end_transaction_command( TXN_ABORTED, [ abortTransaction => 1 ] );
+    };
+    # Ignore all errors thrown by abortTransaction
+
+    return;
+}
+
+sub _send_end_transaction_command {
+    my ( $self, $end_state, $command ) = @_;
+
+    $self->_set__transaction_state( $end_state );
+
+    # Only need to send commit command if the transaction actually sent anything
+    if ( $self->_has_transaction_operations ) {
+        my $op = MongoDB::Op::_Command->_new(
+            db_name             => 'admin',
+            query               => $command,
+            query_flags         => {},
+            bson_codec          => $self->client->bson_codec,
+            session             => $self,
+            monitoring_callback => $self->client->monitoring_callback,
+        );
+
+        my $result = $self->client->send_retryable_write_op( $op, 'force' );
+        # TODO This may be redundant after 918 is merged?
+        $result->assert_no_write_concern_error;
+    }
+
+    # If the commit/abort succeeded, we are no longer in an active transaction
+    $self->_set__active_transaction( 0 );
+}
+
+# For applying connection errors etc
+sub _maybe_apply_error_labels {
+    my ( $self, $err ) = @_;
+
+    if ( $self->_in_transaction_state( TXN_STARTING, TXN_IN_PROGRESS ) ) {
+        push @{ $err->error_labels }, 'TransientTransactionError';
+    }
+    return;
+}
+
 =method end_session
 
     $session->end_session;
@@ -227,6 +462,10 @@ recycling.  Has no effect after calling for the first time.
 sub end_session {
     my ( $self ) = @_;
 
+    if ( $self->_in_transaction_state ( TXN_IN_PROGRESS ) ) {
+        # Ignore all errors
+        eval { $self->abort_transaction };
+    }
     if ( defined $self->_server_session ) {
         $self->client->_server_session_pool->retire_server_session( $self->_server_session );
         $self->__clear_server_session;

lib/MongoDB/CommandResult.pm

@@ -115,6 +115,18 @@ sub last_wtimeout {
         || exists $self->output->{writeConcernError} );
 }
 
+=method last_error_labels
+
+Returns an array of error labels from the command, or an empty array if there
+are none
+
+=cut
+
+sub last_error_labels {
+    my ( $self ) = @_;
+    return $self->output->{errorLabels} || [];
+}
+
 =method assert
 
 Throws an exception if the command failed.