Add rib reconstruction command

CHANGELOG.md

@@ -2,6 +2,21 @@
 
 All notable changes to this project will be documented in this file.
 
+## Unreleased changes
+
+### New Features
+
+* Added `monocle rib` for reconstructing RIB state at arbitrary timestamps
+  * Selects the latest RIB before each requested `rib_ts` and replays updates to the exact timestamp
+  * Supports stdout output by default and SQLite output via `--sqlite-path`
+  * Repeated `--ts` values require `--sqlite-path` and are written to one merged SQLite file keyed by `rib_ts`
+  * Aborts when no RIB exists at or before a requested `rib_ts` for a selected collector
+  * Supports `--country`, `--origin-asn`, `--prefix`, `--as-path`, `--peer-asn`, `--collector`, `--project`, and `--full-feed-only`
+
+### Code Improvements
+
+* Added session-backed SQLite stores for reconstructed RIB working state and merged SQLite export
+
 ## v1.2.0 - 2026-02-28
 
 ### New Features

Cargo.toml

@@ -88,6 +88,7 @@ lib = [
     # Database
     "dep:oneio",
     "dep:ipnet",
+    "dep:tempfile",
     # Lenses
     "dep:chrono-humanize",
     "dep:dateparser",
@@ -98,6 +99,7 @@ lib = [
     "dep:itertools",
     "dep:radar-rs",
     "dep:rayon",
+    "dep:regex",
     # Display (always included with lib)
     "dep:tabled",
     "dep:json_to_table",
@@ -151,6 +153,7 @@ tracing = "0.1"
 # Database
 ipnet = { version = "2.10", features = ["json"], optional = true }
 oneio = { version = "0.20.1", default-features = false, features = ["https", "gz", "bz", "json"], optional = true }
+tempfile = { version = "3", optional = true }
 
 # Lenses
 chrono-humanize = { version = "0.2", optional = true }
@@ -162,6 +165,7 @@ bgpkit-commons = { version = "0.10.2", features = ["asinfo", "rpki", "countries"
 itertools = { version = "0.14", optional = true }
 radar-rs = { version = "0.1.0", optional = true }
 rayon = { version = "1.8", optional = true }
+regex = { version = "1.11", optional = true }
 
 # Display
 tabled = { version = "0.20", optional = true }

README.md

@@ -21,6 +21,7 @@ See through all Border Gateway Protocol (BGP) data with a monocle.
   - [`monocle parse`](#monocle-parse)
     - [Output Format](#output-format)
   - [`monocle search`](#monocle-search)
+  - [`monocle rib`](#monocle-rib)
   - [`monocle time`](#monocle-time)
   - [`monocle inspect`](#monocle-inspect)
   - [`monocle country`](#monocle-country)
@@ -229,6 +230,7 @@ Subcommands:
 
 - `parse`: parse individual MRT files
 - `search`: search for matching messages from all available public MRT files
+- `rib`: reconstruct final RIB state at one or more arbitrary timestamps
 - `server`: start a WebSocket server for programmatic access
 - `inspect`: unified AS and prefix information lookup
 - `country`: utility to look up country name and code
@@ -259,6 +261,7 @@ Usage: monocle [OPTIONS] <COMMAND>
 Commands:
   parse    Parse individual MRT files given a file path, local or remote
   search   Search BGP messages from all available public MRT files
+  rib      Reconstruct final RIB state at one or more arbitrary timestamps
   server   Start the WebSocket server (ws://<address>:<port>/ws, health: http://<address>:<port>/health)
   inspect  Unified AS and prefix information lookup
   country  Country name and code lookup utilities
@@ -701,6 +704,64 @@ Use `--broker-files` to see the list of MRT files that would be queried without
     -c rrc00 --broker-files
 ```
 
+### `monocle rib`
+
+Reconstruct final RIB state at one or more arbitrary timestamps by loading the latest RIB at or before each requested `rib_ts` and replaying updates up to the exact timestamp.
+
+```text
+➜  monocle rib --help
+Reconstruct final RIB state at one or more arbitrary timestamps
+
+Usage: monocle rib [OPTIONS] --ts <RIB_TS>
+
+Options:
+      --ts <RIB_TS>                Target RIB timestamp. Repeat to request multiple snapshots
+      --debug                      Print debug information
+  -o, --origin-asn <ORIGIN_ASN>    Filter by origin AS Number(s), comma-separated. Prefix with ! to exclude
+  -C, --country <COUNTRY>          Filter by origin ASN registration country
+      --format <FORMAT>            Output format: table, markdown, json, json-pretty, json-line, psv (default varies by command)
+      --json                       Output as JSON objects (shortcut for --format json-pretty)
+  -p, --prefix <PREFIX>            Filter by network prefix(es), comma-separated. Prefix with ! to exclude
+      --no-update                  Disable automatic database updates (use existing cached data only)
+  -s, --include-super              Include super-prefixes when filtering
+  -S, --include-sub                Include sub-prefixes when filtering
+  -J, --peer-asn <PEER_ASN>        Filter by peer ASN(s), comma-separated. Prefix with ! to exclude
+  -a, --as-path <AS_PATH>          Filter by AS path regex string
+  -c, --collector <COLLECTOR>      Filter by collector, e.g., rrc00 or route-views2
+  -P, --project <PROJECT>          Filter by route collection project, i.e. riperis or routeviews
+      --full-feed-only             Keep only full-feed peers based on broker peer metadata
+      --sqlite-path <SQLITE_PATH>  SQLite output file path
+  -h, --help                       Print help
+  -V, --version                    Print version
+```
+
+Behavior:
+
+- A single `--ts` writes to stdout by default.
+- Repeated `--ts` values require `--sqlite-path` and are written to one merged SQLite file keyed by `rib_ts`.
+- Providing `--sqlite-path` writes the reconstructed results to that SQLite file instead of stdout.
+- If any selected collector has no RIB at or before a requested `rib_ts`, the command aborts instead of producing a partial result.
+- `--country` uses local ASInfo registration data, and `--full-feed-only` keeps only peers with at least 800k IPv4 prefixes or 100k IPv6 prefixes in broker peer metadata.
+
+Examples:
+
+```bash
+# Print the reconstructed RIB for a single timestamp to stdout
+monocle rib --ts 2025-09-01T12:00:00Z -c rrc00 -o 13335
+
+# Write multiple timestamps to one merged SQLite file in the current directory
+monocle rib \
+  --ts 2025-09-01T12:00:00Z \
+  --ts 2025-09-01T18:00:00Z \
+  --sqlite-path /tmp/rrc00-us.sqlite3 \
+  -c rrc00 \
+  --country US \
+  --full-feed-only
+
+# Write a single reconstructed snapshot to SQLite
+monocle rib --ts 2025-09-01T12:00:00Z --sqlite-path /tmp/route-views2.sqlite3 -c route-views2
+```
+
 ### `monocle time`
 
 Parse and convert time strings between various formats.

src/bin/commands/elem_format.rs

@@ -17,7 +17,9 @@ pub const AVAILABLE_FIELDS: &[&str] = &[
     "peer_ip",
     "peer_asn",
     "prefix",
+    "path_id",
     "as_path",
+    "origin_asns",
     "origin",
     "next_hop",
     "local_pref",
@@ -174,11 +176,26 @@ pub fn get_field_value_with_time_format(
         "peer_ip" => elem.peer_ip.to_string(),
         "peer_asn" => elem.peer_asn.to_string(),
         "prefix" => elem.prefix.to_string(),
+        "path_id" => elem
+            .prefix
+            .path_id
+            .map(|path_id| path_id.to_string())
+            .unwrap_or_default(),
         "as_path" => elem
             .as_path
             .as_ref()
             .map(|p| p.to_string())
             .unwrap_or_default(),
+        "origin_asns" => elem
+            .origin_asns
+            .as_ref()
+            .map(|asns| {
+                asns.iter()
+                    .map(|asn| asn.to_string())
+                    .collect::<Vec<_>>()
+                    .join(" ")
+            })
+            .unwrap_or_default(),
         "origin" => elem
             .origin
             .as_ref()
@@ -288,10 +305,18 @@ pub fn build_json_object(
             "peer_ip" => json!(elem.peer_ip.to_string()),
             "peer_asn" => json!(elem.peer_asn),
             "prefix" => json!(elem.prefix.to_string()),
+            "path_id" => match elem.prefix.path_id {
+                Some(path_id) => json!(path_id),
+                None => serde_json::Value::Null,
+            },
             "as_path" => match &elem.as_path {
                 Some(p) => json!(p.to_string()),
                 None => serde_json::Value::Null,
             },
+            "origin_asns" => match &elem.origin_asns {
+                Some(asns) => json!(asns.iter().map(|asn| asn.to_string()).collect::<Vec<_>>()),
+                None => serde_json::Value::Null,
+            },
             "origin" => match &elem.origin {
                 Some(o) => json!(o.to_string()),
                 None => serde_json::Value::Null,

src/bin/commands/mod.rs

@@ -6,6 +6,7 @@ pub mod inspect;
 pub mod ip;
 pub mod parse;
 pub mod pfx2as;
+pub mod rib;
 pub mod rpki;
 pub mod search;
 pub mod time;

src/bin/commands/rib.rs

@@ -0,0 +1,155 @@
+use std::fs;
+use std::io::Write;
+use std::path::Path;
+
+use anyhow::{anyhow, Result};
+use bgpkit_parser::BgpElem;
+
+use monocle::database::{MonocleDatabase, RibSqliteStore};
+use monocle::lens::rib::RibLens;
+use monocle::utils::{OutputFormat, TimestampFormat};
+use monocle::MonocleConfig;
+
+use super::elem_format::{format_elem, format_elems_table, get_header};
+
+pub use monocle::lens::rib::RibArgs;
+
+const DEFAULT_FIELDS_RIB: &[&str] = &[
+    "type",
+    "timestamp",
+    "peer_ip",
+    "peer_asn",
+    "prefix",
+    "path_id",
+    "as_path",
+    "origin_asns",
+    "origin",
+    "next_hop",
+    "local_pref",
+    "med",
+    "communities",
+    "atomic",
+    "aggr_asn",
+    "aggr_ip",
+    "collector",
+];
+
+pub fn run(config: &MonocleConfig, args: RibArgs, output_format: OutputFormat, no_update: bool) {
+    if let Err(error) = run_inner(config, args, output_format, no_update) {
+        eprintln!("ERROR: {}", error);
+        std::process::exit(1);
+    }
+}
+
+fn run_inner(
+    config: &MonocleConfig,
+    args: RibArgs,
+    output_format: OutputFormat,
+    no_update: bool,
+) -> Result<()> {
+    let sqlite_path = config.sqlite_path();
+    let db = MonocleDatabase::open(&sqlite_path)
+        .map_err(|e| anyhow!("Failed to open database '{}': {}", sqlite_path, e))?;
+    let lens = RibLens::new(&db, config);
+
+    if args.sqlite_path.is_some() {
+        run_sqlite_output(&lens, &args, no_update)
+    } else {
+        run_stdout(&lens, &args, output_format, no_update)
+    }
+}
+
+fn run_stdout(
+    lens: &RibLens<'_>,
+    args: &RibArgs,
+    output_format: OutputFormat,
+    no_update: bool,
+) -> Result<()> {
+    let mut stdout = std::io::stdout();
+
+    if output_format == OutputFormat::Table {
+        let mut elems = Vec::<(BgpElem, Option<String>)>::new();
+        lens.reconstruct_snapshots(args, no_update, |_rib_ts, state_store| {
+            state_store.visit_entries(|entry| {
+                elems.push((entry.elem, Some(entry.collector)));
+                Ok(())
+            })
+        })?;
+
+        if !elems.is_empty() {
+            writeln!(
+                stdout,
+                "{}",
+                format_elems_table(&elems, DEFAULT_FIELDS_RIB, TimestampFormat::Unix)
+            )
+            .map_err(|e| anyhow!("Failed to write table output: {}", e))?;
+        }
+        return Ok(());
+    }
+
+    let mut header_written = false;
+    lens.reconstruct_snapshots(args, no_update, |_rib_ts, state_store| {
+        if !header_written {
+            if let Some(header) = get_header(output_format, DEFAULT_FIELDS_RIB) {
+                writeln!(stdout, "{}", header)
+                    .map_err(|e| anyhow!("Failed to write output header: {}", e))?;
+            }
+            header_written = true;
+        }
+
+        state_store.visit_entries(|entry| {
+            if let Some(line) = format_elem(
+                &entry.elem,
+                output_format,
+                DEFAULT_FIELDS_RIB,
+                Some(entry.collector.as_str()),
+                TimestampFormat::Unix,
+            ) {
+                writeln!(stdout, "{}", line)
+                    .map_err(|e| anyhow!("Failed to write reconstructed RIB row: {}", e))?;
+            }
+            Ok(())
+        })
+    })?;
+
+    Ok(())
+}
+
+fn run_sqlite_output(lens: &RibLens<'_>, args: &RibArgs, no_update: bool) -> Result<()> {
+    args.validate()?;
+    let output_path = args
+        .sqlite_path
+        .as_deref()
+        .ok_or_else(|| anyhow!("Missing --sqlite-path for SQLite output"))?;
+
+    remove_existing_file(output_path)?;
+
+    let sqlite_store = RibSqliteStore::new(path_to_str(output_path)?, true)?;
+    let summary = lens.reconstruct_snapshots(args, no_update, |rib_ts, state_store| {
+        state_store.visit_entries(|entry| sqlite_store.insert_entry(rib_ts, &entry))
+    })?;
+
+    eprintln!(
+        "wrote {} reconstructed RIB snapshot(s) to {}",
+        summary.rib_ts.len(),
+        output_path.display()
+    );
+    Ok(())
+}
+
+fn remove_existing_file(path: &Path) -> Result<()> {
+    match fs::remove_file(path) {
+        Ok(()) => Ok(()),
+        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
+        Err(error) => Err(anyhow!(
+            "Failed to remove existing output file '{}': {}",
+            path.display(),
+            error
+        )),
+    }
+}
+
+fn path_to_str(path: &Path) -> Result<&str> {
+    path.to_str()
+        .ok_or_else(|| anyhow!("Path '{}' contains invalid UTF-8", path.display()))
+}

src/bin/monocle.rs

@@ -17,6 +17,7 @@ use commands::inspect::InspectArgs;
 use commands::ip::IpArgs;
 use commands::parse::ParseArgs;
 use commands::pfx2as::Pfx2asArgs;
+use commands::rib::RibArgs;
 use commands::rpki::RpkiCommands;
 use commands::search::SearchArgs;
 use commands::time::TimeArgs;
@@ -57,6 +58,9 @@ enum Commands {
     /// Search BGP messages from all available public MRT files.
     Search(SearchArgs),
 
+    /// Reconstruct final RIB state at one or more arbitrary timestamps.
+    Rib(RibArgs),
+
     /// Start the WebSocket server (ws://<address>:<port>/ws, health: http://<address>:<port>/health)
     ///
     /// Note: This requires building with the `server` feature enabled.
@@ -176,6 +180,9 @@ fn main() {
     match cli.command {
         Commands::Parse(args) => commands::parse::run(args, streaming_output_format),
         Commands::Search(args) => commands::search::run(&config, args, streaming_output_format),
+        Commands::Rib(args) => {
+            commands::rib::run(&config, args, streaming_output_format, cli.no_update)
+        }
 
         Commands::Server(args) => {
             // The server requires the `server` feature (axum + tokio). Keep the CLI