Fix bug 616: Not possible to use Serializable transactions

src/Marten.Testing/Bugs/Bug_616_not_possible_to_use_Serializable_transactions.cs

@@ -0,0 +1,45 @@
+using System;
+using System.Data;
+using Marten.Services;
+using Xunit;
+
+namespace Marten.Testing.Bugs
+{
+    public class Account
+    {
+        public Guid Id { get; set; }
+        public decimal Amount { get; set; }
+
+        public void Substract(int value)
+        {
+            Amount = Amount - value;
+        }
+    }
+
+    public class Bug_616_not_possible_to_use_Serializable_transactions : DocumentSessionFixture<NulloIdentityMap>
+    {
+        [Fact]
+        public void conccurent_write_should_throw_an_exception()
+        {
+            var accountA = new Account { Id = Guid.NewGuid(), Amount = 100 };
+            theSession.Store(accountA);
+            theSession.SaveChanges();
+
+            using (var session1 = theStore.DirtyTrackedSession(IsolationLevel.Serializable))
+            using (var session2 = theStore.DirtyTrackedSession(IsolationLevel.Serializable))
+            {
+                var session1AcountA = session1.Load<Account>(accountA.Id);
+                session1AcountA.Substract(500);
+
+                var session2AcountA = session2.Load<Account>(accountA.Id);
+                session2AcountA.Substract(350);
+
+                session1.SaveChanges();
+
+                Assert.Throws<ConcurrentUpdateException>(() => {
+                    session2.SaveChanges();
+                });
+            }
+        }
+    }
+}

src/Marten/DocumentSession.cs

@@ -184,11 +184,10 @@ public void SaveChanges()
             var batch = new UpdateBatch(_options, _serializer, _connection, IdentityMap.Versions);
             var changes = _unitOfWork.ApplyChanges(batch);
 
-
             try
             {
                 _connection.Commit();
-            }
+            }           
             catch (Exception)
             {
                 // This code has a try/catch in it to stop

src/Marten/DocumentStore.cs

@@ -263,21 +263,23 @@ private void bulkInsertDocuments<T>(T[] documents, int batchSize, NpgsqlConnecti
 
         public IDocumentSession OpenSession(SessionOptions options)
         {
-            var map = createMap(options.Tracking);
-            var session = new DocumentSession(this, _options, Schema, _serializer,
-                new ManagedConnection(_connectionFactory, CommandRunnerMode.Transactional, options.IsolationLevel, options.Timeout), _parser, map);
-
-            session.Logger = _logger.StartSession(session);
-
-            return session;
+            var connection = new ManagedConnection(_connectionFactory, CommandRunnerMode.Transactional, options.IsolationLevel, options.Timeout);
+            return openSession(options.Tracking, connection);
         }
 
         public IDocumentSession OpenSession(DocumentTracking tracking = DocumentTracking.IdentityOnly,
             IsolationLevel isolationLevel = IsolationLevel.ReadCommitted)
+        {
+            var connection = new ManagedConnection(_connectionFactory, CommandRunnerMode.Transactional, isolationLevel);
+            return openSession(tracking, connection);
+        }
+
+        private IDocumentSession openSession(DocumentTracking tracking, ManagedConnection connection)
         {
             var map = createMap(tracking);
-            var session = new DocumentSession(this, _options, Schema, _serializer,
-                new ManagedConnection(_connectionFactory, CommandRunnerMode.Transactional, isolationLevel), _parser, map);
+            var session = new DocumentSession(this, _options, Schema, _serializer, connection, _parser, map);
+
+            connection.BeginSession();
 
             session.Logger = _logger.StartSession(session);
 

src/Marten/Services/ConcurrentUpdateException.cs

@@ -0,0 +1,11 @@
+using System;
+
+namespace Marten.Services
+{
+    public class ConcurrentUpdateException : Exception
+    {
+        public ConcurrentUpdateException(Exception innerException) : base("Write collission detected while commiting the transaction.", innerException)
+        {
+        }
+    }
+}

src/Marten/Services/ManagedConnection.cs

@@ -9,7 +9,7 @@
 namespace Marten.Services
 {
     public class ManagedConnection : IManagedConnection
-    {
+    {        
         private readonly IConnectionFactory _factory;
         private readonly CommandRunnerMode _mode;
         private readonly IsolationLevel _isolationLevel;
@@ -118,6 +118,14 @@ public async Task RollbackAsync(CancellationToken token)
             }
         }
 
+        public void BeginSession()
+        {
+            if (_isolationLevel == IsolationLevel.Serializable)
+            {
+                BeginTransaction();
+            }
+        }
+
         public void BeginTransaction()
         {
             buildConnection();
@@ -152,6 +160,11 @@ private void handleCommandException(NpgsqlCommand cmd, Exception e)
             this.SafeDispose();
             Logger.LogFailure(cmd, e);
 
+            if ((e as PostgresException)?.SqlState == PostgresErrorCodes.SerializationFailure)
+            {
+                throw new ConcurrentUpdateException(e);
+            }
+
             if (e is NpgsqlException)
             {
                 if (e.Message.IndexOf(EventContracts.UnexpectedMaxEventIdForStream, StringComparison.Ordinal) > -1)
@@ -329,11 +342,14 @@ public T Execute<T>(NpgsqlCommand cmd, Func<NpgsqlCommand, T> func)
             }
         }
 
-
-
         public void Dispose()
         {
             _connection?.Dispose();
         }
     }
+
+    public static class PostgresErrorCodes
+    {
+        public const string SerializationFailure = "40001";
+    }
 }