Mute ACK_FAILURE when sending RESET

It's needed because RESET makes server terminate all running messages
and then moves state machine to idle READY state. That's why it does
not expect clients to ACK_FAILURE for terminated messages. Absence of
this muting caused server to sometimes fail saying: "ACK_FAILURE cannot
be handled by a session in the READY state".

Also removed some unused blocking API classes and better naming for
connection methods.

Author: lutovich

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

@@ -52,7 +52,7 @@ public CompletionStage<Connection> acquireConnection( AccessMode mode )
     @Override
     public CompletionStage<Void> verifyConnectivity()
     {
-        return acquireConnection( READ ).thenCompose( Connection::forceRelease );
+        return acquireConnection( READ ).thenCompose( Connection::releaseNow );
     }
 
     @Override

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

@@ -192,7 +192,7 @@ private BiConsumer<Void,Throwable> transactionClosed( State newState )
         {
             // todo: test that this state transition always happens when commit or rollback
             state = newState;
-            connection.release();
+            connection.releaseInBackground();
             session.setBookmark( bookmark );
         };
     }

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

@@ -492,11 +492,12 @@ private CompletionStage<Void> forceReleaseConnection()
                 {
                     if ( connection != null )
                     {
-                        return connection.forceRelease();
+                        return connection.releaseNow();
                     }
                     return completedFuture( null );
                 } ).exceptionally( error ->
                 {
+                    // todo: this log message looks wrong, should it go to #rollbackTransaction() ?
                     logger.error( "Failed to rollback active transaction", error );
                     return null;
                 } );

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

@@ -25,7 +25,7 @@
 
 import java.util.Map;
 
-import org.neo4j.driver.internal.handlers.AsyncInitResponseHandler;
+import org.neo4j.driver.internal.handlers.InitResponseHandler;
 import org.neo4j.driver.internal.messaging.InitMessage;
 import org.neo4j.driver.v1.Value;
 
@@ -54,7 +54,7 @@ public void operationComplete( ChannelFuture future )
             Channel channel = future.channel();
 
             InitMessage message = new InitMessage( userAgent, authToken );
-            AsyncInitResponseHandler handler = new AsyncInitResponseHandler( connectionInitializedPromise );
+            InitResponseHandler handler = new InitResponseHandler( connectionInitializedPromise );
 
             messageDispatcher( channel ).queue( handler );
             channel.writeAndFlush( message, channel.voidPromise() );

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

@@ -27,6 +27,7 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.neo4j.driver.internal.async.inbound.InboundMessageDispatcher;
+import org.neo4j.driver.internal.handlers.ResetResponseHandler;
 import org.neo4j.driver.internal.messaging.Message;
 import org.neo4j.driver.internal.messaging.PullAllMessage;
 import org.neo4j.driver.internal.messaging.ResetMessage;
@@ -61,6 +62,12 @@ public NettyConnection( Channel channel, ChannelPool channelPool, Clock clock )
         this.clock = clock;
     }
 
