feat: Implement adaptive ZSTD compression levels for DA blob submission based on batch size.

(cherry picked from commit 3ff6211ec69d482437de79a617b5b39a88c91835)

Author: alpe

block/internal/da/client.go

@@ -88,10 +88,20 @@ func (c *client) Submit(ctx context.Context, data [][]byte, _ float64, namespace
 		}
 	}
 
+	// Select compression level based on backlog pressure:
+	// large batch = high backlog = prioritize speed;
+	// small batch = low backlog = prioritize ratio.
+	compLevel := da.LevelBest
+	switch {
+	case len(data) > 10:
+		compLevel = da.LevelFastest
+	case len(data) > 3:
+		compLevel = da.LevelDefault
+	}
+
 	blobs := make([]*blobrpc.Blob, len(data))
 	for i, raw := range data {
-		// Compress blob data before submission to reduce bandwidth and storage costs
-		compressed, compErr := da.Compress(raw)
+		compressed, compErr := da.Compress(raw, compLevel)
 		if compErr != nil {
 			return datypes.ResultSubmit{
 				BaseResult: datypes.BaseResult{
@@ -104,6 +114,7 @@ func (c *client) Submit(ctx context.Context, data [][]byte, _ float64, namespace
 			Int("original_size", len(raw)).
 			Int("compressed_size", len(compressed)).
 			Float64("ratio", float64(len(compressed))/float64(len(raw))).
+			Int("level", int(compLevel)).
 			Msg("compressed blob for DA submission")
 
 		if uint64(len(compressed)) > common.DefaultMaxBlobSize {

pkg/da/compression.go

@@ -10,33 +10,59 @@ import (
 // ASCII "ZSTD" = 0x5A 0x53 0x54 0x44.
 var magic = []byte{0x5A, 0x53, 0x54, 0x44}
 
-// encoder and decoder are package-level singletons. They are safe for
-// concurrent use per the klauspost/compress documentation.
-var (
-	encoder *zstd.Encoder
-	decoder *zstd.Decoder
+// CompressionLevel controls the speed/ratio trade-off for blob compression.
+type CompressionLevel int
+
+const (
+	// LevelFastest prioritizes speed over compression ratio.
+	// Use when backlog is high and throughput matters most.
+	LevelFastest CompressionLevel = iota
+	// LevelDefault balances speed and compression ratio.
+	LevelDefault
+	// LevelBest prioritizes compression ratio over speed.
+	// Use when backlog is low to save bandwidth and storage.
+	LevelBest
 )
 
+// encoders holds one zstd encoder per compression level. Each is safe for
+// concurrent use per the klauspost/compress documentation.
+var encoders [3]*zstd.Encoder
+
+// decoder is a package-level singleton, safe for concurrent use.
+var decoder *zstd.Decoder
+
 func init() {
-	var err error
-	encoder, err = zstd.NewWriter(nil, zstd.WithEncoderLevel(zstd.SpeedDefault))
-	if err != nil {
-		panic(fmt.Sprintf("compression: create zstd encoder: %v", err))
+	levels := [3]zstd.EncoderLevel{
+		zstd.SpeedFastest,
+		zstd.SpeedDefault,
+		zstd.SpeedBestCompression,
+	}
+	for i, lvl := range levels {
+		enc, err := zstd.NewWriter(nil, zstd.WithEncoderLevel(lvl))
+		if err != nil {
+			panic(fmt.Sprintf("compression: create zstd encoder (level %d): %v", i, err))
+		}
+		encoders[i] = enc
 	}
+	var err error
 	decoder, err = zstd.NewReader(nil)
 	if err != nil {
 		panic(fmt.Sprintf("compression: create zstd decoder: %v", err))
 	}
 }
 
-// Compress compresses data using zstd and prepends the magic prefix.
+// Compress compresses data using zstd at the given level and prepends the magic prefix.
 // Returns the original data unchanged if it is empty.
-func Compress(data []byte) ([]byte, error) {
+func Compress(data []byte, level CompressionLevel) ([]byte, error) {
 	if len(data) == 0 {
 		return data, nil
 	}
 
-	compressed := encoder.EncodeAll(data, nil)
+	if level < LevelFastest || level > LevelBest {
+		level = LevelDefault
+	}
+
+	compressed := encoders[level].EncodeAll(data, nil)
 
 	// Prepend magic prefix
 	result := make([]byte, len(magic)+len(compressed))
@@ -57,7 +83,7 @@ func Decompress(data []byte) ([]byte, error) {
 	// Strip magic prefix and decompress
 	decompressed, err := decoder.DecodeAll(data[len(magic):], nil)
 	if err != nil {
-		return nil, fmt.Errorf("compression: zstd decompress failed: %w", err)
+		return nil, fmt.Errorf("compression: zstd decompress: %w", err)
 	}
 
 	return decompressed, nil

pkg/da/compression_test.go

@@ -34,7 +34,7 @@ func TestCompressDecompress_RoundTrip(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			compressed, err := Compress(tt.data)
+			compressed, err := Compress(tt.data, LevelDefault)
 			require.NoError(t, err)
 
 			assert.True(t, IsCompressed(compressed), "compressed data should have magic prefix")
@@ -47,12 +47,46 @@ func TestCompressDecompress_RoundTrip(t *testing.T) {
 	}
 }
 
+func TestCompress_AllLevelsRoundTrip(t *testing.T) {
+	data := bytes.Repeat([]byte("adaptive compression level test "), 5000)
+	levels := []struct {
+		name  string
+		level CompressionLevel
+	}{
+		{"fastest", LevelFastest},
+		{"default", LevelDefault},
+		{"best", LevelBest},
+	}
+
+	var sizes []int
+	for _, lvl := range levels {
+		t.Run(lvl.name, func(t *testing.T) {
+			compressed, err := Compress(data, lvl.level)
+			require.NoError(t, err)
+			assert.True(t, IsCompressed(compressed))
+
+			sizes = append(sizes, len(compressed))
+
+			decompressed, err := Decompress(compressed)
+			require.NoError(t, err)
+			assert.Equal(t, data, decompressed)
+
+			t.Logf("level=%s compressed=%d ratio=%.4f", lvl.name, len(compressed), float64(len(compressed))/float64(len(data)))
+		})
+	}
+
+	// Best should produce equal or smaller output than Fastest
+	if len(sizes) == 3 {
+		assert.LessOrEqual(t, sizes[2], sizes[0], "LevelBest should compress at least as well as LevelFastest")
+	}
+}
+
 func TestCompress_Empty(t *testing.T) {
-	compressed, err := Compress(nil)
+	compressed, err := Compress(nil, LevelDefault)
 	require.NoError(t, err)
 	assert.Nil(t, compressed)
 
-	compressed, err = Compress([]byte{})
+	compressed, err = Compress([]byte{}, LevelDefault)
 	require.NoError(t, err)
 	assert.Empty(t, compressed)
 }
@@ -112,9 +146,8 @@ func TestIsCompressed(t *testing.T) {
 }
 
 func TestCompress_AchievesCompression(t *testing.T) {
-	// Highly compressible data should achieve meaningful compression
 	data := bytes.Repeat([]byte("rollkit block data with repeated content "), 10000)
-	compressed, err := Compress(data)
+	compressed, err := Compress(data, LevelDefault)
 	require.NoError(t, err)
 
 	ratio := float64(len(compressed)) / float64(len(data))
@@ -123,12 +156,11 @@ func TestCompress_AchievesCompression(t *testing.T) {
 }
 
 func TestCompress_RandomDataStillWorks(t *testing.T) {
-	// Random data won't compress well but should still round-trip correctly
 	data := make([]byte, 4096)
 	_, err := rand.Read(data)
 	require.NoError(t, err)
 
-	compressed, err := Compress(data)
+	compressed, err := Compress(data, LevelFastest)
 	require.NoError(t, err)
 
 	decompressed, err := Decompress(compressed)
@@ -137,9 +169,18 @@ func TestCompress_RandomDataStillWorks(t *testing.T) {
 }
 
 func TestDecompress_DataStartingWithMagicButUncompressed(t *testing.T) {
-	// Edge case: data that happens to start with the magic bytes but isn't actually compressed.
-	// This should fail decompression (invalid zstd frame).
 	fakeCompressed := append([]byte{0x5A, 0x53, 0x54, 0x44}, bytes.Repeat([]byte{0x00}, 100)...)
 	_, err := Decompress(fakeCompressed)
 	assert.Error(t, err, "data starting with magic but containing invalid zstd should error")
 }
+
+func TestCompress_InvalidLevel(t *testing.T) {
+	// Out-of-range level should fall back to LevelDefault
+	data := []byte("test data for invalid level")
+	compressed, err := Compress(data, CompressionLevel(99))
+	require.NoError(t, err)
+
+	decompressed, err := Decompress(compressed)
+	require.NoError(t, err)
+	assert.Equal(t, data, decompressed)
+}