neo4j
diff --git a/‎driver/src/main/java/org/neo4j/driver/internal/ExplicitTransaction.java
Copy file name to clipboard
+6-7Lines changed: 6 additions & 7 deletions b/‎driver/src/main/java/org/neo4j/driver/internal/ExplicitTransaction.java
Copy file name to clipboard
+6-7Lines changed: 6 additions & 7 deletions
diff --git a/‎driver/src/main/java/org/neo4j/driver/internal/async/NettyConnection.java
Copy file name to clipboard
-1Lines changed: 0 additions & 1 deletion b/‎driver/src/main/java/org/neo4j/driver/internal/async/NettyConnection.java
Copy file name to clipboard
-1Lines changed: 0 additions & 1 deletion
diff --git a/‎driver/src/main/java/org/neo4j/driver/internal/async/ResultCursorsHolder.java
Copy file name to clipboard
+27-15Lines changed: 27 additions & 15 deletions b/‎driver/src/main/java/org/neo4j/driver/internal/async/ResultCursorsHolder.java
Copy file name to clipboard
+27-15Lines changed: 27 additions & 15 deletions
diff --git a/‎driver/src/main/java/org/neo4j/driver/internal/async/inbound/InboundMessageDispatcher.java
Copy file name to clipboard
+6-54Lines changed: 6 additions & 54 deletions b/‎driver/src/main/java/org/neo4j/driver/internal/async/inbound/InboundMessageDispatcher.java
Copy file name to clipboard
+6-54Lines changed: 6 additions & 54 deletions
diff --git a/‎driver/src/main/java/org/neo4j/driver/internal/handlers/AckFailureResponseHandler.java
Copy file name to clipboard
-63Lines changed: 0 additions & 63 deletions b/‎driver/src/main/java/org/neo4j/driver/internal/handlers/AckFailureResponseHandler.java
Copy file name to clipboard
-63Lines changed: 0 additions & 63 deletions
diff --git a/‎driver/src/main/java/org/neo4j/driver/internal/handlers/ResetResponseHandler.java
Copy file name to clipboard
+10-2Lines changed: 10 additions & 2 deletions b/‎driver/src/main/java/org/neo4j/driver/internal/handlers/ResetResponseHandler.java
Copy file name to clipboard
+10-2Lines changed: 10 additions & 2 deletions
diff --git a/‎driver/src/main/java/org/neo4j/driver/internal/handlers/TransactionPullAllResponseHandler.java
Copy file name to clipboard
+4-9Lines changed: 4 additions & 9 deletions b/‎driver/src/main/java/org/neo4j/driver/internal/handlers/TransactionPullAllResponseHandler.java
Copy file name to clipboard
+4-9Lines changed: 4 additions & 9 deletions
diff --git a/‎driver/src/main/java/org/neo4j/driver/internal/messaging/encode/AckFailureMessageEncoder.java
Copy file name to clipboard
-38Lines changed: 0 additions & 38 deletions b/‎driver/src/main/java/org/neo4j/driver/internal/messaging/encode/AckFailureMessageEncoder.java
Copy file name to clipboard
-38Lines changed: 0 additions & 38 deletions
@@ -207,15 +207,13 @@ else if ( state == State.ROLLED_BACK )
         }
         else if ( state == State.MARKED_FAILED )
         {
-            throw new ClientException(
-                    "Cannot run more statements in this transaction, because previous statements in the " +
-                    "transaction has failed and the transaction has been rolled back. Please start a new " +
-                    "transaction to run another statement."
-            );
+            throw new ClientException( "Cannot run more statements in this transaction, it has been marked for failure. " +
+                                       "Please either rollback or close this transaction" );
         }
         else if ( state == State.TERMINATED )
         {
-            throw new ClientException( "Cannot run more statements in this transaction, it has been terminated" );
+            throw new ClientException( "Cannot run more statements in this transaction, " +
+                                       "it has either experienced an fatal error or was explicitly terminated" );
         }
     }
 
