Added authorization expired response handling

This update adds support for authorization expired response handling. When such error arrives the driver will mark all connections that were created on or before the creation timestamp of the receiving connection for disposal. The connections will be allowed to finish what they are doing and released back to the pool and will be disposed on the next acquire call.

This update also makes sure that we always register the HELLO response handler so that we can handle the response appropriately.

In addition, it adds the ROLLBACK message support to the driver backend.

Author: injectives

driver/src/main/java/org/neo4j/driver/exceptions/AuthorizationExpiredException.java

@@ -0,0 +1,32 @@
+/*
+ * Copyright (c) "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.exceptions;
+
+/**
+ * The authorization info maintained on the server has expired. The client should reconnect.
+ * <p>
+ * Error code: Neo.ClientError.Security.AuthorizationExpired
+ */
+public class AuthorizationExpiredException extends SecurityException
+{
+    public AuthorizationExpiredException( String code, String message )
+    {
+        super( code, message );
+    }
+}

driver/src/main/java/org/neo4j/driver/internal/async/connection/AuthorizationStateListener.java

@@ -0,0 +1,37 @@
+/*
+ * Copyright (c) "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.async.connection;
+
+import io.netty.channel.Channel;
+
+import org.neo4j.driver.exceptions.AuthorizationExpiredException;
+
+/**
+ * Listener for authorization info state maintained on the server side.
+ */
+public interface AuthorizationStateListener
+{
+    /**
+     * Notifies the listener that the credentials stored on the server side have expired.
+     *
+     * @param e       the {@link AuthorizationExpiredException} exception.
+     * @param channel the channel that received the error.
+     */
+    void onExpired( AuthorizationExpiredException e, Channel channel );
+}

driver/src/main/java/org/neo4j/driver/internal/async/connection/ChannelAttributes.java

@@ -40,6 +40,7 @@ public final class ChannelAttributes
     private static final AttributeKey<Long> LAST_USED_TIMESTAMP = newInstance( "lastUsedTimestamp" );
     private static final AttributeKey<InboundMessageDispatcher> MESSAGE_DISPATCHER = newInstance( "messageDispatcher" );
     private static final AttributeKey<String> TERMINATION_REASON = newInstance( "terminationReason" );
+    private static final AttributeKey<AuthorizationStateListener> AUTHORIZATION_STATE_LISTENER = newInstance( "authorizationStateListener" );
 
     private ChannelAttributes()
     {
@@ -145,6 +146,16 @@ public static void setTerminationReason( Channel channel, String reason )
         setOnce( channel, TERMINATION_REASON, reason );
     }
 
+    public static AuthorizationStateListener authorizationStateListener( Channel channel )
+    {
+        return get( channel, AUTHORIZATION_STATE_LISTENER );
+    }
+
+    public static void setAuthorizationStateListener( Channel channel, AuthorizationStateListener authorizationStateListener )
+    {
+        set( channel, AUTHORIZATION_STATE_LISTENER, authorizationStateListener );
+    }
+
     private static <T> T get( Channel channel, AttributeKey<T> key )
     {
         return channel.attr( key ).get();

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

@@ -25,18 +25,19 @@
 import java.util.Map;
 import java.util.Queue;
 
-import org.neo4j.driver.exceptions.ServiceUnavailableException;
+import org.neo4j.driver.Logger;
+import org.neo4j.driver.Logging;
+import org.neo4j.driver.Value;
+import org.neo4j.driver.exceptions.AuthorizationExpiredException;
+import org.neo4j.driver.exceptions.ClientException;
 import org.neo4j.driver.internal.handlers.ResetResponseHandler;
 import org.neo4j.driver.internal.logging.ChannelActivityLogger;
 import org.neo4j.driver.internal.messaging.ResponseMessageHandler;
 import org.neo4j.driver.internal.spi.ResponseHandler;
 import org.neo4j.driver.internal.util.ErrorUtil;
-import org.neo4j.driver.Logger;
-import org.neo4j.driver.Logging;
-import org.neo4j.driver.Value;
-import org.neo4j.driver.exceptions.ClientException;
 
 import static java.util.Objects.requireNonNull;
+import static org.neo4j.driver.internal.async.connection.ChannelAttributes.authorizationStateListener;
 import static org.neo4j.driver.internal.messaging.request.ResetMessage.RESET;
 import static org.neo4j.driver.internal.util.ErrorUtil.addSuppressed;
 
@@ -114,9 +115,17 @@ public void handleFailureMessage( String code, String message )
             return;
         }
 
