diff --git a/adapters/core2sn/class.go b/adapters/core2sn/class.go index c096de74ac..3235177619 100644 --- a/adapters/core2sn/class.go +++ b/adapters/core2sn/class.go @@ -6,6 +6,7 @@ import ( "github.com/NethermindEth/juno/core" "github.com/NethermindEth/juno/starknet" "github.com/NethermindEth/juno/utils" + "github.com/NethermindEth/juno/utils/compression" ) func AdaptSegmentLengths(l core.SegmentLengths) starknet.SegmentLengths { @@ -115,7 +116,7 @@ func AdaptDeprecatedEntryPoint(ep *core.DeprecatedEntryPoint) starknet.EntryPoin func AdaptDeprecatedCairoClass( class *core.DeprecatedCairoClass, ) (starknet.DeprecatedCairoClass, error) { - decompressedProgram, err := utils.Gzip64Decode(class.Program) + decompressedProgram, err := compression.Gzip64Decode(class.Program) if err != nil { return starknet.DeprecatedCairoClass{}, err } diff --git a/adapters/sn2core/sn2core.go b/adapters/sn2core/sn2core.go index 24de85fe6e..d25d3e1eb5 100644 --- a/adapters/sn2core/sn2core.go +++ b/adapters/sn2core/sn2core.go @@ -12,6 +12,7 @@ import ( "github.com/NethermindEth/juno/l1/eth" "github.com/NethermindEth/juno/starknet" "github.com/NethermindEth/juno/utils" + "github.com/NethermindEth/juno/utils/compression" ) // ErrPreConfirmedIdentifierMismatch is returned by AdaptPreConfirmedWithDelta @@ -430,7 +431,7 @@ func AdaptDeprecatedCairoClass( } var err error - class.Program, err = utils.Gzip64Encode(response.Program) + class.Program, err = compression.Gzip64Encode(response.Program) if err != nil { return nil, err } diff --git a/clients/gateway/gateway.go b/clients/gateway/gateway.go index 0c6a9711e0..a3010fdd1d 100644 --- a/clients/gateway/gateway.go +++ b/clients/gateway/gateway.go @@ -2,7 +2,6 @@ package gateway import ( "bytes" - "compress/gzip" "context" "encoding/json" "errors" @@ -12,6 +11,7 @@ import ( "net/url" "time" + "github.com/NethermindEth/juno/utils/compression" "github.com/NethermindEth/juno/utils/log" ) @@ -146,7 +146,8 @@ func prepareRequestBody(jsonBody []byte) (io.Reader, bool, error) { } var buf bytes.Buffer - gzWriter := gzip.NewWriter(&buf) + gzWriter := compression.GzipWriter(&buf) + defer gzWriter.Release() if _, err := gzWriter.Write(jsonBody); err != nil { return nil, false, fmt.Errorf("writing gzip content: %w", err) } diff --git a/core/class_hash.go b/core/class_hash.go index ab1d72ea52..630273a08a 100644 --- a/core/class_hash.go +++ b/core/class_hash.go @@ -5,11 +5,11 @@ import ( "github.com/NethermindEth/juno/core/crypto" "github.com/NethermindEth/juno/core/felt" - "github.com/NethermindEth/juno/utils" + "github.com/NethermindEth/juno/utils/compression" ) func deprecatedCairoClassHash(class *DeprecatedCairoClass) (felt.Felt, error) { - decompressedProgram, err := utils.Gzip64Decode(class.Program) + decompressedProgram, err := compression.Gzip64Decode(class.Program) if err != nil { return felt.Felt{}, err } diff --git a/jsonrpc/http.go b/jsonrpc/http.go index 4cdd8cc87e..9b0ea15fc4 100644 --- a/jsonrpc/http.go +++ b/jsonrpc/http.go @@ -1,7 +1,6 @@ package jsonrpc import ( - "compress/gzip" "context" "errors" "io" @@ -9,20 +8,14 @@ import ( "net/http" "strconv" "strings" - "sync" "time" "github.com/NethermindEth/juno/db" + "github.com/NethermindEth/juno/utils/compression" "github.com/NethermindEth/juno/utils/log" "go.uber.org/zap" ) -// gzipWriterPool holds gzip writers for reuse; -// callers must Reset a writer onto their destination before writing to it. -var gzipWriterPool = sync.Pool{ - New: func() any { return gzip.NewWriter(io.Discard) }, -} - type HTTP struct { rpc *Server logger log.StructuredLogger @@ -131,11 +124,10 @@ func (h *HTTP) ServeHTTP(writer http.ResponseWriter, req *http.Request) { var ioWriter io.Writer = writer if strings.Contains(req.Header.Get("Accept-Encoding"), "gzip") { writer.Header().Set("Content-Encoding", "gzip") - gw := gzipWriterPool.Get().(*gzip.Writer) - gw.Reset(writer) + gw := compression.GzipWriter(writer) defer func() { closeErr := gw.Close() - gzipWriterPool.Put(gw) + gw.Release() if closeErr != nil { http.Error(writer, "gzip close error", http.StatusInternalServerError) return diff --git a/rpc/v10/transaction.go b/rpc/v10/transaction.go index 80112b5c3e..129995c452 100644 --- a/rpc/v10/transaction.go +++ b/rpc/v10/transaction.go @@ -8,7 +8,6 @@ import ( "encoding/json" "errors" "fmt" - "io" "sync" "github.com/NethermindEth/juno/adapters/sn2core" @@ -21,21 +20,14 @@ import ( "github.com/NethermindEth/juno/rpc/rpccore" "github.com/NethermindEth/juno/starknet" "github.com/NethermindEth/juno/starknet/compiler" + "github.com/NethermindEth/juno/utils/compression" "github.com/NethermindEth/juno/utils/throttler" "go.uber.org/zap" ) -var ( - gzPool = sync.Pool{ - New: func() any { - w, _ := gzip.NewWriterLevel(io.Discard, gzip.BestSpeed) - return w - }, - } - bufPool = sync.Pool{ - New: func() any { return new(bytes.Buffer) }, - } -) +var bufPool = sync.Pool{ + New: func() any { return new(bytes.Buffer) }, +} // AdaptTransaction adapts a core.Transaction to a local Transaction. // It's a wrapper around AdaptCoreTransaction that allows to exclude proof facts @@ -244,9 +236,8 @@ func ContractClassToGatewayPayload(class *ContractClass) ([]byte, error) { defer bufPool.Put(sierraBuf) b64 := base64.NewEncoder(base64.StdEncoding, sierraBuf) - gz := gzPool.Get().(*gzip.Writer) - gz.Reset(b64) - defer gzPool.Put(gz) + gz := compression.GzipWriterLevel(b64, gzip.BestSpeed) + defer gz.Release() enc := json.NewEncoder(gz) enc.SetEscapeHTML(false) diff --git a/rpc/v10/transaction_test.go b/rpc/v10/transaction_test.go index 2fdc5a7046..7724c2d659 100644 --- a/rpc/v10/transaction_test.go +++ b/rpc/v10/transaction_test.go @@ -29,7 +29,7 @@ import ( "github.com/NethermindEth/juno/starknet" adaptfeeder "github.com/NethermindEth/juno/starknetdata/feeder" "github.com/NethermindEth/juno/sync/preconfirmed" - "github.com/NethermindEth/juno/utils" + "github.com/NethermindEth/juno/utils/compression" "github.com/NethermindEth/juno/utils/log" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -2285,7 +2285,7 @@ func TestContractClassToGatewayPayload(t *testing.T) { require.Equal(t, class.EntryPoints, decoded.EntryPoints) require.Equal(t, class.ABI, decoded.ABI) - sierraJSON, err := utils.Gzip64Decode(decoded.SierraProgram) + sierraJSON, err := compression.Gzip64Decode(decoded.SierraProgram) require.NoError(t, err, "sierra_program must be gzip+base64 encoded") var roundTripped []felt.Felt diff --git a/rpc/v8/class.go b/rpc/v8/class.go index 184647155a..10ddaf0c61 100644 --- a/rpc/v8/class.go +++ b/rpc/v8/class.go @@ -14,6 +14,7 @@ import ( "github.com/NethermindEth/juno/starknet" "github.com/NethermindEth/juno/starknet/compiler" "github.com/NethermindEth/juno/utils" + "github.com/NethermindEth/juno/utils/compression" ) // https://github.com/starkware-libs/starknet-specs/blob/v0.8.1/api/starknet_api_openrpc.json#L3159 @@ -66,7 +67,7 @@ func adaptDeclaredClass( } base64Program := string(program[1 : len(program)-1]) - feederClass.DeprecatedCairo.Program, err = utils.Gzip64Decode(base64Program) + feederClass.DeprecatedCairo.Program, err = compression.Gzip64Decode(base64Program) if err != nil { return nil, err } diff --git a/rpc/v8/transaction.go b/rpc/v8/transaction.go index 2b17ce5edd..51079d47c9 100644 --- a/rpc/v8/transaction.go +++ b/rpc/v8/transaction.go @@ -20,7 +20,7 @@ import ( "github.com/NethermindEth/juno/rpc/rpccore" "github.com/NethermindEth/juno/starknet" "github.com/NethermindEth/juno/starknet/compiler" - "github.com/NethermindEth/juno/utils" + "github.com/NethermindEth/juno/utils/compression" "github.com/NethermindEth/juno/utils/throttler" "go.uber.org/zap" ) @@ -711,7 +711,7 @@ func (h *Handler) pushToFeederGateway(ctx context.Context, tx *BroadcastedTransa return AddTxResponse{}, jsonrpc.Err(jsonrpc.InternalError, errIn.Error()) } - gwSierraProg, errIn := utils.Gzip64Encode(sierraProgBytes) + gwSierraProg, errIn := compression.Gzip64Encode(sierraProgBytes) if errIn != nil { return AddTxResponse{}, jsonrpc.Err(jsonrpc.InternalError, errIn.Error()) } diff --git a/rpc/v9/class.go b/rpc/v9/class.go index 35973e8a9b..5803cab3ad 100644 --- a/rpc/v9/class.go +++ b/rpc/v9/class.go @@ -13,7 +13,7 @@ import ( "github.com/NethermindEth/juno/rpc/rpccore" "github.com/NethermindEth/juno/starknet" "github.com/NethermindEth/juno/starknet/compiler" - "github.com/NethermindEth/juno/utils" + "github.com/NethermindEth/juno/utils/compression" ) type CalldataInputs = rpccore.LimitSlice[felt.Felt, rpccore.FunctionCalldataLimit] @@ -73,7 +73,7 @@ func AdaptDeclaredClass( } base64Program := string(program[1 : len(program)-1]) - feederClass.DeprecatedCairo.Program, err = utils.Gzip64Decode(base64Program) + feederClass.DeprecatedCairo.Program, err = compression.Gzip64Decode(base64Program) if err != nil { return nil, err } diff --git a/rpc/v9/transaction.go b/rpc/v9/transaction.go index 1aecf1a3df..8b2879cafb 100644 --- a/rpc/v9/transaction.go +++ b/rpc/v9/transaction.go @@ -21,7 +21,7 @@ import ( "github.com/NethermindEth/juno/rpc/rpccore" "github.com/NethermindEth/juno/starknet" "github.com/NethermindEth/juno/starknet/compiler" - "github.com/NethermindEth/juno/utils" + "github.com/NethermindEth/juno/utils/compression" "github.com/NethermindEth/juno/utils/throttler" "go.uber.org/zap" ) @@ -789,7 +789,7 @@ func (h *Handler) pushToFeederGateway( return AddTxResponse{}, jsonrpc.Err(jsonrpc.InternalError, errIn.Error()) } - gwSierraProg, errIn := utils.Gzip64Encode(sierraProgBytes) + gwSierraProg, errIn := compression.Gzip64Encode(sierraProgBytes) if errIn != nil { return AddTxResponse{}, jsonrpc.Err(jsonrpc.InternalError, errIn.Error()) } diff --git a/utils/compression.go b/utils/compression.go deleted file mode 100644 index 6150f28c19..0000000000 --- a/utils/compression.go +++ /dev/null @@ -1,41 +0,0 @@ -package utils - -import ( - "bytes" - "compress/gzip" - "encoding/base64" - "fmt" - "io" -) - -func Gzip64Encode(data []byte) (string, error) { - var compressedBuffer bytes.Buffer - gzipWriter := gzip.NewWriter(&compressedBuffer) - if _, err := gzipWriter.Write(data); err != nil { - return "", fmt.Errorf("gzip data: %v", err) - } - if err := gzipWriter.Close(); err != nil { - return "", fmt.Errorf("close gzip writer: %v", err) - } - return base64.StdEncoding.EncodeToString(compressedBuffer.Bytes()), nil -} - -func Gzip64Decode(data string) ([]byte, error) { - decodedBytes, err := base64.StdEncoding.DecodeString(data) - if err != nil { - return nil, err - } - gzipReader, err := gzip.NewReader(bytes.NewReader(decodedBytes)) - if err != nil { - return nil, err - } - decompressedBytes, err := io.ReadAll(gzipReader) - if err != nil { - return nil, err - } - err = gzipReader.Close() - if err != nil { - return nil, err - } - return decompressedBytes, nil -} diff --git a/utils/compression/compression.go b/utils/compression/compression.go new file mode 100644 index 0000000000..b2ddb182b5 --- /dev/null +++ b/utils/compression/compression.go @@ -0,0 +1,158 @@ +package compression + +import ( + "bytes" + "compress/gzip" + "encoding/base64" + "errors" + "fmt" + "io" + "sync" +) + +// All gzip compression levels +const ( + minLevel = gzip.HuffmanOnly + maxLevel = gzip.BestCompression + levelCount = maxLevel - minLevel + 1 +) + +var ErrWriterNotAcquired = errors.New("using writer after release") + +// gzipWriterPools holds one pool per compression level. +var gzipWriterPools *[levelCount]sync.Pool = func() *[levelCount]sync.Pool { + pool := [levelCount]sync.Pool{} + + for i := range levelCount { + pool[i].New = func() any { return newWriter(minLevel + i) } + } + + return &pool +}() + +// proxy exists to solve the problem of putting a [Writer] `w` back to the pool +// (via [Writer.Release]) while `w` still references live data that is not going to be used +// again. There is the option to drop reference with `w.gz.Reset(io.Discard)` +// but it's expensive (cost in the micro-seconds). +// With `proxy` we create a middle pointer to which `w.gz` will point. +// `proxy` in turn will point to the actual data. +// When releasing a [Writer] it is enough for the proxy to drop the reference. +// When acquired: [Writer.gz] ---> [Writer.proxy] ---> data +// After release: [Writer.gz] ---> [Writer.proxy] ---> nil +type proxy struct { + dst io.Writer +} + +func (d *proxy) Write(p []byte) (int, error) { + if d.dst == nil { + return 0, ErrWriterNotAcquired + } + return d.dst.Write(p) +} + +// Writer is a pooled gzip writer. Its gzip writer is permanently wired to +// proxy, which forwards writes to the caller's dst while the writer is +// acquired. The indirection lets Release detach the caller's destination in +// O(1) instead of paying a second flate reset just to re-point the gzip +// writer at io.Discard. +type Writer struct { + gz *gzip.Writer + proxy proxy // dst of gz + pool *sync.Pool // pool this writer belongs to; nil once released +} + +func newWriter(level int) *Writer { + writer := &Writer{} + gzipWriter, err := gzip.NewWriterLevel(&writer.proxy, level) + if err != nil { + panic(fmt.Sprintf("creating new gzip writer for level %d: %v", level, err)) + } + writer.gz = gzipWriter + return writer +} + +func (w *Writer) Write(p []byte) (int, error) { + if !w.isAcquired() { + return 0, ErrWriterNotAcquired + } + return w.gz.Write(p) +} + +func (w *Writer) Close() error { + if !w.isAcquired() { + return ErrWriterNotAcquired + } + return w.gz.Close() +} + +func (w *Writer) Flush() error { + if !w.isAcquired() { + return ErrWriterNotAcquired + } + return w.gz.Flush() +} + +// Release returns the writer to the pool +func (w *Writer) Release() { + if !w.isAcquired() { + panic("re-releasing writer") + } + pool := w.pool + w.pool = nil + w.proxy.dst = nil + pool.Put(w) +} + +func (w *Writer) isAcquired() bool { + return w.pool != nil +} + +// GzipWriter returns a gzip writer reset onto `dst`, compressing at the default +// level. Once used, it should be sent back to the pool via `Release` +func GzipWriter(dst io.Writer) *Writer { + return GzipWriterLevel(dst, gzip.DefaultCompression) +} + +// GzipWriterLevel returns a gzip writer reset onto `dst`, compressing at `level`. +// Once used, it should be sent back to the pool via `Release` +func GzipWriterLevel(dst io.Writer, level int) *Writer { + pool := &gzipWriterPools[level-minLevel] + writer := pool.Get().(*Writer) + writer.pool = pool + writer.proxy.dst = dst + writer.gz.Reset(&writer.proxy) + return writer +} + +func Gzip64Encode(data []byte) (string, error) { + var compressedBuffer bytes.Buffer + gzipWriter := GzipWriter(&compressedBuffer) + defer gzipWriter.Release() + if _, err := gzipWriter.Write(data); err != nil { + return "", fmt.Errorf("writing data with gzip: %w", err) + } + if err := gzipWriter.Close(); err != nil { + return "", fmt.Errorf("closing gzip writer: %w", err) + } + return base64.StdEncoding.EncodeToString(compressedBuffer.Bytes()), nil +} + +func Gzip64Decode(data string) ([]byte, error) { + decodedBytes, err := base64.StdEncoding.DecodeString(data) + if err != nil { + return nil, err + } + gzipReader, err := gzip.NewReader(bytes.NewReader(decodedBytes)) + if err != nil { + return nil, err + } + decompressedBytes, err := io.ReadAll(gzipReader) + if err != nil { + return nil, err + } + err = gzipReader.Close() + if err != nil { + return nil, err + } + return decompressedBytes, nil +} diff --git a/utils/compression/compression_bench_test.go b/utils/compression/compression_bench_test.go new file mode 100644 index 0000000000..dd0919a413 --- /dev/null +++ b/utils/compression/compression_bench_test.go @@ -0,0 +1,81 @@ +package compression_test + +import ( + "bytes" + "strconv" + "testing" + + "github.com/NethermindEth/juno/utils/compression" +) + +var benchSizes = []int{16 << 10, 256 << 10, 2 << 20} + +func programLike(size int) []byte { + tokens := []string{ + `{"prime":"0x800000000000011000000000000000000000000000000000000000000000001",`, + `"data":["0x480680017fff8000","0x1","0x48127ffe7fff8000","0x208b7fff7fff7ffe"],`, + `"attributes":[],"debug_info":null,"builtins":["range_check","pedersen"],`, + `"hints":{"0":[{"accessible_scopes":["starkware.cairo.common.math"]}]},`, + } + + var buf bytes.Buffer + buf.Grow(size) + for i := 0; buf.Len() < size; i++ { + buf.WriteString(tokens[i%len(tokens)]) + } + return buf.Bytes()[:size] +} + +func BenchmarkGzip64Encode(b *testing.B) { + for _, size := range benchSizes { + data := programLike(size) + b.Run(strconv.Itoa(size>>10)+"KiB", func(b *testing.B) { + b.ReportAllocs() + b.SetBytes(int64(size)) + for b.Loop() { + if _, err := compression.Gzip64Encode(data); err != nil { + b.Fatal(err) + } + } + }) + } +} + +func BenchmarkGzip64Decode(b *testing.B) { + for _, size := range benchSizes { + encoded, err := compression.Gzip64Encode(programLike(size)) + if err != nil { + b.Fatal(err) + } + b.Run(strconv.Itoa(size>>10)+"KiB", func(b *testing.B) { + b.ReportAllocs() + b.SetBytes(int64(size)) + for b.Loop() { + if _, err := compression.Gzip64Decode(encoded); err != nil { + b.Fatal(err) + } + } + }) + } +} + +// GzipWriter is also used directly by callers that stream into a destination +// rather than building a base64 string, e.g. the JSON-RPC HTTP handler. +func BenchmarkGzipWriter(b *testing.B) { + data := programLike(256 << 10) + var sink bytes.Buffer + + b.ReportAllocs() + b.SetBytes(int64(len(data))) + for b.Loop() { + sink.Reset() + gzipWriter := compression.GzipWriter(&sink) + if _, err := gzipWriter.Write(data); err != nil { + b.Fatal(err) + } + if err := gzipWriter.Close(); err != nil { + b.Fatal(err) + } + gzipWriter.Release() + } +} diff --git a/utils/compression/compression_test.go b/utils/compression/compression_test.go new file mode 100644 index 0000000000..d9d1d06671 --- /dev/null +++ b/utils/compression/compression_test.go @@ -0,0 +1,212 @@ +package compression_test + +import ( + "bytes" + "compress/gzip" + "errors" + "io" + "runtime" + "strconv" + "sync" + "testing" + "weak" + + "github.com/NethermindEth/juno/utils/compression" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestGzip64(t *testing.T) { + bytes := []byte{0} + expectedComBytes := "H4sIAAAAAAAA/2IABAAA//+N7wLSAQAAAA==" + comBytes, err := compression.Gzip64Encode(bytes) + require.NoError(t, err) + assert.Equal(t, expectedComBytes, comBytes) + + decompBytes, err := compression.Gzip64Decode(comBytes) + require.NoError(t, err) + assert.Equal(t, bytes, decompBytes) +} + +func FuzzGzip64(f *testing.F) { + f.Fuzz(func(t *testing.T, data []byte) { + compressed, err := compression.Gzip64Encode(data) + require.NoError(t, err) + decompressed, err := compression.Gzip64Decode(compressed) + require.NoError(t, err) + assert.Equal(t, data, decompressed) + }) +} + +// Successive encodes draw the same writer back out of the pool. The sizes below +// are deliberately not monotonic: a writer that carried state over from the +// previous encode would trail bytes of a longer payload into a shorter one, which +// a run of same-or-growing sizes would hide. +func TestGzip64EncodeAcrossSuccessiveCalls(t *testing.T) { + for _, size := range []int{4096, 2048, 64, 8192, 1} { + t.Run(strconv.Itoa(size), func(t *testing.T) { + payload := bytes.Repeat([]byte("a"), size) + + encoded, err := compression.Gzip64Encode(payload) + require.NoError(t, err) + decoded, err := compression.Gzip64Decode(encoded) + require.NoError(t, err) + assert.Equal(t, payload, decoded) + }) + } +} + +type failingWriter struct{} + +func (failingWriter) Write([]byte) (int, error) { + return 0, errors.New("destination unavailable") +} + +// A writer whose destination failed is still safe to hand back: the next caller +// gets output identical to what an untouched writer produces. The expected value +// comes from gzipAtLevel rather than Gzip64Encode so the oracle cannot itself be +// tainted by the pool under test. +func TestGzipWriterAfterFailedDestination(t *testing.T) { + const repeats = 4096 + payload := bytes.Repeat([]byte("compress me "), repeats) + + poisoned := compression.GzipWriter(failingWriter{}) + _, writeErr := poisoned.Write(payload) + closeErr := poisoned.Close() + require.Error(t, errors.Join(writeErr, closeErr), "failing destination must fault the writer") + poisoned.Release() + + var buf bytes.Buffer + reused := compression.GzipWriter(&buf) + _, err := reused.Write(payload) + require.NoError(t, err) + require.NoError(t, reused.Close()) + reused.Release() + + assert.Equal(t, gzipAtLevel(t, payload, gzip.DefaultCompression), buf.Bytes()) +} + +// After Release the pooled writer must hold no reference to the caller's +// destination: the destination has to be collectable while the writer lives on +// in the pool. Holding gzipWriter across the GC is deliberate — it pins the +// writer so the assertion is about the dst reference, not the writer itself +// being collected. +func TestGzipWriterReleaseDropsDestination(t *testing.T) { + dst := &bytes.Buffer{} + gzipWriter := compression.GzipWriter(dst) + _, err := gzipWriter.Write([]byte("payload")) + require.NoError(t, err) + require.NoError(t, gzipWriter.Close()) + + weakDst := weak.Make(dst) + gzipWriter.Release() + dst = nil //nolint:ineffassign // drop the last strong reference so GC can collect the buffer + runtime.GC() + assert.Nil(t, weakDst.Value(), "released writer still references its destination") + runtime.KeepAlive(gzipWriter) +} + +// Releasing twice would hand the same writer to two callers at once, so the +// second Release must fail loudly rather than corrupt a concurrent caller. +func TestGzipWriterDoubleReleasePanics(t *testing.T) { + gzipWriter := compression.GzipWriter(io.Discard) + gzipWriter.Release() + assert.Panics(t, func() { gzipWriter.Release() }) +} + +// A released writer may already be owned by another goroutine, so using it must +// be refused rather than silently corrupt the new owner's output. +func TestGzipWriterUseAfterReleaseErrors(t *testing.T) { + gzipWriter := compression.GzipWriter(io.Discard) + gzipWriter.Release() + + _, err := gzipWriter.Write([]byte("stale")) + assert.ErrorIs(t, err, compression.ErrWriterNotAcquired) + assert.ErrorIs(t, gzipWriter.Close(), compression.ErrWriterNotAcquired) + assert.ErrorIs(t, gzipWriter.Flush(), compression.ErrWriterNotAcquired) +} + +// gzipAtLevel is the reference encoding: a writer built at `level` and used once, +// so no pool can influence the result. +func gzipAtLevel(t *testing.T, data []byte, level int) []byte { + t.Helper() + + var buf bytes.Buffer + gzipWriter, err := gzip.NewWriterLevel(&buf, level) + require.NoError(t, err) + _, err = gzipWriter.Write(data) + require.NoError(t, err) + require.NoError(t, gzipWriter.Close()) + return buf.Bytes() +} + +// Each level keeps its own writers. Returning a writer at one level must never +// let it come back out at another, which would silently compress at the wrong +// level: the only symptom is a differently sized output, never an error. +func TestGzipWriterLevelsDoNotMix(t *testing.T) { + const repeats = 8192 + payload := bytes.Repeat([]byte("mixed levels must not share writers "), repeats) + + // HuffmanOnly and NoCompression are included because they take a different + // reset path inside flate than the levels that build a match chain. + levels := []int{ + gzip.HuffmanOnly, + gzip.NoCompression, + gzip.BestSpeed, + gzip.DefaultCompression, + gzip.BestCompression, + } + + // Cycle the levels so every pool has been drawn from and returned to before the + // assertions below, giving a misfiled writer the chance to surface. + for range 2 { + for _, level := range levels { + compression.GzipWriterLevel(io.Discard, level).Release() + } + } + + for _, level := range levels { + t.Run(strconv.Itoa(level), func(t *testing.T) { + var buf bytes.Buffer + gzipWriter := compression.GzipWriterLevel(&buf, level) + _, err := gzipWriter.Write(payload) + require.NoError(t, err) + require.NoError(t, gzipWriter.Close()) + gzipWriter.Release() + + assert.Equal(t, gzipAtLevel(t, payload, level), buf.Bytes()) + }) + } + + // Guard the premise: these levels really do produce different output, so the + // assertions above could actually fail if writers were shared. + fastest := gzipAtLevel(t, payload, gzip.BestSpeed) + smallest := gzipAtLevel(t, payload, gzip.BestCompression) + require.NotEqual(t, len(fastest), len(smallest), "levels must be distinguishable") +} + +func TestGzipWriterLevelRejectsOutOfRange(t *testing.T) { + assert.Panics(t, func() { compression.GzipWriterLevel(io.Discard, gzip.BestCompression+1) }) + assert.Panics(t, func() { compression.GzipWriterLevel(io.Discard, gzip.HuffmanOnly-1) }) +} + +// Concurrent callers must not end up sharing a writer. Run under -race. +func TestGzip64EncodeConcurrent(t *testing.T) { + const ( + goroutines = 16 + chunk = 512 + ) + + var wg sync.WaitGroup + for i := range goroutines { + wg.Go(func() { + payload := bytes.Repeat([]byte{byte('a' + i)}, chunk*(i+1)) + encoded, err := compression.Gzip64Encode(payload) + assert.NoError(t, err) + decoded, err := compression.Gzip64Decode(encoded) + assert.NoError(t, err) + assert.Equal(t, payload, decoded) + }) + } + wg.Wait() +} diff --git a/utils/compression_test.go b/utils/compression_test.go deleted file mode 100644 index 2aacb58c95..0000000000 --- a/utils/compression_test.go +++ /dev/null @@ -1,31 +0,0 @@ -package utils_test - -import ( - "testing" - - "github.com/NethermindEth/juno/utils" - "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" -) - -func TestGzip64(t *testing.T) { - bytes := []byte{0} - expectedComBytes := "H4sIAAAAAAAA/2IABAAA//+N7wLSAQAAAA==" - comBytes, err := utils.Gzip64Encode(bytes) - require.NoError(t, err) - assert.Equal(t, comBytes, expectedComBytes) - - decompBytes, err := utils.Gzip64Decode(comBytes) - require.NoError(t, err) - assert.Equal(t, bytes, decompBytes) -} - -func FuzzGzip64(f *testing.F) { - f.Fuzz(func(t *testing.T, data []byte) { - compressed, err := utils.Gzip64Encode(data) - require.NoError(t, err) - decompressed, err := utils.Gzip64Decode(compressed) - require.NoError(t, err) - assert.Equal(t, data, decompressed) - }) -}