fix: add FIXME comments for EMFILE risk in unbounded parallel file reads (#2482)

Co-authored-by: laststylebender14 <ran.mahadik14@gmail.com>
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
Co-authored-by: laststylebender <43403528+laststylebender14@users.noreply.github.com>

Author: 3 people

crates/forge_app/src/changed_files.rs

@@ -26,7 +26,8 @@ impl<S: FsReadService + EnvironmentService> ChangedFiles<S> {
     /// duplicate notifications.
     pub async fn update_file_stats(&self, mut conversation: Conversation) -> Conversation {
         use crate::file_tracking::FileChangeDetector;
-        let changes = FileChangeDetector::new(self.services.clone())
+        let parallel_file_reads = self.services.get_environment().parallel_file_reads;
+        let changes = FileChangeDetector::new(self.services.clone(), parallel_file_reads)
             .detect(&conversation.metrics)
             .await;
 

crates/forge_app/src/file_tracking.rs

@@ -1,6 +1,7 @@
 use std::sync::Arc;
 
 use forge_domain::Metrics;
+use futures::StreamExt;
 use tracing::debug;
 
 use crate::FsReadService;
@@ -17,6 +18,7 @@ pub struct FileChange {
 #[derive(Clone)]
 pub struct FileChangeDetector<F> {
     fs_read_service: Arc<F>,
+    parallel_file_reads: usize,
 }
 
 impl<F: FsReadService> FileChangeDetector<F> {
@@ -25,8 +27,9 @@ impl<F: FsReadService> FileChangeDetector<F> {
     /// # Arguments
     ///
     /// * `fs_read_service` - The file system read service implementation
-    pub fn new(fs_read_service: Arc<F>) -> Self {
-        Self { fs_read_service }
+    /// * `parallel_file_reads` - Maximum number of files to hash concurrently
+    pub fn new(fs_read_service: Arc<F>, parallel_file_reads: usize) -> Self {
+        Self { fs_read_service, parallel_file_reads }
     }
 
     /// Detects files that have changed since the last notification
@@ -39,20 +42,34 @@ impl<F: FsReadService> FileChangeDetector<F> {
     /// * `tracked_files` - Map of file paths to their last known hashes (None
     ///   if unreadable)
     pub async fn detect(&self, metrics: &Metrics) -> Vec<FileChange> {
-        let futures: Vec<_> = metrics
+        let fs = self.fs_read_service.clone();
+        // Collect into owned data upfront so the stream futures are 'static-safe
+        let entries: Vec<(std::path::PathBuf, Option<String>)> = metrics
             .file_operations
             .iter()
             .map(|(path, file_metrics)| {
-                let file_path = std::path::PathBuf::from(path);
-                let last_hash = file_metrics.content_hash.clone();
+                (
+                    std::path::PathBuf::from(path),
+                    file_metrics.content_hash.clone(),
+                )
+            })
+            .collect();
+
+        let mut changes: Vec<FileChange> = futures::stream::iter(entries)
+            .map(|(file_path, last_hash)| {
+                let fs = fs.clone();
 
                 async move {
                     // Get current hash from the full raw file content (not the
                     // truncated/formatted content returned to the LLM).
                     // ReadOutput.content_hash is always computed from the
                     // unprocessed file, so it is directly comparable with the
                     // stored hash.
-                    let current_hash = self.read_file_hash(&file_path).await.ok();
+                    let current_hash = fs
+                        .read(file_path.to_string_lossy().to_string(), None, None)
+                        .await
+                        .ok()
+                        .map(|o| o.content_hash);
 
                     // Check if hash has changed
                     if current_hash != last_hash {
@@ -68,30 +85,16 @@ impl<F: FsReadService> FileChangeDetector<F> {
                     }
                 }
             })
-            .collect();
-
-        let mut changes: Vec<FileChange> = futures::future::join_all(futures)
-            .await
-            .into_iter()
-            .flatten()
-            .collect();
+            .buffer_unordered(self.parallel_file_reads)
+            .filter_map(std::future::ready)
+            .collect()
+            .await;
 
         // Sort by path for deterministic ordering
         changes.sort_by(|a, b| a.path.cmp(&b.path));
 
         changes
     }
-
-    /// Reads a file and returns its content hash computed from the full raw
-    /// content, bypassing any line truncation or range limiting.
-    async fn read_file_hash(&self, path: &std::path::Path) -> anyhow::Result<String> {
-        let output = self
-            .fs_read_service
-            .read(path.to_string_lossy().to_string(), None, None)
-            .await?;
-
-        Ok(output.content_hash)
-    }
 }
 
 #[cfg(test)]
@@ -198,7 +201,7 @@ mod tests {
         let content_hash = compute_hash(content);
 
         let fs = MockFsReadService::new().with_file("/test/file.txt", content);
-        let detector = FileChangeDetector::new(Arc::new(fs));
+        let detector = FileChangeDetector::new(Arc::new(fs), 64);
 
         let mut metrics = Metrics::default();
         metrics.file_operations.insert(
@@ -219,7 +222,7 @@ mod tests {
         let new_hash = compute_hash(new_content);
 
         let fs = MockFsReadService::new().with_file("/test/file.txt", new_content);
-        let detector = FileChangeDetector::new(Arc::new(fs));
+        let detector = FileChangeDetector::new(Arc::new(fs), 64);
 
         let mut metrics = Metrics::default();
         metrics.file_operations.insert(
@@ -241,7 +244,7 @@ mod tests {
         let old_hash = compute_hash("old content");
 
         let fs = MockFsReadService::new().with_not_found("/test/file.txt");
-        let detector = FileChangeDetector::new(Arc::new(fs));
+        let detector = FileChangeDetector::new(Arc::new(fs), 64);
 
         let mut metrics = Metrics::default();
         metrics.file_operations.insert(
@@ -265,7 +268,7 @@ mod tests {
         let old_hash = "old_hash".to_string();
 
         let fs = MockFsReadService::new().with_file("/test/file.txt", new_content);
-        let detector = FileChangeDetector::new(Arc::new(fs));
+        let detector = FileChangeDetector::new(Arc::new(fs), 64);
 
         // First call: detect change
         let mut metrics = Metrics::default();
@@ -296,7 +299,7 @@ mod tests {
         let content_hash = compute_hash(content);
 
         let fs = MockFsReadService::new().with_file("/test/file.txt", content);
-        let detector = FileChangeDetector::new(Arc::new(fs));
+        let detector = FileChangeDetector::new(Arc::new(fs), 64);
 
         let mut metrics = Metrics::default();
         metrics.file_operations.insert(
@@ -324,7 +327,7 @@ mod tests {
             &raw_content,
             &displayed_content,
         );
-        let detector = FileChangeDetector::new(Arc::new(fs));
+        let detector = FileChangeDetector::new(Arc::new(fs), 64);
 
         let mut metrics = Metrics::default();
         metrics.file_operations.insert(
@@ -353,7 +356,7 @@ mod tests {
             &raw_content,
             &displayed_content,
         );
-        let detector = FileChangeDetector::new(Arc::new(fs));
+        let detector = FileChangeDetector::new(Arc::new(fs), 64);
 
         let mut metrics = Metrics::default();
         metrics.file_operations.insert(
@@ -378,7 +381,7 @@ mod tests {
         let written_hash = compute_hash(written);
 
         let fs = MockFsReadService::new().with_file("/test/file.txt", written);
-        let detector = FileChangeDetector::new(Arc::new(fs));
+        let detector = FileChangeDetector::new(Arc::new(fs), 64);
 
         // Step 1: Read the file (insert via Metrics::insert like production)
         let metrics = Metrics::default().insert(
@@ -409,7 +412,7 @@ mod tests {
 
         // Disk now has the externally modified content
         let fs = MockFsReadService::new().with_file("/test/file.txt", external);
-        let detector = FileChangeDetector::new(Arc::new(fs));
+        let detector = FileChangeDetector::new(Arc::new(fs), 64);
 
         // Step 1: Read, Step 2: Write
         let metrics = Metrics::default()
@@ -441,7 +444,7 @@ mod tests {
         let content_hash = compute_hash(content);
 
         let fs = MockFsReadService::new().with_file("/test/file.txt", content);
-        let detector = FileChangeDetector::new(Arc::new(fs));
+        let detector = FileChangeDetector::new(Arc::new(fs), 64);
 
         // Step 1: Write, Step 2: Read back (overwrites Write entry)
         let metrics = Metrics::default()
@@ -481,7 +484,7 @@ mod tests {
             .with_file("/test/b.txt", b_external) // user modified B
             .with_file("/test/c.txt", c_content)
             .with_file("/test/d.txt", d_content);
-        let detector = FileChangeDetector::new(Arc::new(fs));
+        let detector = FileChangeDetector::new(Arc::new(fs), 64);
 
         let metrics = Metrics::default()
             .insert(
@@ -527,7 +530,7 @@ mod tests {
         let modified = "someone changed this";
 
         let fs = MockFsReadService::new().with_file("/test/file.txt", modified);
-        let detector = FileChangeDetector::new(Arc::new(fs));
+        let detector = FileChangeDetector::new(Arc::new(fs), 64);
 
         let metrics = Metrics::default().insert(
             "/test/file.txt".to_string(),
@@ -551,7 +554,7 @@ mod tests {
         let final_hash = compute_hash(final_content);
 
         let fs = MockFsReadService::new().with_file("/test/file.txt", final_content);
-        let detector = FileChangeDetector::new(Arc::new(fs));
+        let detector = FileChangeDetector::new(Arc::new(fs), 64);
 
         let metrics = Metrics::default()
             .insert(
@@ -585,7 +588,7 @@ mod tests {
         let original_hash = compute_hash(original);
 
         let fs = MockFsReadService::new().with_file("/test/file.txt", original);
-        let detector = FileChangeDetector::new(Arc::new(fs));
+        let detector = FileChangeDetector::new(Arc::new(fs), 64);
 
         let metrics = Metrics::default()
             .insert(
@@ -617,7 +620,7 @@ mod tests {
 
         // After write, disk has the written content
         let fs = MockFsReadService::new().with_file("/test/file.txt", written_content);
-        let detector = FileChangeDetector::new(Arc::new(fs));
+        let detector = FileChangeDetector::new(Arc::new(fs), 64);
 
         // Read (truncated display), then Write
         let metrics = Metrics::default()

crates/forge_app/src/orch_spec/orch_setup.rs

@@ -94,6 +94,7 @@ impl Default for TestContext {
                 max_image_size: 262144,
                 workspace_server_url: Url::parse("http://localhost:8080").unwrap(),
                 max_extensions: 15,
+                parallel_file_reads: 64,
             },
             title: Some("test-conversation".into()),
             agent: Agent::new(

crates/forge_domain/src/env.rs

@@ -96,6 +96,10 @@ pub struct Environment {
     /// Set to "json" (or "true"/"1"/"yes") for JSON, "html" for HTML, or
     /// unset/other to disable.
     pub auto_dump: Option<AutoDumpFormat>,
+    /// Maximum number of files read concurrently in parallel operations.
+    /// Controlled by FORGE_PARALLEL_FILE_READS environment variable.
+    /// Caps the `buffer_unordered` concurrency to avoid EMFILE errors.
+    pub parallel_file_reads: usize,
 }
 
 /// The output format used when auto-dumping a conversation on task completion.
@@ -334,6 +338,7 @@ fn test_command_path() {
         workspace_server_url: "http://localhost:8080".parse().unwrap(),
         max_extensions: 15,
         auto_dump: None,
+        parallel_file_reads: 64,
     };
 
     let actual = fixture.command_path();
@@ -375,6 +380,7 @@ fn test_command_cwd_path() {
         workspace_server_url: "http://localhost:8080".parse().unwrap(),
         max_extensions: 15,
         auto_dump: None,
+        parallel_file_reads: 64,
     };
 
     let actual = fixture.command_cwd_path();
@@ -416,6 +422,7 @@ fn test_command_cwd_path_independent_from_command_path() {
         workspace_server_url: "http://localhost:8080".parse().unwrap(),
         max_extensions: 15,
         auto_dump: None,
+        parallel_file_reads: 64,
     };
 
     let command_path = fixture.command_path();

crates/forge_infra/src/env.rs

@@ -97,6 +97,13 @@ impl ForgeEnvironmentInfra {
                 .unwrap_or_else(|| Url::parse("https://api.forgecode.dev/").unwrap()),
             max_extensions: parse_env::<usize>("FORGE_MAX_EXTENSIONS").unwrap_or(15),
             auto_dump: parse_env::<AutoDumpFormat>("FORGE_AUTO_DUMP"),
+            parallel_file_reads: parse_env::<usize>("FORGE_PARALLEL_FILE_READS").unwrap_or_else(
+                || {
+                    std::thread::available_parallelism()
+                        .map(|n| n.get() * 2)
+                        .unwrap_or(32)
+                },
+            ),
         }
     }
 

crates/forge_infra/src/forge_infra.rs

@@ -64,7 +64,8 @@ impl ForgeInfra {
         let http_service = Arc::new(ForgeHttpInfra::new(env.clone(), file_write_service.clone()));
         let file_read_service = Arc::new(ForgeFileReadService::new());
         let file_meta_service = Arc::new(ForgeFileMetaService);
-        let directory_reader_service = Arc::new(ForgeDirectoryReaderService);
+        let directory_reader_service =
+            Arc::new(ForgeDirectoryReaderService::new(env.parallel_file_reads));
         let grpc_client = Arc::new(ForgeGrpcClient::new(env.workspace_server_url.clone()));
         let output_printer = Arc::new(StdConsoleWriter::default());