From 277c7c8ef8d15cef405175654c162b133e31d858 Mon Sep 17 00:00:00 2001 From: Oleh Martsokha Date: Mon, 31 Aug 2026 20:02:06 +0200 Subject: [PATCH 1/2] Release DB connections across slow I/O; fail fast on pool exhaustion MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A stress test with a single client exhausted the connection pool and cascaded to 500s. The root cause was handlers and the detection worker holding a pooled connection across slow non-DB work (LLM inference, NATS object I/O, image processing) — with a 10-connection pool and no acquire timeout, a few concurrent requests pinned every connection and later requests hung until the request timeout killed them. Release the connection across the slow phase everywhere it was held: - redact_detection: pre-flight DB work (auth, find detection/file, resolve the audit file row, resolve policies) under a scoped connection, dropped before the audit load, redaction inference, and object staging; re-acquired only for the commit transaction. - DetectionWorker::detect: manages its own connection per phase — reads under a connection, drops it across build/analyze/stage, re-acquires for the fenced finalize. - Avatar upload (account + workspace): authorize under a scoped connection, release before image processing and the NATS put. - Detection-audit read + redaction-review handlers: resolve the audit file row under a connection, release before the object-store load. To support this, split RunBlobStore's audit loading into a connection-bound resolve step (resolve_audit_file / resolve_review_file) and a connection-free load step (load_audit), deduplicating the object load/decode that was copied across two methods and removing the now-unused load_analyzed_document. Fail fast on pool exhaustion instead of hanging: - Default POSTGRES_CONNECTION_TIMEOUT to 10s (was unset = wait forever), below the request timeout, so a starved request fails promptly. - Map a pool wait timeout to a retryable 503 Service Unavailable (new ErrorKind::ServiceUnavailable) rather than a 500; a backend create/recycle timeout stays a 500. Rate limiting is intentionally left to the edge/infra (see issues); the server's job here is to hold shared resources briefly and degrade gracefully. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_018bKk1YEG4tZ69jzYVQvQL8 --- crates/nvisy-postgres/src/client/pg_config.rs | 25 ++++- crates/nvisy-postgres/src/error.rs | 2 +- crates/nvisy-postgres/src/lib.rs | 2 +- crates/nvisy-server/src/handler/accounts.rs | 15 ++- .../src/handler/detection_audits.rs | 54 +++++---- crates/nvisy-server/src/handler/detections.rs | 82 ++++++++------ .../src/handler/error/http_error.rs | 3 + .../src/handler/error/pg_error.rs | 13 ++- crates/nvisy-server/src/handler/redactions.rs | 24 ++-- crates/nvisy-server/src/handler/workspaces.rs | 14 ++- .../src/service/detection/worker.rs | 105 ++++++++++++------ .../src/service/run_blob_store.rs | 101 +++++++---------- 12 files changed, 264 insertions(+), 176 deletions(-) diff --git a/crates/nvisy-postgres/src/client/pg_config.rs b/crates/nvisy-postgres/src/client/pg_config.rs index 715be4c1..0fef0b7d 100644 --- a/crates/nvisy-postgres/src/client/pg_config.rs +++ b/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); } diff --git a/crates/nvisy-postgres/src/error.rs b/crates/nvisy-postgres/src/error.rs index 84cbe390..8566a8ca 100644 --- a/crates/nvisy-postgres/src/error.rs +++ b/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; diff --git a/crates/nvisy-postgres/src/lib.rs b/crates/nvisy-postgres/src/lib.rs index f27663ef..556bf28f 100644 --- a/crates/nvisy-postgres/src/lib.rs +++ b/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}; diff --git a/crates/nvisy-server/src/handler/accounts.rs b/crates/nvisy-server/src/handler/accounts.rs index aa919a01..cc2b3f11 100644 --- a/crates/nvisy-server/src/handler/accounts.rs +++ b/crates/nvisy-server/src/handler/accounts.rs @@ -250,11 +250,18 @@ async fn upload_account_avatar( ) -> Result<(StatusCode, Json)> { 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)))) diff --git a/crates/nvisy-server/src/handler/detection_audits.rs b/crates/nvisy-server/src/handler/detection_audits.rs index a9e23790..43962dba 100644 --- a/crates/nvisy-server/src/handler/detection_audits.rs +++ b/crates/nvisy-server/src/handler/detection_audits.rs @@ -48,18 +48,24 @@ async fn get_detection_analysis( ) -> Result<(StatusCode, Json)> { 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,7 +139,7 @@ async fn download_detection_audit( })?; ( "application/json", - format!("audit-{}.json", detection.id), + format!("audit-{detection_id}.json"), buffer, ) } @@ -133,7 +147,7 @@ async fn download_detection_audit( let archive = build_audit_csv_zip(&audit)?; ( "application/zip", - format!("audit-{}.csv.zip", detection.id), + format!("audit-{detection_id}.csv.zip"), archive, ) } diff --git a/crates/nvisy-server/src/handler/detections.rs b/crates/nvisy-server/src/handler/detections.rs index 850cb51a..cac90773 100644 --- a/crates/nvisy-server/src/handler/detections.rs +++ b/crates/nvisy-server/src/handler/detections.rs @@ -578,48 +578,63 @@ async fn redact_detection( ) -> Result<(StatusCode, Json)> { 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?; diff --git a/crates/nvisy-server/src/handler/error/http_error.rs b/crates/nvisy-server/src/handler/error/http_error.rs index a589a6fc..da9b2891 100644 --- a/crates/nvisy-server/src/handler/error/http_error.rs +++ b/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, } } } diff --git a/crates/nvisy-server/src/handler/error/pg_error.rs b/crates/nvisy-server/src/handler/error/pg_error.rs index 36f7665b..79b4013a 100644 --- a/crates/nvisy-server/src/handler/error/pg_error.rs +++ b/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 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!( diff --git a/crates/nvisy-server/src/handler/redactions.rs b/crates/nvisy-server/src/handler/redactions.rs index 41880e4a..d947beff 100644 --- a/crates/nvisy-server/src/handler/redactions.rs +++ b/crates/nvisy-server/src/handler/redactions.rs @@ -107,18 +107,24 @@ async fn get_redaction_review( ) -> Result<(StatusCode, Json)> { 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))) } diff --git a/crates/nvisy-server/src/handler/workspaces.rs b/crates/nvisy-server/src/handler/workspaces.rs index 88e1bbfa..7162f7c2 100644 --- a/crates/nvisy-server/src/handler/workspaces.rs +++ b/crates/nvisy-server/src/handler/workspaces.rs @@ -469,10 +469,16 @@ async fn upload_workspace_avatar( ) -> Result { tracing::debug!(target: TRACING_TARGET, "Uploading workspace avatar"); - let mut conn = pg_client.get_connection().await?; - auth_state - .authorize_workspace(&mut conn, workspace.id, Permission::UpdateWorkspace) - .await?; + // Authorize under a scoped connection, then release it: `set_workspace_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 mut conn = pg_client.get_connection().await?; + auth_state + .authorize_workspace(&mut conn, workspace.id, Permission::UpdateWorkspace) + .await?; + } avatar.set_workspace_avatar(workspace.id, bytes).await?; diff --git a/crates/nvisy-server/src/service/detection/worker.rs b/crates/nvisy-server/src/service/detection/worker.rs index 381de056..0781b3cf 100644 --- a/crates/nvisy-server/src/service/detection/worker.rs +++ b/crates/nvisy-server/src/service/detection/worker.rs @@ -17,7 +17,7 @@ use nvisy_postgres::query::{ WorkspaceRepository, }; use nvisy_postgres::types::{DetectionStatus, Json, RasterPolicy, WorkspaceSettings}; -use nvisy_postgres::{AsyncConnection, DieselError, PgConn, PgError}; +use nvisy_postgres::{AsyncConnection, DieselError, PgError}; use tokio_util::sync::CancellationToken; use super::job::DetectionJob; @@ -209,11 +209,24 @@ impl DetectionWorker { .broadcast_status(detection.id, DetectionStatus::Executing) .await; - if let Err(err) = self - .detect(&mut conn, &job, &claimed, &pipeline, claim_token) - .await - { + // Release the connection before analysis: `detect` manages its own + // connections across its phases, so holding this one across the (slow) + // inference would pin a pooled connection per in-flight job and starve the + // pool. It is re-acquired below only if the detection fails. + drop(conn); + + if let Err(err) = self.detect(&job, &claimed, &pipeline, claim_token).await { tracing::warn!(target: TRACING_TARGET, error = %err, "Detection failed"); + let mut conn = match self.infra.postgres.get_connection().await { + Ok(conn) => conn, + Err(err) => { + // No connection to persist the failure: the detection stays + // `Executing` with no queued job, so redeliver to drive it to a + // terminal state on a later attempt. + tracing::error!(target: TRACING_TARGET, error = %err, "Failed to get connection to fail detection"); + return JobOutcome::Retry; + } + }; let outcome = fail_detection( &mut conn, &self.detection, @@ -253,46 +266,65 @@ impl DetectionWorker { } /// Performs the analysis and records the detection as `Complete`. + /// + /// Manages its own connection lifecycle in three phases so a pooled + /// connection is never held across the analysis inference: phase 1 reads the + /// inputs under a connection and releases it, phase 2 runs the (slow) document + /// build, analysis, and audit staging with no connection held, and phase 3 + /// re-acquires a connection only for the finalize transaction. async fn detect( &self, - conn: &mut PgConn, job: &DetectionJob, detection: &WorkspaceDetection, pipeline: &WorkspacePipeline, claim_token: jiff::Timestamp, ) -> Result<()> { - let workspace = conn - .find_workspace_by_id(job.workspace_id) - .await? - .ok_or_else(|| ErrorKind::NotFound.with_message("Workspace not found"))?; - let file = conn - .find_file_in_workspace(job.workspace_id, detection.input_file_id) - .await? - .ok_or_else(|| ErrorKind::NotFound.with_message("Input file not found"))?; - - let definition = PipelineDefinition::from_parts(pipeline.definition.clone(), Vec::new()) - .map_err(|err| { - ErrorKind::InternalServerError - .with_message("Failed to decode pipeline definition") - .with_context(err.to_string()) - })?; - - // Parse the workspace settings once; both raster mode and retention read it. - let settings = workspace.settings.or_default(); - let request = - self.engine - .request_context(&definition, job.scope.clone(), raster_mode_of(&settings)); + // Phase 1: read the inputs under a connection, then drop it. + let (file, request, policies, settings) = { + let mut conn = self.infra.postgres.get_connection().await?; + + let workspace = conn + .find_workspace_by_id(job.workspace_id) + .await? + .ok_or_else(|| ErrorKind::NotFound.with_message("Workspace not found"))?; + let file = conn + .find_file_in_workspace(job.workspace_id, detection.input_file_id) + .await? + .ok_or_else(|| ErrorKind::NotFound.with_message("Input file not found"))?; + + let definition = + PipelineDefinition::from_parts(pipeline.definition.clone(), Vec::new()).map_err( + |err| { + ErrorKind::InternalServerError + .with_message("Failed to decode pipeline definition") + .with_context(err.to_string()) + }, + )?; + + // Parse the workspace settings once; both raster mode and retention + // read it. + let settings = workspace.settings.or_default(); + let request = self.engine.request_context( + &definition, + job.scope.clone(), + raster_mode_of(&settings), + ); - let document = self.blob.build_document(&file, detection.id).await?; + let policies = + resolve_policies(&mut conn, &self.infra.crypto, job.workspace_id, pipeline.id) + .await?; + if policies.is_empty() { + return Err(ErrorKind::BadRequest + .with_message("Pipeline has no policies") + .with_resource("pipeline")); + } - let policies = - resolve_policies(conn, &self.infra.crypto, job.workspace_id, pipeline.id).await?; - if policies.is_empty() { - return Err(ErrorKind::BadRequest - .with_message("Pipeline has no policies") - .with_resource("pipeline")); - } + (file, request, policies, settings) + }; + // Phase 2: the slow work — document build, analysis inference, and audit + // staging — runs with no DB connection held. + let document = self.blob.build_document(&file, detection.id).await?; let analyzed = self.engine.analyze(document, &policies, &request).await?; // Write the (non-transactional) audit object first, then commit its file @@ -350,6 +382,9 @@ impl DetectionWorker { &completed_event, )?; + // Phase 3: re-acquire a connection only for the fenced finalize + // transaction, so the pool was free during the analysis above. + let mut conn = self.infra.postgres.get_connection().await?; let staged_audit = audit_file.clone(); let finalized = conn .transaction(async |conn| { diff --git a/crates/nvisy-server/src/service/run_blob_store.rs b/crates/nvisy-server/src/service/run_blob_store.rs index 75011d89..45c386e2 100644 --- a/crates/nvisy-server/src/service/run_blob_store.rs +++ b/crates/nvisy-server/src/service/run_blob_store.rs @@ -256,22 +256,18 @@ impl RunBlobStore { .await } - /// Fetches and decrypts a run's stored [`Audit`]. + /// Resolves the `workspace_files` row holding a detection's analysis blob. /// - /// The `engine` reconstructs the audit's report from its serialized form: an - /// [`Audit`] serializes but does not `Deserialize`, since its report tags each - /// entity group by modality name and only the engine's registry can map those - /// back to concrete types. - /// - /// Errors if the detection never analyzed (409) or its analysis has since - /// been deleted (404). - pub async fn load_analyzed_document( + /// This is the only connection-bound step of loading an analysis, so a caller + /// can resolve the row under a connection, release it, and then load the + /// object with [`load_audit`](Self::load_audit) — keeping the pooled + /// connection off the object-store round-trip. + pub async fn resolve_audit_file( &self, conn: &mut PgConn, - engine: &Engine, workspace_id: Uuid, detection: &WorkspaceDetection, - ) -> Result { + ) -> Result { // A NULL reference means the detection never produced an analysis; a // reference to a now-deleted file means it did, but the analysis has been // removed. These are distinct states, so they map to distinct responses. @@ -280,29 +276,44 @@ impl RunBlobStore { .with_message("Detection has no analysis yet") .with_resource("detection") })?; - let audit_file = conn - .find_file_in_workspace(workspace_id, audit_file_id) + conn.find_file_in_workspace(workspace_id, audit_file_id) .await? .ok_or_else(|| { ErrorKind::NotFound .with_message("The analysis for this detection has been deleted") .with_resource("detection") - })?; + }) + } + + /// Loads and decodes a detection's analysis blob from its already-resolved + /// audit file row. Holds no database connection: only object-store I/O and + /// decryption, so a caller can run it after releasing its connection. + /// + /// The `engine` reconstructs the audit's report from its serialized form: an + /// [`Audit`] serializes but does not `Deserialize`, since its report tags each + /// entity group by modality name and only the engine's registry can map those + /// back to concrete types. + pub async fn load_audit( + &self, + engine: &Engine, + workspace_id: Uuid, + audit_file: &WorkspaceFile, + ) -> Result { let key = AuditKey::from_str(&audit_file.storage_path).map_err(|err| { ErrorKind::InternalServerError - .with_message("Invalid analysis storage key") + .with_message("Invalid audit storage key") .with_context(err.to_string()) })?; let store = self.infra.nats.object_store::().await?; let data = store.get(&key).await?.ok_or_else(|| { - ErrorKind::InternalServerError.with_message("Analysis is missing from storage") + ErrorKind::InternalServerError.with_message("Audit is missing from storage") })?; let mut reader = data.into_reader(); let mut ciphertext = Vec::new(); reader.read_to_end(&mut ciphertext).await.map_err(|err| { ErrorKind::InternalServerError - .with_message("Failed to read analysis") + .with_message("Failed to read audit") .with_context(err.to_string()) })?; @@ -312,14 +323,14 @@ impl RunBlobStore { .decrypt(workspace_id, &ciphertext) .map_err(|err| { ErrorKind::InternalServerError - .with_message("Failed to decrypt analysis") + .with_message("Failed to decrypt audit") .with_context(err.to_string()) })?; engine .deserialize_audit(&mut serde_json::Deserializer::from_slice(&plaintext)) .map_err(|err| { ErrorKind::InternalServerError - .with_message("Failed to decode analysis") + .with_message("Failed to decode audit") .with_context(err.to_string()) }) } @@ -441,65 +452,29 @@ impl RunBlobStore { }) } - /// Fetches and decrypts a redaction's stored review [`Audit`] by its file id. + /// Resolves the `workspace_files` row holding a redaction's review audit blob. /// - /// The review counterpart to - /// [`load_analyzed_document`](Self::load_analyzed_document); the `engine` - /// rebuilds the report from its serialized form. Errors if the redaction has - /// no review audit (409) or it has since been deleted (404). - pub async fn load_review_audit( + /// The connection-bound step of loading a review audit; pair with + /// [`load_audit`](Self::load_audit) to release the connection before the + /// object-store round-trip. Errors if the redaction has no review audit (409) + /// or it has since been deleted (404). + pub async fn resolve_review_file( &self, conn: &mut PgConn, - engine: &Engine, workspace_id: Uuid, review_file_id: Option, - ) -> Result { + ) -> Result { let review_file_id = review_file_id.ok_or_else(|| { ErrorKind::Conflict .with_message("Redaction has no review audit") .with_resource("redaction") })?; - let review_file = conn - .find_file_in_workspace(workspace_id, review_file_id) + conn.find_file_in_workspace(workspace_id, review_file_id) .await? .ok_or_else(|| { ErrorKind::NotFound .with_message("The review audit for this redaction has been deleted") .with_resource("redaction") - })?; - let key = AuditKey::from_str(&review_file.storage_path).map_err(|err| { - ErrorKind::InternalServerError - .with_message("Invalid review audit storage key") - .with_context(err.to_string()) - })?; - - let store = self.infra.nats.object_store::().await?; - let data = store.get(&key).await?.ok_or_else(|| { - ErrorKind::InternalServerError.with_message("Review audit is missing from storage") - })?; - let mut reader = data.into_reader(); - let mut ciphertext = Vec::new(); - reader.read_to_end(&mut ciphertext).await.map_err(|err| { - ErrorKind::InternalServerError - .with_message("Failed to read review audit") - .with_context(err.to_string()) - })?; - - let plaintext = self - .infra - .crypto - .decrypt(workspace_id, &ciphertext) - .map_err(|err| { - ErrorKind::InternalServerError - .with_message("Failed to decrypt review audit") - .with_context(err.to_string()) - })?; - engine - .deserialize_audit(&mut serde_json::Deserializer::from_slice(&plaintext)) - .map_err(|err| { - ErrorKind::InternalServerError - .with_message("Failed to decode review audit") - .with_context(err.to_string()) }) } } From f5198ff62ce2f883d66df4855c59ed452293ac56 Mon Sep 17 00:00:00 2001 From: Oleh Martsokha Date: Mon, 31 Aug 2026 20:09:09 +0200 Subject: [PATCH 2/2] Release the connection across the bulk-delete purge loop With #257 merged, the deferred sixth offender: bulk_delete_files held one pooled connection across N sequential object purges (each a NATS delete). Drop the batch connection after the commit and re-acquire a short-lived connection per purge, so one connection is never pinned across the whole delete sequence. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_018bKk1YEG4tZ69jzYVQvQL8 --- crates/nvisy-server/src/handler/files.rs | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/crates/nvisy-server/src/handler/files.rs b/crates/nvisy-server/src/handler/files.rs index 7cef23d4..32213400 100644 --- a/crates/nvisy-server/src/handler/files.rs +++ b/crates/nvisy-server/src/handler/files.rs @@ -807,6 +807,11 @@ 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, @@ -814,6 +819,18 @@ async fn bulk_delete_files( // 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