fix: deep code review passes 25-28 - 18 issues fixed (#276)

* fix: deep code review passes 25-28 - 18 issues fixed

Pass 25 (6 fixes):
- IPFIX V9 options template validation aligned with V9 parser (reject individual zero-length fields)
- would_accept() false negative causing data truncation fixed
- set_pending_flows_config(None) now records dropped metrics
- IPFIX replay flowset count bound added (consistent with V9)
- YAF DnsSRVTarget (219) type corrected to String
- YAF SslPublicKeyLength (250) type corrected to UnsignedDataNumber

Pass 26 (3 fixes):
- IPFIX "withdraw all" templates implemented per RFC 7011 §8.1
- V9 replay flowset limit uses break instead of continue
- IPFIX withdrawal+redefinition in same flowset no longer drains pending flows

Pass 27 (3 fixes):
- Fuzz round-trip target uses fresh parser for re-parse (no accumulated state)
- README: 3 missing examples added
- Redundant test renamed to test_v5_serialization_single_packet

Pass 28 (3 fixes):
- IPFIX withdraw-all now drains only same-type pending flows (not cross-type)
- test_pcap_iterator_api now only counts successful parses
- README v0.7.0 breaking change note updated for v1.0.0

Additional improvements:
- Dead macro parameter removed from impl_try_from!
- TemplateHooks::clear() method added
- test_parser_builder_with_field_count_limits made non-vacuous
- test_parser_builder_comprehensive strengthened with functional verification

* fix: fmt

* fix: tests

Author: mikemiles-dev

README.md

@@ -317,7 +317,7 @@ The parser also automatically validates:
 
 ### Template TTL (Time-to-Live)
 
-> **⚠️ Breaking Change in v0.7.0:** Packet-based and combined TTL modes have been removed. Only time-based TTL is now supported. See [RELEASES.md](RELEASES.md) for migration guide.
+> **Note:** Only time-based TTL is supported. See [RELEASES.md](RELEASES.md) for details.
 
 Optionally configure templates to expire after a time duration. This is useful for:
 - Handling exporters that reuse template IDs with different schemas
@@ -706,18 +706,18 @@ pub struct NetflowCommon {
     pub flowsets: Vec<NetflowCommonFlowSet>,
 }
 
-#[derive(Debug, Default)]
-struct NetflowCommonFlowSet {
-    src_addr: Option<IpAddr>,
-    dst_addr: Option<IpAddr>,
-    src_port: Option<u16>,
-    dst_port: Option<u16>,
-    protocol_number: Option<u8>,
-    protocol_type: Option<ProtocolTypes>,
-    first_seen: Option<u32>,
-    last_seen: Option<u32>,
-    src_mac: Option<String>,
-    dst_mac: Option<String>,
+#[derive(Debug, Default, Clone)]
+pub struct NetflowCommonFlowSet {
+    pub src_addr: Option<IpAddr>,
+    pub dst_addr: Option<IpAddr>,
+    pub src_port: Option<u16>,
+    pub dst_port: Option<u16>,
+    pub protocol_number: Option<u8>,
+    pub protocol_type: Option<ProtocolTypes>,
+    pub first_seen: Option<u64>,
+    pub last_seen: Option<u64>,
+    pub src_mac: Option<String>,
+    pub dst_mac: Option<String>,
 }
 ```
 
@@ -1194,7 +1194,13 @@ To run:
 
 ```cargo run --example custom_enterprise_fields```
 
-The pcap example also shows how to cache flows that have not yet discovered a template. The custom_enterprise_fields example demonstrates how to register vendor-specific IPFIX fields.
+```cargo run --example template_hooks```
+
+```cargo run --example template_management_demo```
+
+```cargo run --example multi_source_comparison```
+
+The pcap example also shows how to cache flows that have not yet discovered a template. The custom_enterprise_fields example demonstrates how to register vendor-specific IPFIX fields. The template_hooks example shows how to monitor template lifecycle events.
 
 ## Support My Work
 

fuzz/fuzz_targets/fuzz_round_trip.rs

@@ -6,11 +6,13 @@ use netflow_parser::{NetflowPacket, NetflowParser};
 // Round-trip fuzz target: parse → serialize → re-parse → compare.
 // Catches serialization bugs where to_be_bytes() produces output that
 // doesn't re-parse to the same structure.
+//
+// Uses a persistent PARSER to accumulate template state across iterations
+// (exercising caching/collision paths), but re-parses with a FRESH parser
+// each time to ensure the serialized output is self-contained.
 thread_local! {
     static PARSER: std::cell::RefCell<NetflowParser> =
         std::cell::RefCell::new(NetflowParser::default());
-    static REPARSE: std::cell::RefCell<NetflowParser> =
-        std::cell::RefCell::new(NetflowParser::default());
 }
 
 fuzz_target!(|data: &[u8]| {
@@ -33,34 +35,40 @@ fuzz_target!(|data: &[u8]| {
                 _ => continue,
             };
 
-            // Re-parse the serialized output and verify it produces a valid packet
-            REPARSE.with(|rp| {
-                let mut reparser = rp.borrow_mut();
-                let re_result = reparser.parse_bytes(&serialized);
-
-                // Serialized output of a successful parse must re-parse without error
-                assert!(
-                    re_result.error.is_none(),
-                    "Round-trip failed: serialized output of {:?} did not re-parse cleanly: {:?}",
-                    std::mem::discriminant(packet),
-                    re_result.error,
-                );
+            // Re-parse with a fresh parser to ensure serialized output is
+            // self-contained (not relying on accumulated template state).
+            let mut reparser = NetflowParser::default();
+            let re_result = reparser.parse_bytes(&serialized);
 
-                // Must produce at least one packet
-                assert!(
-                    !re_result.packets.is_empty(),
-                    "Round-trip failed: serialized output produced no packets",
-                );
+            // Serialized output of a successful parse must re-parse without error
+            assert!(
+                re_result.error.is_none(),
+                "Round-trip failed: serialized output of {:?} did not re-parse cleanly: {:?}",
+                std::mem::discriminant(packet),
+                re_result.error,
+            );
 
-                // Verify the re-parsed packet is the same variant
-                if let Some(re_packet) = re_result.packets.first() {
-                    assert_eq!(
-                        std::mem::discriminant(packet),
-                        std::mem::discriminant(re_packet),
-                        "Round-trip changed packet type",
+            // Must produce at least one packet (V5/V7 always, V9/IPFIX
+            // may produce template-only packets with no data flowsets).
+            // For V9/IPFIX template-only packets, an empty result is acceptable.
+            match packet {
+                NetflowPacket::V5(_) | NetflowPacket::V7(_) => {
+                    assert!(
+                        !re_result.packets.is_empty(),
+                        "Round-trip failed: serialized V5/V7 output produced no packets",
                     );
                 }
-            });
+                _ => {}
+            }
+
+            // Verify the re-parsed packet is the same variant (when present)
+            if let Some(re_packet) = re_result.packets.first() {
+                assert_eq!(
+                    std::mem::discriminant(packet),
+                    std::mem::discriminant(re_packet),
+                    "Round-trip changed packet type",
+                );
+            }
         }
     });
 });

src/template_events.rs

@@ -190,6 +190,11 @@ impl TemplateHooks {
         self.hooks.push(Arc::new(hook));
     }
 
+    /// Removes all registered hooks.
+    pub fn clear(&mut self) {
+        self.hooks.clear();
+    }
+
     /// Triggers all registered hooks with the given event.
     ///
     /// All hooks are called regardless of whether earlier hooks return errors or panic.

src/variable_versions/field_value.rs

@@ -16,7 +16,7 @@ use serde::ser::Serializer;
 use std::net::{IpAddr, Ipv4Addr, Ipv6Addr};
 
 macro_rules! impl_try_from {
-    ($($t:ty => $v:ident),*; $($s:ty => $sv:ident),*) => {
+    ($($t:ty => $v:ident),*) => {
         $(
             impl TryFrom<&DataNumber> for $t {
                 type Error = DataNumberError;
@@ -159,7 +159,7 @@ impl_try_from!(
     u64 => U64,
     i64 => I64,
     u128 => U128,
-    i128 => I128;
+    i128 => I128
 );
 
 // Manual TryFrom for u32/i32 to also handle U24/I24 variants

src/variable_versions/ipfix/lookup.rs

@@ -240,7 +240,7 @@ pub enum YafIPFixField {
         DnsSRVPriority = 216 => FieldDataType::UnsignedDataNumber,
         DnsSRVWeight = 217 => FieldDataType::UnsignedDataNumber,
         DnsSRVPort = 218 => FieldDataType::UnsignedDataNumber,
-        DnsSRVTarget = 219 => FieldDataType::UnsignedDataNumber,
+        DnsSRVTarget = 219 => FieldDataType::String,
         TcpUrgTotalCount = 223 => FieldDataType::UnsignedDataNumber,
         DnsID = 226 => FieldDataType::UnsignedDataNumber,
         SslCertSerialNumber = 244 => FieldDataType::String,
@@ -249,7 +249,7 @@ pub enum YafIPFixField {
         SslCertValidityNotBefore = 247 => FieldDataType::String,
         SslCertValidityNotAfter = 248 => FieldDataType::String,
         SslPublicKeyAlgorithm = 249 => FieldDataType::String,
-        SslPublicKeyLength = 250 => FieldDataType::String,
+        SslPublicKeyLength = 250 => FieldDataType::UnsignedDataNumber,
         RtpPayloadType = 287 => FieldDataType::UnsignedDataNumber,
         ReverseRtpPayloadType = 288 => FieldDataType::UnsignedDataNumber,
         MptcpInitialDataSequenceNumber = 289 => FieldDataType::UnsignedDataNumber,

src/variable_versions/ipfix/parser.rs

@@ -140,6 +140,13 @@ impl ParserConfig for IPFixParser {
                 }
             }
             None => {
+                // Record all cached entries as dropped before discarding.
+                if let Some(ref cache) = self.pending_flows {
+                    let count = cache.count();
+                    if count > 0 {
+                        self.metrics.record_pending_dropped_n(count as u64);
+                    }
+                }
                 self.pending_flows = None;
             }
         }
@@ -284,6 +291,14 @@ impl IPFixParser {
             let entries = cache.drain(template_id, &mut self.metrics);
             let total_entries = entries.len();
             for (processed, entry) in entries.iter().enumerate() {
+                // Bound flowset count, consistent with V9 replay.
+                if ipfix.flowsets.len() >= u16::MAX as usize {
+                    let remaining = (total_entries - processed) as u64;
+                    for _ in 0..remaining {
+                        self.metrics.record_pending_replay_failed();
+                    }
+                    break;
+                }
                 let flowset_length =
                     u16::try_from(entry.raw_data.len().saturating_add(4)).unwrap_or(u16::MAX);
                 let Some(new_header_length) = ipfix.header.length.checked_add(flowset_length)
@@ -550,29 +565,79 @@ impl IPFixParser {
         );
     }
 
-    /// Remove an IPFIX template by ID (RFC 7011 Section 3.4.3 template withdrawal).
+    /// Remove an IPFIX template by ID (RFC 7011 Section 8.1 template withdrawal).
     /// Also purges any pending flows cached under this template ID to prevent
     /// stale data from being replayed against a replacement template.
+    ///
+    /// Per RFC 7011 §8.1, template_id == DATA_TEMPLATE_IPFIX_ID (2) with
+    /// field_count == 0 signals "withdraw ALL data templates". This method
+    /// handles both individual and bulk withdrawal.
     fn withdraw_ipfix_template(&mut self, template_id: u16) {
-        self.templates.pop(&template_id);
-        if let Some(ref mut cache) = self.pending_flows {
-            let drained = cache.drain(template_id, &mut self.metrics);
-            let n = drained.len() as u64;
-            if n > 0 {
-                self.metrics.record_pending_dropped_n(n);
+        if template_id == DATA_TEMPLATE_IPFIX_ID {
+            // "Withdraw all data templates" — clear entire data template
+            // cache and drain pending flows only for those template IDs.
+            // Pending flows for options template IDs are left untouched
+            // since those templates remain valid.
+            let ids: Vec<u16> = self.templates.iter().map(|(&id, _)| id).collect();
+            for id in &ids {
+                self.templates.pop(id);
+            }
+            if let Some(ref mut cache) = self.pending_flows {
+                for &id in &ids {
+                    let drained = cache.drain(id, &mut self.metrics);
+                    let n = drained.len() as u64;
+                    if n > 0 {
+                        self.metrics.record_pending_dropped_n(n);
+                    }
+                }
+            }
+        } else {
+            self.templates.pop(&template_id);
+            if let Some(ref mut cache) = self.pending_flows {
+                let drained = cache.drain(template_id, &mut self.metrics);
+                let n = drained.len() as u64;
+                if n > 0 {
+                    self.metrics.record_pending_dropped_n(n);
+                }
             }
         }
     }
 
     /// Remove an IPFIX options template by ID (template withdrawal).
     /// Also purges any pending flows cached under this template ID.
+    ///
+    /// Per RFC 7011 §8.1, template_id == OPTIONS_TEMPLATE_IPFIX_ID (3) with
+    /// field_count == 0 signals "withdraw ALL options templates".
     fn withdraw_ipfix_options_template(&mut self, template_id: u16) {
-        self.ipfix_options_templates.pop(&template_id);
-        if let Some(ref mut cache) = self.pending_flows {
-            let drained = cache.drain(template_id, &mut self.metrics);
-            let n = drained.len() as u64;
-            if n > 0 {
-                self.metrics.record_pending_dropped_n(n);
+        if template_id == OPTIONS_TEMPLATE_IPFIX_ID {
+            // "Withdraw all options templates" — clear entire options template
+            // cache and drain pending flows only for those template IDs.
+            // Pending flows for data template IDs are left untouched.
+            let ids: Vec<u16> = self
+                .ipfix_options_templates
+                .iter()
+                .map(|(&id, _)| id)
+                .collect();
+            for id in &ids {
+                self.ipfix_options_templates.pop(id);
+            }
+            if let Some(ref mut cache) = self.pending_flows {
+                for &id in &ids {
+                    let drained = cache.drain(id, &mut self.metrics);
+                    let n = drained.len() as u64;
+                    if n > 0 {
+                        self.metrics.record_pending_dropped_n(n);
+                    }
+                }
+            }
+        } else {
+            self.ipfix_options_templates.pop(&template_id);
+            if let Some(ref mut cache) = self.pending_flows {
+                let drained = cache.drain(template_id, &mut self.metrics);
+                let n = drained.len() as u64;
+                if n > 0 {
+                    self.metrics.record_pending_dropped_n(n);
+                }
             }
         }
     }
@@ -596,13 +661,26 @@ impl FlowSetBody {
     {
         let (i, templates) = many0(complete(parse_fn))(i)?;
 
-        // Handle template withdrawals (RFC 7011 Section 3.4.3):
+        // Handle template withdrawals (RFC 7011 Section 8.1):
         // Templates with field_count=0 signal withdrawal from the cache.
+        // Skip withdrawal for IDs that also have a new definition in the
+        // same flowset — the new definition will simply replace the old one
+        // without needlessly draining pending flows.
         let mut had_withdrawals = false;
         if let Some(withdraw_fn) = withdraw_template {
             for t in &templates {
                 if t.field_count() == 0 {
-                    withdraw_fn(parser, t.template_id());
+                    let id = t.template_id();
+                    // "Withdraw all" IDs (2 for data, 3 for options) always
+                    // take effect regardless of other templates in the batch.
+                    let has_redefinition = id != DATA_TEMPLATE_IPFIX_ID
+                        && id != OPTIONS_TEMPLATE_IPFIX_ID
+                        && templates
+                            .iter()
+                            .any(|other| other.template_id() == id && other.field_count() > 0);
+                    if !has_redefinition {
+                        withdraw_fn(parser, id);
+                    }
                     had_withdrawals = true;
                 }
             }
@@ -696,10 +774,11 @@ impl FlowSetBody {
                         && usize::from(t.get_total_size()) <= p.max_template_total_size
                         && !t.has_duplicate_scope_fields()
                         && !t.has_duplicate_option_fields()
-                        // Reject templates where all fields have zero length to prevent
-                        // zero-byte-per-record parsing issues
-                        && (t.scope_fields.iter().any(|f| f.field_length > 0)
-                            || t.option_fields.iter().any(|f| f.field_length > 0))
+                        // V9 does not support variable-length fields; reject any
+                        // zero-length or variable-length sentinel (65535) fields,
+                        // consistent with the V9 parser's own validation.
+                        && t.scope_fields.iter().all(|f| f.field_length > 0 && f.field_length != 65535)
+                        && t.option_fields.iter().all(|f| f.field_length > 0 && f.field_length != 65535)
                 },
                 |parser, templates| parser.add_v9_options_templates(templates),
                 None, // V9 doesn't support template withdrawal

src/variable_versions/pending_flows.rs

@@ -119,8 +119,10 @@ impl PendingFlowCache {
         if data_len > self.config.max_entry_size_bytes {
             return false;
         }
-        // Reject if adding this entry would exceed the total byte budget
-        if self.total_bytes.saturating_add(data_len) > self.config.max_total_bytes {
+        // Check total byte budget, but allow if eviction could make room.
+        // Only reject if the single entry alone exceeds the total budget
+        // (cache() evicts LRU templates to free space when over budget).
+        if data_len > self.config.max_total_bytes {
             return false;
         }
         match self.cache.peek(&template_id) {