neo4j
diff --git a/‎driver/src/main/java/org/neo4j/driver/internal/async/InternalAsyncTransaction.java
Copy file name to clipboard
-5Lines changed: 0 additions & 5 deletions b/‎driver/src/main/java/org/neo4j/driver/internal/async/InternalAsyncTransaction.java
Copy file name to clipboard
-5Lines changed: 0 additions & 5 deletions
diff --git a/‎driver/src/main/java/org/neo4j/driver/internal/async/NetworkSession.java
Copy file name to clipboard
+1-1Lines changed: 1 addition & 1 deletion b/‎driver/src/main/java/org/neo4j/driver/internal/async/NetworkSession.java
Copy file name to clipboard
+1-1Lines changed: 1 addition & 1 deletion
diff --git a/‎driver/src/main/java/org/neo4j/driver/internal/async/UnmanagedTransaction.java
Copy file name to clipboard
+71-17Lines changed: 71 additions & 17 deletions b/‎driver/src/main/java/org/neo4j/driver/internal/async/UnmanagedTransaction.java
Copy file name to clipboard
+71-17Lines changed: 71 additions & 17 deletions
diff --git a/‎driver/src/main/java/org/neo4j/driver/internal/handlers/TransactionPullResponseCompletionListener.java
Copy file name to clipboard
+1-1Lines changed: 1 addition & 1 deletion b/‎driver/src/main/java/org/neo4j/driver/internal/handlers/TransactionPullResponseCompletionListener.java
Copy file name to clipboard
+1-1Lines changed: 1 addition & 1 deletion
diff --git a/‎driver/src/main/java/org/neo4j/driver/internal/reactive/InternalRxTransaction.java
Copy file name to clipboard
+1-1Lines changed: 1 addition & 1 deletion b/‎driver/src/main/java/org/neo4j/driver/internal/reactive/InternalRxTransaction.java
Copy file name to clipboard
+1-1Lines changed: 1 addition & 1 deletion
diff --git a/‎driver/src/main/java/org/neo4j/driver/internal/retry/ExponentialBackoffRetryLogic.java
Copy file name to clipboard
+57-39Lines changed: 57 additions & 39 deletions b/‎driver/src/main/java/org/neo4j/driver/internal/retry/ExponentialBackoffRetryLogic.java
Copy file name to clipboard
+57-39Lines changed: 57 additions & 39 deletions
@@ -50,11 +50,6 @@ public CompletionStage<ResultCursor> runAsync(Query query)
         return tx.runAsync(query, true );
     }
 
-    public void markTerminated()
-    {
-        tx.markTerminated();
-    }
-
     public boolean isOpen()
     {
         return tx.isOpen();
 
@@ -138,7 +138,7 @@ public CompletionStage<Void> resetAsync()
                 {
                     if ( tx != null )
                     {
-                        tx.markTerminated();
+                        tx.markTerminated( null );
                     }
                 } )
                 .thenCompose( ignore -> connectionStage )
 
@@ -18,6 +18,7 @@
  */
 package org.neo4j.driver.internal.async;
 
+import java.util.EnumSet;
 import java.util.concurrent.CompletionException;
 import java.util.concurrent.CompletionStage;
 import java.util.function.BiFunction;
@@ -58,13 +59,66 @@ private enum State
         ROLLED_BACK
     }
 
