Introduce home database resolution cache (#1600)

This update adds support for Bolt protocol 5.8 and introduces home database resolution cache that works with Bolt protocol 5.8+ when routing scheme is used. The cache is maintained by the driver internally and is not exposed in the API, its purpose is to reduce the number of Bolt exchanges when home database resolution is needed.

Author: injectives

bolt-api-netty/src/main/java/org/neo4j/driver/internal/bolt/basicimpl/NettyBoltConnectionProvider.java

@@ -116,7 +116,8 @@ public CompletionStage<BoltConnection> connect(
             String impersonatedUser,
             BoltProtocolVersion minVersion,
             NotificationConfig notificationConfig,
-            Consumer<DatabaseName> databaseNameConsumer) {
+            Consumer<DatabaseName> databaseNameConsumer,
+            Map<String, Object> additionalParameters) {
         synchronized (this) {
             if (closeFuture != null) {
                 return CompletableFuture.failedFuture(new IllegalStateException("Connection provider is closed."));
@@ -189,7 +190,8 @@ public CompletionStage<Void> verifyConnectivity(SecurityPlan securityPlan, Map<S
                         null,
                         null,
                         null,
-                        (ignored) -> {})
+                        (ignored) -> {},
+                        Collections.emptyMap())
                 .thenCompose(BoltConnection::close);
     }
 
@@ -204,7 +206,8 @@ public CompletionStage<Boolean> supportsMultiDb(SecurityPlan securityPlan, Map<S
                         null,
                         null,
                         null,
-                        (ignored) -> {})
+                        (ignored) -> {},
+                        Collections.emptyMap())
                 .thenCompose(boltConnection -> {
                     var supports = boltConnection.protocolVersion().compareTo(BoltProtocolV4.VERSION) >= 0;
                     return boltConnection.close().thenApply(ignored -> supports);
@@ -222,7 +225,8 @@ public CompletionStage<Boolean> supportsSessionAuth(SecurityPlan securityPlan, M
                         null,
                         null,
                         null,
-                        (ignored) -> {})
+                        (ignored) -> {},
+                        Collections.emptyMap())
                 .thenCompose(boltConnection -> {
                     var supports = BoltProtocolV51.VERSION.compareTo(boltConnection.protocolVersion()) <= 0;
                     return boltConnection.close().thenApply(ignored -> supports);

bolt-api-netty/src/main/java/org/neo4j/driver/internal/bolt/basicimpl/impl/BoltConnectionImpl.java

@@ -80,6 +80,7 @@ public final class BoltConnectionImpl implements BoltConnection {
     private final BoltServerAddress serverAddress;
     private final BoltProtocolVersion protocolVersion;
     private final boolean telemetrySupported;
+    private final boolean serverSideRouting;
     private final AtomicReference<BoltConnectionState> stateRef = new AtomicReference<>(BoltConnectionState.OPEN);
     private final AtomicReference<CompletableFuture<AuthData>> authDataRef;
     private final Map<String, Value> routingContext;
@@ -104,6 +105,7 @@ public BoltConnectionImpl(
         this.serverAddress = Objects.requireNonNull(connection.serverAddress());
         this.protocolVersion = Objects.requireNonNull(connection.protocol().version());
         this.telemetrySupported = connection.isTelemetryEnabled();
+        this.serverSideRouting = connection.isSsrEnabled();
         this.authDataRef = new AtomicReference<>(
                 CompletableFuture.completedFuture(new AuthDataImpl(authMap, latestAuthMillisFuture.join())));
         this.valueFactory = Objects.requireNonNull(valueFactory);
@@ -177,8 +179,8 @@ public void onError(Throwable throwable) {
                             }
 
                             @Override
-                            public void onSummary(Void summary) {
-                                handler.onBeginSummary(BeginSummaryImpl.INSTANCE);
+                            public void onSummary(BeginSummary summary) {
+                                handler.onBeginSummary(summary);
                             }
                         },
                         logging,
@@ -520,6 +522,11 @@ public boolean telemetrySupported() {
         return telemetrySupported;
     }
 
+    @Override
+    public boolean serverSideRoutingEnabled() {
+        return serverSideRouting;
+    }
+
     private CompletionStage<Void> executeInEventLoop(Runnable runnable) {
         var executeFuture = new CompletableFuture<Void>();
         Runnable stageCompletingRunnable = () -> {
@@ -720,10 +727,6 @@ private void runIgnoringError(Runnable runnable) {
         }
     }
 
-    private static class BeginSummaryImpl implements BeginSummary {
-        private static final BeginSummary INSTANCE = new BeginSummaryImpl();
-    }
-
     private static class TelemetrySummaryImpl implements TelemetrySummary {
         private static final TelemetrySummary INSTANCE = new TelemetrySummaryImpl();
     }

bolt-api-netty/src/main/java/org/neo4j/driver/internal/bolt/basicimpl/impl/async/NetworkConnection.java

@@ -53,6 +53,7 @@ public class NetworkConnection implements Connection {
     private final String serverAgent;
     private final BoltServerAddress serverAddress;
     private final boolean telemetryEnabled;
+    private final boolean ssrEnabled;
     private final BoltProtocol protocol;
 
     private final Long connectionReadTimeout;
@@ -67,6 +68,7 @@ public NetworkConnection(Channel channel, LoggingProvider logging) {
         this.serverAgent = ChannelAttributes.serverAgent(channel);
         this.serverAddress = ChannelAttributes.serverAddress(channel);
         this.telemetryEnabled = ChannelAttributes.telemetryEnabled(channel);
+        this.ssrEnabled = ChannelAttributes.ssrEnabled(channel);
         this.protocol = BoltProtocol.forChannel(channel);
         this.connectionReadTimeout =
                 ChannelAttributes.connectionReadTimeout(channel).orElse(null);
@@ -111,6 +113,11 @@ public boolean isTelemetryEnabled() {
         return telemetryEnabled;
     }
 
+    @Override
+    public boolean isSsrEnabled() {
+        return ssrEnabled;
+    }
+
     @Override
     public String serverAgent() {
         return serverAgent;

bolt-api-netty/src/main/java/org/neo4j/driver/internal/bolt/basicimpl/impl/async/connection/BoltProtocolUtil.java

@@ -27,7 +27,7 @@
 import org.neo4j.driver.internal.bolt.basicimpl.impl.messaging.v42.BoltProtocolV42;
 import org.neo4j.driver.internal.bolt.basicimpl.impl.messaging.v44.BoltProtocolV44;
 import org.neo4j.driver.internal.bolt.basicimpl.impl.messaging.v5.BoltProtocolV5;
-import org.neo4j.driver.internal.bolt.basicimpl.impl.messaging.v57.BoltProtocolV57;
+import org.neo4j.driver.internal.bolt.basicimpl.impl.messaging.v58.BoltProtocolV58;
 
 public final class BoltProtocolUtil {
     public static final int BOLT_MAGIC_PREAMBLE = 0x6060B017;
@@ -39,7 +39,7 @@ public final class BoltProtocolUtil {
 
     private static final ByteBuf HANDSHAKE_BUF = unreleasableBuffer(copyInt(
                     BOLT_MAGIC_PREAMBLE,
-                    BoltProtocolV57.VERSION.toIntRange(BoltProtocolV5.VERSION),
+                    BoltProtocolV58.VERSION.toIntRange(BoltProtocolV5.VERSION),
                     BoltProtocolV44.VERSION.toIntRange(BoltProtocolV42.VERSION),
                     BoltProtocolV41.VERSION.toInt(),
                     BoltProtocolV3.VERSION.toInt()))

bolt-api-netty/src/main/java/org/neo4j/driver/internal/bolt/basicimpl/impl/async/connection/ChannelAttributes.java

@@ -45,8 +45,8 @@ public final class ChannelAttributes {
 
     // configuration hints provided by the server
     private static final AttributeKey<Long> CONNECTION_READ_TIMEOUT = newInstance("connectionReadTimeout");
-
     private static final AttributeKey<Boolean> TELEMETRY_ENABLED = newInstance("telemetryEnabled");
+    private static final AttributeKey<Boolean> SSR_ENABLED = newInstance("ssr.enabled");
 
     private ChannelAttributes() {}
 
@@ -153,6 +153,14 @@ public static boolean telemetryEnabled(Channel channel) {
         return Optional.ofNullable(get(channel, TELEMETRY_ENABLED)).orElse(false);
     }
 
+    public static void setSsrEnabled(Channel channel, Boolean telemetryEnabled) {
+        setOnce(channel, SSR_ENABLED, telemetryEnabled);
+    }
+
+    public static boolean ssrEnabled(Channel channel) {
+        return Optional.ofNullable(get(channel, SSR_ENABLED)).orElse(false);
+    }
+
     private static <T> T get(Channel channel, AttributeKey<T> key) {
         return channel.attr(key).get();
     }

bolt-api-netty/src/main/java/org/neo4j/driver/internal/bolt/basicimpl/impl/handlers/BeginTxResponseHandler.java

@@ -20,20 +20,24 @@
 
 import java.util.Arrays;
 import java.util.Map;
+import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
+import org.neo4j.driver.internal.bolt.api.summary.BeginSummary;
 import org.neo4j.driver.internal.bolt.api.values.Value;
 import org.neo4j.driver.internal.bolt.basicimpl.impl.spi.ResponseHandler;
 
 public class BeginTxResponseHandler implements ResponseHandler {
-    private final CompletableFuture<Void> beginTxFuture;
+    private final CompletableFuture<BeginSummary> beginTxFuture;
 
-    public BeginTxResponseHandler(CompletableFuture<Void> beginTxFuture) {
+    public BeginTxResponseHandler(CompletableFuture<BeginSummary> beginTxFuture) {
         this.beginTxFuture = requireNonNull(beginTxFuture);
     }
 
     @Override
     public void onSuccess(Map<String, Value> metadata) {
-        beginTxFuture.complete(null);
+        var db = metadata.get("db");
+        var databaseName = db != null ? db.asString() : null;
+        beginTxFuture.complete(new BeginSummaryImpl(databaseName));
     }
 
     @Override
@@ -46,4 +50,11 @@ public void onRecord(Value[] fields) {
         throw new UnsupportedOperationException(
                 "Transaction begin is not expected to receive records: " + Arrays.toString(fields));
     }
+
+    private record BeginSummaryImpl(String database) implements BeginSummary {
+        @Override
+        public Optional<String> databaseName() {
+            return Optional.ofNullable(database);
+        }
+    }
 }

bolt-api-netty/src/main/java/org/neo4j/driver/internal/bolt/basicimpl/impl/handlers/HelloV51ResponseHandler.java

@@ -19,6 +19,7 @@
 import static org.neo4j.driver.internal.bolt.basicimpl.impl.async.connection.ChannelAttributes.setConnectionId;
 import static org.neo4j.driver.internal.bolt.basicimpl.impl.async.connection.ChannelAttributes.setConnectionReadTimeout;
 import static org.neo4j.driver.internal.bolt.basicimpl.impl.async.connection.ChannelAttributes.setServerAgent;
+import static org.neo4j.driver.internal.bolt.basicimpl.impl.async.connection.ChannelAttributes.setSsrEnabled;
 import static org.neo4j.driver.internal.bolt.basicimpl.impl.async.connection.ChannelAttributes.setTelemetryEnabled;
 import static org.neo4j.driver.internal.bolt.basicimpl.impl.util.MetadataExtractor.extractServer;
 
@@ -35,6 +36,7 @@ public class HelloV51ResponseHandler implements ResponseHandler {
     public static final String CONFIGURATION_HINTS_KEY = "hints";
     public static final String CONNECTION_RECEIVE_TIMEOUT_SECONDS_KEY = "connection.recv_timeout_seconds";
     public static final String TELEMETRY_ENABLED_KEY = "telemetry.enabled";
+    public static final String SSR_ENABLED_KEY = "ssr.enabled";
 
     private final Channel channel;
     private final CompletableFuture<String> helloFuture;
@@ -85,6 +87,12 @@ private void processConfigurationHints(Map<String, Value> metadata) {
                         return !value.isNull() && value.asBoolean();
                     })
                     .ifPresent(telemetryEnabled -> setTelemetryEnabled(channel, telemetryEnabled));
+
+            getFromSupplierOrEmptyOnException(() -> {
+                        var value = configurationHints.get(SSR_ENABLED_KEY);
+                        return !value.isNull() && value.asBoolean();
+                    })
+                    .ifPresent(telemetryEnabled -> setSsrEnabled(channel, telemetryEnabled));
         }
     }
 

bolt-api-netty/src/main/java/org/neo4j/driver/internal/bolt/basicimpl/impl/handlers/RunResponseHandler.java

@@ -18,6 +18,7 @@
 
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 import org.neo4j.driver.internal.bolt.api.summary.RunSummary;
 import org.neo4j.driver.internal.bolt.api.values.Value;
@@ -38,8 +39,10 @@ public void onSuccess(Map<String, Value> metadata) {
         var queryKeys = metadataExtractor.extractQueryKeys(metadata);
         var resultAvailableAfter = metadataExtractor.extractResultAvailableAfter(metadata);
         var queryId = metadataExtractor.extractQueryId(metadata);
+        var db = metadata.get("db");
+        var databaseName = db != null ? db.asString() : null;
 
-        runFuture.complete(new RunResponseImpl(queryId, queryKeys, resultAvailableAfter));
+        runFuture.complete(new RunResponseImpl(queryId, queryKeys, resultAvailableAfter, databaseName));
     }
 
     @Override
@@ -52,5 +55,11 @@ public void onRecord(Value[] fields) {
         throw new UnsupportedOperationException();
     }
 
-    private record RunResponseImpl(long queryId, List<String> keys, long resultAvailableAfter) implements RunSummary {}
+    private record RunResponseImpl(long queryId, List<String> keys, long resultAvailableAfter, String database)
+            implements RunSummary {
+        @Override
+        public Optional<String> databaseName() {
+            return Optional.ofNullable(database);
+        }
+    }
 }

bolt-api-netty/src/main/java/org/neo4j/driver/internal/bolt/basicimpl/impl/messaging/BoltProtocol.java

@@ -34,6 +34,7 @@
 import org.neo4j.driver.internal.bolt.api.RoutingContext;
 import org.neo4j.driver.internal.bolt.api.exception.BoltClientException;
 import org.neo4j.driver.internal.bolt.api.exception.BoltUnsupportedFeatureException;
+import org.neo4j.driver.internal.bolt.api.summary.BeginSummary;
 import org.neo4j.driver.internal.bolt.api.summary.DiscardSummary;
 import org.neo4j.driver.internal.bolt.api.summary.RouteSummary;
 import org.neo4j.driver.internal.bolt.api.summary.RunSummary;
@@ -53,6 +54,7 @@
 import org.neo4j.driver.internal.bolt.basicimpl.impl.messaging.v55.BoltProtocolV55;
 import org.neo4j.driver.internal.bolt.basicimpl.impl.messaging.v56.BoltProtocolV56;
 import org.neo4j.driver.internal.bolt.basicimpl.impl.messaging.v57.BoltProtocolV57;
+import org.neo4j.driver.internal.bolt.basicimpl.impl.messaging.v58.BoltProtocolV58;
 import org.neo4j.driver.internal.bolt.basicimpl.impl.spi.Connection;
 
 public interface BoltProtocol {
@@ -90,7 +92,7 @@ CompletionStage<Void> beginTransaction(
             Map<String, Value> txMetadata,
             String txType,
             NotificationConfig notificationConfig,
-            MessageHandler<Void> handler,
+            MessageHandler<BeginSummary> handler,
             LoggingProvider logging,
             ValueFactory valueFactory);
 
@@ -182,6 +184,8 @@ static BoltProtocol forVersion(BoltProtocolVersion version) {
             return BoltProtocolV56.INSTANCE;
         } else if (BoltProtocolV57.VERSION.equals(version)) {
             return BoltProtocolV57.INSTANCE;
+        } else if (BoltProtocolV58.VERSION.equals(version)) {
+            return BoltProtocolV58.INSTANCE;
         }
         throw new BoltClientException("Unknown protocol version: " + version);
     }

bolt-api-netty/src/main/java/org/neo4j/driver/internal/bolt/basicimpl/impl/messaging/v3/BoltProtocolV3.java

@@ -42,6 +42,7 @@
 import org.neo4j.driver.internal.bolt.api.RoutingContext;
 import org.neo4j.driver.internal.bolt.api.exception.BoltException;
 import org.neo4j.driver.internal.bolt.api.exception.BoltUnsupportedFeatureException;
+import org.neo4j.driver.internal.bolt.api.summary.BeginSummary;
 import org.neo4j.driver.internal.bolt.api.summary.DiscardSummary;
 import org.neo4j.driver.internal.bolt.api.summary.PullSummary;
 import org.neo4j.driver.internal.bolt.api.summary.RouteSummary;
@@ -249,7 +250,7 @@ public CompletionStage<Void> beginTransaction(
             Map<String, Value> txMetadata,
             String txType,
             NotificationConfig notificationConfig,
-            MessageHandler<Void> handler,
+            MessageHandler<BeginSummary> handler,
             LoggingProvider logging,
             ValueFactory valueFactory) {
         var exception = verifyNotificationConfigSupported(notificationConfig);
@@ -262,7 +263,7 @@ public CompletionStage<Void> beginTransaction(
             return CompletableFuture.failedFuture(error);
         }
 
-        var beginTxFuture = new CompletableFuture<Void>();
+        var beginTxFuture = new CompletableFuture<BeginSummary>();
         var beginMessage = new BeginMessage(
                 bookmarks,
                 txTimeout,
@@ -275,11 +276,11 @@ public CompletionStage<Void> beginTransaction(
                 useLegacyNotifications(),
                 logging,
                 valueFactory);
-        beginTxFuture.whenComplete((ignored, throwable) -> {
+        beginTxFuture.whenComplete((summary, throwable) -> {
             if (throwable != null) {
                 handler.onError(throwable);
             } else {
-                handler.onSummary(null);
+                handler.onSummary(summary);
             }
         });
         return connection.write(beginMessage, new BeginTxResponseHandler(beginTxFuture));