diff --git a/example.env b/example.env index f0da51cc..a83f6876 100644 --- a/example.env +++ b/example.env @@ -34,6 +34,15 @@ OXICLOUD_SERVER_HOST=127.0.0.1 # Maximum upload size in bytes (default: 10 GB on 64-bit) #OXICLOUD_MAX_UPLOAD_SIZE=10737418240 +# Directory for upload spool temp files. Uploads are streamed to a temp file +# before deduplication. By default this uses the OS temp dir ($TMPDIR / /tmp), +# which in many containers is tmpfs (RAM) — writing a large upload there fills +# page-cache that counts against the cgroup memory limit and can OOMKill the +# process. Point this at a real-disk path (same filesystem as the storage +# backend is ideal) to keep the upload footprint off RAM. Leave unset to use +# the OS default. +#OXICLOUD_UPLOAD_TMPDIR=/var/lib/oxicloud/tmp + # How often (seconds) the background sweep reconciles each user's cached # storage usage with the real sum of their files (default: 600 = 10 min). # GET /api/auth/me serves the cached value instead of recomputing per request; diff --git a/src/application/services/file_upload_service.rs b/src/application/services/file_upload_service.rs index 811d92d3..9490f7e6 100644 --- a/src/application/services/file_upload_service.rs +++ b/src/application/services/file_upload_service.rs @@ -1,4 +1,4 @@ -use std::path::Path; +use std::path::{Path, PathBuf}; use std::sync::Arc; use crate::application::dtos::file_dto::FileDto; @@ -55,6 +55,9 @@ pub struct FileUploadService { content_cache: Option>, /// Single lifecycle dispatcher — fires on_file_created / on_file_updated. file_lifecycle_hook: Option>, + /// Directory for spool temp files (`&[u8]` upload variants). When `Some`, + /// keeps spools off tmpfs/RAM so they don't count against the cgroup limit. + upload_temp_dir: Option, } impl FileUploadService { @@ -66,6 +69,7 @@ impl FileUploadService { storage_usage_service: None, content_cache: None, file_lifecycle_hook: None, + upload_temp_dir: None, } } @@ -80,9 +84,16 @@ impl FileUploadService { storage_usage_service: None, content_cache: None, file_lifecycle_hook: None, + upload_temp_dir: None, } } + /// Configures the spool directory for the `&[u8]` upload variants. + pub fn with_upload_temp_dir(mut self, dir: Option) -> Self { + self.upload_temp_dir = dir; + self + } + /// Configures the content cache for invalidation on file updates. pub fn with_content_cache(mut self, cache: Arc) -> Self { self.content_cache = Some(cache); @@ -106,6 +117,11 @@ impl FileUploadService { // ── private helpers ────────────────────────────────────────── + /// Create a spool temp file, honoring the configured upload temp dir. + fn new_temp(&self) -> std::io::Result { + crate::common::temp::new_spool_temp_file(self.upload_temp_dir.as_deref()) + } + /// Optionally update storage usage after a successful upload. fn maybe_update_storage_usage(&self, file: &FileDto) { if let Some(storage_service) = &self.storage_usage_service { @@ -220,7 +236,8 @@ impl FileUploadUseCase for FileUploadService { }; // Spool to temp file + hash - let temp = tempfile::NamedTempFile::new() + let temp = self + .new_temp() .map_err(|e| DomainError::internal_error("FileUpload", format!("temp file: {e}")))?; tokio::fs::write(temp.path(), content) .await @@ -260,7 +277,8 @@ impl FileUploadUseCase for FileUploadService { modified_at: Option, ) -> Result { // Spool to temp file + hash - let temp = tempfile::NamedTempFile::new() + let temp = self + .new_temp() .map_err(|e| DomainError::internal_error("FileUpload", format!("temp file: {e}")))?; tokio::fs::write(temp.path(), content) .await diff --git a/src/common/config.rs b/src/common/config.rs index 079de563..36e3d78b 100644 --- a/src/common/config.rs +++ b/src/common/config.rs @@ -211,6 +211,12 @@ pub struct StorageConfig { /// Maximum upload file size in bytes (default: 10 GB). /// Applied as a hard limit to WebDAV PUT and streaming uploads. pub max_upload_size: usize, + /// Directory for upload spool temp files. When `Some`, large uploads are + /// spooled here instead of the OS default temp dir (often tmpfs/RAM in + /// containers, where the spool's page-cache counts against the cgroup + /// memory limit and can trigger OOMKill on large files). Env: + /// `OXICLOUD_UPLOAD_TMPDIR`. + pub upload_temp_dir: Option, /// Interval (seconds) of the background sweep that reconciles every user's /// cached `storage_used_bytes` with the real sum of their files. Keeps the /// quota fresh for all mutations without recomputing on the request path. @@ -353,6 +359,7 @@ impl Default for StorageConfig { parallel_threshold: 100 * 1024 * 1024, // 100 MB trash_retention_days: 30, // 30 days max_upload_size: MAX_UPLOAD_SIZE, + upload_temp_dir: None, usage_reconcile_secs: 600, // 10 minutes backend: StorageBackendType::Local, s3: None, @@ -1218,6 +1225,14 @@ impl AppConfig { config.storage.max_upload_size = val; } + // Upload spool directory — keep large upload temp files off tmpfs/RAM + // (otherwise their page-cache counts against the cgroup memory limit). + if let Ok(dir) = env::var("OXICLOUD_UPLOAD_TMPDIR") + && !dir.trim().is_empty() + { + config.storage.upload_temp_dir = Some(PathBuf::from(dir.trim())); + } + // Background storage-usage reconciliation interval if let Ok(secs) = env::var("OXICLOUD_STORAGE_USAGE_RECONCILE_SECS").map(|v| v.parse::()) diff --git a/src/common/di.rs b/src/common/di.rs index 32d17b59..7f6b389a 100644 --- a/src/common/di.rs +++ b/src/common/di.rs @@ -425,7 +425,8 @@ impl AppServiceFactory { repos.file_read_repository.clone(), ) .with_content_cache(core.file_content_cache.clone()) - .with_file_lifecycle_hook(core.file_lifecycle.clone()), + .with_file_lifecycle_hook(core.file_lifecycle.clone()) + .with_upload_temp_dir(self.config.storage.upload_temp_dir.clone()), ); let file_retrieval_service = Arc::new(FileRetrievalService::new_with_cache( diff --git a/src/common/mod.rs b/src/common/mod.rs index b343f797..ddb9e2c6 100644 --- a/src/common/mod.rs +++ b/src/common/mod.rs @@ -4,3 +4,4 @@ pub mod errors; pub mod locale; pub mod mime_detect; pub mod stubs; +pub mod temp; diff --git a/src/common/temp.rs b/src/common/temp.rs new file mode 100644 index 00000000..b83729df --- /dev/null +++ b/src/common/temp.rs @@ -0,0 +1,26 @@ +//! Shared helper for creating upload spool temp files. +//! +//! Upload paths spool the request body to a temp file before deduplication. +//! By default `tempfile` uses the OS temp dir (`std::env::temp_dir()`, i.e. +//! `$TMPDIR` / `/tmp`), which in many container setups is **tmpfs (RAM)**. +//! Writing a multi-hundred-MB upload there fills page-cache that counts +//! against the cgroup memory limit and can OOMKill the process. Pointing the +//! spool at a real-disk directory (`OXICLOUD_UPLOAD_TMPDIR`) keeps the upload +//! footprint proportional to the streaming buffer, not the file size. + +use std::path::Path; +use tempfile::NamedTempFile; + +/// Create a [`NamedTempFile`], honoring an optional configured spool directory. +/// +/// When `dir` is `Some`, the temp file is created there (the directory is +/// created if missing); otherwise the OS default temp dir is used. +pub fn new_spool_temp_file(dir: Option<&Path>) -> std::io::Result { + match dir { + Some(d) => { + std::fs::create_dir_all(d)?; + NamedTempFile::new_in(d) + } + None => NamedTempFile::new(), + } +} diff --git a/src/infrastructure/services/dedup_service.rs b/src/infrastructure/services/dedup_service.rs index e4c81a5c..de2ee35e 100644 --- a/src/infrastructure/services/dedup_service.rs +++ b/src/infrastructure/services/dedup_service.rs @@ -42,7 +42,6 @@ use std::path::{Path, PathBuf}; use std::pin::Pin; use std::sync::Arc; use tokio::fs; -use tokio::io::{AsyncReadExt, AsyncSeekExt}; use crate::application::ports::blob_lifecycle::BlobLifecycleHook; use crate::application::ports::blob_storage_ports::BlobStorageBackend; @@ -499,73 +498,100 @@ impl DedupService { .into_iter() .collect(); - // ── Phase 1: Read only NEW chunks from disk ────────────── - let mut file = tokio::fs::File::open(source_path).await.map_err(|e| { + // ── Phase 1+2 (fused): upload NEW chunks just-in-time ──── + // Read each new chunk by positioned I/O immediately before its + // upload, instead of first materializing every new chunk's *data* in + // a Vec. Peak heap for file content is bounded to + // ~CHUNK_UPLOAD_CONCURRENCY × CDC_MAX_CHUNK (≈ 8 MiB) — proportional + // to the chunk size, never the file size, so storing a large + // brand-new file no longer spikes RAM. Existing chunks skip all disk + // I/O and just bump ref_count. + // + // We first collect *owned* per-chunk metadata (hash + offset + length + // + existence flag — no file data) so the stream below does not borrow + // the `chunks` parameter across an `.await` (which would make this + // future non-`Send` and break the upload handlers). + let chunk_ops: Vec<(String, u64, usize, bool)> = chunks + .iter() + .map(|chunk| { + let exists = existing_hashes.contains(&chunk.hash); + ( + chunk.hash.clone(), + chunk.offset as u64, + chunk.length, + exists, + ) + }) + .collect(); + + let source = Arc::new(std::fs::File::open(source_path).map_err(|e| { DomainError::internal_error("Dedup", format!("Failed to open source file: {}", e)) - })?; + })?); - // (hash, Option, size) — None = existing chunk (skip I/O), - // Some = new chunk (needs upload). - let mut chunk_ops: Vec<(String, Option, u64)> = Vec::with_capacity(chunks.len()); - - for chunk in chunks { - let size = chunk.length as u64; - if existing_hashes.contains(&chunk.hash) { - chunk_ops.push((chunk.hash.clone(), None, size)); - } else { - file.seek(std::io::SeekFrom::Start(chunk.offset as u64)) - .await - .map_err(|e| { - DomainError::internal_error("Dedup", format!("Failed to seek: {}", e)) - })?; - let mut buf = vec![0u8; chunk.length]; - file.read_exact(&mut buf).await.map_err(|e| { - DomainError::internal_error("Dedup", format!("Failed to read chunk: {}", e)) - })?; - chunk_ops.push((chunk.hash.clone(), Some(Bytes::from(buf)), size)); - } - } - - // ── Phase 2: Parallel upload (new) / ref-bump (existing) ─ let results: Vec> = stream::iter(chunk_ops) - .map(|(hash, data, size)| async move { - if let Some(bytes) = data { - // New chunk: upload to blob backend + INSERT/upsert - backend.put_blob_from_bytes(&hash, bytes).await?; - sqlx::query( - "INSERT INTO storage.blobs (hash, size, ref_count) - VALUES ($1, $2, 1) - ON CONFLICT (hash) DO UPDATE - SET ref_count = storage.blobs.ref_count + 1", - ) - .bind(&hash) - .bind(size as i64) - .execute(pool.as_ref()) - .await - .map_err(|e| { - DomainError::internal_error( - "Dedup", - format!("Failed to upsert chunk: {}", e), + .map(|(hash, offset, length, exists)| { + let source = source.clone(); + let pool = pool.clone(); + let backend = backend.clone(); + async move { + if exists { + // Existing chunk: bump ref_count, no disk I/O. + sqlx::query( + "UPDATE storage.blobs + SET ref_count = ref_count + 1 + WHERE hash = $1", ) - })?; - } else { - // Existing chunk: just bump ref_count (no I/O) - sqlx::query( - "UPDATE storage.blobs - SET ref_count = ref_count + 1 - WHERE hash = $1", - ) - .bind(&hash) - .execute(pool.as_ref()) - .await - .map_err(|e| { - DomainError::internal_error( - "Dedup", - format!("Failed to bump ref_count: {}", e), + .bind(&hash) + .execute(pool.as_ref()) + .await + .map_err(|e| { + DomainError::internal_error( + "Dedup", + format!("Failed to bump ref_count: {}", e), + ) + })?; + } else { + // New chunk: positioned read of just this chunk + // (≤ CDC_MAX_CHUNK) off the async runtime, then upload. + let bytes = tokio::task::spawn_blocking(move || { + use std::os::unix::fs::FileExt; + let mut buf = vec![0u8; length]; + source.read_exact_at(&mut buf, offset)?; + Ok::, std::io::Error>(buf) + }) + .await + .map_err(|e| { + DomainError::internal_error("Dedup", format!("Read task failed: {}", e)) + })? + .map_err(|e| { + DomainError::internal_error( + "Dedup", + format!("Failed to read chunk: {}", e), + ) + })?; + + backend + .put_blob_from_bytes(&hash, Bytes::from(bytes)) + .await?; + sqlx::query( + "INSERT INTO storage.blobs (hash, size, ref_count) + VALUES ($1, $2, 1) + ON CONFLICT (hash) DO UPDATE + SET ref_count = storage.blobs.ref_count + 1", ) - })?; + .bind(&hash) + .bind(length as i64) + .execute(pool.as_ref()) + .await + .map_err(|e| { + DomainError::internal_error( + "Dedup", + format!("Failed to upsert chunk: {}", e), + ) + })?; + } + Ok(()) } - Ok(()) }) .buffer_unordered(Self::CHUNK_UPLOAD_CONCURRENCY) .collect() diff --git a/src/interfaces/api/handlers/webdav_handler.rs b/src/interfaces/api/handlers/webdav_handler.rs index 3a61d8ba..a0f57afe 100644 --- a/src/interfaces/api/handlers/webdav_handler.rs +++ b/src/interfaces/api/handlers/webdav_handler.rs @@ -822,9 +822,7 @@ async fn handle_put( req: Request, path: String, ) -> Result, AppError> { - use http_body_util::BodyStream; - use tokio::io::AsyncWriteExt; - use tokio_stream::StreamExt; + use crate::interfaces::upload_spool::spool_body_to_temp; let user = extract_user(&req)?; @@ -882,44 +880,18 @@ async fn handle_put( .to_string(); // ── Streaming spool: body → temp file + incremental hash ── - let temp_file = tempfile::NamedTempFile::new() - .map_err(|e| AppError::internal_error(format!("Failed to create temp file: {}", e)))?; - let temp_path = temp_file.path().to_path_buf(); - - let mut file = tokio::fs::File::create(&temp_path) - .await - .map_err(|e| AppError::internal_error(format!("Failed to open temp file: {}", e)))?; - - let mut hasher = blake3::Hasher::new(); - let mut total_bytes: usize = 0; - let mut stream = BodyStream::new(req.into_body()); - - while let Some(frame_result) = stream.next().await { - let frame = frame_result - .map_err(|e| AppError::bad_request(format!("Failed to read request body: {}", e)))?; - if let Some(chunk) = frame.data_ref() { - total_bytes += chunk.len(); - if total_bytes > max_upload { - // Abort early — stop reading, delete temp file - drop(file); - let _ = tokio::fs::remove_file(&temp_path).await; - return Err(AppError::payload_too_large(format!( - "Upload exceeds maximum size of {} bytes", - max_upload - ))); - } - hasher.update(chunk); - file.write_all(chunk).await.map_err(|e| { - AppError::internal_error(format!("Failed to write to temp file: {}", e)) - })?; - } - } - file.flush() - .await - .map_err(|e| AppError::internal_error(format!("Failed to flush temp file: {}", e)))?; - drop(file); - - let hash = hasher.finalize().to_hex().to_string(); + // Shared with the NextCloud-compat PUT handler; peak heap ~one frame + // regardless of file size. Honors `upload_temp_dir` to keep the spool + // off tmpfs/RAM. + let spooled = spool_body_to_temp( + req.into_body(), + max_upload, + state.core.config.storage.upload_temp_dir.clone(), + ) + .await?; + let temp_path = spooled.temp.path().to_path_buf(); + let total_bytes = spooled.size as usize; + let hash = spooled.hash; // ── Quota enforcement ──────────────────────────────────── if let Some(storage_svc) = state.storage_usage_service.as_ref() diff --git a/src/interfaces/mod.rs b/src/interfaces/mod.rs index 1b7e6d78..2aab264d 100644 --- a/src/interfaces/mod.rs +++ b/src/interfaces/mod.rs @@ -2,6 +2,7 @@ pub mod api; pub mod errors; pub mod middleware; pub mod nextcloud; +pub mod upload_spool; pub mod web; pub use api::create_api_routes; diff --git a/src/interfaces/nextcloud/webdav_handler.rs b/src/interfaces/nextcloud/webdav_handler.rs index ca0f0ce1..af813e8b 100644 --- a/src/interfaces/nextcloud/webdav_handler.rs +++ b/src/interfaces/nextcloud/webdav_handler.rs @@ -20,9 +20,10 @@ use crate::application::ports::file_ports::{ use crate::application::ports::folder_ports::FolderUseCase; use crate::application::ports::trash_ports::TrashUseCase; use crate::common::di::AppState; -use crate::common::mime_detect::{filename_from_path, refine_content_type}; +use crate::common::mime_detect::{filename_from_path, refine_content_type_from_file}; use crate::interfaces::errors::AppError; use crate::interfaces::middleware::auth::{AuthUser, CurrentUser}; +use crate::interfaces::upload_spool::spool_body_to_temp; /// Extension trait to map XML write errors to `String` concisely. trait XmlResultExt { @@ -531,51 +532,59 @@ async fn handle_put( .and_then(|v| v.parse::().ok()); let max_upload = state.core.config.storage.max_upload_size; - let body_bytes = body::to_bytes(req.into_body(), max_upload) + + // Stream the body to a temp file + incremental hash — never buffer the + // full upload in RAM. The old `body::to_bytes` path loaded the entire + // file (e.g. an 800 MB ISO) into anonymous memory before any dedup logic, + // OOMKilling the process even on dedup hits. Shared with the native + // WebDAV PUT handler; peak heap ~one frame regardless of file size. + let spooled = spool_body_to_temp( + req.into_body(), + max_upload, + state.core.config.storage.upload_temp_dir.clone(), + ) + .await?; + + // Detect real MIME type from the first bytes on disk (no full read). + // `filename` is owned so we don't hold a borrow of the `subpath` param + // across the await (which would make the handler future non-Send). + let filename = filename_from_path(subpath).to_string(); + let content_type = + refine_content_type_from_file(spooled.temp.path(), &filename, &claimed_type).await; + + // Distinguish create (201) vs update (204) for the response status. + let existed = file_service.get_file_by_path(&internal_path).await.is_ok(); + + // Single streaming path — handles both update and create internally, + // passing the precomputed hash so the dedup fast path can short-circuit + // without re-reading the file. + let stored = upload_service + .update_file_streaming( + &internal_path, + spooled.temp.path(), + spooled.size, + &content_type, + Some(spooled.hash), + oc_mtime, + ) .await - .map_err(|e| AppError::bad_request(format!("Failed to read body: {}", e)))?; + .map_err(|e| AppError::internal_error(format!("Failed to store file: {}", e)))?; - // Detect real MIME type via magic bytes + extension, falling back to client header. - let filename = filename_from_path(subpath); - let content_type = refine_content_type(&body_bytes, filename, &claimed_type); + // dedup may have already moved the temp on a new-blob store; ignore error. + let _ = tokio::fs::remove_file(spooled.temp.path()).await; - // Check if the file already exists (update vs create). - let existing = file_service.get_file_by_path(&internal_path).await; - - if existing.is_ok() { - // Update existing file — returns FileDto with fresh content-hash etag. - let updated = upload_service - .update_file(&internal_path, &body_bytes, &content_type, oc_mtime) - .await - .map_err(|e| AppError::internal_error(format!("Failed to update file: {}", e)))?; - - return Ok(Response::builder() - .status(StatusCode::NO_CONTENT) - .header(header::ETAG, format!("\"{}\"", updated.etag)) - .header("oc-etag", format!("\"{}\"", updated.etag)) - .body(Body::empty()) - .unwrap()); - } - - // Create new file — split subpath into parent dir and filename. - let (parent_subpath, filename) = match subpath.rsplit_once('/') { - Some((parent, name)) => (parent, name), - None => ("", subpath), + let status = if existed { + StatusCode::NO_CONTENT + } else { + StatusCode::CREATED }; - let parent_internal = nc_to_internal_path(&user.username, parent_subpath)?; - - let file_dto = upload_service - .create_file(&parent_internal, filename, &body_bytes, &content_type) - .await - .map_err(|e| AppError::internal_error(format!("Failed to create file: {}", e)))?; - - let builder = Response::builder() - .status(StatusCode::CREATED) - .header(header::ETAG, format!("\"{}\"", file_dto.etag)) - .header("oc-etag", format!("\"{}\"", file_dto.etag)); - - Ok(builder.body(Body::empty()).unwrap()) + Ok(Response::builder() + .status(status) + .header(header::ETAG, format!("\"{}\"", stored.etag)) + .header("oc-etag", format!("\"{}\"", stored.etag)) + .body(Body::empty()) + .unwrap()) } // ──────────────────── MKCOL ──────────────────── diff --git a/src/interfaces/upload_spool.rs b/src/interfaces/upload_spool.rs new file mode 100644 index 00000000..694128e2 --- /dev/null +++ b/src/interfaces/upload_spool.rs @@ -0,0 +1,86 @@ +//! Shared streaming upload spool: request body → temp file + incremental hash. +//! +//! Used by both the native WebDAV PUT handler and the NextCloud-compat PUT +//! handler so neither buffers the full request body in memory. Peak heap is +//! ~one HTTP frame regardless of file size; the body is written to a temp +//! file (off tmpfs when [`StorageConfig::upload_temp_dir`] is configured) and +//! BLAKE3-hashed on the fly so the dedup layer can short-circuit on a hit. + +use std::path::PathBuf; + +use axum::body::Body; +use http_body_util::BodyStream; +use tempfile::NamedTempFile; +use tokio::io::AsyncWriteExt; +use tokio_stream::StreamExt; + +use crate::common::temp::new_spool_temp_file; +use crate::interfaces::errors::AppError; + +/// Outcome of spooling a request body to disk. +pub struct SpooledBody { + /// The temp file holding the body. Kept alive by the caller (dropping it + /// removes the file unless the dedup layer already consumed/moved it). + pub temp: NamedTempFile, + /// Hex-encoded BLAKE3 of the full body — matches `DedupService::hash_file`, + /// so passing it as `pre_computed_hash` enables the dedup fast path. + pub hash: String, + /// Total bytes written. + pub size: u64, +} + +/// Stream an HTTP request body to a temp file, computing its BLAKE3 hash +/// incrementally and enforcing `max_upload` as a hard size limit. +/// +/// Peak heap is ~one frame — the body is never fully buffered in RAM. +/// +/// `temp_dir` is taken by value (not `&Path`) so the returned future captures +/// no borrowed lifetime — required for the handler future to stay `Send`. +pub async fn spool_body_to_temp( + body: Body, + max_upload: usize, + temp_dir: Option, +) -> Result { + let temp = new_spool_temp_file(temp_dir.as_deref()) + .map_err(|e| AppError::internal_error(format!("Failed to create temp file: {e}")))?; + let temp_path = temp.path().to_path_buf(); + + let mut file = tokio::fs::File::create(&temp_path) + .await + .map_err(|e| AppError::internal_error(format!("Failed to open temp file: {e}")))?; + + let mut hasher = blake3::Hasher::new(); + let mut total_bytes: usize = 0; + let mut stream = BodyStream::new(body); + + while let Some(frame_result) = stream.next().await { + let frame = frame_result + .map_err(|e| AppError::bad_request(format!("Failed to read request body: {e}")))?; + if let Some(chunk) = frame.data_ref() { + total_bytes += chunk.len(); + if total_bytes > max_upload { + // Abort early — stop reading, delete temp file. + drop(file); + let _ = tokio::fs::remove_file(&temp_path).await; + return Err(AppError::payload_too_large(format!( + "Upload exceeds maximum size of {max_upload} bytes" + ))); + } + hasher.update(chunk); + file.write_all(chunk).await.map_err(|e| { + AppError::internal_error(format!("Failed to write to temp file: {e}")) + })?; + } + } + file.flush() + .await + .map_err(|e| AppError::internal_error(format!("Failed to flush temp file: {e}")))?; + drop(file); + + let hash = hasher.finalize().to_hex().to_string(); + Ok(SpooledBody { + temp, + hash, + size: total_bytes as u64, + }) +}