allow more connections when connecting to the database

Our external cloud database doesn't seem to be able to make do
with just one. It does seem that there is a sqlx bug making this worse:
launchbadge/sqlx#622

But not a lot of options for now, and a large connection pool is good
anyway.

Author: Glauber Costa

server/src/query/dbconn.rs

@@ -41,9 +41,10 @@ pub(crate) struct DbConnection {
 }
 
 impl DbConnection {
-    pub(crate) async fn connect(uri: &str) -> Result<Self> {
+    pub(crate) async fn connect(uri: &str, nr_conn: usize) -> Result<Self> {
         let opts = AnyConnectOptions::from_str(uri).map_err(QueryError::ConnectionFailed)?;
         let pool = AnyPoolOptions::new()
+            .max_connections(nr_conn as _)
             .connect(uri)
             .await
             .map_err(QueryError::ConnectionFailed)?;
@@ -56,9 +57,9 @@ impl DbConnection {
         })
     }
 
-    pub(crate) async fn local_connection(&self) -> Result<Self> {
+    pub(crate) async fn local_connection(&self, nr_conn: usize) -> Result<Self> {
         match self.kind {
-            Kind::Postgres => Self::connect(&self.conn_uri).await,
+            Kind::Postgres => Self::connect(&self.conn_uri, nr_conn).await,
             Kind::Sqlite => Ok(self.clone()),
         }
     }

server/src/query/engine.rs

@@ -163,8 +163,8 @@ impl QueryEngine {
         Self { kind, pool }
     }
 
-    pub(crate) async fn local_connection(conn: &DbConnection) -> Result<Self> {
-        let local = conn.local_connection().await?;
+    pub(crate) async fn local_connection(conn: &DbConnection, nr_conn: usize) -> Result<Self> {
+        let local = conn.local_connection(nr_conn).await?;
         Ok(Self::new(local.kind, local.pool))
     }
 

server/src/query/meta/mod.rs

@@ -161,8 +161,11 @@ impl MetaService {
         Self { kind, pool }
     }
 
-    pub(crate) async fn local_connection(conn: &DbConnection) -> anyhow::Result<Self> {
-        let local = conn.local_connection().await?;
+    pub(crate) async fn local_connection(
+        conn: &DbConnection,
+        nr_conn: usize,
+    ) -> anyhow::Result<Self> {
+        let local = conn.local_connection(nr_conn).await?;
         Ok(Self::new(local.kind, local.pool))
     }
 

server/src/server.rs

@@ -50,6 +50,9 @@ pub struct Opt {
     /// Should we wait for a debugger before executing any JS?
     #[structopt(long)]
     inspect_brk: bool,
+    /// size of database connection pool.
+    #[structopt(short, long, default_value = "1000")]
+    nr_connections: usize,
     /// How many executor threads to create
     #[structopt(short, long, default_value = "1")]
     executor_threads: usize,
@@ -113,6 +116,7 @@ pub struct SharedState {
     data_db: DbConnection,
     metadata_db: DbConnection,
     materialize: ModulesDirectory,
+    nr_connections: usize,
 }
 
 impl SharedState {
@@ -136,7 +140,8 @@ impl SharedTasks {
 async fn run(state: SharedState, mut cmd: ExecutorChannel) -> Result<()> {
     init_deno(&state.materialize.path(), state.inspect_brk).await?;
 
-    let meta = Rc::new(MetaService::local_connection(&state.metadata_db).await?);
+    let meta =
+        Rc::new(MetaService::local_connection(&state.metadata_db, state.nr_connections).await?);
     let ts = meta.load_type_system().await?;
 
     let routes = meta.load_endpoints().await?;
@@ -148,7 +153,8 @@ async fn run(state: SharedState, mut cmd: ExecutorChannel) -> Result<()> {
         Ok(Type::Object(t)) => t,
         _ => anyhow::bail!("Internal error: type {} not found", OAUTHUSER_TYPE_NAME),
     };
-    let query_engine = Arc::new(QueryEngine::local_connection(&state.data_db).await?);
+    let query_engine =
+        Arc::new(QueryEngine::local_connection(&state.data_db, state.nr_connections).await?);
     let mut transaction = query_engine.start_transaction().await?;
     query_engine
         .create_table(&mut transaction, &oauth_user_type)
@@ -239,11 +245,11 @@ pub async fn run_shared_state(
     let file = format!("{}/chisel.ts", materialize.path().display());
     fs::write(file, include_bytes!("../../api/src/chisel.ts")).await?;
 
-    let meta_conn = DbConnection::connect(&opt.metadata_db_uri).await?;
-    let data_db = DbConnection::connect(&opt.data_db_uri).await?;
+    let meta_conn = DbConnection::connect(&opt.metadata_db_uri, opt.nr_connections).await?;
+    let data_db = DbConnection::connect(&opt.data_db_uri, opt.nr_connections).await?;
 
-    let meta = MetaService::local_connection(&meta_conn).await?;
-    let query_engine = QueryEngine::local_connection(&data_db).await?;
+    let meta = MetaService::local_connection(&meta_conn, opt.nr_connections).await?;
+    let query_engine = QueryEngine::local_connection(&data_db, opt.nr_connections).await?;
 
     meta.create_schema().await?;
 
@@ -383,6 +389,7 @@ pub async fn run_shared_state(
         data_db,
         metadata_db: meta_conn,
         materialize: materialize.clone(),
+        nr_connections: opt.nr_connections,
     };
 
     let tasks = SharedTasks { rpc_task, sig_task };