Home Database Cache : Stripped Back Solution (#1235)

Implements a cache mapping users to their last known home database. Allowing more intelligent routing decisions when queries are run without a specified database. This saves up to 33% of round trips in some scenarios.

Author: MaxAake

packages/bolt-connection/src/bolt/bolt-protocol-v5x6.js

@@ -16,7 +16,7 @@
  */
 import BoltProtocolV5x5 from './bolt-protocol-v5x5'
 
-import transformersFactories from './bolt-protocol-v5x5.transformer'
+import transformersFactories from './bolt-protocol-v5x6.transformer'
 import Transformer from './transformer'
 
 import { internal } from 'neo4j-driver-core'

packages/bolt-connection/src/bolt/bolt-protocol-v5x7.js

@@ -16,7 +16,7 @@
  */
 import BoltProtocolV5x6 from './bolt-protocol-v5x6'
 
-import transformersFactories from './bolt-protocol-v5x5.transformer'
+import transformersFactories from './bolt-protocol-v5x7.transformer'
 import Transformer from './transformer'
 
 import { internal } from 'neo4j-driver-core'

packages/bolt-connection/src/bolt/bolt-protocol-v5x8.js

@@ -0,0 +1,104 @@
+/**
+ * Copyright (c) "Neo4j"
+ * Neo4j Sweden AB [https://neo4j.com]
+ *
+ * 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.
+ */
+import BoltProtocolV5x7 from './bolt-protocol-v5x7'
+
+import transformersFactories from './bolt-protocol-v5x8.transformer'
+import Transformer from './transformer'
+import RequestMessage from './request-message'
+import { ResultStreamObserver } from './stream-observers'
+
+import { internal } from 'neo4j-driver-core'
+
+const {
+  constants: { BOLT_PROTOCOL_V5_8, FETCH_ALL }
+} = internal
+
+export default class BoltProtocol extends BoltProtocolV5x7 {
+  get version () {
+    return BOLT_PROTOCOL_V5_8
+  }
+
+  get transformer () {
+    if (this._transformer === undefined) {
+      this._transformer = new Transformer(Object.values(transformersFactories).map(create => create(this._config, this._log)))
+    }
+    return this._transformer
+  }
+
+  run (
+    query,
+    parameters,
+    {
+      bookmarks,
+      txConfig,
+      database,
+      mode,
+      impersonatedUser,
+      notificationFilter,
+      beforeKeys,
+      afterKeys,
+      beforeError,
+      afterError,
+      beforeComplete,
+      afterComplete,
+      flush = true,
+      reactive = false,
+      fetchSize = FETCH_ALL,
+      highRecordWatermark = Number.MAX_VALUE,
+      lowRecordWatermark = Number.MAX_VALUE,
+      onDb
+    } = {}
+  ) {
+    const observer = new ResultStreamObserver({
+      server: this._server,
+      reactive,
+      fetchSize,
+      moreFunction: this._requestMore.bind(this),
+      discardFunction: this._requestDiscard.bind(this),
+      beforeKeys,
+      afterKeys,
+      beforeError,
+      afterError,
+      beforeComplete,
+      afterComplete,
+      highRecordWatermark,
+      lowRecordWatermark,
+      enrichMetadata: this._enrichMetadata,
+      onDb
+    })
+
+    const flushRun = reactive
+    this.write(
+      RequestMessage.runWithMetadata5x5(query, parameters, {
+        bookmarks,
+        txConfig,
+        database,
+        mode,
+        impersonatedUser,
+        notificationFilter
+      }),
+      observer,
+      flushRun && flush
+    )
+
+    if (!reactive) {
+      this.write(RequestMessage.pull({ n: fetchSize }), observer, flush)
+    }
+
+    return observer
+  }
+}

packages/bolt-connection/src/bolt/bolt-protocol-v5x8.transformer.js

@@ -0,0 +1,22 @@
+/**
+ * Copyright (c) "Neo4j"
+ * Neo4j Sweden AB [https://neo4j.com]
+ *
+ * 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.
+ */
+
+import v5x7 from './bolt-protocol-v5x7.transformer'
+
+export default {
+  ...v5x7
+}

packages/bolt-connection/src/bolt/create.js

@@ -32,6 +32,7 @@ import BoltProtocolV5x4 from './bolt-protocol-v5x4'
 import BoltProtocolV5x5 from './bolt-protocol-v5x5'
 import BoltProtocolV5x6 from './bolt-protocol-v5x6'
 import BoltProtocolV5x7 from './bolt-protocol-v5x7'
+import BoltProtocolV5x8 from './bolt-protocol-v5x8'
 // eslint-disable-next-line no-unused-vars
 import { Chunker, Dechunker } from '../channel'
 import ResponseHandler from './response-handler'
@@ -257,6 +258,14 @@ function createProtocol (
         log,
         onProtocolError,
         serversideRouting)
+    case 5.8:
+      return new BoltProtocolV5x8(server,
+        chunker,
+        packingConfig,
+        createResponseHandler,
+        log,
+        onProtocolError,
+        serversideRouting)
     default:
       throw newError('Unknown Bolt protocol version: ' + version)
   }

