Should not retry transaction on connection errors when committing a write transaction.

When committing a transaction, if we get a reply from the server, we know the status of the transaction (success or fail).
However if we fail to get a repy from the server due to database crash or network crash, then we are uncertain about the status of this transaction. Because it is possible that this tx has been committed by server, but it just has failed to send us the reply via wire.
It is okay if this transaction is a read only transaction, however if this is a write transaction, if we retry automatically, we might commit the same data twice.

This PR marks the connection error happened at committing for write tx differently (ServiceUnavailable -> ConnectionBrokenAtCommit) from normal retriable connection errors (ServiceUnavailable -> SessionExpired). As a result, on receiving of this error, the tx will not be retried.

Author: Zhen Li

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

@@ -0,0 +1,25 @@
+/*
+ * Copyright (c) 2002-2019 "Neo4j,"
+ * Neo4j Sweden AB [http://neo4j.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 org.neo4j.driver.internal.spi.ResponseHandler;
+
+interface AbstractCommitTxResponseHandler extends ResponseHandler
+{
+}

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

@@ -0,0 +1,43 @@
+/*
+ * Copyright (c) 2002-2019 "Neo4j,"
+ * Neo4j Sweden AB [http://neo4j.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 java.util.Map;
+
+import org.neo4j.driver.v1.Value;
+
+public class CommitTxNoOpResponseHandler implements AbstractCommitTxResponseHandler
+{
+    public static final CommitTxNoOpResponseHandler INSTANCE = new CommitTxNoOpResponseHandler();
+
+    @Override
+    public void onSuccess( Map<String,Value> metadata )
+    {
+    }
+
+    @Override
+    public void onFailure( Throwable error )
+    {
+    }
+
+    @Override
+    public void onRecord( Value[] fields )
+    {
+    }
+}

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

@@ -23,12 +23,11 @@
 import java.util.concurrent.CompletableFuture;
 
 import org.neo4j.driver.internal.Bookmarks;
-import org.neo4j.driver.internal.spi.ResponseHandler;
 import org.neo4j.driver.v1.Value;
 
 import static java.util.Objects.requireNonNull;
 
-public class CommitTxResponseHandler implements ResponseHandler
+public class CommitTxResponseHandler implements AbstractCommitTxResponseHandler
 {
     private final CompletableFuture<Bookmarks> commitFuture;
 

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

@@ -28,6 +28,7 @@
 import org.neo4j.driver.v1.AccessMode;
 import org.neo4j.driver.v1.Value;
 import org.neo4j.driver.v1.exceptions.ClientException;
+import org.neo4j.driver.v1.exceptions.ConnectionBrokenAtCommitException;
 import org.neo4j.driver.v1.exceptions.ServiceUnavailableException;
 import org.neo4j.driver.v1.exceptions.SessionExpiredException;
 import org.neo4j.driver.v1.exceptions.TransientException;
@@ -106,6 +107,14 @@ else if ( error instanceof TransientException )
     private Throwable handledServiceUnavailableException( ServiceUnavailableException e )
     {
         errorHandler.onConnectionFailure( address );
+        if ( accessMode == AccessMode.WRITE && delegate instanceof AbstractCommitTxResponseHandler )
+        {
+            // cannot be retried
+            return new ConnectionBrokenAtCommitException( format( "Connection with server at %s is broken at commit. The commit status is unknown. " +
+                            "Before retrying your transaction, you might want to double check if the previous transaction has been successfully committed by database or not.",
+                    address ), e );
+        }
+        // can be retried
         return new SessionExpiredException( format( "Server at %s is no longer available", address ), e );
     }
 

driver/src/main/java/org/neo4j/driver/internal/messaging/v1/BoltProtocolV1.java

@@ -30,6 +30,7 @@
 import org.neo4j.driver.internal.ExplicitTransaction;
 import org.neo4j.driver.internal.InternalStatementResultCursor;
 import org.neo4j.driver.internal.handlers.BeginTxResponseHandler;
+import org.neo4j.driver.internal.handlers.CommitTxNoOpResponseHandler;
 import org.neo4j.driver.internal.handlers.CommitTxResponseHandler;
 import org.neo4j.driver.internal.handlers.InitResponseHandler;
 import org.neo4j.driver.internal.handlers.NoOpResponseHandler;
@@ -128,7 +129,7 @@ public CompletionStage<Bookmarks> commitTransaction( Connection connection )
 
         ResponseHandler pullAllHandler = new CommitTxResponseHandler( commitFuture );
         connection.writeAndFlush(
-                COMMIT_MESSAGE, NoOpResponseHandler.INSTANCE,
+                COMMIT_MESSAGE, CommitTxNoOpResponseHandler.INSTANCE,
                 PullAllMessage.PULL_ALL, pullAllHandler );
 
         return commitFuture;

driver/src/main/java/org/neo4j/driver/v1/exceptions/ConnectionBrokenAtCommitException.java

@@ -0,0 +1,27 @@
+/*
+ * Copyright (c) 2002-2019 "Neo4j,"
+ * Neo4j Sweden AB [http://neo4j.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.v1.exceptions;
+
+public class ConnectionBrokenAtCommitException extends Neo4jException
+{
+    public ConnectionBrokenAtCommitException( String message, Throwable e )
+    {
+        super( message, e );
+    }
+}

driver/src/test/java/org/neo4j/driver/internal/RoutingDriverBoltKitTest.java

@@ -46,6 +46,7 @@
 import org.neo4j.driver.v1.StatementResult;
 import org.neo4j.driver.v1.Transaction;
 import org.neo4j.driver.v1.TransactionWork;
+import org.neo4j.driver.v1.exceptions.ConnectionBrokenAtCommitException;
 import org.neo4j.driver.v1.exceptions.ServiceUnavailableException;
 import org.neo4j.driver.v1.exceptions.SessionExpiredException;
 import org.neo4j.driver.v1.net.ServerAddress;
@@ -65,6 +66,7 @@
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.startsWith;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
@@ -764,6 +766,56 @@ void shouldRetryWriteTransactionUntilSuccessWithWhenLeaderIsRemoved() throws Exc
         verify( logger ).warn( startsWith( "Failed to obtain a connection towards address 127.0.0.1:9004" ), any( SessionExpiredException.class ) );
     }
 
+    @Test
+    void shouldNotRetryWriteTransactionWhenFailedToCommitV3() throws Exception
+    {
+        StubServer router = StubServer.start( "acquire_endpoints_v3.script", 9001 );
+        StubServer brokenWriter = StubServer.start( "dead_write_server_at_commit_v3.script", 9007 );
+
+        Logger logger = mock( Logger.class );
+        Config config = Config.builder().withoutEncryption().withLogging( mockedLogging( logger ) ).build();
+        try ( Driver driver = newDriverWithSleeplessClock( "bolt+routing://127.0.0.1:9001", config );
+                Session session = driver.session() )
+        {
+            AtomicInteger invocations = new AtomicInteger();
+            assertThrows( ConnectionBrokenAtCommitException.class,
+                    () -> session.writeTransaction( queryWork( "CREATE (n {name:'Bob'})", invocations ) ) );
+
+            assertEquals( 1, invocations.get() );
+        }
+        finally
+        {
+            assertEquals( 0, router.exitStatus() );
+            assertEquals( 0, brokenWriter.exitStatus() );
+        }
+        verify( logger, never() ).warn( startsWith( "Transaction failed and will be retried in" ), any( SessionExpiredException.class ) );
+    }
+
+    @Test
+    void shouldNotRetryWriteTransactionWhenFailedToCommit() throws Exception
+    {
+        StubServer router = StubServer.start( "acquire_endpoints_v3.script", 9001 );
+        StubServer brokenWriter = StubServer.start( "dead_write_server_at_commit.script", 9007 );
+
+        Logger logger = mock( Logger.class );
+        Config config = Config.builder().withoutEncryption().withLogging( mockedLogging( logger ) ).build();
+        try ( Driver driver = newDriverWithSleeplessClock( "bolt+routing://127.0.0.1:9001", config );
+                Session session = driver.session() )
+        {
+            AtomicInteger invocations = new AtomicInteger();
+            assertThrows( ConnectionBrokenAtCommitException.class,
+                    () -> session.writeTransaction( queryWork( "CREATE (n {name:'Bob'})", invocations ) ) );
+
+            assertEquals( 1, invocations.get() );
+        }
+        finally
+        {
+            assertEquals( 0, router.exitStatus() );
+            assertEquals( 0, brokenWriter.exitStatus() );
+        }
+        verify( logger, never() ).warn( startsWith( "Transaction failed and will be retried in" ), any( SessionExpiredException.class ) );
+    }
+
     @Test
     void shouldRetryReadTransactionUntilFailure() throws Exception
     {

driver/src/test/java/org/neo4j/driver/internal/handlers/RoutingResponseHandlerTest.java

@@ -27,6 +27,7 @@
 import org.neo4j.driver.internal.spi.ResponseHandler;
 import org.neo4j.driver.v1.AccessMode;
 import org.neo4j.driver.v1.exceptions.ClientException;
+import org.neo4j.driver.v1.exceptions.ConnectionBrokenAtCommitException;
 import org.neo4j.driver.v1.exceptions.ServiceUnavailableException;
 import org.neo4j.driver.v1.exceptions.SessionExpiredException;
 import org.neo4j.driver.v1.exceptions.TransientException;
@@ -65,6 +66,18 @@ void shouldHandleServiceUnavailableException()
         verify( errorHandler ).onConnectionFailure( LOCAL_DEFAULT );
     }
 
+    @Test
+    void shouldHandleServiceUnavailableExceptionSpeciallyForWriteOnCommitFailure()
+    {
+        ServiceUnavailableException error = new ServiceUnavailableException( "Hi" );
+        RoutingErrorHandler errorHandler = mock( RoutingErrorHandler.class );
+
+        Throwable handledError = handle( mock( AbstractCommitTxResponseHandler.class ), error, errorHandler, AccessMode.WRITE );
+
+        verify( errorHandler ).onConnectionFailure( LOCAL_DEFAULT );
+        assertThat( handledError, instanceOf( ConnectionBrokenAtCommitException.class ) );
+    }
+
     @Test
     void shouldHandleDatabaseUnavailableError()
     {
@@ -181,13 +194,19 @@ private static Throwable handle( Throwable error, RoutingErrorHandler errorHandl
     private static Throwable handle( Throwable error, RoutingErrorHandler errorHandler, AccessMode accessMode )
     {
         ResponseHandler responseHandler = mock( ResponseHandler.class );
+        return handle( responseHandler, error, errorHandler, accessMode );
+    }
+
+    private static Throwable handle( ResponseHandler responseHandler, Throwable error, RoutingErrorHandler errorHandler, AccessMode mode )
+    {
         RoutingResponseHandler routingResponseHandler =
-                new RoutingResponseHandler( responseHandler, LOCAL_DEFAULT, accessMode, errorHandler );
+                new RoutingResponseHandler( responseHandler, LOCAL_DEFAULT, mode, errorHandler );
 
         routingResponseHandler.onFailure( error );
 
         ArgumentCaptor<Throwable> handledErrorCaptor = ArgumentCaptor.forClass( Throwable.class );
         verify( responseHandler ).onFailure( handledErrorCaptor.capture() );
         return handledErrorCaptor.getValue();
+
     }
 }

driver/src/test/java/org/neo4j/driver/internal/messaging/v1/BoltProtocolV1Test.java

@@ -38,6 +38,7 @@
 import org.neo4j.driver.internal.async.ChannelAttributes;
 import org.neo4j.driver.internal.async.inbound.InboundMessageDispatcher;
 import org.neo4j.driver.internal.handlers.BeginTxResponseHandler;
+import org.neo4j.driver.internal.handlers.CommitTxNoOpResponseHandler;
 import org.neo4j.driver.internal.handlers.CommitTxResponseHandler;
 import org.neo4j.driver.internal.handlers.NoOpResponseHandler;
 import org.neo4j.driver.internal.handlers.RollbackTxResponseHandler;
@@ -192,7 +193,7 @@ void shouldCommitTransaction()
         CompletionStage<Bookmarks> stage = protocol.commitTransaction( connection );
 
         verify( connection ).writeAndFlush(
-                eq( new RunMessage( "COMMIT" ) ), eq( NoOpResponseHandler.INSTANCE ),
+                eq( new RunMessage( "COMMIT" ) ), eq( CommitTxNoOpResponseHandler.INSTANCE ),
                 eq( PullAllMessage.PULL_ALL ), any( CommitTxResponseHandler.class ) );
 
         assertEquals( Bookmarks.from( bookmarkString ), await( stage ) );

driver/src/test/resources/dead_write_server_at_commit.script

@@ -0,0 +1,13 @@
+!: AUTO RESET
+!: AUTO INIT
+
+C: RUN "BEGIN" {}
+   PULL_ALL
+   RUN "CREATE (n {name:'Bob'})" {}
+   PULL_ALL
+S: SUCCESS {}
+   SUCCESS {}
+   SUCCESS {"fields": []}
+   SUCCESS {}
+C: RUN "COMMIT" {}
+S: <EXIT>