Release DB connections across slow I/O; fail fast on pool exhaustion

crates/nvisy-postgres/src/client/pg_config.rs

@@ -42,12 +42,18 @@ pub struct PgConfig {
     )]
     pub postgres_max_connections: u32,
 
-    /// Connection timeout (optional).
+    /// Maximum time to wait for a connection from the pool before failing.
+    ///
+    /// A finite bound is important: with no timeout a request blocks until a
+    /// connection frees, so under pool exhaustion requests pile up until the
+    /// request timeout kills them with a 500. A bound below the request timeout
+    /// turns exhaustion into a prompt, retryable failure instead.
     #[cfg_attr(
         feature = "cli",
         arg(
             long = "postgres-connection-timeout",
             env = "POSTGRES_CONNECTION_TIMEOUT",
+            default_value = "10s",
             value_parser = humantime::parse_duration,
         )
     )]
@@ -89,7 +95,7 @@ impl PgConfig {
         let this = Self {
             postgres_url: database_url.into(),
             postgres_max_connections: 10,
-            postgres_connection_timeout: None,
+            postgres_connection_timeout: Some(Duration::from_secs(10)),
             postgres_idle_timeout: None,
         };
 
@@ -293,7 +299,12 @@ mod tests {
         let config = PgConfig::new("postgresql://user:pass@localhost/db");
         assert_eq!(config.postgres_url, "postgresql://user:pass@localhost/db");
         assert_eq!(config.postgres_max_connections, 10);
-        assert_eq!(config.postgres_connection_timeout, None);
+        // A finite acquire timeout is the default so pool exhaustion fails fast
+        // (a retryable 503) instead of hanging until the request timeout.
+        assert_eq!(
+            config.postgres_connection_timeout,
+            Some(Duration::from_secs(10))
+        );
     }
 
     #[test]
@@ -312,9 +323,13 @@ mod tests {
     }
 
     #[test]
-    fn test_no_timeout() {
+    fn test_default_timeouts() {
         let config = PgConfig::new("postgresql://localhost/db");
-        assert_eq!(config.postgres_connection_timeout, None);
+        // Finite acquire timeout by default; no idle timeout (connections are kept).
+        assert_eq!(
+            config.postgres_connection_timeout,
+            Some(Duration::from_secs(10))
+        );
         assert_eq!(config.postgres_idle_timeout, None);
     }
 

crates/nvisy-postgres/src/error.rs

@@ -5,7 +5,7 @@
 
 use std::borrow::Cow;
 
-use deadpool::managed::TimeoutType;
+pub use deadpool::managed::TimeoutType;
 use diesel::result::{ConnectionError, Error};
 use diesel_async::pooled_connection::PoolError as DieselPoolError;
 use diesel_async::pooled_connection::deadpool::PoolError as DeadpoolError;

crates/nvisy-postgres/src/lib.rs

@@ -37,4 +37,4 @@ pub use crate::client::{
     ConnectionPool, MigrationResult, MigrationStatus, PgClient, PgClientMigrationExt, PgConfig,
     PgConn, PgPoolStatus,
 };
-pub use crate::error::{DieselError, PgError, PgResult};
+pub use crate::error::{DieselError, PgError, PgResult, TimeoutType};

crates/nvisy-server/src/handler/accounts.rs

@@ -250,11 +250,18 @@ async fn upload_account_avatar(
 ) -> Result<(StatusCode, Json<Account>)> {
     tracing::debug!(target: TRACING_TARGET, "Uploading account avatar");
 
-    let mut conn = pg_client.get_connection().await?;
-    let account = find_account(&mut conn, auth_claims.account_id).await?;
-    authorize_self(&account, &path_params.username)?;
+    // Authorize under a scoped connection, then release it: `set_account_avatar`
+    // does image processing and a NATS put (and acquires its own connection for
+    // the DB update), so holding this one across it would pin two pooled
+    // connections for the whole upload.
+    let account_id = {
+        let mut conn = pg_client.get_connection().await?;
+        let account = find_account(&mut conn, auth_claims.account_id).await?;
+        authorize_self(&account, &path_params.username)?;
+        account.id
+    };
 
-    let updated = avatar.set_account_avatar(account.id, bytes).await?;
+    let updated = avatar.set_account_avatar(account_id, bytes).await?;
 
     tracing::info!(target: TRACING_TARGET, "Account avatar set");
     Ok((StatusCode::OK, Json(Account::from_model(updated))))

crates/nvisy-server/src/handler/detection_audits.rs

@@ -48,18 +48,24 @@ async fn get_detection_analysis(
 ) -> Result<(StatusCode, Json<Audit>)> {
     tracing::debug!(target: TRACING_TARGET, "Getting detection analysis");
 
-    let mut conn = pg_client.get_connection().await?;
+    // Resolve the detection and its audit file row under a scoped connection, then
+    // release it before the object-store load below so the pooled connection is
+    // not held across the NATS round-trip.
+    let audit_file = {
+        let mut conn = pg_client.get_connection().await?;
 
-    auth_state
-        .authorize_workspace(&mut conn, workspace.id, Permission::ViewPipelines)
-        .await?;
+        auth_state
+            .authorize_workspace(&mut conn, workspace.id, Permission::ViewPipelines)
+            .await?;
 
-    let (detection, _pipeline) =
-        find_detection(&mut conn, workspace.id, path_params.detection_id.as_uuid()).await?;
+        let (detection, _pipeline) =
+            find_detection(&mut conn, workspace.id, path_params.detection_id.as_uuid()).await?;
 
-    let analyzed = blob
-        .load_analyzed_document(&mut conn, &engine, workspace.id, &detection)
-        .await?;
+        blob.resolve_audit_file(&mut conn, workspace.id, &detection)
+            .await?
+    };
+
+    let analyzed = blob.load_audit(&engine, workspace.id, &audit_file).await?;
 
     tracing::debug!(target: TRACING_TARGET, "Detection analysis retrieved");
 
@@ -104,16 +110,24 @@ async fn download_detection_audit(
 ) -> Result<(StatusCode, HeaderMap, Body)> {
     tracing::debug!(target: TRACING_TARGET, "Downloading detection audit");
 
-    let mut conn = pg_client.get_connection().await?;
-    auth_state
-        .authorize_workspace(&mut conn, workspace.id, Permission::ViewPipelines)
-        .await?;
+    // Resolve the detection and its audit file row under a scoped connection, then
+    // release it before the object-store load below so the pooled connection is
+    // not held across the NATS round-trip.
+    let (detection_id, audit_file) = {
+        let mut conn = pg_client.get_connection().await?;
+        auth_state
+            .authorize_workspace(&mut conn, workspace.id, Permission::ViewPipelines)
+            .await?;
+
+        let (detection, _pipeline) =
+            find_detection(&mut conn, workspace.id, path_params.detection_id.as_uuid()).await?;
+        let audit_file = blob
+            .resolve_audit_file(&mut conn, workspace.id, &detection)
+            .await?;
+        (detection.id, audit_file)
+    };
 
-    let (detection, _pipeline) =
-        find_detection(&mut conn, workspace.id, path_params.detection_id.as_uuid()).await?;
-    let audit = blob
-        .load_analyzed_document(&mut conn, &engine, workspace.id, &detection)
-        .await?;
+    let audit = blob.load_audit(&engine, workspace.id, &audit_file).await?;
 
     let (content_type, filename, body) = match query.format {
         ExportFormat::Json => {
@@ -125,15 +139,15 @@ async fn download_detection_audit(
             })?;
             (
                 "application/json",
-                format!("audit-{}.json", detection.id),
+                format!("audit-{detection_id}.json"),
                 buffer,
             )
         }
         ExportFormat::Csv => {
             let archive = build_audit_csv_zip(&audit)?;
             (
                 "application/zip",
-                format!("audit-{}.csv.zip", detection.id),
+                format!("audit-{detection_id}.csv.zip"),
                 archive,
             )
         }

crates/nvisy-server/src/handler/detections.rs

@@ -578,48 +578,63 @@ async fn redact_detection(
 ) -> Result<(StatusCode, Json<RedactionResult>)> {
     tracing::debug!(target: TRACING_TARGET, "Redacting detection");
 
-    let mut conn = pg_client.get_connection().await?;
+    // Phase 1: the pre-flight DB work under one connection, then release it.
+    // Holding a pooled connection across the audit load, redaction inference, and
+    // object I/O below would pin it for many seconds and starve the pool under
+    // load, so this scope drops the connection before that slow work begins. Only
+    // the audit file row is resolved here; its bytes are loaded in phase 2.
+    let (detection, pipeline, file, audit_file, policies) = {
+        let mut conn = pg_client.get_connection().await?;
+
+        auth_state
+            .authorize_workspace(&mut conn, workspace.id, Permission::RunPipelines)
+            .await?;
 
-    auth_state
-        .authorize_workspace(&mut conn, workspace.id, Permission::RunPipelines)
-        .await?;
+        let (detection, pipeline) =
+            find_detection(&mut conn, workspace.id, path_params.detection_id.as_uuid()).await?;
 
-    let (detection, pipeline) =
-        find_detection(&mut conn, workspace.id, path_params.detection_id.as_uuid()).await?;
+        // A detection can only be redacted once its analysis is complete.
+        if !detection.is_complete() {
+            return Err(ErrorKind::Conflict
+                .with_message("Detection is not ready to redact")
+                .with_resource("detection"));
+        }
 
-    // A detection can only be redacted once its analysis is complete.
-    if !detection.is_complete() {
-        return Err(ErrorKind::Conflict
-            .with_message("Detection is not ready to redact")
-            .with_resource("detection"));
-    }
+        // The source document is normally held back from retention while the
+        // detection is unfinished (see files_due_for_expiry), so this is reachable
+        // only if the input was explicitly deleted; surface a message that names
+        // the cause.
+        let file = conn
+            .find_file_in_workspace(workspace.id, detection.input_file_id)
+            .await?
+            .ok_or_else(|| {
+                ErrorKind::Conflict
+                    .with_message("The detection's source document is no longer available")
+                    .with_resource("detection")
+            })?;
+
+        let audit_file = blob
+            .resolve_audit_file(&mut conn, workspace.id, &detection)
+            .await?;
+        let policies = resolve_policies(&mut conn, &crypto, workspace.id, pipeline.id).await?;
 
-    // The source document is normally held back from retention while the
-    // detection is unfinished (see files_due_for_expiry), so this is reachable
-    // only if the input was explicitly deleted; surface a message that names the
-    // cause.
-    let file = conn
-        .find_file_in_workspace(workspace.id, detection.input_file_id)
-        .await?
-        .ok_or_else(|| {
-            ErrorKind::Conflict
-                .with_message("The detection's source document is no longer available")
-                .with_resource("detection")
-        })?;
+        (detection, pipeline, file, audit_file, policies)
+    };
+
+    // Phase 2: the slow work — loading the analysis, applying reviewer edits, the
+    // redaction inference, and staging the produced objects — runs with no DB
+    // connection held.
 
     // The stored detection analysis is loaded into a working audit and never
     // mutated on disk: reviewer edits and the redaction outcome land on this
     // clone, which is persisted as the redaction's own review audit, leaving the
     // detection analysis immutable and re-redactable.
-    let mut reviewed = blob
-        .load_analyzed_document(&mut conn, &engine, workspace.id, &detection)
-        .await?;
-    let policies = resolve_policies(&mut conn, &crypto, workspace.id, pipeline.id).await?;
+    let mut reviewed = blob.load_audit(&engine, workspace.id, &audit_file).await?;
 
-    // Layer the reviewer's edits onto the working audit's report before
-    // redaction. Validation is report-relative (an unknown target or a
-    // self-contradiction → 400, via `EditError`'s `From` impl) so a reviewer is
-    // never told a decision took effect when the document says otherwise.
+    // Layer the reviewer's edits onto the working audit's report before redaction.
+    // Validation is report-relative (an unknown target or a self-contradiction →
+    // 400, via `EditError`'s `From` impl) so a reviewer is never told a decision
+    // took effect when the document says otherwise.
     if let Some(edits) = &request.edits {
         edits.validate(&reviewed.report)?;
         edits.apply(&mut reviewed.report);
@@ -661,6 +676,9 @@ async fn redact_detection(
         }
     };
 
+    // Phase 3: re-acquire a connection only for the final commit, so the pool was
+    // free during the inference and staging above.
+    let mut conn = pg_client.get_connection().await?;
     let redaction = conn
         .transaction(async |conn| {
             let output_file = conn.create_workspace_file(staged_output.clone()).await?;

crates/nvisy-server/src/handler/error/http_error.rs

@@ -279,6 +279,8 @@ pub enum ErrorKind {
     InternalServerError,
     /// 501 Not Implemented - Feature not yet implemented
     NotImplemented,
+    /// 503 Service Unavailable - The server is temporarily overloaded
+    ServiceUnavailable,
 }
 
 impl ErrorKind {
@@ -342,6 +344,7 @@ impl ErrorKind {
             Self::TooManyRequests => ErrorResponse::TOO_MANY_REQUESTS,
             Self::InternalServerError => ErrorResponse::INTERNAL_SERVER_ERROR,
             Self::NotImplemented => ErrorResponse::NOT_IMPLEMENTED,
+            Self::ServiceUnavailable => ErrorResponse::SERVICE_UNAVAILABLE,
         }
     }
 }

crates/nvisy-server/src/handler/error/pg_error.rs

@@ -6,8 +6,8 @@
 //!
 //! All conversions are implemented via the `From` trait for ergonomic usage.
 
-use nvisy_postgres::PgError;
 use nvisy_postgres::types::ConstraintViolation;
+use nvisy_postgres::{PgError, TimeoutType};
 
 use crate::handler::{Error, ErrorKind};
 
@@ -55,7 +55,16 @@ impl From<PgError> for Error<'static> {
                     timeout = ?timeout,
                     "database timeout",
                 );
-                ErrorKind::InternalServerError.into_error()
+                // A wait timeout means the pool was saturated — the service is
+                // momentarily overloaded, not broken — so surface a retryable 503
+                // rather than a 500. A create/recycle timeout is a backend fault.
+                match timeout {
+                    TimeoutType::Wait => ErrorKind::ServiceUnavailable
+                        .with_message("The server is busy; retry shortly"),
+                    TimeoutType::Create | TimeoutType::Recycle => {
+                        ErrorKind::InternalServerError.into_error()
+                    }
+                }
             }
             PgError::Connection(connection_error) => {
                 tracing::error!(

crates/nvisy-server/src/handler/files.rs

@@ -807,13 +807,30 @@ async fn bulk_delete_files(
         .filter(|id| !deleted_ids.contains(id))
         .collect();
 
+    // Release the batch connection before purging: each purge does a NATS delete
+    // and re-acquires a short-lived connection of its own, so one connection is
+    // never pinned across the whole (potentially long) sequence of object deletes.
+    drop(conn);
+
     // Purge each object, reclaiming storage the same way retention expiry does.
     // The soft-deletes already committed above; `purge_file` re-runs each
     // idempotently, then removes the object. The deletion is the committed result,
     // so a purge failure must not fail the response (that would make a retry see
     // these ids as already-gone `skipped`): log it and leave the object for the
     // reaper to reclaim.
     for file in &files {
+        let mut conn = match pg_client.get_connection().await {
+            Ok(conn) => conn,
+            Err(err) => {
+                tracing::error!(
+                    target: TRACING_TARGET,
+                    file_id = %file.id,
+                    error = %err,
+                    "Failed to get connection to purge a bulk-deleted file's object; left for the reaper",
+                );
+                continue;
+            }
+        };
         if let Err(err) = blob
             .purge_file(&mut conn, file.id, &file.storage_path, &file.storage_bucket)
             .await

crates/nvisy-server/src/handler/redactions.rs

@@ -107,18 +107,24 @@ async fn get_redaction_review(
 ) -> Result<(StatusCode, Json<Audit>)> {
     tracing::debug!(target: TRACING_TARGET, "Getting redaction review audit");
 
-    let mut conn = pg_client.get_connection().await?;
+    // Resolve the redaction and its review audit file row under a scoped
+    // connection, then release it before the object-store load so the pooled
+    // connection is not held across the NATS round-trip.
+    let review_file = {
+        let mut conn = pg_client.get_connection().await?;
 
-    auth_state
-        .authorize_workspace(&mut conn, workspace.id, Permission::ViewPipelines)
-        .await?;
+        auth_state
+            .authorize_workspace(&mut conn, workspace.id, Permission::ViewPipelines)
+            .await?;
 
-    let redaction =
-        find_redaction(&mut conn, workspace.id, path_params.redaction_id.as_uuid()).await?;
+        let redaction =
+            find_redaction(&mut conn, workspace.id, path_params.redaction_id.as_uuid()).await?;
 
-    let review = blob
-        .load_review_audit(&mut conn, &engine, workspace.id, redaction.review_file_id)
-        .await?;
+        blob.resolve_review_file(&mut conn, workspace.id, redaction.review_file_id)
+            .await?
+    };
+
+    let review = blob.load_audit(&engine, workspace.id, &review_file).await?;
 
     Ok((StatusCode::OK, Json(review)))
 }