fix: enforce one NetFlow v9 template owner

README.md

@@ -219,6 +219,8 @@ let parser = NetflowParser::builder()
 - Templates are keyed by template ID (per source)
 - Native and V9-style IPFIX data and Options templates share one Template ID namespace;
   each accepted definition becomes the sole owner of its ID
+- NetFlow v9 ordinary and Options templates share one Template ID namespace;
+  the last valid definition replaces the previous owner
 - Each parser instance maintains its own template cache
 - For multi-source deployments, use `RouterScopedParser` (see Template Management section)
 

src/variable_versions/v9/parser.rs

@@ -39,6 +39,9 @@ use std::num::NonZeroUsize;
 use std::sync::Arc;
 
 /// Stateful NetFlow V9 parser with LRU template caching and optional pending flow support.
+///
+/// Ordinary and Options templates share one protocol-level Template ID namespace.
+/// The last valid definition received for an ID is its sole logical owner.
 #[derive(Debug)]
 pub struct V9Parser {
     pub(crate) templates: LazyLruCache<TemplateId, TemplateWithTtl<Arc<Template>>>,
@@ -193,8 +196,8 @@ impl V9Parser {
         }
     }
 
-    /// Best-effort removal of an LRU-evicted entry from the secondary store.
-    fn evict_template_from_store(&mut self, kind: TemplateKind, template_id: u16) {
+    /// Best-effort removal of a superseded or evicted entry from the secondary store.
+    fn remove_template_from_store(&mut self, kind: TemplateKind, template_id: u16) {
         let Some(store) = self.template_store.as_ref() else {
             return;
         };
@@ -205,6 +208,87 @@ impl V9Parser {
         }
     }
 
+    /// Remove one physical cache entry and report whether it was still a live owner.
+    fn remove_cached_owner<T>(
+        cache: &mut LazyLruCache<TemplateId, TemplateWithTtl<Arc<T>>>,
+        template_id: TemplateId,
+        ttl_config: &Option<TtlConfig>,
+        metrics: &mut CacheMetricsInner,
+    ) -> bool {
+        let Some(entry) = cache.pop(&template_id) else {
+            return false;
+        };
+
+        if ttl_config.as_ref().is_some_and(|ttl| entry.is_expired(ttl)) {
+            metrics.record_expiration();
+            false
+        } else {
+            true
+        }
+    }
+
+    /// Install an ordinary template as the sole owner of its Template ID.
+    fn install_template(&mut self, template: &Template) {
+        let template_id = template.template_id;
+        if Self::remove_cached_owner(
+            &mut self.options_templates,
+            template_id,
+            &self.ttl_config,
+            &mut self.metrics,
+        ) {
+            self.metrics.record_collision();
+        }
+        self.remove_template_from_store(TemplateKind::V9Options, template_id);
+
+        if let Some(existing) = self.templates.peek(&template_id)
+            && existing.template.as_ref() != template
+        {
+            self.metrics.record_collision();
+        }
+
+        let wrapped =
+            TemplateWithTtl::new(Arc::new(template.clone()), self.ttl_config.is_some());
+        if let Some((evicted_key, _)) = self.templates.push(template_id, wrapped)
+            && evicted_key != template_id
+        {
+            self.metrics.record_eviction();
+            self.remove_template_from_store(TemplateKind::V9Data, evicted_key);
+        }
+        self.metrics.record_insertion();
+        self.store_template(template);
+    }
+
+    /// Install an Options template as the sole owner of its Template ID.
+    fn install_options_template(&mut self, template: &OptionsTemplate) {
+        let template_id = template.template_id;
+        if Self::remove_cached_owner(
+            &mut self.templates,
+            template_id,
+            &self.ttl_config,
+            &mut self.metrics,
+        ) {
+            self.metrics.record_collision();
+        }
+        self.remove_template_from_store(TemplateKind::V9Data, template_id);
+
+        if let Some(existing) = self.options_templates.peek(&template_id)
+            && existing.template.as_ref() != template
+        {
+            self.metrics.record_collision();
+        }
+
+        let wrapped =
+            TemplateWithTtl::new(Arc::new(template.clone()), self.ttl_config.is_some());
+        if let Some((evicted_key, _)) = self.options_templates.push(template_id, wrapped)
+            && evicted_key != template_id
+        {
+            self.metrics.record_eviction();
+            self.remove_template_from_store(TemplateKind::V9Options, evicted_key);
+        }
+        self.metrics.record_insertion();
+        self.store_options_template(template);
+    }
+
     /// Insert a read-through-recovered template into the in-process LRU,
     /// mirroring eviction back to the secondary store and tracking the
     /// `Restored` event for hook firing after this packet completes.
@@ -707,29 +791,8 @@ impl FlowSetBody {
                         nom::error::ErrorKind::Verify,
                     )));
                 }
