net: skip a compressed frame that carries no packet

A server may send a compressed frame whose uncompressed length is 0 and
which holds no packet: `ConnReader::recv` read the packet id from the
empty body and failed with "needed 1 bytes, 0 left", which ended the
connection. Vanilla skips such a frame: its PacketDecoder only decodes a
packet from a readable buffer, so an empty frame produces no packet and
the connection stays up.

A frame with no bytes at all is a corrupted frame in vanilla ("Frame
length cannot be zero"), so the frame splitter now says that instead of
failing later with a missing byte.

Fixes #2739

Co-Authored-By: minimax-m3.1-flash (Claude Code) <[email protected]>

crates/horizon-protocol/src/connection.rs

@@ -281,6 +281,13 @@ impl ConnReader {
         loop {
             if let Some(frame) = frame::split(&mut self.buffer)? {
                 let body = frame::decompress(frame, self.threshold)?;
+                if body.is_empty() {
+                    // A frame with no packet in it. Vanilla's PacketDecoder
+                    // only decodes a packet from a readable buffer, so an
+                    // empty compressed frame is skipped and the connection
+                    // stays up.
+                    continue;
+                }
                 let mut r = Reader::new(&body);
                 let id = r.varint()?;
                 let start = r.position();
@@ -496,4 +503,30 @@ mod tests {
         assert_eq!(c.recv().unwrap(), (5, vec![4, 3]));
         client.join().unwrap();
     }
+
+    /// A server may send a compressed frame that carries no packet: the
+    /// uncompressed length is 0 and the frame holds nothing else. Vanilla
+    /// skips it (its PacketDecoder only reads a packet from a readable
+    /// buffer), so the client stays connected.
+    #[test]
+    fn an_empty_compressed_frame_is_skipped() {
+        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
+        let port = listener.local_addr().unwrap().port();
+        let server = std::thread::spawn(move || {
+            let (stream, _) = listener.accept().unwrap();
+            let mut c = Connection::from_stream(stream).unwrap();
+            c.set_compression(64);
+            // The empty frame: its length prefix, then the uncompressed
+            // length 0 and nothing else.
+            c.writer.send_contents(&[0]).unwrap();
+            let mut body = vec![3];
+            body.extend_from_slice(b"after");
+            c.writer.send(&body).unwrap();
+        });
+        let mut c = Connection::connect("127.0.0.1", port, Duration::from_secs(5)).unwrap();
+        c.set_compression(64);
+        let (id, payload) = c.recv().unwrap();
+        assert_eq!((id, payload.as_slice()), (3, &b"after"[..]));
+        server.join().unwrap();
+    }
 }

crates/horizon-protocol/src/frame.rs

@@ -65,6 +65,11 @@ pub fn split(buffer: &mut Vec<u8>) -> Result<Option<Vec<u8>>> {
         };
         len |= usize::from(b & 0x7F) << (7 * i);
         if b & 0x80 == 0 {
+            if len == 0 {
+                // Varint21FrameDecoder: a frame with no bytes is a
+                // corrupted frame, not an empty packet.
+                return Err(Error::new("Frame length cannot be zero"));
+            }
             let start = i + 1;
             if buffer.len() < start + len {
                 return Ok(None);