-        // write a RESET to "acknowledge" the failure
-        enqueue( new ResetResponseHandler( this ) );
-        channel.writeAndFlush( RESET, channel.voidPromise() );
+        Throwable currentError = this.currentError;
+        if ( currentError instanceof AuthorizationExpiredException )
+        {
+            authorizationStateListener( channel ).onExpired( (AuthorizationExpiredException) currentError, channel );
+        }
+        else
+        {
+            // write a RESET to "acknowledge" the failure
+            enqueue( new ResetResponseHandler( this ) );
+            channel.writeAndFlush( RESET, channel.voidPromise() );
+        }
 
         ResponseHandler handler = removeHandler();
         handler.onFailure( currentError );

driver/src/main/java/org/neo4j/driver/internal/async/pool/ConnectionPoolImpl.java

@@ -46,6 +46,7 @@
 import org.neo4j.driver.internal.util.Futures;
 
 import static java.lang.String.format;
+import static org.neo4j.driver.internal.async.connection.ChannelAttributes.setAuthorizationStateListener;
 import static org.neo4j.driver.internal.util.Futures.combineErrors;
 import static org.neo4j.driver.internal.util.Futures.completeWithNullIfNoError;
 
@@ -66,19 +67,22 @@ public class ConnectionPoolImpl implements ConnectionPool
     private final ConnectionFactory connectionFactory;
 
     public ConnectionPoolImpl( ChannelConnector connector, Bootstrap bootstrap, PoolSettings settings, MetricsListener metricsListener, Logging logging,
-            Clock clock, boolean ownsEventLoopGroup )
+                               Clock clock, boolean ownsEventLoopGroup )
     {
-        this( connector, bootstrap, new NettyChannelTracker( metricsListener, bootstrap.config().group().next(), logging ), settings, metricsListener, logging,
-                clock, ownsEventLoopGroup, new NetworkConnectionFactory( clock, metricsListener ) );
+        this( connector, bootstrap, new NettyChannelTracker( metricsListener, bootstrap.config().group().next(), logging ),
+              new NettyChannelHealthChecker( settings, clock, logging ), settings, metricsListener, logging,
+              clock, ownsEventLoopGroup, new NetworkConnectionFactory( clock, metricsListener ) );
     }
 