+    /**
+     * This is a holder so that we can have ony the state volatile in the tx without having to synchronize the whole block.
+     */
+    private static final class StateHolder
+    {
+        private static final EnumSet<State> OPEN_STATES = EnumSet.of( State.ACTIVE, State.TERMINATED );
+        private static final StateHolder ACTIVE_HOLDER = new StateHolder( State.ACTIVE, null );
+        private static final StateHolder COMMITTED_HOLDER = new StateHolder( State.COMMITTED, null );
+        private static final StateHolder ROLLED_BACK_HOLDER = new StateHolder( State.ROLLED_BACK, null );
+
+        /**
+         * The actual state.
+         */
+        final State value;
+
+        /**
+         * If this holder contains a state of {@link State#TERMINATED}, this represents the cause if any.
+         */
+        final Throwable causeOfTermination;
+
+        static StateHolder of( State value )
+        {
+            switch ( value )
+            {
+            case ACTIVE:
+                return ACTIVE_HOLDER;
+            case COMMITTED:
+                return COMMITTED_HOLDER;
+            case ROLLED_BACK:
+                return ROLLED_BACK_HOLDER;
+            case TERMINATED:
+            default:
+                throw new IllegalArgumentException( "Cannot provide a default state holder for state " + value );
+            }
+        }
+
+        static StateHolder terminatedWith( Throwable cause )
+        {
+            return new StateHolder( State.TERMINATED, cause );
+        }
+
+        private StateHolder( State value, Throwable causeOfTermination )
+        {
+            this.value = value;
+            this.causeOfTermination = causeOfTermination;
+        }
+
+        boolean isOpen()
+        {
+            return OPEN_STATES.contains( this.value );
+        }
+    }
+
     private final Connection connection;
     private final BoltProtocol protocol;
     private final BookmarkHolder bookmarkHolder;
     private final ResultCursorsHolder resultCursors;
     private final long fetchSize;
 
-    private volatile State state = State.ACTIVE;
+    private volatile StateHolder state = StateHolder.of( State.ACTIVE );
 
     public UnmanagedTransaction(Connection connection, BookmarkHolder bookmarkHolder, long fetchSize )
     {
@@ -104,11 +158,11 @@ public CompletionStage<Void> closeAsync()
 
     public CompletionStage<Void> commitAsync()
     {
-        if ( state == State.COMMITTED )
+        if ( state.value == State.COMMITTED )
         {
             return failedFuture( new ClientException( "Can't commit, transaction has been committed" ) );
         }
-        else if ( state == State.ROLLED_BACK )
+        else if ( state.value == State.ROLLED_BACK )
         {
             return failedFuture( new ClientException( "Can't commit, transaction has been rolled back" ) );
         }
@@ -122,11 +176,11 @@ else if ( state == State.ROLLED_BACK )
 
     public CompletionStage<Void> rollbackAsync()
     {
-        if ( state == State.COMMITTED )
+        if ( state.value == State.COMMITTED )
         {
             return failedFuture( new ClientException( "Can't rollback, transaction has been committed" ) );
         }
-        else if ( state == State.ROLLED_BACK )
+        else if ( state.value == State.ROLLED_BACK )
         {
             return failedFuture( new ClientException( "Can't rollback, transaction has been rolled back" ) );
         }
@@ -158,12 +212,12 @@ public CompletionStage<RxResultCursor> runRx(Query query)
 
     public boolean isOpen()
     {
-        return state != State.COMMITTED && state != State.ROLLED_BACK;
+        return state.isOpen();
     }
 
-    public void markTerminated()
+    public void markTerminated( Throwable cause )
     {
-        state = State.TERMINATED;
+        state = StateHolder.terminatedWith( cause );
     }
 
     public Connection connection()
@@ -173,34 +227,34 @@ public Connection connection()
 
     private void ensureCanRunQueries()
     {
-        if ( state == State.COMMITTED )
+        if ( state.value == State.COMMITTED )
         {
             throw new ClientException( "Cannot run more queries in this transaction, it has been committed" );
         }
-        else if ( state == State.ROLLED_BACK )
+        else if ( state.value == State.ROLLED_BACK )
         {
             throw new ClientException( "Cannot run more queries in this transaction, it has been rolled back" );
         }
-        else if ( state == State.TERMINATED )
+        else if ( state.value == State.TERMINATED )
         {
             throw new ClientException( "Cannot run more queries in this transaction, " +
-                    "it has either experienced an fatal error or was explicitly terminated" );
+                    "it has either experienced an fatal error or was explicitly terminated", state.causeOfTermination );
         }
     }
 
     private CompletionStage<Void> doCommitAsync()
     {
-        if ( state == State.TERMINATED )
+        if ( state.value == State.TERMINATED )
         {
             return failedFuture( new ClientException( "Transaction can't be committed. " +
-                                                      "It has been rolled back either because of an error or explicit termination" ) );
+                                                      "It has been rolled back either because of an error or explicit termination", state.causeOfTermination ) );
         }
         return protocol.commitTransaction( connection ).thenAccept( bookmarkHolder::setBookmark );
     }
 
     private CompletionStage<Void> doRollbackAsync()
     {
-        if ( state == State.TERMINATED )
+        if ( state.value == State.TERMINATED )
         {
             return completedWithNull();
         }
@@ -224,11 +278,11 @@ private void transactionClosed( boolean isCommitted )
     {
         if ( isCommitted )
         {
-            state = State.COMMITTED;
+            state = StateHolder.of( State.COMMITTED );
         }
         else
         {
-            state = State.ROLLED_BACK;
+            state = StateHolder.of( State.ROLLED_BACK );
         }
         connection.release(); // release in background
     }
 
@@ -45,6 +45,6 @@ public void afterFailure( Throwable error )
         // always mark transaction as terminated because every error is "acknowledged" with a RESET message
         // so database forgets about the transaction after the first error
         // such transaction should not attempt to commit and can be considered as rolled back
-        tx.markTerminated();
+        tx.markTerminated( error );
     }
 }