packages/bolt-connection/src/bolt/handshake.js

@@ -76,7 +76,7 @@ function parseNegotiatedResponse (buffer, log) {
  */
 function newHandshakeBuffer () {
   return createHandshakeMessage([
-    [version(5, 7), version(5, 0)],
+    [version(5, 8), version(5, 0)],
     [version(4, 4), version(4, 2)],
     version(4, 1),
     version(3, 0)

packages/bolt-connection/src/bolt/stream-observers.js

@@ -79,7 +79,8 @@ class ResultStreamObserver extends StreamObserver {
     server,
     highRecordWatermark = Number.MAX_VALUE,
     lowRecordWatermark = Number.MAX_VALUE,
-    enrichMetadata
+    enrichMetadata,
+    onDb
   } = {}) {
     super()
 
@@ -113,6 +114,7 @@ class ResultStreamObserver extends StreamObserver {
     this._paused = false
     this._pulled = !reactive
     this._haveRecordStreamed = false
+    this._onDb = onDb
   }
 
   /**
@@ -319,6 +321,10 @@ class ResultStreamObserver extends StreamObserver {
         }
       }
 
+      if (meta.db !== null && this._onDb !== undefined) {
+        this._onDb(meta.db)
+      }
+
       if (meta.fields != null) {
         // remove fields key from metadata object
         delete meta.fields

packages/bolt-connection/src/connection-provider/connection-provider-routing.js

@@ -78,7 +78,8 @@ export default class RoutingConnectionProvider extends PooledConnectionProvider
         this._createConnectionErrorHandler(),
         this._log,
         await this._clientCertificateHolder.getClientCertificate(),
-        this._routingContext
+        this._routingContext,
+        this._channelSsrCallback.bind(this)
       )
     })
 
@@ -99,6 +100,8 @@ export default class RoutingConnectionProvider extends PooledConnectionProvider
     )
 
     this._refreshRoutingTable = functional.reuseOngoingRequest(this._refreshRoutingTable, this)
+    this._withSSR = 0
+    this._withoutSSR = 0
   }
 
   _createConnectionErrorHandler () {
@@ -139,19 +142,30 @@ export default class RoutingConnectionProvider extends PooledConnectionProvider
    * See {@link ConnectionProvider} for more information about this method and
    * its arguments.
    */
-  async acquireConnection ({ accessMode, database, bookmarks, impersonatedUser, onDatabaseNameResolved, auth } = {}) {
-    let name
-    let address
+  async acquireConnection ({ accessMode, database, bookmarks, impersonatedUser, onDatabaseNameResolved, auth, homeDb } = {}) {
     const context = { database: database || DEFAULT_DB_NAME }
 
     const databaseSpecificErrorHandler = new ConnectionErrorHandler(
       SESSION_EXPIRED,
       (error, address) => this._handleUnavailability(error, address, context.database),
-      (error, address) => this._handleWriteFailure(error, address, context.database),
-      (error, address, conn) =>
-        this._handleSecurityError(error, address, conn, context.database)
+      (error, address) => this._handleWriteFailure(error, address, homeDb ?? context.database),
+      (error, address, conn) => this._handleSecurityError(error, address, conn, context.database)
     )
 
+    let conn
+    if (this.SSREnabled() && homeDb !== undefined && database === '') {
+      const currentRoutingTable = this._routingTableRegistry.get(
+        homeDb,
+        () => new RoutingTable({ database: homeDb })
+      )
+      if (currentRoutingTable && !currentRoutingTable.isStaleFor(accessMode)) {
+        conn = await this.getConnectionFromRoutingTable(currentRoutingTable, auth, accessMode, databaseSpecificErrorHandler)
+        if (this.SSREnabled()) {
+          return conn
+        }
+        conn.release()
+      }
+    }
     const routingTable = await this._freshRoutingTable({
       accessMode,
       database: context.database,
@@ -165,7 +179,12 @@ export default class RoutingConnectionProvider extends PooledConnectionProvider
         }
       }
     })
+    return this.getConnectionFromRoutingTable(routingTable, auth, accessMode, databaseSpecificErrorHandler)
+  }
 
+  async getConnectionFromRoutingTable (routingTable, auth, accessMode, databaseSpecificErrorHandler) {
+    let name
+    let address
     // select a target server based on specified access mode
     if (accessMode === READ) {
       address = this._loadBalancingStrategy.selectReader(routingTable.readers)
@@ -663,6 +682,28 @@ export default class RoutingConnectionProvider extends PooledConnectionProvider
       routingTable.forgetRouter(address)
     }
   }
+
+  _channelSsrCallback (isEnabled, action) {
+    if (action === 'OPEN') {
+      if (isEnabled === true) {
+        this._withSSR = this._withSSR + 1
+      } else {
+        this._withoutSSR = this._withoutSSR + 1
+      }
+    } else if (action === 'CLOSE') {
+      if (isEnabled === true) {
+        this._withSSR = this._withSSR - 1
+      } else {
+        this._withoutSSR = this._withoutSSR - 1
+      }
+    } else {
+      throw newError("Channel SSR Callback invoked with action other than 'OPEN' or 'CLOSE'")
+    }
+  }
+
+  SSREnabled () {
+    return this._withSSR > 0 && this._withoutSSR === 0
+  }
 }
 
 /**

packages/bolt-connection/src/connection/connection-channel.js

@@ -34,6 +34,8 @@ let idGenerator = 0
  * @param {ConnectionErrorHandler} errorHandler - the error handler for connection errors.
  * @param {Logger} log - configured logger.
  * @param {clientCertificate} clientCertificate - configured client certificate
+ * @param ssrCallback - callback function used to update the counts of ssr enabled and disabled connections
+ * @param createChannel - function taking a channelConfig object and creating a channel with it
  * @return {Connection} - new connection.
  */
 export function createChannelConnection (
@@ -43,6 +45,7 @@ export function createChannelConnection (
   log,
   clientCertificate,
   serversideRouting = null,
+  ssrCallback,
   createChannel = channelConfig => new Channel(channelConfig)
 ) {
   const channelConfig = new ChannelConfig(
@@ -89,7 +92,8 @@ export function createChannelConnection (
         chunker,
         config.notificationFilter,
         createProtocol,
-        config.telemetryDisabled
+        config.telemetryDisabled,
+        ssrCallback
       )
 
       // forward all pending bytes to the dechunker
@@ -110,9 +114,11 @@ export default class ChannelConnection extends Connection {
    * @param {ConnectionErrorHandler} errorHandler the error handler.
    * @param {ServerAddress} address - the server address to connect to.
    * @param {Logger} log - the configured logger.
-   * @param {boolean} disableLosslessIntegers if this connection should convert all received integers to native JS numbers.
-   * @param {Chunker} chunker the chunker
-   * @param protocolSupplier Bolt protocol supplier
+   * @param {boolean} disableLosslessIntegers - if this connection should convert all received integers to native JS numbers.
+   * @param {Chunker} chunker - the chunker
+   * @param protocolSupplier - Bolt protocol supplier
+   * @param {boolean} telemetryDisabled - wether telemetry has been disabled in driver config.
+   * @param ssrCallback - callback function used to update the counts of ssr enabled and disabled connections.
    */
   constructor (
     channel,
@@ -124,7 +130,8 @@ export default class ChannelConnection extends Connection {
     chunker, // to be removed,
     notificationFilter,
     protocolSupplier,
-    telemetryDisabled
+    telemetryDisabled,
+    ssrCallback = (_) => {}
   ) {
     super(errorHandler)
     this._authToken = null
@@ -143,6 +150,7 @@ export default class ChannelConnection extends Connection {
     this._notificationFilter = notificationFilter
     this._telemetryDisabledDriverConfig = telemetryDisabled === true
     this._telemetryDisabledConnection = true
+    this._ssrCallback = ssrCallback
 
     // connection from the database, returned in response for HELLO message and might not be available
     this._dbConnectionId = null
@@ -331,7 +339,9 @@ export default class ChannelConnection extends Connection {
               if (telemetryEnabledHint === true) {
                 this._telemetryDisabledConnection = false
               }
+              this.SSREnabledHint = metadata.hints['ssr.enabled']
             }
+            this._ssrCallback(this.SSREnabledHint ?? false, 'OPEN')
           }
           resolve(self)
         }
@@ -538,6 +548,7 @@ export default class ChannelConnection extends Connection {
    * @returns {Promise<void>} - A promise that will be resolved when the underlying channel is closed.
    */
   async close () {
+    this._ssrCallback(this.SSREnabledHint ?? false, 'CLOSE')
     if (this._log.isDebugEnabled()) {
       this._log.debug('closing')
     }