fix(zstd decompression): limit concurrency to 1 to prevent deadlock in zstd library

encoding/codecv7_test.go

@@ -4,8 +4,12 @@
 	"encoding/hex"
 	"encoding/json"
 	"fmt"
+	"log"
 	"math/big"
 	"math/rand"
+	"net/http"
+	_ "net/http"
+	_ "net/http/pprof"
 	"strings"
 	"testing"
 
@@ -17,6 +21,32 @@
 	"github.com/stretchr/testify/require"
 )
 
+// TestDecodeAllDeadlock tests the decompression of random bytes to trigger deadlock in zstd library.
+
+func TestDecodeAllDeadlock(t *testing.T) {
+	t.Skip("Skip test that triggers deadlock in zstd library")
+
+	go func() {
+		log.Println(http.ListenAndServe("localhost:6060", nil))
+	}()
+
+	// generate some random bytes
+	randomBytes := make([]byte, maxBlobBytes)
+	rand.Read(randomBytes)
+
+	c := NewDACodecV8()
+
+	compressed, err := c.CompressScrollBatchBytes(randomBytes)
+	require.NoError(t, err)
+
+	// repeatedly decompress the bytes to trigger deadlock in zstd library
+	for i := 0; i < 100000; i++ {
+		uncompressed, err := decompressV7Bytes(compressed)
+		require.NoError(t, err)
+		require.Equal(t, randomBytes, uncompressed)
+	}
+}
+
 // TestCodecV7DABlockEncodeDecode tests the encoding and decoding of daBlockV7.
 func TestCodecV7DABlockEncodeDecode(t *testing.T) {
 	codecV7, err := CodecFromVersion(CodecV7)

encoding/codecv7_types.go

@@ -482,7 +482,7 @@ func decompressV7Bytes(compressedBytes []byte) ([]byte, error) {
 
 	compressedBytes = append(zstdMagicNumber, compressedBytes...)
 	r := bytes.NewReader(compressedBytes)
-	zr, err := zstd.NewReader(r)
+	zr, err := zstd.NewReader(r, zstd.WithDecoderConcurrency(1))
 	if err != nil {
 		return nil, fmt.Errorf("failed to create zstd reader: %w", err)
 	}