Merge pull request #8 from paritytech/fix/jip3-decoder-alignment

Fix/jip3 decoder alignment

Author: lovelaced

src/decoder.rs

@@ -589,12 +589,13 @@ impl Decode for Vec<ValidatorIndex> {
     }
 }
 
-impl Decode for Event {
-    fn decode(buf: &mut Cursor<&[u8]>) -> Result<Self, DecodingError> {
+impl Event {
+    /// Decode an event from the wire format, using `core_count` (C) from
+    /// the node's ProtocolParameters for fixed-size [u8; C] arrays (JIP-3).
+    pub fn decode_event(buf: &mut Cursor<&[u8]>, core_count: u16) -> Result<Self, DecodingError> {
         let timestamp = Timestamp::decode(buf)?;
         let discriminator = u8::decode(buf)?;
 
-        // Debug logging
         if ![10, 11, 12, 13].contains(&discriminator) {
             tracing::trace!(
                 "Decoding event with discriminator: {}, remaining bytes: {}",
@@ -614,14 +615,12 @@ impl Decode for Event {
                 let num_val_peers = u32::decode(buf)?;
                 let num_sync_peers = u32::decode(buf)?;
 
-                // TODO: JIP-3 specifies num_guarantees as [u8; C] (fixed-size array where
-                // C = core_count), but PolkaJam sends a varint length prefix before the
-                // byte array. We decode it as variable-length to match actual wire format.
-                let core_count = decode_variable_length(buf)? as usize;
-                let mut num_guarantees = vec![0u8; core_count];
-                if buf.remaining() < core_count {
+                // JIP-3: num_guarantees is [u8; C] — fixed-size array, no length prefix
+                let c = core_count as usize;
+                let mut num_guarantees = vec![0u8; c];
+                if buf.remaining() < c {
                     return Err(DecodingError::InsufficientData {
-                        needed: core_count,
+                        needed: c,
                         available: buf.remaining(),
                     });
                 }
@@ -982,10 +981,24 @@ impl Decode for Event {
             }),
 
             // Assurance distribution (126-131)
-            126 => Ok(Event::DistributingAssurance {
-                timestamp,
-                statement: AvailabilityStatement::decode(buf)?,
-            }),
+            126 => {
+                // JIP-3: AvailabilityStatement is anchor (32 bytes) +
+                // bitfield [u8; ceil(C/8)] — fixed-size, no length prefix
+                let anchor = HeaderHash::decode(buf)?;
+                let bitfield_len = (core_count as usize).div_ceil(8);
+                if buf.remaining() < bitfield_len {
+                    return Err(DecodingError::InsufficientData {
+                        needed: bitfield_len,
+                        available: buf.remaining(),
+                    });
+                }
+                let mut bitfield = vec![0u8; bitfield_len];
+                buf.copy_to_slice(&mut bitfield);
+                Ok(Event::DistributingAssurance {
+                    timestamp,
+                    statement: AvailabilityStatement { anchor, bitfield },
+                })
+            }
             127 => Ok(Event::AssuranceSendFailed {
                 timestamp,
                 distributing_id: EventId::decode(buf)?,
@@ -1245,6 +1258,15 @@ impl Decode for Event {
     }
 }
 
+impl Decode for Event {
+    fn decode(buf: &mut Cursor<&[u8]>) -> Result<Self, DecodingError> {
+        // Fallback: decode with core_count=0 (Status/DistributingAssurance will
+        // produce empty arrays). Use decode_event() directly with the real
+        // core_count for production decoding.
+        Event::decode_event(buf, 0)
+    }
+}
+
 pub fn decode_message_frame(data: &[u8]) -> Result<(u32, &[u8]), DecodingError> {
     if data.len() < 4 {
         return Err(DecodingError::InsufficientData {

src/events.rs

@@ -1203,7 +1203,8 @@ impl Encode for Event {
                 num_peers.encode(buf)?;
                 num_val_peers.encode(buf)?;
                 num_sync_peers.encode(buf)?;
-                num_guarantees.encode(buf)?;
+                // JIP-3: num_guarantees is [u8; C] — raw bytes, no length prefix
+                buf.extend_from_slice(num_guarantees);
                 num_shards.encode(buf)?;
                 shards_size.encode(buf)?;
                 num_preimages.encode(buf)?;
@@ -1487,7 +1488,8 @@ impl Encode for Event {
         let specific_size = match self {
             Event::Dropped { .. } => 8 + 8, // last_timestamp + num (u64)
             Event::Status { num_guarantees, .. } => {
-                4 + 4 + 4 + num_guarantees.encoded_size() + 4 + 8 + 4 + 4
+                // JIP-3: num_guarantees is [u8; C] — raw bytes, no length prefix
+                4 + 4 + 4 + num_guarantees.len() + 4 + 8 + 4 + 4
             }
             Event::BestBlockChanged { .. } => 4 + 32, // slot + hash
             Event::FinalizedBlockChanged { .. } => 4 + 32,

src/server.rs

@@ -375,6 +375,9 @@ async fn handle_connection_optimized(
         }
     };
 
+    // Extract core_count from ProtocolParameters for JIP-3 fixed-size arrays
+    let core_count = node_info.params.core_count;
+
     // Generate node ID from peer ID
     let node_id_str = hex::encode(node_info.details.peer_id);
 
@@ -431,7 +434,7 @@ async fn handle_connection_optimized(
                     }
 
                     let mut cursor = Cursor::new(msg_data);
-                    match Event::decode(&mut cursor) {
+                    match Event::decode_event(&mut cursor, core_count) {
                         Ok(event) => {
                             event_count += 1;
 
@@ -500,7 +503,25 @@ async fn handle_connection_optimized(
                             buffer.advance(4 + size as usize);
                         }
                         Err(e) => {
-                            warn!("Failed to decode event from {}: {}", node_id_str, e);
+                            let event_type_hint = if msg_data.len() > 8 {
+                                format!(" (event_type={})", msg_data[8])
+                            } else {
+                                String::new()
+                            };
+                            let hex_preview: String = msg_data
+                                .iter()
+                                .take(32)
+                                .map(|b| format!("{:02x}", b))
+                                .collect::<Vec<_>>()
+                                .join(" ");
+                            warn!(
+                                "Failed to decode event from {}{}: {} [msg_len={}, hex={}]",
+                                node_id_str,
+                                event_type_hint,
+                                e,
+                                msg_data.len(),
+                                hex_preview
+                            );
                             buffer.advance(4 + size as usize);
                         }
                     }

tests/api_tests.rs

@@ -227,7 +227,7 @@ async fn test_events_endpoint_with_data() {
             num_val_peers: 5,
             num_peers: 10,
             num_sync_peers: 8,
-            num_guarantees: vec![1, 2, 3, 4],
+            num_guarantees: vec![0u8; common::TEST_CORE_COUNT],
             num_shards: 50,
             shards_size: 1024 * 1024,
             num_preimages: 3,
@@ -1184,7 +1184,7 @@ async fn test_core_guarantees_with_status_data() {
         num_val_peers: 10,
         num_peers: 20,
         num_sync_peers: 15,
-        num_guarantees: vec![3, 1, 4, 1, 5, 9, 2, 6],
+        num_guarantees: vec![0u8; common::TEST_CORE_COUNT],
         num_shards: 100,
         shards_size: 1024 * 1024,
         num_preimages: 5,
@@ -1217,7 +1217,7 @@ async fn test_da_stats_with_status_data() {
         num_val_peers: 10,
         num_peers: 20,
         num_sync_peers: 15,
-        num_guarantees: vec![1, 2, 3],
+        num_guarantees: vec![0u8; common::TEST_CORE_COUNT],
         num_shards: 150,
         shards_size: 2 * 1024 * 1024,
         num_preimages: 8,

tests/common/mod.rs

@@ -5,6 +5,11 @@ use tart_backend::types::*;
 use tart_backend::TelemetryServer;
 use tokio::time::sleep;
 
+/// Core count used in test_protocol_params(). Status events must have
+/// num_guarantees with exactly this many elements to match the decoder.
+#[allow(dead_code)]
+pub const TEST_CORE_COUNT: usize = 16;
+
 /// Returns a "now" timestamp in JCE-relative microseconds.
 /// Use this in test events so they fall within time-bounded query windows.
 #[allow(dead_code)]

tests/error_tests.rs

@@ -382,7 +382,7 @@ fn test_event_edge_cases() {
     event.encode(&mut buf).unwrap();
 
     let mut cursor = Cursor::new(&buf[..]);
-    let decoded = Event::decode(&mut cursor).unwrap();
+    let decoded = Event::decode_event(&mut cursor, 1000).unwrap();
 
     match decoded {
         Event::Status { num_guarantees, .. } => {

tests/events_tests.rs

@@ -90,8 +90,9 @@ fn test_status_event_encoding_decoding() {
     // Check discriminator
     assert_eq!(buf[8], 10);
 
+    let core_count = guarantees.len() as u16;
     let mut cursor = Cursor::new(&buf[..]);
-    let decoded = Event::decode(&mut cursor).unwrap();
+    let decoded = Event::decode_event(&mut cursor, core_count).unwrap();
 
     match decoded {
         Event::Status {
@@ -717,9 +718,9 @@ fn test_status_event_with_message_frame() {
     let (size, msg_data) = decode_message_frame(&encoded).unwrap();
     println!("Frame size: {}, data len: {}", size, msg_data.len());
 
-    // Decode the event
+    // Decode the event (core_count = 4 to match num_guarantees length)
     let mut cursor = Cursor::new(msg_data);
-    let decoded = Event::decode(&mut cursor).unwrap();
+    let decoded = Event::decode_event(&mut cursor, 4).unwrap();
 
     match decoded {
         Event::Status {

tests/integration_tests.rs

@@ -85,7 +85,7 @@ async fn test_send_multiple_events() {
             num_val_peers: 5,
             num_peers: 10,
             num_sync_peers: 8,
-            num_guarantees: vec![0, 1, 2, 3],
+            num_guarantees: vec![0u8; common::TEST_CORE_COUNT],
             num_shards: 50,
             shards_size: 1024 * 1024,
             num_preimages: 3,
@@ -146,7 +146,7 @@ async fn test_multiple_concurrent_connections() {
                 num_val_peers: i as u32,
                 num_peers: j,
                 num_sync_peers: 0,
-                num_guarantees: vec![],
+                num_guarantees: vec![0u8; common::TEST_CORE_COUNT],
                 num_shards: 0,
                 shards_size: 0,
                 num_preimages: 0,

tests/optimized_server_tests.rs

@@ -72,7 +72,7 @@ async fn test_optimized_server_handles_multiple_connections() {
                     num_val_peers: 10,
                     num_peers: 5,
                     num_sync_peers: 15,
-                    num_guarantees: vec![1, 2, 3],
+                    num_guarantees: vec![0u8; common::TEST_CORE_COUNT],
                     num_shards: 3,
                     shards_size: 1024,
                     num_preimages: 0,