compression: avoid msan crashes

Some of the compression algorithms use hand-written assembly code,
which is "invislble" to the memory sanitizer. Together with the recent
change to use `malloc` instead of `calloc` for the block cache, this
results in uninitialized memory failures with `msan`.

The fix is to call `runtime.MSanWrite()` to inform `msan` that the
memory was written to. Note that `runtime.MSanWrite()` exists only
in builds with the `msan` tag.

Fixes #5275

Author: RaduBerinde

internal/compression/minlz.go

@@ -29,6 +29,7 @@ func (c *minlzCompressor) Compress(dst, src []byte) ([]byte, Setting) {
 	if err != nil {
 		panic(errors.Wrap(err, "minlz compression"))
 	}
+	msanWrite(compressed)
 	return compressed, Setting{Algorithm: MinLZ, Level: uint8(c.level)}
 }
 
@@ -54,11 +55,15 @@ var _ Decompressor = minlzDecompressor{}
 
 func (minlzDecompressor) DecompressInto(buf, compressed []byte) error {
 	result, err := minlz.Decode(buf, compressed)
+	if err != nil {
+		return err
+	}
 	if len(result) != len(buf) || (len(result) > 0 && &result[0] != &buf[0]) {
 		return base.CorruptionErrorf("pebble/table: decompressed into unexpected buffer: %p != %p",
 			errors.Safe(result), errors.Safe(buf))
 	}
-	return err
+	msanWrite(result)
+	return nil
 }
 
 func (minlzDecompressor) DecompressedLen(b []byte) (decompressedLen int, err error) {

internal/compression/msan_off.go

@@ -0,0 +1,11 @@
+// Copyright 2025 The LevelDB-Go and Pebble Authors. All rights reserved. Use
+// of this source code is governed by a BSD-style license that can be found in
+// the LICENSE file.
+
+//go:build !msan
+
+package compression
+
+// msanWrite is used to inform MemorySanitizer that the memory in p was written
+// to; this is necessary for some compression algorithms which use assembly code.
+func msanWrite(p []byte) {}

internal/compression/msan_on.go

@@ -0,0 +1,20 @@
+// Copyright 2025 The LevelDB-Go and Pebble Authors. All rights reserved. Use
+// of this source code is governed by a BSD-style license that can be found in
+// the LICENSE file.
+
+//go:build msan
+
+package compression
+
+import (
+	"runtime"
+	"unsafe"
+)
+
+// msanWrite is used to inform MemorySanitizer that the memory in p was written
+// to; this is necessary for some compression algorithms which use assembly code.
+func msanWrite(p []byte) {
+	if len(p) > 0 {
+		runtime.MSanWrite(unsafe.Pointer(&p[0]), len(p))
+	}
+}

internal/compression/snappy.go

@@ -17,8 +17,9 @@ var _ Compressor = snappyCompressor{}
 func (snappyCompressor) Algorithm() Algorithm { return Snappy }
 
 func (snappyCompressor) Compress(dst, src []byte) ([]byte, Setting) {
-	dst = dst[:cap(dst):cap(dst)]
-	return snappy.Encode(dst, src), SnappySetting
+	result := snappy.Encode(dst[:cap(dst):cap(dst)], src)
+	msanWrite(result)
+	return result, SnappySetting
 }
 
 func (snappyCompressor) Close() {}
@@ -36,6 +37,7 @@ func (snappyDecompressor) DecompressInto(buf, compressed []byte) error {
 		return base.CorruptionErrorf("pebble: decompressed into unexpected buffer: %p != %p",
 			errors.Safe(result), errors.Safe(buf))
 	}
+	msanWrite(buf)
 	return nil
 }
 

internal/compression/zstd_cgo.go

@@ -60,6 +60,7 @@ func (z *zstdCompressor) Compress(compressedBuf []byte, b []byte) ([]byte, Setti
 	if &result[0] != &compressedBuf[varIntLen] {
 		panic("Allocated a new buffer despite checking CompressBound.")
 	}
+	msanWrite(result)
 
 	return compressedBuf[:varIntLen+len(result)], Setting{Algorithm: Zstd, Level: uint8(z.level)}
 }
@@ -80,8 +81,7 @@ type zstdDecompressor struct {
 
 var _ Decompressor = (*zstdDecompressor)(nil)
 
-// DecompressInto decompresses src with the Zstandard algorithm. The destination
-// buffer must already be sufficiently sized, otherwise DecompressInto may error.
+// DecompressInto is part of the Decompressor interface.
 func (z *zstdDecompressor) DecompressInto(dst, src []byte) error {
 	// The payload is prefixed with a varint encoding the length of
 	// the decompressed block.
@@ -93,10 +93,14 @@ func (z *zstdDecompressor) DecompressInto(dst, src []byte) error {
 	if len(dst) == 0 {
 		return errors.Errorf("decodeZstd: empty dst buffer")
 	}
-	_, err := z.ctx.DecompressInto(dst, src)
+	n, err := z.ctx.DecompressInto(dst, src)
 	if err != nil {
 		return err
 	}
+	if n != len(dst) {
+		return base.CorruptionErrorf("ZSTD decompressed %d bytes, expected %d", n, len(dst))
+	}
+	msanWrite(dst)
 	return nil
 }
 

internal/compression/zstd_nocgo.go

@@ -43,6 +43,7 @@ func (z *zstdCompressor) Compress(compressedBuf, b []byte) ([]byte, Setting) {
 	}
 	varIntLen := binary.PutUvarint(compressedBuf, uint64(len(b)))
 	res := z.encoder.EncodeAll(b, compressedBuf[:varIntLen])
+	msanWrite(res)
 	return res, Setting{Algorithm: Zstd, Level: uint8(z.level)}
 }
 
@@ -84,6 +85,7 @@ func (zstdDecompressor) DecompressInto(dst, src []byte) error {
 		return base.CorruptionErrorf("pebble/table: decompressed into unexpected buffer: %p != %p",
 			errors.Safe(result), errors.Safe(dst))
 	}
+	msanWrite(result)
 	return nil
 }