Run response immediate processing update (#897)

* Make run methods wait for successful request acknowledgement

This update makes `run` methods (session, tx) wait for successful request acknowledgement by the server. Specifically, the `Result` object will only be returned when there is a successful response from the server to the `RUN` message, meaning that the request was at least accepted by the server for processing. Otherwise, an appropriate error will be thrown.

Additionally, this update makes the `Result.keys` method non-blocking and free from potential communication errors.

* Update handling of failed queries

* Move common handling to async cursor

* Improve DisposableAsyncResultCursor tests

* Removed duplicate test

* Removed FINE logging

Author: injectives

driver/src/main/java/org/neo4j/driver/internal/FailableCursor.java

@@ -23,12 +23,12 @@
 public interface FailableCursor
 {
     /**
-     * Discarding all unconsumed records and returning failure if there is any to run and/or pulls.
+     * Discarding all unconsumed records and returning failure if there is any pull errors.
      */
     CompletionStage<Throwable> discardAllFailureAsync();
 
     /**
-     * Pulling all unconsumed records into memory and returning failure if there is any to run and/or pulls.
+     * Pulling all unconsumed records into memory and returning failure if there is any pull errors.
      */
     CompletionStage<Throwable> pullAllFailureAsync();
 }

driver/src/main/java/org/neo4j/driver/internal/InternalResult.java

@@ -39,7 +39,6 @@ public class InternalResult implements Result
 {
     private final Connection connection;
     private final ResultCursor cursor;
-    private List<String> keys;
 
     public InternalResult(Connection connection, ResultCursor cursor )
     {
@@ -50,12 +49,7 @@ public InternalResult(Connection connection, ResultCursor cursor )
     @Override
     public List<String> keys()
     {
-        if ( keys == null )
-        {
-            blockingGet( cursor.peekAsync() );
-            keys = cursor.keys();
-        }
-        return keys;
+        return cursor.keys();
     }
 
     @Override

driver/src/main/java/org/neo4j/driver/internal/InternalSession.java

@@ -23,14 +23,14 @@
 import org.neo4j.driver.AccessMode;
 import org.neo4j.driver.Bookmark;
 import org.neo4j.driver.Query;
-import org.neo4j.driver.Session;
 import org.neo4j.driver.Result;
+import org.neo4j.driver.Session;
 import org.neo4j.driver.Transaction;
 import org.neo4j.driver.TransactionConfig;
 import org.neo4j.driver.TransactionWork;
 import org.neo4j.driver.async.ResultCursor;
-import org.neo4j.driver.internal.async.UnmanagedTransaction;
 import org.neo4j.driver.internal.async.NetworkSession;
+import org.neo4j.driver.internal.async.UnmanagedTransaction;
 import org.neo4j.driver.internal.spi.Connection;
 import org.neo4j.driver.internal.util.Futures;
 
@@ -66,8 +66,8 @@ public Result run(String query, Map<String,Object> parameters, TransactionConfig
     @Override
     public Result run(Query query, TransactionConfig config )
     {
-        ResultCursor cursor = Futures.blockingGet( session.runAsync(query, config, false ),
-                () -> terminateConnectionOnThreadInterrupt( "Thread interrupted while running query in session" ) );
+        ResultCursor cursor = Futures.blockingGet( session.runAsync( query, config ),
+                                                   () -> terminateConnectionOnThreadInterrupt( "Thread interrupted while running query in session" ) );
 
         // query executed, it is safe to obtain a connection in a blocking way
         Connection connection = Futures.getNow( session.connectionAsync() );

driver/src/main/java/org/neo4j/driver/internal/InternalTransaction.java

@@ -57,8 +57,8 @@ public void close()
     @Override
     public Result run(Query query)
     {
-        ResultCursor cursor = Futures.blockingGet( tx.runAsync(query, false ),
-                () -> terminateConnectionOnThreadInterrupt( "Thread interrupted while running query in transaction" ) );
+        ResultCursor cursor = Futures.blockingGet( tx.runAsync( query ),
+                                                   () -> terminateConnectionOnThreadInterrupt( "Thread interrupted while running query in transaction" ) );
         return new InternalResult( tx.connection(), cursor );
     }
 

driver/src/main/java/org/neo4j/driver/internal/async/InternalAsyncSession.java

@@ -23,13 +23,13 @@
 import java.util.concurrent.CompletionStage;
 
 import org.neo4j.driver.AccessMode;
+import org.neo4j.driver.Bookmark;
 import org.neo4j.driver.Query;
 import org.neo4j.driver.TransactionConfig;
 import org.neo4j.driver.async.AsyncSession;
 import org.neo4j.driver.async.AsyncTransaction;
 import org.neo4j.driver.async.AsyncTransactionWork;
 import org.neo4j.driver.async.ResultCursor;
-import org.neo4j.driver.Bookmark;
 import org.neo4j.driver.internal.util.Futures;
 
 import static java.util.Collections.emptyMap;
@@ -66,7 +66,7 @@ public CompletionStage<ResultCursor> runAsync(String query, Map<String,Object> p
     @Override
     public CompletionStage<ResultCursor> runAsync(Query query, TransactionConfig config )
     {
-        return session.runAsync(query, config, true );
+        return session.runAsync( query, config );
     }
 
     @Override

driver/src/main/java/org/neo4j/driver/internal/async/InternalAsyncTransaction.java

@@ -47,7 +47,7 @@ public CompletionStage<Void> rollbackAsync()
     @Override
     public CompletionStage<ResultCursor> runAsync(Query query)
     {
-        return tx.runAsync(query, true );
+        return tx.runAsync( query );
     }
 
     public boolean isOpen()

driver/src/main/java/org/neo4j/driver/internal/async/NetworkSession.java

@@ -35,8 +35,8 @@
 import org.neo4j.driver.internal.DatabaseName;
 import org.neo4j.driver.internal.FailableCursor;
 import org.neo4j.driver.internal.cursor.AsyncResultCursor;
-import org.neo4j.driver.internal.cursor.RxResultCursor;
 import org.neo4j.driver.internal.cursor.ResultCursorFactory;
+import org.neo4j.driver.internal.cursor.RxResultCursor;
 import org.neo4j.driver.internal.logging.PrefixedLogger;
 import org.neo4j.driver.internal.retry.RetryLogic;
 import org.neo4j.driver.internal.spi.Connection;
@@ -76,19 +76,19 @@ public NetworkSession( ConnectionProvider connectionProvider, RetryLogic retryLo
         this.fetchSize = fetchSize;
     }
 
-    public CompletionStage<ResultCursor> runAsync(Query query, TransactionConfig config, boolean waitForRunResponse )
+    public CompletionStage<ResultCursor> runAsync( Query query, TransactionConfig config )
     {
         CompletionStage<AsyncResultCursor> newResultCursorStage =
-                buildResultCursorFactory(query, config, waitForRunResponse ).thenCompose( ResultCursorFactory::asyncResult );
+                buildResultCursorFactory( query, config ).thenCompose( ResultCursorFactory::asyncResult );
 
         resultCursorStage = newResultCursorStage.exceptionally( error -> null );
-        return newResultCursorStage.thenApply( cursor -> cursor ); // convert the return type
+        return newResultCursorStage.thenCompose( AsyncResultCursor::mapSuccessfulRunCompletionAsync ).thenApply( cursor -> cursor ); // convert the return type
     }
 
     public CompletionStage<RxResultCursor> runRx(Query query, TransactionConfig config )
     {
         CompletionStage<RxResultCursor> newResultCursorStage =
-                buildResultCursorFactory(query, config, true ).thenCompose( ResultCursorFactory::rxResult );
+                buildResultCursorFactory( query, config ).thenCompose( ResultCursorFactory::rxResult );
 
         resultCursorStage = newResultCursorStage.exceptionally( error -> null );
         return newResultCursorStage;
@@ -223,24 +223,27 @@ protected CompletionStage<Boolean> currentConnectionIsOpen()
                 connection.isOpen() ); // and it's still open
     }
 
-    private CompletionStage<ResultCursorFactory> buildResultCursorFactory(Query query, TransactionConfig config, boolean waitForRunResponse )
+    private CompletionStage<ResultCursorFactory> buildResultCursorFactory( Query query, TransactionConfig config )
     {
         ensureSessionIsOpen();
 
         return ensureNoOpenTxBeforeRunningQuery()
                 .thenCompose( ignore -> acquireConnection( mode ) )
-                .thenCompose( connection -> {
-                    try
-                    {
-                        ResultCursorFactory factory = connection.protocol()
-                                .runInAutoCommitTransaction( connection, query, bookmarkHolder, config, waitForRunResponse, fetchSize );
-                        return completedFuture( factory );
-                    }
-                    catch ( Throwable e )
-                    {
-                        return Futures.failedFuture( e );
-                    }
-                } );
+                .thenCompose(
+                        connection ->
+                        {
+                            try
+                            {
+                                ResultCursorFactory factory = connection
+                                        .protocol()
+                                        .runInAutoCommitTransaction( connection, query, bookmarkHolder, config, fetchSize );
+                                return completedFuture( factory );
+                            }
+                            catch ( Throwable e )
+                            {
+                                return Futures.failedFuture( e );
+                            }
+                        } );
     }
 
     private CompletionStage<Connection> acquireConnection( AccessMode mode )

driver/src/main/java/org/neo4j/driver/internal/async/UnmanagedTransaction.java

@@ -18,6 +18,7 @@
  */
 package org.neo4j.driver.internal.async;
 
+import java.util.Arrays;
 import java.util.EnumSet;
 import java.util.concurrent.CompletionException;
 import java.util.concurrent.CompletionStage;
@@ -204,20 +205,20 @@ else if ( state.value == State.ROLLED_BACK )
         }
     }
 
-    public CompletionStage<ResultCursor> runAsync(Query query, boolean waitForRunResponse )
+    public CompletionStage<ResultCursor> runAsync( Query query )
     {
         ensureCanRunQueries();
         CompletionStage<AsyncResultCursor> cursorStage =
-                protocol.runInUnmanagedTransaction( connection, query, this, waitForRunResponse, fetchSize ).asyncResult();
+                protocol.runInUnmanagedTransaction( connection, query, this, fetchSize ).asyncResult();
         resultCursors.add( cursorStage );
-        return cursorStage.thenApply( cursor -> cursor );
+        return cursorStage.thenCompose( AsyncResultCursor::mapSuccessfulRunCompletionAsync ).thenApply( cursor -> cursor );
     }
 
     public CompletionStage<RxResultCursor> runRx(Query query)
     {
         ensureCanRunQueries();
         CompletionStage<RxResultCursor> cursorStage =
-                protocol.runInUnmanagedTransaction( connection, query, this, false, fetchSize ).rxResult();
+                protocol.runInUnmanagedTransaction( connection, query, this, fetchSize ).rxResult();
         resultCursors.add( cursorStage );
         return cursorStage;
     }
@@ -229,7 +230,29 @@ public boolean isOpen()
 
     public void markTerminated( Throwable cause )
     {
-        state = StateHolder.terminatedWith( cause );
+        if ( state.value == State.TERMINATED )
+        {
+            if ( state.causeOfTermination != null )
+            {
+                addSuppressedWhenNotCaptured( state.causeOfTermination, cause );
+            }
+        }
+        else
+        {
+            state = StateHolder.terminatedWith( cause );
+        }
+    }
+
+    private void addSuppressedWhenNotCaptured( Throwable currentCause, Throwable newCause )
+    {
+        if ( currentCause != newCause )
+        {
+            boolean noneMatch = Arrays.stream( currentCause.getSuppressed() ).noneMatch( suppressed -> suppressed == newCause );
+            if ( noneMatch )
+            {
+                currentCause.addSuppressed( newCause );
+            }
+        }
     }
 
     public Connection connection()

driver/src/main/java/org/neo4j/driver/internal/cluster/SingleDatabaseRoutingProcedureRunner.java

@@ -92,8 +92,8 @@ BookmarkHolder bookmarkHolder( Bookmark ignored )
     CompletionStage<List<Record>> runProcedure(Connection connection, Query procedure, BookmarkHolder bookmarkHolder )
     {
         return connection.protocol()
-                .runInAutoCommitTransaction( connection, procedure, bookmarkHolder, TransactionConfig.empty(), true, UNLIMITED_FETCH_SIZE )
-                .asyncResult().thenCompose( ResultCursor::listAsync );
+                         .runInAutoCommitTransaction( connection, procedure, bookmarkHolder, TransactionConfig.empty(), UNLIMITED_FETCH_SIZE )
+                         .asyncResult().thenCompose( ResultCursor::listAsync );
     }
 
     private CompletionStage<List<Record>> releaseConnection( Connection connection, List<Record> records )

driver/src/main/java/org/neo4j/driver/internal/cursor/AsyncResultCursor.java

@@ -18,9 +18,12 @@
  */
 package org.neo4j.driver.internal.cursor;
 
-import org.neo4j.driver.internal.FailableCursor;
+import java.util.concurrent.CompletableFuture;
+
 import org.neo4j.driver.async.ResultCursor;
+import org.neo4j.driver.internal.FailableCursor;
 
 public interface AsyncResultCursor extends ResultCursor, FailableCursor
 {
+    CompletableFuture<AsyncResultCursor> mapSuccessfulRunCompletionAsync();
 }

driver/src/main/java/org/neo4j/driver/internal/cursor/AsyncResultCursorImpl.java

@@ -33,11 +33,13 @@
 
 public class AsyncResultCursorImpl implements AsyncResultCursor
 {
+    private final Throwable runError;
     private final RunResponseHandler runHandler;
     private final PullAllResponseHandler pullAllHandler;
 
-    public AsyncResultCursorImpl(RunResponseHandler runHandler, PullAllResponseHandler pullAllHandler )
+    public AsyncResultCursorImpl( Throwable runError, RunResponseHandler runHandler, PullAllResponseHandler pullAllHandler )
     {
+        this.runError = runError;
         this.runHandler = runHandler;
         this.pullAllHandler = pullAllHandler;
     }
@@ -113,13 +115,15 @@ public <T> CompletionStage<List<T>> listAsync( Function<Record,T> mapFunction )
     @Override
     public CompletionStage<Throwable> discardAllFailureAsync()
     {
-        return consumeAsync().handle( ( summary, error ) -> error );
+        // runError has priority over other errors and is expected to have been reported to user by now
+        return consumeAsync().handle( ( summary, error ) -> runError != null ? null : error );
     }
 
     @Override
     public CompletionStage<Throwable> pullAllFailureAsync()
     {
-        return pullAllHandler.pullAllFailureAsync();
+        // runError has priority over other errors and is expected to have been reported to user by now
+        return pullAllHandler.pullAllFailureAsync().thenApply( error -> runError != null ? null : error );
     }
 
     private void internalForEachAsync( Consumer<Record> action, CompletableFuture<Void> resultFuture )
@@ -154,4 +158,10 @@ else if ( record != null )
             }
         } );
     }
