Handle very rare stream failure when packets contain magic bytes

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::<protobufs::FromRadio>();
+        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::<protobufs::FromRadio>();