From 9f8a6f51771cedb49c137e5098616e604c46ed37 Mon Sep 17 00:00:00 2001 From: Dionisio Date: Thu, 26 Feb 2026 00:07:10 +0100 Subject: [PATCH] =?UTF-8?q?perf:=20stream=5Ffiles=5Fin=5Fsubtree=20?= =?UTF-8?q?=E2=80=94=20replace=20Vec=20with=20async=20Stream?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace list_files_in_subtree (fetch_all → Vec) with stream_files_in_subtree that returns a Pin>>> 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 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). --- src/application/ports/file_ports.rs | 13 +-- src/application/ports/storage_ports.rs | 16 ++-- src/application/services/batch_operations.rs | 15 ++-- .../services/file_retrieval_service.rs | 11 ++- src/application/services/share_service.rs | 10 +++ .../services/trash_service_test.rs | 8 ++ src/common/stubs.rs | 14 ++++ .../pg/file_blob_read_repository.rs | 82 ++++++++++--------- src/infrastructure/services/zip_service.rs | 27 +++--- 9 files changed, 127 insertions(+), 69 deletions(-) diff --git a/src/application/ports/file_ports.rs b/src/application/ports/file_ports.rs index a6951a76..cfb6fc57 100644 --- a/src/application/ports/file_ports.rs +++ b/src/application/ports/file_ports.rs @@ -154,12 +154,15 @@ pub trait FileRetrievalUseCase: Send + Sync + 'static { end: Option, ) -> Result> + 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). - async fn list_files_in_subtree(&self, folder_id: &str) -> Result, DomainError> { - self.list_files(Some(folder_id)).await - } + /// Returns a streaming cursor — RAM stays O(1) per row. Callers + /// consume incrementally (e.g. group into a HashMap by folder_id) + /// without materializing the full result set. + async fn stream_files_in_subtree( + &self, + folder_id: &str, + ) -> Result> + Send>>, DomainError>; /// Lists files in a folder with LIMIT/OFFSET pagination. /// diff --git a/src/application/ports/storage_ports.rs b/src/application/ports/storage_ports.rs index b1115533..da9938c9 100644 --- a/src/application/ports/storage_ports.rs +++ b/src/application/ports/storage_ports.rs @@ -3,6 +3,7 @@ use bytes::Bytes; use futures::Stream; use serde_json::Value; use std::path::PathBuf; +use std::pin::Pin; use crate::application::dtos::search_dto::SearchCriteriaDto; 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()) } - /// 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 - /// 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). - async fn list_files_in_subtree(&self, folder_id: &str) -> Result, DomainError> { - self.list_files(Some(folder_id)).await - } + /// Callers consume the stream incrementally (e.g. build a HashMap + /// keyed by folder_id) without ever materializing the full Vec. + async fn stream_files_in_subtree( + &self, + folder_id: &str, + ) -> Result> + Send>>, DomainError>; /// Search files with pagination and filtering at database level. /// diff --git a/src/application/services/batch_operations.rs b/src/application/services/batch_operations.rs index da1d1eff..ad4555a0 100644 --- a/src/application/services/batch_operations.rs +++ b/src/application/services/batch_operations.rs @@ -792,29 +792,34 @@ impl BatchOperationService { /// Adds an entire folder subtree to the ZIP using 2 bulk SQL queries /// (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( &self, zip: &mut ZipFileWriter>>, folder_id: &str, root_folder: &FolderDto, ) -> Result<(), BatchOperationError> { - // Bulk-fetch entire subtree (2 queries total) + // Bulk-fetch folder tree (small — one entry per folder) let all_folders = self .folder_service .list_subtree_folders(folder_id) .await .map_err(BatchOperationError::Domain)?; - let all_files = self + // Stream files from DB cursor — O(1) per row + let mut file_stream = self .file_retrieval - .list_files_in_subtree(folder_id) + .stream_files_in_subtree(folder_id) .await .map_err(BatchOperationError::Domain)?; - // Group files by folder_id + // Group files by folder_id incrementally from the stream let mut files_by_folder: HashMap> = 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(); files_by_folder.entry(fid).or_default().push(file); } diff --git a/src/application/services/file_retrieval_service.rs b/src/application/services/file_retrieval_service.rs index a20025eb..296ea047 100644 --- a/src/application/services/file_retrieval_service.rs +++ b/src/application/services/file_retrieval_service.rs @@ -1,6 +1,7 @@ use async_trait::async_trait; use bytes::{Bytes, BytesMut}; use futures::{Stream, StreamExt}; +use std::pin::Pin; use std::sync::Arc; 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 } - async fn list_files_in_subtree(&self, folder_id: &str) -> Result, DomainError> { - let files = self.file_read.list_files_in_subtree(folder_id).await?; - Ok(files.into_iter().map(FileDto::from).collect()) + async fn stream_files_in_subtree( + &self, + folder_id: &str, + ) -> Result> + 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( diff --git a/src/application/services/share_service.rs b/src/application/services/share_service.rs index 7ecb27a4..555c824c 100644 --- a/src/application/services/share_service.rs +++ b/src/application/services/share_service.rs @@ -505,6 +505,16 @@ mod tests { ) -> Result { Ok(0) } + + async fn stream_files_in_subtree( + &self, + _folder_id: &str, + ) -> Result< + std::pin::Pin> + Send>>, + DomainError, + > { + Ok(Box::pin(futures::stream::empty())) + } } #[async_trait] diff --git a/src/application/services/trash_service_test.rs b/src/application/services/trash_service_test.rs index cc5c8ea4..216445ea 100644 --- a/src/application/services/trash_service_test.rs +++ b/src/application/services/trash_service_test.rs @@ -4,6 +4,7 @@ use chrono::Utc; use futures::Stream; use std::collections::HashMap; use std::path::PathBuf; +use std::pin::Pin; use std::sync::{Arc, Mutex}; use uuid::Uuid; @@ -205,6 +206,13 @@ impl FileReadPort for MockFileRepository { ) -> std::result::Result { Ok(0) } + + async fn stream_files_in_subtree( + &self, + _folder_id: &str, + ) -> std::result::Result> + Send>>, DomainError> { + Ok(Box::pin(futures::stream::empty())) + } } #[async_trait] diff --git a/src/common/stubs.rs b/src/common/stubs.rs index c00fe2b3..bfa1ee05 100644 --- a/src/common/stubs.rs +++ b/src/common/stubs.rs @@ -121,6 +121,13 @@ impl FileReadPort for StubFileReadPort { ) -> Result { Ok(0) } + + async fn stream_files_in_subtree( + &self, + _folder_id: &str, + ) -> Result> + 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 { Err(DomainError::not_found("File", "stub")) } + + async fn stream_files_in_subtree( + &self, + _folder_id: &str, + ) -> Result> + Send>>, DomainError> { + Ok(Box::pin(futures::stream::empty())) + } } // --------------------------------------------------------------------------- diff --git a/src/infrastructure/repositories/pg/file_blob_read_repository.rs b/src/infrastructure/repositories/pg/file_blob_read_repository.rs index de61caeb..dbe0c23d 100644 --- a/src/infrastructure/repositories/pg/file_blob_read_repository.rs +++ b/src/infrastructure/repositories/pg/file_blob_read_repository.rs @@ -9,9 +9,10 @@ use async_trait::async_trait; use bytes::Bytes; -use futures::Stream; +use futures::{Stream, TryStreamExt}; use moka::sync::Cache; use sqlx::PgPool; +use std::pin::Pin; use std::sync::Arc; 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 `<@`. - /// Ordered by `(fo.path, fi.name)` so callers iterate in directory order. - async fn list_files_in_subtree(&self, folder_id: &str) -> Result, DomainError> { - let rows: Vec<( - String, - String, - Option, - Option, - i64, - String, - i64, - i64, - Option, - )> = 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}")))?; + /// Single GiST-indexed query via ltree `<@`. Results are delivered + /// through a PostgreSQL cursor — RAM stays O(1) per row. + async fn stream_files_in_subtree( + &self, + folder_id: &str, + ) -> Result> + Send>>, DomainError> { + let pool = Arc::clone(&self.pool); + let folder_id = folder_id.to_owned(); - rows.into_iter() - .map(|(id, name, fid, fpath, size, mime, ca, ma, uid)| { - Self::row_to_file(id, name, fid, fpath, size, mime, ca, ma, uid) - }) - .collect() + let stream = async_stream::try_stream! { + let mut row_stream = sqlx::query_as::<_, ( + String, String, Option, Option, + i64, String, i64, i64, Option, + )>( + 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. diff --git a/src/infrastructure/services/zip_service.rs b/src/infrastructure/services/zip_service.rs index e74a1e57..e4c15586 100644 --- a/src/infrastructure/services/zip_service.rs +++ b/src/infrastructure/services/zip_service.rs @@ -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 .folder_service .list_subtree_folders(folder_id) .await .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 - .list_files_in_subtree(folder_id) + .stream_files_in_subtree(folder_id) .await .map_err(|e| ZipError::FolderContentsError(format!("subtree files: {}", e)))?; + // Group files by folder_id incrementally from the stream + let mut files_by_folder: HashMap> = + 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!( "ZIP subtree: {} folders, {} files", all_folders.len(), - all_files.len() + files_by_folder.values().map(|v| v.len()).sum::() ); - // ── 2. Group files by folder_id ────────────────────────────────── - let mut files_by_folder: HashMap> = - 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 ──────────── // // The root folder's DB path is e.g. "/users/alice/Documents".