CSHARP-2221: Implement transactions.

Author: rstam

evergreen/evergreen.yml

@@ -349,10 +349,10 @@ axes:
   - id: version
     display_name: MongoDB Version
     values:
-#      - id: "latest"
-#        display_name: "latest"
-#        variables:
-#           VERSION: "latest"
+      - id: "latest"
+        display_name: "latest"
+        variables:
+           VERSION: "latest"
       - id: "3.6"
         display_name: "3.6"
         variables:

src/MongoDB.Bson/ObjectModel/RawBsonArray.cs

@@ -16,6 +16,7 @@
 using System;
 using System.Collections;
 using System.Collections.Generic;
+using System.IO;
 using System.Linq;
 using MongoDB.Bson.IO;
 using MongoDB.Bson.Serialization;
@@ -521,6 +522,28 @@ public override void Insert(int index, BsonValue value)
             throw new NotSupportedException("RawBsonArray instances are immutable.");
         }
 
+        /// <summary>
+        /// Materializes the RawBsonArray into a regular BsonArray.
+        /// </summary>
+        /// <param name="binaryReaderSettings">The binary reader settings.</param>
+        /// <returns>A BsonArray.</returns>
+        public BsonArray Materialize(BsonBinaryReaderSettings binaryReaderSettings)
+        {
+            ThrowIfDisposed();
+
+            // because BsonBinaryReader can only read documents at the top level we have to wrap the RawBsonArray in a document
+            var document = new BsonDocument("array", this);
+            var bytes = document.ToBson();
+
+            using (var stream = new MemoryStream(bytes))
+            using (var reader = new BsonBinaryReader(stream, binaryReaderSettings))
+            {
+                var context = BsonDeserializationContext.CreateRoot(reader);
+                var materializedDocument = BsonDocumentSerializer.Instance.Deserialize(context);
+                return materializedDocument["array"].AsBsonArray;
+            }
+        }
+
         /// <summary>
         /// Removes the first occurrence of a value from the array.
         /// </summary>

src/MongoDB.Driver.Core/Core/Bindings/CoreSession.cs

@@ -13,6 +13,9 @@
 * limitations under the License.
 */
 
+using System;
+using System.Threading;
+using System.Threading.Tasks;
 using MongoDB.Bson;
 using MongoDB.Driver.Core.Clusters;
 using MongoDB.Driver.Core.Misc;
