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

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

* skip test

* use crypt/rand

* address review comments

Author: jonastheis

encoding/codecv7_test.go

@@ -1,11 +1,15 @@
 package encoding
 
 import (
+	crand "crypto/rand"
 	"encoding/hex"
 	"encoding/json"
 	"fmt"
+	"log"
 	"math/big"
 	"math/rand"
+	"net/http"
+	_ "net/http/pprof"
 	"strings"
 	"testing"
 
@@ -17,6 +21,33 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+// TestDecodeAllDeadlock tests the decompression of random bytes to trigger deadlock in zstd library
+// with setting of zstd.WithDecoderConcurrency(2).
+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)
+	_, err := crand.Read(randomBytes)
+	require.NoError(t, err)
+
+	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)
 	}

encoding/da.go

@@ -552,7 +552,7 @@ func decompressScrollBlobToBatch(compressedBytes []byte) ([]byte, error) {
 	batchOfBytes := make([]byte, readBatchSize)
 
 	r := bytes.NewReader(compressedBytes)
-	zr, err := zstd.NewReader(r)
+	zr, err := zstd.NewReader(r, zstd.WithDecoderConcurrency(1))
 	if err != nil {
 		return nil, err
 	}