Make Session#reset() only send RESET message

Instead of sending RESET message and returning connection back to the
pool. Queries & transactions should still be in control of the
connection lifecycle even after reset. Previous behaviour was
problematic because reset might race with query execution and make
it try to use connection that has been returned to the pool.

Also moved all RESET tests to a dedicated `SessionResetIT`. Made
explicit transaction remain open after termination so that user code
still needs to close it to return connection to the pool.

Author: lutovich

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

@@ -59,32 +59,25 @@ public class ExplicitTransaction implements Transaction
     private enum State
     {
         /** The transaction is running with no explicit success or failure marked */
-        ACTIVE( true ),
+        ACTIVE,
 
         /** Running, user marked for success, meaning it'll value committed */
-        MARKED_SUCCESS( true ),
+        MARKED_SUCCESS,
 
         /** User marked as failed, meaning it'll be rolled back. */
-        MARKED_FAILED( true ),
+        MARKED_FAILED,
 
         /**
          * This transaction has been terminated either because of explicit {@link Session#reset()} or because of a
          * fatal connection error.
          */
-        TERMINATED( false ),
+        TERMINATED,
 
         /** This transaction has successfully committed */
-        COMMITTED( false ),
+        COMMITTED,
 
         /** This transaction has been rolled back */
-        ROLLED_BACK( false );
-
-        final boolean txOpen;
-
-        State( boolean txOpen )
-        {
-            this.txOpen = txOpen;
-        }
+        ROLLED_BACK
     }
 
     private final Connection connection;
@@ -181,6 +174,7 @@ else if ( state == State.ROLLED_BACK )
         }
         else if ( state == State.TERMINATED )
         {
+            transactionClosed( State.ROLLED_BACK );
             return failedFuture( new ClientException( "Can't commit, transaction has been terminated" ) );
         }
         else
@@ -320,7 +314,7 @@ else if ( state == State.TERMINATED )
     @Override
     public boolean isOpen()
     {
-        return state.txOpen;
+        return state != State.COMMITTED && state != State.ROLLED_BACK;
     }
 
     @Override

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

@@ -258,13 +258,24 @@ public void reset()
 
     private CompletionStage<Void> resetAsync()
     {
-        return existingTransactionOrNull().thenAccept( tx ->
-        {
-            if ( tx != null )
-            {
-                tx.markTerminated();
-            }
-        } ).thenCompose( ignore -> releaseConnection() );
+        return existingTransactionOrNull()
+                .thenAccept( tx ->
+                {
+                    if ( tx != null )
+                    {
+                        tx.markTerminated();
+                    }
+                } )
+                .thenCompose( ignore -> connectionStage )
+                .thenCompose( connection ->
+                {
+                    if ( connection != null )
+                    {
+                        // there exists an active connection, send a RESET message over it
+                        return connection.reset();
+                    }
+                    return completedWithNull();
+                } );
     }
 
     @Override

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

@@ -28,6 +28,7 @@
 
 import org.neo4j.driver.internal.BoltServerAddress;
 import org.neo4j.driver.internal.async.inbound.InboundMessageDispatcher;
+import org.neo4j.driver.internal.handlers.ChannelReleasingResetResponseHandler;
 import org.neo4j.driver.internal.handlers.ResetResponseHandler;
 import org.neo4j.driver.internal.messaging.Message;
 import org.neo4j.driver.internal.messaging.PullAllMessage;
@@ -39,6 +40,7 @@
 import org.neo4j.driver.internal.util.ServerVersion;
 import org.neo4j.driver.v1.Value;
 
+import static java.util.Collections.emptyMap;
 import static org.neo4j.driver.internal.async.ChannelAttributes.setTerminationReason;
 
 public class NettyConnection implements Connection
@@ -108,15 +110,24 @@ public void runAndFlush( String statement, Map<String,Value> parameters, Respons
         }
     }
 
