fix: TemplateStore eviction cleanup, error reporting, and doc accurac…

Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 name = "netflow_parser"
 description = "Parser for Netflow Cisco V5, V7, V9, IPFIX"
-version = "1.0.3"
+version = "1.0.4"
 edition = "2024"
 rust-version = "1.88"
 authors = ["Michael Mileusnich <michael.mileusnich@gmail.com>"]

RELEASES.md

@@ -1,3 +1,76 @@
+# 1.0.4
+
+## Fixes
+
+* **`TemplateStore`: source-eviction in `AutoScopedParser` no longer leaks
+  store entries.** When `max_sources` capacity was reached and a per-source
+  parser was popped from the LRU, every template that source had written
+  under its scope (e.g. `v9:10.0.0.1:2055/0`) was orphaned in the store
+  forever. `evict_global_lru` now calls `clear_v9_templates` /
+  `clear_ipfix_templates` on the evicted parser before dropping it, so the
+  external store keyspace tracks the live source set.
+
+* **`TemplateStore`: `clear_v9_templates` / `clear_ipfix_templates` now record
+  backend errors.** Previously these methods called `let _ = store.remove(...)`,
+  silently swallowing failures. They now bump
+  `template_store_backend_errors` on each failed `remove`, matching every
+  other store call site.
+
+* **`TemplateStore`: misleading inline doc comment on `clear_*_templates`
+  removed.** The comment claimed that after a clear, "subsequent reads do
+  not transparently repopulate the in-process cache via read-through" — true
+  only for templates that were in the in-process LRU at clear time. Templates
+  evicted from the LRU before the call, or written by another parser instance
+  under the same scope, remain reachable via read-through. The trait
+  intentionally exposes only `get`/`put`/`remove`, not a per-scope wipe; the
+  comment now documents this honestly.
+
+* **`set_template_store_scope` rustdoc**: clarified that the scope must be
+  set before the first `parse_bytes` call to avoid orphaning entries written
+  under the previous scope, and that `with_template_store_scope` on a builder
+  fed into `AutoScopedParser` is overridden by the auto-derived per-source
+  scope.
+
+* **Read-through hit semantics documented.** Clarified that a successful
+  `TemplateStore` read-through increments `hits` (counted as a hit, not a
+  miss) and that restored templates have their TTL re-stamped to
+  `Instant::now()`.
+
+## Tests
+
+* **Rewrote `read_through_drives_pending_flow_replay`.** The previous test
+  never queued a pending flow before the template arrived — it would have
+  passed even if the read-through-driven pending-flow replay code were
+  deleted. Now exercises the full path: data record arrives before any
+  template is known → queued → template is written to the store by another
+  replica → next data record's read-through restores the template AND
+  triggers replay of the queued flow.
+
+* **Strengthened `auto_scoped_parser_uses_per_source_scope`.** Now also
+  validates cross-replica round-trip: a fresh `AutoScopedParser` reading the
+  store must decode each source's data record against the correctly-scoped
+  template.
+
+* **Added eviction-cleanup test for `AutoScopedParser`** verifying that
+  evicted source parsers' store entries are removed.
+
+* **Coverage filled in** for backend `remove` failures (`inject_remove_failures`
+  is now exercised), IPFIX-side codec corruption rejection, IPFIX-side LRU
+  eviction propagation to the store, and IPFIX-side `TemplateEvent::Restored`
+  firing.
+
+## Examples
+
+* **New example: `horizontal_scale_out_template_store`.** Demonstrates the
+  feature's headline use case — two `NetflowParser` instances sharing an
+  `InMemoryTemplateStore`. Replica A learns a template and goes away;
+  replica B starts cold and decodes a data record against the template via
+  read-through. Run with:
+
+  ```sh
+  cargo run --example horizontal_scale_out_template_store
+  ```
+
 # 1.0.3
 
 ## Features

examples/horizontal_scale_out_template_store.rs

