diff --git a/src/ingest.rs b/src/ingest.rs index 385d004..bdd461f 100644 --- a/src/ingest.rs +++ b/src/ingest.rs @@ -1,8 +1,14 @@ +use std::path::{Path, PathBuf}; +use std::time::Duration; + use crate::engine::{EngineMessage, IngestedLine}; use crate::timestamp::Timestamp; +use tokio::fs::File; use tokio::io::{AsyncBufReadExt, BufReader}; use tokio::sync::mpsc; +const POLL_INTERVAL: Duration = Duration::from_millis(200); + /// Reads stdin line by line, sequences and timestamps each line, and forwards it /// to the engine task. pub async fn run(engine: mpsc::Sender) { @@ -23,3 +29,207 @@ pub async fn run(engine: mpsc::Sender) { } } } + +/// Identifies a specific file instance on disk, so a recreated path (e.g. a +/// tailed process restarting and reopening its log file) can be told apart from +/// the same file simply growing. +#[derive(PartialEq, Eq, Clone, Copy)] +struct FileId(#[cfg(unix)] u64, #[cfg(not(unix))] ()); + +fn file_id(meta: &std::fs::Metadata) -> FileId { + #[cfg(unix)] + { + use std::os::unix::fs::MetadataExt; + FileId(meta.ino()) + } + #[cfg(not(unix))] + { + let _ = meta; + FileId(()) + } +} + +async fn open(path: &Path) -> Option<(File, FileId)> { + let file = File::open(path).await.ok()?; + let meta = file.metadata().await.ok()?; + Some((file, file_id(&meta))) +} + +/// Tails `path` like `tail -f`: reads existing content, then keeps reading as +/// new lines are appended. If the file is deleted and recreated at the same +/// path (or truncated in place) — e.g. the tailed process restarted and +/// reopened its log — tailing resumes against the new file from its start, +/// rather than exiting or getting stuck. `sequence` keeps incrementing across +/// a recreation event; it's ottyl's own line counter, not tied to the source +/// file. +pub async fn run_file(engine: mpsc::Sender, path: PathBuf) { + let mut sequence: u64 = 0; + + 'reopen: loop { + let (file, mut id) = loop { + if let Some(opened) = open(&path).await { + break opened; + } + tokio::time::sleep(POLL_INTERVAL).await; + }; + + let mut reader = BufReader::new(file); + let mut offset: u64 = 0; + let mut pending = String::new(); + + loop { + let mut chunk = String::new(); + match reader.read_line(&mut chunk).await { + Ok(0) => { + // EOF: decide whether the file was truncated/recreated + // (tail the new instance from scratch) or just hasn't + // grown yet (keep polling). + match tokio::fs::metadata(&path).await { + Ok(meta) => { + let new_id = file_id(&meta); + if new_id != id || meta.len() < offset { + continue 'reopen; + } + id = new_id; + } + Err(_) => { + // Path gone (mid-recreation); wait for it to + // reappear and reopen fresh. + } + } + tokio::time::sleep(POLL_INTERVAL).await; + } + Ok(n) => { + offset += n as u64; + pending.push_str(&chunk); + if pending.ends_with('\n') { + let text = pending + .trim_end_matches(['\n', '\r']) + .to_string(); + pending.clear(); + + let line = IngestedLine { + sequence, + timestamp: Timestamp::now(), + text, + }; + sequence += 1; + + if engine.send(EngineMessage::Ingest(line)).await.is_err() { + return; + } + } + // else: partial line at EOF (write in progress) — keep + // buffering and read again without treating it as final. + } + Err(_) => continue 'reopen, + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::io::Write; + use tokio::sync::mpsc; + + async fn recv_text(rx: &mut mpsc::Receiver) -> String { + match tokio::time::timeout(Duration::from_secs(2), rx.recv()) + .await + .expect("timed out waiting for ingested line") + .expect("engine channel closed") + { + EngineMessage::Ingest(line) => line.text, + _ => panic!("expected Ingest"), + } + } + + #[tokio::test] + async fn tails_existing_and_appended_lines() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("app.log"); + std::fs::write(&path, "first\n").unwrap(); + + let (tx, mut rx) = mpsc::channel(16); + tokio::spawn(run_file(tx, path.clone())); + + assert_eq!(recv_text(&mut rx).await, "first"); + + let mut f = std::fs::OpenOptions::new().append(true).open(&path).unwrap(); + writeln!(f, "second").unwrap(); + assert_eq!(recv_text(&mut rx).await, "second"); + } + + #[tokio::test] + async fn resumes_after_delete_and_recreate() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("app.log"); + std::fs::write(&path, "before-restart\n").unwrap(); + + let (tx, mut rx) = mpsc::channel(16); + tokio::spawn(run_file(tx, path.clone())); + + assert_eq!(recv_text(&mut rx).await, "before-restart"); + + std::fs::remove_file(&path).unwrap(); + std::fs::write(&path, "after-restart\n").unwrap(); + + assert_eq!(recv_text(&mut rx).await, "after-restart"); + } + + #[tokio::test] + async fn resumes_after_truncation_in_place() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("app.log"); + std::fs::write(&path, "long line before truncation\n").unwrap(); + + let (tx, mut rx) = mpsc::channel(16); + tokio::spawn(run_file(tx, path.clone())); + + assert_eq!(recv_text(&mut rx).await, "long line before truncation"); + + // Truncate in place (same inode, shorter content) then write new data. + let mut f = std::fs::OpenOptions::new() + .write(true) + .truncate(true) + .open(&path) + .unwrap(); + writeln!(f, "new").unwrap(); + + assert_eq!(recv_text(&mut rx).await, "new"); + } + + #[tokio::test] + async fn sequence_keeps_incrementing_across_recreation() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("app.log"); + std::fs::write(&path, "one\n").unwrap(); + + let (tx, mut rx) = mpsc::channel(16); + tokio::spawn(run_file(tx, path.clone())); + + let first = match tokio::time::timeout(Duration::from_secs(2), rx.recv()) + .await + .unwrap() + .unwrap() + { + EngineMessage::Ingest(line) => line.sequence, + _ => panic!("expected Ingest"), + }; + assert_eq!(first, 0); + + std::fs::remove_file(&path).unwrap(); + std::fs::write(&path, "two\n").unwrap(); + + let second = match tokio::time::timeout(Duration::from_secs(2), rx.recv()) + .await + .unwrap() + .unwrap() + { + EngineMessage::Ingest(line) => line.sequence, + _ => panic!("expected Ingest"), + }; + assert_eq!(second, 1); + } +} diff --git a/src/main.rs b/src/main.rs index 8650b71..4399c27 100644 --- a/src/main.rs +++ b/src/main.rs @@ -38,6 +38,12 @@ struct Args { /// (e.g. `~/.config/ottyl/settings.json` on Linux). #[arg(long)] settings_path: Option, + + /// File to tail for log lines, instead of reading from stdin. Behaves like + /// `tail -f`: reads existing content then keeps following appended lines, + /// and keeps tailing if the file is deleted/recreated at the same path. + #[arg(long)] + file: Option, } #[tokio::main] @@ -86,7 +92,14 @@ async fn main() { } } - tokio::spawn(ingest::run(engine_tx.clone())); + match args.file.clone() { + Some(path) => { + tokio::spawn(ingest::run_file(engine_tx.clone(), path)); + } + None => { + tokio::spawn(ingest::run(engine_tx.clone())); + } + } let app = api::router(engine_tx, &args.static_dir, settings, settings_path); let listener = TcpListener::bind(("0.0.0.0", args.port)) @@ -97,9 +110,13 @@ async fn main() { .expect("failed to read back bound address") .port(); + let source = match &args.file { + Some(path) => format!("tailing {}", path.display()), + None => "reading logs from stdin".to_string(), + }; eprintln!( - "ottyl serving on http://localhost:{} (reading logs from stdin)", - bound_port + "ottyl serving on http://localhost:{} ({})", + bound_port, source ); axum::serve(listener, app)