@@ -247,7 +245,8 @@ private CompletionStage<Void> doCommitAsync()
     {
         if ( state == State.TERMINATED )
         {
-            return failedFuture( new ClientException( "Can't commit, transaction has been terminated" ) );
+            return failedFuture( new ClientException( "Transaction can't be committed. " +
+                                                      "It has been rolled back either because of an error or explicit termination" ) );
         }
         return protocol.commitTransaction( connection, this );
     }
 
@@ -178,7 +178,6 @@ private void writeResetMessageIfNeeded( ResponseHandler resetHandler, boolean is
             }
             else
             {
-                messageDispatcher.muteAckFailure();
                 // auto-read could've been disabled, re-enable it to automatically receive response for RESET
                 setAutoRead( true );
                 writeAndFlushMessage( ResetMessage.RESET, resetHandler );
 
@@ -21,11 +21,11 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
 
 import org.neo4j.driver.internal.InternalStatementResultCursor;
 
-import static java.util.concurrent.CompletableFuture.completedFuture;
 import static org.neo4j.driver.internal.util.Futures.completedWithNull;
 
 public class ResultCursorsHolder
@@ -40,28 +40,40 @@ public void add( CompletionStage<InternalStatementResultCursor> cursorStage )
 
     public CompletionStage<Throwable> retrieveNotConsumedError()
     {
-        return cursorStages.stream()
-                .map( this::retrieveFailure )
-                .reduce( completedWithNull(), this::nonNullFailureFromEither );
+        CompletableFuture<Throwable>[] failures = retrieveAllFailures();
+
+        return CompletableFuture.allOf( failures )
+                .thenApply( ignore -> findFirstFailure( failures ) );
     }
 
-    private CompletionStage<Throwable> retrieveFailure( CompletionStage<InternalStatementResultCursor> cursorStage )
+    @SuppressWarnings( "unchecked" )
+    private CompletableFuture<Throwable>[] retrieveAllFailures()
     {
-        return cursorStage
-                .exceptionally( cursor -> null )
-                .thenCompose( cursor -> cursor == null ? completedWithNull() : cursor.failureAsync() );
+        return cursorStages.stream()
+                .map( ResultCursorsHolder::retrieveFailure )
+                .map( CompletionStage::toCompletableFuture )
+                .toArray( CompletableFuture[]::new );
     }
 
-    private CompletionStage<Throwable> nonNullFailureFromEither( CompletionStage<Throwable> stage1,
-            CompletionStage<Throwable> stage2 )
+    private static Throwable findFirstFailure( CompletableFuture<Throwable>[] completedFailureFutures )
     {
-        return stage1.thenCompose( value ->
+        // all given futures should be completed, it is thus safe to get their values
+
+        for ( CompletableFuture<Throwable> failureFuture : completedFailureFutures )
         {
-            if ( value != null )
+            Throwable failure = failureFuture.getNow( null ); // does not block
+            if ( failure != null )
             {
-                return completedFuture( value );
+                return failure;
             }
-            return stage2;
-        } );
+        }
+        return null;
+    }
+
+    private static CompletionStage<Throwable> retrieveFailure( CompletionStage<InternalStatementResultCursor> cursorStage )
+    {
+        return cursorStage
+                .exceptionally( cursor -> null )
+                .thenCompose( cursor -> cursor == null ? completedWithNull() : cursor.failureAsync() );
     }
 }
@@ -25,7 +25,7 @@
 import java.util.Map;
 import java.util.Queue;
 
-import org.neo4j.driver.internal.handlers.AckFailureResponseHandler;
+import org.neo4j.driver.internal.handlers.ResetResponseHandler;
 import org.neo4j.driver.internal.logging.ChannelActivityLogger;
 import org.neo4j.driver.internal.messaging.ResponseMessageHandler;
 import org.neo4j.driver.internal.spi.ResponseHandler;