@@ -27,7 +30,9 @@ namespace MongoDB.Driver.Core.Bindings
     public sealed class CoreSession : ICoreSession
     {
         // private fields
+        private readonly ICluster _cluster;
         private readonly IClusterClock _clusterClock = new ClusterClock();
+        private CoreTransaction _currentTransaction;
         private bool _disposed;
         private readonly IOperationClock _operationClock = new OperationClock();
         private readonly CoreSessionOptions _options;
@@ -37,20 +42,29 @@ public sealed class CoreSession : ICoreSession
         /// <summary>
         /// Initializes a new instance of the <see cref="CoreSession" /> class.
         /// </summary>
+        /// <param name="cluster">The cluster.</param>
         /// <param name="serverSession">The server session.</param>
         /// <param name="options">The options.</param>
         public CoreSession(
+            ICluster cluster,
             ICoreServerSession serverSession,
             CoreSessionOptions options)
         {
+            _cluster = Ensure.IsNotNull(cluster, nameof(cluster));
             _serverSession = Ensure.IsNotNull(serverSession, nameof(serverSession));
             _options = Ensure.IsNotNull(options, nameof(options));
         }
 
         // public properties
+        /// <inheritdoc />
+        public ICluster Cluster => _cluster;
+
         /// <inheritdoc />
         public BsonDocument ClusterTime => _clusterClock.ClusterTime;
 
+        /// <inheritdoc />
+        public CoreTransaction CurrentTransaction => _currentTransaction;
+
         /// <inheritdoc />
         public BsonDocument Id => _serverSession.Id;
 
@@ -60,6 +74,9 @@ public CoreSession(
         /// <inheritdoc />
         public bool IsImplicit => _options.IsImplicit;
 
+        /// <inheritdoc />
+        public bool IsInTransaction => _currentTransaction != null;
+
         /// <inheritdoc />
         public BsonTimestamp OperationTime => _operationClock.OperationTime;
 
@@ -70,6 +87,100 @@ public CoreSession(
         public ICoreServerSession ServerSession => _serverSession;
 
         // public methods
+        /// <inheritdoc />
+        public void AbortTransaction(CancellationToken cancellationToken = default(CancellationToken))
+        {
+            EnsureIsInTransaction(nameof(AbortTransaction));
+
+            try
+            {
+                if (_currentTransaction.StatementId == 0)
+                {
+                    return;
+                }
+
+                try
+                {
+                    var firstAttempt = CreateAbortTransactionOperation();
+                    ExecuteEndTransactionOnPrimary(firstAttempt, cancellationToken);
+                    return;
+                }
+                catch (Exception exception) when (ShouldIgnoreAbortTransactionException(exception))
+                {
+                    return; // ignore exception and return
+                }
+                catch (Exception exception) when (ShouldRetryEndTransactionException(exception))
+                {
+                    // ignore exception and retry
+                }
+                catch
+                {
+                    return; // ignore exception and return
+                }
+
+                try
+                {
+                    var secondAttempt = CreateAbortTransactionOperation();
+                    ExecuteEndTransactionOnPrimary(secondAttempt, cancellationToken);
+                }
+                catch
+                {
+                    return; // ignore exception and return
+                }
+            }
+            finally
+            {
+                _currentTransaction = null;
+            }
+        }
+
+        /// <inheritdoc />
+        public async Task AbortTransactionAsync(CancellationToken cancellationToken = default(CancellationToken))
+        {
+            EnsureIsInTransaction(nameof(AbortTransaction));
+
+            try
+            {
+                if (_currentTransaction.StatementId == 0)
+                {
+                    return;
+                }
+
+                try
+                {
+                    var firstAttempt = CreateAbortTransactionOperation();
+                    await ExecuteEndTransactionOnPrimaryAsync(firstAttempt, cancellationToken).ConfigureAwait(false);
+                    return;
+                }
+                catch (Exception exception) when (ShouldIgnoreAbortTransactionException(exception))
+                {
+                    return; // ignore exception and return
+                }
+                catch (Exception exception) when (ShouldRetryEndTransactionException(exception))
+                {
+                    // ignore exception and retry
+                }
+                catch
+                {
+                    return; // ignore exception and return
+                }
+
+                try
+                {
+                    var secondAttempt = CreateAbortTransactionOperation();
+                    await ExecuteEndTransactionOnPrimaryAsync(secondAttempt, cancellationToken).ConfigureAwait(false);
+                }
+                catch
+                {
+                    return; // ignore exception and return
+                }
+            }
+            finally
+            {
+                _currentTransaction = null;
+            }
+        }
+
         /// <inheritdoc />
         public void AdvanceClusterTime(BsonDocument newClusterTime)
         {
@@ -88,20 +199,175 @@ public long AdvanceTransactionNumber()
             return _serverSession.AdvanceTransactionNumber();
         }
 
+        /// <inheritdoc />
+        public void CommitTransaction(CancellationToken cancellationToken = default(CancellationToken))
+        {
+            EnsureIsInTransaction(nameof(CommitTransaction));
+
+            try
+            {
+                if (_currentTransaction.StatementId == 0)
+                {
+                    return;
+                }
+
+                try
+                {
+                    var firstAttempt = CreateCommitTransactionOperation();
+                    ExecuteEndTransactionOnPrimary(firstAttempt, cancellationToken);
+                    return;
+                }
+                catch (Exception exception) when (ShouldRetryEndTransactionException(exception))
+                {
+                    // ignore exception and retry
+                }
+
+                var secondAttempt = CreateCommitTransactionOperation();
+                ExecuteEndTransactionOnPrimary(secondAttempt, cancellationToken);
+            }
+            finally
+            {
+                _currentTransaction = null;
+            }
+        }
+
+        /// <inheritdoc />
+        public async Task CommitTransactionAsync(CancellationToken cancellationToken = default(CancellationToken))
+        {
+            EnsureIsInTransaction(nameof(CommitTransaction));
+
+            try
+            {
+                if (_currentTransaction.StatementId == 0)
+                {
+                    return;
+                }
+
+                try
+                {
+                    var firstAttempt = CreateCommitTransactionOperation();
+                    await ExecuteEndTransactionOnPrimaryAsync(firstAttempt, cancellationToken).ConfigureAwait(false);
+                    return;
+                }
+                catch (Exception exception) when (ShouldRetryEndTransactionException(exception))
+                {
+                    // ignore exception and retry
+                }
+
+                var secondAttempt = CreateCommitTransactionOperation();
+                await ExecuteEndTransactionOnPrimaryAsync(secondAttempt, cancellationToken).ConfigureAwait(false);
+            }
+            finally
+            {
+                _currentTransaction = null;
+            }
+        }
+
         /// <inheritdoc />
         public void Dispose()
         {
             if (!_disposed)
             {
+                if (_currentTransaction != null)
+                {
+                    try
+                    {
+                        AbortTransaction(CancellationToken.None);
+                    }
+                    catch
+                    {
+                        // ignore exceptions
+                    }
+                }
+
                 _serverSession.Dispose();
                 _disposed = true;
             }
         }
 
+        /// <inheritdoc />
+        public void StartTransaction(TransactionOptions transactionOptions = null)
+        {
+            if (_currentTransaction != null)
+            {
+                throw new InvalidOperationException("Transaction already in progress.");
+            }
+
+            var transactionNumber = AdvanceTransactionNumber();
+            var readConcern = transactionOptions?.ReadConcern ?? _options.DefaultTransactionOptions?.ReadConcern ?? ReadConcern.Default;
+            var writeConcern = transactionOptions?.WriteConcern ?? _options.DefaultTransactionOptions?.WriteConcern ?? new WriteConcern();
+            var effectiveTransactionOptions = new TransactionOptions(readConcern, writeConcern);
+            var transaction = new CoreTransaction(transactionNumber, effectiveTransactionOptions);
+
+            _currentTransaction = transaction;
+        }
+
         /// <inheritdoc />
         public void WasUsed()
         {
             _serverSession.WasUsed();
         }
+
+        // private methods
+        private IReadOperation<BsonDocument> CreateAbortTransactionOperation()
+        {
+            return new AbortTransactionOperation(GetTransactionWriteConcern());
+        }
+
+        private IReadOperation<BsonDocument> CreateCommitTransactionOperation()
+        {
+            return new CommitTransactionOperation(GetTransactionWriteConcern());
+        }
+
+        private void EnsureIsInTransaction(string methodName)
+        {
+            this.AutoStartTransactionIfApplicable();
+            if (_currentTransaction == null)
+            {
+                throw new InvalidOperationException("No transaction started.");
+            }
+        }
+
+        private TResult ExecuteEndTransactionOnPrimary<TResult>(IReadOperation<TResult> operation, CancellationToken cancellationToken)
+        {
+            using (var sessionHandle = new NonDisposingCoreSessionHandle(this))
+            using (var binding = new WritableServerBinding(_cluster, sessionHandle))
+            {
+                return operation.Execute(binding, cancellationToken);
+            }
+        }
+
+        private async Task<TResult> ExecuteEndTransactionOnPrimaryAsync<TResult>(IReadOperation<TResult> operation, CancellationToken cancellationToken)
+        {
+            using (var sessionHandle = new NonDisposingCoreSessionHandle(this))
+            using (var binding = new WritableServerBinding(_cluster, sessionHandle))
+            {
+                return await operation.ExecuteAsync(binding, cancellationToken).ConfigureAwait(false);
+            }
+        }
+
+        private WriteConcern GetTransactionWriteConcern()
+        {
+            return
+                _currentTransaction.TransactionOptions?.WriteConcern ??
+                _options.DefaultTransactionOptions?.WriteConcern ??
+                WriteConcern.WMajority;
+        }
+
+        private bool ShouldIgnoreAbortTransactionException(Exception exception)
+        {
+            var commandException = exception as MongoCommandException;
+            if (commandException != null)
+            {
+                return true;
+            }
+
+            return false;
+        }
+
+        private bool ShouldRetryEndTransactionException(Exception exception)
+        {
+            return RetryabilityHelper.IsRetryableWriteException(exception);
+        }
     }
 }