Add --read-logs-from for asynchronous log output

Author: 9999years

Cargo.lock

@@ -44,6 +44,7 @@ clearscreen = "2.0.1"
 command-group = { version = "2.1.0", features = ["tokio", "with-tokio"] }
 crossterm = { version = "0.27.0", features = ["event-stream"] }
 enum-iterator = "1.4.1"
+fs-err = { version = "2.11.0", features = ["tokio"] }
 humantime = "2.1.0"
 ignore = "0.4.20"
 indoc = "1.0.6"

Cargo.toml

@@ -60,6 +60,21 @@ pub struct Opts {
     #[arg(long)]
     pub no_interrupt_reloads: bool,
 
+    /// Extra paths to read logs from.
+    ///
+    /// `ghciwatch` needs to parse `ghci`'s output to determine when reloads have finished and to
+    /// parse compiler errors, so libraries like Yesod that asynchronously print to stdout or
+    /// stderr are not supported.
+    ///
+    /// Instead, you should have your program write its logs to a file and use its path as an
+    /// argument to this option. `ghciwatch` will read from the file and output logs inline with
+    /// the rest of its output.
+    ///
+    /// See: https://github.com/ndmitchell/ghcid/issues/137
+    #[allow(rustdoc::bare_urls)]
+    #[arg(long)]
+    pub read_logs_from: Vec<Utf8PathBuf>,
+
     /// Enable TUI mode (experimental).
     #[arg(long, hide = true)]
     pub tui: bool,

src/cli.rs

@@ -1,6 +1,6 @@
 use camino::Utf8PathBuf;
+use fs_err::tokio::File;
 use miette::IntoDiagnostic;
-use tokio::fs::File;
 use tokio::io::AsyncWriteExt;
 use tokio::io::BufWriter;
 use tracing::instrument;

src/ghci/error_log.rs

@@ -20,6 +20,7 @@ use tokio::task::JoinHandle;
 use aho_corasick::AhoCorasick;
 use camino::Utf8Path;
 use camino::Utf8PathBuf;
+use fs_err::tokio as fs;
 use miette::miette;
 use miette::IntoDiagnostic;
 use miette::WrapErr;
@@ -641,7 +642,7 @@ impl Ghci {
     /// Read and parse eval commands from the given `path`.
     #[instrument(level = "trace")]
     async fn parse_eval_commands(path: &Utf8Path) -> miette::Result<Vec<EvalCommand>> {
-        let contents = tokio::fs::read_to_string(path)
+        let contents = fs::read_to_string(path)
             .await
             .into_diagnostic()
             .wrap_err_with(|| format!("Failed to read {path}"))?;

src/ghci/mod.rs

@@ -28,6 +28,7 @@ mod ignore;
 mod incremental_reader;
 mod maybe_async_command;
 mod normal_path;
+mod read_logs_from;
 mod shutdown;
 mod string_case;
 mod tracing;
@@ -43,6 +44,7 @@ pub use ghci::manager::run_ghci;
 pub use ghci::Ghci;
 pub use ghci::GhciOpts;
 pub use ghci::GhciWriter;
+pub use read_logs_from::ReadLogsFrom;
 pub use shutdown::ShutdownError;
 pub use shutdown::ShutdownHandle;
 pub use shutdown::ShutdownManager;

src/lib.rs

@@ -12,6 +12,7 @@ use ghciwatch::run_ghci;
 use ghciwatch::run_tui;
 use ghciwatch::run_watcher;
 use ghciwatch::GhciOpts;
+use ghciwatch::ReadLogsFrom;
 use ghciwatch::ShutdownManager;
 use ghciwatch::TracingOpts;
 use ghciwatch::WatcherOpts;
@@ -45,6 +46,18 @@ async fn main() -> miette::Result<()> {
             .await;
     }
 
+    for path in opts.read_logs_from {
+        manager
+            .spawn("read-logs", |handle| {
+                ReadLogsFrom {
+                    shutdown: handle,
+                    path,
+                }
+                .run()
+            })
+            .await;
+    }
+
     manager
         .spawn("run_ghci", |handle| {
             run_ghci(handle, ghci_opts, ghci_receiver)

src/main.rs

@@ -0,0 +1,148 @@
+use std::io::SeekFrom;
+#[cfg(unix)]
+use std::os::unix::fs::MetadataExt;
+use std::time::Duration;
+
+use backoff::backoff::Backoff;
+use backoff::ExponentialBackoff;
+use camino::Utf8Path;
+use camino::Utf8PathBuf;
+use fs_err::tokio as fs;
+use fs_err::tokio::File;
+use miette::miette;
+use miette::IntoDiagnostic;
+use tap::TryConv;
+use tokio::io::AsyncBufReadExt;
+use tokio::io::AsyncSeekExt;
+use tokio::io::AsyncWriteExt;
+use tokio::io::BufReader;
+use tracing::instrument;
+
+use crate::ShutdownHandle;
+
+/// Maximum number of bytes to print near the end of a log file, if it already has data when it's
+/// opened.
+const MAX_BYTES_PRINT_FROM_END: u64 = 0x200; // = 512
+
+/// Me: Can we have `tail(1)`?
+///
+/// `ghciwatch`: We have `tail(1)` at home.
+///
+/// `tail(1)` at home:
+pub struct ReadLogsFrom {
+    /// Shutdown handle.
+    pub shutdown: ShutdownHandle,
+    /// Path to read logs from.
+    pub path: Utf8PathBuf,
+}
+
+impl ReadLogsFrom {
+    /// Read logs from the given path and output them to stdout.
+    #[instrument(skip_all, name = "read-logs", level = "debug", fields(path = %self.path))]
+    pub async fn run(mut self) -> miette::Result<()> {
+        let mut backoff = ExponentialBackoff {
+            max_elapsed_time: None,
+            max_interval: Duration::from_secs(1),
+            ..Default::default()
+        };
+        while let Some(duration) = backoff.next_backoff() {
+            match self.run_inner().await {
+                Ok(()) => {
+                    // Graceful exit.
+                    break;
+                }
+                Err(err) => {
+                    // These errors are often like "the file doesn't exist yet" so we don't want
+                    // them to be noisy.
+                    tracing::debug!("{err:?}");
+                }
+            }
+
+            tracing::debug!("Waiting {duration:?} before retrying");
+            tokio::time::sleep(duration).await;
+        }
+
+        Ok(())
+    }
+
+    async fn run_inner(&mut self) -> miette::Result<()> {
+        loop {
+            tokio::select! {
+                result = Self::read(&self.path) => {
+                    result?;
+                }
+                _ = self.shutdown.on_shutdown_requested() => {
+                    // Graceful exit.
+                    break;
+                }
+                else => {
+                    // Graceful exit.
+                    break;
+                }
+            }
+        }
+        Ok(())
+    }
+
+    async fn read(path: &Utf8Path) -> miette::Result<()> {
+        let file = File::open(&path).await.into_diagnostic()?;
+        let mut metadata = file.metadata().await.into_diagnostic()?;
+        let mut size = metadata.len();
+        let mut reader = BufReader::new(file);
+
+        if size > MAX_BYTES_PRINT_FROM_END {
+            tracing::debug!("Log file too big, skipping to end");
+            reader
+                .seek(SeekFrom::End(
+                    -MAX_BYTES_PRINT_FROM_END
+                        .try_conv::<i64>()
+                        .expect("Constant is not bigger than i64::MAX"),
+                ))
+                .await
+                .into_diagnostic()?;
+        }
+
+        let mut lines = reader.lines();
+
+        let mut backoff = ExponentialBackoff {
+            max_elapsed_time: None,
+            max_interval: Duration::from_millis(1000),
+            ..Default::default()
+        };
+
+        let mut stdout = tokio::io::stdout();
+
+        while let Some(duration) = backoff.next_backoff() {
+            while let Some(line) = lines.next_line().await.into_diagnostic()? {
+                // TODO: Lock stdout here and for ghci output.
+                let _ = stdout.write_all(line.as_bytes()).await;
+                let _ = stdout.write_all(b"\n").await;
+            }
+
+            // Note: This will fail if the file has been removed. The inode/device number check is
+            // a secondary heuristic.
+            let new_metadata = fs::metadata(&path).await.into_diagnostic()?;
+            #[cfg(unix)]
+            if new_metadata.dev() != metadata.dev() || new_metadata.ino() != metadata.ino() {
+                return Err(miette!("Log file was replaced or removed: {path}"));
+            }
+
+            let new_size = new_metadata.len();
+            if new_size < size {
+                tracing::info!(%path, "Log file truncated");
+                let mut reader = lines.into_inner();
+                reader.seek(SeekFrom::Start(0)).await.into_diagnostic()?;
+                lines = reader.lines();
+            }
+            size = new_size;
+            metadata = new_metadata;
+
+            tracing::trace!("Caught up to log file");
+
+            tracing::trace!("Waiting {duration:?} before retrying");
+            tokio::time::sleep(duration).await;
+        }
+
+        Ok(())
+    }
+}