perf: make database connection pools resilient under concurrent load

config/config.docker.yaml

@@ -17,6 +17,16 @@ database:
     # Connection Pool Settings (optional - defaults shown)
     # max_connections: 100
     # min_connections: 5
+    # Per-request query fan-out bound: how many related-table queries a single
+    # list/detail request runs at once (caps connections held per request).
+    # Higher than SQLite since Postgres runs queries truly concurrently.
+    # batch_fan_out: 8
+    # Separate pool for in-process background work (task workers, scheduler,
+    # pollers). NOTE: additive to max_connections; ensure the Postgres server's
+    # max_connections covers the total when workers run in the same process.
+    # Multi-pod deployments run serve with CODEX_DISABLE_WORKERS=true, so no
+    # extra pool is created there.
+    # background_max_connections: 16
     # acquire_timeout_seconds: 30
     # idle_timeout_seconds: 600
     # max_lifetime_seconds: 3600

config/config.sqlite.yaml

@@ -15,8 +15,15 @@ database:
       synchronous: NORMAL
       # Note: foreign_keys is ALWAYS ON for data integrity (cannot be disabled)
     # Connection Pool Settings (optional - defaults shown)
-    # max_connections: 16
+    # max_connections: 64
     # min_connections: 2
+    # Per-request query fan-out bound: how many related-table queries a single
+    # list/detail request runs at once (caps connections held per request).
+    # Kept low for SQLite (writes serialize; queries are cheap lookups).
+    # batch_fan_out: 4
+    # Separate pool for in-process background work (task workers, scheduler,
+    # pollers) so heavy background queries cannot drain the API pool.
+    # background_max_connections: 4
     # acquire_timeout_seconds: 30
     # idle_timeout_seconds: 300
     # max_lifetime_seconds: 1800

crates/codex-api/src/db_batch.rs

@@ -0,0 +1,130 @@
+//! Bounded fan-out for per-request batched reads.
+//!
+//! List endpoints enrich their rows from many related tables. Running those
+//! queries with an unbounded `tokio::join!` lets a single request hold one
+//! pool connection *per query* simultaneously (14 for the full series DTO).
+//! Under concurrent load that exhausts the connection pool — acutely on SQLite,
+//! whose pool is small — so requests block for seconds on `acquire()`.
+//!
+//! Gating each query on a shared [`Semaphore`] caps how many run, and therefore
+//! how many connections one request holds, at once. The concurrency benefit is
+//! preserved up to the bound while the pathological amplification is removed.
+//!
+//! Each `tokio::join!` arm keeps its own return type — this gates heterogeneous
+//! futures (the arms return different map types) without the boxing/type-erasure
+//! a homogeneous `buffer_unordered` stream would require.
+
+use std::future::Future;
+use std::sync::OnceLock;
+use tokio::sync::Semaphore;
+
+/// Fallback per-request fan-out bound used until [`set_fan_out`] is called
+/// (e.g. in tests that exercise handlers without going through `serve`).
+pub const DEFAULT_BATCH_FAN_OUT: usize = 4;
+
+/// Process-wide resolved fan-out bound, set once at startup from the per-backend
+/// database config. It is a process constant (the backend never changes at
+/// runtime), so a global avoids threading the value through every converter and
+/// call site. Unset → [`DEFAULT_BATCH_FAN_OUT`].
+static CONFIGURED_FAN_OUT: OnceLock<usize> = OnceLock::new();
+
+/// Set the process-wide fan-out bound from configuration. Called once during
+/// startup. Clamped to at least 1. Subsequent calls are ignored (the first
+/// value wins), which keeps it stable across the process lifetime.
+pub fn set_fan_out(bound: usize) {
+    let _ = CONFIGURED_FAN_OUT.set(bound.max(1));
+}
+
+/// The configured fan-out bound, or [`DEFAULT_BATCH_FAN_OUT`] if unset.
+pub fn configured_fan_out() -> usize {
+    CONFIGURED_FAN_OUT
+        .get()
+        .copied()
+        .unwrap_or(DEFAULT_BATCH_FAN_OUT)
+}
+
+/// Build a [`Semaphore`] that bounds concurrent batched reads to `bound`.
+///
+/// `bound` is clamped to at least 1 so a misconfigured `0` cannot deadlock the
+/// request (which would otherwise never acquire a permit).
+pub fn fan_out_limiter(bound: usize) -> Semaphore {
+    Semaphore::new(bound.max(1))
+}
+
+/// Run `fut` once a permit is available, holding the permit for the whole query.
+///
+/// Used to wrap each arm of a `tokio::join!` so the number of arms actively
+/// executing (and thus pool connections held) never exceeds the `limiter`'s
+/// bound. Arms beyond the bound are parked on `acquire()` holding no connection.
+pub async fn with_permit<F: Future>(limiter: &Semaphore, fut: F) -> F::Output {
+    let _permit = limiter
+        .acquire()
+        .await
+        .expect("fan-out limiter semaphore is never closed");
+    fut.await
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::sync::Arc;
+    use std::sync::atomic::{AtomicUsize, Ordering::SeqCst};
+
+    #[test]
+    fn fan_out_limiter_clamps_zero_to_one() {
+        assert_eq!(fan_out_limiter(0).available_permits(), 1);
+        assert_eq!(fan_out_limiter(4).available_permits(), 4);
+    }
+
+    #[test]
+    fn configured_fan_out_is_always_positive() {
+        // Whether or not set_fan_out has run in this process, the bound must be
+        // >= 1 so a request can always acquire a permit. (We avoid calling
+        // set_fan_out here: CONFIGURED_FAN_OUT is process-global and a write
+        // would leak into other tests in this binary.)
+        assert!(configured_fan_out() >= 1);
+    }
+
+    /// `with_permit` must (a) return each arm's value in order and (b) never let
+    /// more than `bound` arms execute concurrently.
+    #[tokio::test]
+    async fn with_permit_bounds_concurrency_and_preserves_results() {
+        let limiter = fan_out_limiter(3);
+        let current = Arc::new(AtomicUsize::new(0));
+        let max_seen = Arc::new(AtomicUsize::new(0));
+
+        let task = |i: usize| {
+            let current = current.clone();
+            let max_seen = max_seen.clone();
+            async move {
+                let now = current.fetch_add(1, SeqCst) + 1;
+                max_seen.fetch_max(now, SeqCst);
+                // Yield repeatedly so other arms get a chance to run; this is
+                // what would let concurrency exceed the bound if it were unbounded.
+                for _ in 0..8 {
+                    tokio::task::yield_now().await;
+                }
+                current.fetch_sub(1, SeqCst);
+                i
+            }
+        };
+
+        let results = tokio::join!(
+            with_permit(&limiter, task(0)),
+            with_permit(&limiter, task(1)),
+            with_permit(&limiter, task(2)),
+            with_permit(&limiter, task(3)),
+            with_permit(&limiter, task(4)),
+            with_permit(&limiter, task(5)),
+            with_permit(&limiter, task(6)),
+            with_permit(&limiter, task(7)),
+        );
+
+        assert_eq!(results, (0, 1, 2, 3, 4, 5, 6, 7));
+        assert!(
+            max_seen.load(SeqCst) <= 3,
+            "observed {} concurrent arms, expected <= 3",
+            max_seen.load(SeqCst)
+        );
+    }
+}

