feat: detect log rotation and reopen the rotated file

README.md

@@ -190,18 +190,24 @@ cargo clippy --all-targets --all-features -- -D warnings
 
 CI runs all three on every pull request and on pushes to `main`.
 
+Tests are organized by what they're proving, one file per concern under `tests/` (each file is its own binary, so this costs nothing to keep separate):
+
+- `filter_test.rs`, `parser_test.rs`, `source_test.rs`, `sink_test.rs`, `notify_test.rs` — one stage or behavior at a time.
+- `integration_test.rs`, `cli_test.rs` — the full pipeline wired together, the second one through the actual compiled binary.
+- `stress_test.rs` — load/failure-shaped: proving an outcome (nothing lost) under conditions designed to trigger backpressure, not just checking one function's return value.
+
 ## Roadmap
 
 ### v1 acceptance criteria
 
 v1 is done when all of the following hold:
 
 - [x] No batch is ever lost silently — every failed send is retried or persisted locally.
-- [ ] The process survives log rotation without manual intervention.
-- [ ] No channel grows unbounded under any tested load condition.
-- [ ] Every config field is validated with a clear error message at startup — never fails silently at runtime.
+- [x] The process survives log rotation without manual intervention (standard rename-then-create logrotate behavior; `copytruncate` isn't handled yet).
+- [x] No channel grows unbounded under load — guaranteed by construction (every channel has a fixed `channel_capacity`), and backed by [`tests/stress_test.rs`](tests/stress_test.rs), which proves the *outcome* (a burst of logs during a prolonged outage is never lost) rather than measuring memory directly.
+- [ ] Every config field is validated with a clear error message at startup — never fails silently at runtime. *(deferred to v1.1 — Config::load already rejects a missing/malformed file with a clean error, just not individual bad field values like `batch_size = 0`)*
 - [x] Sensitive data is masked by default, even with no user-configured rules.
-- [ ] External visibility (metrics) into what the pipeline is doing, without reading logtap's own stderr output.
+- [ ] External visibility (metrics) into what the pipeline is doing, without reading logtap's own stderr output. *(deferred to v1.1 — stderr logging already covers retries, dead-letter writes, and rotation events)*
 - [x] End-to-end integration test covering the full pipeline, plus unit tests per stage.
 - [x] Installing and running the project requires no source reading — README and example config are enough.
 

src/source.rs

@@ -3,26 +3,71 @@ use anyhow::Result;
 use notify::{Event, EventKind, RecommendedWatcher, RecursiveMode, Watcher};
 use std::fs::File;
 use std::io::{BufRead, BufReader, Seek, SeekFrom};
+use std::os::unix::fs::MetadataExt;
+use std::path::Path;
 use std::sync::mpsc as std_mpsc;
 use std::time::Duration;
 use tokio::sync::mpsc::Sender;
 
 // How often the watch loop wakes up even without a filesystem event, just to
-// check whether the downstream receiver is gone. Without this, the loop can
-// block forever on `notify_rx.recv()` after the rest of the pipeline shut
-// down, since there is no more filesystem activity left to notice it.
+// check whether the downstream receiver is gone, and whether the file at
+// source_path has been swapped out from under us (log rotation). Without
+// this, the loop can block forever on `notify_rx.recv()` once there's no
+// more filesystem activity on the file it's actually still holding open.
 const SHUTDOWN_POLL_INTERVAL: Duration = Duration::from_millis(200);
 
+/// Whether the caller should keep going or the downstream receiver is gone.
+enum DrainOutcome {
+    Continue,
+    ReceiverGone,
+}
+
+/// Reads every complete line available from `offset` onward and forwards it
+/// down the channel, advancing `offset` as it goes.
+fn drain_new_lines(
+    reader: &mut BufReader<File>,
+    offset: &mut u64,
+    tx: &Sender<String>,
+    line: &mut String,
+) -> Result<DrainOutcome> {
+    reader.seek(SeekFrom::Start(*offset))?;
+
+    loop {
+        line.clear();
+        let bytes_read = reader.read_line(line)?;
+
+        if bytes_read == 0 {
+            return Ok(DrainOutcome::Continue);
+        }
+
+        *offset += bytes_read as u64;
+        let trimmed = line.trim_end().to_string();
+
+        if tx.blocking_send(trimmed).is_err() {
+            return Ok(DrainOutcome::ReceiverGone);
+        }
+    }
+}
+
+fn watch(
+    path: &Path,
+    notify_tx: std_mpsc::Sender<notify::Result<Event>>,
+) -> Result<RecommendedWatcher> {
+    let mut watcher: RecommendedWatcher = notify::recommended_watcher(move |res| {
+        let _ = notify_tx.send(res);
+    })?;
+    watcher.watch(path, RecursiveMode::NonRecursive)?;
+    Ok(watcher)
+}
+
 pub fn run_source(cfg: Config, tx: Sender<String>) -> Result<()> {
     let file = File::open(&cfg.source_path)?;
+    let mut inode = file.metadata()?.ino();
     let mut reader = BufReader::new(file);
     let mut offset = reader.seek(SeekFrom::End(0))?;
 
     let (notify_tx, notify_rx) = std_mpsc::channel::<notify::Result<Event>>();
-    let mut watcher: RecommendedWatcher = notify::recommended_watcher(move |res| {
-        let _ = notify_tx.send(res);
-    })?;
-    watcher.watch(&cfg.source_path, RecursiveMode::NonRecursive)?;
+    let mut _watcher = watch(&cfg.source_path, notify_tx.clone())?;
 
     let mut line = String::new();
 
@@ -33,6 +78,33 @@ pub fn run_source(cfg: Config, tx: Sender<String>) -> Result<()> {
                 if tx.is_closed() {
                     return Ok(());
                 }
+
+                // The path may briefly not resolve mid-rotation (rename in
+                // progress) — that's transient, just retry on the next tick.
+                let Ok(current_inode) = std::fs::metadata(&cfg.source_path).map(|m| m.ino()) else {
+                    continue;
+                };
+
+                if current_inode == inode {
+                    continue;
+                }
+
+                // The file at source_path is no longer the one we have open
+                // (standard logrotate "rename + create new" behavior). Flush
+                // whatever was written to the old file right before the
+                // swap, then switch to the new one from the start.
+                if let DrainOutcome::ReceiverGone =
+                    drain_new_lines(&mut reader, &mut offset, &tx, &mut line)?
+                {
+                    return Ok(());
+                }
+
+                let new_file = File::open(&cfg.source_path)?;
+                inode = new_file.metadata()?.ino();
+                reader = BufReader::new(new_file);
+                offset = 0;
+                _watcher = watch(&cfg.source_path, notify_tx.clone())?;
+
                 continue;
             }
             Err(std_mpsc::RecvTimeoutError::Disconnected) => return Ok(()),
@@ -42,22 +114,10 @@ pub fn run_source(cfg: Config, tx: Sender<String>) -> Result<()> {
             continue;
         }
 
-        reader.seek(SeekFrom::Start(offset))?;
-
-        loop {
-            line.clear();
-            let bytes_read = reader.read_line(&mut line)?;
-
-            if bytes_read == 0 {
-                break;
-            }
-
-            offset += bytes_read as u64;
-            let trimmed = line.trim_end().to_string();
-
-            if tx.blocking_send(trimmed).is_err() {
-                return Ok(());
-            }
+        if let DrainOutcome::ReceiverGone =
+            drain_new_lines(&mut reader, &mut offset, &tx, &mut line)?
+        {
+            return Ok(());
         }
     }
 }

tests/source_test.rs

@@ -48,3 +48,65 @@ async fn test_run_source_emite_linhas_novas() {
 
     fs::remove_file(&path).ok();
 }
+
+#[tokio::test]
+async fn test_run_source_detects_log_rotation() {
+    let path = PathBuf::from("teste_run_source_rotacao.log");
+    let rotated_path = PathBuf::from("teste_run_source_rotacao.log.1");
+    fs::remove_file(&rotated_path).ok();
+    fs::write(&path, "").unwrap();
+
+    let (tx, mut rx) = mpsc::channel::<String>(100);
+
+    let cfg = Config {
+        source_path: path.clone(),
+        sink_url: "http://localhost:8080".to_string(),
+        batch_size: 100,
+        flush_interval_secs: 5,
+        channel_capacity: 100,
+        max_retries: 3,
+        retry_backoff_initial_ms: 100,
+        retry_backoff_max_secs: 5,
+        filter_rules: vec![],
+        mask_common_patterns: false,
+        dead_letter_max_bytes: 1024 * 1024,
+        dead_letter_max_files: 5,
+    };
+
+    let _handle = tokio::task::spawn_blocking(move || run_source(cfg, tx));
+
+    tokio::time::sleep(Duration::from_millis(100)).await;
+
+    {
+        let mut file = fs::OpenOptions::new().append(true).open(&path).unwrap();
+        writeln!(file, "antes da rotacao").unwrap();
+    }
+
+    let before = timeout(Duration::from_secs(2), rx.recv())
+        .await
+        .expect("timeout: linha de antes da rotação não chegou")
+        .expect("canal fechou antes de receber a linha");
+    assert_eq!(before, "antes da rotacao");
+
+    // Simulate standard logrotate behavior: rename the current file away,
+    // then create a brand new empty file at the same path.
+    fs::rename(&path, &rotated_path).unwrap();
+    fs::write(&path, "").unwrap();
+
+    // Give the 200ms rotation-detection tick room to notice the inode swap.
+    tokio::time::sleep(Duration::from_millis(400)).await;
+
+    {
+        let mut file = fs::OpenOptions::new().append(true).open(&path).unwrap();
+        writeln!(file, "depois da rotacao").unwrap();
+    }
+
+    let after = timeout(Duration::from_secs(2), rx.recv())
+        .await
+        .expect("timeout: linha de depois da rotação não chegou — rotação não detectada")
+        .expect("canal fechou antes de receber a linha");
+    assert_eq!(after, "depois da rotacao");
+
+    fs::remove_file(&path).ok();
+    fs::remove_file(&rotated_path).ok();
+}

tests/stress_test.rs

@@ -0,0 +1,103 @@
+// Stress / load-shaped tests — distinct from the correctness-per-function
+// tests elsewhere in this directory. These exercise the full pipeline under
+// conditions meant to trigger backpressure and sustained failure, not just
+// verify a single behavior in isolation.
+
+use std::fs;
+use std::io::Write;
+use std::path::PathBuf;
+use std::time::Duration;
+
+use logtap::config::Config;
+use tokio::net::TcpListener;
+
+const LOG_COUNT: usize = 20;
+
+#[tokio::test]
+async fn outage_does_not_lose_logs_when_destination_is_down() {
+    // Bind a port, then drop the listener immediately — nothing answers
+    // there, so every send fails fast with "connection refused", simulating
+    // the destination being down for the entire test.
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let address = listener.local_addr().unwrap();
+    drop(listener);
+
+    let source_path = PathBuf::from("stress_outage.log");
+    let dead_letter_path = PathBuf::from("logtap.failed.jsonl");
+    fs::remove_file(&dead_letter_path).ok();
+    fs::write(&source_path, "").unwrap();
+
+    let cfg = Config {
+        source_path: source_path.clone(),
+        sink_url: format!("http://{address}/logs"),
+        batch_size: 1,
+        flush_interval_secs: 60,
+        // Deliberately small — with 20 logs written at once, this forces the
+        // channels to fill up and back the source's reads up behind them,
+        // rather than everything sailing through unobstructed.
+        channel_capacity: 5,
+        max_retries: 2,
+        retry_backoff_initial_ms: 1,
+        retry_backoff_max_secs: 1,
+        filter_rules: vec![],
+        mask_common_patterns: false,
+        dead_letter_max_bytes: 1024 * 1024,
+        dead_letter_max_files: 5,
+    };
+
+    let app = tokio::spawn(async move {
+        logtap::run(cfg).await.unwrap();
+    });
+
+    tokio::time::sleep(Duration::from_millis(300)).await;
+
+    // Write all 20 lines in a single burst, well beyond channel_capacity, so
+    // the source can't just breeze through them one at a time.
+    {
+        let mut file = fs::OpenOptions::new()
+            .append(true)
+            .open(&source_path)
+            .unwrap();
+
+        for seq in 0..LOG_COUNT {
+            writeln!(file, r#"{{"seq":{seq}}}"#).unwrap();
+        }
+    }
+
+    // Poll the dead-letter file instead of guessing a fixed sleep — every
+    // batch here is doomed to fail, so this is where all 20 logs should
+    // eventually land.
+    let deadline = tokio::time::Instant::now() + Duration::from_secs(10);
+    let mut seen = Vec::new();
+
+    while tokio::time::Instant::now() < deadline {
+        if let Ok(contents) = fs::read_to_string(&dead_letter_path) {
+            seen = contents
+                .lines()
+                .map(|line| serde_json::from_str::<serde_json::Value>(line).unwrap())
+                .collect();
+
+            if seen.len() >= LOG_COUNT {
+                break;
+            }
+        }
+
+        tokio::time::sleep(Duration::from_millis(50)).await;
+    }
+
+    app.abort();
+    fs::remove_file(&source_path).ok();
+    fs::remove_file(&dead_letter_path).ok();
+
+    let mut seqs: Vec<u64> = seen
+        .iter()
+        .map(|v| v["seq"].as_u64().expect("dead-letter entry missing seq"))
+        .collect();
+    seqs.sort_unstable();
+
+    let expected: Vec<u64> = (0..LOG_COUNT as u64).collect();
+    assert_eq!(
+        seqs, expected,
+        "dead-letter file should contain exactly the {LOG_COUNT} logs written, no more, no fewer, no duplicates"
+    );
+}