@@ -0,0 +1,195 @@
+//! Horizontal scale-out via a shared `TemplateStore`.
+//!
+//! Demonstrates the use case the `TemplateStore` extension point exists for:
+//! running multiple stateless parser replicas behind a UDP load balancer
+//! without source-IP-affinity routing.
+//!
+//! Replica A learns a template, writes it through to a shared store, and
+//! goes away. Replica B starts cold — its in-process template cache is
+//! empty — and immediately receives a data record for that template.
+//! With the store configured, replica B transparently restores the template
+//! from the store and decodes the record. Without the store, the same data
+//! record would queue in pending flows or fail to decode.
+//!
+//! In production you would back the `TemplateStore` with Redis, NATS KV,
+//! or similar; the in-memory store used here keeps the example self
+//! contained. The trait sees only opaque `Vec<u8>` payloads, so the
+//! protocol is identical regardless of backend.
+//!
+//! Run with:
+//! ```sh
+//! cargo run --example horizontal_scale_out_template_store
+//! ```
+
+use netflow_parser::{
+    InMemoryTemplateStore, NetflowPacket, NetflowParser, TemplateEvent, TemplateProtocol,
+};
+use std::sync::Arc;
+use std::sync::Mutex;
+
+fn main() {
+    println!("=== Horizontal scale-out demo: two parsers sharing a TemplateStore ===\n");
+
+    // The store any production deployment would back with Redis, NATS KV,
+    // DynamoDB, etc. Implements the `TemplateStore` trait — get / put /
+    // remove on opaque byte payloads.
+    let store = Arc::new(InMemoryTemplateStore::new());
+
+    // ------------------------------------------------------------------
+    // Replica A: learns a template, persists it via write-through.
+    // ------------------------------------------------------------------
+    println!("[replica A] starting up, will learn one template");
+    let mut replica_a = NetflowParser::builder()
+        .with_template_store(Arc::clone(&store) as _)
+        .build()
+        .expect("build replica A");
+
+    let template_packet = build_v9_template_packet(256, &[(8, 4), (12, 4), (1, 8)]);
+    let result = replica_a.parse_bytes(&template_packet);
+    if let Some(err) = result.error {
+        panic!("template parse failed: {err}");
+    }
+    println!(
+        "[replica A] learned template 256, store now has {} entr(ies)\n",
+        store.len()
+    );
+
+    // Replica A goes away — drop it. The store is the only surviving
+    // record of the template. A real deployment might drop replica A
+    // because it crashed, scaled down, or rolled.
+    drop(replica_a);
+
+    // ------------------------------------------------------------------
+    // Replica B: starts cold, no in-process templates. Receives a data
+    // record for template 256 and must decode it via read-through.
+    // ------------------------------------------------------------------
+    println!("[replica B] starting cold (no in-process template cache)");
+
+    // Wire up a hook so we can observe the Restored event. In production
+    // this is how an observability system would distinguish "template
+    // recovered from secondary tier" from "template freshly learned from
+    // exporter announce" — both look like cache hits in the basic metric.
+    let restored_log: Arc<Mutex<Vec<(TemplateProtocol, u16)>>> =
+        Arc::new(Mutex::new(Vec::new()));
+    let restored_log_for_hook = Arc::clone(&restored_log);
+
+    let mut replica_b = NetflowParser::builder()
+        .with_template_store(Arc::clone(&store) as _)
+        .on_template_event(move |event| {
+            if let TemplateEvent::Restored {
+                template_id: Some(id),
+                protocol,
+            } = event
+            {
+                restored_log_for_hook
+                    .lock()
+                    .expect("poisoned")
+                    .push((*protocol, *id));
+            }
+            Ok(())
+        })
+        .build()
+        .expect("build replica B");
+
+    // 16 bytes = three fields (4 + 4 + 8) matching the template above.
+    let data_payload = [
+        // src IP = 10.0.0.1
+        0x0A, 0x00, 0x00, 0x01, // dst IP = 10.0.0.2
+        0x0A, 0x00, 0x00, 0x02, // bytes = 4096
+        0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x10, 0x00,
+    ];
+    let data_packet = build_v9_data_packet(256, &data_payload);
+
+    let result = replica_b.parse_bytes(&data_packet);
+    if let Some(err) = result.error {
+        panic!("data parse on replica B failed: {err}");
+    }
+
+    let v9 = result
+        .packets
+        .into_iter()
+        .find_map(|p| match p {
+            NetflowPacket::V9(v) => Some(v),
+            _ => None,
+        })
+        .expect("expected a V9 packet");
+    let flowset_count = v9.flowsets.len();
+    println!(
+        "[replica B] decoded data packet against restored template ({} flowset(s))",
+        flowset_count
+    );
+
+    // ------------------------------------------------------------------
+    // Observability — what metrics and events fired?
+    // ------------------------------------------------------------------
+    let metrics = replica_b.v9_cache_info().metrics;
+    println!("\n[replica B] cache metrics after read-through:");
+    println!("    hits                     = {}", metrics.hits);
+    println!("    misses                   = {}", metrics.misses);
+    println!(
+        "    template_store_restored  = {}",
+        metrics.template_store_restored
+    );
+    println!(
+        "    template_store_codec_err = {}",
+        metrics.template_store_codec_errors
+    );
+    println!(
+        "    template_store_backend_err = {}",
+        metrics.template_store_backend_errors
+    );
+
+    let restored = restored_log.lock().expect("poisoned");
+    println!("\n[replica B] TemplateEvent::Restored events:");
+    for (protocol, id) in restored.iter() {
+        println!("    {:?} template_id={}", protocol, id);
+    }
+
+    println!("\nDone. The same protocol works for IPFIX and IPFIX-options templates.");
+    println!(
+        "Hot-path overhead when no store is configured is a single Option::is_none branch."
+    );
+}
+
+// --- packet builders --------------------------------------------------------
+// Minimal V9 packet construction for the demo. In production these come from
+// the wire — exporters announce templates, then send data records that
+// reference them.
+
+fn build_v9_template_packet(template_id: u16, fields: &[(u16, u16)]) -> Vec<u8> {
+    let template_record_len = 4 + fields.len() * 4; // template header + fields
+    let flowset_len = 4 + template_record_len; // set header + record
+    let mut pkt = Vec::new();
+    // V9 header (20 bytes)
+    pkt.extend_from_slice(&9u16.to_be_bytes()); // version
+    pkt.extend_from_slice(&1u16.to_be_bytes()); // count
+    pkt.extend_from_slice(&0u32.to_be_bytes()); // sys_up_time
+    pkt.extend_from_slice(&0u32.to_be_bytes()); // unix_secs
+    pkt.extend_from_slice(&0u32.to_be_bytes()); // sequence
+    pkt.extend_from_slice(&0u32.to_be_bytes()); // source_id
+    // Template flowset
+    pkt.extend_from_slice(&0u16.to_be_bytes()); // flowset_id = 0 (template)
+    pkt.extend_from_slice(&(flowset_len as u16).to_be_bytes());
+    pkt.extend_from_slice(&template_id.to_be_bytes());
+    pkt.extend_from_slice(&(fields.len() as u16).to_be_bytes());
+    for &(ft, fl) in fields {
+        pkt.extend_from_slice(&ft.to_be_bytes());
+        pkt.extend_from_slice(&fl.to_be_bytes());
+    }
+    pkt
+}
+
+fn build_v9_data_packet(template_id: u16, payload: &[u8]) -> Vec<u8> {
+    let flowset_len = 4 + payload.len();
+    let mut pkt = Vec::new();
+    pkt.extend_from_slice(&9u16.to_be_bytes());
+    pkt.extend_from_slice(&1u16.to_be_bytes());
+    pkt.extend_from_slice(&0u32.to_be_bytes());
+    pkt.extend_from_slice(&0u32.to_be_bytes());
+    pkt.extend_from_slice(&0u32.to_be_bytes());
+    pkt.extend_from_slice(&0u32.to_be_bytes());
+    pkt.extend_from_slice(&template_id.to_be_bytes());
+    pkt.extend_from_slice(&(flowset_len as u16).to_be_bytes());
+    pkt.extend_from_slice(payload);
+    pkt
+}