fix(oxfmt): Use tokio::stdin/out to handle WouldBlock error

apps/oxfmt/Cargo.toml

@@ -50,7 +50,7 @@ serde_json = { workspace = true }
 simdutf8 = { workspace = true }
 sort-package-json = { workspace = true }
 oxc-toml = { workspace = true }
-tokio = { workspace = true, features = ["rt-multi-thread", "macros"] }
+tokio = { workspace = true, features = ["rt-multi-thread", "macros", "io-util", "io-std"] }
 tracing = { workspace = true }
 tracing-subscriber = { workspace = true, features = [] } # Omit the `regex` feature
 tower-lsp-server = { workspace = true, features = ["proposed"] }

apps/oxfmt/src/cli/format.rs

@@ -1,4 +1,4 @@
-use std::{env, io::BufWriter, path::PathBuf, sync::mpsc, time::Instant};
+use std::{env, path::PathBuf, sync::mpsc, time::Instant};
 
 use oxc_diagnostics::DiagnosticService;
 
@@ -47,12 +47,7 @@ impl FormatRunner {
 
     /// # Panics
     /// Panics if `napi` feature is enabled but external_formatter is not set.
-    pub fn run(self) -> CliRunResult {
-        // stdio is blocked by `LineWriter`, use a `BufWriter` to reduce syscalls.
-        // See https://github.com/rust-lang/rust/issues/60673
-        let stdout = &mut BufWriter::new(std::io::stdout());
-        let stderr = &mut BufWriter::new(std::io::stderr());
-
+    pub async fn run(self) -> CliRunResult {
         let start_time = Instant::now();
 
         let cwd = self.cwd;
@@ -78,17 +73,14 @@ impl FormatRunner {
         ) {
             Ok(r) => r,
             Err(err) => {
-                utils::print_and_flush(
-                    stderr,
-                    &format!("Failed to load configuration file.\n{err}\n"),
-                );
+                utils::print_stderr(&format!("Failed to load configuration file.\n{err}\n")).await;
                 return CliRunResult::InvalidOptionConfig;
             }
         };
         let ignore_patterns = match config_resolver.build_and_validate() {
             Ok(patterns) => patterns,
             Err(err) => {
-                utils::print_and_flush(stderr, &format!("Failed to parse configuration.\n{err}\n"));
+                utils::print_stderr(&format!("Failed to parse configuration.\n{err}\n")).await;
                 return CliRunResult::InvalidOptionConfig;
             }
         };
@@ -107,10 +99,7 @@ impl FormatRunner {
             // - Pass `parser` to `SourceFormatter`
             Ok(_) => {}
             Err(err) => {
-                utils::print_and_flush(
-                    stderr,
-                    &format!("Failed to setup external formatter.\n{err}\n"),
-                );
+                utils::print_stderr(&format!("Failed to setup external formatter.\n{err}\n")).await;
                 return CliRunResult::InvalidOptionConfig;
             }
         }
@@ -127,100 +116,111 @@ impl FormatRunner {
             // All target paths are ignored
             Ok(None) => {
                 if runtime_options.no_error_on_unmatched_pattern {
-                    utils::print_and_flush(stderr, "No files found matching the given patterns.\n");
+                    utils::print_stderr("No files found matching the given patterns.\n").await;
                     return CliRunResult::None;
                 }
-                utils::print_and_flush(stderr, "Expected at least one target file\n");
+                utils::print_stderr("Expected at least one target file\n").await;
                 return CliRunResult::NoFilesFound;
             }
             Err(err) => {
-                utils::print_and_flush(
-                    stderr,
-                    &format!("Failed to parse target paths or ignore settings.\n{err}\n"),
-                );
+                utils::print_stderr(&format!(
+                    "Failed to parse target paths or ignore settings.\n{err}\n"
+                ))
+                .await;
                 return CliRunResult::InvalidOptionConfig;
             }
         };
 
-        // Get the receiver for streaming entries
-        let rx_entry = walker.stream_entries();
-        // Collect format results (changed paths or unchanged count)
-        let (tx_success, rx_success) = mpsc::channel();
-        // Diagnostic from formatting service
-        let (mut diagnostic_service, tx_error) =
-            DiagnosticService::new(Box::new(DefaultReporter::default()));
-
         if matches!(format_mode, OutputMode::Check) {
-            utils::print_and_flush(stdout, "Checking formatting...\n");
-            utils::print_and_flush(stdout, "\n");
+            utils::print_stdout("Checking formatting...\n\n").await;
         }
 
-        // Create `SourceFormatter` instance
-        let source_formatter = SourceFormatter::new(num_of_threads);
+        // Run formatting in a sync block to avoid holding non-Send DiagnosticService across await points
         #[cfg(feature = "napi")]
-        let source_formatter = source_formatter.with_external_formatter(self.external_formatter);
+        let external_formatter = self.external_formatter;
+
+        let (changed_paths, unchanged_count, diagnostic_output, error_count) = {
+            // Get the receiver for streaming entries
+            let rx_entry = walker.stream_entries();
+            // Collect format results (changed paths or unchanged count)
+            let (tx_success, rx_success) = mpsc::channel();
+            // Diagnostic from formatting service
+            let (mut diagnostic_service, tx_error) =
+                DiagnosticService::new(Box::new(DefaultReporter::default()));
+
+            // Create `SourceFormatter` instance
+            let source_formatter = SourceFormatter::new(num_of_threads);
+            #[cfg(feature = "napi")]
+            let source_formatter = source_formatter.with_external_formatter(external_formatter);
+
+            let format_mode_clone = format_mode.clone();
+
+            // Spawn a thread to run formatting service with streaming entries
+            rayon::spawn(move || {
+                let format_service =
+                    FormatService::new(cwd, format_mode_clone, source_formatter, config_resolver);
+                format_service.run_streaming(rx_entry, &tx_error, &tx_success);
+            });
+
+            // Collect results and separate changed paths from unchanged count
+            let mut changed_paths: Vec<String> = vec![];
+            let mut unchanged_count: usize = 0;
+            for result in rx_success {
+                match result {
+                    SuccessResult::Changed(path) => changed_paths.push(path),
+                    SuccessResult::Unchanged => unchanged_count += 1,
+                }
+            }
 
-        let format_mode_clone = format_mode.clone();
+            // Sort changed paths for deterministic output
+            if !changed_paths.is_empty() {
+                changed_paths.sort_unstable();
+            }
 
-        // Spawn a thread to run formatting service with streaming entries
-        rayon::spawn(move || {
-            let format_service =
-                FormatService::new(cwd, format_mode_clone, source_formatter, config_resolver);
-            format_service.run_streaming(rx_entry, &tx_error, &tx_success);
-        });
+            // Run diagnostics service and collect output (sync)
+            let mut diagnostic_output = Vec::new();
+            let diagnostics = diagnostic_service.run(&mut diagnostic_output);
+            // NOTE: We are not using `DiagnosticService` for warnings
+            let error_count = diagnostics.errors_count();
 
-        // Collect results and separate changed paths from unchanged count
-        let mut changed_paths: Vec<String> = vec![];
-        let mut unchanged_count: usize = 0;
-        for result in rx_success {
-            match result {
-                SuccessResult::Changed(path) => changed_paths.push(path),
-                SuccessResult::Unchanged => unchanged_count += 1,
-            }
-        }
+            (changed_paths, unchanged_count, diagnostic_output, error_count)
+        };
 
         // Print sorted changed file paths to stdout
         if !changed_paths.is_empty() {
-            changed_paths.sort_unstable();
-            utils::print_and_flush(stdout, &changed_paths.join("\n"));
+            utils::print_stdout(&changed_paths.join("\n")).await;
         }
 
         // Then, output diagnostics errors to stderr
-        // NOTE: This is blocking and print errors
-        let diagnostics = diagnostic_service.run(stderr);
-        // NOTE: We are not using `DiagnosticService` for warnings
-        let error_count = diagnostics.errors_count();
+        if !diagnostic_output.is_empty() {
+            utils::write_stderr(&diagnostic_output).await;
+        }
 
         // Count the processed files
         let total_target_files_count = changed_paths.len() + unchanged_count + error_count;
-        let print_stats = |stdout| {
+        let print_stats = || async {
             let elapsed_ms = start_time.elapsed().as_millis();
-            utils::print_and_flush(
-                stdout,
-                &format!(
-                    "Finished in {elapsed_ms}ms on {total_target_files_count} files using {num_of_threads} threads.\n",
-                ),
-            );
+            utils::print_stdout(&format!(
+                "Finished in {elapsed_ms}ms on {total_target_files_count} files using {num_of_threads} threads.\n",
+            )).await;
         };
 
         // Check if no files were found
         if total_target_files_count == 0 {
             if runtime_options.no_error_on_unmatched_pattern {
-                utils::print_and_flush(stderr, "No files found matching the given patterns.\n");
-                print_stats(stdout);
+                utils::print_stderr("No files found matching the given patterns.\n").await;
+                print_stats().await;
                 return CliRunResult::None;
             }
 
-            utils::print_and_flush(stderr, "Expected at least one target file\n");
+            utils::print_stderr("Expected at least one target file\n").await;
             return CliRunResult::NoFilesFound;
         }
 
         if 0 < error_count {
             // Each error is already printed in reporter
-            utils::print_and_flush(
-                stderr,
-                "Error occurred when checking code style in the above files.\n",
-            );
+            utils::print_stderr("Error occurred when checking code style in the above files.\n")
+                .await;
             return CliRunResult::FormatFailed;
         }
 
@@ -230,19 +230,13 @@ impl FormatRunner {
             (OutputMode::ListDifferent, _) => CliRunResult::FormatMismatch,
             // `--check` outputs friendly summary
             (OutputMode::Check, 0) => {
-                utils::print_and_flush(stdout, "All matched files use the correct format.\n");
-                print_stats(stdout);
+                utils::print_stdout("All matched files use the correct format.\n").await;
+                print_stats().await;
                 CliRunResult::FormatSucceeded
             }
             (OutputMode::Check, changed_count) => {
-                utils::print_and_flush(stdout, "\n\n");
-                utils::print_and_flush(
-                    stdout,
-                    &format!(
-                        "Format issues found in above {changed_count} files. Run without `--check` to fix.\n",
-                    ),
-                );
-                print_stats(stdout);
+                utils::print_stdout(&format!("\n\nFormat issues found in above {changed_count} files. Run without `--check` to fix.\n")).await;
+                print_stats().await;
                 CliRunResult::FormatMismatch
             }
             // Default (write) does not output anything

apps/oxfmt/src/core/utils.rs

@@ -1,4 +1,6 @@
-use std::{fs, io, io::Write, path::Path};
+use std::{fs, io, path::Path};
+
+use tokio::io::{AsyncWrite, AsyncWriteExt, BufWriter, stderr, stdout};
 
 /// To debug `oxc_formatter`:
 /// `OXC_LOG=oxc_formatter oxfmt`
@@ -38,17 +40,25 @@ pub fn read_to_string(path: &Path) -> io::Result<String> {
     Ok(unsafe { String::from_utf8_unchecked(bytes) })
 }
 
-pub fn print_and_flush(writer: &mut dyn Write, message: &str) {
-    use std::io::{Error, ErrorKind};
-    fn check_for_writer_error(error: Error) -> Result<(), Error> {
-        // Do not panic when the process is killed (e.g. piping into `less`).
-        if matches!(error.kind(), ErrorKind::Interrupted | ErrorKind::BrokenPipe) {
-            Ok(())
-        } else {
-            Err(error)
-        }
-    }
+/// Prints a string to stdout with buffering.
+pub async fn print_stdout(message: &str) {
+    write_buffered(stdout(), message.as_bytes()).await;
+}
+
+/// Prints a string to stderr with buffering.
+pub async fn print_stderr(message: &str) {
+    write_buffered(stderr(), message.as_bytes()).await;
+}
+
+/// Writes bytes to stderr with buffering.
+pub async fn write_stderr(message: &[u8]) {
+    write_buffered(stderr(), message).await;
+}
 
-    writer.write_all(message.as_bytes()).or_else(check_for_writer_error).unwrap();
-    writer.flush().unwrap();
+async fn write_buffered<W: AsyncWrite + Unpin>(writer: W, message: &[u8]) {
+    // stdio is blocked by `LineWriter`, use a `BufWriter` to reduce syscalls.
+    // See https://github.com/rust-lang/rust/issues/60673
+    let mut writer = BufWriter::new(writer);
+    let _ = writer.write_all(message).await;
+    let _ = writer.flush().await;
 }

apps/oxfmt/src/main.rs

@@ -14,5 +14,5 @@ async fn main() -> CliRunResult {
     init_tracing();
     init_miette();
     init_rayon(command.runtime_options.threads);
-    FormatRunner::new(command).run()
+    FormatRunner::new(command).run().await
 }

apps/oxfmt/src/main_napi.rs

@@ -95,17 +95,21 @@ pub async fn run_cli(
             init_miette();
 
             // TODO: `.with_external_formatter()` is not needed, just pass with `new(command, external_formatter)`
-            let result =
-                StdinRunner::new(command).with_external_formatter(Some(external_formatter)).run();
+            let result = StdinRunner::new(command)
+                .with_external_formatter(Some(external_formatter))
+                .run()
+                .await;
 
             ("stdin".to_string(), Some(result.exit_code()))
         }
         Mode::Cli(_) => {
             init_miette();
             init_rayon(command.runtime_options.threads);
 
-            let result =
-                FormatRunner::new(command).with_external_formatter(Some(external_formatter)).run();
+            let result = FormatRunner::new(command)
+                .with_external_formatter(Some(external_formatter))
+                .run()
+                .await;
 
             ("cli".to_string(), Some(result.exit_code()))
         }