fix(rislive)!: error on unparsed message bodies; next-hop consistency fixes

.gitignore

@@ -26,6 +26,7 @@ docs/
 !tests/fixtures/ripe/**/*.md
 !tests/fixtures/ripe/**/*.gz
 !tests/fixtures/packetlife/**/*.md
+!tests/fixtures/rislive/**/*.md
 
 # WASM build artifacts
 pkg/

CHANGELOG.md

@@ -7,7 +7,9 @@ All notable changes to this project will be documented in this file.
 ### Breaking changes
 
 * **MSRV raised to Rust 1.88** ([#338](https://github.com/bgpkit/bgpkit-parser/pull/338)): `rust-version` in `Cargo.toml` is now 1.88.0. The `idna_adapter` dev-dependency pin is removed as it is no longer needed.
-* **`rislive::messages::Announcement::next_hop` is now the verbatim RIS Live string** (`String`, was `IpAddr`): the field can carry a comma-joined RFC 2545 global + link-local pair (see Fixed below), so like `prefixes` it now keeps the wire value — parse it with `NextHopAddress`'s new `FromStr`. Related edges: `NextHopAddress`'s `Display` renders a pair comma-joined (round-tripping with `FromStr`), and `BgpModelsError` gains a `NextHopParsingError` variant.
+* **`Announcement::next_hop` is now the verbatim RIS Live string** (`String`, was `IpAddr`): the field can carry a comma-joined RFC 2545 global + link-local pair (see Fixed below), so like `prefixes` it now keeps the wire value — parse it with `NextHopAddress`'s new `FromStr`. Related edges: `NextHopAddress`'s `Display` renders a pair comma-joined (round-tripping with `FromStr`), and `BgpModelsError` gains a `NextHopParsingError` variant.
+* **`BgpModelsError` and `ParserRisliveError` are now `#[non_exhaustive]`**: both enums gain a variant in this cycle (`NextHopParsingError`, `UnparsedMessageBody`), and future additions will no longer break exhaustive matches downstream.
+* **`parse_ris_live_message` now errors on a body it could not deserialize**: a frame that declares a message type this crate decodes but whose body fails to parse returns `ParserRisliveError::UnparsedMessageBody` with the underlying reason, where it previously returned `Ok(vec![])`. The flattened `RisMessage::msg` `Option` hid such failures, which is how the comma-joined next-hop bug (see Fixed below) dropped a frame's routes without a trace. Frames whose message type this crate does not decode still return no elems.
 
 ### Dependencies
 
@@ -22,8 +24,9 @@ All notable changes to this project will be documented in this file.
 
 ### Fixed
 
-* **RIS Live link-local next hops no longer drop UPDATEs**: an RFC 2545 global + link-local pair arrives as one comma-joined string (`"next_hop": "2001:db8::1,fe80::1"`), which failed `IpAddr` deserialisation; because `RisMessage::msg` is a flattened `Option`, the error surfaced as `msg: None` and the frame's announcements and withdrawals were silently discarded — ~20% of UPDATE frames (15,706 of 77,075 in a 20-second full-feed capture on 2026-09-10). `next_hop` is now kept verbatim; `parse_ris_live_message` resolves pairs by scope and returns `ElemIncorrectIp` for unparseable values instead of dropping the frame. Regression fixtures with captured frames live in `tests/fixtures/rislive/`; `tests/rislive_frames.rs` also cross-checks the JSON and raw-bytes parsers against each other.
+* **RIS Live link-local next hops no longer drop UPDATEs**: an RFC 2545 global + link-local pair arrives as one comma-joined string (`"next_hop": "2001:db8::1,fe80::1"`), which failed `IpAddr` deserialisation; because `RisMessage::msg` is a flattened `Option`, the error surfaced as `msg: None` and the frame's announcements and withdrawals were silently discarded — ~20% of UPDATE frames (15,706 of 77,075 in a 20-second full-feed capture on 2026-09-10). `next_hop` is now kept verbatim; `parse_ris_live_message` resolves pairs by scope and returns `ElemIncorrectIp` for unparseable values instead of silently returning no elems. Regression fixtures with captured frames live in `tests/fixtures/rislive/`; `tests/rislive_frames.rs` also cross-checks the JSON and raw-bytes parsers against each other.
 * **Two-address next hops resolve by scope in elem conversion**: the MRT/raw path took the first address of an `Ipv6LinkLocal`/`VpnIpv6LinkLocal` next hop positionally, so a pair with the link-local address first yielded `fe80::`. The MRT/raw path, the RIS Live JSON path, and `Nlri::next_hop_addr()` now share `NextHopAddress::global_addr()`.
+* **BGP-LS next hops resolve RFC 2545 pairs by scope**: `parse_link_state_nlri` took the first address of an `Ipv6LinkLocal`/`VpnIpv6LinkLocal` next hop positionally, so a reversed pair put the link-local address in the BGP-LS NLRI; it now uses `NextHopAddress::global_addr()`, matching the other elem conversion paths.
 * **RIS Live peer-state message type** ([#338](https://github.com/bgpkit/bgpkit-parser/pull/338)): the live stream sends peer-state messages with `"type": "STATE"`, not `"type": "RIS_PEER_STATE"` as documented in the RIPE RIS Live documentation and schema. `RisMessageEnum::RIS_PEER_STATE` now deserializes from both `STATE` and `RIS_PEER_STATE` and serializes as `STATE`, matching the live stream. To be re-validated against RIPE's documentation and schema on/after January 2027.
 
 ## v0.21.0 - 2026-08-21

src/models/err.rs

@@ -3,6 +3,7 @@ use std::error::Error;
 use std::fmt::{Display, Formatter};
 
 #[derive(Debug)]
+#[non_exhaustive]
 pub enum BgpModelsError {
     PrefixParsingError(String),
     NextHopParsingError(String),

src/parser/bgp/attributes/attr_29_linkstate.rs

@@ -111,7 +111,9 @@ pub fn parse_link_state_nlri(
     }
 
     let nlri = if is_reachable {
-        Nlri::new_link_state_reachable(next_hop.map(|nh| nh.addr()), safi, nlri_list)
+        // an MP_REACH next hop can carry an RFC 2545 global + link-local pair; the NLRI keeps the
+        // global half, as the MRT/raw and RIS Live elem paths do
+        Nlri::new_link_state_reachable(next_hop.map(|nh| nh.global_addr()), safi, nlri_list)
     } else {
         Nlri::new_link_state_unreachable(safi, nlri_list)
     };
@@ -523,6 +525,25 @@ mod tests {
         }
     }
 
+    #[test]
+    fn test_link_state_next_hop_resolves_pair_by_scope() {
+        // RFC 2545 pair in reverse wire order: the NLRI must carry the global address, not the
+        // link-local address that arrives first
+        let global: Ipv6Addr = "2001:db8::1".parse().unwrap();
+        let link_local: Ipv6Addr = "fe80::1".parse().unwrap();
+
+        let nlri = parse_link_state_nlri(
+            Bytes::new(),
+            Afi::LinkState,
+            Safi::LinkState,
+            Some(NextHopAddress::Ipv6LinkLocal(link_local, global)),
+            true,
+        )
+        .unwrap();
+
+        assert_eq!(nlri.next_hop_addr(), std::net::IpAddr::V6(global));
+    }
+
     #[test]
     fn test_link_state_attribute_encoding() {
         let mut attr = LinkStateAttribute::new();

src/parser/rislive/error.rs

@@ -3,6 +3,7 @@ use std::error::Error;
 use std::fmt::{Display, Formatter};
 
 #[derive(Debug)]
+#[non_exhaustive]
 pub enum ParserRisliveError {
     IncorrectJson(String),
     IncorrectRawBytes,
@@ -13,6 +14,10 @@ pub enum ParserRisliveError {
     ElemIncorrectAggregator(String),
     ElemIncorrectPrefix(String),
     ElemIncorrectIp(String),
+    /// A frame declared a message type this crate decodes, but its body did not deserialize
+    /// into that type. The flattened `RisMessage::msg` `Option` hides such failures as `None`,
+    /// so without this error the frame's elems would be dropped without a trace.
+    UnparsedMessageBody(String),
 }
 
 impl Display for ParserRisliveError {
@@ -45,6 +50,9 @@ impl Display for ParserRisliveError {
             ParserRisliveError::ElemEndOfRibPrefix => {
                 write!(f, "found 'eor' (End of RIB) prefix")
             }
+            ParserRisliveError::UnparsedMessageBody(msg) => {
+                write!(f, "message body failed to deserialize: {msg}")
+            }
         }
     }
 }
@@ -100,6 +108,13 @@ mod tests {
 
         let err = ParserRisliveError::ElemEndOfRibPrefix;
         assert_eq!(err.to_string(), "found 'eor' (End of RIB) prefix");
+
+        let err =
+            ParserRisliveError::UnparsedMessageBody("UPDATE: missing field `path`".to_string());
+        assert_eq!(
+            err.to_string(),
+            "message body failed to deserialize: UPDATE: missing field `path`"
+        );
     }
     #[test]
     fn test_ris_live_error_debug() {
@@ -129,5 +144,8 @@ mod tests {
 
         let err = ParserRisliveError::ElemEndOfRibPrefix;
         assert_eq!(format!("{err:?}"), "ElemEndOfRibPrefix");
+
+        let err = ParserRisliveError::UnparsedMessageBody("UPDATE: test".to_string());
+        assert_eq!(format!("{err:?}"), "UnparsedMessageBody(\"UPDATE: test\")");
     }
 }

src/parser/rislive/mod.rs

@@ -87,11 +87,54 @@ fn parse_prefix(prefix_str: &str) -> Result<IpNet, ParserRisliveError> {
     Ok(p)
 }
 
+/// Message types this crate decodes, as RIS Live spells them in the body's `type` field.
+///
+/// Frames that declare one of these must produce a body. Keep in sync with [`RisMessageEnum`]:
+/// a missing entry only costs the loud failure for that type, never a wrong result.
+const DECODED_MESSAGE_TYPES: [&str; 6] = [
+    "UPDATE",
+    "KEEPALIVE",
+    "OPEN",
+    "NOTIFICATION",
+    "STATE",
+    "RIS_PEER_STATE",
+];
+
+/// Explain a `RisMessage::msg` that came out `None` although the frame declares a decoded
+/// message type.
+///
+/// The flattened `Option<RisMessageEnum>` reports a body-level deserialisation failure as
+/// `None`, so the caller sees an empty frame and loses every route it carried. Re-deserialising
+/// the body here keeps the underlying reason.
+fn unparsed_body_error(msg_str: &str) -> Option<ParserRisliveError> {
+    #[derive(serde::Deserialize)]
+    struct Envelope {
+        data: Option<serde_json::Value>,
+    }
+
+    let envelope: Envelope = serde_json::from_str(msg_str).ok()?;
+    let data = envelope.data?;
+    let message_type = data.get("type")?.as_str()?.to_string();
+    if !DECODED_MESSAGE_TYPES.contains(&message_type.as_str()) {
+        // a message type this crate does not decode yet: nothing was lost
+        return None;
+    }
+
+    serde_json::from_value::<RisMessageEnum>(data)
+        .err()
+        .map(|e| ParserRisliveError::UnparsedMessageBody(format!("{message_type}: {e}")))
+}
+
 /// Parse one RIS Live message using RIS Live's JSON-projected UPDATE fields.
 ///
 /// This parser is convenient and does not require `socketOptions.includeRaw`, but RIS Live's JSON
 /// schema exposes only a subset of BGP path attributes. Use [`parse_ris_live_message_raw`] when you
 /// need attributes that are only present in the raw BGP message.
+///
+/// A frame that declares a message type this crate decodes but whose body fails to deserialize
+/// returns [`ParserRisliveError::UnparsedMessageBody`] rather than no elems: callers streaming
+/// frames should log and skip it, and can fall back to [`parse_ris_live_message_raw`], which reads
+/// the `raw` bytes instead of the projection.
 pub fn parse_ris_live_message(msg_str: &str) -> Result<Vec<BgpElem>, ParserRisliveError> {
     let msg_string = msg_str.to_string();
 
@@ -108,6 +151,11 @@ pub fn parse_ris_live_message(msg_str: &str) -> Result<Vec<BgpElem>, ParserRisli
             // thus for now will be ignored.
 
             if ris_msg.msg.is_none() {
+                // `msg` is flattened, so a body-level deserialisation failure arrives here as
+                // `None`: indistinguishable from a frame without a body, and silently empty.
+                if let Some(err) = unparsed_body_error(msg_str) {
+                    return Err(err);
+                }
                 return Ok(vec![]);
             }
 
@@ -334,6 +382,58 @@ mod tests {
         }
     }
 
+    #[test]
+    fn test_unparsed_body_is_an_error_not_empty_elems() {
+        // A frame that declares a message type this crate decodes must produce a body: the
+        // flattened `Option` reports a body-level deserialisation failure as `None`, which used
+        // to come back as `Ok(vec![])` with every route in the frame silently dropped.
+        let broken_bodies = [
+            (
+                "UPDATE",
+                r#"{"type":"ris_message","data":{"timestamp":1789019601.670,"peer":"2001:7f8:4::1","peer_asn":"207841","id":"x-1","host":"rrc01.ripe.net","type":"UPDATE","path":[207841,6939],"med":"high","announcements":[{"next_hop":"2001:db8::1","prefixes":["2001:db8::/32"]}]}}"#,
+            ),
+            (
+                "OPEN",
+                r#"{"type":"ris_message","data":{"timestamp":1789019601.670,"peer":"2001:7f8:4::1","peer_asn":"207841","id":"x-2","host":"rrc01.ripe.net","type":"OPEN"}}"#,
+            ),
+            (
+                "NOTIFICATION",
+                r#"{"type":"ris_message","data":{"timestamp":1789019601.670,"peer":"2001:7f8:4::1","peer_asn":"207841","id":"x-3","host":"rrc01.ripe.net","type":"NOTIFICATION","notification":"code 6"}}"#,
+            ),
+            (
+                "STATE",
+                r#"{"type":"ris_message","data":{"timestamp":1789019601.670,"peer":"2001:7f8:4::1","peer_asn":"207841","id":"x-4","host":"rrc01.ripe.net","type":"STATE","state":7}}"#,
+            ),
+        ];
+        for (message_type, frame) in broken_bodies {
+            let err = parse_ris_live_message(frame).unwrap_err();
+            assert!(
+                matches!(&err, ParserRisliveError::UnparsedMessageBody(_)),
+                "expected UnparsedMessageBody for {message_type}, got {err:?}"
+            );
+            assert!(
+                err.to_string().starts_with(&format!(
+                    "message body failed to deserialize: {message_type}: "
+                )),
+                "reason should name the declared type: {err}"
+            );
+        }
+    }
+
+    #[test]
+    fn test_undecoded_frames_still_yield_no_elems() {
+        // nothing is lost for a message type this crate does not decode, or for an empty
+        // UPDATE: those keep returning no elems rather than erroring
+        for frame in [
+            r#"{"type":"ris_message","data":{"timestamp":1789019601.670,"peer":"2001:7f8:4::1","peer_asn":"207841","id":"x-5","host":"rrc01.ripe.net","type":"SOMETHING_NEW","payload":{}}}"#,
+            r#"{"type":"ris_message","data":{"timestamp":1789019601.670,"peer":"2001:7f8:4::1","peer_asn":"207841","id":"x-6","host":"rrc01.ripe.net","type":"UPDATE","path":[],"announcements":[],"withdrawals":[]}}"#,
+            r#"{"type":"ris_error","data":{"message":"client too slow"}}"#,
+        ] {
+            let elems = parse_ris_live_message(frame).unwrap();
+            assert!(elems.is_empty(), "expected no elems for {frame}");
+        }
+    }
+
     #[test]
     fn test_parse_prefix() {
         // parse correct ipv4 prefix

tests/fixtures/rislive/README.md

@@ -0,0 +1,40 @@
+# RIS Live frame fixtures
+
+Captured RIS Live `ris_message` frames, one JSON object per line, used by
+`tests/rislive_frames.rs`. Tests must use these local copies and must not
+download data.
+
+The fixture guards a failure mode that is easy to reintroduce: `RisMessage::msg`
+is a flattened `Option`, so a body-level deserialisation failure looks exactly
+like a frame without a body, and the frame's routes disappear without an error.
+These frames pin the forms the live stream actually sends, in particular
+RFC 2545 next hops that RIS Live comma-joins into one string
+(`"next_hop": "2001:7f8:4::3:2be1:1,fe80::3efd:feff:feee:62ca"`).
+
+| File | Original source | Size | SHA-256 |
+| --- | --- | ---: | --- |
+| `ris-live-frames.jsonl` | RIS Live full stream, <https://ris-live.ripe.net/v1/stream/?format=json>, captured 2026-09-10 | 3,801 bytes | `a2930691c91a81ebfea6854c0b8c469e0ef2c6e6a5dfa5954b1f1ac4f0dfe5ca` |
+
+## What the tests require of the file
+
+- one frame per message type this crate decodes: UPDATE, KEEPALIVE, OPEN,
+  NOTIFICATION, STATE
+- at least one UPDATE whose announcement next hop is a comma-joined pair
+- `raw` present on the UPDATE frames, so the JSON and raw-bytes parsers can be
+  compared against each other
+
+Tests select frames by content, never by line position, so the file can be
+extended freely.
+
+## Regenerating
+
+Capture a sample of the full stream:
+
+```sh
+timeout 20 curl -s "https://ris-live.ripe.net/v1/stream/?format=json&client=<client-id>" -o capture.jsonl
+```
+
+`timeout` kills curl mid-frame, so the capture ends with a partial line: drop it
+to keep the file valid NDJSON. Keep one frame per form a test asserts on rather
+than the whole capture, and update the size and SHA-256 above when the file
+changes.