@@ -56,7 +56,7 @@ public RxResult run(Query query)
                     // The logic here shall be the same as `TransactionPullResponseHandler#afterFailure` as that is where cursor handling failure
                     // This is optional as tx still holds a reference to all cursor futures and they will be clean up properly in commit
                     Throwable error = Futures.completionExceptionCause( completionError );
-                    tx.markTerminated();
+                    tx.markTerminated( error );
                     cursorFuture.completeExceptionally( error );
                 }
             } );
 
@@ -39,6 +39,7 @@
 
 import org.neo4j.driver.Logger;
 import org.neo4j.driver.Logging;
+import org.neo4j.driver.exceptions.ClientException;
 import org.neo4j.driver.exceptions.ServiceUnavailableException;
 import org.neo4j.driver.exceptions.SessionExpiredException;
 import org.neo4j.driver.exceptions.TransientException;
@@ -100,8 +101,9 @@ public <T> T retry( Supplier<T> work )
             {
                 return work.get();
             }
-            catch ( Throwable error )
+            catch ( Throwable throwable )
             {
+                Throwable error = extractPossibleTerminationCause( throwable );
                 if ( canRetryOn( error ) )
                 {
                     long currentTime = clock.millis();
@@ -122,8 +124,10 @@ public <T> T retry( Supplier<T> work )
                         continue;
                     }
                 }
-                addSuppressed( error, errors );
-                throw error;
+
+                // Add the original error in case we didn't continue the loop from within the if above.
+                addSuppressed( throwable, errors );
+                throw throwable;
             }
         }
     }