+    @Override
+    public CompletionStage<Void> reset()
+    {
+        CompletableFuture<Void> result = new CompletableFuture<>();
+        ResetResponseHandler handler = new ResetResponseHandler( messageDispatcher, result );
+        reset( handler, true );
+        return result;
+    }
+
     @Override
     public CompletionStage<Void> release()
     {
         if ( status.compareAndSet( Status.OPEN, Status.RELEASED ) )
         {
-            // auto-read could've been disabled, re-enable it to automatically receive response for RESET
-            setAutoRead( true );
+            ChannelReleasingResetResponseHandler handler = new ChannelReleasingResetResponseHandler( channel,
+                    channelPool, messageDispatcher, clock, releaseFuture );
 
-            reset( new ResetResponseHandler( channel, channelPool, messageDispatcher, clock, releaseFuture ) );
+            reset( handler, false );
         }
         return releaseFuture;
     }
@@ -152,12 +163,21 @@ private void run( String statement, Map<String,Value> parameters, ResponseHandle
                 pullAllHandler, flush );
     }
 
-    private void reset( ResponseHandler resetHandler )
+    private void reset( ResponseHandler resetHandler, boolean sendNothingIfClosed )
     {
         channel.eventLoop().execute( () ->
         {
-            messageDispatcher.muteAckFailure();
-            writeAndFlushMessage( ResetMessage.RESET, resetHandler );
+            if ( sendNothingIfClosed && !isOpen() )
+            {
+                resetHandler.onSuccess( emptyMap() );
+            }
+            else
+            {
+                messageDispatcher.muteAckFailure();
+                // auto-read could've been disabled, re-enable it to automatically receive response for RESET
+                setAutoRead( true );
+                writeAndFlushMessage( ResetMessage.RESET, resetHandler );
+            }
         } );
     }
 

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

@@ -71,6 +71,12 @@ public void runAndFlush( String statement, Map<String,Value> parameters, Respons
                 newRoutingResponseHandler( pullAllHandler ) );
     }
 
+    @Override
+    public CompletionStage<Void> reset()
+    {
+        return delegate.reset();
+    }
+
     @Override
     public boolean isOpen()
     {

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

@@ -33,6 +33,7 @@
 import org.neo4j.driver.v1.Logger;
 import org.neo4j.driver.v1.Logging;
 import org.neo4j.driver.v1.Value;
+import org.neo4j.driver.v1.exceptions.ClientException;
 
 import static java.util.Objects.requireNonNull;
 import static org.neo4j.driver.internal.messaging.AckFailureMessage.ACK_FAILURE;
@@ -157,14 +158,24 @@ public void handleIgnoredMessage()
         log.debug( "S: IGNORED" );
 
         ResponseHandler handler = handlers.remove();
+
+        Throwable error;
         if ( currentError != null )
         {
-            handler.onFailure( currentError );
+            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", handler );
+            log.warn( "Received IGNORED message for handler %s but error is missing and RESET is not in progress. " +
+                      "Current handlers %s", handler, handlers );
+
+            error = new ClientException( "Database ignored the request" );
         }
+        handler.onFailure( error );
     }
 
     public void handleFatalError( Throwable error )
@@ -212,15 +223,9 @@ public void muteAckFailure()
      * {@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.
-     *
-     * @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;
     }
 

driver/src/main/java/org/neo4j/driver/internal/handlers/ChannelReleasingResetResponseHandler.java