-    public ConnectionPoolImpl( ChannelConnector connector, Bootstrap bootstrap, NettyChannelTracker nettyChannelTracker, PoolSettings settings,
-            MetricsListener metricsListener, Logging logging, Clock clock, boolean ownsEventLoopGroup, ConnectionFactory connectionFactory )
+    protected ConnectionPoolImpl( ChannelConnector connector, Bootstrap bootstrap, NettyChannelTracker nettyChannelTracker,
+                                  NettyChannelHealthChecker nettyChannelHealthChecker, PoolSettings settings,
+                                  MetricsListener metricsListener, Logging logging, Clock clock, boolean ownsEventLoopGroup,
+                                  ConnectionFactory connectionFactory )
     {
         this.connector = connector;
         this.bootstrap = bootstrap;
         this.nettyChannelTracker = nettyChannelTracker;
-        this.channelHealthChecker = new NettyChannelHealthChecker( settings, clock, logging );
+        this.channelHealthChecker = nettyChannelHealthChecker;
         this.settings = settings;
         this.metricsListener = metricsListener;
         this.log = logging.getLog( ConnectionPool.class.getSimpleName() );
@@ -104,6 +108,7 @@ public CompletionStage<Connection> acquire( BoltServerAddress address )
             {
                 processAcquisitionError( pool, address, error );
                 assertNotClosed( address, channel, pool );
+                setAuthorizationStateListener( channel, channelHealthChecker );
                 Connection connection = connectionFactory.createConnection( channel, pool );
 
                 metricsListener.afterAcquiredOrCreated( pool.id(), acquireEvent );

driver/src/main/java/org/neo4j/driver/internal/async/pool/NettyChannelHealthChecker.java

@@ -23,27 +23,34 @@
 import io.netty.util.concurrent.Future;
 import io.netty.util.concurrent.Promise;
 
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.neo4j.driver.Logger;
+import org.neo4j.driver.Logging;
+import org.neo4j.driver.exceptions.AuthorizationExpiredException;
+import org.neo4j.driver.internal.async.connection.AuthorizationStateListener;
 import org.neo4j.driver.internal.handlers.PingResponseHandler;
 import org.neo4j.driver.internal.messaging.request.ResetMessage;
 import org.neo4j.driver.internal.util.Clock;
-import org.neo4j.driver.Logger;
-import org.neo4j.driver.Logging;
 
 import static org.neo4j.driver.internal.async.connection.ChannelAttributes.creationTimestamp;
 import static org.neo4j.driver.internal.async.connection.ChannelAttributes.lastUsedTimestamp;
 import static org.neo4j.driver.internal.async.connection.ChannelAttributes.messageDispatcher;
 
-public class NettyChannelHealthChecker implements ChannelHealthChecker
+public class NettyChannelHealthChecker implements ChannelHealthChecker, AuthorizationStateListener
 {
     private final PoolSettings poolSettings;
     private final Clock clock;
     private final Logger log;
+    private final AtomicReference<Optional<Long>> minCreationTimestampMillisOpt;
 
     public NettyChannelHealthChecker( PoolSettings poolSettings, Clock clock, Logging logging )
     {
         this.poolSettings = poolSettings;
         this.clock = clock;
         this.log = logging.getLog( getClass().getSimpleName() );
+        this.minCreationTimestampMillisOpt = new AtomicReference<>( Optional.empty() );
     }
 
     @Override
@@ -60,11 +67,27 @@ public Future<Boolean> isHealthy( Channel channel )
         return ACTIVE.isHealthy( channel );
     }
 
+    @Override
+    public void onExpired( AuthorizationExpiredException e, Channel channel )
+    {
+        long ts = creationTimestamp( channel );
+        // Override current value ONLY if the new one is greater
+        minCreationTimestampMillisOpt.getAndUpdate( prev -> Optional.of( prev.filter( prevTs -> ts <= prevTs ).orElse( ts ) ) );
+    }
+
     private boolean isTooOld( Channel channel )
     {
-        if ( poolSettings.maxConnectionLifetimeEnabled() )
+        long creationTimestampMillis = creationTimestamp( channel );
+        Optional<Long> minCreationTimestampMillisOpt = this.minCreationTimestampMillisOpt.get();
+
+        if ( minCreationTimestampMillisOpt.isPresent() && creationTimestampMillis <= minCreationTimestampMillisOpt.get() )
+        {
+            log.trace( "The channel %s is marked for closure as its creation timestamp is older than or equal to the acceptable minimum timestamp: %s <= %s",
+                       channel, creationTimestampMillis, minCreationTimestampMillisOpt.get() );
+            return true;
+        }
+        else if ( poolSettings.maxConnectionLifetimeEnabled() )
         {
-            long creationTimestampMillis = creationTimestamp( channel );
             long currentTimestampMillis = clock.millis();
 
             long ageMillis = currentTimestampMillis - creationTimestampMillis;
@@ -74,7 +97,7 @@ private boolean isTooOld( Channel channel )
             if ( tooOld )
             {
                 log.trace( "Failed acquire channel %s from the pool because it is too old: %s > %s",
-                        channel, ageMillis, maxAgeMillis );
+                           channel, ageMillis, maxAgeMillis );
             }
 
             return tooOld;

driver/src/main/java/org/neo4j/driver/internal/messaging/v3/BoltProtocolV3.java

@@ -21,15 +21,13 @@
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelPromise;
 
-import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
 
 import org.neo4j.driver.AuthToken;
 import org.neo4j.driver.Bookmark;
 import org.neo4j.driver.Query;
 import org.neo4j.driver.TransactionConfig;
-import org.neo4j.driver.Value;
 import org.neo4j.driver.internal.BookmarkHolder;
 import org.neo4j.driver.internal.DatabaseName;
 import org.neo4j.driver.internal.async.UnmanagedTransaction;
@@ -123,19 +121,10 @@ public CompletionStage<Void> beginTransaction( Connection connection, Bookmark b
             return Futures.failedFuture( error );
         }
 
+        CompletableFuture<Void> beginTxFuture = new CompletableFuture<>();
         BeginMessage beginMessage = new BeginMessage( bookmark, config, connection.databaseName(), connection.mode() );
-
-        if ( bookmark.isEmpty() )
-        {
-            connection.write( beginMessage, NoOpResponseHandler.INSTANCE );
-            return Futures.completedWithNull();
-        }
-        else
-        {
-            CompletableFuture<Void> beginTxFuture = new CompletableFuture<>();
-            connection.writeAndFlush( beginMessage, new BeginTxResponseHandler( beginTxFuture ) );
-            return beginTxFuture;
-        }
+        connection.writeAndFlush( beginMessage, new BeginTxResponseHandler( beginTxFuture ) );
+        return beginTxFuture;
     }
 
     @Override

driver/src/main/java/org/neo4j/driver/internal/retry/ExponentialBackoffRetryLogic.java

@@ -39,6 +39,7 @@
 
 import org.neo4j.driver.Logger;
 import org.neo4j.driver.Logging;
+import org.neo4j.driver.exceptions.AuthorizationExpiredException;
 import org.neo4j.driver.exceptions.ClientException;
 import org.neo4j.driver.exceptions.ServiceUnavailableException;
 import org.neo4j.driver.exceptions.SessionExpiredException;
@@ -155,7 +156,8 @@ protected boolean canRetryOn( Throwable error )
     @Experimental
     public static boolean isRetryable( Throwable error )
     {
-        return error instanceof SessionExpiredException || error instanceof ServiceUnavailableException || isTransientError( error );
+        return error instanceof SessionExpiredException || error instanceof ServiceUnavailableException || error instanceof AuthorizationExpiredException ||
+               isTransientError( error );
     }
 
     /**

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

@@ -25,6 +25,7 @@
 import java.util.stream.Stream;
 
 import org.neo4j.driver.exceptions.AuthenticationException;
+import org.neo4j.driver.exceptions.AuthorizationExpiredException;
 import org.neo4j.driver.exceptions.ClientException;
 import org.neo4j.driver.exceptions.DatabaseException;
 import org.neo4j.driver.exceptions.FatalDiscoveryException;
@@ -75,6 +76,10 @@ else if ( code.equalsIgnoreCase( "Neo.ClientError.Database.DatabaseNotFound" ) )
             {
                 return new FatalDiscoveryException( code, message );
             }
+            else if ( code.equalsIgnoreCase( "Neo.ClientError.Security.AuthorizationExpired" ) )
+            {
+                return new AuthorizationExpiredException( code, message );
+            }
             else
             {
                 return new ClientException( code, message );

driver/src/test/java/org/neo4j/driver/internal/async/connection/ChannelAttributesTest.java

@@ -30,6 +30,7 @@
 import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.mockito.Mockito.mock;
+import static org.neo4j.driver.internal.async.connection.ChannelAttributes.authorizationStateListener;
 import static org.neo4j.driver.internal.async.connection.ChannelAttributes.connectionId;
 import static org.neo4j.driver.internal.async.connection.ChannelAttributes.creationTimestamp;
 import static org.neo4j.driver.internal.async.connection.ChannelAttributes.lastUsedTimestamp;
@@ -38,6 +39,7 @@
 import static org.neo4j.driver.internal.async.connection.ChannelAttributes.serverAddress;
 import static org.neo4j.driver.internal.async.connection.ChannelAttributes.serverAgent;
 import static org.neo4j.driver.internal.async.connection.ChannelAttributes.serverVersion;
+import static org.neo4j.driver.internal.async.connection.ChannelAttributes.setAuthorizationStateListener;
 import static org.neo4j.driver.internal.async.connection.ChannelAttributes.setConnectionId;
 import static org.neo4j.driver.internal.async.connection.ChannelAttributes.setCreationTimestamp;
 import static org.neo4j.driver.internal.async.connection.ChannelAttributes.setLastUsedTimestamp;
@@ -197,4 +199,23 @@ void shouldFailToSetTerminationReasonTwice()
 
         assertThrows( IllegalStateException.class, () -> setTerminationReason( channel, "Reason 2" ) );
     }
+
+    @Test
+    void shouldSetAndGetAuthorizationStateListener()
+    {
+        AuthorizationStateListener listener = mock( AuthorizationStateListener.class );
+        setAuthorizationStateListener( channel, listener );
+        assertEquals( listener, authorizationStateListener( channel ) );
+    }
+
+    @Test
+    void shouldAllowOverridingAuthorizationStateListener()
+    {
+        AuthorizationStateListener listener = mock( AuthorizationStateListener.class );
+        setAuthorizationStateListener( channel, listener );
+        assertEquals( listener, authorizationStateListener( channel ) );
+        AuthorizationStateListener newListener = mock( AuthorizationStateListener.class );
+        setAuthorizationStateListener( channel, newListener );
+        assertEquals( newListener, authorizationStateListener( channel ) );
+    }
 }