@@ -144,54 +148,67 @@ public <T> Publisher<T> retryRx( Publisher<T> work )
 
     protected boolean canRetryOn( Throwable error )
     {
-        return error instanceof SessionExpiredException ||
-               error instanceof ServiceUnavailableException ||
-               isTransientError( error );
+        return error instanceof SessionExpiredException || error instanceof ServiceUnavailableException || isTransientError( error );
+    }
+
+    /**
+     * Extracts the possible cause of a transaction that has been marked terminated.
+     *
+     * @param error
+     * @return
+     */
+    private static Throwable extractPossibleTerminationCause( Throwable error )
+    {
+
+        // Having a dedicated "TerminatedException" inheriting from ClientException might be a good idea.
+        if ( error instanceof ClientException && error.getCause() != null )
+        {
+            return error.getCause();
+        }
+        return error;
     }
 
     private Function<Flux<Throwable>,Publisher<Context>> retryRxCondition()
     {
-        return errorCurrentAttempt -> errorCurrentAttempt.flatMap( e -> Mono.subscriberContext().map( ctx -> Tuples.of( e, ctx ) ) ).flatMap( t2 -> {
-            Throwable lastError = t2.getT1();
+        return errorCurrentAttempt -> errorCurrentAttempt.flatMap( e -> Mono.subscriberContext().map( ctx -> Tuples.of( e, ctx ) ) ).flatMap( t2 ->
+        {
+
+            Throwable throwable = t2.getT1();
+            Throwable error = extractPossibleTerminationCause( throwable );
+
             Context ctx = t2.getT2();
 
             List<Throwable> errors = ctx.getOrDefault( "errors", null );
 
-            long startTime  = ctx.getOrDefault( "startTime", -1L );
+            long startTime = ctx.getOrDefault( "startTime", -1L );
             long nextDelayMs = ctx.getOrDefault( "nextDelayMs", initialRetryDelayMs );
 
-            if( !canRetryOn( lastError ) )
+            if ( canRetryOn( error ) )
             {
-                addSuppressed( lastError, errors );
-                return Mono.error( lastError );
-            }
-
-            long currentTime = clock.millis();
-            if ( startTime == -1 )
-            {
-                startTime = currentTime;
-            }
+                long currentTime = clock.millis();
+                if ( startTime == -1 )
+                {
+                    startTime = currentTime;
+                }
 
-            long elapsedTime = currentTime - startTime;
-            if ( elapsedTime < maxRetryTimeMs )
-            {
-                long delayWithJitterMs = computeDelayWithJitter( nextDelayMs );
-                log.warn( "Reactive transaction failed and is scheduled to retry in " + delayWithJitterMs + "ms", lastError );
+                long elapsedTime = currentTime - startTime;
+                if ( elapsedTime < maxRetryTimeMs )
+                {
+                    long delayWithJitterMs = computeDelayWithJitter( nextDelayMs );
+                    log.warn( "Reactive transaction failed and is scheduled to retry in " + delayWithJitterMs + "ms", error );
 
-                nextDelayMs = (long) (nextDelayMs * multiplier);
-                errors = recordError( lastError, errors );
+                    nextDelayMs = (long) (nextDelayMs * multiplier);
+                    errors = recordError( error, errors );
 
-                // retry on netty event loop thread
-                EventExecutor eventExecutor = eventExecutorGroup.next();
-                return Mono.just( ctx.put( "errors", errors ).put( "startTime", startTime ).put( "nextDelayMs", nextDelayMs ) )
-                        .delayElement( Duration.ofMillis( delayWithJitterMs ), Schedulers.fromExecutorService( eventExecutor ) );
+                    // retry on netty event loop thread
+                    EventExecutor eventExecutor = eventExecutorGroup.next();
+                    return Mono.just( ctx.put( "errors", errors ).put( "startTime", startTime ).put( "nextDelayMs", nextDelayMs ) ).delayElement(
+                            Duration.ofMillis( delayWithJitterMs ), Schedulers.fromExecutorService( eventExecutor ) );
+                }
             }
-            else
-            {
-                addSuppressed( lastError, errors );
+            addSuppressed( throwable, errors );
 
-                return Mono.error( lastError );
-            }
+            return Mono.error( throwable );
         } );
     }
 
@@ -249,9 +266,10 @@ private <T> void executeWork( CompletableFuture<T> resultFuture, Supplier<Comple
         } );
     }
 
-    private <T> void retryOnError( CompletableFuture<T> resultFuture, Supplier<CompletionStage<T>> work,
-            long startTime, long retryDelayMs, Throwable error, List<Throwable> errors )
+    private <T> void retryOnError( CompletableFuture<T> resultFuture, Supplier<CompletionStage<T>> work, long startTime, long retryDelayMs, Throwable throwable,
+            List<Throwable> errors )
     {
+        Throwable error = extractPossibleTerminationCause( throwable );
         if ( canRetryOn( error ) )
         {
             long currentTime = clock.millis();
@@ -269,8 +287,8 @@ private <T> void retryOnError( CompletableFuture<T> resultFuture, Supplier<Compl
             }
         }
 
-        addSuppressed( error, errors );
-        resultFuture.completeExceptionally( error );
+        addSuppressed( throwable, errors );
+        resultFuture.completeExceptionally( throwable );
     }
 
     private long computeDelayWithJitter( long delayMs )
