fix metadata encoding

pkg/experiment/block/metadata/metadata.go

@@ -164,7 +164,8 @@ var castagnoli = crc32.MakeTable(crc32.Castagnoli)
 //	be_uint32 | size of the raw metadata
 //	be_uint32 | CRC32 of the raw metadata and size
 func Encode(w io.Writer, md *metastorev1.BlockMeta) error {
-	ww := crc32.New(castagnoli)
+	crc := crc32.New(castagnoli)
+	w = io.MultiWriter(w, crc)
 	b, _ := md.MarshalVT()
 	n, err := w.Write(b)
 	if err != nil {
@@ -173,7 +174,7 @@ func Encode(w io.Writer, md *metastorev1.BlockMeta) error {
 	if err = binary.Write(w, binary.BigEndian, uint32(n)); err != nil {
 		return err
 	}
-	return binary.Write(w, binary.BigEndian, ww.Sum32())
+	return binary.Write(w, binary.BigEndian, crc.Sum32())
 }
 
 // Decode metadata encoded with Encode.
@@ -183,11 +184,12 @@ func Decode(b []byte, md *metastorev1.BlockMeta) error {
 	}
 	crc := binary.BigEndian.Uint32(b[len(b)-4:])
 	size := binary.BigEndian.Uint32(b[len(b)-8 : len(b)-4])
-	if size != uint32(len(b)-8) {
+	off := len(b) - 8 - int(size)
+	if off < 0 {
 		return fmt.Errorf("%w: invalid size", ErrMetadataInvalid)
 	}
-	if crc32.Checksum(b[:len(b)-4], castagnoli) != crc {
+	if crc32.Checksum(b[off:len(b)-4], castagnoli) != crc {
 		return fmt.Errorf("%w: invalid CRC", ErrMetadataInvalid)
 	}
-	return md.UnmarshalVT(b[:len(b)-8])
+	return md.UnmarshalVT(b[off : len(b)-8])
 }

pkg/experiment/block/metadata/metadata_test.go

@@ -6,6 +6,7 @@ import (
 
 	"github.com/oklog/ulid"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 
 	metastorev1 "github.com/grafana/pyroscope/api/gen/proto/go/metastore/v1"
 )
@@ -216,3 +217,28 @@ func TestMetadataStrings_Export(t *testing.T) {
 
 	assert.Equal(t, expected, md)
 }
+
+func TestMetadata_EncodeDecode(t *testing.T) {
+	md := &metastorev1.BlockMeta{
+		Id:        "1",
+		Tenant:    0,
+		CreatedBy: 1,
+		Datasets: []*metastorev1.Dataset{
+			{Tenant: 2, Name: 3, Labels: []int32{2, 4, 3, 5, 6, 2, 4, 3, 5, 7, 2, 4, 3, 5, 8}},
+			{Tenant: 9, Name: 10, Labels: []int32{2, 4, 10, 5, 7, 2, 4, 10, 5, 8, 2, 4, 10, 5, 11}},
+		},
+		StringTable: []string{
+			"", "ingester",
+			"tenant-a", "dataset-a", "service_name", "__profile_type__", "1", "2", "3",
+			"tenant-b", "dataset-b", "4",
+		},
+	}
+
+	var buf bytes.Buffer
+	require.NoError(t, Encode(&buf, md))
+
+	var d metastorev1.BlockMeta
+	raw := append([]byte("garbage"), buf.Bytes()...)
+	require.NoError(t, Decode(raw, &d))
+	assert.Equal(t, md, &d)
+}