crates/codex-api/src/lib.rs

@@ -1,3 +1,4 @@
+pub mod db_batch;
 pub mod docs;
 pub mod error;
 pub mod extractors;

crates/codex-api/src/routes/komga/handlers/books.rs

@@ -276,8 +276,10 @@ pub async fn get_books_ondeck(
     require_permission!(auth, Permission::BooksRead)?;
 
     let user_id = auth.user_id;
+    // Komga pagination is 0-indexed; convert to a 0-indexed row offset.
     let page = query.page.max(0) as u64;
     let size = query.size.clamp(1, 500) as u64;
+    let offset = page * size;
 
     let content_filter = ContentFilter::for_user(&state.db, user_id)
         .await
@@ -290,7 +292,7 @@ pub async fn get_books_ondeck(
         &state.db,
         user_id,
         query.library_id,
-        page,
+        offset,
         size,
         visibility.as_ref(),
     )

crates/codex-api/src/routes/v1/handlers/books.rs

@@ -206,11 +206,22 @@ pub async fn books_to_dtos(
 
     // Fan out one batched query per related table. Previous per-row loops
     // produced O(books) sequential DB roundtrips here.
+    let limiter = crate::db_batch::fan_out_limiter(crate::db_batch::configured_fan_out());
+    use crate::db_batch::with_permit;
     let (series_metadata_result, book_metadata_result, libraries_result, progress_result) = tokio::join!(
-        SeriesMetadataRepository::get_by_series_ids(db, &series_ids),
-        BookMetadataRepository::get_by_book_ids(db, &book_ids),
-        LibraryRepository::get_by_ids(db, &library_ids),
-        ReadProgressRepository::get_for_user_books(db, user_id, &book_ids),
+        with_permit(
+            &limiter,
+            SeriesMetadataRepository::get_by_series_ids(db, &series_ids)
+        ),
+        with_permit(
+            &limiter,
+            BookMetadataRepository::get_by_book_ids(db, &book_ids)
+        ),
+        with_permit(&limiter, LibraryRepository::get_by_ids(db, &library_ids)),
+        with_permit(
+            &limiter,
+            ReadProgressRepository::get_for_user_books(db, user_id, &book_ids)
+        ),
     );
 
     let series_metadata_map = series_metadata_result
@@ -337,12 +348,21 @@ pub async fn books_to_full_dtos_batched(
         .into_iter()
         .collect();
 
-    // Fetch all related data in parallel
+    // Fetch all related data in parallel, bounded so a single request cannot
+    // hold one pool connection per query and exhaust the pool.
+    let limiter = crate::db_batch::fan_out_limiter(crate::db_batch::configured_fan_out());
+    use crate::db_batch::with_permit;
     let (metadata_map, series_metadata_map, library_map, progress_map, genres_map, tags_map) = tokio::join!(
-        BookMetadataRepository::get_by_book_ids(db, &book_ids),
-        SeriesMetadataRepository::get_by_series_ids(db, &series_ids),
-        LibraryRepository::get_by_ids(db, &library_ids),
-        async {
+        with_permit(
+            &limiter,
+            BookMetadataRepository::get_by_book_ids(db, &book_ids)
+        ),
+        with_permit(
+            &limiter,
+            SeriesMetadataRepository::get_by_series_ids(db, &series_ids)
+        ),
+        with_permit(&limiter, LibraryRepository::get_by_ids(db, &library_ids)),
+        with_permit(&limiter, async {
             // Fetch read progress for all books
             let mut map = HashMap::new();
             for book_id in &book_ids {
@@ -353,9 +373,15 @@ pub async fn books_to_full_dtos_batched(
                 }
             }
             Ok::<_, anyhow::Error>(map)
-        },
-        GenreRepository::get_genres_for_book_ids(db, &book_ids),
-        TagRepository::get_tags_for_book_ids(db, &book_ids),
+        }),
+        with_permit(
+            &limiter,
+            GenreRepository::get_genres_for_book_ids(db, &book_ids)
+        ),
+        with_permit(
+            &limiter,
+            TagRepository::get_tags_for_book_ids(db, &book_ids)
+        ),
     );
 
     // Handle errors

crates/codex-api/src/routes/v1/handlers/bulk.rs

@@ -847,7 +847,10 @@ pub async fn bulk_reset_series_metadata(
             _ => continue,
         };
 
-        // Clear all associated data
+        // Clear all associated data, bounded so this does not hold one pool
+        // connection per query at once.
+        let limiter = crate::db_batch::fan_out_limiter(crate::db_batch::configured_fan_out());
+        use crate::db_batch::with_permit;
         let (
             genres_res,
             tags_res,
@@ -858,14 +861,38 @@ pub async fn bulk_reset_series_metadata(
             covers_res,
             sharing_res,
         ) = tokio::join!(
-            GenreRepository::set_genres_for_series(&state.db, *series_id, vec![]),
-            TagRepository::set_tags_for_series(&state.db, *series_id, vec![]),
-            AlternateTitleRepository::delete_all_for_series(&state.db, *series_id, None),
-            SeriesExternalIdRepository::delete_all_for_series(&state.db, *series_id),
-            ExternalRatingRepository::delete_all_for_series(&state.db, *series_id),
-            ExternalLinkRepository::delete_all_for_series(&state.db, *series_id),
-            SeriesCoversRepository::delete_by_series(&state.db, *series_id),
-            SharingTagRepository::set_tags_for_series(&state.db, *series_id, vec![]),
+            with_permit(
+                &limiter,
+                GenreRepository::set_genres_for_series(&state.db, *series_id, vec![])
+            ),
+            with_permit(
+                &limiter,
+                TagRepository::set_tags_for_series(&state.db, *series_id, vec![])
+            ),
+            with_permit(
+                &limiter,
+                AlternateTitleRepository::delete_all_for_series(&state.db, *series_id, None)
+            ),
+            with_permit(
+                &limiter,
+                SeriesExternalIdRepository::delete_all_for_series(&state.db, *series_id)
+            ),
+            with_permit(
+                &limiter,
+                ExternalRatingRepository::delete_all_for_series(&state.db, *series_id)
+            ),
+            with_permit(
+                &limiter,
+                ExternalLinkRepository::delete_all_for_series(&state.db, *series_id)
+            ),
+            with_permit(
+                &limiter,
+                SeriesCoversRepository::delete_by_series(&state.db, *series_id)
+            ),
+            with_permit(
+                &limiter,
+                SharingTagRepository::set_tags_for_series(&state.db, *series_id, vec![])
+            ),
         );
 
         // Log errors but continue with other series