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

perf: Phase 4+5 optimizations — uploads 10x, downloads 2x, concurrent 2x. moka cache, 512KB buffers, remove sync_all, hash-on-write, preloaded queries, bench.sh v3, gitignore storage/. 500MB upload 12.6s->1.3s (392MB/s). RSS 69-113MB, 0 swap.

Browse Source
This commit is contained in:
Dionisio
2026-02-15 17:53:25 +01:00
parent fac0b5e77b
commit 1ed20f425f
57 changed files with 1700 additions and 1184 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/infrastructure/adapters/contact_storage_adapter.rs
+14 -14
View File
@@ -491,24 +491,24 @@ impl ContactUseCase for ContactStorageAdapter {
for line in vcard_data.lines() {
let trimmed = line.trim();
if trimmed.starts_with("UID:") {
uid = Some(trimmed[4..].trim().to_string());
} else if trimmed.starts_with("FN:") {
full_name = Some(trimmed[3..].trim().to_string());
} else if trimmed.starts_with("N:") {
let parts: Vec<&str> = trimmed[2..].split(';').collect();
if let Some(stripped) = trimmed.strip_prefix("UID:") {
uid = Some(stripped.trim().to_string());
} else if let Some(stripped) = trimmed.strip_prefix("FN:") {
full_name = Some(stripped.trim().to_string());
} else if let Some(stripped) = trimmed.strip_prefix("N:") {
let parts: Vec<&str> = stripped.split(';').collect();
if parts.len() >= 2 {
last_name = Some(parts[0].trim().to_string()).filter(|s| !s.is_empty());
first_name = Some(parts[1].trim().to_string()).filter(|s| !s.is_empty());
}
} else if trimmed.starts_with("NICKNAME:") {
nickname = Some(trimmed[9..].trim().to_string());
} else if trimmed.starts_with("ORG:") {
organization = Some(trimmed[4..].trim().to_string());
} else if trimmed.starts_with("TITLE:") {
title = Some(trimmed[6..].trim().to_string());
} else if trimmed.starts_with("NOTE:") {
notes = Some(trimmed[5..].trim().to_string());
} else if let Some(stripped) = trimmed.strip_prefix("NICKNAME:") {
nickname = Some(stripped.trim().to_string());
} else if let Some(stripped) = trimmed.strip_prefix("ORG:") {
organization = Some(stripped.trim().to_string());
} else if let Some(stripped) = trimmed.strip_prefix("TITLE:") {
title = Some(stripped.trim().to_string());
} else if let Some(stripped) = trimmed.strip_prefix("NOTE:") {
notes = Some(stripped.trim().to_string());
} else if trimmed.starts_with("EMAIL") {
if let Some(value) = trimmed.split(':').nth(1)
&& !value.is_empty()
src/infrastructure/repositories/pg/file_blob_read_repository.rs
+137 -21
View File
@@ -8,6 +8,7 @@ use async_trait::async_trait;
use bytes::Bytes;
use futures::Stream;
use sqlx::PgPool;
use std::collections::HashMap;
use std::sync::Arc;
use crate::application::ports::dedup_ports::DedupPort;
@@ -24,6 +25,10 @@ pub struct FileBlobReadRepository {
pool: Arc<PgPool>,
dedup: Arc<dyn DedupPort>,
folder_repo: Arc<FolderDbRepository>,
/// Lightweight cache: file_id → blob_hash.
/// Populated by `get_file()`, consumed by `get_blob_hash()`.
/// Avoids an extra SQL round-trip on the hot download path.
hash_cache: std::sync::Mutex<HashMap<String, String>>,
}
impl FileBlobReadRepository {
@@ -36,6 +41,7 @@ impl FileBlobReadRepository {
pool,
dedup,
folder_repo,
hash_cache: std::sync::Mutex::new(HashMap::new()),
}
}
@@ -53,7 +59,7 @@ impl FileBlobReadRepository {
}
}
/// Convert a database row into a `File` domain entity.
#[allow(clippy::too_many_arguments)]
async fn row_to_file(
&self,
id: String,
@@ -79,7 +85,13 @@ impl FileBlobReadRepository {
}
/// Get the blob hash for a file.
/// Checks the in-memory cache first (populated by `get_file`).
async fn get_blob_hash(&self, file_id: &str) -> Result<String, DomainError> {
// Fast path: already cached from a prior get_file call
if let Some(hash) = self.hash_cache.lock().unwrap().remove(file_id) {
return Ok(hash);
}
// Slow path: DB round-trip
sqlx::query_scalar::<_, String>(
"SELECT blob_hash FROM storage.files WHERE id = $1::uuid AND NOT is_trashed",
)
@@ -94,11 +106,12 @@ impl FileBlobReadRepository {
#[async_trait]
impl FileReadPort for FileBlobReadRepository {
async fn get_file(&self, id: &str) -> Result<File, DomainError> {
let row = sqlx::query_as::<_, (String, String, Option<String>, i64, String, i64, i64)>(
let row = sqlx::query_as::<_, (String, String, Option<String>, i64, String, i64, i64, String)>(
r#"
SELECT id::text, name, folder_id::text, size, mime_type,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint
EXTRACT(EPOCH FROM updated_at)::bigint,
blob_hash
FROM storage.files
WHERE id = $1::uuid AND NOT is_trashed
"#,
@@ -109,6 +122,13 @@ impl FileReadPort for FileBlobReadRepository {
.map_err(|e| DomainError::internal_error("FileBlobRead", format!("get: {e}")))?
.ok_or_else(|| DomainError::not_found("File", id))?;
// Cache blob_hash so the subsequent get_file_stream / get_file_content
// call doesn't need a separate DB round-trip.
self.hash_cache
.lock()
.unwrap()
.insert(id.to_string(), row.7.clone());
self.row_to_file(row.0, row.1, row.2, row.3, row.4, row.5, row.6)
.await
}
@@ -145,9 +165,32 @@ impl FileReadPort for FileBlobReadRepository {
}
.map_err(|e| DomainError::internal_error("FileBlobRead", format!("list: {e}")))?;
// ── N+1 fix: resolve the folder path ONCE (all rows share
// the same folder_id when listing a specific folder). ──
let shared_folder_path = if let Some(fid) = folder_id {
Some(self.folder_repo.get_folder_path(fid).await?)
} else {
None
};
let mut files = Vec::with_capacity(rows.len());
for (id, name, fid, size, mime, ca, ma) in rows {
files.push(self.row_to_file(id, name, fid, size, mime, ca, ma).await?);
let storage_path = match &shared_folder_path {
Some(fp) => fp.join(&name),
None => StoragePath::from_string(&name),
};
let file = File::with_timestamps(
id,
name,
storage_path,
size as u64,
mime,
fid,
ca as u64,
ma as u64,
)
.map_err(|e| DomainError::internal_error("FileBlobRead", format!("entity: {e}")))?;
files.push(file);
}
Ok(files)
}
@@ -161,13 +204,10 @@ impl FileReadPort for FileBlobReadRepository {
&self,
id: &str,
) -> Result<Box<dyn Stream<Item = Result<Bytes, std::io::Error>> + Send>, DomainError> {
// Read blob as bytes and wrap in a single-chunk stream.
// For very large files, a true streaming implementation from the
// blob file would be better, but DedupPort API currently returns bytes.
// True streaming: reads the blob file in 64 KB chunks.
// Memory usage is ~64 KB regardless of file size.
let blob_hash = self.get_blob_hash(id).await?;
let content = self.dedup.read_blob_bytes(&blob_hash).await?;
let stream = futures::stream::once(async move { Ok(content) });
let stream = self.dedup.read_blob_stream(&blob_hash).await?;
Ok(Box::new(stream))
}
@@ -177,22 +217,19 @@ impl FileReadPort for FileBlobReadRepository {
start: u64,
end: Option<u64>,
) -> Result<Box<dyn Stream<Item = Result<Bytes, std::io::Error>> + Send>, DomainError> {
// True range streaming: seeks to `start` and reads only the requested range.
// A 1 MB range on a 1 GB file uses ~64 KB of RAM.
let blob_hash = self.get_blob_hash(id).await?;
let content = self.dedup.read_blob_bytes(&blob_hash).await?;
let start = start as usize;
let end = end.map_or(content.len(), |e| e as usize).min(content.len());
if start >= content.len() {
return Ok(Box::new(futures::stream::empty()));
}
let slice = content.slice(start..end);
let stream = futures::stream::once(async move { Ok(slice) });
let stream = self
.dedup
.read_blob_range_stream(&blob_hash, start, end)
.await?;
Ok(Box::new(stream))
}
async fn get_file_mmap(&self, id: &str) -> Result<Bytes, DomainError> {
// For RPi targets, mmap is less beneficial than streaming.
// Keep as a fallback that loads content for small/medium files.
let blob_hash = self.get_blob_hash(id).await?;
self.dedup.read_blob_bytes(&blob_hash).await
}
@@ -262,4 +299,83 @@ impl FileReadPort for FileBlobReadRepository {
current_parent
.ok_or_else(|| DomainError::not_found("Folder", format!("parent for path: {path}")))
}
/// Direct SQL lookup: split path into folder segments + filename,
/// walk the folder hierarchy, then match the file by name + folder_id.
/// O(depth) queries instead of O(total_files).
async fn find_file_by_path(&self, path: &str) -> Result<Option<File>, DomainError> {
let path = path.trim_start_matches('/').trim_end_matches('/');
let segments: Vec<&str> = path.split('/').filter(|s| !s.is_empty()).collect();
if segments.is_empty() {
return Ok(None);
}
// Last segment is the filename, preceding segments are folders
let filename = segments[segments.len() - 1];
let folder_segments = &segments[..segments.len() - 1];
// Walk folder hierarchy to find parent folder_id
let mut current_parent: Option<String> = None;
for segment in folder_segments {
let row = if let Some(ref pid) = current_parent {
sqlx::query_scalar::<_, String>(
"SELECT id::text FROM storage.folders WHERE name = $1 AND parent_id = $2::uuid AND NOT is_trashed",
)
.bind(segment)
.bind(pid)
.fetch_optional(self.pool.as_ref())
.await
} else {
sqlx::query_scalar::<_, String>(
"SELECT id::text FROM storage.folders WHERE name = $1 AND parent_id IS NULL AND NOT is_trashed",
)
.bind(segment)
.fetch_optional(self.pool.as_ref())
.await
}
.map_err(|e| DomainError::internal_error("FileBlobRead", format!("path walk: {e}")))?;
match row {
Some(id) => current_parent = Some(id),
None => return Ok(None), // Folder not found → file doesn't exist at this path
}
}
// Now find the file by name + folder_id
let row = if let Some(ref fid) = current_parent {
sqlx::query_as::<_, (String, String, Option<String>, i64, String, i64, i64)>(
r#"
SELECT id::text, name, folder_id::text, size, mime_type,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint
FROM storage.files
WHERE name = $1 AND folder_id = $2::uuid AND NOT is_trashed
"#,
)
.bind(filename)
.bind(fid)
.fetch_optional(self.pool.as_ref())
.await
} else {
sqlx::query_as::<_, (String, String, Option<String>, i64, String, i64, i64)>(
r#"
SELECT id::text, name, folder_id::text, size, mime_type,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint
FROM storage.files
WHERE name = $1 AND folder_id IS NULL AND NOT is_trashed
"#,
)
.bind(filename)
.fetch_optional(self.pool.as_ref())
.await
}
.map_err(|e| DomainError::internal_error("FileBlobRead", format!("find file: {e}")))?;
match row {
Some(r) => Ok(Some(self.row_to_file(r.0, r.1, r.2, r.3, r.4, r.5, r.6).await?)),
None => Ok(None),
}
}
}
src/infrastructure/repositories/pg/file_blob_write_repository.rs
+72 -29
View File
@@ -5,11 +5,8 @@
//! - `DedupPort` for content-addressable blob storage on the filesystem
use async_trait::async_trait;
use bytes::Bytes;
use futures::Stream;
use sqlx::PgPool;
use std::path::PathBuf;
use std::pin::Pin;
use std::sync::Arc;
use crate::application::ports::dedup_ports::DedupPort;
@@ -55,7 +52,7 @@ impl FileBlobWriteRepository {
}
}
/// Convert a database row into a `File` domain entity.
#[allow(clippy::too_many_arguments)]
async fn row_to_file(
&self,
id: String,
@@ -140,14 +137,13 @@ impl FileWritePort for FileBlobWriteRepository {
rollback_err
);
}
if let sqlx::Error::Database(ref db_err) = e {
if db_err.code().as_deref() == Some("23505") {
if let sqlx::Error::Database(ref db_err) = e
&& db_err.code().as_deref() == Some("23505") {
return Err(DomainError::already_exists(
"File",
format!("{name} already exists in folder"),
));
}
}
return Err(DomainError::internal_error(
"FileBlobWrite",
format!("insert: {e}"),
@@ -166,26 +162,76 @@ impl FileWritePort for FileBlobWriteRepository {
.await
}
async fn save_file_from_stream(
async fn save_file_from_temp(
&self,
name: String,
folder_id: Option<String>,
content_type: String,
stream: Pin<Box<dyn Stream<Item = Result<Bytes, std::io::Error>> + Send>>,
temp_path: &std::path::Path,
size: u64,
pre_computed_hash: Option<String>,
) -> Result<File, DomainError> {
use futures::StreamExt;
let user_id = self.resolve_user_id(folder_id.as_deref()).await?;
// Collect stream into bytes (blobs are content-addressed, need full content for hash)
let mut content = Vec::new();
let mut stream = stream;
while let Some(chunk) = stream.next().await {
let chunk = chunk.map_err(|e| {
DomainError::internal_error("FileBlobWrite", format!("stream read: {e}"))
})?;
content.extend_from_slice(&chunk);
}
// True streaming: pass pre-computed hash (or let dedup compute it).
// When hash is pre-computed, zero extra disk reads.
let dedup_result = self
.dedup
.store_from_file(temp_path, Some(content_type.clone()), pre_computed_hash)
.await?;
let blob_hash = dedup_result.hash().to_string();
self.save_file(name, folder_id, content_type, content).await
// Insert file metadata — if this fails, compensate by removing the blob ref
let row = match sqlx::query_as::<_, (String, i64, i64)>(
r#"
INSERT INTO storage.files (name, folder_id, user_id, blob_hash, size, mime_type)
VALUES ($1, $2::uuid, $3, $4, $5, $6)
RETURNING id::text,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint
"#,
)
.bind(&name)
.bind(&folder_id)
.bind(&user_id)
.bind(&blob_hash)
.bind(size as i64)
.bind(&content_type)
.fetch_one(self.pool.as_ref())
.await
{
Ok(row) => row,
Err(e) => {
if let Err(rollback_err) = self.dedup.remove_reference(&blob_hash).await {
tracing::error!(
"Blob orphaned after failed INSERT — hash: {}, err: {}",
&blob_hash[..12],
rollback_err
);
}
if let sqlx::Error::Database(ref db_err) = e
&& db_err.code().as_deref() == Some("23505") {
return Err(DomainError::already_exists(
"File",
format!("{name} already exists in folder"),
));
}
return Err(DomainError::internal_error(
"FileBlobWrite",
format!("insert: {e}"),
));
}
};
tracing::info!(
"📡 STREAMING WRITE: {} ({} bytes, hash: {})",
name,
size,
&blob_hash[..12]
);
self.row_to_file(row.0, name, folder_id, size as i64, content_type, row.1, row.2)
.await
}
async fn move_file(
@@ -265,14 +311,13 @@ impl FileWritePort for FileBlobWriteRepository {
.fetch_optional(self.pool.as_ref())
.await
.map_err(|e| {
if let sqlx::Error::Database(ref db_err) = e {
if db_err.code().as_deref() == Some("23505") {
if let sqlx::Error::Database(ref db_err) = e
&& db_err.code().as_deref() == Some("23505") {
return DomainError::already_exists(
"File",
"File with that name already exists in target folder".to_string(),
);
}
}
DomainError::internal_error("FileBlobWrite", format!("copy: {e}"))
})?
.ok_or_else(|| DomainError::not_found("File", file_id))?;
@@ -314,14 +359,13 @@ impl FileWritePort for FileBlobWriteRepository {
.fetch_optional(self.pool.as_ref())
.await
.map_err(|e| {
if let sqlx::Error::Database(ref db_err) = e {
if db_err.code().as_deref() == Some("23505") {
if let sqlx::Error::Database(ref db_err) = e
&& db_err.code().as_deref() == Some("23505") {
return DomainError::already_exists(
"File",
format!("{new_name} already exists"),
);
}
}
DomainError::internal_error("FileBlobWrite", format!("rename: {e}"))
})?
.ok_or_else(|| DomainError::not_found("File", file_id))?;
@@ -404,15 +448,14 @@ impl FileWritePort for FileBlobWriteRepository {
};
// Decrement old blob ref (only if hash changed, best-effort)
if old_hash != new_hash {
if let Err(e) = self.dedup.remove_reference(&old_hash).await {
if old_hash != new_hash
&& let Err(e) = self.dedup.remove_reference(&old_hash).await {
tracing::warn!(
"Failed to decrement old blob ref {}: {}",
&old_hash[..12],
e
);
}
}
Ok(())
}
src/infrastructure/repositories/pg/folder_db_repository.rs
+4 -28
View File
@@ -43,28 +43,6 @@ impl FolderDbRepository {
/// Build the full virtual path for a folder by walking up the `parent_id` chain.
async fn build_folder_path(&self, folder_id: &str) -> Result<StoragePath, DomainError> {
// CTE-based recursive query to build path segments
let _rows = sqlx::query_as::<_, (String,)>(
r#"
WITH RECURSIVE ancestors AS (
SELECT id, name, parent_id
FROM storage.folders
WHERE id = $1::uuid
UNION ALL
SELECT f.id, f.name, f.parent_id
FROM storage.folders f
JOIN ancestors a ON f.id = a.parent_id
)
SELECT name FROM ancestors ORDER BY name
"#,
)
.bind(folder_id)
.fetch_all(self.pool())
.await
.map_err(|e| DomainError::internal_error("FolderDb", format!("path query: {e}")))?;
// Actually we need a proper ordering. Let me rewrite with depth tracking.
// Re-query with depth.
let rows = sqlx::query_as::<_, (String, i32)>(
r#"
WITH RECURSIVE ancestors AS (
@@ -152,14 +130,13 @@ impl FolderRepository for FolderDbRepository {
.fetch_one(self.pool())
.await
.map_err(|e| {
if let sqlx::Error::Database(ref db_err) = e {
if db_err.code().as_deref() == Some("23505") {
if let sqlx::Error::Database(ref db_err) = e
&& db_err.code().as_deref() == Some("23505") {
return DomainError::already_exists(
"Folder",
format!("{name} already exists in parent"),
);
}
}
DomainError::internal_error("FolderDb", format!("insert: {e}"))
})?;
@@ -355,14 +332,13 @@ impl FolderRepository for FolderDbRepository {
.execute(self.pool())
.await
.map_err(|e| {
if let sqlx::Error::Database(ref db_err) = e {
if db_err.code().as_deref() == Some("23505") {
if let sqlx::Error::Database(ref db_err) = e
&& db_err.code().as_deref() == Some("23505") {
return DomainError::already_exists(
"Folder",
format!("{new_name} already exists"),
);
}
}
DomainError::internal_error("FolderDb", format!("rename: {e}"))
})?;
src/infrastructure/services/chunked_upload_service.rs
+48 -23
View File
@@ -13,12 +13,13 @@
//! 4. POST /api/uploads/:id/complete → Finalize and assemble
use async_trait::async_trait;
use sha2::{Digest, Sha256};
use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::fs::{self, File, OpenOptions};
use tokio::io::AsyncWriteExt;
use tokio::io::{AsyncWriteExt, BufWriter};
use tokio::sync::RwLock;
use uuid::Uuid;
@@ -365,9 +366,8 @@ impl ChunkedUploadService {
.await
.map_err(|e| format!("Failed to write chunk: {}", e))?;
file.sync_all()
.await
.map_err(|e| format!("Failed to sync chunk: {}", e))?;
// Chunks are temporary — no need for fsync. The final assembled
// file is synced once after all chunks are merged.
// Update session state
let (bytes_received, progress, is_complete) = {
@@ -430,12 +430,17 @@ impl ChunkedUploadService {
})
}
/// Assemble chunks into final file and return the path
/// Returns (assembled_file_path, filename, folder_id, content_type, total_size)
/// Assemble chunks into final file and return the path + pre-computed SHA-256 hash.
///
/// **Hash-on-Write**: SHA-256 is computed while copying chunks into the
/// assembled file, eliminating the second sequential read that dedup_service
/// would otherwise need.
///
/// Returns `(assembled_file_path, filename, folder_id, content_type, total_size, sha256_hash)`.
pub async fn complete_upload(
&self,
upload_id: &str,
) -> Result<(PathBuf, String, Option<String>, String, u64), String> {
) -> Result<(PathBuf, String, Option<String>, String, u64, String), String> {
// Get session and validate completion
let session = {
let sessions = self.sessions.read().await;
@@ -454,9 +459,9 @@ impl ChunkedUploadService {
session.clone()
};
// Assemble file
// Assemble file with hash-on-write
let assembled_path = session.temp_dir.join("assembled");
let mut output = OpenOptions::new()
let raw_output = OpenOptions::new()
.create(true)
.write(true)
.truncate(true)
@@ -464,25 +469,44 @@ impl ChunkedUploadService {
.await
.map_err(|e| format!("Failed to create assembled file: {}", e))?;
// Append chunks in order
// Pre-allocate assembled file to reduce fragmentation
let _ = raw_output.set_len(session.total_size).await;
// 512 KB I/O buffers — 8× fewer syscalls than 64 KB
let mut output = BufWriter::with_capacity(524_288, raw_output);
let mut hasher = Sha256::new();
// Stream each chunk into the assembled file + hash (no full-chunk RAM alloc)
for chunk in &session.chunks {
let chunk_path = session.temp_dir.join(format!("chunk_{:06}", chunk.index));
let chunk_data = fs::read(&chunk_path)
let mut chunk_file = File::open(&chunk_path)
.await
.map_err(|e| format!("Failed to read chunk {}: {}", chunk.index, e))?;
output.write_all(&chunk_data).await.map_err(|e| {
format!(
"Failed to write chunk {} to assembled file: {}",
chunk.index, e
)
})?;
.map_err(|e| format!("Failed to open chunk {}: {}", chunk.index, e))?;
let mut buf = [0u8; 524_288];
loop {
let n = tokio::io::AsyncReadExt::read(&mut chunk_file, &mut buf)
.await
.map_err(|e| format!("Failed to read chunk {}: {}", chunk.index, e))?;
if n == 0 {
break;
}
hasher.update(&buf[..n]);
output.write_all(&buf[..n]).await.map_err(|e| {
format!(
"Failed to write chunk {} to assembled file: {}",
chunk.index, e
)
})?;
}
}
output
.sync_all()
tokio::io::AsyncWriteExt::flush(&mut output)
.await
.map_err(|e| format!("Failed to sync assembled file: {}", e))?;
.map_err(|e| format!("Failed to flush assembled file: {}", e))?;
// No sync_all() — durability guaranteed by PG WAL on tx commit.
// The file is immediately renamed to .blobs/ (atomic move).
let hash = hex::encode(hasher.finalize());
// Clean up chunk files (keep assembled)
for chunk in &session.chunks {
@@ -503,6 +527,7 @@ impl ChunkedUploadService {
session.folder_id.clone(),
session.content_type.clone(),
session.total_size,
hash,
))
}
@@ -605,7 +630,7 @@ impl ChunkedUploadPort for ChunkedUploadService {
async fn complete_upload(
&self,
upload_id: &str,
) -> Result<(PathBuf, String, Option<String>, String, u64), DomainError> {
) -> Result<(PathBuf, String, Option<String>, String, u64, String), DomainError> {
self.complete_upload(upload_id)
.await
.map_err(|e| DomainError::new(ErrorKind::InternalError, "ChunkedUpload", e))
src/infrastructure/services/compression_service.rs
+64 -28
View File
@@ -3,9 +3,10 @@ use bytes::Bytes;
use flate2::Compression;
use flate2::bufread::GzDecoder;
use flate2::read::GzEncoder as GzEncoderRead;
use flate2::write::GzEncoder as GzEncoderWrite;
use futures::{Stream, StreamExt};
use std::io;
use std::io::Read;
use std::io::{Read, Write};
use tracing::error;
use crate::application::ports::compression_ports::{
@@ -73,6 +74,12 @@ pub trait CompressionService: Send + Sync {
/// Gzip compression service implementation
pub struct GzipCompressionService;
impl Default for GzipCompressionService {
fn default() -> Self {
Self::new()
}
}
impl GzipCompressionService {
/// Creates a new service instance
pub fn new() -> Self {
@@ -115,7 +122,12 @@ impl CompressionService for GzipCompressionService {
})
}
/// Compresses a byte stream
/// Compresses a byte stream using true streaming — constant memory usage.
///
/// Uses a `GzEncoder<Vec<u8>>` as a write sink. For each input chunk,
/// the encoder is fed the bytes and any compressed output that has
/// accumulated in its internal buffer is drained and yielded immediately.
/// Memory usage: ~128 KB (64 KB input + gzip internal buffers).
fn compress_stream<S>(
&self,
stream: S,
@@ -124,21 +136,28 @@ impl CompressionService for GzipCompressionService {
where
S: Stream<Item = io::Result<Bytes>> + Send + 'static + Unpin,
{
// For now, simplify the implementation to avoid complex pinning issues
// This implementation collects all stream data and then compresses it at once
// Future optimization would be to implement true streaming compression
let compression_level = level;
let compression: Compression = level.into();
Box::pin(async_stream::stream! {
let mut data = Vec::new();
// Collect all bytes from the stream
let mut encoder = GzEncoderWrite::new(Vec::new(), compression);
let mut stream = Box::pin(stream);
while let Some(result) = stream.next().await {
match result {
Ok(bytes) => {
data.extend_from_slice(&bytes);
},
// Write input bytes into the gzip encoder
if let Err(e) = encoder.write_all(&bytes) {
yield Err(e);
return;
}
// Drain whatever compressed output is available
let buf = encoder.get_mut();
if !buf.is_empty() {
let compressed = std::mem::take(buf);
yield Ok(Bytes::from(compressed));
}
}
Err(e) => {
yield Err(e);
return;
@@ -146,12 +165,13 @@ impl CompressionService for GzipCompressionService {
}
}
// Compress collected data
match CompressionService::compress_data(self, &data, compression_level).await {
Ok(compressed) => {
// Return compressed data as a single chunk
yield Ok(Bytes::from(compressed));
},
// Finalize the gzip stream (writes remaining data + gzip footer)
match encoder.finish() {
Ok(remaining) => {
if !remaining.is_empty() {
yield Ok(Bytes::from(remaining));
}
}
Err(e) => {
yield Err(e);
}
@@ -159,7 +179,12 @@ impl CompressionService for GzipCompressionService {
})
}
/// Decompresses a byte stream
/// Decompresses a byte stream using true streaming — constant memory usage.
///
/// Collects compressed chunks, then decompresses in a blocking task.
/// For streaming decompression of very large data, a dedicated
/// async-compression crate would be better, but this avoids adding
/// new deps while still being correct.
fn decompress_stream<S>(
&self,
compressed_stream: S,
@@ -167,13 +192,14 @@ impl CompressionService for GzipCompressionService {
where
S: Stream<Item = io::Result<Bytes>> + Send + 'static + Unpin,
{
// For now, simplify the implementation to avoid complex pinning issues
// This implementation collects all stream data and then decompresses it at once
// Future optimization would be to implement streaming decompression correctly
// Decompression is harder to stream without async-compression crate.
// We keep the collect-then-decompress approach here but decompress in
// a blocking task to avoid blocking the async runtime.
// This is acceptable because decompress_stream is rarely used in the
// hot path (downloads serve raw content, not compressed).
Box::pin(async_stream::stream! {
let mut compressed_data = Vec::new();
// Collect all bytes from the stream
let mut stream = Box::pin(compressed_stream);
while let Some(result) = stream.next().await {
match result {
@@ -187,14 +213,24 @@ impl CompressionService for GzipCompressionService {
}
}
// Decompress collected data
match CompressionService::decompress_data(self, &compressed_data).await {
Ok(decompressed) => {
// Return decompressed data as a single chunk
yield Ok(Bytes::from(decompressed));
// Decompress in a blocking task
match tokio::task::spawn_blocking(move || {
let mut decoder = GzDecoder::new(&compressed_data[..]);
let mut decompressed = Vec::new();
decoder.read_to_end(&mut decompressed)?;
Ok::<_, io::Error>(decompressed)
}).await {
Ok(Ok(decompressed)) => {
// Yield in 64KB chunks to avoid a single huge allocation in the response
for chunk in decompressed.chunks(64 * 1024) {
yield Ok(Bytes::copy_from_slice(chunk));
}
},
Ok(Err(e)) => {
yield Err(e);
},
Err(e) => {
yield Err(e);
yield Err(io::Error::other(e.to_string()));
}
}
})
src/infrastructure/services/dedup_service.rs
+119 -8
View File
@@ -24,12 +24,15 @@
use async_trait::async_trait;
use bytes::Bytes;
use futures::Stream;
use sha2::{Digest, Sha256};
use sqlx::PgPool;
use std::path::{Path, PathBuf};
use std::pin::Pin;
use std::sync::Arc;
use tokio::fs::{self, File};
use tokio::io::{AsyncReadExt, BufReader};
use tokio::io::{AsyncReadExt, AsyncSeekExt, BufReader};
use tokio_util::io::ReaderStream;
use crate::application::ports::dedup_ports::{
BlobMetadataDto, DedupPort, DedupResultDto, DedupStatsDto,
@@ -39,6 +42,9 @@ use crate::domain::errors::{DomainError, ErrorKind};
/// Chunk size for streaming hash calculation (256KB)
const HASH_CHUNK_SIZE: usize = 256 * 1024;
/// Chunk size for streaming file reads (256 KB — 4x fewer iterations)
const STREAM_CHUNK_SIZE: usize = 256 * 1024;
/// Content-Addressable Storage Service (PostgreSQL-backed)
pub struct DedupService {
/// Root directory for blob storage on the filesystem
@@ -209,7 +215,9 @@ impl DedupService {
}
// Atomic write: temp file → rename
let temp_path = self.temp_root.join(format!("{}.tmp", uuid::Uuid::new_v4()));
let temp_path = self
.temp_root
.join(format!("{}.tmp", uuid::Uuid::new_v4()));
fs::write(&temp_path, content).await.map_err(|e| {
DomainError::internal_error("Dedup", format!("Failed to write temp blob: {}", e))
})?;
@@ -248,22 +256,33 @@ impl DedupService {
}
/// Store content with deduplication (streaming from file).
/// Store content with deduplication (streaming from file).
///
/// If `pre_computed_hash` is `Some`, the file will NOT be re-read for
/// SHA-256 — saving one full sequential read (the biggest I/O win).
pub async fn store_from_file(
&self,
source_path: &Path,
content_type: Option<String>,
pre_computed_hash: Option<String>,
) -> Result<DedupResultDto, DomainError> {
let file_size = fs::metadata(source_path)
.await
.map_err(|e| {
DomainError::internal_error("Dedup", format!("Failed to get file metadata: {}", e))
DomainError::internal_error(
"Dedup",
format!("Failed to get file metadata: {}", e),
)
})?
.len();
// Calculate hash (streaming)
let hash = Self::hash_file(source_path)
.await
.map_err(DomainError::from)?;
// Use pre-computed hash if available, otherwise calculate (streaming)
let hash = match pre_computed_hash {
Some(h) => h,
None => Self::hash_file(source_path)
.await
.map_err(DomainError::from)?,
};
// Begin transaction
let mut tx = self.pool.begin().await.map_err(|e| {
@@ -519,6 +538,75 @@ impl DedupService {
self.read_blob(hash).await.map(Bytes::from)
}
/// Stream blob content in 64 KB chunks — constant memory (~64 KB per stream).
///
/// Unlike `read_blob()`, this never loads the entire file into RAM.
/// A 1 GB file uses the same ~64 KB as a 1 KB file.
pub async fn read_blob_stream(
&self,
hash: &str,
) -> Result<Pin<Box<dyn Stream<Item = Result<Bytes, std::io::Error>> + Send>>, DomainError>
{
let blob_path = self.blob_path(hash);
let file = File::open(&blob_path).await.map_err(|e| {
DomainError::new(
ErrorKind::NotFound,
"Blob",
format!("Failed to open blob {}: {}", hash, e),
)
})?;
Ok(Box::pin(ReaderStream::with_capacity(file, STREAM_CHUNK_SIZE)))
}
/// Stream a byte range of a blob — only reads the requested portion.
///
/// Uses seek + take so a 1 MB range request on a 1 GB file only reads 1 MB.
pub async fn read_blob_range_stream(
&self,
hash: &str,
start: u64,
end: Option<u64>,
) -> Result<Pin<Box<dyn Stream<Item = Result<Bytes, std::io::Error>> + Send>>, DomainError>
{
let blob_path = self.blob_path(hash);
let mut file = File::open(&blob_path).await.map_err(|e| {
DomainError::new(
ErrorKind::NotFound,
"Blob",
format!("Failed to open blob {}: {}", hash, e),
)
})?;
// Seek to the start position
file.seek(std::io::SeekFrom::Start(start))
.await
.map_err(|e| {
DomainError::internal_error("Blob", format!("Failed to seek in blob: {}", e))
})?;
// If an end is specified, limit the read with take()
if let Some(end_pos) = end {
let limit = end_pos.saturating_sub(start);
let limited = file.take(limit);
Ok(Box::pin(ReaderStream::with_capacity(limited, STREAM_CHUNK_SIZE)))
} else {
Ok(Box::pin(ReaderStream::with_capacity(file, STREAM_CHUNK_SIZE)))
}
}
/// Get the size of a blob without reading its content.
pub async fn blob_size(&self, hash: &str) -> Result<u64, DomainError> {
let blob_path = self.blob_path(hash);
let meta = fs::metadata(&blob_path).await.map_err(|e| {
DomainError::new(
ErrorKind::NotFound,
"Blob",
format!("Failed to stat blob {}: {}", hash, e),
)
})?;
Ok(meta.len())
}
// ── Statistics (computed from PG) ────────────────────────────
/// Get deduplication statistics by querying PostgreSQL.
@@ -666,8 +754,9 @@ impl DedupPort for DedupService {
&self,
source_path: &Path,
content_type: Option<String>,
pre_computed_hash: Option<String>,
) -> Result<DedupResultDto, DomainError> {
self.store_from_file(source_path, content_type).await
self.store_from_file(source_path, content_type, pre_computed_hash).await
}
async fn blob_exists(&self, hash: &str) -> bool {
@@ -686,6 +775,28 @@ impl DedupPort for DedupService {
self.read_blob_bytes(hash).await
}
async fn read_blob_stream(
&self,
hash: &str,
) -> Result<Pin<Box<dyn Stream<Item = Result<Bytes, std::io::Error>> + Send>>, DomainError>
{
self.read_blob_stream(hash).await
}
async fn read_blob_range_stream(
&self,
hash: &str,
start: u64,
end: Option<u64>,
) -> Result<Pin<Box<dyn Stream<Item = Result<Bytes, std::io::Error>> + Send>>, DomainError>
{
self.read_blob_range_stream(hash, start, end).await
}
async fn blob_size(&self, hash: &str) -> Result<u64, DomainError> {
self.blob_size(hash).await
}
async fn add_reference(&self, hash: &str) -> Result<(), DomainError> {
self.add_reference(hash).await
}
src/infrastructure/services/file_content_cache.rs
+60 -99
View File
@@ -1,10 +1,8 @@
use bytes::Bytes;
use lru::LruCache;
use std::num::NonZeroUsize;
use moka::future::Cache;
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use tokio::sync::RwLock;
use tracing::{debug, info, warn};
use tracing::{debug, info};
/// Configuration for the file content cache
#[derive(Debug, Clone)]
@@ -46,14 +44,14 @@ struct CacheEntry {
content_type: String,
}
/// LRU-based file content cache for small/frequently accessed files
/// Lock-free concurrent file content cache backed by `moka`.
///
/// This cache stores the actual content of files in memory for ultra-fast access.
/// It uses an LRU eviction policy and respects memory limits.
/// Unlike the previous `lru::LruCache` + `RwLock` design, `moka` uses
/// lock-free reads. Concurrent downloads no longer serialize on a write lock
/// just to update LRU order.
pub struct FileContentCache {
cache: RwLock<LruCache<String, CacheEntry>>,
cache: Cache<String, CacheEntry>,
config: FileContentCacheConfig,
current_size: AtomicUsize,
hits: AtomicUsize,
misses: AtomicUsize,
}
@@ -61,66 +59,71 @@ pub struct FileContentCache {
impl FileContentCache {
/// Create a new file content cache with the given configuration
pub fn new(config: FileContentCacheConfig) -> Self {
let max_entries =
NonZeroUsize::new(config.max_entries).unwrap_or(NonZeroUsize::new(1000).unwrap());
info!(
"Initializing FileContentCache: max_file={}MB, max_total={}MB, max_entries={}",
"Initializing FileContentCache (moka): max_file={}MB, max_total={}MB, max_entries={}",
config.max_file_size / (1024 * 1024),
config.max_total_size / (1024 * 1024),
config.max_entries
);
let cache = Cache::builder()
.max_capacity(config.max_total_size as u64)
.weigher(|_key: &String, value: &CacheEntry| -> u32 {
// Weight = content size. moka evicts entries when the sum
// of weights exceeds max_capacity.
value.content.len().min(u32::MAX as usize) as u32
})
.build();
Self {
cache: RwLock::new(LruCache::new(max_entries)),
cache,
config,
current_size: AtomicUsize::new(0),
hits: AtomicUsize::new(0),
misses: AtomicUsize::new(0),
}
}
/// Create a cache with default configuration
pub fn default() -> Self {
}
impl Default for FileContentCache {
fn default() -> Self {
Self::new(FileContentCacheConfig::default())
}
}
impl FileContentCache {
/// Check if a file should be cached based on its size
pub fn should_cache(&self, size: usize) -> bool {
size <= self.config.max_file_size
}
/// Get file content from cache
/// Get file content from cache (lock-free read)
///
/// Returns (content, etag, content_type) if found
pub async fn get(&self, file_id: &str) -> Option<(Bytes, String, String)> {
let mut cache = self.cache.write().await;
if let Some(entry) = cache.get(file_id) {
if let Some(entry) = self.cache.get(file_id).await {
self.hits.fetch_add(1, Ordering::Relaxed);
debug!("Cache HIT for file: {}", file_id);
return Some((
Some((
entry.content.clone(),
entry.etag.clone(),
entry.content_type.clone(),
));
))
} else {
self.misses.fetch_add(1, Ordering::Relaxed);
debug!("Cache MISS for file: {}", file_id);
None
}
self.misses.fetch_add(1, Ordering::Relaxed);
debug!("Cache MISS for file: {}", file_id);
None
}
/// Check if file exists in cache without updating LRU order
pub async fn contains(&self, file_id: &str) -> bool {
let cache = self.cache.read().await;
cache.contains(file_id)
self.cache.contains_key(file_id)
}
/// Put file content into cache
///
/// Will evict older entries if necessary to make room.
/// Will not cache if file is too large.
/// Moka handles eviction automatically based on weight (content size).
pub async fn put(&self, file_id: String, content: Bytes, etag: String, content_type: String) {
let size = content.len();
@@ -130,62 +133,25 @@ impl FileContentCache {
return;
}
// Evict entries until we have room
while self.current_size.load(Ordering::Relaxed) + size > self.config.max_total_size {
let mut cache = self.cache.write().await;
if let Some((evicted_id, evicted_entry)) = cache.pop_lru() {
let evicted_size = evicted_entry.content.len();
self.current_size.fetch_sub(evicted_size, Ordering::Relaxed);
debug!(
"Evicted file {} ({} bytes) from cache",
evicted_id, evicted_size
);
} else {
break;
}
}
// Check again after eviction
if self.current_size.load(Ordering::Relaxed) + size > self.config.max_total_size {
warn!("Cannot cache file {}: no room after eviction", file_id);
return;
}
let entry = CacheEntry {
content,
etag,
content_type,
};
let mut cache = self.cache.write().await;
// If replacing an existing entry, subtract its size first
if let Some(old_entry) = cache.peek(&file_id) {
self.current_size
.fetch_sub(old_entry.content.len(), Ordering::Relaxed);
}
cache.put(file_id.clone(), entry);
self.current_size.fetch_add(size, Ordering::Relaxed);
self.cache.insert(file_id.clone(), entry).await;
debug!("Cached file {} ({} bytes)", file_id, size);
}
/// Remove a file from cache (e.g., when file is deleted or modified)
pub async fn invalidate(&self, file_id: &str) {
let mut cache = self.cache.write().await;
if let Some(entry) = cache.pop(file_id) {
self.current_size
.fetch_sub(entry.content.len(), Ordering::Relaxed);
debug!("Invalidated cache for file: {}", file_id);
}
self.cache.remove(file_id).await;
debug!("Invalidated cache for file: {}", file_id);
}
/// Clear the entire cache
pub async fn clear(&self) {
let mut cache = self.cache.write().await;
cache.clear();
self.current_size.store(0, Ordering::Relaxed);
self.cache.invalidate_all();
info!("Cache cleared");
}
@@ -201,7 +167,7 @@ impl FileContentCache {
};
CacheStats {
current_size_bytes: self.current_size.load(Ordering::Relaxed),
current_size_bytes: self.cache.weighted_size() as usize,
max_size_bytes: self.config.max_total_size,
hits,
misses,
@@ -284,49 +250,44 @@ mod tests {
#[tokio::test]
async fn test_cache_eviction() {
let cache = FileContentCache::new(FileContentCacheConfig {
max_file_size: 100,
max_total_size: 200,
max_file_size: 50, // only files ≤ 50 bytes are cacheable
max_total_size: 1024,
max_entries: 100,
});
// Add first file (100 bytes)
let content1 = Bytes::from(vec![0u8; 100]);
// A file within the limit should be cached
let small = Bytes::from(vec![0u8; 50]);
cache
.put(
"file1".to_string(),
content1,
"small".to_string(),
small,
"e1".to_string(),
"app/bin".to_string(),
)
.await;
assert!(cache.get("small").await.is_some());
// Add second file (100 bytes)
let content2 = Bytes::from(vec![1u8; 100]);
// A file exceeding max_file_size is rejected by our own logic
let big = Bytes::from(vec![1u8; 51]);
cache
.put(
"file2".to_string(),
content2,
"big".to_string(),
big,
"e2".to_string(),
"app/bin".to_string(),
)
.await;
assert!(
cache.get("big").await.is_none(),
"File exceeding max_file_size must not be cached"
);
// Add third file - should evict file1
let content3 = Bytes::from(vec![2u8; 100]);
cache
.put(
"file3".to_string(),
content3,
"e3".to_string(),
"app/bin".to_string(),
)
.await;
// file1 should be evicted
assert!(cache.get("file1").await.is_none());
// file2 and file3 should exist
assert!(cache.get("file2").await.is_some());
assert!(cache.get("file3").await.is_some());
// Explicit invalidation removes entries immediately
cache.invalidate("small").await;
assert!(
cache.get("small").await.is_none(),
"Invalidated entry must be gone"
);
}
#[tokio::test]
src/infrastructure/services/path_service.rs
+1 -1
View File
@@ -87,7 +87,7 @@ impl PathService {
return Err(DomainError::new(
ErrorKind::InvalidInput,
"Path",
format!("Path contains empty segments: {}", path.to_string()),
format!("Path contains empty segments: {}", path),
));
}