Treat request errors as fatal

Fatal errors result in network channel being closed. Previously only
some errors that originate in the driver were treated as fatal.
However some errors received from the database mean communication
is broken.

This commit makes protocol errors that start with
`Neo.ClientError.Request` be treated as fatal. No ACK_FAILURE will be
send for such errors and netty channel will be closed.

Author: lutovich

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

@@ -21,27 +21,29 @@
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelFutureListener;
+import io.netty.channel.ChannelPipeline;
 import io.netty.channel.ChannelPromise;
 
 import org.neo4j.driver.v1.Logging;
 import org.neo4j.driver.v1.exceptions.ServiceUnavailableException;
 
 import static java.lang.String.format;
-import static java.util.Objects.requireNonNull;
 import static org.neo4j.driver.internal.async.ProtocolUtil.handshake;
 
 public class ChannelConnectedListener implements ChannelFutureListener
 {
     private final BoltServerAddress address;
+    private final ChannelPipelineBuilder pipelineBuilder;
     private final ChannelPromise handshakeCompletedPromise;
     private final Logging logging;
 
-    public ChannelConnectedListener( BoltServerAddress address, ChannelPromise handshakeCompletedPromise,
-            Logging logging )
+    public ChannelConnectedListener( BoltServerAddress address, ChannelPipelineBuilder pipelineBuilder,
+            ChannelPromise handshakeCompletedPromise, Logging logging )
     {
-        this.address = requireNonNull( address );
-        this.handshakeCompletedPromise = requireNonNull( handshakeCompletedPromise );
-        this.logging = requireNonNull( logging );
+        this.address = address;
+        this.pipelineBuilder = pipelineBuilder;
+        this.handshakeCompletedPromise = handshakeCompletedPromise;
+        this.logging = logging;
     }
 
     @Override
@@ -51,7 +53,8 @@ public void operationComplete( ChannelFuture future )
 
         if ( future.isSuccess() )
         {
-            channel.pipeline().addLast( new HandshakeResponseHandler( handshakeCompletedPromise, logging ) );
+            ChannelPipeline pipeline = channel.pipeline();
+            pipeline.addLast( new HandshakeResponseHandler( pipelineBuilder, handshakeCompletedPromise, logging ) );
             ChannelFuture handshakeFuture = channel.writeAndFlush( handshake() );
 
             handshakeFuture.addListener( channelFuture ->

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

@@ -43,17 +43,25 @@ public class ChannelConnectorImpl implements ChannelConnector
     private final String userAgent;
     private final Map<String,Value> authToken;
     private final SecurityPlan securityPlan;
+    private final ChannelPipelineBuilder pipelineBuilder;
     private final int connectTimeoutMillis;
     private final Logging logging;
     private final Clock clock;
 
     public ChannelConnectorImpl( ConnectionSettings connectionSettings, SecurityPlan securityPlan, Logging logging,
             Clock clock )
+    {
+        this( connectionSettings, securityPlan, new ChannelPipelineBuilderImpl(), logging, clock );
+    }
+
+    public ChannelConnectorImpl( ConnectionSettings connectionSettings, SecurityPlan securityPlan,
+            ChannelPipelineBuilder pipelineBuilder, Logging logging, Clock clock )
     {
         this.userAgent = connectionSettings.userAgent();
         this.authToken = tokenAsMap( connectionSettings.authToken() );
         this.connectTimeoutMillis = connectionSettings.connectTimeoutMillis();
         this.securityPlan = requireNonNull( securityPlan );
+        this.pipelineBuilder = pipelineBuilder;
         this.logging = requireNonNull( logging );
         this.clock = requireNonNull( clock );
     }
@@ -70,8 +78,10 @@ public ChannelFuture connect( BoltServerAddress address, Bootstrap bootstrap )
         ChannelPromise handshakeCompleted = channel.newPromise();
         ChannelPromise connectionInitialized = channel.newPromise();
 
-        channelConnected.addListener( new ChannelConnectedListener( address, handshakeCompleted, logging ) );
-        handshakeCompleted.addListener( new HandshakeCompletedListener( userAgent, authToken, connectionInitialized ) );
+        channelConnected.addListener(
+                new ChannelConnectedListener( address, pipelineBuilder, handshakeCompleted, logging ) );
+        handshakeCompleted.addListener(
+                new HandshakeCompletedListener( userAgent, authToken, connectionInitialized ) );
 
         return connectionInitialized;
     }

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

@@ -93,7 +93,7 @@ private void fail( ChannelHandlerContext ctx, Throwable error )
     {
         Throwable cause = transformError( error );
         messageDispatcher.handleFatalError( cause );
-        log.debug( "Closing channel: %s", ctx.channel() );
+        log.debug( "Closing channel because of an error: %s", ctx.channel() );
         ctx.close();
     }
 

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

@@ -18,35 +18,12 @@
  */
 package org.neo4j.driver.internal.async;
 
-import io.netty.channel.Channel;
 import io.netty.channel.ChannelPipeline;
 
-import org.neo4j.driver.internal.async.inbound.ChunkDecoder;
-import org.neo4j.driver.internal.async.inbound.InboundMessageHandler;
-import org.neo4j.driver.internal.async.inbound.MessageDecoder;
-import org.neo4j.driver.internal.async.outbound.OutboundMessageHandler;
 import org.neo4j.driver.internal.messaging.MessageFormat;
 import org.neo4j.driver.v1.Logging;
 
-public final class ChannelPipelineBuilder
+public interface ChannelPipelineBuilder
 {
-    private ChannelPipelineBuilder()
-    {
-    }
-
-    public static void buildPipeline( Channel channel, MessageFormat messageFormat, Logging logging )
-    {
-        ChannelPipeline pipeline = channel.pipeline();
-
-        // inbound handlers
-        pipeline.addLast( new ChunkDecoder() );
-        pipeline.addLast( new MessageDecoder() );
-        pipeline.addLast( new InboundMessageHandler( messageFormat, logging ) );
-
-        // outbound handlers
-        pipeline.addLast( OutboundMessageHandler.NAME, new OutboundMessageHandler( messageFormat, logging ) );
-
-        // last one - error handler
-        pipeline.addLast( new ChannelErrorHandler( logging ) );
-    }
+    void build( MessageFormat messageFormat, ChannelPipeline pipeline, Logging logging );
 }

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

@@ -0,0 +1,46 @@
+/*
+ * Copyright (c) 2002-2017 "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.async;
+
+import io.netty.channel.ChannelPipeline;
+
+import org.neo4j.driver.internal.async.inbound.ChunkDecoder;
+import org.neo4j.driver.internal.async.inbound.InboundMessageHandler;
+import org.neo4j.driver.internal.async.inbound.MessageDecoder;
+import org.neo4j.driver.internal.async.outbound.OutboundMessageHandler;
+import org.neo4j.driver.internal.messaging.MessageFormat;
+import org.neo4j.driver.v1.Logging;
+
+public class ChannelPipelineBuilderImpl implements ChannelPipelineBuilder
+{
+    @Override
+    public void build( MessageFormat messageFormat, ChannelPipeline pipeline, Logging logging )
+    {
+        // inbound handlers
+        pipeline.addLast( new ChunkDecoder() );
+        pipeline.addLast( new MessageDecoder() );
+        pipeline.addLast( new InboundMessageHandler( messageFormat, logging ) );
+
+        // outbound handlers
+        pipeline.addLast( OutboundMessageHandler.NAME, new OutboundMessageHandler( messageFormat, logging ) );
+
+        // last one - error handler
+        pipeline.addLast( new ChannelErrorHandler( logging ) );
+    }
+}

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

@@ -41,12 +41,15 @@
 
 public class HandshakeResponseHandler extends ReplayingDecoder<Void>
 {
+    private final ChannelPipelineBuilder pipelineBuilder;
     private final ChannelPromise handshakeCompletedPromise;
     private final Logging logging;
     private final Logger log;
 
-    public HandshakeResponseHandler( ChannelPromise handshakeCompletedPromise, Logging logging )
+    public HandshakeResponseHandler( ChannelPipelineBuilder pipelineBuilder, ChannelPromise handshakeCompletedPromise,
+            Logging logging )
     {
+        this.pipelineBuilder = pipelineBuilder;
         this.handshakeCompletedPromise = handshakeCompletedPromise;
         this.logging = logging;
         this.log = logging.getLog( getClass().getSimpleName() );
@@ -80,10 +83,9 @@ protected void decode( ChannelHandlerContext ctx, ByteBuf in, List<Object> out )
         switch ( serverSuggestedVersion )
         {
         case PROTOCOL_VERSION_1:
-            MessageFormat format = new PackStreamMessageFormatV1();
-            ChannelPipelineBuilder.buildPipeline( ctx.channel(), format, logging );
+            MessageFormat messageFormat = new PackStreamMessageFormatV1();
+            pipelineBuilder.build( messageFormat, pipeline, logging );
             handshakeCompletedPromise.setSuccess();
-
             break;
         case NO_PROTOCOL_VERSION:
             fail( ctx, protocolNoSupportedByServerError() );

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

@@ -134,6 +134,14 @@ 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 );
 
+        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
+            channel.pipeline().fireExceptionCaught( currentError );
+            return;
+        }
+
         // try to write ACK_FAILURE before notifying the next response handler
         ackFailureIfNeeded();
 
@@ -180,6 +188,11 @@ public Throwable currentError()
         return currentError;
     }
 
+    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"

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

@@ -61,9 +61,16 @@ public void handlerRemoved( ChannelHandlerContext ctx )
     @Override
     protected void channelRead0( ChannelHandlerContext ctx, ByteBuf msg )
     {
+        if ( messageDispatcher.fatalErrorOccurred() )
+        {
+            log.warn( "Message ignored because of the previous fatal error. Channel will be closed. Message:\n%s\n",
+                    prettyHexDump( msg ) );
+            return;
+        }
+
         if ( log.isTraceEnabled() )
         {
-            log.trace( "Inbound message received: \n%s\n", prettyHexDump( msg ) );
+            log.trace( "Inbound message received:\n%s\n", prettyHexDump( msg ) );
         }
 
         input.start( msg );

driver/src/main/java/org/neo4j/driver/internal/async/outbound/OutboundMessageHandler.java

@@ -68,11 +68,7 @@ protected void encode( ChannelHandlerContext ctx, Message msg, List<Object> out
         }
         catch ( Throwable error )
         {
-            EncoderException exception = new EncoderException( "Failed to write outbound message: " + msg, error );
-            // tell ChannelErrorHandler which is the last handler in the pipeline about this error
-            ctx.fireExceptionCaught( exception );
-            // rethrow, encoder contract requires handler to either fail or populate out list
-            throw exception;
+            throw new EncoderException( "Failed to write outbound message: " + msg, error );
         }
         finally
         {

driver/src/main/java/org/neo4j/driver/internal/util/ErrorUtil.java

@@ -51,22 +51,21 @@ public static Neo4jException newNeo4jError( String code, String message )
         }
     }
 
-    // todo: use this method and close channel after unrecoverable error
-    public static boolean isRecoverable( Throwable error )
+    public static boolean isFatal( Throwable error )
     {
         if ( error instanceof Neo4jException )
         {
             if ( isProtocolViolationError( ((Neo4jException) error) ) )
             {
-                return false;
+                return true;
             }
 
             if ( isClientOrTransientError( ((Neo4jException) error) ) )
             {
-                return true;
+                return false;
             }
         }
-        return false;
+        return true;
     }
 
     private static boolean isProtocolViolationError( Neo4jException error )
@@ -84,6 +83,10 @@ private static boolean isClientOrTransientError( Neo4jException error )
     private static String extractClassification( String code )
     {
         String[] parts = code.split( "\\." );
+        if ( parts.length < 2 )
+        {
+            return "";
+        }
         return parts[1];
     }
 }