@@ -0,0 +1,55 @@
+/*
+ * Copyright (c) 2002-2018 "Neo Technology,"
+ * Network Engine for Objects in Lund AB [http://neotechnology.com]
+ *
+ * This file is part of Neo4j.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.neo4j.driver.internal.handlers;
+
+import io.netty.channel.Channel;
+import io.netty.channel.pool.ChannelPool;
+import io.netty.util.concurrent.Future;
+
+import java.util.concurrent.CompletableFuture;
+
+import org.neo4j.driver.internal.async.inbound.InboundMessageDispatcher;
+import org.neo4j.driver.internal.util.Clock;
+
+import static org.neo4j.driver.internal.async.ChannelAttributes.setLastUsedTimestamp;
+
+public class ChannelReleasingResetResponseHandler extends ResetResponseHandler
+{
+    private final Channel channel;
+    private final ChannelPool pool;
+    private final Clock clock;
+
+    public ChannelReleasingResetResponseHandler( Channel channel, ChannelPool pool,
+            InboundMessageDispatcher messageDispatcher, Clock clock, CompletableFuture<Void> releaseFuture )
+    {
+        super( messageDispatcher, releaseFuture );
+        this.channel = channel;
+        this.pool = pool;
+        this.clock = clock;
+    }
+
+    @Override
+    protected void resetCompleted( CompletableFuture<Void> completionFuture )
+    {
+        setLastUsedTimestamp( channel, clock.millis() );
+
+        Future<Void> released = pool.release( channel );
+        released.addListener( ignore -> completionFuture.complete( null ) );
+    }
+}

driver/src/main/java/org/neo4j/driver/internal/handlers/ResetResponseHandler.java

@@ -18,62 +18,50 @@
  */
 package org.neo4j.driver.internal.handlers;
 
-import io.netty.channel.Channel;
-import io.netty.channel.pool.ChannelPool;
-import io.netty.util.concurrent.Future;
-
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 
 import org.neo4j.driver.internal.async.inbound.InboundMessageDispatcher;
 import org.neo4j.driver.internal.spi.ResponseHandler;
-import org.neo4j.driver.internal.util.Clock;
 import org.neo4j.driver.v1.Value;
 
-import static org.neo4j.driver.internal.async.ChannelAttributes.setLastUsedTimestamp;
-
 public class ResetResponseHandler implements ResponseHandler
 {
-    private final Channel channel;
-    private final ChannelPool pool;
     private final InboundMessageDispatcher messageDispatcher;
-    private final Clock clock;
-    private final CompletableFuture<Void> releaseFuture;
+    private final CompletableFuture<Void> completionFuture;
 
-    public ResetResponseHandler( Channel channel, ChannelPool pool, InboundMessageDispatcher messageDispatcher,
-            Clock clock, CompletableFuture<Void> releaseFuture )
+    public ResetResponseHandler( InboundMessageDispatcher messageDispatcher, CompletableFuture<Void> completionFuture )
     {
-        this.channel = channel;
-        this.pool = pool;
         this.messageDispatcher = messageDispatcher;
-        this.clock = clock;
-        this.releaseFuture = releaseFuture;
+        this.completionFuture = completionFuture;
     }
 
     @Override
-    public void onSuccess( Map<String,Value> metadata )
+    public final void onSuccess( Map<String,Value> metadata )
     {
-        releaseChannel();
+        resetCompleted();
     }
 
     @Override
-    public void onFailure( Throwable error )
+    public final void onFailure( Throwable error )
     {
-        releaseChannel();
+        resetCompleted();
     }
 
     @Override
-    public void onRecord( Value[] fields )
+    public final void onRecord( Value[] fields )
     {
         throw new UnsupportedOperationException();
     }
 
-    private void releaseChannel()
+    private void resetCompleted()
     {
         messageDispatcher.unMuteAckFailure();
-        setLastUsedTimestamp( channel, clock.millis() );
+        resetCompleted( completionFuture );
+    }
 
-        Future<Void> released = pool.release( channel );
-        released.addListener( ignore -> releaseFuture.complete( null ) );
+    protected void resetCompleted( CompletableFuture<Void> completionFuture )
+    {
+        completionFuture.complete( null );
     }
 }

driver/src/main/java/org/neo4j/driver/internal/spi/Connection.java

@@ -39,6 +39,8 @@ void run( String statement, Map<String,Value> parameters, ResponseHandler runHan
     void runAndFlush( String statement, Map<String,Value> parameters, ResponseHandler runHandler,
             ResponseHandler pullAllHandler );
 
+    CompletionStage<Void> reset();
+
     CompletionStage<Void> release();
 
     void terminateAndRelease( String reason );