diff --git a/.gitignore b/.gitignore index 885c704..d0cb370 100644 --- a/.gitignore +++ b/.gitignore @@ -1,4 +1,4 @@ /target .idea -claude/ +.claude/ .vscode/ \ No newline at end of file diff --git a/src/config.rs b/src/config.rs index 0d67b6d..aa3f708 100644 --- a/src/config.rs +++ b/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 { let text = std::fs::read_to_string(path) diff --git a/src/sink.rs b/src/sink.rs index fb3178d..8362425 100644 --- a/src/sink.rs +++ b/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) { let client = reqwest::Client::new(); let mut batch: Vec = Vec::with_capacity(cfg.batch_size); @@ -41,7 +47,7 @@ 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() @@ -49,7 +55,7 @@ async fn flush_with_retry(client: &reqwest::Client, cfg: &Config, batch: &mut Ve } 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}"); + } +} diff --git a/tests/integration_test.rs b/tests/integration_test.rs index cf21d46..4031b4e 100644 --- a/tests/integration_test.rs +++ b/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 { diff --git a/tests/sink_test.rs b/tests/sink_test.rs index 25e44a2..8a08199 100644 --- a/tests/sink_test.rs +++ b/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::(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::(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 = 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(¤t).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::(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(¤t).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(¤t).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" }) + ); +} diff --git a/tests/source_test.rs b/tests/source_test.rs index f4e7d60..c11e0e0 100644 --- a/tests/source_test.rs +++ b/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));