Add a recovering parser

CHANGELOG.md

@@ -49,6 +49,7 @@ All notable changes to this project will be documented in this file.
 #### Test data
 
 * **Historical RIPE regression fixtures**: Added original RRC00 update and bview gzip files from 1999 and January 2000 as repository-only, offline integration fixtures.
+* **Opt-in MRT framing recovery**: Added `into_recovering_record_iter`, `into_recovering_elem_iter`, and CLI `--recover` support for salvaging records after damaged MRT framing. Recovery validates a three-record chain, uses exact embedded BGP headers as BGP4MP anchors, and reports every skipped decompressed byte range as a typed gap event. A record of any MRT type whose framing is intact but whose body fails to parse is skipped exactly (`AlignedRecordChain` evidence) when intact records follow its declared boundary, and damage extending to the end of the stream — e.g. a truncated final record — is reported as a terminal gap (`EndOfStream` evidence) instead of an error. Text-dump parsers, which have no MRT record representation, yield an explicit `Unsupported` error.
 
 #### Examples
 
@@ -66,6 +67,7 @@ All notable changes to this project will be documented in this file.
 * **BGP OPEN parameter type 255 rejected**: RFC 9072 reserves type 255 as the extended-length marker; encoding it as a real parameter produced output that round-tripped to a structurally different message.
 * **BGP OPEN optional-parameter encoding**: Encode the Optional Parameters Length as the total byte length required by RFC 4271 instead of the number of parameters. OPEN messages now also use the extended length format from RFC 9072 when requested or required.
 * **Historical Quagga state changes**: Parse BGP4MP state-change records containing Quagga's `Clearing` (7) and `Deleted` (8) FSM states instead of logging an error and dropping the MRT record.
+* **CLI broken-pipe exit status**: the CLI now exits 0 when its stdout consumer closes the pipe early (e.g. `bgpkit-parser updates.gz | head`), matching Unix convention; previous releases exited 1. Other write errors still exit 1.
 
 ### Contributors
 

Cargo.toml

@@ -16,6 +16,8 @@ exclude = [
     "tests/fixtures/ripe/**",
     "tests/legacy_ripe_2000.rs",
     "tests/quagga_bgp_states.rs",
+    "tests/recovery.rs",
+    "tests/torn_mrt_records.rs",
 ]
 
 [[bin]]

README.md

@@ -246,6 +246,32 @@ match process_mrt_file("http://example.com/updates.bz2") {
 }
 ```
 
+**Recovering After Damaged MRT Framing**
+
+Recovery is opt-in and never reconstructs a damaged record. It reports the skipped decompressed
+byte range before resuming at a conservatively validated record chain. A correctly framed record
+whose body fails to parse is skipped exactly, and damage extending to the end of the stream
+(e.g. a truncated final record) is reported as a terminal gap rather than an error. Use
+`into_recovering_elem_iter` for the same events at the `BgpElem` level.
+
+```rust
+use bgpkit_parser::{BgpkitParser, RecoveryConfig, RecoveryEvent};
+
+fn recover(path: &str) -> Result<(), Box<dyn std::error::Error>> {
+    let parser = BgpkitParser::new(path)?;
+    for event in parser.into_recovering_record_iter(RecoveryConfig::default()) {
+        match event? {
+            RecoveryEvent::Item(record) => println!("{}", record),
+            RecoveryEvent::Gap(gap) => eprintln!(
+                "skipped bytes {}..{}: {}",
+                gap.start_offset, gap.end_offset, gap.cause
+            ),
+        }
+    }
+    Ok(())
+}
+```
+
 ### Advanced Examples
 
 #### Parsing Real-time Data Streams
@@ -481,6 +507,7 @@ Options:
       --pretty                   Pretty-print JSON output
   -e, --elems-count              Count BGP elems
   -r, --records-count            Count MRT records
+      --recover                  Recover after damaged MRT framing and report skipped byte ranges on stderr
   -o, --origin-asn <ORIGIN_ASN>  Filter by origin AS Number
   -f, --filter <FILTERS>         Generic filter expression (key=value or key!=value)
   -p, --prefix <PREFIX>          Filter by network prefix
@@ -526,6 +553,11 @@ bgpkit-parser --json updates.20211001.0000.bz2 > output.json
 bgpkit-parser -e updates.20211001.0000.bz2
 ```
 
