feat: add dead-letter when reach max retries

.gitignore

@@ -1,4 +1,4 @@
 /target
 .idea
-claude/
+.claude/
 .vscode/

src/config.rs

@@ -31,6 +31,12 @@ pub struct Config {
 
     #[serde(default = "default_mask_common_patterns")]
     pub mask_common_patterns: bool,
+
+    #[serde(default = "default_dead_letter_max_bytes")]
+    pub dead_letter_max_bytes: usize,
+
+    #[serde(default = "default_dead_letter_max_files")]
+    pub dead_letter_max_files: u64,
 }
 
 fn default_batch_size() -> usize {
@@ -55,6 +61,14 @@ fn default_mask_common_patterns() -> bool {
     true
 }
 
+fn default_dead_letter_max_bytes() -> usize {
+    1024 * 1024 * 1024
+}
+
+fn default_dead_letter_max_files() -> u64 {
+    5
+}
+
 impl Config {
     pub fn load(path: &str) -> Result<Self> {
         let text = std::fs::read_to_string(path)

src/sink.rs

@@ -1,9 +1,15 @@
 use crate::config::Config;
 use crate::record::LogLine;
+use std::fs;
+use std::fs::OpenOptions;
+use std::io::Write;
+use std::path::PathBuf;
 use std::time::Duration;
 use tokio::sync::mpsc::Receiver;
 use tokio::time::{interval, sleep};
 
+const DEAD_LETTER_PATH: &str = "logtap.failed.jsonl";
+
 pub async fn run_sink(cfg: Config, mut rx: Receiver<LogLine>) {
     let client = reqwest::Client::new();
     let mut batch: Vec<LogLine> = Vec::with_capacity(cfg.batch_size);
@@ -41,15 +47,15 @@ async fn flush_with_retry(client: &reqwest::Client, cfg: &Config, batch: &mut Ve
             }
             Ok(resp) => {
                 eprintln!(
-                    "logtap: tentativa {}/{} falhou — destino respondeu com status {}",
+                    "logtap: attempt {}/{} failed — server responded with status {}",
                     attempt + 1,
                     cfg.max_retries,
                     resp.status()
                 );
             }
             Err(err) => {
                 eprintln!(
-                    "logtap: tentativa {}/{} falhou — erro ao enviar lote ({} itens): {err}",
+                    "logtap: attempt {}/{} failed — error sending batch ({} items): {err}",
                     attempt + 1,
                     cfg.max_retries,
                     batch.len()
@@ -60,13 +66,13 @@ async fn flush_with_retry(client: &reqwest::Client, cfg: &Config, batch: &mut Ve
         attempt += 1;
 
         if attempt >= cfg.max_retries {
-            // TODO(fase 1 do roadmap): dead-letter em vez de descartar aqui.
-            // Por enquanto o lote é perdido depois de esgotar as tentativas.
             eprintln!(
-                "logtap: desistindo do lote ({} itens) após {} tentativas",
+                "logtap: abandoning batch of {} items after {} attempts — writing to {DEAD_LETTER_PATH}",
                 batch.len(),
                 cfg.max_retries
             );
+
+            write_dead_letter(cfg, batch);
             batch.clear();
             return;
         }
@@ -77,7 +83,96 @@ async fn flush_with_retry(client: &reqwest::Client, cfg: &Config, batch: &mut Ve
         let backoff =
             Duration::from_millis(backoff_ms).min(Duration::from_secs(cfg.retry_backoff_max_secs));
 
-        eprintln!("logtap: esperando {backoff:?} antes da próxima tentativa");
+        eprintln!("logtap: waiting {backoff:?} before next attempt");
         sleep(backoff).await;
     }
 }
+
+// Reopens the file on every call instead of keeping a long-lived handle —
+// a handle stays bound to the underlying inode even if the file is later
+// renamed out from under it (e.g. for a manual replay), which would make
+// new failures silently land in the renamed-away file.
+fn write_dead_letter(cfg: &Config, batch: &[LogLine]) {
+    rotate_dead_letter_if_full(cfg);
+
+    let mut file = match OpenOptions::new()
+        .create(true)
+        .append(true)
+        .open(DEAD_LETTER_PATH)
+    {
+        Ok(file) => file,
+        Err(err) => {
+            eprintln!(
+                "logtap: could not open {DEAD_LETTER_PATH} ({err}) — {} item(s) lost for good",
+                batch.len()
+            );
+            return;
+        }
+    };
+
+    for log in batch {
+        if let Err(err) = writeln!(file, "{log}") {
+            eprintln!("logtap: failed writing to {DEAD_LETTER_PATH}: {err}");
+            return;
+        }
+    }
+}
+
+fn rotated_dead_letter_path(n: u64) -> PathBuf {
+    PathBuf::from(format!("{DEAD_LETTER_PATH}.{n}"))
+}
+
+// Caps how big logtap.failed.jsonl is allowed to get. Same idea as
+// logrotate: once the current file crosses dead_letter_max_bytes, it's
+// shifted into .1, the old .1 becomes .2, and so on up to
+// dead_letter_max_files — at which point the oldest file is discarded to
+// make room. That eviction is real, permanent data loss, so it's always
+// logged loudly rather than happening quietly.
+fn rotate_dead_letter_if_full(cfg: &Config) {
+    let current_size = match fs::metadata(DEAD_LETTER_PATH) {
+        Ok(meta) => meta.len(),
+        Err(_) => return, // no file yet — nothing to rotate
+    };
+
+    if current_size < cfg.dead_letter_max_bytes as u64 {
+        return;
+    }
+
+    if cfg.dead_letter_max_files == 0 {
+        if let Err(err) = fs::remove_file(DEAD_LETTER_PATH) {
+            eprintln!("logtap: failed to reset full dead-letter file: {err}");
+        } else {
+            eprintln!(
+                "logtap: dead-letter file hit {current_size} bytes and dead_letter_max_files is 0 — discarding it entirely"
+            );
+        }
+        return;
+    }
+
+    let oldest = rotated_dead_letter_path(cfg.dead_letter_max_files);
+    if oldest.exists() {
+        eprintln!(
+            "logtap: dead-letter rotation limit ({} files) reached — discarding oldest file {}",
+            cfg.dead_letter_max_files,
+            oldest.display()
+        );
+    }
+
+    for n in (1..cfg.dead_letter_max_files).rev() {
+        let from = rotated_dead_letter_path(n);
+        if from.exists() {
+            let to = rotated_dead_letter_path(n + 1);
+            if let Err(err) = fs::rename(&from, &to) {
+                eprintln!(
+                    "logtap: failed to rotate {} -> {}: {err}",
+                    from.display(),
+                    to.display()
+                );
+            }
+        }
+    }
+
+    if let Err(err) = fs::rename(DEAD_LETTER_PATH, rotated_dead_letter_path(1)) {
+        eprintln!("logtap: failed to rotate {DEAD_LETTER_PATH}: {err}");
+    }
+}

tests/integration_test.rs

@@ -80,6 +80,8 @@ async fn integration_source_parser_filter_sink_sends_log_to_http_server() {
         retry_backoff_max_secs: 0,
         filter_rules: vec![],
         mask_common_patterns: false,
+        dead_letter_max_bytes: 1024 * 1024,
+        dead_letter_max_files: 5,
     };
 
     let app = tokio::spawn(async move {

tests/sink_test.rs

@@ -1,11 +1,19 @@
 use std::path::PathBuf;
 use tokio::io::{AsyncReadExt, AsyncWriteExt};
 use tokio::net::TcpListener;
+use tokio::sync::Mutex;
 
 use logtap::config::Config;
 use logtap::record::LogLine;
 use logtap::sink::run_sink;
 
+// The dead-letter path is a fixed constant in sink.rs, not a Config field,
+// so every test that exercises it shares the same file on disk. cargo test
+// runs tests in the same binary concurrently by default — this lock keeps
+// the dead-letter tests from stepping on each other's file. Needs to be a
+// tokio Mutex (not std) since the guard is held across await points.
+static DEAD_LETTER_TEST_LOCK: Mutex<()> = Mutex::const_new(());
+
 async fn read_http_request_body(stream: &mut tokio::net::TcpStream) -> String {
     let mut buffer = Vec::new();
     let mut temp = [0_u8; 1024];
@@ -76,6 +84,8 @@ async fn sink_posts_logs_when_batch_size_is_reached() {
         retry_backoff_max_secs: 5,
         filter_rules: vec![],
         mask_common_patterns: false,
+        dead_letter_max_bytes: 0,
+        dead_letter_max_files: 0,
     };
 
     let (tx, rx) = tokio::sync::mpsc::channel::<LogLine>(cfg.channel_capacity);
@@ -107,3 +117,130 @@ async fn sink_posts_logs_when_batch_size_is_reached() {
         ])
     );
 }
+
+#[tokio::test]
+async fn sink_writes_batch_to_dead_letter_file_after_exhausting_retries() {
+    let _guard = DEAD_LETTER_TEST_LOCK.lock().await;
+
+    let dead_letter_path = PathBuf::from("logtap.failed.jsonl");
+    std::fs::remove_file(&dead_letter_path).ok();
+
+    // Bind a port, then drop the listener immediately — nothing will ever
+    // answer there, so every attempt fails fast with "connection refused".
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let address = listener.local_addr().unwrap();
+    drop(listener);
+
+    let cfg = Config {
+        source_path: PathBuf::from("unused.log"),
+        sink_url: format!("http://{address}/logs"),
+        batch_size: 1,
+        flush_interval_secs: 60,
+        channel_capacity: 10,
+        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 (tx, rx) = tokio::sync::mpsc::channel::<LogLine>(cfg.channel_capacity);
+
+    let sink = tokio::spawn(run_sink(cfg, rx));
+
+    tx.send(serde_json::json!({ "message": "never delivered" }))
+        .await
+        .unwrap();
+
+    // 2 attempts with ~1ms backoff exhaust almost immediately; give it some slack.
+    tokio::time::sleep(std::time::Duration::from_millis(300)).await;
+
+    sink.abort();
+
+    let contents = std::fs::read_to_string(&dead_letter_path)
+        .expect("expected dead-letter file to have been created");
+
+    std::fs::remove_file(&dead_letter_path).ok();
+
+    let lines: Vec<serde_json::Value> = contents
+        .lines()
+        .map(|line| serde_json::from_str(line).unwrap())
+        .collect();
+
+    assert_eq!(
+        lines,
+        vec![serde_json::json!({ "message": "never delivered" })]
+    );
+}
+
+#[tokio::test]
+async fn sink_rotates_dead_letter_file_once_it_exceeds_max_bytes() {
+    let _guard = DEAD_LETTER_TEST_LOCK.lock().await;
+
+    let current = PathBuf::from("logtap.failed.jsonl");
+    let rotated = PathBuf::from("logtap.failed.jsonl.1");
+    std::fs::remove_file(&current).ok();
+    std::fs::remove_file(&rotated).ok();
+
+    // Bind a port, then drop the listener immediately — nothing will ever
+    // answer there, so every attempt fails fast with "connection refused".
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let address = listener.local_addr().unwrap();
+    drop(listener);
+
+    let cfg = Config {
+        source_path: PathBuf::from("unused.log"),
+        sink_url: format!("http://{address}/logs"),
+        batch_size: 1,
+        flush_interval_secs: 60,
+        channel_capacity: 10,
+        max_retries: 1,
+        retry_backoff_initial_ms: 1,
+        retry_backoff_max_secs: 1,
+        filter_rules: vec![],
+        mask_common_patterns: false,
+        // Tiny on purpose: the very first failed batch already exceeds this,
+        // so the *second* failure is guaranteed to trigger a rotation.
+        dead_letter_max_bytes: 10,
+        dead_letter_max_files: 1,
+    };
+
+    let (tx, rx) = tokio::sync::mpsc::channel::<LogLine>(cfg.channel_capacity);
+
+    let sink = tokio::spawn(run_sink(cfg, rx));
+
+    tx.send(serde_json::json!({ "message": "first failure" }))
+        .await
+        .unwrap();
+    tx.send(serde_json::json!({ "message": "second failure" }))
+        .await
+        .unwrap();
+
+    // Each batch gives up after a single failed attempt (max_retries: 1),
+    // and the sink processes messages one at a time, so this is plenty.
+    tokio::time::sleep(std::time::Duration::from_millis(500)).await;
+
+    sink.abort();
+
+    let current_contents =
+        std::fs::read_to_string(&current).expect("expected a fresh logtap.failed.jsonl");
+    let rotated_contents =
+        std::fs::read_to_string(&rotated).expect("expected logtap.failed.jsonl.1 from rotation");
+
+    std::fs::remove_file(&current).ok();
+    std::fs::remove_file(&rotated).ok();
+
+    let current_log: serde_json::Value = serde_json::from_str(current_contents.trim()).unwrap();
+    let rotated_log: serde_json::Value = serde_json::from_str(rotated_contents.trim()).unwrap();
+
+    assert_eq!(
+        current_log,
+        serde_json::json!({ "message": "second failure" })
+    );
+    assert_eq!(
+        rotated_log,
+        serde_json::json!({ "message": "first failure" })
+    );
+}

tests/source_test.rs

@@ -26,6 +26,8 @@ async fn test_run_source_emite_linhas_novas() {
         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));