Limit data ingestion to one database connection

Hopefully this will prevent the timeouts we're getting during data
ingestion pretty frequently.

Author: loafofpiecrust

graphql/src/lambda.rs

@@ -15,11 +15,12 @@ async fn main() -> Result<(), Error> {
     // Share database connection between executions.
     // This prevents each lambda invocation from creating a new connection to
     // the database.
-    let database = dailp::Database::connect().await?;
+    let connections = Some(16);
+    let database = dailp::Database::connect(connections).await?;
     let schema = {
         Schema::build(Query, Mutation, EmptySubscription)
             .data(DataLoader::new(
-                dailp::Database::connect().await?,
+                dailp::Database::connect(connections).await?,
                 tokio::spawn,
             ))
             .finish()

graphql/src/server.rs

@@ -24,14 +24,14 @@ async fn main() -> tide::Result<()> {
     // create schema
     let schema = Schema::build(query::Query, query::Mutation, EmptySubscription)
         .data(DataLoader::new(
-            dailp::Database::connect().await?,
+            dailp::Database::connect(None).await?,
             tokio::spawn,
         ))
         .finish();
 
     let authed_schema = Schema::build(query::Query, query::Mutation, EmptySubscription)
         .data(DataLoader::new(
-            dailp::Database::connect().await?,
+            dailp::Database::connect(None).await?,
             tokio::spawn,
         ))
         .data(UserInfo::new_test_admin())

migration/src/main.rs

@@ -27,7 +27,7 @@ async fn main() -> Result<()> {
     println!("Validating manuscript spreadsheets...");
     validate_documents().await?;
 
-    let db = Database::connect().await?;
+    let db = Database::connect(Some(1)).await?;
 
     println!("Migrating Image Sources...");
     migrate_image_sources(&db).await?;

types/src/database_sql.rs

@@ -26,11 +26,13 @@ pub struct Database {
     client: sqlx::Pool<sqlx::Postgres>,
 }
 impl Database {
-    pub async fn connect() -> Result<Self> {
+    pub async fn connect(num_connections: Option<u32>) -> Result<Self> {
         let db_url = std::env::var("DATABASE_URL")?;
         let conn = PgPoolOptions::new()
-            .max_connections(std::thread::available_parallelism().map_or(2, |x| x.get() as u32))
-            .acquire_timeout(Duration::from_secs(60 * 4))
+            .max_connections(num_connections.unwrap_or_else(|| {
+                std::thread::available_parallelism().map_or(2, |x| x.get() as u32)
+            }))
+            .acquire_timeout(Duration::from_secs(60 * 8))
             // Disable excessive pings to the database.
             .test_before_acquire(false)
             .connect(&db_url)