修复 FEC 变长批次解码必失败且连带丢包

- 编码端把所有数据 shard 填充到 max_len 再做 RS 编码，解码端此前按
  实际长度存放数据 shard，批次内包长不一致时 reconstruct 必报
  IncorrectShardSize。现在记录 shard_size（校验包长度），
  无论校验包先到还是后到都把数据 shard 补齐到统一长度
- 解码失败不再向上传播错误：坏 group 直接移除，
  当前完好包与后续数据包正常透传
- 补充变长批次恢复、校验包先到恢复、解码失败不吞好包三个测试

vnt-core/src/fec/decoder.rs

@@ -26,6 +26,8 @@ struct FecDecoderInner {
 struct FecGroup {
     data_shards: usize,
     parity_shards: usize,
+    /// shard 统一长度（校验包到达后才能确定，等于编码端填充后的 max_len）
+    shard_size: usize,
     received_original_count: usize,
     received_shards: Vec<Option<Vec<u8>>>,
     last_update: Instant,
@@ -44,6 +46,7 @@ impl Default for FecGroup {
         Self {
             data_shards: 0,
             parity_shards: 0,
+            shard_size: 0,
             received_original_count: 0,
             received_shards: Vec::with_capacity(16),
             last_update: Instant::now(),
@@ -120,12 +123,21 @@ impl FecDecoder {
             }
             group.data_shards = data_shards;
             group.parity_shards = parity_shards;
+            // 校验包在线上即编码端填充后的 shard 统一长度
+            group.shard_size = payload.len();
             if group.received_shards.len() < data_shards + parity_shards {
                 group
                     .received_shards
                     .resize(data_shards + parity_shards, None);
             }
 
+            // RS 要求所有 shard 等长：把先到的数据 shard 补齐到 shard_size
+            // （编码端发送前把所有数据 shard 填充到 max_len）
+            for shard in group.received_shards[..data_shards].iter_mut().flatten() {
+                if shard.len() < group.shard_size {
+                    shard.resize(group.shard_size, 0);
+                }
+            }
             group.received_shards[packet_index] = Some(payload);
         } else {
             let buffer = TransmissionBytes::zeroed(HEAD_LENGTH + payload.len());
@@ -147,6 +159,10 @@ impl FecDecoder {
             batch_data[1] = flags_byte;
             batch_data[2..4].copy_from_slice(&(payload.len() as u16).to_be_bytes());
             batch_data[4..].copy_from_slice(&payload);
+            // 校验包先到时 shard 尺寸已知，补齐保持所有 shard 等长
+            if group.shard_size != 0 && batch_data.len() < group.shard_size {
+                batch_data.resize(group.shard_size, 0);
+            }
             group.received_shards[packet_index] = Some(batch_data);
             group.received_original_count += 1;
         }
@@ -161,7 +177,26 @@ impl FecDecoder {
             return Ok(packet.map(|v| vec![v]));
         }
 
-        let result = Self::try_decode(group, (src_ip, group_id), &net_packet)?;
+        // 解码失败（如对端构造的异常 group）只放弃该 group，
+        // 不能向上传播错误——否则当前完好接收的包会被连带丢弃
+        let decode_failed;
+        let result = match Self::try_decode(group, (src_ip, group_id), &net_packet) {
+            Ok(result) => {
+                decode_failed = false;
+                result
+            }
+            Err(e) => {
+                log::warn!(
+                    "fec decode failed, drop group, src={src_ip},group_id={group_id}: {e:?}"
+                );
+                decode_failed = true;
+                None
+            }
+        };
+        if decode_failed {
+            // 移除坏 group，后续该 group 的完好数据包按新 group 正常透传
+            inner.groups.remove(&(src_ip, group_id));
+        }
 
         if inner.last_cleanup.elapsed() > Duration::from_secs(1) {
             Self::cleanup_old_groups(&mut inner.groups);
@@ -300,3 +335,198 @@ impl FecDecoder {
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::fec::encoder::ParityData;
+
+    const SRC: u32 = 0x0A000001;
+    const DST: u32 = 0x0A000002;
+
+    /// 构造线上的 FEC 数据包（payload 为 prost 编码的 FecPacket）
+    fn build_data_packet(
+        group_id: u64,
+        packet_index: u32,
+        type_byte: u8,
+        flags_byte: u8,
+        payload: &[u8],
+    ) -> NetPacket<TransmissionBytes> {
+        let fec = FecPacket {
+            group_id,
+            packet_index,
+            payload: payload.to_vec(),
+            parity_data: None,
+        };
+        build_packet(fec, type_byte, flags_byte)
+    }
+
+    fn build_parity_packet(
+        group_id: u64,
+        packet_index: u32,
+        payload: Vec<u8>,
+        data_shards: u32,
+        parity_shards: u32,
+    ) -> NetPacket<TransmissionBytes> {
+        let fec = FecPacket {
+            group_id,
+            packet_index,
+            payload,
+            parity_data: Some(ParityData {
+                data_shards,
+                parity_shards,
+            }),
+        };
+        // 校验包头部的 type/flags 不参与数据重建，固定取值
+        build_packet(fec, 0x81, 0)
+    }
+
+    fn build_packet(
+        fec: FecPacket,
+        type_byte: u8,
+        flags_byte: u8,
+    ) -> NetPacket<TransmissionBytes> {
+        let fec_payload = fec.encode_to_vec();
+        let buffer = TransmissionBytes::zeroed(HEAD_LENGTH + fec_payload.len());
+        let mut pkt = NetPacket::new(buffer).unwrap();
+        pkt.head_mut()[0] = type_byte;
+        pkt.head_mut()[2] = flags_byte;
+        pkt.set_src_id(SRC);
+        pkt.set_dest_id(DST);
+        pkt.set_ttl(5);
+        pkt.set_fec_flag(true);
+        pkt.set_payload(&fec_payload).unwrap();
+        pkt
+    }
+
+    /// 按编码端格式组装 shard：[type_byte, flags_byte, len(u16), payload...]，填充到 max_len
+    fn make_shard(type_byte: u8, flags_byte: u8, payload: &[u8], max_len: usize) -> Vec<u8> {
+        let mut shard = vec![0u8; max_len];
+        shard[0] = type_byte;
+        shard[1] = flags_byte;
+        shard[2..4].copy_from_slice(&(payload.len() as u16).to_be_bytes());
+        shard[4..4 + payload.len()].copy_from_slice(payload);
+        shard
+    }
+
+    /// 变长批次：丢一个数据包，靠校验包必须能恢复（修复前必报 IncorrectShardSize）
+    #[test]
+    fn test_reconstruct_variable_length_batch() {
+        let decoder = FecDecoder::new();
+        let group_id = 1u64;
+
+        let payloads: [&[u8]; 3] = [&[0xAA; 10], &[0xBB; 30], &[0xCC; 20]];
+        let type_bytes = [0x81u8, 0x82, 0x83];
+        let max_len = 4 + 30; // 编码端按最长 shard 填充
+
+        // 编码端：3 数据 + 1 校验
+        let mut shards: Vec<Vec<u8>> = payloads
+            .iter()
+            .zip(type_bytes)
+            .map(|(p, t)| make_shard(t, 0, p, max_len))
+            .collect();
+        shards.push(vec![0u8; max_len]);
+        let rs = ReedSolomon::new(3, 1).unwrap();
+        {
+            let mut refs: Vec<&mut [u8]> = shards.iter_mut().map(|s| s.as_mut()).collect();
+            rs.encode(&mut refs).unwrap();
+        }
+        let parity = shards[3].clone();
+
+        // 收到 pkt0、pkt1，pkt2 丢失，随后收到校验包
+        assert!(
+            decoder
+                .receive(build_data_packet(group_id, 0, type_bytes[0], 0, payloads[0]))
+                .unwrap()
+                .is_some()
+        );
+        assert!(
+            decoder
+                .receive(build_data_packet(group_id, 1, type_bytes[1], 0, payloads[1]))
+                .unwrap()
+                .is_some()
+        );
+        let recovered = decoder
+            .receive(build_parity_packet(group_id, 3, parity, 3, 1))
+            .unwrap()
+            .expect("should reconstruct the lost packet");
+
+        assert_eq!(recovered.len(), 1);
+        assert_eq!(recovered[0].payload(), &payloads[2][..]);
+        assert_eq!(recovered[0].head()[0], type_bytes[2]);
+        assert_eq!(recovered[0].src_id(), SRC);
+        assert_eq!(recovered[0].dest_id(), DST);
+        assert!(!recovered[0].is_fec());
+    }
+
+    /// 校验包先到时，后到的数据 shard 也要补齐（等长）后能恢复
+    #[test]
+    fn test_reconstruct_when_parity_arrives_first() {
+        let decoder = FecDecoder::new();
+        let group_id = 2u64;
+
+        let payloads: [&[u8]; 2] = [&[0x11; 8], &[0x22; 24]];
+        let max_len = 4 + 24;
+
+        let mut shards = vec![
+            make_shard(0x81, 0, payloads[0], max_len),
+            make_shard(0x82, 0, payloads[1], max_len),
+            vec![0u8; max_len],
+        ];
+        let rs = ReedSolomon::new(2, 1).unwrap();
+        {
+            let mut refs: Vec<&mut [u8]> = shards.iter_mut().map(|s| s.as_mut()).collect();
+            rs.encode(&mut refs).unwrap();
+        }
+        let parity = shards[2].clone();
+
+        // 校验包先到（此时丢失 pkt0 还未知），再收到 pkt1
+        assert!(
+            decoder
+                .receive(build_parity_packet(group_id, 2, parity, 2, 1))
+                .unwrap()
+                .is_none()
+        );
+        let recovered = decoder
+            .receive(build_data_packet(group_id, 1, 0x82, 0, payloads[1]))
+            .unwrap()
+            .expect("should reconstruct pkt0 after parity-first");
+
+        // 校验包先到时 pkt1 会透传，恢复的 pkt0 也应在返回列表中
+        let recovered_pkt0 = recovered
+            .iter()
+            .find(|p| p.payload() == &payloads[0][..])
+            .expect("recovered pkt0 missing");
+        assert_eq!(recovered_pkt0.head()[0], 0x81);
+    }
+
+    /// 异常 group（校验 shard 比数据 shard 短）导致重建失败时：
+    /// 当前包不被连带丢弃，坏 group 被移除，后续数据包正常透传
+    #[test]
+    fn test_decode_failure_does_not_swallow_good_packets() {
+        let decoder = FecDecoder::new();
+        let group_id = 3u64;
+
+        // 先到一个较大的数据包
+        assert!(
+            decoder
+                .receive(build_data_packet(group_id, 0, 0x81, 0, &[0xAA; 40]))
+                .unwrap()
+                .is_some()
+        );
+        // 异常校验包：shard 只有 4 字节，比已存数据 shard 短，reconstruct 必失败
+        assert!(
+            decoder
+                .receive(build_parity_packet(group_id, 2, vec![0u8; 4], 2, 1))
+                .unwrap()
+                .is_none()
+        );
+        // 坏 group 已被移除，后续数据包按新 group 正常透传
+        let passed = decoder
+            .receive(build_data_packet(group_id, 1, 0x82, 0, &[0xBB; 8]))
+            .unwrap()
+            .expect("good packet must not be swallowed");
+        assert_eq!(passed.len(), 1);
+        assert_eq!(passed[0].payload(), &[0xBB; 8][..]);
+    }
+}

vnt-core/src/fec/encoder.rs

@@ -15,7 +15,7 @@ mod fec_proto {
     include!(concat!(env!("OUT_DIR"), "/protocol.fec.rs"));
 }
 use crate::utils::task_control::TaskGroup;
-pub use fec_proto::FecPacket;
+pub use fec_proto::{FecPacket, ParityData};
 
 const BATCH_SIZE: usize = 10;
 const REDUNDANCY_RATE: f32 = 0.2;