Support multiple routing tables (one per database) (#244)

* Keep a map of registries to store a routing table per database

* Small changes

* Refresh table when empty

* Use empty string for default db

* Use the new fetch single table command

* Do not lock bookmarks

* Fix endless loop

* Create a pool for each server

* Fix server removal

* Get the pool from the pool map given the server instance

* Fix load balancing strategy for multi database support

* When servers list is empty for a db/operation, always force a RT refresh

* Maintain the global ttl for refreshing routing tables, even if we ask for routing tables different from the one used in the connect configuration

* Fix server ordering to ensure consistent load balancer outcome

* Fix compilation issue after rebase with main

* Use error level when refreshing routing table fails

Author: madchicken

lib/src/connection.rs

@@ -448,19 +448,18 @@ impl ConnectionInfo {
         tls_config: &ConnectionTLSConfig,
     ) -> Result<(TlsConnector, ServerName<'static>)> {
         let mut root_cert_store = RootCertStore::empty();
-        match rustls_native_certs::load_native_certs() {
-            Ok(certs) => {
-                root_cert_store.add_parsable_certificates(certs);
-            }
-            Err(e) => {
-                warn!("Failed to load native certificates: {e}");
-            }
-        }
 
         let builder = ClientConfig::builder();
         let config = match tls_config {
             ConnectionTLSConfig::None => {
-                warn!("TLS config set to None but required from the URI. Using default config.");
+                match rustls_native_certs::load_native_certs() {
+                    Ok(certs) => {
+                        root_cert_store.add_parsable_certificates(certs);
+                    }
+                    Err(e) => {
+                        warn!("Failed to load native certificates: {e}");
+                    }
+                }
                 builder
                     .with_root_certificates(root_cert_store)
                     .with_no_client_auth()

lib/src/graph.rs

@@ -34,10 +34,14 @@ pub(crate) enum ConnectionPoolManager {
 
 impl ConnectionPoolManager {
     #[allow(unused_variables)]
-    pub(crate) async fn get(&self, operation: Option<Operation>) -> Result<ManagedConnection> {
+    pub(crate) async fn get(
+        &self,
+        operation: Option<Operation>,
+        db: Option<Database>,
+    ) -> Result<ManagedConnection> {
         match self {
             #[cfg(feature = "unstable-bolt-protocol-impl-v2")]
-            Routed(manager) => manager.get(operation).await,
+            Routed(manager) => manager.get(operation, db).await,
             Direct(pool) => pool.get().await.map_err(crate::Error::from),
         }
     }
@@ -166,7 +170,7 @@ impl Graph {
         operation: Operation,
         bookmarks: &[String],
     ) -> Result<Txn> {
-        let connection = self.pool.get(Some(operation)).await?;
+        let connection = self.pool.get(Some(operation), db.clone()).await?;
         #[cfg(feature = "unstable-bolt-protocol-impl-v2")]
         {
             Txn::new(db, self.config.fetch_size, connection, operation, bookmarks).await

lib/src/pool.rs

@@ -68,8 +68,8 @@ pub fn create_pool(config: &Config) -> Result<ConnectionPool> {
         &config.tls_config,
     )?;
     info!(
-        "creating connection pool with max size {}",
-        config.max_connections
+        "creating connection pool for node {} with max size {}",
+        config.uri, config.max_connections
     );
     Ok(ConnectionPool::builder(mgr)
         .max_size(config.max_connections)

lib/src/query.rs

@@ -117,6 +117,7 @@ impl Query {
             query,
             operation,
             fetch_size,
+            db,
         }
     }
 
@@ -233,6 +234,7 @@ pub(crate) struct RetryableQuery<'a> {
     query: Query,
     operation: Operation,
     fetch_size: Option<usize>,
+    db: Option<Database>,
 }
 
 impl<'a> RetryableQuery<'a> {
@@ -270,7 +272,10 @@ impl<'a> RetryableQuery<'a> {
 
     async fn connect(&self) -> QueryResult<ManagedConnection> {
         // an error when retrieving a connection is considered permanent
-        self.pool.get(Some(self.operation)).await.map_err(Retry::No)
+        self.pool
+            .get(Some(self.operation), self.db.clone())
+            .await
+            .map_err(Retry::No)
     }
 }