improve slow operations

CHANGELOG.md

@@ -12,17 +12,25 @@ All notable changes to this project will be documented in this file.
 
 * **BGP Domain Path (D-PATH) attribute support** ([RFC 10039](https://datatracker.ietf.org/doc/html/rfc10039)): path attribute type 36 now parses into `AttributeValue::DomainPath(DomainPathAttribute)` — segments of `DomainPathDomain { global_admin, local_admin, isf_safi_type }` — and re-encodes byte-identically instead of falling into the `Unknown` raw catch-all. Malformed values, and a D-PATH on a family other than IPVPN or EVPN, become RFC 7606 validation findings with the bytes retained raw, and the encoder refuses values it cannot represent. Text rendering and byte-level dissection label the attribute `BGP_DOMAIN_PATH`.
 * **`--color auto|always|never` for `--format text`**: colors session keys, section headers, prefixes, and next-hop values with ANSI accents that follow the terminal theme. `auto` (the default) colors only when stdout is a terminal; `NO_COLOR` disables coloring and `CLICOLOR_FORCE` forces it. Library callers opt in with `render::text::Style::ansi()` plus the new `format_record_with_style` / `format_record_with_hex_and_style`; `format_record` and the unstyled output are unchanged, since styling is a post-pass over the rendered block.
+* **`parse_ris_live_message_iter`**: parses one RIS Live JSON message into an iterator over its elems instead of a `Vec`. Every elem carries its own copy of the message's AS path and communities, so collecting a message that announces many prefixes needs memory proportional to prefixes × path length; the iterator makes those copies one elem at a time. `parse_ris_live_message` is unchanged and now builds on it.
 
 ### Changed
 
 * **Internal cleanups, no behavior change**: the MP next-hop encoder (`encode_mp_next_hop`) is shared by the `NEXT_HOP` and MP_REACH encoders instead of being duplicated, `AsPathSegment` hashing skips the sort when a set is already ordered, `Elementor::record_to_elems` logs peer-table conversion errors as its documentation promises, the invariant `unreachable!()` arms state their invariant, and both the crate-wide `uninlined_format_args` allow and a module-wide `#![allow(unused)]` are removed.
 * **`--format text` session labels are now `PEER`/`LOCAL`**: the endpoint lines read `PEER: <peer_ip> AS<peer_asn>` and `LOCAL: <local_ip> AS<local_asn>`, matching the peer/local names MRT ([RFC 6396](https://www.rfc-editor.org/rfc/rfc6396.html)) uses for the same fields instead of the ambiguous `FROM`/`TO`. The rendered values are unchanged.
+* **Record parse errors are logged once per parser**: the default, raw, update and route iterators log the first `ParseError` of a parser at `error` level and later ones at `debug` level. A corrupted stream re-synchronises header by header, which previously emitted one `error` line per 12 bytes of damaged input. Other error classes are logged as before.
 * **Faster elem conversion**: `into_elem_iter` is about 24% faster on updates and 21% faster on RIB dumps, mostly from fewer copies and allocations when building elems; output is unchanged.
 
 ### Fixed
 
+* **RIS Live JSON elems carry `origin_asns`**: `parse_ris_live_message` left `origin_asns` empty, so an `origin_asn` filter never matched an elem parsed from RIS Live's JSON fields. It is now derived from the AS path, as `parse_ris_live_message_raw` and the MRT iterators already did.
+* **Dissection span correlation is linear in the number of warnings**: `span_record_warnings` (used by `DiagnosticIterator::with_dissection`) walked the whole dissection tree once per warning, so a BGP UPDATE carrying thousands of duplicate attributes took tens of seconds to diagnose. The tree is now indexed by field name once per record. Spans are unchanged.
 * **Fallible iterators stop after a fatal stream read error**: `into_fallible_record_iter`, `into_fallible_elem_iter`, `into_fallible_update_iter` and `into_fallible_route_iter` now yield a framing I/O or decompression error once and then end, instead of polling a spent decoder indefinitely. An `Interrupted`/`WouldBlock` read stays retryable only when it consumed no framing bytes, and skippable malformed records are unchanged ([#350](https://github.com/bgpkit/bgpkit-parser/pull/350)).
 
+### Contributors
+
+* @ties — faster elem conversion ([#348](https://github.com/bgpkit/bgpkit-parser/pull/348)), `parse_ris_live_message_iter` and RIS Live origin ASNs, once-per-parser parse error logging, linear-time dissection span correlation
+
 ## v0.22.0 - 2026-09-10
 
 ### Breaking changes

src/parser/iters/diagnostic.rs

@@ -242,27 +242,62 @@ pub fn span_record_warnings(
     warnings: &[BgpValidationWarning],
     tree: &DissectionNode,
 ) -> Vec<SpannedWarning> {
+    // Most records are clean: don't walk the tree for them at all.
+    if warnings.is_empty() {
+        return Vec::new();
+    }
+    // Index the tree once: a record can carry as many warnings as it has
+    // attributes, and walking the whole tree per warning is quadratic.
+    let index = SpanIndex::build(tree);
     let mut duplicate_counts: HashMap<u8, usize> = HashMap::new();
     warnings
         .iter()
         .map(|warning| SpannedWarning {
-            span: locate_warning_span(warning, tree, &mut duplicate_counts),
+            span: locate_warning_span(warning, &index, &mut duplicate_counts)
+                .unwrap_or_else(|| tree.span()),
             warning: warning.clone(),
         })
         .collect()
 }
 
+/// Spans of every node in a dissection tree, keyed by field name, in tree
+/// (pre-order) order: `nth(field, 0)` is the node `DissectionNode::find`
+/// would return, and `nth(field, n)` the `n`th match of `find_all`.
+struct SpanIndex<'a> {
+    spans: HashMap<&'a str, Vec<Span>>,
+}
+
+impl<'a> SpanIndex<'a> {
+    fn build(tree: &'a DissectionNode) -> Self {
+        let mut spans = HashMap::new();
+        Self::walk(tree, &mut spans);
+        Self { spans }
+    }
+
+    fn walk(node: &'a DissectionNode, spans: &mut HashMap<&'a str, Vec<Span>>) {
+        spans
+            .entry(node.field.as_str())
+            .or_default()
+            .push(node.span());
+        for child in &node.children {
+            Self::walk(child, spans);
+        }
+    }
+
+    fn nth(&self, field: &str, occurrence: usize) -> Option<Span> {
+        self.spans.get(field)?.get(occurrence).copied()
+    }
+}
+
 fn locate_warning_span(
     warning: &BgpValidationWarning,
-    tree: &DissectionNode,
+    index: &SpanIndex<'_>,
     duplicate_counts: &mut HashMap<u8, usize>,
-) -> Span {
+) -> Option<Span> {
     use BgpValidationWarning as W;
 
     let attr_span = |code: u8, occurrence: usize| -> Option<Span> {
-        let mut nodes = Vec::new();
-        tree.find_all(&format!("bgp.attr.{code}"), &mut nodes);
-        nodes.get(occurrence).map(|node| node.span())
+        index.nth(&format!("bgp.attr.{code}"), occurrence)
     };
     // Anchor to the attribute occurrence the warning is about. The parser
     // emits warnings while observing headers in wire order and reports
@@ -273,9 +308,9 @@ fn locate_warning_span(
         let occurrence = duplicate_counts.get(&code).copied().unwrap_or(0);
         attr_span(code, occurrence)
     };
-    let section_span = |field: &str| -> Option<Span> { tree.find(field).map(|node| node.span()) };
+    let section_span = |field: &str| -> Option<Span> { index.nth(field, 0) };
 
-    let span = match warning {
+    match warning {
         W::AttributeFlagsError { attr_type, .. }
         | W::AttributeLengthError { attr_type, .. }
         | W::OptionalAttributeError { attr_type, .. }
@@ -307,8 +342,7 @@ fn locate_warning_span(
         W::UnknownRouteRefreshSubtype { .. } | W::InvalidRouteRefreshLength { .. } => {
             section_span("bgp.route_refresh")
         }
-    };
-    span.unwrap_or_else(|| tree.span())
+    }
 }
 
 #[cfg(test)]

src/parser/iters/mod.rs

@@ -156,7 +156,7 @@ pub(crate) fn handle_record_parse_error<R>(
             true
         }
         ParserError::ParseError(err_str) => {
-            error!("parser error: {}", err_str);
+            parser.log_parse_error_once(&err_str);
             write_mrt_core_dump(parser.core_dump, bytes);
             // stop after writing the dump so later failures don't overwrite it
             !parser.core_dump

src/parser/iters/raw.rs

@@ -11,14 +11,9 @@ when possible and continue processing the remaining data. It also supports confi
 warning messages and core dump generation for debugging purposes.
 */
 
-use crate::parser::iters::{
-    handle_record_parse_error, record_matches_filters, write_mrt_core_dump,
-};
+use crate::parser::iters::{handle_record_parse_error, record_matches_filters};
 use crate::parser::mrt::mrt_record::raw_record_uses_zebra_compat;
-use crate::{
-    chunk_mrt_record, BgpkitParser, Elementor, Filter, MrtRecord, ParserError, RawMrtRecord,
-};
-use log::{error, warn};
+use crate::{chunk_mrt_record, BgpkitParser, Elementor, Filter, MrtRecord, RawMrtRecord};
 use std::io::Read;
 
 pub struct RawRecordIterator<R> {
@@ -44,52 +39,11 @@ impl<R: Read> Iterator for RawRecordIterator<R> {
         loop {
             match chunk_mrt_record(&mut self.parser.reader) {
                 Ok(raw_record) => return Some(raw_record),
-                Err(e) => match e.error {
-                    ParserError::TruncatedMsg(err_str) | ParserError::Unsupported(err_str) => {
-                        if self.parser.options.show_warnings {
-                            warn!("parser warn: {}", err_str);
-                        }
-                        write_mrt_core_dump(self.parser.core_dump, e.bytes);
-                        continue;
-                    }
-                    ParserError::ParseError(err_str) => {
-                        error!("parser error: {}", err_str);
-                        write_mrt_core_dump(self.parser.core_dump, e.bytes);
-                        if self.parser.core_dump {
-                            return None;
-                        } else {
-                            continue;
-                        }
-                    }
-                    ParserError::EofExpected => {
-                        // normal end of file
-                        return None;
-                    }
-                    ParserError::IoError(err) | ParserError::EofError(err) => {
-                        // when reaching IO error, stop iterating
-                        error!("{:?}", err);
-                        write_mrt_core_dump(self.parser.core_dump, e.bytes);
+                Err(e) => {
+                    if !handle_record_parse_error(&mut self.parser, e.error, e.bytes) {
                         return None;
                     }
-                    #[cfg(feature = "oneio")]
-                    ParserError::OneIoError(_) => return None,
-                    ParserError::FilterError(_) => {
-                        // this should not happen at this stage
-                        return None;
-                    }
-                    // Labeled NLRI parsing errors - treat as malformed and skip
-                    ParserError::InvalidLabeledNlriLength
-                    | ParserError::TruncatedLabeledNlri
-                    | ParserError::TruncatedPrefix
-                    | ParserError::MaxLabelStackDepthExceeded
-                    | ParserError::PeerMaxLabelsExceeded
-                    | ParserError::InvalidPrefix => {
-                        if self.parser.options.show_warnings {
-                            warn!("parser warn: labeled NLRI parsing error: {:?}", e.error);
-                        }
-                        continue;
-                    }
-                },
+                }
             }
         }
     }

src/parser/iters/route.rs

@@ -2,7 +2,7 @@ use crate::error::{ParserError, ParserErrorWithBytes};
 use crate::models::*;
 use crate::parser::bgp::attributes::{parse_as_path, parse_nlri, AttributeValidationState};
 use crate::parser::bgp::messages::read_and_validate_bgp_marker;
-use crate::parser::iters::write_mrt_core_dump;
+use crate::parser::iters::{handle_record_parse_error, write_mrt_core_dump};
 use crate::parser::mrt::messages::bgp4mp::{bgp4mp_message_payload_len, is_short_zebra_open};
 use crate::parser::mrt::messages::legacy_bgp::{
     BGP_KEEPALIVE, BGP_NOTIFY, BGP_OPEN, BGP_STATE_CHANGE, BGP_UPDATE,
@@ -679,6 +679,16 @@ impl<R> RouteIterator<R> {
     }
 }
 
+/// Log an error from parsing a record body into routes. Only a `ParseError` goes through the
+/// once-per-parser limiter, as in the other iterators, so that another error class cannot use up
+/// the one `error`-level parse error line.
+fn log_route_body_error<R>(parser: &mut BgpkitParser<R>, err: &ParserError) {
+    match err {
+        ParserError::ParseError(err_str) => parser.log_parse_error_once(err_str),
+        other => error!("parser error: {}", other),
+    }
+}
+
 impl<R: Read> Iterator for RouteIterator<R> {
     type Item = BgpRouteElem;
 
@@ -699,7 +709,7 @@ impl<R: Read> Iterator for RouteIterator<R> {
                     self.pending_raw_bytes = None;
                 }
                 Err(err) => {
-                    error!("parser error: {}", err);
+                    log_route_body_error(&mut self.parser, &err);
                     self.pending_routes = RouteRecordIter::Empty;
                     write_mrt_core_dump(self.parser.core_dump, self.pending_raw_bytes.take());
                     if self.parser.core_dump {
@@ -711,43 +721,12 @@ impl<R: Read> Iterator for RouteIterator<R> {
 
             let raw_record = match chunk_mrt_record(&mut self.parser.reader) {
                 Ok(raw_record) => raw_record,
-                Err(e) => match e.error {
-                    ParserError::TruncatedMsg(err_str) | ParserError::Unsupported(err_str) => {
-                        if self.parser.options.show_warnings {
-                            warn!("parser warn: {}", err_str);
-                        }
-                        write_mrt_core_dump(self.parser.core_dump, e.bytes);
-                        continue;
-                    }
-                    ParserError::ParseError(err_str) => {
-                        error!("parser error: {}", err_str);
-                        write_mrt_core_dump(self.parser.core_dump, e.bytes);
-                        if self.parser.core_dump {
-                            return None;
-                        }
-                        continue;
-                    }
-                    ParserError::EofExpected => return None,
-                    ParserError::IoError(err) | ParserError::EofError(err) => {
-                        error!("{:?}", err);
-                        write_mrt_core_dump(self.parser.core_dump, e.bytes);
+                Err(e) => {
+                    if !handle_record_parse_error(&mut self.parser, e.error, e.bytes) {
                         return None;
                     }
-                    #[cfg(feature = "oneio")]
-                    ParserError::OneIoError(_) => return None,
-                    ParserError::FilterError(_) => return None,
-                    ParserError::InvalidLabeledNlriLength
-                    | ParserError::TruncatedLabeledNlri
-                    | ParserError::TruncatedPrefix
-                    | ParserError::MaxLabelStackDepthExceeded
-                    | ParserError::PeerMaxLabelsExceeded
-                    | ParserError::InvalidPrefix => {
-                        if self.parser.options.show_warnings {
-                            warn!("parser warn: labeled NLRI parsing error: {:?}", e.error);
-                        }
-                        continue;
-                    }
-                },
+                    continue;
+                }
             };
 
             let used_zebra_compat = raw_record_uses_zebra_compat(&raw_record);
@@ -761,7 +740,7 @@ impl<R: Read> Iterator for RouteIterator<R> {
                     self.pending_raw_bytes = Some(raw_bytes);
                 }
                 Err(err) => {
-                    error!("parser error: {}", err);
+                    log_route_body_error(&mut self.parser, &err);
                     write_mrt_core_dump(self.parser.core_dump, Some(raw_bytes));
                     if self.parser.core_dump {
                         return None;

src/parser/iters/update.rs

@@ -44,12 +44,10 @@ for announcement in parser.into_update_iter() {
 }
 ```
 */
-use crate::error::ParserError;
 use crate::models::*;
-use crate::parser::iters::write_mrt_core_dump;
+use crate::parser::iters::handle_record_parse_error;
 use crate::parser::BgpkitParser;
 use crate::Elementor;
-use log::{error, warn};
 use std::io::Read;
 use std::net::IpAddr;
 
@@ -179,44 +177,12 @@ impl<R: Read> Iterator for UpdateIterator<R> {
             }
             let record = match self.parser.next_record() {
                 Ok(record) => record,
-                Err(e) => match e.error {
-                    ParserError::TruncatedMsg(err_str) | ParserError::Unsupported(err_str) => {
-                        if self.parser.options.show_warnings {
-                            warn!("parser warn: {}", err_str);
-                        }
-                        write_mrt_core_dump(self.parser.core_dump, e.bytes);
-                        continue;
-                    }
-                    ParserError::ParseError(err_str) => {
-                        error!("parser error: {}", err_str);
-                        write_mrt_core_dump(self.parser.core_dump, e.bytes);
-                        if self.parser.core_dump {
-                            return None;
-                        }
-                        continue;
-                    }
-                    ParserError::EofExpected => return None,
-                    ParserError::IoError(err) | ParserError::EofError(err) => {
-                        error!("{:?}", err);
-                        write_mrt_core_dump(self.parser.core_dump, e.bytes);
+                Err(e) => {
+                    if !handle_record_parse_error(&mut self.parser, e.error, e.bytes) {
                         return None;
                     }
-                    #[cfg(feature = "oneio")]
-                    ParserError::OneIoError(_) => return None,
-                    ParserError::FilterError(_) => return None,
-                    // Labeled NLRI parsing errors - treat as malformed and skip
-                    ParserError::InvalidLabeledNlriLength
-                    | ParserError::TruncatedLabeledNlri
-                    | ParserError::TruncatedPrefix
-                    | ParserError::MaxLabelStackDepthExceeded
-                    | ParserError::PeerMaxLabelsExceeded
-                    | ParserError::InvalidPrefix => {
-                        if self.parser.options.show_warnings {
-                            warn!("parser warn: labeled NLRI parsing error: {:?}", e.error);
-                        }
-                        continue;
-                    }
-                },
+                    continue;
+                }
             };
 
             let t = record.common_header.timestamp;