feat(watch): live RIS Live streaming with filter pushdown and MRT recording

[digizeph]

Adds monocle watch (issue #100, RIS Live portion): streams live BGP messages from RIPE RIS Live, normalized to BgpElem so the parse filter grammar, output formats, and field selection all apply unchanged.

Examples

Watch your own network's propagation seconds after a policy change:

$ monocle watch --origin-asn 400644
A|1788630482.36|2a04:ccc2::63fe|34800|2400:cb00:da93::/48|34800 29632 6762 13335|IGP|...

Capture an incident live, replay it offline with the same filters:

$ monocle watch --host rrc00 --prefix 45.57.60.0/24 -S --record incident.mrt.bz2
^C
$ monocle parse incident.mrt.bz2 --origin-asn 2906

Scope server-side to cut bandwidth (multi-prefix/peer fan out to one subscription per combination):

$ monocle watch --host rrc10 --prefix 1.1.1.0/24,8.8.8.0/24 --json

Pipe into other tools; broken pipes exit cleanly:

$ monocle watch --origin-asn 13335 --json | jq -c 'select(.prefix | contains("2400:cb00"))' | head -20

Behavior

• Server-side pushdown of host, prefix (one subscription per prefix, both specificity flags explicit), peer (one per peer IP), and elem type via require — purely to reduce traffic. Origin ASN, peer ASN, community, and AS-path predicates always run client-side with parser Filter semantics: the RIS path pattern cannot express AS_SET origins, and RIS selects whole UPDATE messages while elements expand per prefix, so every element predicate is re-checked locally. Watch matches exactly what monocle parse would match.
• Any filter is accepted without opt-in (measured full feed: ~5k msgs/s, ~28 MB client RSS). Only a completely bare invocation requires --firehose to drink the full unfiltered stream. Server-side scope (--host/--prefix/--peer-ip) is recommended to cut bandwidth and RIPE-side load.
• Raw-BGP parsing (includeRaw + parse_ris_live_message_raw) preserves all path attributes, so --community large-community filters work and recordings keep raw-only attributes.
• Subscriptions request acknowledgement; success is reported after ris_subscribe_ok; ris_error aborts with the server message.
• --record PATH writes the filtered stream as BGP4MP MRT updates (batched flushes of 500) for offline replay with monocle parse and the same filters. The record writer opens before the async runtime (oneio wraps reqwest::blocking) and only after argument/filter validation; every exit path (success, Ctrl-C, fatal error, broken pipe) finalizes the recording.
• Reconnect with 5s backoff (--no-reconnect to disable); one Ctrl-C handler spans connect, subscribe, read, and backoff windows; non-BrokenPipe output errors abort nonzero.
• --help notes that live vantage is RIS collectors only, not global visibility.

RouteViews BMP/Kafka is intentionally out of scope for this PR; it follows later on the same normalization.

Verified

cargo fmt --check, cargo clippy --all-targets --all-features -- -D warnings, full test suite (298 lib tests incl. new watch-lens tests), plus live runs against rrc00: multi-prefix subscription fan-out confirmed in subscription JSON with acks, --record + monocle parse round trips element-for-element, firehose measured over 60s (311k messages, 708k elements, 28 MB RSS), and markdown/PSV headers pipe to head without panics.

[Copilot]

🟡 Changes recommended

Multiple unresolved critical and moderate issues affect data integrity, filtering correctness, transport security, and command behavior.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Pull request overview

Adds monocle watch for streaming RIPE RIS Live BGP messages with filtering, formatting, reconnection, and MRT recording.

Changes:

• Adds the watch lens and CLI command.
• Adds filter pushdown and client-side filtering.
• Adds MRT recording, WebSocket dependencies, and documentation.

File summaries

File	Review
src/lens/watch.rs	Critical: use WSS/TLS (1 vote); use raw messages for filter/record parity (1); avoid unsafe origin pushdown with AS_SET origins (1); reapply pushed predicates per element (1). Moderate: explicitly preserve exact-prefix semantics (2); propagate malformed-message errors (1).
src/lens/mod.rs	Exports the watch lens.
src/bin/monocle.rs	Registers and dispatches watch.
src/bin/commands/watch.rs	Critical: validate before opening/truncating recordings (2); abort on recording failures (2). Moderate: reject or implement table output (2); validate positive origin filters (1); await subscription acknowledgement/errors (1). Nits: fix both 5ss messages (1 vote each).
src/bin/commands/mod.rs	Exports the watch command.
CHANGELOG.md	Documents the feature.
Cargo.toml	Adds feature configuration and WebSocket dependencies.
Cargo.lock	Locks the new transitive dependencies.

Review details

Suppressed comments (6)

src/bin/commands/watch.rs:171

• Formatting a Duration with {:?} already includes the unit (5s), and the literal adds another s, producing “retrying in 5ss”. Remove the extra suffix.

                    eprintln!(
                        "connection failed ({e}); retrying in {:?}s",
                        RECONNECT_BACKOFF
                    );

src/bin/commands/watch.rs:273

• RECONNECT_BACKOFF formats as 5s with Debug, so this message prints “reconnecting in 5ss”. Do not append a second unit.

        eprintln!("reconnecting in {:?}s", RECONNECT_BACKOFF);

src/lens/watch.rs:141

• RisSubscribe::path stores a single Option<String>, so each iteration replaces the previous origin. For --origin-asn 13335,15169, only 15169$ reaches RIS Live, and positive origins are not reapplied client-side, so matching 13335 elements cannot be emitted. Preserve OR semantics by sending separate subscriptions or by leaving multi-value origins entirely client-side.

            sub = sub.path(&pattern);
            report.origin_path_patterns.push(pattern);

src/lens/watch.rs:137

• Negated origins should not be pushed down here. The implementation itself notes that RIS Live ! path behavior is unreliable, and a client-side filter cannot recover messages that the server has already discarded, so --origin-asn !N does not have guaranteed parser semantics. Skip negated values in the subscription and apply them only through the local filter.

        for value in &self.origin_asn {
            let (asn, negated) = strip_negation(value);
            let pattern = if negated {
                format!("!{asn}$")
            } else {

src/lens/watch.rs:197

• Stripping ! turns an exclusion such as --peer-asn !174 into an inclusion. Creating one filter per value also ANDs repeated positive peers, although a BGP element has one peer and parse semantics treat the list as OR. Build one peer_asns filter from the original comma-joined values.

        for value in &self.peer_asn {
            let v = strip_negation(value);
            filters.push(Filter::new("peer_asns", &v.0)?);
        }

src/lens/watch.rs:151

• The public option help says prefixes may be prefixed with ! to exclude, and the command claims parse-compatible filter semantics, but every negative prefix is rejected here. Keep negative prefixes out of the RIS subscription and apply the canonical prefix predicate client-side instead.

            if negated {
                bail!(
                    "negative prefix filters are not supported for live subscriptions: '{value}'"
                );

• Files reviewed: 7/8 changed files
• Comments generated: 11
• Review effort level: Balanced

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[digizeph]

Stale anchor: in the current head (15084d9) the record writer opens at src/bin/commands/watch.rs:124, after filter validation (compile_client_filters, line 86), field parsing (line 96), and the firehose guard (lines 104-118). An invalid --record existing.mrt invocation exits before MrtRecorder::new runs, so the existing file is never truncated. Verified live: monocle watch --origin-asn bogus --record /tmp/precious.mrt.bz2 exits with the validation error and leaves the file intact.

[digizeph]

Review round 1 addressed in 4258d90 (all 11 inline comments + maintainer feedback on the feature gate):

• Origin ASN pushdown removed: RIS path patterns can't express AS_SET origins, so origin filters client-side only. Host/prefix/peer/require still push down as traffic reduction, and every element predicate also runs client-side, so semantics match monocle parse exactly (RIS selects whole UPDATEs; elements expand per prefix).
• includeRaw subscription + parse_ris_live_message_raw: all path attributes preserved (large-community filters work; recorded MRT no longer drops raw-only attributes).
• wss:// with rustls webpki roots; ring CryptoProvider installed explicitly since the tree enables two providers and auto-detection panics at runtime.
• acknowledge(true); subscription success is only reported after ris_subscribe_ok; ris_error aborts with the server message.
• Filter validation (invalid ASN/prefix, mixed positive/negative) runs before the record file opens, so bad invocations can't truncate an existing recording; record write failures abort instead of being swallowed.
• --format table rejected for the unbounded stream; malformed frames surface as errors instead of silent drops; prefix subscriptions set both specificity flags explicitly (RIS defaults moreSpecific=true).
• The separate watch cargo feature is gone (it was only reachable via cli, so the gate had no consumer); the lens is part of lib.

Gates green locally and CI passes. Live check: ack received, 30s rrc00 origin-13335 run: 235 elements streamed, record + monocle parse round trip 235/235, clean SIGINT.

[Copilot]

🟡 Changes recommended

Critical and moderate issues remain in subscription coverage, filter semantics, signal handling, output behavior, and dependency configuration.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Review details

Suppressed comments (2)

src/bin/commands/watch.rs:213

• Duration's debug representation already includes its unit (5s), so this prints retrying in 5ss. Remove the extra literal s.

                    eprintln!("connection failed ({e}); retrying in {RECONNECT_BACKOFF:?}s");

src/bin/commands/watch.rs:329

• RECONNECT_BACKOFF formats as 5s with :?, making this message read 5ss. Remove the appended s.

        eprintln!("reconnecting in {RECONNECT_BACKOFF:?}s");

• Files reviewed: 7/8 changed files
• Comments generated: 12
• Review effort level: Balanced

[digizeph]

Stale anchor: the loop at src/lens/watch.rs:184 only builds the display report. The subscriptions themselves come from the prefix x peer cross product at src/lens/watch.rs:207-208 — one RisSubscribe per (prefix, peer) combination, so --peer-ip A --peer-ip B produces subscriptions covering both. Covered by test_multi_prefix_multi_peer_expands_subscriptions (asserts the 2x2 fan-out); verified live with --prefix 1.1.1.0/24,8.8.8.0/24 sending two ris_subscribe messages, both acknowledged.

[Copilot]

🟡 Changes recommended

Recorder data loss and multiple stream-control issues must be fixed before approval.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Review details

Suppressed comments (1)

src/lens/watch.rs:370

• The central data-frame branch has no automated test: current tests cover only control frames and host extraction. Add a representative includeRaw RIS message assertion that verifies LiveFrame::Data, host propagation, element expansion, and a raw-only attribute such as a large community, since this behavior is the basis for parse-equivalent filtering and recording.

    match parse_ris_live_message_raw(msg_str) {
        Ok(elems) => Ok(LiveFrame::Data(LiveMessage { host, elems })),
        Err(e) => Err(anyhow!("failed to parse RIS Live message: {e}")),

• Files reviewed: 8/9 changed files
• Comments generated: 4
• Review effort level: Balanced

[Copilot]

🟡 Changes recommended

Multiple moderate issues affect output, acknowledgements, error handling, broken pipes, JSON parsing, and recording guarantees.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Review details

Suppressed comments (6)

Previously missed (2) — in code that hasn't changed since the last review.

src/bin/commands/watch.rs:488

• This encoder reconstructs each MRT UPDATE from BgpElem; it does not retain every attribute from the raw BGP message. In bgpkit-parser 0.20, the conversion omits recognized attributes not represented by BgpElem (for example ORIGINATOR_ID/CLUSTER_LIST, and the encoder conversion also omits ATOMIC_AGGREGATE), so recordings do not satisfy the stated raw-attribute preservation guarantee. Keep the parsed raw attributes/update alongside each element when recording, or narrow the documented guarantee.
  src/lens/watch.rs:384
• This scanner treats the next quote as the end of the value and does not decode JSON escapes. A valid ris_error message containing an escaped quote (or an escaped host value) is therefore truncated; parse the envelope with serde_json or otherwise handle JSON string escaping.

src/bin/commands/watch.rs:80

• --pretty must not bypass this guard. With --format table --pretty, out_format remains Table, format_elem always returns None, and the command connects forever while silently emitting no elements. Reject Table regardless of --pretty (and do not advertise --pretty as a replacement format).

    if !args.pretty && output_format == OutputFormat::Table {
        eprintln!(
            "watch: --format table is not supported for an unbounded stream; \
             use the default PSV, JSON, or --pretty"
        );

src/bin/commands/watch.rs:254

• Duration's debug representation already includes the unit (5s), so the appended s produces 5ss in this retry message.

                        eprintln!("connection failed ({e}); retrying in {RECONNECT_BACKOFF:?}s");

src/bin/commands/watch.rs:417

• Duration's debug representation already includes s, so this prints reconnecting in 5ss. Remove the extra suffix.

        eprintln!("reconnecting in {RECONNECT_BACKOFF:?}s");

src/lens/watch.rs:247

• This duplicates the parse lens's filter construction instead of using the canonical ParseFilters::validate/to_filters path (src/lens/parse/mod.rs:430-478,678-713,834-843). Because watch promises identical semantics, future parser filter changes can silently diverge here; construct a ParseFilters from the shared fields and delegate validation/conversion, which also avoids exposing the two community helpers solely for watch.

    pub fn compile_client_filters(&self) -> Result<Vec<bgpkit_parser::parser::filter::Filter>> {
        use bgpkit_parser::parser::filter::Filter;

        let mut filters = Vec::new();

        if !self.origin_asn.is_empty() {
            filters.push(Filter::new("origin_asns", &self.origin_asn.join(","))?);

• Files reviewed: 8/9 changed files
• Comments generated: 4
• Review effort level: Balanced

[Copilot]

🟡 Changes recommended

Unresolved moderate issues affect output, recording integrity and fidelity, shutdown behavior, and error status.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Review details

Suppressed comments (7)

Previously missed (3) — in code that hasn't changed since the last review.

src/bin/commands/watch.rs:82

• --format table --pretty bypasses this guard, but pretty only upgrades Json later; the format therefore remains Table, and format_elem returns None for every element, producing a silent stream. Reject table output regardless of --pretty and point users to an actual streaming format.
  src/bin/commands/watch.rs:490
• This re-encodes the BgpElem projection rather than preserving the original raw UPDATE. In bgpkit-parser, MrtUpdatesEncoder rebuilds attributes via Attributes::from(elem), while attributes not represented by BgpElem (for example ORIGINATOR_ID, CLUSTER_LIST, AIGP, and Prefix-SID) are lost; therefore --record does not meet the PR's stated guarantee that raw-only path attributes survive recording. Preserve the full/raw attributes through the recording path or narrow that guarantee.
  src/lens/watch.rs:369
• This elem-only parser does not preserve all raw path attributes as claimed: converting a raw UPDATE to BgpElem drops attributes not represented by that type (for example ORIGINATOR_ID and CLUSTER_LIST), and MrtUpdatesEncoder later reconstructs the recording solely from those elems. Consequently --record cannot retain every raw-only attribute. Retain a full/raw UPDATE representation for recording (while deriving elems for filtering), or narrow the documented guarantee if lossy replay is intended.

src/bin/commands/watch.rs:139

• The recorder has already opened/truncated the target when runtime creation is attempted. If Runtime::new() fails, process::exit skips MrtRecorder/compression-writer destruction, leaving the requested recording damaged even though streaming never started. Construct the runtime before opening the recorder (it is still outside block_on) so this failure occurs before touching the output path.

    let rt = match tokio::runtime::Runtime::new() {
        Ok(rt) => rt,
        Err(e) => {
            eprintln!("Failed to create async runtime: {e}");
            std::process::exit(1);
        }

src/bin/commands/watch.rs:261

• Duration's debug representation already includes the unit (5s), so the extra literal s prints 5ss in this retry message.

                        eprintln!("connection failed ({e}); retrying in {RECONNECT_BACKOFF:?}s");

src/bin/commands/watch.rs:424

• Duration's debug representation already includes the unit, so this message reports the backoff as 5ss.

        eprintln!("reconnecting in {RECONNECT_BACKOFF:?}s");

src/bin/commands/watch.rs:327

• With --no-reconnect, a WebSocket read failure only reaches the outer no_reconnect break, leaving fatal unset, so run_async returns Ok(()) and the command exits successfully despite the stream error. Preserve the error as fatal when reconnection is disabled; otherwise automation cannot distinguish a failed stream from a clean stop.

                    Some(Err(e)) => {
                        eprintln!("stream error: {e}");
                        break;

• Files reviewed: 8/9 changed files
• Comments generated: 1
• Review effort level: Balanced

[Copilot]

🟡 Changes recommended

Critical attribute-loss issues and moderate output and error-handling defects remain unresolved.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Review details

Suppressed comments (2)

src/bin/commands/watch.rs:70

• The binary sets SIGPIPE to SIG_DFL in src/bin/monocle.rs:149-155, so on Unix a closed pipeline terminates the process at the write instead of returning BrokenPipe. Consequently the handling below is unreachable and --record can be killed before its compression stream is finalized. Restore ignored SIGPIPE handling for watch before writing output so the existing BrokenPipe branches can finish the recorder.

pub fn run(mut args: WatchArgs, output_format: OutputFormat) {
    // The tree enables two rustls CryptoProviders (aws-lc-rs via oneio, ring
    // via reqwest); pick ring once so TLS setup cannot fail ambiguously.
    install_crypto_provider();

src/bin/commands/watch.rs:328

• A WebSocket read error is only logged. With --no-reconnect, line 422 then exits the session with fatal == None, so run_async returns success and automation cannot distinguish a failed stream from a clean stop. Record this as fatal when reconnection is disabled, consistent with connect/send failures.

                    Some(Err(e)) => {
                        eprintln!("stream error: {e}");
                        break;

• Files reviewed: 8/9 changed files
• Comments generated: 3
• Review effort level: Balanced

[Copilot]

🟡 Changes recommended

Multiple moderate correctness and contract issues remain unresolved.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Review details

Suppressed comments (8)

Previously missed (3) — in code that hasn't changed since the last review.

src/lens/watch.rs:235

• This reimplements the canonical parse filter conversion instead of using ParseFilters::validate and ParseFilters::to_filters. The duplicated key selection, validation, negation, and community conversion can drift from monocle parse, despite this API promising identical semantics; the new visibility changes to parse's private helpers are already a symptom. Convert the overlapping fields into a ParseFilters value and delegate validation/conversion to that API.
  src/lens/watch.rs:375
• This stops at the first quote even when it is escaped, so a valid ris_error message such as bad \"quoted\" subscription is truncated and JSON escapes are never decoded. It also searches textual keys without respecting object boundaries. Parse the envelope as JSON when extracting protocol fields so the reported server message and host are decoded correctly.
  src/bin/commands/watch.rs:263
• Duration's debug representation already includes its unit (5s), so the extra literal s renders this as 5ss.

This issue also appears on line 426 of the same file.

src/bin/commands/watch.rs:79

• --format table --pretty bypasses this rejection, but --pretty only upgrades OutputFormat::Json later. The resulting format remains Table, for which format_elem always returns None, so the command connects and silently emits no rows. Reject table output regardless of --pretty (or explicitly convert this combination to JSON).

    if !args.pretty && output_format == OutputFormat::Table {
        eprintln!(
            "watch: --format table is not supported for an unbounded stream; \
             use the default PSV, JSON, or --pretty"
        );

src/bin/commands/watch.rs:373

• Data arriving before the acknowledgement sets subscribed_ok, so the subsequent real ris_subscribe_ok is silently suppressed. This defeats the stated guarantee that subscription success is reported only after the server acknowledgement; data is not an acknowledgement, especially when several subscriptions were sent.

                    if !subscribed_ok {
                        // Data before an ack still means the subscription works.
                        subscribed_ok = true;
                    }

src/bin/commands/watch.rs:426

• Duration's debug representation already includes its unit (5s), so the extra literal s renders this as 5ss.

        eprintln!("reconnecting in {RECONNECT_BACKOFF:?}s");

src/bin/commands/watch.rs:309

• This reports the subscription as successful immediately after writing the request, before any ris_subscribe_ok is received. Since the command separately requests acknowledgements to distinguish acceptance from rejection, label this as sent rather than subscribed; the acknowledgment branch can remain the success report.

                    eprintln!("subscribed: {sub_msg}");

src/lens/watch.rs:359

• This is the elem-only raw parser, so it does not preserve every path attribute as claimed. bgpkit-parser 0.20 provides parse_ris_live_message_raw_full specifically because conversion to BgpElem drops attributes such as originator ID, cluster list, AIGP, Prefix-SID, and raw-retained attributes; MrtUpdatesEncoder then reconstructs updates only from those reduced elements. Consequently, --record cannot retain all raw-only attributes. Carry the full parsed attributes/raw update into the recording path, or narrow the advertised recording contract.

    let host = extract_host(msg_str);
    match parse_ris_live_message_raw(msg_str) {
        Ok(elems) => Ok(LiveFrame::Data(LiveMessage { host, elems })),
        Err(e) => Err(anyhow!("failed to parse RIS Live message: {e}")),

• Files reviewed: 8/9 changed files
• Comments generated: 2
• Review effort level: Balanced

[Copilot]

🔵 Needs a closer look

Six moderate correctness and behavior issues remain unresolved.

Review details

Suppressed comments (6)

Previously missed (1) — in code that hasn't changed since the last review.

src/bin/commands/watch.rs:142

• The recorder has already opened/truncated the target when runtime creation is attempted, and process::exit skips destructors. If runtime creation fails, the output is therefore left unfinalized despite the stated all-exit-path guarantee. Create the runtime before opening the recorder (creation itself does not enter an async context).

src/bin/commands/watch.rs:75

• --pretty should not bypass the table-format rejection. With --format table --pretty, out_format remains Table, format_elem returns None for every element, and watch runs indefinitely without displaying output. Reject Table regardless of pretty.

    if !args.pretty && output_format == OutputFormat::Table {

src/bin/commands/watch.rs:430

• If Ctrl-C arrives during this reconnect backoff, running becomes false but the session loop starts another iteration and polls the already-completed ctrl_c future again. Completed async futures cannot safely be re-polled and may panic, defeating the promised clean shutdown and recorder finalization. Break the session directly from this branch.

        tokio::select! {
            _ = &mut sig => running = false,
            _ = tokio::time::sleep(RECONNECT_BACKOFF) => {}
        }

src/bin/commands/watch.rs:329

• With --no-reconnect, a WebSocket read error is only printed here; the loop then reaches result with fatal == None, so the command exits successfully. This prevents scripts from detecting an interrupted/failed stream. Preserve the error as fatal when reconnection is disabled.

                    Some(Err(e)) => {
                        eprintln!("stream error: {e}");
                        break;

src/bin/monocle.rs:105

• The visible command help tells users to pass --all, but WatchArgs defines the opt-in as --firehose; following this help produces an unknown-argument error. Update the help text to name the actual flag.

    /// refuses an unfiltered stream unless --all is passed.

src/lens/watch.rs:340

• parse_ris_live_message_raw does not preserve all path attributes: bgpkit-parser 0.20 converts the raw UPDATE to BgpElem, which lacks fields such as ORIGINATOR_ID and CLUSTER_LIST, and MrtUpdatesEncoder reconstructs the recording from that reduced model. Those attributes are therefore lost despite this full-fidelity claim. Carry the raw/full attribute representation through recording, or explicitly narrow the advertised recording fidelity.

/// Data frames are parsed from the raw BGP message bytes (the subscription
/// requests `includeRaw`), preserving all path attributes; the JSON-projected
/// parser drops attributes such as large communities. Element parse failures

• Files reviewed: 8/9 changed files
• Comments generated: 0 new
• Review effort level: Balanced

[Copilot]

🔵 Needs a closer look

SIGPIPE can bypass recording finalization, and recorder boundary/finalization tests are missing.

Review details

Suppressed comments (4)

Previously missed (1) — in code that hasn't changed since the last review.

src/bin/commands/watch.rs:516

• The new recorder's batching boundary and final partial-batch finalization are not covered by automated tests, even though comparable command modules contain unit tests. Add temporary-file round-trip cases around 500 elements (for example 499/500/501) and verify finish() produces replayable MRT output; otherwise regressions here can silently truncate incident recordings.

src/bin/commands/watch.rs:69

• main resets SIGPIPE to SIG_DFL before dispatching commands, so on Unix a closed pipe terminates watch at the kernel write instead of returning BrokenPipe. That bypasses this command's cleanup path (including MRT/compression finalization) and exits with signal status 141. Restore ignored SIGPIPE for watch before any output, or remove the global reset and handle write errors consistently.

pub fn run(mut args: WatchArgs, output_format: OutputFormat) {

src/bin/commands/watch.rs:430

• Duration's debug format already includes its unit, so this prints reconnecting in 5ss. Remove the extra s.

        eprintln!("reconnecting in {RECONNECT_BACKOFF:?}s");

src/bin/commands/watch.rs:266

• Duration's debug format already includes its unit (5s), so the appended s makes this log read retrying in 5ss. Remove the extra suffix.

                        eprintln!("connection failed ({e}); retrying in {RECONNECT_BACKOFF:?}s");

• Files reviewed: 8/9 changed files
• Comments generated: 0 new
• Review effort level: Balanced

[Copilot]

🔵 Needs a closer look

Four moderate issues involving validation, error handling, and unnecessary live traffic remain unresolved.

Review details

Suppressed comments (6)

Previously missed (4) — in code that hasn't changed since the last review.

src/bin/commands/watch.rs:256

• The explicit stdout flush result is discarded. A redirected output can accept the buffered write and fail only during flush(), causing the command to continue or exit successfully despite losing output; non-BrokenPipe flush errors should be propagated, while BrokenPipe should follow the existing clean-finalization path.
  src/bin/commands/watch.rs:264
• The ctrl_c() future returns io::Result<()>, but every place that polls this shared future treats both Ok(()) and Err(_) as a clean Ctrl-C. If signal-handler registration fails, watch therefore finalizes an empty recording and exits successfully instead of reporting that it cannot handle shutdown safely. Preserve the result in the connect/read select! branches and make the backoff helper return a result so registration errors become fatal.
  src/lens/watch.rs:199
• Each subscription leaves the RIS message type unset, so the server also sends OPEN, KEEPALIVE, NOTIFICATION, and peer-state traffic even though watch can only emit UPDATE-derived BgpElems. Set the subscription type to UPDATE to avoid transporting and parsing frames that can never produce output.
  src/lens/watch.rs:234
• This bypasses ParseFilters::validate() and reimplements only its conversion logic, so watch does not actually enforce parse's input semantics. For example, bgpkit-parser's plural ASN parser removes spaces, meaning --origin-asn '13 335' is accepted here as ASN 13335, while monocle parse rejects it during validation. Build the overlapping ParseFilters, call validate(), then use to_filters() so validation and matching cannot drift; the duplicated community helpers can then remain private.

src/bin/commands/watch.rs:91

• The suggested --pretty option does not make table output streamable: it only upgrades --format json, while this invocation still exits here. Recommend an option that actually changes the format, such as --json or --format json-line.

            "watch: --format table is not supported for an unbounded stream; \
             use the default PSV, JSON, or --pretty"

src/lens/watch.rs:363

• The primary Data branch has no test: the existing frame tests cover only subscribe/error/pong classification. Add a representative ris_message fixture with valid data.raw and assert both the extracted host and emitted elements, so changes to the hand-written classifier cannot silently break the main live-data path while all current tests still pass.

    let host = extract_host(msg_str);
    match parse_ris_live_message_raw(msg_str) {
        Ok(elems) => Ok(LiveFrame::Data(LiveMessage { host, elems })),
        Err(e) => Err(anyhow!("failed to parse RIS Live message: {e}")),

• Files reviewed: 8/9 changed files
• Comments generated: 0 new
• Review effort level: Balanced

[Copilot]

🟡 Changes recommended

Compressed recordings are not finalized with propagated encoder errors, risking corrupt output with a successful exit status.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Review details

• Files reviewed: 8/9 changed files
• Comments generated: 1
• Review effort level: Balanced

[Copilot]

🟡 Changes recommended

The feature-gating build break, silent malformed-frame loss, and incomplete recording verification must be addressed before approval.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Review details

Suppressed comments (2)

Previously missed (1) — in code that hasn't changed since the last review.

src/lens/watch.rs:308

• A ris_message without a raw member currently falls through as Other and is silently dropped, even though this function promises to surface element parse failures. Classify by the frame type and invoke parse_ris_live_message_raw for every data frame; bgpkit-parser 0.20 then reports the missing raw payload as an error instead of losing an update.

src/bin/commands/watch.rs:639

• This count-only check can report a semantically changed recording as “verified.” With the pinned bgpkit-parser 0.20.0, MrtUpdatesEncoder converts through Attributes::from(&BgpElem), which does not emit AtomicAggregate; an element with atomic == true therefore replays with false while this check still passes. Preserve all supported element attributes and verify their values (not only the count) before claiming the recording is faithful.

        let replayed = parser.into_iter().count();
        if replayed != self.count as usize {
            return Err(anyhow!(
                "recorded file verification failed: wrote {} elements but replayed {replayed}",
                self.count

• Files reviewed: 8/9 changed files
• Comments generated: 1
• Review effort level: Balanced

[Copilot]

🟡 Changes recommended

Four moderate issues remain in subscription acknowledgement handling, shutdown interruption, JSON parsing, and recording fidelity.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Review details

Suppressed comments (3)

Previously missed (1) — in code that hasn't changed since the last review.

src/lens/watch.rs:332

• This scanner does not respect JSON nesting or object-member ordering. A valid RIS frame with data serialized before the outer type (the data object itself contains "type":"UPDATE") is classified as Other, and escaped quotes in string values are truncated. Parse the envelope once with serde_json and read the top-level type plus data.host/data.message explicitly.

src/bin/commands/watch.rs:413

• A multi-prefix/peer plan requests one acknowledgement per subscription, but this single flag treats the first acknowledgement—and even an unrelated data frame—as confirmation for the entire plan. If only part of the plan is accepted, watch continues while silently missing the unacknowledged filter combinations. Track the expected acknowledgement count and only report the subscription plan ready after all acknowledgements arrive; data should not increment that count.

                    if !subscribed_ok {
                        eprintln!("subscription acknowledged by server");
                        subscribed_ok = true;
                    }
                }
                LiveFrame::Error(err_text) => {
                    fatal = Some(anyhow!("RIS Live rejected the subscription: {err_text}"));
                    break;
                }
                LiveFrame::Other => {}
                LiveFrame::Data(live) => {
                    if !subscribed_ok {
                        // Data before an ack still means the subscription works.
                        subscribed_ok = true;

src/lens/watch.rs:295

• The recording is not BgpElem-faithful: the pinned bgpkit-parser 0.20 encoder does not re-encode the element's atomic flag, as this PR itself notes in MrtRecorder::verify_replayable. Consequently replay can differ from displayed live elements, so this guarantee should be qualified.

/// drops. Attributes outside the `BgpElem` model (e.g. ORIGINATOR_ID,
/// CLUSTER_LIST) are still not represented downstream; recordings are
/// `BgpElem`-faithful (exactly what filtering and output see), not
/// byte-faithful to the original UPDATE. Element parse failures are returned

• Files reviewed: 8/9 changed files
• Comments generated: 1
• Review effort level: Balanced