fix: ring buffer O(n)→VecDeque, TUI layout guard, mutex poison recovery, cache filter regex, log task panics, SSE serialize error

Author: akosidencio

crates/tailflow-core/src/ingestion/process.rs

@@ -81,8 +81,12 @@ impl ProcessSource {
         });
 
         let status = child.wait().await?;
-        stdout_task.await.ok();
-        stderr_task.await.ok();
+        if let Err(e) = stdout_task.await {
+            tracing::warn!(label = %self.label, err = ?e, "stdout reader task panicked");
+        }
+        if let Err(e) = stderr_task.await {
+            tracing::warn!(label = %self.label, err = ?e, "stderr reader task panicked");
+        }
 
         Ok(status)
     }

crates/tailflow-daemon/src/routes.rs

@@ -75,10 +75,13 @@ async fn sse_handler(
     let filter = params.into_filter();
     let rx = state.tx.subscribe();
     let stream = BroadcastStream::new(rx).filter_map(move |res| match res {
-        Ok(record) if filter.matches(&record) => {
-            let data = serde_json::to_string(&record).unwrap_or_default();
-            Some(Ok(Event::default().data(data)))
-        }
+        Ok(record) if filter.matches(&record) => match serde_json::to_string(&record) {
+            Ok(data) => Some(Ok(Event::default().data(data))),
+            Err(e) => {
+                tracing::error!(err = %e, "failed to serialize log record for SSE");
+                None
+            }
+        },
         Ok(_) => None,
         Err(tokio_stream::wrappers::errors::BroadcastStreamRecvError::Lagged(n)) => {
             tracing::warn!(dropped = n, "SSE client lagged");
@@ -99,7 +102,7 @@ async fn records_handler(
     let records: Vec<_> = state
         .ring
         .lock()
-        .unwrap()
+        .unwrap_or_else(|p| p.into_inner()) // recover from poisoned mutex
         .iter()
         .filter(|r| filter.matches(r))
         .cloned()

crates/tailflow-daemon/src/state.rs

@@ -1,4 +1,7 @@
-use std::sync::{Arc, Mutex};
+use std::{
+    collections::VecDeque,
+    sync::{Arc, Mutex},
+};
 use tailflow_core::{LogReceiver, LogRecord, LogSender};
 use tokio::sync::broadcast;
 
@@ -9,7 +12,7 @@ pub struct AppState {
     /// Subscribe to the live stream by calling `tx.subscribe()`.
     pub tx: LogSender,
     /// Rolling buffer of the last RING_SIZE records (for `/api/records`).
-    pub ring: Mutex<Vec<LogRecord>>,
+    pub ring: Mutex<VecDeque<LogRecord>>,
 }
 
 impl AppState {
@@ -18,7 +21,7 @@ impl AppState {
         let (tx, _) = broadcast::channel(tailflow_core::BUS_CAPACITY);
         let state = Arc::new(AppState {
             tx: tx.clone(),
-            ring: Mutex::new(Vec::with_capacity(RING_SIZE)),
+            ring: Mutex::new(VecDeque::with_capacity(RING_SIZE)),
         });
         let state2 = state.clone();
 
@@ -27,11 +30,11 @@ impl AppState {
                 match source_rx.recv().await {
                     Ok(record) => {
                         {
-                            let mut buf = state2.ring.lock().unwrap();
+                            let mut buf = state2.ring.lock().unwrap_or_else(|p| p.into_inner());
                             if buf.len() >= RING_SIZE {
-                                buf.remove(0);
+                                buf.pop_front(); // O(1) vs Vec::remove(0) O(n)
                             }
-                            buf.push(record.clone());
+                            buf.push_back(record.clone());
                         }
                         let _ = tx.send(record);
                     }

crates/tailflow-tui/src/app.rs

@@ -6,6 +6,7 @@ use crossterm::{
     terminal::{disable_raw_mode, enable_raw_mode, EnterAlternateScreen, LeaveAlternateScreen},
 };
 use ratatui::{backend::CrosstermBackend, Terminal};
+use regex::Regex;
 use std::{io, time::Duration};
 use tailflow_core::{LogReceiver, LogRecord};
 use tokio::sync::broadcast;
@@ -15,6 +16,8 @@ const MAX_RECORDS: usize = 2000;
 pub struct App {
     pub records: Vec<LogRecord>,
     pub filter: String,
+    /// Compiled regex for `filter`; kept in sync whenever `filter` changes.
+    pub filter_re: Option<Regex>,
     pub filter_mode: bool,
     rx: LogReceiver,
     pub scroll: usize,
@@ -66,6 +69,7 @@ impl App {
         Self {
             records: Vec::with_capacity(MAX_RECORDS),
             filter: String::new(),
+            filter_re: None,
             filter_mode: false,
             rx,
             scroll: 0,
@@ -159,9 +163,11 @@ impl App {
                         }
                         (true, KeyCode::Backspace) => {
                             self.filter.pop();
+                            self.filter_re = Regex::new(&self.filter).ok();
                         }
                         (true, KeyCode::Char(c)) => {
                             self.filter.push(c);
+                            self.filter_re = Regex::new(&self.filter).ok();
                         }
 
                         _ => {}

crates/tailflow-tui/src/ui/mod.rs

@@ -20,21 +20,20 @@ pub fn draw(f: &mut Frame, app: &mut App) {
         ])
         .split(area);
 
+    // Guard against degenerate terminal sizes
+    if chunks.len() < 3 {
+        return;
+    }
     let list_height = chunks[1].height as usize;
 
-    // ── Build filter predicate ─────────────────────────────────────────────
-    let filter_re = if !app.filter.is_empty() {
-        regex::Regex::new(&app.filter).ok()
-    } else {
-        None
-    };
+    // ── Build filter predicate (uses pre-compiled regex from App) ──────────
     let filter_lower = app.filter.to_lowercase();
 
     let matches = |payload: &str, source: &str| -> bool {
         if app.filter.is_empty() {
             return true;
         }
-        if let Some(re) = &filter_re {
+        if let Some(re) = &app.filter_re {
             re.is_match(payload) || re.is_match(source)
         } else {
             payload.to_lowercase().contains(&filter_lower)