Original file line number	Diff line number	Diff line change
`@@ -50,11 +50,6 @@ public CompletionStage<ResultCursor> runAsync(Query query)`
`50`	`50`	`return tx.runAsync(query, true );`
`51`	`51`	`}`
`52`	`52`
`53`		`- public void markTerminated()`
`54`		`- {`
`55`		`- tx.markTerminated();`
`56`		`- }`
`57`		`-`
`58`	`53`	`public boolean isOpen()`
`59`	`54`	`{`
`60`	`55`	`return tx.isOpen();`
Original file line number	Diff line number	Diff line change
`@@ -138,7 +138,7 @@ public CompletionStage<Void> resetAsync()`
`138`	`138`	`{`
`139`	`139`	`if ( tx != null )`
`140`	`140`	`{`
`141`		`- tx.markTerminated();`
	`141`	`+ tx.markTerminated( null );`
`142`	`142`	`}`
`143`	`143`	`} )`
`144`	`144`	`.thenCompose( ignore -> connectionStage )`
Original file line number	Diff line number	Diff line change
`@@ -18,6 +18,7 @@`
`18`	`18`	`*/`
`19`	`19`	`package org.neo4j.driver.internal.async;`
`20`	`20`
	`21`	`+import java.util.EnumSet;`
`21`	`22`	`import java.util.concurrent.CompletionException;`
`22`	`23`	`import java.util.concurrent.CompletionStage;`
`23`	`24`	`import java.util.function.BiFunction;`
`@@ -58,13 +59,66 @@ private enum State`
`58`	`59`	`ROLLED_BACK`
`59`	`60`	`}`
`60`	`61`
	`62`	`+ /**`
	`63`	`+ * This is a holder so that we can have ony the state volatile in the tx without having to synchronize the whole block.`
	`64`	`+ */`
	`65`	`+ private static final class StateHolder`
	`66`	`+ {`
	`67`	`+ private static final EnumSet<State> OPEN_STATES = EnumSet.of( State.ACTIVE, State.TERMINATED );`
	`68`	`+ private static final StateHolder ACTIVE_HOLDER = new StateHolder( State.ACTIVE, null );`
	`69`	`+ private static final StateHolder COMMITTED_HOLDER = new StateHolder( State.COMMITTED, null );`
	`70`	`+ private static final StateHolder ROLLED_BACK_HOLDER = new StateHolder( State.ROLLED_BACK, null );`
	`71`	`+`
	`72`	`+ /**`
	`73`	`+ * The actual state.`
	`74`	`+ */`
	`75`	`+ final State value;`
	`76`	`+`
	`77`	`+ /**`
	`78`	`+ * If this holder contains a state of {@link State#TERMINATED}, this represents the cause if any.`
	`79`	`+ */`
	`80`	`+ final Throwable causeOfTermination;`
	`81`	`+`
	`82`	`+ static StateHolder of( State value )`
	`83`	`+ {`
	`84`	`+ switch ( value )`
	`85`	`+ {`
	`86`	`+ case ACTIVE:`
	`87`	`+ return ACTIVE_HOLDER;`
	`88`	`+ case COMMITTED:`
	`89`	`+ return COMMITTED_HOLDER;`
	`90`	`+ case ROLLED_BACK:`
	`91`	`+ return ROLLED_BACK_HOLDER;`
	`92`	`+ case TERMINATED:`
	`93`	`+ default:`
	`94`	`+ throw new IllegalArgumentException( "Cannot provide a default state holder for state " + value );`
	`95`	`+ }`
	`96`	`+ }`
	`97`	`+`
	`98`	`+ static StateHolder terminatedWith( Throwable cause )`
	`99`	`+ {`
	`100`	`+ return new StateHolder( State.TERMINATED, cause );`
	`101`	`+ }`
	`102`	`+`
	`103`	`+ private StateHolder( State value, Throwable causeOfTermination )`
	`104`	`+ {`
	`105`	`+ this.value = value;`
	`106`	`+ this.causeOfTermination = causeOfTermination;`
	`107`	`+ }`
	`108`	`+`
	`109`	`+ boolean isOpen()`
	`110`	`+ {`
	`111`	`+ return OPEN_STATES.contains( this.value );`
	`112`	`+ }`
	`113`	`+ }`
	`114`	`+`
