feat: enhance parsers with validation, performance, and safety improvements

chore: bump version to 0.2.4
fix: skip consecutive invalid bytes and file size validation
feat: parser shutdown handling and message count estimation
feat: validation in parsers and concurrent components
feat: unchecked read methods
enhance: memcpy_nontemporal with alignment check
feat: Direct Listing (O) message support
fix: prevent integer overflow in ZeroCopyParser
fix: replace swap_bytes with to_be
feat: message type validation in ZeroCopyParser
feat: input availability tracking to SpscParser
refactor: replaced atomic status with OnceLock
fix: prevent data loss in compact_buffer
refactor: remove manual padding
refactor: safe pointer arithmetic in zerocopy_types
refactor: Default derive for ParserBuilder
perf: cache SIMD level at startup
refactor: replace manual byte search with memchr
refactor: reorder tempfile dependency
refactor: error handling in SpscParser, ParallelParser, and WorkStealingParser
refactor: remove duplicated WorkUnit slice extraction
refactor: simplify work unit handling
refactor: remove UTF-8 validation in hot path

Author: sundanc

Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "lunary"
-version = "0.2.3"
+version = "0.2.4"
 authors = ["Lunyn <contact@lunyn.com>"]
 edition = "2024"
 license = "AGPL-3.0-or-later"
@@ -27,7 +27,6 @@ crossbeam-channel = "0.5"
 crossbeam-utils = "0.8"
 crossbeam-deque = "0.8"
 core_affinity = { version = "0.8", optional = true }
-tempfile = "3.23.0"
 zerocopy = { version = "0.8", features = ["derive"] }
 memchr = "2.7"
 clap = "4.0"
@@ -43,6 +42,7 @@ mimalloc = { version = "0.1", optional = true }
 criterion = { version = "0.5", default-features = false }
 arbitrary = "1"
 libfuzzer-sys = "0.4"
+tempfile = "3.23.0"
 
 [features]
 default = ["simd"]

Cargo.toml

@@ -288,7 +288,7 @@ pub fn run_feature_comparison(data: &[u8]) -> Result<()> {
         }
     }
     let parallel_time = t0.elapsed();
-    parser.shutdown();
+    let _ = parser.shutdown();
     let parallel_mps = parallel_count as f64 / parallel_time.as_secs_f64() / 1_000_000.0;
 
     println!(

src/bench/diagnostics.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;
@@ -62,7 +64,7 @@ pub fn bench_worksteal(data: &[u8]) -> Result<(u64, f64, f64)> {
     }
 
     let wall = t0.elapsed();
-    parser.shutdown();
+    let _ = parser.shutdown();
 
     let (elapsed_ms, mps) = calculate_throughput(total_messages as u64, wall);
     Ok((total_messages as u64, elapsed_ms, mps))
@@ -206,7 +208,7 @@ pub fn run_worksteal(data: &[u8]) -> Result<()> {
         stats.errors()
     );
 
-    parser.shutdown();
+    let _ = parser.shutdown();
     Ok(())
 }
 
@@ -353,6 +355,6 @@ pub fn run_worker_stats(data: &[u8]) -> Result<()> {
         println!("Load balance: {:.1}% (100% = perfect)", balance);
     }
 
-    parser.shutdown();
+    let _ = parser.shutdown();
     Ok(())
 }

src/bench/parallel.rs

@@ -238,5 +238,6 @@ fn get_message_type_name(msg: &Message) -> &'static str {
         Message::NetOrderImbalance(_) => "NetOrderImbalance",
         Message::RetailPriceImprovement(_) => "RetailPriceImprovement",
         Message::LuldAuctionCollar(_) => "LuldAuctionCollar",
+        Message::DirectListing(_) => "DirectListing",
     }
 }

src/bench/single.rs

@@ -51,6 +51,7 @@ impl Default for ParserConfig {
     }
 }
 
+#[derive(Default)]
 pub struct ParserBuilder {
     config: ParserConfig,
 }
@@ -198,12 +199,6 @@ impl ParserBuilder {
     }
 }
 
-impl Default for ParserBuilder {
-    fn default() -> Self {
-        Self::new()
-    }
-}
-
 pub enum AnyParser<'a> {
     Simple(Box<Parser>),
     Batch(Box<BatchProcessor>),