perf: stream_files_in_subtree — replace Vec<File> with async Stream
Replace list_files_in_subtree (fetch_all → Vec) with stream_files_in_subtree that returns a Pin<Box<dyn Stream<Item = Result<File/FileDto>>>> backed by a PostgreSQL cursor via sqlx::fetch(). Changes: - FileReadPort::stream_files_in_subtree() returns streaming cursor (no default) - FileRetrievalUseCase::stream_files_in_subtree() maps File→FileDto on the fly - FileBlobReadRepository: async_stream::try_stream! + sqlx::fetch() cursor - batch_operations: consume stream into HashMap incrementally - zip_service: consume stream into HashMap incrementally - All stubs/mocks updated (return empty stream) Eliminates: - Double allocation: Vec<(9-tuple)> + Vec<File> materialized simultaneously - Unbounded RAM proportional to subtree size (was ~500 bytes × N files) - Latency: callers blocked until last row fetched from PG RAM is now O(folders) for the HashMap, not O(files).
This commit is contained in:
@@ -154,12 +154,15 @@ pub trait FileRetrievalUseCase: Send + Sync + 'static {
|
|||||||
end: Option<u64>,
|
end: Option<u64>,
|
||||||
) -> Result<Box<dyn Stream<Item = Result<Bytes, std::io::Error>> + Send>, DomainError>;
|
) -> Result<Box<dyn Stream<Item = Result<Bytes, std::io::Error>> + Send>, DomainError>;
|
||||||
|
|
||||||
/// Lists every file in the subtree rooted at `folder_id`.
|
/// Streams every file in the subtree rooted at `folder_id`.
|
||||||
///
|
///
|
||||||
/// Default: falls back to `list_files(Some(folder_id))` (one level).
|
/// Returns a streaming cursor — RAM stays O(1) per row. Callers
|
||||||
async fn list_files_in_subtree(&self, folder_id: &str) -> Result<Vec<FileDto>, DomainError> {
|
/// consume incrementally (e.g. group into a HashMap by folder_id)
|
||||||
self.list_files(Some(folder_id)).await
|
/// without materializing the full result set.
|
||||||
}
|
async fn stream_files_in_subtree(
|
||||||
|
&self,
|
||||||
|
folder_id: &str,
|
||||||
|
) -> Result<Pin<Box<dyn Stream<Item = Result<FileDto, DomainError>> + Send>>, DomainError>;
|
||||||
|
|
||||||
/// Lists files in a folder with LIMIT/OFFSET pagination.
|
/// Lists files in a folder with LIMIT/OFFSET pagination.
|
||||||
///
|
///
|
||||||
|
|||||||
@@ -3,6 +3,7 @@ use bytes::Bytes;
|
|||||||
use futures::Stream;
|
use futures::Stream;
|
||||||
use serde_json::Value;
|
use serde_json::Value;
|
||||||
use std::path::PathBuf;
|
use std::path::PathBuf;
|
||||||
|
use std::pin::Pin;
|
||||||
|
|
||||||
use crate::application::dtos::search_dto::SearchCriteriaDto;
|
use crate::application::dtos::search_dto::SearchCriteriaDto;
|
||||||
use crate::common::errors::DomainError;
|
use crate::common::errors::DomainError;
|
||||||
@@ -94,15 +95,18 @@ pub trait FileReadPort: Send + Sync + 'static {
|
|||||||
Ok(all.into_iter().skip(start).take(end - start).collect())
|
Ok(all.into_iter().skip(start).take(end - start).collect())
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Lists every file in the subtree rooted at `folder_id`.
|
/// Streams every file in the subtree rooted at `folder_id`.
|
||||||
///
|
///
|
||||||
/// Uses an ltree `<@` join against `storage.folders` so the entire
|
/// Uses an ltree `<@` join against `storage.folders` so the entire
|
||||||
/// subtree is fetched in a single GiST-indexed query.
|
/// subtree is resolved in a single GiST-indexed query, but rows are
|
||||||
|
/// delivered via a PostgreSQL cursor — RAM stays O(1) per row.
|
||||||
///
|
///
|
||||||
/// Default: falls back to `list_files(Some(folder_id))` (one level).
|
/// Callers consume the stream incrementally (e.g. build a HashMap
|
||||||
async fn list_files_in_subtree(&self, folder_id: &str) -> Result<Vec<File>, DomainError> {
|
/// keyed by folder_id) without ever materializing the full Vec.
|
||||||
self.list_files(Some(folder_id)).await
|
async fn stream_files_in_subtree(
|
||||||
}
|
&self,
|
||||||
|
folder_id: &str,
|
||||||
|
) -> Result<Pin<Box<dyn Stream<Item = Result<File, DomainError>> + Send>>, DomainError>;
|
||||||
|
|
||||||
/// Search files with pagination and filtering at database level.
|
/// Search files with pagination and filtering at database level.
|
||||||
///
|
///
|
||||||
|
|||||||
@@ -792,29 +792,34 @@ impl BatchOperationService {
|
|||||||
|
|
||||||
/// Adds an entire folder subtree to the ZIP using 2 bulk SQL queries
|
/// Adds an entire folder subtree to the ZIP using 2 bulk SQL queries
|
||||||
/// (ltree `<@`) instead of N+1 per-folder traversal.
|
/// (ltree `<@`) instead of N+1 per-folder traversal.
|
||||||
|
///
|
||||||
|
/// Files are streamed from the DB cursor — RAM for the file list is
|
||||||
|
/// proportional to the number of *folders* (HashMap keys), not files.
|
||||||
async fn add_folder_subtree_to_zip(
|
async fn add_folder_subtree_to_zip(
|
||||||
&self,
|
&self,
|
||||||
zip: &mut ZipFileWriter<tokio_util::compat::Compat<BufWriter<tokio::fs::File>>>,
|
zip: &mut ZipFileWriter<tokio_util::compat::Compat<BufWriter<tokio::fs::File>>>,
|
||||||
folder_id: &str,
|
folder_id: &str,
|
||||||
root_folder: &FolderDto,
|
root_folder: &FolderDto,
|
||||||
) -> Result<(), BatchOperationError> {
|
) -> Result<(), BatchOperationError> {
|
||||||
// Bulk-fetch entire subtree (2 queries total)
|
// Bulk-fetch folder tree (small — one entry per folder)
|
||||||
let all_folders = self
|
let all_folders = self
|
||||||
.folder_service
|
.folder_service
|
||||||
.list_subtree_folders(folder_id)
|
.list_subtree_folders(folder_id)
|
||||||
.await
|
.await
|
||||||
.map_err(BatchOperationError::Domain)?;
|
.map_err(BatchOperationError::Domain)?;
|
||||||
|
|
||||||
let all_files = self
|
// Stream files from DB cursor — O(1) per row
|
||||||
|
let mut file_stream = self
|
||||||
.file_retrieval
|
.file_retrieval
|
||||||
.list_files_in_subtree(folder_id)
|
.stream_files_in_subtree(folder_id)
|
||||||
.await
|
.await
|
||||||
.map_err(BatchOperationError::Domain)?;
|
.map_err(BatchOperationError::Domain)?;
|
||||||
|
|
||||||
// Group files by folder_id
|
// Group files by folder_id incrementally from the stream
|
||||||
let mut files_by_folder: HashMap<String, Vec<FileDto>> =
|
let mut files_by_folder: HashMap<String, Vec<FileDto>> =
|
||||||
HashMap::with_capacity(all_folders.len());
|
HashMap::with_capacity(all_folders.len());
|
||||||
for file in all_files {
|
while let Some(file) = file_stream.next().await {
|
||||||
|
let file = file.map_err(BatchOperationError::Domain)?;
|
||||||
let fid = file.folder_id.clone().unwrap_or_default();
|
let fid = file.folder_id.clone().unwrap_or_default();
|
||||||
files_by_folder.entry(fid).or_default().push(file);
|
files_by_folder.entry(fid).or_default().push(file);
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1,6 +1,7 @@
|
|||||||
use async_trait::async_trait;
|
use async_trait::async_trait;
|
||||||
use bytes::{Bytes, BytesMut};
|
use bytes::{Bytes, BytesMut};
|
||||||
use futures::{Stream, StreamExt};
|
use futures::{Stream, StreamExt};
|
||||||
|
use std::pin::Pin;
|
||||||
use std::sync::Arc;
|
use std::sync::Arc;
|
||||||
|
|
||||||
use crate::application::dtos::file_dto::FileDto;
|
use crate::application::dtos::file_dto::FileDto;
|
||||||
@@ -308,9 +309,13 @@ impl FileRetrievalUseCase for FileRetrievalService {
|
|||||||
self.file_read.get_file_range_stream(id, start, end).await
|
self.file_read.get_file_range_stream(id, start, end).await
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn list_files_in_subtree(&self, folder_id: &str) -> Result<Vec<FileDto>, DomainError> {
|
async fn stream_files_in_subtree(
|
||||||
let files = self.file_read.list_files_in_subtree(folder_id).await?;
|
&self,
|
||||||
Ok(files.into_iter().map(FileDto::from).collect())
|
folder_id: &str,
|
||||||
|
) -> Result<Pin<Box<dyn Stream<Item = Result<FileDto, DomainError>> + Send>>, DomainError> {
|
||||||
|
let inner = self.file_read.stream_files_in_subtree(folder_id).await?;
|
||||||
|
let mapped = inner.map(|r| r.map(FileDto::from));
|
||||||
|
Ok(Box::pin(mapped))
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn list_files_batch(
|
async fn list_files_batch(
|
||||||
|
|||||||
@@ -505,6 +505,16 @@ mod tests {
|
|||||||
) -> Result<usize, DomainError> {
|
) -> Result<usize, DomainError> {
|
||||||
Ok(0)
|
Ok(0)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
async fn stream_files_in_subtree(
|
||||||
|
&self,
|
||||||
|
_folder_id: &str,
|
||||||
|
) -> Result<
|
||||||
|
std::pin::Pin<Box<dyn futures::Stream<Item = Result<crate::domain::entities::file::File, DomainError>> + Send>>,
|
||||||
|
DomainError,
|
||||||
|
> {
|
||||||
|
Ok(Box::pin(futures::stream::empty()))
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
#[async_trait]
|
#[async_trait]
|
||||||
|
|||||||
@@ -4,6 +4,7 @@ use chrono::Utc;
|
|||||||
use futures::Stream;
|
use futures::Stream;
|
||||||
use std::collections::HashMap;
|
use std::collections::HashMap;
|
||||||
use std::path::PathBuf;
|
use std::path::PathBuf;
|
||||||
|
use std::pin::Pin;
|
||||||
use std::sync::{Arc, Mutex};
|
use std::sync::{Arc, Mutex};
|
||||||
use uuid::Uuid;
|
use uuid::Uuid;
|
||||||
|
|
||||||
@@ -205,6 +206,13 @@ impl FileReadPort for MockFileRepository {
|
|||||||
) -> std::result::Result<usize, DomainError> {
|
) -> std::result::Result<usize, DomainError> {
|
||||||
Ok(0)
|
Ok(0)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
async fn stream_files_in_subtree(
|
||||||
|
&self,
|
||||||
|
_folder_id: &str,
|
||||||
|
) -> std::result::Result<Pin<Box<dyn Stream<Item = std::result::Result<File, DomainError>> + Send>>, DomainError> {
|
||||||
|
Ok(Box::pin(futures::stream::empty()))
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
#[async_trait]
|
#[async_trait]
|
||||||
|
|||||||
@@ -121,6 +121,13 @@ impl FileReadPort for StubFileReadPort {
|
|||||||
) -> Result<usize, DomainError> {
|
) -> Result<usize, DomainError> {
|
||||||
Ok(0)
|
Ok(0)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
async fn stream_files_in_subtree(
|
||||||
|
&self,
|
||||||
|
_folder_id: &str,
|
||||||
|
) -> Result<Pin<Box<dyn Stream<Item = Result<File, DomainError>> + Send>>, DomainError> {
|
||||||
|
Ok(Box::pin(futures::stream::empty()))
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// ---------------------------------------------------------------------------
|
// ---------------------------------------------------------------------------
|
||||||
@@ -526,6 +533,13 @@ impl FileRetrievalUseCase for StubFileRetrievalUseCase {
|
|||||||
async fn get_file_by_path(&self, _path: &str) -> Result<FileDto, DomainError> {
|
async fn get_file_by_path(&self, _path: &str) -> Result<FileDto, DomainError> {
|
||||||
Err(DomainError::not_found("File", "stub"))
|
Err(DomainError::not_found("File", "stub"))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
async fn stream_files_in_subtree(
|
||||||
|
&self,
|
||||||
|
_folder_id: &str,
|
||||||
|
) -> Result<Pin<Box<dyn Stream<Item = Result<FileDto, DomainError>> + Send>>, DomainError> {
|
||||||
|
Ok(Box::pin(futures::stream::empty()))
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// ---------------------------------------------------------------------------
|
// ---------------------------------------------------------------------------
|
||||||
|
|||||||
@@ -9,9 +9,10 @@
|
|||||||
|
|
||||||
use async_trait::async_trait;
|
use async_trait::async_trait;
|
||||||
use bytes::Bytes;
|
use bytes::Bytes;
|
||||||
use futures::Stream;
|
use futures::{Stream, TryStreamExt};
|
||||||
use moka::sync::Cache;
|
use moka::sync::Cache;
|
||||||
use sqlx::PgPool;
|
use sqlx::PgPool;
|
||||||
|
use std::pin::Pin;
|
||||||
use std::sync::Arc;
|
use std::sync::Arc;
|
||||||
use std::time::Duration;
|
use std::time::Duration;
|
||||||
|
|
||||||
@@ -437,45 +438,50 @@ impl FileReadPort for FileBlobReadRepository {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Lists every file in the subtree rooted at `folder_id` (inclusive).
|
/// Streams every file in the subtree rooted at `folder_id`.
|
||||||
///
|
///
|
||||||
/// Single GiST-indexed query via ltree `<@`.
|
/// Single GiST-indexed query via ltree `<@`. Results are delivered
|
||||||
/// Ordered by `(fo.path, fi.name)` so callers iterate in directory order.
|
/// through a PostgreSQL cursor — RAM stays O(1) per row.
|
||||||
async fn list_files_in_subtree(&self, folder_id: &str) -> Result<Vec<File>, DomainError> {
|
async fn stream_files_in_subtree(
|
||||||
let rows: Vec<(
|
&self,
|
||||||
String,
|
folder_id: &str,
|
||||||
String,
|
) -> Result<Pin<Box<dyn Stream<Item = Result<File, DomainError>> + Send>>, DomainError> {
|
||||||
Option<String>,
|
let pool = Arc::clone(&self.pool);
|
||||||
Option<String>,
|
let folder_id = folder_id.to_owned();
|
||||||
i64,
|
|
||||||
String,
|
|
||||||
i64,
|
|
||||||
i64,
|
|
||||||
Option<String>,
|
|
||||||
)> = sqlx::query_as(
|
|
||||||
r#"
|
|
||||||
SELECT fi.id::text, fi.name, fi.folder_id::text, fo.path,
|
|
||||||
fi.size, fi.mime_type,
|
|
||||||
EXTRACT(EPOCH FROM fi.created_at)::bigint,
|
|
||||||
EXTRACT(EPOCH FROM fi.updated_at)::bigint,
|
|
||||||
fi.user_id::text
|
|
||||||
FROM storage.files fi
|
|
||||||
JOIN storage.folders fo ON fo.id = fi.folder_id
|
|
||||||
WHERE fo.lpath <@ (SELECT lpath FROM storage.folders WHERE id = $1::uuid)
|
|
||||||
AND NOT fi.is_trashed
|
|
||||||
ORDER BY fo.path, fi.name
|
|
||||||
"#,
|
|
||||||
)
|
|
||||||
.bind(folder_id)
|
|
||||||
.fetch_all(self.pool.as_ref())
|
|
||||||
.await
|
|
||||||
.map_err(|e| DomainError::internal_error("FileBlobRead", format!("subtree files: {e}")))?;
|
|
||||||
|
|
||||||
rows.into_iter()
|
let stream = async_stream::try_stream! {
|
||||||
.map(|(id, name, fid, fpath, size, mime, ca, ma, uid)| {
|
let mut row_stream = sqlx::query_as::<_, (
|
||||||
Self::row_to_file(id, name, fid, fpath, size, mime, ca, ma, uid)
|
String, String, Option<String>, Option<String>,
|
||||||
})
|
i64, String, i64, i64, Option<String>,
|
||||||
.collect()
|
)>(
|
||||||
|
r#"
|
||||||
|
SELECT fi.id::text, fi.name, fi.folder_id::text, fo.path,
|
||||||
|
fi.size, fi.mime_type,
|
||||||
|
EXTRACT(EPOCH FROM fi.created_at)::bigint,
|
||||||
|
EXTRACT(EPOCH FROM fi.updated_at)::bigint,
|
||||||
|
fi.user_id::text
|
||||||
|
FROM storage.files fi
|
||||||
|
JOIN storage.folders fo ON fo.id = fi.folder_id
|
||||||
|
WHERE fo.lpath <@ (SELECT lpath FROM storage.folders WHERE id = $1::uuid)
|
||||||
|
AND NOT fi.is_trashed
|
||||||
|
ORDER BY fo.path, fi.name
|
||||||
|
"#,
|
||||||
|
)
|
||||||
|
.bind(&folder_id)
|
||||||
|
.fetch(pool.as_ref());
|
||||||
|
|
||||||
|
while let Some(row) = row_stream.try_next().await.map_err(|e| {
|
||||||
|
DomainError::internal_error("FileBlobRead", format!("subtree stream: {e}"))
|
||||||
|
})? {
|
||||||
|
let (id, name, fid, fpath, size, mime, ca, ma, uid) = row;
|
||||||
|
let file = FileBlobReadRepository::row_to_file(
|
||||||
|
id, name, fid, fpath, size, mime, ca, ma, uid,
|
||||||
|
)?;
|
||||||
|
yield file;
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
Ok(Box::pin(stream))
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Search files with filtering and pagination at database level.
|
/// Search files with filtering and pagination at database level.
|
||||||
|
|||||||
@@ -94,33 +94,36 @@ impl ZipService {
|
|||||||
}
|
}
|
||||||
};
|
};
|
||||||
|
|
||||||
// ── 1. Bulk-fetch the entire subtree (2 queries total) ───────────
|
// ── 1. Bulk-fetch folder tree (small — one entry per folder) ────
|
||||||
let all_folders = self
|
let all_folders = self
|
||||||
.folder_service
|
.folder_service
|
||||||
.list_subtree_folders(folder_id)
|
.list_subtree_folders(folder_id)
|
||||||
.await
|
.await
|
||||||
.map_err(|e| ZipError::FolderContentsError(format!("subtree folders: {}", e)))?;
|
.map_err(|e| ZipError::FolderContentsError(format!("subtree folders: {}", e)))?;
|
||||||
|
|
||||||
let all_files = self
|
// ── 2. Stream files from DB cursor — O(1) per row ───────────────
|
||||||
|
let mut file_stream = self
|
||||||
.file_service
|
.file_service
|
||||||
.list_files_in_subtree(folder_id)
|
.stream_files_in_subtree(folder_id)
|
||||||
.await
|
.await
|
||||||
.map_err(|e| ZipError::FolderContentsError(format!("subtree files: {}", e)))?;
|
.map_err(|e| ZipError::FolderContentsError(format!("subtree files: {}", e)))?;
|
||||||
|
|
||||||
|
// Group files by folder_id incrementally from the stream
|
||||||
|
let mut files_by_folder: HashMap<String, Vec<FileDto>> =
|
||||||
|
HashMap::with_capacity(all_folders.len());
|
||||||
|
while let Some(file) = file_stream.next().await {
|
||||||
|
let file =
|
||||||
|
file.map_err(|e| ZipError::FolderContentsError(format!("subtree file: {}", e)))?;
|
||||||
|
let fid = file.folder_id.clone().unwrap_or_default();
|
||||||
|
files_by_folder.entry(fid).or_default().push(file);
|
||||||
|
}
|
||||||
|
|
||||||
info!(
|
info!(
|
||||||
"ZIP subtree: {} folders, {} files",
|
"ZIP subtree: {} folders, {} files",
|
||||||
all_folders.len(),
|
all_folders.len(),
|
||||||
all_files.len()
|
files_by_folder.values().map(|v| v.len()).sum::<usize>()
|
||||||
);
|
);
|
||||||
|
|
||||||
// ── 2. Group files by folder_id ──────────────────────────────────
|
|
||||||
let mut files_by_folder: HashMap<String, Vec<FileDto>> =
|
|
||||||
HashMap::with_capacity(all_folders.len());
|
|
||||||
for file in all_files {
|
|
||||||
let fid = file.folder_id.clone().unwrap_or_default();
|
|
||||||
files_by_folder.entry(fid).or_default().push(file);
|
|
||||||
}
|
|
||||||
|
|
||||||
// ── 3. Build a mapping: folder_id → ZIP-relative path ────────────
|
// ── 3. Build a mapping: folder_id → ZIP-relative path ────────────
|
||||||
//
|
//
|
||||||
// The root folder's DB path is e.g. "/users/alice/Documents".
|
// The root folder's DB path is e.g. "/users/alice/Documents".
|
||||||
|
|||||||
Reference in New Issue
Block a user