-                let ttl_enabled = parser.ttl_config.is_some();
                 for template in &valid_templates {
-                    let arc_template = Arc::new(template.clone());
-                    let wrapped = TemplateWithTtl::new(arc_template, ttl_enabled);
-                    // Check for collision (same ID, different definition)
-                    // Use peek() to avoid affecting LRU ordering
-                    if let Some(existing) = parser.templates.peek(&template.template_id)
-                        && existing.template.as_ref() != template
-                    {
-                        parser.metrics.record_collision();
-                    }
-                    // push() returns Some in two cases: (1) a different key was LRU-evicted
-                    // to make room, or (2) the same key existed and its value was replaced.
-                    // Only count case (1) as an eviction.
-                    if let Some((evicted_key, _evicted)) =
-                        parser.templates.push(template.template_id, wrapped)
-                        && evicted_key != template.template_id
-                    {
-                        parser.metrics.record_eviction();
-                        parser.evict_template_from_store(TemplateKind::V9Data, evicted_key);
-                    }
-                    parser.metrics.record_insertion();
-                    parser.store_template(template);
+                    parser.install_template(template);
                 }
                 let result = Templates {
                     templates: valid_templates,
@@ -752,30 +815,8 @@ impl FlowSetBody {
                         nom::error::ErrorKind::Verify,
                     )));
                 }
