feat: validation in parsers and concurrent components

Author: sundanc

src/bench/parallel.rs

@@ -51,7 +51,9 @@ pub fn bench_worksteal(data: &[u8]) -> Result<(u64, f64, f64)> {
     let chunks = split_into_chunks(data, chunk_size);
 
     for (start, end) in &chunks {
-        parser.submit_arc(Arc::clone(&data_arc), *start, *end);
+        parser
+            .submit_arc(Arc::clone(&data_arc), *start, *end)
+            .unwrap();
     }
 
     let mut total_messages = 0;

src/concurrent.rs

@@ -309,7 +309,7 @@ impl ConcurrentParser for SpscParser {
             if let Ok(msgs) = self.output_receiver.try_recv() {
                 return Some(msgs);
             }
-            if self.shutdown.load(Ordering::Relaxed) {
+            if self.shutdown.load(Ordering::Acquire) {
                 return None;
             }
             let (lock, cvar) = &*self.has_data;
@@ -1062,7 +1062,8 @@ impl AdaptiveBatchProcessor {
         }
         let n = self.throughput_count as f64;
         let mean = self.throughput_sum / n;
-        (self.throughput_sum_sq / n) - (mean * mean)
+        let variance = (self.throughput_sum_sq / n) - (mean * mean);
+        variance.max(0.0)
     }
 
     #[inline]
@@ -1714,11 +1715,17 @@ impl WorkStealingParser {
         self.injector.push(WorkUnit::Owned(data));
     }
 
-    pub fn submit_arc(&self, data: Arc<[u8]>, start: usize, end: usize) {
+    pub fn submit_arc(&self, data: Arc<[u8]>, start: usize, end: usize) -> Result<()> {
         if start >= end || end > data.len() {
-            return;
+            return Err(ParseError::InvalidArgument(format!(
+                "invalid range {}..{} for data of length {}",
+                start,
+                end,
+                data.len()
+            )));
         }
         self.injector.push(WorkUnit::ArcSlice(data, start, end));
+        Ok(())
     }
 
     pub fn submit_chunks(&self, data: Arc<[u8]>, chunk_size: usize) -> usize {

src/config.rs

@@ -3,6 +3,7 @@ pub struct Config {
     pub max_buffer_size: usize,
     pub max_message_size: usize,
     pub initial_capacity: usize,
+    pub strict_validation: bool,
 }
 
 impl Default for Config {
@@ -17,6 +18,7 @@ impl Config {
             max_buffer_size: 1024 * 1024 * 1024, // 1 GB
             max_message_size: 64 * 1024,         // 64 KB
             initial_capacity: 2 * 1024 * 1024,   // 2 MB
+            strict_validation: false,
         }
     }
 
@@ -40,6 +42,11 @@ impl Config {
         self
     }
 
+    pub const fn with_strict_validation(mut self, strict: bool) -> Self {
+        self.strict_validation = strict;
+        self
+    }
+
     pub fn from_size_mb(size_mb: usize) -> Self {
         Self::new().with_max_buffer_size_mb(size_mb)
     }

src/error.rs

@@ -32,6 +32,9 @@ pub enum ParseError {
     #[error("Invalid UTF-8 in field '{field}'")]
     InvalidUtf8 { field: &'static str },
 
+    #[error("Invalid timestamp: {value} nanoseconds (must be <= 86,400,000,000,000)")]
+    InvalidTimestamp { value: u64 },
+
     #[error("Parser state error: {reason}")]
     StateError { reason: &'static str },
 
@@ -59,6 +62,7 @@ impl ParseError {
                 | ParseError::InvalidHeader { .. }
                 | ParseError::LengthMismatch { .. }
                 | ParseError::InvalidUtf8 { .. }
+                | ParseError::InvalidTimestamp { .. }
         )
     }
 

src/parser.rs

@@ -212,12 +212,18 @@ impl Parser {
         }
 
         let message_type = remaining[2];
-        let expected = EXPECTED_LENGTHS[message_type as usize] as usize;
-        if expected != 0 && expected != length {
+        let expected = EXPECTED_LENGTHS[message_type as usize];
+        if expected == NO_VALIDATION {
+            #[cfg(debug_assertions)]
+            eprintln!(
+                "Warning: no expected length for message type 0x{:02X}",
+                message_type
+            );
+        } else if expected as usize != length {
             return Err(ParseError::LengthMismatch {
                 msg_type: message_type,
                 declared: length,
-                expected,
+                expected: expected as usize,
             });
         }
 
@@ -486,6 +492,10 @@ impl Parser {
             data[*pos + 5],
         ]);
         *pos += 6;
+        const MAX_VALID_TIMESTAMP: u64 = 86_400_000_000_000;
+        if self.config.strict_validation && value > MAX_VALID_TIMESTAMP {
+            return Err(ParseError::InvalidTimestamp { value });
+        }
         Ok(value)
     }
 
@@ -917,8 +927,10 @@ impl Parser {
     }
 }
 
+const NO_VALIDATION: u16 = u16::MAX;
+
 const EXPECTED_LENGTHS: [u16; 256] = {
-    let mut arr = [0u16; 256];
+    let mut arr = [NO_VALIDATION; 256];
     arr[b'S' as usize] = 12;
     arr[b'R' as usize] = 39;
     arr[b'H' as usize] = 25;

src/simd.rs

@@ -934,7 +934,7 @@ unsafe fn memcpy_nontemporal_avx2(dst: *mut u8, src: *const u8, len: usize) {
 ///
 /// Caller must ensure:
 /// - `src` is valid for reads of `len` bytes
-/// - `dst` is valid for writes of `len` bytes and is 16-byte aligned
+/// - `dst` is valid for writes of `len` bytes
 /// - the regions do not overlap
 pub unsafe fn memcpy_nontemporal(dst: *mut u8, src: *const u8, len: usize) {
     unsafe {