+#### Recover records after damaged framing
+```bash
+bgpkit-parser --recover -e damaged-updates.gz
+```
+
 #### Cache remote files for faster repeated access
 ```bash
 bgpkit-parser -c ~/.bgpkit-cache http://example.com/updates.mrt.bz2

src/bin/main.rs

@@ -4,7 +4,7 @@ use std::io::Write;
 use std::net::IpAddr;
 use std::path::PathBuf;
 
-use bgpkit_parser::{BgpElem, BgpkitParser, Elementor};
+use bgpkit_parser::{BgpElem, BgpkitParser, Elementor, RecoveryConfig, RecoveryEvent, RecoveryGap};
 use clap::{Parser, ValueEnum};
 use ipnet::IpNet;
 
@@ -72,6 +72,10 @@ struct Opts {
     #[clap(short, long)]
     records_count: bool,
 
+    /// Recover after damaged MRT framing and report skipped byte ranges on stderr
+    #[clap(long)]
+    recover: bool,
+
     #[clap(flatten)]
     filters: Filters,
 }
@@ -248,52 +252,202 @@ fn main() {
         opts.format
     };
 
-    match (opts.elems_count, opts.records_count) {
-        (true, true) => {
-            let mut elementor = Elementor::new();
-            let (mut records_count, mut elems_count) = (0, 0);
-            for record in parser.into_record_iter() {
-                records_count += 1;
-                elems_count += elementor.record_to_elems(record).len();
-            }
-            println!("total records: {records_count}");
-            println!("total elems:   {elems_count}");
+    let recovery_config = RecoveryConfig::default();
+    // Element-level runs (element output or counting only elements) use the elem
+    // iterators, which apply filters per element; everything else stays at the record
+    // level. Counting both (-e -r) iterates records and converts once per record.
+    let use_elem_stream = (opts.elems_count && !opts.records_count)
+        || (!opts.elems_count && !opts.records_count && matches!(opts.level, OutputLevel::Elems));
+
+    let result = match (opts.recover, use_elem_stream) {
+        (true, true) => run_elems(
+            parser
+                .into_recovering_elem_iter(recovery_config)
+                .map(|event| event.map_err(|error| error.to_string())),
+            output_format,
+            opts.elems_count,
+            true,
+        ),
+        (false, true) => run_elems(
+            parser
+                .into_elem_iter()
+                .map(|elem| Ok(RecoveryEvent::Item(elem))),
+            output_format,
+            opts.elems_count,
+            false,
+        ),
+        (true, false) => run_records(
+            parser
+                .into_recovering_record_iter(recovery_config)
+                .map(|event| event.map_err(|error| error.to_string())),
+            output_format,
+            opts.elems_count,
+            opts.records_count,
+            true,
+        ),
+        (false, false) => run_records(
+            parser
+                .into_record_iter()
+                .map(|record| Ok(RecoveryEvent::Item(record))),
+            output_format,
+            opts.elems_count,
+            opts.records_count,
+            false,
+        ),
+    };
+    if let Err(error) = result {
+        eprintln!("{error}");
+        std::process::exit(1);
+    }
+}
+
+/// Per-gap reporting and end-of-run summary for `--recover`.
+struct RecoveryStats {
+    enabled: bool,
+    gap_count: usize,
+    skipped_bytes: u64,
+}
+
+impl RecoveryStats {
+    fn new(enabled: bool) -> Self {
+        Self {
+            enabled,
+            gap_count: 0,
+            skipped_bytes: 0,
         }
-        (false, true) => {
-            println!("total records: {}", parser.into_record_iter().count());
+    }
+
+    fn observe(&mut self, gap: &RecoveryGap) {
+        self.gap_count += 1;
+        self.skipped_bytes += gap.skipped_bytes();
+        eprintln!(
+            "recovered MRT framing: skipped bytes {}..{} ({} bytes, {:?}, {} confirming records): {}",
+            gap.start_offset,
+            gap.end_offset,
+            gap.skipped_bytes(),
+            gap.evidence,
+            gap.confirmed_records,
+            gap.cause
+        );
+    }
+
+    fn print_summary(&self) {
+        if self.enabled {
+            eprintln!(
+                "recovery summary: {} gaps, {} bytes skipped",
+                self.gap_count, self.skipped_bytes
+            );
         }
-        (true, false) => {
-            println!("total elems: {}", parser.into_elem_iter().count());
+    }
+}
+
+fn run_elems<I>(
+    events: I,
+    output_format: OutputFormat,
+    count_requested: bool,
+    report_recovery: bool,
+) -> Result<(), String>
+where
+    I: IntoIterator<Item = Result<RecoveryEvent<BgpElem>, String>>,
+{
+    let mut stdout = std::io::stdout();
+    let mut elems_count = 0usize;
+    let mut elem_index = 0usize;
+    let mut stats = RecoveryStats::new(report_recovery);
+    let mut terminal_error = None;
+
+    for event in events {
+        match event {
+            Err(error) => {
+                terminal_error = Some(error);
+                break;
+            }
+            Ok(RecoveryEvent::Gap(gap)) => stats.observe(&gap),
+            Ok(RecoveryEvent::Item(elem)) => {
+                elems_count += 1;
+                if count_requested {
+                    continue;
+                }
+                let output = format_elem(&elem, output_format, elem_index);
+                elem_index += 1;
+                if !write_output(&mut stdout, &output)? {
+                    return Ok(());
+                }
+            }
         }
-        (false, false) => {
-            let mut stdout = std::io::stdout();
-
-            match opts.level {
-                OutputLevel::Elems => {
-                    for (index, elem) in parser.into_elem_iter().enumerate() {
-                        let output_str = format_elem(&elem, output_format, index);
-                        if let Err(e) = writeln!(stdout, "{}", output_str) {
-                            if e.kind() != std::io::ErrorKind::BrokenPipe {
-                                eprintln!("{e}");
-                            }
-                            std::process::exit(1);
-                        }
-                    }
+    }
+
+    if count_requested {
+        println!("total elems: {elems_count}");
+    }
+    stats.print_summary();
+    terminal_error.map_or(Ok(()), Err)
+}
+
+fn run_records<I>(
+    events: I,
+    output_format: OutputFormat,
+    elems_count_requested: bool,
+    records_count_requested: bool,
+    report_recovery: bool,
+) -> Result<(), String>
+where
+    I: IntoIterator<Item = Result<RecoveryEvent<bgpkit_parser::MrtRecord>, String>>,
+{
+    let mut stdout = std::io::stdout();
+    let mut elementor = Elementor::new();
+    let mut records_count = 0usize;
+    let mut elems_count = 0usize;
+    let mut stats = RecoveryStats::new(report_recovery);
+    let mut terminal_error = None;
+
+    for event in events {
+        match event {
+            Err(error) => {
+                terminal_error = Some(error);
+                break;
+            }
+            Ok(RecoveryEvent::Gap(gap)) => stats.observe(&gap),
+            Ok(RecoveryEvent::Item(record)) => {
+                records_count += 1;
+                if elems_count_requested {
+                    // Counting both (-e -r): count every element of records that passed
+                    // record-level filtering, matching the historical CLI behavior.
+                    elems_count += elementor.record_to_elems(record).len();
+                    continue;
                 }
-                OutputLevel::Records => {
-                    for record in parser.into_record_iter() {
-                        let output_str = format_record(&record, output_format);
-                        if let Err(e) = writeln!(stdout, "{}", output_str) {
-                            if e.kind() != std::io::ErrorKind::BrokenPipe {
-                                eprintln!("{e}");
-                            }
-                            std::process::exit(1);
-                        }
-                    }
+                if records_count_requested {
+                    continue;
+                }
+                let output = format_record(&record, output_format);
+                if !write_output(&mut stdout, &output)? {
+                    return Ok(());
                 }
             }
         }
     }
+
+    match (elems_count_requested, records_count_requested) {
+        (true, true) => {
+            println!("total records: {records_count}");
+            println!("total elems:   {elems_count}");
+        }
+        (false, true) => println!("total records: {records_count}"),
+        (true, false) => println!("total elems: {elems_count}"),
+        (false, false) => {}
+    }
+    stats.print_summary();
+    terminal_error.map_or(Ok(()), Err)
+}
+
+fn write_output(stdout: &mut std::io::Stdout, output: &str) -> Result<bool, String> {
+    if let Err(error) = writeln!(stdout, "{output}") {
+        if error.kind() == std::io::ErrorKind::BrokenPipe {
+            return Ok(false);
+        }
+        return Err(error.to_string());
+    }
+    Ok(true)
 }
 
 fn format_elem(elem: &BgpElem, format: OutputFormat, index: usize) -> String {

src/lib.rs

@@ -236,6 +236,29 @@ match process_mrt_file("http://example.com/updates.bz2") {
 }
 ```
 
