fix: Cleanup pass — 7 audit findings

crates/logfwd-core/src/format.rs

@@ -108,8 +108,13 @@ impl FormatParser for RawParser {
                     match b {
                         b'"' => out.extend_from_slice(b"\\\""),
                         b'\\' => out.extend_from_slice(b"\\\\"),
+                        b'\n' => out.extend_from_slice(b"\\n"),
                         b'\r' => out.extend_from_slice(b"\\r"),
                         b'\t' => out.extend_from_slice(b"\\t"),
+                        b if b < 0x20 => {
+                            // Escape control characters per RFC 8259.
+                            let _ = std::io::Write::write_fmt(out, format_args!("\\u{:04x}", b));
+                        }
                         _ => out.push(b),
                     }
                 }

crates/logfwd-core/src/streaming_builder.rs

@@ -173,8 +173,8 @@ impl StreamingBuilder {
         }
         self.written_bits |= bit;
         let fc = &mut self.fields[idx];
-        fc.has_int = true;
         if let Some(v) = parse_int_fast(value) {
+            fc.has_int = true;
             fc.int_values.push((self.row_count, v));
         }
     }
@@ -187,8 +187,8 @@ impl StreamingBuilder {
         }
         self.written_bits |= bit;
         let fc = &mut self.fields[idx];
-        fc.has_float = true;
         if let Some(v) = parse_float_fast(value) {
+            fc.has_float = true;
             fc.float_values.push((self.row_count, v));
         }
     }

crates/logfwd-output/src/fanout.rs

@@ -21,17 +21,35 @@ impl FanOut {
 
 impl OutputSink for FanOut {
     fn send_batch(&mut self, batch: &RecordBatch, meta: &BatchMetadata) -> io::Result<()> {
+        // Try ALL sinks before returning an error. Don't short-circuit.
+        let mut first_err: Option<io::Error> = None;
         for sink in &mut self.sinks {
-            sink.send_batch(batch, meta)?;
+            if let Err(e) = sink.send_batch(batch, meta) {
+                eprintln!("fanout: sink '{}' failed: {e}", sink.name());
+                if first_err.is_none() {
+                    first_err = Some(e);
+                }
+            }
+        }
+        match first_err {
+            Some(e) => Err(e),
+            None => Ok(()),
         }
-        Ok(())
     }
 
     fn flush(&mut self) -> io::Result<()> {
+        let mut first_err: Option<io::Error> = None;
         for sink in &mut self.sinks {
-            sink.flush()?;
+            if let Err(e) = sink.flush()
+                && first_err.is_none()
+            {
+                first_err = Some(e);
+            }
+        }
+        match first_err {
+            Some(e) => Err(e),
+            None => Ok(()),
         }
-        Ok(())
     }
 
     fn name(&self) -> &str {

crates/logfwd-output/src/json_lines.rs

@@ -15,15 +15,21 @@ pub struct JsonLinesSink {
     url: String,
     headers: Vec<(String, String)>,
     pub batch_buf: Vec<u8>,
+    http_agent: ureq::Agent,
 }
 
 impl JsonLinesSink {
     pub fn new(name: String, url: String, headers: Vec<(String, String)>) -> Self {
+        let http_agent = ureq::config::Config::builder()
+            .timeout_global(Some(std::time::Duration::from_secs(30)))
+            .build()
+            .new_agent();
         JsonLinesSink {
             name,
             url,
             headers,
             batch_buf: Vec::with_capacity(64 * 1024),
+            http_agent,
         }
     }
 
@@ -87,7 +93,7 @@ impl OutputSink for JsonLinesSink {
             return Ok(());
         }
 
-        let mut req = ureq::post(&self.url);
+        let mut req = self.http_agent.post(&self.url);
         for (k, v) in &self.headers {
             req = req.header(k.as_str(), v.as_str());
         }

crates/logfwd-output/src/otlp_sink.rs

@@ -32,6 +32,7 @@ pub struct OtlpSink {
     pub encoder_buf: Vec<u8>,
     compress_buf: Vec<u8>,
     compressor: Option<ChunkCompressor>,
+    http_agent: ureq::Agent,
 }
 
 impl OtlpSink {
@@ -45,6 +46,10 @@ impl OtlpSink {
             Compression::Zstd => Some(ChunkCompressor::new(1)),
             _ => None,
         };
+        let http_agent = ureq::config::Config::builder()
+            .timeout_global(Some(std::time::Duration::from_secs(30)))
+            .build()
+            .new_agent();
         OtlpSink {
             name,
             endpoint,
@@ -53,6 +58,7 @@ impl OtlpSink {
             encoder_buf: Vec::with_capacity(64 * 1024),
             compress_buf: Vec::with_capacity(64 * 1024),
             compressor,
+            http_agent,
         }
     }
 
@@ -155,7 +161,7 @@ impl OutputSink for OtlpSink {
             OtlpProtocol::Http => "application/x-protobuf",
         };
 
-        let mut req = ureq::post(&self.endpoint);
+        let mut req = self.http_agent.post(&self.endpoint);
         req = req.header("Content-Type", content_type);
         if self.compression == Compression::Zstd {
             req = req.header("Content-Encoding", "zstd");

crates/logfwd-transform/src/lib.rs

@@ -528,22 +528,21 @@ impl SqlTransform {
     }
 
     /// Add an enrichment table that will be registered in each DataFusion
-    /// session alongside the `logs` table. Panics if a table with the same
-    /// name is already registered or if the name conflicts with "logs".
+    /// session alongside the `logs` table. Returns an error if a table with
+    /// the same name is already registered or if the name conflicts with "logs".
     pub fn add_enrichment_table(
         &mut self,
         table: Arc<dyn logfwd_core::enrichment::EnrichmentTable>,
-    ) {
+    ) -> Result<(), String> {
         let name = table.name();
-        assert!(
-            name != "logs",
-            "enrichment table cannot be named 'logs' (reserved)"
-        );
-        assert!(
-            !self.enrichment_tables.iter().any(|t| t.name() == name),
-            "duplicate enrichment table name: '{name}'"
-        );
+        if name == "logs" {
+            return Err("enrichment table cannot be named 'logs' (reserved)".to_string());
+        }
+        if self.enrichment_tables.iter().any(|t| t.name() == name) {
+            return Err(format!("duplicate enrichment table name: '{name}'"));
+        }
         self.enrichment_tables.push(table);
+        Ok(())
     }
 
     /// Execute the SQL transform on a RecordBatch.