extract database pool into new database subcrate

Author: syphar

Cargo.lock

@@ -27,6 +27,7 @@ axum-extra = { version = "0.12.0", features = ["typed-header", "routing", "middl
 bincode = "2.0.1"
 chrono = { version = "0.4.11", default-features = false, features = ["clock", "serde"] }
 derive_more = { version = "2.0.0", features = ["display", "deref", "from", "into", "from_str"] }
+futures-util = "0.3.5"
 http = "1.0.0"
 itertools = { version = "0.14.0" }
 mockito = "1.0.2"
@@ -41,12 +42,14 @@ serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
 sqlx = { version = "0.8", features = [ "runtime-tokio", "postgres", "sqlite", "chrono" ] }
 test-case = "3.0.0"
+thiserror = "2.0.3"
 tokio = { version = "1.0", features = ["rt-multi-thread", "signal", "macros", "process", "sync"] }
 tracing = "0.1.37"
 url = { version = "2.1.1", features = ["serde"] }
 
 [dependencies]
 docs_rs_cargo_metadata = { path = "crates/lib/docs_rs_cargo_metadata" }
+docs_rs_database = { path = "crates/lib/docs_rs_database" }
 docs_rs_env_vars = { path = "crates/lib/docs_rs_env_vars" }
 docs_rs_fastly = { path = "crates/lib/docs_rs_fastly" }
 docs_rs_headers = { path = "crates/lib/docs_rs_headers" }
@@ -73,7 +76,7 @@ sqlx = { workspace = true }
 url = { workspace = true }
 docsrs-metadata = { path = "crates/lib/metadata" }
 anyhow = { workspace = true }
-thiserror = "2.0.3"
+thiserror = { workspace = true }
 comrak = { version = "0.49.0", default-features = false }
 syntect = { version = "5.0.0", default-features = false, features = ["parsing", "html", "dump-load", "regex-onig"] }
 toml = "0.9.2"
@@ -104,7 +107,7 @@ async-compression = { version = "0.4.32", features = ["tokio", "bzip2", "zstd",
 tokio = { workspace = true }
 tokio-util = { version = "0.7.15", default-features = false, features = ["io"] }
 tracing-futures= { version = "0.2.5", features = ["std-future", "futures-03"] }
-futures-util = "0.3.5"
+futures-util = { workspace = true }
 async-stream = "0.3.5"
 aws-config = { version = "1.0.0", default-features = false, features = ["rt-tokio", "default-https-client"] }
 aws-sdk-s3 = "1.3.0"
@@ -140,6 +143,7 @@ constant_time_eq = "0.4.2"
 
 [dev-dependencies]
 docs_rs_cargo_metadata = { path = "crates/lib/docs_rs_cargo_metadata", features = ["testing"] }
+docs_rs_database = { path = "crates/lib/docs_rs_database", features = ["testing"] }
 docs_rs_fastly = { path = "crates/lib/docs_rs_fastly", features = ["testing"] }
 docs_rs_headers = { path = "crates/lib/docs_rs_headers", features = ["testing"] }
 docs_rs_opentelemetry = { path = "crates/lib/docs_rs_opentelemetry", features = ["testing"] }

Cargo.toml

@@ -0,0 +1,18 @@
+[package]
+name = "docs_rs_database"
+version = "0.1.0"
+edition = "2024"
+
+[dependencies]
+anyhow = { workspace = true }
+docs_rs_env_vars = { path = "../docs_rs_env_vars" }
+docs_rs_opentelemetry = { path = "../docs_rs_opentelemetry" }
+futures-util = { workspace = true }
+opentelemetry = { workspace = true }
+sqlx = { workspace = true }
+thiserror = { workspace = true }
+tokio = { workspace = true }
+tracing = { workspace = true }
+
+[features]
+testing = []

crates/lib/docs_rs_database/Cargo.toml

@@ -0,0 +1,18 @@
+use docs_rs_env_vars::{env, require_env};
+
+#[derive(Debug)]
+pub struct Config {
+    pub database_url: String,
+    pub max_pool_size: u32,
+    pub min_pool_idle: u32,
+}
+
+impl Config {
+    pub fn from_environment() -> anyhow::Result<Self> {
+        Ok(Self {
+            database_url: require_env("DOCSRS_DATABASE_URL")?,
+            max_pool_size: env("DOCSRS_MAX_POOL_SIZE", 90u32)?,
+            min_pool_idle: env("DOCSRS_MIN_POOL_IDLE", 10u32)?,
+        })
+    }
+}

crates/lib/docs_rs_database/src/config.rs

@@ -0,0 +1,8 @@
+#[derive(Debug, thiserror::Error)]
+pub enum PoolError {
+    #[error("failed to create the database connection pool")]
+    AsyncPoolCreationFailed(#[source] sqlx::Error),
+
+    #[error("failed to get a database connection")]
+    AsyncClientError(#[source] sqlx::Error),
+}

crates/lib/docs_rs_database/src/errors.rs

@@ -0,0 +1,8 @@
+mod config;
+mod errors;
+mod metrics;
+mod pool;
+
+pub use config::Config;
+pub use errors::PoolError;
+pub use pool::{AsyncPoolClient, Pool};

crates/lib/docs_rs_database/src/lib.rs

@@ -0,0 +1,54 @@
+use docs_rs_opentelemetry::AnyMeterProvider;
+use opentelemetry::metrics::{Counter, ObservableGauge};
+
+#[derive(Debug)]
+pub(crate) struct PoolMetrics {
+    pub(crate) failed_connections: Counter<u64>,
+    _idle_connections: ObservableGauge<u64>,
+    _used_connections: ObservableGauge<u64>,
+    _max_connections: ObservableGauge<u64>,
+}
+
+impl PoolMetrics {
+    pub(crate) fn new(pool: sqlx::PgPool, meter_provider: &AnyMeterProvider) -> Self {
+        let meter = meter_provider.meter("pool");
+        const PREFIX: &str = "docsrs.db.pool";
+        Self {
+            failed_connections: meter
+                .u64_counter(format!("{PREFIX}.failed_connections"))
+                .with_unit("1")
+                .build(),
+            _idle_connections: meter
+                .u64_observable_gauge(format!("{PREFIX}.idle_connections"))
+                .with_unit("1")
+                .with_callback({
+                    let pool = pool.clone();
+                    move |observer| {
+                        observer.observe(pool.num_idle() as u64, &[]);
+                    }
+                })
+                .build(),
+            _used_connections: meter
+                .u64_observable_gauge(format!("{PREFIX}.used_connections"))
+                .with_unit("1")
+                .with_callback({
+                    let pool = pool.clone();
+                    move |observer| {
+                        let used = pool.size() as u64 - pool.num_idle() as u64;
+                        observer.observe(used, &[]);
+                    }
+                })
+                .build(),
+            _max_connections: meter
+                .u64_observable_gauge(format!("{PREFIX}.max_connections"))
+                .with_unit("1")
+                .with_callback({
+                    let pool = pool.clone();
+                    move |observer| {
+                        observer.observe(pool.size() as u64, &[]);
+                    }
+                })
+                .build(),
+        }
+    }
+}

crates/lib/docs_rs_database/src/metrics.rs

@@ -1,7 +1,6 @@
-use crate::Config;
+use crate::{Config, errors::PoolError, metrics::PoolMetrics};
 use docs_rs_opentelemetry::AnyMeterProvider;
 use futures_util::{future::BoxFuture, stream::BoxStream};
-use opentelemetry::metrics::{Counter, ObservableGauge};
 use sqlx::{Executor, postgres::PgPoolOptions};
 use std::{
     ops::{Deref, DerefMut},
@@ -13,58 +12,6 @@ use tracing::debug;
 
 const DEFAULT_SCHEMA: &str = "public";
 
-#[derive(Debug)]
-struct PoolMetrics {
-    failed_connections: Counter<u64>,
-    _idle_connections: ObservableGauge<u64>,
-    _used_connections: ObservableGauge<u64>,
-    _max_connections: ObservableGauge<u64>,
-}
-
-impl PoolMetrics {
-    fn new(pool: sqlx::PgPool, meter_provider: &AnyMeterProvider) -> Self {
-        let meter = meter_provider.meter("pool");
-        const PREFIX: &str = "docsrs.db.pool";
-        Self {
-            failed_connections: meter
-                .u64_counter(format!("{PREFIX}.failed_connections"))
-                .with_unit("1")
-                .build(),
-            _idle_connections: meter
-                .u64_observable_gauge(format!("{PREFIX}.idle_connections"))
-                .with_unit("1")
-                .with_callback({
-                    let pool = pool.clone();
-                    move |observer| {
-                        observer.observe(pool.num_idle() as u64, &[]);
-                    }
-                })
-                .build(),
-            _used_connections: meter
-                .u64_observable_gauge(format!("{PREFIX}.used_connections"))
-                .with_unit("1")
-                .with_callback({
-                    let pool = pool.clone();
-                    move |observer| {
-                        let used = pool.size() as u64 - pool.num_idle() as u64;
-                        observer.observe(used, &[]);
-                    }
-                })
-                .build(),
-            _max_connections: meter
-                .u64_observable_gauge(format!("{PREFIX}.max_connections"))
-                .with_unit("1")
-                .with_callback({
-                    let pool = pool.clone();
-                    move |observer| {
-                        observer.observe(pool.size() as u64, &[]);
-                    }
-                })
-                .build(),
-        }
-    }
-}
-
 #[derive(Debug, Clone)]
 pub struct Pool {
     async_pool: sqlx::PgPool,
@@ -83,8 +30,8 @@ impl Pool {
         Self::new_inner(config, DEFAULT_SCHEMA, otel_meter_provider).await
     }
 
-    #[cfg(test)]
-    pub(crate) async fn new_with_schema(
+    #[cfg(feature = "testing")]
+    pub async fn new_with_schema(
         config: &Config,
         schema: &str,
         otel_meter_provider: &AnyMeterProvider,
@@ -233,12 +180,3 @@ impl Drop for AsyncPoolClient {
         drop(self.inner.take())
     }
 }
-
-#[derive(Debug, thiserror::Error)]
-pub enum PoolError {
-    #[error("failed to create the database connection pool")]
-    AsyncPoolCreationFailed(#[source] sqlx::Error),
-
-    #[error("failed to get a database connection")]
-    AsyncClientError(#[source] sqlx::Error),
-}

src/db/pool.rs crates/lib/docs_rs_database/src/pool.rssrc/db/pool.rs renamed to crates/lib/docs_rs_database/src/pool.rs

@@ -1,6 +1,6 @@
 use crate::{
     BuildPackageSummary, Config, Context, Index, RustwideBuilder,
-    db::{AsyncPoolClient, Pool, delete_crate, delete_version, update_latest_version_id},
+    db::{delete_crate, delete_version, update_latest_version_id},
     docbuilder::{BuilderMetrics, PackageKind},
     error::Result,
     storage::AsyncStorage,
@@ -9,6 +9,7 @@ use crate::{
 use anyhow::Context as _;
 use chrono::NaiveDate;
 use crates_index_diff::{Change, CrateVersion};
+use docs_rs_database::{AsyncPoolClient, Pool};
 use docs_rs_fastly::{Cdn, CdnBehaviour as _};
 use docs_rs_opentelemetry::AnyMeterProvider;
 use docs_rs_types::{CrateId, KrateName, Version};

src/build_queue.rs

@@ -17,11 +17,6 @@ pub struct Config {
     /// How long to wait between registry checks
     pub(crate) delay_between_registry_fetches: Duration,
 
-    // Database connection params
-    pub(crate) database_url: String,
-    pub(crate) max_pool_size: u32,
-    pub(crate) min_pool_idle: u32,
-
     // Storage params
     pub(crate) storage_backend: StorageKind,
 
@@ -131,6 +126,7 @@ pub struct Config {
     pub(crate) fastly: docs_rs_fastly::Config,
     pub(crate) opentelemetry: docs_rs_opentelemetry::Config,
     pub(crate) registry_api: docs_rs_registry_api::Config,
+    pub(crate) database: docs_rs_database::Config,
 }
 
 impl Config {
@@ -170,9 +166,6 @@ impl Config {
             .registry_index_path(env("REGISTRY_INDEX_PATH", prefix.join("crates.io-index"))?)
             .registry_url(maybe_env("REGISTRY_URL")?)
             .prefix(prefix.clone())
-            .database_url(require_env("DOCSRS_DATABASE_URL")?)
-            .max_pool_size(env("DOCSRS_MAX_POOL_SIZE", 90u32)?)
-            .min_pool_idle(env("DOCSRS_MIN_POOL_IDLE", 10u32)?)
             .storage_backend(env("DOCSRS_STORAGE_BACKEND", StorageKind::Database)?)
             .aws_sdk_max_retries(env("DOCSRS_AWS_SDK_MAX_RETRIES", 6u32)?)
             .s3_bucket(env("DOCSRS_S3_BUCKET", "rust-docs-rs".to_string())?)
@@ -229,7 +222,8 @@ impl Config {
                     .context("error reading fastly config from environment")?,
             )
             .opentelemetry(docs_rs_opentelemetry::Config::from_environment()?)
-            .registry_api(docs_rs_registry_api::Config::from_environment()?))
+            .registry_api(docs_rs_registry_api::Config::from_environment()?)
+            .database(docs_rs_database::Config::from_environment()?))
     }
 
     pub fn max_file_size_for(&self, path: impl AsRef<Path>) -> usize {