-                // Store templates efficiently using Arc for zero-cost sharing
-                let ttl_enabled = parser.ttl_config.is_some();
                 for template in &valid_templates {
-                    let arc_template = Arc::new(template.clone());
-                    let wrapped = TemplateWithTtl::new(arc_template, ttl_enabled);
-                    // Check for collision (same ID, different definition)
-                    // Use peek() to avoid affecting LRU ordering
-                    if let Some(existing) = parser.options_templates.peek(&template.template_id)
-                        && existing.template.as_ref() != template
-                    {
-                        parser.metrics.record_collision();
-                    }
-                    // push() returns Some in two cases: (1) a different key was LRU-evicted
-                    // to make room, or (2) the same key existed and its value was replaced.
-                    // Only count case (1) as an eviction.
-                    if let Some((evicted_key, _evicted)) =
-                        parser.options_templates.push(template.template_id, wrapped)
-                        && evicted_key != template.template_id
-                    {
-                        parser.metrics.record_eviction();
-                        parser.evict_template_from_store(TemplateKind::V9Options, evicted_key);
-                    }
-                    parser.metrics.record_insertion();
-                    parser.store_options_template(template);
+                    parser.install_options_template(template);
                 }
                 let result = OptionsTemplates {
                     templates: valid_templates,

tests/v9_template_id_ownership.rs

@@ -0,0 +1,165 @@
+use netflow_parser::variable_versions::v9::FlowSetBody;
+use netflow_parser::{
+    InMemoryTemplateStore, NetflowPacket, NetflowParser, TemplateKind, TemplateStore,
+    TemplateStoreKey, TtlConfig,
+};
+use std::sync::Arc;
+use std::thread;
+use std::time::Duration;
+
+const TEMPLATE_ID: u16 = 256;
+const STORE_SCOPE: &str = "v9-template-owner";
+
+fn v9_header(count: u16) -> Vec<u8> {
+    let mut packet = Vec::with_capacity(20);
+    packet.extend_from_slice(&9u16.to_be_bytes());
+    packet.extend_from_slice(&count.to_be_bytes());
+    packet.extend_from_slice(&0u32.to_be_bytes());
+    packet.extend_from_slice(&0u32.to_be_bytes());
+    packet.extend_from_slice(&0u32.to_be_bytes());
+    packet.extend_from_slice(&42u32.to_be_bytes());
+    packet
+}
+
+fn flowset(id: u16, body: &[u8]) -> Vec<u8> {
+    let mut bytes = Vec::with_capacity(4 + body.len());
+    bytes.extend_from_slice(&id.to_be_bytes());
+    bytes.extend_from_slice(&u16::try_from(4 + body.len()).unwrap().to_be_bytes());
+    bytes.extend_from_slice(body);
+    bytes
+}
+
+fn data_template() -> Vec<u8> {
+    let mut body = Vec::with_capacity(8);
+    body.extend_from_slice(&TEMPLATE_ID.to_be_bytes());
+    body.extend_from_slice(&1u16.to_be_bytes());
+    body.extend_from_slice(&1u16.to_be_bytes());
+    body.extend_from_slice(&4u16.to_be_bytes());
+    flowset(0, &body)
+}
+
+fn options_template() -> Vec<u8> {
+    let mut body = Vec::with_capacity(16);
+    body.extend_from_slice(&TEMPLATE_ID.to_be_bytes());
+    body.extend_from_slice(&4u16.to_be_bytes());
+    body.extend_from_slice(&4u16.to_be_bytes());
+    body.extend_from_slice(&1u16.to_be_bytes());
+    body.extend_from_slice(&4u16.to_be_bytes());
+    body.extend_from_slice(&2u16.to_be_bytes());
+    body.extend_from_slice(&4u16.to_be_bytes());
+    flowset(1, &body)
+}
+
+fn data_flowset() -> Vec<u8> {
+    flowset(TEMPLATE_ID, &[0, 0, 0, 1, 0, 0, 0, 2])
+}
+
+fn packet(flowsets: &[&[u8]]) -> Vec<u8> {
+    let mut packet = v9_header(u16::try_from(flowsets.len()).unwrap());
+    for flowset in flowsets {
+        packet.extend_from_slice(flowset);
+    }
+    packet
+}
+
+fn parsed_v9(
+    result: &netflow_parser::ParseResult,
+) -> &netflow_parser::variable_versions::v9::V9 {
+    assert!(result.error.is_none(), "{:?}", result.error);
+    assert_eq!(result.packets.len(), 1);
+    let NetflowPacket::V9(packet) = &result.packets[0] else {
+        panic!("expected NetFlow v9 packet");
+    };
+    packet
+}
+
+#[test]
+fn last_template_kind_owns_the_id_in_wire_order() {
+    for options_last in [true, false] {
+        let ordinary = data_template();
+        let options = options_template();
+        let data = data_flowset();
+        let definitions = if options_last {
+            [&ordinary[..], &options[..], &data[..]]
+        } else {
+            [&options[..], &ordinary[..], &data[..]]
+        };
+
+        let mut parser = NetflowParser::default();
+        let result = parser.parse_bytes(&packet(&definitions));
+        let v9 = parsed_v9(&result);
+
+        if options_last {
+            assert!(matches!(v9.flowsets[2].body, FlowSetBody::OptionsData(_)));
+        } else {
+            assert!(matches!(v9.flowsets[2].body, FlowSetBody::Data(_)));
+        }
+
+        let cache = parser.v9_cache_info();
+        assert_eq!(cache.current_size, 1);
+        assert_eq!(cache.num_caches, 2);
+        assert_eq!(cache.metrics.insertions, 2);
+        assert_eq!(cache.metrics.collisions, 1);
+        assert_eq!(cache.metrics.evictions, 0);
+    }
+}
+
+#[test]
+fn template_store_retains_only_the_last_kind() {
+    for options_last in [true, false] {
+        let store = Arc::new(InMemoryTemplateStore::new());
+        let builder = NetflowParser::builder()
+            .with_template_store(store.clone())
+            .with_template_store_scope(STORE_SCOPE);
+        let mut writer = builder.clone().build().unwrap();
+
+        let ordinary = data_template();
+        let options = options_template();
+        let definitions = if options_last {
+            [&ordinary[..], &options[..]]
+        } else {
+            [&options[..], &ordinary[..]]
+        };
+        let result = writer.parse_bytes(&packet(&definitions));
+        parsed_v9(&result);
+
+        let data_key = TemplateStoreKey::new(STORE_SCOPE, TemplateKind::V9Data, TEMPLATE_ID);
+        let options_key =
+            TemplateStoreKey::new(STORE_SCOPE, TemplateKind::V9Options, TEMPLATE_ID);
+        assert_eq!(store.get(&data_key).unwrap().is_some(), !options_last);
+        assert_eq!(store.get(&options_key).unwrap().is_some(), options_last);
+
+        let mut reader = builder.build().unwrap();
+        let data = data_flowset();
+        let result = reader.parse_bytes(&packet(&[&data]));
+        let v9 = parsed_v9(&result);
+        if options_last {
+            assert!(matches!(v9.flowsets[0].body, FlowSetBody::OptionsData(_)));
+        } else {
+            assert!(matches!(v9.flowsets[0].body, FlowSetBody::Data(_)));
+        }
+        assert_eq!(reader.v9_cache_info().current_size, 1);
+    }
+}
+
+#[test]
+fn expired_opposite_kind_is_not_counted_as_a_collision() {
+    let mut parser = NetflowParser::builder()
+        .with_v9_ttl(TtlConfig::new(Duration::from_millis(1)))
+        .build()
+        .unwrap();
+
+    let ordinary = data_template();
+    parsed_v9(&parser.parse_bytes(&packet(&[&ordinary])));
+    thread::sleep(Duration::from_millis(5));
+
+    let options = options_template();
+    parsed_v9(&parser.parse_bytes(&packet(&[&options])));
+
+    let cache = parser.v9_cache_info();
+    assert_eq!(cache.current_size, 1);
+    assert_eq!(cache.metrics.insertions, 2);
+    assert_eq!(cache.metrics.expired, 1);
+    assert_eq!(cache.metrics.collisions, 0);
+    assert_eq!(cache.metrics.evictions, 0);
+}