+
+    @Override
+    public CompletableFuture<AsyncResultCursor> mapSuccessfulRunCompletionAsync()
+    {
+        return runError != null ? Futures.failedFuture( runError ) : CompletableFuture.completedFuture( this );
+    }
 }

driver/src/main/java/org/neo4j/driver/internal/cursor/AsyncResultCursorOnlyFactory.java

@@ -18,6 +18,7 @@
  */
 package org.neo4j.driver.internal.cursor;
 
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
 
 import org.neo4j.driver.exceptions.ClientException;
@@ -28,7 +29,6 @@
 import org.neo4j.driver.internal.util.Futures;
 
 import static java.util.Objects.requireNonNull;
-import static java.util.concurrent.CompletableFuture.completedFuture;
 
 /**
  * Used by Bolt V1, V2, V3
@@ -38,23 +38,24 @@ public class AsyncResultCursorOnlyFactory implements ResultCursorFactory
     protected final Connection connection;
     protected final Message runMessage;
     protected final RunResponseHandler runHandler;
+    private final CompletableFuture<Void> runFuture;
     protected final PullAllResponseHandler pullAllHandler;
-    private final boolean waitForRunResponse;
 
-    public AsyncResultCursorOnlyFactory(Connection connection, Message runMessage, RunResponseHandler runHandler,
-                                        PullAllResponseHandler pullHandler, boolean waitForRunResponse )
+    public AsyncResultCursorOnlyFactory( Connection connection, Message runMessage, RunResponseHandler runHandler, CompletableFuture<Void> runFuture,
+                                         PullAllResponseHandler pullHandler )
     {
         requireNonNull( connection );
         requireNonNull( runMessage );
         requireNonNull( runHandler );
+        requireNonNull( runFuture );
         requireNonNull( pullHandler );
 
         this.connection = connection;
         this.runMessage = runMessage;
         this.runHandler = runHandler;
+        this.runFuture = runFuture;
 
         this.pullAllHandler = pullHandler;
-        this.waitForRunResponse = waitForRunResponse;
     }
 
     public CompletionStage<AsyncResultCursor> asyncResult()
@@ -63,16 +64,7 @@ public CompletionStage<AsyncResultCursor> asyncResult()
         connection.write( runMessage, runHandler ); // queues the run message, will be flushed with pull message together
         pullAllHandler.prePopulateRecords();
 
-        if ( waitForRunResponse )
-        {
-            // wait for response of RUN before proceeding
-            return runHandler.runFuture().thenApply( ignore ->
-                    new DisposableAsyncResultCursor( new AsyncResultCursorImpl( runHandler, pullAllHandler ) ) );
-        }
-        else
-        {
-            return completedFuture( new DisposableAsyncResultCursor( new AsyncResultCursorImpl( runHandler, pullAllHandler ) ) );
-        }
+        return runFuture.handle( ( ignored, error ) -> new DisposableAsyncResultCursor( new AsyncResultCursorImpl( error, runHandler, pullAllHandler ) ) );
     }
 
     public CompletionStage<RxResultCursor> rxResult()

driver/src/main/java/org/neo4j/driver/internal/cursor/DisposableAsyncResultCursor.java

@@ -36,7 +36,7 @@ public class DisposableAsyncResultCursor implements AsyncResultCursor
     private final AsyncResultCursor delegate;
     private boolean isDisposed;
 
-    public DisposableAsyncResultCursor(AsyncResultCursor delegate )
+    public DisposableAsyncResultCursor( AsyncResultCursor delegate )
     {
         this.delegate = delegate;
     }
@@ -118,4 +118,10 @@ boolean isDisposed()
     {
         return this.isDisposed;
     }
+
+    @Override
+    public CompletableFuture<AsyncResultCursor> mapSuccessfulRunCompletionAsync()
+    {
+        return this.delegate.mapSuccessfulRunCompletionAsync().thenApply( ignored -> this );
+    }
 }