diff --git a/src/connections/stream_buffer.rs b/src/connections/stream_buffer.rs index 78e228c..eaabc4d 100644 --- a/src/connections/stream_buffer.rs +++ b/src/connections/stream_buffer.rs @@ -323,6 +323,15 @@ impl StreamBuffer { .map(|idx| idx + packet_data_start_index); if let Some(next_packet_start_idx) = next_packet_start_index { + // Only malformed if it doesn't decode at the declared length. + let mut declared_packet = + &self.buffer[packet_data_start_index..packet_data_start_index + packet_data_size]; + if protobufs::FromRadio::decode(&mut declared_packet).is_ok() + && declared_packet.is_empty() + { + return Ok(()); + } + // Remove malformed packet from buffer self.buffer.drain(..next_packet_start_idx); @@ -718,6 +727,59 @@ mod tests { #[tokio::test] async fn detect_malformed_packets_with_internal_header_sequence() {} + /// `outer_packet`'s serialized bytes coincidentally contain [0x94, 0xc3], + /// the same collision that caused a real hang. Identity fields are + /// placeholders; device_metrics is unchanged + /// since that's what produces the collision. Expected behavior is that both + /// packets decode, since the coincidental match doesn't mean outer_packet + /// is malformed. + #[tokio::test] + async fn resync_after_coincidental_header_match_inside_packet_payload() { + let outer_packet = protobufs::FromRadio { + payload_variant: Some(protobufs::from_radio::PayloadVariant::NodeInfo( + protobufs::NodeInfo { + device_metrics: Some(protobufs::DeviceMetrics { + uptime_seconds: Some(14655892), + ..Default::default() + }), + ..Default::default() + }, + )), + ..Default::default() + }; + let encoded_outer = format_data_packet(outer_packet.encode_to_vec().into()).unwrap(); + assert!( + encoded_outer.data_vec()[PACKET_HEADER_SIZE..] + .windows(2) + .any(|w| w == [0x94, 0xc3]), + "outer_packet no longer reproduces the coincidental header collision" + ); + + let recovery_packet = protobufs::FromRadio { + id: 0, + payload_variant: Some(protobufs::from_radio::PayloadVariant::NodeInfo( + protobufs::NodeInfo { + num: 987654321, + ..Default::default() + }, + )), + }; + let encoded_recovery = format_data_packet(recovery_packet.encode_to_vec().into()).unwrap(); + + let (mock_tx, mut mock_rx) = unbounded_channel::(); + let mut buffer = StreamBuffer::new(mock_tx); + + let mut data = encoded_outer.data_vec(); + data.extend_from_slice(encoded_recovery.data()); + buffer.process_incoming_bytes(data.into()); + + assert_eq!(timeout_test(mock_rx.recv(), None).await, Some(outer_packet)); + assert_eq!( + timeout_test(mock_rx.recv(), None).await, + Some(recovery_packet) + ); + } + #[tokio::test] async fn process_log_lines() { let (mock_tx, mut _mock_rx) = unbounded_channel::();