fix: validate declared bucket count in TreemapReader

TreemapReader::new previously read only the 8-byte bucket count and
trusted the declared size, so a corrupt buffer that declares N buckets
but carries fewer (or none) was silently treated as valid.

This diverged from deserialize_bitmap, which will rejects it.

A lazy check in TreeMapIter::next cannot close this gap, because
early-return callers such as bitmap_min stop calling next before
reaching the missing trailing

Walk all `size` bucket headers in TreemapReader::new so that truncation
is rejected up front. The cost paid to keep consistent with
deserialize_bitmap on corrupt buffers.

Author: harry-hao

src/common/io/src/bitmap/reader.rs

@@ -47,6 +47,31 @@ impl<'a> TreemapReader<'a> {
             .read_u64::<LittleEndian>()
             .map_err(|_| Error::other("fail to read size"))?;
 
+        // Validate that `size` buckets fit in `buf` before returning.
+        //
+        // Without this check, the fast path diverges from `deserialize_bitmap`
+        // on corrupt input. Consider `bitmap_has_all(lhs, corrupt_rhs)` where
+        // `corrupt_rhs` declares buckets but carries no bucket data. The fast
+        // path would treat `corrupt_rhs` as empty and return `true`, while
+        // `deserialize_bitmap(corrupt_rhs)` rejects the same buffer as
+        // BadBytes.
+        //
+        // A lazy check in `TreeMapIter::next` cannot close this gap. It fires
+        // only when `next` is called. Early-return callers such as
+        // `bitmap_min` (stops at the first non-empty container) stop calling
+        // `next` before they ever reach a missing trailing bucket. The check
+        // must run before iteration begins.
+        //
+        // `BitmapReader::decode` reads only the cookie, the last container
+        // description, and the last offset, then seeks over container bodies.
+        // This cost is paid to keep consistent with `deserialize_bitmap` on
+        // corrupted buffers.
+        let mut probe = buf;
+        for _ in 0..size {
+            let header = BitmapReader::decode(probe)?;
+            probe = &probe[header.buf.len()..];
+        }
+
         Ok(Self { buf, _size: size })
     }
 
@@ -992,6 +1017,31 @@ mod tests {
         Ok(())
     }
 
+    // A declared bucket count with no bucket data must be rejected up front.
+    #[test]
+    fn test_treemap_reader_rejects_truncated_buf() -> io::Result<()> {
+        let corrupt = 1u64.to_le_bytes();
+        assert!(TreemapReader::new(&corrupt).is_err());
+        Ok(())
+    }
+
+    // Partial truncation: the first bucket is intact and non-empty, but the
+    // declared count claims more buckets than exist. Eager validation in
+    // `TreemapReader::new` catches this before any caller can early-return.
+    #[test]
+    fn test_treemap_reader_rejects_partial_truncated_buf() -> io::Result<()> {
+        let mut tree = RoaringTreemap::new();
+        tree.insert(7);
+        let mut buf = Vec::new();
+        tree.serialize_into(&mut buf)?;
+
+        // Overwrite the leading u64 to claim 2 buckets while only 1 exists.
+        buf[..8].copy_from_slice(&2u64.to_le_bytes());
+
+        assert!(TreemapReader::new(&buf).is_err());
+        Ok(())
+    }
+
     #[test]
     fn test_intersection() -> io::Result<()> {
         let v1 = create_bitmap(123);