`61`	`115`	`private final Connection connection;`
`62`	`116`	`private final BoltProtocol protocol;`
`63`	`117`	`private final BookmarkHolder bookmarkHolder;`
`64`	`118`	`private final ResultCursorsHolder resultCursors;`
`65`	`119`	`private final long fetchSize;`
`66`	`120`
`67`		`- private volatile State state = State.ACTIVE;`
	`121`	`+ private volatile StateHolder state = StateHolder.of( State.ACTIVE );`
`68`	`122`
`69`	`123`	`public UnmanagedTransaction(Connection connection, BookmarkHolder bookmarkHolder, long fetchSize )`
`70`	`124`	`{`
`@@ -104,11 +158,11 @@ public CompletionStage<Void> closeAsync()`
`104`	`158`
`105`	`159`	`public CompletionStage<Void> commitAsync()`
`106`	`160`	`{`
`107`		`- if ( state == State.COMMITTED )`
	`161`	`+ if ( state.value == State.COMMITTED )`
`108`	`162`	`{`
`109`	`163`	`return failedFuture( new ClientException( "Can't commit, transaction has been committed" ) );`
`110`	`164`	`}`
`111`		`- else if ( state == State.ROLLED_BACK )`
	`165`	`+ else if ( state.value == State.ROLLED_BACK )`
`112`	`166`	`{`
`113`	`167`	`return failedFuture( new ClientException( "Can't commit, transaction has been rolled back" ) );`
`114`	`168`	`}`
`@@ -122,11 +176,11 @@ else if ( state == State.ROLLED_BACK )`
`122`	`176`
`123`	`177`	`public CompletionStage<Void> rollbackAsync()`
`124`	`178`	`{`
`125`		`- if ( state == State.COMMITTED )`
	`179`	`+ if ( state.value == State.COMMITTED )`
`126`	`180`	`{`
`127`	`181`	`return failedFuture( new ClientException( "Can't rollback, transaction has been committed" ) );`
`128`	`182`	`}`
`129`		`- else if ( state == State.ROLLED_BACK )`
	`183`	`+ else if ( state.value == State.ROLLED_BACK )`
`130`	`184`	`{`
`131`	`185`	`return failedFuture( new ClientException( "Can't rollback, transaction has been rolled back" ) );`
`132`	`186`	`}`
`@@ -158,12 +212,12 @@ public CompletionStage<RxResultCursor> runRx(Query query)`
`158`	`212`
`159`	`213`	`public boolean isOpen()`
`160`	`214`	`{`
`161`		`- return state != State.COMMITTED && state != State.ROLLED_BACK;`
	`215`	`+ return state.isOpen();`
`162`	`216`	`}`
`163`	`217`
`164`		`- public void markTerminated()`
	`218`	`+ public void markTerminated( Throwable cause )`
`165`	`219`	`{`
`166`		`- state = State.TERMINATED;`
	`220`	`+ state = StateHolder.terminatedWith( cause );`
`167`	`221`	`}`
`168`	`222`
`169`	`223`	`public Connection connection()`
`@@ -173,34 +227,34 @@ public Connection connection()`
`173`	`227`
`174`	`228`	`private void ensureCanRunQueries()`
`175`	`229`	`{`
`176`		`- if ( state == State.COMMITTED )`
	`230`	`+ if ( state.value == State.COMMITTED )`
