From e9e1618db3a64b0430d6e474a52f0ff5539f5a2e Mon Sep 17 00:00:00 2001 From: Oleh Martsokha Date: Sun, 30 Aug 2026 18:25:31 +0200 Subject: [PATCH 1/4] Add bulk file delete and configurable upload limits MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Bulk delete: POST /workspaces/{slug}/files/delete/ deletes several files in one call. It is idempotent — ids that resolve to live files in the workspace are removed and returned in `deleted`; ids that are unknown, already deleted, or in another workspace are returned in `skipped`. Each delete soft-deletes the row and records its FileDeleted event in one transaction, then purges the object best-effort, matching the single-file delete. Repurposes the dead find_workspace_files_by_ids query into a workspace-scoped find_files_in_workspace and drops the unused plural delete query. Upload limits, two layers: - Hard limit (server-wide, config). A new UploadConfig (MAX_BODY_BYTES / MAX_FILE_BODY_BYTES) replaces the hardcoded constants, driving the global request-body layer and the per-route upload limit. It is stored on ServiceState so handlers read it via State, with a max_file_bytes() accessor. - Soft cap (per workspace, DB). WorkspaceSettings gains max_upload_bytes (Option, no migration — settings are JSON). A new LimitedReader in the upload pipe aborts an oversized upload before its excess is encrypted and stored, returning 413 (new ErrorKind::PayloadTooLarge). The workspace response resolves maxUploadBytes to the effective per-file limit — min(soft ?? hard, hard) — so a client always reads one concrete number to enforce; the raw server limit is never exposed. Cleanups: rename OcrPolicy to RasterPolicy (Force to Always, field ocr to raster) to align with the engine's RasterMode; remove the legacy WorkspaceSettings::require_approval. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_018bKk1YEG4tZ69jzYVQvQL8 --- crates/nvisy-cli/src/config/middleware.rs | 13 +- crates/nvisy-cli/src/config/mod.rs | 1 + crates/nvisy-cli/src/main.rs | 5 +- .../src/query/workspace_file.rs | 57 ++----- crates/nvisy-postgres/src/types/json/mod.rs | 2 +- .../src/types/json/workspace_settings.rs | 63 +++---- crates/nvisy-postgres/src/types/mod.rs | 8 +- .../src/handler/error/http_error.rs | 3 + crates/nvisy-server/src/handler/files.rs | 161 ++++++++++++++++-- crates/nvisy-server/src/handler/mod.rs | 26 ++- .../nvisy-server/src/handler/request/files.rs | 15 ++ .../src/handler/response/files.rs | 15 ++ .../src/handler/response/workspaces.rs | 39 ++++- crates/nvisy-server/src/handler/workspaces.rs | 30 +++- crates/nvisy-server/src/middleware/mod.rs | 1 + .../nvisy-server/src/middleware/security.rs | 67 +++++++- .../src/service/crypto/limited_reader.rs | 125 ++++++++++++++ crates/nvisy-server/src/service/crypto/mod.rs | 2 + .../src/service/detection/worker.rs | 14 +- crates/nvisy-server/src/service/mod.rs | 9 +- 20 files changed, 518 insertions(+), 138 deletions(-) create mode 100644 crates/nvisy-server/src/service/crypto/limited_reader.rs diff --git a/crates/nvisy-cli/src/config/middleware.rs b/crates/nvisy-cli/src/config/middleware.rs index 45c7f5ea..4d4c7d5e 100644 --- a/crates/nvisy-cli/src/config/middleware.rs +++ b/crates/nvisy-cli/src/config/middleware.rs @@ -14,7 +14,7 @@ //! ``` use clap::Args; -use nvisy_server::middleware::{CorsConfig, OpenApiConfig, RecoveryConfig}; +use nvisy_server::middleware::{CorsConfig, OpenApiConfig, RecoveryConfig, UploadConfig}; use super::TRACING_TARGET_CONFIG; @@ -28,6 +28,10 @@ pub struct MiddlewareConfig { #[clap(flatten)] pub cors: CorsConfig, + /// Request body size limits. + #[clap(flatten)] + pub upload: UploadConfig, + /// OpenAPI documentation configuration. #[clap(flatten)] pub openapi: OpenApiConfig, @@ -47,6 +51,13 @@ impl MiddlewareConfig { "CORS configuration" ); + tracing::info!( + target: TRACING_TARGET_CONFIG, + max_body_bytes = self.upload.max_body_bytes, + max_file_body_bytes = self.upload.max_file_body_bytes, + "Upload configuration" + ); + tracing::info!( target: TRACING_TARGET_CONFIG, openapi_path = %self.openapi.open_api_json, diff --git a/crates/nvisy-cli/src/config/mod.rs b/crates/nvisy-cli/src/config/mod.rs index edff6ad3..316954dc 100644 --- a/crates/nvisy-cli/src/config/mod.rs +++ b/crates/nvisy-cli/src/config/mod.rs @@ -172,6 +172,7 @@ impl Cli { service.health, service.sync, webhook, + self.middleware.upload.clone(), ) .await?) } diff --git a/crates/nvisy-cli/src/main.rs b/crates/nvisy-cli/src/main.rs index 74e6bd65..c93fe64b 100644 --- a/crates/nvisy-cli/src/main.rs +++ b/crates/nvisy-cli/src/main.rs @@ -63,12 +63,13 @@ async fn run() -> anyhow::Result<()> { /// Creates the router with all middleware layers applied. fn create_router(state: ServiceState, middleware: &MiddlewareConfig) -> Router { - let api_routes = routes(CustomRoutes::new(), state.clone()).with_state(state); + let api_routes = + routes(CustomRoutes::new(), state.clone(), &middleware.upload).with_state(state); api_routes .with_open_api(&middleware.openapi) .with_metrics() - .with_security(&middleware.cors, &Default::default()) + .with_security(&middleware.cors, &middleware.upload, &Default::default()) .with_observability() .with_recovery(&middleware.recovery) } diff --git a/crates/nvisy-postgres/src/query/workspace_file.rs b/crates/nvisy-postgres/src/query/workspace_file.rs index c4c57ede..b7d10e00 100644 --- a/crates/nvisy-postgres/src/query/workspace_file.rs +++ b/crates/nvisy-postgres/src/query/workspace_file.rs @@ -164,15 +164,6 @@ pub trait WorkspaceFileRepository { fn delete_workspace_file(&mut self, file_id: Uuid) -> impl Future> + Send; - /// Soft deletes multiple workspace files by setting deletion timestamps. - /// - /// Returns the number of files deleted. - fn delete_workspace_files( - &mut self, - workspace_id: Uuid, - file_ids: &[Uuid], - ) -> impl Future> + Send; - /// Lists all files in a workspace with sorting and filtering options. /// /// Supports filtering by file format and sorting by name, date, or size. @@ -206,9 +197,14 @@ pub trait WorkspaceFileRepository { account_id: Uuid, ) -> impl Future> + Send; - /// Finds multiple workspace files by their IDs. - fn find_workspace_files_by_ids( + /// Finds the live files among `file_ids` that belong to `workspace_id`. + /// + /// Workspace-scoped so a caller can only resolve files in the workspace it + /// addressed; ids that are unknown, soft-deleted, or in another workspace are + /// simply absent from the result rather than an error. + fn find_files_in_workspace( &mut self, + workspace_id: Uuid, file_ids: &[Uuid], ) -> impl Future>> + Send; @@ -619,41 +615,6 @@ impl WorkspaceFileRepository for PgConnection { .await } - async fn delete_workspace_files( - &mut self, - workspace_id: Uuid, - file_ids: &[Uuid], - ) -> PgResult { - use diesel_async::AsyncConnection; - use schema::{workspace_file_imports, workspace_files}; - - let ids = file_ids.to_vec(); - self.transaction(async |conn| { - let count = diesel::update( - workspace_files::table - .filter(workspace_files::id.eq_any(&ids)) - .filter(workspace_files::workspace_id.eq(workspace_id)) - .filter(workspace_files::deleted_at.is_null()), - ) - .set(workspace_files::deleted_at.eq(diesel::dsl::now)) - .execute(conn) - .await - .map_err(PgError::from)?; - - // Drop import-origin rows so re-import is never blocked (see - // `delete_workspace_file`). - diesel::delete( - workspace_file_imports::table.filter(workspace_file_imports::file_id.eq_any(&ids)), - ) - .execute(conn) - .await - .map_err(PgError::from)?; - - Ok::<_, PgError>(count) - }) - .await - } - async fn offset_list_workspace_files( &mut self, workspace_id: Uuid, @@ -864,14 +825,16 @@ impl WorkspaceFileRepository for PgConnection { Ok(usage.unwrap_or_else(|| BigDecimal::from(0))) } - async fn find_workspace_files_by_ids( + async fn find_files_in_workspace( &mut self, + workspace_id: Uuid, file_ids: &[Uuid], ) -> PgResult> { use schema::workspace_files::{self, dsl}; let files = workspace_files::table .filter(dsl::id.eq_any(file_ids)) + .filter(dsl::workspace_id.eq(workspace_id)) .filter(dsl::deleted_at.is_null()) .select(WorkspaceFile::as_select()) .load(self) diff --git a/crates/nvisy-postgres/src/types/json/mod.rs b/crates/nvisy-postgres/src/types/json/mod.rs index f4ef1128..5a38f30b 100644 --- a/crates/nvisy-postgres/src/types/json/mod.rs +++ b/crates/nvisy-postgres/src/types/json/mod.rs @@ -30,4 +30,4 @@ pub use retention::{Retention, RetentionScope, RetentionSettings}; pub use typed_json::Json; pub use webhook_headers::{InvalidHeader, WebhookHeaders}; pub use workspace_metadata::WorkspaceMetadata; -pub use workspace_settings::{OcrPolicy, WorkspaceSettings}; +pub use workspace_settings::{RasterPolicy, WorkspaceSettings}; diff --git a/crates/nvisy-postgres/src/types/json/workspace_settings.rs b/crates/nvisy-postgres/src/types/json/workspace_settings.rs index a9919151..62ad177a 100644 --- a/crates/nvisy-postgres/src/types/json/workspace_settings.rs +++ b/crates/nvisy-postgres/src/types/json/workspace_settings.rs @@ -9,55 +9,44 @@ use serde::{Deserialize, Serialize}; use super::retention::RetentionSettings; -/// Whether processed files require approval before becoming visible, on by -/// default (the safe choice for a redaction workflow). -#[must_use] -const fn default_require_approval() -> bool { - true -} - -/// How a workspace's documents are turned into images for OCR during detection. +/// How a workspace's document pages are rasterised to images for OCR during +/// detection. /// -/// A workspace-level policy over the engine's per-run OCR mode: `Auto` lets the -/// engine decide from the text layer, `Force` always renders every page (for +/// A workspace-level policy over the engine's per-run raster mode: `Auto` lets +/// the engine decide from the text layer, `Always` renders every page (for /// documents with unreliable text layers — scans, watermarks), and `Never` /// relies on the text layer only. #[derive(Debug, Default, Clone, Copy, PartialEq, Eq)] #[cfg_attr(feature = "schema", derive(schemars::JsonSchema))] #[derive(Serialize, Deserialize)] #[serde(rename_all = "camelCase")] -pub enum OcrPolicy { - /// Extract the text layer where present and OCR only pages that lack it. +pub enum RasterPolicy { + /// Extract the text layer where present and rasterise only pages that lack it. #[default] Auto, - /// Always render every page to images for OCR, ignoring any text layer. - Force, - /// Rely on the text layer only; never render pages for OCR. + /// Always render every page to images, ignoring any text layer. + Always, + /// Rely on the text layer only; never rasterise pages. Never, } /// Typed workspace settings, the JSON stored in the `workspaces.settings` column. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)] #[cfg_attr(feature = "schema", derive(schemars::JsonSchema))] #[derive(Serialize, Deserialize)] #[serde(rename_all = "camelCase", default)] pub struct WorkspaceSettings { - /// Whether approval is required before processed files become visible. - pub require_approval: bool, - /// How documents are rendered for OCR during detection. - pub ocr: OcrPolicy, + /// How document pages are rasterised for OCR during detection. + pub raster: RasterPolicy, /// Data-retention rules for the workspace. pub retention: RetentionSettings, -} - -impl Default for WorkspaceSettings { - fn default() -> Self { - Self { - require_approval: default_require_approval(), - ocr: OcrPolicy::Auto, - retention: RetentionSettings::default(), - } - } + /// A soft per-file upload cap in bytes: an upload larger than this is + /// rejected for this workspace. `None` imposes no workspace-specific cap. + /// + /// The server-wide hard limit still applies regardless; the effective cap is + /// the smaller of the two. + #[serde(skip_serializing_if = "Option::is_none")] + pub max_upload_bytes: Option, } #[cfg(test)] @@ -75,31 +64,31 @@ mod tests { #[test] fn empty_settings_blob_is_default() { let settings = column(json!({})).or_default(); - assert!(settings.require_approval); + assert_eq!(settings.raster, RasterPolicy::Auto); assert!(settings.retention.is_noop()); } #[test] fn malformed_settings_blob_falls_back_to_default() { let settings = column(json!({ "retention": "nonsense" })).or_default(); - assert!(settings.require_approval); + assert_eq!(settings.raster, RasterPolicy::Auto); assert!(settings.retention.is_noop()); } #[test] - fn require_approval_survives_round_trip() { + fn raster_policy_defaults_to_auto_and_round_trips() { + assert_eq!(column(json!({})).or_default().raster, RasterPolicy::Auto); let settings = WorkspaceSettings { - require_approval: false, + raster: RasterPolicy::Always, ..Default::default() }; assert_eq!(Json::encode(&settings).or_default(), settings); } #[test] - fn ocr_policy_defaults_to_auto_and_round_trips() { - assert_eq!(column(json!({})).or_default().ocr, OcrPolicy::Auto); + fn max_upload_bytes_round_trips() { let settings = WorkspaceSettings { - ocr: OcrPolicy::Force, + max_upload_bytes: Some(25 * 1024 * 1024), ..Default::default() }; assert_eq!(Json::encode(&settings).or_default(), settings); diff --git a/crates/nvisy-postgres/src/types/mod.rs b/crates/nvisy-postgres/src/types/mod.rs index 29a80ad8..1dce059b 100644 --- a/crates/nvisy-postgres/src/types/mod.rs +++ b/crates/nvisy-postgres/src/types/mod.rs @@ -34,10 +34,10 @@ pub use json::{ ConnectionSyncFailedParams, DetectionActivityParams, DetectionCompletedParams, DetectionFailedParams, DetectionMetadata, FileActivityParams, InvalidHeader, InviteActivityParams, Json, MemberActivityParams, MemberInvitedParams, MemberJoinedParams, - NotificationPayload, OcrPolicy, PipelineActivityParams, PipelineMetadata, PolicyActivityParams, - RedactionActivityParams, RedactionCreatedParams, Retention, RetentionOverride, RetentionScope, - RetentionSettings, WebhookActivityParams, WebhookHeaders, WorkspaceActivityParams, - WorkspaceMetadata, WorkspaceSettings, + NotificationPayload, PipelineActivityParams, PipelineMetadata, PolicyActivityParams, + RasterPolicy, RedactionActivityParams, RedactionCreatedParams, Retention, RetentionOverride, + RetentionScope, RetentionSettings, WebhookActivityParams, WebhookHeaders, + WorkspaceActivityParams, WorkspaceMetadata, WorkspaceSettings, }; pub use pagination::{Cursor, CursorPage, CursorPagination, OffsetPage, OffsetPagination}; pub use prefixed_id::{ConnectionId, DetectionId, PrefixedIdError, RedactionId, WebhookId}; diff --git a/crates/nvisy-server/src/handler/error/http_error.rs b/crates/nvisy-server/src/handler/error/http_error.rs index 681cfcc1..a589a6fc 100644 --- a/crates/nvisy-server/src/handler/error/http_error.rs +++ b/crates/nvisy-server/src/handler/error/http_error.rs @@ -268,6 +268,8 @@ pub enum ErrorKind { NotFound, /// 409 Conflict - Conflicting resource state Conflict, + /// 413 Payload Too Large - Request body exceeds the allowed size + PayloadTooLarge, /// 429 Too Many Requests - Rate limit exceeded TooManyRequests, @@ -336,6 +338,7 @@ impl ErrorKind { Self::Forbidden => ErrorResponse::FORBIDDEN, Self::NotFound => ErrorResponse::NOT_FOUND, Self::Conflict => ErrorResponse::CONFLICT, + Self::PayloadTooLarge => ErrorResponse::PAYLOAD_TOO_LARGE, Self::TooManyRequests => ErrorResponse::TOO_MANY_REQUESTS, Self::InternalServerError => ErrorResponse::INTERNAL_SERVER_ERROR, Self::NotImplemented => ErrorResponse::NOT_IMPLEMENTED, diff --git a/crates/nvisy-server/src/handler/files.rs b/crates/nvisy-server/src/handler/files.rs index 7a209c52..395559db 100644 --- a/crates/nvisy-server/src/handler/files.rs +++ b/crates/nvisy-server/src/handler/files.rs @@ -4,6 +4,7 @@ //! including upload, download, metadata management, and file operations. All //! operations are secured with workspace-level authorization. +use std::collections::BTreeSet; use std::str::FromStr; use aide::axum::ApiRouter; @@ -26,14 +27,15 @@ use crate::extract::{ AuthProvider, AuthState, Json, Multipart, Path, Permission, Query, SecurityContext, ValidateJson, WorkspaceContext, }; -use crate::handler::request::{CursorPagination, ListFiles, UpdateFile, WorkspaceFilePathParams}; +use crate::handler::request::{ + CursorPagination, DeleteFiles, ListFiles, UpdateFile, WorkspaceFilePathParams, +}; use crate::handler::response::{self, ErrorResponse, File, Files, FilesPage}; use crate::handler::utility::{DownloadResponseExt, attachment_headers, resolve_account_ref}; use crate::handler::{Error, ErrorKind, Result}; -use crate::middleware::DEFAULT_MAX_FILE_BODY_SIZE; use crate::service::{ - CryptoService, EngineService, EventEmitter, EventOrigin, FileRef, HashingReader, RunBlobStore, - ServiceState, WorkspaceEvent, + CryptoService, EngineService, EventEmitter, EventOrigin, FileRef, HashingReader, LimitedReader, + RunBlobStore, ServiceState, WorkspaceEvent, }; /// Tracing target for workspace file operations. @@ -127,6 +129,11 @@ struct FileUploadContext { engine: EngineService, /// Retention expiry for uploaded originals (`None` = keep indefinitely). expires_at: Option, + /// The workspace's soft per-file upload cap in bytes, if set. A file that + /// streams past it is rejected before its excess reaches storage. `None` + /// leaves only the server-wide hard limit (enforced by the request body + /// layer) in force. + max_upload_bytes: Option, } /// Streams one multipart file to storage and builds its unsaved row. @@ -169,13 +176,30 @@ async fn process_single_file( "Streaming file to storage" ); - // Step 1: Encrypt the plaintext as it streams to NATS. The measured reader - // captures the plaintext size and hash (NATS only sees ciphertext). + // Step 1: Encrypt the plaintext as it streams to NATS. The limited reader + // aborts an oversized upload before its excess is encrypted and stored; with + // no workspace soft cap it is set to an unreachable budget, leaving the + // server-wide hard limit (enforced upstream by the request body layer) as the + // only bound. The measured reader captures the plaintext size and hash (NATS + // only sees ciphertext). + let cap = ctx.max_upload_bytes.unwrap_or(u64::MAX); let source = StreamReader::new(field.map(|result| result.map_err(std::io::Error::other))); - let (measured, measurements) = HashingReader::new(source); + let (limited, limit_state) = LimitedReader::new(source, cap); + let (measured, measurements) = HashingReader::new(limited); let encrypted = ctx.crypto.encrypt_reader(ctx.workspace_id, measured); - ctx.file_store.put(&file_key, Box::pin(encrypted)).await?; + if let Err(err) = ctx.file_store.put(&file_key, Box::pin(encrypted)).await { + // The limited reader aborts the stream, which fails the `put`. When that + // is why it failed, report the size limit (413) rather than a storage + // error; the reader's error is stringified in transit, so consult the + // shared state instead of inspecting the error. + if limit_state.is_exceeded() { + return Err(ErrorKind::PayloadTooLarge.with_message(format!( + "File exceeds the {cap}-byte upload limit for this workspace" + ))); + } + return Err(err.into()); + } tracing::debug!( target: TRACING_TARGET, @@ -234,11 +258,10 @@ async fn upload_file( // The uploader is the caller; resolve their identity once for every file below. let uploaded_by = resolve_account_ref(&mut conn, auth_claims.account_id).await?; - // Precompute the retention expiry for uploaded originals from workspace - // settings, so every file in this batch carries the same expiry. - let expires_at = workspace - .settings - .or_default() + // Read workspace settings once for the whole batch: the retention expiry for + // uploaded originals, and the soft per-file upload cap. + let settings = workspace.settings.or_default(); + let expires_at = settings .retention .original_documents .expires_at(jiff::Timestamp::now()); @@ -250,6 +273,7 @@ async fn upload_file( crypto, engine, expires_at, + max_upload_bytes: settings.max_upload_bytes, }; let mut uploaded_files = Vec::new(); @@ -654,10 +678,110 @@ fn delete_file_docs(op: TransformOperation) -> TransformOperation { .response::<404, Json>() } +/// Deletes several files in one call (soft delete). +/// +/// Idempotent: each requested id that resolves to a live file in the workspace +/// is deleted; ids that are unknown, already deleted, or in another workspace are +/// reported as skipped rather than failing the request. +#[tracing::instrument( + skip_all, + fields( + account_id = %auth_claims.account_id, + workspace_id = %workspace.id, + requested = request.file_ids.len(), + ) +)] +async fn bulk_delete_files( + State(pg_client): State, + State(blob): State, + WorkspaceContext(workspace): WorkspaceContext, + AuthState(auth_claims): AuthState, + security: SecurityContext, + ValidateJson(request): ValidateJson, +) -> Result<(StatusCode, Json)> { + tracing::debug!(target: TRACING_TARGET, "Bulk-deleting files"); + + let mut conn = pg_client.get_connection().await?; + + auth_claims + .authorize_workspace(&mut conn, workspace.id, Permission::DeleteFiles) + .await?; + + // De-duplicate the requested ids, then resolve the live files among them that + // belong to this workspace. Whatever does not come back is skipped: unknown, + // already deleted, or another workspace's — the delete is idempotent. + let requested: BTreeSet = request.file_ids.into_iter().collect(); + let requested: Vec = requested.into_iter().collect(); + let files = conn + .find_files_in_workspace(workspace.id, &requested) + .await?; + + let found: BTreeSet = files.iter().map(|file| file.id).collect(); + let skipped: Vec = requested + .into_iter() + .filter(|id| !found.contains(id)) + .collect(); + + // Soft-delete every resolved file and record its deletion event in one + // transaction, so no event is lost, nor recorded for a delete that rolled + // back. Either the whole batch's rows and events commit, or none do. + conn.transaction(async |conn| { + for file in &files { + conn.delete_workspace_file(file.id).await?; + conn.emit_event( + EventOrigin { + workspace_id: workspace.id, + account_id: auth_claims.account_id, + security: &security, + }, + WorkspaceEvent::FileDeleted(FileRef { + file_id: file.id, + file_name: file.display_name.clone(), + }), + ) + .await?; + } + Ok::<_, Error>(()) + }) + .await?; + + // 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. A pending purge is the reaper's to + // retry, so its outcome is not surfaced to the caller. + for file in &files { + let _ = blob + .purge_file(&mut conn, file.id, &file.storage_path, &file.storage_bucket) + .await?; + } + + let deleted: Vec = files.iter().map(|file| file.id).collect(); + tracing::info!( + target: TRACING_TARGET, + deleted = deleted.len(), + skipped = skipped.len(), + "Files bulk-deleted", + ); + + Ok(( + StatusCode::OK, + Json(response::DeletedFiles { deleted, skipped }), + )) +} + +fn bulk_delete_files_docs(op: TransformOperation) -> TransformOperation { + op.summary("Delete files") + .description("Deletes several files in one call. Idempotent: ids that resolve to live files in the workspace are removed and returned in `deleted`; ids that are unknown, already deleted, or in another workspace are returned in `skipped`. Deletion is permanent — the files' content cannot be recovered.") + .response::<200, Json>() + .response::<400, Json>() + .response::<401, Json>() + .response::<403, Json>() +} + /// Returns a [`Router`] with all related routes. /// /// [`Router`]: axum::routing::Router -pub fn routes() -> ApiRouter { +pub fn routes(max_file_body_bytes: usize) -> ApiRouter { use aide::axum::routing::*; ApiRouter::new() @@ -665,9 +789,16 @@ pub fn routes() -> ApiRouter { .api_route( "/workspaces/{workspaceSlug}/files/", post_with(upload_file, upload_file_docs) - .layer(DefaultBodyLimit::max(DEFAULT_MAX_FILE_BODY_SIZE)) + // Raise this route's default body limit to the upload ceiling; the + // global `RequestBodyLimitLayer` still caps every route at the same + // hard limit. + .layer(DefaultBodyLimit::max(max_file_body_bytes)) .get_with(list_files, list_files_docs), ) + .api_route( + "/workspaces/{workspaceSlug}/files/delete/", + post_with(bulk_delete_files, bulk_delete_files_docs), + ) .api_route( "/workspaces/{workspaceSlug}/files/{fileId}/", get_with(read_file, read_file_docs) diff --git a/crates/nvisy-server/src/handler/mod.rs b/crates/nvisy-server/src/handler/mod.rs index f367ed90..344fae72 100644 --- a/crates/nvisy-server/src/handler/mod.rs +++ b/crates/nvisy-server/src/handler/mod.rs @@ -40,7 +40,7 @@ pub use error::{Error, ErrorKind, Result}; pub use invites::{CreatedInvite, InviteOutcome, create_invite}; pub use utility::{BuiltinModule, CustomRoutes, RouterMapFn}; -use crate::middleware::{require_authentication, validate_token_middleware}; +use crate::middleware::{UploadConfig, require_authentication, validate_token_middleware}; use crate::service::ServiceState; /// Tracing target for unmatched-route fallbacks. @@ -70,6 +70,7 @@ fn private_routes( additional_routes: Option>, excluded: &HashSet, service_state: ServiceState, + upload: &UploadConfig, ) -> ApiRouter { let mut router = ApiRouter::new(); @@ -88,7 +89,7 @@ fn private_routes( .merge(connections::routes()) .merge(chat::routes()) .merge(connection_syncs::routes()) - .merge(files::routes()) + .merge(files::routes(upload.max_file_body_bytes)) .merge(pipelines::routes()) .merge(detections::routes()) .merge(detection_audits::routes()) @@ -144,14 +145,23 @@ fn public_routes( } /// Returns an [`ApiRouter`] with all routes. -pub fn routes(mut routes: CustomRoutes, state: ServiceState) -> ApiRouter { +pub fn routes( + mut routes: CustomRoutes, + state: ServiceState, + upload: &UploadConfig, +) -> ApiRouter { let require_authentication = from_fn_with_state(state.clone(), require_authentication); let validate_token_middleware = from_fn_with_state(state.clone(), validate_token_middleware); let excluded = std::mem::take(&mut routes.excluded_modules); // Private routes. - let mut private_router = private_routes(routes.private_routes.take(), &excluded, state.clone()); + let mut private_router = private_routes( + routes.private_routes.take(), + &excluded, + state.clone(), + upload, + ); private_router = routes.map_private_before_middleware(private_router); private_router = private_router .route_layer(require_authentication) @@ -184,6 +194,7 @@ mod test { use nvisy_webhook::reqwest::ReqwestClient; use crate::handler::{CustomRoutes, routes}; + use crate::middleware::UploadConfig; use crate::service::{ CryptoConfig, EngineConfig, HealthConfig, ServiceState, SessionKeysConfig, SyncConfig, }; @@ -228,6 +239,7 @@ mod test { HealthConfig::default(), SyncConfig::default(), webhook_service, + UploadConfig::default(), ) .await?; let router = router(state.clone()); @@ -246,7 +258,10 @@ mod test { /// Returns a new [`TestServer`] with the default router and state. pub async fn create_test_server() -> anyhow::Result { - create_test_server_with_router(|state| routes(CustomRoutes::new(), state)).await + create_test_server_with_router(|state| { + routes(CustomRoutes::new(), state, &UploadConfig::default()) + }) + .await } #[tokio::test] @@ -283,6 +298,7 @@ mod test { .exclude(BuiltinModule::Invites) .add_private_routes(custom.clone()), state, + &UploadConfig::default(), ) }) .await?; diff --git a/crates/nvisy-server/src/handler/request/files.rs b/crates/nvisy-server/src/handler/request/files.rs index 8e1f7445..ee523d11 100644 --- a/crates/nvisy-server/src/handler/request/files.rs +++ b/crates/nvisy-server/src/handler/request/files.rs @@ -9,6 +9,7 @@ use nvisy_postgres::model::UpdateWorkspaceFile as UpdateFileModel; use nvisy_postgres::types::FileFilter; use schemars::{JsonSchema, Schema, SchemaGenerator}; use serde::{Deserialize, Serialize}; +use uuid::Uuid; use validator::Validate; use crate::service::{EngineService, UnknownFormatToken}; @@ -35,6 +36,20 @@ impl UpdateFile { } } +/// Request to delete several files in one call. +/// +/// The `100`-id cap bounds the work one call fans out into: the resolve query, +/// the delete transaction, and one best-effort object purge per file. +#[must_use] +#[derive(Debug, Default, Serialize, Deserialize, Validate, JsonSchema)] +#[serde(rename_all = "camelCase")] +pub struct DeleteFiles { + /// Ids of the files to delete. Ids that are unknown, already deleted, or in + /// another workspace are skipped rather than failing the request. + #[validate(length(min = 1, max = 100))] + pub file_ids: Vec, +} + /// Defines a transparent string newtype whose OpenAPI schema enumerates the /// values the built-in codec registry supports, so the API advertises exactly /// which values are accepted (each is validated again at request time). diff --git a/crates/nvisy-server/src/handler/response/files.rs b/crates/nvisy-server/src/handler/response/files.rs index 81cfb83b..16435b91 100644 --- a/crates/nvisy-server/src/handler/response/files.rs +++ b/crates/nvisy-server/src/handler/response/files.rs @@ -60,6 +60,21 @@ impl File { } } +/// Result of a bulk file deletion. +/// +/// The deletion is idempotent: `deleted` holds the ids that resolved to live +/// files in the workspace and were removed, and `skipped` holds the requested +/// ids that did not (unknown, already deleted, or in another workspace). +#[must_use] +#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] +#[serde(rename_all = "camelCase")] +pub struct DeletedFiles { + /// Ids that were deleted. + pub deleted: Vec, + /// Requested ids that did not resolve to a live file and were skipped. + pub skipped: Vec, +} + /// Response for file uploads (simple list without pagination). pub type Files = Vec; diff --git a/crates/nvisy-server/src/handler/response/workspaces.rs b/crates/nvisy-server/src/handler/response/workspaces.rs index c4151958..95ef997f 100644 --- a/crates/nvisy-server/src/handler/response/workspaces.rs +++ b/crates/nvisy-server/src/handler/response/workspaces.rs @@ -2,7 +2,7 @@ use jiff::Timestamp; use nvisy_postgres::model; -use nvisy_postgres::types::{Handle, NotificationEvent, WorkspaceRole, WorkspaceSettings}; +use nvisy_postgres::types::{Handle, Json, NotificationEvent, WorkspaceRole, WorkspaceSettings}; use schemars::JsonSchema; use serde::{Deserialize, Serialize}; @@ -23,7 +23,11 @@ pub struct Workspace { /// Serve path of the workspace's avatar (logo), when set. #[serde(skip_serializing_if = "Option::is_none")] pub avatar_url: Option, - /// Workspace settings (approval requirement, data-retention rules). + /// Workspace settings (raster policy, data-retention rules, upload cap). + /// + /// `maxUploadBytes` is resolved to the effective per-file limit — the smaller + /// of the workspace's own cap and the server-wide hard limit — so a client + /// always reads a concrete number to enforce. pub settings: WorkspaceSettings, /// Account that created this workspace. pub created_by: AccountRef, @@ -37,14 +41,17 @@ pub struct Workspace { impl Workspace { /// Creates a new instance of [`Workspace`] as an owner. - pub fn from_model(workspace: model::Workspace, created_by: AccountRef) -> Self { - let settings = workspace.settings.or_default(); + pub fn from_model( + workspace: model::Workspace, + created_by: AccountRef, + hard_max_upload_bytes: u64, + ) -> Self { Self { slug: workspace.slug, display_name: workspace.display_name, description: workspace.description, avatar_url: workspace.avatar_url, - settings, + settings: resolve_settings(&workspace.settings, hard_max_upload_bytes), created_by, member_role: WorkspaceRole::Owner, created_at: workspace.created_at.into(), @@ -57,14 +64,14 @@ impl Workspace { workspace: model::Workspace, member: model::WorkspaceMember, created_by: AccountRef, + hard_max_upload_bytes: u64, ) -> Self { - let settings = workspace.settings.or_default(); Self { slug: workspace.slug, display_name: workspace.display_name, description: workspace.description, avatar_url: workspace.avatar_url, - settings, + settings: resolve_settings(&workspace.settings, hard_max_upload_bytes), created_by, member_role: member.member_role, created_at: workspace.created_at.into(), @@ -73,6 +80,24 @@ impl Workspace { } } +/// Resolves the stored settings for the response, replacing the raw soft cap with +/// the effective per-file upload limit: the smaller of the workspace's own cap +/// (the hard limit when it set none) and the server-wide hard limit. The result +/// is always a concrete number a client can enforce. +fn resolve_settings( + settings: &Json, + hard_max_upload_bytes: u64, +) -> WorkspaceSettings { + let mut settings = settings.or_default(); + let effective = settings + .max_upload_bytes + .map_or(hard_max_upload_bytes, |soft| { + soft.min(hard_max_upload_bytes) + }); + settings.max_upload_bytes = Some(effective); + settings +} + /// Paginated list of workspaces. pub type WorkspacesPage = Page; diff --git a/crates/nvisy-server/src/handler/workspaces.rs b/crates/nvisy-server/src/handler/workspaces.rs index 9d207bc5..88e1bbfa 100644 --- a/crates/nvisy-server/src/handler/workspaces.rs +++ b/crates/nvisy-server/src/handler/workspaces.rs @@ -28,6 +28,7 @@ use crate::handler::response::{ }; use crate::handler::utility::resolve_account_ref; use crate::handler::{Error, ErrorKind, Result}; +use crate::middleware::UploadConfig; use crate::service::{ AvatarService, EventEmitter, EventOrigin, MAX_AVATAR_UPLOAD_BYTES, ServiceState, WorkspaceEvent, WorkspaceRef, @@ -69,6 +70,7 @@ async fn backfill_retention( #[tracing::instrument(skip_all, fields(account_id = %auth_state.account_id))] async fn create_workspace( State(pg_client): State, + State(upload): State, AuthState(auth_state): AuthState, security: SecurityContext, ValidateJson(request): ValidateJson, @@ -105,7 +107,12 @@ async fn create_workspace( // The creator is the authenticated caller; resolve their identity directly. let creator = resolve_account_ref(&mut conn, creator_id).await?; - let response = Workspace::from_model_with_membership(workspace, membership, creator); + let response = Workspace::from_model_with_membership( + workspace, + membership, + creator, + upload.max_file_bytes(), + ); tracing::info!( target: TRACING_TARGET, @@ -131,6 +138,7 @@ fn create_workspace_docs(op: TransformOperation) -> TransformOperation { #[tracing::instrument(skip_all, fields(account_id = %auth_state.account_id))] async fn list_workspaces( State(pg_client): State, + State(upload): State, AuthState(auth_state): AuthState, Query(pagination): Query, ) -> Result<(StatusCode, Json)> { @@ -139,8 +147,14 @@ async fn list_workspaces( .cursor_list_account_workspaces_with_details(auth_state.account_id, pagination.into()) .await?; + let hard_max_upload_bytes = upload.max_file_bytes(); let response = Page::from_cursor_page(page, |(workspace, member, creator)| { - Workspace::from_model_with_membership(workspace, member, creator.into()) + Workspace::from_model_with_membership( + workspace, + member, + creator.into(), + hard_max_upload_bytes, + ) }); tracing::debug!( @@ -171,6 +185,7 @@ fn list_workspaces_docs(op: TransformOperation) -> TransformOperation { )] async fn read_workspace( State(pg_client): State, + State(upload): State, AuthState(auth_state): AuthState, WorkspaceContext(workspace): WorkspaceContext, ) -> Result<(StatusCode, Json)> { @@ -183,9 +198,10 @@ async fn read_workspace( tracing::info!(target: TRACING_TARGET, "Workspace read"); + let hard = upload.max_file_bytes(); let response = match member { - Some(member) => Workspace::from_model_with_membership(workspace, member, creator), - None => Workspace::from_model(workspace, creator), + Some(member) => Workspace::from_model_with_membership(workspace, member, creator, hard), + None => Workspace::from_model(workspace, creator, hard), }; Ok((StatusCode::OK, Json(response))) } @@ -211,6 +227,7 @@ fn read_workspace_docs(op: TransformOperation) -> TransformOperation { )] async fn update_workspace( State(pg_client): State, + State(upload): State, AuthState(auth_state): AuthState, WorkspaceContext(workspace): WorkspaceContext, security: SecurityContext, @@ -260,9 +277,10 @@ async fn update_workspace( tracing::info!(target: TRACING_TARGET, "Workspace updated"); + let hard = upload.max_file_bytes(); let response = match member { - Some(member) => Workspace::from_model_with_membership(updated, member, creator), - None => Workspace::from_model(updated, creator), + Some(member) => Workspace::from_model_with_membership(updated, member, creator, hard), + None => Workspace::from_model(updated, creator, hard), }; Ok((StatusCode::OK, Json(response))) diff --git a/crates/nvisy-server/src/middleware/mod.rs b/crates/nvisy-server/src/middleware/mod.rs index d077ed1a..b286d2aa 100644 --- a/crates/nvisy-server/src/middleware/mod.rs +++ b/crates/nvisy-server/src/middleware/mod.rs @@ -64,6 +64,7 @@ pub use recovery::{RecoveryConfig, RouterRecoveryExt}; pub use route_category::RouteCategory; pub use security::{ CorsConfig, FrameOptions, ReferrerPolicy, RouterSecurityExt, SecurityHeadersConfig, + UploadConfig, }; pub use specification::{OpenApiConfig, RouterOpenApiExt}; pub use sunset::{SunsetConfig, sunset_headers}; diff --git a/crates/nvisy-server/src/middleware/security.rs b/crates/nvisy-server/src/middleware/security.rs index a91278d8..4cf1d7c2 100644 --- a/crates/nvisy-server/src/middleware/security.rs +++ b/crates/nvisy-server/src/middleware/security.rs @@ -27,7 +27,12 @@ pub trait RouterSecurityExt { /// /// This middleware stack applies CORS rules, security headers including /// HSTS and CSP, response compression, and request body size limits. - fn with_security(self, cors: &CorsConfig, headers: &SecurityHeadersConfig) -> Self; + fn with_security( + self, + cors: &CorsConfig, + upload: &UploadConfig, + headers: &SecurityHeadersConfig, + ) -> Self; /// Layers security middlewares with default configurations. /// @@ -41,7 +46,12 @@ impl RouterSecurityExt for Router where S: Clone + Send + Sync + 'static, { - fn with_security(self, cors: &CorsConfig, headers: &SecurityHeadersConfig) -> Self { + fn with_security( + self, + cors: &CorsConfig, + upload: &UploadConfig, + headers: &SecurityHeadersConfig, + ) -> Self { let cors_layer = CorsLayer::new() .allow_origin(cors.to_header_values()) .allow_methods([ @@ -57,8 +67,8 @@ where .max_age(cors.max_age); let mut router = self - .layer(DefaultBodyLimit::max(DEFAULT_MAX_BODY_SIZE)) - .layer(RequestBodyLimitLayer::new(DEFAULT_MAX_FILE_BODY_SIZE)) + .layer(DefaultBodyLimit::max(upload.max_body_bytes)) + .layer(RequestBodyLimitLayer::new(upload.max_file_body_bytes)) .layer(CompressionLayer::new()) .layer(cors_layer) .layer(SetResponseHeaderLayer::overriding( @@ -89,7 +99,11 @@ where } fn with_default_security(self) -> Self { - self.with_security(&CorsConfig::default(), &SecurityHeadersConfig::default()) + self.with_security( + &CorsConfig::default(), + &UploadConfig::default(), + &SecurityHeadersConfig::default(), + ) } } @@ -155,6 +169,49 @@ impl CorsConfig { } } +/// Request body size limits. +/// +/// Bounds how large an incoming request body may be before it is rejected, +/// guarding against denial-of-service via oversized payloads. Two limits: a +/// tighter default for ordinary JSON requests, and a larger one for file +/// uploads (which stream to object storage rather than buffer in memory). +#[derive(Debug, Clone)] +#[cfg_attr(feature = "cli", derive(clap::Args))] +#[must_use = "config does nothing unless you use it"] +pub struct UploadConfig { + /// Maximum size in bytes for an ordinary request body (non-upload routes). + #[cfg_attr( + feature = "cli", + arg(long, env = "MAX_BODY_BYTES", default_value_t = DEFAULT_MAX_BODY_SIZE) + )] + pub max_body_bytes: usize, + + /// Maximum size in bytes for a file upload request body. + #[cfg_attr( + feature = "cli", + arg(long, env = "MAX_FILE_BODY_BYTES", default_value_t = DEFAULT_MAX_FILE_BODY_SIZE) + )] + pub max_file_body_bytes: usize, +} + +impl Default for UploadConfig { + fn default() -> Self { + Self { + max_body_bytes: DEFAULT_MAX_BODY_SIZE, + max_file_body_bytes: DEFAULT_MAX_FILE_BODY_SIZE, + } + } +} + +impl UploadConfig { + /// The server-wide hard file-upload limit in bytes, as a `u64` for comparison + /// against measured upload sizes and workspace caps. + #[must_use] + pub fn max_file_bytes(&self) -> u64 { + self.max_file_body_bytes as u64 + } +} + /// Security headers configuration for the application. /// /// Configures various HTTP security headers that protect against diff --git a/crates/nvisy-server/src/service/crypto/limited_reader.rs b/crates/nvisy-server/src/service/crypto/limited_reader.rs new file mode 100644 index 00000000..5dee1caf --- /dev/null +++ b/crates/nvisy-server/src/service/crypto/limited_reader.rs @@ -0,0 +1,125 @@ +//! An [`AsyncRead`] wrapper that fails once more than a byte budget is read. +//! +//! Placed ahead of the upload pipe's hashing and encryption stages, it aborts an +//! oversized stream as soon as the limit is crossed — before the excess is +//! encrypted and written to storage — rather than measuring the size only after +//! the whole body has streamed through. +//! +//! The failing read raises an [`io::Error`], but intervening stages (the +//! encryptor, the object store) may stringify that error and lose its cause. So +//! the reader also records the trip in a shared [`LimitState`] handle the caller +//! can consult afterwards to tell an over-limit abort apart from a genuine I/O or +//! storage failure. + +use std::io; +use std::pin::Pin; +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::task::{Context, Poll}; + +use pin_project_lite::pin_project; +use tokio::io::{AsyncRead, ReadBuf}; + +/// A shared handle to whether a [`LimitedReader`] exceeded its budget. +/// +/// Cloneable and readable after the reader has been consumed, which is when the +/// caller — whose downstream `put`/encrypt may have swallowed the reader's error +/// into a generic failure — needs to know the cause was the size limit. +#[derive(Clone, Default)] +pub struct LimitState { + exceeded: Arc, +} + +impl LimitState { + /// Whether the reader read past its budget. + pub fn is_exceeded(&self) -> bool { + self.exceeded.load(Ordering::Relaxed) + } +} + +pin_project! { + /// An [`AsyncRead`] that yields an error once its byte budget is exceeded. + /// + /// Reads pass through untouched until the cumulative total would exceed + /// `limit`; the read that crosses the budget marks the shared [`LimitState`] + /// and fails with an [`io::Error`]. A stream that stays within the budget is + /// unaffected. + pub struct LimitedReader { + #[pin] + inner: R, + limit: u64, + read: u64, + state: LimitState, + } +} + +impl LimitedReader { + /// Wraps `inner`, allowing at most `limit` bytes, and returns it alongside a + /// handle that reports whether the budget was exceeded. + pub fn new(inner: R, limit: u64) -> (Self, LimitState) { + let state = LimitState::default(); + let reader = Self { + inner, + limit, + read: 0, + state: state.clone(), + }; + (reader, state) + } +} + +impl AsyncRead for LimitedReader { + fn poll_read( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &mut ReadBuf<'_>, + ) -> Poll> { + let this = self.project(); + let before = buf.filled().len(); + let poll = this.inner.poll_read(cx, buf); + if let Poll::Ready(Ok(())) = &poll { + *this.read += (buf.filled().len() - before) as u64; + if *this.read > *this.limit { + this.state.exceeded.store(true, Ordering::Relaxed); + // Roll the just-read bytes back out of the buffer: an `AsyncRead` + // that returns an error must not also report bytes read on the + // same poll (tokio's read helpers assert this). + buf.set_filled(before); + return Poll::Ready(Err(io::Error::other(format!( + "upload exceeds the {}-byte limit", + *this.limit + )))); + } + } + poll + } +} + +#[cfg(test)] +mod tests { + use tokio::io::AsyncReadExt; + + use super::LimitedReader; + + #[tokio::test] + async fn reads_within_the_limit_pass_through() { + let data = [7u8; 64]; + let (mut reader, state) = LimitedReader::new(&data[..], 64); + let mut out = Vec::new(); + reader.read_to_end(&mut out).await.expect("within limit"); + assert_eq!(out, data); + assert!(!state.is_exceeded()); + } + + #[tokio::test] + async fn reading_past_the_limit_fails_and_marks_the_state() { + let data = [7u8; 65]; + let (mut reader, state) = LimitedReader::new(&data[..], 64); + let mut out = Vec::new(); + reader + .read_to_end(&mut out) + .await + .expect_err("over limit must fail"); + assert!(state.is_exceeded()); + } +} diff --git a/crates/nvisy-server/src/service/crypto/mod.rs b/crates/nvisy-server/src/service/crypto/mod.rs index 4d353adb..16d4faed 100644 --- a/crates/nvisy-server/src/service/crypto/mod.rs +++ b/crates/nvisy-server/src/service/crypto/mod.rs @@ -8,6 +8,7 @@ mod error; mod generation; mod hashing_reader; mod key; +mod limited_reader; mod service; pub(crate) use encryption::{ @@ -17,4 +18,5 @@ pub use error::{CryptoError, CryptoResult}; pub(crate) use generation::generate_secret; pub(crate) use hashing_reader::{HashingReader, Measurements}; pub(crate) use key::EncryptionKey; +pub(crate) use limited_reader::LimitedReader; pub use service::{CryptoConfig, CryptoService}; diff --git a/crates/nvisy-server/src/service/detection/worker.rs b/crates/nvisy-server/src/service/detection/worker.rs index 67a305f3..381de056 100644 --- a/crates/nvisy-server/src/service/detection/worker.rs +++ b/crates/nvisy-server/src/service/detection/worker.rs @@ -16,7 +16,7 @@ use nvisy_postgres::query::{ EventOutboxRepository, WorkspaceDetectionRepository, WorkspaceFileRepository, WorkspaceRepository, }; -use nvisy_postgres::types::{DetectionStatus, Json, OcrPolicy, WorkspaceSettings}; +use nvisy_postgres::types::{DetectionStatus, Json, RasterPolicy, WorkspaceSettings}; use nvisy_postgres::{AsyncConnection, DieselError, PgConn, PgError}; use tokio_util::sync::CancellationToken; @@ -277,7 +277,7 @@ impl DetectionWorker { .with_context(err.to_string()) })?; - // Parse the workspace settings once; both OCR mode and retention read it. + // Parse the workspace settings once; both raster mode and retention read it. let settings = workspace.settings.or_default(); let request = self.engine @@ -412,12 +412,12 @@ enum JobOutcome { Retry, } -/// Maps a workspace's OCR policy to the engine's per-detection +/// Maps a workspace's raster policy to the engine's per-detection /// page-rasterisation mode. fn raster_mode_of(settings: &WorkspaceSettings) -> RasterMode { - match settings.ocr { - OcrPolicy::Auto => RasterMode::Auto, - OcrPolicy::Force => RasterMode::always(), - OcrPolicy::Never => RasterMode::Never, + match settings.raster { + RasterPolicy::Auto => RasterMode::Auto, + RasterPolicy::Always => RasterMode::always(), + RasterPolicy::Never => RasterMode::Never, } } diff --git a/crates/nvisy-server/src/service/mod.rs b/crates/nvisy-server/src/service/mod.rs index de5ff09e..923a41b3 100644 --- a/crates/nvisy-server/src/service/mod.rs +++ b/crates/nvisy-server/src/service/mod.rs @@ -28,11 +28,12 @@ use nvisy_postgres::{PgClient, PgClientMigrationExt, PgConfig}; use nvisy_webhook::WebhookService; use tokio_util::sync::CancellationToken; +use crate::middleware::UploadConfig; pub use crate::service::avatar::{AVATAR_CONTENT_TYPE, AvatarService, MAX_AVATAR_UPLOAD_BYTES}; pub use crate::service::chat::{ChatService, TurnLocation}; pub use crate::service::connection_config::ConnectionConfig; pub use crate::service::crypto::{CryptoConfig, CryptoService}; -pub(crate) use crate::service::crypto::{CryptoError, HashingReader, Measurements}; +pub(crate) use crate::service::crypto::{CryptoError, HashingReader, LimitedReader, Measurements}; pub(crate) use crate::service::detection::resolve_policies; pub use crate::service::detection::{ DetectionJob, DetectionOutboxDrainer, DetectionQueue, DetectionStatusEvent, DetectionWorker, @@ -99,6 +100,9 @@ pub struct ServiceState { pub password: PasswordService, pub session_keys: SessionKeys, pub user_agent_parser: UserAgentParser, + + // Request body size limits (server-wide hard caps): + pub upload: UploadConfig, } impl ServiceState { @@ -114,6 +118,7 @@ impl ServiceState { health_config: HealthConfig, sync_config: SyncConfig, webhook_service: WebhookService, + upload_config: UploadConfig, ) -> Result { let postgres_client = connect_postgres(postgres_config).await?; let nats_client = connect_nats(nats_config).await?; @@ -145,6 +150,7 @@ impl ServiceState { password: PasswordService::new(), session_keys, user_agent_parser: UserAgentParser::new(), + upload: upload_config, }; Ok(service_state) @@ -265,6 +271,7 @@ impl_di_field!( password: PasswordService, session_keys: SessionKeys, user_agent_parser: UserAgentParser, + upload: UploadConfig, ); // Stateless services, composed from `Infra` on extraction: From 1f65d97d188382631334685c0080ce4a4cfbcdbd Mon Sep 17 00:00:00 2001 From: Oleh Martsokha Date: Sun, 30 Aug 2026 18:43:41 +0200 Subject: [PATCH 2/4] Address review: atomic bulk delete, per-file cap, resolved() method MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Make bulk delete's resolution and deletion atomic: a single guarded `UPDATE ... RETURNING` (delete_files_in_workspace) transitions and returns only the rows it actually changed, run on the handler's connection so the deletion and its FileDeleted events commit together. A row a concurrent request already deleted is never double-reported or double-emitted. - Do not fail the bulk-delete response on an object-purge error after the transaction commits: log it and leave the object for the reaper, so a retry does not see the ids as already-gone `skipped`. - Document the 413 response on the upload endpoint. - Enforce the effective per-file cap (min(soft, hard)) in the LimitedReader for every upload, not only when a workspace soft cap is set — the request-body layer bounds the whole multipart request, not a single file. - Preserve legacy workspace raster settings with serde aliases (`ocr` -> raster, `force` -> always) plus a regression test, so pre-rename rows keep behavior. - Move the effective-cap resolution onto WorkspaceSettings: add effective_max_upload_bytes() and resolved(), replacing the response-side helper. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_018bKk1YEG4tZ69jzYVQvQL8 --- .../src/query/workspace_file.rs | 59 ++++++--- .../src/types/json/workspace_settings.rs | 58 +++++++++ crates/nvisy-server/src/handler/files.rs | 116 ++++++++++-------- .../src/handler/response/workspaces.rs | 30 ++--- 4 files changed, 176 insertions(+), 87 deletions(-) diff --git a/crates/nvisy-postgres/src/query/workspace_file.rs b/crates/nvisy-postgres/src/query/workspace_file.rs index b7d10e00..02be24fb 100644 --- a/crates/nvisy-postgres/src/query/workspace_file.rs +++ b/crates/nvisy-postgres/src/query/workspace_file.rs @@ -197,12 +197,19 @@ pub trait WorkspaceFileRepository { account_id: Uuid, ) -> impl Future> + Send; - /// Finds the live files among `file_ids` that belong to `workspace_id`. + /// Soft-deletes the live files among `file_ids` that belong to + /// `workspace_id`, returning the rows it actually transitioned. /// - /// Workspace-scoped so a caller can only resolve files in the workspace it - /// addressed; ids that are unknown, soft-deleted, or in another workspace are - /// simply absent from the result rather than an error. - fn find_files_in_workspace( + /// Resolution and deletion are one atomic step: the `UPDATE ... RETURNING` + /// guarded on `deleted_at IS NULL` transitions and returns only rows it + /// changed, so a row concurrently deleted by another request is absent from + /// the result and never double-reported. Ids that are unknown, already + /// deleted, or in another workspace are simply absent rather than an error. + /// Also drops each returned file's import-origin row so re-import is never + /// blocked (see [`delete_workspace_file`]). + /// + /// [`delete_workspace_file`]: WorkspaceFileRepository::delete_workspace_file + fn delete_files_in_workspace( &mut self, workspace_id: Uuid, file_ids: &[Uuid], @@ -825,23 +832,43 @@ impl WorkspaceFileRepository for PgConnection { Ok(usage.unwrap_or_else(|| BigDecimal::from(0))) } - async fn find_files_in_workspace( + async fn delete_files_in_workspace( &mut self, workspace_id: Uuid, file_ids: &[Uuid], ) -> PgResult> { - use schema::workspace_files::{self, dsl}; + use schema::{workspace_file_imports, workspace_files}; - let files = workspace_files::table - .filter(dsl::id.eq_any(file_ids)) - .filter(dsl::workspace_id.eq(workspace_id)) - .filter(dsl::deleted_at.is_null()) - .select(WorkspaceFile::as_select()) - .load(self) - .await - .map_err(PgError::from)?; + // Transition and return only the live rows in this workspace, in one + // atomic statement. The `deleted_at IS NULL` guard means a row a concurrent + // request already deleted is not returned here, so it is never + // double-counted or double-emitted. This runs on the caller's connection + // (not its own transaction) so the caller can commit the deletion together + // with the events it emits for the returned rows. + let deleted: Vec = diesel::update( + workspace_files::table + .filter(workspace_files::id.eq_any(file_ids)) + .filter(workspace_files::workspace_id.eq(workspace_id)) + .filter(workspace_files::deleted_at.is_null()), + ) + .set(workspace_files::deleted_at.eq(diesel::dsl::now)) + .returning(WorkspaceFile::as_returning()) + .get_results(self) + .await + .map_err(PgError::from)?; + + // Drop the import-origin rows of exactly the files just deleted, so + // re-import is never blocked (see `delete_workspace_file`). + let deleted_ids: Vec = deleted.iter().map(|file| file.id).collect(); + diesel::delete( + workspace_file_imports::table + .filter(workspace_file_imports::file_id.eq_any(&deleted_ids)), + ) + .execute(self) + .await + .map_err(PgError::from)?; - Ok(files) + Ok(deleted) } async fn list_workspace_file_versions( diff --git a/crates/nvisy-postgres/src/types/json/workspace_settings.rs b/crates/nvisy-postgres/src/types/json/workspace_settings.rs index 62ad177a..f4d8ecf9 100644 --- a/crates/nvisy-postgres/src/types/json/workspace_settings.rs +++ b/crates/nvisy-postgres/src/types/json/workspace_settings.rs @@ -25,6 +25,7 @@ pub enum RasterPolicy { #[default] Auto, /// Always render every page to images, ignoring any text layer. + #[serde(alias = "force")] Always, /// Rely on the text layer only; never rasterise pages. Never, @@ -37,6 +38,7 @@ pub enum RasterPolicy { #[serde(rename_all = "camelCase", default)] pub struct WorkspaceSettings { /// How document pages are rasterised for OCR during detection. + #[serde(alias = "ocr")] pub raster: RasterPolicy, /// Data-retention rules for the workspace. pub retention: RetentionSettings, @@ -49,6 +51,28 @@ pub struct WorkspaceSettings { pub max_upload_bytes: Option, } +impl WorkspaceSettings { + /// The effective per-file upload limit in bytes: the smaller of this + /// workspace's own soft cap (the hard limit when it set none) and the + /// server-wide `hard_max_upload_bytes`. Always a concrete number a client can + /// enforce. + #[must_use] + pub fn effective_max_upload_bytes(&self, hard_max_upload_bytes: u64) -> u64 { + self.max_upload_bytes.map_or(hard_max_upload_bytes, |soft| { + soft.min(hard_max_upload_bytes) + }) + } + + /// Returns these settings with `max_upload_bytes` replaced by the effective + /// per-file limit for `hard_max_upload_bytes`, so a response always exposes a + /// single concrete cap a client can enforce rather than the raw soft value. + #[must_use] + pub fn resolved(mut self, hard_max_upload_bytes: u64) -> Self { + self.max_upload_bytes = Some(self.effective_max_upload_bytes(hard_max_upload_bytes)); + self + } +} + #[cfg(test)] mod tests { use serde_json::json; @@ -93,4 +117,38 @@ mod tests { }; assert_eq!(Json::encode(&settings).or_default(), settings); } + + #[test] + fn legacy_ocr_field_and_force_variant_still_deserialize() { + // Settings persisted before the `ocr`/`force` rename must keep their + // behaviour: the `ocr` key aliases `raster`, and `force` aliases `always`. + let settings = column(json!({ "ocr": "force" })).or_default(); + assert_eq!(settings.raster, RasterPolicy::Always); + + let settings = column(json!({ "ocr": "never" })).or_default(); + assert_eq!(settings.raster, RasterPolicy::Never); + } + + #[test] + fn effective_max_upload_bytes_is_the_smaller_of_soft_and_hard() { + let hard = 12 * 1024 * 1024; + + // No soft cap: the hard limit governs. + let settings = WorkspaceSettings::default(); + assert_eq!(settings.effective_max_upload_bytes(hard), hard); + + // Soft cap below the hard limit wins. + let settings = WorkspaceSettings { + max_upload_bytes: Some(8 * 1024 * 1024), + ..Default::default() + }; + assert_eq!(settings.effective_max_upload_bytes(hard), 8 * 1024 * 1024); + + // Soft cap above the hard limit is clamped to it. + let settings = WorkspaceSettings { + max_upload_bytes: Some(64 * 1024 * 1024), + ..Default::default() + }; + assert_eq!(settings.effective_max_upload_bytes(hard), hard); + } } diff --git a/crates/nvisy-server/src/handler/files.rs b/crates/nvisy-server/src/handler/files.rs index 395559db..d6d38b51 100644 --- a/crates/nvisy-server/src/handler/files.rs +++ b/crates/nvisy-server/src/handler/files.rs @@ -33,6 +33,7 @@ use crate::handler::request::{ use crate::handler::response::{self, ErrorResponse, File, Files, FilesPage}; use crate::handler::utility::{DownloadResponseExt, attachment_headers, resolve_account_ref}; use crate::handler::{Error, ErrorKind, Result}; +use crate::middleware::UploadConfig; use crate::service::{ CryptoService, EngineService, EventEmitter, EventOrigin, FileRef, HashingReader, LimitedReader, RunBlobStore, ServiceState, WorkspaceEvent, @@ -129,11 +130,11 @@ struct FileUploadContext { engine: EngineService, /// Retention expiry for uploaded originals (`None` = keep indefinitely). expires_at: Option, - /// The workspace's soft per-file upload cap in bytes, if set. A file that - /// streams past it is rejected before its excess reaches storage. `None` - /// leaves only the server-wide hard limit (enforced by the request body - /// layer) in force. - max_upload_bytes: Option, + /// The effective per-file upload cap in bytes — the smaller of the workspace's + /// soft cap and the server-wide hard limit. A file streaming past it is + /// rejected before its excess reaches storage. This is a true per-file bound, + /// unlike the request-body layer, which limits the whole multipart request. + max_upload_bytes: u64, } /// Streams one multipart file to storage and builds its unsaved row. @@ -177,12 +178,11 @@ async fn process_single_file( ); // Step 1: Encrypt the plaintext as it streams to NATS. The limited reader - // aborts an oversized upload before its excess is encrypted and stored; with - // no workspace soft cap it is set to an unreachable budget, leaving the - // server-wide hard limit (enforced upstream by the request body layer) as the - // only bound. The measured reader captures the plaintext size and hash (NATS - // only sees ciphertext). - let cap = ctx.max_upload_bytes.unwrap_or(u64::MAX); + // aborts an oversized upload before its excess is encrypted and stored, + // enforcing the effective per-file cap directly (the request-body layer only + // bounds the whole multipart request). The measured reader captures the + // plaintext size and hash (NATS only sees ciphertext). + let cap = ctx.max_upload_bytes; let source = StreamReader::new(field.map(|result| result.map_err(std::io::Error::other))); let (limited, limit_state) = LimitedReader::new(source, cap); let (measured, measurements) = HashingReader::new(limited); @@ -194,9 +194,8 @@ async fn process_single_file( // error; the reader's error is stringified in transit, so consult the // shared state instead of inspecting the error. if limit_state.is_exceeded() { - return Err(ErrorKind::PayloadTooLarge.with_message(format!( - "File exceeds the {cap}-byte upload limit for this workspace" - ))); + return Err(ErrorKind::PayloadTooLarge + .with_message(format!("File exceeds the {cap}-byte upload limit"))); } return Err(err.into()); } @@ -240,6 +239,7 @@ async fn upload_file( State(nats_client): State, State(crypto): State, State(engine): State, + State(upload): State, WorkspaceContext(workspace): WorkspaceContext, AuthState(auth_claims): AuthState, security: SecurityContext, @@ -259,12 +259,14 @@ async fn upload_file( let uploaded_by = resolve_account_ref(&mut conn, auth_claims.account_id).await?; // Read workspace settings once for the whole batch: the retention expiry for - // uploaded originals, and the soft per-file upload cap. + // uploaded originals, and the effective per-file upload cap (the workspace's + // soft cap clamped to the server-wide hard limit). let settings = workspace.settings.or_default(); let expires_at = settings .retention .original_documents .expires_at(jiff::Timestamp::now()); + let max_upload_bytes = settings.effective_max_upload_bytes(upload.max_file_bytes()); let ctx = FileUploadContext { workspace_id: workspace.id, @@ -273,7 +275,7 @@ async fn upload_file( crypto, engine, expires_at, - max_upload_bytes: settings.max_upload_bytes, + max_upload_bytes, }; let mut uploaded_files = Vec::new(); @@ -363,6 +365,7 @@ fn upload_file_docs(op: TransformOperation) -> TransformOperation { .response::<400, Json>() .response::<401, Json>() .response::<403, Json>() + .response::<413, Json>() } /// Gets file metadata without downloading the content. @@ -707,55 +710,68 @@ async fn bulk_delete_files( .authorize_workspace(&mut conn, workspace.id, Permission::DeleteFiles) .await?; - // De-duplicate the requested ids, then resolve the live files among them that - // belong to this workspace. Whatever does not come back is skipped: unknown, - // already deleted, or another workspace's — the delete is idempotent. + // De-duplicate the requested ids. let requested: BTreeSet = request.file_ids.into_iter().collect(); let requested: Vec = requested.into_iter().collect(); + + // Atomically soft-delete the live files among them and record a deletion event + // for each in one transaction: the delete resolves and transitions the rows in + // a single guarded statement (so a row a concurrent request already deleted is + // never double-reported), and the events commit with it — none lost, none + // recorded for a delete that rolled back. `files` holds exactly the rows this + // request deleted. let files = conn - .find_files_in_workspace(workspace.id, &requested) + .transaction(async |conn| { + let files = conn + .delete_files_in_workspace(workspace.id, &requested) + .await?; + for file in &files { + conn.emit_event( + EventOrigin { + workspace_id: workspace.id, + account_id: auth_claims.account_id, + security: &security, + }, + WorkspaceEvent::FileDeleted(FileRef { + file_id: file.id, + file_name: file.display_name.clone(), + }), + ) + .await?; + } + Ok::<_, Error>(files) + }) .await?; - let found: BTreeSet = files.iter().map(|file| file.id).collect(); + // Whatever was not deleted is skipped: unknown, already deleted, or another + // workspace's — the delete is idempotent. + let deleted_ids: BTreeSet = files.iter().map(|file| file.id).collect(); let skipped: Vec = requested .into_iter() - .filter(|id| !found.contains(id)) + .filter(|id| !deleted_ids.contains(id)) .collect(); - // Soft-delete every resolved file and record its deletion event in one - // transaction, so no event is lost, nor recorded for a delete that rolled - // back. Either the whole batch's rows and events commit, or none do. - conn.transaction(async |conn| { - for file in &files { - conn.delete_workspace_file(file.id).await?; - conn.emit_event( - EventOrigin { - workspace_id: workspace.id, - account_id: auth_claims.account_id, - security: &security, - }, - WorkspaceEvent::FileDeleted(FileRef { - file_id: file.id, - file_name: file.display_name.clone(), - }), - ) - .await?; - } - Ok::<_, Error>(()) - }) - .await?; - // 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. A pending purge is the reaper's to - // retry, so its outcome is not surfaced to the caller. + // 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 _ = blob + if let Err(err) = blob .purge_file(&mut conn, file.id, &file.storage_path, &file.storage_bucket) - .await?; + .await + { + tracing::error!( + target: TRACING_TARGET, + file_id = %file.id, + error = %err, + "Failed to purge a bulk-deleted file's object; left for the reaper to retry", + ); + } } - let deleted: Vec = files.iter().map(|file| file.id).collect(); + let deleted: Vec = deleted_ids.into_iter().collect(); tracing::info!( target: TRACING_TARGET, deleted = deleted.len(), diff --git a/crates/nvisy-server/src/handler/response/workspaces.rs b/crates/nvisy-server/src/handler/response/workspaces.rs index 95ef997f..1e1aa4ff 100644 --- a/crates/nvisy-server/src/handler/response/workspaces.rs +++ b/crates/nvisy-server/src/handler/response/workspaces.rs @@ -2,7 +2,7 @@ use jiff::Timestamp; use nvisy_postgres::model; -use nvisy_postgres::types::{Handle, Json, NotificationEvent, WorkspaceRole, WorkspaceSettings}; +use nvisy_postgres::types::{Handle, NotificationEvent, WorkspaceRole, WorkspaceSettings}; use schemars::JsonSchema; use serde::{Deserialize, Serialize}; @@ -51,7 +51,10 @@ impl Workspace { display_name: workspace.display_name, description: workspace.description, avatar_url: workspace.avatar_url, - settings: resolve_settings(&workspace.settings, hard_max_upload_bytes), + settings: workspace + .settings + .or_default() + .resolved(hard_max_upload_bytes), created_by, member_role: WorkspaceRole::Owner, created_at: workspace.created_at.into(), @@ -71,7 +74,10 @@ impl Workspace { display_name: workspace.display_name, description: workspace.description, avatar_url: workspace.avatar_url, - settings: resolve_settings(&workspace.settings, hard_max_upload_bytes), + settings: workspace + .settings + .or_default() + .resolved(hard_max_upload_bytes), created_by, member_role: member.member_role, created_at: workspace.created_at.into(), @@ -80,24 +86,6 @@ impl Workspace { } } -/// Resolves the stored settings for the response, replacing the raw soft cap with -/// the effective per-file upload limit: the smaller of the workspace's own cap -/// (the hard limit when it set none) and the server-wide hard limit. The result -/// is always a concrete number a client can enforce. -fn resolve_settings( - settings: &Json, - hard_max_upload_bytes: u64, -) -> WorkspaceSettings { - let mut settings = settings.or_default(); - let effective = settings - .max_upload_bytes - .map_or(hard_max_upload_bytes, |soft| { - soft.min(hard_max_upload_bytes) - }); - settings.max_upload_bytes = Some(effective); - settings -} - /// Paginated list of workspaces. pub type WorkspacesPage = Page; From 04f8f41e5e7b3e2a849871338a20a449ece7bfde Mon Sep 17 00:00:00 2001 From: Oleh Martsokha Date: Sun, 30 Aug 2026 18:51:03 +0200 Subject: [PATCH 3/4] Redesign upload flow: atomic batch with single-site cleanup MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Rework the multi-file upload into a two-phase, all-or-nothing batch: - Phase 1 (stage_all): stream every file to object storage, collecting each object key with its unsaved row. Non-file fields are skipped. On any staging error, the objects staged so far are removed before returning. - Phase 2: insert every row and emit every FileCreated event in ONE transaction; on failure, discard all staged objects best-effort. The upload is now atomic — it records every file or, on any failure, none, and never leaves an object behind with no row to reclaim it. This replaces the previous per-file transaction loop, which committed files one at a time (so a mid-batch failure left earlier files persisted) and repeated the orphan-object cleanup inline for every file. Object cleanup now lives in one place (discard_staged), and process_single_file becomes stage_file returning a StagedFile. Documents the atomic contract on the endpoint. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_018bKk1YEG4tZ69jzYVQvQL8 --- crates/nvisy-server/src/handler/files.rs | 190 ++++++++++++++--------- 1 file changed, 118 insertions(+), 72 deletions(-) diff --git a/crates/nvisy-server/src/handler/files.rs b/crates/nvisy-server/src/handler/files.rs index d6d38b51..b1294c91 100644 --- a/crates/nvisy-server/src/handler/files.rs +++ b/crates/nvisy-server/src/handler/files.rs @@ -137,15 +137,17 @@ struct FileUploadContext { max_upload_bytes: u64, } -/// Streams one multipart file to storage and builds its unsaved row. +/// A file streamed to object storage whose row has not yet been inserted. /// -/// Returns the object's storage key alongside the row (rather than inserting it), -/// so the caller can persist the row and record its creation event in one -/// transaction, and reclaim the already-stored object if that transaction fails. -async fn process_single_file( - ctx: &FileUploadContext, - field: Field<'_>, -) -> Result<(FileKey, NewWorkspaceFile)> { +/// Pairs the object's storage key with its unsaved row so the batch can persist +/// every row together, and reclaim every staged object if that fails. +struct StagedFile { + key: FileKey, + record: NewWorkspaceFile, +} + +/// Streams one multipart file to storage and builds its unsaved row. +async fn stage_file(ctx: &FileUploadContext, field: Field<'_>) -> Result { let filename = field .file_name() .map(ToString::to_string) @@ -207,8 +209,9 @@ async fn process_single_file( "File encrypted and streamed to storage" ); - // Step 2: Create DB record with all storage info (Postgres generates its own id) - let file_record = NewWorkspaceFile { + // Step 2: Build the unsaved row from the storage location and the measured + // plaintext (Postgres generates the row's own id on insert). + let record = NewWorkspaceFile { workspace_id: ctx.workspace_id, account_id: ctx.account_id, display_name: Some(filename.clone()), @@ -223,7 +226,69 @@ async fn process_single_file( ..Default::default() }; - Ok((file_key, file_record)) + Ok(StagedFile { + key: file_key, + record, + }) +} + +impl FileUploadContext { + /// Streams every file field in the multipart body to storage, returning the + /// staged files (object key + unsaved row). Non-file fields are skipped. + /// + /// On any error, the objects staged so far are removed before returning, so a + /// failed batch never leaves an object behind with no row to reclaim it. + async fn stage_all(&self, multipart: &mut Multipart) -> Result> { + let mut staged: Vec = Vec::new(); + loop { + let field = match multipart.next_field().await { + Ok(Some(field)) => field, + Ok(None) => break, + Err(err) => { + self.discard_staged(&staged).await; + tracing::error!(target: TRACING_TARGET, error = %err, "Failed to read multipart field"); + return Err(ErrorKind::BadRequest + .with_message("Invalid multipart data") + .with_context(format!("Failed to parse multipart form: {err}"))); + } + }; + + if field.file_name().is_none() { + tracing::debug!( + target: TRACING_TARGET, + name = ?field.name(), + "Skipping non-file multipart field" + ); + continue; + } + + match stage_file(self, field).await { + Ok(file) => staged.push(file), + Err(err) => { + self.discard_staged(&staged).await; + return Err(err); + } + } + } + Ok(staged) + } + + /// Removes staged objects best-effort, for when the batch does not commit. + /// Each object was written before any row exists, so nothing else can reclaim + /// it; a failed removal is logged and left for no one — an acceptable leak in + /// the rare storage-error case, not worth failing the response over. + async fn discard_staged(&self, staged: &[StagedFile]) { + for file in staged { + if let Err(err) = self.file_store.delete(&file.key).await { + tracing::warn!( + target: TRACING_TARGET, + error = %err, + object_id = %file.key.object_id, + "Failed to remove staged object after an aborted upload", + ); + } + } + } } /// Uploads input files to a workspace for processing. @@ -243,7 +308,7 @@ async fn upload_file( WorkspaceContext(workspace): WorkspaceContext, AuthState(auth_claims): AuthState, security: SecurityContext, - Multipart(mut multipart): Multipart, + mut multipart: Multipart, ) -> Result<(StatusCode, Json)> { tracing::info!(target: TRACING_TARGET, "Uploading files"); @@ -278,76 +343,57 @@ async fn upload_file( max_upload_bytes, }; - let mut uploaded_files = Vec::new(); - while let Some(field) = multipart.next_field().await.map_err(|err| { - tracing::error!(target: TRACING_TARGET, error = %err, "Failed to read multipart field"); - ErrorKind::BadRequest - .with_message("Invalid multipart data") - .with_context(format!("Failed to parse multipart form: {}", err)) - })? { - if field.file_name().is_none() { - tracing::debug!( - target: TRACING_TARGET, - name = ?field.name(), - "Skipping non-file multipart field" - ); - continue; - } + // Stream every file to storage first, then persist all their rows and events + // in one transaction. The upload is atomic: it either records every file or, + // on any failure, records none and reclaims every staged object — never a + // partial batch, and never an object left behind with no row. + let staged = ctx.stage_all(&mut multipart).await?; - let (file_key, file_record) = process_single_file(&ctx, field).await?; + if staged.is_empty() { + return Err(ErrorKind::BadRequest.with_message("No files provided in multipart request")); + } - // Persist the row and record its creation event in one transaction, so - // the event is never lost, nor recorded for a row that rolled back. - let created_file = match conn - .transaction(async |conn| { - let created_file = conn.create_workspace_file(file_record).await?; + let origin = EventOrigin { + workspace_id: workspace.id, + account_id: auth_claims.account_id, + security: &security, + }; + let created = match conn + .transaction(async |conn| { + let mut created = Vec::with_capacity(staged.len()); + for file in &staged { + let record = conn.create_workspace_file(file.record.clone()).await?; conn.emit_event( - EventOrigin { - workspace_id: workspace.id, - account_id: auth_claims.account_id, - security: &security, - }, + origin, WorkspaceEvent::FileCreated { file: FileRef { - file_id: created_file.id, - file_name: created_file.display_name.clone(), + file_id: record.id, + file_name: record.display_name.clone(), }, - file_size_bytes: created_file.file_size_bytes, + file_size_bytes: record.file_size_bytes, }, ) .await?; - Ok::<_, Error>(created_file) - }) - .await - { - Ok(created_file) => created_file, - Err(err) => { - // The object was streamed to storage before this transaction, so a - // rollback leaves it with no row and nothing can reclaim it later - // (the reaper works from file rows). Remove it best-effort before - // surfacing the error. - if let Err(cleanup) = ctx.file_store.delete(&file_key).await { - tracing::warn!( - target: TRACING_TARGET, - error = %cleanup, - object_id = %file_key.object_id, - "Failed to remove orphaned object after rolled-back file insert", - ); - } - return Err(err); + created.push(record); } - }; - - uploaded_files.push(response::File::from_model( - created_file, - workspace.slug.clone(), - uploaded_by.clone(), - )); - } + Ok::<_, Error>(created) + }) + .await + { + Ok(created) => created, + Err(err) => { + // The objects were streamed before this transaction, so a rollback + // leaves them with no rows and nothing to reclaim them later (the + // reaper works from file rows). Remove them best-effort first. + ctx.discard_staged(&staged).await; + return Err(err); + } + }; - if uploaded_files.is_empty() { - return Err(ErrorKind::BadRequest.with_message("No files provided in multipart request")); - } + let uploaded_files: Files = created + .into_iter() + .map(|file| response::File::from_model(file, workspace.slug.clone(), uploaded_by.clone())) + .collect(); tracing::info!( target: TRACING_TARGET, @@ -360,7 +406,7 @@ async fn upload_file( fn upload_file_docs(op: TransformOperation) -> TransformOperation { op.summary("Upload files") - .description("Uploads one or more files to a workspace. Each file is encrypted, streamed to storage, and recorded.") + .description("Uploads one or more files to a workspace. Each file is encrypted and streamed to storage. The batch is atomic: either every file is recorded, or on any failure none are and the request fails.") .response::<201, Json>() .response::<400, Json>() .response::<401, Json>() From 8046b2dbef298991eb161d8a3bc7e8bc2633ea87 Mon Sep 17 00:00:00 2001 From: Oleh Martsokha Date: Sun, 30 Aug 2026 20:22:23 +0200 Subject: [PATCH 4/4] Address review: protect active-detection files; body-limit ceiling - Bulk delete now holds back files an in-progress detection still needs: the delete UPDATE excludes any file referenced as an input_file_id or audit_file_id of a detection in IN_PROGRESS, mirroring the expiry sweep's hold in files_due_for_expiry. A held file is not transitioned, so no FileDeleted event or object purge occurs and it is reported in `skipped`. Documented on the response type. - Make the router-wide RequestBodyLimitLayer the larger of the two configured limits (request_body_ceiling), so a configuration where max_body_bytes exceeds max_file_body_bytes can no longer 413 an ordinary request the per-route default would allow. Adds a unit test. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_018bKk1YEG4tZ69jzYVQvQL8 --- .../src/query/workspace_file.rs | 29 ++++++++++--- crates/nvisy-server/src/handler/files.rs | 4 +- .../src/handler/response/files.rs | 6 ++- .../nvisy-server/src/middleware/security.rs | 42 ++++++++++++++++++- 4 files changed, 71 insertions(+), 10 deletions(-) diff --git a/crates/nvisy-postgres/src/query/workspace_file.rs b/crates/nvisy-postgres/src/query/workspace_file.rs index 02be24fb..5a31ebde 100644 --- a/crates/nvisy-postgres/src/query/workspace_file.rs +++ b/crates/nvisy-postgres/src/query/workspace_file.rs @@ -837,11 +837,29 @@ impl WorkspaceFileRepository for PgConnection { workspace_id: Uuid, file_ids: &[Uuid], ) -> PgResult> { - use schema::{workspace_file_imports, workspace_files}; + use diesel::dsl::{exists, not}; + use schema::workspace_detections::dsl as detections; + use schema::{workspace_detections, workspace_file_imports, workspace_files}; + + // A detection still analyzing (pending or executing) needs its input + // document and audit blob, so a file either references is held back from + // deletion — otherwise purging it would strand the in-flight detection with + // no source or analysis. This mirrors the expiry sweep's hold + // (`files_due_for_expiry`); a held file is simply not transitioned, so it + // is absent from the result and the caller reports it as skipped. + let active_detection_holds_file = exists( + workspace_detections::table.filter( + detections::status.eq_any(DetectionStatus::IN_PROGRESS).and( + detections::input_file_id + .eq(workspace_files::id) + .or(detections::audit_file_id.eq(workspace_files::id.nullable())), + ), + ), + ); - // Transition and return only the live rows in this workspace, in one - // atomic statement. The `deleted_at IS NULL` guard means a row a concurrent - // request already deleted is not returned here, so it is never + // Transition and return only the live, non-held rows in this workspace, in + // one atomic statement. The `deleted_at IS NULL` guard means a row a + // concurrent request already deleted is not returned here, so it is never // double-counted or double-emitted. This runs on the caller's connection // (not its own transaction) so the caller can commit the deletion together // with the events it emits for the returned rows. @@ -849,7 +867,8 @@ impl WorkspaceFileRepository for PgConnection { workspace_files::table .filter(workspace_files::id.eq_any(file_ids)) .filter(workspace_files::workspace_id.eq(workspace_id)) - .filter(workspace_files::deleted_at.is_null()), + .filter(workspace_files::deleted_at.is_null()) + .filter(not(active_detection_holds_file)), ) .set(workspace_files::deleted_at.eq(diesel::dsl::now)) .returning(WorkspaceFile::as_returning()) diff --git a/crates/nvisy-server/src/handler/files.rs b/crates/nvisy-server/src/handler/files.rs index b1294c91..826cf43d 100644 --- a/crates/nvisy-server/src/handler/files.rs +++ b/crates/nvisy-server/src/handler/files.rs @@ -789,8 +789,8 @@ async fn bulk_delete_files( }) .await?; - // Whatever was not deleted is skipped: unknown, already deleted, or another - // workspace's — the delete is idempotent. + // Whatever was not deleted is skipped: unknown, already deleted, another + // workspace's, or held by an in-progress detection — the delete is idempotent. let deleted_ids: BTreeSet = files.iter().map(|file| file.id).collect(); let skipped: Vec = requested .into_iter() diff --git a/crates/nvisy-server/src/handler/response/files.rs b/crates/nvisy-server/src/handler/response/files.rs index 16435b91..40442a84 100644 --- a/crates/nvisy-server/src/handler/response/files.rs +++ b/crates/nvisy-server/src/handler/response/files.rs @@ -64,14 +64,16 @@ impl File { /// /// The deletion is idempotent: `deleted` holds the ids that resolved to live /// files in the workspace and were removed, and `skipped` holds the requested -/// ids that did not (unknown, already deleted, or in another workspace). +/// ids that did not — unknown, already deleted, in another workspace, or held by +/// an in-progress detection that still needs the file. #[must_use] #[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] #[serde(rename_all = "camelCase")] pub struct DeletedFiles { /// Ids that were deleted. pub deleted: Vec, - /// Requested ids that did not resolve to a live file and were skipped. + /// Requested ids that were skipped: unknown, already deleted, in another + /// workspace, or held by an in-progress detection. pub skipped: Vec, } diff --git a/crates/nvisy-server/src/middleware/security.rs b/crates/nvisy-server/src/middleware/security.rs index 4cf1d7c2..101f8aaa 100644 --- a/crates/nvisy-server/src/middleware/security.rs +++ b/crates/nvisy-server/src/middleware/security.rs @@ -68,7 +68,11 @@ where let mut router = self .layer(DefaultBodyLimit::max(upload.max_body_bytes)) - .layer(RequestBodyLimitLayer::new(upload.max_file_body_bytes)) + // The router-wide hard ceiling must not sit below any per-route + // default, or an ordinary request within `max_body_bytes` would be + // rejected here first; use the larger of the two limits regardless of + // how they are configured relative to each other. + .layer(RequestBodyLimitLayer::new(upload.request_body_ceiling())) .layer(CompressionLayer::new()) .layer(cors_layer) .layer(SetResponseHeaderLayer::overriding( @@ -210,6 +214,18 @@ impl UploadConfig { pub fn max_file_bytes(&self) -> u64 { self.max_file_body_bytes as u64 } + + /// The router-wide request-body ceiling: the larger of the two limits. + /// + /// This backs the single `RequestBodyLimitLayer` wrapping every route, so it + /// must never sit below the per-route default (`max_body_bytes`) — otherwise + /// an ordinary request the default would allow gets rejected by the ceiling + /// first. The per-route `DefaultBodyLimit`s enforce the finer limits beneath + /// it. + #[must_use] + pub fn request_body_ceiling(&self) -> usize { + self.max_body_bytes.max(self.max_file_body_bytes) + } } /// Security headers configuration for the application. @@ -308,3 +324,27 @@ impl ReferrerPolicy { } } } + +#[cfg(test)] +mod tests { + use super::UploadConfig; + + #[test] + fn request_body_ceiling_is_never_below_the_default_body_limit() { + // A misconfiguration where the ordinary-request default exceeds the file + // limit must still admit ordinary requests: the router-wide ceiling is the + // larger of the two, so it never rejects a request the default allows. + let config = UploadConfig { + max_body_bytes: 8 * 1024 * 1024, + max_file_body_bytes: 4 * 1024 * 1024, + }; + assert_eq!(config.request_body_ceiling(), 8 * 1024 * 1024); + + // In the usual configuration (file limit larger), the file limit governs. + let config = UploadConfig { + max_body_bytes: 4 * 1024 * 1024, + max_file_body_bytes: 12 * 1024 * 1024, + }; + assert_eq!(config.request_body_ceiling(), 12 * 1024 * 1024); + } +}