cjw/Oxicloud
1
0
Fork 0
Code Issues Pull Requests Actions 4 Packages Projects Releases Wiki Activity

fix(upload): stream WebDAV/NextCloud PUT to disk to prevent OOM on large files

Browse Source 
Large uploads (e.g. ~800 MB ISOs) could OOMKill the process, even on
dedup hits, due to three separate full-file-in-memory paths:

- NextCloud PUT (/remote.php/dav) buffered the entire body in RAM via
  body::to_bytes before any dedup logic, then re-wrote and re-hashed it.
  Now streams the body to a temp file with incremental BLAKE3 and goes
  through update_file_streaming (shared spool helper with the native
  WebDAV PUT handler); peak heap is ~one HTTP frame regardless of size.

- DedupService::store_chunks materialized every new chunk's data in a Vec
  before uploading. Now reads each new chunk by positioned I/O
  (read_exact_at, off the runtime via spawn_blocking) just before its
  upload; peak heap bounded to ~CHUNK_UPLOAD_CONCURRENCY x CDC_MAX_CHUNK.

- The upload spool used the OS temp dir, often tmpfs/RAM in containers
  where its page-cache counts against the cgroup memory limit. Add
  OXICLOUD_UPLOAD_TMPDIR to point the spool at real disk.

Also collapse a pre-existing clippy collapsible_else_if in carddav_handler.

Refs #404

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
DioCrafts
2026-06-07 01:03:52 +02:00
parent c5e0800336
commit 061306cc84
12 changed files with 315 additions and 153 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/interfaces/api/handlers/carddav_handler.rs
+3 -5
View File
@@ -141,12 +141,10 @@ fn strip_username_prefix(path: &str) -> &str {
} else {
&path[pos + 1..]
}
} else if uuid::Uuid::parse_str(path).is_ok() {
path
} else {
if uuid::Uuid::parse_str(path).is_ok() {
path
} else {
""
}
""
}
}
src/interfaces/api/handlers/webdav_handler.rs
+13 -41
View File
@@ -822,9 +822,7 @@ async fn handle_put(
req: Request<Body>,
path: String,
) -> Result<Response<Body>, 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()
src/interfaces/mod.rs
+1
View File
@@ -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;
src/interfaces/nextcloud/webdav_handler.rs
+50 -41
View File
@@ -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<T> {
@@ -531,51 +532,59 @@ async fn handle_put(
.and_then(|v| v.parse::<i64>().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 ────────────────────
src/interfaces/upload_spool.rs
+86
View File
@@ -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<PathBuf>,
) -> Result<SpooledBody, AppError> {
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,
})
}