`177`	`231`	`{`
`178`	`232`	`throw new ClientException( "Cannot run more queries in this transaction, it has been committed" );`
`179`	`233`	`}`
`180`		`- else if ( state == State.ROLLED_BACK )`
	`234`	`+ else if ( state.value == State.ROLLED_BACK )`
`181`	`235`	`{`
`182`	`236`	`throw new ClientException( "Cannot run more queries in this transaction, it has been rolled back" );`
`183`	`237`	`}`
`184`		`- else if ( state == State.TERMINATED )`
	`238`	`+ else if ( state.value == State.TERMINATED )`
`185`	`239`	`{`
`186`	`240`	`throw new ClientException( "Cannot run more queries in this transaction, " +`
`187`		`- "it has either experienced an fatal error or was explicitly terminated" );`
	`241`	`+ "it has either experienced an fatal error or was explicitly terminated", state.causeOfTermination );`
`188`	`242`	`}`
`189`	`243`	`}`
`190`	`244`
`191`	`245`	`private CompletionStage<Void> doCommitAsync()`
`192`	`246`	`{`
`193`		`- if ( state == State.TERMINATED )`
	`247`	`+ if ( state.value == State.TERMINATED )`
`194`	`248`	`{`
`195`	`249`	`return failedFuture( new ClientException( "Transaction can't be committed. " +`
`196`		`- "It has been rolled back either because of an error or explicit termination" ) );`
	`250`	`+ "It has been rolled back either because of an error or explicit termination", state.causeOfTermination ) );`
`197`	`251`	`}`
`198`	`252`	`return protocol.commitTransaction( connection ).thenAccept( bookmarkHolder::setBookmark );`
`199`	`253`	`}`
`200`	`254`
`201`	`255`	`private CompletionStage<Void> doRollbackAsync()`
`202`	`256`	`{`
`203`		`- if ( state == State.TERMINATED )`
	`257`	`+ if ( state.value == State.TERMINATED )`
`204`	`258`	`{`
`205`	`259`	`return completedWithNull();`
`206`	`260`	`}`
`@@ -224,11 +278,11 @@ private void transactionClosed( boolean isCommitted )`
`224`	`278`	`{`
`225`	`279`	`if ( isCommitted )`
`226`	`280`	`{`
`227`		`- state = State.COMMITTED;`
	`281`	`+ state = StateHolder.of( State.COMMITTED );`
`228`	`282`	`}`
`229`	`283`	`else`
`230`	`284`	`{`
`231`		`- state = State.ROLLED_BACK;`
	`285`	`+ state = StateHolder.of( State.ROLLED_BACK );`
`232`	`286`	`}`
`233`	`287`	`connection.release(); // release in background`
`234`	`288`	`}`
Original file line number	Diff line number	Diff line change
`@@ -45,6 +45,6 @@ public void afterFailure( Throwable error )`
`45`	`45`	`// always mark transaction as terminated because every error is "acknowledged" with a RESET message`
`46`	`46`	`// so database forgets about the transaction after the first error`
`47`	`47`	`// such transaction should not attempt to commit and can be considered as rolled back`
`48`		`- tx.markTerminated();`
	`48`	`+ tx.markTerminated( error );`
`49`	`49`	`}`
`50`	`50`	`}`
Original file line number	Diff line number	Diff line change
`@@ -56,7 +56,7 @@ public RxResult run(Query query)`
`56`	`56`	// The logic here shall be the same as `TransactionPullResponseHandler#afterFailure` as that is where cursor handling failure
`57`	`57`	`// This is optional as tx still holds a reference to all cursor futures and they will be clean up properly in commit`
`58`	`58`	`Throwable error = Futures.completionExceptionCause( completionError );`
`59`		`- tx.markTerminated();`
	`59`	`+ tx.markTerminated( error );`
`60`	`60`	`cursorFuture.completeExceptionally( error );`
`61`	`61`	`}`
`62`	`62`	`} );`