+**Recovering After Damaged MRT Framing**
+
+Recovery is opt-in and never reconstructs a damaged record. It reports the skipped decompressed
+byte range before resuming at a conservatively validated record chain.
+
+```no_run
+use bgpkit_parser::{BgpkitParser, RecoveryConfig, RecoveryEvent};
+
+fn recover(path: &str) -> Result<(), Box<dyn std::error::Error>> {
+    let parser = BgpkitParser::new(path)?;
+    for event in parser.into_recovering_record_iter(RecoveryConfig::default()) {
+        match event? {
+            RecoveryEvent::Item(record) => println!("{}", record),
+            RecoveryEvent::Gap(gap) => eprintln!(
+                "skipped bytes {}..{}: {}",
+                gap.start_offset, gap.end_offset, gap.cause
+            ),
+        }
+    }
+    Ok(())
+}
+```
+
 ## Advanced Examples
 
 ### Parsing Real-time Data Streams
@@ -477,6 +500,7 @@ Options:
       --pretty                   Pretty-print JSON output
   -e, --elems-count              Count BGP elems
   -r, --records-count            Count MRT records
+      --recover                  Recover after damaged MRT framing and report skipped byte ranges on stderr
   -o, --origin-asn <ORIGIN_ASN>  Filter by origin AS Number
   -f, --filter <FILTERS>         Generic filter expression (key=value or key!=value)
   -p, --prefix <PREFIX>          Filter by network prefix
@@ -522,6 +546,11 @@ bgpkit-parser --json updates.20211001.0000.bz2 > output.json
 bgpkit-parser -e updates.20211001.0000.bz2
 ```
 
+### Recover records after damaged framing
+```bash
+bgpkit-parser --recover -e damaged-updates.gz
+```
+
 ### Cache remote files for faster repeated access
 ```bash
 bgpkit-parser -c ~/.bgpkit-cache http://example.com/updates.mrt.bz2