@@ -36,7 +36,7 @@
 import org.neo4j.driver.v1.exceptions.ClientException;
 
 import static java.util.Objects.requireNonNull;
-import static org.neo4j.driver.internal.messaging.request.AckFailureMessage.ACK_FAILURE;
+import static org.neo4j.driver.internal.messaging.request.ResetMessage.RESET;
 
 public class InboundMessageDispatcher implements ResponseMessageHandler
 {
@@ -46,7 +46,6 @@ public class InboundMessageDispatcher implements ResponseMessageHandler
 
     private Throwable currentError;
     private boolean fatalErrorOccurred;
-    private boolean ackFailureMuted;
 
     public InboundMessageDispatcher( Channel channel, Logging logging )
     {
@@ -104,13 +103,14 @@ public void handleFailureMessage( String code, String message )
         if ( ErrorUtil.isFatal( currentError ) )
         {
             // we should not continue using channel after a fatal error
-            // fire error event back to the pipeline and avoid sending ACK_FAILURE
+            // fire error event back to the pipeline and avoid sending RESET
             channel.pipeline().fireExceptionCaught( currentError );
             return;
         }
 
-        // try to write ACK_FAILURE before notifying the next response handler
-        ackFailureIfNeeded();
+        // write a RESET to "acknowledge" the failure
+        queue( new ResetResponseHandler( this ) );
+        channel.writeAndFlush( RESET, channel.voidPromise() );
 
         ResponseHandler handler = handlers.remove();
         handler.onFailure( currentError );
@@ -128,10 +128,6 @@ public void handleIgnoredMessage()
         {
             error = currentError;
         }
-        else if ( ackFailureMuted )
-        {
-            error = new ClientException( "Database ignored the request because session has been reset" );
-        }
         else
         {
             log.warn( "Received IGNORED message for handler %s but error is missing and RESET is not in progress. " +
@@ -169,48 +165,4 @@ public boolean fatalErrorOccurred()
         return fatalErrorOccurred;
     }
 
-    /**
-     * Makes this message dispatcher not send ACK_FAILURE in response to FAILURE until it's un-muted using
-     * {@link #unMuteAckFailure()}. Muting ACK_FAILURE is needed <b>only</b> when sending RESET message. RESET "jumps"
-     * over all queued messages on server and makes them fail. Received failures do not need to be acknowledge because
-     * RESET moves server's state machine to READY state.
-     * <p>
-     * <b>This method is not thread-safe</b> and should only be executed by the event loop thread.
-     */
-    public void muteAckFailure()
-    {
-        ackFailureMuted = true;
-    }
-
-    /**
-     * Makes this message dispatcher send ACK_FAILURE in response to FAILURE. Should be used in combination with
-     * {@link #muteAckFailure()} when sending RESET message.
-     * <p>
-     * <b>This method is not thread-safe</b> and should only be executed by the event loop thread.
-     */
-    public void unMuteAckFailure()
-    {
-        ackFailureMuted = false;
-    }
-
-    /**
-     * Check if ACK_FAILURE is muted.
-     * <p>
-     * <b>This method is not thread-safe</b> and should only be executed by the event loop thread.
-     *
-     * @return {@code true} if ACK_FAILURE has been muted via {@link #muteAckFailure()}, {@code false} otherwise.
-     */
-    public boolean isAckFailureMuted()
-    {
-        return ackFailureMuted;
-    }
-
-    private void ackFailureIfNeeded()
-    {
-        if ( !ackFailureMuted )
-        {
-            queue( new AckFailureResponseHandler( this ) );
-            channel.writeAndFlush( ACK_FAILURE, channel.voidPromise() );
-        }
-    }
 }
@@ -30,6 +30,11 @@ public class ResetResponseHandler implements ResponseHandler
     private final InboundMessageDispatcher messageDispatcher;
     private final CompletableFuture<Void> completionFuture;
 
+    public ResetResponseHandler( InboundMessageDispatcher messageDispatcher )
+    {
+        this( messageDispatcher, null );
+    }
+
     public ResetResponseHandler( InboundMessageDispatcher messageDispatcher, CompletableFuture<Void> completionFuture )
     {
         this.messageDispatcher = messageDispatcher;
@@ -56,8 +61,11 @@ public final void onRecord( Value[] fields )
 
     private void resetCompleted( boolean success )
     {
-        messageDispatcher.unMuteAckFailure();
-        resetCompleted( completionFuture, success );
+        messageDispatcher.clearCurrentError();
+        if ( completionFuture != null )
+        {
+            resetCompleted( completionFuture, success );
+        }
     }
 
     protected void resetCompleted( CompletableFuture<Void> completionFuture, boolean success )
 
@@ -20,7 +20,6 @@
 
 import org.neo4j.driver.internal.ExplicitTransaction;
 import org.neo4j.driver.internal.spi.Connection;
-import org.neo4j.driver.internal.util.ErrorUtil;
 import org.neo4j.driver.v1.Statement;
 
 import static java.util.Objects.requireNonNull;
@@ -44,13 +43,9 @@ protected void afterSuccess()
     @Override
     protected void afterFailure( Throwable error )
     {
-        if ( ErrorUtil.isFatal( error ) )
-        {
-            tx.markTerminated();
-        }
-        else
-        {
-            tx.failure();
-        }
+        // 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();
     }
 }
Original file line number	Diff line number	Diff line change
`@@ -207,15 +207,13 @@ else if ( state == State.ROLLED_BACK )`
`207`	`207`	`}`
`208`	`208`	`else if ( state == State.MARKED_FAILED )`
`209`	`209`	`{`
`210`		`- throw new ClientException(`
`211`		`- "Cannot run more statements in this transaction, because previous statements in the " +`
`212`		`- "transaction has failed and the transaction has been rolled back. Please start a new " +`
`213`		`- "transaction to run another statement."`
`214`		`- );`
	`210`	`+ throw new ClientException( "Cannot run more statements in this transaction, it has been marked for failure. " +`
	`211`	`+ "Please either rollback or close this transaction" );`
`215`	`212`	`}`
`216`	`213`	`else if ( state == State.TERMINATED )`
`217`	`214`	`{`
`218`		`- throw new ClientException( "Cannot run more statements in this transaction, it has been terminated" );`
	`215`	`+ throw new ClientException( "Cannot run more statements in this transaction, " +`
	`216`	`+ "it has either experienced an fatal error or was explicitly terminated" );`
`219`	`217`	`}`
`220`	`218`	`}`
`221`	`219`
`@@ -247,7 +245,8 @@ private CompletionStage<Void> doCommitAsync()`
`247`	`245`	`{`
`248`	`246`	`if ( state == State.TERMINATED )`
`249`	`247`	`{`
`250`		`- return failedFuture( new ClientException( "Can't commit, transaction has been terminated" ) );`
	`248`	`+ return failedFuture( new ClientException( "Transaction can't be committed. " +`
	`249`	`+ "It has been rolled back either because of an error or explicit termination" ) );`
`251`	`250`	`}`
`252`	`251`	`return protocol.commitTransaction( connection, this );`
`253`	`252`	`}`
Original file line number	Diff line number	Diff line change
`@@ -178,7 +178,6 @@ private void writeResetMessageIfNeeded( ResponseHandler resetHandler, boolean is`
`178`	`178`	`}`
`179`	`179`	`else`
`180`	`180`	`{`
`181`		`- messageDispatcher.muteAckFailure();`
`182`	`181`	`// auto-read could've been disabled, re-enable it to automatically receive response for RESET`
`183`	`182`	`setAutoRead( true );`
`184`	`183`	`writeAndFlushMessage( ResetMessage.RESET, resetHandler );`