+    @Override
+    public boolean isInUse()
+    {
+        return state.isInUse();
+    }
+
     @Override
     public boolean tryMarkInUse()
     {
@@ -99,29 +106,22 @@ public void runAndFlush( String statement, Map<String,Value> parameters, Respons
         run( statement, parameters, runHandler, pullAllHandler, true );
     }
 
-    // TODO change this to return future or say that it does stuff in the background in the method name?
     @Override
-    public void release()
+    public void releaseInBackground()
     {
         if ( state.release() )
         {
-            reset( new ReleaseChannelHandler( channel, channelPool, clock ) );
+            reset( new ResetResponseHandler( channel, channelPool, messageDispatcher, clock ) );
         }
     }
 
     @Override
-    public boolean isInUse()
-    {
-        return state.isInUse();
-    }
-
-    @Override
-    public CompletionStage<Void> forceRelease()
+    public CompletionStage<Void> releaseNow()
     {
         if ( state.forceRelease() )
         {
             Promise<Void> releasePromise = channel.eventLoop().newPromise();
-            reset( new ReleaseChannelHandler( channel, channelPool, clock, releasePromise ) );
+            reset( new ResetResponseHandler( channel, channelPool, messageDispatcher, clock, releasePromise ) );
             return asCompletionStage( releasePromise );
         }
         else
@@ -151,7 +151,11 @@ private void run( String statement, Map<String,Value> parameters, ResponseHandle
 
     private void reset( ResponseHandler resetHandler )
     {
-        writeAndFlushMessageInEventLoop( ResetMessage.RESET, resetHandler );
+        channel.eventLoop().execute( () ->
+        {
+            messageDispatcher.muteAckFailure();
+            writeAndFlushMessage( ResetMessage.RESET, resetHandler );
+        } );
     }
 
     private void writeMessagesInEventLoop( Message message1, ResponseHandler handler1, Message message2,
@@ -160,11 +164,6 @@ private void writeMessagesInEventLoop( Message message1, ResponseHandler handler
         channel.eventLoop().execute( () -> writeMessages( message1, handler1, message2, handler2, flush ) );
     }
 
-    private void writeAndFlushMessageInEventLoop( Message message, ResponseHandler handler )
-    {
-        channel.eventLoop().execute( () -> writeAndFlushMessage( message, handler ) );
-    }
-
     private void writeMessages( Message message1, ResponseHandler handler1, Message message2, ResponseHandler handler2,
             boolean flush )
     {

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

@@ -76,9 +76,9 @@ public void runAndFlush( String statement, Map<String,Value> parameters, Respons
     }
 
     @Override
-    public void release()
+    public void releaseInBackground()
     {
-        delegate.release();
+        delegate.releaseInBackground();
     }
 
     @Override
@@ -88,9 +88,9 @@ public boolean isInUse()
     }
 
     @Override
-    public CompletionStage<Void> forceRelease()
+    public CompletionStage<Void> releaseNow()
     {
-        return delegate.forceRelease();
+        return delegate.releaseNow();
     }
 
     @Override

driver/src/main/java/org/neo4j/driver/internal/async/inbound/InboundMessageDispatcher.java

@@ -26,7 +26,6 @@
 import java.util.Queue;
 
 import org.neo4j.driver.internal.handlers.AckFailureResponseHandler;
-import org.neo4j.driver.internal.messaging.AckFailureMessage;
 import org.neo4j.driver.internal.messaging.MessageHandler;
 import org.neo4j.driver.internal.spi.ResponseHandler;
 import org.neo4j.driver.internal.util.ErrorUtil;
@@ -35,6 +34,7 @@
 import org.neo4j.driver.v1.Value;
 
 import static java.util.Objects.requireNonNull;
+import static org.neo4j.driver.internal.messaging.AckFailureMessage.ACK_FAILURE;
 
 public class InboundMessageDispatcher implements MessageHandler
 {
@@ -44,6 +44,7 @@ public class InboundMessageDispatcher implements MessageHandler
 
     private Throwable currentError;
     private boolean fatalErrorOccurred;
+    private boolean ackFailureMuted;
 
     public InboundMessageDispatcher( Channel channel, Logging logging )
     {
@@ -133,9 +134,8 @@ public void handleFailureMessage( String code, String message )
         log.debug( "Received FAILURE message with code '%s' and message '%s'", code, message );
         currentError = ErrorUtil.newNeo4jError( code, message );
 
-        // queue ACK_FAILURE before notifying the next response handler
-        queue( new AckFailureResponseHandler( this ) );
-        channel.writeAndFlush( AckFailureMessage.ACK_FAILURE, channel.voidPromise() );
+        // try to write ACK_FAILURE before notifying the next response handler
+        ackFailureIfNeeded();
 
         ResponseHandler handler = handlers.remove();
         handler.onFailure( currentError );
@@ -179,4 +179,39 @@ public Throwable currentError()
     {
         return currentError;
     }
+
+    /**
+     * 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.
+     */
+    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.
+     *
+     * @throws IllegalStateException if ACK_FAILURE is not muted right now.
+     */
+    public void unMuteAckFailure()
+    {
+        if ( !ackFailureMuted )
+        {
+            throw new IllegalStateException( "Can't un-mute ACK_FAILURE because it's not muted" );
+        }
+        ackFailureMuted = false;
+    }
+
+    private void ackFailureIfNeeded()
+    {
+        if ( !ackFailureMuted )
+        {
+            queue( new AckFailureResponseHandler( this ) );
+            channel.writeAndFlush( ACK_FAILURE, channel.voidPromise() );
+        }
+    }
 }